[fix](iceberg) Honor standard row-level isolation properties (#69702)

### What problem does this PR solve?

Issue Number: N/A

Related PR: #69551

Problem Summary:
Doris Iceberg DELETE, UPDATE, and MERGE all used the non-standard
`delete_isolation_level` property. As a result, the standard
per-operation table properties `write.delete.isolation-level`,
`write.update.isolation-level`, and `write.merge.isolation-level` were
ignored. A table configured for `snapshot` isolation therefore still
performed serializable data-file conflict validation and rejected normal
concurrent appends.

This change selects the Iceberg `TableProperties` key based on
`WriteOperation` and parses it with `IsolationLevel.fromName()`. The
existing delete-file, base-snapshot, conflict-filter, and
referenced-data-file validations remain enabled; only
`validateNoConflictingDataFiles` is gated on serializable isolation.
Invalid isolation-level values now fail explicitly instead of being
treated as snapshot isolation.

The three disabled regressions from #69551 are enabled. Unit coverage
also verifies per-operation key selection, serializable defaults, and
invalid values.

### Release note

Iceberg DELETE, UPDATE, and MERGE now honor their standard per-operation
isolation-level table properties.

### Check List (For Author)

- Test
    - [ ] Regression test
- [x] Unit Test: `IcebergConnectorTransactionTest` and
`IcebergWriteKnownBugsTest` pass, including all three concurrent-append
snapshot-isolation regressions.
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason

- Behavior changed:
    - [ ] No.
- [x] Yes. Snapshot isolation now permits concurrent data-file appends
while retaining the other row-delta validations.

- Does this need documentation?
    - [x] No.
    - [ ] Yes.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java

@@ -38,6 +38,7 @@ import org.apache.iceberg.FileContent;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.IsolationLevel;
 import org.apache.iceberg.ManifestFile;
 import org.apache.iceberg.ManifestFiles;
 import org.apache.iceberg.ManifestReader;
@@ -129,8 +130,7 @@ import java.util.Set;
 public class IcebergConnectorTransaction implements ConnectorTransaction, RewriteCapableTransaction {
 
     private static final Logger LOG = LogManager.getLogger(IcebergConnectorTransaction.class);
-    private static final String DELETE_ISOLATION_LEVEL = "delete_isolation_level";
-    private static final String DELETE_ISOLATION_LEVEL_DEFAULT = "serializable";
+    private static final String LEGACY_DELETE_ISOLATION_LEVEL = "delete_isolation_level";
     // Scope of a bare INSERT OVERWRITE (no PARTITION clause). Key + values are byte-identical to the fe-core
     // SessionVariable constants (the connector cannot import fe-core); the value reaches the commit through
     // ConnectorSession.getProperty, which reads the session-variable registry by name.
@@ -170,6 +170,7 @@ public class IcebergConnectorTransaction implements ConnectorTransaction, Rewrit
     // Op context captured at begin time, consumed by commit() (the volatile transaction write at the end of
     // beginWrite publishes these plain writes to the commit thread).
     private WriteOperation writeOperation = WriteOperation.INSERT;
+    private IsolationLevel rowLevelIsolationLevel;
     private boolean staticPartitionOverwrite;
     // True when iceberg_partition_overwrite_mode = 'static': a bare INSERT OVERWRITE clears the whole table
     // (every partition spec) instead of replacing only the partitions the new data lands in. Captured at begin
@@ -292,15 +293,23 @@ public class IcebergConnectorTransaction implements ConnectorTransaction, Rewrit
                     if (writeSchemaContext != null) {
                         writeSchemaContext.validateCurrentSchema(loaded, ctx.isOverwrite());
                     }
+                    // Reject invalid isolation before a sink can dispatch BE file writes. Read the SDK
+                    // transaction's refreshed metadata and keep that selection for this statement's commit.
+                    IsolationLevel isolationLevel = null;
+                    if (writeOperation == WriteOperation.DELETE || writeOperation == WriteOperation.UPDATE
+                            || writeOperation == WriteOperation.MERGE) {
+                        isolationLevel = resolveIsolationLevel(opened.table(), writeOperation);
+                    }
                     synchronized (beginLock) {
                         if (closed) {
                             this.table = null;
                             throw new IllegalStateException(
                                     "Iceberg transaction closed while its write table was loading");
                         }
-                        this.transaction = opened;
+                        this.rowLevelIsolationLevel = isolationLevel;
                         this.tableLease = retained;
                         retained = null;
+                        this.transaction = opened;
                         writeStarted = true;
                     }
                     return null;
@@ -1235,7 +1244,7 @@ public class IcebergConnectorTransaction implements ConnectorTransaction, Rewrit
             List<TIcebergCommitData> commitData, List<String> referencedDataFiles) {
         applyBaseSnapshotValidation(rowDelta);
         applyConflictDetectionFilter(rowDelta, icebergTable, commitData);
-        if (isSerializableIsolationLevel(icebergTable)) {
+        if (rowLevelIsolationLevel == IsolationLevel.SERIALIZABLE) {
             rowDelta.validateNoConflictingDataFiles();
         }
         rowDelta.validateDeletedFiles();
@@ -1390,13 +1399,32 @@ public class IcebergConnectorTransaction implements ConnectorTransaction, Rewrit
         return Collections.emptyList();
     }
 
-    boolean isSerializableIsolationLevel(Table icebergTable) {
-        if (icebergTable == null) {
-            return true;
+    IsolationLevel resolveIsolationLevel(Table icebergTable, WriteOperation operation) {
+        String property;
+        String defaultLevel;
+        switch (operation) {
+            case DELETE:
+                property = TableProperties.DELETE_ISOLATION_LEVEL;
+                defaultLevel = TableProperties.DELETE_ISOLATION_LEVEL_DEFAULT;
+                break;
+            case UPDATE:
+                property = TableProperties.UPDATE_ISOLATION_LEVEL;
+                defaultLevel = TableProperties.UPDATE_ISOLATION_LEVEL_DEFAULT;
+                break;
+            case MERGE:
+                property = TableProperties.MERGE_ISOLATION_LEVEL;
+                defaultLevel = TableProperties.MERGE_ISOLATION_LEVEL_DEFAULT;
+                break;
+            default:
+                throw new IllegalArgumentException(
+                        "Unsupported iceberg row-level write operation: " + operation);
         }
-        String level = icebergTable.properties()
-                .getOrDefault(DELETE_ISOLATION_LEVEL, DELETE_ISOLATION_LEVEL_DEFAULT);
-        return "serializable".equalsIgnoreCase(level);
+        // Standard per-operation settings take precedence; existing legacy-only tables retain their
+        // configured isolation until they migrate. Parse only the selected value, including invalid ones.
+        Map<String, String> properties = icebergTable.properties();
+        String level = properties.getOrDefault(property,
+                properties.getOrDefault(LEGACY_DELETE_ISOLATION_LEVEL, defaultLevel));
+        return IsolationLevel.fromName(level);
     }
 
     // ─────────────────── V3 deletion-vector "rewrite previous delete files" (legacy :786-851) ───────────────────

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java

@@ -43,12 +43,14 @@ import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileMetadata;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.IsolationLevel;
 import org.apache.iceberg.PartitionData;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -2032,31 +2034,209 @@ public class IcebergConnectorTransactionTest {
     }
 
     @Test
-    public void isSerializableIsolationLevelDefaultsAndReadsProperty() {
+    public void resolveIsolationLevelUsesOperationDefaultsAndStandardKeyPrecedence() {
         InMemoryCatalog catalog = freshCatalog();
         IcebergConnectorTransaction txn = txnFor(opsReturning(null), new RecordingConnectorContext());
+        Table defaults = catalog.createTable(TableIdentifier.of("db1", "defaults"),
+                SCHEMA, PartitionSpec.unpartitioned());
+        Table legacy = catalog.createTable(TableIdentifier.of("db1", "legacy"),
+                SCHEMA, PartitionSpec.unpartitioned(), props("delete_isolation_level", "snapshot"));
+        Table standard = catalog.createTable(TableIdentifier.of("db1", "standard"),
+                SCHEMA, PartitionSpec.unpartitioned(), props(
+                        "delete_isolation_level", "invalid",
+                        TableProperties.DELETE_ISOLATION_LEVEL, "snapshot",
+                        TableProperties.UPDATE_ISOLATION_LEVEL, "serializable",
+                        TableProperties.MERGE_ISOLATION_LEVEL, "SNAPSHOT"));
+        Table override = catalog.createTable(TableIdentifier.of("db1", "override"),
+                SCHEMA, PartitionSpec.unpartitioned(), props(
+                        "delete_isolation_level", "snapshot",
+                        TableProperties.DELETE_ISOLATION_LEVEL, "serializable",
+                        TableProperties.UPDATE_ISOLATION_LEVEL, "serializable",
+                        TableProperties.MERGE_ISOLATION_LEVEL, "serializable"));
+        for (WriteOperation operation : Arrays.asList(
+                WriteOperation.DELETE, WriteOperation.UPDATE, WriteOperation.MERGE)) {
+            Assertions.assertEquals(IsolationLevel.SERIALIZABLE, txn.resolveIsolationLevel(defaults, operation));
+            Assertions.assertEquals(IsolationLevel.SNAPSHOT, txn.resolveIsolationLevel(legacy, operation));
+            Assertions.assertEquals(IsolationLevel.SERIALIZABLE, txn.resolveIsolationLevel(override, operation));
+        }
+        Assertions.assertEquals(IsolationLevel.SNAPSHOT, txn.resolveIsolationLevel(standard, WriteOperation.DELETE));
+        Assertions.assertEquals(IsolationLevel.SERIALIZABLE, txn.resolveIsolationLevel(standard, WriteOperation.UPDATE));
+        Assertions.assertEquals(IsolationLevel.SNAPSHOT, txn.resolveIsolationLevel(standard, WriteOperation.MERGE));
+        Table invalid = catalog.createTable(TableIdentifier.of("db1", "invalid"),
+                SCHEMA, PartitionSpec.unpartitioned(), props("delete_isolation_level", "snapshot",
+                        TableProperties.UPDATE_ISOLATION_LEVEL, "invalid"));
+        Assertions.assertEquals(IsolationLevel.SNAPSHOT, txn.resolveIsolationLevel(invalid, WriteOperation.DELETE));
+        Assertions.assertEquals(IsolationLevel.SNAPSHOT, txn.resolveIsolationLevel(invalid, WriteOperation.MERGE));
+        Assertions.assertThrows(IllegalArgumentException.class,
+                () -> txn.resolveIsolationLevel(invalid, WriteOperation.UPDATE));
+    }
 
-        Table dflt = catalog.createTable(TableIdentifier.of("db1", "d"), SCHEMA, PartitionSpec.unpartitioned());
-        Assertions.assertTrue(txn.isSerializableIsolationLevel(dflt), "missing property defaults to serializable");
+    @Test
+    public void legacySnapshotIsolationCommitsConcurrentAppendsForEveryRowOperation() throws Exception {
+        for (WriteOperation operation : Arrays.asList(
+                WriteOperation.DELETE, WriteOperation.UPDATE, WriteOperation.MERGE)) {
+            InMemoryCatalog catalog = freshCatalog();
+            TableIdentifier id = TableIdentifier.of("db1", "legacy");
+            Table table = catalog.createTable(id, SCHEMA, PartitionSpec.unpartitioned(),
+                    props("format-version", "2", "delete_isolation_level", "snapshot"));
+            String seed = "s3://b/db1/legacy/seed.parquet";
+            String appended = "s3://b/db1/legacy/appended.parquet";
+            String replacement = "s3://b/db1/legacy/replacement.parquet";
+            table.newAppend().appendFile(dataFile(table.spec(), seed, 5L)).commit();
+            IcebergConnectorTransaction txn = txnFor(opsReturning(table), new RecordingConnectorContext());
+            txn.beginWrite(SESSION, "db1", "legacy", new IcebergWriteContext(
+                    operation, false, Collections.emptyMap(), Optional.empty()));
+            catalog.loadTable(id).newAppend().appendFile(dataFile(table.spec(), appended, 7L)).commit();
+            long appendSnapshot = catalog.loadTable(id).currentSnapshot().snapshotId();
+            txn.addCommitData(commitBytes(positionDeleteItem("s3://b/db1/legacy/delete.parquet", 2L, seed)));
+            if (operation != WriteOperation.DELETE) {
+                txn.addCommitData(commitBytes(dataFileItem(replacement, 2L, 1024L)));
+            }
 
-        Table ser = catalog.createTable(TableIdentifier.of("db1", "s"), SCHEMA, PartitionSpec.unpartitioned(),
-                props("delete_isolation_level", "serializable"));
-        Assertions.assertTrue(txn.isSerializableIsolationLevel(ser));
+            txn.commit();
 
-        Table snap = catalog.createTable(TableIdentifier.of("db1", "n"), SCHEMA, PartitionSpec.unpartitioned(),
-                props("delete_isolation_level", "snapshot"));
-        Assertions.assertFalse(txn.isSerializableIsolationLevel(snap), "snapshot isolation is not serializable");
+            Table committed = catalog.loadTable(id);
+            Assertions.assertNotEquals(appendSnapshot, committed.currentSnapshot().snapshotId());
+            Assertions.assertEquals("1", committed.currentSnapshot().summary().get("added-delete-files"));
+            Set<String> paths = new HashSet<>();
+            for (DataFile file : currentDataFiles(committed)) {
+                paths.add(file.path().toString());
+            }
+            Set<String> expected = new HashSet<>(Arrays.asList(seed, appended));
+            if (operation != WriteOperation.DELETE) {
+                expected.add(replacement);
+                Assertions.assertEquals("1", committed.currentSnapshot().summary().get("added-data-files"));
+            }
+            Assertions.assertEquals(expected, paths, "the row delta must publish without losing the append");
+            txn.close();
+        }
+    }
+
+    @Test
+    public void invalidIsolationFailsBeginAndReleasesRetainedCatalogResources() {
+        for (String property : Arrays.asList(TableProperties.DELETE_ISOLATION_LEVEL,
+                TableProperties.UPDATE_ISOLATION_LEVEL, TableProperties.MERGE_ISOLATION_LEVEL,
+                "delete_isolation_level")) {
+            InMemoryCatalog catalog = freshCatalog();
+            Table table = catalog.createTable(TableIdentifier.of("db1", "invalid"),
+                    SCHEMA, PartitionSpec.unpartitioned(), props("format-version", "2", property, "invalid"));
+            WriteOperation operation = TableProperties.UPDATE_ISOLATION_LEVEL.equals(property)
+                    ? WriteOperation.UPDATE : TableProperties.MERGE_ISOLATION_LEVEL.equals(property)
+                            ? WriteOperation.MERGE : WriteOperation.DELETE;
+            IcebergCatalogResourceTracker tracker = new IcebergCatalogResourceTracker();
+            IcebergConnectorTransaction txn = new IcebergConnectorTransaction(
+                    42L, opsReturning(table), new RecordingConnectorContext(), tracker);
+
+            DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class,
+                    () -> txn.beginWrite(SESSION, "db1", "invalid", new IcebergWriteContext(
+                            operation, false, Collections.emptyMap(), Optional.empty())));
+
+            Assertions.assertTrue(ex.getMessage().contains("Invalid isolation level: invalid"));
+            Assertions.assertNull(txn.getTransaction(), "failed begin must not publish the SDK transaction");
+            Assertions.assertNull(txn.getTable());
+            Assertions.assertTrue(txn.getCommitDataList().isEmpty());
+            AtomicInteger cleanup = new AtomicInteger();
+            tracker.close(cleanup::incrementAndGet);
+            Assertions.assertEquals(1, cleanup.get(), "failed begin must release both local tracked-table leases");
+            txn.close();
+            Assertions.assertEquals(1, cleanup.get(), "close must not retain or release the failed lease twice");
+        }
+    }
+
+    @Test
+    public void beginWriteResolvesIsolationFromOpenedTransactionMetadata() {
+        InMemoryCatalog catalog = freshCatalog();
+        Table table = catalog.createTable(TableIdentifier.of("db1", "refreshed"),
+                SCHEMA, PartitionSpec.unpartitioned(), props("format-version", "2",
+                        TableProperties.DELETE_ISOLATION_LEVEL, "snapshot"));
+        AtomicInteger opens = new AtomicInteger();
+        Table refreshing = (Table) Proxy.newProxyInstance(Table.class.getClassLoader(), new Class<?>[] {Table.class},
+                (proxy, method, args) -> {
+                    if ("newTransaction".equals(method.getName())) {
+                        if (opens.incrementAndGet() == 1) {
+                            table.updateProperties().set(TableProperties.DELETE_ISOLATION_LEVEL, "invalid").commit();
+                        }
+                    }
+                    try {
+                        return method.invoke(table, args);
+                    } catch (InvocationTargetException e) {
+                        throw e.getCause();
+                    }
+                });
+        IcebergConnectorTransaction txn = txnFor(opsReturning(refreshing), new RecordingConnectorContext());
+
+        DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class,
+                () -> txn.beginWrite(SESSION, "db1", "refreshed", deleteCtx()));
+
+        Assertions.assertEquals(1, opens.get());
+        Assertions.assertTrue(ex.getMessage().contains("Invalid isolation level: invalid"));
+        Assertions.assertNull(txn.getTransaction());
+        table.updateProperties().set(TableProperties.DELETE_ISOLATION_LEVEL, "snapshot").commit();
+        // A failed begin must also release the begin-once guard, so the same transaction can be prepared again.
+        txn.beginWrite(SESSION, "db1", "refreshed", deleteCtx());
+        Assertions.assertNotNull(txn.getTransaction());
+        txn.close();
+    }
+
+    @Test
+    public void rowLevelIsolationSelectionStaysFixedAfterBegin() throws Exception {
+        for (String initial : Arrays.asList("snapshot", "serializable")) {
+            InMemoryCatalog catalog = freshCatalog();
+            TableIdentifier id = TableIdentifier.of("db1", "fixed");
+            Table table = catalog.createTable(id, SCHEMA, PartitionSpec.unpartitioned(), props(
+                    "format-version", "2", TableProperties.UPDATE_ISOLATION_LEVEL, initial));
+            String seed = "s3://b/db1/fixed/seed.parquet";
+            table.newAppend().appendFile(dataFile(table.spec(), seed, 5L)).commit();
+            IcebergConnectorTransaction txn = txnFor(opsReturning(table), new RecordingConnectorContext());
+            txn.beginWrite(SESSION, "db1", "fixed", new IcebergWriteContext(
+                    WriteOperation.UPDATE, false, Collections.emptyMap(), Optional.empty()));
+            // Change even the SDK transaction table's properties: commit must use the parsed begin-time enum.
+            txn.getTransaction().table().updateProperties().set(TableProperties.UPDATE_ISOLATION_LEVEL,
+                    "snapshot".equals(initial) ? "invalid" : "snapshot").commit();
+            catalog.loadTable(id).newAppend()
+                    .appendFile(dataFile(table.spec(), "s3://b/db1/fixed/appended.parquet", 7L)).commit();
+            txn.addCommitData(commitBytes(positionDeleteItem("s3://b/db1/fixed/delete.parquet", 2L, seed)));
+            txn.addCommitData(commitBytes(dataFileItem("s3://b/db1/fixed/replacement.parquet", 2L, 1024L)));
+
+            if ("snapshot".equals(initial)) {
+                txn.commit();
+                Assertions.assertEquals("1", reloadCurrentSnapshot(catalog, id).summary().get("added-delete-files"));
+            } else {
+                Assertions.assertThrows(DorisConnectorException.class, txn::commit,
+                        "begin-time serializable isolation must still reject the concurrent append");
+            }
+            txn.close();
+        }
+    }
+
+    @Test
+    public void nonRowLevelBeginDoesNotParseRowLevelIsolation() {
+        InMemoryCatalog catalog = freshCatalog();
+        Table table = catalog.createTable(TableIdentifier.of("db1", "other"),
+                SCHEMA, PartitionSpec.unpartitioned(), props("format-version", "2",
+                        "delete_isolation_level", "invalid",
+                        TableProperties.DELETE_ISOLATION_LEVEL, "invalid",
+                        TableProperties.UPDATE_ISOLATION_LEVEL, "invalid",
+                        TableProperties.MERGE_ISOLATION_LEVEL, "invalid"));
+        for (WriteOperation operation : Arrays.asList(
+                WriteOperation.INSERT, WriteOperation.OVERWRITE, WriteOperation.REWRITE)) {
+            IcebergConnectorTransaction txn = txnFor(opsReturning(table), new RecordingConnectorContext());
+            txn.beginWrite(SESSION, "db1", "other", new IcebergWriteContext(
+                    operation, operation == WriteOperation.OVERWRITE, Collections.emptyMap(), Optional.empty()));
+            Assertions.assertNotNull(txn.getTransaction());
+            txn.close();
+        }
     }
 
     @Test
     public void deleteSnapshotIsolationSkipsConcurrentDataFileValidation() {
-        // parity: at delete_isolation_level=snapshot (non-serializable) validateNoConflictingDataFiles is NOT
+        // parity: at write.delete.isolation-level=snapshot (non-serializable) validateNoConflictingDataFiles is NOT
         // applied, so a concurrent data-file append since the base snapshot does NOT fail the delete — the inverse
         // of deleteDetectsConcurrentDataFileConflict (which runs at the default serializable level).
         InMemoryCatalog catalog = freshCatalog();
         TableIdentifier id = TableIdentifier.of("db1", "t1");
         Table table = catalog.createTable(id, SCHEMA, PartitionSpec.unpartitioned(),
-                props("format-version", "2", "delete_isolation_level", "snapshot"));
+                props("format-version", "2", TableProperties.DELETE_ISOLATION_LEVEL, "snapshot"));
         table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db1/t1/seed.parquet", 5L)).commit();
 
         IcebergConnectorTransaction txn = txnFor(opsReturning(catalog.loadTable(id)), new RecordingConnectorContext());

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java

@@ -50,19 +50,16 @@ public class IcebergWriteKnownBugsTest {
     private static final String COMMIT_LOCATION = "s3://bucket/events";
 
     @Test
-    @Disabled("Known issue: DELETE ignores the standard snapshot isolation property.")
     public void deleteHonorsStandardSnapshotIsolationProperty() throws Exception {
         assertStandardSnapshotIsolation(WriteOperation.DELETE, "write.delete.isolation-level");
     }
 
     @Test
-    @Disabled("Known issue: UPDATE ignores the standard snapshot isolation property.")
     public void updateHonorsStandardSnapshotIsolationProperty() throws Exception {
         assertStandardSnapshotIsolation(WriteOperation.UPDATE, "write.update.isolation-level");
     }
 
     @Test
-    @Disabled("Known issue: MERGE ignores the standard snapshot isolation property.")
     public void mergeHonorsStandardSnapshotIsolationProperty() throws Exception {
         assertStandardSnapshotIsolation(WriteOperation.MERGE, "write.merge.isolation-level");
     }

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java

@@ -402,6 +402,38 @@ public class IcebergWritePlanProviderTest {
         return plan.getDataSink().getIcebergTableSink();
     }
 
+    @Test
+    public void invalidSelectedIsolationRejectsPlanBeforeReturningBackendSink() {
+        Map<WriteOperation, String> isolationKeys = new LinkedHashMap<>();
+        isolationKeys.put(WriteOperation.DELETE, TableProperties.DELETE_ISOLATION_LEVEL);
+        isolationKeys.put(WriteOperation.UPDATE, TableProperties.UPDATE_ISOLATION_LEVEL);
+        isolationKeys.put(WriteOperation.MERGE, TableProperties.MERGE_ISOLATION_LEVEL);
+        for (Map.Entry<WriteOperation, String> entry : isolationKeys.entrySet()) {
+            Table table = unpartitionedUnsortedTable(freshCatalog());
+            table.updateProperties().set("delete_isolation_level", "snapshot")
+                    .set(entry.getValue(), "invalid").commit();
+            RecordingConnectorContext context = contextWithStorage();
+            RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+            ops.table = table;
+            IcebergConnectorTransaction txn = new IcebergConnectorTransaction(42L, ops, context);
+            WriteHandle handle = new WriteHandle(new IcebergTableHandle("db1", "t2"))
+                    .writeOperation(entry.getKey());
+
+            DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class,
+                    () -> providerFor(table, context).planWrite(new WriteSession(txn), handle));
+
+            Assertions.assertTrue(ex.getMessage().contains("Invalid isolation level: invalid"),
+                    "an invalid selected standard key must not fall back to the valid legacy value");
+            Assertions.assertNull(txn.getTransaction(), "a failed preparation cannot return an executable BE sink");
+            Assertions.assertNull(txn.getTable());
+            Assertions.assertTrue(txn.getCommitDataList().isEmpty());
+            Assertions.assertEquals(0, context.normalizeCount,
+                    "invalid isolation must fail before output paths are bound into a backend sink");
+            Assertions.assertNull(table.currentSnapshot(), "preparation must not create write snapshots");
+            txn.close();
+        }
+    }
+
     @Test
     public void getWriteColumnsCarriesPinnedTypedDefaultsAndNullability() {
         Schema writeSchema = new Schema(

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java

@@ -609,19 +609,19 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
             outputExprs.add(slotRef);
         }
         rootFragment.setOutputExprs(outputExprs);
-        // The MERGE/UPDATE target is a PluginDrivenExternalTable: route through the connector's
-        // PluginDrivenTableSink with WriteOperation.MERGE so the connector's planWrite emits its
-        // TIcebergMergeSink dialect (which threads its own sort_fields).
-        // SQL MERGE INTO carries the cardinality requirement onto the write handle; UPDATE shares this
-        // sink dialect but has no such rule, so it threads false (see PhysicalExternalRowLevelMergeSink).
-        rootFragment.setSink(buildPluginRowLevelDmlSink(mergeSink, WriteOperation.MERGE,
+        // UPDATE and SQL MERGE share the plugin BE sink dialect, but their per-operation connector
+        // behavior differs. The existing SQL-MERGE cardinality requirement is set only by the MERGE
+        // builder, so preserve the statement identity when selecting the connector write operation.
+        WriteOperation rowLevelOperation = mergeSink.isRequireMergeCardinalityCheck()
+                ? WriteOperation.MERGE : WriteOperation.UPDATE;
+        rootFragment.setSink(buildPluginRowLevelDmlSink(mergeSink, rowLevelOperation,
                 mergeSink.isWritesDataFiles(), mergeSink.isRequireMergeCardinalityCheck(),
                 mergeSink.getBoundWriteMetadataIdentity(), context));
         return rootFragment;
     }
 
     /**
-     * Builds the plugin-driven sink for an iceberg row-level DML (DELETE / MERGE) whose target has
+     * Builds the plugin-driven sink for an iceberg row-level DML (DELETE / UPDATE / MERGE) whose target has
      * been flipped to a {@link PluginDrivenExternalTable}. Mirrors {@link #visitPhysicalConnectorTableSink}'s
      * connector resolution (catalog -&gt; connector -&gt; metadata -&gt; pinned table handle) but threads the
      * given {@link WriteOperation} so the connector's {@code planWrite} dispatches to its DELETE / MERGE BE sink
@@ -664,10 +664,10 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
         Set<WriteOperation> writeOps = writePlanProvider == null
                 ? EnumSet.noneOf(WriteOperation.class)
                 : writePlanProvider.supportedOperations();
-        if (!(writeOps.contains(WriteOperation.DELETE) || writeOps.contains(WriteOperation.MERGE))) {
+        if (!writeOps.contains(writeOperation)) {
             throw new AnalysisException(
                     "Connector '" + catalog.getName() + "' (type: " + catalog.getType()
-                            + ") does not support row-level DML operations");
+                            + ") does not support " + writeOperation + " row-level DML");
         }
         // writeSortInfo == null: a row-level DML has no engine-resolved write sort (MERGE's sort lives in the
         // connector's TIcebergMergeSink.sort_fields, DELETE is unsorted).

fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorAdmissionGateTest.java

@@ -125,7 +125,7 @@ public class PhysicalPlanTranslatorAdmissionGateTest {
 
     @Test
     public void rowLevelDmlGateRejectsConnectorDeclaringOnlyInsertWithDistinctMessage() {
-        // Declares INSERT (would pass the INSERT gate above) but neither DELETE nor MERGE: the row-level DML
+        // Declares INSERT (would pass the INSERT gate above) but not DELETE: the row-level DML
         // helper must reject it, and with a message DISTINCT from the INSERT gate's, so logs/callers can tell
         // "this connector can't do row-level DML at all" apart from "this connector can't write at all".
         PlanTranslatorContext context = new PlanTranslatorContext();
@@ -141,7 +141,7 @@ public class PhysicalPlanTranslatorAdmissionGateTest {
         PhysicalPlanTranslator translator = new PhysicalPlanTranslator(context, null);
         AnalysisException ex = Assertions.assertThrows(AnalysisException.class,
                 () -> translator.visitPhysicalExternalRowLevelDeleteSink(sink, context));
-        Assertions.assertTrue(ex.getMessage().contains("does not support row-level DML operations"),
+        Assertions.assertTrue(ex.getMessage().contains("does not support DELETE row-level DML"),
                 "got: " + ex.getMessage());
         Assertions.assertFalse(ex.getMessage().contains("does not support INSERT operations"),
                 "the row-level DML rejection must be a message DISTINCT from the INSERT gate's, got: "

fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorIcebergRowLevelDmlTest.java

@@ -142,6 +142,7 @@ public class PhysicalPlanTranslatorIcebergRowLevelDmlTest {
         Mockito.doReturn(plugin.table).when(sink).getTargetTable();
         Mockito.doReturn(ImmutableList.of(DATA)).when(sink).getCols();
         Mockito.doReturn(ImmutableList.<Slot>of(dataSlot, opSlot, rowidSlot)).when(sink).getOutput();
+        Mockito.doReturn(true).when(sink).isRequireMergeCardinalityCheck();
 
         PhysicalPlanTranslator translator = new PhysicalPlanTranslator(context, null);
         translator.visitPhysicalExternalRowLevelMergeSink(sink, context);
@@ -179,6 +180,8 @@ public class PhysicalPlanTranslatorIcebergRowLevelDmlTest {
     @Test
     public void mergePluginArmThreadsTheSqlMergeCardinalityRequirementOntoTheWriteHandle() {
         PluginDrivenTableSink pluginSink = translateMergeWithCardinalityRequirement(true);
+        Assertions.assertEquals(WriteOperation.MERGE, Deencapsulation.getField(pluginSink, "writeOperation"),
+                "SQL MERGE must select write.merge.isolation-level through WriteOperation.MERGE");
         Assertions.assertEquals(true,
                 Deencapsulation.getField(pluginSink, "requireMergeCardinalityCheck"),
                 "a SQL MERGE must thread requireMergeCardinalityCheck=true so the connector stamps it onto"
@@ -188,6 +191,8 @@ public class PhysicalPlanTranslatorIcebergRowLevelDmlTest {
     @Test
     public void updatePluginArmThreadsNoCardinalityRequirementOntoTheWriteHandle() {
         PluginDrivenTableSink pluginSink = translateMergeWithCardinalityRequirement(false);
+        Assertions.assertEquals(WriteOperation.UPDATE, Deencapsulation.getField(pluginSink, "writeOperation"),
+                "UPDATE must select write.update.isolation-level through WriteOperation.UPDATE");
         Assertions.assertEquals(false,
                 Deencapsulation.getField(pluginSink, "requireMergeCardinalityCheck"),
                 "an UPDATE reaches the same sink dialect but carries no SQL cardinality rule");
@@ -392,9 +397,9 @@ public class PhysicalPlanTranslatorIcebergRowLevelDmlTest {
         // Production selects the write provider per-handle; a plain mock does not run the interface default.
         Mockito.when(connector.getWritePlanProvider(Mockito.any())).thenReturn(provider);
         // The row-level DML gate (buildPluginRowLevelDmlSink) resolves the handle, fetches the per-handle
-        // provider and admits on ITS supportedOperations containing DELETE/MERGE.
+        // provider and admits on ITS supportedOperations containing the concrete operation.
         Mockito.when(provider.supportedOperations())
-                .thenReturn(EnumSet.of(WriteOperation.DELETE, WriteOperation.MERGE));
+                .thenReturn(EnumSet.of(WriteOperation.DELETE, WriteOperation.UPDATE, WriteOperation.MERGE));
         Mockito.when(provider.planWrite(Mockito.any(), Mockito.any()))
                 .thenReturn(new ConnectorSinkPlan(new TDataSink()));
         Mockito.when(connector.getMetadata(Mockito.any())).thenReturn(metadata);