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 f8f70684560 [FLINK-40948][table] Maintain window properties for
different joins
f8f70684560 is described below
commit f8f70684560edc5bbd507fd2b76a74821a6e19d9
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Thu Oct 29 17:33:39 2026 +0100
[FLINK-40948][table] Maintain window properties for different joins
---
.../plan/metadata/FlinkRelMdWindowProperties.scala | 15 ++-
.../stream/WindowJoinSemanticTestPrograms.java | 150 +++++++++++++++++++++
.../nodes/exec/stream/WindowJoinSemanticTests.java | 37 +++++
.../planner/plan/stream/sql/WindowRankTest.xml | 62 +++++++++
.../plan/stream/sql/agg/WindowAggregateTest.xml | 98 ++++++++++++++
.../plan/stream/sql/join/WindowJoinTest.xml | 68 ++++++++++
.../planner/plan/stream/sql/WindowRankTest.scala | 26 ++++
.../plan/stream/sql/agg/WindowAggregateTest.scala | 32 +++++
.../plan/stream/sql/join/WindowJoinTest.scala | 23 ++++
9 files changed, 507 insertions(+), 4 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
index 80cce27a559..ca07a4fdc66 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
@@ -413,7 +413,7 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
def getWindowProperties(rel: FlinkLogicalJoin, mq: RelMetadataQuery):
RelWindowProperties = {
if (satisfyWindowJoin(rel)) {
- getJoinWindowProperties(rel.getLeft, rel.getRight, mq)
+ getJoinWindowProperties(rel.getJoinType, rel.getLeft, rel.getRight, mq)
} else {
null
}
@@ -422,10 +422,11 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
def getWindowProperties(
rel: StreamPhysicalWindowJoin,
mq: RelMetadataQuery): RelWindowProperties = {
- getJoinWindowProperties(rel.getLeft, rel.getRight, mq)
+ getJoinWindowProperties(rel.getJoinType, rel.getLeft, rel.getRight, mq)
}
private def getJoinWindowProperties(
+ joinType: JoinRelType,
left: RelNode,
right: RelNode,
mq: RelMetadataQuery): RelWindowProperties = {
@@ -435,10 +436,16 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
def inferWindowPropertyAfterWindowJoin(
leftWindowProperty: ImmutableBitSet,
rightWindowProperty: ImmutableBitSet): ImmutableBitSet = {
+ // window columns on an outer join's null-generating side become
nullable, so they are no
+ // longer valid window properties
+ val preservedLeft =
+ if (joinType.generatesNullsOnLeft) ImmutableBitSet.of() else
leftWindowProperty
+ val preservedRight =
+ if (joinType.generatesNullsOnRight) ImmutableBitSet.of() else
rightWindowProperty
val fieldMapping = new JHashMap[Integer, Integer]()
(0 until rightFieldCnt).foreach(idx => fieldMapping.put(idx,
leftFieldCnt + idx))
- val rightWindowPropertyAfterWindowJoin =
rightWindowProperty.permute(fieldMapping)
- leftWindowProperty.union(rightWindowPropertyAfterWindowJoin)
+ val rightWindowPropertyAfterWindowJoin =
preservedRight.permute(fieldMapping)
+ preservedLeft.union(rightWindowPropertyAfterWindowJoin)
}
val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinSemanticTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinSemanticTestPrograms.java
new file mode 100644
index 00000000000..e77af3bc7a2
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinSemanticTestPrograms.java
@@ -0,0 +1,150 @@
+/*
+ * 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.nodes.exec.stream;
+
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+
+/**
+ * Semantic {@link TableTestProgram} definitions for a window aggregate placed
on top of a {@link
+ * StreamExecWindowJoin}.
+ */
+public class WindowJoinSemanticTestPrograms {
+
+ // window W1 = [00:00, 00:15), two left rows, one right row -> matched on
both sides
+ // window W2 = [00:15, 00:30), two left rows, no right rows -> left-only
+ // window W3 = [00:30, 00:45), no left rows, two right rows -> right-only
+ private static final Row[] SOURCE_ONE_DATA = {
+ Row.of("2020-10-10 00:01:00"),
+ Row.of("2020-10-10 00:02:00"),
+ Row.of("2020-10-10 00:16:00"),
+ Row.of("2020-10-10 00:17:00")
+ };
+
+ private static final Row[] SOURCE_TWO_DATA = {
+ Row.of("2020-10-10 00:03:00"), Row.of("2020-10-10 00:31:00"),
Row.of("2020-10-10 00:32:00")
+ };
+
+ private static final String[] SOURCE_SCHEMA = {
+ "ts STRING",
+ "rowtime AS TO_TIMESTAMP(`ts`)",
+ "WATERMARK for `rowtime` AS `rowtime` - INTERVAL '1' SECOND"
+ };
+
+ private static final String[] SINK_SCHEMA = {
+ "window_start TIMESTAMP(3)", "window_end TIMESTAMP(3)", "cnt BIGINT"
+ };
+
+ private static final SourceTestStep SOURCE_ONE =
+ SourceTestStep.newBuilder("source_one_t")
+ .addSchema(SOURCE_SCHEMA)
+ .producedValues(SOURCE_ONE_DATA)
+ .build();
+
+ private static final SourceTestStep SOURCE_TWO =
+ SourceTestStep.newBuilder("source_two_t")
+ .addSchema(SOURCE_SCHEMA)
+ .producedValues(SOURCE_TWO_DATA)
+ .build();
+
+ private static String windowAggregateOnWindowJoin(String joinType, String
windowSide) {
+ return String.format(
+ "INSERT INTO sink_t\n"
+ + "SELECT %1$s.window_start, %1$s.window_end, COUNT(*)
AS cnt\n"
+ + "FROM TUMBLE(TABLE source_one_t,
DESCRIPTOR(rowtime), INTERVAL '15' MINUTE) t_left\n"
+ + "%2$s TUMBLE(TABLE source_two_t,
DESCRIPTOR(rowtime), INTERVAL '15' MINUTE) t_right\n"
+ + "ON t_left.window_start = t_right.window_start"
+ + " AND t_left.window_end = t_right.window_end\n"
+ + "GROUP BY %1$s.window_start, %1$s.window_end",
+ windowSide, joinType);
+ }
+
+ public static final TableTestProgram WINDOW_AGGREGATE_ON_INNER_WINDOW_JOIN
=
+ TableTestProgram.of(
+ "window-aggregate-on-inner-window-join",
+ "validates a window aggregate on top of an inner
window join keeps the"
+ + " window properties, so no row carries
null window columns")
+ .setupTableSource(SOURCE_ONE)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SINK_SCHEMA)
+ .testMaterializedData()
+ .consumedValues("+I[2020-10-10T00:00,
2020-10-10T00:15, 2]")
+ .build())
+ .runSql(windowAggregateOnWindowJoin("JOIN", "t_right"))
+ .build();
+
+ public static final TableTestProgram WINDOW_AGGREGATE_ON_LEFT_WINDOW_JOIN =
+ TableTestProgram.of(
+ "window-aggregate-on-left-window-join",
+ "validates a window aggregate on top of a left
window join groups the"
+ + " unmatched rows whose (right-side)
window columns are null")
+ .setupTableSource(SOURCE_ONE)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SINK_SCHEMA)
+ .testMaterializedData()
+ .consumedValues(
+ "+I[2020-10-10T00:00,
2020-10-10T00:15, 2]",
+ "+I[null, null, 2]")
+ .build())
+ .runSql(windowAggregateOnWindowJoin("LEFT JOIN",
"t_right"))
+ .build();
+
+ public static final TableTestProgram WINDOW_AGGREGATE_ON_RIGHT_WINDOW_JOIN
=
+ TableTestProgram.of(
+ "window-aggregate-on-right-window-join",
+ "validates a window aggregate on top of a right
window join groups the"
+ + " unmatched rows whose (left-side)
window columns are null")
+ .setupTableSource(SOURCE_ONE)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SINK_SCHEMA)
+ .testMaterializedData()
+ .consumedValues(
+ "+I[2020-10-10T00:00,
2020-10-10T00:15, 2]",
+ "+I[null, null, 2]")
+ .build())
+ .runSql(windowAggregateOnWindowJoin("RIGHT JOIN",
"t_left"))
+ .build();
+
+ public static final TableTestProgram WINDOW_AGGREGATE_ON_FULL_WINDOW_JOIN =
+ TableTestProgram.of(
+ "window-aggregate-on-full-window-join",
+ "validates a window aggregate on top of a full
window join groups the"
+ + " unmatched rows whose window columns
are null")
+ .setupTableSource(SOURCE_ONE)
+ .setupTableSource(SOURCE_TWO)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SINK_SCHEMA)
+ .testMaterializedData()
+ .consumedValues(
+ "+I[2020-10-10T00:00,
2020-10-10T00:15, 2]",
+ "+I[2020-10-10T00:30,
2020-10-10T00:45, 2]",
+ "+I[null, null, 2]")
+ .build())
+ .runSql(windowAggregateOnWindowJoin("FULL JOIN",
"t_right"))
+ .build();
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinSemanticTests.java
new file mode 100644
index 00000000000..f432fcb7b1d
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/WindowJoinSemanticTests.java
@@ -0,0 +1,37 @@
+/*
+ * 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.nodes.exec.stream;
+
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.SemanticTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+
+/** Semantic tests for a window aggregate on top of a {@link
StreamExecWindowJoin}. */
+public class WindowJoinSemanticTests extends SemanticTestBase {
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return List.of(
+
WindowJoinSemanticTestPrograms.WINDOW_AGGREGATE_ON_INNER_WINDOW_JOIN,
+
WindowJoinSemanticTestPrograms.WINDOW_AGGREGATE_ON_LEFT_WINDOW_JOIN,
+
WindowJoinSemanticTestPrograms.WINDOW_AGGREGATE_ON_RIGHT_WINDOW_JOIN,
+
WindowJoinSemanticTestPrograms.WINDOW_AGGREGATE_ON_FULL_WINDOW_JOIN);
+ }
+}
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 07098bba996..b444868954f 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
@@ -502,6 +502,68 @@ Calc(select=[a, window_start, window_end, EXPR$3, EXPR$4,
EXPR$5])
+- Calc(select=[a, d, IS TRUE(>(b, 1000)) AS
$f5, b, e, c, rowtime])
+- WatermarkAssigner(rowtime=[rowtime],
watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+-
TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, d, e, rowtime])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testWindowTopNOnNullSideFallsBackToRegularRank">
+ <Resource name="sql">
+ <![CDATA[
+SELECT *
+FROM (
+ SELECT *,
+ ROW_NUMBER() OVER (
+ PARTITION BY null_window_start, null_window_end ORDER BY a DESC) AS
rownum
+ FROM (
+ SELECT
+ null_side.window_start AS null_window_start,
+ null_side.window_end AS null_window_end,
+ preserved_side.a AS a
+ FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
preserved_side
+ LEFT JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
null_side
+ ON preserved_side.window_start = null_side.window_start
+ AND preserved_side.window_end = null_side.window_end
+ AND preserved_side.a = null_side.a
+ )
+)
+WHERE rownum <= 3
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(null_window_start=[$0], null_window_end=[$1], a=[$2],
rownum=[$3])
++- LogicalFilter(condition=[<=($3, 3)])
+ +- LogicalProject(null_window_start=[$0], null_window_end=[$1], a=[$2],
rownum=[ROW_NUMBER() OVER (PARTITION BY $0, $1 ORDER BY $2 DESC NULLS LAST)])
+ +- LogicalProject(null_window_start=[$17], null_window_end=[$18], a=[$0])
+ +- LogicalJoin(condition=[AND(=($7, $17), =($8, $18), =($0, $10))],
joinType=[left])
+ :- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3)
d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME*
proctime, 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], proctime=[$6])
+ : +- LogicalWatermarkAssigner(rowtime=[rowtime],
watermark=[-($5, 1000:INTERVAL SECOND)])
+ : +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4],
rowtime=[$5], proctime=[PROCTIME()])
+ : +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3)
d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME*
proctime, 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], proctime=[$6])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime],
watermark=[-($5, 1000:INTERVAL SECOND)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4],
rowtime=[$5], proctime=[PROCTIME()])
+ +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[window_start0 AS null_window_start, window_end0 AS
null_window_end, a, rownum])
++- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
rankRange=[rankStart=1, rankEnd=3], partitionBy=[window_start0, window_end0],
orderBy=[a DESC], select=[a, window_start0, window_end0, rownum])
+ +- Exchange(distribution=[hash[window_start0, window_end0]])
+ +- Calc(select=[a, window_start0, window_end0])
+ +- WindowJoin(leftWindow=[TUMBLE(win_start=[window_start],
win_end=[window_end], size=[15 min])],
rightWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15
min])], joinType=[LeftOuterJoin], where=[=(a, a0)], select=[a, b, c, d, e,
rowtime, proctime, window_start, window_end, window_time, a0, b0, c0, d0, e0,
rowtime0, proctime0, window_start0, window_end0, window_time0])
+ :- Exchange(distribution=[hash[a]])
+ : +- 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])
+ +- Exchange(distribution=[hash[a]])
+ +- 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])
]]>
</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 68e4171ca14..ab610430755 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
@@ -2266,6 +2266,104 @@ Sink(table=[default_catalog.default_database.s1],
fields=[window_start, window_e
+- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime -
1000:INTERVAL SECOND)])
+- Calc(select=[rowtime])
+- Reused(reference_id=[1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testOnNullSideOfOuterWindowJoin[aggPhaseEnforcer=TWO_PHASE]">
+ <Resource name="sql">
+ <![CDATA[
+SELECT null_side.window_start, null_side.window_end, COUNT(*) AS cnt
+FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
preserved_side
+LEFT JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
null_side
+ON preserved_side.window_start = null_side.window_start
+ AND preserved_side.window_end = null_side.window_end
+ AND preserved_side.a = null_side.a
+GROUP BY null_side.window_start, null_side.window_end
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{0, 1}], cnt=[COUNT()])
++- LogicalProject(window_start=[$17], window_end=[$18])
+ +- LogicalJoin(condition=[AND(=($7, $17), =($8, $18), =($0, $10))],
joinType=[left])
+ :- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3)
d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME*
proctime, 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], proctime=[$6])
+ : +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5,
1000:INTERVAL SECOND)])
+ : +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4],
rowtime=[$5], proctime=[PROCTIME()])
+ : +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3)
d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME*
proctime, 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], proctime=[$6])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5,
1000:INTERVAL SECOND)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4],
rowtime=[$5], proctime=[PROCTIME()])
+ +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupAggregate(groupBy=[window_start, window_end], select=[window_start,
window_end, COUNT(*) AS cnt])
++- Exchange(distribution=[hash[window_start, window_end]])
+ +- Calc(select=[window_start0 AS window_start, window_end0 AS window_end])
+ +- WindowJoin(leftWindow=[TUMBLE(win_start=[window_start],
win_end=[window_end], size=[15 min])],
rightWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15
min])], joinType=[LeftOuterJoin], where=[=(a, a0)], select=[a, window_start,
window_end, a0, window_start0, window_end0])
+ :- Exchange(distribution=[hash[a]])
+ : +- Calc(select=[a, window_start, window_end])
+ : +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime],
size=[15 min])])
+ : +- WatermarkAssigner(rowtime=[rowtime],
watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, MyTable, project=[a, rowtime]]], fields=[a, rowtime])
+ +- Exchange(distribution=[hash[a]])
+ +- Calc(select=[a, window_start, window_end])
+ +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime],
size=[15 min])])
+ +- WatermarkAssigner(rowtime=[rowtime],
watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog,
default_database, MyTable, project=[a, rowtime]]], fields=[a, rowtime])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase
name="testOnPreservedSideOfOuterWindowJoin[aggPhaseEnforcer=TWO_PHASE]">
+ <Resource name="sql">
+ <![CDATA[
+SELECT preserved_side.window_start, preserved_side.window_end, COUNT(*) AS cnt
+FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
preserved_side
+LEFT JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
null_side
+ON preserved_side.window_start = null_side.window_start
+ AND preserved_side.window_end = null_side.window_end
+ AND preserved_side.a = null_side.a
+GROUP BY preserved_side.window_start, preserved_side.window_end
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{0, 1}], cnt=[COUNT()])
++- LogicalProject(window_start=[$7], window_end=[$8])
+ +- LogicalJoin(condition=[AND(=($7, $17), =($8, $18), =($0, $10))],
joinType=[left])
+ :- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3)
d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME*
proctime, 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], proctime=[$6])
+ : +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5,
1000:INTERVAL SECOND)])
+ : +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4],
rowtime=[$5], proctime=[PROCTIME()])
+ : +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, DECIMAL(10, 3)
d, BIGINT e, TIMESTAMP(3) *ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME*
proctime, 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], proctime=[$6])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($5,
1000:INTERVAL SECOND)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4],
rowtime=[$5], proctime=[PROCTIME()])
+ +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[window_start, window_end, cnt])
++- GlobalWindowAggregate(window=[TUMBLE(win_end=[$window_end], size=[15
min])], select=[COUNT(count1$0) AS cnt, start('w$) AS window_start, end('w$) AS
window_end])
+ +- Exchange(distribution=[single])
+ +- LocalWindowAggregate(window=[TUMBLE(win_start=[window_start],
win_end=[window_end], size=[15 min])], select=[COUNT(*) AS count1$0,
slice_end('w$) AS $window_end])
+ +- Calc(select=[window_start, window_end])
+ +- WindowJoin(leftWindow=[TUMBLE(win_start=[window_start],
win_end=[window_end], size=[15 min])],
rightWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15
min])], joinType=[LeftOuterJoin], where=[=(a, a0)], select=[a, window_start,
window_end, a0, window_start0, window_end0])
+ :- Exchange(distribution=[hash[a]])
+ : +- Calc(select=[a, window_start, window_end])
+ : +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime],
size=[15 min])])
+ : +- WatermarkAssigner(rowtime=[rowtime],
watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, MyTable, project=[a, rowtime]]], fields=[a, rowtime])
+ +- Exchange(distribution=[hash[a]])
+ +- Calc(select=[a, window_start, window_end])
+ +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime],
size=[15 min])])
+ +- WatermarkAssigner(rowtime=[rowtime],
watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog,
default_database, MyTable, project=[a, rowtime]]], fields=[a, 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 ca0cd46423c..80983fc90fa 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
@@ -2288,6 +2288,74 @@ Calc(select=[rowtime, a, l_cnt, l_uv, r_cnt, r_uv, a0,
b, c, rowtime0, PROCTIME_
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime,
1000:INTERVAL SECOND)])
+- Calc(select=[a, b, c, rowtime, PROCTIME() AS proctime])
+- TableSourceScan(table=[[default_catalog, default_database,
MyTable4]], fields=[a, b, c, rowtime])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testWindowJoinOnNullSideFallsBackToRegularJoin">
+ <Resource name="sql">
+ <![CDATA[
+SELECT outer_join.null_window_start, outer_join.null_window_end, probe_side.a
+FROM (
+ SELECT
+ null_side.window_start AS null_window_start,
+ null_side.window_end AS null_window_end,
+ preserved_side.a AS a
+ FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
preserved_side
+ LEFT JOIN TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
null_side
+ ON preserved_side.window_start = null_side.window_start
+ AND preserved_side.window_end = null_side.window_end
+ AND preserved_side.a = null_side.a
+) outer_join
+JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
probe_side
+ON outer_join.null_window_start = probe_side.window_start
+ AND outer_join.null_window_end = probe_side.window_end
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(null_window_start=[$0], null_window_end=[$1], a=[$3])
++- LogicalJoin(condition=[AND(=($0, $8), =($1, $9))], joinType=[inner])
+ :- LogicalProject(null_window_start=[$13], null_window_end=[$14], a=[$0])
+ : +- LogicalJoin(condition=[AND(=($5, $13), =($6, $14), =($0, $8))],
joinType=[left])
+ : :- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, VARCHAR(2147483647) b, BIGINT c, TIMESTAMP(3)
*ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME* proctime, TIMESTAMP(3)
window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ : : +- LogicalProject(a=[$0], b=[$1], c=[$2], rowtime=[$3],
proctime=[$4])
+ : : +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($3,
1000:INTERVAL SECOND)])
+ : : +- LogicalProject(a=[$0], b=[$1], c=[$2], rowtime=[$3],
proctime=[PROCTIME()])
+ : : +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable]])
+ : +- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, VARCHAR(2147483647) b, BIGINT c, TIMESTAMP(3)
*ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME* proctime, TIMESTAMP(3)
window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ : +- LogicalProject(a=[$0], b=[$1], c=[$2], rowtime=[$3],
proctime=[$4])
+ : +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($3,
1000:INTERVAL SECOND)])
+ : +- LogicalProject(a=[$0], b=[$1], c=[$2], rowtime=[$3],
proctime=[PROCTIME()])
+ : +- LogicalTableScan(table=[[default_catalog,
default_database, MyTable2]])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rowtime'), 900000:INTERVAL MINUTE)],
rowType=[RecordType(INTEGER a, VARCHAR(2147483647) b, BIGINT c, TIMESTAMP(3)
*ROWTIME* rowtime, TIMESTAMP_LTZ(3) *PROCTIME* proctime, TIMESTAMP(3)
window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], rowtime=[$3], proctime=[$4])
+ +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[-($3,
1000:INTERVAL SECOND)])
+ +- LogicalProject(a=[$0], b=[$1], c=[$2], rowtime=[$3],
proctime=[PROCTIME()])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
MyTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[null_window_start, null_window_end, a])
++- Join(joinType=[InnerJoin], where=[AND(=(null_window_start, window_start),
=(null_window_end, window_end))], select=[null_window_start, null_window_end,
a, window_start, window_end], leftInputSpec=[NoUniqueKey],
rightInputSpec=[NoUniqueKey])
+ :- Exchange(distribution=[hash[null_window_start, null_window_end]])
+ : +- Calc(select=[window_start0 AS null_window_start, window_end0 AS
null_window_end])
+ : +- WindowJoin(leftWindow=[TUMBLE(win_start=[window_start],
win_end=[window_end], size=[15 min])],
rightWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15
min])], joinType=[LeftOuterJoin], where=[=(a, a0)], select=[a, window_start,
window_end, a0, window_start0, window_end0])
+ : :- Exchange(distribution=[hash[a]])
+ : : +- Calc(select=[a, window_start, window_end])
+ : : +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime],
size=[15 min])])
+ : : +- WatermarkAssigner(rowtime=[rowtime],
watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ : : +- TableSourceScan(table=[[default_catalog,
default_database, MyTable, project=[a, rowtime]]], fields=[a, rowtime])
+ : +- Exchange(distribution=[hash[a]])
+ : +- Calc(select=[a, window_start, window_end])
+ : +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime],
size=[15 min])])
+ : +- WatermarkAssigner(rowtime=[rowtime],
watermark=[-(rowtime, 1000:INTERVAL SECOND)])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, MyTable2, project=[a, rowtime]]], fields=[a, rowtime])
+ +- Exchange(distribution=[hash[window_start, window_end]])
+ +- Calc(select=[a, window_start, window_end])
+ +- WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15
min])])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime,
1000:INTERVAL SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database,
MyTable, project=[a, rowtime]]], fields=[a, rowtime])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.scala
index fde6fbd50d6..704dd574a4e 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/WindowRankTest.scala
@@ -557,4 +557,30 @@ class WindowRankTest extends TableTestBase {
"Rank strategy rankEnd=max_b is not supported on window rank
currently.")
.isInstanceOf[TableException]
}
+
+ @Test
+ def testWindowTopNOnNullSideFallsBackToRegularRank(): Unit = {
+ val sql =
+ """
+ |SELECT *
+ |FROM (
+ | SELECT *,
+ | ROW_NUMBER() OVER (
+ | PARTITION BY null_window_start, null_window_end ORDER BY a
DESC) AS rownum
+ | FROM (
+ | SELECT
+ | null_side.window_start AS null_window_start,
+ | null_side.window_end AS null_window_end,
+ | preserved_side.a AS a
+ | FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
MINUTE) preserved_side
+ | LEFT JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL
'15' MINUTE) null_side
+ | ON preserved_side.window_start = null_side.window_start
+ | AND preserved_side.window_end = null_side.window_end
+ | AND preserved_side.a = null_side.a
+ | )
+ |)
+ |WHERE rownum <= 3
+ """.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 56987570274..8a89d76fc3f 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
@@ -1753,6 +1753,38 @@ class WindowAggregateTest(aggPhaseEnforcer:
AggregatePhaseStrategy) extends Tabl
|group by window_start, window_end, c, window_time
|""".stripMargin)
}
+
+ @TestTemplate
+ def testOnPreservedSideOfOuterWindowJoin(): Unit = {
+ assumeThat(isTwoPhase).isTrue
+ val sql =
+ """
+ |SELECT preserved_side.window_start, preserved_side.window_end,
COUNT(*) AS cnt
+ |FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
preserved_side
+ |LEFT JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
MINUTE) null_side
+ |ON preserved_side.window_start = null_side.window_start
+ | AND preserved_side.window_end = null_side.window_end
+ | AND preserved_side.a = null_side.a
+ |GROUP BY preserved_side.window_start, preserved_side.window_end
+ """.stripMargin
+ util.verifyRelPlan(sql)
+ }
+
+ @TestTemplate
+ def testOnNullSideOfOuterWindowJoin(): Unit = {
+ assumeThat(isTwoPhase).isTrue
+ val sql =
+ """
+ |SELECT null_side.window_start, null_side.window_end, COUNT(*) AS cnt
+ |FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
preserved_side
+ |LEFT JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
MINUTE) null_side
+ |ON preserved_side.window_start = null_side.window_start
+ | AND preserved_side.window_end = null_side.window_end
+ | AND preserved_side.a = null_side.a
+ |GROUP BY null_side.window_start, null_side.window_end
+ """.stripMargin
+ util.verifyRelPlan(sql)
+ }
}
object WindowAggregateTest {
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala
index 8ad6d45cdc8..59247a047ed 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala
@@ -1348,4 +1348,27 @@ class WindowJoinTest extends TableTestBase {
statementSet.addInsertSql("INSERT INTO sink2 SELECT * FROM food_view")
util.verifyRelPlan(statementSet)
}
+
+ @Test
+ def testWindowJoinOnNullSideFallsBackToRegularJoin(): Unit = {
+ val sql =
+ """
+ |SELECT outer_join.null_window_start, outer_join.null_window_end,
probe_side.a
+ |FROM (
+ | SELECT
+ | null_side.window_start AS null_window_start,
+ | null_side.window_end AS null_window_end,
+ | preserved_side.a AS a
+ | FROM TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15'
MINUTE) preserved_side
+ | LEFT JOIN TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15'
MINUTE) null_side
+ | ON preserved_side.window_start = null_side.window_start
+ | AND preserved_side.window_end = null_side.window_end
+ | AND preserved_side.a = null_side.a
+ |) outer_join
+ |JOIN TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE)
probe_side
+ |ON outer_join.null_window_start = probe_side.window_start
+ | AND outer_join.null_window_end = probe_side.window_end
+ """.stripMargin
+ util.verifyRelPlan(sql)
+ }
}
