You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by ga...@apache.org on 2026/10/23 07:20:46 UTC
(doris) branch master updated: [feature](iceberg) Support whole-file metadata delete for DELETE (#69176)
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 <ch...@selectdb.com>
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: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org