From a374262c7a9be4e2857ba22a91ba370129356041 Mon Sep 17 00:00:00 2001 From: Sergey Nuyanzin Date: Tue, 21 Jul 2026 12:50:38 +0200 Subject: [PATCH 1/3] [FLINK-40205][table] PTF are failing in case of usage multiple `PARTITION`/`ORDER BY` --- .../src/main/codegen/templates/Parser.jj | 3 +- .../sql/SqlSetSemanticsTableOperator.java | 91 +++++++++++++ .../ProcessTableFunctionSemanticTests.java | 4 + .../ProcessTableFunctionTestPrograms.java | 125 ++++++++++++++++++ 4 files changed, 222 insertions(+), 1 deletion(-) create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlSetSemanticsTableOperator.java diff --git a/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj b/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj index cb9121e97d183a..a031434e8ca709 100644 --- a/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj +++ b/flink-table/flink-sql-parser/src/main/codegen/templates/Parser.jj @@ -1658,7 +1658,8 @@ SqlNode PartitionedQueryOrQueryOrExpr(ExprContext exprContext) : SqlNode e; } { - e = OrderedQueryOrExpr(exprContext) + // QueryOrExpr, not OrderedQueryOrExpr: ORDER BY is handled by PartitionedByAndOrderBy below. + e = QueryOrExpr(exprContext) e = PartitionedByAndOrderBy(e) { return e; } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlSetSemanticsTableOperator.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlSetSemanticsTableOperator.java new file mode 100644 index 00000000000000..385b866c724c1b --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/SqlSetSemanticsTableOperator.java @@ -0,0 +1,91 @@ +/* + * 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.calcite.sql; + +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.sql.parser.SqlParserPos; +import org.apache.calcite.sql.validate.SqlValidator; +import org.apache.calcite.sql.validate.SqlValidatorScope; +import org.checkerframework.checker.nullness.qual.Nullable; + +import java.util.List; + +import static java.util.Objects.requireNonNull; + +/** + * Copied from Calcite because of CALCITE-7660. Should be removed after upgrade to Calcite 1.43.0. + */ +public class SqlSetSemanticsTableOperator extends SqlInternalOperator { + + // ~ Constructors ----------------------------------------------------------- + + public SqlSetSemanticsTableOperator() { + super("SET_SEMANTICS_TABLE", SqlKind.SET_SEMANTICS_TABLE); + } + + @Override + public SqlCall createCall( + @Nullable SqlLiteral functionQualifier, + SqlParserPos pos, + @Nullable SqlNode... operands) { + assert operands.length == 3; + SqlNode partitionList = operands[1]; + SqlNode orderList = operands[2]; + assert (partitionList != null && !SqlNodeList.isEmptyList(partitionList)) + || (orderList != null && !SqlNodeList.isEmptyList(orderList)); + return super.createCall(functionQualifier, pos, operands); + } + + @Override + public void unparse(SqlWriter writer, SqlCall call, int leftPrec, int rightPrec) { + call.operand(0).unparse(writer, 0, 0); + + SqlNodeList partitionList = call.operand(1); + if (!partitionList.isEmpty()) { + writer.sep("PARTITION BY"); + // FLINK MODIFICATION BEGIN + final SqlWriter.Frame partitionFrame = + partitionList.size() == 1 + ? writer.startList("", "") + : writer.startList("(", ")"); + // FLINK MODIFICATION END + partitionList.unparse(writer, 0, 0); + writer.endList(partitionFrame); + } + SqlNodeList orderList = call.operand(2); + if (!orderList.isEmpty()) { + writer.sep("ORDER BY"); + // FLINK MODIFICATION BEGIN + final SqlWriter.Frame orderFrame = + orderList.size() == 1 ? writer.startList("", "") : writer.startList("(", ")"); + orderList.unparse(writer, 0, 0); + writer.endList(orderFrame); + // FLINK MODIFICATION END + } + } + + @Override + public RelDataType deriveType(SqlValidator validator, SqlValidatorScope scope, SqlCall call) { + final List operands = call.getOperandList(); + return requireNonNull(validator.deriveType(scope, operands.get(0))); + } + + @Override + public boolean argumentMustBeScalar(int ordinal) { + return false; + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java index 39131ff4ed3c97..078e91067f0f60 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java @@ -42,6 +42,10 @@ public List programs() { ProcessTableFunctionTestPrograms.PROCESS_SET_SEMANTIC_TABLE_TABLE_API_INLINE, ProcessTableFunctionTestPrograms.PROCESS_SET_SEMANTIC_TABLE_TABLE_API_INLINE_NAMED, ProcessTableFunctionTestPrograms.PROCESS_TYPED_SET_SEMANTIC_TABLE, + ProcessTableFunctionTestPrograms + .PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY, + ProcessTableFunctionTestPrograms + .PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY_AND_ORDER_BY, ProcessTableFunctionTestPrograms.PROCESS_TYPED_SET_SEMANTIC_TABLE_TABLE_API, ProcessTableFunctionTestPrograms.PROCESS_POJO_ARGS, ProcessTableFunctionTestPrograms.PROCESS_INTERVAL_DAY_ARGS, diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java index 7ea4950c23fe25..ae412c20de7162 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java @@ -1961,4 +1961,129 @@ public class ProcessTableFunctionTestPrograms { // Also in constructed types: ROW (table input) vs. STRUCTURED (expected). .runSql("INSERT INTO sink SELECT * FROM f(p => TABLE v, b => 42)") .build(); + + public static final TableTestProgram + PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY = + TableTestProgram.of( + "process-set-from-session-view-with-multi-partition-by", + "set semantic table partitioned by multiple columns, sourced from a view wrapping a SESSION window aggregate") + .setupTemporarySystemFunction("f", SetSemanticTableFunction.class) + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema( + "suite_name STRING", + "test_name STRING", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND") + .producedValues( + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(0)), + Row.of( + "suiteB", + "test2", + Instant.ofEpochMilli(1)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(2)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(3)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(4)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(5)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(6))) + .build()) + .setupSql( + "CREATE VIEW v AS " + + "SELECT suite_name, test_name, COUNT(*) AS c " + + "FROM SESSION(TABLE t PARTITION BY (suite_name, test_name), DESCRIPTOR(ts), INTERVAL '0.002' SECOND) " + + "GROUP BY suite_name, test_name, window_start, window_end") + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema( + "`suite_name` STRING", + "`test_name` STRING", + "`out` STRING") + .consumedValues( + "+I[suiteB, test2, {+I[suiteB, test2, 1], 1}]", + "+I[suiteA, test1, {+I[suiteA, test1, 6], 1}]") + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM f(r => TABLE v PARTITION BY (suite_name, test_name), i => 1)") + .build(); + + public static final TableTestProgram + PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY_AND_ORDER_BY = + TableTestProgram.of( + "process-order-by-multi-partition-key-and-order-by", + "set semantic table partitioned and ordered by multiple columns, sourced from a view wrapping a SESSION window aggregate") + .setupTemporarySystemFunction("f", SetSemanticTableFunction.class) + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema( + "suite_name STRING", + "test_name STRING", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND") + .producedValues( + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(0)), + Row.of( + "suiteB", + "test2", + Instant.ofEpochMilli(1)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(2)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(3)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(4)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(5)), + Row.of( + "suiteA", + "test1", + Instant.ofEpochMilli(6))) + .build()) + .setupSql( + "CREATE VIEW v AS " + + "SELECT suite_name, test_name, window_time, COUNT(*) AS c " + + "FROM TABLE(SESSION(TABLE t PARTITION BY (suite_name, test_name), DESCRIPTOR(ts), INTERVAL '0.002' SECOND)) " + + "GROUP BY suite_name, test_name, window_start, window_end, window_time") + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema( + "`suite_name` STRING", + "`test_name` STRING", + "`out` STRING") + .consumedValues( + "+I[suiteB, test2, {+I[suiteB, test2, 1970-01-01T00:00:00.002Z, 1], 1}]", + "+I[suiteA, test1, {+I[suiteA, test1, 1970-01-01T00:00:00.007Z, 6], 1}]") + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM f(" + + "r => TABLE v PARTITION BY (suite_name, test_name) ORDER BY (window_time ASC, c DESC), i => 1)") + .build(); } From e0c5586062276e6b9092c92b24c60ea76f075378 Mon Sep 17 00:00:00 2001 From: Sergey Nuyanzin Date: Tue, 21 Jul 2026 16:55:57 +0200 Subject: [PATCH 2/3] Address feedback --- .../ProcessTableFunctionSemanticTests.java | 6 +- .../ProcessTableFunctionTestPrograms.java | 204 +++++++----------- 2 files changed, 82 insertions(+), 128 deletions(-) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java index 078e91067f0f60..2357235b11864b 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java @@ -42,10 +42,8 @@ public List programs() { ProcessTableFunctionTestPrograms.PROCESS_SET_SEMANTIC_TABLE_TABLE_API_INLINE, ProcessTableFunctionTestPrograms.PROCESS_SET_SEMANTIC_TABLE_TABLE_API_INLINE_NAMED, ProcessTableFunctionTestPrograms.PROCESS_TYPED_SET_SEMANTIC_TABLE, - ProcessTableFunctionTestPrograms - .PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY, - ProcessTableFunctionTestPrograms - .PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY_AND_ORDER_BY, + ProcessTableFunctionTestPrograms.PROCESS_MULTI_PARTITION_BY, + ProcessTableFunctionTestPrograms.PROCESS_MULTI_PARTITION_BY_AND_ORDER_BY, ProcessTableFunctionTestPrograms.PROCESS_TYPED_SET_SEMANTIC_TABLE_TABLE_API, ProcessTableFunctionTestPrograms.PROCESS_POJO_ARGS, ProcessTableFunctionTestPrograms.PROCESS_INTERVAL_DAY_ARGS, diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java index ae412c20de7162..048d2449efa87d 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java @@ -1962,128 +1962,84 @@ public class ProcessTableFunctionTestPrograms { .runSql("INSERT INTO sink SELECT * FROM f(p => TABLE v, b => 42)") .build(); - public static final TableTestProgram - PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY = - TableTestProgram.of( - "process-set-from-session-view-with-multi-partition-by", - "set semantic table partitioned by multiple columns, sourced from a view wrapping a SESSION window aggregate") - .setupTemporarySystemFunction("f", SetSemanticTableFunction.class) - .setupTableSource( - SourceTestStep.newBuilder("t") - .addSchema( - "suite_name STRING", - "test_name STRING", - "ts TIMESTAMP_LTZ(3)", - "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND") - .producedValues( - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(0)), - Row.of( - "suiteB", - "test2", - Instant.ofEpochMilli(1)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(2)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(3)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(4)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(5)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(6))) - .build()) - .setupSql( - "CREATE VIEW v AS " - + "SELECT suite_name, test_name, COUNT(*) AS c " - + "FROM SESSION(TABLE t PARTITION BY (suite_name, test_name), DESCRIPTOR(ts), INTERVAL '0.002' SECOND) " - + "GROUP BY suite_name, test_name, window_start, window_end") - .setupTableSink( - SinkTestStep.newBuilder("sink") - .addSchema( - "`suite_name` STRING", - "`test_name` STRING", - "`out` STRING") - .consumedValues( - "+I[suiteB, test2, {+I[suiteB, test2, 1], 1}]", - "+I[suiteA, test1, {+I[suiteA, test1, 6], 1}]") - .build()) - .runSql( - "INSERT INTO sink SELECT * FROM f(r => TABLE v PARTITION BY (suite_name, test_name), i => 1)") - .build(); - - public static final TableTestProgram - PROCESS_SET_SEMANTIC_TABLE_FROM_SESSION_VIEW_WITH_MULTI_PARTITION_BY_AND_ORDER_BY = - TableTestProgram.of( - "process-order-by-multi-partition-key-and-order-by", - "set semantic table partitioned and ordered by multiple columns, sourced from a view wrapping a SESSION window aggregate") - .setupTemporarySystemFunction("f", SetSemanticTableFunction.class) - .setupTableSource( - SourceTestStep.newBuilder("t") - .addSchema( - "suite_name STRING", - "test_name STRING", - "ts TIMESTAMP_LTZ(3)", - "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND") - .producedValues( - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(0)), - Row.of( - "suiteB", - "test2", - Instant.ofEpochMilli(1)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(2)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(3)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(4)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(5)), - Row.of( - "suiteA", - "test1", - Instant.ofEpochMilli(6))) - .build()) - .setupSql( - "CREATE VIEW v AS " - + "SELECT suite_name, test_name, window_time, COUNT(*) AS c " - + "FROM TABLE(SESSION(TABLE t PARTITION BY (suite_name, test_name), DESCRIPTOR(ts), INTERVAL '0.002' SECOND)) " - + "GROUP BY suite_name, test_name, window_start, window_end, window_time") - .setupTableSink( - SinkTestStep.newBuilder("sink") - .addSchema( - "`suite_name` STRING", - "`test_name` STRING", - "`out` STRING") - .consumedValues( - "+I[suiteB, test2, {+I[suiteB, test2, 1970-01-01T00:00:00.002Z, 1], 1}]", - "+I[suiteA, test1, {+I[suiteA, test1, 1970-01-01T00:00:00.007Z, 6], 1}]") - .build()) - .runSql( - "INSERT INTO sink SELECT * FROM f(" - + "r => TABLE v PARTITION BY (suite_name, test_name) ORDER BY (window_time ASC, c DESC), i => 1)") - .build(); + public static final TableTestProgram PROCESS_MULTI_PARTITION_BY = + TableTestProgram.of( + "process-set-from-session-view-with-multi-partition-by", + "set semantic table partitioned by multiple columns, sourced from a view wrapping a SESSION window aggregate") + .setupTemporarySystemFunction("f", SetSemanticTableFunction.class) + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema( + "suite_name STRING", + "test_name STRING", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND") + .producedValues( + Row.of("suiteA", "test1", Instant.ofEpochMilli(0)), + Row.of("suiteB", "test2", Instant.ofEpochMilli(1)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(2)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(3)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(4)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(5)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(6))) + .build()) + .setupSql( + "CREATE VIEW v AS " + + "SELECT suite_name, test_name, COUNT(*) AS c " + + "FROM SESSION(TABLE t PARTITION BY (suite_name, test_name), DESCRIPTOR(ts), INTERVAL '0.002' SECOND) " + + "GROUP BY suite_name, test_name, window_start, window_end") + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema( + "`suite_name` STRING", + "`test_name` STRING", + "`out` STRING") + .consumedValues( + "+I[suiteB, test2, {+I[suiteB, test2, 1], 1}]", + "+I[suiteA, test1, {+I[suiteA, test1, 6], 1}]") + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM f(r => TABLE v PARTITION BY (suite_name, test_name), i => 1)") + .build(); + + public static final TableTestProgram PROCESS_MULTI_PARTITION_BY_AND_ORDER_BY = + TableTestProgram.of( + "process-order-by-multi-partition-key-and-order-by", + "set semantic table partitioned and ordered by multiple columns, sourced from a view wrapping a SESSION window aggregate") + .setupTemporarySystemFunction("f", SetSemanticTableFunction.class) + .setupTableSource( + SourceTestStep.newBuilder("t") + .addSchema( + "suite_name STRING", + "test_name STRING", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND") + .producedValues( + Row.of("suiteA", "test1", Instant.ofEpochMilli(0)), + Row.of("suiteB", "test2", Instant.ofEpochMilli(1)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(2)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(3)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(4)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(5)), + Row.of("suiteA", "test1", Instant.ofEpochMilli(6))) + .build()) + .setupSql( + "CREATE VIEW v AS " + + "SELECT suite_name, test_name, window_time, COUNT(*) AS c " + + "FROM SESSION(TABLE t PARTITION BY (suite_name, test_name), DESCRIPTOR(ts), INTERVAL '0.002' SECOND) " + + "GROUP BY suite_name, test_name, window_start, window_end, window_time") + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema( + "`suite_name` STRING", + "`test_name` STRING", + "`out` STRING") + .consumedValues( + "+I[suiteB, test2, {+I[suiteB, test2, 1970-01-01T00:00:00.002Z, 1], 1}]", + "+I[suiteA, test1, {+I[suiteA, test1, 1970-01-01T00:00:00.007Z, 6], 1}]") + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM f(" + + "r => TABLE v PARTITION BY (suite_name, test_name) ORDER BY (window_time ASC, c DESC), i => 1)") + .build(); } From da35a01b99971eb324b7123a982b8e0e00217da3 Mon Sep 17 00:00:00 2001 From: Sergey Nuyanzin Date: Tue, 21 Jul 2026 20:46:33 +0200 Subject: [PATCH 3/3] Address feedback --- .../ProcessTableFunctionTestPrograms.java | 21 ++++++++++--------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java index 048d2449efa87d..327c3167fe53dc 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java @@ -2012,22 +2012,22 @@ public class ProcessTableFunctionTestPrograms { .addSchema( "suite_name STRING", "test_name STRING", + "`group` STRING", "ts TIMESTAMP_LTZ(3)", "WATERMARK FOR ts AS ts - INTERVAL '0.001' SECOND") .producedValues( - Row.of("suiteA", "test1", Instant.ofEpochMilli(0)), - Row.of("suiteB", "test2", Instant.ofEpochMilli(1)), - Row.of("suiteA", "test1", Instant.ofEpochMilli(2)), - Row.of("suiteA", "test1", Instant.ofEpochMilli(3)), - Row.of("suiteA", "test1", Instant.ofEpochMilli(4)), - Row.of("suiteA", "test1", Instant.ofEpochMilli(5)), - Row.of("suiteA", "test1", Instant.ofEpochMilli(6))) + // group is only used to force two independent SESSION + // windows for suiteA/test1 whose window_time ties. + Row.of("suiteA", "test1", "x", Instant.ofEpochMilli(0)), + Row.of("suiteB", "test2", "x", Instant.ofEpochMilli(1)), + Row.of("suiteA", "test1", "x", Instant.ofEpochMilli(2)), + Row.of("suiteA", "test1", "y", Instant.ofEpochMilli(2))) .build()) .setupSql( "CREATE VIEW v AS " + "SELECT suite_name, test_name, window_time, COUNT(*) AS c " - + "FROM SESSION(TABLE t PARTITION BY (suite_name, test_name), DESCRIPTOR(ts), INTERVAL '0.002' SECOND) " - + "GROUP BY suite_name, test_name, window_start, window_end, window_time") + + "FROM SESSION(TABLE t PARTITION BY (suite_name, test_name, `group`), DESCRIPTOR(ts), INTERVAL '0.002' SECOND) " + + "GROUP BY suite_name, test_name, `group`, window_start, window_end, window_time") .setupTableSink( SinkTestStep.newBuilder("sink") .addSchema( @@ -2036,7 +2036,8 @@ public class ProcessTableFunctionTestPrograms { "`out` STRING") .consumedValues( "+I[suiteB, test2, {+I[suiteB, test2, 1970-01-01T00:00:00.002Z, 1], 1}]", - "+I[suiteA, test1, {+I[suiteA, test1, 1970-01-01T00:00:00.007Z, 6], 1}]") + "+I[suiteA, test1, {+I[suiteA, test1, 1970-01-01T00:00:00.003Z, 2], 1}]", + "+I[suiteA, test1, {+I[suiteA, test1, 1970-01-01T00:00:00.003Z, 1], 1}]") .build()) .runSql( "INSERT INTO sink SELECT * FROM f("