This is an automated email from the ASF dual-hosted git repository.
csun5285 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7488bc7a19e [fix](ccr) Disallow Row Binlog on MOW tables with cluster
keys (#66093)
7488bc7a19e is described below
commit 7488bc7a19e5d2e4856ab69319240a0943bd630a
Author: Chenyang Sun <[email protected]>
AuthorDate: Tue Jul 28 12:18:30 2026 +0800
[fix](ccr) Disallow Row Binlog on MOW tables with cluster keys (#66093)
Disallow Row Binlog on MOW tables with cluster keys
---
.../java/org/apache/doris/catalog/Database.java | 20 +++++
.../apache/doris/datasource/InternalCatalog.java | 18 ++---
.../trees/plans/commands/info/CreateTableInfo.java | 10 +++
.../row_binlog_p0/test_row_binlog_cluster_key.out | 8 ++
.../test_row_binlog_cluster_key.groovy | 88 ++++++++++++++++++++++
5 files changed, 135 insertions(+), 9 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
index f6f21be4977..ac6d53271f2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Database.java
@@ -1049,6 +1049,26 @@ public class Database extends MetaObject implements
Writable, DatabaseIf<Table>,
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()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index 6a43988ccfd..4b44421e583 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -2310,15 +2310,10 @@ public class InternalCatalog implements
CatalogIf<Database> {
}
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 @@ public class InternalCatalog implements
CatalogIf<Database> {
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");
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
index fd30dacc9d1..ad4a526cb6a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
@@ -26,6 +26,7 @@ import org.apache.doris.analysis.SlotRef;
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 class CreateTableInfo {
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 class CreateTableInfo {
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 -> {
diff --git a/regression-test/data/row_binlog_p0/test_row_binlog_cluster_key.out
b/regression-test/data/row_binlog_p0/test_row_binlog_cluster_key.out
new file mode 100644
index 00000000000..0690e49b8a1
--- /dev/null
+++ b/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_ [...]
+
diff --git
a/regression-test/suites/row_binlog_p0/test_row_binlog_cluster_key.groovy
b/regression-test/suites/row_binlog_p0/test_row_binlog_cluster_key.groovy
new file mode 100644
index 00000000000..39eb4d3c2c3
--- /dev/null
+++ b/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
+ """
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
