This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 33029143730 [FLINK-40449][table] Reject non-temporal streaming sort
during planning
33029143730 is described below
commit 330291437309c829378006f5109c947d1bf4216e
Author: Martijn Visser <[email protected]>
AuthorDate: Fri Aug 21 11:14:36 2026 +0200
[FLINK-40449][table] Reject non-temporal streaming sort during planning
A streaming sort on a non-time attribute is unsupported, but the check
lived in StreamExecSort.translateToPlanInternal, so it only fired at job
(re)submission, not during optimization: COMPILE PLAN succeeded and the
query failed later.
Reject it in StreamPhysicalSortRule.convert() during optimization, with
two targeted messages (non-time-attribute key, and descending time
attribute) built in SortUtil. A sort key that is an expression is
projected into a generated column (EXPR$n from SQL, $fn from the Table
API); the message describes it as "the sort key expression" instead of
leaking that name.
The StreamExecSort check is kept as a backstop for compiled plans loaded
via loadPlan, which bypasses optimization. It uses
LogicalTypeChecks.isTimeAttribute so a TIMESTAMP_LTZ rowtime is
recognized after deserialization, and guards against a plan without sort
keys.
Existing tests whose ORDER BY became invalid in streaming are adapted:
those that sorted only incidentally keep asserting their feature
(TableSinkTest.testDistribution, GroupingSetsTest.testFromBlogspot), the
rest assert the rejection. SortValidationTest covers the rule, the
expression cases, and compile/load round trips for the backstop.
Generated-by: Claude Code (Claude Opus 4.8, Claude Fable 5.1)
---
.../plan/nodes/exec/stream/StreamExecSort.java | 22 ++-
.../physical/stream/StreamPhysicalSortRule.scala | 22 +++
.../flink/table/planner/plan/utils/SortUtil.scala | 28 ++-
.../plan/stream/sql/SortValidationTest.java | 196 +++++++++++++++++++++
.../planner/plan/common/PartialInsertTest.xml | 23 ---
.../table/planner/plan/stream/sql/SortTest.xml | 146 ++-------------
.../planner/plan/stream/sql/TableSinkTest.xml | 11 +-
.../plan/stream/sql/agg/GroupingSetsTest.xml | 79 ++-------
.../planner/plan/common/PartialInsertTest.scala | 13 +-
.../table/planner/plan/stream/sql/SortTest.scala | 35 +---
.../planner/plan/stream/sql/TableSinkTest.scala | 2 +-
.../plan/stream/sql/agg/GroupingSetsTest.scala | 17 +-
.../planner/runtime/stream/sql/SortITCase.scala | 7 +-
13 files changed, 321 insertions(+), 280 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSort.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSort.java
index fb9f38ed967..e3a5dd99af9 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSort.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSort.java
@@ -33,11 +33,14 @@ import
org.apache.flink.table.planner.plan.nodes.exec.ExecNodeMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
import org.apache.flink.table.planner.plan.nodes.exec.spec.SortSpec;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
+import org.apache.flink.table.planner.plan.utils.SortUtil;
import org.apache.flink.table.planner.utils.InternalConfigOptions;
import org.apache.flink.table.runtime.generated.GeneratedRecordComparator;
import org.apache.flink.table.runtime.operators.sort.StreamSortOperator;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -100,12 +103,25 @@ public class StreamExecSort extends ExecNodeBase<RowData>
implements StreamExecN
@Override
protected Transformation<RowData> translateToPlanInternal(
PlannerBase planner, ExecNodeConfig config) {
+ ExecEdge inputEdge = getInputEdges().get(0);
+ RowType inputType = (RowType) inputEdge.getOutputType();
if
(!config.get(InternalConfigOptions.TABLE_EXEC_NON_TEMPORAL_SORT_ENABLED)) {
- throw new TableException("Sort on a non-time-attribute field is
not supported.");
+ // Backstop for compiled plans loaded without passing through
StreamPhysicalSortRule.
+ if (sortSpec.getFieldSize() == 0) {
+ throw new TableException(
+ "Compiled plan contains a streaming sort without sort
keys.");
+ }
+ int firstSortField = sortSpec.getFieldIndices()[0];
+ String column = inputType.getFieldNames().get(firstSortField);
+ LogicalType type = inputType.getTypeAt(firstSortField);
+ if (LogicalTypeChecks.isTimeAttribute(type)
+ && !sortSpec.getFieldSpecs()[0].getIsAscendingOrder()) {
+ throw new TableException(
+
SortUtil.sortKeyTimeAttributeMustBeAscendingMessage(column));
+ }
+ throw new
TableException(SortUtil.sortKeyNotTimeAttributeMessage(column, type));
}
- ExecEdge inputEdge = getInputEdges().get(0);
- RowType inputType = (RowType) inputEdge.getOutputType();
// sort code gen
GeneratedRecordComparator rowComparator =
ComparatorCodeGenerator.gen(
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalSortRule.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalSortRule.scala
index 63958f7235c..85ef8544732 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalSortRule.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalSortRule.scala
@@ -17,10 +17,14 @@
*/
package org.apache.flink.table.planner.plan.rules.physical.stream
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalSort
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalSort
+import org.apache.flink.table.planner.plan.utils.SortUtil
+import org.apache.flink.table.planner.utils.{InternalConfigOptions,
ShortcutUtils}
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
import org.apache.calcite.rel.RelNode
@@ -41,6 +45,24 @@ class StreamPhysicalSortRule(config: Config) extends
ConverterRule(config) {
override def convert(rel: RelNode): RelNode = {
val sort: FlinkLogicalSort = rel.asInstanceOf[FlinkLogicalSort]
+ // Reject a non-time-attribute streaming sort here instead of deferring to
StreamExecSort.
+ if (
+ !ShortcutUtils
+ .unwrapTableConfig(sort)
+ .get(InternalConfigOptions.TABLE_EXEC_NON_TEMPORAL_SORT_ENABLED)
+ ) {
+ val field = SortUtil.getFirstSortField(sort.getCollation,
sort.getInput.getRowType)
+ // A time-attribute first key only reaches here when descending;
ascending goes to
+ // StreamPhysicalTemporalSort via matches().
+ val message = if (FlinkTypeFactory.isTimeIndicatorType(field.getType)) {
+ SortUtil.sortKeyTimeAttributeMustBeAscendingMessage(field.getName)
+ } else {
+ SortUtil.sortKeyNotTimeAttributeMessage(
+ field.getName,
+ FlinkTypeFactory.toLogicalType(field.getType))
+ }
+ throw new TableException(message)
+ }
val input = sort.getInput(0)
val requiredTraitSet = input.getTraitSet
.replace(FlinkRelDistribution.SINGLETON)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/SortUtil.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/SortUtil.scala
index da6c1e9fbaf..dc38d99c96c 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/SortUtil.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/SortUtil.scala
@@ -23,7 +23,7 @@ import org.apache.flink.table.api.TableException
import org.apache.flink.table.planner.calcite.FlinkPlannerImpl
import org.apache.flink.table.planner.codegen.sort.SortCodeGenerator
import org.apache.flink.table.planner.plan.nodes.exec.spec.SortSpec
-import org.apache.flink.table.types.logical.RowType
+import org.apache.flink.table.types.logical.{LogicalType, RowType}
import org.apache.calcite.rel.`type`._
import org.apache.calcite.rel.{RelCollation, RelFieldCollation}
@@ -74,6 +74,32 @@ object SortUtil {
rowType.getFieldList.get(idx)
}
+ /**
+ * Name pattern of the column a sort expression is projected into: `EXPR$n`
by the SQL converter,
+ * `$fn` by the Table API's RelBuilder. Such names are meaningless to users.
+ */
+ private val GENERATED_SORT_KEY_NAME = "^(EXPR\\$|\\$f)\\d+$".r
+
+ private def describeSortKey(column: String): String =
+ if (GENERATED_SORT_KEY_NAME.findFirstIn(column).isDefined) "the sort key
expression"
+ else s"'$column'"
+
+ /** Error message when the primary streaming sort key is not a time
attribute. */
+ def sortKeyNotTimeAttributeMessage(column: String, tpe: LogicalType): String
=
+ s"Streaming ORDER BY requires the primary sort key to be a time attribute
in ascending " +
+ s"order, but ${describeSortKey(column)} is ${tpe.asSummaryString}. A
time attribute is an " +
+ s"event-time column (a TIMESTAMP or TIMESTAMP_LTZ column with a
WATERMARK) or a " +
+ s"processing-time column. Otherwise use LIMIT for Top-N, sort within a
window, or run in " +
+ s"batch mode."
+
+ /**
+ * Error message when the primary streaming sort key is a time attribute but
sorted descending.
+ */
+ def sortKeyTimeAttributeMustBeAscendingMessage(column: String): String =
+ s"Streaming ORDER BY on time attribute '$column' must be sorted in
ascending order; " +
+ s"descending order is not supported. Otherwise use LIMIT for Top-N, sort
within a window, " +
+ s"or run in batch mode."
+
/** Returns the default null direction if not specified. */
def getNullDefaultOrders(ascendings: Array[Boolean]): Array[Boolean] = {
ascendings.map(asc => FlinkPlannerImpl.defaultNullCollation.last(!asc))
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SortValidationTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SortValidationTest.java
new file mode 100644
index 00000000000..1489aff0858
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SortValidationTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.stream.sql;
+
+import org.apache.flink.table.api.CompiledPlan;
+import org.apache.flink.table.api.PlanReference;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.internal.CompiledPlanUtils;
+import org.apache.flink.table.planner.utils.InternalConfigOptions;
+import org.apache.flink.table.planner.utils.JavaStreamTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Plan-time validation tests for streaming sort. A non-time-attribute sort is
unsupported and is
+ * rejected by {@code StreamPhysicalSortRule} during optimization, so it
surfaces at {@code COMPILE
+ * PLAN} / planning time. {@code StreamExecSort} keeps the same check as a
backstop for compiled
+ * plans loaded via {@code loadPlan}, which bypasses optimization. Valid
temporal sorts are covered
+ * by {@code SortTest}.
+ */
+class SortValidationTest extends TableTestBase {
+
+ private static final String MESSAGE =
+ "requires the primary sort key to be a time attribute in ascending
order";
+ private static final String MESSAGE_DESC =
+ "must be sorted in ascending order; descending order is not
supported";
+
+ private final JavaStreamTableTestUtil util = javaStreamTestUtil();
+
+ @BeforeEach
+ void setup() {
+ util.addTable(
+ "CREATE TABLE MyTable (\n"
+ + " a INT,\n"
+ + " b STRING,\n"
+ + " c BIGINT,\n"
+ + " proctime AS PROCTIME(),\n"
+ + " rowtime TIMESTAMP(3),\n"
+ + " WATERMARK FOR rowtime AS rowtime\n"
+ + ") WITH ('connector' = 'values')");
+ }
+
+ static Stream<Arguments> nonTemporalSorts() {
+ return Stream.of(
+ // primary sort key is not a time attribute -> message A
+ Arguments.of("SELECT a FROM MyTable ORDER BY c", MESSAGE),
+ Arguments.of("SELECT a FROM MyTable ORDER BY c, proctime",
MESSAGE),
+ Arguments.of("SELECT a FROM MyTable ORDER BY c, rowtime",
MESSAGE),
+ Arguments.of("SELECT a FROM MyTable ORDER BY c, proctime
DESC", MESSAGE),
+ Arguments.of("SELECT a FROM MyTable ORDER BY c, rowtime DESC",
MESSAGE),
+ // primary sort key is a time attribute but sorted descending
-> message B
+ Arguments.of("SELECT a FROM MyTable ORDER BY proctime DESC,
c", MESSAGE_DESC),
+ Arguments.of("SELECT a FROM MyTable ORDER BY rowtime DESC, c",
MESSAGE_DESC),
+ // ordinals and aliases are expanded by the validator and hit
the same rule
+ Arguments.of("SELECT a FROM MyTable ORDER BY 1", MESSAGE),
+ Arguments.of("SELECT c AS x FROM MyTable ORDER BY x", MESSAGE),
+ Arguments.of("SELECT rowtime AS t, a FROM MyTable ORDER BY t
DESC", MESSAGE_DESC),
+ // an expression is projected into a generated column below
the sort
+ Arguments.of("SELECT a FROM MyTable ORDER BY c + 1", MESSAGE));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("nonTemporalSorts")
+ void testNonTemporalSortRejected(String query, String expectedMessage) {
+ assertThatThrownBy(() -> util.verifyExecPlan(query))
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining(expectedMessage);
+ }
+
+ @Test
+ void testMessageNamesColumnAndType() {
+ assertThatThrownBy(() -> util.verifyExecPlan("SELECT a FROM MyTable
ORDER BY c"))
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining(MESSAGE)
+ .hasMessageContaining("'c' is BIGINT");
+ }
+
+ @Test
+ void testMessageDescribesSqlExpressionKey() {
+ // The SQL converter projects the ORDER BY expression as EXPR$n; the
message must not
+ // leak that alias. The Volcano wrapper prepends the plan (which does
contain the
+ // alias), so assert on the rule's own exception.
+ assertThatThrownBy(() -> util.verifyExecPlan("SELECT a FROM MyTable
ORDER BY c + 1"))
+ .rootCause()
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining("the sort key expression is BIGINT")
+ .hasMessageNotContaining("EXPR$");
+ }
+
+ @Test
+ void testMessageDescribesTableApiExpressionKey() {
+ // The Table API projects the ORDER BY expression as $fn instead of
EXPR$n.
+ Table table =
util.getTableEnv().from("MyTable").orderBy($("c").plus(1));
+ assertThatThrownBy(() -> util.verifyExecPlan(table))
+ .rootCause()
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining("the sort key expression is BIGINT")
+ .hasMessageNotContaining("$f");
+ }
+
+ @Test
+ void testNonTemporalSortAllowedWhenEnabled() {
+ util.getTableEnv()
+ .getConfig()
+
.set(InternalConfigOptions.TABLE_EXEC_NON_TEMPORAL_SORT_ENABLED, true);
+ // With the internal flag enabled, the plan-time check is skipped and
the sort is accepted.
+ assertThatCode(() -> util.getTableEnv().explainSql("SELECT a FROM
MyTable ORDER BY c"))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testCompilePlanRejectsNonTemporalSort() {
+ util.addTable("CREATE TABLE MySink (a INT) WITH ('connector' =
'values')");
+ assertThatThrownBy(
+ () ->
+ util.getTableEnv()
+ .compilePlanSql(
+ "INSERT INTO MySink SELECT a
FROM MyTable ORDER BY c"))
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining(MESSAGE);
+ }
+
+ @Test
+ void testLoadedPlanRejectsNonTemporalSort() {
+ String json =
+ compileWithNonTemporalSortEnabled(
+ "INSERT INTO MySink SELECT a FROM MyTable ORDER BY c");
+ assertTranslationRejected(json, MESSAGE);
+ }
+
+ @Test
+ void testLoadedPlanRejectsDescendingTimeAttributeSort() {
+ String json =
+ compileWithNonTemporalSortEnabled(
+ "INSERT INTO MySink SELECT a FROM MyTable ORDER BY
rowtime DESC");
+ // The backstop can only tell the two cases apart if the time
attribute survives the
+ // round trip; fail on that premise rather than silently drifting to
the other message.
+ assertThat(json).containsPattern("\"kind\"\\s*:\\s*\"ROWTIME\"");
+ assertTranslationRejected(json, MESSAGE_DESC);
+ }
+
+ /**
+ * Compiles the insert with the internal flag enabled so the sort passes
{@code
+ * StreamPhysicalSortRule}. The flag is not persisted in the plan.
+ */
+ private String compileWithNonTemporalSortEnabled(String insert) {
+ util.addTable("CREATE TABLE MySink (a INT) WITH ('connector' =
'values')");
+ util.getTableEnv()
+ .getConfig()
+
.set(InternalConfigOptions.TABLE_EXEC_NON_TEMPORAL_SORT_ENABLED, true);
+ return util.getTableEnv().compilePlanSql(insert).asJsonString();
+ }
+
+ /**
+ * Loads the plan with the flag disabled and translates it, which reaches
the backstop in {@code
+ * StreamExecSort} without passing through {@code StreamPhysicalSortRule}.
+ */
+ private void assertTranslationRejected(String json, String
expectedMessage) {
+ TableEnvironment tEnv = util.getTableEnv();
+
tEnv.getConfig().set(InternalConfigOptions.TABLE_EXEC_NON_TEMPORAL_SORT_ENABLED,
false);
+ CompiledPlan loaded =
tEnv.loadPlan(PlanReference.fromJsonString(json));
+ assertThatThrownBy(() -> CompiledPlanUtils.toTransformations(tEnv,
loaded))
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining(expectedMessage);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/common/PartialInsertTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/common/PartialInsertTest.xml
index 9f23ebd1a81..d691886784f 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/common/PartialInsertTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/common/PartialInsertTest.xml
@@ -312,29 +312,6 @@
Sink(table=[default_catalog.default_database.partitioned_sink], targetColumns=[[
+- Exchange(distribution=[hash[a, b, c, d, e]])
+- LocalHashAggregate(groupBy=[a, b, c, d,
e], select=[a, b, c, d, e])
+-
TableSourceScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, d, e])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testPartialInsertWithOrderBy[isBatch: false]">
- <Resource name="sql">
- <![CDATA[INSERT INTO partitioned_sink (e,a,g,f,c,d) SELECT
e,a,456,123,c,d FROM MyTable ORDER BY a,e,c,d]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalSink(table=[default_catalog.default_database.partitioned_sink],
targetColumns=[[4],[0],[6],[5],[2],[3]], fields=[a, c, d, e, f, g])
-+- LogicalProject(a=[$0], c=[$1], d=[$2], e=[$3], f=[CAST($4):BIGINT],
g=[CAST($5):INTEGER])
- +- LogicalSort(sort0=[$0], sort1=[$3], sort2=[$1], sort3=[$2],
dir0=[ASC-nulls-first], dir1=[ASC-nulls-first], dir2=[ASC-nulls-first],
dir3=[ASC-nulls-first])
- +- LogicalProject(a=[$0], c=[$2], d=[$3], e=[$4], EXPR$4=[123],
EXPR$5=[456])
- +- LogicalTableScan(table=[[default_catalog, default_database,
MyTable]])
-]]>
- </Resource>
- <Resource name="optimized rel plan">
- <![CDATA[
-Sink(table=[default_catalog.default_database.partitioned_sink],
targetColumns=[[4],[0],[6],[5],[2],[3]], fields=[a, c, d, e, f, g])
-+- Calc(select=[a, c, d, e, CAST(123 AS BIGINT) AS f, CAST(456 AS INTEGER) AS
g])
- +- Sort(orderBy=[a ASC, e ASC, c ASC, d ASC])
- +- Exchange(distribution=[single])
- +- TableSourceScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c, d, e])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SortTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SortTest.xml
index 22a2cfc58fc..c92a3bcfa7b 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SortTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SortTest.xml
@@ -16,85 +16,43 @@ See the License for the specific language governing
permissions and
limitations under the License.
-->
<Root>
- <TestCase name="testSortProcessingTime">
+ <TestCase name="testSortOnRowTimeAlias">
<Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY proctime, c]]>
+ <![CDATA[SELECT a, rowtime AS t FROM MyTable ORDER BY t, c]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalProject(a=[$0])
+LogicalProject(a=[$0], t=[$1])
+- LogicalSort(sort0=[$1], sort1=[$2], dir0=[ASC-nulls-first],
dir1=[ASC-nulls-first])
- +- LogicalProject(a=[$0], proctime=[$3], c=[$2])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Calc(select=[a])
-+- TemporalSort(orderBy=[proctime ASC, c ASC])
- +- Exchange(distribution=[single])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testSortProcessingTimeDesc">
- <Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY proctime desc, c]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(a=[$0])
-+- LogicalSort(sort0=[$1], sort1=[$2], dir0=[DESC-nulls-last],
dir1=[ASC-nulls-first])
- +- LogicalProject(a=[$0], proctime=[$3], c=[$2])
+ +- LogicalProject(a=[$0], t=[$4], c=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[a])
-+- Sort(orderBy=[proctime DESC, c ASC])
+Calc(select=[a, rowtime AS t])
++- TemporalSort(orderBy=[rowtime ASC, c ASC])
+- Exchange(distribution=[single])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
]]>
</Resource>
</TestCase>
- <TestCase name="testSortProcessingTimeSecond">
+ <TestCase name="testSortProcessingTime">
<Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY c, proctime]]>
+ <![CDATA[SELECT a FROM MyTable ORDER BY proctime, c]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(a=[$0])
+- LogicalSort(sort0=[$1], sort1=[$2], dir0=[ASC-nulls-first],
dir1=[ASC-nulls-first])
- +- LogicalProject(a=[$0], c=[$2], proctime=[$3])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Calc(select=[a])
-+- Sort(orderBy=[c ASC, proctime ASC])
- +- Exchange(distribution=[single])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testSortProcessingTimeSecondDesc">
- <Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY c, proctime desc]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(a=[$0])
-+- LogicalSort(sort0=[$1], sort1=[$2], dir0=[ASC-nulls-first],
dir1=[DESC-nulls-last])
- +- LogicalProject(a=[$0], c=[$2], proctime=[$3])
+ +- LogicalProject(a=[$0], proctime=[$3], c=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[a])
-+- Sort(orderBy=[c ASC, proctime DESC])
++- TemporalSort(orderBy=[proctime ASC, c ASC])
+- Exchange(distribution=[single])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
]]>
@@ -118,90 +76,6 @@ Calc(select=[a])
+- TemporalSort(orderBy=[rowtime ASC, c ASC])
+- Exchange(distribution=[single])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testSortRowTimeDesc">
- <Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY rowtime desc, c]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(a=[$0])
-+- LogicalSort(sort0=[$1], sort1=[$2], dir0=[DESC-nulls-last],
dir1=[ASC-nulls-first])
- +- LogicalProject(a=[$0], rowtime=[$4], c=[$2])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Calc(select=[a])
-+- Sort(orderBy=[rowtime DESC, c ASC])
- +- Exchange(distribution=[single])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testSortRowTimeSecond">
- <Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY c, rowtime]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(a=[$0])
-+- LogicalSort(sort0=[$1], sort1=[$2], dir0=[ASC-nulls-first],
dir1=[ASC-nulls-first])
- +- LogicalProject(a=[$0], c=[$2], rowtime=[$4])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Calc(select=[a])
-+- Sort(orderBy=[c ASC, rowtime ASC])
- +- Exchange(distribution=[single])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testSortRowTimeSecondDesc">
- <Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY c, rowtime desc]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(a=[$0])
-+- LogicalSort(sort0=[$1], sort1=[$2], dir0=[ASC-nulls-first],
dir1=[DESC-nulls-last])
- +- LogicalProject(a=[$0], c=[$2], rowtime=[$4])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Calc(select=[a])
-+- Sort(orderBy=[c ASC, rowtime DESC])
- +- Exchange(distribution=[single])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testSortWithoutTime">
- <Resource name="sql">
- <![CDATA[SELECT a FROM MyTable ORDER BY c]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(a=[$0])
-+- LogicalSort(sort0=[$1], dir0=[ASC-nulls-first])
- +- LogicalProject(a=[$0], c=[$2])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Calc(select=[a])
-+- Sort(orderBy=[c ASC])
- +- Exchange(distribution=[single])
- +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c, proctime, rowtime])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml
index 486e4d045ec..493a6f1f932 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.xml
@@ -489,18 +489,15 @@ Sink(table=[default_catalog.default_database.sink],
fields=[id, city_name, ts, r
<Resource name="ast">
<![CDATA[
LogicalSink(table=[default_catalog.default_database.sink], fields=[a, b])
-+- LogicalSort(sort0=[$0], dir0=[ASC-nulls-first])
- +- LogicalProject(a=[$0], b=[$1])
- +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
++- LogicalProject(a=[$0], b=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
Sink(table=[default_catalog.default_database.sink], fields=[a, b])
-+- Sort(orderBy=[a ASC])
- +- Exchange(distribution=[single])
- +- Calc(select=[a, b])
- +- DataStreamScan(table=[[default_catalog, default_database,
MyTable]], fields=[a, b, c])
++- Calc(select=[a, b])
+ +- DataStreamScan(table=[[default_catalog, default_database, MyTable]],
fields=[a, b, c])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.xml
index a56cec43528..54d07e3fb38 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.xml
@@ -402,28 +402,24 @@ SELECT deptno, job, empno, ename, SUM(sal) sumsal,
END gr_text
from scott_emp
GROUP BY ROLLUP(deptno, job, (empno,ename))
- ORDER BY deptno, job, empno
]]>
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalSort(sort0=[$0], sort1=[$1], sort2=[$2], dir0=[ASC-nulls-first],
dir1=[ASC-nulls-first], dir2=[ASC-nulls-first])
-+- LogicalProject(deptno=[$0], job=[$1], empno=[$2], ename=[$3], sumsal=[$4],
gr_text=[CASE(=($5, 0), _UTF-16LE'grouped by
deptno,job,empno,ename':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", =($5, 1),
_UTF-16LE'grouped by deptno,job':VARCHAR(2147483647) CHARACTER SET "UTF-16LE",
=($5, 3), _UTF-16LE'grouped by deptno':VARCHAR(2147483647) CHARACTER SET
"UTF-16LE", =($5, 7), _UTF-16LE'grouped by ()':VARCHAR(2147483647) CHARACTER
SET "UTF-16LE", null:VARCHAR(2147483647) CHARACTER SET "UTF- [...]
- +- LogicalAggregate(group=[{0, 1, 2, 3}], groups=[[{0, 1, 2, 3}, {0, 1},
{0}, {}]], sumsal=[SUM($4)], agg#1=[GROUPING_ID($0, $1, $2)])
- +- LogicalProject(deptno=[$7], job=[$2], empno=[$0], ename=[$1],
sal=[$5])
- +- LogicalTableScan(table=[[default_catalog, default_database,
scott_emp]])
+LogicalProject(deptno=[$0], job=[$1], empno=[$2], ename=[$3], sumsal=[$4],
gr_text=[CASE(=($5, 0), _UTF-16LE'grouped by
deptno,job,empno,ename':VARCHAR(2147483647) CHARACTER SET "UTF-16LE", =($5, 1),
_UTF-16LE'grouped by deptno,job':VARCHAR(2147483647) CHARACTER SET "UTF-16LE",
=($5, 3), _UTF-16LE'grouped by deptno':VARCHAR(2147483647) CHARACTER SET
"UTF-16LE", =($5, 7), _UTF-16LE'grouped by ()':VARCHAR(2147483647) CHARACTER
SET "UTF-16LE", null:VARCHAR(2147483647) CHARACTER SET "UTF-16LE")])
++- LogicalAggregate(group=[{0, 1, 2, 3}], groups=[[{0, 1, 2, 3}, {0, 1}, {0},
{}]], sumsal=[SUM($4)], agg#1=[GROUPING_ID($0, $1, $2)])
+ +- LogicalProject(deptno=[$7], job=[$2], empno=[$0], ename=[$1], sal=[$5])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
scott_emp]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Sort(orderBy=[deptno ASC, job ASC, empno ASC])
-+- Exchange(distribution=[single])
- +- Calc(select=[deptno, job, empno, ename, sumsal, CASE((CASE(($e = 0), 0,
($e = 3), 1, ($e = 7), 3, 7) = 0), 'grouped by deptno,job,empno,ename',
(CASE(($e = 0), 0, ($e = 3), 1, ($e = 7), 3, 7) = 1), 'grouped by deptno,job',
(CASE(($e = 0), 0, ($e = 3), 1, ($e = 7), 3, 7) = 3), 'grouped by deptno',
(CASE(($e = 0), 0, ($e = 3), 1, ($e = 7), 3, 7) = 7), 'grouped by ()',
null:VARCHAR(2147483647)) AS gr_text])
- +- GroupAggregate(groupBy=[deptno, job, empno, ename, $e],
select=[deptno, job, empno, ename, $e, SUM(sal) AS sumsal])
- +- Exchange(distribution=[hash[deptno, job, empno, ename, $e]])
- +- Expand(projects=[{deptno, job, empno, ename, sal, 0 AS $e},
{deptno, job, null AS empno, null AS ename, sal, 3 AS $e}, {deptno, null AS
job, null AS empno, null AS ename, sal, 7 AS $e}, {null AS deptno, null AS job,
null AS empno, null AS ename, sal, 15 AS $e}])
- +- Calc(select=[deptno, job, empno, ename, sal])
- +- TableSourceScan(table=[[default_catalog,
default_database, scott_emp]], fields=[empno, ename, job, mgr, hiredate, sal,
comm, deptno])
+Calc(select=[deptno, job, empno, ename, sumsal, CASE((CASE(($e = 0), 0, ($e =
3), 1, ($e = 7), 3, 7) = 0), 'grouped by deptno,job,empno,ename', (CASE(($e =
0), 0, ($e = 3), 1, ($e = 7), 3, 7) = 1), 'grouped by deptno,job', (CASE(($e =
0), 0, ($e = 3), 1, ($e = 7), 3, 7) = 3), 'grouped by deptno', (CASE(($e = 0),
0, ($e = 3), 1, ($e = 7), 3, 7) = 7), 'grouped by ()',
null:VARCHAR(2147483647)) AS gr_text])
++- GroupAggregate(groupBy=[deptno, job, empno, ename, $e], select=[deptno,
job, empno, ename, $e, SUM(sal) AS sumsal])
+ +- Exchange(distribution=[hash[deptno, job, empno, ename, $e]])
+ +- Expand(projects=[{deptno, job, empno, ename, sal, 0 AS $e}, {deptno,
job, null AS empno, null AS ename, sal, 3 AS $e}, {deptno, null AS job, null AS
empno, null AS ename, sal, 7 AS $e}, {null AS deptno, null AS job, null AS
empno, null AS ename, sal, 15 AS $e}])
+ +- Calc(select=[deptno, job, empno, ename, sal])
+ +- TableSourceScan(table=[[default_catalog, default_database,
scott_emp]], fields=[empno, ename, job, mgr, hiredate, sal, comm, deptno])
]]>
</Resource>
</TestCase>
@@ -525,36 +521,6 @@ Calc(select=[deptno, gender, CASE(SEARCH($e, Sarg[0, 1]),
0, 1) AS gd, CASE(($e
+- Expand(projects=[{deptno, gender, 0 AS $e}, {deptno, null AS gender,
1 AS $e}, {null AS deptno, gender, 2 AS $e}, {null AS deptno, null AS gender, 3
AS $e}])
+- Calc(select=[deptno, gender])
+- TableSourceScan(table=[[default_catalog, default_database,
emp]], fields=[ename, deptno, gender])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testGroupingInOrderByClause">
- <Resource name="sql">
- <![CDATA[
-SELECT COUNT(*) AS c FROM emp GROUP BY ROLLUP(deptno) ORDER BY
GROUPING(deptno), c
- ]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalProject(c=[$0])
-+- LogicalSort(sort0=[$1], sort1=[$0], dir0=[ASC-nulls-first],
dir1=[ASC-nulls-first])
- +- LogicalProject(c=[$1], EXPR$1=[$2])
- +- LogicalAggregate(group=[{0}], groups=[[{0}, {}]], c=[COUNT()],
agg#1=[GROUPING($0)])
- +- LogicalProject(deptno=[$1])
- +- LogicalTableScan(table=[[default_catalog, default_database,
emp]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Calc(select=[c])
-+- Sort(orderBy=[EXPR$1 ASC, c ASC])
- +- Exchange(distribution=[single])
- +- Calc(select=[c, CASE(($e = 0), 0, 1) AS EXPR$1])
- +- GroupAggregate(groupBy=[deptno, $e], select=[deptno, $e, COUNT(*)
AS c])
- +- Exchange(distribution=[hash[deptno, $e]])
- +- Expand(projects=[{deptno, 0 AS $e}, {null AS deptno, 1 AS
$e}])
- +- Calc(select=[deptno])
- +- TableSourceScan(table=[[default_catalog,
default_database, emp]], fields=[ename, deptno, gender])
]]>
</Resource>
</TestCase>
@@ -979,31 +945,6 @@ Calc(select=[gender, c])
+- Expand(projects=[{gender, 0 AS $e}, {null AS gender, 1 AS $e}])
+- Calc(select=[gender])
+- TableSourceScan(table=[[default_catalog, default_database,
emp]], fields=[ename, deptno, gender])
-]]>
- </Resource>
- </TestCase>
- <TestCase name="testRollupPlusOrderBy">
- <Resource name="sql">
- <![CDATA[SELECT gender, COUNT(*) AS c FROM emp GROUP BY ROLLUP(gender)
ORDER BY c DESC]]>
- </Resource>
- <Resource name="ast">
- <![CDATA[
-LogicalSort(sort0=[$1], dir0=[DESC-nulls-last])
-+- LogicalAggregate(group=[{0}], groups=[[{0}, {}]], c=[COUNT()])
- +- LogicalProject(gender=[$2])
- +- LogicalTableScan(table=[[default_catalog, default_database, emp]])
-]]>
- </Resource>
- <Resource name="optimized exec plan">
- <![CDATA[
-Sort(orderBy=[c DESC])
-+- Exchange(distribution=[single])
- +- Calc(select=[gender, c])
- +- GroupAggregate(groupBy=[gender, $e], select=[gender, $e, COUNT(*) AS
c])
- +- Exchange(distribution=[hash[gender, $e]])
- +- Expand(projects=[{gender, 0 AS $e}, {null AS gender, 1 AS $e}])
- +- Calc(select=[gender])
- +- TableSourceScan(table=[[default_catalog,
default_database, emp]], fields=[ename, deptno, gender])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/PartialInsertTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/PartialInsertTest.scala
index 574d07b3ecf..4e509882fa7 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/PartialInsertTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/common/PartialInsertTest.scala
@@ -158,9 +158,18 @@ class PartialInsertTest(isBatch: Boolean) extends
TableTestBase {
@TestTemplate
def testPartialInsertWithOrderBy(): Unit = {
- util.verifyRelPlanInsert(
+ val insert =
"INSERT INTO partitioned_sink (e,a,g,f,c,d) " +
- "SELECT e,a,456,123,c,d FROM MyTable ORDER BY a,e,c,d")
+ "SELECT e,a,456,123,c,d FROM MyTable ORDER BY a,e,c,d"
+ // ORDER BY on a non-time attribute is supported in batch but rejected
during streaming
+ // optimization.
+ if (isBatch) {
+ util.verifyRelPlanInsert(insert)
+ } else {
+ assertThatThrownBy(() => util.verifyRelPlanInsert(insert))
+ .hasMessageContaining(
+ "requires the primary sort key to be a time attribute in ascending
order")
+ }
}
@TestTemplate
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SortTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SortTest.scala
index 042842e7d2e..0ec77e0ebe6 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SortTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SortTest.scala
@@ -41,37 +41,10 @@ class SortTest extends TableTestBase {
}
@Test
- def testSortProcessingTimeDesc(): Unit = {
- util.verifyExecPlan("SELECT a FROM MyTable ORDER BY proctime desc, c")
+ def testSortOnRowTimeAlias(): Unit = {
+ util.verifyExecPlan("SELECT a, rowtime AS t FROM MyTable ORDER BY t, c")
}
- @Test
- def testSortRowTimeDesc(): Unit = {
- util.verifyExecPlan("SELECT a FROM MyTable ORDER BY rowtime desc, c")
- }
-
- @Test
- def testSortProcessingTimeSecond(): Unit = {
- util.verifyExecPlan("SELECT a FROM MyTable ORDER BY c, proctime")
- }
-
- @Test
- def testSortRowTimeSecond(): Unit = {
- util.verifyExecPlan("SELECT a FROM MyTable ORDER BY c, rowtime")
- }
-
- @Test
- def testSortProcessingTimeSecondDesc(): Unit = {
- util.verifyExecPlan("SELECT a FROM MyTable ORDER BY c, proctime desc")
- }
-
- @Test
- def testSortRowTimeSecondDesc(): Unit = {
- util.verifyExecPlan("SELECT a FROM MyTable ORDER BY c, rowtime desc")
- }
-
- @Test
- def testSortWithoutTime(): Unit = {
- util.verifyExecPlan("SELECT a FROM MyTable ORDER BY c")
- }
+ // Non-temporal streaming sorts (first sort field is not an ascending time
attribute) are now
+ // rejected during optimization; see SortValidationTest for the
corresponding negative cases.
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
index a2bad91048f..346db891909 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/TableSinkTest.scala
@@ -800,7 +800,7 @@ class TableSinkTest extends TableTestBase {
|)
|""".stripMargin)
val stmtSet = util.tableEnv.createStatementSet()
- stmtSet.addInsertSql("INSERT INTO sink SELECT a,b FROM MyTable ORDER BY a")
+ stmtSet.addInsertSql("INSERT INTO sink SELECT a,b FROM MyTable")
util.verifyExecPlan(stmtSet)
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.scala
index 5e68ef39dd4..ab4c0ffd463 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/GroupingSetsTest.scala
@@ -21,7 +21,7 @@ import org.apache.flink.table.api._
import org.apache.flink.table.planner.plan.utils.FlinkRelOptUtil
import org.apache.flink.table.planner.utils.{TableTestBase, TableTestUtil}
-import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.{assertThat, assertThatThrownBy}
import org.junit.jupiter.api.Test
import java.sql.Date
@@ -213,8 +213,13 @@ class GroupingSetsTest extends TableTestBase {
@Test
def testRollupPlusOrderBy(): Unit = {
- util.verifyExecPlan(
- "SELECT gender, COUNT(*) AS c FROM emp GROUP BY ROLLUP(gender) ORDER BY
c DESC")
+ // A non-time-attribute streaming sort is rejected during optimization.
+ assertThatThrownBy(
+ () =>
+ util.verifyExecPlan(
+ "SELECT gender, COUNT(*) AS c FROM emp GROUP BY ROLLUP(gender) ORDER
BY c DESC"))
+ .hasMessageContaining(
+ "requires the primary sort key to be a time attribute in ascending
order")
}
@Test
@@ -318,7 +323,10 @@ class GroupingSetsTest extends TableTestBase {
"""
|SELECT COUNT(*) AS c FROM emp GROUP BY ROLLUP(deptno) ORDER BY
GROUPING(deptno), c
""".stripMargin
- util.verifyExecPlan(sqlQuery)
+ // A non-time-attribute streaming sort is rejected during optimization.
+ assertThatThrownBy(() => util.verifyExecPlan(sqlQuery))
+ .hasMessageContaining(
+ "requires the primary sort key to be a time attribute in ascending
order")
}
@Test
@@ -424,7 +432,6 @@ class GroupingSetsTest extends TableTestBase {
| END gr_text
|from scott_emp
| GROUP BY ROLLUP(deptno, job, (empno,ename))
- | ORDER BY deptno, job, empno
""".stripMargin
util.verifyExecPlan(sqlQuery)
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala
index 8ba436b52eb..d1d92ce040d 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/SortITCase.scala
@@ -46,9 +46,12 @@ class SortITCase(mode: StateBackendMode) extends
StreamingWithStateTestBase(mode
val da = StreamingEnvUtil.fromCollection(env, data).toTable(tEnv, 'a1, 'a2)
tEnv.createTemporaryView("a", da)
+ // The rejection is raised during optimization and wrapped by the Volcano
program, so assert
+ // on a contained substring rather than an exact message.
assertThatThrownBy(() => tEnv.sqlQuery(sqlQuery).toRetractStream[Row])
- .hasMessage("Sort on a non-time-attribute field is not supported.")
- .isInstanceOf[TableException]
+ .isExactlyInstanceOf(classOf[TableException])
+ .hasMessageContaining(
+ "requires the primary sort key to be a time attribute in ascending
order")
}
@TestTemplate
