Wei Zhong created FLINK-28764:
---------------------------------
Summary: Support more than 64 distinct aggregate function calls in
one aggregate SQL query
Key: FLINK-28764
URL: https://issues.apache.org/jira/browse/FLINK-28764
Project: Flink
Issue Type: Improvement
Components: Table SQL / Planner
Affects Versions: 1.15.1, 1.14.5, 1.13.6
Reporter: Wei Zhong
Currently Flink SQL does not support more than 64 distinct aggregate function
calls in one aggregate SQL query. We encountered this problem while migrating
batch jobs from spark to flink. The spark job has 79 distinct aggregate
function calls in one aggregate SQL query.
Reproduce code:
{code:java}
public class Test64Distinct {
public static void main(String[] args) {
TableEnvironment tableEnv =
TableEnvironment.create(EnvironmentSettings.inBatchMode());
tableEnv.executeSql("create table datagen_source(id BIGINT, val BIGINT)
with " +
"('connector'='datagen', 'number-of-rows'='1000')");
tableEnv.executeSql("select " +
"count(distinct val * 1), " +
"count(distinct val * 2), " +
"count(distinct val * 3), " +
"count(distinct val * 4), " +
"count(distinct val * 5), " +
"count(distinct val * 6), " +
"count(distinct val * 7), " +
"count(distinct val * 8), " +
"count(distinct val * 9), " +
"count(distinct val * 10), " +
"count(distinct val * 11), " +
"count(distinct val * 12), " +
"count(distinct val * 13), " +
"count(distinct val * 14), " +
"count(distinct val * 15), " +
"count(distinct val * 16), " +
"count(distinct val * 17), " +
"count(distinct val * 18), " +
"count(distinct val * 19), " +
"count(distinct val * 20), " +
"count(distinct val * 21), " +
"count(distinct val * 22), " +
"count(distinct val * 23), " +
"count(distinct val * 24), " +
"count(distinct val * 25), " +
"count(distinct val * 26), " +
"count(distinct val * 27), " +
"count(distinct val * 28), " +
"count(distinct val * 29), " +
"count(distinct val * 30), " +
"count(distinct val * 31), " +
"count(distinct val * 32), " +
"count(distinct val * 33), " +
"count(distinct val * 34), " +
"count(distinct val * 35), " +
"count(distinct val * 36), " +
"count(distinct val * 37), " +
"count(distinct val * 38), " +
"count(distinct val * 39), " +
"count(distinct val * 40), " +
"count(distinct val * 41), " +
"count(distinct val * 42), " +
"count(distinct val * 43), " +
"count(distinct val * 44), " +
"count(distinct val * 45), " +
"count(distinct val * 46), " +
"count(distinct val * 47), " +
"count(distinct val * 48), " +
"count(distinct val * 49), " +
"count(distinct val * 50), " +
"count(distinct val * 51), " +
"count(distinct val * 52), " +
"count(distinct val * 53), " +
"count(distinct val * 54), " +
"count(distinct val * 55), " +
"count(distinct val * 56), " +
"count(distinct val * 57), " +
"count(distinct val * 58), " +
"count(distinct val * 59), " +
"count(distinct val * 60), " +
"count(distinct val * 61), " +
"count(distinct val * 62), " +
"count(distinct val * 63), " +
"count(distinct val * 64), " +
"count(distinct val * 65) from datagen_source").print();
}
} {code}
Exception:
{code:java}
Exception in thread "main" org.apache.flink.table.api.TableException: Sql
optimization: Cannot generate a valid execution plan for the given query:
LogicalSink(table=[*anonymous_collect$1*], fields=[EXPR$0, EXPR$1, EXPR$2,
EXPR$3, EXPR$4, EXPR$5, EXPR$6, EXPR$7, EXPR$8, EXPR$9, EXPR$10, EXPR$11,
EXPR$12, EXPR$13, EXPR$14, EXPR$15, EXPR$16, EXPR$17, EXPR$18, EXPR$19,
EXPR$20, EXPR$21, EXPR$22, EXPR$23, EXPR$24, EXPR$25, EXPR$26, EXPR$27,
EXPR$28, EXPR$29, EXPR$30, EXPR$31, EXPR$32, EXPR$33, EXPR$34, EXPR$35,
EXPR$36, EXPR$37, EXPR$38, EXPR$39, EXPR$40, EXPR$41, EXPR$42, EXPR$43,
EXPR$44, EXPR$45, EXPR$46, EXPR$47, EXPR$48, EXPR$49, EXPR$50, EXPR$51,
EXPR$52, EXPR$53, EXPR$54, EXPR$55, EXPR$56, EXPR$57, EXPR$58, EXPR$59,
EXPR$60, EXPR$61, EXPR$62, EXPR$63, EXPR$64])
+- LogicalAggregate(group=[{}], EXPR$0=[COUNT(DISTINCT $0)],
EXPR$1=[COUNT(DISTINCT $1)], EXPR$2=[COUNT(DISTINCT $2)],
EXPR$3=[COUNT(DISTINCT $3)], EXPR$4=[COUNT(DISTINCT $4)],
EXPR$5=[COUNT(DISTINCT $5)], EXPR$6=[COUNT(DISTINCT $6)],
EXPR$7=[COUNT(DISTINCT $7)], EXPR$8=[COUNT(DISTINCT $8)],
EXPR$9=[COUNT(DISTINCT $9)], EXPR$10=[COUNT(DISTINCT $10)],
EXPR$11=[COUNT(DISTINCT $11)], EXPR$12=[COUNT(DISTINCT $12)],
EXPR$13=[COUNT(DISTINCT $13)], EXPR$14=[COUNT(DISTINCT $14)],
EXPR$15=[COUNT(DISTINCT $15)], EXPR$16=[COUNT(DISTINCT $16)],
EXPR$17=[COUNT(DISTINCT $17)], EXPR$18=[COUNT(DISTINCT $18)],
EXPR$19=[COUNT(DISTINCT $19)], EXPR$20=[COUNT(DISTINCT $20)],
EXPR$21=[COUNT(DISTINCT $21)], EXPR$22=[COUNT(DISTINCT $22)],
EXPR$23=[COUNT(DISTINCT $23)], EXPR$24=[COUNT(DISTINCT $24)],
EXPR$25=[COUNT(DISTINCT $25)], EXPR$26=[COUNT(DISTINCT $26)],
EXPR$27=[COUNT(DISTINCT $27)], EXPR$28=[COUNT(DISTINCT $28)],
EXPR$29=[COUNT(DISTINCT $29)], EXPR$30=[COUNT(DISTINCT $30)],
EXPR$31=[COUNT(DISTINCT $31)], EXPR$32=[COUNT(DISTINCT $32)],
EXPR$33=[COUNT(DISTINCT $33)], EXPR$34=[COUNT(DISTINCT $34)],
EXPR$35=[COUNT(DISTINCT $35)], EXPR$36=[COUNT(DISTINCT $36)],
EXPR$37=[COUNT(DISTINCT $37)], EXPR$38=[COUNT(DISTINCT $38)],
EXPR$39=[COUNT(DISTINCT $39)], EXPR$40=[COUNT(DISTINCT $40)],
EXPR$41=[COUNT(DISTINCT $41)], EXPR$42=[COUNT(DISTINCT $42)],
EXPR$43=[COUNT(DISTINCT $43)], EXPR$44=[COUNT(DISTINCT $44)],
EXPR$45=[COUNT(DISTINCT $45)], EXPR$46=[COUNT(DISTINCT $46)],
EXPR$47=[COUNT(DISTINCT $47)], EXPR$48=[COUNT(DISTINCT $48)],
EXPR$49=[COUNT(DISTINCT $49)], EXPR$50=[COUNT(DISTINCT $50)],
EXPR$51=[COUNT(DISTINCT $51)], EXPR$52=[COUNT(DISTINCT $52)],
EXPR$53=[COUNT(DISTINCT $53)], EXPR$54=[COUNT(DISTINCT $54)],
EXPR$55=[COUNT(DISTINCT $55)], EXPR$56=[COUNT(DISTINCT $56)],
EXPR$57=[COUNT(DISTINCT $57)], EXPR$58=[COUNT(DISTINCT $58)],
EXPR$59=[COUNT(DISTINCT $59)], EXPR$60=[COUNT(DISTINCT $60)],
EXPR$61=[COUNT(DISTINCT $61)], EXPR$62=[COUNT(DISTINCT $62)],
EXPR$63=[COUNT(DISTINCT $63)], EXPR$64=[COUNT(DISTINCT $64)])
+- LogicalProject(exprs=[[*($1, 1), *($1, 2), *($1, 3), *($1, 4), *($1, 5),
*($1, 6), *($1, 7), *($1, 8), *($1, 9), *($1, 10), *($1, 11), *($1, 12), *($1,
13), *($1, 14), *($1, 15), *($1, 16), *($1, 17), *($1, 18), *($1, 19), *($1,
20), *($1, 21), *($1, 22), *($1, 23), *($1, 24), *($1, 25), *($1, 26), *($1,
27), *($1, 28), *($1, 29), *($1, 30), *($1, 31), *($1, 32), *($1, 33), *($1,
34), *($1, 35), *($1, 36), *($1, 37), *($1, 38), *($1, 39), *($1, 40), *($1,
41), *($1, 42), *($1, 43), *($1, 44), *($1, 45), *($1, 46), *($1, 47), *($1,
48), *($1, 49), *($1, 50), *($1, 51), *($1, 52), *($1, 53), *($1, 54), *($1,
55), *($1, 56), *($1, 57), *($1, 58), *($1, 59), *($1, 60), *($1, 61), *($1,
62), *($1, 63), *($1, 64), *($1, 65)]])
+- LogicalTableScan(table=[[default_catalog, default_database,
datagen_source]])group count must be less than 64.
Please check the documentation for the set of currently supported SQL features.
at
org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:86)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59)
at
scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
at
scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
at scala.collection.Iterator.foreach(Iterator.scala:937)
at scala.collection.Iterator.foreach$(Iterator.scala:937)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
at scala.collection.IterableLike.foreach(IterableLike.scala:70)
at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:55)
at
org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeTree(BatchCommonSubGraphBasedOptimizer.scala:92)
at
org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeBlock(BatchCommonSubGraphBasedOptimizer.scala:57)
at
org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1(BatchCommonSubGraphBasedOptimizer.scala:44)
at
org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1$adapted(BatchCommonSubGraphBasedOptimizer.scala:44)
at scala.collection.immutable.List.foreach(List.scala:388)
at
org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.doOptimize(BatchCommonSubGraphBasedOptimizer.scala:44)
at
org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:78)
at
org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:312)
at
org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:192)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1688)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:840)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1342)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:708)
at com.shopee.di.Test64Distinct.main(Test64Distinct.java:11)
Caused by: org.apache.flink.table.api.TableException: group count must be less
than 64.
at
org.apache.flink.table.planner.plan.rules.logical.DecomposeGroupingSetsRule.onMatch(DecomposeGroupingSetsRule.scala:177)
at
org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:229)
at
org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
at
org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
at
org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62)
... 27 more {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
