[FLINK-XXXXX] add ORDER BY ALL support

docs/layouts/shortcodes/generated/table_config_configuration.html

@@ -74,6 +74,12 @@
             <td>Boolean</td>
             <td>If enabled, executing a CREATE OR ALTER MATERIALIZED TABLE on an existing regular table converts it in place to a materialized table, preserving identity and storage. The option is read at planning time from the session's root configuration, so it must be set when the TableEnvironment session is initialized (for example in the cluster configuration file config.yaml, or in the configuration used to create the session); a session-level SET statement has no effect. When disabled (the default), CREATE OR ALTER MATERIALIZED TABLE against a regular table is rejected.</td>
         </tr>
+        <tr>
+            <td><h5>table.order-by-all-enabled</h5><br> <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Enables the 'ORDER BY ALL' clause, a shorthand that sorts by every expression in the SELECT list, from left to right. A trailing ASC/DESC and NULLS FIRST/LAST applies to all sort keys. Disabled by default during the initial rollout; will be enabled by default in a future release.</td>
+        </tr>
         <tr>
             <td><h5>table.plan.compile.catalog-objects</h5><br> <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span></td>
             <td style="word-wrap: break-word;">ALL</td>

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java

@@ -185,6 +185,18 @@ private TableConfigOptions() {}
                                     + "For example, it prevented using rows in computed columns or join keys. "
                                     + "The new behavior takes the nullability into consideration.");
 
+    @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
+    public static final ConfigOption<Boolean> TABLE_ORDER_BY_ALL_ENABLED =
+            key("table.order-by-all-enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Enables the 'ORDER BY ALL' clause, a shorthand that sorts by every"
+                                    + " expression in the SELECT list, from left to right. A trailing"
+                                    + " ASC/DESC and NULLS FIRST/LAST applies to all sort keys."
+                                    + " Disabled by default during the initial rollout; will be"
+                                    + " enabled by default in a future release.");
+
     // ------------------------------------------------------------------------------------------
     // Options for plan handling
     // ------------------------------------------------------------------------------------------

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java

@@ -198,6 +198,57 @@ protected void validateJoin(SqlJoin join, SqlValidatorScope scope) {
         super.validateJoin(join, scope);
     }
 
