github-actions[bot] commented on code in PR #69176:
URL: https://github.com/apache/doris/pull/69176#discussion_r4312052770
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlCommand.java:
##########
@@ -80,10 +87,26 @@ public void run(ConnectContext ctx, StmtExecutor
stmtExecutor) throws Exception
Callable<Void> execute = () -> {
LogicalPlanAdapter logicalPlanAdapter = new
LogicalPlanAdapter(plan, ctx.getStatementContext());
NereidsPlanner planner = new
NereidsPlanner(ctx.getStatementContext());
+ if (op == RowLevelDmlOp.DELETE) {
+ captureAnalyzedPlan(ctx);
+ }
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 (op == RowLevelDmlOp.DELETE &&
args.getPartitions().isEmpty() && !args.isTempPart()) {
+ OptionalLong metadataDeletedRows =
+ transform.deleteUsingMetadata(ctx, args,
planner.getAnalyzedPlan());
Review Comment:
[P1] Preserve cancellation before committing a metadata delete
If KILL QUERY or the timeout checker cancels a DELETE while planner.plan is
still loading remote metadata, StmtExecutor retains the cancellation reason
without interrupting the planning thread. The ordinary row-level path publishes
its coordinator in AbstractInsertExecutor.executeSingleInsert, replays that
cancellation, and checks terminal status again before commit. This branch
bypasses those fences and directly invokes a committing connector method, so an
already-cancelled whole-partition DELETE can still remove data and report OK.
Cancellation during the whole-file preflight is also invisible to the
connector. Check the retained status before dispatch and carry a cancellation
check through to the remote commit, preserving the original cancellation result.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1938,6 +1970,324 @@ 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, String
boundWriteMetadataIdentity) {
+ 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)));
+ } 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) {
+ 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();
+ }
+ long baseSnapshotId = base.snapshotId();
+ List<DeleteFile> liveDeletes = readLiveDeleteFiles(table, base);
+ List<DataFile> candidates = new ArrayList<>();
+ if (!matchesWholeFilesOnly(table, deleteExpr, !liveDeletes.isEmpty(),
candidates)) {
+ return OptionalLong.empty();
+ }
+ if (!liveDeletes.isEmpty() && !onlyExactlyCountedDeletesApply(table,
liveDeletes, candidates)) {
+ // The affected rows cannot be derived exactly from the snapshot
summary; keeping the row-level path
+ // keeps the reported affected-row count exact.
+ LOG.info("Skipping metadata delete of {}: affected files carry
delete files whose row count is not "
+ + "exact (legacy position or equality deletes).",
table.name());
+ 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.
+ table.refresh();
+ validateBoundWriteGeneration(table, boundWriteMetadataIdentity);
+
+ // 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.
+ Transaction transaction =
IcebergConnectorTransaction.openTransaction(table, context, null, false);
+ DeleteFiles deleteFiles =
transaction.newDelete().deleteFromRowFilter(deleteExpr);
+ deleteFiles.commit();
+ transaction.commitTransaction();
+ Snapshot committed = transaction.table().currentSnapshot();
+ if (committed == null || committed.snapshotId() == baseSnapshotId) {
+ // The predicate matched nothing; DeleteFiles does not create an
empty snapshot.
+ return OptionalLong.of(0L);
+ }
+ return OptionalLong.of(deletedRowCount(committed));
+ }
+
+ /** 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.
+ *
+ * @param collectCandidates whether the candidate data files must be
materialized (needed by the affected-row
+ * check when the table already has delete files)
+ */
+ private static boolean matchesWholeFilesOnly(Table table, Expression
deleteExpr, boolean collectCandidates,
+ List<DataFile> candidates) {
+ boolean caseSensitive = false;
+ boolean wholePartitions = ExpressionUtil.selectsPartitions(deleteExpr,
table, caseSensitive);
+ if (wholePartitions && !collectCandidates) {
+ return true;
+ }
+ TableScan scan = table.newScan()
+ .filter(deleteExpr)
+ .caseSensitive(caseSensitive)
+ .includeColumnStats()
+ .ignoreResiduals();
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
Review Comment:
[P2] Authenticate the candidate scan before planning manifests
On secured HDFS with multiple manifests, a non-partition predicate (or any
live delete file requiring candidate collection) reaches this scan on the raw
table from withTable. executeAuthenticated covers the caller, but Iceberg
DataTableScan/ManifestGroup opens manifest inputs on its shared workers; those
workers need IcebergAuthenticatedFileIO, as the existing wrapTableForScan path
documents. The wrapper installed later by openTransaction cannot cover this
preflight, and its worker RuntimeException bypasses the IOException fallback
and fails the DELETE. This is separate from the earlier commit-worker issue:
that commit site is now wrapped, while this earlier scan is not. Prepare an
authenticated table/FileIO view before candidate planning and reuse the
existing scan-side pattern.
--
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]
