From 0eb5131db62cd7939788508377fc148df6ac1ef1 Mon Sep 17 00:00:00 2001 From: daidai Date: Wed, 4 Nov 2026 15:20:10 +0800 Subject: [PATCH] [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 --- .../iceberg/IcebergConnectorTransaction.java | 48 ++++- .../IcebergConnectorTransactionTest.java | 202 +++++++++++++++++- .../iceberg/IcebergWriteKnownBugsTest.java | 3 - .../iceberg/IcebergWritePlanProviderTest.java | 32 +++ .../translator/PhysicalPlanTranslator.java | 18 +- ...ysicalPlanTranslatorAdmissionGateTest.java | 4 +- ...lPlanTranslatorIcebergRowLevelDmlTest.java | 9 +- 7 files changed, 279 insertions(+), 37 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java index b6ca61f312f..9db7005897d 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java +++ b/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 commitData, List 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 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) ─────────────────── diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java index 771588ecca5..dbf2f0df886 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java +++ b/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 paths = new HashSet<>(); + for (DataFile file : currentDataFiles(committed)) { + paths.add(file.path().toString()); + } + Set 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()); diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java index a1ae89bc1ef..131e09cd9f4 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java +++ b/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"); } diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java index c34c2da06ff..9df779895b0 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java +++ b/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 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 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( diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 12b8157dc8e..48da7623c5b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java @@ -609,19 +609,19 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor 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). diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorAdmissionGateTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorAdmissionGateTest.java index ff82ee634a9..c406886c406 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorAdmissionGateTest.java +++ b/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: " diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorIcebergRowLevelDmlTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorIcebergRowLevelDmlTest.java index a789c2bb359..447d41f3a25 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorIcebergRowLevelDmlTest.java +++ b/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.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);