You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by "Wei Zhong (Jira)" <ji...@apache.org> on 2022/08/01 09:06:00 UTC
[jira] [Created] (FLINK-28764) Support more than 64 distinct aggregate function calls in one aggregate SQL query
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)