ObjectHashAggregateExec Physical Operator

Author: jaceklaskowski

docs/AggUtils.md

@@ -60,6 +60,51 @@ createAggregate(
 
 1. [SortAggregateExec](physical-operators/SortAggregateExec.md)
 
+---
+
 `createAggregate` is used when:
 
-* `AggUtils` is used to [planAggregateWithoutDistinct](#planAggregateWithoutDistinct), [planAggregateWithOneDistinct](#planAggregateWithOneDistinct), and `planStreamingAggregation`
+* `AggUtils` is used to [createStreamingAggregate](#createStreamingAggregate), [planAggregateWithoutDistinct](#planAggregateWithoutDistinct), [planAggregateWithOneDistinct](#planAggregateWithOneDistinct)
+
+## <span id="planStreamingAggregation"> Planning Execution of Streaming Aggregation
+
+```scala
+planStreamingAggregation(
+  groupingExpressions: Seq[NamedExpression],
+  functionsWithoutDistinct: Seq[AggregateExpression],
+  resultExpressions: Seq[NamedExpression],
+  stateFormatVersion: Int,
+  child: SparkPlan): Seq[SparkPlan]
+```
+
+`planStreamingAggregation`...FIXME
+
+---
+
+`planStreamingAggregation` is used when:
+
+* `StatefulAggregationStrategy` ([Spark Structured Streaming]({{ book.structured_streaming }}/StatefulAggregationStrategy)) execution planning strategy is requested to plan a logical plan of a streaming aggregation (a streaming query with [Aggregate](logical-operators/Aggregate.md) operator)
+
+## <span id="createStreamingAggregate"> Creating Streaming Aggregate Physical Operator
+
+```scala
+createStreamingAggregate(
+  requiredChildDistributionExpressions: Option[Seq[Expression]] = None,
+  groupingExpressions: Seq[NamedExpression] = Nil,
+  aggregateExpressions: Seq[AggregateExpression] = Nil,
+  aggregateAttributes: Seq[Attribute] = Nil,
+  initialInputBufferOffset: Int = 0,
+  resultExpressions: Seq[NamedExpression] = Nil,
+  child: SparkPlan): SparkPlan
+```
+
+`createStreamingAggregate` [creates an aggregate physical operator](#createAggregate) (with `isStreaming` flag enabled).
+
+!!! note
+    `createStreamingAggregate` is exactly [createAggregate](#createAggregate) with `isStreaming` flag enabled.
+
+---
+
+`createStreamingAggregate` is used when:
+
+* `AggUtils` is requested to plan a [regular](#planStreamingAggregation) and [session-windowed](#planStreamingAggregationForSession) streaming aggregation

docs/ObjectAggregationIterator.md

@@ -16,7 +16,7 @@
 * <span id="newMutableProjection"> Function to create a new `MutableProjection` given expressions and attributes (`(Seq[Expression], Seq[Attribute]) => MutableProjection`)
 * <span id="originalInputAttributes"> Original Input [Attribute](expressions/Attribute.md)s
 * <span id="inputRows"> Input [InternalRow](InternalRow.md)s
-* <span id="fallbackCountThreshold"> `fallbackCountThreshold`
+* <span id="fallbackCountThreshold"> [spark.sql.objectHashAggregate.sortBased.fallbackThreshold](configuration-properties.md#spark.sql.objectHashAggregate.sortBased.fallbackThreshold)
 * <span id="numOutputRows"> `numOutputRows` [SQLMetric](physical-operators/SQLMetric.md)
 
 `ObjectAggregationIterator` is created when:

docs/SQLConf.md

@@ -713,6 +713,10 @@ Used when:
 
 * [OptimizeSkewedJoin](physical-optimizations/OptimizeSkewedJoin.md) physical optimization is executed
 
+## <span id="objectAggSortBasedFallbackThreshold"><span id="OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD"> objectAggSortBasedFallbackThreshold
+
+[spark.sql.objectHashAggregate.sortBased.fallbackThreshold](configuration-properties.md#spark.sql.objectHashAggregate.sortBased.fallbackThreshold)
+
 ## <span id="offHeapColumnVectorEnabled"> offHeapColumnVectorEnabled
 
 [spark.sql.columnVector.offheap.enabled](configuration-properties.md#spark.sql.columnVector.offheap.enabled)

docs/configuration-properties.md

@@ -50,6 +50,14 @@ Since: `3.2.0`
 
 Use [SQLConf.ADAPTIVE_CUSTOM_COST_EVALUATOR_CLASS](SQLConf.md#ADAPTIVE_CUSTOM_COST_EVALUATOR_CLASS) method to access the property (in a type-safe way).
 
+## <span id="spark.sql.objectHashAggregate.sortBased.fallbackThreshold"> spark.sql.objectHashAggregate.sortBased.fallbackThreshold
+
+**(internal)** The number of rows of an in-memory hash map (to store aggregation buffer) before [ObjectHashAggregateExec](physical-operators/ObjectHashAggregateExec.md) ([ObjectAggregationIterator](ObjectAggregationIterator.md#processInputs) precisely) falls back to sort-based aggregation
+
+Default: `128`
+
+Use [SQLConf.objectAggSortBasedFallbackThreshold](SQLConf.md#objectAggSortBasedFallbackThreshold) for the current value
+
 ## <span id="spark.sql.optimizer.decorrelateInnerQuery.enabled"> spark.sql.optimizer.decorrelateInnerQuery.enabled
 
 **(internal)** Decorrelates inner queries by eliminating correlated references and build domain joins

docs/physical-operators/ObjectHashAggregateExec.md

@@ -2,26 +2,17 @@
 
 `ObjectHashAggregateExec` is an [aggregate unary physical operator](BaseAggregateExec.md) for **object aggregation**.
 
-![ObjectHashAggregateExec in web UI (Details for Query)](../images/ObjectHashAggregateExec-webui-details-for-query.png)
-
-## <span id="supportsAggregate"> Selection Requirements
-
-```scala
-supportsAggregate(
-  aggregateExpressions: Seq[AggregateExpression]): Boolean
-```
-
-`supportsAggregate` is enabled (`true`) when there is a `TypedImperativeAggregate` aggregate function among the [AggregateFunction](../expressions/AggregateFunction.md)s of the given [AggregateExpression](../expressions/AggregateExpression.md)s.
-
-`supportsAggregate` is used when:
+`ObjectHashAggregateExec` uses [ObjectAggregationIterator](../ObjectAggregationIterator.md) for [aggregation](#doExecute) (one per partition).
 
-* `AggUtils` utility is used to [select an aggregate physical operator](../AggUtils.md#createAggregate)
+![ObjectHashAggregateExec in web UI (Details for Query)](../images/ObjectHashAggregateExec-webui-details-for-query.png)
 
 ## Creating Instance
 
 `ObjectHashAggregateExec` takes the following to be created:
 
-* <span id="requiredChildDistributionExpressions"> (optional) Required Child Distribution [Expression](../expressions/Expression.md)s
+* <span id="requiredChildDistributionExpressions"> Required Child Distribution [Expression](../expressions/Expression.md)s
+* [isStreaming](#isStreaming) flag
+* <span id="numShufflePartitions"> Number of Shuffle Partitions (always `None`)
 * <span id="groupingExpressions"> Grouping [NamedExpression](../expressions/NamedExpression.md)s
 * <span id="aggregateExpressions"> [AggregateExpression](../expressions/AggregateExpression.md)s
 * <span id="aggregateAttributes"> Aggregate [Attribute](../expressions/Attribute.md)s
@@ -31,14 +22,32 @@ supportsAggregate(
 
 `ObjectHashAggregateExec` is created when:
 
-* `AggUtils` utility is used to [create a physical operator for aggregation](../AggUtils.md#createAggregate)
+* `AggUtils` is requested to [create a physical operator for aggregation](../AggUtils.md#createAggregate)
+
+### <span id="isStreaming"> isStreaming Flag
+
+`ObjectHashAggregateExec` is given `isStreaming` flag when [created](#creating-instance).
+
+The `isStreaming` is always `false` but when `AggUtils` is requested to [create a streaming aggregate physical operator](../AggUtils.md#createStreamingAggregate).
 
 ## <span id="metrics"> Performance Metrics
 
-Key             | Name (in web UI)
-----------------|--------------------------
-numOutputRows   | number of output rows
-aggTime         | time in aggregation build
+### <span id="aggTime"> time in aggregation build
+
+The time to [doExecute](#doExecute) of a single partition.
+
+### <span id="numOutputRows"> number of output rows
+
+* `1` when there is no input rows in a partition and no [groupingExpressions](#groupingExpressions).
+* Used to create an [ObjectAggregationIterator](../ObjectAggregationIterator.md#numOutputRows).
+
+### <span id="numTasksFallBacked"> number of sort fallback tasks
+
+Used to create a [ObjectAggregationIterator](../ObjectAggregationIterator.md#numTasksFallBacked).
+
+### <span id="spillSize"> spill size
+
+Used to create a [ObjectAggregationIterator](../ObjectAggregationIterator.md#spillSize).
 
 ## <span id="doExecute"> Executing Physical Operator
 
@@ -48,9 +57,35 @@ doExecute(): RDD[InternalRow]
 
 `doExecute` is part of the [SparkPlan](SparkPlan.md#doExecute) abstraction.
 
-`doExecute` uses [ObjectAggregationIterator](../ObjectAggregationIterator.md) for aggregation (one per partition).
+---
+
+`doExecute` requests the [child physical operator](#child) to [execute](SparkPlan.md#execute) (and generate an `RDD[InternalRow]`) that is `mapPartitionsWithIndexInternal` to process partitions.
+
+!!! note
+    `doExecute` adds a new `MapPartitionsRDD` ([Spark Core]({{ book.spark_core }}/rdd/MapPartitionsRDD)) to the RDD lineage.
+
+For no input records (in a partition) and non-empty [groupingExpressions](#groupingExpressions), `doExecute` returns an empty `Iterator`.
+
+Otherwise, `doExecute` creates a [ObjectAggregationIterator](../ObjectAggregationIterator.md).
 
-`doExecute`...FIXME
+For no input records (in a partition) and no [groupingExpressions](#groupingExpressions), `doExecute` increments the [numOutputRows](#numOutputRows) metric (so it's just `1`) and requests the `ObjectAggregationIterator` for [outputForEmptyGroupingKeyWithoutInput](../ObjectAggregationIterator.md#outputForEmptyGroupingKeyWithoutInput).
+
+Otherwise, `doExecute` returns the `ObjectAggregationIterator`.
+
+## <span id="supportsAggregate"> Selection Requirements
+
+```scala
+supportsAggregate(
+  aggregateExpressions: Seq[AggregateExpression]): Boolean
+```
+
+`supportsAggregate` is enabled (`true`) when there is a `TypedImperativeAggregate` aggregate function among the [AggregateFunction](../expressions/AggregateFunction.md)s of the given [AggregateExpression](../expressions/AggregateExpression.md)s.
+
+---
+
+`supportsAggregate` is used when:
+
+* `AggUtils` utility is used to [select an aggregate physical operator](../AggUtils.md#createAggregate)
 
 ## Demo