Module org.elasticsearch.compute

Package org.elasticsearch.compute.operator

Class HashAggregationOperator

java.lang.Object

org.elasticsearch.compute.operator.HashAggregationOperator

All Implemented Interfaces:

Closeable, AutoCloseable, Operator, Releasable

Direct Known Subclasses:

TimeSeriesAggregationOperator, TimeSeriesCollapseOperator

public class HashAggregationOperator
extends Object
implements Operator

Aggregates input Pages into many output rows, grouping by some values.

Receives input pages until we call finish(), telling it there are no more pages. Each aggregation is an instance of GroupingAggregatorFunction.

For

| STATS MIN(discovered), MAX(discovered) BY class

this'd look like:

Before first page:
                                    ┌─────────────────┬─────────────────┬──────────┐
                                    │ MIN(discovered) │ MAX(discovered) │ class    │
                                    ├─────────────────┼─────────────────┼──────────┤
                                    └─────────────────┴─────────────────┴──────────┘

First page:
┌──────┬──────────┬────────────┐    ┌─────────────────┬─────────────────┬──────────┐
│  ref │ class    │ discovered │    │ MIN(discovered) │ MAX(discovered) │ class    │
├──────┼──────────┼────────────┤    ├─────────────────┼─────────────────┼──────────┤
│  079 │ Euclid   │ 1988-01-01 │ -> │ 1988-01-01      │ 1993-01-01      │ Euclid   │
│  173 │ Euclid   │ 1993-01-01 │    │ 1967-01-01      │ 1967-01-01      │ Keter    │
│ 1313 │ Keter    │ 1967-01-01 │    │ 1991-01-01      │ 1991-01-01      │ Safe     │
│ 1981 │ Safe     │ 1991-01-01 │    └─────────────────┴─────────────────┴──────────┘
└──────┴──────────┴────────────┘

Second page:
┌──────┬──────────┬────────────┐    ┌─────────────────┬─────────────────┬──────────┐
│  ref │ class    │ discovered │    │ MIN(discovered) │ MAX(discovered) │ class    │
├──────┼──────────┼────────────┤    ├─────────────────┼─────────────────┼──────────┤
│ 2317 │ Keter    │ 1922-01-01 │ -> │ 1988-01-01      │ 2010-01-01      │ Euclid   │
│ 2639 │ Euclid   │ 2010-01-01 │    │ 1922-01-01      │ 2016-04-28      │ Keter    │
│ 2935 │ Keter    │ 2016-04-28 │    │ 1991-01-01      │ 1991-01-01      │ Safe     │
│ 3000 │ Thaumiel │ 1971-01-01 │    │ 1971-01-01      │ 1971-01-01      │ Thaumiel │
└──────┴──────────┴────────────┘    └─────────────────┴─────────────────┴──────────┘

Third page:
┌──────┬──────────┬────────────┐    ┌─────────────────┬─────────────────┬──────────┐
│  ref │ class    │ discovered │    │ MIN(discovered) │ MAX(discovered) │ class    │
├──────┼──────────┼────────────┤    ├─────────────────┼─────────────────┼──────────┤
│ 3001 │ Euclid   │ 2000-01-02 │ -> │ 1988-01-01      │ 2010-01-01      │ Euclid   │
│ 4999 │ Keter    │ 1998-11-27 │    │ 1922-01-01      │ 2016-04-28      │ Keter    │
│ 5000 │ Safe     │ 2020-12-04 │    │ 1991-01-01      │ 2020-12-04      │ Safe     │
└──────┴──────────┴────────────┘    │ 1971-01-01      │ 1971-01-01      │ Thaumiel │
                                    └─────────────────┴─────────────────┴──────────┘

finish():
                                    ┌─────────────────┬─────────────────┬──────────┐
                                    │ MIN(discovered) │ MAX(discovered) │ class    │
                                    ├─────────────────┼─────────────────┼──────────┤
                                    │ 1988-01-01      │ 2010-01-01      │ Euclid   │
                                    │ 1922-01-01      │ 2016-04-28      │ Keter    │
                                    │ 1991-01-01      │ 2020-12-04      │ Safe     │
                                    │ 1971-01-01      │ 1971-01-01      │ Thaumiel │
                                    └─────────────────┴─────────────────┴──────────┘

