This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new 71f23066562 [FLINK-31424][table-planner] Fix NPE produced by multiple
sink with local-global window aggregate
71f23066562 is described below
commit 71f23066562018910e5599719d2a1bc0eb17f430
Author: Jane Chan <55568005+ladyfor...@users.noreply.github.com>
AuthorDate: Thu Mar 23 12:04:42 2023 +0800
[FLINK-31424][table-planner] Fix NPE produced by multiple sink with
local-global window aggregate
This closes #22238
---
.../plan/metadata/FlinkRelMdWindowProperties.scala | 40 ++-
.../table/planner/plan/stats/FlinkStatistic.scala | 6 +-
.../plan/stream/sql/join/WindowJoinTest.xml | 72 +++++
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 318 ++++++++++++++++++++-
.../metadata/FlinkRelMdWindowPropertiesTest.scala | 184 ++++++++++++
.../planner/plan/metadata/MetadataTestUtil.scala | 36 +++
.../plan/stream/sql/join/WindowJoinTest.scala | 60 ++++
7 files changed, 708 insertions(+), 8 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 30087585974..a688302e914 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
@@ -19,6 +19,7 @@ package org.apache.flink.table.planner.plan.metadata
import org.apache.flink.table.planner.{JArrayList, JHashMap, JList}
import org.apache.flink.table.planner.plan.`trait`.RelWindowProperties
+import org.apache.flink.table.planner.plan.logical.WindowSpec
import org.apache.flink.table.planner.plan.nodes.calcite.{Expand,
WatermarkAssigner}
import
org.apache.flink.table.planner.plan.nodes.logical.{FlinkLogicalAggregate,
FlinkLogicalCorrelate, FlinkLogicalJoin, FlinkLogicalRank}
import
org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin
@@ -27,6 +28,7 @@ import
org.apache.flink.table.planner.plan.schema.FlinkPreparingTableBase
import
org.apache.flink.table.planner.plan.utils.WindowJoinUtil.satisfyWindowJoin
import
org.apache.flink.table.planner.plan.utils.WindowUtil.{convertToWindowingStrategy,
groupingContainsWindowStartEnd, isWindowTableFunctionCall}
import org.apache.flink.table.runtime.groupwindow._
+import org.apache.flink.table.types.logical.LogicalType
import org.apache.calcite.plan.hep.HepRelVertex
import org.apache.calcite.plan.volcano.RelSubset
@@ -254,11 +256,34 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
def getWindowProperties(
rel: StreamPhysicalWindowAggregate,
mq: RelMetadataQuery): RelWindowProperties = {
+ getWindowAggregateWindowProperties(
+ rel.grouping.length + rel.aggCalls.size(),
+ rel.namedWindowProperties,
+ rel.windowing.getWindow,
+ rel.windowing.getTimeAttributeType
+ )
+ }
+
+ def getWindowProperties(
+ rel: StreamPhysicalGlobalWindowAggregate,
+ mq: RelMetadataQuery): RelWindowProperties = {
+ getWindowAggregateWindowProperties(
+ rel.grouping.length + rel.aggCalls.size(),
+ rel.namedWindowProperties,
+ rel.windowing.getWindow,
+ rel.windowing.getTimeAttributeType
+ )
+ }
+
+ private def getWindowAggregateWindowProperties(
+ propertyOffset: Int,
+ windowProperties: Seq[NamedWindowProperty],
+ windowSpec: WindowSpec,
+ timeAttributeType: LogicalType): RelWindowProperties = {
val starts = ArrayBuffer[Int]()
val ends = ArrayBuffer[Int]()
val times = ArrayBuffer[Int]()
- val propertyOffset = rel.grouping.length + rel.aggCalls.size()
- rel.namedWindowProperties.map(_.getProperty).zipWithIndex.foreach {
+ windowProperties.map(_.getProperty).zipWithIndex.foreach {
case (p, index) =>
p match {
case _: WindowStart =>
@@ -275,11 +300,18 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
ImmutableBitSet.of(starts: _*),
ImmutableBitSet.of(ends: _*),
ImmutableBitSet.of(times: _*),
- rel.windowing.getWindow,
- rel.windowing.getTimeAttributeType
+ windowSpec,
+ timeAttributeType
)
}
+ def getWindowProperties(
+ rel: StreamPhysicalLocalWindowAggregate,
+ mq: RelMetadataQuery): RelWindowProperties = {
+ val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
+ fmq.getRelWindowProperties(rel.getInput)
+ }
+
def getWindowProperties(
rel: StreamPhysicalWindowRank,
mq: RelMetadataQuery): RelWindowProperties = {
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/stats/FlinkStatistic.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/stats/FlinkStatistic.scala
index 4d5b4ecf2f7..7158a8039c9 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/stats/FlinkStatistic.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/stats/FlinkStatistic.scala
@@ -17,7 +17,7 @@
*/
package org.apache.flink.table.planner.plan.stats
-import org.apache.flink.table.catalog.{ContextResolvedTable, ResolvedSchema,
UniqueConstraint}
+import org.apache.flink.table.catalog.{ResolvedSchema, UniqueConstraint}
import org.apache.flink.table.plan.stats.{ColumnStats, TableStats}
import org.apache.flink.table.planner.plan.`trait`.{RelModifiedMonotonicity,
RelWindowProperties}
@@ -29,7 +29,6 @@ import org.apache.calcite.util.ImmutableBitSet
import javax.annotation.Nullable
import java.util
-import java.util.{HashSet, Optional, Set}
import scala.collection.JavaConversions._
@@ -206,6 +205,7 @@ object FlinkStatistic {
this.tableStats = statistic.getTableStats
this.uniqueKeys = statistic.getUniqueKeys
this.relModifiedMonotonicity = statistic.getRelModifiedMonotonicity
+ this.windowProperties = statistic.getRelWindowProperties
this
}
@@ -218,7 +218,7 @@ object FlinkStatistic {
) {
UNKNOWN
} else {
- new FlinkStatistic(tableStats, uniqueKeys, relModifiedMonotonicity)
+ new FlinkStatistic(tableStats, uniqueKeys, relModifiedMonotonicity,
windowProperties)
}
}
}
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 39a7abd3fd2..f94791d3b5a 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
@@ -154,6 +154,78 @@ WindowJoin(leftWindow=[TUMBLE(win_start=[window_start],
win_end=[window_end], si
+- LocalWindowAggregate(groupBy=[a],
window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[a, slice_end('w$)
AS $slice_end])
+- WatermarkAssigner(rowtime=[rowtime], watermark=[-(rowtime,
1000:INTERVAL SECOND)])
+- TableSourceScan(table=[[default_catalog,
default_database, MyTable2, project=[a, c, rowtime], metadata=[]]], fields=[a,
c, rowtime])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJoinToMultiSink">
+ <Resource name="ast">
+ <![CDATA[
+LogicalSink(table=[default_catalog.default_database.sink1],
fields=[window_start, window_end, user_id, dt, hour])
++- LogicalProject(window_start=[$1], window_end=[$2], user_id=[$0],
dt=[DATE_FORMAT(+($2, 25200000:INTERVAL HOUR), _UTF-16LE'yyyyMMdd')],
hour=[DATE_FORMAT(+($2, 25200000:INTERVAL HOUR), _UTF-16LE'HH')])
+ +- LogicalJoin(condition=[AND(=($0, $3), =($1, $4), =($2, $5))],
joinType=[left])
+ :- LogicalAggregate(group=[{0, 1, 2}])
+ : +- LogicalProject(user_id=[$0], window_start=[$4], window_end=[$5])
+ : +- LogicalTableFunctionScan(invocation=[TUMBLE($3, DESCRIPTOR($3),
60000:INTERVAL MINUTE)], rowType=[RecordType(VARCHAR(2147483647) user_id,
VARCHAR(2147483647) order_id, INTEGER amount, TIMESTAMP(3) *ROWTIME*
event_time, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3)
*ROWTIME* window_time)])
+ : +- LogicalProject(user_id=[$0], order_id=[$1], amount=[$2],
event_time=[$3])
+ : +- LogicalWatermarkAssigner(rowtime=[event_time],
watermark=[$3])
+ : +- LogicalTableScan(table=[[default_catalog,
default_database, food_order]])
+ +- LogicalAggregate(group=[{0, 1, 2}])
+ +- LogicalProject(user_id=[$0], window_start=[$4], window_end=[$5])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE($3, DESCRIPTOR($3),
60000:INTERVAL MINUTE)], rowType=[RecordType(VARCHAR(2147483647) user_id,
VARCHAR(2147483647) order_id, INTEGER amount, TIMESTAMP(3) *ROWTIME*
event_time, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3)
*ROWTIME* window_time)])
+ +- LogicalProject(user_id=[$0], order_id=[$1], amount=[$2],
event_time=[$3])
+ +- LogicalWatermarkAssigner(rowtime=[event_time],
watermark=[$3])
+ +- LogicalTableScan(table=[[default_catalog,
default_database, food_order]])
+
+LogicalSink(table=[default_catalog.default_database.sink2],
fields=[window_start, window_end, user_id, dt, hour])
++- LogicalProject(window_start=[$1], window_end=[$2], user_id=[$0],
dt=[DATE_FORMAT(+($2, 25200000:INTERVAL HOUR), _UTF-16LE'yyyyMMdd')],
hour=[DATE_FORMAT(+($2, 25200000:INTERVAL HOUR), _UTF-16LE'HH')])
+ +- LogicalJoin(condition=[AND(=($0, $3), =($1, $4), =($2, $5))],
joinType=[left])
+ :- LogicalAggregate(group=[{0, 1, 2}])
+ : +- LogicalProject(user_id=[$0], window_start=[$4], window_end=[$5])
+ : +- LogicalTableFunctionScan(invocation=[TUMBLE($3, DESCRIPTOR($3),
60000:INTERVAL MINUTE)], rowType=[RecordType(VARCHAR(2147483647) user_id,
VARCHAR(2147483647) order_id, INTEGER amount, TIMESTAMP(3) *ROWTIME*
event_time, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3)
*ROWTIME* window_time)])
+ : +- LogicalProject(user_id=[$0], order_id=[$1], amount=[$2],
event_time=[$3])
+ : +- LogicalWatermarkAssigner(rowtime=[event_time],
watermark=[$3])
+ : +- LogicalTableScan(table=[[default_catalog,
default_database, food_order]])
+ +- LogicalAggregate(group=[{0, 1, 2}])
+ +- LogicalProject(user_id=[$0], window_start=[$4], window_end=[$5])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE($3, DESCRIPTOR($3),
60000:INTERVAL MINUTE)], rowType=[RecordType(VARCHAR(2147483647) user_id,
VARCHAR(2147483647) order_id, INTEGER amount, TIMESTAMP(3) *ROWTIME*
event_time, TIMESTAMP(3) window_start, TIMESTAMP(3) window_end, TIMESTAMP(3)
*ROWTIME* window_time)])
+ +- LogicalProject(user_id=[$0], order_id=[$1], amount=[$2],
event_time=[$3])
+ +- LogicalWatermarkAssigner(rowtime=[event_time],
watermark=[$3])
+ +- LogicalTableScan(table=[[default_catalog,
default_database, food_order]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Sink(table=[default_catalog.default_database.sink1], fields=[window_start,
window_end, user_id, dt, hour])
++- Calc(select=[window_start, window_end, user_id, DATE_FORMAT(+(window_end,
25200000:INTERVAL HOUR), 'yyyyMMdd') AS dt, DATE_FORMAT(+(window_end,
25200000:INTERVAL HOUR), 'HH') AS hour])
+ +- WindowJoin(leftWindow=[TUMBLE(win_start=[window_start],
win_end=[window_end], size=[1 min])],
rightWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[1
min])], joinType=[LeftOuterJoin], where=[=(user_id, user_id0)],
select=[user_id, window_start, window_end, user_id0, window_start0,
window_end0])
+ :- Exchange(distribution=[hash[user_id]])
+ : +- GlobalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(slice_end=[$slice_end], size=[1 min])], select=[user_id,
start('w$) AS window_start, end('w$) AS window_end])
+ : +- Exchange(distribution=[hash[user_id]])
+ : +- LocalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(time_col=[event_time], size=[1 min])], select=[user_id,
slice_end('w$) AS $slice_end])
+ : +- WatermarkAssigner(rowtime=[event_time],
watermark=[event_time])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, food_order, project=[user_id, event_time], metadata=[]]],
fields=[user_id, event_time])
+ +- Exchange(distribution=[hash[user_id]])
+ +- GlobalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(slice_end=[$slice_end], size=[1 min])], select=[user_id,
start('w$) AS window_start, end('w$) AS window_end])
+ +- Exchange(distribution=[hash[user_id]])
+ +- LocalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(time_col=[event_time], size=[1 min])], select=[user_id,
slice_end('w$) AS $slice_end])
+ +- WatermarkAssigner(rowtime=[event_time],
watermark=[event_time])
+ +- TableSourceScan(table=[[default_catalog,
default_database, food_order, project=[user_id, event_time], metadata=[]]],
fields=[user_id, event_time])
+
+Sink(table=[default_catalog.default_database.sink2], fields=[window_start,
window_end, user_id, dt, hour])
++- Calc(select=[window_start, window_end, user_id, DATE_FORMAT(+(window_end,
25200000:INTERVAL HOUR), 'yyyyMMdd') AS dt, DATE_FORMAT(+(window_end,
25200000:INTERVAL HOUR), 'HH') AS hour])
+ +- WindowJoin(leftWindow=[TUMBLE(win_start=[window_start],
win_end=[window_end], size=[1 min])],
rightWindow=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[1
min])], joinType=[LeftOuterJoin], where=[=(user_id, user_id0)],
select=[user_id, window_start, window_end, user_id0, window_start0,
window_end0])
+ :- Exchange(distribution=[hash[user_id]])
+ : +- GlobalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(slice_end=[$slice_end], size=[1 min])], select=[user_id,
start('w$) AS window_start, end('w$) AS window_end])
+ : +- Exchange(distribution=[hash[user_id]])
+ : +- LocalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(time_col=[event_time], size=[1 min])], select=[user_id,
slice_end('w$) AS $slice_end])
+ : +- WatermarkAssigner(rowtime=[event_time],
watermark=[event_time])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, food_order, project=[user_id, event_time], metadata=[]]],
fields=[user_id, event_time])
+ +- Exchange(distribution=[hash[user_id]])
+ +- GlobalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(slice_end=[$slice_end], size=[1 min])], select=[user_id,
start('w$) AS window_start, end('w$) AS window_end])
+ +- Exchange(distribution=[hash[user_id]])
+ +- LocalWindowAggregate(groupBy=[user_id],
window=[TUMBLE(time_col=[event_time], size=[1 min])], select=[user_id,
slice_end('w$) AS $slice_end])
+ +- WatermarkAssigner(rowtime=[event_time],
watermark=[event_time])
+ +- TableSourceScan(table=[[default_catalog,
default_database, food_order, project=[user_id, event_time], metadata=[]]],
fields=[user_id, event_time])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
index 5535fc0671e..48d2dc204e0 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.metadata
import org.apache.flink.table.api.{TableConfig, TableException}
+import org.apache.flink.table.catalog.ContextResolvedFunction
import org.apache.flink.table.data.RowData
import org.apache.flink.table.expressions._
import org.apache.flink.table.expressions.ApiExpressionUtils.intervalOfMillis
@@ -25,6 +26,7 @@ import org.apache.flink.table.functions.{FunctionIdentifier,
UserDefinedFunction
import org.apache.flink.table.operations.TableSourceQueryOperation
import org.apache.flink.table.planner.calcite.{FlinkRelBuilder,
FlinkTypeFactory}
import org.apache.flink.table.planner.delegation.PlannerContext
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction
import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable
import org.apache.flink.table.planner.functions.utils.AggSqlFunction
import org.apache.flink.table.planner.plan.`trait`.{FlinkRelDistribution,
FlinkRelDistributionTraitDef}
@@ -39,6 +41,7 @@ import
org.apache.flink.table.planner.plan.nodes.physical.stream._
import org.apache.flink.table.planner.plan.schema.{FlinkPreparingTableBase,
IntermediateRelTable, TableSourceTable}
import org.apache.flink.table.planner.plan.stream.sql.join.TestTemporalTable
import org.apache.flink.table.planner.plan.utils._
+import
org.apache.flink.table.planner.runtime.utils.JavaUserDefinedTableFunctions
import org.apache.flink.table.planner.utils.{PlannerMocks, Top3}
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapContext
import org.apache.flink.table.runtime.groupwindow._
@@ -48,6 +51,7 @@ import org.apache.flink.table.types.logical._
import org.apache.flink.table.types.utils.TypeConversions
import com.google.common.collect.{ImmutableList, Lists}
+import org.apache.calcite.avatica.util.TimeUnit
import org.apache.calcite.jdbc.CalciteSchema
import org.apache.calcite.plan._
import org.apache.calcite.prepare.CalciteCatalogReader
@@ -60,7 +64,7 @@ import
org.apache.calcite.rel.metadata.{JaninoRelMetadataProvider, RelMetadataQu
import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.{BasicSqlType, SqlTypeName}
import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.{SqlAggFunction, SqlWindow}
+import org.apache.calcite.sql.{SqlAggFunction, SqlIntervalQualifier, SqlWindow}
import org.apache.calcite.sql.fun.{SqlCountAggFunction, SqlStdOperatorTable}
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.calcite.sql.parser.SqlParserPos
@@ -197,6 +201,13 @@ class FlinkRelMdHandlerTestBase {
protected lazy val tableSourceTableNonKeyStreamScan:
StreamPhysicalDataStreamScan =
createTableSourceTable(ImmutableList.of("TableSourceTable3"),
streamPhysicalTraits)
+ protected lazy val temporalTableLogicalScan: LogicalTableScan =
+ createDataStreamScan(ImmutableList.of("TemporalTable4"), logicalTraits)
+ protected lazy val temporalTableFlinkLogicalScan:
FlinkLogicalDataStreamTableScan =
+ createDataStreamScan(ImmutableList.of("TemporalTable4"),
flinkLogicalTraits)
+ protected lazy val temporalTableStreamScan: StreamPhysicalDataStreamScan =
+ createDataStreamScan(ImmutableList.of("TemporalTable4"),
streamPhysicalTraits)
+
private lazy val valuesType = relBuilder.getTypeFactory
.builder()
.add("a", SqlTypeName.BIGINT)
@@ -3156,6 +3167,10 @@ class FlinkRelMdHandlerTestBase {
protected lazy val batchCumulateWindowTVFRel = createWindowTVFRel(false,
cumulateWindowSpec)
protected lazy val streamCumulateWindowTVFRel = createWindowTVFRel(true,
cumulateWindowSpec)
+ protected lazy val timeAttributeType = new TimestampType(true,
TimestampKind.ROWTIME, 3)
+ protected lazy val proctimeType = new LocalZonedTimestampType(true,
TimestampKind.PROCTIME, 3)
+ protected lazy val windowStartEndType = new TimestampType(false, 3)
+
protected def createWindowTVFRel(
isStreamingMode: Boolean,
windowSpec: WindowSpec): CommonPhysicalWindowTableFunction = {
@@ -3195,6 +3210,307 @@ class FlinkRelMdHandlerTestBase {
}
}
+ // equivalent SQL is
+ // SELECT * FROM
+ // TABLE(TUMBLE(TABLE tmp, DESCRIPTOR(ptime), INTERVAL '10' MINUTE))
+ // CREATE TEMPORARY VIEW tmp AS
+ // SELECT `id`, `name`, PROCTIME() AS ptime FROM student
+ protected lazy val windowTableFunctionScan: TableFunctionScan =
createTableFunctionScan(true)
+
+ // equivalent SQL is
+ // SELECT `name`, `val` FROM student,
+ // LATERAL TABLE(STRING_SPLIT(`name`, CAST(`id` AS STRING))) AS T(`val`);
+ protected lazy val lateralTableFunctionScan: TableFunctionScan =
createTableFunctionScan(false)
+
+ protected def createTableFunctionScan(windowFunctionCall: Boolean):
TableFunctionScan = {
+ relBuilder.push(studentLogicalScan)
+
+ if (windowFunctionCall) {
+ val projects = List(
+ relBuilder.field(0),
+ relBuilder.field(1),
+ relBuilder.call(FlinkSqlOperatorTable.PROCTIME))
+ val outputRowType = typeFactory.buildRelNodeRowType(
+ Array("id", "name", "ptime"),
+ Array(
+ new BigIntType,
+ new VarCharType,
+ proctimeType
+ )
+ )
+ val calcOnStudentScan =
+ createLogicalCalc(
+ studentLogicalScan,
+ outputRowType,
+ projects,
+ null
+ )
+ new FlinkLogicalTableFunctionScan(
+ cluster,
+ logicalTraits,
+ ImmutableList.of(calcOnStudentScan),
+ relBuilder.call(
+ FlinkSqlOperatorTable.TUMBLE,
+ relBuilder.field(2),
+ relBuilder.call(FlinkSqlOperatorTable.DESCRIPTOR,
relBuilder.field(2)),
+ rexBuilder.makeIntervalLiteral(
+ bd(600000L),
+ new SqlIntervalQualifier(TimeUnit.MILLISECOND, null,
SqlParserPos.ZERO))
+ ),
+ null,
+ typeFactory.builder
+ .kind(outputRowType.getStructKind)
+ .addAll(outputRowType.getFieldList)
+ .add("window_start", SqlTypeName.TIMESTAMP, 3)
+ .add("window_end", SqlTypeName.TIMESTAMP, 3)
+ .add("window_time", outputRowType.getFieldList.get(2).getType)
+ .build,
+ null)
+ } else {
+ val correlVar = rexBuilder.makeCorrel(
+ typeFactory.buildRelNodeRowType(
+ Array("id", "name", "val"),
+ Array(
+ new BigIntType,
+ new VarCharType,
+ new VarCharType
+ )
+ ),
+ new CorrelationId(0))
+ val tableFunctionCall = relBuilder.call(
+ BridgingSqlFunction.of(
+ relBuilder.getCluster,
+ ContextResolvedFunction.temporary(
+ FunctionIdentifier.of("STRING_SPLIT"),
+ new JavaUserDefinedTableFunctions.StringSplit())),
+ rexBuilder.makeFieldAccess(correlVar, 1),
+ rexBuilder.makeCall(stringType, CAST,
List(rexBuilder.makeFieldAccess(correlVar, 0)))
+ )
+ new FlinkLogicalTableFunctionScan(
+ cluster,
+ logicalTraits,
+ ImmutableList.of(),
+ tableFunctionCall,
+ null,
+ typeFactory.buildRelNodeRowType(Array("EXPR$0"), Array(new
VarCharType)),
+ null
+ )
+ }
+ }
+
+ // equivalent SQL is
+ // SELECT b, window_end AS my_window_end, window_start FROM
+ // TABLE(TUMBLE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE))
+ protected lazy val keepWindowCalcOnTumbleWindowTVF: Calc =
+ createCalcOnWindowTVF(streamTumbleWindowTVFRel, true)
+
+ // equivalent SQL is
+ // SELECT b, CAST(window_end AS STRING) AS my_end, CAST(window_start AS
STRING) AS my_start FROM
+ // TABLE(TUMBLE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE))
+ protected lazy val discardWindowCalcOnTumbleWindowTVF: Calc =
+ createCalcOnWindowTVF(streamTumbleWindowTVFRel, false)
+ // equivalent SQL is
+ // SELECT b, window_end AS my_window_end, window_start FROM
+ // TABLE(HOP (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE, INTERVAL '1' HOUR))
+ protected lazy val keepWindowCalcOnHopWindowTVF: Calc =
+ createCalcOnWindowTVF(streamHopWindowTVFRel, true)
+
+ // equivalent SQL is
+ // SELECT b, CAST(window_end AS STRING) AS my_end, CAST(window_start AS
STRING) AS my_start FROM
+ // TABLE(HOP (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE, INTERVAL '1' HOUR))
+ protected lazy val discardWindowCalcOnHopWindowTVF: Calc =
+ createCalcOnWindowTVF(streamHopWindowTVFRel, false)
+ // equivalent SQL is
+ // SELECT b, window_end AS my_window_end, window_start FROM
+ // TABLE(CUMULATE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL
'10' MINUTE, INTERVAL '1' HOUR))
+ protected lazy val keepWindowCalcOnCumulateWindowTVF: Calc =
+ createCalcOnWindowTVF(streamCumulateWindowTVFRel, true)
+
+ // equivalent SQL is
+ // SELECT b, CAST(window_end AS STRING) AS my_end, CAST(window_start AS
STRING) AS my_start FROM
+ // TABLE(CUMULATE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL
'10' MINUTE, INTERVAL '1' HOUR))
+ protected lazy val discardWindowCalcOnCumulateWindowTVF: Calc =
+ createCalcOnWindowTVF(streamCumulateWindowTVFRel, false)
+ protected def createCalcOnWindowTVF(
+ tvf: CommonPhysicalWindowTableFunction,
+ projectKeepWindow: Boolean): Calc = {
+ relBuilder.push(tvf)
+ val (projects, outputType) = {
+ if (projectKeepWindow) {
+ (
+ List(relBuilder.field(1), relBuilder.field(6), relBuilder.field(5)),
+ typeFactory.buildRelNodeRowType(
+ Array("b", "my_window_end", "window_start"),
+ Array(
+ VarCharType.STRING_TYPE,
+ windowStartEndType,
+ windowStartEndType
+ )
+ ))
+ } else {
+ (
+ List(
+ relBuilder.field(1),
+ rexBuilder.makeCast(stringType, relBuilder.field(6)),
+ rexBuilder.makeCast(stringType, relBuilder.field(5))),
+ typeFactory.buildRelNodeRowType(
+ Array("b", "my_window_end", "window_start"),
+ Array(
+ VarCharType.STRING_TYPE,
+ VarCharType.STRING_TYPE,
+ VarCharType.STRING_TYPE
+ )
+ ))
+ }
+ }
+ val program = RexProgram.create(tvf.getRowType, projects, null,
outputType, rexBuilder)
+ new StreamPhysicalCalc(
+ cluster,
+ streamPhysicalTraits,
+ tvf,
+ program,
+ program.getOutputRowType
+ )
+ }
+
+ // equivalent SQL is
+ // SELECT * FROM
+ // TABLE(TUMBLE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE))
+ // UNION ALL
+ // (SELECT * FROM
+ // TABLE(TUMBLE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE)))
+ protected lazy val unionOnWindowTVFWithSameWindowSpec: Union =
+ createUnionOnWindowTVF(streamTumbleWindowTVFRel, streamTumbleWindowTVFRel)
+
+ // SELECT * FROM
+ // TABLE(TUMBLE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE))
+ // UNION ALL
+ // (SELECT * FROM
+ // TABLE(HOP (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE, INTERVAL '1' HOUR)))
+ protected lazy val unionOnWindowTVFWithDifferentWindowSpec: Union =
+ createUnionOnWindowTVF(streamTumbleWindowTVFRel, streamHopWindowTVFRel)
+
+ protected def createUnionOnWindowTVF(
+ tvf1: CommonPhysicalWindowTableFunction,
+ tvf2: CommonPhysicalWindowTableFunction): Union = {
+ new StreamPhysicalUnion(cluster, streamPhysicalTraits, List(tvf1, tvf2),
true, tvf1.getRowType)
+ }
+
+ // hash by field a
+ protected lazy val hashOnTumbleWindowTVF =
createExchangeOnWindowTVF(streamTumbleWindowTVFRel)
+ protected lazy val hashOnHopWindowTVF =
createExchangeOnWindowTVF(streamHopWindowTVFRel)
+ protected lazy val hashOnCumulateWindowTVF =
createExchangeOnWindowTVF(streamCumulateWindowTVFRel)
+
+ protected def createExchangeOnWindowTVF(tvf:
CommonPhysicalWindowTableFunction): Exchange = {
+ val hash = FlinkRelDistribution.hash(Array(0), requireStrict = true)
+ new StreamPhysicalExchange(
+ cluster,
+ streamPhysicalTraits.replace(hash),
+ tvf,
+ hash
+ )
+ }
+
+ // equivalent SQL is
+ // CREATE TEMPORARY VIEW tmp AS
+ // SELECT `id`, `name`, PROCTIME() AS ptime FROM student
+ // SELECT `id`, `window_start`, `window_end`, COUNT(DISTINCT `name`) AS cnt
FROM
+ // (TABLE(TUMBLE(TABLE tmp, DESCRIPTOR(ptime), INTERVAL '10' MINUTE)))
alias GROUP BY `id`, `window_start`, `window_end`
+ protected lazy val logicalGroupWindowAggOnTumbleWindowTVF =
createLogicalAggregateOnWindowTVF(
+ true)
+
+ // equivalent SQL is
+ // CREATE TEMPORARY VIEW tmp AS
+ // SELECT `id`, `name`, PROCTIME() AS ptime FROM student
+ // SELECT `id`, COUNT(DISTINCT `name`) AS cnt FROM
+ // (TABLE(TUMBLE(TABLE tmp, DESCRIPTOR(ptime), INTERVAL '10' MINUTE)))
alias GROUP BY `id`
+ protected lazy val logicalGroupAggOnTumbleWindowTVF =
createLogicalAggregateOnWindowTVF(false)
+ protected def createLogicalAggregateOnWindowTVF(groupByWindow: Boolean):
FlinkLogicalAggregate = {
+ relBuilder.push(windowTableFunctionScan)
+ val groupKey =
+ if (groupByWindow)
+ List(relBuilder.field(0), relBuilder.field(3), relBuilder.field(4))
+ else List(relBuilder.field(0))
+ val logicalAgg =
+ relBuilder
+ .aggregate(
+ relBuilder.groupKey(groupKey),
+ relBuilder.count(true, "cnt", relBuilder.field(1)))
+ .build()
+ .asInstanceOf[LogicalAggregate]
+ new FlinkLogicalAggregate(
+ cluster,
+ flinkLogicalTraits,
+ windowTableFunctionScan,
+ logicalAgg.getGroupSet,
+ logicalAgg.getGroupSets,
+ logicalAgg.getAggCallList
+ )
+ }
+
+ // equivalent SQL is
+ // SELECT window_start, window_end, SUM(a) AS sum_a, COUNT(b) AS cnt_b FROM
+ // TABLE(TUMBLE (TABLE TemporalTable1, DESCRIPTOR(rowtime), INTERVAL '10'
MINUTE))
+ // GROUP BY window_start, window_end
+ protected lazy val (
+ streamWindowAggOnWindowTVF,
+ streamLocalWindowAggOnWindowTVF,
+ streamGlobalWindowAggOnWindowTVF) = {
+ val logicalAgg =
+ relBuilder
+ .push(streamTumbleWindowTVFRel)
+ .aggregate(
+ relBuilder.groupKey(relBuilder.field(5), relBuilder.field(6)),
+ relBuilder.sum(false, "sum_a", relBuilder.field(0)),
+ relBuilder.count(false, "cnt_b", relBuilder.field(1))
+ )
+ .build()
+ .asInstanceOf[LogicalAggregate]
+
+ val windowRef = new WindowReference("w$", timeAttributeType)
+ val namedWindowProperties: Seq[NamedWindowProperty] = Seq(
+ new NamedWindowProperty("window_start", new WindowStart(windowRef)),
+ new NamedWindowProperty("window_end", new WindowEnd(windowRef)))
+ val traitSet = streamPhysicalTraits.replace(FlinkRelDistribution.SINGLETON)
+ val streamWindowAggOnWindowTVF = new StreamPhysicalWindowAggregate(
+ cluster,
+ traitSet,
+ streamTumbleWindowTVFRel,
+ Array(),
+ logicalAgg.getAggCallList,
+ new WindowAttachedWindowingStrategy(tumbleWindowSpec, timeAttributeType,
5, 6),
+ namedWindowProperties)
+
+ val streamLocalWindowAggOnWindowTVF = new
StreamPhysicalLocalWindowAggregate(
+ cluster,
+ traitSet,
+ streamTumbleWindowTVFRel,
+ streamWindowAggOnWindowTVF.grouping,
+ streamWindowAggOnWindowTVF.aggCalls,
+ streamWindowAggOnWindowTVF.windowing)
+
+ val exchange = new StreamPhysicalExchange(
+ cluster,
+ traitSet,
+ streamLocalWindowAggOnWindowTVF,
+ FlinkRelDistribution.SINGLETON)
+ val globalWindowing = new SliceAttachedWindowingStrategy(
+ tumbleWindowSpec,
+ timeAttributeType,
+ streamLocalWindowAggOnWindowTVF.getRowType.getFieldCount - 1)
+ val streamGlobalWindowAggOnWindowTVF = new
StreamPhysicalGlobalWindowAggregate(
+ cluster,
+ traitSet,
+ exchange,
+ streamTumbleWindowTVFRel.getRowType,
+ Array(),
+ streamWindowAggOnWindowTVF.aggCalls,
+ globalWindowing,
+ namedWindowProperties)
+
+ (streamWindowAggOnWindowTVF, streamLocalWindowAggOnWindowTVF,
streamGlobalWindowAggOnWindowTVF)
+ }
+
// select * from TableSourceTable1
// left join TableSourceTable2 on TableSourceTable1.b = TableSourceTable2.b
protected lazy val logicalLeftJoinOnContainedUniqueKeys: RelNode = relBuilder
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowPropertiesTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowPropertiesTest.scala
new file mode 100644
index 00000000000..3ef62d0ff43
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowPropertiesTest.scala
@@ -0,0 +1,184 @@
+/*
+ * 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.metadata
+
+import org.apache.flink.table.planner.plan.`trait`.RelWindowProperties
+import org.apache.flink.table.planner.plan.logical.WindowSpec
+import org.apache.flink.table.types.logical.LogicalType
+
+import org.apache.calcite.util.ImmutableBitSet
+import org.junit.Assert._
+import org.junit.Test
+
+/** Test for [[FlinkRelMdWindowProperties]]. */
+class FlinkRelMdWindowPropertiesTest extends FlinkRelMdHandlerTestBase {
+
+ @Test
+ def testGetWindowPropertiesOnTableScan(): Unit = {
+ // get window properties from statistics
+ Array(studentLogicalScan, studentFlinkLogicalScan,
studentStreamScan).foreach(
+ scan => assertEquals(null, mq.getRelWindowProperties(scan)))
+
+ Array(temporalTableLogicalScan, temporalTableFlinkLogicalScan,
temporalTableStreamScan).foreach(
+ scan =>
+ assertEquals(
+ createRelWindowProperties(0, 1, 2, tumbleWindowSpec,
timeAttributeType),
+ mq.getRelWindowProperties(scan)
+ ))
+ }
+
+ @Test
+ def testGetWindowPropertiesOnTableFunctionScan(): Unit = {
+ Array(windowTableFunctionScan,
lateralTableFunctionScan).zipWithIndex.foreach {
+ case (scan, idx) =>
+ assertEquals(
+ Array(createRelWindowProperties(3, 4, 5, tumbleWindowSpec,
proctimeType), null).apply(
+ idx),
+ mq.getRelWindowProperties(scan)
+ )
+ }
+ }
+
+ @Test
+ def testGetWindowPropertiesOnWindowTableFunction(): Unit = {
+ Array(streamTumbleWindowTVFRel, streamHopWindowTVFRel,
streamCumulateWindowTVFRel).zipWithIndex
+ .foreach {
+ case (tvf, idx) =>
+ assertEquals(
+ createRelWindowProperties(
+ 5,
+ 6,
+ 7,
+ Array(tumbleWindowSpec, hopWindowSpec,
cumulateWindowSpec).apply(idx),
+ timeAttributeType),
+ mq.getRelWindowProperties(tvf)
+ )
+ }
+ }
+
+ @Test
+ def testGetWindowPropertiesOnCalc(): Unit = {
+ Array(
+ keepWindowCalcOnTumbleWindowTVF,
+ keepWindowCalcOnHopWindowTVF,
+ keepWindowCalcOnCumulateWindowTVF).zipWithIndex
+ .foreach {
+ case (calc, idx) =>
+ assertEquals(
+ createRelWindowProperties(
+ 2,
+ 1,
+ -1,
+ Array(tumbleWindowSpec, hopWindowSpec,
cumulateWindowSpec).apply(idx),
+ proctimeType),
+ mq.getRelWindowProperties(calc)
+ )
+ }
+ Array(
+ discardWindowCalcOnTumbleWindowTVF,
+ discardWindowCalcOnHopWindowTVF,
+ discardWindowCalcOnCumulateWindowTVF)
+ .foreach(
+ calc =>
+ assertEquals(
+ null,
+ mq.getRelWindowProperties(calc)
+ ))
+ }
+
+ @Test
+ def testGetWindowPropertiesOnUnion(): Unit = {
+ Array(unionOnWindowTVFWithSameWindowSpec,
unionOnWindowTVFWithDifferentWindowSpec).zipWithIndex
+ .foreach {
+ case (union, idx) =>
+ assertEquals(
+ Array(createRelWindowProperties(5, 6, 7, tumbleWindowSpec,
timeAttributeType), null)
+ .apply(idx),
+ mq.getRelWindowProperties(union)
+ )
+ }
+ }
+
+ @Test
+ def testGetWindowPropertiesOnExchange(): Unit = {
+ Array(hashOnTumbleWindowTVF, hashOnHopWindowTVF,
hashOnCumulateWindowTVF).zipWithIndex.foreach {
+ case (exchange, idx) =>
+ assertEquals(
+ createRelWindowProperties(
+ 5,
+ 6,
+ 7,
+ Array(
+ tumbleWindowSpec,
+ hopWindowSpec,
+ cumulateWindowSpec
+ ).apply(idx),
+ timeAttributeType),
+ mq.getRelWindowProperties(exchange)
+ )
+ }
+ }
+
+ @Test
+ def testGetWindowPropertiesOnLogicalAggregate(): Unit = {
+ Array(logicalGroupWindowAggOnTumbleWindowTVF,
logicalGroupAggOnTumbleWindowTVF).zipWithIndex
+ .foreach {
+ case (groupAgg, idx) =>
+ assertEquals(
+ Array(
+ createRelWindowProperties(1, 2, -1, tumbleWindowSpec,
proctimeType),
+ null
+ ).apply(idx),
+ mq.getRelWindowProperties(groupAgg)
+ )
+ }
+ }
+
+ @Test
+ def testGetWindowPropertiesOnPhysicalAggregate(): Unit = {
+ Array(
+ streamWindowAggOnWindowTVF,
+ streamLocalWindowAggOnWindowTVF,
+ streamGlobalWindowAggOnWindowTVF).zipWithIndex.foreach {
+ case (groupAgg, idx) =>
+ assertEquals(
+ Array(
+ createRelWindowProperties(2, 3, -1, tumbleWindowSpec,
timeAttributeType),
+ createRelWindowProperties(5, 6, 7, tumbleWindowSpec,
timeAttributeType),
+ createRelWindowProperties(2, 3, -1, tumbleWindowSpec,
timeAttributeType)
+ ).apply(idx),
+ mq.getRelWindowProperties(groupAgg)
+ )
+ }
+ }
+
+ private def createRelWindowProperties(
+ start: Int,
+ end: Int,
+ time: Int,
+ spec: WindowSpec,
+ timeAttributeType: LogicalType): RelWindowProperties = {
+ RelWindowProperties.create(
+ ImmutableBitSet.of(start),
+ ImmutableBitSet.of(end),
+ if (time >= 0) ImmutableBitSet.of(time) else ImmutableBitSet.of(),
+ spec,
+ timeAttributeType)
+ }
+
+}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala
index 671c4b854c9..af245270093 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/MetadataTestUtil.scala
@@ -25,6 +25,8 @@ import
org.apache.flink.table.connector.source.{DynamicTableSource, ScanTableSou
import org.apache.flink.table.module.ModuleManager
import org.apache.flink.table.plan.stats.{ColumnStats, TableStats}
import org.apache.flink.table.planner.calcite.{FlinkContext, FlinkContextImpl,
FlinkTypeFactory}
+import org.apache.flink.table.planner.plan.`trait`.RelWindowProperties
+import org.apache.flink.table.planner.plan.logical.TumblingWindowSpec
import org.apache.flink.table.planner.plan.schema.{FlinkPreparingTableBase,
TableSourceTable}
import org.apache.flink.table.planner.plan.stats.FlinkStatistic
import
org.apache.flink.table.runtime.types.TypeInfoLogicalTypeConverter.fromTypeInfoToLogicalType
@@ -37,7 +39,9 @@ import org.apache.calcite.rel.`type`.{RelDataType,
RelDataTypeFactory}
import org.apache.calcite.schema.{Schema, SchemaPlus, Table}
import org.apache.calcite.schema.Schema.TableType
import org.apache.calcite.sql.{SqlCall, SqlNode}
+import org.apache.calcite.util.ImmutableBitSet
+import java.time.Duration
import java.util
import java.util.Collections
@@ -57,6 +61,7 @@ object MetadataTestUtil {
rootSchema.add("TemporalTable1", createTemporalTable1())
rootSchema.add("TemporalTable2", createTemporalTable2())
rootSchema.add("TemporalTable3", createTemporalTable3())
+ rootSchema.add("TemporalTable4", createTemporalTable4())
rootSchema.add("TableSourceTable1", createTableSourceTable1())
rootSchema.add("TableSourceTable2", createTableSourceTable2())
rootSchema.add("TableSourceTable3", createTableSourceTable3())
@@ -265,6 +270,37 @@ object MetadataTestUtil {
getMetadataTable(fieldNames, fieldTypes, new FlinkStatistic(tableStats))
}
+ private def createTemporalTable4(): Table = {
+ val fieldNames = Array("window_start", "window_end", "window_time", "a",
"b", "c")
+ val fieldTypes = Array[LogicalType](
+ new TimestampType(false, 3),
+ new TimestampType(false, 3),
+ new TimestampType(true, TimestampKind.ROWTIME, 3),
+ new IntType(),
+ new BigIntType(),
+ VarCharType.STRING_TYPE
+ )
+
+ val windowProperties = RelWindowProperties.create(
+ ImmutableBitSet.of(0),
+ ImmutableBitSet.of(1),
+ ImmutableBitSet.of(2),
+ new TumblingWindowSpec(Duration.ofMinutes(10L), null),
+ fieldTypes.apply(2))
+
+ val colStatsMap = Map[String, ColumnStats](
+ "a" -> new ColumnStats(3740000000L, 0L, 4d, 4, null, null),
+ "b" -> new ColumnStats(53252726L, 1474L, 8d, 8, 100000000L, -100000000L),
+ "c" -> new ColumnStats(null, 0L, 18.6, 64, null, null)
+ )
+
+ val tableStats = new TableStats(4000000000L, colStatsMap)
+ getMetadataTable(
+ fieldNames,
+ fieldTypes,
+ new FlinkStatistic(tableStats, relWindowProperties = windowProperties))
+ }
+
private val flinkContext = new FlinkContextImpl(
false,
TableConfig.getDefault,
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 9788791e781..9fe0b3e7fbf 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
@@ -1273,4 +1273,64 @@ class WindowJoinTest extends TableTestBase {
""".stripMargin
util.verifyRelPlan(sql)
}
+
+ @Test
+ def testJoinToMultiSink(): Unit = {
+ val sourceDdl =
+ """
+ |CREATE TABLE food_order (
+ | user_id STRING,
+ | order_id STRING,
+ | amount INT,
+ | event_time TIMESTAMP(3),
+ | WATERMARK FOR event_time AS event_time
+ |) WITH (
+ |'connector' = 'values')
+ |""".stripMargin
+ util.tableEnv.executeSql(sourceDdl)
+
+ val query =
+ """
+ |CREATE TEMPORARY VIEW food_view AS
+ |WITH food AS (
+ | SELECT user_id,
+ | window_start,
+ | window_end
+ | FROM TABLE(TUMBLE(TABLE food_order, DESCRIPTOR(event_time),
INTERVAL '1' MINUTES))
+ | GROUP BY
+ | user_id,
+ | window_start,
+ | window_end)
+ |SELECT food.window_start
+ | ,food.window_end
+ | ,food.user_id
+ | ,DATE_FORMAT(food.window_end + INTERVAL '7' HOUR, 'yyyyMMdd') AS
dt
+ | ,DATE_FORMAT(food.window_end + INTERVAL '7' HOUR, 'HH') AS `hour`
+ |FROM food
+ |LEFT JOIN food AS a ON food.user_id = a.user_id
+ |AND food.window_start = a.window_start
+ |AND food.window_end = a.window_end
+ |""".stripMargin
+
+ util.tableEnv.executeSql(query)
+
+ val sinkDdl =
+ """
+ |CREATE TABLE %s (
+ | window_start TIMESTAMP(3),
+ | window_end TIMESTAMP(3),
+ | user_id STRING,
+ | dt STRING,
+ | `hour` STRING
+ |) WITH (
+ | 'connector' = 'values')
+ |""".stripMargin
+ util.tableEnv.executeSql(sinkDdl.format("sink1"))
+ util.tableEnv.executeSql(sinkDdl.format("sink2"))
+
+ val statementSet = util.tableEnv.createStatementSet()
+ statementSet.addInsertSql("INSERT INTO sink1 SELECT * FROM food_view")
+ statementSet.addInsertSql("INSERT INTO sink2 SELECT * FROM food_view")
+ util.verifyRelPlan(statementSet)
+ }
}
