sunchao commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3859501548
########## contrib/delta-spark/src/main/scala/org/apache/spark/sql/comet/CometDeltaNativeScanExec.scala: ########## @@ -0,0 +1,301 @@ +/* + * 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 + * `CometPlanAdaptiveDynamicPruningFilters`'s rewrite (via [[CometScanWithPlanData]]) survives + * plan copies, the lesson from CometIcebergNativeScanExec (a transient field is dropped by + * `TreeNode.makeCopy` on MERGE re-planning). + */ +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, our + // own withDynamicPruningFilters, or a generic Catalyst expression rewrite going through + // TreeNode.makeCopy, automatically gets a helper consistent with ITS runtimeFilters. A + // stored helper field would desync from rewritten filters (the #3510 class of bug). The + // cost is that file listing runs once per executed instance (planning listed separately in + // the rule extension); correctness over the duplicate driver-side listing. + // + // Forcing invariant: this lazy val is forced by the `metrics` override below (via + // `scanHelper.metrics`), 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` just below). Constructing + // `scanHelper` here, and reading `CometScanExec.metrics` off it, is safe ONLY because that + // construction is a cheap case-class build with no file listing, and core's + // `CometScanExec.metrics` (spark/.../CometScanExec.scala) touches only `wrapped.driverMetrics` + // -- populated by Spark's own planning, not by this scan -- plus a static metric-node + // constructor. Neither does file listing (`selectedPartitions`/`getFilePartitions`) or + // subquery resolution. If core's `metrics` is ever changed to touch either, forcing + // `scanHelper` from this `metrics` override would resurrect the two AQE mid-planning crashes + // this invariant was written to prevent. + @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 (planningHelper passes + // partitionFilters = runtimeFilters into CometScanExec's plain constructor field below), so + // checking runtimeFilters here avoids constructing/forcing the derived scanHelper just to + // read partitioning. The InSubqueryExec placeholder shapes mirror + // CometPlanAdaptiveDynamicPruningFilters.extractSABData + hasWrappedSAB -- keep these two + // sets in sync; if that rule learns to unwrap a new wrapper form, mirror it here too. The + // ScalarSubquery case (e.g. `p = (SELECT max(p) FROM dim ...)`, a partition-column filter, + // so it lands in runtimeFilters rather than dataFilters) is presence-based, unlike the + // InSubqueryExec check above: Spark exposes no public finished/updated probe on + // ExecSubqueryExpression, so any partition filter containing one forces UnknownPartitioning(0) + // even once it has actually resolved. That is safe one-directionally only -- worst case an + // extra shuffle from an overly conservative partitioning, never a wrong answer, since + // execution reads perPartitionData directly and never goes through this getter. + 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 _: ScalarSubquery => true Review Comment: **[P2] Restore the execution partition count after scalar-subquery resolution** Could we preserve the planning-time guard while exposing the actual partition count during execution? For a partition filter such as `p = (SELECT max(p) FROM thresholds)`, the expression remains a `ScalarSubquery` after its result is available, so this branch keeps `outputPartitioning` at zero. A fused native parent reads that zero in [buildNativeContext](https://github.com/apache/datafusion-comet/blob/1d3557cf7eef766056d4c082f8be89355118bf53/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala#L784-L795), and [NativeExecContext's validation](https://github.com/apache/datafusion-comet/blob/1d3557cf7eef766056d4c082f8be89355118bf53/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala#L550-L556) rejects nonempty scan partition data with `All per-partition arrays must have length 0`. This also affects contexts without broadcast inputs. A separate Spark/Delta probe confirmed that the scalar expression remains in the partition filter after `collect()`, with an evaluable result and matching rows. The native-context failure is source-traced, not a full current-head JNI reproduction. Please add a regression that requires the native Delta scan beneath a native parent; the existing scalar-filter test permits fallback. ########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,539 @@ +/* + * 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 surviving rows are + * by construction not deleted: 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 the native side 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. + 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 all positional 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 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. createPhysicalSchema also stamps + // parquet.field.id metadata, but files written before the column-mapping upgrade have + // no field ids and would fail the reader's id expectations, strip the ids so the + // reader stays purely name-based (id mode, when enabled, will keep them). + 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. + */ + def convert(scanExec: FileSourceScanExec, scanHelper: CometScanExec): Option[Operator] = { + val relation = scanExec.relation + + val firstFileUri = scanHelper.selectedPartitions + .flatMap(_.files.headOption) + .headOption + .map(_.getPath.toUri) + + val hadoopConf = relation.sparkSession.sessionState + .newHadoopConfWithOptions(relation.options) + + 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 => + // 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. + val dvDescriptors = DeltaScanSupport.selectedDvDescriptors(scanHelper, tableRoot) + commonBuilder.putAllObjectStoreOptions( + mergedObjectStoreOptions( + hadoopConf, + storeUris(dvDescriptors, tableRootPath, firstFileUri)).asJava) + + val common = commonBuilder.build() + val deltaCommon = OperatorOuterClass.DeltaSparkScanCommon + .newBuilder() + .setTableRoot(tableRoot) + .setColumnMappingMode(columnMappingMode(scanExec)) + .setSourceKey(DeltaPlanDataInjector.sourceKey(tableRoot, common)) + .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 it is cheap to include even when absent), + * and every distinct on-disk DV authority from `descriptors`. Inline DVs are filtered out -- + * they carry no external URI, only embedded bytes. Deduping by authority (rather than by full + * URI) keeps this O(distinct authorities) instead of O(files): a table with N deletion-vector + * files on the same external store previously produced ~N distinct URIs here, each + * independently fed into `mergedObjectStoreOptions`'s `extractObjectStoreOptions` walk over + * `hadoopConf`. Candidates are deduped keeping the FIRST URI seen per authority, so callers can + * rely on `firstFileUri`/the table root winning over any DV path that happens to share their + * authority. Factored out of [[convert]] so the URI-assembly logic (in particular the + * `storageType` filter and `absolutePath` resolution) is directly unit-testable with hand-built + * [[DeletionVectorDescriptor]] fixtures, without a Spark session or real selected files (a + * `file://` scan alone can't exercise a foreign-authority DV). + */ + 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 simply + * union rather than pick one: the 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. Factored out of [[convert]] so it is directly + * unit-testable without a Spark session. + */ + 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 Spark + * version; the dedup keeps Spark 4.x from carrying duplicates. + * + * Safety comes from the plan walk, not just the reference guard: a filter is only harvested + * when every operator between it and the scan commutes with pushing the predicate into the scan + * (see `spineToScan`). Reference containment alone proves the predicate is expressible over the + * scan's output, not that moving it there is semantics-preserving -- an intervening LIMIT/TopN + * (or Sort, Aggregate, Window, join, ...) can change which rows the predicate would have + * applied to, so those stop the walk and the filter is left where Spark placed it. + */ + 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 on the path from `node` down to `scanExec` is one that pushdown can + // safely cross: a deterministic ProjectExec is 1:1 on rows (an Alias mints a new exprId, so + // it cannot alias over the scan's own output attributes, which is what the reference guard + // below requires) and a deterministic FilterExec only removes rows, so moving a predicate + // expressed over the scan's output through either preserves the query's semantics. A + // nondeterministic projection breaks that: a deterministic conjunct does not commute with it, + // because pushing the predicate into the scan changes which rows survive to have + // nondeterministic expressions (e.g. monotonically_increasing_id()) evaluated over them, + // changing the result -- so both guards require `.deterministic`, mirroring Spark's own + // PushPredicateThroughNonJoin/CollapseProject rules. Anything else (LIMIT/TopN, Sort, + // Aggregate, Window, joins, Union, Sample, ...) can reorder or drop rows in ways that make + // "push the predicate down to the scan" change the result, so an unrecognized node stops the + // walk: the filter is left uncollected (missed pruning only, never a correctness issue). + 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) + } + + /** + * 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, + * never a correctness issue. + * + * Known core-parity limitation: when the scan is fused under a parent native operator, + * `ensureSubqueriesResolved` has already called `updateResult()` on these subqueries and this + * path calls it again (Spark's ScalarSubquery.updateResult re-executes unconditionally). Core's + * CometNativeScanExec has the identical double-execution; benign for Delta (the subquery's + * snapshot is pinned at analysis), wasteful for expensive subqueries. 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) + : Seq[org.apache.comet.serde.ExprOuterClass.Expr] = { + if (!conf.getConf(org.apache.spark.sql.internal.SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) { + return Seq.empty + } + val subqueryFilters = dataFilters.filter(_.exists(_.isInstanceOf[ExecScalarSubquery])) + if (subqueryFilters.isEmpty) { + return Seq.empty + } + // Same binding guard as the DV shape's plan-time filters: references limited to the + // data-column prefix of the output, where positions agree between the output and the + // native read schema. For the plain shape strippedLen is the full required 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 _ => + }) + 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 + } + } + + /** + * 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( + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + firstFileUri: Option[java.net.URI], + hadoopConf: org.apache.hadoop.conf.Configuration) + : Option[OperatorOuterClass.NativeScanCommon.Builder] = { + val relation = scanExec.relation + val output = scanExec.output + val requiredSchema = scanExec.requiredSchema + + val commonBuilder = OperatorOuterClass.NativeScanCommon.newBuilder() + commonBuilder.setSource(scanExec.simpleStringWithNodeId()) + + val scanTypes = output.flatMap(attr => serializeDataType(attr.dataType)) + if (scanTypes.length != output.length) { + return None + } + commonBuilder.addAllFields(scanTypes.asJava) + + val strippedRequired = + StructType(requiredSchema.filterNot(f => internalColumnNames.contains(f.name))) + val strippedLen = strippedRequired.length + val requiredLen = requiredSchema.length + + // Keep only data filters that bind identically in the output and the native + // (strippedRequired ++ partitionFields) index spaces: references limited to the first + // strippedLen output attributes. Internal-column filters (is_row_deleted = 0) are + // trivially true after native DV application, and Spark's Filter above the scan + // re-evaluates everything anyway. + if (scanExec.conf.getConf( + org.apache.spark.sql.internal.SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) { + val dataColIds = output.take(strippedLen).map(_.exprId).toSet + val filterProtos = scanHelper.supportedDataFilters + .filter(_.references.forall(r => dataColIds.contains(r.exprId))) + .flatMap(f => exprToProto(f, output)) + commonBuilder.addAllDataFilters(filterProtos.asJava) + } + + // Constant metadata columns and real partition columns follow the required schema in the + // output, exactly like the plain shape. + val constantMetadataFields = scanExec.fileConstantMetadataColumns.map(attr => + StructField( + s"${CometNativeScan.constantMetadataFieldPrefix}${attr.name}", + attr.dataType, + attr.nullable)) + val internalFields = requiredSchema.fields.toSeq + .filter(f => internalColumnNames.contains(f.name)) + .map(f => StructField(s"$deltaConstFieldPrefix${f.name}", f.dataType, f.nullable)) + // The real partition columns must carry physical names in the proto, same as the data/ + // required schemas: a retained physical data name can otherwise collide with a partition + // column's LOGICAL name after a rename history (e.g. a->b then p->a), and DataFusion's + // name-based partition rewrite would then replace the data projection with the partition + // constant. constantMetadataFields/internalFields are synthetic slots the + // native side allocates for this scan, not table columns, so they are not physicalized. + val partitionSchemaFields = toPhysical(scanExec, relation.partitionSchema).fields.toSeq ++ + constantMetadataFields ++ internalFields + + // Protos carry physical names (column mapping); index math below stays logical. + val partitionSchemaProto = schema2Proto(partitionSchemaFields) + val requiredSchemaProto = schema2Proto(toPhysical(scanExec, strippedRequired)) + val dataSchemaProto = schema2Proto(toPhysical(scanExec, relation.dataSchema)) + + // Projection: data columns from the (stripped) read schema; internal columns from their + // constants slots at the END of the partition fields; the output tail (real partitions + + // constant metadata) positionally from the head of the partition fields. + val dataSchema = relation.dataSchema + val internalBase = dataSchema.length + partitionSchemaFields.length - internalFields.length + val internalIndexByName = requiredSchema.fields.toSeq + .filter(f => internalColumnNames.contains(f.name)) + .zipWithIndex + .map { case (f, i) => f.name -> (internalBase + i) } + .toMap + val projectionVector = output.zipWithIndex.map { case (attr, i) => + val idx = if (internalColumnNames.contains(attr.name)) { + internalIndexByName(attr.name) + } else if (i < requiredLen) { + dataSchema.fieldIndex(attr.name) + } else { + dataSchema.length + (i - requiredLen) + } + idx.toLong.asInstanceOf[java.lang.Long] + } + commonBuilder.addAllProjectionVector(projectionVector.asJava) + + commonBuilder.addAllDataSchema(dataSchemaProto.asJava) + commonBuilder.addAllRequiredSchema(requiredSchemaProto.asJava) + commonBuilder.addAllPartitionSchema(partitionSchemaProto.asJava) + + CometNativeScan.populateScanConfFlags( + commonBuilder, + strippedRequired, + firstFileUri, + hadoopConf, + scanExec.conf) + + Some(commonBuilder) + } + + /** Serialize one file partition into a DeltaSparkScan proto with per-file DV descriptors. */ + def serializePartition( + filePartition: FilePartition, + scanExec: FileSourceScanExec, + tableRoot: String): Array[Byte] = { + val relation = scanExec.relation + val sparkPartition = partition2Proto( + filePartition, + relation.partitionSchema, + scanExec.fileConstantMetadataColumns, + ShimFileFormat.fileConstantMetadataExtractors(relation.fileFormat)) + + val dvShape = isDvShape(scanExec) + + val deltaPartition = OperatorOuterClass.DeltaSparkFilePartition.newBuilder() + sparkPartition.getPartitionedFileList.asScala.zip(filePartition.files.toSeq).foreach { + case (fileProto, file) => + val fileBuilder = fileProto.toBuilder + if (dvShape) { + // Append the internal-constant values after the real partition/constant-metadata + // values, matching the order of the appended partition-schema fields. + scanExec.requiredSchema.fields + .filter(f => internalColumnNames.contains(f.name)) + .foreach { f => + val lit = f.dataType match { + case ByteType => Literal(0.toByte, ByteType) + case LongType => Literal(0L, LongType) + case other => + // Fixed internal invariant (observed Delta 3.3 types); fail loudly on + // drift rather than emit a plausible-looking constant. + throw new IllegalStateException( + s"Unexpected type $other for Delta internal column ${f.name}") + } + fileBuilder.addPartitionValues( + literalToProto(lit, s"delta internal constant ${f.name}")) + } + } + val dfb = OperatorOuterClass.DeltaSparkPartitionedFile + .newBuilder() + .setFile(fileBuilder.build()) + extractDvDescriptor(file, tableRoot).foreach(dfb.setDv) + deltaPartition.addPartitionedFile(dfb.build()) + } + + OperatorOuterClass.DeltaSparkScan + .newBuilder() + .setFilePartition(deltaPartition.build()) + .build() + .toByteArray + } + + /** + * Pull the DV descriptor Delta attached to this file (base64 under + * `row_index_filter_id_encoded`), resolving UUID-relative paths to absolute URLs and + * Z85-decoding inline bitmaps here on the JVM where delta-spark's codecs live. + */ + private def extractDvDescriptor( + file: PartitionedFile, + tableRoot: String): Option[OperatorOuterClass.DeltaSparkDvDescriptor] = { + val encoded = file.otherConstantMetadataColumnValues + .get(DeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED) + val filterType = file.otherConstantMetadataColumnValues + .get(DeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE) + encoded.map { enc => + filterType match { + case Some(RowIndexFilterType.IF_CONTAINED) | None => + case other => + // DeltaScanSupport declines CDF reads, the only source of inverted filters; + // reaching here means a gate was bypassed -- fail loudly rather than corrupt. + throw new IllegalStateException( + s"Native Delta scan cannot apply row index filter type $other") + } + val desc = DeletionVectorDescriptor.deserializeFromBase64(enc.asInstanceOf[String]) + val builder = OperatorOuterClass.DeltaSparkDvDescriptor + .newBuilder() + .setStorageType(desc.storageType) + .setSizeInBytes(desc.sizeInBytes) + .setCardinality(desc.cardinality) + if (desc.storageType == DeletionVectorDescriptor.INLINE_DV_MARKER) { + // The comet-spark jar relocates protobuf, so use the shaded ByteString. + builder.setInlineData( + org.apache.comet.shaded.protobuf.ByteString.copyFrom(desc.inlineData)) Review Comment: **[P2] Support the unshaded classpath during reactor compilation** Could we make the contrib's compile/shading arrangement work before core's package phase as well? A clean root `./mvnw -Pspark-4.0,delta compile` or `test` sees unshaded `spark/target/classes`. The `org.apache.comet.shaded.protobuf` namespace is created only by [core's package-phase shade execution](https://github.com/apache/datafusion-comet/blob/1d3557cf7eef766056d4c082f8be89355118bf53/spark/pom.xml#L567-L578), so this hard-coded reference cannot resolve in that reactor path. The separate core-install/contrib-test CI sequence masks it. An isolated Scala compilation against Java types generated from the current protocol reproduced `object shaded is not a member of package org.apache.comet`; the unshaded ByteString control compiled. The full reactor attempt was blocked earlier in dependency resolution, so this is compiler/classpath evidence rather than a completed reactor reproduction. Simply changing the import would invert the problem for packaged-core consumers; the module needs a consistent arrangement for both classpaths. ########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,539 @@ +/* + * 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 surviving rows are + * by construction not deleted: 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 the native side 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. + 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 all positional 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 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. createPhysicalSchema also stamps + // parquet.field.id metadata, but files written before the column-mapping upgrade have + // no field ids and would fail the reader's id expectations, strip the ids so the + // reader stays purely name-based (id mode, when enabled, will keep them). + 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. + */ + def convert(scanExec: FileSourceScanExec, scanHelper: CometScanExec): Option[Operator] = { + val relation = scanExec.relation + + val firstFileUri = scanHelper.selectedPartitions + .flatMap(_.files.headOption) + .headOption + .map(_.getPath.toUri) + + val hadoopConf = relation.sparkSession.sessionState + .newHadoopConfWithOptions(relation.options) + + 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 => + // 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. + val dvDescriptors = DeltaScanSupport.selectedDvDescriptors(scanHelper, tableRoot) + commonBuilder.putAllObjectStoreOptions( + mergedObjectStoreOptions( + hadoopConf, + storeUris(dvDescriptors, tableRootPath, firstFileUri)).asJava) + + val common = commonBuilder.build() + val deltaCommon = OperatorOuterClass.DeltaSparkScanCommon + .newBuilder() + .setTableRoot(tableRoot) + .setColumnMappingMode(columnMappingMode(scanExec)) + .setSourceKey(DeltaPlanDataInjector.sourceKey(tableRoot, common)) + .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 it is cheap to include even when absent), + * and every distinct on-disk DV authority from `descriptors`. Inline DVs are filtered out -- + * they carry no external URI, only embedded bytes. Deduping by authority (rather than by full + * URI) keeps this O(distinct authorities) instead of O(files): a table with N deletion-vector + * files on the same external store previously produced ~N distinct URIs here, each + * independently fed into `mergedObjectStoreOptions`'s `extractObjectStoreOptions` walk over + * `hadoopConf`. Candidates are deduped keeping the FIRST URI seen per authority, so callers can + * rely on `firstFileUri`/the table root winning over any DV path that happens to share their + * authority. Factored out of [[convert]] so the URI-assembly logic (in particular the + * `storageType` filter and `absolutePath` resolution) is directly unit-testable with hand-built + * [[DeletionVectorDescriptor]] fixtures, without a Spark session or real selected files (a + * `file://` scan alone can't exercise a foreign-authority DV). + */ + 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 simply + * union rather than pick one: the 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. Factored out of [[convert]] so it is directly + * unit-testable without a Spark session. + */ + 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 Spark + * version; the dedup keeps Spark 4.x from carrying duplicates. + * + * Safety comes from the plan walk, not just the reference guard: a filter is only harvested + * when every operator between it and the scan commutes with pushing the predicate into the scan + * (see `spineToScan`). Reference containment alone proves the predicate is expressible over the + * scan's output, not that moving it there is semantics-preserving -- an intervening LIMIT/TopN + * (or Sort, Aggregate, Window, join, ...) can change which rows the predicate would have + * applied to, so those stop the walk and the filter is left where Spark placed it. + */ + 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 on the path from `node` down to `scanExec` is one that pushdown can + // safely cross: a deterministic ProjectExec is 1:1 on rows (an Alias mints a new exprId, so + // it cannot alias over the scan's own output attributes, which is what the reference guard + // below requires) and a deterministic FilterExec only removes rows, so moving a predicate + // expressed over the scan's output through either preserves the query's semantics. A + // nondeterministic projection breaks that: a deterministic conjunct does not commute with it, + // because pushing the predicate into the scan changes which rows survive to have + // nondeterministic expressions (e.g. monotonically_increasing_id()) evaluated over them, + // changing the result -- so both guards require `.deterministic`, mirroring Spark's own + // PushPredicateThroughNonJoin/CollapseProject rules. Anything else (LIMIT/TopN, Sort, + // Aggregate, Window, joins, Union, Sample, ...) can reorder or drop rows in ways that make + // "push the predicate down to the scan" change the result, so an unrecognized node stops the + // walk: the filter is left uncollected (missed pruning only, never a correctness issue). + 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) + } + + /** + * 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, + * never a correctness issue. + * + * Known core-parity limitation: when the scan is fused under a parent native operator, + * `ensureSubqueriesResolved` has already called `updateResult()` on these subqueries and this + * path calls it again (Spark's ScalarSubquery.updateResult re-executes unconditionally). Core's + * CometNativeScanExec has the identical double-execution; benign for Delta (the subquery's + * snapshot is pinned at analysis), wasteful for expensive subqueries. 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) + : Seq[org.apache.comet.serde.ExprOuterClass.Expr] = { + if (!conf.getConf(org.apache.spark.sql.internal.SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) { + return Seq.empty + } + val subqueryFilters = dataFilters.filter(_.exists(_.isInstanceOf[ExecScalarSubquery])) + if (subqueryFilters.isEmpty) { + return Seq.empty + } + // Same binding guard as the DV shape's plan-time filters: references limited to the + // data-column prefix of the output, where positions agree between the output and the + // native read schema. For the plain shape strippedLen is the full required 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 _ => + }) + 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 + } + } + + /** + * 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( + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + firstFileUri: Option[java.net.URI], + hadoopConf: org.apache.hadoop.conf.Configuration) + : Option[OperatorOuterClass.NativeScanCommon.Builder] = { + val relation = scanExec.relation + val output = scanExec.output + val requiredSchema = scanExec.requiredSchema + + val commonBuilder = OperatorOuterClass.NativeScanCommon.newBuilder() + commonBuilder.setSource(scanExec.simpleStringWithNodeId()) + + val scanTypes = output.flatMap(attr => serializeDataType(attr.dataType)) + if (scanTypes.length != output.length) { + return None + } + commonBuilder.addAllFields(scanTypes.asJava) + + val strippedRequired = + StructType(requiredSchema.filterNot(f => internalColumnNames.contains(f.name))) + val strippedLen = strippedRequired.length + val requiredLen = requiredSchema.length + + // Keep only data filters that bind identically in the output and the native + // (strippedRequired ++ partitionFields) index spaces: references limited to the first + // strippedLen output attributes. Internal-column filters (is_row_deleted = 0) are + // trivially true after native DV application, and Spark's Filter above the scan + // re-evaluates everything anyway. + if (scanExec.conf.getConf( + org.apache.spark.sql.internal.SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) { + val dataColIds = output.take(strippedLen).map(_.exprId).toSet + val filterProtos = scanHelper.supportedDataFilters + .filter(_.references.forall(r => dataColIds.contains(r.exprId))) + .flatMap(f => exprToProto(f, output)) + commonBuilder.addAllDataFilters(filterProtos.asJava) + } + + // Constant metadata columns and real partition columns follow the required schema in the + // output, exactly like the plain shape. + val constantMetadataFields = scanExec.fileConstantMetadataColumns.map(attr => + StructField( + s"${CometNativeScan.constantMetadataFieldPrefix}${attr.name}", + attr.dataType, + attr.nullable)) + val internalFields = requiredSchema.fields.toSeq + .filter(f => internalColumnNames.contains(f.name)) + .map(f => StructField(s"$deltaConstFieldPrefix${f.name}", f.dataType, f.nullable)) Review Comment: **[P2] Make DV synthetic field names unique against the physical schemas** Could we choose collision-free synthetic names, or decline conflicting scans? A legal user column named `_comet_delta___delta_internal_is_row_deleted` collides with the partition constant generated here. DataFusion substitutes partition constants by column name, so the real data projection is replaced by the bookkeeping value `0`. A separate Spark/Delta probe accepted that TINYINT column with value `7`, created a real DV through DELETE, and showed the expected internal-column suffix in the scan schema. A standalone probe against the locked public DataFusion 54.1.0 reader then read real Parquet values `(1,7),(2,7)` as `(1,0),(2,0)` with this colliding constant; a distinct-name control preserved `7`. These are separate plan-shape and reader probes, not a full current-head Spark/JNI run. Please cover the collision in the native Delta differential suite, including the physical data and partition namespaces when allocating the synthetic fields. ########## contrib/delta-spark/dev/run-delta-regression.sh: ########## @@ -0,0 +1,163 @@ +#!/bin/bash +# +# 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. +# +# Run Delta Lake's own Spark test suites against a Comet build with the +# native Delta scan enabled. Clones delta at $DELTA_VERSION into $WORKDIR, +# injects Comet into the test SparkSession (DeltaSQLCommandTest) and the +# test classpath (unmanagedJars), then runs the given testOnly selectors. +# +# Usage: +# COMET_JARS=/path/comet-spark.jar,/path/comet-contrib-delta.jar,/path/flatbuffers.jar \ +# ./run-delta-regression.sh <workdir> 'org.apache.spark.sql.delta.DeletionVectorsSuite' [...] +# +# Env: +# DELTA_VERSION delta tag to test against (default 3.3.2) +# COMET_JARS comma-separated jars added to the test classpath (required) +# JAVA_HOME JDK for sbt (17 recommended) +set -euo pipefail + +DELTA_VERSION="${DELTA_VERSION:-3.3.2}" +WORKDIR="${1:?usage: run-delta-regression.sh <workdir> <suite> [...suites]}" +shift +[ $# -ge 1 ] || { echo "no suites given" >&2; exit 2; } +: "${COMET_JARS:?COMET_JARS must list the comet jars}" + +IFS=',' read -ra _jars <<< "$COMET_JARS" +for j in "${_jars[@]}"; do + [ -f "$j" ] || { echo "COMET_JARS entry not found: $j" >&2; exit 2; } +done + +DELTA_DIR="$WORKDIR/delta-$DELTA_VERSION" +if [ ! -d "$DELTA_DIR" ]; then + git clone --depth 1 --branch "v$DELTA_VERSION" https://github.com/delta-io/delta.git "$DELTA_DIR" +elif [ ! -d "$DELTA_DIR/.git" ]; then + echo "stale/partial checkout at $DELTA_DIR; remove it (rm -rf) and rerun" >&2 + exit 2 +fi +cd "$DELTA_DIR" + +# Add COMET_EXTRA_JARS to every project's test classpath, plus the JDK-17 +# module-access flags Spark needs (both for forked test JVMs and sbt's own JVM). +if ! grep -q "COMET_EXTRA_JARS" build.sbt; then + python3 - <<'EOF' +s = open('build.sbt').read() +marker = 'lazy val commonSettings = Seq(' +opens = [ + "--add-opens=java.base/java.lang=ALL-UNNAMED", + "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED", + "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED", + "--add-opens=java.base/java.io=ALL-UNNAMED", + "--add-opens=java.base/java.net=ALL-UNNAMED", + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-opens=java.base/java.util=ALL-UNNAMED", + "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED", + "--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED", + "--add-opens=java.base/jdk.internal.ref=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.cs=ALL-UNNAMED", + "--add-opens=java.base/sun.security.action=ALL-UNNAMED", + "--add-opens=java.base/sun.util.calendar=ALL-UNNAMED", + "--add-exports=java.base/sun.nio.ch=ALL-UNNAMED", +] +opts = ", ".join('"%s"' % o for o in opens) +inject = ( + 'lazy val commonSettings = Seq(\n' + ' Test / unmanagedJars ++= sys.env.get("COMET_EXTRA_JARS").toSeq\n' + ' .flatMap(_.split(",")).map(p => Attributed.blank(file(p))),\n' + ' Test / fork := true,\n' + ' Test / javaOptions ++= Seq(%s),\n' % opts +) +assert marker in s, 'commonSettings marker not found' +open('build.sbt', 'w').write(s.replace(marker, inject, 1)) +EOF +fi + +# Inject Comet into the shared test SparkSession when COMET_EXTRA_JARS is set. +TEST_BASE=spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaSQLCommandTest.scala +if ! grep -q "CometSparkSessionExtensions" "$TEST_BASE"; then + python3 - "$TEST_BASE" <<'EOF' +import sys +p = sys.argv[1] +s = open(p).read() +old = ''' override protected def sparkConf: SparkConf = { + super.sparkConf + .set(StaticSQLConf.SPARK_SESSION_EXTENSIONS.key, + classOf[DeltaSparkSessionExtension].getName) + .set(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key, + classOf[DeltaCatalog].getName) + }''' +new = ''' override protected def sparkConf: SparkConf = { + val conf = super.sparkConf + .set(StaticSQLConf.SPARK_SESSION_EXTENSIONS.key, + classOf[DeltaSparkSessionExtension].getName) + .set(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key, + classOf[DeltaCatalog].getName) + if (sys.env.contains("COMET_EXTRA_JARS")) { + conf + .set(StaticSQLConf.SPARK_SESSION_EXTENSIONS.key, + classOf[DeltaSparkSessionExtension].getName + + ",org.apache.comet.CometSparkSessionExtensions") + .set("spark.comet.enabled", "true") + .set("spark.comet.exec.enabled", "true") + .set("spark.comet.exec.shuffle.enabled", "true") + .set("spark.shuffle.manager", + "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager") + .set("spark.memory.offHeap.enabled", "true") + .set("spark.memory.offHeap.size", "2g") + .set("spark.comet.scan.delta.enabled", "true") + } else conf + }''' +assert old in s, 'sparkConf block not found' +open(p, 'w').write(s.replace(old, new)) +EOF +fi + +# ScanReportHelper is a test-only trait that counts scans by pattern-matching +# FileSourceScanExec in the executed plan. The Comet Delta scan replaces those +# nodes, so claimed scans would go uncounted ("0 did not equal 2" in +# MergeIntoSuiteBase's insert-only data-skipping test). Map the Comet node back +# to the FileSourceScanExec it was built from: originalPlan carries the same +# PreparedDeltaFileIndex, so the reported paths and skipping stats are identical. +SCAN_HELPER=spark/src/test/scala/org/apache/spark/sql/delta/test/ScanReportHelper.scala +if [ -f "$SCAN_HELPER" ] && ! grep -q "CometDeltaNativeScanExec" "$SCAN_HELPER"; then + python3 - "$SCAN_HELPER" <<'EOF' +import sys +p = sys.argv[1] +s = open(p).read() +old = " case fs: FileSourceScanExec => Seq(fs)\n" +new = (" case fs: FileSourceScanExec => Seq(fs)\n" + " case c: org.apache.spark.sql.comet.CometDeltaNativeScanExec =>\n" + " Seq(c.originalPlan)\n") +assert s.count(old) == 1, s.count(old) +open(p, 'w').write(s.replace(old, new)) +EOF +fi + +export COMET_EXTRA_JARS="$COMET_JARS" +export SPARK_LOCAL_IP=127.0.0.1 +export RUST_BACKTRACE=1 + +cmds=() +for sel in "$@"; do + cmds+=("spark/testOnly $sel") +done + +LOG="$WORKDIR/delta-regression-$(date +%Y%m%d-%H%M%S).log" +echo "==> logging to $LOG" +build/sbt "${cmds[@]}" 2>&1 | tee "$LOG" | grep -E "^\[info\] (Tests:|Suites:|All tests|.*\*\*\* FAILED| - )" | tail -80 Review Comment: **[P3] Resolve the work directory before changing into the Delta checkout** Could we normalize `WORKDIR` to an absolute path before `cd "$DELTA_DIR"`? A relative argument such as `work` is initially valid, but this log path is later evaluated from inside `work/delta-<version>`, so `tee` tries to create `work/delta-<version>/work/delta-regression-...log`. That directory normally does not exist, and `set -o pipefail` makes the harness fail even when the test command succeeds. Verified with the unchanged script and a stubbed `sbt` command to isolate path handling: the relative-workdir case exited 1 with `tee: ... No such file or directory`, while the absolute-workdir control exited 0. No actual Delta test suite was run by that path-handling probe. ########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,981 @@ +/* + * 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 + +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.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. Note `deletionVectors` and `columnMapping` are declined separately (below) so their + * fallback reasons are specific. + */ + 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`: this is + * the first gate on every V1 scan and must stay inert when delta-spark is absent from the + * classpath, where `classOf[DeltaParquetFileFormat]` would raise NoClassDefFoundError inside + * CometScanRule and take down every parquet scan in the session. A name match proves + * delta-spark is present, so Delta types used past this gate are safe. + */ + def isDeltaScan(scanExec: FileSourceScanExec): Boolean = + scanExec.relation.fileFormat.getClass.getName == + "org.apache.spark.sql.delta.DeltaParquetFileFormat" + + /** + * Returns the first reason this Delta scan cannot go native, or None when claimable. Only + * called when [[isDeltaScan]] is true. `scanHelper` is the same [[CometScanExec]] the caller + * builds to drive [[CometDeltaNativeScan.convert]] on a claim, reused here to resolve the + * scan's selected files for the multi-store gate below. + */ + def declineReason( + plan: SparkPlan, + scanExec: FileSourceScanExec, + scanHelper: CometScanExec): Option[String] = { + val format = scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + val protocol = format.protocol + val metadata = format.metadata + // Descriptor deserialization is expensive; hoisted once so it runs at most once per claim + // attempt. `lazy` since most scans are not DV-shaped and gates that return earlier should not + // pay for it. + val tableRoot = scanExec.relation.location.rootPaths.head.toString + lazy val dvDescriptors: Seq[DeletionVectorDescriptor] = + selectedDvDescriptors(scanHelper, tableRoot) + + // Mirrors core's CometScanRule.isSchemaSupported (the same type checker core's own + // FileSourceScanExec path runs) so scan-time type gates -- notably the default-on + // COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK safety fallback for ShortType, plus the + // collation and shredded-variant-struct gates -- apply identically whether the scan is + // claimed through core or through this contrib. A pure in-memory schema check with no + // file or store I/O, so it runs first, ahead of every other 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") + } + + // Name mode is supported via physical-name schemas; id mode needs the field-id path and + // stays declined until validated. + val cmMode = metadata.columnMappingMode.name + 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 uses the + // required schema verbatim as output: name-sensitive expressions (e.g. to_json) would leak + // physical names into results. Needs a rename adapter for the logical schema; decline until + // then. + 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") + } + // Bound native's memory for expanded DV row selectors (delta_dv.rs), bounded above by + // 2*cardinality + #row-groups; the descriptor's cardinality is a sound, pessimistic upper + // bound. 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 + // does not populate it, and Delta's DML find-touched-files scans use it (mirrors core's check + // in CometScanRule.nativeScan). + if (plan.exists(node => + node.expressions.exists(_.exists { + case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength => true + case _ => false + }))) { + return Some( + "Native Delta scan is not compatible with input_file_name, " + + "input_file_block_start, or input_file_block_length") + } + + // Row-index metadata columns are generated per-row by Spark's reader (mirrors core); the DV + // shape's trailing row-index column is exempt since the gates above already proved it dead. + if (!CometDeltaNativeScan.isDvShape(scanExec) && + ShimFileFormat.findRowIndexColumnIndexInSchema(scanExec.requiredSchema) >= 0) { + return Some("Native Delta scan does not support row index generation") + } + + // Mirror core's vectorized-reader compatibility gate. + if (!SQLConf.get.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED) && + !CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.get()) { + return Some( + "Native Delta scan is incompatible with " + + s"${SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key}=false") + } + + // Decline ALL encrypted-parquet configurations (stricter than core): the exec node does not + // yet wire the decryption-key broadcast to executors. + val hadoopConf = scanExec.relation.sparkSession.sessionState + .newHadoopConfWithOptions(scanExec.relation.options) + if (CometParquetUtils.encryptionEnabled(hadoopConf)) { + return Some("Native Delta scan does not support encrypted parquet") + } + + // Nested-type column defaults cannot be serialized; a dropped default would misalign the + // value/index lists consumed positionally on the native side. Mirrors core's + // transformV1Scan gate. + val possibleDefaultValues = getExistenceDefaultValues(scanExec.requiredSchema) + if (possibleDefaultValues.exists(d => + d != null && (d.isInstanceOf[ArrayBasedMapData] || d + .isInstanceOf[GenericInternalRow] || d.isInstanceOf[GenericArrayData]))) { + return Some("Native Delta scan does not support default values for nested types") + } + + // Only claim scans whose root paths object_store (or the configured libhdfs schemes) can + // actually read (mirrors core's unsupportedFsSchemes gate). + val libhdfs = libhdfsSchemes + val unsupportedRootSchemes = + unsupportedSchemes(scanExec.relation.location.rootPaths.map(_.toUri), libhdfs) + if (unsupportedRootSchemes.nonEmpty) { + return Some( + "Native Delta scan does not support filesystem scheme(s) " + + s"${unsupportedRootSchemes.mkString(", ")}") + } + + // A shallow clone can span multiple object-store authorities; the native builder resolves the + // scan's ObjectStoreUrl from the FIRST selected file only, so a later file under a different + // store would silently read through the wrong handle. Force file listing (scanHelper is + // already built for the claim path) and decline rather than risk it. + val dataFileUris = + scanHelper.selectedPartitions.iterator.flatMap(_.files).map(_.getPath.toUri).toSeq + + // Both gates below need the DV absolute-path URIs; dvDescriptors is already memoized. + val dvUris = dvDescriptors + .filter(_.storageType != DeletionVectorDescriptor.INLINE_DV_MARKER) + .map(_.absolutePath(new Path(tableRoot)).toUri) + + // The root-path gate above only inspects the table root(s); selected files can resolve + // through a different scheme (e.g. `viewfs:`). Checked before the authority gates below, + // which presume every URI is natively resolvable. + val unsupportedSelected = unsupportedSelectedSchemeReason(dataFileUris ++ dvUris, libhdfs) + if (unsupportedSelected.isDefined) { + return unsupportedSelected + } + + // Checked before multiStoreReason, which presumes every URI resolves to a single store + // identity -- a userinfo-bearing authority provably does not (store keying drops userinfo). + val userInfoReason = userInfoBearingAuthorityReason(dataFileUris ++ dvUris) + if (userInfoReason.isDefined) { + return userInfoReason + } + + val multiStore = multiStoreReason(dataFileUris) + if (multiStore.isDefined) { + return multiStore + } + + // Credentials that resolve ONLY through a Hadoop credential provider (JCEKS et al.) are + // invisible to the plain-conf extraction forwarded to the native S3 client; reuses hadoopConf + // from the encryption gate above. + val credentialReason = credentialAliasReason(hadoopConf, dataFileUris ++ dvUris) + if (credentialReason.isDefined) { + return credentialReason + } + + // A credential-provider class native's build_aws_credential_provider_metadata (s3.rs) does + // not recognize errors at scan EXECUTION time, after the scan was already claimed; decline + // eagerly instead. + val providerReason = providerClassGateReason(hadoopConf, dataFileUris ++ dvUris) + if (providerReason.isDefined) { + return providerReason + } + + // Reuse core's generic native-scan gates (ignoreCorruptFiles/ignoreMissingFiles, AQE DPP on + // Spark 3.4, exec enabled); tags its own fallback reasons. + if (!CometNativeScan.isSupported(scanExec)) { + return Some("Core native scan gates rejected the scan (see reasons above)") + } + + None + } + + /** + * Deletion-vector descriptors for every file this DV-shape scan selected, deserialized once and + * normalized to absolute on-disk paths via `copyWithAbsolutePath`. Returns `Seq.empty` for the + * plain shape ([[CometDeltaNativeScan.isDvShape]] false). Shared by the DV cardinality gate and + * [[CometDeltaNativeScan.convert]]'s object-store option merge. + */ + private[delta] def selectedDvDescriptors( + scanHelper: CometScanExec, + tableRoot: String): Seq[DeletionVectorDescriptor] = { + if (!CometDeltaNativeScan.isDvShape(scanHelper.wrapped)) { + return Seq.empty + } + val tableRootPath = new Path(tableRoot) + scanHelper.selectedPartitions.iterator + .flatMap(_.files) + .flatMap { file => + file.metadata + .get(DeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED) + .map(enc => DeletionVectorDescriptor.deserializeFromBase64(enc.asInstanceOf[String])) + } + .map(_.copyWithAbsolutePath(tableRootPath)) + .toSeq + } + + /** + * The libhdfs scheme exemption set from [[org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES]], + * lowercased and defaulting to `Set("hdfs")` when unset. Hoisted so [[declineReason]]'s + * root-path and selected-file scheme gates share one parsed set. + */ + private[delta] def libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get() match { + case Some(s) => + s.split(",").map(_.trim.toLowerCase(Locale.ROOT)).filter(_.nonEmpty).toSet + case None => Set("hdfs") + } + + /** + * The lowercased, deduplicated schemes among `uris` that neither `libhdfs` nor Comet's native + * object_store layer ([[CometScanRule.isNativelyReadableScheme]]) can read. A `null` scheme is + * tolerated, not flagged: such a URI cannot come from a Hadoop-backed source. Factored out of + * [[declineReason]]'s two scheme gates so both share one predicate, unit-testable without a + * Spark session. + */ + private[delta] def unsupportedSchemes(uris: Seq[URI], libhdfs: Set[String]): Set[String] = { + uris + .filter { uri => + val sch = uri.getScheme + sch != null && { + val sl = sch.toLowerCase(Locale.ROOT) + !libhdfs.contains(sl) && !CometScanRule.isNativelyReadableScheme(uri) + } + } + .map(_.getScheme.toLowerCase(Locale.ROOT)) + .toSet + } + + /** + * Decline reason when any of `uris` -- the scan's selected data-file and deletion-vector URIs + * -- use a scheme [[unsupportedSchemes]] flags, or `None` when every URI is natively readable + * (or libhdfs-exempt). See [[declineReason]]'s call site for ordering vs. the authority gates + * below. + */ + private[delta] def unsupportedSelectedSchemeReason( + uris: Seq[URI], + libhdfs: Set[String]): Option[String] = { + val schemes = unsupportedSchemes(uris, libhdfs) + if (schemes.isEmpty) { + None + } else { + Some( + "Native Delta scan does not support selected data file or deletion vector filesystem " + + s"scheme(s) ${schemes.mkString(", ")}") + } + } + + /** + * Decline reason when `uris` span more than one object-store authority (scheme plus the raw + * authority -- userinfo, host, port -- lowercased so `S3A://Bucket:1234` and + * `s3a://bucket:1234` collapse), or `None` when they share one. `file://` paths carry no + * authority, so local scans across many directories are unaffected. + */ + private[delta] def multiStoreReason(uris: Seq[URI]): Option[String] = { + val authorities = uris.map(uriAuthority).distinct + if (authorities.size > 1) { + Some( + "Native Delta scan does not support data files spanning multiple object stores " + + s"(found: ${authorities.sorted.mkString(", ")})") + } else { + None + } + } + + /** + * Normalizes `uri` to a lowercased `scheme://authority` string, keyed on the raw `getAuthority` + * rather than the individually-parsed host/port/userinfo fields: `getAuthority` already + * includes userinfo, so containers on the same storage account don't collapse together; and + * `getHost` (and `getUserInfo`/`getPort`) return `null` for the whole authority when it doesn't + * conform to RFC 3986's `reg-name` syntax (e.g. an underscore in a GCS bucket name, + * `gs://my_bucket`), silently collapsing distinct buckets into one empty-host key. A `null` + * authority normalizes to the empty string. + */ + private[delta] def uriAuthority(uri: URI): String = { + val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("") + val authority = Option(uri.getAuthority).map(_.toLowerCase(Locale.ROOT)).getOrElse("") + s"$scheme://$authority" + } + + /** + * The raw userinfo component of `uri`'s authority, or empty when none. Splits at the LAST `@` + * rather than using `URI#getUserInfo`: as with [[uriAuthority]], the structured getters return + * `null` for the whole authority when it fails RFC 3986 `reg-name` syntax (e.g. an underscore + * in a GCS bucket name), hiding a real userinfo component. Never lowercased: userinfo is + * case-sensitive. + */ + private[delta] def uriUserInfo(uri: URI): String = { + val authority = Option(uri.getAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + if (at >= 0) authority.substring(0, at) else "" + } + + /** + * Redacts `uri`'s authority to `scheme` + `://` + a literal `***` + `@host[:port]`, for + * embedding in a decline reason. NEVER interpolate `uri.getAuthority` or [[uriUserInfo]] + * directly into a reason string: doing so would leak credentials embedded as URI userinfo (e.g. + * `s3a://AKIA...:secret@bucket`) into the SQL plan's explain output, fallback-reason logging, + * or the Spark UI. + */ + private[delta] def redactedAuthority(uri: URI): String = { + val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("") + val authority = Option(uri.getAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + val hostPort = if (at >= 0) authority.substring(at + 1) else authority + s"$scheme://***@$hostPort" + } + + /** + * Decline reason when any of `uris` carries userinfo in its authority (e.g. the container in an + * abfss:// path), or `None` when none do. The native store cache, `ObjectStoreUrl`, and + * DataFusion registry all key on scheme/host/port only, dropping userinfo, so two authorities + * differing only in userinfo collide onto the same store handle -- normally a missing-object + * error, but silently wrong data if a same-named object exists under both. + */ + private[delta] def userInfoBearingAuthorityReason(uris: Seq[URI]): Option[String] = { + val offending = uris.filter(uri => uriUserInfo(uri).nonEmpty).map(redactedAuthority).distinct + if (offending.isEmpty) { + None + } else { + Some("Native Delta scan does not support object-store paths whose authority carries " + + "userinfo (e.g. the container in an abfss:// path): the native object-store cache, " + + "ObjectStoreUrl and DataFusion registry all key on scheme, host and port only, so two " + + "containers on one storage account share a single store handle " + + s"(found: ${offending.sorted.mkString(", ")})") + } + } + + /** + * String-literal Hadoop conf keys consulted by [[credentialAliasReason]] below. `hadoop-aws` is + * NOT on this module's runtime classpath (`spark-hadoop-cloud` is test-scope only), so + * `org.apache.hadoop.fs.s3a.Constants` must never be referenced here -- doing so would raise + * `NoClassDefFoundError` and take down every Delta scan in a session with no S3 dependency at + * all, not just S3 ones. + */ + private val HadoopCredentialProviderPathKey = "hadoop.security.credential.provider.path" + private val S3aCredentialProviderPathKey = "fs.s3a.security.credential.provider.path" + + private def s3aBucketProviderPathKey(bucket: String): String = + s"fs.s3a.bucket.$bucket.security.credential.provider.path" + + /** + * The LONG form of [[s3aBucketProviderPathKey]]: `S3AUtils#lookupPassword` resolves per-bucket + * overrides through both a long key (`fs.s3a.bucket.B.<full base key>`) and a short key + * (`fs.s3a.bucket.B.<base key minus its "fs.s3a." prefix>`) -- see [[s3aCredentialAliases]] for + * why both must be covered here too. + */ + private def s3aBucketLongProviderPathKey(bucket: String): String = + s"fs.s3a.bucket.$bucket.fs.s3a.security.credential.provider.path" + + /** + * The aliases the native S3 credentials provider chain tries per bucket (short bucket form, + * then global), PLUS the LONG bucket form Hadoop's `S3AUtils#lookupPassword` also consults. + * Hadoop resolves long before short before global, keeping a non-empty long value over the + * short/global ones (`S3AUtils#getPassword` returns a non-empty `val` unchanged); native reads + * only short+global. So a JCEKS entry set ONLY under the long alias is resolved by Hadoop but + * invisible to native -- a shadowed value under ANY alias here means the caller must decline. + * List order only affects which alias name appears in the reason. + */ + private def s3aCredentialAliases(bucket: String): Seq[String] = + Seq( + s"fs.s3a.bucket.$bucket.fs.s3a.access.key", + s"fs.s3a.bucket.$bucket.fs.s3a.secret.key", + s"fs.s3a.bucket.$bucket.fs.s3a.session.token", + s"fs.s3a.bucket.$bucket.access.key", + s"fs.s3a.bucket.$bucket.secret.key", + s"fs.s3a.bucket.$bucket.session.token", + "fs.s3a.access.key", + "fs.s3a.secret.key", + "fs.s3a.session.token") + + private def nonEmptyConf(hadoopConf: Configuration, key: String): Boolean = + Option(hadoopConf.get(key)).exists(_.nonEmpty) + + /** + * The lowercase-scheme-checked S3/S3A bucket name from `uri`'s authority (host, minus any + * userinfo or port), or `None` when `uri`'s scheme is not `s3`/`s3a`. Parses the raw authority + * manually (last `@`, then last `:`) rather than using `URI#getHost`: the same RFC 3986 + * `reg-name` pitfall as [[uriAuthority]] applies (an underscore, valid in an S3 bucket name, + * makes `getHost` return `null` for the whole authority). + */ + private def s3Bucket(uri: URI): Option[String] = { + val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)) + if (scheme.contains("s3") || scheme.contains("s3a")) { + val authority = Option(uri.getAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + val hostAndPort = if (at >= 0) authority.substring(at + 1) else authority + val colon = hostAndPort.lastIndexOf(':') + val host = if (colon >= 0) hostAndPort.substring(0, colon) else hostAndPort + if (host.isEmpty) None else Some(host) + } else { + None + } + } + + /** + * The three per-bucket S3A credential base keys the native S3 client's `get_config` (`s3.rs`) + * resolves: short bucket key first, then global (native never reads the long bucket key). + */ + private val PlainCredentialBaseKeys: Seq[String] = + Seq("fs.s3a.access.key", "fs.s3a.secret.key", "fs.s3a.session.token") + + private def plainValue(hadoopConf: Configuration, key: String): Option[String] = + Option(hadoopConf.get(key)).filter(_.nonEmpty) + + /** + * The short-bucket-then-global value native's `get_config` (s3.rs) resolves for `baseKey` under + * `bucket`. Used by the credential-provider-class gates below, which read general S3A options + * the same way native does. + */ + private def effectiveOptionValue( + hadoopConf: Configuration, + bucket: String, + baseKey: String): Option[String] = { + val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.") + plainValue(hadoopConf, shortKey).orElse(plainValue(hadoopConf, baseKey)) + } + + private def plainLongFormCredentialDivergenceReason(bucket: String, longKey: String): String = + "Native Delta scan cannot forward long-form bucket credentials for " + + s"$bucket ($longKey is set but the native S3 client only reads the short bucket and " + + "global keys, so its credentials would differ from Hadoop's)" + + /** + * Zero-I/O plain-value divergence check (see [[s3aCredentialAliases]] for the long-before-short + * precedence). Native's `get_config` never reads the long form, so it can silently diverge from + * Hadoop's effective value. Declines when the long form is set and the two sides disagree + * (short set to the SAME value as long passes). Never interpolates a resolved value, only the + * key name. + * + * This applies to credentials ONLY. Hadoop resolves them via `S3AUtils#lookupPassword`, where + * the long bucket form (`fs.s3a.bucket.B.fs.s3a.<key>`) WINS when set -- a real, silent + * divergence from native, which never reads it. Every OTHER `fs.s3a` option instead flows + * through `S3AUtils#propagateBucketOptions`, which strips only ONE `fs.s3a.bucket.B.` prefix + * layer: a long-form key there folds into `fs.s3a.fs.s3a.<key>`, a key nothing else in Hadoop + * ever reads, so the long form is inert for those options -- Hadoop's own effective value + * already reduces to short.orElse(global), matching native. No gate is needed (or present) for + * that wider set; see DeltaScanContribSuite's pinned control test for the long-form endpoint + * case. + */ + private def plainLongFormCredentialReason( + hadoopConf: Configuration, + bucket: String): Option[String] = { + PlainCredentialBaseKeys.foldLeft(Option.empty[String]) { (declined, baseKey) => + if (declined.isDefined) { + declined + } else { + val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.") + val longKey = s"fs.s3a.bucket.$bucket.$baseKey" + val long = plainValue(hadoopConf, longKey) + if (long.isEmpty) { + None + } else { + val short = plainValue(hadoopConf, shortKey) + val global = plainValue(hadoopConf, baseKey) + val hadoopEffective = long.orElse(short).orElse(global) + val nativeEffective = short.orElse(global) + if (hadoopEffective != nativeEffective) { + Some(plainLongFormCredentialDivergenceReason(bucket, longKey)) + } else { + None + } + } + } + } + } + + /** + * String-literal mirror of every credential-provider class name s3.rs's + * `build_aws_credential_provider_metadata` and `is_anonymous_credential_provider` recognize + * (Hadoop S3A names plus AWS SDK v1/v2 names). `hadoop-aws` is NOT on this module's runtime + * classpath (see the note above [[HadoopCredentialProviderPathKey]]), so these are string + * literals, never `org.apache.hadoop.fs.s3a.auth.*` or AWS SDK provider `classOf` references. + */ + private val SupportedCredentialProviderClasses: Set[String] = Set( + "org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider", + "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider", + "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider", + "org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider", + "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider", + "software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider", + "com.amazonaws.auth.ContainerCredentialsProvider", + "com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper", + "software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider", + "com.amazonaws.auth.InstanceProfileCredentialsProvider", + "software.amazon.awssdk.auth.credentials.EnvironmentVariableCredentialsProvider", + "com.amazonaws.auth.EnvironmentVariableCredentialsProvider", + "software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider", + "com.amazonaws.auth.WebIdentityTokenCredentialsProvider", + "software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider", + "com.amazonaws.auth.profile.ProfileCredentialsProvider", + "software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider", + "com.amazonaws.auth.AnonymousAWSCredentials") + + private val AnonymousCredentialProviderClasses: Set[String] = Set( + "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider", + "software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider", + "com.amazonaws.auth.AnonymousAWSCredentials") + + private val HadoopAssumedRoleProviderClass = + "org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider" + + private val AwsCredentialsProviderKey = "fs.s3a.aws.credentials.provider" + private val AssumedRoleCredentialsProviderKey = "fs.s3a.assumed.role.credentials.provider" + + /** + * Splits a Hadoop-style comma-separated credential-provider-class list the same way s3.rs's + * `parse_credential_provider_names` does: split on comma, trim each entry, drop empties. + */ + private def parseProviderClassNames(value: String): Seq[String] = + value.split(",").map(_.trim).filter(_.nonEmpty).toSeq + + private def unsupportedProviderReason(bucket: String, key: String, className: String): String = + s"Native Delta scan does not support the credential provider class $className " + + s"configured via $key for $bucket (the native S3 client only supports a fixed set of " + + "provider classes; an unsupported class would fail at scan execution time, after the " + + "scan was already claimed, rather than at planning time)" + + private def mixedAnonymousProviderReason(bucket: String, key: String): String = + s"Native Delta scan does not support $key for $bucket naming an anonymous credential " + + "provider together with any other provider (the native S3 client rejects this " + + "combination at scan execution time)" + + private def anonymousAssumedRoleProviderReason(bucket: String, key: String): String = + s"Native Delta scan does not support an anonymous credential provider in $key for " + + s"$bucket (the native S3 client does not allow an anonymous provider as the base " + + "credentials for an assumed-role chain)" + + private def unsupportedProviderNameReason( + bucket: String, + key: String, + names: Seq[String]): Option[String] = + names + .find(name => !SupportedCredentialProviderClasses.contains(name)) + .map(unsupportedProviderReason(bucket, key, _)) + + /** + * Decline reason when `bucket`'s effective (short-bucket-then-global; see + * [[effectiveOptionValue]]) `assumed.role.credentials.provider` names an unsupported class, or + * an anonymous one (native's `build_assume_role_credential_provider_metadata` rejects ANY + * anonymous entry here, not just a mix). Unset defaults to native's own hardcoded + * `[SimpleAWSCredentialsProvider, EnvironmentVariableCredentialsProvider]` fallback, both + * always supported, so `None` is safe. + */ + private def assumedRoleProviderClassReason( + hadoopConf: Configuration, + bucket: String): Option[String] = { + effectiveOptionValue(hadoopConf, bucket, AssumedRoleCredentialsProviderKey) match { + case None => None + case Some(value) => + val names = parseProviderClassNames(value) + unsupportedProviderNameReason(bucket, AssumedRoleCredentialsProviderKey, names).orElse { + if (names.exists(AnonymousCredentialProviderClasses.contains)) { + Some(anonymousAssumedRoleProviderReason(bucket, AssumedRoleCredentialsProviderKey)) + } else { + None + } + } + } + } + + /** + * Decline reason when `bucket`'s effective `aws.credentials.provider` names a class native's + * `build_aws_credential_provider_metadata` (s3.rs) does not recognize, mixes an anonymous + * provider with any other provider (native's `build_credential_provider` rejects this + * combination), or -- when `AssumedRoleCredentialProvider` is among the names -- its + * `assumed.role.credentials.provider` sub-chain has the same problem. An unset/empty value is + * fine: native falls back to its own default AWS SDK provider chain. Checked so an unsupported + * class or invalid combination declines at planning time instead of erroring during scan + * execution, after the scan was already claimed. + */ + private def providerClassReason(hadoopConf: Configuration, bucket: String): Option[String] = { + effectiveOptionValue(hadoopConf, bucket, AwsCredentialsProviderKey).flatMap { value => + val names = parseProviderClassNames(value) + unsupportedProviderNameReason(bucket, AwsCredentialsProviderKey, names) + .orElse { + if (names.length > 1 && names.exists(AnonymousCredentialProviderClasses.contains)) { + Some(mixedAnonymousProviderReason(bucket, AwsCredentialsProviderKey)) + } else { + None + } + } + .orElse { + if (names.contains(HadoopAssumedRoleProviderClass)) { + assumedRoleProviderClassReason(hadoopConf, bucket) + } else { + None + } + } + } + } + + /** + * Returns the first reason any bucket among `uris` names an unsupported (or invalidly combined) + * credential-provider class, or `None` when every bucket's provider configuration is one native + * can build. See [[providerClassReason]]. + */ + private[delta] def providerClassGateReason( + hadoopConf: Configuration, + uris: Seq[URI]): Option[String] = { + val buckets = uris.flatMap(s3Bucket).distinct + buckets.foldLeft(Option.empty[String]) { (declined, bucket) => + if (declined.isDefined) declined else providerClassReason(hadoopConf, bucket) + } + } + + private def s3aScopedProviderPathReason(bucket: String, providerPathKey: String): String = + "Native Delta scan cannot forward Hadoop credential-provider aliases for " + + s"$bucket ($providerPathKey configures an S3A-scoped Hadoop credential provider that " + + "Configuration#getPassword does not consult, so the native S3 client's credentials " + + "cannot be verified)" + + private def shadowedCredentialAliasReason(bucket: String, alias: String): String = + "Native Delta scan cannot forward Hadoop credential-provider aliases for " + + s"$bucket ($alias resolves through $HadoopCredentialProviderPathKey but is not present " + + "as a plain configuration value, so the native S3 client would have no credentials)" + + private def unverifiableCredentialProviderReason(bucket: String, error: Throwable): String = + "Native Delta scan cannot verify Hadoop credential-provider aliases for " + + s"$bucket (reading $HadoopCredentialProviderPathKey raised " + + s"${error.getClass.getName}), declining rather than risk missing credentials" + + /** + * Compares each [[s3aCredentialAliases]] alias's plain `Configuration#get` value against + * `Configuration#getPassword` (providers first, plain conf as fallback); a resolved value that + * differs means the alias is invisible or wrong to native's plain-conf extraction, so decline. + * Only called once the GLOBAL provider path is established as the sole path for `bucket`. + * + * Runs inside try/catch: `getPassword` performs real keystore I/O, and a corrupt or unreadable + * store must decline this bucket rather than abort planning for the whole session. + */ + private def verifyGlobalProviderAliases( + hadoopConf: Configuration, + bucket: String): Option[String] = { + try { + s3aCredentialAliases(bucket).foldLeft(Option.empty[String]) { (declined, alias) => + if (declined.isDefined) { + declined + } else { + val resolved = + Option(hadoopConf.getPassword(alias)).map(new String(_)).filter(_.nonEmpty) + resolved match { + case Some(value) if !Option(hadoopConf.get(alias)).contains(value) => + Some(shadowedCredentialAliasReason(bucket, alias)) + case _ => None + } + } + } + } catch { + case e @ (_: IOException | _: RuntimeException) => + Some(unverifiableCredentialProviderReason(bucket, e)) + } + } + + /** + * The decline reason for `bucket` alone, or `None` when its credentials are safe to forward. + * `globalPathSet`/`s3aPathSet` are hoisted by the caller since they don't vary per bucket. + * + * Checked FIRST, regardless of provider-path config: [[plainLongFormCredentialReason]] is a + * zero-I/O, provider-unrelated check that is decisive on its own. + * + * Zero-I/O precheck: `None` immediately when none of the four provider-path keys are set -- a + * config-map lookup only, no keystore access. + * + * Arm A: the S3A-scoped provider-path keys point S3A's own resolution at a provider + * `Configuration#getPassword` does not consult, so any being set declines with no keystore + * read. + * + * Arm B: only the global path is set, which `getPassword` DOES consult; delegates to + * [[verifyGlobalProviderAliases]]. + */ + private def bucketCredentialAliasReason( + hadoopConf: Configuration, + bucket: String, + globalPathSet: Boolean, + s3aPathSet: Boolean): Option[String] = { + plainLongFormCredentialReason(hadoopConf, bucket).orElse { + val bucketPathKey = s3aBucketProviderPathKey(bucket) + val bucketLongPathKey = s3aBucketLongProviderPathKey(bucket) + val bucketPathSet = nonEmptyConf(hadoopConf, bucketPathKey) + val bucketLongPathSet = nonEmptyConf(hadoopConf, bucketLongPathKey) + if (!globalPathSet && !s3aPathSet && !bucketPathSet && !bucketLongPathSet) { + None + } else if (s3aPathSet || bucketPathSet || bucketLongPathSet) { + val offendingKey = + if (s3aPathSet) S3aCredentialProviderPathKey + else if (bucketPathSet) bucketPathKey + else bucketLongPathKey + Some(s3aScopedProviderPathReason(bucket, offendingKey)) + } else { + verifyGlobalProviderAliases(hadoopConf, bucket) + } + } + } + + /** + * Returns the first reason a native S3 scan cannot faithfully forward this table's Hadoop + * credentials, or `None` when claimable. `uris` is the scan's data-file and DV URIs; only + * `s3`/`s3a` authorities matter here (ABFS/WASB mooted by the userinfo gate, GCS out of scope). + * `Configuration#getPassword` checks every provider FIRST, falling back to plain conf only when + * unset, so a keystore-only or shadowing alias is invisible/wrong to native's plain-conf + * extraction; when in doubt, decline. Never interpolates a resolved value, only key names. + */ + private[delta] def credentialAliasReason( + hadoopConf: Configuration, + uris: Seq[URI]): Option[String] = { + val buckets = uris.flatMap(s3Bucket).distinct + if (buckets.isEmpty) { + return None Review Comment: **[P2] Decline Hadoop-only GCS authentication until it can be forwarded** Could we also gate GCS authentication compatibility? A scan with local/S3 data files and an absolute `gs://private-bucket/...` DV can pass the scheme and authority gates while relying only on Hadoop's `fs.gs.auth.service.account.json.keyfile` for GCS access. This function and the provider-class gate consider only S3 authorities. Although conversion forwards `fs.gs.*`, the native resolver's [GCS/default branch](https://github.com/apache/datafusion-comet/blob/1d3557cf7eef766056d4c082f8be89355118bf53/native/core/src/parquet/parquet_support.rs#L565-L574) calls `parse_url(&url)` without those options. With no alternative ADC or metadata-service credentials, Spark can read the sidecar but the claimed native scan fails authentication instead of falling back. The core GCS options omission predates this PR; the finding here is the Delta admission/fallback gap exposing that limitation. This is a source-traced case; no cloud credentials or GCS requests were used. A conservative fallback would be sufficient until native GCS option translation is supported. -- 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]
