Apache Calcite流式SQL示例执行失败，请求技术支持

按照官方文档编写的流式SQL查询如下：

SELECT STREAM TUMBLE_END(ROWTIME, INTERVAL '1' HOUR) AS rowtime,
  PRODUCT,
  COUNT(*) AS c,
  SUM(UNITS) AS units
FROM STREAMS.ORDERS

执行时出现核心错误：cannot translate call $TUMBLE($t0, $t4)，完整报错堆栈如下：

Unable to implement EnumerableCalc(expr#0..3=[{inputs}],
expr#4=[3600000:INTERVAL HOUR], expr#5=[+($t0, $t4)], rowtime=[$t5],
PRODUCT=[$t1], c=[$t2], units=[$t3]): rowcount = 10.0, cumulative cost
= {172.5 rows, 1050.0 cpu, 0.0 io}, id = 102 EnumerableAggregate(group=[{0, 1}], c=[COUNT()], units=[$SUM0($2)]):
rowcount = 10.0, cumulative cost = {162.5 rows, 950.0 cpu, 0.0 io}, id
= 98
EnumerableCalc(expr#0..3=[{inputs}], expr#4=[3600000:INTERVAL HOUR], expr#5=[$TUMBLE($t0, $t4)], $f0=[$t5], PRODUCT=[$t2],
UNITS=[$t3]): rowcount = 100.0, cumulative cost = {150.0 rows, 950.0
cpu, 0.0 io}, id = 104
EnumerableInterpreter: rowcount = 100.0, cumulative cost = {50.0 rows, 50.0 cpu, 0.0 io}, id = 94
BindableTableScan(table=[[STREAMS, ORDERS, (STREAM)]]): rowcount = 100.0, cumulative cost = {100.0 rows, 101.0 cpu, 0.0 io},
id = 85

at org.apache.calcite.avatica.Helper.createException(Helper.java:56)
at org.apache.calcite.avatica.Helper.createException(Helper.java:41)
at
org.apache.calcite.avatica.AvaticaStatement.executeInternal(AvaticaStatement.java:164)
at
org.apache.calcite.avatica.AvaticaStatement.executeQuery(AvaticaStatement.java:228)
at
com.calcite.example.example.StreamExample.testStream(StreamExample.java:57)
at
com.calcite.example.example.StreamExample.main(StreamExample.java:22)
Caused by: java.lang.IllegalStateException: Unable to implement
EnumerableCalc(expr#0..3=[{inputs}], expr#4=[3600000:INTERVAL HOUR],
expr#5=[+($t0, $t4)], rowtime=[$t5], PRODUCT=[$t1], c=[$t2],
units=[$t3]): rowcount = 10.0, cumulative cost = {172.5 rows, 1050.0
cpu, 0.0 io}, id = 102 EnumerableAggregate(group=[{0, 1}],
c=[COUNT()], units=[$SUM0($2)]): rowcount = 10.0, cumulative cost =
{162.5 rows, 950.0 cpu, 0.0 io}, id = 98
EnumerableCalc(expr#0..3=[{inputs}], expr#4=[3600000:INTERVAL HOUR], expr#5=[$TUMBLE($t0, $t4)], $f0=[$t5], PRODUCT=[$t2],
UNITS=[$t3]): rowcount = 100.0, cumulative cost = {150.0 rows, 950.0
cpu, 0.0 io}, id = 104

EnumerableInterpreter: rowcount = 100.0, cumulative cost = {50.0 rows, 50.0 cpu, 0.0 io}, id = 94
    BindableTableScan(table=[[STREAMS, ORDERS, (STREAM)]]): rowcount = 100.0, cumulative cost = {100.0 rows, 101.0 cpu, 0.0 io}, id = 85

