apache · yiguolei · Jun 2, 2026 · Jun 1, 2026
@@ -55,10 +55,12 @@
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.iceberg.ManageSnapshots;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowLevelOperationMode;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.SnapshotRef;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.UpdatePartitionSpec;
 import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.catalog.Catalog;
@@ -359,6 +361,10 @@ public boolean performCreateTable(CreateTableInfo createTableInfo) throws UserEx
         Schema schema = new Schema(visit.asNestedType().asStructType().fields());
         Map<String, String> properties = createTableInfo.getProperties();
         properties.put(ExternalCatalog.DORIS_VERSION, ExternalCatalog.DORIS_VERSION_VALUE);
+        properties.putIfAbsent(TableProperties.FORMAT_VERSION, "2");
+        properties.putIfAbsent(TableProperties.DELETE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
+        properties.putIfAbsent(TableProperties.UPDATE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
+        properties.putIfAbsent(TableProperties.MERGE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
         PartitionSpec partitionSpec = IcebergUtils.solveIcebergPartitionSpec(createTableInfo.getPartitionDesc(),
                 schema);
         // Build and create table with optional sort order

@@ -114,6 +114,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         }
 
         IcebergExternalTable icebergTable = (IcebergExternalTable) table;
+        IcebergDmlCommandUtils.checkDeleteMode(icebergTable);
 
         // Verify table format version (must be v2+ for delete support)
         // org.apache.iceberg.Table icebergTableObj = icebergTable.getIcebergTable();
@@ -283,10 +284,12 @@ public Plan getExplainPlan(ConnectContext ctx) {
         if (!(table instanceof IcebergExternalTable)) {
             throw new AnalysisException("Table must be IcebergExternalTable in DELETE command");
         }
+        IcebergExternalTable icebergTable = (IcebergExternalTable) table;
+        IcebergDmlCommandUtils.checkDeleteMode(icebergTable);
         long previousTargetTableId = ctx.getIcebergRowIdTargetTableId();
         ctx.setIcebergRowIdTargetTableId(table.getId());
         try {
-            return completeQueryPlan(ctx, logicalQuery, (IcebergExternalTable) table);
+            return completeQueryPlan(ctx, logicalQuery, icebergTable);
         } finally {
             ctx.setIcebergRowIdTargetTableId(previousTargetTableId);
         }

@@ -0,0 +1,61 @@
+// 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.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+
+import org.apache.iceberg.RowLevelOperationMode;
+import org.apache.iceberg.TableProperties;
+
+import java.util.Map;
+
+/**
+ * Helpers for Iceberg row-level DML commands.
+ */
+final class IcebergDmlCommandUtils {
+    private IcebergDmlCommandUtils() {
+    }
+
+    static void checkDeleteMode(IcebergExternalTable table) {
+        checkNotCopyOnWrite(table, "DELETE", TableProperties.DELETE_MODE,
+                TableProperties.DELETE_MODE_DEFAULT);
+    }
+
+    static void checkUpdateMode(IcebergExternalTable table) {
+        checkNotCopyOnWrite(table, "UPDATE", TableProperties.UPDATE_MODE,
+                TableProperties.UPDATE_MODE_DEFAULT);
+    }
+
+    static void checkMergeMode(IcebergExternalTable table) {
+        checkNotCopyOnWrite(table, "MERGE INTO", TableProperties.MERGE_MODE,
+                TableProperties.MERGE_MODE_DEFAULT);
+    }
+
+    private static void checkNotCopyOnWrite(IcebergExternalTable table, String operation,
+            String modeProperty, String defaultMode) {
+        Map<String, String> properties = table.getIcebergTable().properties();
+        String mode = properties.getOrDefault(modeProperty, defaultMode);
+        if (RowLevelOperationMode.COPY_ON_WRITE.modeName().equalsIgnoreCase(mode)) {
+            throw new AnalysisException(String.format(
+                    "Doris does not support %s on Iceberg copy-on-write tables. "
+                            + "Set table property '%s' to 'merge-on-read'.",
+                    operation, modeProperty));
+        }
+    }
+}
@@ -139,6 +139,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
                     + "Table " + Util.getTempTableDisplayName(table.getName()) + " is not an Iceberg table.");
         }
         IcebergExternalTable icebergTable = (IcebergExternalTable) table;
+        IcebergDmlCommandUtils.checkMergeMode(icebergTable);
         long previousTargetTableId = ctx.getIcebergRowIdTargetTableId();
         ctx.setIcebergRowIdTargetTableId(icebergTable.getId());
         try {
@@ -156,10 +157,12 @@ public Plan getExplainPlan(ConnectContext ctx) {
             throw new AnalysisException("MERGE INTO can only be used on Iceberg tables. "
                     + "Table " + Util.getTempTableDisplayName(table.getName()) + " is not an Iceberg table.");
         }
+        IcebergExternalTable icebergTable = (IcebergExternalTable) table;
+        IcebergDmlCommandUtils.checkMergeMode(icebergTable);
         long previousTargetTableId = ctx.getIcebergRowIdTargetTableId();
-        ctx.setIcebergRowIdTargetTableId(((IcebergExternalTable) table).getId());
+        ctx.setIcebergRowIdTargetTableId(icebergTable.getId());
         try {
-            return buildMergePlan(ctx, (IcebergExternalTable) table);
+            return buildMergePlan(ctx, icebergTable);
         } finally {
             ctx.setIcebergRowIdTargetTableId(previousTargetTableId);
         }

@@ -114,6 +114,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         }
 
         IcebergExternalTable icebergTable = (IcebergExternalTable) table;
+        IcebergDmlCommandUtils.checkUpdateMode(icebergTable);
 
         // Verify table format version (must be v2+ for update support)
         // org.apache.iceberg.Table icebergTableObj = icebergTable.getIcebergTable();
@@ -305,10 +306,12 @@ public Plan getExplainPlan(ConnectContext ctx) {
         if (!(table instanceof IcebergExternalTable)) {
             throw new AnalysisException("Table must be IcebergExternalTable in UPDATE command");
         }
+        IcebergExternalTable icebergTable = (IcebergExternalTable) table;
+        IcebergDmlCommandUtils.checkUpdateMode(icebergTable);
         long previousTargetTableId = ctx.getIcebergRowIdTargetTableId();
         ctx.setIcebergRowIdTargetTableId(table.getId());
         try {
-            return buildMergePlan(ctx, logicalQuery, assignments, (IcebergExternalTable) table);
+            return buildMergePlan(ctx, logicalQuery, assignments, icebergTable);
         } finally {
             ctx.setIcebergRowIdTargetTableId(previousTargetTableId);
         }

@@ -28,9 +28,12 @@
 import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.Maps;
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowLevelOperationMode;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
@@ -102,6 +105,40 @@ public void testProperties() throws UserException {
         Assert.assertEquals("b", table.properties().get("a"));
     }
 
+    @Test
+    public void testDefaultProperties() throws UserException {
+        TableIdentifier tb = TableIdentifier.of(dbName, getTableName());
+        String sql = "create table " + tb + " (id int) engine = iceberg";
+        createTable(sql);
+        Table table = ops.getCatalog().loadTable(tb);
+        Assert.assertEquals(2, getFormatVersion(table));
+        Assert.assertEquals(RowLevelOperationMode.MERGE_ON_READ.modeName(),
+                table.properties().get(TableProperties.DELETE_MODE));
+        Assert.assertEquals(RowLevelOperationMode.MERGE_ON_READ.modeName(),
+                table.properties().get(TableProperties.UPDATE_MODE));
+        Assert.assertEquals(RowLevelOperationMode.MERGE_ON_READ.modeName(),
+                table.properties().get(TableProperties.MERGE_MODE));
+    }
+
+    @Test
+    public void testExplicitProperties() throws UserException {
+        TableIdentifier tb = TableIdentifier.of(dbName, getTableName());
+        String sql = "create table " + tb + " (id int) engine = iceberg properties("
+                + "\"format-version\"=\"1\", "
+                + "\"write.delete.mode\"=\"copy-on-write\", "
+                + "\"write.update.mode\"=\"copy-on-write\", "
+                + "\"write.merge.mode\"=\"copy-on-write\")";
+        createTable(sql);
+        Table table = ops.getCatalog().loadTable(tb);
+        Assert.assertEquals(1, getFormatVersion(table));
+        Assert.assertEquals(RowLevelOperationMode.COPY_ON_WRITE.modeName(),
+                table.properties().get(TableProperties.DELETE_MODE));
+        Assert.assertEquals(RowLevelOperationMode.COPY_ON_WRITE.modeName(),
+                table.properties().get(TableProperties.UPDATE_MODE));
+        Assert.assertEquals(RowLevelOperationMode.COPY_ON_WRITE.modeName(),
+                table.properties().get(TableProperties.MERGE_MODE));
+    }
+
     @Test
     public void testType() throws UserException {
         TableIdentifier tb = TableIdentifier.of(dbName, getTableName());
@@ -191,6 +228,11 @@ public String getTableName() {
         return s.replaceAll("-", "");
     }
 
+    private int getFormatVersion(Table table) {
+        Assert.assertTrue(table instanceof BaseTable);
+        return ((BaseTable) table).operations().current().formatVersion();
+    }
+
     @Test
     public void testDropDB() {
         try {

@@ -60,8 +60,10 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowLevelOperationMode;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
@@ -142,7 +144,11 @@ protected void runBeforeAll() throws Exception {
                 TableIdentifier.of(dbName, tableName),
                 icebergSchema,
                 PartitionSpec.unpartitioned(),
-                ImmutableMap.of("format-version", "2"));
+                ImmutableMap.of(
+                        TableProperties.FORMAT_VERSION, "2",
+                        TableProperties.DELETE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName(),
+                        TableProperties.UPDATE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName(),
+                        TableProperties.MERGE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName()));
 
         List<Column> schema = ImmutableList.of(
                 new Column("id", PrimitiveType.INT),
@@ -172,7 +178,12 @@ protected void runBeforeAll() throws Exception {
         Table mockedIcebergTable = Mockito.mock(Table.class);
         PartitionSpec mockedSpec = Mockito.mock(PartitionSpec.class);
         Mockito.doReturn(false).when(mockedSpec).isPartitioned();
-        Mockito.doReturn(ImmutableMap.of("format-version", "2")).when(mockedIcebergTable).properties();
+        Mockito.doReturn(ImmutableMap.of(
+                TableProperties.FORMAT_VERSION, "2",
+                TableProperties.DELETE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName(),
+                TableProperties.UPDATE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName(),
+                TableProperties.MERGE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName()))
+                .when(mockedIcebergTable).properties();
         Mockito.doReturn(mockedSpec).when(mockedIcebergTable).spec();
         Mockito.doReturn(ImmutableMap.<Integer, PartitionSpec>of()).when(mockedIcebergTable).specs();
         Mockito.doReturn(icebergSchema).when(mockedIcebergTable).schema();

@@ -0,0 +1,91 @@
+// 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.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+
+import org.apache.iceberg.RowLevelOperationMode;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class IcebergDmlCommandUtilsTest {
+
+    @Test
+    public void testDefaultModesRejectCopyOnWriteOperations() {
+        IcebergExternalTable table = mockIcebergExternalTable(new HashMap<>());
+
+        assertCopyOnWriteException(() -> IcebergDmlCommandUtils.checkDeleteMode(table),
+                "DELETE", TableProperties.DELETE_MODE);
+        assertCopyOnWriteException(() -> IcebergDmlCommandUtils.checkUpdateMode(table),
+                "UPDATE", TableProperties.UPDATE_MODE);
+        assertCopyOnWriteException(() -> IcebergDmlCommandUtils.checkMergeMode(table),
+                "MERGE INTO", TableProperties.MERGE_MODE);
+    }
+
+    @Test
+    public void testExplicitCopyOnWriteModeRejectsOperation() {
+        Map<String, String> properties = new HashMap<>();
+        properties.put(TableProperties.DELETE_MODE, RowLevelOperationMode.COPY_ON_WRITE.modeName());
+        properties.put(TableProperties.UPDATE_MODE, RowLevelOperationMode.COPY_ON_WRITE.modeName());
+        properties.put(TableProperties.MERGE_MODE, RowLevelOperationMode.COPY_ON_WRITE.modeName());
+        IcebergExternalTable table = mockIcebergExternalTable(properties);
+
+        assertCopyOnWriteException(() -> IcebergDmlCommandUtils.checkDeleteMode(table),
+                "DELETE", TableProperties.DELETE_MODE);
+        assertCopyOnWriteException(() -> IcebergDmlCommandUtils.checkUpdateMode(table),
+                "UPDATE", TableProperties.UPDATE_MODE);
+        assertCopyOnWriteException(() -> IcebergDmlCommandUtils.checkMergeMode(table),
+                "MERGE INTO", TableProperties.MERGE_MODE);
+    }
+
+    @Test
+    public void testMergeOnReadModeAllowsOperation() {
+        Map<String, String> properties = new HashMap<>();
+        properties.put(TableProperties.DELETE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
+        properties.put(TableProperties.UPDATE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
+        properties.put(TableProperties.MERGE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
+        IcebergExternalTable table = mockIcebergExternalTable(properties);
+
+        Assertions.assertDoesNotThrow(() -> IcebergDmlCommandUtils.checkDeleteMode(table));
+        Assertions.assertDoesNotThrow(() -> IcebergDmlCommandUtils.checkUpdateMode(table));
+        Assertions.assertDoesNotThrow(() -> IcebergDmlCommandUtils.checkMergeMode(table));
+    }
+
+    private static void assertCopyOnWriteException(Runnable action, String operation, String property) {
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class, action::run);
+        Assertions.assertTrue(exception.getMessage().contains(operation));
+        Assertions.assertTrue(exception.getMessage().contains("copy-on-write"));
+        Assertions.assertTrue(exception.getMessage().contains(property));
+    }
+
+    private static IcebergExternalTable mockIcebergExternalTable(Map<String, String> properties) {
+        Table icebergTable = Mockito.mock(Table.class);
+        Mockito.when(icebergTable.properties()).thenReturn(properties);
+
+        IcebergExternalTable table = Mockito.mock(IcebergExternalTable.class);
+        Mockito.when(table.getIcebergTable()).thenReturn(icebergTable);
+        return table;
+    }
+}