Aggregations can also filter which rows they receive using FilteredGroupingAggregatorFunction. In ESQL that looks like:

| STATS min3 = MIN(discovered) WHERE LENGTH(ref) == 3,
        max3 = MAX(discovered) WHERE LENGTH(ref) == 3,
        min4 = MIN(discovered) WHERE LENGTH(ref) == 4
     BY class

this'd look like:

Before first page:
                                    ┌────────────┬────────────┬────────────┬──────────┐
                                    │ min3       │ max3       │ min4       │ class    │
                                    ├────────────┼────────────┼────────────┼──────────┤
                                    └────────────┴────────────┴────────────┴──────────┘

First page:
┌──────┬──────────┬────────────┐    ┌────────────┬────────────┬────────────┬──────────┐
│  ref │ class    │ discovered │    │ min3       │ max3       │ min4       │ class    │
├──────┼──────────┼────────────┤    ├────────────┼────────────┼────────────┼──────────┤
│  079 │ Euclid   │ 1988-01-01 │ -> │ 1988-01-01 │ 1993-01-01 │ null       │ Euclid   │
│  173 │ Euclid   │ 1993-01-01 │    │ null       │ null       │ 1967-01-01 │ Keter    │
│ 1313 │ Keter    │ 1967-01-01 │    │ null       │ null       │ 1991-01-01 │ Safe     │
│ 1981 │ Safe     │ 1991-01-01 │    └────────────┴────────────┴────────────┴──────────┘
└──────┴──────────┴────────────┘

Second page:
┌──────┬──────────┬────────────┐    ┌────────────┬────────────┬────────────┬──────────┐
│  ref │ class    │ discovered │    │ min3       │ max3       │ min4       │ class    │
├──────┼──────────┼────────────┤    ├────────────┼────────────┼────────────┼──────────┤
│ 2317 │ Keter    │ 1922-01-01 │ -> │ 1988-01-01 │ 1993-01-01 │ 2010-01-01 │ Euclid   │
│ 2639 │ Euclid   │ 2010-01-01 │    │ null       │ null       │ 1922-01-01 │ Keter    │
│ 2935 │ Keter    │ 2016-04-28 │    │ null       │ null       │ 1991-01-01 │ Safe     │
│ 3000 │ Thaumiel │ 1971-01-01 │    │ null       │ null       │ 1971-01-01 │ Thaumiel │
└──────┴──────────┴────────────┘    └────────────┴────────────┴────────────┴──────────┘

Third page:
┌──────┬──────────┬────────────┐    ┌────────────┬────────────┬────────────┬──────────┐
│  ref │ class    │ discovered │    │ min3       │ max3       │ min4       │ class    │
├──────┼──────────┼────────────┤    ├────────────┼────────────┼────────────┼──────────┤
│ 3001 │ Euclid   │ 2000-01-02 │ -> │ 1988-01-01 │ 1993-01-01 │ 2000-01-02 │ Euclid   │
│ 4999 │ Keter    │ 1998-11-27 │    │ null       │ null       │ 1922-01-01 │ Keter    │
│ 5000 │ Safe     │ 2020-12-04 │    │ null       │ null       │ 1991-01-01 │ Safe     │
└──────┴──────────┴────────────┘    │ null       │ null       │ 1971-01-01 │ Thaumiel │
                                    └────────────┴────────────┴────────────┴──────────┘

