You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2021/08/16 19:47:25 UTC
[flink] branch master updated: [FLINK-21290][table-planner] Support
project push down for window TVF
This is an automated email from the ASF dual-hosted git repository.
godfrey 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 25bce2f [FLINK-21290][table-planner] Support project push down for window TVF
25bce2f is described below
commit 25bce2ff3a9f94665e0ad552b4dc91a1e478d36b
Author: Jing <be...@126.com>
AuthorDate: Tue Aug 3 17:31:15 2021 +0800
[FLINK-21290][table-planner] Support project push down for window TVF
This closes #16689
---
.../ProjectWindowTableFunctionTransposeRule.java | 208 +++++++++++++
.../planner/plan/rules/FlinkStreamRuleSets.scala | 4 +-
.../testDistinctSplitEnabled.out | 29 +-
.../testEventTimeCumulateWindow.out | 29 +-
.../testEventTimeCumulateWindowWithOffset.out | 29 +-
.../testEventTimeHopWindow.out | 29 +-
.../testEventTimeHopWindowWithOffset.out | 29 +-
.../testEventTimeTumbleWindow.out | 29 +-
.../testEventTimeTumbleWindowWithOffset.out | 29 +-
.../testProcTimeCumulateWindow.out | 96 +++---
.../testProcTimeTumbleWindow.out | 98 +++----
.../testEventTimeTumbleWindow.out | 294 ++++---------------
.../testFollowedByWindowJoin.out | 80 +----
...ProjectWindowTableFunctionTransposeRuleTest.xml | 94 ++++++
.../ExpandWindowTableFunctionTransposeRuleTest.xml | 9 +-
.../planner/plan/stream/sql/WindowRankTest.xml | 19 +-
.../plan/stream/sql/agg/WindowAggregateTest.xml | 321 +++++++--------------
.../plan/stream/sql/join/WindowJoinTest.xml | 284 +++++++-----------
...ojectWindowTableFunctionTransposeRuleTest.scala | 103 +++++++
.../plan/stream/sql/agg/WindowAggregateTest.scala | 1 -
20 files changed, 794 insertions(+), 1020 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRule.java
new file mode 100644
index 0000000..165ea77
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRule.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.functions.sql.SqlWindowTableFunction;
+import org.apache.flink.table.planner.plan.logical.TimeAttributeWindowingStrategy;
+import org.apache.flink.table.planner.plan.utils.WindowUtil;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rel.logical.LogicalTableFunctionScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.Pair;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Planner rule that pushes a {@link LogicalProject} into a {@link LogicalTableFunctionScan} which
+ * contains a Window table function call by splitting the projection into a projection on top of
+ * child of the TableFunctionScan.
+ */
+public class ProjectWindowTableFunctionTransposeRule extends RelOptRule {
+
+ public static final ProjectWindowTableFunctionTransposeRule INSTANCE =
+ new ProjectWindowTableFunctionTransposeRule();
+
+ public ProjectWindowTableFunctionTransposeRule() {
+ super(
+ operand(LogicalProject.class, operand(LogicalTableFunctionScan.class, any())),
+ "ProjectWindowTableFunctionTransposeRule");
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ LogicalTableFunctionScan scan = call.rel(1);
+ return WindowUtil.isWindowTableFunctionCall(scan.getCall());
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ LogicalProject project = call.rel(0);
+ LogicalTableFunctionScan scan = call.rel(1);
+ RelNode scanInput = scan.getInput(0);
+ TimeAttributeWindowingStrategy windowingStrategy =
+ WindowUtil.convertToWindowingStrategy(
+ (RexCall) scan.getCall(), scanInput.getRowType());
+ // 1. get fields to push down
+ ImmutableBitSet projectFields = RelOptUtil.InputFinder.bits(project.getProjects(), null);
+ int scanInputFieldCount = scanInput.getRowType().getFieldCount();
+ ImmutableBitSet toPushFields =
+ ImmutableBitSet.range(0, scanInputFieldCount)
+ .intersect(projectFields)
+ .set(windowingStrategy.getTimeAttributeIndex());
+ if (toPushFields.cardinality() == scanInputFieldCount) {
+ return;
+ }
+
+ // 2. create new input of window table function scan
+ RelBuilder relBuilder = call.builder();
+ RelNode newScanInput = createInnerProject(relBuilder, scanInput, toPushFields);
+
+ // mapping origin field index to new field index, used to rewrite WindowTableFunction and
+ // top project
+ Map<Integer, Integer> mapping =
+ getFieldMapping(
+ scan.getRowType().getFieldCount(), scanInputFieldCount, toPushFields);
+
+ // 3. create new window table function scan
+ LogicalTableFunctionScan newScan =
+ createNewTableFunctionScan(
+ relBuilder,
+ scan,
+ windowingStrategy.getTimeAttributeType(),
+ newScanInput,
+ mapping);
+
+ // 4. create top project
+ RelNode topProject = createTopProject(relBuilder, project, newScan, mapping);
+ call.transformTo(topProject);
+ }
+
+ private Map<Integer, Integer> getFieldMapping(
+ int scanFieldCount, int scanInputFieldCount, ImmutableBitSet toPushFields) {
+ int toPushFieldCount = toPushFields.cardinality();
+ Map<Integer, Integer> mapping = new HashMap<>();
+ IntStream.range(0, scanFieldCount)
+ .forEach(
+ idx -> {
+ int newPosition;
+ if (idx < scanInputFieldCount) {
+ newPosition = toPushFields.indexOf(idx);
+ } else {
+ newPosition = toPushFieldCount + idx - scanInputFieldCount;
+ }
+ mapping.put(idx, newPosition);
+ });
+ return mapping;
+ }
+
+ private RelNode createInnerProject(
+ RelBuilder relBuilder, RelNode scanInput, ImmutableBitSet toPushFields) {
+ relBuilder.push(scanInput);
+ List<RexInputRef> newProjects =
+ toPushFields.toList().stream().map(relBuilder::field).collect(Collectors.toList());
+ return relBuilder.project(newProjects).build();
+ }
+
+ private LogicalTableFunctionScan createNewTableFunctionScan(
+ RelBuilder relBuilder,
+ LogicalTableFunctionScan oldScan,
+ LogicalType timeAttributeType,
+ RelNode newInput,
+ Map<Integer, Integer> mapping) {
+ relBuilder.push(newInput);
+ RexNode newCall = rewriteWindowCall((RexCall) oldScan.getCall(), mapping, relBuilder);
+ RelOptCluster cluster = oldScan.getCluster();
+ FlinkTypeFactory typeFactory = (FlinkTypeFactory) cluster.getTypeFactory();
+ RelDataType newScanOutputType =
+ SqlWindowTableFunction.inferRowType(
+ typeFactory,
+ newInput.getRowType(),
+ typeFactory.createFieldTypeFromLogicalType(timeAttributeType));
+ return LogicalTableFunctionScan.create(
+ cluster,
+ new ArrayList<>(Collections.singleton(newInput)),
+ newCall,
+ oldScan.getElementType(),
+ newScanOutputType,
+ oldScan.getColumnMappings());
+ }
+
+ private RexNode rewriteWindowCall(
+ RexCall windowCall, Map<Integer, Integer> mapping, RelBuilder relBuilder) {
+ List<RexNode> newOperands = new ArrayList<>();
+ Iterator<RexNode> operandsItr = windowCall.getOperands().iterator();
+ // Note: skip to rewrite the first operand of window table function because it is a special
+ // ref to a table instead of a normal input ref, if process it as a regular input ref, an
+ // exception would be thrown out. It's safe to use first operand of function because
+ // framework never use it (or avoid to use it).
+ newOperands.add(operandsItr.next());
+ while (operandsItr.hasNext()) {
+ newOperands.add(adjustInputRef(operandsItr.next(), mapping));
+ }
+ return relBuilder.call(windowCall.getOperator(), newOperands);
+ }
+
+ private RelNode createTopProject(
+ RelBuilder relBuilder,
+ LogicalProject oldProject,
+ LogicalTableFunctionScan newInput,
+ Map<Integer, Integer> mapping) {
+ List<Pair<RexNode, String>> newTopProjects =
+ oldProject.getNamedProjects().stream()
+ .map(r -> Pair.of(adjustInputRef(r.left, mapping), r.right))
+ .collect(Collectors.toList());
+ return relBuilder
+ .push(newInput)
+ .project(Pair.left(newTopProjects), Pair.right(newTopProjects))
+ .build();
+ }
+
+ private RexNode adjustInputRef(RexNode expr, Map<Integer, Integer> mapping) {
+ return expr.accept(
+ new RexShuttle() {
+
+ @Override
+ public RexNode visitInputRef(RexInputRef inputRef) {
+ Integer newIndex = mapping.get(inputRef.getIndex());
+ return new RexInputRef(newIndex, inputRef.getType());
+ }
+ });
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
index 2ce8b9c..6a06e6d 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
@@ -215,7 +215,9 @@ object FlinkStreamRuleSets {
//removes constant keys from an Agg
CoreRules.AGGREGATE_PROJECT_PULL_UP_CONSTANTS,
// push project through a Union
- CoreRules.PROJECT_SET_OP_TRANSPOSE
+ CoreRules.PROJECT_SET_OP_TRANSPOSE,
+ // push a projection to the child of a WindowTableFunctionScan
+ ProjectWindowTableFunctionTransposeRule.INSTANCE
)
val JOIN_REORDER_PREPARE_RULES: RuleSet = RuleSets.ofList(
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out
index 7e131f5..df0f6d2 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testDistinctSplitEnabled.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindow.out
index 5a48981..60984eb 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindow.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindowWithOffset.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindowWithOffset.out
index eb4e2fa..9f57c7a 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindowWithOffset.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeCumulateWindowWithOffset.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindow.out
index 9906d70..10901cb 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindow.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindowWithOffset.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindowWithOffset.out
index 4f168c4..b0d63ee 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindowWithOffset.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeHopWindowWithOffset.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
index 011f243..3cc1eb7 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out
index da5f735..c757f85 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindowWithOffset.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeCumulateWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeCumulateWindow.out
index cd6f15b..e9b78ce 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeCumulateWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeCumulateWindow.out
@@ -25,21 +25,32 @@
"schema.3.expr" : "TO_TIMESTAMP(`c`)",
"schema.4.data-type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL",
"schema.0.name" : "a"
- }
+ },
+ "sourceAbilitySpecs" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 1 ], [ 2 ] ],
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "b" : "BIGINT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ }
+ } ]
},
"id" : 1,
"outputType" : {
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "a" : "INT"
- }, {
"b" : "BIGINT"
}, {
"c" : "VARCHAR(2147483647)"
} ]
},
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c]]], fields=[b, c])",
"inputProperties" : [ ]
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
@@ -47,19 +58,12 @@
"kind" : "INPUT_REF",
"inputIndex" : 0,
"type" : {
- "typeName" : "INTEGER",
- "nullable" : true
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : {
"typeName" : "BIGINT",
"nullable" : true
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "VARCHAR",
"nullable" : true,
@@ -68,13 +72,26 @@
}, {
"kind" : "REX_CALL",
"operator" : {
+ "name" : "PROCTIME",
+ "kind" : "OTHER_FUNCTION",
+ "syntax" : "FUNCTION"
+ },
+ "operands" : [ ],
+ "type" : {
+ "timestampKind" : "PROCTIME",
+ "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false
+ }
+ }, {
+ "kind" : "REX_CALL",
+ "operator" : {
"name" : "TO_TIMESTAMP",
"kind" : "OTHER_FUNCTION",
"syntax" : "FUNCTION"
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "VARCHAR",
"nullable" : true,
@@ -86,19 +103,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -113,28 +117,26 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "a" : "INT"
- }, {
"b" : "BIGINT"
}, {
"c" : "VARCHAR(2147483647)"
}, {
- "rowtime" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : true,
- "precision" : 3,
- "kind" : "REGULAR"
- }
- }, {
"proctime" : {
"type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
"nullable" : false,
"precision" : 3,
"kind" : "PROCTIME"
}
+ }, {
+ "rowtime" : {
+ "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
+ "nullable" : true,
+ "precision" : 3,
+ "kind" : "REGULAR"
+ }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[b, c, PROCTIME() AS proctime, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -181,25 +183,23 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "a" : "INT"
- }, {
"b" : "BIGINT"
}, {
"c" : "VARCHAR(2147483647)"
}, {
- "rowtime" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : true,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- }, {
"proctime" : {
"type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
"nullable" : false,
"precision" : 3,
"kind" : "PROCTIME"
}
+ }, {
+ "rowtime" : {
+ "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
+ "nullable" : true,
+ "precision" : 3,
+ "kind" : "ROWTIME"
+ }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
@@ -207,14 +207,14 @@
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 1,
+ "inputIndex" : 0,
"type" : {
"typeName" : "BIGINT",
"nullable" : true
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "VARCHAR",
"nullable" : true,
@@ -222,7 +222,7 @@
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 4,
+ "inputIndex" : 2,
"type" : {
"timestampKind" : "PROCTIME",
"typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
index 957e7d0..20bcc03 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
@@ -25,21 +25,32 @@
"schema.3.expr" : "TO_TIMESTAMP(`c`)",
"schema.4.data-type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL",
"schema.0.name" : "a"
- }
+ },
+ "sourceAbilitySpecs" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 1 ], [ 2 ] ],
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "b" : "BIGINT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ }
+ } ]
},
"id" : 1,
"outputType" : {
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "a" : "INT"
- }, {
"b" : "BIGINT"
}, {
"c" : "VARCHAR(2147483647)"
} ]
},
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c]]], fields=[b, c])",
"inputProperties" : [ ]
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
@@ -47,23 +58,21 @@
"kind" : "INPUT_REF",
"inputIndex" : 0,
"type" : {
- "typeName" : "INTEGER",
- "nullable" : true
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 1,
- "type" : {
"typeName" : "BIGINT",
"nullable" : true
}
}, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "kind" : "REX_CALL",
+ "operator" : {
+ "name" : "PROCTIME",
+ "kind" : "OTHER_FUNCTION",
+ "syntax" : "FUNCTION"
+ },
+ "operands" : [ ],
"type" : {
- "typeName" : "VARCHAR",
- "nullable" : true,
- "precision" : 2147483647
+ "timestampKind" : "PROCTIME",
+ "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false
}
}, {
"kind" : "REX_CALL",
@@ -74,7 +83,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "VARCHAR",
"nullable" : true,
@@ -86,19 +95,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -113,11 +109,14 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "a" : "INT"
- }, {
"b" : "BIGINT"
}, {
- "c" : "VARCHAR(2147483647)"
+ "proctime" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
}, {
"rowtime" : {
"type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
@@ -125,16 +124,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[b, PROCTIME() AS proctime, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -146,7 +138,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 3,
+ "inputIndex" : 2,
"type" : {
"typeName" : "TIMESTAMP",
"nullable" : true,
@@ -168,7 +160,7 @@
"precision" : 3
}
},
- "rowtimeFieldIndex" : 3,
+ "rowtimeFieldIndex" : 2,
"id" : 3,
"inputProperties" : [ {
"requiredDistribution" : {
@@ -181,11 +173,14 @@
"type" : "ROW",
"nullable" : true,
"fields" : [ {
- "a" : "INT"
- }, {
"b" : "BIGINT"
}, {
- "c" : "VARCHAR(2147483647)"
+ "proctime" : {
+ "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
+ "nullable" : false,
+ "precision" : 3,
+ "kind" : "PROCTIME"
+ }
}, {
"rowtime" : {
"type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
@@ -193,13 +188,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
@@ -207,14 +195,14 @@
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 1,
+ "inputIndex" : 0,
"type" : {
"typeName" : "BIGINT",
"nullable" : true
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 4,
+ "inputIndex" : 1,
"type" : {
"timestampKind" : "PROCTIME",
"typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
index f432747..e2443ea 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
@@ -25,7 +25,20 @@
"schema.3.expr" : "TO_TIMESTAMP(`c`)",
"schema.4.data-type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL",
"schema.0.name" : "a"
- }
+ },
+ "sourceAbilitySpecs" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 2 ] ],
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "a" : "INT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ }
+ } ]
},
"id" : 1,
"outputType" : {
@@ -34,12 +47,10 @@
"fields" : [ {
"a" : "INT"
}, {
- "b" : "BIGINT"
- }, {
"c" : "VARCHAR(2147483647)"
} ]
},
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c]]], fields=[a, c])",
"inputProperties" : [ ]
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
@@ -54,13 +65,6 @@
"kind" : "INPUT_REF",
"inputIndex" : 1,
"type" : {
- "typeName" : "BIGINT",
- "nullable" : true
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : {
"typeName" : "VARCHAR",
"nullable" : true,
"precision" : 2147483647
@@ -74,7 +78,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "VARCHAR",
"nullable" : true,
@@ -86,19 +90,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -115,8 +106,6 @@
"fields" : [ {
"a" : "INT"
}, {
- "b" : "BIGINT"
- }, {
"c" : "VARCHAR(2147483647)"
}, {
"rowtime" : {
@@ -125,16 +114,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -146,7 +128,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 3,
+ "inputIndex" : 2,
"type" : {
"typeName" : "TIMESTAMP",
"nullable" : true,
@@ -168,7 +150,7 @@
"precision" : 3
}
},
- "rowtimeFieldIndex" : 3,
+ "rowtimeFieldIndex" : 2,
"id" : 3,
"inputProperties" : [ {
"requiredDistribution" : {
@@ -183,8 +165,6 @@
"fields" : [ {
"a" : "INT"
}, {
- "b" : "BIGINT"
- }, {
"c" : "VARCHAR(2147483647)"
}, {
"rowtime" : {
@@ -193,69 +173,10 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
}, {
- "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : {
- "typeName" : "INTEGER",
- "nullable" : true
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : {
- "typeName" : "VARCHAR",
- "nullable" : true,
- "precision" : 2147483647
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : {
- "timestampKind" : "ROWTIME",
- "typeName" : "TIMESTAMP",
- "nullable" : true
- }
- } ],
- "condition" : null,
- "id" : 4,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "nullable" : true,
- "fields" : [ {
- "a" : "INT"
- }, {
- "c" : "VARCHAR(2147483647)"
- }, {
- "rowtime" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : true,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "Calc(select=[a, c, rowtime])"
- }, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLocalWindowAggregate",
"grouping" : [ 0 ],
"aggCalls" : [ {
@@ -306,7 +227,7 @@
"timeAttributeIndex" : 2,
"isRowtime" : true
},
- "id" : 5,
+ "id" : 4,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -344,7 +265,7 @@
"description" : "LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
- "id" : 6,
+ "id" : 5,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "HASH",
@@ -475,7 +396,7 @@
}
}
} ],
- "id" : 7,
+ "id" : 6,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -573,7 +494,7 @@
}
} ],
"condition" : null,
- "id" : 8,
+ "id" : 7,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -609,7 +530,7 @@
"description" : "Calc(select=[a, window_start, window_end, cnt, uv])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
- "id" : 9,
+ "id" : 8,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "HASH",
@@ -669,21 +590,32 @@
"schema.3.expr" : "TO_TIMESTAMP(`c`)",
"schema.4.data-type" : "TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL",
"schema.0.name" : "a"
- }
+ },
+ "sourceAbilitySpecs" : [ {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ], [ 2 ] ],
+ "producedType" : {
+ "type" : "ROW",
+ "nullable" : false,
+ "fields" : [ {
+ "a" : "INT"
+ }, {
+ "c" : "VARCHAR(2147483647)"
+ } ]
+ }
+ } ]
},
- "id" : 10,
+ "id" : 9,
"outputType" : {
"type" : "ROW",
"nullable" : true,
"fields" : [ {
"a" : "INT"
}, {
- "b" : "BIGINT"
- }, {
"c" : "VARCHAR(2147483647)"
} ]
},
- "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c])",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c]]], fields=[a, c])",
"inputProperties" : [ ]
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
@@ -698,13 +630,6 @@
"kind" : "INPUT_REF",
"inputIndex" : 1,
"type" : {
- "typeName" : "BIGINT",
- "nullable" : true
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : {
"typeName" : "VARCHAR",
"nullable" : true,
"precision" : 2147483647
@@ -718,7 +643,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 2,
+ "inputIndex" : 1,
"type" : {
"typeName" : "VARCHAR",
"nullable" : true,
@@ -730,22 +655,9 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
- "id" : 11,
+ "id" : 10,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -759,8 +671,6 @@
"fields" : [ {
"a" : "INT"
}, {
- "b" : "BIGINT"
- }, {
"c" : "VARCHAR(2147483647)"
}, {
"rowtime" : {
@@ -769,16 +679,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -790,7 +693,7 @@
},
"operands" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 3,
+ "inputIndex" : 2,
"type" : {
"typeName" : "TIMESTAMP",
"nullable" : true,
@@ -812,8 +715,8 @@
"precision" : 3
}
},
- "rowtimeFieldIndex" : 3,
- "id" : 12,
+ "rowtimeFieldIndex" : 2,
+ "id" : 11,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -827,8 +730,6 @@
"fields" : [ {
"a" : "INT"
}, {
- "b" : "BIGINT"
- }, {
"c" : "VARCHAR(2147483647)"
}, {
"rowtime" : {
@@ -837,69 +738,10 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
}, {
- "class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
- "projection" : [ {
- "kind" : "INPUT_REF",
- "inputIndex" : 0,
- "type" : {
- "typeName" : "INTEGER",
- "nullable" : true
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 2,
- "type" : {
- "typeName" : "VARCHAR",
- "nullable" : true,
- "precision" : 2147483647
- }
- }, {
- "kind" : "INPUT_REF",
- "inputIndex" : 3,
- "type" : {
- "timestampKind" : "ROWTIME",
- "typeName" : "TIMESTAMP",
- "nullable" : true
- }
- } ],
- "condition" : null,
- "id" : 13,
- "inputProperties" : [ {
- "requiredDistribution" : {
- "type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
- } ],
- "outputType" : {
- "type" : "ROW",
- "nullable" : true,
- "fields" : [ {
- "a" : "INT"
- }, {
- "c" : "VARCHAR(2147483647)"
- }, {
- "rowtime" : {
- "type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
- "nullable" : true,
- "precision" : 3,
- "kind" : "ROWTIME"
- }
- } ]
- },
- "description" : "Calc(select=[a, c, rowtime])"
- }, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLocalWindowAggregate",
"grouping" : [ 0 ],
"aggCalls" : [ {
@@ -950,7 +792,7 @@
"timeAttributeIndex" : 2,
"isRowtime" : true
},
- "id" : 14,
+ "id" : 12,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -988,7 +830,7 @@
"description" : "LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
- "id" : 15,
+ "id" : 13,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "HASH",
@@ -1119,7 +961,7 @@
}
}
} ],
- "id" : 16,
+ "id" : 14,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -1217,7 +1059,7 @@
}
} ],
"condition" : null,
- "id" : 17,
+ "id" : 15,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -1253,7 +1095,7 @@
"description" : "Calc(select=[a, window_start, window_end, cnt, uv])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecExchange",
- "id" : 18,
+ "id" : 16,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "HASH",
@@ -1329,7 +1171,7 @@
"windowEnd" : 2,
"isRowtime" : true
},
- "id" : 19,
+ "id" : 17,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -1451,7 +1293,7 @@
}
} ],
"condition" : null,
- "id" : 20,
+ "id" : 18,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -1520,7 +1362,7 @@
}
},
"inputChangelogMode" : [ "INSERT" ],
- "id" : 21,
+ "id" : 19,
"inputProperties" : [ {
"requiredDistribution" : {
"type" : "UNKNOWN"
@@ -1611,8 +1453,8 @@
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 8,
- "target" : 9,
+ "source" : 9,
+ "target" : 10,
"shuffle" : {
"type" : "FORWARD"
},
@@ -1660,22 +1502,22 @@
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 16,
+ "source" : 8,
"target" : 17,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 17,
- "target" : 18,
+ "source" : 16,
+ "target" : 17,
"shuffle" : {
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
}, {
- "source" : 9,
- "target" : 19,
+ "source" : 17,
+ "target" : 18,
"shuffle" : {
"type" : "FORWARD"
},
@@ -1687,19 +1529,5 @@
"type" : "FORWARD"
},
"shuffleMode" : "PIPELINED"
- }, {
- "source" : 19,
- "target" : 20,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
- }, {
- "source" : 20,
- "target" : 21,
- "shuffle" : {
- "type" : "FORWARD"
- },
- "shuffleMode" : "PIPELINED"
} ]
}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowJoin.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowJoin.out
index e297776..b65c760 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowJoin.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowTableFunctionJsonPlanTest_jsonplan/testFollowedByWindowJoin.out
@@ -86,19 +86,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 2,
@@ -125,16 +112,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -193,13 +173,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
@@ -246,13 +219,6 @@
"kind" : "ROWTIME"
}
}, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- }, {
"window_start" : {
"type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
"nullable" : false,
@@ -280,7 +246,7 @@
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 5,
+ "inputIndex" : 4,
"type" : {
"typeName" : "TIMESTAMP",
"nullable" : false,
@@ -288,7 +254,7 @@
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 6,
+ "inputIndex" : 5,
"type" : {
"typeName" : "TIMESTAMP",
"nullable" : false,
@@ -501,19 +467,6 @@
"nullable" : true,
"precision" : 3
}
- }, {
- "kind" : "REX_CALL",
- "operator" : {
- "name" : "PROCTIME",
- "kind" : "OTHER_FUNCTION",
- "syntax" : "FUNCTION"
- },
- "operands" : [ ],
- "type" : {
- "timestampKind" : "PROCTIME",
- "typeName" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false
- }
} ],
"condition" : null,
"id" : 8,
@@ -540,16 +493,9 @@
"precision" : 3,
"kind" : "REGULAR"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
- "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime, PROCTIME() AS proctime])"
+ "description" : "Calc(select=[a, b, c, TO_TIMESTAMP(c) AS rowtime])"
}, {
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecWatermarkAssigner",
"watermarkExpr" : {
@@ -608,13 +554,6 @@
"precision" : 3,
"kind" : "ROWTIME"
}
- }, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
} ]
},
"description" : "WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])"
@@ -661,13 +600,6 @@
"kind" : "ROWTIME"
}
}, {
- "proctime" : {
- "type" : "TIMESTAMP_WITH_LOCAL_TIME_ZONE",
- "nullable" : false,
- "precision" : 3,
- "kind" : "PROCTIME"
- }
- }, {
"window_start" : {
"type" : "TIMESTAMP_WITHOUT_TIME_ZONE",
"nullable" : false,
@@ -695,7 +627,7 @@
"class" : "org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecCalc",
"projection" : [ {
"kind" : "INPUT_REF",
- "inputIndex" : 5,
+ "inputIndex" : 4,
"type" : {
"typeName" : "TIMESTAMP",
"nullable" : false,
@@ -703,7 +635,7 @@
}
}, {
"kind" : "INPUT_REF",
- "inputIndex" : 6,
+ "inputIndex" : 5,
"type" : {
"typeName" : "TIMESTAMP",
"nullable" : false,
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRuleTest.xml
new file mode 100644
index 0000000..bba27f8
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRuleTest.xml
@@ -0,0 +1,94 @@
+<?xml version="1.0" ?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to you under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<Root>
+ <TestCase name="testNoColumnToPrune">
+ <Resource name="sql">
+ <![CDATA[
+SELECT
+ a,
+ window_start,
+ window_end,
+ count(*),
+ sum(d),
+ max(d) filter (where b > 1000),
+ weightedAvg(b, e) AS wAvg,
+ count(distinct c) AS uv
+FROM TABLE(
+ CUMULATE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '10' MINUTE, INTERVAL '1' HOUR))
+GROUP BY a, window_start, window_end
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{0, 1, 2}], EXPR$3=[COUNT()], EXPR$4=[SUM($3)], EXPR$5=[MAX($3) FILTER $4], wAvg=[weightedAvg($5, $6)], uv=[COUNT(DISTINCT $7)])
++- LogicalProject(a=[$0], window_start=[$6], window_end=[$7], d=[$3], $f4=[IS TRUE(>($1, 1000))], b=[$1], e=[$4], c=[$2])
+ +- LogicalTableFunctionScan(invocation=[CUMULATE($5, DESCRIPTOR($5), 600000:INTERVAL MINUTE, 3600000:INTERVAL HOUR)], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3) d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], rowtime=[$5])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5, 1000:INTERVAL SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalAggregate(group=[{0, 1, 2}], EXPR$3=[COUNT()], EXPR$4=[SUM($3)], EXPR$5=[MAX($3) FILTER $4], wAvg=[weightedAvg($5, $6)], uv=[COUNT(DISTINCT $7)])
++- LogicalProject(a=[$0], window_start=[$6], window_end=[$7], d=[$3], $f4=[IS TRUE(>($1, 1000))], b=[$1], e=[$4], c=[$2])
+ +- LogicalTableFunctionScan(invocation=[CUMULATE($5, DESCRIPTOR($5), 600000:INTERVAL MINUTE, 3600000:INTERVAL HOUR)], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3) d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], rowtime=[$5])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5, 1000:INTERVAL SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testPruneUnusedColumn">
+ <Resource name="sql">
+ <![CDATA[
+SELECT
+ window_start,
+ window_end,
+ count(*),
+ sum(d),
+ max(d) filter (where b > 1000),
+ count(distinct c) AS uv
+FROM TABLE(
+ CUMULATE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '10' MINUTE, INTERVAL '1' HOUR))
+GROUP BY window_start, window_end
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{0, 1}], EXPR$2=[COUNT()], EXPR$3=[SUM($2)], EXPR$4=[MAX($2) FILTER $3], uv=[COUNT(DISTINCT $4)])
++- LogicalProject(window_start=[$6], window_end=[$7], d=[$3], $f3=[IS TRUE(>($1, 1000))], c=[$2])
+ +- LogicalTableFunctionScan(invocation=[CUMULATE($5, DESCRIPTOR($5), 600000:INTERVAL MINUTE, 3600000:INTERVAL HOUR)], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3) d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], rowtime=[$5])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5, 1000:INTERVAL SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalAggregate(group=[{0, 1}], EXPR$2=[COUNT()], EXPR$3=[SUM($2)], EXPR$4=[MAX($2) FILTER $3], uv=[COUNT(DISTINCT $4)])
++- LogicalProject(window_start=[$4], window_end=[$5], d=[$2], $f3=[IS TRUE(>($0, 1000))], c=[$1])
+ +- LogicalTableFunctionScan(invocation=[CUMULATE($5, DESCRIPTOR($3), 600000:INTERVAL MINUTE, 3600000:INTERVAL HOUR)], rowType=[RecordType(BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3) d, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ +- LogicalProject(b=[$1], c=[$2], d=[$3], rowtime=[$5])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], rowtime=[$5])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5, 1000:INTERVAL SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ExpandWindowTableFunctionTransposeRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ExpandWindowTableFunctionTransposeRuleTest.xml
index e06c9e1..623033d 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ExpandWindowTableFunctionTransposeRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/ExpandWindowTableFunctionTransposeRuleTest.xml
@@ -56,8 +56,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -100,8 +99,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -145,8 +143,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.xml
index b54632f..c3061ad 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.xml
@@ -69,8 +69,7 @@ Calc(select=[window_start, window_end, window_time, a, cnt, sum_d, max_d, wAvg,
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f5 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f5, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -125,8 +124,7 @@ Calc(select=[window_start, window_end, window_time, a, cnt, sum_d, max_d, wAvg,
+- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f5 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f5, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -179,8 +177,7 @@ Calc(select=[window_start, window_end, window_time, a, cnt, sum_d, max_d, wAvg,
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f5, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -234,8 +231,7 @@ Calc(select=[window_start, window_end, window_time, a, cnt, sum_d, max_d, wAvg,
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f5 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f5, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -314,16 +310,15 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[1 h])], select=[a, SUM(cnt) AS sum$0, SUM(sum_d) AS sum$1, MAX(max_d) AS max$2, slice_end('w$) AS $slice_end])
+- Calc(select=[a, cnt, sum_d, max_d, window_time AS rowtime])
- +- WindowRank(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3], partitionBy=[a], orderBy=[cnt DESC], select=[a, window_start, window_end, window_time, cnt, sum_d, max_d, wAvg, uv])
+ +- WindowRank(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3], partitionBy=[a], orderBy=[cnt DESC], select=[a, window_start, window_end, window_time, cnt, sum_d, max_d])
+- Exchange(distribution=[hash[a]])
- +- Calc(select=[a, window_start, window_end, window_time, cnt, sum_d, max_d, wAvg, uv])
+ +- Calc(select=[a, window_start, window_end, window_time, cnt, sum_d, max_d])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, SUM(sum$1) AS sum_d, MAX(max$2) AS max_d, weightedAvg(weightedavg$3) AS wAvg, COUNT(distinct$0 count$4) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f5 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f5, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml
index 215d370..5e37c42 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml
@@ -55,8 +55,7 @@ Calc(select=[window_start, window_end, EXPR$2, EXPR$3, EXPR$4, wAvg, uv])
+- Calc(select=[window_start, window_end, d, IS TRUE(>(b, 1000)) AS $f3, b, e, c], where=[>=(window_start, 2021-01-01 10:10:00:TIMESTAMP(3))])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, e, rowtime]]], fields=[b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -100,8 +99,7 @@ Calc(select=[window_start, window_end, EXPR$2, EXPR$3, EXPR$4, wAvg, uv])
+- Calc(select=[window_start, window_end, d, IS TRUE(>(b, 1000)) AS $f3, b, e, c], where=[>=(window_start, 2021-01-01 10:10:00:TIMESTAMP(3))])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, e, rowtime]]], fields=[b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -145,8 +143,7 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS EXPR$3, $f5 AS uv])
+- Calc(select=[a, b, window_start, window_end, d, IS TRUE(>(b, 1000)) AS $f5, window_time])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, d, rowtime]]], fields=[a, b, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -192,8 +189,7 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS EXPR$3, $f5 AS uv])
+- Calc(select=[a, b, window_start, window_end, d, IS TRUE(>(b, 1000)) AS $f5, window_time])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, d, rowtime]]], fields=[a, b, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -322,8 +318,7 @@ Calc(select=[a, b, uv])
+- Calc(select=[a, b, window_start, window_end, c])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -358,8 +353,7 @@ Calc(select=[a, uv])
+- Calc(select=[a, window_start, c])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -394,8 +388,7 @@ Calc(select=[a, uv])
+- Calc(select=[a, window_start, window_end, c])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -431,8 +424,7 @@ GroupAggregate(groupBy=[window_start], select=[window_start, COUNT(*) AS EXPR$1,
+- Calc(select=[window_start, d, IS TRUE(>(b, 1000)) AS $f2, b, e, c])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, e, rowtime]]], fields=[b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -465,8 +457,7 @@ PythonGroupAggregate(groupBy=[window_start, window_end], select=[window_start, w
+- Calc(select=[window_start, window_end, 1 AS $f2])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[rowtime]]], fields=[rowtime])
]]>
</Resource>
</TestCase>
@@ -502,8 +493,7 @@ Calc(select=[a, b, uv])
+- Calc(select=[a, b, window_start, window_end, c])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -537,10 +527,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -575,10 +563,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -621,8 +607,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -667,8 +652,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -702,10 +686,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[25 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -740,10 +722,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[25 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -782,10 +762,8 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS uv])
+- Exchange(distribution=[hash[a, b, $e, $f6]])
+- Calc(select=[a, b, c, $e, MOD(HASH_CODE(c), 1024) AS $f6, rowtime])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -826,10 +804,8 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS uv])
+- LocalWindowAggregate(groupBy=[a, b, $e, $f6], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, b, $e, $f6, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, b, c, $e, MOD(HASH_CODE(c), 1024) AS $f6, rowtime])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -908,8 +884,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -949,8 +924,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f4 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -984,10 +958,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1022,10 +994,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1059,10 +1029,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1097,10 +1065,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, COUNT(*) AS count1$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1143,8 +1109,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -1189,8 +1154,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -1224,10 +1188,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1262,10 +1224,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, COUNT(*) AS count1$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1304,10 +1264,8 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS uv])
+- Exchange(distribution=[hash[a, b, $e, $f6]])
+- Calc(select=[a, b, c, $e, MOD(HASH_CODE(c), 1024) AS $f6, rowtime])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1348,10 +1306,8 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS uv])
+- LocalWindowAggregate(groupBy=[a, b, $e, $f6], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, b, $e, $f6, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, b, c, $e, MOD(HASH_CODE(c), 1024) AS $f6, rowtime])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1430,8 +1386,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -1471,8 +1426,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f4 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -1506,10 +1460,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1544,10 +1496,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, COUNT(*) AS count1$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1575,23 +1525,19 @@ LogicalSink(table=[default_catalog.default_database.s1], fields=[window_start, w
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])(reuse_id=[1])
-+- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
-
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, wAvg])
+- Calc(select=[window_start, window_end, wAvg])
+- WindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[weightedAvg(b, e) AS wAvg, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
- +- Calc(select=[b, e, rowtime])
- +- Reused(reference_id=[1])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, e, rowtime]]], fields=[b, e, rowtime])
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, EXPR$2])
+- Calc(select=[window_start, window_end, EXPR$2])
+- WindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[COUNT(*) AS EXPR$2, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
- +- Calc(select=[rowtime])
- +- Reused(reference_id=[1])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[rowtime]]], fields=[rowtime])
]]>
</Resource>
</TestCase>
@@ -1619,25 +1565,21 @@ LogicalSink(table=[default_catalog.default_database.s1], fields=[window_start, w
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])(reuse_id=[1])
-+- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
-
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, wAvg])
+- Calc(select=[window_start, window_end, wAvg])
+- GlobalWindowAggregate(window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[weightedAvg(weightedavg$0) AS wAvg, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
+- LocalWindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[weightedAvg(b, e) AS weightedavg$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[b, e, rowtime])
- +- Reused(reference_id=[1])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, e, rowtime]]], fields=[b, e, rowtime])
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, EXPR$2])
+- Calc(select=[window_start, window_end, EXPR$2])
+- GlobalWindowAggregate(window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[COUNT(count1$0) AS EXPR$2, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
+- LocalWindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[COUNT(*) AS count1$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[rowtime])
- +- Reused(reference_id=[1])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[rowtime]]], fields=[rowtime])
]]>
</Resource>
</TestCase>
@@ -1675,8 +1617,8 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- Calc(select=[a, b, c, d, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -1714,8 +1656,8 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- Calc(select=[a, b, c, d, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -1834,8 +1776,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -1874,8 +1815,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -1918,8 +1858,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS NOT NULL(b) AS $f4, b, e, c, rowtime], where=[>(b, 1000)])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -1963,8 +1902,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f4 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS NOT NULL(b) AS $f4, b, e, c, rowtime], where=[>(b, 1000)])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -2005,10 +1943,8 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5])
+- Calc(select=[a, cnt, sum_d, max_d, window_time AS rowtime])
+- WindowAggregate(groupBy=[a, b], window=[TUMBLE(time_col=[rowtime], size=[5 min])], select=[a, b, COUNT(*) AS cnt, SUM(d) AS sum_d, MAX(d) AS max_d, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a, b]])
- +- Calc(select=[a, b, d, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, d, rowtime]]], fields=[a, b, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2051,10 +1987,8 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5])
+- GlobalWindowAggregate(groupBy=[a, b], window=[TUMBLE(slice_end=[$slice_end], size=[5 min])], select=[a, b, COUNT(count1$0) AS cnt, SUM(sum$1) AS sum_d, MAX(max$2) AS max_d, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a, b]])
+- LocalWindowAggregate(groupBy=[a, b], window=[TUMBLE(time_col=[rowtime], size=[5 min])], select=[a, b, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) AS max$2, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, b, d, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, d, rowtime]]], fields=[a, b, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2096,8 +2030,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5])
+- Calc(select=[a, window_start, window_end, d, IS TRUE(>(b, 1000)) AS $f4, window_time, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(CAST(window_time)), 1024) AS $f7])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, d, rowtime]]], fields=[a, b, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2141,8 +2074,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5])
+- Calc(select=[a, window_start, window_end, d, IS TRUE(>(b, 1000)) AS $f4, window_time, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(CAST(window_time)), 1024) AS $f7])
+- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, d, rowtime]]], fields=[a, b, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2184,8 +2116,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2229,8 +2160,7 @@ Calc(select=[a, window_start, window_end, $f1 AS $f3, $f2 AS $f4, $f3 AS $f5, $f
+- Expand(projects=[{a, d, $f4, c, $f6, null AS $f7, 1 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, $f7, 2 AS $e, rowtime}, {a, d, $f4, c, null AS $f6, null AS $f7, 3 AS $e, rowtime}])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, MOD(HASH_CODE(d), 1024) AS $f6, MOD(HASH_CODE(c), 1024) AS $f7, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2267,8 +2197,8 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- Calc(select=[a, b, c, d, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2301,10 +2231,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -2338,10 +2266,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, null AS b, c, 4 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -2374,10 +2300,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, null AS b, c, 1 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -2411,10 +2335,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, null AS b, c, 1 AS $e, rowtime}, {null AS a, b, c, 8 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -2458,8 +2380,7 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS EXPR$3, $f5 AS EXPR$4, $f6 AS uv])
+- Expand(projects=[{a, null AS b, d, $f5, c, 4 AS $e, rowtime}, {null AS a, b, d, $f5, c, 8 AS $e, rowtime}])
+- Calc(select=[a, b, d, IS TRUE(>(b, 1000)) AS $f5, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2505,8 +2426,7 @@ Calc(select=[a, b, $f3 AS EXPR$2, $f4 AS EXPR$3, $f5 AS EXPR$4, $f6 AS uv])
+- Expand(projects=[{a, null AS b, d, $f5, c, 4 AS $e, rowtime}, {null AS a, b, d, $f5, c, 8 AS $e, rowtime}])
+- Calc(select=[a, b, d, IS TRUE(>(b, 1000)) AS $f5, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2545,8 +2465,7 @@ Calc(select=[a, b, window_start, window_end, EXPR$4, EXPR$5, EXPR$6, uv])
+- Exchange(distribution=[hash[a, b]])
+- Calc(select=[a, b, d, IS TRUE(>(b, 1000)) AS $f5, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2586,8 +2505,7 @@ Calc(select=[a, b, window_start, window_end, EXPR$4, EXPR$5, EXPR$6, uv])
+- LocalWindowAggregate(groupBy=[a, b], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, b, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f5 AS max$2, COUNT(distinct$0 c) AS count$3, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, b, d, IS TRUE(>(b, 1000)) AS $f5, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, d, rowtime]]], fields=[a, b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2623,8 +2541,7 @@ Calc(select=[window_start, window_end, EXPR$2, EXPR$3, EXPR$4, uv])
+- Exchange(distribution=[single])
+- Calc(select=[d, IS TRUE(>(b, 1000)) AS $f3, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, rowtime]]], fields=[b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2668,8 +2585,7 @@ Calc(select=[a, window_start, window_start AS ws, window_end, window_time, EXPR$
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f6, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -2714,8 +2630,7 @@ Calc(select=[a, window_start, window_start AS ws, window_end, window_time, EXPR$
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f6 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f6, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -2752,8 +2667,7 @@ Calc(select=[window_start, window_end, EXPR$2, EXPR$3, EXPR$4, uv])
+- Exchange(distribution=[single])
+- Calc(select=[d, IS TRUE(>(b, 1000)) AS $f4, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, rowtime]]], fields=[b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2791,8 +2705,7 @@ Calc(select=[window_start, window_end, EXPR$2, EXPR$3, EXPR$4, uv])
+- LocalWindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f4 AS max$2, COUNT(distinct$0 c) AS count$3, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[d, IS TRUE(>(b, 1000)) AS $f4, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, rowtime]]], fields=[b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2829,8 +2742,7 @@ Calc(select=[window_start, window_end, EXPR$2, EXPR$3, EXPR$4, uv])
+- LocalWindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f3 AS max$2, COUNT(distinct$0 c) AS count$3, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[d, IS TRUE(>(b, 1000)) AS $f3, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, rowtime]]], fields=[b, c, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -2866,8 +2778,7 @@ Calc(select=[EXPR$0, EXPR$1, EXPR$2, wAvg, uv])
+- Exchange(distribution=[single])
+- Calc(select=[d, IS TRUE(>(b, 1000)) AS $f3, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, e, rowtime]]], fields=[b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -2904,8 +2815,7 @@ Calc(select=[EXPR$0, EXPR$1, EXPR$2, wAvg, uv])
+- LocalWindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f3 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[d, IS TRUE(>(b, 1000)) AS $f3, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, c, d, e, rowtime]]], fields=[b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -2982,8 +2892,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -3022,8 +2931,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f4 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -3056,10 +2964,8 @@ LogicalAggregate(group=[{0, 1, 2}], EXPR$3=[COUNT()], EXPR$4=[SUM($3)])
Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4])
+- WindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS EXPR$3, SUM(d) AS EXPR$4, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a]])
- +- Calc(select=[a, d, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, d, rowtime]]], fields=[a, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -3093,10 +2999,8 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS EXPR$3, SUM(sum$1) AS EXPR$4, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, d, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, d, rowtime]]], fields=[a, d, rowtime])
]]>
</Resource>
</TestCase>
@@ -3129,10 +3033,8 @@ Calc(select=[a, b, uv])
+- WindowAggregate(groupBy=[a, b, $e], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, b, $e, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[a, b, $e]])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -3166,10 +3068,8 @@ Calc(select=[a, b, uv])
+- Exchange(distribution=[hash[a, b, $e]])
+- LocalWindowAggregate(groupBy=[a, b, $e], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, b, $e, COUNT(distinct$0 c) AS count$0, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Expand(projects=[{a, b, c, 0 AS $e, rowtime}, {a, null AS b, c, 4 AS $e, rowtime}, {null AS a, null AS b, c, 12 AS $e, rowtime}])
- +- Calc(select=[a, b, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, rowtime]]], fields=[a, b, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -3207,8 +3107,7 @@ WindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])],
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -3246,8 +3145,7 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, wAvg, uv])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
@@ -3286,8 +3184,7 @@ GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, SUM(d) AS sum$1, MAX(d) FILTER $f4 AS max$2, weightedAvg(b, e) AS weightedavg$3, COUNT(distinct$0 c) AS count$4, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS $f4, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.xml
index 42508e7..6d44ddf 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.xml
@@ -74,19 +74,15 @@ WindowJoin(leftWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], si
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -149,19 +145,15 @@ WindowJoin(leftWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], si
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[single])
+- Calc(select=[a, window_start, window_end])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -223,19 +215,15 @@ WindowJoin(leftWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], si
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -300,19 +288,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_start, window_start0), =(a, a0))]
: +- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_start, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -375,19 +359,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_start, window_start0), =(a, a0))]
: +- GlobalWindowAggregate(groupBy=[a], window=[HOP(slice_end=[$slice_end], size=[10 min], slide=[5 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_start, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[HOP(slice_end=[$slice_end], size=[10 min], slide=[5 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -448,19 +428,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_start, window_start0), =(a, a0))]
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_start, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -525,19 +501,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_end, window_end0), =(a, a0))], se
: +- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_end, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -600,19 +572,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_end, window_end0), =(a, a0))], se
: +- GlobalWindowAggregate(groupBy=[a], window=[HOP(slice_end=[$slice_end], size=[10 min], slide=[5 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_end, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[HOP(slice_end=[$slice_end], size=[10 min], slide=[5 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -673,19 +641,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_end, window_end0), =(a, a0))], se
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_end, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -751,17 +715,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_start, window_start0), =(window_e
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, c, proctime])
: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_start, window_end, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -824,19 +786,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_start, window_start0), =(window_e
: +- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[2 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[2 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_start, window_end, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -899,19 +857,15 @@ Join(joinType=[InnerJoin], where=[AND(=(window_start, window_start0), =(window_e
: +- GlobalWindowAggregate(groupBy=[a], window=[HOP(slice_end=[$slice_end], size=[10 min], slide=[5 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[window_start, window_end, a]])
+- Calc(select=[a, window_start, window_end, CAST(window_time) AS window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -976,19 +930,15 @@ WindowJoin(leftWindow=[CUMULATE(win_start=[window_start], win_end=[window_end],
: +- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1055,16 +1005,16 @@ Calc(select=[a, window_start, window_end, PROCTIME_MATERIALIZE(window_time) AS w
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, c, proctime])
: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- WindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[proctime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS cnt, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end, proctime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1127,19 +1077,15 @@ WindowJoin(leftWindow=[HOP(win_start=[window_start], win_end=[window_end], size=
: +- GlobalWindowAggregate(groupBy=[a], window=[HOP(slice_end=[$slice_end], size=[10 min], slide=[5 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[HOP(slice_end=[$slice_end], size=[10 min], slide=[5 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[HOP(time_col=[rowtime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1204,16 +1150,16 @@ Calc(select=[a, window_start, window_end, PROCTIME_MATERIALIZE(window_time) AS w
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, c, proctime])
: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- WindowAggregate(groupBy=[a], window=[HOP(time_col=[proctime], size=[10 min], slide=[5 min])], select=[a, COUNT(*) AS cnt, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end, proctime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1273,18 +1219,14 @@ WindowJoin(leftWindow=[SESSION(win_start=[window_start], win_end=[window_end], g
: +- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
: +- WindowAggregate(groupBy=[a], window=[SESSION(time_col=[rowtime], gap=[15 min])], select=[a, COUNT(*) AS cnt, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- WindowAggregate(groupBy=[a], window=[SESSION(time_col=[rowtime], gap=[15 min])], select=[a, COUNT(*) AS cnt, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1347,16 +1289,16 @@ Calc(select=[a, window_start, window_end, PROCTIME_MATERIALIZE(window_time) AS w
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, c, proctime])
: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- WindowAggregate(groupBy=[a], window=[SESSION(time_col=[proctime], gap=[15 min])], select=[a, COUNT(*) AS cnt, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end, proctime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1417,19 +1359,15 @@ WindowJoin(leftWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], si
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1492,16 +1430,16 @@ Calc(select=[a, window_start, window_end, PROCTIME_MATERIALIZE(window_time) AS w
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, c, proctime])
: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- WindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[proctime], size=[15 min])], select=[a, COUNT(*) AS cnt, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end, proctime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1563,19 +1501,15 @@ WindowJoin(leftWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], si
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -1638,19 +1572,15 @@ WindowJoin(leftWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], si
: +- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
-: +- Calc(select=[a, c, rowtime])
-: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
-: +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
-: +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+: +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end])
+- GlobalWindowAggregate(groupBy=[a], window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[a, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -2244,19 +2174,15 @@ Calc(select=[rowtime, a, l_cnt, l_uv, r_cnt, r_uv, a0, b, c, rowtime0, PROCTIME_
: : +- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: : +- Exchange(distribution=[hash[a]])
: : +- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- : : +- Calc(select=[a, c, rowtime])
- : : +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : : +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ : : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, window_start, window_end, cnt, uv])
: +- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- : +- Calc(select=[a, c, rowtime])
- : +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ : +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
@@ -2417,14 +2343,12 @@ Calc(select=[rowtime, a, l_b, l_c, r_b, r_c, a0, b, c, rowtime0, PROCTIME_MATERI
: : +- Calc(select=[a, b, c, window_start, window_end, window_time])
: : +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])], emitPerRecord=[true])
: : +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, b, c, window_start, window_end])
: +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])], emitPerRecord=[true])
: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
@@ -2496,16 +2420,16 @@ Calc(select=[a, window_start, window_end, PROCTIME_MATERIALIZE(window_time) AS w
: +- Exchange(distribution=[hash[a]])
: +- Calc(select=[a, c, proctime])
: +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- WindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[proctime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS cnt, COUNT(DISTINCT c) AS uv, start('w$) AS window_start, end('w$) AS window_end, proctime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, c, proctime])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- Calc(select=[a, c, PROCTIME() AS proctime, rowtime])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
@@ -2555,19 +2479,15 @@ Calc(select=[window_start, window_end, a, cnt AS l_cnt, uv AS l_uv, cnt0 AS r_cn
: +- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
: +- Exchange(distribution=[hash[a]])
: +- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- : +- Calc(select=[a, c, rowtime])
- : +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- : +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- : +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, rowtime])
+ : +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ : +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, c, rowtime]]], fields=[a, c, rowtime])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, window_start, window_end, window_time, cnt, uv])
+- GlobalWindowAggregate(groupBy=[a], window=[CUMULATE(slice_end=[$slice_end], max_size=[1 h], step=[10 min])], select=[a, COUNT(count1$0) AS cnt, COUNT(distinct$0 count$1) AS uv, start('w$) AS window_start, end('w$) AS window_end, rowtime('w$) AS window_time])
+- Exchange(distribution=[hash[a]])
+- LocalWindowAggregate(groupBy=[a], window=[CUMULATE(time_col=[rowtime], max_size=[1 h], step=[10 min])], select=[a, COUNT(*) AS count1$0, COUNT(distinct$0 c) AS count$1, DISTINCT(c) AS distinct$0, slice_end('w$) AS $slice_end])
- +- Calc(select=[a, c, rowtime])
- +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b, c, rowtime])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, c, rowtime]]], fields=[a, c, rowtime])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRuleTest.scala
new file mode 100644
index 0000000..20e8aef
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectWindowTableFunctionTransposeRuleTest.scala
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.logical
+
+import org.apache.flink.table.planner.plan.optimize.program.{FlinkChainedProgram, FlinkHepRuleSetProgramBuilder, FlinkStreamProgram, HEP_RULES_EXECUTION_TYPE, StreamOptimizeContext}
+import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.WeightedAvgWithMerge
+import org.apache.flink.table.planner.utils.TableTestBase
+
+import org.apache.calcite.plan.hep.HepMatchOrder
+import org.apache.calcite.tools.RuleSets
+
+import org.junit.{Before, Test}
+
+/**
+ * Test for [[ProjectWindowTableFunctionTransposeRule]].
+ */
+class ProjectWindowTableFunctionTransposeRuleTest extends TableTestBase {
+ private val util = streamTestUtil()
+
+ @Before
+ def setup(): Unit = {
+ val programs = new FlinkChainedProgram[StreamOptimizeContext]()
+ programs.addLast(
+ "rules",
+ FlinkHepRuleSetProgramBuilder.newBuilder
+ .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE)
+ .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+ .add(RuleSets.ofList(ProjectWindowTableFunctionTransposeRule.INSTANCE))
+ .build())
+ util.replaceStreamProgram(programs)
+
+ util.tableEnv.executeSql(
+ s"""
+ |CREATE TABLE MyTable (
+ | a INT,
+ | b BIGINT,
+ | c STRING NOT NULL,
+ | d DECIMAL(10, 3),
+ | e BIGINT,
+ | rowtime TIMESTAMP(3),
+ | WATERMARK FOR rowtime AS rowtime - INTERVAL '1' SECOND
+ |) with (
+ | 'connector' = 'values'
+ |)
+ |""".stripMargin)
+ }
+
+ @Test
+ def testPruneUnusedColumn(): Unit = {
+ val sql =
+ """
+ |SELECT
+ | window_start,
+ | window_end,
+ | count(*),
+ | sum(d),
+ | max(d) filter (where b > 1000),
+ | count(distinct c) AS uv
+ |FROM TABLE(
+ | CUMULATE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '10' MINUTE, INTERVAL '1' HOUR))
+ |GROUP BY window_start, window_end
+ """.stripMargin
+ util.verifyRelPlan(sql)
+ }
+
+ @Test
+ def testNoColumnToPrune(): Unit = {
+ util.addTemporarySystemFunction("weightedAvg", classOf[WeightedAvgWithMerge])
+
+ val sql =
+ """
+ |SELECT
+ | a,
+ | window_start,
+ | window_end,
+ | count(*),
+ | sum(d),
+ | max(d) filter (where b > 1000),
+ | weightedAvg(b, e) AS wAvg,
+ | count(distinct c) AS uv
+ |FROM TABLE(
+ | CUMULATE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '10' MINUTE, INTERVAL '1' HOUR))
+ |GROUP BY a, window_start, window_end
+ """.stripMargin
+ util.verifyRelPlan(sql)
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.scala
index 8588f56..c3d1a25 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.scala
@@ -236,7 +236,6 @@ class WindowAggregateTest(aggPhaseEnforcer: AggregatePhaseStrategy) extends Tabl
@Test
def testTumble_ProjectionPushDown(): Unit = {
- // TODO: FLINK-21290 [b, c, e, proctime] are never used, should be pruned in source
val sql =
"""
|SELECT