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 bf278a5b210 [feature](iceberg) Support whole-file metadata delete for
DELETE (#69176)
bf278a5b210 is described below
commit bf278a5b2106b92ebe8f8d91d9d339e9628bfdb4
Author: daidai <[email protected]>
AuthorDate: Fri Oct 23 15:20:38 2026 +0800
[feature](iceberg) Support whole-file metadata delete for DELETE (#69176)
### What problem does this PR solve?
Issue Number: N/A
Problem Summary:
For an Iceberg table, a row-level DELETE always scans the matching rows
and writes delete files (position deletes on v2, deletion vectors on
v3), so a data file whose rows are all gone stays in the snapshot and
every later scan still reads it.
This PR adds the whole-file "metadata delete" fast path that Spark /
Trino use for whole-partition and whole-file predicates: the DELETE is
committed as one metadata operation that drops the matching data files,
so the statement reads no rows, writes no delete files, and later scans
no longer read those files.
- The engine only tries the fast path when the statement predicate can
be extracted exactly (single-table DELETE, leaf is the target table
scan, every conjunct representable). Anything else keeps the current
row-level path. `PARTITION(...)` scoped deletes, UPDATE and MERGE are
unchanged.
- The Iceberg connector lowers the predicate with the strictly
all-or-nothing rewrite matrix and only commits when every candidate file
is matched in full (whole-partition shortcut, strict partition
projection or file metrics); otherwise it falls back to the row-level
delete, so a partially matched file is never dropped.
- Like the engines above, the committed snapshot applies the predicate
to the current table state, and the reported row count is iceberg's
`deleted-records` minus the rows the removed delete files had already
hidden.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] 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.
- 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
---
.../connector/hive/HiveConnectorMetadata.java | 16 +
...HiveConnectorMetadataSiblingDelegationTest.java | 42 ++-
.../iceberg/IcebergConnectorMetadata.java | 333 ++++++++++++++++++
.../iceberg/IcebergConnectorTransaction.java | 57 +++-
.../iceberg/IcebergMetadataDeleteTest.java | 378 +++++++++++++++++++++
.../doris/connector/spi/ConnectorWriteOps.java | 43 +++
.../connector/spi/ConnectorPluginSurfaceTest.java | 6 +-
.../test/resources/connector-metadata-methods.txt | 1 +
fe/fe-connector/pom.xml | 2 +-
.../NereidsToConnectorExpressionConverter.java | 100 ++++--
.../converter/WriteConstraintExtractor.java | 106 ++++++
.../PositionDeleteRowLevelDmlTransform.java | 42 ++-
.../trees/plans/commands/RowLevelDmlCommand.java | 87 +++++
.../trees/plans/commands/RowLevelDmlTransform.java | 21 ++
.../NereidsToConnectorExpressionConverterTest.java | 59 ++++
.../converter/WriteConstraintExtractorTest.java | 199 +++++++++++
.../PositionDeleteRowLevelDmlTransformTest.java | 95 ++++++
.../plans/commands/RowLevelDmlCommandTest.java | 50 +++
.../test_iceberg_write_dml_modes_evolution.out | 5 +-
...t_iceberg_v3_row_lineage_dml_mode_matrix.groovy | 26 +-
...t_iceberg_delete_unpartitioned_evolution.groovy | 7 +-
21 files changed, 1629 insertions(+), 46 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java
index 6971d3763f5..4c8f56c58b5 100644
---
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java
@@ -100,6 +100,7 @@ import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeMap;
+import java.util.function.BooleanSupplier;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.function.ToLongBiFunction;
@@ -2149,6 +2150,21 @@ public class HiveConnectorMetadata implements
ConnectorMetadata {
// permissive here -- a throw would newly reject legal plain-hive
row-level DML.
}
+ /**
+ * A foreign Iceberg handle must use the sibling's whole-file
metadata-delete implementation. The Hive gateway
+ * itself has no data-file delete contract, so a real Hive handle keeps
the SPI default of falling back.
+ */
+ @Override
+ public OptionalLong deleteUsingMetadata(ConnectorSession session,
ConnectorTableHandle handle,
+ Optional<ConnectorExpression> predicate, String
boundWriteMetadataIdentity,
+ BooleanSupplier isCancelled) {
+ if (!(handle instanceof HiveTableHandle)) {
+ return siblingMetadata(session, handle).deleteUsingMetadata(
+ session, handle, predicate, boundWriteMetadataIdentity,
isCancelled);
+ }
+ return OptionalLong.empty();
+ }
+
/**
* Rejects an illegal static-partition column on {@code INSERT [OVERWRITE]
... PARTITION (col=val)}: the
* column must be a partition key of this (partitioned) table. Port of
upstream #65991's
diff --git
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java
index 928fd132811..d2f340f0f31 100644
---
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java
+++
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorMetadataSiblingDelegationTest.java
@@ -70,6 +70,7 @@ import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeSet;
+import java.util.function.BooleanSupplier;
import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -356,12 +357,12 @@ public class HiveConnectorMetadataSiblingDelegationTest {
}
@Test
- public void
everyAlterDdlAndValidateMethodForwardsAForeignHandleToTheSibling() {
+ public void everyForeignWriteOperationForwardsAForeignHandleToTheSibling()
{
HiveConnectorMetadata md = withSibling();
- // The 14 ALTER-DDL mutators + 2 write validators: a foreign
(iceberg-on-HMS) handle must divert, never
- // run the hive branch and never be cast. Change objects are null —
the guard fires on the handle type
- // before any param is touched.
+ // The 14 ALTER-DDL mutators + 2 write validators + metadata delete: a
foreign (iceberg-on-HMS) handle
+ // must divert, never run the hive branch and never be cast. Change
objects are null — the guard fires on
+ // the handle type before any param is touched.
md.renameTable(session, foreignHandle, "new");
md.setTableProperties(session, foreignHandle, Collections.emptyMap());
md.addColumn(session, foreignHandle, null, null);
@@ -387,9 +388,21 @@ public class HiveConnectorMetadataSiblingDelegationTest {
// Empty list on purpose: a foreign handle must forward REGARDLESS of
emptiness (the empty-early-return is
// hive-only) — this would fail if the empty check were placed before
the foreign-handle divert.
md.validateWritePartitionNames(session, foreignHandle,
Collections.emptyList());
+ ConnectorExpression predicate = new ConnectorColumnRef("id",
ConnectorType.of("INT"));
+ BooleanSupplier cancellation = () -> true;
+ OptionalLong deleted = md.deleteUsingMetadata(
+ session, foreignHandle, Optional.of(predicate),
"bound-generation", cancellation);
Assertions.assertEquals(RecordingSiblingMetadata.EXPECTED_WRITE_METHODS,
siblingMetadata.calls,
"every ALTER-DDL mutator + write validator must forward a
foreign handle to the sibling");
+
Assertions.assertEquals(OptionalLong.of(RecordingSiblingMetadata.SENTINEL_DELETED_ROWS),
deleted,
+ "metadata delete must return the sibling's result");
+ Assertions.assertSame(predicate, siblingMetadata.lastDeletePredicate,
+ "metadata delete must forward the exact predicate");
+ Assertions.assertEquals("bound-generation",
siblingMetadata.lastBoundWriteMetadataIdentity,
+ "metadata delete must forward the write-generation fence");
+ Assertions.assertSame(cancellation, siblingMetadata.lastCancellation,
+ "metadata delete must forward the cancellation callback");
}
@Test
@@ -451,6 +464,9 @@ public class HiveConnectorMetadataSiblingDelegationTest {
// plain-hive row-level DML / static-partition INSERTs.
md.validateRowLevelDmlMode(session, hive, null);
md.validateStaticPartitionColumns(session, hive,
Collections.emptyList());
+ Assertions.assertEquals(OptionalLong.empty(),
+ md.deleteUsingMetadata(session, hive, Optional.empty(), null,
() -> false),
+ "a real Hive handle must keep the metadata-delete fallback");
Assertions.assertEquals(0, siblingConnector.getMetadataCount,
"a hive handle must never build/consult the iceberg sibling
for ALTER-DDL / validate");
@@ -721,6 +737,7 @@ public class HiveConnectorMetadataSiblingDelegationTest {
static final ConnectorTransaction SIBLING_TXN = new
NoOpConnectorTransaction(4243L, "ICEBERG");
static final long SENTINEL_SIZE = 4242L;
static final long SENTINEL_SNAPSHOT_ID = 99L;
+ static final long SENTINEL_DELETED_ROWS = 31L;
static final List<Long> SENTINEL_FILE_SIZES =
Collections.unmodifiableList(Arrays.asList(7L, 11L));
static final List<ConnectorExpression> SIBLING_PREDICATES =
Collections.singletonList(
new ConnectorColumnRef("sibling-pred",
ConnectorType.of("STRING")));
@@ -747,9 +764,13 @@ public class HiveConnectorMetadataSiblingDelegationTest {
"modifyNestedColumn", "modifyColumnComment",
"createOrReplaceBranch", "createOrReplaceTag", "dropBranch",
"dropTag",
"addPartitionField", "dropPartitionField",
"replacePartitionField",
- "validateRowLevelDmlMode", "validateStaticPartitionColumns",
"validateWritePartitionNames"));
+ "validateRowLevelDmlMode", "validateStaticPartitionColumns",
"validateWritePartitionNames",
+ "deleteUsingMetadata"));
final List<String> calls = new ArrayList<>();
+ ConnectorExpression lastDeletePredicate;
+ String lastBoundWriteMetadataIdentity;
+ BooleanSupplier lastCancellation;
final Optional<FilterApplicationResult<ConnectorTableHandle>>
filterResult =
Optional.of(new FilterApplicationResult<>(SIBLING_HANDLE,
null, false));
@@ -1069,5 +1090,16 @@ public class HiveConnectorMetadataSiblingDelegationTest {
List<String> partitionNames) {
calls.add("validateWritePartitionNames");
}
+
+ @Override
+ public OptionalLong deleteUsingMetadata(ConnectorSession session,
ConnectorTableHandle handle,
+ Optional<ConnectorExpression> predicate, String
boundWriteMetadataIdentity,
+ BooleanSupplier isCancelled) {
+ calls.add("deleteUsingMetadata");
+ lastDeletePredicate = predicate.orElse(null);
+ lastBoundWriteMetadataIdentity = boundWriteMetadataIdentity;
+ lastCancellation = isCancelled;
+ return OptionalLong.of(SENTINEL_DELETED_ROWS);
+ }
}
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index 3864846d2b3..084ff2fe8fb 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -46,7 +46,14 @@ import org.apache.doris.connector.spi.handle.WriteOperation;
import org.apache.doris.connector.spi.mvcc.ConnectorMvccPartitionView;
import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot;
import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec;
+import org.apache.doris.connector.spi.pushdown.ConnectorAnd;
+import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
+import org.apache.doris.connector.spi.pushdown.ConnectorIn;
+import org.apache.doris.connector.spi.pushdown.ConnectorIsNull;
+import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
+import org.apache.doris.connector.spi.pushdown.ConnectorNot;
+import org.apache.doris.kerberos.HadoopAuthenticator;
import org.apache.doris.thrift.THiveTable;
import org.apache.doris.thrift.TIcebergTable;
import org.apache.doris.thrift.TTableDescriptor;
@@ -54,6 +61,13 @@ import org.apache.doris.thrift.TTableType;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.DeleteFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.MetadataTableUtils;
import org.apache.iceberg.PartitionField;
@@ -62,12 +76,24 @@ import org.apache.iceberg.RowLevelOperationMode;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionUtil;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Projections;
+import org.apache.iceberg.expressions.StrictMetricsEvaluator;
+import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.SnapshotUtil;
@@ -77,6 +103,9 @@ import org.apache.iceberg.view.ViewVersion;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -86,10 +115,12 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.Callable;
+import java.util.function.BooleanSupplier;
import java.util.function.Predicate;
/**
@@ -1968,6 +1999,308 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
}
}
+ /**
+ * Whole-file metadata delete for {@code DELETE} (the engine's fast path):
commit the delete as one atomic
+ * metadata operation that removes every matching data file, instead of
scanning rows and writing position
+ * delete files / deletion vectors. This is what Spark, StarRocks and
Trino do for whole-partition and
+ * whole-file predicates, and it is the only way to get rid of a data file
whose rows are all gone: a
+ * row-level Doris DELETE keeps the file (plus a redundant deletion
vector) in the snapshot, and every later
+ * scan still opens and reads it.
+ *
+ * <p>The predicate is lowered with {@link
IcebergPredicateConverter.Mode#REWRITE} — the strictly
+ * all-or-nothing matrix, because a dropped conjunct would widen the
delete — and then every candidate data
+ * file must be proven matched <i>in full</i> (strict partition projection
or file metrics), exactly the
+ * check Spark's {@code SparkTable.canDeleteUsingMetadata} and StarRocks's
+ * {@code IcebergMetadata.canDeleteUsingMetadata} perform. Only then is
the delete committed: without that
+ * proof iceberg's {@code deleteFromRowFilter} would fail the statement
with "Cannot delete file where some,
+ * but not all, rows match filter".</p>
+ *
+ * <p>Anything that cannot be proven returns empty and the engine runs the
ordinary row-level delete; a
+ * {@link ValidationException} raised at commit (the table moved between
the check and the commit, so a
+ * candidate file is no longer fully matched) does the same, because
nothing was committed.</p>
+ *
+ * <p>The returned row count is iceberg's {@code deleted-records} summary
minus the rows that the delete
+ * files removed with those data files had already hidden (mirroring
Trino's {@code executeDelete}), i.e. the
+ * rows this statement deleted. As in the engines above, the commit
applies the predicate to the table's
+ * current state: rows concurrently appended into a deleted partition are
deleted as well, whereas a
+ * row-level Doris DELETE would only delete the rows it scanned.</p>
+ */
+ @Override
+ public OptionalLong deleteUsingMetadata(ConnectorSession session,
ConnectorTableHandle handle,
+ Optional<ConnectorExpression> predicate, String
boundWriteMetadataIdentity,
+ BooleanSupplier isCancelled) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ String qualifiedName = iceHandle.getDbName() + "." +
iceHandle.getTableName();
+ try {
+ // withTable runs the configured cached-table cleanup in finally:
the fresh load below owns its
+ // FileIO on Glue / S3Tables / vended-credential REST catalogs, so
it must not outlive this call.
+ return executeAuthenticated(() -> catalogOps.withTable(
+ iceHandle.getDbName(), iceHandle.getTableName(),
+ table -> deleteUsingMetadata(table, session, predicate,
+ boundWriteMetadataIdentity, isCancelled)));
+ } catch (ValidationException e) {
+ // A racing writer turned a candidate file into a partial match
(or another writer already removed
+ // it): nothing was committed, so the row-level path still
produces the right result.
+ LOG.info("Metadata delete of {} is no longer whole-file ({});
falling back to the row-level delete.",
+ qualifiedName, e.getMessage());
+ return OptionalLong.empty();
+ } catch (Exception e) {
+ throw new DorisConnectorException("Failed to metadata-delete
Iceberg table " + qualifiedName
+ + ": " + e.getMessage(), e);
+ }
+ }
+
+ /** Body of the whole-file delete, run against a fresh table load inside
{@code withTable}. */
+ private OptionalLong deleteUsingMetadata(Table table, ConnectorSession
session,
+ Optional<ConnectorExpression> predicate, String
boundWriteMetadataIdentity,
+ BooleanSupplier isCancelled) {
+ if (isCancelled.getAsBoolean()) {
+ return OptionalLong.empty();
+ }
+ if (!(table instanceof BaseTable)
+ || ((BaseTable) table).operations().current().formatVersion()
< 2) {
+ // A v1 table has no row-level delete contract (the row-level path
fails with
+ // "must have format version 2 or higher for position deletes");
keep that behavior.
+ return OptionalLong.empty();
+ }
+ validateBoundWriteGeneration(table, boundWriteMetadataIdentity);
+ Expression deleteExpr = predicate.isPresent()
+ ? wholeFileDeleteExpression(table, session, predicate.get())
+ : Expressions.alwaysTrue();
+ if (deleteExpr == null) {
+ return OptionalLong.empty();
+ }
+ Snapshot base = table.currentSnapshot();
+ if (base == null) {
+ // Nothing to delete; leave the empty-table path to the row-level
plan.
+ return OptionalLong.empty();
+ }
+ // Candidate planning and delete-manifest reads fan out onto Iceberg's
shared worker pool, which needs the
+ // plugin Kerberos FileIO; the caller-thread executeAuthenticated
scope does not cover those workers.
+ Table scanTable = wrapTableForScan(table);
+ if (hasLiveDeleteFiles(scanTable, base)) {
+ // The affected-row count is only exact when nothing was hidden
before: legacy position deletes,
+ // equality deletes (a key is not a row) and stale-delete cleanup
all break the summary arithmetic.
+ LOG.info("Skipping metadata delete of {}: the table already
carries delete files.", table.name());
+ return OptionalLong.empty();
+ }
+ if (!matchesWholeFilesOnly(scanTable, deleteExpr) ||
isCancelled.getAsBoolean()) {
+ return OptionalLong.empty();
+ }
+ // The load above is a metadata snapshot; re-read it immediately
before committing so a drop/recreate or
+ // schema rebind between planning and commit aborts instead of hitting
the new table, and so a
+ // concurrently added delete file cannot slip into the affected-row
summary.
+ table.refresh();
+ validateBoundWriteGeneration(table, boundWriteMetadataIdentity);
+ Snapshot refreshed = table.currentSnapshot();
+ if (refreshed == null || hasLiveDeleteFiles(scanTable, refreshed)) {
+ return OptionalLong.empty();
+ }
+ long refreshedSnapshotId = refreshed.snapshotId();
+
+ // Use a transaction rather than table.newDelete(): its ops carry the
plugin Kerberos FileIO for Iceberg's
+ // worker-pool manifest work, and its table view keeps the metadata it
committed, so currentSnapshot()
+ // below is this delete's own snapshot even if another writer
committed meanwhile. The bound identity
+ // and required snapshot fence the final CAS and every retry replay: a
concurrent commit aborts the
+ // metadata delete (the row-level fallback then reports an exact
affected-row count).
+ Transaction transaction = IcebergConnectorTransaction.openTransaction(
+ table, context, null, false, boundWriteMetadataIdentity,
refreshedSnapshotId);
+ DeleteFiles deleteFiles =
transaction.newDelete().deleteFromRowFilter(deleteExpr);
+ if (isCancelled.getAsBoolean()) {
+ return OptionalLong.empty();
+ }
+ deleteFiles.commit();
+ if (isCancelled.getAsBoolean()) {
+ // The snapshot is staged on the transaction only, so dropping it
leaves the table unchanged (the
+ // staged metadata files are unreferenced and are reclaimed by the
existing orphan-file cleanup).
+ LOG.info("Metadata delete of {} was cancelled before the remote
commit.", table.name());
+ return OptionalLong.empty();
+ }
+ transaction.commitTransaction();
+ Snapshot committed = transaction.table().currentSnapshot();
+ if (committed == null || committed.snapshotId() ==
refreshedSnapshotId) {
+ // The predicate matched nothing; DeleteFiles does not create an
empty snapshot.
+ return OptionalLong.of(0L);
+ }
+ return OptionalLong.of(deletedRowCount(committed));
+ }
+
+ /**
+ * Authenticated view of {@code table} for candidate-scan manifest IO —
the metadata-delete mirror of
+ * {@code IcebergScanPlanProvider.wrapTableForScan}. Non-Kerberos catalogs
and offline fakes get the table
+ * unchanged; the wrapper delegates reads to the same underlying
operations.
+ */
+ private Table wrapTableForScan(Table table) {
+ if (!(context instanceof TcclPinningConnectorContext) || !(table
instanceof BaseTable)) {
+ return table;
+ }
+ HadoopAuthenticator auth = ((TcclPinningConnectorContext)
context).getPluginAuthenticator();
+ if (auth == null) {
+ return table;
+ }
+ BaseTable baseTable = (BaseTable) table;
+ TableOperations rawOps = baseTable.operations();
+ return new BaseTable(new IcebergAuthenticatedTableOperations(
+ rawOps, new IcebergAuthenticatedFileIO(rawOps.io(), auth)),
+ table.name(), baseTable.reporter());
+ }
+
+ /** Aborts a stale statement when the table was dropped/recreated (or
rebound) after it was planned. */
+ private static void validateBoundWriteGeneration(Table table, String
boundWriteMetadataIdentity) {
+ if (boundWriteMetadataIdentity != null
+ &&
!boundWriteMetadataIdentity.equals(IcebergWritePlanProvider.writeMetadataIdentity(table)))
{
+ throw new DorisConnectorException(
+ "Iceberg write metadata changed after the write was bound;
retry the statement");
+ }
+ }
+
+ /**
+ * Lowers the statement predicate to an iceberg expression with the
strictly all-or-nothing
+ * {@link IcebergPredicateConverter.Mode#REWRITE} matrix, or {@code null}
when it is not fully representable
+ * (the caller then falls back to the row-level delete). A bare {@code
true} — an unfiltered
+ * {@code DELETE FROM t} — is {@link Expressions#alwaysTrue()}; the
rewrite matrix rejects bare literals, so
+ * that case is handled here.
+ *
+ * <p>{@code forTable} also carries the schema/type history: a DATE field
promoted to TIMESTAMP_NS makes old
+ * four-byte bounds unreadable, and such predicates must fall back instead
of pruning on mis-decoded metrics.</p>
+ */
+ private Expression wholeFileDeleteExpression(Table table, ConnectorSession
session,
+ ConnectorExpression predicate) {
+ if (!isExactDeletePredicate(predicate)) {
+ // Negation other than IS NOT NULL (or a negated IN) changes NULL
semantics: iceberg's notEqual/notIn
+ // match NULL rows while SQL's NOT(UNKNOWN) does not, so such a
predicate must not drive this path.
+ return null;
+ }
+ if (predicate instanceof ConnectorLiteral) {
+ return Boolean.TRUE.equals(((ConnectorLiteral)
predicate).getValue())
+ ? Expressions.alwaysTrue() : null;
+ }
+ ZoneId zone = IcebergTimeUtils.resolveSessionZone(session);
+ List<Expression> predicates = IcebergPredicateConverter.forTable(
+ table, table.schema(), zone,
IcebergPredicateConverter.Mode.REWRITE).convert(predicate);
+ // Same guard as RewriteDataFilePlanner: that matrix drops an
unconvertible top-level conjunct instead of
+ // failing, and a dropped conjunct would widen the delete here.
+ if (predicates.isEmpty() || predicates.size() <
countTopLevelConjuncts(predicate)) {
+ return null;
+ }
+ Expression combined = null;
+ for (Expression converted : predicates) {
+ combined = combined == null ? converted :
Expressions.and(combined, converted);
+ }
+ return combined;
+ }
+
+ /** Null-safe lowering guard for the destructive path; see {@link
#wholeFileDeleteExpression}. */
+ private static boolean isExactDeletePredicate(ConnectorExpression
expression) {
+ if (expression instanceof ConnectorNot) {
+ return ((ConnectorNot) expression).getOperand() instanceof
ConnectorIsNull;
+ }
+ if (expression instanceof ConnectorIn && ((ConnectorIn)
expression).isNegated()) {
+ return false;
+ }
+ if (expression instanceof ConnectorComparison
+ && ((ConnectorComparison) expression).getOperator() ==
ConnectorComparison.Operator.NE) {
+ return false;
+ }
+ for (ConnectorExpression child : expression.getChildren()) {
+ if (!isExactDeletePredicate(child)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /** Number of top-level conjuncts of a neutral predicate (a top-level
AND's arity, else 1). */
+ private static int countTopLevelConjuncts(ConnectorExpression predicate) {
+ return predicate instanceof ConnectorAnd
+ ? ((ConnectorAnd) predicate).getConjuncts().size() : 1;
+ }
+
+ /**
+ * Whether every candidate data file of {@code deleteExpr} is matched in
full, so that iceberg's
+ * {@link DeleteFiles#deleteFromRowFilter} drops files instead of
rejecting a partially matched one. Whole
+ * partitions short-circuit ({@link ExpressionUtil#selectsPartitions}
holds for every spec, so partition
+ * evolution is covered); otherwise each candidate file — as enumerated by
the very scan the engine would
+ * run — must satisfy its own spec's strict partition projection or the
strict metrics evaluator.
+ *
+ */
+ private static boolean matchesWholeFilesOnly(Table table, Expression
deleteExpr) {
+ boolean caseSensitive = false;
+ if (ExpressionUtil.selectsPartitions(deleteExpr, table,
caseSensitive)) {
+ return true;
+ }
+ TableScan scan = table.newScan()
+ .filter(deleteExpr)
+ .caseSensitive(caseSensitive)
+ .includeColumnStats()
+ .ignoreResiduals();
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ Map<Integer, Evaluator> evaluators = new HashMap<>();
+ StrictMetricsEvaluator metricsEvaluator =
+ new StrictMetricsEvaluator(table.schema(), deleteExpr,
caseSensitive);
+ for (FileScanTask task : tasks) {
+ DataFile file = task.file();
+ PartitionSpec spec = task.spec();
+ Evaluator evaluator =
evaluators.computeIfAbsent(spec.specId(), specId -> new Evaluator(
+ spec.partitionType(), Projections.strict(spec,
caseSensitive).project(deleteExpr)));
+ if (!evaluator.eval(file.partition()) &&
!metricsEvaluator.eval(file)) {
+ return false;
+ }
+ }
+ return true;
+ } catch (IOException e) {
+ LOG.warn("Failed to evaluate whole-file delete candidates for {}:
{}",
+ table.name(), e.getMessage());
+ return false;
+ }
+ }
+
+ /**
+ * Whether the snapshot has any live delete file. Both callers only need
presence, so stop at the first live
+ * entry instead of materializing every delete file of every unrelated
partition.
+ */
+ private static boolean hasLiveDeleteFiles(Table table, Snapshot snapshot) {
+ try {
+ for (ManifestFile manifest : snapshot.deleteManifests(table.io()))
{
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(),
table.specs())) {
+ if (reader.iterator().hasNext()) {
+ return true;
+ }
+ }
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ return false;
+ }
+
+ /**
+ * Rows deleted by a metadata delete, derived from this delete's own
committed snapshot. The pre-commit gate
+ * requires that no delete file existed on the table (checked again after
the refresh), so no hidden rows need
+ * subtracting; a deletion vector added by a concurrent row-level delete
between that check and our commit is
+ * still exact because its record count is its cardinality.
Equality-delete records are keys, never rows, so
+ * they are not subtracted (a concurrent equality delete only logs a
warning).
+ */
+ private static long deletedRowCount(Snapshot committed) {
+ if (committed == null || committed.summary() == null) {
+ return 0L;
+ }
+ Map<String, String> summary = committed.summary();
+ long deletedRecords =
Long.parseLong(summary.getOrDefault(SnapshotSummary.DELETED_RECORDS_PROP, "0"));
+ long removedPosDeletes =
+
Long.parseLong(summary.getOrDefault(SnapshotSummary.REMOVED_POS_DELETES_PROP,
"0"));
+ long removedEqDeletes =
+
Long.parseLong(summary.getOrDefault(SnapshotSummary.REMOVED_EQ_DELETES_PROP,
"0"));
+ if (removedEqDeletes > 0) {
+ // A concurrent writer removed an equality delete between the gate
and the commit: its records are
+ // keys, so the exact live-row count is no longer derivable.
+ LOG.warn("Metadata delete {} removed equality deletes;
affected-row count may be approximate: {}.",
+ committed.snapshotId(), committed.summary());
+ }
+ return Math.max(0L, deletedRecords - removedPosDeletes);
+ }
+
/**
* Rejects an illegal static-partition column on {@code INSERT [OVERWRITE]
... PARTITION (col=val)}: the name
* must be the SOURCE column of an <em>identity</em> partition field of
this (partitioned) table.
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 ddfa9d9fae1..ad17c6d353e 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
@@ -60,6 +60,7 @@ import org.apache.iceberg.TableScan;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.Transactions;
import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.io.CloseableIterable;
@@ -356,6 +357,17 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
* {@code io()} differs. Non-Kerberos catalogs ({@code
getPluginAuthenticator() == null}) keep the legacy path.
*/
private Transaction openTransaction(Table loaded, boolean
requireCurrentPartitionSpec) {
+ return openTransaction(loaded, context, writeSchemaContext,
requireCurrentPartitionSpec, null, null);
+ }
+
+ /**
+ * Transaction-opening seam shared by the row-level write path and the
whole-file metadata delete: wraps the
+ * table's {@link TableOperations#io()} in the plugin Kerberos {@code
doAs} so manifest work fanned onto
+ * Iceberg's shared worker pool still authenticates, and optionally
installs the pinned write-schema validator.
+ */
+ static Transaction openTransaction(Table loaded, ConnectorContext context,
+ IcebergWriteSchemaContext writeSchemaContext, boolean
requireCurrentPartitionSpec,
+ String boundWriteMetadataIdentity, Long requiredSnapshotId) {
HadoopAuthenticator auth = context instanceof
TcclPinningConnectorContext
? ((TcclPinningConnectorContext)
context).getPluginAuthenticator()
: null;
@@ -370,9 +382,10 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
FileIO authIo = new IcebergAuthenticatedFileIO(realOps.io(), auth);
writeOps = new IcebergAuthenticatedTableOperations(realOps,
authIo);
}
- if (writeSchemaContext != null) {
+ if (writeSchemaContext != null || boundWriteMetadataIdentity != null
|| requiredSnapshotId != null) {
writeOps = new ValidatingTableOperations(
- writeOps, writeSchemaContext, requireCurrentPartitionSpec,
loaded.name());
+ writeOps, writeSchemaContext, requireCurrentPartitionSpec,
loaded.name(),
+ boundWriteMetadataIdentity, requiredSnapshotId);
}
if (writeOps == realOps) {
return loaded.newTransaction();
@@ -399,14 +412,23 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
private final IcebergWriteSchemaContext writeSchemaContext;
private final boolean requireCurrentPartitionSpec;
private final String tableName;
+ // Non-null for a whole-file metadata delete: fences the final CAS
(including retry replays) against a
+ // same-name table replacement or a schema/partition rebind that
happened after planning.
+ private final String boundWriteMetadataIdentity;
+ // Non-null for a whole-file metadata delete: requires the commit to
apply to the exact snapshot the
+ // affected-row preflight inspected, so a concurrent commit aborts
instead of changing the removed set.
+ private final Long requiredSnapshotId;
private ValidatingTableOperations(TableOperations delegate,
IcebergWriteSchemaContext writeSchemaContext,
- boolean requireCurrentPartitionSpec, String tableName) {
+ boolean requireCurrentPartitionSpec, String tableName,
+ String boundWriteMetadataIdentity, Long requiredSnapshotId) {
this.delegate = delegate;
this.writeSchemaContext = writeSchemaContext;
this.requireCurrentPartitionSpec = requireCurrentPartitionSpec;
this.tableName = tableName;
+ this.boundWriteMetadataIdentity = boundWriteMetadataIdentity;
+ this.requiredSnapshotId = requiredSnapshotId;
}
@Override
@@ -416,7 +438,9 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
@Override
public TableMetadata refresh() {
- return delegate.refresh();
+ TableMetadata refreshed = delegate.refresh();
+ checkRequiredSnapshot(refreshed);
+ return refreshed;
}
@Override
@@ -434,11 +458,32 @@ public class IcebergConnectorTransaction implements
ConnectorTransaction, Rewrit
new StaticTableOperations(
currentMetadata, delegate.io(),
delegate.locationProvider()),
tableName);
- writeSchemaContext.validateCurrentSchema(
- currentTable, requireCurrentPartitionSpec);
+ if (writeSchemaContext != null) {
+ writeSchemaContext.validateCurrentSchema(
+ currentTable, requireCurrentPartitionSpec);
+ }
+ if (boundWriteMetadataIdentity != null
+ && !boundWriteMetadataIdentity.equals(
+
IcebergWritePlanProvider.writeMetadataIdentity(currentTable))) {
+ throw new DorisConnectorException(
+ "Iceberg write metadata changed after the write was
bound; retry the statement");
+ }
+ checkRequiredSnapshot(currentMetadata);
delegate.commit(base, metadata);
}
+ /** Aborts a retry replay that would apply the delete to a snapshot
other than the preflighted one. */
+ private void checkRequiredSnapshot(TableMetadata metadata) {
+ if (requiredSnapshotId == null) {
+ return;
+ }
+ Snapshot current = metadata.currentSnapshot();
+ if (current == null || current.snapshotId() != requiredSnapshotId)
{
+ throw new ValidationException(
+ "Table %s changed since the metadata delete preflight;
retry the statement", tableName);
+ }
+ }
+
@Override
public FileIO io() {
return delegate.io();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergMetadataDeleteTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergMetadataDeleteTest.java
new file mode 100644
index 00000000000..c6b08bdcdf0
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergMetadataDeleteTest.java
@@ -0,0 +1,378 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.connector.iceberg;
+
+import org.apache.doris.connector.spi.ConnectorSession;
+import org.apache.doris.connector.spi.ConnectorType;
+import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
+import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef;
+import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
+import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
+
+import com.google.common.collect.Iterables;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+
+/**
+ * Pins {@link IcebergConnectorMetadata#deleteUsingMetadata}: the whole-file
{@code DELETE} fast path that
+ * Spark / StarRocks / Trino use for whole-partition and whole-file predicates.
+ *
+ * <p>Mirrors the no-Mockito, real-{@link InMemoryCatalog} style of {@code
IcebergConnectorTransactionTest}: the
+ * delete is committed against a real iceberg table, but — unlike a row-level
delete — it never reads a data
+ * file, so the fixtures only need data-file metadata (path + record count),
not real parquet files.</p>
+ */
+public class IcebergMetadataDeleteTest {
+
+ private static final TableIdentifier TABLE_ID = TableIdentifier.of("db1",
"t1");
+ private static final ConnectorTableHandle HANDLE = new
IcebergTableHandle("db1", "t1");
+ private static final ConnectorSession SESSION = new FakeSession("UTC");
+
+ @Test
+ public void wholePartitionDeleteDropsMatchingFilesOnly() {
+ InMemoryCatalog catalog = freshCatalog();
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("region").build();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA, spec,
props("format-version", "2"));
+ table.newAppend()
+ .appendFile(partitionedDataFile(spec,
"s3://b/db1/t1/region=us/a.parquet", 10L, "region=us"))
+ .appendFile(partitionedDataFile(spec,
"s3://b/db1/t1/region=us/b.parquet", 10L, "region=us"))
+ .appendFile(partitionedDataFile(spec,
"s3://b/db1/t1/region=eu/c.parquet", 20L, "region=eu"))
+ .commit();
+
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(SESSION,
HANDLE, Optional.of(
+ new ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("region",
ConnectorType.of("STRING")), ConnectorLiteral.ofString("us"))),
+ null, () -> false);
+
+ Assertions.assertEquals(OptionalLong.of(20L), deleted,
+ "the two region=us files (10 records each) are the deleted
rows");
+
Assertions.assertEquals(Collections.singletonList("s3://b/db1/t1/region=eu/c.parquet"),
+ dataFilePaths(reload(catalog)),
+ "a whole-partition delete removes only the matching data
files");
+ Assertions.assertEquals(0, liveDeleteFileCount(reload(catalog)),
+ "a metadata delete writes no delete files (no position
deletes, no deletion vectors)");
+ }
+
+ @Test
+ public void unfilteredDeleteRemovesEveryFile() {
+ InMemoryCatalog catalog = freshCatalog();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA, spec,
props("format-version", "2"));
+ table.newAppend()
+ .appendFile(dataFile(spec, "s3://b/db1/t1/a.parquet", 10L))
+ .appendFile(dataFile(spec, "s3://b/db1/t1/b.parquet", 5L))
+ .commit();
+
+ // The engine passes no predicate for "DELETE FROM t": every row of
the table is deleted.
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(SESSION,
HANDLE, Optional.empty(), null, () -> false);
+
+ Assertions.assertEquals(OptionalLong.of(15L), deleted);
+ Assertions.assertTrue(dataFilePaths(reload(catalog)).isEmpty(),
+ "an unfiltered metadata delete drops every file");
+ }
+
+ @Test
+ public void partialMatchFallsBackWithoutCommitting() {
+ InMemoryCatalog catalog = freshCatalog();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA, spec,
props("format-version", "2"));
+ // No column metrics on the file: nothing proves that ALL of its rows
satisfy `id > 5`.
+ table.newAppend().appendFile(dataFile(spec, "s3://b/db1/t1/a.parquet",
10L)).commit();
+ long snapshotBefore = table.currentSnapshot().snapshotId();
+
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(SESSION,
HANDLE, Optional.of(
+ new ConnectorComparison(ConnectorComparison.Operator.GT,
+ new ConnectorColumnRef("id", ConnectorType.of("INT")),
ConnectorLiteral.ofLong(5L))),
+ null, () -> false);
+
+ Assertions.assertFalse(deleted.isPresent(),
+ "a candidate file that is only partially matched must fall
back to the row-level delete");
+ Assertions.assertEquals(snapshotBefore,
reload(catalog).currentSnapshot().snapshotId(),
+ "the fallback must not commit anything");
+ }
+
+ @Test
+ public void unpushablePredicateFallsBack() {
+ InMemoryCatalog catalog = freshCatalog();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA,
PartitionSpec.unpartitioned(),
+ props("format-version", "2"));
+ table.newAppend().appendFile(dataFile(PartitionSpec.unpartitioned(),
"s3://b/db1/t1/a.parquet", 10L))
+ .commit();
+ long snapshotBefore = table.currentSnapshot().snapshotId();
+
+ // A column the table does not have: the strict matrix cannot lower
it, and dropping it would delete
+ // every row of every file.
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(SESSION,
HANDLE, Optional.of(
+ new ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("no_such_column",
ConnectorType.of("BIGINT")), ConnectorLiteral.ofLong(1L))),
+ null, () -> false);
+
+ Assertions.assertFalse(deleted.isPresent(), "an unliftable predicate
must not be treated as always-true");
+ Assertions.assertEquals(snapshotBefore,
reload(catalog).currentSnapshot().snapshotId());
+ }
+
+ @Test
+ public void existingDeleteFilesFallBackToTheRowLevelPath() {
+ InMemoryCatalog catalog = freshCatalog();
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("region").build();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA, spec,
props("format-version", "3"));
+ DataFile dataFile = partitionedDataFile(spec,
"s3://b/db1/t1/region=us/a.parquet", 10L, "region=us");
+ table.newAppend().appendFile(dataFile).commit();
+ // Three of the ten rows are already deleted by a deletion vector for
that data file.
+ table.newRowDelta().addDeletes(deletionVector(spec,
"s3://b/db1/t1/dv.puffin", dataFile, 3L)).commit();
+ Assertions.assertEquals(1, attachedDeleteFileCount(table,
dataFile.path().toString()),
+ "the deletion vector must be attached to its data file");
+ long snapshotBefore = table.currentSnapshot().snapshotId();
+
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(SESSION,
HANDLE, Optional.of(
+ new ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("region",
ConnectorType.of("STRING")), ConnectorLiteral.ofString("us"))),
+ null, () -> false);
+
+ Assertions.assertFalse(deleted.isPresent(),
+ "the affected-row count cannot be derived exactly once the
file is hidden by a delete file");
+ Assertions.assertEquals(snapshotBefore,
reload(catalog).currentSnapshot().snapshotId(),
+ "the fallback must not commit");
+
Assertions.assertEquals(Collections.singletonList(dataFile.path().toString()),
+ dataFilePaths(reload(catalog)));
+ Assertions.assertEquals(1, attachedDeleteFileCount(reload(catalog),
dataFile.path().toString()));
+ }
+
+ @Test
+ public void cancelledDeleteDoesNotCommit() {
+ InMemoryCatalog catalog = freshCatalog();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA, spec,
props("format-version", "2"));
+ table.newAppend()
+ .appendFile(dataFile(spec, "s3://b/db1/t1/a.parquet", 10L))
+ .appendFile(dataFile(spec, "s3://b/db1/t1/b.parquet", 5L))
+ .commit();
+ long snapshotBefore = table.currentSnapshot().snapshotId();
+
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(
+ SESSION, HANDLE, Optional.empty(), null, () -> true);
+
+ Assertions.assertFalse(deleted.isPresent(), "a cancelled delete must
not commit");
+ Assertions.assertEquals(snapshotBefore,
reload(catalog).currentSnapshot().snapshotId());
+ Assertions.assertEquals(2, dataFilePaths(reload(catalog)).size());
+ }
+
+ @Test
+ public void v1TableKeepsTheFormatVersionContract() {
+ InMemoryCatalog catalog = freshCatalog();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA, spec,
props("format-version", "1"));
+ table.newAppend().appendFile(dataFile(spec, "s3://b/db1/t1/a.parquet",
10L)).commit();
+ long snapshotBefore = table.currentSnapshot().snapshotId();
+
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(
+ SESSION, HANDLE, Optional.empty(), null, () -> false);
+
+ Assertions.assertFalse(deleted.isPresent(),
+ "v1 tables must fall back so the row-level path keeps
rejecting row-level deletes");
+ Assertions.assertEquals(snapshotBefore,
reload(catalog).currentSnapshot().snapshotId());
+ }
+
+ private static IcebergConnectorMetadata metadata(InMemoryCatalog catalog) {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = reload(catalog);
+ return new IcebergConnectorMetadata(ops,
IcebergCatalogProperties.of(Collections.emptyMap()),
+ new RecordingConnectorContext());
+ }
+
+ /** Re-reads the table so the assertions observe the snapshot the metadata
delete committed. */
+ private static Table reload(InMemoryCatalog catalog) {
+ return catalog.loadTable(TABLE_ID);
+ }
+
+ private static List<String> dataFilePaths(Table table) {
+ List<String> paths = new ArrayList<>();
+ try (CloseableIterable<FileScanTask> tasks =
table.newScan().planFiles()) {
+ for (FileScanTask task : tasks) {
+ paths.add(task.file().path().toString());
+ }
+ } catch (Exception e) {
+ throw new AssertionError(e);
+ }
+ Collections.sort(paths);
+ return paths;
+ }
+
+ /** Live delete-file entries in the current snapshot (an emptied manifest
may legitimately remain). */
+ private static int liveDeleteFileCount(Table table) {
+ Snapshot snapshot = table.currentSnapshot();
+ if (snapshot == null) {
+ return 0;
+ }
+ int live = 0;
+ for (ManifestFile manifest : snapshot.deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(),
table.specs())) {
+ live += Iterables.size(reader);
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ }
+ }
+ return live;
+ }
+
+ private static final Schema SCHEMA = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "region", Types.StringType.get()));
+
+ private static InMemoryCatalog freshCatalog() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ return catalog;
+ }
+
+ private static Map<String, String> props(String... kv) {
+ Map<String, String> properties = new HashMap<>();
+ for (int i = 0; i + 1 < kv.length; i += 2) {
+ properties.put(kv[i], kv[i + 1]);
+ }
+ return properties;
+ }
+
+ private static DataFile dataFile(PartitionSpec spec, String path, long
records) {
+ return DataFiles.builder(spec)
+ .withPath(path)
+ .withFileSizeInBytes(1024)
+ .withRecordCount(records)
+ .withFormat(FileFormat.PARQUET)
+ .build();
+ }
+
+ private static DataFile partitionedDataFile(PartitionSpec spec, String
path, long records,
+ String partitionPath) {
+ return DataFiles.builder(spec)
+ .withPath(path)
+ .withFileSizeInBytes(1024)
+ .withRecordCount(records)
+ .withFormat(FileFormat.PARQUET)
+ .withPartitionPath(partitionPath)
+ .build();
+ }
+
+ /** A v3 deletion vector for {@code referencedDataFile} that already hides
{@code cardinality} rows. */
+ private static DeleteFile deletionVector(PartitionSpec spec, String
puffin, DataFile referencedDataFile,
+ long cardinality) {
+ return FileMetadata.deleteFileBuilder(spec)
+ .ofPositionDeletes()
+ .withPath(puffin)
+ .withFormat(FileFormat.PUFFIN)
+ .withFileSizeInBytes(128L)
+ .withRecordCount(cardinality)
+ // The deletion vector must carry the data file's partition
tuple; otherwise iceberg treats the
+ // pair as corrupt metadata and a scan cannot attach the
vector to its data file.
+ .withPartition(referencedDataFile.partition())
+ .withReferencedDataFile(referencedDataFile.path().toString())
+ .withContentOffset(0L)
+ .withContentSizeInBytes(64L)
+ .build();
+ }
+
+ private static int attachedDeleteFileCount(Table table, String
dataFilePath) {
+ try (CloseableIterable<FileScanTask> tasks =
table.newScan().planFiles()) {
+ for (FileScanTask task : tasks) {
+ if (task.file().path().toString().equals(dataFilePath)) {
+ return task.deletes().size();
+ }
+ }
+ } catch (IOException e) {
+ throw new AssertionError(e);
+ }
+ return -1;
+ }
+
+ /** Minimal {@link ConnectorSession} exposing a time zone
(partition-timestamp parsing); no Mockito. */
+ private static final class FakeSession implements ConnectorSession {
+ private final String timeZone;
+
+ FakeSession(String timeZone) {
+ this.timeZone = timeZone;
+ }
+
+ @Override
+ public String getQueryId() {
+ return "q";
+ }
+
+ @Override
+ public String getUser() {
+ return "u";
+ }
+
+ @Override
+ public String getTimeZone() {
+ return timeZone;
+ }
+
+ @Override
+ public String getLocale() {
+ return "en_US";
+ }
+
+ @Override
+ public <T> T getProperty(String name, Class<T> type) {
+ return null;
+ }
+
+ @Override
+ public long getCatalogId() {
+ return 0;
+ }
+
+ @Override
+ public String getCatalogName() {
+ return "test";
+ }
+
+ @Override
+ public Map<String, String> getCatalogProperties() {
+ return Collections.emptyMap();
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorWriteOps.java
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorWriteOps.java
index ef47bd92875..92b81a4a09b 100644
---
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorWriteOps.java
+++
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorWriteOps.java
@@ -20,8 +20,12 @@ package org.apache.doris.connector.spi;
import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
import org.apache.doris.connector.spi.handle.ConnectorTransaction;
import org.apache.doris.connector.spi.handle.WriteOperation;
+import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
import java.util.List;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.function.BooleanSupplier;
/**
* Write (DML) operations that a connector may support.
@@ -55,6 +59,45 @@ public interface ConnectorWriteOps {
// default: no per-table mode constraint
}
+ /**
+ * Attempts to satisfy a {@code DELETE} entirely by dropping whole data
files, without reading any rows
+ * ("metadata delete"): a remote atomic commit that removes the files
whose rows must all disappear. Spark,
+ * StarRocks and Trino all take this path for whole-partition and
whole-file predicates, and it is what makes
+ * {@code DELETE FROM t} / {@code DELETE FROM t WHERE <partition>} cheap.
+ *
+ * <p>The engine calls this before executing the row-level
(position-delete) plan of a single-table
+ * {@code DELETE}; a returned value means the connector already committed
the delete, so the engine skips the
+ * scan and the sink entirely. Returning empty is always legal and means
"not applicable": the engine then
+ * falls back to the ordinary row-level delete.</p>
+ *
+ * <p><b>Contract on {@code predicate}.</b> The engine only calls this
method when it could extract the
+ * statement's predicate <i>exactly</i> (every conjunct, target table
only, no partition-clause scope), so an
+ * implementation may treat it as complete. That matters because a wider
predicate would drop files that still
+ * hold live rows. The current engine represents an unfiltered DELETE as a
present true literal; implementations
+ * may also accept {@code Optional.empty()} as always true. An
implementation MUST itself verify that the
+ * predicate only ever removes whole files (e.g. by checking file
statistics / partition projections) and fall
+ * back — return empty, never a partial delete — when any candidate file
is only partially matched.</p>
+ *
+ * <p>The default returns empty, so connectors without a whole-file delete
path are unaffected.</p>
+ *
+ * @param session the current session
+ * @param handle the target table handle
+ * @param predicate the statement's complete delete predicate over the
target's own columns; the engine sends a
+ * true literal for an unfiltered delete
+ * @param boundWriteMetadataIdentity the write-metadata generation the
statement was planned against, or
+ * {@code null} when the engine has none; a connector
whose writes are fenced by a table
+ * generation MUST reject a mismatch before committing
+ * @param isCancelled checked before the remote commit; returning {@code
true} must abort without committing
+ * (the engine then reports the retained cancellation
reason)
+ * @return the number of deleted rows when the delete was fully executed
as a metadata delete; empty to fall
+ * back to row-level DML
+ */
+ default OptionalLong deleteUsingMetadata(ConnectorSession session,
ConnectorTableHandle handle,
+ Optional<ConnectorExpression> predicate, String
boundWriteMetadataIdentity,
+ BooleanSupplier isCancelled) {
+ return OptionalLong.empty();
+ }
+
/**
* Validates that every column named in a static-partition spec ({@code
INSERT [OVERWRITE] ... PARTITION
* (col=val)}) is a legal static-partition target on {@code handle},
throwing a {@link DorisConnectorException}
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
index cce4173053f..83944f58185 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
+++
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
@@ -88,9 +88,9 @@ public class ConnectorPluginSurfaceTest {
Assertions.assertNotNull(in, "missing connector plugin API version
resource");
version.load(in);
}
- // Major 19 adds
ConnectorColumn#isSourceNullable()/withSourceNullable(boolean) - the
nullability the
- // source declares, kept separate from the nullability the engine
reads the column with.
- Assertions.assertEquals("19.0", version.getProperty("api.version"));
+ // Major 19 adds
ConnectorColumn#isSourceNullable()/withSourceNullable(boolean); major 20 adds
+ // deleteUsingMetadata to the connector write surface.
+ Assertions.assertEquals("20.0", version.getProperty("api.version"));
}
/** Root entry points plus provider/handle types returned to connector
plugins. */
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt
index 80057cd9be8..8c48399c75f 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt
+++
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-metadata-methods.txt
@@ -17,6 +17,7 @@
createOrReplaceBranch(org.apache.doris.connector.spi.ConnectorSession,org.apache
createOrReplaceTag(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.handle.ConnectorTableHandle,org.apache.doris.connector.spi.ddl.TagChange)
createTable(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.ddl.ConnectorCreateTableRequest)
databaseExists(org.apache.doris.connector.spi.ConnectorSession,java.lang.String)
+deleteUsingMetadata(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.handle.ConnectorTableHandle,java.util.Optional,java.lang.String,java.util.function.BooleanSupplier)
dropBranch(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.handle.ConnectorTableHandle,org.apache.doris.connector.spi.ddl.DropRefChange)
dropColumn(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.handle.ConnectorTableHandle,java.lang.String)
dropDatabase(org.apache.doris.connector.spi.ConnectorSession,java.lang.String,boolean,boolean)
diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml
index 50adfde9967..d1d8f2fb4c4 100644
--- a/fe/fe-connector/pom.xml
+++ b/fe/fe-connector/pom.xml
@@ -55,7 +55,7 @@ under the License.
of the latter two means bumping this property as well (and
fe-extension-spi means bumping
all five families).
-->
- <connector.plugin.api.version>19.0</connector.plugin.api.version>
+ <connector.plugin.api.version>20.0</connector.plugin.api.version>
</properties>
<modules>
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverter.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverter.java
index ad3db2b4869..7e5dc31130e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverter.java
@@ -57,8 +57,9 @@ import java.util.List;
* IcebergNereidsUtils.convertNereidsToIcebergExpression}): {@code And}/{@code
Or}/{@code Not}, the five
* comparisons ({@code EqualTo}/{@code GreaterThan}/{@code
GreaterThanEqual}/{@code LessThan}/{@code
* LessThanEqual}), {@code InPredicate}, {@code IsNull}, {@code Between}.
Comparisons require a bare
- * {@link SlotReference} on one side and a {@link Literal} on the other
(operands are normalised to
- * column-on-left, the operator unchanged — mirroring legacy {@code
convertNereidsBinaryPredicate}).</p>
+ * {@link SlotReference} on one side and a {@link Literal} on the other. Loose
conversion preserves the
+ * legacy operator behavior; strict conversion mirrors the operator when it
normalises operands to
+ * column-on-left.</p>
*
* <p><b>Anything else yields {@code null}</b> — {@code NullSafeEqual}, {@code
Cast}-wrapped columns,
* column-to-column comparisons, bare literals, etc. The legacy conflict path
drops exactly these. Dropping a
@@ -83,26 +84,49 @@ public final class NereidsToConnectorExpressionConverter {
/** Convert a Nereids predicate to a neutral {@link ConnectorExpression},
or {@code null} if not
* representable in the legacy iceberg conflict matrix (a safe
over-approximation). */
public static ConnectorExpression convert(Expression expr) {
+ return convert(expr, false);
+ }
+
+ /**
+ * Strict variant of {@link #convert}: it refuses to <i>drop</i> anything.
The loose variant is a safe
+ * over-approximation for its consumers (a dropped {@code AND} arm only
widens a conflict-detection filter),
+ * but the whole-file "metadata delete" fast path consumes the result as
the <i>exact</i> statement predicate:
+ * a dropped conjunct would widen the delete and drop data files that
still hold live rows. Strict mode
+ * therefore collapses the whole expression to {@code null} — "not usable"
— as soon as any node outside the
+ * matrix is met, so the caller falls back to the row-level path instead
of guessing.
+ */
+ public static ConnectorExpression convertStrict(Expression expr) {
+ return convert(expr, true);
+ }
+
+ private static ConnectorExpression convert(Expression expr, boolean
strict) {
if (expr == null) {
return null;
}
if (expr instanceof And) {
- return convertAnd((And) expr);
+ return convertAnd((And) expr, strict);
} else if (expr instanceof Or) {
- return convertOr((Or) expr);
+ return convertOr((Or) expr, strict);
} else if (expr instanceof Not) {
- ConnectorExpression child = convert(((Not) expr).child());
+ // SQL negation is not exact for a destructive whole-file
predicate: Iceberg's notEqual/notIn
+ // matching treats NULL rows as matching, while SQL NOT(UNKNOWN)
is UNKNOWN (deletes nothing), so
+ // lowering NOT(col = x) would drop files whose rows are all NULL.
Only IS NOT NULL keeps SQL
+ // semantics under negation; every other negated form falls back
to the row-level path in strict mode.
+ if (strict && !(((Not) expr).child() instanceof IsNull)) {
+ return null;
+ }
+ ConnectorExpression child = convert(((Not) expr).child(), strict);
return child == null ? null : new ConnectorNot(child);
} else if (expr instanceof EqualTo) {
- return convertComparison(expr, ConnectorComparison.Operator.EQ);
+ return convertComparison(expr, ConnectorComparison.Operator.EQ,
strict);
} else if (expr instanceof GreaterThan) {
- return convertComparison(expr, ConnectorComparison.Operator.GT);
+ return convertComparison(expr, ConnectorComparison.Operator.GT,
strict);
} else if (expr instanceof GreaterThanEqual) {
- return convertComparison(expr, ConnectorComparison.Operator.GE);
+ return convertComparison(expr, ConnectorComparison.Operator.GE,
strict);
} else if (expr instanceof LessThan) {
- return convertComparison(expr, ConnectorComparison.Operator.LT);
+ return convertComparison(expr, ConnectorComparison.Operator.LT,
strict);
} else if (expr instanceof LessThanEqual) {
- return convertComparison(expr, ConnectorComparison.Operator.LE);
+ return convertComparison(expr, ConnectorComparison.Operator.LE,
strict);
} else if (expr instanceof InPredicate) {
return convertIn((InPredicate) expr);
} else if (expr instanceof IsNull) {
@@ -114,9 +138,12 @@ public final class NereidsToConnectorExpressionConverter {
return null;
}
- private static ConnectorExpression convertAnd(And and) {
+ private static ConnectorExpression convertAnd(And and, boolean strict) {
List<ConnectorExpression> conjuncts = new ArrayList<>();
- flattenAnd(and, conjuncts);
+ boolean complete = flattenAnd(and, conjuncts, strict);
+ if (!complete) {
+ return null;
+ }
conjuncts.removeIf(c -> c == null);
if (conjuncts.isEmpty()) {
return null;
@@ -124,34 +151,42 @@ public final class NereidsToConnectorExpressionConverter {
return conjuncts.size() == 1 ? conjuncts.get(0) : new
ConnectorAnd(conjuncts);
}
- private static void flattenAnd(Expression expr, List<ConnectorExpression>
out) {
+ private static boolean flattenAnd(Expression expr,
List<ConnectorExpression> out, boolean strict) {
if (expr instanceof And) {
for (Expression child : expr.children()) {
- flattenAnd(child, out);
+ if (!flattenAnd(child, out, strict)) {
+ return false;
+ }
}
+ return true;
} else {
- out.add(convert(expr));
+ ConnectorExpression converted = convert(expr, strict);
+ if (converted == null && strict) {
+ return false;
+ }
+ out.add(converted);
+ return true;
}
}
- private static ConnectorExpression convertOr(Or or) {
+ private static ConnectorExpression convertOr(Or or, boolean strict) {
List<ConnectorExpression> disjuncts = new ArrayList<>();
- if (!flattenOr(or, disjuncts) || disjuncts.isEmpty()) {
+ if (!flattenOr(or, disjuncts, strict) || disjuncts.isEmpty()) {
return null;
}
return disjuncts.size() == 1 ? disjuncts.get(0) : new
ConnectorOr(disjuncts);
}
- private static boolean flattenOr(Expression expr,
List<ConnectorExpression> out) {
+ private static boolean flattenOr(Expression expr,
List<ConnectorExpression> out, boolean strict) {
if (expr instanceof Or) {
for (Expression child : expr.children()) {
- if (!flattenOr(child, out)) {
+ if (!flattenOr(child, out, strict)) {
return false;
}
}
return true;
}
- ConnectorExpression c = convert(expr);
+ ConnectorExpression c = convert(expr, strict);
if (c == null) {
return false;
}
@@ -159,7 +194,8 @@ public final class NereidsToConnectorExpressionConverter {
return true;
}
- private static ConnectorExpression convertComparison(Expression cmp,
ConnectorComparison.Operator op) {
+ private static ConnectorExpression convertComparison(
+ Expression cmp, ConnectorComparison.Operator op, boolean strict) {
Expression left = cmp.child(0);
Expression right = cmp.child(1);
SlotReference slot;
@@ -170,6 +206,11 @@ public final class NereidsToConnectorExpressionConverter {
} else if (left instanceof Literal && right instanceof SlotReference) {
slot = (SlotReference) right;
literal = (Literal) left;
+ if (strict) {
+ // Exact destructive predicates must commute the operator with
the operands: `1 < id` is
+ // `id > 1`. Loose conflict-filter behavior is deliberately
unchanged by this feature.
+ op = commute(op);
+ }
} else {
return null;
}
@@ -180,6 +221,23 @@ public final class NereidsToConnectorExpressionConverter {
return new ConnectorComparison(op, columnRef(slot), litExpr);
}
+ /** Mirrors {@code op} across the operands: {@code 1 < id} is {@code id >
1}. */
+ private static ConnectorComparison.Operator
commute(ConnectorComparison.Operator op) {
+ switch (op) {
+ case LT:
+ return ConnectorComparison.Operator.GT;
+ case LE:
+ return ConnectorComparison.Operator.GE;
+ case GT:
+ return ConnectorComparison.Operator.LT;
+ case GE:
+ return ConnectorComparison.Operator.LE;
+ default:
+ // EQ / NE / EQ_FOR_NULL are symmetric.
+ return op;
+ }
+ }
+
private static ConnectorExpression convertIn(InPredicate in) {
if (!(in.child(0) instanceof SlotReference)) {
return null;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractor.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractor.java
index 2897ad8c7b4..251380db7af 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractor.java
@@ -17,15 +17,22 @@
package org.apache.doris.datasource.connector.converter;
+import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.connector.spi.pushdown.ConnectorAnd;
import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
+import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
import org.apache.doris.connector.spi.pushdown.ConnectorPredicate;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSink;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
import java.util.ArrayList;
import java.util.List;
@@ -102,6 +109,105 @@ public final class WriteConstraintExtractor {
}
}
+ /**
+ * Strict, whole-plan variant of {@link #extract} used by the connector's
whole-file "metadata delete" fast
+ * path (Spark/StarRocks-style {@code DELETE} that drops data files
instead of writing delete files): it
+ * returns the statement's complete predicate, or empty when completeness
cannot be proven.
+ *
+ * <p>The two extractors deliberately differ. {@link #extract} drops
anything it cannot represent, which only
+ * ever <i>widens</i> a conflict-detection filter and is therefore safe
there. Here a dropped conjunct would
+ * widen the <i>delete</i> and drop whole data files that still hold live
rows, so this variant bails out
+ * instead of guessing whenever a conjunct is not provably part of the
statement predicate — anything that
+ * references another table (join / subquery), any conjunct whose Nereids
form is outside the neutral matrix,
+ * or any plan node that could restrict/reshape rows without appearing as
a {@link LogicalFilter}.</p>
+ *
+ * <p>Every conjunct must be a physical target-table predicate. A conjunct
is rejected — not silently
+ * dropped — when it references a synthetic column: the exclusion list is
name-based, and a name such as the
+ * connector's {@code $partition_spec_id} can also be a legitimate user
column, so provenance cannot be
+ * proven here. A conjunct whose slot name differs from the physical
column it came from is rejected too:
+ * an alias/mask projection renames a slot while keeping its original
column, and emitting the alias name
+ * would target a different physical column.</p>
+ *
+ * @param analyzedPlan the analyzed DELETE plan
+ * @param targetTableId the id of the DELETE target table
+ * @param exclusion a predicate marking synthetic (non-statement)
slots, may be {@code null}
+ * @return the complete predicate — a {@code true} literal when the
statement has no {@code WHERE} clause at
+ * all — or empty when it cannot be extracted exactly
+ */
+ public static Optional<ConnectorExpression> extractCompletePredicate(Plan
analyzedPlan, long targetTableId,
+ Predicate<SlotReference> exclusion) {
+ if (analyzedPlan == null) {
+ return Optional.empty();
+ }
+ List<Expression> targetConjuncts = new ArrayList<>();
+ if (!collectCompletePredicate(analyzedPlan, targetTableId, exclusion,
targetConjuncts)) {
+ return Optional.empty();
+ }
+ if (targetConjuncts.isEmpty()) {
+ // No WHERE clause at all: the statement deletes every row.
+ return Optional.of(ConnectorLiteral.ofBoolean(true));
+ }
+ List<ConnectorExpression> converted = new
ArrayList<>(targetConjuncts.size());
+ for (Expression conjunct : targetConjuncts) {
+ ConnectorExpression neutral =
NereidsToConnectorExpressionConverter.convertStrict(conjunct);
+ if (neutral == null) {
+ return Optional.empty();
+ }
+ converted.add(neutral);
+ }
+ return Optional.of(converted.size() == 1 ? converted.get(0) : new
ConnectorAnd(converted));
+ }
+
+ /**
+ * Walks the analyzed plan along its single-input chain, collecting every
statement conjunct, and returns
+ * {@code false} as soon as the plan proves to be more than "target table
→ filter(s) → projection(s) → sink".
+ * Only nodes that cannot change which rows of the target are produced are
allowed; any branching node (a
+ * join, set operation, ...) fails the walk, because its predicate would
not be visible here.
+ */
+ private static boolean collectCompletePredicate(Plan plan, long
targetTableId,
+ Predicate<SlotReference> exclusion, List<Expression> output) {
+ if (plan instanceof LogicalFilter) {
+ for (Expression conjunct : ((LogicalFilter<?>)
plan).getConjuncts()) {
+ if (!isExactPhysicalTargetPredicate(conjunct, targetTableId,
exclusion)) {
+ return false;
+ }
+ output.add(conjunct);
+ }
+ } else if (!(plan instanceof LogicalSink) && !(plan instanceof
LogicalProject)
+ && !(plan instanceof LogicalCheckPolicy) && !(plan instanceof
LogicalSubQueryAlias)
+ && !(plan instanceof LogicalCatalogRelation)) {
+ return false;
+ }
+
+ List<Plan> children = plan.children();
+ if (plan instanceof LogicalCatalogRelation) {
+ return children.isEmpty() && ((LogicalCatalogRelation)
plan).getTable().getId() == targetTableId;
+ }
+ if (children.size() != 1) {
+ return false;
+ }
+ return collectCompletePredicate(children.get(0), targetTableId,
exclusion, output);
+ }
+
+ /**
+ * A conjunct that can be lowered to a physical predicate with no loss:
target-only, and every slot still
+ * carries the name of the column it was created from (no alias/mask
rename).
+ */
+ private static boolean isExactPhysicalTargetPredicate(Expression
predicate, long targetTableId,
+ Predicate<SlotReference> exclusion) {
+ if (!isTargetOnlyPredicate(predicate, targetTableId, exclusion)) {
+ return false;
+ }
+ for (Slot slot : predicate.getInputSlots()) {
+ SlotReference reference = (SlotReference) slot;
+ Optional<Column> originalColumn = reference.getOriginalColumn();
+ if (!originalColumn.isPresent() ||
!reference.getName().equals(originalColumn.get().getName())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
private static boolean isTargetOnlyPredicate(Expression predicate, long
targetTableId,
Predicate<SlotReference> exclusion) {
if (predicate == null) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransform.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransform.java
index 9996e3e8915..d97a7c00801 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransform.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransform.java
@@ -23,6 +23,7 @@ import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
import org.apache.doris.connector.spi.handle.WriteOperation;
+import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
import org.apache.doris.connector.spi.pushdown.ConnectorPredicate;
import org.apache.doris.connector.spi.write.ConnectorRowChangeStyle;
import org.apache.doris.datasource.ExternalTable;
@@ -45,8 +46,10 @@ import org.apache.doris.planner.PlanFragment;
import org.apache.doris.qe.ConnectContext;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeSet;
+import java.util.function.BooleanSupplier;
import java.util.function.Predicate;
/**
@@ -209,10 +212,45 @@ public class PositionDeleteRowLevelDmlTransform
implements RowLevelDmlTransform
@Override
public Optional<ConnectorPredicate> extractWriteConstraint(Plan
analyzedPlan, TableIf table) {
+ return WriteConstraintExtractor.extract(analyzedPlan, table.getId(),
syntheticColumnExclusion(table));
+ }
+
+ @Override
+ public OptionalLong deleteUsingMetadata(ConnectContext ctx,
RowLevelDmlArgs args, Plan analyzedPlan,
+ BooleanSupplier isCancelled) {
+ TableIf table = args.getTable();
+ // The statement's predicate must be extractable *exactly*: the
connector turns it into a whole-file
+ // delete, where a dropped conjunct would drop data files that still
hold live rows. extractCompletePredicate
+ // returns empty whenever completeness cannot be proven, and the
row-level path then runs unchanged.
+ Optional<ConnectorExpression> predicate =
WriteConstraintExtractor.extractCompletePredicate(
+ analyzedPlan, table.getId(), syntheticColumnExclusion(table));
+ if (!predicate.isPresent()) {
+ return OptionalLong.empty();
+ }
+ ExternalTable externalTable = (ExternalTable) table;
+ PluginDrivenExternalCatalog catalog =
+ (PluginDrivenExternalCatalog) ((PluginDrivenExternalTable)
table).getCatalog();
+ ConnectorSession session = catalog.buildConnectorSession();
+ ConnectorMetadata metadata = PluginDrivenMetadata.get(session,
catalog.getConnector());
+ ConnectorTableHandle handle = metadata.getTableHandle(
+ session, externalTable.getRemoteDbName(),
externalTable.getRemoteName())
+ .orElseThrow(() -> new AnalysisException("Table not found: "
+ + externalTable.getRemoteDbName() + "." +
externalTable.getRemoteName()
+ + " in catalog " + catalog.getName()));
+ // Fence the fresh by-name load against a drop/recreate (or schema
rebind) that happened after analysis,
+ // exactly like the row-level write path does with its pinned
write-metadata identity.
+ PluginDrivenExternalTable.WriteSchemaSnapshot writeSchema =
+ ((PluginDrivenExternalTable) table).getWriteSchemaSnapshot();
+ String boundWriteMetadataIdentity =
+ writeSchema == null ? null :
writeSchema.getWriteMetadataIdentity();
+ return metadata.deleteUsingMetadata(session, handle, predicate,
boundWriteMetadataIdentity, isCancelled);
+ }
+
+ /** The connector-declared synthetic (non-statement) columns, excluded
from a statement predicate. */
+ private static Predicate<SlotReference> syntheticColumnExclusion(TableIf
table) {
Set<String> excludedColumns = new
TreeSet<>(String.CASE_INSENSITIVE_ORDER);
excludedColumns.addAll(((PluginDrivenExternalTable) table)
.getConnectorRowLevelWriteConstraintExcludedColumns());
- Predicate<SlotReference> exclusion = slot ->
excludedColumns.contains(slot.getName());
- return WriteConstraintExtractor.extract(analyzedPlan, table.getId(),
exclusion);
+ return slot -> excludedColumns.contains(slot.getName());
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommand.java
index 9d3911f5422..cb1884eb572 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommand.java
@@ -17,10 +17,14 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.Status;
+import org.apache.doris.common.UserException;
import org.apache.doris.connector.spi.handle.ConnectorTransaction;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.nereids.PlannerHook;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
import org.apache.doris.nereids.trees.plans.Plan;
import
org.apache.doris.nereids.trees.plans.commands.insert.BaseExternalTableInsertExecutor;
@@ -34,8 +38,12 @@ import org.apache.doris.qe.StmtExecutor;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import java.util.OptionalLong;
import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicReference;
/**
* Generic shell for row-level DML ({@code DELETE}/{@code UPDATE}/{@code MERGE
INTO}) against external tables.
@@ -53,6 +61,8 @@ import java.util.concurrent.Callable;
*/
public class RowLevelDmlCommand {
+ private static final Logger LOG =
LogManager.getLogger(RowLevelDmlCommand.class);
+
private final RowLevelDmlTransform transform;
private final RowLevelDmlArgs args;
private final RowLevelDmlOp op;
@@ -80,10 +90,37 @@ public class RowLevelDmlCommand {
Callable<Void> execute = () -> {
LogicalPlanAdapter logicalPlanAdapter = new
LogicalPlanAdapter(plan, ctx.getStatementContext());
NereidsPlanner planner = new
NereidsPlanner(ctx.getStatementContext());
+ boolean tryMetadataDelete =
+ op == RowLevelDmlOp.DELETE &&
args.getPartitions().isEmpty() && !args.isTempPart();
+ AtomicReference<Plan> metadataDeletePlan = new
AtomicReference<>();
+ if (tryMetadataDelete) {
+ captureAnalyzedPlan(ctx, metadataDeletePlan);
+ }
planner.plan(logicalPlanAdapter,
ctx.getSessionVariable().toThrift());
stmtExecutor.setPlanner(planner);
stmtExecutor.checkBlockRules();
+ // Whole-file metadata delete (Spark/StarRocks-style): when
the connector can satisfy the DELETE
+ // by dropping whole data files, it commits that remote
metadata operation here and the
+ // row-level scan/sink never runs. Skipped for a
PARTITION(...) scope, whose row restriction is
+ // carried by the scan rather than by the statement predicate
the connector receives.
+ if (tryMetadataDelete) {
+ // The ordinary row-level path replays this retained
cancellation when the executor publishes
+ // its coordinator; this branch has no coordinator, so do
it explicitly before touching the
+ // remote table.
+ throwIfCancelled(stmtExecutor);
+ OptionalLong metadataDeletedRows =
+ transform.deleteUsingMetadata(ctx, args,
metadataDeletePlan.get(),
+ () ->
stmtExecutor.getPendingCancelReason() != null);
+ if (metadataDeletedRows.isPresent()) {
+ reportMetadataDelete(ctx, table,
metadataDeletedRows.getAsLong());
+ return null;
+ }
+ // The connector aborts before committing when the query
was cancelled; surface the original
+ // cancellation instead of falling back to a row-level
write.
+ throwIfCancelled(stmtExecutor);
+ }
+
PhysicalSink<?> physicalSink =
transform.requirePhysicalSink(planner, op);
PlanFragment fragment = planner.getFragments().get(0);
DataSink dataSink = fragment.getSink();
@@ -180,4 +217,54 @@ public class RowLevelDmlCommand {
return sink.children() != null && sink.children().size() == 1
&& sink.child(0) instanceof PhysicalEmptyRelation;
}
+
+ /** Replays a cancellation retained by the executor before the statement
may mutate the remote table. */
+ @VisibleForTesting
+ static void throwIfCancelled(StmtExecutor stmtExecutor) throws
UserException {
+ Status cancelReason = stmtExecutor.getPendingCancelReason();
+ if (cancelReason != null) {
+ throw new UserException(cancelReason.getErrorMsg());
+ }
+ }
+
+ /**
+ * Registers the analyzed-plan capture the whole-file DELETE fast path
consumes.
+ *
+ * <p>{@link NereidsPlanner} only retains {@code analyzedPlan} for {@code
EXPLAIN ANALYZED}/{@code ALL} or when
+ * a {@link PlannerHook} records it; a plain DELETE leaves {@code
getAnalyzedPlan()} null, which would make
+ * {@code extractCompletePredicate} return empty and silently disable the
optimization. Keep the capture local
+ * to the metadata-delete attempt: publishing it through {@link
NereidsPlanner#setAnalyzedPlan} would also
+ * activate the ordinary row-level conflict-filter consumer when this fast
path falls back.</p>
+ */
+ @VisibleForTesting
+ static void captureAnalyzedPlan(ConnectContext ctx, AtomicReference<Plan>
analyzedPlan) {
+ ctx.getStatementContext().addPlannerHook(new PlannerHook() {
+ @Override
+ public void afterAnalyze(NereidsPlanner planner) {
+ if (planner.getCascadesContext() != null
+ && planner.getCascadesContext().getRewritePlan() !=
null) {
+
analyzedPlan.set(planner.getCascadesContext().getRewritePlan());
+ }
+ }
+ });
+ }
+
+ /**
+ * Publishes the result of a whole-file metadata delete. The connector has
already committed the remote
+ * delete as one atomic metadata operation, so no coordinator or connector
transaction is involved here:
+ * mirror the surfaces the row-level path's {@code afterExec} feeds —
{@link org.apache.doris.qe.QueryState}
+ * for the client's "Query OK, N rows affected" and the fe.audit.log
return-rows counter — plus its
+ * post-commit refresh, which a remote commit needs so later queries do
not read a stale cached snapshot.
+ */
+ private static void reportMetadataDelete(ConnectContext ctx, TableIf
table, long deletedRows) {
+ ctx.getState().setOk(deletedRows, 0, "{'status':'COMMITTED'}");
+ ctx.updateReturnRows((int) deletedRows);
+ try {
+
Env.getCurrentEnv().getRefreshManager().refreshTableAfterExternalMutation((ExternalTable)
table);
+ } catch (Exception e) {
+ // The remote delete is already durable: a refresh failure must
not make the client retry it.
+ LOG.warn("Post-commit refresh failed for table {}. Data was
committed successfully.",
+ table.getName(), e);
+ }
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlTransform.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlTransform.java
index 841350fd8be..13ecbef0f17 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlTransform.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlTransform.java
@@ -29,6 +29,8 @@ import org.apache.doris.planner.PlanFragment;
import org.apache.doris.qe.ConnectContext;
import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.function.BooleanSupplier;
/**
* Per-table strategy that turns a row-level DML ({@code DELETE}/{@code
UPDATE}/{@code MERGE INTO}) against an
@@ -77,4 +79,23 @@ public interface RowLevelDmlTransform {
* when no target-only conjunct survives.
*/
Optional<ConnectorPredicate> extractWriteConstraint(Plan analyzedPlan,
TableIf table);
+
+ /**
+ * Whole-file "metadata delete" fast path for {@code DELETE}, attempted
once the analyzed plan is available
+ * and before the row-level (position-delete) plan is executed.
+ *
+ * <p>The implementation asks the connector whether the statement's
predicate can be honoured by dropping
+ * whole data files ({@code ConnectorWriteOps.deleteUsingMetadata}). A
returned count means the connector
+ * already committed the remote delete, so {@link RowLevelDmlCommand}
reports the result and skips the scan,
+ * the sink and the connector transaction entirely. Returning empty — the
default for transforms whose
+ * connector has no such path — runs the ordinary row-level delete.</p>
+ *
+ * <p>Callers must only invoke this for a DELETE whose statement scope is
exactly its predicate (no
+ * {@code PARTITION(...)} clause), because a predicate that silently
ignores another row restriction would
+ * delete too much.</p>
+ */
+ default OptionalLong deleteUsingMetadata(ConnectContext ctx,
RowLevelDmlArgs args, Plan analyzedPlan,
+ BooleanSupplier isCancelled) {
+ return OptionalLong.empty();
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverterTest.java
index 3d2e9e96eea..f016579131e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverterTest.java
@@ -295,4 +295,63 @@ public class NereidsToConnectorExpressionConverterTest {
public void nullInputReturnsNull() {
Assertions.assertNull(NereidsToConnectorExpressionConverter.convert(null));
}
+
+ // ---- strict variant: never drops an arm (the whole-file metadata-delete
predicate) ----
+
+ @Test
+ public void strictConversionRefusesToDropAnAndArm() {
+ // The loose variant drops the column-to-column arm, which is safe
only for a consumer that treats the
+ // result as an over-approximation (a conflict-detection filter). The
whole-file delete fast path consumes
+ // the result as the exact statement predicate instead, so the strict
variant must refuse the whole AND.
+ SlotReference a = slot("a", ScalarType.INT);
+ SlotReference b = slot("b", ScalarType.INT);
+ SlotReference c = slot("c", ScalarType.INT);
+ And and = new And(new EqualTo(a, new IntegerLiteral(1)), new
EqualTo(b, c));
+
+ ConnectorExpression loose =
NereidsToConnectorExpressionConverter.convert(and);
+ Assertions.assertInstanceOf(ConnectorComparison.class, loose, "the
loose variant keeps the arm it can");
+
+
Assertions.assertNull(NereidsToConnectorExpressionConverter.convertStrict(and),
+ "the strict variant must not widen an exact predicate");
+ }
+
+ @Test
+ public void strictConversionKeepsFullyRepresentablePredicates() {
+ SlotReference a = slot("a", ScalarType.INT);
+ SlotReference b = slot("b", ScalarType.INT);
+
+ ConnectorExpression strict =
NereidsToConnectorExpressionConverter.convertStrict(
+ new And(new EqualTo(a, new IntegerLiteral(1)), new
GreaterThan(b, new IntegerLiteral(2))));
+
+ Assertions.assertInstanceOf(ConnectorAnd.class, strict);
+ Assertions.assertEquals(2, ((ConnectorAnd)
strict).getConjuncts().size());
+ Assertions.assertEquals(NereidsToConnectorExpressionConverter.convert(
+ new And(new EqualTo(a, new IntegerLiteral(1)), new
GreaterThan(b, new IntegerLiteral(2)))),
+ strict, "with nothing to drop, strict and loose agree");
+ }
+
+ @Test
+ public void strictConversionCommutesLiteralLeftComparisons() {
+ // Analysis keeps the operand order as written, so `1 < id` reaches
the converter literal-first. The
+ // exact delete predicate must become `id > 1`; keeping `<` would
select the wrong files.
+ SlotReference id = slot("id", ScalarType.INT);
+
+ ConnectorComparison strict = (ConnectorComparison)
NereidsToConnectorExpressionConverter.convertStrict(
+ new LessThan(new IntegerLiteral(1), id));
+
+ Assertions.assertEquals(ConnectorComparison.Operator.GT,
strict.getOperator());
+ Assertions.assertEquals("id", ((ConnectorColumnRef)
strict.getLeft()).getColumnName());
+ }
+
+ @Test
+ public void strictConversionRejectsSqlUnsafeNegations() {
+ // NOT(col = 1) becomes iceberg notEqual, which treats NULL-only files
as matching, while SQL
+ // NOT(UNKNOWN) deletes nothing. IS NOT NULL is the only negation that
keeps SQL semantics.
+ SlotReference id = slot("id", ScalarType.INT);
+
+
Assertions.assertNull(NereidsToConnectorExpressionConverter.convertStrict(
+ new Not(new EqualTo(id, new IntegerLiteral(1)))));
+ Assertions.assertInstanceOf(ConnectorNot.class,
+ NereidsToConnectorExpressionConverter.convertStrict(new
Not(new IsNull(id))));
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractorTest.java
index 911131f9f13..108d1811721 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractorTest.java
@@ -24,19 +24,27 @@ import org.apache.doris.connector.spi.pushdown.ConnectorAnd;
import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef;
import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
+import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
import org.apache.doris.connector.spi.pushdown.ConnectorPredicate;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.GreaterThan;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
@@ -45,6 +53,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.function.Predicate;
@@ -227,4 +236,194 @@ public class WriteConstraintExtractorTest {
public void nullPlanReturnsEmpty() {
Assertions.assertFalse(WriteConstraintExtractor.extract(null,
TARGET_ID, NO_EXCLUSION).isPresent());
}
+
+ // ───────── complete-predicate extraction (the whole-file "metadata
delete" fast path) ─────────
+
+ /**
+ * The complete variant only accepts a plan that is provably "target table
→ filter(s) → sink": the leaf must
+ * be the target {@link
org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation}, so a
+ * {@link LogicalEmptyRelation} (which cannot even arise from a DELETE,
but would mean "no rows") is rejected
+ * rather than read as an unfiltered delete of everything.
+ */
+ private Plan completePlanOver(Set<Expression> conjuncts, SlotReference
output) {
+ ExternalTable externalTarget = targetTableWithId();
+ return new LogicalFilter<>(conjuncts, new TargetScan(externalTarget,
ImmutableList.of(output)));
+ }
+
+ private SlotReference externalSlot(ExternalTable table, String name) {
+ return SlotReference.fromColumn(StatementScopeIdGenerator.newExprId(),
table,
+ new Column(name, ScalarType.INT), ImmutableList.of());
+ }
+
+ private ExternalTable targetTableWithId() {
+ ExternalTable externalTarget = Mockito.mock(ExternalTable.class);
+ Mockito.when(externalTarget.getId()).thenReturn(TARGET_ID);
+ return externalTarget;
+ }
+
+ /** A minimal concrete target-table scan standing in for the DELETE's
{@code LogicalFileScan} leaf. */
+ private static final class TargetScan extends LogicalCatalogRelation {
+ private final List<Slot> output;
+
+ TargetScan(TableIf table, List<Slot> output) {
+ super(new RelationId(0), PlanType.LOGICAL_FILE_SCAN, table,
ImmutableList.of("db1"));
+ this.output = ImmutableList.copyOf(output);
+ }
+
+ @Override
+ public List<Slot> computeOutput() {
+ return output;
+ }
+
+ @Override
+ public LogicalCatalogRelation withRelationId(RelationId relationId) {
+ return this;
+ }
+
+ @Override
+ public LogicalCatalogRelation withTableAlias(String tableAlias) {
+ return this;
+ }
+
+ @Override
+ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
+ Optional<LogicalProperties> logicalProperties, List<Plan>
children) {
+ return this;
+ }
+
+ @Override
+ public Plan withGroupExpression(Optional<GroupExpression>
groupExpression) {
+ return this;
+ }
+ }
+
+ @Test
+ public void completePredicateIsExtractedOverTheTargetScan() {
+ ExternalTable externalTarget = targetTableWithId();
+ SlotReference id = externalSlot(externalTarget, "id");
+
+ Optional<ConnectorExpression> result =
WriteConstraintExtractor.extractCompletePredicate(
+ completePlanOver(ImmutableSet.of(new EqualTo(id, new
IntegerLiteral(1))), id),
+ TARGET_ID, NO_EXCLUSION);
+
+ Assertions.assertTrue(result.isPresent());
+ Assertions.assertTrue(result.get() instanceof ConnectorComparison);
+ Assertions.assertEquals("id",
+ ((ConnectorColumnRef) ((ConnectorComparison)
result.get()).getLeft()).getColumnName());
+ }
+
+ @Test
+ public void completePredicateWithoutWhereClauseIsAlwaysTrue() {
+ ExternalTable externalTarget = targetTableWithId();
+ TargetScan scan = new TargetScan(externalTarget, ImmutableList.of());
+
+ Optional<ConnectorExpression> result =
+ WriteConstraintExtractor.extractCompletePredicate(scan,
TARGET_ID, NO_EXCLUSION);
+
+ Assertions.assertTrue(result.isPresent(), "DELETE without WHERE is an
always-true predicate");
+ Assertions.assertTrue(result.get() instanceof ConnectorLiteral);
+ Assertions.assertEquals(Boolean.TRUE, ((ConnectorLiteral)
result.get()).getValue());
+ }
+
+ @Test
+ public void completePredicateBailsOutWhenAnyConjunctIsUnrepresentable() {
+ // The loose variant drops the column-to-column arm (safe there: it
only widens a conflict filter), but a
+ // whole-file delete must not: a dropped conjunct would drop files
that still hold live rows.
+ ExternalTable externalTarget = targetTableWithId();
+ SlotReference id = externalSlot(externalTarget, "id");
+ SlotReference v = externalSlot(externalTarget, "v");
+ SlotReference w = externalSlot(externalTarget, "w");
+ Plan plan = completePlanOver(ImmutableSet.of(
+ new EqualTo(id, new IntegerLiteral(1)),
+ new EqualTo(v, w)), id);
+
+ Assertions.assertTrue(WriteConstraintExtractor.extract(plan,
TARGET_ID, NO_EXCLUSION).isPresent(),
+ "the loose extractor keeps the representable arm");
+ Assertions.assertFalse(
+ WriteConstraintExtractor.extractCompletePredicate(plan,
TARGET_ID, NO_EXCLUSION).isPresent(),
+ "the complete extractor must refuse the whole predicate
instead of guessing");
+ }
+
+ @Test
+ public void completePredicateBailsOutOnCrossTableConjunct() {
+ ExternalTable externalTarget = targetTableWithId();
+ TableIf other = Mockito.mock(TableIf.class);
+ Mockito.when(other.getId()).thenReturn(2L);
+ SlotReference id = externalSlot(externalTarget, "id");
+ SlotReference otherId = slot(other, "id", ScalarType.INT);
+
+ Plan plan = completePlanOver(ImmutableSet.of(
+ new EqualTo(id, new IntegerLiteral(1)),
+ new EqualTo(otherId, new IntegerLiteral(2))), id);
+
+
Assertions.assertFalse(WriteConstraintExtractor.extractCompletePredicate(plan,
TARGET_ID, NO_EXCLUSION)
+ .isPresent(), "a conjunct over another table cannot be
honoured by a whole-file delete");
+ }
+
+ @Test
+ public void completePredicateBailsOutOnSyntheticConjuncts() {
+ // Provenance between an engine-injected $row_id conjunct and a user
column that merely shares a
+ // synthetic-looking name cannot be established here, so the exact
extractor must fall back instead of
+ // dropping the conjunct (dropping it would widen the delete).
+ ExternalTable externalTarget = targetTableWithId();
+ SlotReference id = externalSlot(externalTarget, "id");
+ SlotReference rowId = externalSlot(externalTarget,
"__DORIS_ICEBERG_ROWID_COL__");
+ Plan plan = completePlanOver(ImmutableSet.of(
+ new EqualTo(id, new IntegerLiteral(1)),
+ new EqualTo(rowId, new IntegerLiteral(0))), id);
+ Predicate<SlotReference> excludeRowId = s ->
"__DORIS_ICEBERG_ROWID_COL__".equalsIgnoreCase(s.getName());
+
+ Optional<ConnectorExpression> result =
+ WriteConstraintExtractor.extractCompletePredicate(plan,
TARGET_ID, excludeRowId);
+
+ Assertions.assertFalse(result.isPresent(), "a synthetic conjunct must
disable the fast path");
+ }
+
+ @Test
+ public void completePredicateBailsOutOnNonLinearPlan() {
+ // Anything that can restrict rows without showing up as a
LogicalFilter (a join, set operation,
+ // aggregate, limit, ...) must disable the fast path: its predicate is
not part of what we can see.
+ Plan join = Mockito.mock(LogicalJoin.class);
+
+
Assertions.assertFalse(WriteConstraintExtractor.extractCompletePredicate(join,
TARGET_ID, NO_EXCLUSION)
+ .isPresent());
+ }
+
+ @Test
+ public void completePredicateBailsOutOnEmptyRelationLeaf() {
+ SlotReference id = externalSlot(targetTableWithId(), "id");
+ Plan plan = filterOver(ImmutableSet.of(new EqualTo(id, new
IntegerLiteral(1))), id);
+
+
Assertions.assertFalse(WriteConstraintExtractor.extractCompletePredicate(plan,
TARGET_ID, NO_EXCLUSION)
+ .isPresent(), "a non-table leaf cannot prove the delete
touches whole table files");
+ }
+
+ @Test
+ public void completePredicateBailsOutOnAliasRenamedSlots() {
+ // Alias.toSlot keeps the original table/column but renames the slot.
Emitting that name would target a
+ // different physical column, so the exact extractor must fall back
instead of trusting it.
+ ExternalTable externalTarget = targetTableWithId();
+ SlotReference physical = externalSlot(externalTarget, "b");
+ SlotReference renamed = physical.withName("a");
+ Plan plan = completePlanOver(ImmutableSet.of(new EqualTo(renamed, new
IntegerLiteral(1))), renamed);
+
+
Assertions.assertFalse(WriteConstraintExtractor.extractCompletePredicate(plan,
TARGET_ID, NO_EXCLUSION)
+ .isPresent(), "a renamed slot no longer denotes its physical
column");
+ }
+
+ @Test
+ public void completePredicateBailsOutOnSyntheticNameCollision() {
+ // $partition_spec_id is both an Iceberg-injected name and a legal
physical column name. The exclusion
+ // list is name-based, so a matching conjunct must be rejected rather
than silently dropped as
+ // "synthetic" (which would turn the predicate into always-true and
delete every file).
+ ExternalTable externalTarget = targetTableWithId();
+ SlotReference physical = externalSlot(externalTarget,
"$partition_spec_id");
+ Plan plan = completePlanOver(ImmutableSet.of(new EqualTo(physical, new
IntegerLiteral(1))), physical);
+ Predicate<SlotReference> excludeSyntheticNames =
+ slot -> "$partition_spec_id".equalsIgnoreCase(slot.getName());
+
+
Assertions.assertFalse(WriteConstraintExtractor.extractCompletePredicate(
+ plan, TARGET_ID, excludeSyntheticNames).isPresent(),
+ "a name collision must fall back; dropping the conjunct would
widen the delete to everything");
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransformTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransformTest.java
index b838ad91ce6..aec0e088afc 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransformTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransformTest.java
@@ -35,6 +35,8 @@ import
org.apache.doris.datasource.plugin.PluginDrivenExternalCatalog;
import org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
@@ -42,9 +44,11 @@ import
org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.RelationId;
import
org.apache.doris.nereids.trees.plans.commands.insert.BaseExternalTableInsertExecutor;
import
org.apache.doris.nereids.trees.plans.commands.insert.PluginDrivenInsertExecutor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation;
import
org.apache.doris.nereids.trees.plans.logical.LogicalExternalRowLevelDeleteSink;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
@@ -58,10 +62,13 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import java.util.EnumSet;
+import java.util.List;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
/**
@@ -360,4 +367,92 @@ public class PositionDeleteRowLevelDmlTransformTest {
Mockito.verify(t,
Mockito.never()).extractWriteConstraint(Mockito.any(), Mockito.any());
}
+
+ // ─────────────── whole-file metadata-delete fast path ───────────────
+
+ /** A minimal concrete target-table scan: the leaf an analyzed
single-table DELETE filters over. */
+ private static final class TargetScan extends LogicalCatalogRelation {
+ private final List<SlotReference> output;
+
+ TargetScan(TableIf table, List<SlotReference> output) {
+ super(new RelationId(0), PlanType.LOGICAL_FILE_SCAN, table,
ImmutableList.of("db1"));
+ this.output = ImmutableList.copyOf(output);
+ }
+
+ @Override
+ public List<org.apache.doris.nereids.trees.expressions.Slot>
computeOutput() {
+ return ImmutableList.copyOf(output);
+ }
+
+ @Override
+ public LogicalCatalogRelation withRelationId(RelationId relationId) {
+ return this;
+ }
+
+ @Override
+ public LogicalCatalogRelation withTableAlias(String tableAlias) {
+ return this;
+ }
+
+ @Override
+ public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
+ Optional<LogicalProperties> logicalProperties, List<Plan>
children) {
+ return this;
+ }
+
+ @Override
+ public Plan withGroupExpression(Optional<GroupExpression>
groupExpression) {
+ return this;
+ }
+ }
+
+ private RowLevelDmlArgs deleteArgs(TableIf table) {
+ return RowLevelDmlArgs.forDelete(table, ImmutableList.of("db1", "t1"),
null, false,
+ ImmutableList.of(), null);
+ }
+
+ @Test
+ public void
deleteUsingMetadataForwardsTheCompletePredicateToTheConnector() {
+ ConnectorSession session = Mockito.mock(ConnectorSession.class);
+ ConnectorMetadata metadata = Mockito.mock(ConnectorMetadata.class);
+ ConnectorTableHandle handle = Mockito.mock(ConnectorTableHandle.class);
+ Mockito.when(metadata.getTableHandle(session, "db1",
"t1")).thenReturn(Optional.of(handle));
+ Mockito.when(metadata.deleteUsingMetadata(
+ Mockito.eq(session), Mockito.eq(handle),
Mockito.any(), Mockito.any(), Mockito.any()))
+ .thenReturn(OptionalLong.of(3L));
+ PluginDrivenExternalTable table = pluginTableWithMetadata(metadata,
session);
+ SlotReference slot = slot(table, "id");
+ Plan plan = new LogicalFilter<>(ImmutableSet.of(new EqualTo(slot, new
IntegerLiteral(1))),
+ new TargetScan(table, ImmutableList.of(slot)));
+
+ OptionalLong deleted = transform.deleteUsingMetadata(
+ Mockito.mock(ConnectContext.class), deleteArgs(table), plan,
() -> false);
+
+ Assertions.assertEquals(OptionalLong.of(3L), deleted);
+ @SuppressWarnings("unchecked")
+
ArgumentCaptor<Optional<org.apache.doris.connector.spi.pushdown.ConnectorExpression>>
captor =
+ ArgumentCaptor.forClass(Optional.class);
+ Mockito.verify(metadata).deleteUsingMetadata(
+ Mockito.eq(session), Mockito.eq(handle), captor.capture(),
Mockito.any(), Mockito.any());
+ Assertions.assertTrue(captor.getValue().isPresent(), "the statement
predicate must reach the connector");
+
Assertions.assertInstanceOf(org.apache.doris.connector.spi.pushdown.ConnectorComparison.class,
+ captor.getValue().get());
+ }
+
+ @Test
+ public void
deleteUsingMetadataStaysOutWhenThePredicateIsNotProvablyComplete() {
+ ConnectorSession session = Mockito.mock(ConnectorSession.class);
+ ConnectorMetadata metadata = Mockito.mock(ConnectorMetadata.class);
+ ConnectorTableHandle handle = Mockito.mock(ConnectorTableHandle.class);
+ Mockito.when(metadata.getTableHandle(session, "db1",
"t1")).thenReturn(Optional.of(handle));
+ PluginDrivenExternalTable table = pluginTableWithMetadata(metadata,
session);
+
+ // The leaf is not the target catalog relation, so completeness cannot
be proven.
+ OptionalLong deleted = transform.deleteUsingMetadata(
+ Mockito.mock(ConnectContext.class), deleteArgs(table),
filterOver(table, "id"), () -> false);
+
+ Assertions.assertEquals(OptionalLong.empty(), deleted);
+ Mockito.verify(metadata, Mockito.never())
+ .deleteUsingMetadata(Mockito.any(), Mockito.any(),
Mockito.any(), Mockito.any(), Mockito.any());
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommandTest.java
index 81ef3a894d0..3d168db1ce2 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommandTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommandTest.java
@@ -18,19 +18,30 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.Status;
+import org.apache.doris.common.UserException;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.nereids.PlannerHook;
+import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.trees.plans.Plan;
import
org.apache.doris.nereids.trees.plans.commands.insert.BaseExternalTableInsertExecutor;
import org.apache.doris.nereids.trees.plans.physical.PhysicalSink;
import org.apache.doris.planner.DataSink;
import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.Coordinator;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.thrift.TStatusCode;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mockito;
+import java.util.concurrent.atomic.AtomicReference;
+
/**
* Pins the begin→finalize guarded window of {@link RowLevelDmlCommand}:
{@code beginTransaction} registers
* the transaction with the transaction manager and the global
external-transaction registry, and the
@@ -100,4 +111,43 @@ public class RowLevelDmlCommandTest {
Mockito.verify(coordinator).setTxnId(42L);
Mockito.verify(executor, Mockito.never()).onFail(Mockito.any());
}
+
+ @Test
+ public void deleteCapturesThePreRewriteAnalyzedPlanLocally() {
+ // NereidsPlanner leaves analyzedPlan null for a plain DELETE. Capture
the plan for the metadata-delete
+ // attempt only; publishing it through setAnalyzedPlan would also
change row-level fallback semantics.
+ ConnectContext ctx = Mockito.mock(ConnectContext.class);
+ StatementContext statementContext =
Mockito.mock(StatementContext.class);
+ Mockito.when(ctx.getStatementContext()).thenReturn(statementContext);
+ AtomicReference<Plan> captured = new AtomicReference<>();
+
+ RowLevelDmlCommand.captureAnalyzedPlan(ctx, captured);
+
+ ArgumentCaptor<PlannerHook> captor =
ArgumentCaptor.forClass(PlannerHook.class);
+ Mockito.verify(statementContext).addPlannerHook(captor.capture());
+ NereidsPlanner planner = Mockito.mock(NereidsPlanner.class);
+ CascadesContext cascadesContext = Mockito.mock(CascadesContext.class);
+ Plan analyzed = Mockito.mock(Plan.class);
+ Mockito.when(planner.getCascadesContext()).thenReturn(cascadesContext);
+ Mockito.when(cascadesContext.getRewritePlan()).thenReturn(analyzed);
+
+ captor.getValue().afterAnalyze(planner);
+
+ Assertions.assertSame(analyzed, captured.get());
+ Mockito.verify(planner,
Mockito.never()).setAnalyzedPlan(Mockito.any());
+ }
+
+ @Test
+ public void retainedCancellationIsReplayedBeforeTheMetadataDelete() {
+ // The row-level path replays the retained cancellation when the
coordinator is published. The metadata
+ // branch has no coordinator, so it must replay it explicitly before
any remote mutation.
+ StmtExecutor stmtExecutor = Mockito.mock(StmtExecutor.class);
+ Mockito.when(stmtExecutor.getPendingCancelReason())
+ .thenReturn(new Status(TStatusCode.CANCELLED, "query was
killed"));
+
+ UserException thrown = Assertions.assertThrows(UserException.class,
+ () -> RowLevelDmlCommand.throwIfCancelled(stmtExecutor));
+
+ Assertions.assertTrue(thrown.getMessage().contains("query was
killed"), thrown.getMessage());
+ }
}
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_dml_modes_evolution.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_dml_modes_evolution.out
index 98c60ec0eb8..eca58838d6b 100644
---
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_dml_modes_evolution.out
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_dml_modes_evolution.out
@@ -22,11 +22,10 @@
7 C golf 2026-03-01T07:00 70 \N
-- !mor_delete_files --
-0 4 4
-2 2 2
+0 3 3
+2 1 1
-- !cow_after_rejections --
1 A alpha 2026-01-01T01:00 10 base
2 \N null-key \N 20 null-partition
3 B beta 2026-02-01T03:00 30 new-spec
-
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_dml_mode_matrix.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_dml_mode_matrix.groovy
index fede01f4fb4..9a9c7c0eec4 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_dml_mode_matrix.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_dml_mode_matrix.groovy
@@ -101,7 +101,7 @@ suite("test_iceberg_v3_row_lineage_dml_mode_matrix",
"p0,external,iceberg,extern
}
}
- def assertDeleteFilesForVersion = { tableName, formatVersion, format,
scenario ->
+ def assertDeleteFilesForVersion = { tableName, formatVersion, format,
partitioned, scenario ->
def deleteFiles = sql("""
select file_path, lower(file_format), record_count
from ${tableName}\$delete_files
@@ -113,8 +113,26 @@ suite("test_iceberg_v3_row_lineage_dml_mode_matrix",
"p0,external,iceberg,extern
"COW scenario ${scenario.name} should not create delete files
for ${tableName}, actual=${deleteFiles}")
}
if (scenario.mode == "merge-on-read" && formatVersion == 3) {
- assertTrue(deleteFiles.size() > 0,
- "MOR scenario ${scenario.name} should create delete files
for ${tableName}")
+ if (scenario.operation == "delete") {
+ // Partitioned fixtures write one row per data file, so
deleting id=2 drops a whole file and
+ // writes no DV. The unpartitioned fixture has one three-row
data file, so it must fall back.
+ long dataFileCount =
+ sql("""select count(*) from
${tableName}\$data_files""")[0][0].toString().toLong()
+ if (partitioned) {
+ assertEquals(0, deleteFiles.size(),
+ "partitioned MOR delete should not create delete
files for ${tableName}")
+ assertEquals(2L, dataFileCount,
+ "partitioned MOR delete should drop one whole data
file for ${tableName}")
+ } else {
+ assertEquals(1, deleteFiles.size(),
+ "unpartitioned partial-file delete should create
one DV for ${tableName}")
+ assertEquals(1L, dataFileCount,
+ "unpartitioned partial-file delete should retain
its data file for ${tableName}")
+ }
+ } else {
+ assertTrue(deleteFiles.size() > 0,
+ "MOR scenario ${scenario.name} should create delete
files for ${tableName}")
+ }
}
deleteFiles.each { row ->
String filePath = row[0].toString().toLowerCase()
@@ -264,7 +282,7 @@ suite("test_iceberg_v3_row_lineage_dml_mode_matrix",
"p0,external,iceberg,extern
}
}
- assertDeleteFilesForVersion(tableName,
formatVersion, format, scenario)
+ assertDeleteFilesForVersion(tableName,
formatVersion, format, partitioned, scenario)
} finally {
sql """drop table if exists ${tableName}"""
}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_delete_unpartitioned_evolution.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_delete_unpartitioned_evolution.groovy
index 7ae07744445..03a0003f353 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_delete_unpartitioned_evolution.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_delete_unpartitioned_evolution.groovy
@@ -109,8 +109,13 @@ suite("test_iceberg_delete_unpartitioned_evolution",
"""
sql """delete from mixed_partition_specs where record_key in (1, 3)"""
checkRows("mixed_partition_specs", expectedAfterMixedDelete)
+ // `record_key in (1, 3)` covers whole data files in this fixture, so
Doris commits it as a metadata
+ // delete: no delete files are written and the surviving rows keep both
historical spec ids.
+ assertEquals(0, sql("""
+ select count(*) from mixed_partition_specs\$delete_files
+ """)[0][0].toString().toLong())
assertEquals(mixedSpecs, sql("""
- select distinct spec_id from mixed_partition_specs\$delete_files order
by spec_id
+ select distinct spec_id from mixed_partition_specs\$data_files order
by spec_id
"""))
def expectedAfterUpdate = spark_iceberg """
select record_key, metric + 100 from
demo.${dbName}.mixed_partition_specs order by record_key
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