finish():
                                    ┌────────────┬────────────┬────────────┬──────────┐
                                    │ min3       │ max3       │ min4       │ class    │
                                    ├────────────┼────────────┼────────────┼──────────┤
                                    │ 1988-01-01 │ 1993-01-01 │ 2000-01-02 │ Euclid   │
                                    │ null       │ null       │ 1922-01-01 │ Keter    │
                                    │ null       │ null       │ 1991-01-01 │ Safe     │
                                    │ null       │ null       │ 1971-01-01 │ Thaumiel │
                                    └────────────┴────────────┴────────────┴──────────┘

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	HashAggregationOperator.Builder	Builder for HashAggregationOperator.
static class	HashAggregationOperator.Factory
static class	HashAggregationOperator.Status

Nested classes/interfaces inherited from interface org.elasticsearch.compute.operator.Operator

Operator.OperatorFactory

Field Summary

Fields

Modifier and Type	Field	Description
protected final List<GroupingAggregator.Factory>	aggregatorFactories
protected final AggregatorMode	aggregatorMode
protected final List<GroupingAggregator>	aggregators
protected BlockHash	blockHash
protected final Supplier<BlockHash>	blockHashSupplier
static final int	DEFAULT_PARTIAL_EMIT_KEYS_THRESHOLD
static final double	DEFAULT_PARTIAL_EMIT_UNIQUENESS_THRESHOLD
protected final DriverContext	driverContext
protected long	emitCount
protected long	emitNanos	Total nanos for emitting the output
protected ReleasableIterator<Page>	output
protected final int	partialEmitKeysThreshold
protected final double	partialEmitUniquenessThreshold
protected long	rowsAddedInCurrentBatch

Fields inherited from interface org.elasticsearch.compute.operator.Operator

MIN_TARGET_PAGE_SIZE, NOT_BLOCKED, TARGET_PAGE_SIZE

Constructor Summary

Constructors

Constructor	Description
HashAggregationOperator(AggregatorMode aggregatorMode, List<GroupingAggregator.Factory> aggregatorFactories, Supplier<BlockHash> blockHashSupplier, int partialEmitKeysThreshold, double partialEmitUniquenessThreshold, int maxPageSize, DriverContext driverContext)	Build the operator.

Method Summary

Modifier and Type	Method	Description
void	addInput(Page page)	adds an input page to the operator.
boolean	canProduceMoreDataWithoutExtraInput()	Returns true if the operator can produce more output pages without requiring additional input pages.
protected static void	checkState(boolean condition, String msg)
void	close()	notifies the operator that it won't be used anymore (i.e.
protected IntVector	customizeSelected(GroupingAggregator aggregator, IntVector selected)	Customize the selected groupIds that are sent to the agg's GroupingAggregatorFunction.prepareEvaluateIntermediate(org.elasticsearch.compute.data.IntVector, org.elasticsearch.compute.aggregation.GroupingAggregatorEvaluationContext) and GroupingAggregatorFunction.prepareEvaluateFinal(org.elasticsearch.compute.data.IntVector, org.elasticsearch.compute.aggregation.GroupingAggregatorEvaluationContext).
protected void	emit()
protected GroupingAggregatorEvaluationContext	evaluationContext(BlockHash blockHash)
void	finish()	notifies the operator that it won't receive any more input pages
Page	getOutput()	returns non-null if output page available.
boolean	isFinished()	whether the operator has finished processing all input pages and made the corresponding output pages available
boolean	needsInput()	whether the given operator can accept more input pages
protected boolean	shouldEmitPartialResultsPeriodically()
Operator.Status	status()	The status of the operator.
String	toString()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.elasticsearch.compute.operator.Operator

isBlocked, tryPromote

Field Details

DEFAULT_PARTIAL_EMIT_KEYS_THRESHOLD

public static final int DEFAULT_PARTIAL_EMIT_KEYS_THRESHOLD

See Also:

• Constant Field Values

DEFAULT_PARTIAL_EMIT_UNIQUENESS_THRESHOLD

public static final double DEFAULT_PARTIAL_EMIT_UNIQUENESS_THRESHOLD

See Also:

• Constant Field Values

blockHashSupplier

protected final Supplier<BlockHash> blockHashSupplier

aggregatorMode

protected final AggregatorMode aggregatorMode

aggregatorFactories

