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 fb9f38ed96707f..511827f5b9671c 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
@@ -23,6 +23,7 @@
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
 import org.apache.flink.table.planner.codegen.sort.ComparatorCodeGenerator;
 import org.apache.flink.table.planner.delegation.PlannerBase;
 import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
@@ -33,10 +34,12 @@
 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.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
@@ -100,12 +103,21 @@ public StreamExecSort(
     @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.
+            int firstSortField = sortSpec.getFieldIndices()[0];
+            String column = inputType.getFieldNames().get(firstSortField);
+            LogicalType type = inputType.getTypeAt(firstSortField);
+            if (FlinkTypeFactory.isTimeIndicatorType(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 63958f7235c715..85ef8544732e0d 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 da6c1e9fbaf452..a24553f9ff6aef 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,21 @@ object SortUtil {
     rowType.getFieldList.get(idx)
   }
 
+  /** 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 '$column' is ${tpe.asSummaryString}. A time attribute is an event-time column " +
+      s"(a TIMESTAMP with a WATERMARK) or a processing-time column. Otherwise use LIMIT for " +
+      s"Top-N, sort within a window, or run in 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 00000000000000..048f287f1bcfc6
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SortValidationTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.TableException;
+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.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 must
+ * be rejected during optimization (so it surfaces at {@code COMPILE PLAN} / planning time), not
+ * deferred to execution-plan translation. 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));
+    }
+
+    @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 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);
+    }
+}
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 9f23ebd1a81b92..d691886784fed2 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 22a2cfc58fcb1a..c92a3bcfa7b592 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 486e4d045ec2cc..493a6f1f9329fd 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 a56cec435289a9..54d07e3fb382b3 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-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]])
+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 574d07b3ecf6a2..4e509882fa7d37 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 042842e7d2eaef..0ec77e0ebe6eff 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 a2bad91048f99d..346db891909e13 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 5e68ef39dd439e..ab4c0ffd463ad8 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 8ba436b52ebe87..d1d92ce040d624 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