[fix](ccr) Disallow Row Binlog on MOW tables with cluster keys

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

@@ -1049,6 +1049,26 @@ public BinlogConfig getBinlogConfig() {
         return binlogConfig;
     }
 
+    /**
+     * Get the database binlog config snapshot and the effective table binlog config for creating a table.
+     *
+     * <p>The first value is the database binlog config snapshot, and the second value is the effective
+     * table binlog config after applying table properties.
+     */
+    public Pair<BinlogConfig, BinlogConfig> getBinlogConfigsForCreateTable(
+            Map<String, String> tableProperties) {
+        BinlogConfig dbBinlogConfig;
+        readLock();
+        try {
+            dbBinlogConfig = new BinlogConfig(binlogConfig);
+        } finally {
+            readUnlock();
+        }
+        BinlogConfig createTableBinlogConfig = new BinlogConfig(dbBinlogConfig);
+        createTableBinlogConfig.mergeFromProperties(tableProperties);
+        return Pair.of(dbBinlogConfig, createTableBinlogConfig);
+    }
+
     public void checkStorageVault(Map<String, String> properties) throws DdlException {
         Env env = Env.getCurrentEnv();
         if (!Config.isCloudMode() || !((CloudEnv) env).getEnableStorageVault()) {

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

@@ -2310,15 +2310,10 @@ private boolean createOlapTable(Database db, CreateTableInfo createTableInfo) th
         }
 
         boolean tableHasExist = false;
-        BinlogConfig dbBinlogConfig;
-        db.readLock();
-        try {
-            dbBinlogConfig = new BinlogConfig(db.getBinlogConfig());
-        } finally {
-            db.readUnlock();
-        }
-        BinlogConfig createTableBinlogConfig = new BinlogConfig(dbBinlogConfig);
-        createTableBinlogConfig.mergeFromProperties(createTableInfo.getProperties());
+        Pair<BinlogConfig, BinlogConfig> binlogConfigs =
+                db.getBinlogConfigsForCreateTable(createTableInfo.getProperties());
+        BinlogConfig dbBinlogConfig = binlogConfigs.first;
+        BinlogConfig createTableBinlogConfig = binlogConfigs.second;
         if (dbBinlogConfig.getEnable() && !createTableBinlogConfig.isEnableForCCR() && !createTableInfo.isTemp()) {
             throw new DdlException("Cannot create table with binlog disabled when database binlog enable");
         }
@@ -2840,6 +2835,11 @@ private boolean createOlapTable(Database db, CreateTableInfo createTableInfo) th
                     if (Config.isCloudMode()) {
                         throw new AnalysisException("Binlog<Row> is not supported in the cloud mode yet");
                     }
+                    if (keysType == KeysType.UNIQUE_KEYS && enableUniqueKeyMergeOnWrite
+                            && !CollectionUtils.isEmpty(keysDesc.getOrderByKeysColumnNames())) {
+                        throw new AnalysisException(
+                                "Unique merge-on-write tables with cluster keys do not support binlog<Row>");
+                    }
                     if (keysType == KeysType.DUP_KEYS && binlogConfig.getNeedHistoricalValue()) {
                         throw new AnalysisException("Duplicate table model don't support record historical value");
                     }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java

@@ -26,6 +26,7 @@
 import org.apache.doris.catalog.AggregateType;
 import org.apache.doris.catalog.BinlogConfig;
 import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.Index;
 import org.apache.doris.catalog.KeysType;
@@ -298,6 +299,14 @@ public List<SortFieldInfo> getSortOrderFields() {
         return sortOrderFields;
     }
 
+    private boolean isEffectiveRowBinlogEnabled() {
+        Database db = Env.getCurrentInternalCatalog().getDbNullable(dbName);
+        BinlogConfig binlogConfig = db == null
+                ? BinlogConfig.fromProperties(properties)
+                : db.getBinlogConfigsForCreateTable(properties).second;
+        return binlogConfig.isEnableForStreaming();
+    }
+
     public void setRollups(List<RollupDefinition> rollups) {
         this.rollups = rollups;
     }
@@ -611,6 +620,7 @@ public void validate(ConnectContext ctx) {
 
             try {
                 if (Config.random_add_order_by_keys_for_mow && isEnableMergeOnWrite && sortOrderFields.isEmpty()
+                        && !isEffectiveRowBinlogEnabled()
                         && PropertyAnalyzer.analyzeUseLightSchemaChange(new HashMap<>(properties))) {
                     // exclude columns whose data type can not be order key, see {@link ColumnDefinition#validate}
                     List<ColumnDefinition> orderKeysCandidates = columns.stream().filter(c -> {

regression-test/data/row_binlog_p0/test_row_binlog_cluster_key.out

@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !effective_row_binlog_properties --
+binlog.enable	true
+binlog.format	ROW
+
+-- !show_create_without_random_cluster_key --
+test_row_binlog_without_cluster_key	CREATE TABLE `test_row_binlog_without_cluster_key` (\n  `id` bigint NOT NULL,\n  `cluster_value` int NOT NULL,\n  `payload` varchar(32) NOT NULL\n) ENGINE=OLAP\nUNIQUE KEY(`id`)\nDISTRIBUTED BY HASH(`id`) BUCKETS 1\nPROPERTIES (\n"replication_allocation" = "tag.location.default: 1",\n"min_load_replica_num" = "-1",\n"is_being_synced" = "false",\n"storage_medium" = "hdd",\n"storage_format" = "V2",\n"inverted_index_storage_format" = "V3",\n"enable_unique_key_merge_on_write" = "true",\n"light_schema_change" = "true",\n"disable_auto_compaction" = "false",\n"binlog.enable" = "true",\n"binlog.ttl_seconds" = "86400",\n"binlog.max_bytes" = "9223372036854775807",\n"binlog.max_history_nums" = "9223372036854775807",\n"binlog.format" = "ROW",\n"binlog.need_historical_value" = "false",\n"group_commit_interval_ms" = "10000",\n"group_commit_data_bytes" = "134217728",\n"enable_mow_light_delete" = "false"\n);
+

regression-test/suites/row_binlog_p0/test_row_binlog_cluster_key.groovy

@@ -0,0 +1,88 @@
+// 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.
+
+suite("test_row_binlog_cluster_key", "nonConcurrent") {
+    if (isCloudMode()) {
+        return
+    }
+
+    def inheritedBinlogDb = "test_row_binlog_cluster_key_db"
+
+    sql "DROP TABLE IF EXISTS test_row_binlog_cluster_key FORCE"
+    sql "DROP DATABASE IF EXISTS ${inheritedBinlogDb} FORCE"
+    sql "CREATE DATABASE ${inheritedBinlogDb}"
+    sql """
+        ALTER DATABASE ${inheritedBinlogDb} SET PROPERTIES (
+            "binlog.enable" = "false",
+            "binlog.format" = "ROW"
+        )
+    """
+
+    setFeConfigTemporary([random_add_order_by_keys_for_mow: true]) {
+        test {
+            sql """
+                CREATE TABLE test_row_binlog_cluster_key (
+                    id BIGINT NOT NULL,
+                    cluster_value INT NOT NULL,
+                    payload VARCHAR(32) NOT NULL
+                )
+                UNIQUE KEY(id)
+                ORDER BY(cluster_value)
+                DISTRIBUTED BY HASH(id) BUCKETS 1
+                PROPERTIES (
+                    "replication_num" = "1",
+                    "enable_unique_key_merge_on_write" = "true",
+                    "light_schema_change" = "true",
+                    "binlog.enable" = "true",
+                    "binlog.format" = "ROW"
+                )
+            """
+            exception "Unique merge-on-write tables with cluster keys do not support binlog<Row>"
+        }
+
+        sql """
+            CREATE TABLE ${inheritedBinlogDb}.test_row_binlog_without_cluster_key (
+                id BIGINT NOT NULL,
+                cluster_value INT NOT NULL,
+                payload VARCHAR(32) NOT NULL
+            )
+            UNIQUE KEY(id)
+            DISTRIBUTED BY HASH(id) BUCKETS 1
+            PROPERTIES (
+                "replication_num" = "1",
+                "enable_unique_key_merge_on_write" = "true",
+                "light_schema_change" = "true",
+                "storage_format" = "V2",
+                "binlog.enable" = "true"
+            )
+        """
+
+        order_qt_effective_row_binlog_properties """
+            SELECT PROPERTY_NAME, PROPERTY_VALUE
+            FROM information_schema.table_properties
+            WHERE TABLE_CATALOG = 'internal'
+                AND TABLE_SCHEMA = '${inheritedBinlogDb}'
+                AND TABLE_NAME = 'test_row_binlog_without_cluster_key'
+                AND PROPERTY_NAME IN ('binlog.enable', 'binlog.format')
+            ORDER BY PROPERTY_NAME
+        """
+
+        qt_show_create_without_random_cluster_key """
+            SHOW CREATE TABLE ${inheritedBinlogDb}.test_row_binlog_without_cluster_key
+        """
+    }
+}