protected final List<GroupingAggregator.Factory> aggregatorFactories

aggregators

protected final List<GroupingAggregator> aggregators

partialEmitKeysThreshold

protected final int partialEmitKeysThreshold

partialEmitUniquenessThreshold

protected final double partialEmitUniquenessThreshold

driverContext

protected final DriverContext driverContext

blockHash

protected BlockHash blockHash

output

protected ReleasableIterator<Page> output

emitNanos

protected long emitNanos

Total nanos for emitting the output

emitCount

protected long emitCount

rowsAddedInCurrentBatch

protected long rowsAddedInCurrentBatch

Constructor Details

HashAggregationOperator

public HashAggregationOperator(AggregatorMode aggregatorMode,
 List<GroupingAggregator.Factory> aggregatorFactories,
 Supplier<BlockHash> blockHashSupplier,
 int partialEmitKeysThreshold,
 double partialEmitUniquenessThreshold,
 int maxPageSize,
 DriverContext driverContext)

Build the operator. Instead of calling this directly, build the HashAggregationOperator.Builder, then the HashAggregationOperator.Builder.build() then HashAggregationOperator.Factory.get(org.elasticsearch.compute.operator.DriverContext).

Method Details

needsInput

public boolean needsInput()

Description copied from interface: Operator

whether the given operator can accept more input pages

Specified by:

needsInput in interface Operator

addInput

public void addInput(Page page)

Description copied from interface: Operator

adds an input page to the operator. only called when needsInput() == true and isFinished() == false

Specified by:

addInput in interface Operator

getOutput

public Page getOutput()

Description copied from interface: Operator

returns non-null if output page available. Only called when isFinished() == false

Specified by:

getOutput in interface Operator

finish

public void finish()

Description copied from interface: Operator

notifies the operator that it won't receive any more input pages

Specified by:

finish in interface Operator

emit

protected void emit()

customizeSelected

protected IntVector customizeSelected(GroupingAggregator aggregator,
 IntVector selected)

Customize the selected groupIds that are sent to the agg's GroupingAggregatorFunction.prepareEvaluateIntermediate(org.elasticsearch.compute.data.IntVector, org.elasticsearch.compute.aggregation.GroupingAggregatorEvaluationContext) and GroupingAggregatorFunction.prepareEvaluateFinal(org.elasticsearch.compute.data.IntVector, org.elasticsearch.compute.aggregation.GroupingAggregatorEvaluationContext). TSDB uses this to do less work later on.

shouldEmitPartialResultsPeriodically

protected boolean shouldEmitPartialResultsPeriodically()

evaluationContext

protected GroupingAggregatorEvaluationContext evaluationContext(BlockHash blockHash)

isFinished

public boolean isFinished()

Description copied from interface: Operator

whether the operator has finished processing all input pages and made the corresponding output pages available

Specified by:

isFinished in interface Operator

canProduceMoreDataWithoutExtraInput

public boolean canProduceMoreDataWithoutExtraInput()

Description copied from interface: Operator

Returns true if the operator can produce more output pages without requiring additional input pages. This is useful for operators that buffer data or have internal state that can produce multiple output pages.

Operators that do not buffer data should return false - they cannot produce pages out of thin air. Examples of operators that may return true:

• Operators with internal buffers (e.g., AsyncOperator with pending results)
• Operators processing a single input page into multiple output pages
• Aggregation operators that buffer partial results

Specified by:

canProduceMoreDataWithoutExtraInput in interface Operator

Returns:

true if the operator has buffered data that can produce output, false otherwise

close

public void close()

Description copied from interface: Operator

notifies the operator that it won't be used anymore (i.e. none of the other methods called), and its resources can be cleaned up

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

Specified by:

close in interface Operator

Specified by:

close in interface Releasable

status

public Operator.Status status()

Description copied from interface: Operator

The status of the operator.

Specified by:

status in interface Operator

checkState

protected static void checkState(boolean condition,
 String msg)

toString

public String toString()

Overrides:

toString in class Object