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..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,6 +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_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 7ea4950c23fe25..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 @@ -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(); }