[opt](nereids) support order by clause in create table

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -644,9 +644,9 @@ public class Config extends ConfigBase {
             "Default timeout for insert load job, in seconds."})
     public static int insert_load_default_timeout_second = 14400; // 4 hour
 
-    @ConfField(mutable = true, masterOnly = true, description = {"对 mow 表随机设置 cluster keys，用于测试",
-            "random set cluster keys for mow table for test"})
-    public static boolean random_add_cluster_keys_for_mow = false;
+    @ConfField(mutable = true, masterOnly = true, description = {"对 mow 表随机设置 order by keys，用于测试",
+            "random set order by keys for mow table for test"})
+    public static boolean random_add_order_by_keys_for_mow = false;
 
     @ConfField(mutable = true, masterOnly = true, description = {
             "在 fuzzy 测试中随机选择部分表使用 V3 storage_format（ext_meta），用于增强覆盖",

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -185,8 +185,8 @@ supportedCreateStatement
     : CREATE (EXTERNAL | TEMPORARY)? TABLE (IF NOT EXISTS)? name=multipartIdentifier
         ((ctasCols=identifierList)? | (LEFT_PAREN columnDefs (COMMA indexDefs)? COMMA? RIGHT_PAREN))
         (ENGINE EQ engine=identifier)?
-        ((AGGREGATE | UNIQUE | DUPLICATE) KEY keys=identifierList
-        (CLUSTER BY clusterKeys=identifierList)?)?
+        ((AGGREGATE | UNIQUE | DUPLICATE) KEY keys=identifierList)?
+        (ORDER BY LEFT_PAREN sortItems+=sortItem (COMMA sortItems+=sortItem)* RIGHT_PAREN)?
         (COMMENT STRING_LITERAL)?
         (partition=partitionTable)?
         (DISTRIBUTED BY (HASH hashKeys=identifierList | RANDOM)

fe/fe-core/src/main/java/org/apache/doris/DorisFE.java

@@ -592,8 +592,8 @@ private static void fuzzyConfigs() {
         // Keep global fuzzy knobs that are not session-based.
         if (Config.fuzzy_test_type.equalsIgnoreCase("daily")
                 || Config.fuzzy_test_type.equalsIgnoreCase("rqg")) {
-            Config.random_add_cluster_keys_for_mow = (LocalDate.now().getDayOfMonth() % 2 == 0);
-            LOG.info("fuzzy set random_add_cluster_keys_for_mow={}", Config.random_add_cluster_keys_for_mow);
+            Config.random_add_order_by_keys_for_mow = (LocalDate.now().getDayOfMonth() % 2 == 0);
+            LOG.info("fuzzy set random_add_order_by_keys_for_mow={}", Config.random_add_order_by_keys_for_mow);
         }
 
         setFuzzyForCatalog();

fe/fe-core/src/main/java/org/apache/doris/analysis/KeysDesc.java

@@ -28,7 +28,7 @@
 public class KeysDesc {
     private KeysType type;
     private List<String> keysColumnNames;
-    private List<String> clusterKeysColumnNames;
+    private List<String> orderByKeysColumnNames;
 
     public KeysDesc() {
         this.type = KeysType.AGG_KEYS;
@@ -40,9 +40,9 @@ public KeysDesc(KeysType type, List<String> keysColumnNames) {
         this.keysColumnNames = keysColumnNames;
     }
 
-    public KeysDesc(KeysType type, List<String> keysColumnNames, List<String> clusterKeyColumnNames) {
+    public KeysDesc(KeysType type, List<String> keysColumnNames, List<String> orderByKeysColumnNames) {
         this(type, keysColumnNames);
-        this.clusterKeysColumnNames = clusterKeyColumnNames;
+        this.orderByKeysColumnNames = orderByKeysColumnNames;
     }
 
     public KeysType getKeysType() {
@@ -53,8 +53,8 @@ public int keysColumnSize() {
         return keysColumnNames.size();
     }
 
-    public List<String> getClusterKeysColumnNames() {
-        return clusterKeysColumnNames;
+    public List<String> getOrderByKeysColumnNames() {
+        return orderByKeysColumnNames;
     }
 
     public boolean containsCol(String colName) {
@@ -107,39 +107,39 @@ public void analyze(List<ColumnDef> cols) throws AnalysisException {
             }
         }
 
-        if (clusterKeysColumnNames != null) {
-            analyzeClusterKeys(cols);
+        if (orderByKeysColumnNames != null) {
+            analyzeOrderByKeys(cols);
         }
     }
 
-    private void analyzeClusterKeys(List<ColumnDef> cols) throws AnalysisException {
+    private void analyzeOrderByKeys(List<ColumnDef> cols) throws AnalysisException {
         if (type != KeysType.UNIQUE_KEYS) {
-            throw new AnalysisException("Cluster keys only support unique keys table");
+            throw new AnalysisException("Order by keys only support unique keys table");
         }
-        // check that cluster keys is not duplicated
-        for (int i = 0; i < clusterKeysColumnNames.size(); i++) {
-            String name = clusterKeysColumnNames.get(i);
+        // check that order by keys is not duplicated
+        for (int i = 0; i < orderByKeysColumnNames.size(); i++) {
+            String name = orderByKeysColumnNames.get(i);
             for (int j = 0; j < i; j++) {
-                if (clusterKeysColumnNames.get(j).equalsIgnoreCase(name)) {
-                    throw new AnalysisException("Duplicate cluster key column[" + name + "].");
+                if (orderByKeysColumnNames.get(j).equalsIgnoreCase(name)) {
+                    throw new AnalysisException("Duplicate order by key column[" + name + "].");
                 }
             }
         }
-        // check that cluster keys is not equal to primary keys
-        int minKeySize = Math.min(keysColumnNames.size(), clusterKeysColumnNames.size());
+        // check that order by keys is not equal to primary keys
+        int minKeySize = Math.min(keysColumnNames.size(), orderByKeysColumnNames.size());
         boolean sameKey = true;
         for (int i = 0; i < minKeySize; i++) {
-            if (!keysColumnNames.get(i).equalsIgnoreCase(clusterKeysColumnNames.get(i))) {
+            if (!keysColumnNames.get(i).equalsIgnoreCase(orderByKeysColumnNames.get(i))) {
                 sameKey = false;
                 break;
             }
         }
-        if (sameKey && !Config.random_add_cluster_keys_for_mow) {
+        if (sameKey && !Config.random_add_order_by_keys_for_mow) {
             throw new AnalysisException("Unique keys and cluster keys should be different.");
         }
         // check that cluster key column exists
-        for (int i = 0; i < clusterKeysColumnNames.size(); i++) {
-            String name = clusterKeysColumnNames.get(i);
+        for (int i = 0; i < orderByKeysColumnNames.size(); i++) {
+            String name = orderByKeysColumnNames.get(i);
             for (int j = 0; j < cols.size(); j++) {
                 if (cols.get(j).getName().equalsIgnoreCase(name)) {
                     cols.get(j).setClusterKeyId(i);
@@ -164,10 +164,10 @@ public String toSql() {
             i++;
         }
         stringBuilder.append(")");
-        if (clusterKeysColumnNames != null) {
-            stringBuilder.append("\nCLUSTER BY (");
+        if (orderByKeysColumnNames != null) {
+            stringBuilder.append("\nORDER BY (");
             i = 0;
-            for (String columnName : clusterKeysColumnNames) {
+            for (String columnName : orderByKeysColumnNames) {
                 if (i != 0) {
                     stringBuilder.append(", ");
                 }

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -4036,9 +4036,9 @@ public static void getCreateTableLikeStmt(CreateTableLikeInfo createTableLikeInf
                     }
                 }
                 sb.append(Joiner.on(", ").join(keysColumnNames)).append(")");
-                // show cluster keys
+                // show order keys
                 if (!clusterKeysColumnNamesToId.isEmpty()) {
-                    sb.append("\n").append("CLUSTER BY (`");
+                    sb.append("\n").append("ORDER BY (`");
                     sb.append(Joiner.on("`, `").join(clusterKeysColumnNamesToId.values())).append("`)");
                 }
             }
@@ -4263,6 +4263,9 @@ public static void getCreateTableLikeStmt(CreateTableLikeInfo createTableLikeInf
         } else if (table.getType() == TableType.ICEBERG_EXTERNAL_TABLE) {
             addTableComment(table, sb);
             IcebergExternalTable icebergExternalTable = (IcebergExternalTable) table;
+            if (icebergExternalTable.hasSortOrder()) {
+                sb.append("\n").append(icebergExternalTable.getSortOrderSql());
+            }
             sb.append("\nLOCATION '").append(icebergExternalTable.location()).append("'");
             sb.append("\nPROPERTIES (");
             Iterator<Entry<String, String>> iterator = icebergExternalTable.properties().entrySet().iterator();
@@ -4445,9 +4448,9 @@ public static void getDdlStmt(Command command, String dbName, TableIf table, Lis
                     }
                 }
                 sb.append(Joiner.on(", ").join(keysColumnNames)).append(")");
-                // show cluster keys
+                // show order keys
                 if (!clusterKeysColumnNamesToId.isEmpty()) {
-                    sb.append("\n").append("CLUSTER BY (`");
+                    sb.append("\n").append("ORDER BY (`");
                     sb.append(Joiner.on("`, `").join(clusterKeysColumnNamesToId.values())).append("`)");
                 }
             }
@@ -4678,6 +4681,9 @@ public static void getDdlStmt(Command command, String dbName, TableIf table, Lis
         } else if (table.getType() == TableType.ICEBERG_EXTERNAL_TABLE) {
             addTableComment(table, sb);
             IcebergExternalTable icebergExternalTable = (IcebergExternalTable) table;
+            if (icebergExternalTable.hasSortOrder()) {
+                sb.append("\n").append(icebergExternalTable.getSortOrderSql());
+            }
             sb.append("\nLOCATION '").append(icebergExternalTable.location()).append("'");
             sb.append("\nPROPERTIES (");
             Iterator<Entry<String, String>> iterator = icebergExternalTable.properties().entrySet().iterator();
@@ -5375,7 +5381,7 @@ public static short calcShortKeyColumnCount(List<Column> columns, Map<String, St
                     break;
                 }
             }
-            if (sameKey && !Config.random_add_cluster_keys_for_mow) {
+            if (sameKey && !Config.random_add_order_by_keys_for_mow) {
                 throw new DdlException(shortKeyColumnCount + " short keys is a part of unique keys");
             }
         }

fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java

@@ -2549,8 +2549,8 @@ private boolean createOlapTable(Database db, CreateTableInfo createTableInfo) th
         olapTable.setStorageDictPageSize(storageDictPageSize);
 
         // check data sort properties
-        int keyColumnSize = CollectionUtils.isEmpty(keysDesc.getClusterKeysColumnNames()) ? keysDesc.keysColumnSize() :
-                keysDesc.getClusterKeysColumnNames().size();
+        int keyColumnSize = CollectionUtils.isEmpty(keysDesc.getOrderByKeysColumnNames()) ? keysDesc.keysColumnSize() :
+                keysDesc.getOrderByKeysColumnNames().size();
         DataSortInfo dataSortInfo = PropertyAnalyzer.analyzeDataSortInfo(properties, keysType,
                 keyColumnSize, storageFormat);
         olapTable.setDataSortInfo(dataSortInfo);
@@ -2563,7 +2563,7 @@ private boolean createOlapTable(Database db, CreateTableInfo createTableInfo) th
                 throw new DdlException(e.getMessage());
             }
             if (enableUniqueKeyMergeOnWrite && !enableLightSchemaChange && !CollectionUtils.isEmpty(
-                    keysDesc.getClusterKeysColumnNames())) {
+                    keysDesc.getOrderByKeysColumnNames())) {
                 throw new DdlException(
                     "Unique merge-on-write tables with cluster keys require light schema change to be enabled.");
             }

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalTable.java

@@ -38,6 +38,7 @@
 import org.apache.doris.mtmv.MTMVRelatedTableIf;
 import org.apache.doris.mtmv.MTMVSnapshotIdSnapshot;
 import org.apache.doris.mtmv.MTMVSnapshotIf;
+import org.apache.doris.nereids.trees.plans.commands.info.SortFieldInfo;
 import org.apache.doris.statistics.AnalysisInfo;
 import org.apache.doris.statistics.BaseAnalysisTask;
 import org.apache.doris.statistics.ExternalAnalysisTask;
@@ -377,4 +378,38 @@ public boolean isPartitionedTable() {
         Table table = getIcebergTable();
         return table.spec().isPartitioned();
     }
+
+    /**
+     * Get sort order SQL clause from iceberg table
+     * @return SQL string representing ORDER BY clause, or empty string if no sort order
+     */
+    public String getSortOrderSql() {
+        Table table = getIcebergTable();
+        org.apache.iceberg.SortOrder sortOrder = table.sortOrder();
+        if (sortOrder == null || sortOrder.isUnsorted() || sortOrder.fields().isEmpty()) {
+            return "";
+        }
+
+        List<String> sortItems = new java.util.ArrayList<>();
+        for (org.apache.iceberg.SortField sortField : sortOrder.fields()) {
+            String columnName = table.schema().findColumnName(sortField.sourceId());
+            if (columnName != null) {
+                boolean isAscending = sortField.direction() != org.apache.iceberg.SortDirection.DESC;
+                boolean isNullFirst = sortField.nullOrder() == org.apache.iceberg.NullOrder.NULLS_FIRST;
+                SortFieldInfo sortFieldInfo = new SortFieldInfo(columnName, isAscending, isNullFirst);
+                sortItems.add(sortFieldInfo.toSql());
+            }
+        }
+        return "ORDER BY (" + String.join(", ", sortItems) + ")";
+    }
+
+    /**
+     * Check if table has sort order defined
+     * @return true if table has sort order
+     */
+    public boolean hasSortOrder() {
+        Table table = getIcebergTable();
+        org.apache.iceberg.SortOrder sortOrder = table.sortOrder();
+        return sortOrder != null && !sortOrder.isUnsorted();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java

@@ -44,6 +44,7 @@
 import org.apache.doris.nereids.trees.plans.commands.info.DropPartitionFieldOp;
 import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.ReplacePartitionFieldOp;
+import org.apache.doris.nereids.trees.plans.commands.info.SortFieldInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.TagOptions;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -355,7 +356,17 @@ public boolean performCreateTable(CreateTableInfo createTableInfo) throws UserEx
         properties.put(ExternalCatalog.DORIS_VERSION, ExternalCatalog.DORIS_VERSION_VALUE);
         PartitionSpec partitionSpec = IcebergUtils.solveIcebergPartitionSpec(createTableInfo.getPartitionDesc(),
                 schema);
-        catalog.createTable(getTableIdentifier(dbName, tableName), schema, partitionSpec, properties);
+        // Build and create table with optional sort order
+        org.apache.iceberg.SortOrder sortOrder = buildSortOrder(createTableInfo.getSortOrderFields(), schema);
+        if (sortOrder != null && !sortOrder.isUnsorted()) {
+            catalog.buildTable(getTableIdentifier(dbName, tableName), schema)
+                    .withPartitionSpec(partitionSpec)
+                    .withProperties(properties)
+                    .withSortOrder(sortOrder)
+                    .create();
+        } else {
+            catalog.createTable(getTableIdentifier(dbName, tableName), schema, partitionSpec, properties);
+        }
         return false;
     }
 
@@ -1012,6 +1023,34 @@ private void performDropView(String remoteDbName, String remoteViewName) throws
         ViewCatalog viewCatalog = (ViewCatalog) catalog;
         viewCatalog.dropView(getTableIdentifier(remoteDbName, remoteViewName));
     }
+
+    /**
+     * Build Iceberg SortOrder from SortFieldInfo list
+     */
+    private org.apache.iceberg.SortOrder buildSortOrder(List<SortFieldInfo> sortFields, Schema schema) {
+        if (sortFields == null || sortFields.isEmpty()) {
+            return null;
+        }
+
+        org.apache.iceberg.SortOrder.Builder builder = org.apache.iceberg.SortOrder.builderFor(schema);
+        for (SortFieldInfo sortField : sortFields) {
+            String columnName = sortField.getColumnName();
+            if (sortField.isAscending()) {
+                if (sortField.isNullFirst()) {
+                    builder.asc(columnName, org.apache.iceberg.NullOrder.NULLS_FIRST);
+                } else {
+                    builder.asc(columnName, org.apache.iceberg.NullOrder.NULLS_LAST);
+                }
+            } else {
+                if (sortField.isNullFirst()) {
+                    builder.desc(columnName, org.apache.iceberg.NullOrder.NULLS_FIRST);
+                } else {
+                    builder.desc(columnName, org.apache.iceberg.NullOrder.NULLS_LAST);
+                }
+            }
+        }
+        return builder.build();
+    }
 }