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 c762aa5948c [FLINK-40205][table] PTF are failing in case of usage
multiple `PARTITION`/`ORDER BY`
c762aa5948c is described below
commit c762aa5948c277d130f68daaa5d0443f73da8f01
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Wed Jul 22 09:44:50 2026 +0200
[FLINK-40205][table] PTF are failing in case of usage multiple
`PARTITION`/`ORDER BY`
---
.../src/main/codegen/templates/Parser.jj | 3 +-
.../calcite/sql/SqlSetSemanticsTableOperator.java | 91 ++++++++++++++++++++++
.../stream/ProcessTableFunctionSemanticTests.java | 2 +
.../stream/ProcessTableFunctionTestPrograms.java | 82 +++++++++++++++++++
4 files changed, 177 insertions(+), 1 deletion(-)
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 cb9121e97d1..a031434e8ca 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 00000000000..385b866c724
--- /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<SqlNode> 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 39131ff4ed3..2357235b118 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,8 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
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_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 7ea4950c23f..327c3167fe5 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,86 @@ 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_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",
+ "`group` STRING",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR ts AS ts - INTERVAL
'0.001' SECOND")
+ .producedValues(
+ // 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, `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(
+ "`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.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("
+ + "r => TABLE v PARTITION BY (suite_name,
test_name) ORDER BY (window_time ASC, c DESC), i => 1)")
+ .build();
}