at org.apache.calcite.adapter.enumerable.EnumerableRelImplementor.implementRoot(EnumerableRelImplementor.java:117)
at org.apache.calcite.adapter.enumerable.EnumerableInterpretable.toBindable(EnumerableInterpretable.java:112)
at org.apache.calcite.prepare.CalcitePrepareImpl$CalcitePreparingStmt.implement(CalcitePrepareImpl.java:1159)
at org.apache.calcite.prepare.Prepare.prepareSql(Prepare.java:324)
at org.apache.calcite.prepare.Prepare.prepareSql(Prepare.java:220)
at org.apache.calcite.prepare.CalcitePrepareImpl.prepare2_(CalcitePrepareImpl.java:665)
at org.apache.calcite.prepare.CalcitePrepareImpl.prepare_(CalcitePrepareImpl.java:519)
at org.apache.calcite.prepare.CalcitePrepareImpl.prepareSql(CalcitePrepareImpl.java:487)
at org.apache.calcite.jdbc.CalciteConnectionImpl.parseQuery(CalciteConnectionImpl.java:236)
at org.apache.calcite.jdbc.CalciteMetaImpl.prepareAndExecute(CalciteMetaImpl.java:621)
at org.apache.calcite.avatica.AvaticaConnection.prepareAndExecuteInternal(AvaticaConnection.java:677)
at org.apache.calcite.avatica.AvaticaStatement.executeInternal(AvaticaStatement.java:157)
... 3 more
Suppressed: java.lang.RuntimeException: cannot translate call $TUMBLE($t0, $t4)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.visitCall(RexToLixTranslator.java:1172)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.visitCall(RexToLixTranslator.java:98)
    at org.apache.calcite.rex.RexCall.accept(RexCall.java:189)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.visitLocalRef(RexToLixTranslator.java:1059)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.visitLocalRef(RexToLixTranslator.java:98)
    at org.apache.calcite.rex.RexLocalRef.accept(RexLocalRef.java:77)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.translate(RexToLixTranslator.java:259)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.translate(RexToLixTranslator.java:253)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.translateList(RexToLixTranslator.java:899)
    at org.apache.calcite.adapter.enumerable.RexToLixTranslator.translateProjects(RexToLixTranslator.java:207)
    at org.apache.calcite.adapter.enumerable.EnumerableCalc.implement(EnumerableCalc.java:192)
    at org.apache.calcite.adapter.enumerable.EnumerableRelImplementor.visitChild(EnumerableRelImplementor.java:107)
    at org.apache.calcite.adapter.enumerable.EnumerableAggregate.implement(EnumerableAggregate.java:105)
    at org.apache.calcite.adapter.enumerable.EnumerableRelImplementor.visitChild(EnumerableRelImplementor.java:107)
    at org.apache.calcite.adapter.enumerable.EnumerableCalc.implement(EnumerableCalc.java:118)
    at org.apache.calcite.adapter.enumerable.EnumerableRelImplementor.implementRoot(EnumerableRelImplementor.java:114)
    ... 14 more

1. 补全GROUP BY窗口分组定义 #

TUMBLE窗口函数必须配合GROUP BY子句使用，Calcite需要明确的分组逻辑才能解析窗口计算。修正后的SQL如下：

SELECT STREAM TUMBLE_END(ROWTIME, INTERVAL '1' HOUR) AS rowtime,
  PRODUCT,
  COUNT(*) AS c,
  SUM(UNITS) AS units
FROM STREAMS.ORDERS
GROUP BY TUMBLE(ROWTIME, INTERVAL '1' HOUR), PRODUCT

2. 验证Calcite依赖与版本 #

确保项目依赖中包含流式查询所需的组件：

• 核心依赖：calcite-core
• 适配器依赖：calcite-linq4j（若使用Enumerable执行器）
  同时确认Calcite版本不低于1.20.0（该版本后流式窗口支持更完善）。

3. 确认流表属性配置 #

确保STREAMS.ORDERS表在元数据中被标记为流表，表定义需包含stream: true属性。例如在JSON模型中配置：

{
  "tables": [
    {
      "name": "ORDERS",
      "type": "stream",
      "columns": [
        {"name": "ROWTIME", "type": "TIMESTAMP"},
        {"name": "PRODUCT", "type": "VARCHAR"},
        {"name": "UNITS", "type": "INTEGER"}
      ]
    }
  ]
}

内容的提问来源于stack exchange，提问作者xiongyunkun