This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 0eb5131db62 [fix](iceberg) Honor standard row-level isolation
properties (#69702)
0eb5131db62 is described below
commit 0eb5131db62cd7939788508377fc148df6ac1ef1
Author: daidai <[email protected]>
AuthorDate: Wed Nov 4 15:20:10 2026 +0800
[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 | 50 +++--
.../iceberg/IcebergConnectorTransactionTest.java | 202 +++++++++++++++++++--
.../iceberg/IcebergWriteKnownBugsTest.java | 3 -
.../iceberg/IcebergWritePlanProviderTest.java | 32 ++++
.../glue/translator/PhysicalPlanTranslator.java | 18 +-
.../PhysicalPlanTranslatorAdmissionGateTest.java | 4 +-
...ysicalPlanTranslatorIcebergRowLevelDmlTest.java | 9 +-
7 files changed, 280 insertions(+), 38 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<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;
- }
- String level = icebergTable.properties()
- .getOrDefault(DELETE_ISOLATION_LEVEL,
DELETE_ISOLATION_LEVEL_DEFAULT);
- return "serializable".equalsIgnoreCase(level);
+ 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);
+ }
+ // 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) ───────────────────
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));
+ }
+
+ @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)));
+ }
+
+ txn.commit();
+
+ 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();
+ }
+ }
- Table dflt = catalog.createTable(TableIdentifier.of("db1", "d"),
SCHEMA, PartitionSpec.unpartitioned());
- Assertions.assertTrue(txn.isSerializableIsolationLevel(dflt), "missing
property defaults to serializable");
+ @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);
- Table ser = catalog.createTable(TableIdentifier.of("db1", "s"),
SCHEMA, PartitionSpec.unpartitioned(),
- props("delete_isolation_level", "serializable"));
- Assertions.assertTrue(txn.isSerializableIsolationLevel(ser));
+ 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");
+ }
+ }
- 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");
+ @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<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(
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<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 -> connector -> metadata ->
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).
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.<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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
