This is an automated email from the ASF dual-hosted git repository.
snuyanzin 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 3871786003a [FLINK-25802][FLINK-30499][table] Fix TIMESTAMP codegen
for `RANGE OVER` window bounds
3871786003a is described below
commit 3871786003a5cdb52748d83bec614b05bf2423fd
Author: Aleksandr Iushmanov <[email protected]>
AuthorDate: Thu Aug 13 09:45:24 2026 +0100
[FLINK-25802][FLINK-30499][table] Fix TIMESTAMP codegen for `RANGE OVER`
window bounds
RangeBoundComparatorCodeGenerator bound TIMESTAMP columns as BigIntType
but passed the raw TimestampData term to ExprCodeGenerator, causing an
invalid (Long) cast at Janino compile time (FLINK-25802).
TIMESTAMP_WITH_LOCAL_TIME_ZONE fell through to the default case, and
reuseMemberCode() was emitted inside compare() instead of the class scope
(FLINK-30499).
Fix: use LogicalTypeFamily.TIMESTAMP to cover all three TIMESTAMP variants,
extract epoch time with microseconds precision, and promote child member
code
to the parent context via addReusableMember().
Generated-by: Claude Code
---
.../over/RangeBoundComparatorCodeGenerator.scala | 46 ++-
.../exec/batch/OverAggregateBatchRestoreTest.java | 9 +-
.../exec/common/OverAggregateTestPrograms.java | 63 +++
.../over-aggregate-batch-range-timestamp-ltz.json | 214 ++++++++++
.../plan/over-aggregate-batch-range-timestamp.json | 214 ++++++++++
...ver-aggregate-bounded-non-partitioned-rows.json | 432 ++++++++++++++++++++
.../over-aggregate-bounded-partitioned-rows.json | 440 +++++++++++++++++++++
7 files changed, 1402 insertions(+), 16 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
index c1f2965c80f..5ab4ac94c97 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/over/RangeBoundComparatorCodeGenerator.scala
@@ -23,7 +23,7 @@ import
org.apache.flink.table.planner.codegen.{CodeGeneratorContext, CodeGenUtil
import org.apache.flink.table.planner.codegen.CodeGenUtils.{newName, ROW_DATA}
import org.apache.flink.table.planner.codegen.Indenter.toISC
import org.apache.flink.table.runtime.generated.{GeneratedRecordComparator,
RecordComparator}
-import org.apache.flink.table.types.logical.{BigIntType, IntType, LogicalType,
LogicalTypeRoot, RowType}
+import org.apache.flink.table.types.logical.{BigIntType, IntType, LogicalType,
LogicalTypeFamily, LogicalTypeRoot, RowType}
import org.apache.calcite.avatica.util.DateTimeUtils
import org.apache.calcite.rex.{RexInputRef, RexWindowBound}
@@ -135,16 +135,32 @@ class RangeBoundComparatorCodeGenerator(
inputValue: String,
currentValue: String,
parentCtx: CodeGeneratorContext): String = {
- val (realBoundValue, realKeyType) = keyType.getTypeRoot match {
- case LogicalTypeRoot.DATE =>
- // The constant about time is expressed based millisecond unit in
calcite, but
- // the field about date is expressed based day unit. So here should
keep the same unit for
- // comparator.
+ val (realBoundValue, realKeyType) =
+ if (keyType.is(LogicalTypeFamily.TIMESTAMP)) {
+ // Scale millis bound to microseconds to preserve sub-millisecond
TIMESTAMP(n>3) precision
+ val microsBound = bound match {
+ case l: Long => l * 1000L
+ case bg: BigDecimal => bg.multiply(BigDecimal.valueOf(1000))
+ }
+ (microsBound, new BigIntType())
+ } else if (keyType.is(LogicalTypeFamily.TIME)) {
+ (bound, new IntType())
+ } else if (keyType.is(LogicalTypeRoot.DATE)) {
+ // Calcite bound is in millis; DATE field is in days
(bound.asInstanceOf[Long] / DateTimeUtils.MILLIS_PER_DAY, new
IntType())
- case LogicalTypeRoot.TIME_WITHOUT_TIME_ZONE => (bound, new IntType())
- case LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE => (bound, new
BigIntType())
- case _ => (bound, keyType)
- }
+ } else {
+ (bound, keyType)
+ }
+
+ val (realInputValue, realCurrentValue) =
+ if (keyType.is(LogicalTypeFamily.TIMESTAMP)) {
+ // Epoch microseconds: millis * 1000 + nanoOfMillis / 1000 preserves
TIMESTAMP(6) precision
+ (
+ s"($inputValue.getMillisecond() * 1000L +
$inputValue.getNanoOfMillisecond() / 1000)",
+ s"($currentValue.getMillisecond() * 1000L +
$currentValue.getNanoOfMillisecond() / 1000)")
+ } else {
+ (inputValue, currentValue)
+ }
val typeFactory = relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory]
val relKeyType = typeFactory.createFieldTypeFromLogicalType(realKeyType)
@@ -157,7 +173,9 @@ class RangeBoundComparatorCodeGenerator(
} else {
relBuilder.call(MINUS, new RexInputRef(1, relKeyType), new
RexInputRef(0, relKeyType))
}
- exprCodeGenerator.bindInput(realKeyType,
inputValue).bindSecondInput(realKeyType, currentValue)
+ exprCodeGenerator
+ .bindInput(realKeyType, realInputValue)
+ .bindSecondInput(realKeyType, realCurrentValue)
val literal = relBuilder.literal(realBoundValue)
// In order to avoid the loss of precision in long cast to int.
@@ -169,8 +187,12 @@ class RangeBoundComparatorCodeGenerator(
val comExpr = exprCodeGenerator.generateExpression(comCall)
+ val childMemberCode = ctx.reuseMemberCode()
+ if (childMemberCode.nonEmpty) {
+ parentCtx.addReusableMember(childMemberCode)
+ }
+
j"""
- ${ctx.reuseMemberCode()}
${ctx.reuseLocalVariableCode()}
${ctx.reuseInputUnboxingCode()}
${comExpr.code}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/OverAggregateBatchRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/OverAggregateBatchRestoreTest.java
index 1308cd8ff82..169f053b222 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/OverAggregateBatchRestoreTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/OverAggregateBatchRestoreTest.java
@@ -35,10 +35,11 @@ public class OverAggregateBatchRestoreTest extends
BatchRestoreTestBase {
@Override
public List<TableTestProgram> programs() {
return Arrays.asList(
- // These tests fail due to FLINK-25802
- //
OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_PARTITIONED_ROWS,
- //
OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_NON_PARTITIONED_ROWS
+
OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_PARTITIONED_ROWS,
+
OverAggregateTestPrograms.OVER_AGGREGATE_TIME_BOUNDED_NON_PARTITIONED_ROWS,
OverAggregateTestPrograms.OVER_AGGREGATE_UNBOUNDED_PARTITIONED_ROWS,
-
OverAggregateTestPrograms.OVER_AGGREGATE_ROW_BOUNDED_PARTITIONED_PRECEDING_ROWS);
+
OverAggregateTestPrograms.OVER_AGGREGATE_ROW_BOUNDED_PARTITIONED_PRECEDING_ROWS,
+ OverAggregateTestPrograms.OVER_AGGREGATE_RANGE_TIMESTAMP,
+ OverAggregateTestPrograms.OVER_AGGREGATE_RANGE_TIMESTAMP_LTZ);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/OverAggregateTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/OverAggregateTestPrograms.java
index 748dd9c5b1d..b38ad2f8983 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/OverAggregateTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/OverAggregateTestPrograms.java
@@ -25,6 +25,9 @@ import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.types.Row;
+import java.time.Instant;
+import java.time.LocalDateTime;
+
import static
org.apache.flink.table.api.config.TableConfigOptions.LOCAL_TIME_ZONE;
/**
@@ -383,6 +386,66 @@ public class OverAggregateTestPrograms {
BEFORE_RESTORE_DATA_PRECEDING_ROWS_WITH_OUT_OF_ORDER_RECORDS,
AFTER_RESTORE_DATA_PRECEDING_ROWS_WITH_OUT_OF_ORDER_RECORDS);
+ /**
+ * Regression test for FLINK-25802: RANGE OVER with TIMESTAMP ORDER BY.
Uses TIMESTAMP(6) with a
+ * sub-millisecond row to verify that the range comparator uses
microsecond precision: row 3 is
+ * 10s+600µs after row 1, so row 1 falls just outside row 3's 10-second
window (count=2, not 3).
+ */
+ public static final TableTestProgram OVER_AGGREGATE_RANGE_TIMESTAMP =
+ TableTestProgram.of(
+ "over-aggregate-batch-range-timestamp",
+ "RANGE OVER with TIMESTAMP(6) ORDER BY")
+ .setupTableSource(
+ SourceTestStep.newBuilder("timestamp_range_source")
+ .addSchema("ts TIMESTAMP(6)", "val INT")
+ .producedValues(
+ Row.of(LocalDateTime.of(2021, 1,
1, 0, 0, 0, 0), 1),
+ Row.of(LocalDateTime.of(2021, 1,
1, 0, 0, 5, 0), 2),
+ Row.of(
+ LocalDateTime.of(2021, 1,
1, 0, 0, 10, 600_000),
+ 3))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("timestamp_range_sink")
+ .addSchema("val INT", "cnt BIGINT")
+ .consumedValues(Row.of(1, 1L), Row.of(2,
2L), Row.of(3, 2L))
+ .build())
+ .runSql(
+ "INSERT INTO timestamp_range_sink"
+ + " SELECT val, COUNT(val) OVER (ORDER BY
ts"
+ + " RANGE BETWEEN INTERVAL '10' SECOND
PRECEDING"
+ + " AND CURRENT ROW)"
+ + " FROM timestamp_range_source")
+ .build();
+
+ /**
+ * Regression test for FLINK-30499: RANGE OVER with TIMESTAMP_LTZ(3) ORDER
BY failed to compile.
+ */
+ public static final TableTestProgram OVER_AGGREGATE_RANGE_TIMESTAMP_LTZ =
+ TableTestProgram.of(
+ "over-aggregate-batch-range-timestamp-ltz",
+ "RANGE OVER with TIMESTAMP_LTZ(3) ORDER BY")
+ .setupTableSource(
+
SourceTestStep.newBuilder("timestamp_ltz_range_source")
+ .addSchema("ts TIMESTAMP_LTZ(3)", "val
INT")
+ .producedValues(
+
Row.of(Instant.parse("2021-01-01T00:00:00Z"), 1),
+
Row.of(Instant.parse("2021-01-01T00:00:05Z"), 2),
+
Row.of(Instant.parse("2021-01-01T00:00:12Z"), 3))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("timestamp_ltz_range_sink")
+ .addSchema("val INT", "cnt BIGINT")
+ .consumedValues(Row.of(1, 1L), Row.of(2,
2L), Row.of(3, 2L))
+ .build())
+ .runSql(
+ "INSERT INTO timestamp_ltz_range_sink"
+ + " SELECT val, COUNT(val) OVER (ORDER BY
ts"
+ + " RANGE BETWEEN INTERVAL '10' SECOND
PRECEDING"
+ + " AND CURRENT ROW)"
+ + " FROM timestamp_ltz_range_source")
+ .build();
+
private static SourceTestStep getSourceTestStep(Row[] data, Row[]
afterData) {
return SourceTestStep.newBuilder("MyTable")
.addSchema(
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-batch-range-timestamp-ltz/plan/over-aggregate-batch-range-timestamp-ltz.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-batch-range-timestamp-ltz/plan/over-aggregate-batch-range-timestamp-ltz.json
new file mode 100644
index 00000000000..59afff65144
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-batch-range-timestamp-ltz/plan/over-aggregate-batch-range-timestamp-ltz.json
@@ -0,0 +1,214 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 7,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" :
"`default_catalog`.`default_database`.`timestamp_ltz_range_source`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "ts",
+ "dataType" : "TIMESTAMP(3) WITH LOCAL TIME ZONE"
+ }, {
+ "name" : "val",
+ "dataType" : "INT"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, timestamp_ltz_range_source]], fields=[ts, val])",
+ "dynamicFilteringDataListenerID" : "4368c400-2178-4dfa-8997-886f640a518c"
+ }, {
+ "id" : 8,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT>",
+ "description" : "Exchange(distribution=[single])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 9,
+ "type" : "batch-exec-sort_1",
+ "configuration" : {
+ "table.exec.resource.sort.memory" : "128 mb",
+ "table.exec.sort.async-merge-enabled" : "true",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 0,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT>",
+ "description" : "Sort(orderBy=[ts ASC])"
+ }, {
+ "id" : 10,
+ "type" : "batch-exec-over-aggregate_1",
+ "configuration" : {
+ "table.exec.resource.external-buffer-memory" : "10 mb"
+ },
+ "overSpec" : {
+ "partition" : {
+ "fields" : [ ]
+ },
+ "groups" : [ {
+ "orderBy" : {
+ "fields" : [ {
+ "index" : 0,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "isRows" : false,
+ "lowerBound" : {
+ "kind" : "BOUNDED_WINDOW",
+ "isPreceding" : true,
+ "offset" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ }
+ },
+ "upperBound" : {
+ "kind" : "CURRENT_ROW"
+ },
+ "aggCalls" : [ {
+ "name" : "w0$o0",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ]
+ } ],
+ "constants" : [ {
+ "kind" : "LITERAL",
+ "value" : "10000",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "originalInputFields" : 2
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` TIMESTAMP(3) WITH LOCAL TIME ZONE, `val` INT,
`w0$o0` BIGINT NOT NULL>",
+ "description" : "OverAggregate(orderBy=[ts ASC], window#0=[COUNT(val) AS
w0$o0 RANGE BETWEEN 10000 PRECEDING AND CURRENT ROW], select=[ts, val, w0$o0])"
+ }, {
+ "id" : 11,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`val` INT, `$1` BIGINT NOT NULL>",
+ "description" : "Calc(select=[val, w0$o0 AS $1])"
+ }, {
+ "id" : 12,
+ "type" : "batch-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" :
"`default_catalog`.`default_database`.`timestamp_ltz_range_sink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "val",
+ "dataType" : "INT"
+ }, {
+ "name" : "cnt",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`val` INT, `$1` BIGINT NOT NULL>",
+ "description" :
"Sink(table=[default_catalog.default_database.timestamp_ltz_range_sink],
fields=[val, $1])"
+ } ],
+ "edges" : [ {
+ "source" : 7,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 9,
+ "target" : 10,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 10,
+ "target" : 11,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-batch-range-timestamp/plan/over-aggregate-batch-range-timestamp.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-batch-range-timestamp/plan/over-aggregate-batch-range-timestamp.json
new file mode 100644
index 00000000000..69e352f9f73
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-batch-range-timestamp/plan/over-aggregate-batch-range-timestamp.json
@@ -0,0 +1,214 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" :
"`default_catalog`.`default_database`.`timestamp_range_source`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "ts",
+ "dataType" : "TIMESTAMP(6)"
+ }, {
+ "name" : "val",
+ "dataType" : "INT"
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`ts` TIMESTAMP(6), `val` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, timestamp_range_source]], fields=[ts, val])",
+ "dynamicFilteringDataListenerID" : "227ea629-cfe7-4e32-9b5f-44d2da57286e"
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` TIMESTAMP(6), `val` INT>",
+ "description" : "Exchange(distribution=[single])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-sort_1",
+ "configuration" : {
+ "table.exec.resource.sort.memory" : "128 mb",
+ "table.exec.sort.async-merge-enabled" : "true",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 0,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` TIMESTAMP(6), `val` INT>",
+ "description" : "Sort(orderBy=[ts ASC])"
+ }, {
+ "id" : 4,
+ "type" : "batch-exec-over-aggregate_1",
+ "configuration" : {
+ "table.exec.resource.external-buffer-memory" : "10 mb"
+ },
+ "overSpec" : {
+ "partition" : {
+ "fields" : [ ]
+ },
+ "groups" : [ {
+ "orderBy" : {
+ "fields" : [ {
+ "index" : 0,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "isRows" : false,
+ "lowerBound" : {
+ "kind" : "BOUNDED_WINDOW",
+ "isPreceding" : true,
+ "offset" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ }
+ },
+ "upperBound" : {
+ "kind" : "CURRENT_ROW"
+ },
+ "aggCalls" : [ {
+ "name" : "w0$o0",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ]
+ } ],
+ "constants" : [ {
+ "kind" : "LITERAL",
+ "value" : "10000",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "originalInputFields" : 2
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` TIMESTAMP(6), `val` INT, `w0$o0` BIGINT NOT
NULL>",
+ "description" : "OverAggregate(orderBy=[ts ASC], window#0=[COUNT(val) AS
w0$o0 RANGE BETWEEN 10000 PRECEDING AND CURRENT ROW], select=[ts, val, w0$o0])"
+ }, {
+ "id" : 5,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`val` INT, `$1` BIGINT NOT NULL>",
+ "description" : "Calc(select=[val, w0$o0 AS $1])"
+ }, {
+ "id" : 6,
+ "type" : "batch-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" :
"`default_catalog`.`default_database`.`timestamp_range_sink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "val",
+ "dataType" : "INT"
+ }, {
+ "name" : "cnt",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`val` INT, `$1` BIGINT NOT NULL>",
+ "description" :
"Sink(table=[default_catalog.default_database.timestamp_range_sink],
fields=[val, $1])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-bounded-non-partitioned-rows/plan/over-aggregate-bounded-non-partitioned-rows.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-bounded-non-partitioned-rows/plan/over-aggregate-bounded-non-partitioned-rows.json
new file mode 100644
index 00000000000..58353091767
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-bounded-non-partitioned-rows/plan/over-aggregate-bounded-non-partitioned-rows.json
@@ -0,0 +1,432 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 10,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MyTable`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "ts",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "a",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b",
+ "dataType" : "INT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "rowtime",
+ "kind" : "COMPUTED",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$FROM_UNIXTIME$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ },
+ "serializableString" : "TO_TIMESTAMP(FROM_UNIXTIME(`ts`))"
+ }
+ } ],
+ "watermarkSpecs" : [ {
+ "rowtimeAttribute" : "rowtime",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "TIMESTAMP(3)"
+ },
+ "serializableString" : "`rowtime`"
+ }
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, MyTable]], fields=[ts, a, b, c])",
+ "dynamicFilteringDataListenerID" : "5a529672-b088-4321-9f17-47023adbbbbc"
+ }, {
+ "id" : 11,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[single])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 12,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$FROM_UNIXTIME$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 4,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Calc(select=[ts, a, b, c, TO_TIMESTAMP(FROM_UNIXTIME(ts))
AS rowtime, 4 AS $5])"
+ }, {
+ "id" : 17,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "KEEP_INPUT_AS_IS",
+ "inputDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "isStrict" : true
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Exchange(distribution=[forward])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 13,
+ "type" : "batch-exec-sort_1",
+ "configuration" : {
+ "table.exec.resource.sort.memory" : "128 mb",
+ "table.exec.sort.async-merge-enabled" : "true",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 4,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Sort(orderBy=[rowtime ASC])"
+ }, {
+ "id" : 18,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "KEEP_INPUT_AS_IS",
+ "inputDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "isStrict" : true
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Exchange(distribution=[forward])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 14,
+ "type" : "batch-exec-over-aggregate_1",
+ "configuration" : {
+ "table.exec.resource.external-buffer-memory" : "10 mb"
+ },
+ "overSpec" : {
+ "partition" : {
+ "fields" : [ ]
+ },
+ "groups" : [ {
+ "orderBy" : {
+ "fields" : [ {
+ "index" : 4,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "isRows" : false,
+ "lowerBound" : {
+ "kind" : "BOUNDED_WINDOW",
+ "isPreceding" : true,
+ "offset" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ }
+ },
+ "upperBound" : {
+ "kind" : "CURRENT_ROW"
+ },
+ "aggCalls" : [ {
+ "name" : "w0$o0",
+ "systemName" : "LTCNT",
+ "argList" : [ 1, 5 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "w0$o1",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "w0$o2",
+ "internalName" : "$SUM$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ]
+ } ],
+ "constants" : [ {
+ "kind" : "LITERAL",
+ "value" : "10000",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "originalInputFields" : 6
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "SINGLETON"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL, `w0$o0`
BIGINT, `w0$o1` BIGINT NOT NULL, `w0$o2` BIGINT NOT NULL>",
+ "description" : "OverAggregate(orderBy=[rowtime ASC], window#0=[LTCNT(a,
$5) AS w0$o0, COUNT(a) AS w0$o1, SUM(a) AS w0$o2 RANGE BETWEEN 10000 PRECEDING
AND CURRENT ROW], select=[ts, a, b, c, rowtime, $5, w0$o0, w0$o1, w0$o2])"
+ }, {
+ "id" : 15,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "TIMESTAMP(3)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CASE$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 0,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 8,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "BIGINT"
+ } ],
+ "type" : "BIGINT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`c` VARCHAR(2147483647), `ts` BIGINT, `rowtime`
TIMESTAMP(3), `b` INT, `EXPR$4` BIGINT, `EXPR$5` BIGINT NOT NULL, `EXPR$6`
BIGINT>",
+ "description" : "Calc(select=[c, ts, rowtime, b, w0$o0 AS EXPR$4, w0$o1 AS
EXPR$5, CASE((w0$o1 > 0), w0$o2, null:BIGINT) AS EXPR$6])"
+ }, {
+ "id" : 16,
+ "type" : "batch-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "rowtime",
+ "dataType" : "TIMESTAMP(3)"
+ }, {
+ "name" : "b",
+ "dataType" : "INT"
+ }, {
+ "name" : "c",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "d",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`c` VARCHAR(2147483647), `ts` BIGINT, `rowtime`
TIMESTAMP(3), `b` INT, `EXPR$4` BIGINT, `EXPR$5` BIGINT NOT NULL, `EXPR$6`
BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[c, ts, rowtime, b, EXPR$4, EXPR$5, EXPR$6])"
+ } ],
+ "edges" : [ {
+ "source" : 10,
+ "target" : 11,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 11,
+ "target" : 12,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 12,
+ "target" : 17,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 17,
+ "target" : 13,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 13,
+ "target" : 18,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 18,
+ "target" : 14,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 14,
+ "target" : 15,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 15,
+ "target" : 16,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-bounded-partitioned-rows/plan/over-aggregate-bounded-partitioned-rows.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-bounded-partitioned-rows/plan/over-aggregate-bounded-partitioned-rows.json
new file mode 100644
index 00000000000..38fbd193f19
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/batch-exec-over-aggregate_1/over-aggregate-bounded-partitioned-rows/plan/over-aggregate-bounded-partitioned-rows.json
@@ -0,0 +1,440 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "batch-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MyTable`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "ts",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "a",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "b",
+ "dataType" : "INT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "rowtime",
+ "kind" : "COMPUTED",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$FROM_UNIXTIME$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ },
+ "serializableString" : "TO_TIMESTAMP(FROM_UNIXTIME(`ts`))"
+ }
+ } ],
+ "watermarkSpecs" : [ {
+ "rowtimeAttribute" : "rowtime",
+ "expression" : {
+ "rexNode" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "TIMESTAMP(3)"
+ },
+ "serializableString" : "`rowtime`"
+ }
+ } ]
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, MyTable]], fields=[ts, a, b, c])",
+ "dynamicFilteringDataListenerID" : "ab4ea66b-59ca-4617-b714-7e3bcb9f581c"
+ }, {
+ "id" : 2,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 3 ]
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[c]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 3,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "CALL",
+ "internalName" : "$TO_TIMESTAMP$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "internalName" : "$FROM_UNIXTIME$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "TIMESTAMP(3)"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 4,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Calc(select=[ts, a, b, c, TO_TIMESTAMP(FROM_UNIXTIME(ts))
AS rowtime, 4 AS $5])"
+ }, {
+ "id" : 8,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "KEEP_INPUT_AS_IS",
+ "inputDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 3 ]
+ },
+ "isStrict" : false
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Exchange(distribution=[keep_input_as_is[hash[c]]])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 4,
+ "type" : "batch-exec-sort_1",
+ "configuration" : {
+ "table.exec.resource.sort.memory" : "128 mb",
+ "table.exec.sort.async-merge-enabled" : "true",
+ "table.exec.sort.max-num-file-handles" : "128",
+ "table.exec.spill-compression.block-size" : "64 kb",
+ "table.exec.spill-compression.enabled" : "true"
+ },
+ "sortSpec" : {
+ "fields" : [ {
+ "index" : 3,
+ "isAscending" : true,
+ "nullIsLast" : false
+ }, {
+ "index" : 4,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "END_INPUT",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Sort(orderBy=[c ASC, rowtime ASC])"
+ }, {
+ "id" : 9,
+ "type" : "batch-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "KEEP_INPUT_AS_IS",
+ "inputDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 3 ]
+ },
+ "isStrict" : true
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL>",
+ "description" : "Exchange(distribution=[forward])",
+ "requiredExchangeMode" : "UNDEFINED"
+ }, {
+ "id" : 5,
+ "type" : "batch-exec-over-aggregate_1",
+ "configuration" : {
+ "table.exec.resource.external-buffer-memory" : "10 mb"
+ },
+ "overSpec" : {
+ "partition" : {
+ "fields" : [ 3 ]
+ },
+ "groups" : [ {
+ "orderBy" : {
+ "fields" : [ {
+ "index" : 4,
+ "isAscending" : true,
+ "nullIsLast" : false
+ } ]
+ },
+ "isRows" : false,
+ "lowerBound" : {
+ "kind" : "BOUNDED_WINDOW",
+ "isPreceding" : true,
+ "offset" : {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ }
+ },
+ "upperBound" : {
+ "kind" : "CURRENT_ROW"
+ },
+ "aggCalls" : [ {
+ "name" : "w0$o0",
+ "systemName" : "LTCNT",
+ "argList" : [ 1, 5 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT"
+ }, {
+ "name" : "w0$o1",
+ "syntax" : "FUNCTION_STAR",
+ "internalName" : "$COUNT$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "name" : "w0$o2",
+ "internalName" : "$SUM$1",
+ "argList" : [ 1 ],
+ "filterArg" : -1,
+ "distinct" : false,
+ "approximate" : false,
+ "ignoreNulls" : false,
+ "type" : "BIGINT NOT NULL"
+ } ]
+ } ],
+ "constants" : [ {
+ "kind" : "LITERAL",
+ "value" : "10000",
+ "type" : "INTERVAL SECOND(6) NOT NULL"
+ } ],
+ "originalInputFields" : 6
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 3 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`ts` BIGINT, `a` BIGINT, `b` INT, `c`
VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `$5` BIGINT NOT NULL, `w0$o0`
BIGINT, `w0$o1` BIGINT NOT NULL, `w0$o2` BIGINT NOT NULL>",
+ "description" : "OverAggregate(partitionBy=[c], orderBy=[rowtime ASC],
window#0=[LTCNT(a, $5) AS w0$o0, COUNT(a) AS w0$o1, SUM(a) AS w0$o2 RANGE
BETWEEN 10000 PRECEDING AND CURRENT ROW], select=[ts, a, b, c, rowtime, $5,
w0$o0, w0$o1, w0$o2])"
+ }, {
+ "id" : 6,
+ "type" : "batch-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "TIMESTAMP(3)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "BIGINT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CASE$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 7,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 0,
+ "type" : "BIGINT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 8,
+ "type" : "BIGINT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : null,
+ "type" : "BIGINT"
+ } ],
+ "type" : "BIGINT"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`c` VARCHAR(2147483647), `ts` BIGINT, `rowtime`
TIMESTAMP(3), `b` INT, `EXPR$4` BIGINT, `EXPR$5` BIGINT NOT NULL, `EXPR$6`
BIGINT>",
+ "description" : "Calc(select=[c, ts, rowtime, b, w0$o0 AS EXPR$4, w0$o1 AS
EXPR$5, CASE((w0$o1 > 0), w0$o2, null:BIGINT) AS EXPR$6])"
+ }, {
+ "id" : 7,
+ "type" : "batch-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`MySink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "ts",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "rowtime",
+ "dataType" : "TIMESTAMP(3)"
+ }, {
+ "name" : "b",
+ "dataType" : "INT"
+ }, {
+ "name" : "c",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "d",
+ "dataType" : "BIGINT"
+ }, {
+ "name" : "e",
+ "dataType" : "BIGINT"
+ } ]
+ }
+ }
+ }
+ },
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "BLOCKING",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`c` VARCHAR(2147483647), `ts` BIGINT, `rowtime`
TIMESTAMP(3), `b` INT, `EXPR$4` BIGINT, `EXPR$5` BIGINT NOT NULL, `EXPR$6`
BIGINT>",
+ "description" : "Sink(table=[default_catalog.default_database.MySink],
fields=[c, ts, rowtime, b, EXPR$4, EXPR$5, EXPR$6])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 8,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 8,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 4,
+ "target" : 9,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 9,
+ "target" : 5,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 5,
+ "target" : 6,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 6,
+ "target" : 7,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
