sunchao commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3998601057
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,583 @@ +/* + * 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.comet.contrib.delta + +import scala.jdk.CollectionConverters._ + +import org.apache.hadoop.fs.Path +import org.apache.spark.internal.Logging +import org.apache.spark.sql.catalyst.expressions.Literal +import org.apache.spark.sql.comet.{CometScanExec, DeltaPlanDataInjector} +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.RowIndexFilterType +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ScalarSubquery => ExecScalarSubquery} +import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile} +import org.apache.spark.sql.types.{ByteType, LongType, MetadataBuilder, StructField, StructType} + +import org.apache.comet.objectstore.NativeConfig +import org.apache.comet.serde.OperatorOuterClass +import org.apache.comet.serde.OperatorOuterClass.Operator +import org.apache.comet.serde.QueryPlanSerde.{exprToProto, serializeDataType} +import org.apache.comet.serde.operator.{literalToProto, partition2Proto, schema2Proto, CometNativeScan} +import org.apache.comet.shims.ShimFileFormat + +/** + * Serde for the native Delta scan. Two shapes: + * - Plain reads reuse core's `NativeScanCommon` builder wholesale. + * - Deletion-vector reads: Delta's planner appends `__delta_internal_is_row_deleted` (tinyint) + * and Spark's row-index temp column (bigint) to the read schema and filters on is_row_deleted + * above the scan. The native reader applies the DV as a row selection, so both internal + * columns are emitted as per-file constants (0), the parquet read schema is stripped to the + * real data columns, and the DV descriptor ships per file for native to fetch and decode. + */ +object CometDeltaNativeScan + extends Logging + with org.apache.spark.sql.catalyst.expressions.PredicateHelper { + + val IsRowDeletedColumn: String = DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME + val RowIndexColumn: String = ShimFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME + + private[delta] val internalColumnNames: Set[String] = Set(IsRowDeletedColumn, RowIndexColumn) + + // Prefix for the internal columns' slots in the partition schema, mirroring core's + // _comet_metadata_ prefix rationale: DataFusion matches partition columns by name. + // [[allocateUniqueInternalFields]] additionally suffixes on collision with a real column. + private val deltaConstFieldPrefix = "_comet_delta_" + + def isDvShape(scanExec: FileSourceScanExec): Boolean = + scanExec.requiredSchema.exists(f => internalColumnNames.contains(f.name)) + + private def deltaFormat(scanExec: FileSourceScanExec): DeltaParquetFileFormat = + scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + + private def columnMappingMode(scanExec: FileSourceScanExec): String = + deltaFormat(scanExec).metadata.columnMappingMode.name + + /** + * Under column mapping, parquet files store physical column names (stable UUIDs / ids), so the + * schemas passed to the native parquet reader must be physical. Positions and structure are + * preserved, so output binding and projection are unaffected. The scan's internal DV columns + * are not part of the table schema and must be stripped before calling this. + * + * `private[delta]` (not `private`): [[DeltaScanSupport.declineReason]]'s non-ASCII + * case-insensitive name gate reuses this exact conversion to compute the names native sees + * under column mapping, rather than re-deriving physical names with separate logic. + */ + private[delta] def toPhysical(scanExec: FileSourceScanExec, schema: StructType): StructType = { + val format = deltaFormat(scanExec) + if (format.metadata.columnMappingMode.name == "none") { + schema + } else { + // Name mode matches file columns by physical NAME. Strip the parquet.field.id metadata + // createPhysicalSchema also stamps: files written before the column-mapping upgrade have + // no field ids and would fail the reader's id expectations. + stripFieldIds(org.apache.spark.sql.delta.DeltaColumnMapping + .createPhysicalSchema(schema, format.metadata.schema, format.metadata.columnMappingMode)) + } + } + + private def stripFieldIds(schema: StructType): StructType = { + import org.apache.spark.sql.types._ + def stripType(dt: DataType): DataType = dt match { + case s: StructType => stripFieldIds(s) + case a: ArrayType => a.copy(elementType = stripType(a.elementType)) + case m: MapType => + m.copy(keyType = stripType(m.keyType), valueType = stripType(m.valueType)) + case other => other + } + StructType(schema.fields.map { f => + val metadata = new MetadataBuilder() + .withMetadata(f.metadata) + .remove("parquet.field.id") + // Sibling key Delta stamps on array/map fields under IcebergCompat/Uniform. + .remove("parquet.field.nested.ids") + .build() + f.copy(dataType = stripType(f.dataType), metadata = metadata) + }) + } + + /** + * Build the planning-time `DeltaScan` operator (common data only; file partitions are injected + * lazily at execution). Returns None when an output data type cannot be serialized or the plan + * shape is not one we can translate faithfully. `memo` is the same claim-memo instance + * [[DeltaScanSupport.declineReason]] populated on this claim; its `hadoopConf` and + * `dvDescriptors` are reused here rather than recomputed. + */ + def convert( + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + memo: DeltaScanSupport.DeltaClaimMemo): Option[Operator] = { + val relation = scanExec.relation + + val firstFileUri = scanHelper.selectedPartitions + .flatMap(_.files.headOption) + .headOption + .map(_.getPath.toUri) + + val hadoopConf = memo.hadoopConf + + val tableRootPath = relation.location.rootPaths.head + val tableRoot = tableRootPath.toString + + val commonOpt = if (!isDvShape(scanExec)) { + // Under column mapping (name mode) the parquet reader must see physical names; + // positions are preserved so output binding and projection stay untouched. + CometNativeScan.buildNativeScanCommon( + source = scanExec.simpleStringWithNodeId(), + output = scanExec.output, + requiredSchema = toPhysical(scanExec, scanExec.requiredSchema), + dataSchema = toPhysical(scanExec, relation.dataSchema), + partitionSchema = toPhysical(scanExec, relation.partitionSchema), + fileConstantMetadataColumns = scanExec.fileConstantMetadataColumns, + dataFilters = scanHelper.supportedDataFilters, + firstFileUri = firstFileUri, + hadoopConf = hadoopConf, + conf = scanExec.conf) + } else { + buildDvScanCommon(scanExec, scanHelper, firstFileUri, hadoopConf) + } + + commonOpt.map { commonBuilder => + // Already forced by declineReason on this claim; reused rather than deserialized again. + val dvDescriptors = memo.dvDescriptors + // Union object-store options over every authority a partition of this scan may need a + // store for, not just the first data file's scheme. + commonBuilder.putAllObjectStoreOptions( + mergedObjectStoreOptions( + hadoopConf, + storeUris(dvDescriptors, tableRootPath, firstFileUri)).asJava) + + val common = commonBuilder.build() + // Effective session rebase read modes, resolved through ParquetOptions exactly as + // ParquetFileFormat.buildReaderWithPartitionValues resolves them (per-relation + // `datetimeRebaseMode` / `int96RebaseMode` options win over the session conf, whose + // per-Spark-version default -- EXCEPTION on 3.x, CORRECTED on 4.0 -- SQLConf supplies). + // Native consults them only for files whose footer metadata does not decide the rebase + // policy on its own, mirroring DataSourceUtils.getRebaseSpec's modeByConfig fallback. + val parquetReadOptions = + new org.apache.spark.sql.execution.datasources.parquet.ParquetOptions( + relation.options, + scanExec.conf) + val deltaCommon = OperatorOuterClass.DeltaSparkScanCommon + .newBuilder() + .setTableRoot(tableRoot) + .setColumnMappingMode(columnMappingMode(scanExec)) + .setSourceKey(DeltaPlanDataInjector.sourceKey(tableRoot, common)) + .setDatetimeRebaseModeInRead(parquetReadOptions.datetimeRebaseModeInRead) + .setInt96RebaseModeInRead(parquetReadOptions.int96RebaseModeInRead) + .build() + val deltaScan = OperatorOuterClass.DeltaSparkScan + .newBuilder() + .setCommon(common) + .setDeltaCommon(deltaCommon) + Operator + .newBuilder() + .setPlanId(scanExec.id) + .setContribScan(DeltaSparkScanEnvelope.pack(deltaScan.build())) + .build() + } + } + + /** + * One representative store URI per distinct object-store authority this scan's partitions may + * need options for: the data-file authority (`firstFileUri`), the table root unconditionally + * (UUID-relative DV sidecars resolve against it), and every distinct on-disk DV authority from + * `descriptors` (inline DVs carry no external URI and are filtered out). Deduping by authority + * rather than full URI keeps this O(distinct authorities) instead of O(files), keeping the + * FIRST URI seen per authority so `firstFileUri`/the table root win over a same-authority DV + * path. + */ + private[delta] def storeUris( + descriptors: Seq[DeletionVectorDescriptor], + tableRootPath: Path, + firstFileUri: Option[java.net.URI]): Seq[java.net.URI] = { + val dvAuthorityUris = descriptors + .filter(_.storageType != DeletionVectorDescriptor.INLINE_DV_MARKER) + .map(_.absolutePath(tableRootPath).toUri) + val candidates = firstFileUri.toSeq ++ Seq(tableRootPath.toUri) ++ dvAuthorityUris + val byAuthority = scala.collection.mutable.LinkedHashMap.empty[String, java.net.URI] + candidates.foreach(uri => + byAuthority.getOrElseUpdate(DeltaScanSupport.uriAuthority(uri), uri)) + byAuthority.values.toSeq + } + + /** + * Unions `NativeConfig.extractObjectStoreOptions` over every `uris` authority. Safe to union + * rather than pick one: extracted keys are scheme-disjoint prefixes (`fs.s3a.*` vs + * `fs.azure.*`, ...), so options for different schemes never collide, and re-extracting the + * same scheme from two URIs is idempotent. + */ + private[delta] def mergedObjectStoreOptions( + hadoopConf: org.apache.hadoop.conf.Configuration, + uris: Seq[java.net.URI]): Map[String, String] = + uris.foldLeft(Map.empty[String, String]) { (merged, uri) => + merged ++ NativeConfig.extractObjectStoreOptions(hadoopConf, uri) + } + + /** + * Harvest subquery-bearing predicates for this scan from its covering FilterExec. Spark 3.x + * strips them from a scan's `dataFilters` at planning (`FileSourceStrategy` routes them to the + * post-scan filter only), while Spark 4.x keeps them in `dataFilters`; collecting them here at + * claim time gives the execution-time resolve-and-push path the same inputs on every version, + * and the dedup below keeps Spark 4.x from carrying duplicates. Reference containment alone + * does not prove pushing a predicate down is semantics-preserving, so `spineToScan` also + * requires every intervening operator to commute with the push (an intervening + * LIMIT/Sort/Aggregate/join etc. stops the walk and leaves the filter where Spark placed it: + * missed pruning only). + */ + def subqueryFiltersFromParent( + plan: org.apache.spark.sql.execution.SparkPlan, + scanExec: FileSourceScanExec): Seq[org.apache.spark.sql.catalyst.expressions.Expression] = { + import org.apache.spark.sql.catalyst.expressions.{PlanExpression, SubqueryExpression} + import org.apache.spark.sql.execution.{FilterExec, ProjectExec, SparkPlan} + + // Whether every node from `node` down to `scanExec` is one pushdown can safely cross: a + // deterministic ProjectExec is 1:1 on rows and a deterministic FilterExec only removes rows, + // so moving a predicate over the scan's output through either preserves semantics -- mirroring + // Spark's own PushPredicateThroughNonJoin/CollapseProject rules. A nondeterministic node (or + // anything else: LIMIT/TopN, Sort, Aggregate, Window, joins, ...) can change which rows survive + // to matter, so it stops the walk and the filter is left uncollected (missed pruning only). + def spineToScan(node: SparkPlan): Boolean = node match { + case n if n eq scanExec => true + case p: ProjectExec if p.projectList.forall(_.deterministic) => spineToScan(p.child) + case f: FilterExec if f.condition.deterministic => spineToScan(f.child) + case _ => false + } + + // Nearest FilterExec whose spine down to the scan is Project/Filter-only (the DV shape + // interposes such nodes between them, so do not require a direct parent-child edge). + val filtersAboveScan = plan.collect { + case f: FilterExec if spineToScan(f.child) => f + } + filtersAboveScan.lastOption + .map { f => + splitConjunctivePredicates(f.condition) + .filter(_.deterministic) + .filter(_.references.subsetOf(scanExec.outputSet)) + .filter(p => + SubqueryExpression.hasSubquery(p) || p.exists(_.isInstanceOf[PlanExpression[_]])) + .filterNot(p => scanExec.dataFilters.exists(_.semanticEquals(p))) + } + .getOrElse(Seq.empty) + } + + /** + * Execution-time scalar-subquery data filters of a scan. `hasResolvedFilters` is true whenever + * pushdown is enabled and such filters exist, whether or not they bind or serialize; `protos` + * holds only the ones that serialized. + */ + case class ResolvedSubqueryFilters( + hasResolvedFilters: Boolean, + protos: Seq[org.apache.comet.serde.ExprOuterClass.Expr]) + + private val NoResolvedSubqueryFilters = ResolvedSubqueryFilters(false, Seq.empty) + + /** + * Resolve scalar-subquery data filters at execution time and serialize them for native + * pushdown, mirroring `CometNativeScanExec.serializedPartitionData`. `supportedDataFilters` + * excludes PlanExpressions at planning time (subquery results do not exist yet), so these + * bounds reach the native reader only through this path. Filters that fail to serialize are + * skipped: Spark keeps a covering FilterExec above the scan, so this is missed pruning only. + * Their presence is still reported, since native keys the safe timestamp conversion on the scan + * being filtered at all, as the core scan does for its resolved filters. + * + * Known core-parity limitation: when fused under a parent native operator, + * `ensureSubqueriesResolved` has already called `updateResult()` on these subqueries and this + * path calls it again (`ScalarSubquery.updateResult` re-executes unconditionally); benign here + * since the subquery's snapshot is pinned at analysis, but wasteful. Fix belongs in core. + */ + def resolvedSubqueryFilters( + dataFilters: Seq[org.apache.spark.sql.catalyst.expressions.Expression], + output: Seq[org.apache.spark.sql.catalyst.expressions.Attribute], + requiredSchema: StructType, + conf: org.apache.spark.sql.internal.SQLConf): ResolvedSubqueryFilters = { + if (!conf.getConf(org.apache.spark.sql.internal.SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) { + return NoResolvedSubqueryFilters + } + val subqueryFilters = dataFilters.filter(_.exists(_.isInstanceOf[ExecScalarSubquery])) + if (subqueryFilters.isEmpty) { + return NoResolvedSubqueryFilters + } + // Same binding guard as the DV shape's plan-time filters: references limited to the + // data-column prefix of the output, where positions agree with the native read schema. + // Guard BEFORE updateResult so discarded filters never execute their subqueries. + val strippedLen = requiredSchema.count(f => !internalColumnNames.contains(f.name)) + val dataColIds = output.take(strippedLen).map(_.exprId).toSet + val pushableFilters = + subqueryFilters.filter(_.references.forall(r => dataColIds.contains(r.exprId))) + pushableFilters.foreach(_.foreach { + case s: ExecScalarSubquery => s.updateResult() + case _ => + }) + val protos = pushableFilters + .flatMap { filter => + // MergeScalarSubqueries can fuse several scalar subqueries into one struct-returning + // subquery accessed via GetStructField; fold that whole subtree to a literal (a bare + // GetStructField-over-Literal would not serialize). + val resolved = filter.transform { + case g @ org.apache.spark.sql.catalyst.expressions + .GetStructField(_: ExecScalarSubquery, _, _) => + Literal.create(g.eval(null), g.dataType) + case s: ExecScalarSubquery => + Literal.create(s.eval(null), s.dataType) + } + val proto = exprToProto(resolved, output) + if (proto.isEmpty) { + logWarning(s"Could not serialize resolved scalar subquery filter: $resolved") + } + proto + } + ResolvedSubqueryFilters(hasResolvedFilters = true, protos) + } + + /** + * Allocate the partition-schema slots for the DV shape's internal columns + * (`internalColumnNames`), with names collision-free against the physical data schema, the + * physical partition schema, and the constant-metadata slots already allocated for this scan + * (plus each other): DataFusion substitutes partition constants BY NAME, so an unprefixed, + * un-uniquified slot could collide with a real column and silently replace its data with the + * bookkeeping constant. `buildDvScanCommon` keys `internalIndexByName` by each field's ORIGINAL + * name from `requiredSchema`, so the renaming here only changes the proto's field name. + */ + private[delta] def allocateUniqueInternalFields( + requiredSchema: StructType, + physicalDataSchema: StructType, + physicalPartitionSchema: StructType, + constantMetadataFields: Seq[StructField]): Seq[StructField] = { + val reserved = scala.collection.mutable.LinkedHashSet[String]() + reserved ++= physicalDataSchema.fields.map(_.name) + reserved ++= physicalPartitionSchema.fields.map(_.name) + reserved ++= constantMetadataFields.map(_.name) + requiredSchema.fields.toSeq + .filter(f => internalColumnNames.contains(f.name)) + .map { f => + var name = s"$deltaConstFieldPrefix${f.name}" + while (reserved.contains(name)) { + name = name + "_" + } + reserved += name + StructField(name, f.dataType, f.nullable) + } + } + + /** + * DV shape common builder. Layout invariants (declined by DeltaScanSupport when violated): scan + * output = requiredSchema attrs (data columns, then the internal columns as a suffix) followed + * by partition and constant-metadata columns. The parquet read schema strips the internal + * columns; they are appended to the partition schema as per-file constants, so the projection + * vector routes them from the constants block. + */ + private def buildDvScanCommon( Review Comment: ### 3. Reuse ordinary-column construction in the DV builder Could ordinary-column construction in this path share the same builder as `CometNativeScan.buildNativeScanCommon`, with an explicit input or result describing the generated columns and visible projection? Configuration flags and collision-free metadata-name allocation are already shared; filter binding, output types, schemas, and projection assembly still have a separate DV implementation. One possible approach is to build a view without generated columns, reuse the common builder, and then append their slots and restore output order by attribute identity. A smaller extraction is also reasonable if it gives ordinary and DV scans one implementation of the shared serialization rules. This can preserve the current suffix requirement, defaults fallback, and DML/liveness checks. Could tests compare ordinary-column construction across both paths and retain predicate-presence, projection-order, and synthetic-name collision coverage? That would let later serializer fixes reach both paths without expanding DV admission in this PR. ########## contrib/delta-spark/src/main/scala/org/apache/spark/sql/comet/DeltaPlanDataInjector.scala: ########## @@ -0,0 +1,86 @@ +/* + * 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.spark.sql.comet + +import scala.jdk.CollectionConverters._ + +import org.apache.comet.contrib.delta.DeltaSparkScanEnvelope +import org.apache.comet.serde.{OperatorOuterClass, QueryContextInterner} +import org.apache.comet.serde.OperatorOuterClass.Operator + +/** + * PlanDataInjector for the Delta contrib scan, discovered by core's ServiceLoader (see the + * `META-INF/services` resource). Lives in this package because [[PlanDataInjector]] is + * `private[comet]`. + */ +class DeltaPlanDataInjector extends PlanDataInjector { + + override val opStructCase: Operator.OpStructCase = Operator.OpStructCase.CONTRIB_SCAN + + override def canInject(op: Operator): Boolean = Review Comment: ### 6. Add focused envelope and provider contract tests As a non-blocking maintainability improvement, could we add focused tests for the serialized Delta Spark envelope and the real `DeltaPlanDataInjector` through the registry? These could assert operator slot 200, the distinct Spark/Kernel type URLs, and expected schema, projection, and filter values in a representative common payload. Provider tests could check common/partition assembly, preservation of outer operator fields and children, rejection of another contrib's envelope, and an empty injected partition no longer qualifying for injection. The generic suites and successful scan tests already exercise the extension machinery, and injection completion and native feature-off rejection are implemented. These additional assertions would pin the Delta-specific boundary during refactoring without introducing a new capability API or a general mixed-version compatibility commitment. ########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,1838 @@ +/* + * 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.comet.contrib.delta + +import java.io.IOException +import java.net.URI +import java.util.Locale + +import scala.collection.mutable.{ListBuffer, Map => MutableMap} +import scala.jdk.CollectionConverters._ + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.Path +import org.apache.spark.sql.catalyst.expressions.{Alias, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName} +import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData} +import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues +import org.apache.spark.sql.comet.CometScanExec +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec, SparkPlan} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType} + +import org.apache.comet.CometConf +import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES +import org.apache.comet.objectstore.NativeConfig +import org.apache.comet.parquet.CometParquetUtils +import org.apache.comet.rules.{CometScanRule, CometScanTypeChecker} +import org.apache.comet.serde.operator.CometNativeScan +import org.apache.comet.shims.ShimFileFormat + +/** + * Claim/decline gates for the native Delta scan. Correctness rule: when in doubt, decline, + * Spark's Delta reader handles the scan and results stay correct, just unaccelerated. + */ +object DeltaScanSupport { + + /** + * Reader features the native path understands; anything else on the protocol declines the + * table. `deletionVectors`/`columnMapping` are declined separately below for specific reasons. + */ + private val understoodReaderFeatures: Set[String] = + Set("columnMapping", "deletionVectors", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck") + + /** + * Is this exactly Delta's DSv1 parquet format? Compared by class name, not `classOf`: a + * `classOf` reference would raise `NoClassDefFoundError` and break every parquet scan when + * delta-spark is absent from the classpath. + */ + def isDeltaScan(scanExec: FileSourceScanExec): Boolean = + scanExec.relation.fileFormat.getClass.getName == + "org.apache.spark.sql.delta.DeltaParquetFileFormat" + + /** + * Claim-time artifacts [[declineReason]] already computes but [[CometDeltaNativeScan.convert]] + * also needs -- threaded through by reference (populated only on the claimable path, right + * before `declineReason` returns `None`) so a claimed scan does not pay to recompute either: + * the Hadoop conf ([[org.apache.spark.sql.internal.SessionState#newHadoopConfWithOptions]] is + * not cheap) and the deletion-vector descriptors (base64-decoded, non-trivial only for DV-shape + * scans). One instance is created per claim attempt in `DeltaScanContrib` and passed to both + * `declineReason` and `convert`. + */ + private[delta] final class DeltaClaimMemo { + var hadoopConf: Configuration = _ + var dvDescriptors: Seq[DeletionVectorDescriptor] = Seq.empty + } + + /** + * First reason this Delta scan cannot go native, or None when claimable (in which case `memo` + * is populated for [[CometDeltaNativeScan.convert]] to reuse). Only called when [[isDeltaScan]] + * is true. `scanHelper` is the [[CometScanExec]] built to drive `convert` on a claim, reused + * for the multi-store gate below. + */ + def declineReason( + plan: SparkPlan, + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + memo: DeltaClaimMemo): Option[String] = { + val format = scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + val protocol = format.protocol + val metadata = format.metadata + // Name mode is supported via physical-name schemas; id mode needs the field-id path and + // stays declined until validated. Hoisted here since several gates below reuse it. + val cmMode = metadata.columnMappingMode.name + // Descriptor deserialization is expensive, so hoist it into a `lazy val`, forced at most + // once in this method; on the claimable path the result is handed to `convert` through + // `memo` below, so a claimed scan deserializes the descriptors exactly once end to end. + val tableRoot = scanExec.relation.location.rootPaths.head.toString + lazy val dvDescriptors: Seq[DeletionVectorDescriptor] = + selectedDvDescriptors(scanHelper, tableRoot) + + // Mirrors core's CometScanRule.isSchemaSupported so scan-time type gates (unsigned-small-int + // fallback, collation, shredded-variant-struct) apply identically here. Pure in-memory check, + // so it runs first, ahead of every I/O-bearing gate below. + val schemaFallbackReasons = new ListBuffer[String]() + val typeChecker = CometScanTypeChecker() + val requiredSchemaSupported = + typeChecker.isSchemaSupported(scanExec.requiredSchema, schemaFallbackReasons) + val partitionSchemaSupported = + typeChecker.isSchemaSupported(scanExec.relation.partitionSchema, schemaFallbackReasons) + if (!requiredSchemaSupported || !partitionSchemaSupported) { + return Some( + "Native Delta scan does not support the schema: " + schemaFallbackReasons.mkString(", ")) + } + + if (format.isCDCRead) { + return Some("Native Delta scan does not support Change Data Feed reads") + } + + // Delta's DML machinery (findTouchedFiles) disables reader optimizations and needs real + // row indexes from Spark's reader; claiming here would feed NULL indexes into DV construction. + if (!format.optimizationsEnabled) { + return Some("Native Delta scan does not support reads with reader optimizations disabled") + } + if (scanExec.requiredSchema.exists(_.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME) || + scanExec.relation.dataSchema.exists( + _.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME)) { + return Some("Native Delta scan does not support Delta's generated row-index column") + } + + if (cmMode != "none" && cmMode != "name") { + return Some(s"Native Delta scan does not support column mapping mode $cmMode") + } + // createPhysicalSchema wholesale-replaces field metadata, silently dropping EXISTS_DEFAULT. + if (cmMode == "name" && + getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with column mapping") + } + // createPhysicalSchema rewrites nested StructField names too, and the native builder emits the + // required schema verbatim as output, so name-sensitive expressions (e.g. to_json) would leak + // physical names. Decline until a rename adapter exists. + if (cmMode == "name" && + scanExec.requiredSchema.exists(f => containsNestedStruct(f.dataType))) { + return Some("Native Delta scan does not support column mapping with nested struct fields") + } + + val readerFeatures = protocol.readerFeatureNames + val unknownFeatures = readerFeatures -- understoodReaderFeatures + if (unknownFeatures.nonEmpty) { + return Some( + s"Native Delta scan does not support reader feature(s) ${unknownFeatures.mkString(", ")}") + } + + // Non-constant metadata columns are generated per-row by Spark's reader and unsupported, + // except Delta's DV bookkeeping columns, which the native path emits as constants. + val knownColNames = + scanExec.relation.dataSchema.map(_.name).toSet ++ + scanExec.relation.partitionSchema.map(_.name).toSet ++ + scanExec.fileConstantMetadataColumns.map(_.name).toSet ++ + CometDeltaNativeScan.internalColumnNames + val unknownOutput = scanExec.output.map(_.name).filterNot(knownColNames.contains) + if (unknownOutput.nonEmpty) { + return Some( + s"Native Delta scan does not support generated column(s) ${unknownOutput.mkString(", ")}") + } + + // Deletion-vector shape invariants (see CometDeltaNativeScan.buildDvScanCommon). + if (CometDeltaNativeScan.isDvShape(scanExec)) { + // A row-index column WITHOUT is_row_deleted is Delta DML bookkeeping (real row indexes), + // not a DV read; claiming it with a constant would corrupt the DVs being written. + val hasIsRowDeleted = + scanExec.requiredSchema.exists(_.name == CometDeltaNativeScan.IsRowDeletedColumn) + val hasRowIndex = + scanExec.requiredSchema.exists(_.name == CometDeltaNativeScan.RowIndexColumn) + if (hasRowIndex && !hasIsRowDeleted) { + return Some( + "Native Delta scan does not support row-index reads outside a deletion-vector scan") + } + // Internal columns must form a suffix of the read schema so data-column positions agree + // between Spark's output and the stripped native schema. + val names = scanExec.requiredSchema.fields.map(_.name) + val firstInternal = names.indexWhere(CometDeltaNativeScan.internalColumnNames.contains) + if (!names.drop(firstInternal).forall(CometDeltaNativeScan.internalColumnNames.contains)) { + return Some("Native Delta scan requires DV bookkeeping columns to trail the read schema") + } + // Native applies the DV itself and emits a dead constant for row-index, so the real value + // must be provably unused above the scan. + if (!rowIndexUnusedAbove(plan, scanExec)) { + return Some( + "Native Delta scan cannot supply _metadata.row_index values consumed by the query") + } + // The DV common builder does not serialize existence defaults yet. + if (getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with deletion vectors") + } + // Bounds native's memory for expanded DV row selectors (delta_dv.rs), pessimistically + // bounded by 2*cardinality + #row-groups; the conf below makes an over-pessimistic decline + // recoverable. + val maxDeletedRowsPerFile = DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.get() + val oversizedCardinalities = dvDescriptors + .map(_.cardinality) + .filter(_ > maxDeletedRowsPerFile) + if (oversizedCardinalities.nonEmpty) { + return Some( + "Native Delta scan does not support a deletion vector deleting " + + s"${oversizedCardinalities.max} rows in a single file, exceeding " + + s"${DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.key}=$maxDeletedRowsPerFile") + } + } + + // input_file_name & friends read from a thread-local Spark's FileScanRDD sets; the native scan Review Comment: ### 2. Share common native-reader admission checks Could the input-file-expression, vectorized-reader-setting, and nested-default checks come from a shared native-reader admission helper? These checks mirror core today, so a future fix otherwise needs a separate Delta update. The shared entry point can leave Delta's protocol checks, DV row-index exception, liveness checks, and stricter encryption fallback explicit in the contrib. Filesystem policy also differs; its broader consolidation can stay in #5658. The aim here is to share the checks that already have the same semantics. A parity test covering the common rejection cases for ordinary Parquet and Delta, together with supported-scan controls, would protect this boundary. Existing Delta-specific metadata fallbacks should remain covered separately. ########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,583 @@ +/* + * 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.comet.contrib.delta + +import scala.jdk.CollectionConverters._ + +import org.apache.hadoop.fs.Path +import org.apache.spark.internal.Logging +import org.apache.spark.sql.catalyst.expressions.Literal +import org.apache.spark.sql.comet.{CometScanExec, DeltaPlanDataInjector} +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.RowIndexFilterType +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ScalarSubquery => ExecScalarSubquery} +import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile} +import org.apache.spark.sql.types.{ByteType, LongType, MetadataBuilder, StructField, StructType} + +import org.apache.comet.objectstore.NativeConfig +import org.apache.comet.serde.OperatorOuterClass +import org.apache.comet.serde.OperatorOuterClass.Operator +import org.apache.comet.serde.QueryPlanSerde.{exprToProto, serializeDataType} +import org.apache.comet.serde.operator.{literalToProto, partition2Proto, schema2Proto, CometNativeScan} +import org.apache.comet.shims.ShimFileFormat + +/** + * Serde for the native Delta scan. Two shapes: + * - Plain reads reuse core's `NativeScanCommon` builder wholesale. + * - Deletion-vector reads: Delta's planner appends `__delta_internal_is_row_deleted` (tinyint) + * and Spark's row-index temp column (bigint) to the read schema and filters on is_row_deleted + * above the scan. The native reader applies the DV as a row selection, so both internal + * columns are emitted as per-file constants (0), the parquet read schema is stripped to the + * real data columns, and the DV descriptor ships per file for native to fetch and decode. + */ +object CometDeltaNativeScan + extends Logging + with org.apache.spark.sql.catalyst.expressions.PredicateHelper { + + val IsRowDeletedColumn: String = DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME + val RowIndexColumn: String = ShimFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME + + private[delta] val internalColumnNames: Set[String] = Set(IsRowDeletedColumn, RowIndexColumn) + + // Prefix for the internal columns' slots in the partition schema, mirroring core's + // _comet_metadata_ prefix rationale: DataFusion matches partition columns by name. + // [[allocateUniqueInternalFields]] additionally suffixes on collision with a real column. + private val deltaConstFieldPrefix = "_comet_delta_" + + def isDvShape(scanExec: FileSourceScanExec): Boolean = + scanExec.requiredSchema.exists(f => internalColumnNames.contains(f.name)) + + private def deltaFormat(scanExec: FileSourceScanExec): DeltaParquetFileFormat = + scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + + private def columnMappingMode(scanExec: FileSourceScanExec): String = + deltaFormat(scanExec).metadata.columnMappingMode.name + + /** + * Under column mapping, parquet files store physical column names (stable UUIDs / ids), so the + * schemas passed to the native parquet reader must be physical. Positions and structure are + * preserved, so output binding and projection are unaffected. The scan's internal DV columns + * are not part of the table schema and must be stripped before calling this. + * + * `private[delta]` (not `private`): [[DeltaScanSupport.declineReason]]'s non-ASCII + * case-insensitive name gate reuses this exact conversion to compute the names native sees + * under column mapping, rather than re-deriving physical names with separate logic. + */ + private[delta] def toPhysical(scanExec: FileSourceScanExec, schema: StructType): StructType = { + val format = deltaFormat(scanExec) + if (format.metadata.columnMappingMode.name == "none") { + schema + } else { + // Name mode matches file columns by physical NAME. Strip the parquet.field.id metadata + // createPhysicalSchema also stamps: files written before the column-mapping upgrade have + // no field ids and would fail the reader's id expectations. + stripFieldIds(org.apache.spark.sql.delta.DeltaColumnMapping + .createPhysicalSchema(schema, format.metadata.schema, format.metadata.columnMappingMode)) + } + } + + private def stripFieldIds(schema: StructType): StructType = { + import org.apache.spark.sql.types._ + def stripType(dt: DataType): DataType = dt match { + case s: StructType => stripFieldIds(s) + case a: ArrayType => a.copy(elementType = stripType(a.elementType)) + case m: MapType => + m.copy(keyType = stripType(m.keyType), valueType = stripType(m.valueType)) + case other => other + } + StructType(schema.fields.map { f => + val metadata = new MetadataBuilder() + .withMetadata(f.metadata) + .remove("parquet.field.id") + // Sibling key Delta stamps on array/map fields under IcebergCompat/Uniform. + .remove("parquet.field.nested.ids") + .build() + f.copy(dataType = stripType(f.dataType), metadata = metadata) + }) + } + + /** + * Build the planning-time `DeltaScan` operator (common data only; file partitions are injected + * lazily at execution). Returns None when an output data type cannot be serialized or the plan + * shape is not one we can translate faithfully. `memo` is the same claim-memo instance + * [[DeltaScanSupport.declineReason]] populated on this claim; its `hadoopConf` and + * `dvDescriptors` are reused here rather than recomputed. + */ + def convert( + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + memo: DeltaScanSupport.DeltaClaimMemo): Option[Operator] = { + val relation = scanExec.relation + + val firstFileUri = scanHelper.selectedPartitions + .flatMap(_.files.headOption) + .headOption + .map(_.getPath.toUri) + + val hadoopConf = memo.hadoopConf + + val tableRootPath = relation.location.rootPaths.head + val tableRoot = tableRootPath.toString + + val commonOpt = if (!isDvShape(scanExec)) { + // Under column mapping (name mode) the parquet reader must see physical names; + // positions are preserved so output binding and projection stay untouched. + CometNativeScan.buildNativeScanCommon( + source = scanExec.simpleStringWithNodeId(), Review Comment: ### 1. Use a stable scan identity for task injection Could `common.source` use the original physical scan's `SparkPlan.id` in both builders, for example `s"${scanExec.nodeName} (${scanExec.id})"`? `simpleStringWithNodeId()` reads Spark's explain-local ID map and can produce `(unknown)` during ordinary planning. Two independently converted scans of the same table, with identical common fields but partition selections `p = 1` and `p = 2`, can therefore receive the same injection key. Partition predicates are absent from the key; if both scans contribute to one collection scope, the map merge retains only one payload. Exchange boundaries isolate common self-join plans, so this is a key-contract concern rather than an established end-to-end SQL wrong answer. Could a regression clear the explain map, use disjoint selected files, and assert distinct keys? An equivalent-scan control should preserve `sameResult` and semantic hashes. The key should remain excluded from semantic equality, as it is today. ########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,583 @@ +/* + * 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.comet.contrib.delta + +import scala.jdk.CollectionConverters._ + +import org.apache.hadoop.fs.Path +import org.apache.spark.internal.Logging +import org.apache.spark.sql.catalyst.expressions.Literal +import org.apache.spark.sql.comet.{CometScanExec, DeltaPlanDataInjector} +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.RowIndexFilterType +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ScalarSubquery => ExecScalarSubquery} +import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile} +import org.apache.spark.sql.types.{ByteType, LongType, MetadataBuilder, StructField, StructType} + +import org.apache.comet.objectstore.NativeConfig +import org.apache.comet.serde.OperatorOuterClass +import org.apache.comet.serde.OperatorOuterClass.Operator +import org.apache.comet.serde.QueryPlanSerde.{exprToProto, serializeDataType} +import org.apache.comet.serde.operator.{literalToProto, partition2Proto, schema2Proto, CometNativeScan} +import org.apache.comet.shims.ShimFileFormat + +/** + * Serde for the native Delta scan. Two shapes: + * - Plain reads reuse core's `NativeScanCommon` builder wholesale. + * - Deletion-vector reads: Delta's planner appends `__delta_internal_is_row_deleted` (tinyint) + * and Spark's row-index temp column (bigint) to the read schema and filters on is_row_deleted + * above the scan. The native reader applies the DV as a row selection, so both internal + * columns are emitted as per-file constants (0), the parquet read schema is stripped to the + * real data columns, and the DV descriptor ships per file for native to fetch and decode. + */ +object CometDeltaNativeScan + extends Logging + with org.apache.spark.sql.catalyst.expressions.PredicateHelper { + + val IsRowDeletedColumn: String = DeltaParquetFileFormat.IS_ROW_DELETED_COLUMN_NAME + val RowIndexColumn: String = ShimFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME + + private[delta] val internalColumnNames: Set[String] = Set(IsRowDeletedColumn, RowIndexColumn) + + // Prefix for the internal columns' slots in the partition schema, mirroring core's + // _comet_metadata_ prefix rationale: DataFusion matches partition columns by name. + // [[allocateUniqueInternalFields]] additionally suffixes on collision with a real column. + private val deltaConstFieldPrefix = "_comet_delta_" + + def isDvShape(scanExec: FileSourceScanExec): Boolean = + scanExec.requiredSchema.exists(f => internalColumnNames.contains(f.name)) + + private def deltaFormat(scanExec: FileSourceScanExec): DeltaParquetFileFormat = + scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + + private def columnMappingMode(scanExec: FileSourceScanExec): String = + deltaFormat(scanExec).metadata.columnMappingMode.name + + /** + * Under column mapping, parquet files store physical column names (stable UUIDs / ids), so the + * schemas passed to the native parquet reader must be physical. Positions and structure are + * preserved, so output binding and projection are unaffected. The scan's internal DV columns + * are not part of the table schema and must be stripped before calling this. + * + * `private[delta]` (not `private`): [[DeltaScanSupport.declineReason]]'s non-ASCII + * case-insensitive name gate reuses this exact conversion to compute the names native sees + * under column mapping, rather than re-deriving physical names with separate logic. + */ + private[delta] def toPhysical(scanExec: FileSourceScanExec, schema: StructType): StructType = { Review Comment: ### 9. Reuse Delta's schema preparation and retain a logical-output boundary Could `toPhysical` delegate to `DeltaParquetFileFormat.prepareSchemaForRead` instead of duplicating `createPhysicalSchema` and name-mode field-ID stripping? That helper exists in the supported Delta versions and is used by Delta for its data, required, and partition schemas. Generated bookkeeping columns should remain excluded from the physical schema passed to it. For the [existing nested-mapping follow-up](https://github.com/apache/datafusion-comet/pull/5365#discussion_r3826296728), could we retain a clear division between those physical read schemas and the logical output types already present in `common.fields`? Native output adaptation can restore logical nested types before parent expressions run; mapped complex predicates should remain residual until they bind safely to the physical schema. The current ID-mode and mapped-nested fallbacks can remain until that adaptation is implemented and tested. Useful follow-up controls include upgrade/rename, `to_json` of mapped structs, missing nested children, and version-specific null-struct behavior. #5661 can document the support boundary and link the implementation work. ########## native/core/src/execution/delta_dv.rs: ########## @@ -0,0 +1,2096 @@ +// 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. + +//! Delta Lake deletion-vector decoding and translation into DataFusion +//! [`ParquetAccessPlan`]s (feature = "delta"). +//! +//! Wire formats implemented here (from delta-spark's `DeletionVectorStore` / +//! `RoaringBitmapArray`, v3.3.2): +//! - On-disk DV file: 1 version byte at the start of the file; at +//! `descriptor.offset`: `[i32 BE size][data: size bytes][i32 BE CRC32(data)]`. +//! - `data`: `[i32 LE magic]` then either +//! - magic 1681511376 ("native"): `[i32 LE count]`, then per bitmap +//! `[i32 LE size][standard 32-bit RoaringBitmap]`, keys implicit (index); +//! - magic 1681511377 ("portable", the spec's 64-bit extension): `[i64 LE +//! count]`, then per bitmap `[i32 LE key][standard 32-bit RoaringBitmap]` +//! with keys ascending -- exactly [`RoaringTreemap`]'s serialized form. + +use std::mem::size_of; +use std::sync::Arc; + +use datafusion::datasource::listing::PartitionedFile; +use datafusion::datasource::physical_plan::parquet::metadata::DFParquetMetadata; +use datafusion::datasource::physical_plan::parquet::{ParquetAccessPlan, RowGroupAccess}; +use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation}; +use datafusion::execution::runtime_env::RuntimeEnv; +use futures::{StreamExt, TryStreamExt}; +use object_store::path::Path; +use object_store::{ObjectStore, ObjectStoreExt}; +use parquet::arrow::arrow_reader::{RowSelection, RowSelector}; +use parquet::file::metadata::{PageIndexPolicy, ParquetMetaData}; +use roaring::{RoaringBitmap, RoaringTreemap}; + +use crate::execution::operators::ExecutionError; +use crate::execution::operators::ExecutionError::GeneralError; +use datafusion_comet_proto::spark_operator::DeltaSparkDvDescriptor; + +const NATIVE_MAGIC: i32 = 1681511376; +const PORTABLE_MAGIC: i32 = 1681511377; + +/// Unframe a DV blob read from `descriptor.offset` of a DV file: +/// `[i32 BE size][data][i32 BE crc]`. Verifies both the size against the +/// descriptor's `size_in_bytes` and the CRC32 checksum. +pub fn unframe_dv_blob(blob: &[u8], expected_size: usize) -> Result<&[u8], ExecutionError> { + if blob.len() < 8 { + return Err(GeneralError(format!( + "Deletion vector blob too short: {} bytes", + blob.len() + ))); + } + let size = i32::from_be_bytes(blob[0..4].try_into().unwrap()); + if size < 0 || size as usize != expected_size { + return Err(GeneralError(format!( + "Deletion vector size mismatch: file says {size}, descriptor says {expected_size}" + ))); + } + let end = 4 + size as usize; + if blob.len() < end + 4 { + return Err(GeneralError(format!( + "Deletion vector blob truncated: need {} bytes, have {}", + end + 4, + blob.len() + ))); + } + let data = &blob[4..end]; + let expected_crc = i32::from_be_bytes(blob[end..end + 4].try_into().unwrap()); + let actual_crc = crc32fast::hash(data) as i32; + if expected_crc != actual_crc { + return Err(GeneralError( + "Deletion vector checksum mismatch".to_string(), + )); + } + Ok(data) +} + +/// Deserialize the magic-prefixed RoaringBitmapArray into a 64-bit treemap of +/// deleted row indexes. +pub fn deserialize_dv_bitmap(data: &[u8]) -> Result<RoaringTreemap, ExecutionError> { + if data.len() < 4 { + return Err(GeneralError( + "Deletion vector bitmap too short for magic number".to_string(), + )); + } + let magic = i32::from_le_bytes(data[0..4].try_into().unwrap()); + let rest = &data[4..]; + match magic { + PORTABLE_MAGIC => RoaringTreemap::deserialize_from(rest) + .map_err(|e| GeneralError(format!("Invalid portable deletion vector bitmap: {e}"))), + NATIVE_MAGIC => { + if rest.len() < 4 { + return Err(GeneralError( + "Native deletion vector bitmap missing count".to_string(), + )); + } + let count = i32::from_le_bytes(rest[0..4].try_into().unwrap()); + if count < 0 { + return Err(GeneralError(format!( + "Invalid RoaringBitmapArray length ({count} < 0)" + ))); + } + let mut pos = 4usize; + let mut treemap = RoaringTreemap::new(); + for key in 0..count as u64 { + if rest.len() < pos + 4 { + return Err(GeneralError( + "Native deletion vector bitmap truncated".to_string(), + )); + } + let size = i32::from_le_bytes(rest[pos..pos + 4].try_into().unwrap()); + pos += 4; + if size < 0 || rest.len() < pos + size as usize { + return Err(GeneralError( + "Native deletion vector bitmap truncated".to_string(), + )); + } + let bitmap = RoaringBitmap::deserialize_from(&rest[pos..pos + size as usize]) + .map_err(|e| { + GeneralError(format!("Invalid deletion vector sub-bitmap: {e}")) + })?; + pos += size as usize; + for value in bitmap { + treemap.insert((key << 32) | value as u64); + } + } + Ok(treemap) + } + other => Err(GeneralError(format!( + "Unexpected RoaringBitmapArray magic number {other}" + ))), + } +} + +/// Translate deleted row indexes into a [`ParquetAccessPlan`]: fully-deleted +/// row groups become `Skip`, untouched groups stay `Scan`, and partially +/// deleted groups get a `RowSelection` selecting the complement of the deleted +/// rows. Page-index pruning later INTERSECTS with these selections, so DV +/// skips and page skips compose. +pub fn build_access_plan( + row_group_row_counts: &[i64], + deleted: &RoaringTreemap, +) -> Result<ParquetAccessPlan, ExecutionError> { + let mut plan = ParquetAccessPlan::new_all(row_group_row_counts.len()); + // Single sweep over the (sorted) deleted row indexes, bucketing by row group. + let mut deleted_iter = deleted.iter().peekable(); + let mut group_start = 0u64; + for (idx, &num_rows) in row_group_row_counts.iter().enumerate() { + // A corrupt footer can report a negative row count. `num_rows as u64` would otherwise + // wrap it into a huge positive value, silently corrupting every row-group boundary + // computed from `group_start`/`group_end` below (and therefore which deleted row indexes + // land in which row group) instead of failing loudly. + if num_rows < 0 { + return Err(GeneralError(format!( + "Parquet footer reports a negative row count ({num_rows}) for row group {idx}" + ))); + } + let num_rows = num_rows as u64; + let group_end = group_start + num_rows; + let mut selectors: Vec<RowSelector> = Vec::new(); + let mut cursor = group_start; + let mut deleted_in_group = 0u64; + while let Some(&row) = deleted_iter.peek() { + if row >= group_end { + break; + } + deleted_iter.next(); + deleted_in_group += 1; + if row > cursor { + selectors.push(RowSelector::select((row - cursor) as usize)); + } + // Merge runs of consecutive deleted rows into one skip. + match selectors.last_mut() { + Some(last) if last.skip => last.row_count += 1, + _ => selectors.push(RowSelector::skip(1)), + } + cursor = row + 1; + } + if deleted_in_group == num_rows && num_rows > 0 { + plan.skip(idx); + } else if deleted_in_group > 0 { + if group_end > cursor { + selectors.push(RowSelector::select((group_end - cursor) as usize)); + } + plan.scan_selection(idx, RowSelection::from(selectors)); + } + group_start = group_end; + } + // A deleted index beyond the file's total row count means the DV does not + // belong to this file (stale or corrupted metadata); silently dropping it + // would under-apply deletions. + if let Some(&row) = deleted_iter.peek() { + return Err(GeneralError(format!( + "Deletion vector marks row {row} but the file only has {group_start} rows" + ))); + } + Ok(plan) +} + +/// Verify a decoded deletion vector's row count matches the descriptor's +/// declared `cardinality`, mirroring Delta's JVM reader +/// (`StoredBitmap.validateCardinality`). The CRC and framing checks catch +/// corruption but not a stale, otherwise well-formed bitmap whose row count +/// no longer matches the descriptor -- that would silently under- or +/// over-delete rows. +fn validate_cardinality( + file_path: &str, + expected: i64, + deleted: &RoaringTreemap, +) -> Result<(), ExecutionError> { + let actual = deleted.len(); + if actual != expected as u64 { + return Err(GeneralError(format!( + "Deletion vector for {file_path} has cardinality mismatch: descriptor says {expected}, decoded bitmap has {actual} deleted rows" + ))); + } + Ok(()) +} + +/// One data file plus everything needed to apply its deletion vector. The +/// file's size comes from `file.object_meta.size` (built by the planner from +/// the proto's `file_size`). +/// +/// `data_store` and `dv_store` are resolved by the caller *before* entering +/// the async `attach_access_plans` runtime (see its doc comment): building an +/// object store is sync I/O that, for a cold S3 authority, internally issues +/// its own `Handle::block_on` calls, which panics if nested inside another +/// `block_on`. Resolving up front means this module never constructs a +/// store itself. +pub struct DvScanFile { + pub file: PartitionedFile, + /// Full URL of the data file (proto `file_path`). + pub file_path: String, + pub dv: Option<DeltaSparkDvDescriptor>, + /// Object store for `file_path`, pre-resolved by the caller. Only read + /// when `dv` is `Some` (files without a deletion vector never open their + /// footer here), but every file carries one so the struct's shape + /// doesn't depend on whether a deletion vector is present. + pub data_store: Arc<dyn ObjectStore>, + /// Store and within-store path for an on-disk deletion vector's absolute + /// path, pre-resolved by the caller. `None` when the file has no + /// deletion vector or the deletion vector is stored inline. + pub dv_store: Option<(Arc<dyn ObjectStore>, Path)>, +} + +/// Execution-memory-pool reservation covering one file's expanded DV row selectors across +/// their *entire* lifetime attached to a scan -- from `build_access_plan`'s construction +/// through DataFusion 54.1's reader normalizing the attached [`ParquetAccessPlan`] +/// (`create_initial_plan`'s deep clone plus `into_overall_row_selection`'s combined +/// `RowSelection`; see [`reader_peak_bytes`]) -- attached to the file's [`PartitionedFile`] +/// extensions alongside its [`ParquetAccessPlan`]. The reservation's lifetime is tied to the +/// `PartitionedFile` it is attached to, so it is released back to the pool exactly when the +/// plan is dropped (query completion or an early-terminated scan), never held open longer. +/// Newtype-wrapped so it occupies its own slot in the multi-slot, type-keyed `extensions` map +/// (`datafusion_common::extensions::Extensions`) alongside the plan, rather than a bare +/// `MemoryReservation` colliding with one some other extension might attach. +pub struct DvAccessPlanReservation(pub MemoryReservation); + +/// Total number of [`RowSelector`]s materialized across `plan`'s per-row-group +/// selections (`RowGroupAccess::Selection`); `Scan`/`Skip` row groups +/// contribute none. An alternating deleted/retained bitmap produces one +/// non-coalescing selector per row (see [`reader_peak_bytes`]'s doc comment +/// for the worst-case accounting), so this count -- not the deletion +/// vector's cardinality -- is the thing that must be bounded and reserved +/// against the execution memory pool. +fn total_selectors(plan: &ParquetAccessPlan) -> usize { + plan.inner() + .iter() + .map(|access| match access { + RowGroupAccess::Selection(selection) => selection.iter().count(), + _ => 0, + }) + .sum() +} + +/// Multiplier bounding the peak allocation live *during construction* of one +/// file's [`RowSelection`]s, relative to the conservative selector-count +/// bound `S = 2 * cardinality + num_row_groups` (one non-coalescing selector +/// per deleted row in the worst-case alternating pattern, doubled, plus up to +/// one extra boundary selector per row group). Split `S` into `r`, the +/// selectors already retained from row groups `build_access_plan` has +/// finished, and `c`, the selectors accumulated so far in the current row +/// group's source `Vec`; `r` and `c` partition the selectors counted toward +/// `S`, so `r + c <= S` always. While the current group is being built, the +/// `Vec`'s doubling growth strategy can leave its backing allocation at up to +/// `2 * c` (the next power-of-two capacity above `c`). Once the group +/// finishes, `RowSelection::from(Vec)` (parquet's `FromIterator` impl, +/// `with_capacity` + copy) builds a second, separate `Vec` of size `c` from +/// that source while the source is still alive, so at the moment the copy +/// begins, the retained selectors, the current group's doubled source `Vec`, +/// and the copy are all live simultaneously: `r + 2c + c = r + 3c`. Since +/// `r >= 0`, `r + 3c <= 3r + 3c = 3(r + c) <= 3S`. 3x covers that peak. +const CONSTRUCTION_PEAK_FACTOR: usize = 3; + +/// Upper bound on how much larger a `Vec`'s backing allocation can be than its element count +/// after being built by repeated pushes: `std`'s doubling growth strategy never leaves a `Vec` +/// of `n` elements with a backing allocation larger than the next power of two above `n`, which +/// is at most `2 * n` for any `n >= 1`. +const VEC_GROWTH_CAPACITY_FACTOR: usize = 2; + +/// `RawVec`'s minimum non-zero capacity for element sizes `<= 1024` bytes ([`RowSelector`] is +/// 16 bytes on 64-bit platforms: a `usize` row count plus a padded `bool`). Applied once per +/// row group (or per contiguous run of row groups) a fresh `from_fn`/`FlatMap`-driven `Vec` +/// gets built for (see [`reader_peak_bytes`]), so even a group or run whose true selector count +/// is tiny still pays this floor. +const MIN_VEC_CAPACITY_SELECTORS: usize = 4; + +/// Conservative upper bound, in bytes, on the peak allocation live while DataFusion 54.1's +/// reader normalizes one file's attached [`ParquetAccessPlan`] -- the allocation this module's +/// steady-state reservation must cover, not merely the plan's own retained selector bytes. +/// THREE allocations can be live simultaneously by the time `into_overall_row_selection` +/// returns, not two -- the clone is only exact when page-index pruning never touches it: +/// +/// 1. **Attached original** (`selectors`, exact): `create_initial_plan` deep-clones the +/// attached plan while the original remains reachable from the file's `extensions` until +/// the scan consumes it. The ORIGINAL's own selector `Vec`s are exact -- a coalesced +/// [`RowSelection`] built via `RowSelection::from(Vec<RowSelector>)` (what +/// `build_access_plan` uses) has no excess capacity, because that conversion is a plain +/// `with_capacity(len)` copy, not a `size_hint`-blind fold. +/// 2. **The clone, possibly capacity-inflated** (`<= VEC_GROWTH_CAPACITY_FACTOR * selectors + +/// MIN_VEC_CAPACITY_SELECTORS * num_row_groups`): if page-index pruning fires +/// (`PagePruningAccessPlanFilter`; `access_plan.rs`'s `scan_selection` on a row group that +/// already carries a `RowGroupAccess::Selection` calls `existing.intersection(&page_derived)` +/// -- `RowSelection::intersection` -> `intersect_row_selections`), it replaces the CLONE's +/// per-row-group selection with that intersection's output. `intersect_row_selections` is +/// ANOTHER `from_fn` generator with `size_hint() == (0, None)`, so each intersected row +/// group's backing `Vec` starts at `with_capacity(0)` and doubles as it grows, independent +/// of whatever capacity the pre-intersection selection had. This inflated clone is still +/// live when `into_overall_row_selection` later moves its buffer. Term 1's exactness +/// guarantee holds for the ORIGINAL always, and for the clone only when page-index pruning +/// never fires against it -- once it does, the clone must be charged at the SAME +/// growth-capped bound as a fresh combined-selection `Vec` (term 3), summed once per row +/// group rather than once per run, since each row group's `Selection` is intersected +/// independently. +/// 3. **Per-run combined-selection allocation** (`<= VEC_GROWTH_CAPACITY_FACTOR * (selectors + +/// num_row_groups) + MIN_VEC_CAPACITY_SELECTORS * num_row_groups`): `into_overall_row_selection` +/// collects each contiguous run of row groups' selectors into a *new* `RowSelection` via a +/// `FlatMap` whose `size_hint().0 == 0`, so that run's `Vec` starts at `with_capacity(0)` +/// and doubles as it grows -- capping its backing allocation at +/// `max(MIN_VEC_CAPACITY_SELECTORS, next_power_of_two(len))`, which is at most +/// `MIN_VEC_CAPACITY_SELECTORS + VEC_GROWTH_CAPACITY_FACTOR * len` for a run of `len` +/// selectors. `len` is at most that run's share of `selectors` plus one boundary selector +/// per `RowGroupAccess::Scan` row group in the run (`Scan` always contributes exactly one +/// `RowSelector::select(num_rows)`; see `access_plan.rs`'s `into_overall_row_selection`). +/// Summing across at most `num_row_groups` runs (each spans >= 1 row group) bounds the total +/// at `VEC_GROWTH_CAPACITY_FACTOR * selectors + (MIN_VEC_CAPACITY_SELECTORS + +/// VEC_GROWTH_CAPACITY_FACTOR) * num_row_groups`. +/// +/// Summing all three terms and converting to bytes: `((1 + 2 * VEC_GROWTH_CAPACITY_FACTOR) * +/// selectors + (2 * MIN_VEC_CAPACITY_SELECTORS + VEC_GROWTH_CAPACITY_FACTOR) * num_row_groups) +/// * size_of::<RowSelector>()` -- with the constants above, `(5 * selectors + 10 * +/// num_row_groups) * size_of::<RowSelector>()`. Checked against two measured worst cases: +/// +/// - No page-index pruning (the original P2 report; term 2 stays exact): one 2,000,000-row +/// group, 1,000,000 alternating deletions, `selectors = 2,000,000`. Measured allocator peak +/// 97,554,457 B; the byte-for-byte accounting for the attached original plus the (here, +/// exact) clone plus the inflated combined selection explains 97,554,432 B of that, a 25 B +/// residue we did not attribute. This bound gives 160,000,160 B -- much looser here because +/// it must also cover the next case, where the clone is NOT exact. +/// - Page-index pruning fires against the clone: one 1,048,577-row group, `selectors = +/// 1,048,577`. Measured peak 83,886,096 B; this bound gives 83,886,320 B (a 224 B, <1% +/// margin -- deliberately tight, since this is the case that drives the bound). +/// +/// Uses checked arithmetic throughout: a selector or row-group count large enough to overflow +/// `usize` indicates a corrupted or malicious input, reported as a clean error rather than +/// panicking. +fn reader_peak_bytes(selectors: usize, num_row_groups: usize) -> Result<usize, ExecutionError> { + let overflow = || { + GeneralError(format!( + "Deletion vector reader-peak bound overflowed for {selectors} selectors and \ + {num_row_groups} row groups" + )) + }; + // Term 1: the attached original -- exact, untouched by page-index pruning (only the clone + // is ever intersected; see the doc comment above). + let attached_term = selectors; + // Term 2: the clone, bounded as if page-index pruning DID fire against every row group + // (safe even when it doesn't: term 2's bound is always >= `selectors`, so it never + // undershoots the exact case either). + let clone_growth = selectors + .checked_mul(VEC_GROWTH_CAPACITY_FACTOR) + .ok_or_else(overflow)?; + let clone_floor = num_row_groups + .checked_mul(MIN_VEC_CAPACITY_SELECTORS) + .ok_or_else(overflow)?; + let clone_term = clone_growth.checked_add(clone_floor).ok_or_else(overflow)?; + // Term 3: into_overall_row_selection's per-run combined-selection allocation. + let combined_growth = selectors + .checked_mul(VEC_GROWTH_CAPACITY_FACTOR) + .ok_or_else(overflow)?; + let combined_floor = num_row_groups + .checked_mul(MIN_VEC_CAPACITY_SELECTORS + VEC_GROWTH_CAPACITY_FACTOR) + .ok_or_else(overflow)?; + let combined_term = combined_growth + .checked_add(combined_floor) + .ok_or_else(overflow)?; + + let selector_bound = attached_term + .checked_add(clone_term) + .and_then(|sum| sum.checked_add(combined_term)) + .ok_or_else(overflow)?; + selector_bound + .checked_mul(size_of::<RowSelector>()) + .ok_or_else(overflow) +} + +/// Upper bound, in [`RowSelector`]s, on how many extra selectors the parquet reader's +/// page-index pruning can add on top of the deletion vector's own selection when normalizing +/// one file, from that file's already-fetched [`ParquetMetaData`]. +/// +/// `intersect_row_selections` (parquet's `selection.rs`), which combines a page-pruning +/// selection with the deletion vector's selection, is a `from_fn` generator whose +/// `size_hint()` is `(0, None)`: for inputs of length `a` and `b`, its output can have up to +/// `a + b` selectors -- longer than either input. Bounding the page-pruning side of that sum +/// requires knowing how many selectors a page-index-derived selection could produce: at most +/// two per data page (one skip, one select, in the worst case of alternating page-level +/// pruning decisions), summed over every column of every row group. +/// +/// Returns `0` when `metadata` carries no offset index (`metadata.offset_index()` is `None`). +/// This is provably safe, not merely a convenient default: page-index pruning cannot produce a +/// page-level selection without the offset index to locate pages by, so there are no +/// page-pruning selectors to bound. The offset index is fetched with +/// `PageIndexPolicy::Optional` from the same `FileMetadataCache` entry the scan's reader later +/// reopens (see [`attach_access_plan`]'s footer-fetch comment), so this function observes +/// exactly what the reader will see. +/// +/// Uses checked arithmetic throughout for the same reason as [`admission_bound_bytes`]. +fn page_selection_bound_selectors(metadata: &ParquetMetaData) -> Result<usize, ExecutionError> { + let Some(offset_index) = metadata.offset_index() else { + return Ok(0); + }; + let overflow = || { + GeneralError( + "Deletion vector page-selection bound overflowed while summing offset-index page \ + locations" + .to_string(), + ) + }; + let mut total_page_locations = 0usize; + for row_group in offset_index { + for column in row_group { + total_page_locations = total_page_locations + .checked_add(column.page_locations().len()) + .ok_or_else(overflow)?; + } + } + total_page_locations.checked_mul(2).ok_or_else(overflow) +} + +/// Execution-memory-pool admission bound, in bytes, for one file's deletion-vector access +/// plan -- reserved *before* calling `build_access_plan` (see [`attach_access_plan`]'s +/// pre-reserve call site) to cover the larger of two peaks live at different points in the +/// plan's lifetime. In practice the reader-normalization peak below dominates the construction +/// peak unconditionally for any non-trivial input (`reader_peak_bytes(S, G) = (5S + 10G) * +/// size_of::<RowSelector>()` always exceeds `CONSTRUCTION_PEAK_FACTOR * S * +/// size_of::<RowSelector>() = 3S * size_of::<RowSelector>()` once `S >= 1`, since the `5S` term +/// alone already exceeds `3S`); the construction term is retained as a documented floor rather +/// than dropped, since it is cheap to compute and keeps this bound correct even if the reader's +/// growth factors ever shrink below construction's. +/// +/// - **Construction peak** (`CONSTRUCTION_PEAK_FACTOR * S`, see that constant's doc comment): +/// live while `build_access_plan` builds the plan's `RowSelection`s. Construction's +/// transient allocations fully unwind before `build_access_plan` returns, so this peak never +/// overlaps the reader-normalization peak below. +/// - **Reader-normalization peak** (`reader_peak_bytes(S + page_bound_selectors, +/// num_row_groups)`, see that function): live later, once DataFusion's reader normalizes the +/// attached plan. `S = 2 * cardinality + num_row_groups` is the same conservative bound on +/// the plan's final retained selector count used for the construction peak -- it provably +/// bounds `R = total_selectors(&plan)` (`R <= S`, from `build_access_plan`'s +/// one-non-coalescing-selector-per-deleted-row worst case plus one boundary selector per row +/// group), so `S + page_bound_selectors` bounds `R` after page-index inflation the same way +/// `S` bounds `R` before it. +/// +/// These two peaks never overlap in time, so `max` -- not `sum` -- is the correct combinator: +/// reserving their sum would over-reserve for no safety benefit. +/// +/// Deliberately not clamped by the file's total row count here, unlike the reader-peak target +/// `attach_access_plan` resizes down to after construction (see that call site): `S`'s +/// `+ num_row_groups` boundary term is a worst-case padding margin that can legitimately exceed +/// the total row count for a small, heavily-deleted file, and admission sizing has no actual +/// retained-selector count yet to clamp against -- only after construction, once `R` is known, +/// is clamping to the total row count both meaningful and strictly tighter. Leaving this bound +/// unclamped only ever makes admission more conservative, never less safe. +/// +/// Uses checked arithmetic throughout: a cardinality, row-group count, or page bound large +/// enough to overflow `usize` while computing this bound indicates a corrupted or malicious +/// descriptor, reported as a clean error rather than panicking. +fn admission_bound_bytes( + cardinality: i64, + num_row_groups: usize, + page_bound_selectors: usize, +) -> Result<usize, ExecutionError> { + let overflow = || { + GeneralError(format!( + "Deletion vector admission bound overflowed for cardinality {cardinality}, \ + {num_row_groups} row groups, and page bound {page_bound_selectors} selectors" + )) + }; + let cardinality_usize = usize::try_from(cardinality).map_err(|_| overflow())?; + // S: the conservative bound on the plan's final *retained* selector count (what + // `total_selectors(&plan)` cannot exceed) -- unchanged from the pre-existing + // construction-only bound this function replaces. + let s = cardinality_usize + .checked_mul(2) + .and_then(|doubled| doubled.checked_add(num_row_groups)) + .ok_or_else(overflow)?; + + let construction_bytes = s + .checked_mul(size_of::<RowSelector>()) + .and_then(|bytes| bytes.checked_mul(CONSTRUCTION_PEAK_FACTOR)) + .ok_or_else(overflow)?; + + let s_plus_page = s.checked_add(page_bound_selectors).ok_or_else(overflow)?; + let reader_bytes = reader_peak_bytes(s_plus_page, num_row_groups)?; + + Ok(construction_bytes.max(reader_bytes)) +} + +/// Upper bound on concurrent DV-blob and footer fetches per partition. Both +/// are small ranged reads, so a modest fan-out hides object-store latency +/// without flooding the store client. +const DV_FETCH_CONCURRENCY: usize = 8; + +/// Called via `block_on` at plan-creation time on the executor task: DV blobs +/// are small ranged reads and footers are needed to learn row-group +/// boundaries. Files are fetched concurrently (bounded by +/// [`DV_FETCH_CONCURRENCY`]) with input order preserved. Footer fetches go +/// through the scan's shared FileMetadataCache, so the scan's subsequent open +/// of the same file is served from cache. That reuse relies on each input +/// [`PartitionedFile`] being returned as-is (only `with_extension` applied), +/// never rebuilt: the cache entry is keyed by this exact `object_meta` and the +/// scan later looks it up through the same struct. +/// +/// Deliberately takes no object-store options map and imports no +/// store-construction helper: every [`DvScanFile`] arrives with its stores +/// already resolved by the caller (see its doc comment), so this async path +/// structurally cannot build an object store -- only `runtime_env` is still +/// threaded through, for the shared `FileMetadataCache` and (per file) the +/// execution `MemoryPool` each expanded access plan's row selectors are +/// reserved against -- see [`DvAccessPlanReservation`]. +pub async fn attach_access_plans( + runtime_env: Arc<RuntimeEnv>, + files: Vec<DvScanFile>, +) -> Result<Vec<PartitionedFile>, ExecutionError> { + futures::stream::iter(files) + .map(|scan_file| attach_access_plan(Arc::clone(&runtime_env), scan_file)) + .buffered(DV_FETCH_CONCURRENCY) + .try_collect() + .await +} + +/// Resolve one file's deletion vector into an attached [`ParquetAccessPlan`]; +/// files without a DV pass through untouched. +async fn attach_access_plan( + runtime_env: Arc<RuntimeEnv>, + scan_file: DvScanFile, +) -> Result<PartitionedFile, ExecutionError> { + let DvScanFile { + file, + file_path, + dv, + data_store, + dv_store, + } = scan_file; + let dv = match dv { + Some(dv) => dv, + None => return Ok(file), + }; + // Delta's canonical `DeletionVectorDescriptor.EMPTY`: inline storage, empty + // payload, size 0, cardinality 0. Spark's reader returns all rows for it; + // decoding would fail (the empty payload is too short for a magic + // number), so pass the file through unchanged before attempting to read it. + if dv.cardinality == 0 && dv.size_in_bytes == 0 { + return Ok(file); + } + if dv.size_in_bytes < 0 { + return Err(GeneralError(format!( + "Deletion vector for {file_path} has negative size {}", + dv.size_in_bytes + ))); + } + if dv.cardinality < 0 { + return Err(GeneralError(format!( + "Deletion vector for {file_path} has negative cardinality {}", + dv.cardinality + ))); + } + + let data: Vec<u8> = if let Some(inline) = dv.inline_data { + inline + } else if let Some(dv_path) = &dv.absolute_path { + let offset = dv + .offset + .ok_or_else(|| GeneralError("On-disk deletion vector missing offset".into()))?; + if offset < 0 { + return Err(GeneralError(format!( + "Deletion vector for {file_path} has negative offset {offset}" + ))); + } + let offset = offset as u64; + // [i32 BE size][data: size_in_bytes][i32 BE crc] + let framed_len = 4 + dv.size_in_bytes as u64 + 4; + let (store, dv_store_path) = dv_store.ok_or_else(|| { + GeneralError(format!( + "Deletion vector for {file_path} has an absolute path but no pre-resolved object store" + )) + })?; + let blob = store + .get_range(&dv_store_path, offset..offset + framed_len) + .await + .map_err(|e| GeneralError(format!("Failed to read deletion vector {dv_path}: {e}")))?; + unframe_dv_blob(&blob, dv.size_in_bytes as usize)?.to_vec() + } else { + return Err(GeneralError( + "Deletion vector descriptor has neither inline data nor a path".into(), + )); + }; + let deleted = deserialize_dv_bitmap(&data) + .map_err(|e| GeneralError(format!("Invalid deletion vector for {file_path}: {e}")))?; + validate_cardinality(&file_path, dv.cardinality, &deleted)?; + + // Row-group boundaries come from the data file's footer, fetched through the scan's + // shared FileMetadataCache with the page index loaded eagerly and the scan's metadata + // size hint (mirroring EagerPageIndexReaderFactory): the one fetch here also serves the + // subsequent data-file open, so DV files pay no extra footer round-trip. Keyed by + // `file.object_meta`, the exact ObjectMeta the scan's reader factory will look up. + let metadata_cache = runtime_env.cache_manager.get_file_metadata_cache(); + let metadata = DFParquetMetadata::new(data_store.as_ref(), &file.object_meta) Review Comment: ### 7. Share metadata preparation and the final scan's accounting Could DV preparation use a shared metadata-reading boundary connected to the final scan's reader configuration and metrics? This direct `DFParquetMetadata` fetch correctly fills the shared cache, but runs before the scan creates its instrumented reader. Preparation footer/page-index I/O therefore bypasses the final scan's `scan_io_*` counters, while the later reader open can report a cache hit. A preparation hook in the shared scan builder, or a shared metadata helper with the same metrics owner, would give preparation and normal reads one place for metadata policy and instrumentation fixes. The important part is carrying preparation accounting into the final scan; creating a separate factory whose counters are discarded would retain the gap. Could cold-cache and warm-cache DV tests verify that preparation I/O is counted, the footer is reused, and warm metadata opens avoid storage reads? Please preserve the current INT96 stamping on cached metadata. The accounting gap follows from the source path; no runtime counter totals or performance improvement have been measured here. ########## contrib/delta-spark/src/main/scala/org/apache/spark/sql/comet/CometDeltaNativeScanExec.scala: ########## @@ -0,0 +1,309 @@ +/* + * 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.spark.sql.comet + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.plans.QueryPlan +import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, UnknownPartitioning} +import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, ReusedSubqueryExec, ScalarSubquery, SparkPlan, SubqueryAdaptiveBroadcastExec} +import org.apache.spark.sql.execution.datasources.HadoopFsRelation +import org.apache.spark.sql.execution.metric.SQLMetric +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.vectorized.ColumnarBatch + +import org.apache.comet.contrib.delta.DeltaSparkScanEnvelope +import org.apache.comet.serde.OperatorOuterClass +import org.apache.comet.serde.OperatorOuterClass.Operator + +/** + * Native scan node for Delta Lake tables (contrib). Delta's own planning (log replay, snapshot + * resolution, partition pruning) has already run inside delta-spark by the time this node is + * created from the DSv1 [[FileSourceScanExec]]; file listing and split planning are delegated to + * a [[CometScanExec]] helper, and data reads execute through Comet's native DataFusion parquet + * machinery, inheriting row-group and page-index pruning. + * + * DPP: `runtimeFilters` is a constructor field included in equality, so its rewrite (via + * [[CometScanWithPlanData]]) survives plan copies -- a transient field would be dropped by + * `TreeNode.makeCopy` on MERGE re-planning (the CometIcebergNativeScanExec lesson). + */ +case class CometDeltaNativeScanExec( + override val nativeOp: Operator, + override val output: Seq[Attribute], + requiredSchema: StructType, + runtimeFilters: Seq[Expression], + dataFilters: Seq[Expression], + @transient relation: HadoopFsRelation, + originalPlan: FileSourceScanExec, + override val serializedPlanOpt: SerializedPlan, + sourceKey: String) + extends CometLeafExec + with CometScanWithPlanData { + + override val nodeName: String = s"CometDeltaNativeScan $relation" + + // Derived from (originalPlan, runtimeFilters), never stored: any copy of this node + // automatically gets a helper consistent with ITS runtimeFilters, avoiding the #3510 class of + // bug where a stored helper field desyncs from rewritten filters. Costs one extra file listing + // per executed instance; correctness over the duplicate driver-side listing. + // + // Forcing invariant: this lazy val is forced by the `metrics` override below, and AQE's UI + // plan-walk calls `.metrics` on every node MID-PLANNING, including while a DPP subquery is + // still an adaptive placeholder or a partition filter holds an unresolved ScalarSubquery (see + // `hasUnevaluableSubqueryFilter` below). That's safe ONLY because constructing `scanHelper` is a + // cheap case-class build with no file listing, and core's `CometScanExec.metrics` touches only + // `wrapped.driverMetrics` (populated by Spark's own planning) plus a static metric-node + // constructor -- neither file listing nor subquery resolution. If core's `metrics` ever touches + // either, forcing `scanHelper` here would resurrect the AQE mid-planning crashes this invariant + // prevents. + @transient private lazy val scanHelper: CometScanExec = + CometDeltaNativeScanExec.planningHelper(originalPlan, runtimeFilters) + + // NOT lazy val: while a DPP subquery is still an adaptive placeholder, or a partition filter + // holds an unresolved scalar subquery, this returns a temporary value that must not be + // memoized -- after CometPlanAdaptiveDynamicPruningFilters rewrites the filters (DPP case) or + // AQE resolves the subquery (scalar case), later reads must see the real post-pruning + // partition count. + override def outputPartitioning: Partitioning = + if (hasUnevaluableSubqueryFilter) UnknownPartitioning(0) + else UnknownPartitioning(perPartitionData.length) + + // runtimeFilters IS scanHelper.partitionFilters element-for-element, so checking runtimeFilters + // here avoids constructing/forcing the derived scanHelper just to read partitioning. The + // InSubqueryExec placeholder shapes mirror + // CometPlanAdaptiveDynamicPruningFilters.extractSABData + hasWrappedSAB -- keep in sync. The + // ScalarSubquery case is probed rather than treated as permanently unevaluable: Spark exposes no + // public finished/updated flag on ExecSubqueryExpression, but `eval()` doubles as one -- it only + // reads the cached `result` behind a `require(updated, ...)` guard, while the subquery is + // actually run by `updateResult()` (invoked separately during prepare/AQE), never by `eval()`. + // Once resolved, outputPartitioning below reports the real perPartitionData.length instead of + // staying at zero -- a fused native parent's buildNativeContext requires that count to match. + private def hasUnevaluableSubqueryFilter: Boolean = + runtimeFilters.exists(_.exists { + // Match `e: InSubqueryExec` and dispatch on e.plan rather than unapplying InSubqueryExec + // directly: its unapply arity differs across Spark versions and this module ships no + // version shim. + case e: InSubqueryExec => isAdaptivePlaceholder(e.plan) + case s: ScalarSubquery => !isScalarSubqueryResolved(s) + case _ => false + }) + + // `eval()` never triggers the subquery's execution: on a resolved subquery it is a pure cached + // read of `result` (verified against bytecode: `Predef.require(updated(), ...)` then a plain + // field read), so this probe is safe to call repeatedly, including from AQE's mid-planning plan + // walks. Pre-resolution, the ONLY throw is `require`'s `IllegalArgumentException`; catch exactly + // that, since anything else escaping is a genuine bug we must not mask as unpartitioned. + private def isScalarSubqueryResolved(s: ScalarSubquery): Boolean = + try { + s.eval() + true + } catch { + case _: IllegalArgumentException => false + } + + private def isAdaptivePlaceholder(p: SparkPlan): Boolean = p match { + case ReusedSubqueryExec(inner) => isAdaptivePlaceholder(inner) + case _: CometSubqueryAdaptiveBroadcastExec => true + case _: SubqueryAdaptiveBroadcastExec => true + case _ => false + } + + override lazy val outputOrdering: Seq[SortOrder] = originalPlan.outputOrdering + + override def dynamicPruningFilters: Seq[Expression] = runtimeFilters + + override def withDynamicPruningFilters(filters: Seq[Expression]): SparkPlan = { + // A real copy: runtimeFilters is a constructor field included in equality, so the copy + // survives enclosing-block rebuilds, and the derived scanHelper picks up the rewritten + // filters automatically. + copy(runtimeFilters = filters) + } + + /** + * Lazy split-mode serialization, mirroring CometNativeScanExec: common data was serialized at + * planning; per-partition file lists serialize here, at execution time. + */ + @transient private lazy val serializedPartitionData + : (Array[Byte], Array[Array[Byte]], Array[Seq[String]]) = { + // Resolve the helper's DPP subqueries: it holds its own InSubqueryExec instances that + // Spark's expressions walk does not see (the helper is derived, not a child). + scanHelper.partitionFilters.foreach { + case DynamicPruningExpression(e: InSubqueryExec) if e.values().isEmpty => + e.updateResult() + case _ => + } + + val commonBytes = { + val deltaScan = DeltaSparkScanEnvelope.unpack(nativeOp) + // Scalar subqueries in dataFilters were unresolved at planning; resolve them now and + // append them as pushed filters, as CometNativeScanExec.serializedPartitionData does. + // has_data_filters follows their presence, not the serialized count: a filter that fails + // to serialize still keeps native on the safe timestamp conversion for a filtered scan. + val resolved = org.apache.comet.contrib.delta.CometDeltaNativeScan + .resolvedSubqueryFilters(dataFilters, output, requiredSchema, conf) + val common = if (!resolved.hasResolvedFilters) { + deltaScan.getCommon + } else { + val builder = deltaScan.getCommon.toBuilder + builder.setHasDataFilters(true) + resolved.protos.foreach(builder.addDataFilters) + builder.build() + } + OperatorOuterClass.DeltaSparkScan + .newBuilder() + .setCommon(common) + .setDeltaCommon(deltaScan.getDeltaCommon) + .build() + .toByteArray + } + + val filePartitions = scanHelper.getFilePartitions() + + val tableRoot = DeltaSparkScanEnvelope.unpack(nativeOp).getDeltaCommon.getTableRoot + val perPartitionBytes = filePartitions.map { filePartition => + org.apache.comet.contrib.delta.CometDeltaNativeScan + .serializePartition(filePartition, originalPlan, tableRoot) + }.toArray + + val perPartitionPaths = filePartitions.map(_.files.map(_.filePath.toString).toSeq).toArray + + (commonBytes, perPartitionBytes, perPartitionPaths) + } + + override def commonData: Array[Byte] = serializedPartitionData._1 + + override def perPartitionData: Array[Array[Byte]] = serializedPartitionData._2 + + def perPartitionFilePaths: Array[Seq[String]] = serializedPartitionData._3 + + override def doExecuteColumnar(): RDD[ColumnarBatch] = { + val nativeMetrics = CometMetricNode.fromCometPlan(this) + val serializedPlan = CometExec.serializeNativePlan(nativeOp) + + new CometExecRDD( + sparkContext, + Seq.empty, + Map(sourceKey -> commonData), + Map(sourceKey -> perPartitionData), + serializedPlan, + perPartitionData.length, + output.length, + nativeMetrics, + Seq.empty, + None, + Seq.empty, + perPartitionFilePaths = perPartitionFilePaths, + reportScanInputMetrics = true) + } + + override def doCanonicalize(): CometDeltaNativeScanExec = { + val canonOriginal = if (originalPlan != null) { + val stripped = originalPlan.copy(partitionFilters = + CometScanUtils.filterUnusedDynamicPruningExpressions(originalPlan.partitionFilters)) + stripped.doCanonicalize() + } else { + null + } + CometDeltaNativeScanExec( + nativeOp, + output.map(QueryPlan.normalizeExpressions(_, output)), + requiredSchema, + QueryPlan.normalizePredicates( + CometScanUtils.filterUnusedDynamicPruningExpressions(runtimeFilters), + output), + QueryPlan.normalizePredicates(dataFilters, output), + relation, + canonOriginal, + SerializedPlan(None), + "") + } + + override def stringArgs: Iterator[Any] = Iterator(output, runtimeFilters) + + override def equals(obj: Any): Boolean = obj match { + case other: CometDeltaNativeScanExec => + this.originalPlan == other.originalPlan && + this.serializedPlanOpt == other.serializedPlanOpt && + this.runtimeFilters == other.runtimeFilters && + this.dataFilters == other.dataFilters + case _ => false + } + + override def hashCode(): Int = + java.util.Objects.hash(originalPlan, serializedPlanOpt, runtimeFilters, dataFilters) + + private val driverMetricKeys = + Set( + "numFiles", + "filesSize", + "numPartitions", + "metadataTime", + "staticFilesNum", + "staticFilesSize", + "pruningTime") + + // Forces `scanHelper` (see its doc above for why that -- and reading `.metrics` off it -- is + // safe even when AQE calls `.metrics` mid-planning against an unresolved DPP/scalar subquery). + override lazy val metrics: Map[String, SQLMetric] = { + CometMetricNode.nativeScanMetrics(session.sparkContext) ++ + scanHelper.metrics.filter { case (k, _) => driverMetricKeys.contains(k) } + } +} + +object CometDeltaNativeScanExec { + + /** File-planning helper: reuses CometScanExec's listing/splitting/DPP machinery. */ Review Comment: ### 4. Make the existing file-planning boundary explicit `planningHelper` already reuses core's listing, splitting, and dynamic partition pruning. Could we make its contract explicitly take the original scan plus the current partition and data filters, and return the final `FilePartition` objects with their metadata? Today partition filters are supplied separately while data filters come from `originalPlan`. Keeping both inputs explicit, and distinguishing predicates used for file selection from predicates evaluated by the reader, would localize later pruning changes. The existing reader-free helper can remain the implementation; the serializer would consume only its final result. Plan copies and filter rewrites should retain the current protection against stale cached selection. Could regression tests compare serialized paths, order, split starts and lengths, partition values, and metadata against Spark's final partitions? Include an ordinary non-DV file with a nonzero split start and dynamic partition pruning with AQE on and off. This would extend the existing result, file-count, and unresolved-subquery coverage. -- 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]
