github-actions[bot] commented on code in PR #69176:
URL: https://github.com/apache/doris/pull/69176#discussion_r4293045501
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommand.java:
##########
@@ -84,6 +90,19 @@ public void run(ConnectContext ctx, StmtExecutor
stmtExecutor) throws Exception
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 (op == RowLevelDmlOp.DELETE &&
args.getPartitions().isEmpty() && !args.isTempPart()) {
+ OptionalLong metadataDeletedRows =
+ transform.deleteUsingMetadata(ctx, args,
planner.getAnalyzedPlan());
Review Comment:
[P2] Capture the analyzed DELETE plan before invoking the fast path
A normal DELETE creates a fresh planner without an analyzed-plan hook.
NereidsPlanner only fills analyzedPlan for ANALYZED_PLAN/ALL_PLAN EXPLAIN or
the hook registered by the insert commands, so getAnalyzedPlan() is null here.
extractCompletePredicate(null) returns empty and every potentially eligible
DELETE falls back to the existing row-level path. Capture the pre-rewrite
analyzed plan and test the real command/planner call path; the added tests pass
constructed plans directly and miss this integration.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractor.java:
##########
@@ -102,6 +108,100 @@ private static void collectTargetConjuncts(Plan plan,
long targetTableId,
}
}
+ /**
+ * 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>Conjuncts over the excluded synthetic columns (the connector's
{@code $row_id} metadata column) are
+ * ignored rather than rejected: they are injected by the engine, not
authored by the statement.</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 (or only engine-injected conjuncts): 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 (isTargetOnlyPredicate(conjunct, targetTableId, exclusion))
{
+ output.add(conjunct);
+ } else if (!isSyntheticOnlyPredicate(conjunct, exclusion)) {
Review Comment:
[P1] Keep predicates on physical columns with synthetic-looking names
isSyntheticOnlyPredicate uses the name-only exclusion supplied by the
transform. Iceberg excludes $partition_spec_id, but that is also a permitted
physical user column name: schema parsing preserves it and only _file/_pos are
reserved metadata names. For Filter(t.`$partition_spec_id` = 1) -> Scan(t) on
such a table, the user conjunct is discarded here and an empty predicate list
becomes alwaysTrue, deleting files containing other values too. Require proven
engine-injection provenance before skipping a conjunct; otherwise reject it and
fall back.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/NereidsToConnectorExpressionConverter.java:
##########
@@ -83,15 +83,31 @@ private 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);
Review Comment:
[P1] Commute inequalities when converting literal-left comparisons
The new exact converter delegates to convertComparison, which swaps a
literal-left comparison without reversing its operator: convertStrict(1 < id)
produces id < 1. A valid analyzed tree is DeleteSink ->
Project(operation,row_id) -> Filter(1 < id) -> Scan(t); analysis preserves
operand order and normalization happens later. With files containing id=0 and
id=2, this exact-delete predicate removes the wrong file. Commute LT/LE/GT/GE
when swapping operands, or reject these forms. The old loose converter also has
this behavior, but using it for the new destructive API introduces the
wrong-file deletion consequence.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/PositionDeleteRowLevelDmlTransform.java:
##########
@@ -209,10 +211,38 @@ public void finalizeSink(BaseExternalTableInsertExecutor
executor, RowLevelDmlOp
@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) {
+ 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()));
+ return metadata.deleteUsingMetadata(session, handle, predicate);
Review Comment:
[P1] Carry the bound write generation into the metadata-delete request
The synthesized delete sink pins a write metadata identity, but this call
forwards only a newly resolved name-based handle and predicate. If another
client drops/recreates the table after analysis, the connector reloads the
replacement by name and commits the old statement against its new UUID.
Dropping/readding a same-name field similarly rebinds the name-only predicate
to a different field. The normal write path rejects these races through
validateBoundWriteGeneration and the UUID/schema identity. Carry and validate
that identity on the fresh load and final metadata commit; including new
snapshots of the same table need not discard the generation fence.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/connector/converter/WriteConstraintExtractor.java:
##########
@@ -102,6 +108,100 @@ private static void collectTargetConjuncts(Plan plan,
long targetTableId,
}
}
+ /**
+ * 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>Conjuncts over the excluded synthetic columns (the connector's
{@code $row_id} metadata column) are
+ * ignored rather than rejected: they are injected by the engine, not
authored by the statement.</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 (or only engine-injected conjuncts): 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 (isTargetOnlyPredicate(conjunct, targetTableId, exclusion))
{
+ output.add(conjunct);
+ } else if (!isSyntheticOnlyPredicate(conjunct, exclusion)) {
+ return false;
+ }
+ }
+ } else if (!(plan instanceof LogicalSink) && !(plan instanceof
LogicalProject)
Review Comment:
[P1] Resolve projected slots before treating filters as physical predicates
Allowing every LogicalProject loses the meaning of aliases used above it. A
valid analyzed mask plan is DeleteSink -> Project(operation,row_id) ->
Filter(a#masked = 1) -> Project(b AS a#masked, b, row_id) -> Scan(t).
Alias.toSlot preserves b's original table/column but uses the alias name a, so
isTargetOnlyPredicate accepts it and columnRef emits physical a=1. With files
(a=0,b=1) and (a=1,b=0), the API selects the wrong file. DELETE targets do pass
through CheckPolicy, which can produce this bare-slot mask projection. Rewrite
through projections using slot identity, or reject non-identity projections and
fall back.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1889,6 +1907,153 @@ public void validateRowLevelDmlMode(ConnectorSession
session, ConnectorTableHand
}
}
+ /**
+ * 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) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ String qualifiedName = iceHandle.getDbName() + "." +
iceHandle.getTableName();
+ try {
+ return executeAuthenticated(() -> {
+ // Load fresh rather than through the read-side caches: the
delete commits against the current
+ // metadata, never a possibly stale cached snapshot.
+ Table table = catalogOps.loadTable(iceHandle.getDbName(),
iceHandle.getTableName());
+ Expression deleteExpr = predicate.isPresent()
+ ? wholeFileDeleteExpression(table, session,
predicate.get())
+ : Expressions.alwaysTrue();
+ if (deleteExpr == null || !matchesWholeFilesOnly(table,
deleteExpr)) {
+ return OptionalLong.empty();
+ }
+ DeleteFiles deleteFiles =
table.newDelete().deleteFromRowFilter(deleteExpr);
+ deleteFiles.commit();
+ return
OptionalLong.of(deletedRowCount(table.currentSnapshot()));
+ });
+ } catch (ValidationException e) {
+ // A racing writer turned a candidate file into a partial match
(or another delete 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);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ private Expression wholeFileDeleteExpression(Table table, ConnectorSession
session,
+ ConnectorExpression predicate) {
+ if (predicate instanceof ConnectorLiteral) {
+ return Boolean.TRUE.equals(((ConnectorLiteral)
predicate).getValue())
+ ? Expressions.alwaysTrue() : null;
+ }
+ ZoneId zone = IcebergTimeUtils.resolveSessionZone(session);
+ List<Expression> predicates = new IcebergPredicateConverter(
+ table.schema(), zone,
IcebergPredicateConverter.Mode.REWRITE).convert(predicate);
Review Comment:
[P1] Preserve SQL null semantics when lowering negated predicates
REWRITE conversion is not an exact SQL DELETE predicate for nullable
negation. An accepted analyzed Filter(NOT(id = 1)) becomes Iceberg
not(equal(id,1)); Iceberg rewrites this to notEqual, and its
StrictEvalVisitor.notEq explicitly treats a NULL-only file as fully matching.
The API therefore drops that file although SQL NOT(UNKNOWN) is UNKNOWN and
deletes no rows. NOT IN has the same issue. Add SQL null guards with correct
nested boolean semantics, or reject unsafe negative forms and fall back.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1889,6 +1907,153 @@ public void validateRowLevelDmlMode(ConnectorSession
session, ConnectorTableHand
}
}
+ /**
+ * 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) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ String qualifiedName = iceHandle.getDbName() + "." +
iceHandle.getTableName();
+ try {
+ return executeAuthenticated(() -> {
+ // Load fresh rather than through the read-side caches: the
delete commits against the current
+ // metadata, never a possibly stale cached snapshot.
+ Table table = catalogOps.loadTable(iceHandle.getDbName(),
iceHandle.getTableName());
+ Expression deleteExpr = predicate.isPresent()
+ ? wholeFileDeleteExpression(table, session,
predicate.get())
+ : Expressions.alwaysTrue();
+ if (deleteExpr == null || !matchesWholeFilesOnly(table,
deleteExpr)) {
+ return OptionalLong.empty();
+ }
+ DeleteFiles deleteFiles =
table.newDelete().deleteFromRowFilter(deleteExpr);
+ deleteFiles.commit();
+ return
OptionalLong.of(deletedRowCount(table.currentSnapshot()));
+ });
+ } catch (ValidationException e) {
+ // A racing writer turned a candidate file into a partial match
(or another delete 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);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ private Expression wholeFileDeleteExpression(Table table, ConnectorSession
session,
+ ConnectorExpression predicate) {
+ if (predicate instanceof ConnectorLiteral) {
+ return Boolean.TRUE.equals(((ConnectorLiteral)
predicate).getValue())
+ ? Expressions.alwaysTrue() : null;
+ }
+ ZoneId zone = IcebergTimeUtils.resolveSessionZone(session);
+ List<Expression> predicates = new IcebergPredicateConverter(
Review Comment:
[P1] Retain the timestamp history guard for metadata deletes
The plain constructor leaves unsafeNanosFields empty, bypassing the
DATE-history/incomplete-history checks in IcebergPredicateConverter.forTable.
For a retained DATE file with ts=2024-01-02 whose field was externally promoted
to TIMESTAMP_NS in format v3, Iceberg decodes its four-byte epoch-day bound as
19724 nanoseconds. A predicate ts < TIMESTAMP_NS '2000-01-01 00:00:00' can then
pass both candidate pruning and the strict whole-file proof and delete the
nonmatching 2024 file. Use the history-aware forTable factory so these fields
force fallback, as RewriteDataFilePlanner already does.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1889,6 +1907,153 @@ public void validateRowLevelDmlMode(ConnectorSession
session, ConnectorTableHand
}
}
+ /**
+ * 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) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ String qualifiedName = iceHandle.getDbName() + "." +
iceHandle.getTableName();
+ try {
+ return executeAuthenticated(() -> {
+ // Load fresh rather than through the read-side caches: the
delete commits against the current
+ // metadata, never a possibly stale cached snapshot.
+ Table table = catalogOps.loadTable(iceHandle.getDbName(),
iceHandle.getTableName());
+ Expression deleteExpr = predicate.isPresent()
+ ? wholeFileDeleteExpression(table, session,
predicate.get())
+ : Expressions.alwaysTrue();
+ if (deleteExpr == null || !matchesWholeFilesOnly(table,
deleteExpr)) {
+ return OptionalLong.empty();
+ }
+ DeleteFiles deleteFiles =
table.newDelete().deleteFromRowFilter(deleteExpr);
Review Comment:
[P2] Reuse authenticated FileIO for the metadata commit workers
On secured HDFS, executeAuthenticated authenticates the caller only.
DeleteFiles commits dispatch manifest filtering/writing to Iceberg's shared
worker pool, where raw table operations resolve HadoopFileIO without that
Kerberos UGI. With workers created outside this call, a whole-file delete fails
authentication and is wrapped as a statement error.
IcebergConnectorTransaction.openTransaction already wraps TableOperations.io in
IcebergAuthenticatedFileIO for this exact worker-thread failure. Reuse that
preparation for the new commit path; the InMemoryCatalog tests do not cover it.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1889,6 +1907,153 @@ public void validateRowLevelDmlMode(ConnectorSession
session, ConnectorTableHand
}
}
+ /**
+ * 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) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ String qualifiedName = iceHandle.getDbName() + "." +
iceHandle.getTableName();
+ try {
+ return executeAuthenticated(() -> {
+ // Load fresh rather than through the read-side caches: the
delete commits against the current
+ // metadata, never a possibly stale cached snapshot.
+ Table table = catalogOps.loadTable(iceHandle.getDbName(),
iceHandle.getTableName());
Review Comment:
[P2] Release the freshly loaded table on every exit
This extra load is neither registered with a statement/cache cleanup scope
nor passed through catalogOps.withTable. Glue/S3Tables tables own their FileIO,
and REST tables can own vended FileIO; CatalogBackedIcebergCatalogOps.withTable
runs the configured cachedTableCleanup in finally for exactly these resources.
Success, unsupported-predicate fallback, partial-file fallback and exceptions
all leave this fresh FileIO unclosed, accumulating clients and
credential-provider resources across API calls. Use the bounded withTable
lifecycle while retaining a fresh table load.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1889,6 +1907,153 @@ public void validateRowLevelDmlMode(ConnectorSession
session, ConnectorTableHand
}
}
+ /**
+ * 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) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ String qualifiedName = iceHandle.getDbName() + "." +
iceHandle.getTableName();
+ try {
+ return executeAuthenticated(() -> {
+ // Load fresh rather than through the read-side caches: the
delete commits against the current
+ // metadata, never a possibly stale cached snapshot.
+ Table table = catalogOps.loadTable(iceHandle.getDbName(),
iceHandle.getTableName());
+ Expression deleteExpr = predicate.isPresent()
+ ? wholeFileDeleteExpression(table, session,
predicate.get())
+ : Expressions.alwaysTrue();
+ if (deleteExpr == null || !matchesWholeFilesOnly(table,
deleteExpr)) {
+ return OptionalLong.empty();
+ }
+ DeleteFiles deleteFiles =
table.newDelete().deleteFromRowFilter(deleteExpr);
+ deleteFiles.commit();
+ return
OptionalLong.of(deletedRowCount(table.currentSnapshot()));
Review Comment:
[P2] Read the summary of this delete's committed snapshot
DeleteFiles.commit refreshes its TableOperations after committing; Iceberg
explicitly looks up its own saved snapshot by ID because another writer may
have committed before that refresh. Thus table.currentSnapshot() here can be
another writer's snapshot. If this delete removes 100 rows and an append wins
that interval, deletedRowCount sees the append summary and reports zero; a
concurrent delete can supply its count instead. Retain the actual committed
snapshot identity and read that summary across commit retries.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1889,6 +1907,153 @@ public void validateRowLevelDmlMode(ConnectorSession
session, ConnectorTableHand
}
}
+ /**
+ * 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) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ String qualifiedName = iceHandle.getDbName() + "." +
iceHandle.getTableName();
+ try {
+ return executeAuthenticated(() -> {
+ // Load fresh rather than through the read-side caches: the
delete commits against the current
+ // metadata, never a possibly stale cached snapshot.
+ Table table = catalogOps.loadTable(iceHandle.getDbName(),
iceHandle.getTableName());
+ Expression deleteExpr = predicate.isPresent()
+ ? wholeFileDeleteExpression(table, session,
predicate.get())
+ : Expressions.alwaysTrue();
+ if (deleteExpr == null || !matchesWholeFilesOnly(table,
deleteExpr)) {
+ return OptionalLong.empty();
+ }
+ DeleteFiles deleteFiles =
table.newDelete().deleteFromRowFilter(deleteExpr);
+ deleteFiles.commit();
+ return
OptionalLong.of(deletedRowCount(table.currentSnapshot()));
+ });
+ } catch (ValidationException e) {
+ // A racing writer turned a candidate file into a partial match
(or another delete 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);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ private Expression wholeFileDeleteExpression(Table table, ConnectorSession
session,
+ ConnectorExpression predicate) {
+ if (predicate instanceof ConnectorLiteral) {
+ return Boolean.TRUE.equals(((ConnectorLiteral)
predicate).getValue())
+ ? Expressions.alwaysTrue() : null;
+ }
+ ZoneId zone = IcebergTimeUtils.resolveSessionZone(session);
+ List<Expression> predicates = new IcebergPredicateConverter(
+ 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;
+ }
+
+ /** 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;
+ }
+ }
+
+ /**
+ * Rows deleted by a metadata delete: the removed data files' records
minus the records of the delete files
+ * removed with them (rows that were already deleted). Mirrors Trino's
{@code executeDelete} accounting.
+ */
+ private static long deletedRowCount(Snapshot snapshot) {
+ if (snapshot == null || snapshot.summary() == null) {
+ return 0L;
+ }
+ Map<String, String> summary = snapshot.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"));
+ return Math.max(0L, deletedRecords - removedPosDeletes -
removedEqDeletes);
Review Comment:
[P2] Do not equate delete-file records with hidden data rows
An equality-delete record is a key, not a count of matching data rows. For a
ten-row v2 data file with eight rows matching id=7 and a newer one-record
equality-delete file for that key, only two rows remain visible. An unfiltered
metadata delete removes both files, but this calculation reports 10-1=9
affected rows. Overlapping position deletes can also hide one row more than
once. Restrict exact metadata counting to cases whose cardinality is
guaranteed, or fall back when the exact number of live rows cannot be derived.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorWriteOps.java:
##########
@@ -55,6 +58,39 @@ default void validateRowLevelDmlMode(ConnectorSession
session, ConnectorTableHan
// 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. {@code Optional.empty()} means the statement has no
{@code WHERE} clause at all, i.e. every
+ * row of the table is deleted. 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, or empty for an
+ * unfiltered delete (always true)
+ * @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,
Review Comment:
[P2] Advance the connector API major for the new SPI method
This adds a public ConnectorMetadata method through ConnectorWriteOps, but
connector.plugin.api.version and its pinned test remain 17.0. The loader
requires equal majors and resolves SPI types from the FE, so the same declared
version now describes two different APIs and cannot reject a plugin built
against the new method on an older FE. Increment the major and update the
version pin as required by the explicit connector API contract. The metadata
baseline is updated; the second baseline does not enumerate this interface and
has no corresponding signature drift.
##########
fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergMetadataDeleteTest.java:
##########
@@ -0,0 +1,320 @@
+// 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"))));
+
+ 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());
+
+ 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))));
+
+ 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))));
+
+ Assertions.assertFalse(deleted.isPresent(), "an unliftable predicate
must not be treated as always-true");
+ Assertions.assertEquals(snapshotBefore,
reload(catalog).currentSnapshot().snapshotId());
+ }
+
+ @Test
+ public void deletedRowCountExcludesRowsHiddenByExistingDeleteFiles() {
+ InMemoryCatalog catalog = freshCatalog();
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("region").build();
+ Table table = catalog.createTable(TABLE_ID, SCHEMA, spec,
props("format-version", "3"));
+ table.newAppend()
+ .appendFile(partitionedDataFile(spec,
"s3://b/db1/t1/region=us/a.parquet", 10L, "region=us"))
+ .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",
+ "s3://b/db1/t1/region=us/a.parquet", 3L)).commit();
+
+ OptionalLong deleted = metadata(catalog).deleteUsingMetadata(SESSION,
HANDLE, Optional.of(
+ new ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("region",
ConnectorType.of("STRING")), ConnectorLiteral.ofString("us"))));
+
+ Assertions.assertEquals(OptionalLong.of(7L), deleted,
+ "deleted rows = the dropped file's records minus the rows its
deletion vector already hid");
+ Assertions.assertTrue(dataFilePaths(reload(catalog)).isEmpty());
+ Assertions.assertEquals(0, liveDeleteFileCount(reload(catalog)),
+ "the deletion vector of the dropped data file is removed as
well");
+ }
+
+ 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, String referencedDataFile,
+ long cardinality) {
+ return FileMetadata.deleteFileBuilder(spec)
+ .ofPositionDeletes()
+ .withPath(puffin)
+ .withFormat(FileFormat.PUFFIN)
+ .withFileSizeInBytes(128L)
+ .withRecordCount(cardinality)
+ .withReferencedDataFile(referencedDataFile)
Review Comment:
[P2] Give the deletion vector its referenced file's partition
The data file in this test has region=us, but this helper never sets the DV
partition, so FileMetadata builds it with region=null. Iceberg requires the DV
and data file to have the same spec and partition tuple; normal DeleteFileIndex
lookup rejects this pair as corrupted metadata. The test avoids planning the
original table and can still pass after the whole-partition fast path removes
both entries, so it does not prove the advertised seven-live-row case. Copy/set
the data file partition on the DV and check that a pre-delete scan attaches it
to the data file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
