[FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime #21586

swuferhong · 2023-01-03T03:49:44Z

What is the purpose of the change

This pr is aims to introduce adaptive local hash aggregate in batch mode to adaptively determine whether local hash aggregate is required according to the aggregation degree of local hash aggregate. adaptive local hash agg will first sampling the data, and then decide whether local hash agg is required according to the degree of aggregation of the sampled data. If requiring local hash agg, it will not take any action. If not required, we will transfer the output rowType of data to same with aggregated results and pass to downstream.

Brief change log

Add codeGen code in HashAggCodeGenerator,
Add ITCase in HashAggITCase.

Verifying this change

Add ITCase in HashAggITCase.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? no document now.

flinkbot · 2023-01-03T03:52:35Z

CI report:

7c472c7 Azure: SUCCESS
1669e00 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

lsyldliu

@swuferhong Thanks for contribution, I left some comments.

...link-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

...la/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalHashAggregate.scala

...r/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala

...link-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

...-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala

...r/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala

swuferhong · 2023-01-15T03:58:38Z

@flinkbot run azure

swuferhong · 2023-01-16T15:54:57Z

@flinkbot run azure

lsyldliu

@swuferhong Thanks for contribution, I left some comments.

...r/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala

...-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala

lsyldliu · 2023-01-19T02:33:59Z

...-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala

+      aggInfo =>
+        aggInfo.function match {
+          case _: SumAggFunction =>
+            fieldExprs += GenerateUtils.generateFieldAccess(


Assuming the sum field type is bigint, if some element of the field value is null, we emit the result to downstream global agg is default value -1L, whether the final is corrected?

...nner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/HashAggITCase.scala

godfreyhe

Thanks for the contribution, I left some comments

...c/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecHashAggregate.java

...r/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala

godfreyhe · 2023-01-19T03:12:51Z

...scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalHashAggRule.scala

@@ -69,7 +69,7 @@ class EnforceLocalHashAggRule
    val agg: BatchPhysicalHashAggregate = call.rel(0)
    val expand: BatchPhysicalExpand = call.rel(2)

-    val localAgg = createLocalAgg(agg, expand)
+    val localAgg = createLocalAgg(agg, expand, true)


why supportAdaptiveLocalHashAgg is always true

...scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRule.scala

...nner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/agg/HashAggITCase.scala

…to adaptively determine whether local hash aggregate is required at runtime

swuferhong · 2023-01-30T09:16:40Z

@flinkbot run azure

swuferhong · 2023-01-30T11:42:44Z

@flinkbot run azure

swuferhong · 2023-01-30T14:07:25Z

@flinkbot run azure

swuferhong · 2023-01-30T15:21:58Z

@flinkbot run azure

swuferhong · 2023-01-31T00:59:50Z

@flinkbot run azure

godfreyhe

LGTM

…to adaptively determine whether local hash aggregate is required at runtime This closes apache#21586

flinkbot added the component=Connectors/ElasticSearch label Jan 3, 2023

flinkbot added component=TableSQL/Runtime and removed component=Connectors/ElasticSearch labels Jan 3, 2023

swuferhong force-pushed the FLINK-30542 branch 2 times, most recently from 78fb096 to 968c4cb Compare January 10, 2023 01:24

lsyldliu reviewed Jan 10, 2023

View reviewed changes

swuferhong force-pushed the FLINK-30542 branch from 968c4cb to fe3ddc2 Compare January 11, 2023 13:57

lsyldliu reviewed Jan 11, 2023

View reviewed changes

swuferhong force-pushed the FLINK-30542 branch from aa0d21b to 45efc3e Compare January 16, 2023 11:38

lsyldliu reviewed Jan 19, 2023

View reviewed changes

godfreyhe reviewed Jan 19, 2023

View reviewed changes

swuferhong added 3 commits January 28, 2023 13:36

[FLINK-30542][table-runtime] Introduce adaptive local hash aggregate …

6373910

…to adaptively determine whether local hash aggregate is required at runtime

Adding some comments and rename some variable

37a393f

Adding tests and parameters

7bbfbb3

swuferhong force-pushed the FLINK-30542 branch from 45efc3e to 60b3cfe Compare January 28, 2023 05:37

swuferhong force-pushed the FLINK-30542 branch 2 times, most recently from a7eb550 to 8d3265c Compare January 29, 2023 01:44

Rename some variables and add more tests

409cea5

swuferhong force-pushed the FLINK-30542 branch 2 times, most recently from 224964e to ab48c32 Compare January 30, 2023 08:20

Convert value type to agg func target type for sumAggFunc

50ad6e1

swuferhong force-pushed the FLINK-30542 branch from ab48c32 to 50ad6e1 Compare January 30, 2023 08:22

Small fix

7c472c7

swuferhong force-pushed the FLINK-30542 branch from ff0095b to 7c472c7 Compare January 30, 2023 14:06

godfreyhe approved these changes Jan 31, 2023

View reviewed changes

small change

1669e00

godfreyhe closed this in 122ba8f Jan 31, 2023

akkinenivijay pushed a commit to krisnaru/flink that referenced this pull request Feb 11, 2023

[FLINK-30542][table-planner] Introduce adaptive local hash aggregate …

24c91d8

…to adaptively determine whether local hash aggregate is required at runtime This closes apache#21586

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime #21586

[FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime #21586

swuferhong commented Jan 3, 2023 •

edited

flinkbot commented Jan 3, 2023 •

edited

lsyldliu left a comment

swuferhong commented Jan 15, 2023

swuferhong commented Jan 16, 2023

lsyldliu left a comment

lsyldliu Jan 19, 2023

godfreyhe left a comment

godfreyhe Jan 19, 2023

swuferhong commented Jan 30, 2023

swuferhong commented Jan 30, 2023

swuferhong commented Jan 30, 2023

swuferhong commented Jan 30, 2023

swuferhong commented Jan 31, 2023

godfreyhe left a comment

[FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime #21586

[FLINK-30542][table-runtime] Introduce adaptive local hash aggregate to adaptively determine whether local hash aggregate is required at runtime #21586

Conversation

swuferhong commented Jan 3, 2023 • edited

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jan 3, 2023 • edited

CI report:

lsyldliu left a comment

Choose a reason for hiding this comment

swuferhong commented Jan 15, 2023

swuferhong commented Jan 16, 2023

lsyldliu left a comment

Choose a reason for hiding this comment

lsyldliu Jan 19, 2023

Choose a reason for hiding this comment

godfreyhe left a comment

Choose a reason for hiding this comment

godfreyhe Jan 19, 2023

Choose a reason for hiding this comment

swuferhong commented Jan 30, 2023

swuferhong commented Jan 30, 2023

swuferhong commented Jan 30, 2023

swuferhong commented Jan 30, 2023

swuferhong commented Jan 31, 2023

godfreyhe left a comment

Choose a reason for hiding this comment

swuferhong commented Jan 3, 2023 •

edited

flinkbot commented Jan 3, 2023 •

edited