github-actions[bot] commented on code in PR #69176:
URL: https://github.com/apache/doris/pull/69176#discussion_r4312505273
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -1938,6 +1969,299 @@ 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,
+ 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();
+ }
+ 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 (!readLiveDeleteFiles(scanTable, base).isEmpty()) {
+ // 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 || !readLiveDeleteFiles(scanTable,
refreshed).isEmpty()) {
+ 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
+ // fences the final CAS and every retry replay against a table/schema
generation change.
+ Transaction transaction = IcebergConnectorTransaction.openTransaction(
+ table, context, null, false, boundWriteMetadataIdentity);
+ 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;
+ }
+ }
+
+ /** All live delete files of the base snapshot, copied so they survive the
manifest readers. */
+ private static List<DeleteFile> readLiveDeleteFiles(Table table, Snapshot
snapshot) {
+ List<DeleteFile> files = new ArrayList<>();
+ try {
+ for (ManifestFile manifest : snapshot.deleteManifests(table.io()))
{
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(),
table.specs())) {
+ for (DeleteFile file : reader) {
+ files.add(file.copy());
Review Comment:
[P2] Stop reading delete manifests after the first live entry
Both callers only test whether this list is empty, and the first live delete
file already forces the row-level fallback. On a partitioned v2/v3 table with
many position deletes or deletion vectors, even a DELETE of one small partition
now reads every delete manifest and retains a copy of every live entry across
unrelated partitions before returning empty. This adds full-table manifest IO
and O(total live delete files) FE heap to ordinary deletes that cannot use the
optimization. Replace this collecting helper with a presence check that returns
on the first live entry while still closing the reader.
--
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]
