[FLINK-20895] [flink-table-planner] support local aggregate push down in table planner #17344

iyupeng · 2021-09-23T15:26:22Z

What is the purpose of the change

This pull request follows #14894 which has been inactive for months.
This pull request addresses review comments in the existing pull request and is based on the existing one.
All the code changes are targeted to make it possible to push down local aggregations from computing layer to storage layer for better computing & I/O performance.

Brief change log

Add a new configuration of table.optimizer.source.aggregate-pushdown-enabled to control the aggregation push down.
Add an abstract class of PushLocalAggIntoScanRuleBase for supporting both with and without sort operator.
Add a new rule of PushLocalSortAggIntoScanRule for sort aggregation case.
Add a new rule of PushLocalSortAggWithSortIntoScanRule for sort aggregation with sort case.
Add a new rule of PushLocalSortAggWithSortAndCalcIntoScanRule for sort aggregation with sort & calc case.
Add a new rule of PushLocalHashAggIntoScanRule for hash aggregation case.
Add a new rule of PushLocalHashAggWithCalcIntoScanRule for hash aggregation with calc case.
Add new push down spec AggregatePushDownSpec to serialize/deserialize aggregations.
Implement the SupportsAggregatePushDown interface for TestValuesScanTableSource.
Some other unit tests and integration tests.

Verifying this change

This change added tests and can be verified as follows:

Added unit test class PushLocalAggIntoTableSourceScanRuleTest for plan verifications
Added integration test class LocalAggregatePushDownITCase for end-to-end functional test

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? (docs)

