You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2023/12/06 09:36:00 UTC
(flink) branch master updated: [FLINK-33691][planner] Support agg push down for 'count(*)/count(1)/count(column not null)' (#23828)
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3060ccd49cc [FLINK-33691][planner] Support agg push down for 'count(*)/count(1)/count(column not null)' (#23828)
3060ccd49cc is described below
commit 3060ccd49cc8d19634b431dbf0f09ac875d0d422
Author: yunhong <33...@qq.com>
AuthorDate: Wed Dec 6 17:35:52 2023 +0800
[FLINK-33691][planner] Support agg push down for 'count(*)/count(1)/count(column not null)' (#23828)
Co-authored-by: zhengyunhong.zyh <zh...@alibaba-inc.com>
---
.../batch/PushLocalAggIntoScanRuleBase.java | 36 +++--
.../batch/PushLocalHashAggIntoScanRule.java | 4 +-
.../PushLocalHashAggWithCalcIntoScanRule.java | 10 +-
.../batch/PushLocalSortAggIntoScanRule.java | 4 +-
.../PushLocalSortAggWithCalcIntoScanRule.java | 10 +-
...ushLocalSortAggWithSortAndCalcIntoScanRule.java | 10 +-
.../PushLocalSortAggWithSortIntoScanRule.java | 4 +-
.../PushLocalAggIntoTableSourceScanRuleTest.java | 22 ++-
.../sql/agg/LocalAggregatePushDownITCase.java | 42 +++++-
.../plan/batch/sql/agg/HashAggregateTest.xml | 134 ++++++++---------
.../plan/batch/sql/agg/SortAggregateTest.xml | 132 ++++++++---------
.../PushLocalAggIntoTableSourceScanRuleTest.xml | 164 +++++++++++++++------
12 files changed, 350 insertions(+), 222 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java
index 81bdaf68767..c576e5b684d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoScanRuleBase.java
@@ -45,8 +45,11 @@ import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexProgram;
+import org.apache.calcite.sql.SqlKind;
import org.apache.commons.lang3.ArrayUtils;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -76,7 +79,8 @@ public abstract class PushLocalAggIntoScanRuleBase extends RelOptRule {
protected boolean canPushDown(
RelOptRuleCall call,
BatchPhysicalGroupAggregateBase aggregate,
- BatchPhysicalTableSourceScan tableSourceScan) {
+ BatchPhysicalTableSourceScan tableSourceScan,
+ @Nullable BatchPhysicalCalc calc) {
TableConfig tableConfig = ShortcutUtils.unwrapContext(call.getPlanner()).getTableConfig();
if (!tableConfig.get(
OptimizerConfigOptions.TABLE_OPTIMIZER_SOURCE_AGGREGATE_PUSHDOWN_ENABLED)) {
@@ -86,9 +90,24 @@ public abstract class PushLocalAggIntoScanRuleBase extends RelOptRule {
if (aggregate.isFinal() || aggregate.getAggCallList().isEmpty()) {
return false;
}
+
List<AggregateCall> aggCallList =
JavaScalaConversionUtil.toJava(aggregate.getAggCallList());
for (AggregateCall aggCall : aggCallList) {
+ // We should try best to push down count(*) and count(n) even if it was optimized to a
+ // calc as '0 AS $f0' to reduce read cost.
+ if (aggCall.getAggregation().kind != SqlKind.COUNT && calc != null) {
+ if (!isInputRefOnly(calc) || !isProjectionNotPushedDown(tableSourceScan)) {
+ return false;
+ }
+ } else if (aggCall.getAggregation().kind == SqlKind.COUNT && calc != null) {
+ // count(*) push down shouldn't have filter condition upon source. For example
+ // 'select count(*) from source where a > 10' cannot push down now.
+ if (calc.getProgram().getCondition() != null) {
+ return false;
+ }
+ }
+
if (aggCall.isDistinct()
|| aggCall.isApproximate()
|| aggCall.getArgList().size() > 1
@@ -105,24 +124,23 @@ public abstract class PushLocalAggIntoScanRuleBase extends RelOptRule {
.noneMatch(spec -> spec instanceof AggregatePushDownSpec);
}
- protected void pushLocalAggregateIntoScan(
- RelOptRuleCall call,
- BatchPhysicalGroupAggregateBase localAgg,
- BatchPhysicalTableSourceScan oldScan) {
- pushLocalAggregateIntoScan(call, localAgg, oldScan, null);
- }
-
protected void pushLocalAggregateIntoScan(
RelOptRuleCall call,
BatchPhysicalGroupAggregateBase localAgg,
BatchPhysicalTableSourceScan oldScan,
- int[] calcRefFields) {
+ @Nullable BatchPhysicalCalc calc) {
RowType inputType = FlinkTypeFactory.toLogicalRowType(oldScan.getRowType());
List<int[]> groupingSets =
Collections.singletonList(
ArrayUtils.addAll(localAgg.grouping(), localAgg.auxGrouping()));
List<AggregateCall> aggCallList = JavaScalaConversionUtil.toJava(localAgg.getAggCallList());
+ // For count(*) and count(n) we need to ignore the calc.
+ int[] calcRefFields = null;
+ if (calc != null && isInputRefOnly(calc) && isProjectionNotPushedDown(oldScan)) {
+ calcRefFields = getRefFiledIndex(calc);
+ }
+
// map arg index in aggregate to field index in scan through referred fields by calc.
if (calcRefFields != null) {
groupingSets = translateGroupingArgIndex(groupingSets, calcRefFields);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java
index 0678162929e..e800a77a56f 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggIntoScanRule.java
@@ -67,13 +67,13 @@ public class PushLocalHashAggIntoScanRule extends PushLocalAggIntoScanRuleBase {
public boolean matches(RelOptRuleCall call) {
BatchPhysicalLocalHashAggregate localAggregate = call.rel(1);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(2);
- return canPushDown(call, localAggregate, tableSourceScan);
+ return canPushDown(call, localAggregate, tableSourceScan, null);
}
@Override
public void onMatch(RelOptRuleCall call) {
BatchPhysicalLocalHashAggregate localHashAgg = call.rel(1);
BatchPhysicalTableSourceScan oldScan = call.rel(2);
- pushLocalAggregateIntoScan(call, localHashAgg, oldScan);
+ pushLocalAggregateIntoScan(call, localHashAgg, oldScan, null);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java
index 87f47c51541..2ec53eb5304 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalHashAggWithCalcIntoScanRule.java
@@ -73,10 +73,7 @@ public class PushLocalHashAggWithCalcIntoScanRule extends PushLocalAggIntoScanRu
BatchPhysicalLocalHashAggregate localHashAgg = call.rel(1);
BatchPhysicalCalc calc = call.rel(2);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(3);
-
- return isInputRefOnly(calc)
- && isProjectionNotPushedDown(tableSourceScan)
- && canPushDown(call, localHashAgg, tableSourceScan);
+ return canPushDown(call, localHashAgg, tableSourceScan, calc);
}
@Override
@@ -84,9 +81,6 @@ public class PushLocalHashAggWithCalcIntoScanRule extends PushLocalAggIntoScanRu
BatchPhysicalLocalHashAggregate localHashAgg = call.rel(1);
BatchPhysicalCalc calc = call.rel(2);
BatchPhysicalTableSourceScan oldScan = call.rel(3);
-
- int[] calcRefFields = getRefFiledIndex(calc);
-
- pushLocalAggregateIntoScan(call, localHashAgg, oldScan, calcRefFields);
+ pushLocalAggregateIntoScan(call, localHashAgg, oldScan, calc);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java
index ca101ca3cd8..fe2a2738035 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggIntoScanRule.java
@@ -67,13 +67,13 @@ public class PushLocalSortAggIntoScanRule extends PushLocalAggIntoScanRuleBase {
public boolean matches(RelOptRuleCall call) {
BatchPhysicalLocalSortAggregate localAggregate = call.rel(1);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(2);
- return canPushDown(call, localAggregate, tableSourceScan);
+ return canPushDown(call, localAggregate, tableSourceScan, null);
}
@Override
public void onMatch(RelOptRuleCall call) {
BatchPhysicalLocalSortAggregate localHashAgg = call.rel(1);
BatchPhysicalTableSourceScan oldScan = call.rel(2);
- pushLocalAggregateIntoScan(call, localHashAgg, oldScan);
+ pushLocalAggregateIntoScan(call, localHashAgg, oldScan, null);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java
index e56e3aa9dd8..a61d88db6d9 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithCalcIntoScanRule.java
@@ -73,10 +73,7 @@ public class PushLocalSortAggWithCalcIntoScanRule extends PushLocalAggIntoScanRu
BatchPhysicalLocalSortAggregate localAggregate = call.rel(1);
BatchPhysicalCalc calc = call.rel(2);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(3);
-
- return isInputRefOnly(calc)
- && isProjectionNotPushedDown(tableSourceScan)
- && canPushDown(call, localAggregate, tableSourceScan);
+ return canPushDown(call, localAggregate, tableSourceScan, calc);
}
@Override
@@ -84,9 +81,6 @@ public class PushLocalSortAggWithCalcIntoScanRule extends PushLocalAggIntoScanRu
BatchPhysicalLocalSortAggregate localHashAgg = call.rel(1);
BatchPhysicalCalc calc = call.rel(2);
BatchPhysicalTableSourceScan oldScan = call.rel(3);
-
- int[] calcRefFields = getRefFiledIndex(calc);
-
- pushLocalAggregateIntoScan(call, localHashAgg, oldScan, calcRefFields);
+ pushLocalAggregateIntoScan(call, localHashAgg, oldScan, calc);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java
index d9c340a00ed..0bc7fca86aa 100755
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortAndCalcIntoScanRule.java
@@ -82,10 +82,7 @@ public class PushLocalSortAggWithSortAndCalcIntoScanRule extends PushLocalAggInt
BatchPhysicalGroupAggregateBase localAggregate = call.rel(1);
BatchPhysicalCalc calc = call.rel(3);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(4);
-
- return isInputRefOnly(calc)
- && isProjectionNotPushedDown(tableSourceScan)
- && canPushDown(call, localAggregate, tableSourceScan);
+ return canPushDown(call, localAggregate, tableSourceScan, calc);
}
@Override
@@ -93,9 +90,6 @@ public class PushLocalSortAggWithSortAndCalcIntoScanRule extends PushLocalAggInt
BatchPhysicalGroupAggregateBase localSortAgg = call.rel(1);
BatchPhysicalCalc calc = call.rel(3);
BatchPhysicalTableSourceScan oldScan = call.rel(4);
-
- int[] calcRefFields = getRefFiledIndex(calc);
-
- pushLocalAggregateIntoScan(call, localSortAgg, oldScan, calcRefFields);
+ pushLocalAggregateIntoScan(call, localSortAgg, oldScan, calc);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java
index 9d952b253af..87e433ceabd 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalSortAggWithSortIntoScanRule.java
@@ -75,13 +75,13 @@ public class PushLocalSortAggWithSortIntoScanRule extends PushLocalAggIntoScanRu
public boolean matches(RelOptRuleCall call) {
BatchPhysicalGroupAggregateBase localAggregate = call.rel(1);
BatchPhysicalTableSourceScan tableSourceScan = call.rel(3);
- return canPushDown(call, localAggregate, tableSourceScan);
+ return canPushDown(call, localAggregate, tableSourceScan, null);
}
@Override
public void onMatch(RelOptRuleCall call) {
BatchPhysicalGroupAggregateBase localSortAgg = call.rel(1);
BatchPhysicalTableSourceScan oldScan = call.rel(3);
- pushLocalAggregateIntoScan(call, localSortAgg, oldScan);
+ pushLocalAggregateIntoScan(call, localSortAgg, oldScan, null);
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.java
index 01558291d80..6aa1c5a0ac3 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.java
@@ -44,7 +44,7 @@ class PushLocalAggIntoTableSourceScanRuleTest extends TableTestBase {
String ddl =
"CREATE TABLE inventory (\n"
+ " id BIGINT,\n"
- + " name STRING,\n"
+ + " name STRING not null,\n"
+ " amount BIGINT,\n"
+ " price BIGINT,\n"
+ " type STRING\n"
@@ -156,6 +156,26 @@ class PushLocalAggIntoTableSourceScanRuleTest extends TableTestBase {
+ "FROM inventory");
}
+ @Test
+ public void testCanPushDownLocalHashAggForCount() {
+ util.verifyRelPlan("SELECT count(*) FROM inventory");
+ }
+
+ @Test
+ public void testCanPushDownLocalHashAggForCount1() {
+ util.verifyRelPlan("SELECT count(1) FROM inventory");
+ }
+
+ @Test
+ public void testCanPushDownLocalHashAggForCountNullableColumn() {
+ util.verifyRelPlan("SELECT count(id) FROM inventory");
+ }
+
+ @Test
+ public void testCanPushDownLocalHashAggForCountNotNullColumn() {
+ util.verifyRelPlan("SELECT count(name) FROM inventory");
+ }
+
@Test
void testCanPushDownLocalSortAggWithoutSort() {
// enable sort agg
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/agg/LocalAggregatePushDownITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/agg/LocalAggregatePushDownITCase.java
index ef758c7e646..14cd2c19b2a 100755
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/agg/LocalAggregatePushDownITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/agg/LocalAggregatePushDownITCase.java
@@ -44,7 +44,7 @@ class LocalAggregatePushDownITCase extends BatchTestBase {
String ddl =
"CREATE TABLE AggregatableTable (\n"
+ " id int,\n"
- + " age int,\n"
+ + " age int not null,\n"
+ " name string,\n"
+ " height int,\n"
+ " gender string,\n"
@@ -166,6 +166,46 @@ class LocalAggregatePushDownITCase extends BatchTestBase {
false);
}
+ @Test
+ public void testCanPushDownLocalHashAggForCount() {
+ checkResult(
+ "SELECT count(*) FROM AggregatableTable",
+ JavaScalaConversionUtil.toScala(Collections.singletonList(Row.of(11))),
+ false);
+ }
+
+ @Test
+ public void testCannotPushDownLocalHashAggForCountWithFilterCondition() {
+ checkResult(
+ "SELECT count(*) FROM AggregatableTable where id > 8",
+ JavaScalaConversionUtil.toScala(Collections.singletonList(Row.of(3))),
+ false);
+ }
+
+ @Test
+ public void testCanPushDownLocalHashAggForCount1() {
+ checkResult(
+ "SELECT count(1) FROM AggregatableTable",
+ JavaScalaConversionUtil.toScala(Collections.singletonList(Row.of(11))),
+ false);
+ }
+
+ @Test
+ public void testCanPushDownLocalHashAggForCountNullableColumn() {
+ checkResult(
+ "SELECT count(id) FROM AggregatableTable",
+ JavaScalaConversionUtil.toScala(Collections.singletonList(Row.of(11))),
+ false);
+ }
+
+ @Test
+ public void testCanPushDownLocalHashAggForCountNotNullColumn() {
+ checkResult(
+ "SELECT count(age) FROM AggregatableTable",
+ JavaScalaConversionUtil.toScala(Collections.singletonList(Row.of(11))),
+ false);
+ }
+
@Test
void testPushDownLocalSortAggWithoutSort() {
// enable sort agg
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/HashAggregateTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/HashAggregateTest.xml
index fcb0b600a74..d05228dfaaa 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/HashAggregateTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/HashAggregateTest.xml
@@ -514,25 +514,6 @@ HashAggregate(isMerge=[true], select=[Final_COUNT(count$0) AS EXPR$0, Final_COUN
+- Exchange(distribution=[single]), rowType=[RecordType(BIGINT count$0, BIGINT count$1, BIGINT count$2, BIGINT count$3, BIGINT count$4, BIGINT count$5, BIGINT count$6, BIGINT count$7, BIGINT count$8, BIGINT count$9, BIGINT count$10, BIGINT count$11, BIGINT count$12)]
+- LocalHashAggregate(select=[Partial_COUNT(byte) AS count$0, Partial_COUNT(short) AS count$1, Partial_COUNT(int) AS count$2, Partial_COUNT(long) AS count$3, Partial_COUNT(float) AS count$4, Partial_COUNT(double) AS count$5, Partial_COUNT(decimal3020) AS count$6, Partial_COUNT(decimal105) AS count$7, Partial_COUNT(boolean) AS count$8, Partial_COUNT(date) AS count$9, Partial_COUNT(time) AS count$10, Partial_COUNT(timestamp) AS count$11, Partial_COUNT(string) AS count$12]), rowType=[Rec [...]
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TI [...]
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testGroupAggregate[aggStrategy=AUTO]">
- <Resource name="sql">
- <![CDATA[SELECT a, SUM(b), COUNT(c) FROM MyTable1 GROUP BY a]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)], EXPR$2=[COUNT($2)])
-+- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_SUM(sum$0) AS EXPR$1, Final_COUNT(count$1) AS EXPR$2])
-+- Exchange(distribution=[hash[a]])
- +- LocalHashAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0, Partial_COUNT(c) AS count$1])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
@@ -613,9 +594,7 @@ LogicalAggregate(group=[{}], EXPR$0=[COUNT()]), rowType=[RecordType(BIGINT EXPR$
<![CDATA[
HashAggregate(isMerge=[true], select=[Final_COUNT(count1$0) AS EXPR$0]), rowType=[RecordType(BIGINT EXPR$0)]
+- Exchange(distribution=[single]), rowType=[RecordType(BIGINT count1$0)]
- +- LocalHashAggregate(select=[Partial_COUNT(*) AS count1$0]), rowType=[RecordType(BIGINT count1$0)]
- +- Calc(select=[0 AS $f0]), rowType=[RecordType(INTEGER $f0)]
- +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[]]], fields=[id]), rowType=[RecordType(VARCHAR(2147483647) id)]
+ +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[Count1AggFunction()]]]], fields=[count1$0]), rowType=[RecordType(BIGINT count1$0)]
]]>
</Resource>
</TestCase>
@@ -654,9 +633,26 @@ LogicalAggregate(group=[{}], EXPR$0=[COUNT()]), rowType=[RecordType(BIGINT EXPR$
<![CDATA[
HashAggregate(isMerge=[true], select=[Final_COUNT(count1$0) AS EXPR$0]), rowType=[RecordType(BIGINT EXPR$0)]
+- Exchange(distribution=[single]), rowType=[RecordType(BIGINT count1$0)]
- +- LocalHashAggregate(select=[Partial_COUNT(*) AS count1$0]), rowType=[RecordType(BIGINT count1$0)]
- +- Calc(select=[0 AS $f0]), rowType=[RecordType(INTEGER $f0)]
- +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[]]], fields=[id]), rowType=[RecordType(VARCHAR(2147483647) id)]
+ +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[Count1AggFunction()]]]], fields=[count1$0]), rowType=[RecordType(BIGINT count1$0)]
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testGroupAggregate[aggStrategy=AUTO]">
+ <Resource name="sql">
+ <![CDATA[SELECT a, SUM(b), COUNT(c) FROM MyTable1 GROUP BY a]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)], EXPR$2=[COUNT($2)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_SUM(sum$0) AS EXPR$1, Final_COUNT(count$1) AS EXPR$2])
++- Exchange(distribution=[hash[a]])
+ +- LocalHashAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0, Partial_COUNT(c) AS count$1])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
@@ -937,38 +933,42 @@ HashAggregate(isMerge=[true], select=[Final_MAX(max$0) AS EXPR$0, Final_MAX(max$
]]>
</Resource>
</TestCase>
- <TestCase name="testSum[aggStrategy=AUTO]">
+ <TestCase name="testMinWithFixLengthType[aggStrategy=AUTO]">
<Resource name="sql">
<![CDATA[
-SELECT SUM(`byte`),
- SUM(`short`),
- SUM(`int`),
- SUM(`long`),
- SUM(`float`),
- SUM(`double`),
- SUM(`decimal3020`),
- SUM(`decimal105`)
+SELECT MIN(`byte`),
+ MIN(`short`),
+ MIN(`int`),
+ MIN(`long`),
+ MIN(`float`),
+ MIN(`double`),
+ MIN(`decimal3020`),
+ MIN(`decimal105`),
+ MIN(`boolean`),
+ MIN(`date`),
+ MIN(`time`),
+ MIN(`timestamp`)
FROM MyTable
]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalAggregate(group=[{}], EXPR$0=[SUM($0)], EXPR$1=[SUM($1)], EXPR$2=[SUM($2)], EXPR$3=[SUM($3)], EXPR$4=[SUM($4)], EXPR$5=[SUM($5)], EXPR$6=[SUM($6)], EXPR$7=[SUM($7)]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
-+- LogicalProject(byte=[$0], short=[$1], int=[$2], long=[$3], float=[$4], double=[$5], decimal3020=[$11], decimal105=[$12]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+LogicalAggregate(group=[{}], EXPR$0=[MIN($0)], EXPR$1=[MIN($1)], EXPR$2=[MIN($2)], EXPR$3=[MIN($3)], EXPR$4=[MIN($4)], EXPR$5=[MIN($5)], EXPR$6=[MIN($6)], EXPR$7=[MIN($7)], EXPR$8=[MIN($8)], EXPR$9=[MIN($9)], EXPR$10=[MIN($10)], EXPR$11=[MIN($11)]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(30, 20) EXPR$6, DECIMAL(10, 5) EXPR$7, BOOLEAN EXPR$8, DATE EXPR$9, TIME(0) EXPR$10, TIMESTAMP(3) EXPR$11)]
++- LogicalProject(byte=[$0], short=[$1], int=[$2], long=[$3], float=[$4], double=[$5], decimal3020=[$11], decimal105=[$12], boolean=[$6], date=[$8], time=[$9], timestamp=[$10]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TIME(0) time, TIMESTAMP(3) timestamp, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-HashAggregate(isMerge=[true], select=[Final_SUM(sum$0) AS EXPR$0, Final_SUM(sum$1) AS EXPR$1, Final_SUM(sum$2) AS EXPR$2, Final_SUM(sum$3) AS EXPR$3, Final_SUM(sum$4) AS EXPR$4, Final_SUM(sum$5) AS EXPR$5, Final_SUM(sum$6) AS EXPR$6, Final_SUM(sum$7) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
-+- Exchange(distribution=[single]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
- +- LocalHashAggregate(select=[Partial_SUM(byte) AS sum$0, Partial_SUM(short) AS sum$1, Partial_SUM(int) AS sum$2, Partial_SUM(long) AS sum$3, Partial_SUM(float) AS sum$4, Partial_SUM(double) AS sum$5, Partial_SUM(decimal3020) AS sum$6, Partial_SUM(decimal105) AS sum$7]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
- +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+HashAggregate(isMerge=[true], select=[Final_MIN(min$0) AS EXPR$0, Final_MIN(min$1) AS EXPR$1, Final_MIN(min$2) AS EXPR$2, Final_MIN(min$3) AS EXPR$3, Final_MIN(min$4) AS EXPR$4, Final_MIN(min$5) AS EXPR$5, Final_MIN(min$6) AS EXPR$6, Final_MIN(min$7) AS EXPR$7, Final_MIN(min$8) AS EXPR$8, Final_MIN(min$9) AS EXPR$9, Final_MIN(min$10) AS EXPR$10, Final_MIN(min$11) AS EXPR$11]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, [...]
++- Exchange(distribution=[single]), rowType=[RecordType(TINYINT min$0, SMALLINT min$1, INTEGER min$2, BIGINT min$3, FLOAT min$4, DOUBLE min$5, DECIMAL(30, 20) min$6, DECIMAL(10, 5) min$7, BOOLEAN min$8, DATE min$9, TIME(0) min$10, TIMESTAMP(3) min$11)]
+ +- LocalHashAggregate(select=[Partial_MIN(byte) AS min$0, Partial_MIN(short) AS min$1, Partial_MIN(int) AS min$2, Partial_MIN(long) AS min$3, Partial_MIN(float) AS min$4, Partial_MIN(double) AS min$5, Partial_MIN(decimal3020) AS min$6, Partial_MIN(decimal105) AS min$7, Partial_MIN(boolean) AS min$8, Partial_MIN(date) AS min$9, Partial_MIN(time) AS min$10, Partial_MIN(timestamp) AS min$11]), rowType=[RecordType(TINYINT min$0, SMALLINT min$1, INTEGER min$2, BIGINT min$3, FLOAT min$4, DO [...]
+ +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105, boolean, date, time, timestamp]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, [...]
]]>
</Resource>
</TestCase>
- <TestCase name="testMinWithFixLengthType[aggStrategy=AUTO]">
+ <TestCase name="testMinWithFixLengthType[aggStrategy=ONE_PHASE]">
<Resource name="sql">
<![CDATA[
SELECT MIN(`byte`),
@@ -995,11 +995,10 @@ LogicalAggregate(group=[{}], EXPR$0=[MIN($0)], EXPR$1=[MIN($1)], EXPR$2=[MIN($2)
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-HashAggregate(isMerge=[true], select=[Final_MIN(min$0) AS EXPR$0, Final_MIN(min$1) AS EXPR$1, Final_MIN(min$2) AS EXPR$2, Final_MIN(min$3) AS EXPR$3, Final_MIN(min$4) AS EXPR$4, Final_MIN(min$5) AS EXPR$5, Final_MIN(min$6) AS EXPR$6, Final_MIN(min$7) AS EXPR$7, Final_MIN(min$8) AS EXPR$8, Final_MIN(min$9) AS EXPR$9, Final_MIN(min$10) AS EXPR$10, Final_MIN(min$11) AS EXPR$11]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, [...]
-+- Exchange(distribution=[single]), rowType=[RecordType(TINYINT min$0, SMALLINT min$1, INTEGER min$2, BIGINT min$3, FLOAT min$4, DOUBLE min$5, DECIMAL(30, 20) min$6, DECIMAL(10, 5) min$7, BOOLEAN min$8, DATE min$9, TIME(0) min$10, TIMESTAMP(3) min$11)]
- +- LocalHashAggregate(select=[Partial_MIN(byte) AS min$0, Partial_MIN(short) AS min$1, Partial_MIN(int) AS min$2, Partial_MIN(long) AS min$3, Partial_MIN(float) AS min$4, Partial_MIN(double) AS min$5, Partial_MIN(decimal3020) AS min$6, Partial_MIN(decimal105) AS min$7, Partial_MIN(boolean) AS min$8, Partial_MIN(date) AS min$9, Partial_MIN(time) AS min$10, Partial_MIN(timestamp) AS min$11]), rowType=[RecordType(TINYINT min$0, SMALLINT min$1, INTEGER min$2, BIGINT min$3, FLOAT min$4, DO [...]
- +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105, boolean, date, time, timestamp]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, [...]
+HashAggregate(isMerge=[false], select=[MIN(byte) AS EXPR$0, MIN(short) AS EXPR$1, MIN(int) AS EXPR$2, MIN(long) AS EXPR$3, MIN(float) AS EXPR$4, MIN(double) AS EXPR$5, MIN(decimal3020) AS EXPR$6, MIN(decimal105) AS EXPR$7, MIN(boolean) AS EXPR$8, MIN(date) AS EXPR$9, MIN(time) AS EXPR$10, MIN(timestamp) AS EXPR$11]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(30, 20) EXPR$6, DECIMAL(10, 5) EXPR$7, BOOLEAN EXPR$ [...]
++- Exchange(distribution=[single]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
+ +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105, boolean, date, time, timestamp]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TI [...]
]]>
</Resource>
</TestCase>
@@ -1038,7 +1037,7 @@ HashAggregate(isMerge=[true], select=[Final_MIN(min$0) AS EXPR$0, Final_MIN(min$
]]>
</Resource>
</TestCase>
- <TestCase name="testSum[aggStrategy=ONE_PHASE]">
+ <TestCase name="testSum[aggStrategy=AUTO]">
<Resource name="sql">
<![CDATA[
SELECT SUM(`byte`),
@@ -1061,43 +1060,40 @@ LogicalAggregate(group=[{}], EXPR$0=[SUM($0)], EXPR$1=[SUM($1)], EXPR$2=[SUM($2)
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-HashAggregate(isMerge=[false], select=[SUM(byte) AS EXPR$0, SUM(short) AS EXPR$1, SUM(int) AS EXPR$2, SUM(long) AS EXPR$3, SUM(float) AS EXPR$4, SUM(double) AS EXPR$5, SUM(decimal3020) AS EXPR$6, SUM(decimal105) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
-+- Exchange(distribution=[single]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
- +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TI [...]
+HashAggregate(isMerge=[true], select=[Final_SUM(sum$0) AS EXPR$0, Final_SUM(sum$1) AS EXPR$1, Final_SUM(sum$2) AS EXPR$2, Final_SUM(sum$3) AS EXPR$3, Final_SUM(sum$4) AS EXPR$4, Final_SUM(sum$5) AS EXPR$5, Final_SUM(sum$6) AS EXPR$6, Final_SUM(sum$7) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
++- Exchange(distribution=[single]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
+ +- LocalHashAggregate(select=[Partial_SUM(byte) AS sum$0, Partial_SUM(short) AS sum$1, Partial_SUM(int) AS sum$2, Partial_SUM(long) AS sum$3, Partial_SUM(float) AS sum$4, Partial_SUM(double) AS sum$5, Partial_SUM(decimal3020) AS sum$6, Partial_SUM(decimal105) AS sum$7]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
+ +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, [...]
]]>
</Resource>
</TestCase>
- <TestCase name="testMinWithFixLengthType[aggStrategy=ONE_PHASE]">
+ <TestCase name="testSum[aggStrategy=ONE_PHASE]">
<Resource name="sql">
<![CDATA[
-SELECT MIN(`byte`),
- MIN(`short`),
- MIN(`int`),
- MIN(`long`),
- MIN(`float`),
- MIN(`double`),
- MIN(`decimal3020`),
- MIN(`decimal105`),
- MIN(`boolean`),
- MIN(`date`),
- MIN(`time`),
- MIN(`timestamp`)
+SELECT SUM(`byte`),
+ SUM(`short`),
+ SUM(`int`),
+ SUM(`long`),
+ SUM(`float`),
+ SUM(`double`),
+ SUM(`decimal3020`),
+ SUM(`decimal105`)
FROM MyTable
]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalAggregate(group=[{}], EXPR$0=[MIN($0)], EXPR$1=[MIN($1)], EXPR$2=[MIN($2)], EXPR$3=[MIN($3)], EXPR$4=[MIN($4)], EXPR$5=[MIN($5)], EXPR$6=[MIN($6)], EXPR$7=[MIN($7)], EXPR$8=[MIN($8)], EXPR$9=[MIN($9)], EXPR$10=[MIN($10)], EXPR$11=[MIN($11)]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(30, 20) EXPR$6, DECIMAL(10, 5) EXPR$7, BOOLEAN EXPR$8, DATE EXPR$9, TIME(0) EXPR$10, TIMESTAMP(3) EXPR$11)]
-+- LogicalProject(byte=[$0], short=[$1], int=[$2], long=[$3], float=[$4], double=[$5], decimal3020=[$11], decimal105=[$12], boolean=[$6], date=[$8], time=[$9], timestamp=[$10]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
+LogicalAggregate(group=[{}], EXPR$0=[SUM($0)], EXPR$1=[SUM($1)], EXPR$2=[SUM($2)], EXPR$3=[SUM($3)], EXPR$4=[SUM($4)], EXPR$5=[SUM($5)], EXPR$6=[SUM($6)], EXPR$7=[SUM($7)]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
++- LogicalProject(byte=[$0], short=[$1], int=[$2], long=[$3], float=[$4], double=[$5], decimal3020=[$11], decimal105=[$12]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TIME(0) time, TIMESTAMP(3) timestamp, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-HashAggregate(isMerge=[false], select=[MIN(byte) AS EXPR$0, MIN(short) AS EXPR$1, MIN(int) AS EXPR$2, MIN(long) AS EXPR$3, MIN(float) AS EXPR$4, MIN(double) AS EXPR$5, MIN(decimal3020) AS EXPR$6, MIN(decimal105) AS EXPR$7, MIN(boolean) AS EXPR$8, MIN(date) AS EXPR$9, MIN(time) AS EXPR$10, MIN(timestamp) AS EXPR$11]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(30, 20) EXPR$6, DECIMAL(10, 5) EXPR$7, BOOLEAN EXPR$ [...]
-+- Exchange(distribution=[single]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
- +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105, boolean, date, time, timestamp]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105, BOOLEAN boolean, DATE date, TIME(0) time, TIMESTAMP(3) timestamp)]
+HashAggregate(isMerge=[false], select=[SUM(byte) AS EXPR$0, SUM(short) AS EXPR$1, SUM(int) AS EXPR$2, SUM(long) AS EXPR$3, SUM(float) AS EXPR$4, SUM(double) AS EXPR$5, SUM(decimal3020) AS EXPR$6, SUM(decimal105) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
++- Exchange(distribution=[single]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+ +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TI [...]
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/SortAggregateTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/SortAggregateTest.xml
index 0049f5705e3..1ccd27b5af3 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/SortAggregateTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/SortAggregateTest.xml
@@ -637,29 +637,6 @@ SortAggregate(isMerge=[true], select=[Final_COUNT(count$0) AS EXPR$0, Final_COUN
+- Exchange(distribution=[single]), rowType=[RecordType(BIGINT count$0, BIGINT count$1, BIGINT count$2, BIGINT count$3, BIGINT count$4, BIGINT count$5, BIGINT count$6, BIGINT count$7, BIGINT count$8, BIGINT count$9, BIGINT count$10, BIGINT count$11, BIGINT count$12)]
+- LocalSortAggregate(select=[Partial_COUNT(byte) AS count$0, Partial_COUNT(short) AS count$1, Partial_COUNT(int) AS count$2, Partial_COUNT(long) AS count$3, Partial_COUNT(float) AS count$4, Partial_COUNT(double) AS count$5, Partial_COUNT(decimal3020) AS count$6, Partial_COUNT(decimal105) AS count$7, Partial_COUNT(boolean) AS count$8, Partial_COUNT(date) AS count$9, Partial_COUNT(time) AS count$10, Partial_COUNT(timestamp) AS count$11, Partial_COUNT(string) AS count$12]), rowType=[Rec [...]
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TI [...]
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testGroupAggregate[aggStrategy=AUTO]">
- <Resource name="sql">
- <![CDATA[SELECT a, SUM(b), COUNT(c) FROM MyTable1 GROUP BY a]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)], EXPR$2=[COUNT($2)])
-+- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-SortAggregate(isMerge=[true], groupBy=[a], select=[a, Final_SUM(sum$0) AS EXPR$1, Final_COUNT(count$1) AS EXPR$2])
-+- Exchange(distribution=[forward])
- +- Sort(orderBy=[a ASC])
- +- Exchange(distribution=[hash[a]])
- +- LocalSortAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0, Partial_COUNT(c) AS count$1])
- +- Exchange(distribution=[forward])
- +- Sort(orderBy=[a ASC])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
@@ -740,9 +717,7 @@ LogicalAggregate(group=[{}], EXPR$0=[COUNT()]), rowType=[RecordType(BIGINT EXPR$
<![CDATA[
SortAggregate(isMerge=[true], select=[Final_COUNT(count1$0) AS EXPR$0]), rowType=[RecordType(BIGINT EXPR$0)]
+- Exchange(distribution=[single]), rowType=[RecordType(BIGINT count1$0)]
- +- LocalSortAggregate(select=[Partial_COUNT(*) AS count1$0]), rowType=[RecordType(BIGINT count1$0)]
- +- Calc(select=[0 AS $f0]), rowType=[RecordType(INTEGER $f0)]
- +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[]]], fields=[id]), rowType=[RecordType(VARCHAR(2147483647) id)]
+ +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[Count1AggFunction()]]]], fields=[count1$0]), rowType=[RecordType(BIGINT count1$0)]
]]>
</Resource>
</TestCase>
@@ -781,9 +756,30 @@ LogicalAggregate(group=[{}], EXPR$0=[COUNT()]), rowType=[RecordType(BIGINT EXPR$
<![CDATA[
SortAggregate(isMerge=[true], select=[Final_COUNT(count1$0) AS EXPR$0]), rowType=[RecordType(BIGINT EXPR$0)]
+- Exchange(distribution=[single]), rowType=[RecordType(BIGINT count1$0)]
- +- LocalSortAggregate(select=[Partial_COUNT(*) AS count1$0]), rowType=[RecordType(BIGINT count1$0)]
- +- Calc(select=[0 AS $f0]), rowType=[RecordType(INTEGER $f0)]
- +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[]]], fields=[id]), rowType=[RecordType(VARCHAR(2147483647) id)]
+ +- TableSourceScan(table=[[default_catalog, default_database, src, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[Count1AggFunction()]]]], fields=[count1$0]), rowType=[RecordType(BIGINT count1$0)]
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testGroupAggregate[aggStrategy=AUTO]">
+ <Resource name="sql">
+ <![CDATA[SELECT a, SUM(b), COUNT(c) FROM MyTable1 GROUP BY a]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{0}], EXPR$1=[SUM($1)], EXPR$2=[COUNT($2)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+SortAggregate(isMerge=[true], groupBy=[a], select=[a, Final_SUM(sum$0) AS EXPR$1, Final_COUNT(count$1) AS EXPR$2])
++- Exchange(distribution=[forward])
+ +- Sort(orderBy=[a ASC])
+ +- Exchange(distribution=[hash[a]])
+ +- LocalSortAggregate(groupBy=[a], select=[a, Partial_SUM(b) AS sum$0, Partial_COUNT(c) AS count$1])
+ +- Exchange(distribution=[forward])
+ +- Sort(orderBy=[a ASC])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
@@ -1318,38 +1314,29 @@ SortAggregate(isMerge=[true], select=[Final_MIN(min$0) AS EXPR$0]), rowType=[Rec
]]>
</Resource>
</TestCase>
- <TestCase name="testSum[aggStrategy=AUTO]">
+ <TestCase name="testPojoAccumulator[aggStrategy=AUTO]">
<Resource name="sql">
- <![CDATA[
-SELECT SUM(`byte`),
- SUM(`short`),
- SUM(`int`),
- SUM(`long`),
- SUM(`float`),
- SUM(`double`),
- SUM(`decimal3020`),
- SUM(`decimal105`)
-FROM MyTable
- ]]>
+ <![CDATA[SELECT b, var_sum(a) FROM MyTable1 GROUP BY b]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalAggregate(group=[{}], EXPR$0=[SUM($0)], EXPR$1=[SUM($1)], EXPR$2=[SUM($2)], EXPR$3=[SUM($3)], EXPR$4=[SUM($4)], EXPR$5=[SUM($5)], EXPR$6=[SUM($6)], EXPR$7=[SUM($7)]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
-+- LogicalProject(byte=[$0], short=[$1], int=[$2], long=[$3], float=[$4], double=[$5], decimal3020=[$11], decimal105=[$12]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TIME(0) time, TIMESTAMP(3) timestamp, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+LogicalAggregate(group=[{0}], EXPR$1=[var_sum($1)])
++- LogicalProject(b=[$1], a=[$0])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
]]>
</Resource>
- <Resource name="optimized rel plan">
+ <Resource name="optimized exec plan">
<![CDATA[
-SortAggregate(isMerge=[true], select=[Final_SUM(sum$0) AS EXPR$0, Final_SUM(sum$1) AS EXPR$1, Final_SUM(sum$2) AS EXPR$2, Final_SUM(sum$3) AS EXPR$3, Final_SUM(sum$4) AS EXPR$4, Final_SUM(sum$5) AS EXPR$5, Final_SUM(sum$6) AS EXPR$6, Final_SUM(sum$7) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
-+- Exchange(distribution=[single]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
- +- LocalSortAggregate(select=[Partial_SUM(byte) AS sum$0, Partial_SUM(short) AS sum$1, Partial_SUM(int) AS sum$2, Partial_SUM(long) AS sum$3, Partial_SUM(float) AS sum$4, Partial_SUM(double) AS sum$5, Partial_SUM(decimal3020) AS sum$6, Partial_SUM(decimal105) AS sum$7]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
- +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, [...]
+SortAggregate(isMerge=[false], groupBy=[b], select=[b, var_sum(a) AS EXPR$1])
++- Exchange(distribution=[forward])
+ +- Sort(orderBy=[b ASC])
+ +- Exchange(distribution=[hash[b]])
+ +- Calc(select=[b, a])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
- <TestCase name="testPojoAccumulator[aggStrategy=AUTO]">
+ <TestCase name="testPojoAccumulator[aggStrategy=ONE_PHASE]">
<Resource name="sql">
<![CDATA[SELECT b, var_sum(a) FROM MyTable1 GROUP BY b]]>
</Resource>
@@ -1393,7 +1380,7 @@ SortAggregate(isMerge=[false], groupBy=[b], select=[b, var_sum(a) AS EXPR$1])
]]>
</Resource>
</TestCase>
- <TestCase name="testSum[aggStrategy=ONE_PHASE]">
+ <TestCase name="testSum[aggStrategy=AUTO]">
<Resource name="sql">
<![CDATA[
SELECT SUM(`byte`),
@@ -1416,32 +1403,41 @@ LogicalAggregate(group=[{}], EXPR$0=[SUM($0)], EXPR$1=[SUM($1)], EXPR$2=[SUM($2)
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-SortAggregate(isMerge=[false], select=[SUM(byte) AS EXPR$0, SUM(short) AS EXPR$1, SUM(int) AS EXPR$2, SUM(long) AS EXPR$3, SUM(float) AS EXPR$4, SUM(double) AS EXPR$5, SUM(decimal3020) AS EXPR$6, SUM(decimal105) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
-+- Exchange(distribution=[single]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
- +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TI [...]
+SortAggregate(isMerge=[true], select=[Final_SUM(sum$0) AS EXPR$0, Final_SUM(sum$1) AS EXPR$1, Final_SUM(sum$2) AS EXPR$2, Final_SUM(sum$3) AS EXPR$3, Final_SUM(sum$4) AS EXPR$4, Final_SUM(sum$5) AS EXPR$5, Final_SUM(sum$6) AS EXPR$6, Final_SUM(sum$7) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
++- Exchange(distribution=[single]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
+ +- LocalSortAggregate(select=[Partial_SUM(byte) AS sum$0, Partial_SUM(short) AS sum$1, Partial_SUM(int) AS sum$2, Partial_SUM(long) AS sum$3, Partial_SUM(float) AS sum$4, Partial_SUM(double) AS sum$5, Partial_SUM(decimal3020) AS sum$6, Partial_SUM(decimal105) AS sum$7]), rowType=[RecordType(TINYINT sum$0, SMALLINT sum$1, INTEGER sum$2, BIGINT sum$3, FLOAT sum$4, DOUBLE sum$5, DECIMAL(38, 20) sum$6, DECIMAL(38, 5) sum$7)]
+ +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, [...]
]]>
</Resource>
</TestCase>
- <TestCase name="testPojoAccumulator[aggStrategy=ONE_PHASE]">
+ <TestCase name="testSum[aggStrategy=ONE_PHASE]">
<Resource name="sql">
- <![CDATA[SELECT b, var_sum(a) FROM MyTable1 GROUP BY b]]>
+ <![CDATA[
+SELECT SUM(`byte`),
+ SUM(`short`),
+ SUM(`int`),
+ SUM(`long`),
+ SUM(`float`),
+ SUM(`double`),
+ SUM(`decimal3020`),
+ SUM(`decimal105`)
+FROM MyTable
+ ]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalAggregate(group=[{0}], EXPR$1=[var_sum($1)])
-+- LogicalProject(b=[$1], a=[$0])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
+LogicalAggregate(group=[{}], EXPR$0=[SUM($0)], EXPR$1=[SUM($1)], EXPR$2=[SUM($2)], EXPR$3=[SUM($3)], EXPR$4=[SUM($4)], EXPR$5=[SUM($5)], EXPR$6=[SUM($6)], EXPR$7=[SUM($7)]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
++- LogicalProject(byte=[$0], short=[$1], int=[$2], long=[$3], float=[$4], double=[$5], decimal3020=[$11], decimal105=[$12]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TIME(0) time, TIMESTAMP(3) timestamp, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
]]>
</Resource>
- <Resource name="optimized exec plan">
+ <Resource name="optimized rel plan">
<![CDATA[
-SortAggregate(isMerge=[false], groupBy=[b], select=[b, var_sum(a) AS EXPR$1])
-+- Exchange(distribution=[forward])
- +- Sort(orderBy=[b ASC])
- +- Exchange(distribution=[hash[b]])
- +- Calc(select=[b, a])
- +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+SortAggregate(isMerge=[false], select=[SUM(byte) AS EXPR$0, SUM(short) AS EXPR$1, SUM(int) AS EXPR$2, SUM(long) AS EXPR$3, SUM(float) AS EXPR$4, SUM(double) AS EXPR$5, SUM(decimal3020) AS EXPR$6, SUM(decimal105) AS EXPR$7]), rowType=[RecordType(TINYINT EXPR$0, SMALLINT EXPR$1, INTEGER EXPR$2, BIGINT EXPR$3, FLOAT EXPR$4, DOUBLE EXPR$5, DECIMAL(38, 20) EXPR$6, DECIMAL(38, 5) EXPR$7)]
++- Exchange(distribution=[single]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+ +- Calc(select=[byte, short, int, long, float, double, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, DECIMAL(30, 20) decimal3020, DECIMAL(10, 5) decimal105)]
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105)]]], fields=[byte, short, int, long, float, double, boolean, string, date, time, timestamp, decimal3020, decimal105]), rowType=[RecordType(TINYINT byte, SMALLINT short, INTEGER int, BIGINT long, FLOAT float, DOUBLE double, BOOLEAN boolean, VARCHAR(2147483647) string, DATE date, TI [...]
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.xml
index f34df4c45fa..7ef0c180f08 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/PushLocalAggIntoTableSourceScanRuleTest.xml
@@ -227,32 +227,6 @@ Calc(select=[id, amount, CASE(>(w0$o0, 0), w0$o1, null:BIGINT) AS EXPR$2, name])
+- Sort(orderBy=[name ASC])
+- Exchange(distribution=[hash[name]])
+- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[id, name, amount, price], metadata=[]]], fields=[id, name, amount, price])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testCanPushDownLocalHashAggWithGroup">
- <Resource name="sql">
- <![CDATA[SELECT
- sum(amount),
- name,
- type
-FROM inventory
- group by name, type]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(EXPR$0=[$2], name=[$0], type=[$1])
-+- LogicalAggregate(group=[{0, 1}], EXPR$0=[SUM($2)])
- +- LogicalProject(name=[$1], type=[$4], amount=[$2])
- +- LogicalTableScan(table=[[default_catalog, default_database, inventory]])
-]]>
- </Resource>
- <Resource name="optimized rel plan">
- <![CDATA[
-Calc(select=[EXPR$0, name, type])
-+- HashAggregate(isMerge=[true], groupBy=[name, type], select=[name, type, Final_SUM(sum$0) AS EXPR$0])
- +- Exchange(distribution=[hash[name, type]])
- +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[name, type, amount], metadata=[], aggregates=[grouping=[name,type], aggFunctions=[LongSumAggFunction(amount)]]]], fields=[name, type, sum$0])
]]>
</Resource>
</TestCase>
@@ -334,6 +308,34 @@ Calc(select=[EXPR$0, EXPR$1, name, type])
+- HashAggregate(isMerge=[true], groupBy=[name, type], select=[name, type, Final_SUM(sum$0) AS EXPR$0, Final_MAX(max$1) AS EXPR$1])
+- Exchange(distribution=[hash[name, type]])
+- TableSourceScan(table=[[default_catalog, default_database, inventory_meta, filter=[=(id, 123:BIGINT)], project=[name, type, amount], metadata=[metadata_1], aggregates=[grouping=[name,type], aggFunctions=[LongSumAggFunction(amount),LongMaxAggFunction(metadata_1)]]]], fields=[name, type, sum$0, max$1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCanPushDownLocalAggWithoutProjectionPushDown">
+ <Resource name="sql">
+ <![CDATA[SELECT
+ sum(amount),
+ name,
+ type
+FROM inventory_no_proj
+ where id = 123
+ group by name, type]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$2], name=[$0], type=[$1])
++- LogicalAggregate(group=[{0, 1}], EXPR$0=[SUM($2)])
+ +- LogicalProject(name=[$1], type=[$4], amount=[$2])
+ +- LogicalFilter(condition=[=($0, 123)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, inventory_no_proj]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[EXPR$0, name, type])
++- HashAggregate(isMerge=[true], groupBy=[name, type], select=[name, type, Final_SUM(sum$0) AS EXPR$0])
+ +- Exchange(distribution=[hash[name, type]])
+ +- TableSourceScan(table=[[default_catalog, default_database, inventory_no_proj, filter=[=(id, 123:BIGINT)], aggregates=[grouping=[name,type], aggFunctions=[LongSumAggFunction(amount)]]]], fields=[name, type, sum$0])
]]>
</Resource>
</TestCase>
@@ -365,39 +367,89 @@ Calc(select=[EXPR$0, type, name])
]]>
</Resource>
</TestCase>
- <TestCase name="testCanPushDownLocalHashAggWithoutGroup">
+ <TestCase name="testCanPushDownLocalHashAggForCount">
<Resource name="sql">
- <![CDATA[SELECT
- min(id),
- max(amount),
- sum(price),
- avg(price),
- count(id)
-FROM inventory]]>
+ <![CDATA[SELECT count(*) FROM inventory]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalAggregate(group=[{}], EXPR$0=[MIN($0)], EXPR$1=[MAX($1)], EXPR$2=[SUM($2)], EXPR$3=[AVG($2)], EXPR$4=[COUNT($0)])
-+- LogicalProject(id=[$0], amount=[$2], price=[$3])
+LogicalAggregate(group=[{}], EXPR$0=[COUNT()])
++- LogicalProject($f0=[0])
+- LogicalTableScan(table=[[default_catalog, default_database, inventory]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-HashAggregate(isMerge=[true], select=[Final_MIN(min$0) AS EXPR$0, Final_MAX(max$1) AS EXPR$1, Final_SUM(sum$2) AS EXPR$2, Final_AVG(sum$3, count$4) AS EXPR$3, Final_COUNT(count$5) AS EXPR$4])
+HashAggregate(isMerge=[true], select=[Final_COUNT(count1$0) AS EXPR$0])
+- Exchange(distribution=[single])
- +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[id, amount, price], metadata=[], aggregates=[grouping=[], aggFunctions=[LongMinAggFunction(id),LongMaxAggFunction(amount),LongSumAggFunction(price),LongSum0AggFunction(price),CountAggFunction(price),CountAggFunction(id)]]]], fields=[min$0, max$1, sum$2, sum$3, count$4, count$5])
+ +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[Count1AggFunction()]]]], fields=[count1$0])
]]>
</Resource>
</TestCase>
- <TestCase name="testCanPushDownLocalAggWithoutProjectionPushDown">
+ <TestCase name="testCanPushDownLocalHashAggForCount1">
+ <Resource name="sql">
+ <![CDATA[SELECT count(1) FROM inventory]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{}], EXPR$0=[COUNT()])
++- LogicalProject($f0=[0])
+ +- LogicalTableScan(table=[[default_catalog, default_database, inventory]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashAggregate(isMerge=[true], select=[Final_COUNT(count1$0) AS EXPR$0])
++- Exchange(distribution=[single])
+ +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[Count1AggFunction()]]]], fields=[count1$0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCanPushDownLocalHashAggForCountNotNullColumn">
+ <Resource name="sql">
+ <![CDATA[SELECT count(name) FROM inventory]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{}], EXPR$0=[COUNT()])
++- LogicalProject($f0=[0])
+ +- LogicalTableScan(table=[[default_catalog, default_database, inventory]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashAggregate(isMerge=[true], select=[Final_COUNT(count1$0) AS EXPR$0])
++- Exchange(distribution=[single])
+ +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[Count1AggFunction()]]]], fields=[count1$0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCanPushDownLocalHashAggForCountNullableColumn">
+ <Resource name="sql">
+ <![CDATA[SELECT count(id) FROM inventory]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{}], EXPR$0=[COUNT($0)])
++- LogicalProject(id=[$0])
+ +- LogicalTableScan(table=[[default_catalog, default_database, inventory]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashAggregate(isMerge=[true], select=[Final_COUNT(count$0) AS EXPR$0])
++- Exchange(distribution=[single])
+ +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[id], metadata=[], aggregates=[grouping=[], aggFunctions=[CountAggFunction(id)]]]], fields=[count$0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCanPushDownLocalHashAggWithGroup">
<Resource name="sql">
<![CDATA[SELECT
sum(amount),
name,
type
-FROM inventory_no_proj
- where id = 123
+FROM inventory
group by name, type]]>
</Resource>
<Resource name="ast">
@@ -405,8 +457,7 @@ FROM inventory_no_proj
LogicalProject(EXPR$0=[$2], name=[$0], type=[$1])
+- LogicalAggregate(group=[{0, 1}], EXPR$0=[SUM($2)])
+- LogicalProject(name=[$1], type=[$4], amount=[$2])
- +- LogicalFilter(condition=[=($0, 123)])
- +- LogicalTableScan(table=[[default_catalog, default_database, inventory_no_proj]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, inventory]])
]]>
</Resource>
<Resource name="optimized rel plan">
@@ -414,7 +465,32 @@ LogicalProject(EXPR$0=[$2], name=[$0], type=[$1])
Calc(select=[EXPR$0, name, type])
+- HashAggregate(isMerge=[true], groupBy=[name, type], select=[name, type, Final_SUM(sum$0) AS EXPR$0])
+- Exchange(distribution=[hash[name, type]])
- +- TableSourceScan(table=[[default_catalog, default_database, inventory_no_proj, filter=[=(id, 123:BIGINT)], aggregates=[grouping=[name,type], aggFunctions=[LongSumAggFunction(amount)]]]], fields=[name, type, sum$0])
+ +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[name, type, amount], metadata=[], aggregates=[grouping=[name,type], aggFunctions=[LongSumAggFunction(amount)]]]], fields=[name, type, sum$0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCanPushDownLocalHashAggWithoutGroup">
+ <Resource name="sql">
+ <![CDATA[SELECT
+ min(id),
+ max(amount),
+ sum(price),
+ avg(price),
+ count(id)
+FROM inventory]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{}], EXPR$0=[MIN($0)], EXPR$1=[MAX($1)], EXPR$2=[SUM($2)], EXPR$3=[AVG($2)], EXPR$4=[COUNT($0)])
++- LogicalProject(id=[$0], amount=[$2], price=[$3])
+ +- LogicalTableScan(table=[[default_catalog, default_database, inventory]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashAggregate(isMerge=[true], select=[Final_MIN(min$0) AS EXPR$0, Final_MAX(max$1) AS EXPR$1, Final_SUM(sum$2) AS EXPR$2, Final_AVG(sum$3, count$4) AS EXPR$3, Final_COUNT(count$5) AS EXPR$4])
++- Exchange(distribution=[single])
+ +- TableSourceScan(table=[[default_catalog, default_database, inventory, project=[id, amount, price], metadata=[], aggregates=[grouping=[], aggFunctions=[LongMinAggFunction(id),LongMaxAggFunction(amount),LongSumAggFunction(price),LongSum0AggFunction(price),CountAggFunction(price),CountAggFunction(id)]]]], fields=[min$0, max$1, sum$2, sum$3, count$4, count$5])
]]>
</Resource>
</TestCase>