+    @Override
+    protected void validateOrderList(SqlSelect select) {
+        checkOrderByAllEnabled(select);
+        super.validateOrderList(select);
+    }
+
+    /**
+     * {@code ORDER BY ALL} is parsed and expanded by Calcite's validator (see {@code
+     * SqlValidatorImpl#rewriteOrderByAll}, invoked from {@code validateOrderList}). Flink only gates
+     * it behind {@link TableConfigOptions#TABLE_ORDER_BY_ALL_ENABLED}, since Calcite provides no
+     * conformance flag for it; the expansion itself is left to {@code super}.
+     */
+    private void checkOrderByAllEnabled(SqlSelect select) {
+        if (!isOrderByAll(select)) {
+            return;
+        }
+        final boolean enabled =
+                ShortcutUtils.unwrapTableConfig(relOptCluster)
+                        .get(TableConfigOptions.TABLE_ORDER_BY_ALL_ENABLED);
+        if (!enabled) {
+            throw new ValidationException(
+                    "ORDER BY ALL is not enabled. Set '"
+                            + TableConfigOptions.TABLE_ORDER_BY_ALL_ENABLED.key()
+                            + "' to true to enable it.");
+        }
+    }
+
+    /**
+     * Detects the {@code ORDER BY ALL} marker exactly as Calcite's {@code rewriteOrderByAll} does:
+     * the order list holds a single element, {@code ORDER_BY_ALL}, optionally wrapped by a trailing
+     * {@code DESC} and/or {@code NULLS FIRST}/{@code NULLS LAST} that applies to every sort key.
+     */
+    private static boolean isOrderByAll(SqlSelect select) {
+        final SqlNodeList orderList = select.getOrderList();
+        if (orderList == null || orderList.size() != 1) {
+            return false;
+        }
+        SqlNode node = orderList.get(0);
+        while (node instanceof SqlCall) {
+            final SqlKind kind = node.getKind();
+            if (kind == SqlKind.NULLS_FIRST
+                    || kind == SqlKind.NULLS_LAST
+                    || kind == SqlKind.DESCENDING) {
+                node = ((SqlCall) node).operand(0);
+            } else {
+                break;
+            }
+        }
+        return node.getKind() == SqlKind.ORDER_BY_ALL;
+    }
+
     @Override
     protected void registerNamespace(
             @Nullable SqlValidatorScope usingScope,

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/OrderByAllTest.java

@@ -0,0 +1,112 @@
+/*
+ * 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.batch.sql;
+
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+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.List;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the {@code ORDER BY ALL} clause. */
+class OrderByAllTest {
+
+    private TableEnvironment tEnv;
+
+    @BeforeEach
+    void setUp() {
+        // Batch mode: ORDER BY ALL performs a global sort over bounded VALUES inputs, so results
+        // are asserted in order. Streaming coverage (where ORDER BY requires a leading ascending
+        // time attribute) is tracked separately.
+        tEnv = TableEnvironment.create(EnvironmentSettings.inBatchMode());
+        tEnv.getConfig().set(TableConfigOptions.TABLE_ORDER_BY_ALL_ENABLED, true);
+    }
+
+    static Stream<Arguments> orderByAllQueries() {
+        return Stream.of(
+                Arguments.of(
+                        "sorts by every column, left to right",
+                        "SELECT x, y "
+                                + "FROM (VALUES (2, 'b'), (1, 'a'), (1, 'c')) AS t(x, y) "
+                                + "ORDER BY ALL",
+                        new Row[] {Row.of(1, "a"), Row.of(1, "c"), Row.of(2, "b")}),
+                Arguments.of(
+                        "a trailing DESC applies to all keys",
+                        "SELECT x, y "
+                                + "FROM (VALUES (2, 'b'), (1, 'a'), (1, 'c')) AS t(x, y) "
+                                + "ORDER BY ALL DESC",
+                        new Row[] {Row.of(2, "b"), Row.of(1, "c"), Row.of(1, "a")}),
+                Arguments.of(
+                        "expands SELECT * to every column before sorting",
+                        "SELECT * "
+                                + "FROM (VALUES (2, 'b'), (1, 'a'), (1, 'c')) AS t(x, y) "
+                                + "ORDER BY ALL",
+                        new Row[] {Row.of(1, "a"), Row.of(1, "c"), Row.of(2, "b")}),
+                Arguments.of(
+                        "expands a qualified star (t.*) before sorting",
+                        "SELECT t.* "
+                                + "FROM (VALUES (2, 'b'), (1, 'a'), (1, 'c')) AS t(x, y) "
+                                + "ORDER BY ALL",
+                        new Row[] {Row.of(1, "a"), Row.of(1, "c"), Row.of(2, "b")}));
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("orderByAllQueries")
+    void testOrderByAll(String description, String sql, Row[] expected) {
+        final List<Row> actual = CollectionUtil.iteratorToList(tEnv.executeSql(sql).collect());
+        assertThat(actual).containsExactly(expected);
+    }
+
+    @Test
+    void testOrderByAllNullsFirstAppliesToAllKeys() {
+        // The trailing NULLS FIRST applies to the (single) expanded sort key.
+        final List<Row> actual =
+                CollectionUtil.iteratorToList(
+                        tEnv.executeSql(
+                                        "SELECT x "
+                                                + "FROM (VALUES (2), (CAST(NULL AS INT)), (1)) AS t(x) "
+                                                + "ORDER BY ALL NULLS FIRST")
+                                .collect());
+        assertThat(actual).containsExactly(Row.of((Integer) null), Row.of(1), Row.of(2));
+    }
+
+    @Test
+    void testOrderByAllDisabledThrows() {
+        tEnv.getConfig().set(TableConfigOptions.TABLE_ORDER_BY_ALL_ENABLED, false);
+        assertThatThrownBy(
+                        () ->
+                                tEnv.executeSql(
+                                                "SELECT x, y "
+                                                        + "FROM (VALUES (2, 'b'), (1, 'a')) AS t(x, y) "
+                                                        + "ORDER BY ALL")
+                                        .collect())
+                .hasMessageContaining("ORDER BY ALL is not enabled");
+    }
+}