flinkbot · 2021-09-23T15:29:59Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 479edc3 (Thu Sep 23 15:29:59 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!
This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2021-09-23T16:06:24Z

CI report:

aec0bc1 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

...a/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java

...k-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java

...rc/main/java/org/apache/flink/table/planner/plan/abilities/source/AggregatePushDownSpec.java

...k-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java

...a/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java

iyupeng · 2021-09-29T02:34:28Z

@wuchong @godfreyhe Please take a look, thanks a lot.

godfreyhe · 2021-10-20T09:34:49Z

@iyupeng sorry for the late response, I start to review it now

godfreyhe

Thanks for the contribution @iyupeng , I left some comment

...a/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java

...a/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java

...e/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithoutSortIntoScanRule.java

...e-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java

...e/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.java

iyupeng · 2021-10-22T00:15:17Z

@godfreyhe Thanks for your comments. I plan to push new commits in a few days.

godfreyhe

Thanks for the quick update @iyupeng, I left some comments about your changes

...rc/main/java/org/apache/flink/table/planner/plan/abilities/source/AggregatePushDownSpec.java

.../java/org/apache/flink/table/planner/runtime/batch/sql/agg/LocalAggregatePushDownITCase.java

...a/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java

...ache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java

...ink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java

...e-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java

...e/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.java

.../java/org/apache/flink/table/planner/runtime/batch/sql/agg/LocalAggregatePushDownITCase.java

godfreyhe · 2021-10-28T14:44:35Z

.../flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala

+  // Disable push down aggregates to avoid conflicts with existing test cases that verify plans.
+  tableEnv.getConfig
+    .getConfiguration
+    .setBoolean(
+      OptimizerConfigOptions.TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED, false)


Currently, this pr do not change all tests except you added, so remove it ?

Hello @godfreyhe, thanks for the comments.

2 existing test cases with local aggregate and 'connector' = 'values' can be affected if we enable aggregate push down:

RankTest.scala#L172, testCreateViewWithRowNumber

TableSourceTest.scala#L118, testProjectWithoutInputRef

We need to change the plans of these 2 test cases in .xml files if we remove above configuration.

pushed a new commit 2ee07dc to solve this.

godfreyhe

LGTM overall except one minor comment

godfreyhe

Thanks for the quick update, LGTM

twalthr · 2021-11-01T09:02:29Z

...c/main/java/org/apache/flink/table/connector/source/abilities/SupportsAggregatePushDown.java

 * <p>Regardless if this interface is implemented or not, a final aggregation is always applied in a
 * subsequent operation after the source.
- *
- * <p>Note: currently, the {@link SupportsAggregatePushDown} is not supported by planner.


@iyupeng are we sure that the JavaDocs don't need more updates for this PR?

For example, is this description still correct?

* <p>Note: The local aggregate push down strategy is all or nothing, it can only be pushed down if * all aggregate functions are supported.

Also maybe we should mention the config option that was added in this PR.

Hi @twalthr, the strategy is still all or nothing here. applyAggregates returns a bool value, meaning whether all local aggregates are accepted by underlying data source or not at all.

The docs could be improved, like mentioning the new option, etc.

I could push a hotfix for docs.

That would be great. The more information in the interface description, the better for all implementers.

Hi @twalthr I pushed this for doc improvement: #17630

…le planner Co-authored-by: Sebastian Liu <liuyang0704@gmail.com> This closes apache#17344

rmetzger added review=description? component=TableSQL/Planner labels Sep 23, 2021

Airblader reviewed Sep 23, 2021

View reviewed changes

iyupeng force-pushed the FLINK-20895 branch from 70104a7 to 4ad2900 Compare September 25, 2021 02:28

godfreyhe reviewed Oct 20, 2021

View reviewed changes

godfreyhe reviewed Oct 28, 2021

View reviewed changes

sebastianliu and others added 7 commits October 29, 2021 15:00

[FLINK-20895] support local aggregate push down in blink planner

95a1865

[FLINK-20895] make changes according to review comments

f468eca

[FLINK-20895] code and comment optimizations

4ceb389

[FLINK-20895] fix test errors in CI pipelines

1dbeace

[FLINK-20895] add more rules and tests for local agg push down

289a257

[FLINK-20895] optimize code and add test cases for auxGrouping

0b0584f

[FLINK-20895] enable push down local aggregate for existing unit tests

aec0bc1

iyupeng force-pushed the FLINK-20895 branch from 2ee07dc to aec0bc1 Compare October 29, 2021 07:01

godfreyhe approved these changes Oct 31, 2021

View reviewed changes

godfreyhe closed this in 34d581b Oct 31, 2021

twalthr reviewed Nov 1, 2021

View reviewed changes

niklassemmler pushed a commit to niklassemmler/flink that referenced this pull request Feb 3, 2022

[FLINK-20895][table-planner] Support local aggregate push down in tab…

c5667a6

…le planner Co-authored-by: Sebastian Liu <liuyang0704@gmail.com> This closes apache#17344

[FLINK-20895] [flink-table-planner] support local aggregate push down in table planner #17344

[FLINK-20895] [flink-table-planner] support local aggregate push down in table planner #17344

Uh oh!

Conversation

iyupeng commented Sep 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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

Uh oh!

flinkbot commented Sep 23, 2021

Automated Checks

Review Progress

Uh oh!

flinkbot commented Sep 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

iyupeng commented Sep 29, 2021

Uh oh!

godfreyhe commented Oct 20, 2021

Uh oh!

godfreyhe left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

iyupeng commented Oct 22, 2021

Uh oh!

godfreyhe left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

godfreyhe Oct 28, 2021

Choose a reason for hiding this comment

Uh oh!

iyupeng Oct 29, 2021

Choose a reason for hiding this comment

Uh oh!

iyupeng Oct 29, 2021

Choose a reason for hiding this comment

Uh oh!

godfreyhe left a comment

Choose a reason for hiding this comment

Uh oh!

godfreyhe left a comment

Choose a reason for hiding this comment

Uh oh!

twalthr Nov 1, 2021

Choose a reason for hiding this comment

Uh oh!

iyupeng Nov 1, 2021

Choose a reason for hiding this comment

Uh oh!

twalthr Nov 1, 2021

Choose a reason for hiding this comment

iyupeng commented Sep 23, 2021 •

edited

Loading

flinkbot commented Sep 23, 2021 •

edited

Loading