sunchao commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3827550396
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala: ########## @@ -0,0 +1,450 @@ +/* + * 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.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 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 = relation.partitionSchema, + fileConstantMetadataColumns = scanExec.fileConstantMetadataColumns, + dataFilters = scanHelper.supportedDataFilters, + firstFileUri = firstFileUri, + hadoopConf = hadoopConf, + conf = scanExec.conf) + } else { + buildDvScanCommon(scanExec, scanHelper, firstFileUri, hadoopConf) + } + + commonOpt.map { commonBuilder => + val common = commonBuilder.build() + val tableRoot = relation.location.rootPaths.head.toString + 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() + } + } + + /** + * 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. + */ + 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} + // Nearest FilterExec above the scan (the DV shape interposes nodes between them, so do + // not require a direct parent-child edge). Safety comes from the reference guard, not the + // plan walk: only conjuncts expressed directly over the scan's own output attributes + // survive, so filters from other branches or over aliased projections contribute nothing. + val filtersAboveScan = plan.collect { + case f: org.apache.spark.sql.execution.FilterExec if f.find(_ eq scanExec).isDefined => f + } + filtersAboveScan.lastOption + .map { f => + splitConjunctivePredicates(f.condition) + .filter(_.references.subsetOf(scanExec.outputSet)) Review Comment: **[P1] Also stop at nondeterministic projections** The LIMIT/TopN case is fixed at `7e09e04f`, but `spineToScan` still crosses every `ProjectExec`. A deterministic conjunct does not commute with a nondeterministic projection. For a single-file Delta table `t` containing IDs 0–4: ```sql SELECT id FROM (SELECT id, monotonically_increasing_id() AS seq FROM t) q WHERE id > (SELECT max(id) FROM range(1)) AND seq = 1 ``` An isolated probe using the unchanged current-head harvesting method on a real Spark 4.0.2 / Delta 4.0.0 physical plan harvests `id > scalar-subquery`; applying that resolved predicate immediately above the scan changes the result from ID 1 to ID 2. With `spark.comet.parquet.rowFilterPushdown.enabled=true`, the native path can perform that same early filtering. Keeping the covering filter does not restore the sequence values assigned to the surviving rows. This was an exact-method/physical-plan probe, not a full Comet/JNI run. Could we require deterministic projection expressions before crossing a `ProjectExec`, as Spark's predicate-pushdown rule does, and add this regression? ########## native/core/src/execution/planner/delta_spark_scan.rs: ########## @@ -0,0 +1,145 @@ +// 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. + +//! JVM-planned Delta handler for the generic `OpStruct::ContribScan` dispatcher, feature-gated +//! behind `delta`. +//! +//! delta-spark has already done log replay, snapshot resolution, and partition pruning by the +//! time the scan reaches Comet, so the envelope carries a concrete file list (plus deletion +//! vector descriptors) and the read path reuses the exact same shared parquet scan builder as +//! `NativeScan` -- inheriting row-group stats pruning, page-index pruning, and filter pushdown. +//! Sibling of the kernel-planned handler in `delta_scan.rs`; the two claim different +//! `type_url`s within the same `ContribScan` envelope. + +use std::collections::HashMap; + +use datafusion_comet_proto::spark_operator::{ + ContribScan, DeltaSparkScan, Operator, SparkFilePartition, +}; +use prost::Message; + +use crate::execution::operators::ExecutionError::GeneralError; +use crate::execution::planner::PhysicalPlanner; +use crate::execution::planner::PlanCreationResult; + +/// Type name the JVM-planned Delta contrib claims within the `ContribScan` envelope. The +/// contrib jar packs a `DeltaSparkScan` with a `type_url` of +/// `type.googleapis.com/comet.contrib.delta_spark.DeltaSparkScan`; dispatch keys on the +/// contrib-owned suffix, same convention as the kernel path's `delta_scan.rs`. +const DELTA_SPARK_SCAN_TYPE_NAME: &str = "comet.contrib.delta_spark.DeltaSparkScan"; + +/// Contrib entry point for the `OpStruct::ContribScan` dispatcher. Returns `Some(result)` when +/// the envelope carries a JVM-planned Delta scan, or `None` when the `type_url` belongs to some +/// other contrib. +pub(crate) fn try_plan_contrib_scan( + planner: &PhysicalPlanner, + spark_plan: &Operator, + contrib: &ContribScan, +) -> Option<PlanCreationResult> { + if !contrib.type_url.ends_with(DELTA_SPARK_SCAN_TYPE_NAME) { + return None; + } + Some( + DeltaSparkScan::decode(contrib.value.as_slice()) + .map_err(|e| { + GeneralError(format!( + "Failed to decode DeltaSparkScan from contrib_scan: {e}" + )) + }) + .and_then(|scan| plan_delta_spark_scan(planner, spark_plan, &scan)), + ) +} + +fn plan_delta_spark_scan( + planner: &PhysicalPlanner, + spark_plan: &Operator, + scan: &DeltaSparkScan, +) -> PlanCreationResult { + // Delta data files are plain parquet; the read path deliberately reuses + // the same shared parquet scan builder as NativeScan so Delta inherits + // row-group stats pruning, page-index pruning, and filter pushdown. Only + // the file list arrives in Delta-specific form. Note delta_common's + // column_mapping_mode is informational in M1: the actual field-id + // matching switch is common.use_field_id, same as the Iceberg path. + let common = scan + .common + .as_ref() + .ok_or_else(|| GeneralError("DeltaSparkScan missing common data".into()))?; + + let delta_partition = scan + .file_partition + .as_ref() + .ok_or_else(|| GeneralError("DeltaSparkScan missing file_partition".into()))?; + + let spark_partition = SparkFilePartition { + partitioned_file: delta_partition + .partitioned_file + .iter() + .map(|f| { + f.file + .clone() + .ok_or_else(|| GeneralError("DeltaSparkPartitionedFile missing inner file".into())) + }) + .collect::<Result<Vec<_>, _>>()?, + }; + + let (object_store_url, mut files) = + planner.prepare_scan_store_and_files(common, &spark_partition)?; Review Comment: **[P1] Decline the residual cross-container DV case before claiming the scan** The mixed-data-file guard fixes the original case. For the residual you identified, could we add a Delta-side decline guard until store identity is fixed? A shallow clone from `abfss://[email protected]/...` to `abfss://[email protected]/...`, followed by DELETE, can keep all data files in `source` while putting the new DV in `clone`. Both current authority gates accept it because they inspect only data files. At `7e09e04f`, an offline probe using the exact cache/prepare functions, new resolver, and real Azure store builder confirms that both URLs map to the same cache/registry identity: the DV resolves to `MicrosoftAzure { container: source }`, and its handle is pointer-identical to the data store. The DV read therefore requests the clone-relative path from the wrong container. Normally that fails with a missing object; a matching, well-formed object could instead supply the wrong deletion bitmap. The helper predates this PR, but the new single-scan data-plus-DV path makes this reachable. Declining sidecars whose full authority differs but collides in the current store identity would contain it without a broad cache refactor. Fixing only the local `resolved_stores` key is insufficient because the shared cache and DataFusion registry also collapse the container. No live Azure request was used in the probe. ########## native/core/src/execution/delta_dv.rs: ########## @@ -0,0 +1,587 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Delta Lake deletion-vector decoding and translation into DataFusion +//! [`ParquetAccessPlan`]s (feature = "delta"). +//! +//! Wire formats implemented here (from delta-spark's `DeletionVectorStore` / +//! `RoaringBitmapArray`, v3.3.2): +//! - On-disk DV file: 1 version byte at the start of the file; at +//! `descriptor.offset`: `[i32 BE size][data: size bytes][i32 BE CRC32(data)]`. +//! - `data`: `[i32 LE magic]` then either +//! - magic 1681511376 ("native"): `[i32 LE count]`, then per bitmap +//! `[i32 LE size][standard 32-bit RoaringBitmap]`, keys implicit (index); +//! - magic 1681511377 ("portable", the spec's 64-bit extension): `[i64 LE +//! count]`, then per bitmap `[i32 LE key][standard 32-bit RoaringBitmap]` +//! with keys ascending -- exactly [`RoaringTreemap`]'s serialized form. + +use std::collections::HashMap; +use std::sync::Arc; + +use datafusion::datasource::listing::PartitionedFile; +use datafusion::datasource::physical_plan::parquet::metadata::DFParquetMetadata; +use datafusion::datasource::physical_plan::parquet::ParquetAccessPlan; +use datafusion::execution::runtime_env::RuntimeEnv; +use futures::{StreamExt, TryStreamExt}; +use object_store::ObjectStoreExt; +use parquet::arrow::arrow_reader::{RowSelection, RowSelector}; +use parquet::file::metadata::PageIndexPolicy; +use roaring::{RoaringBitmap, RoaringTreemap}; + +use crate::execution::operators::ExecutionError; +use crate::execution::operators::ExecutionError::GeneralError; +use crate::parquet::parquet_support::prepare_object_store_with_configs; +use datafusion_comet_proto::spark_operator::DeltaSparkDvDescriptor; + +const NATIVE_MAGIC: i32 = 1681511376; +const PORTABLE_MAGIC: i32 = 1681511377; + +/// Unframe a DV blob read from `descriptor.offset` of a DV file: +/// `[i32 BE size][data][i32 BE crc]`. Verifies both the size against the +/// descriptor's `size_in_bytes` and the CRC32 checksum. +pub fn unframe_dv_blob(blob: &[u8], expected_size: usize) -> Result<&[u8], ExecutionError> { + if blob.len() < 8 { + return Err(GeneralError(format!( + "Deletion vector blob too short: {} bytes", + blob.len() + ))); + } + let size = i32::from_be_bytes(blob[0..4].try_into().unwrap()); + if size < 0 || size as usize != expected_size { + return Err(GeneralError(format!( + "Deletion vector size mismatch: file says {size}, descriptor says {expected_size}" + ))); + } + let end = 4 + size as usize; + if blob.len() < end + 4 { + return Err(GeneralError(format!( + "Deletion vector blob truncated: need {} bytes, have {}", + end + 4, + blob.len() + ))); + } + let data = &blob[4..end]; + let expected_crc = i32::from_be_bytes(blob[end..end + 4].try_into().unwrap()); + let actual_crc = crc32fast::hash(data) as i32; + if expected_crc != actual_crc { + return Err(GeneralError( + "Deletion vector checksum mismatch".to_string(), + )); + } + Ok(data) +} + +/// Deserialize the magic-prefixed RoaringBitmapArray into a 64-bit treemap of +/// deleted row indexes. +pub fn deserialize_dv_bitmap(data: &[u8]) -> Result<RoaringTreemap, ExecutionError> { + if data.len() < 4 { + return Err(GeneralError( + "Deletion vector bitmap too short for magic number".to_string(), + )); + } + let magic = i32::from_le_bytes(data[0..4].try_into().unwrap()); + let rest = &data[4..]; + match magic { + PORTABLE_MAGIC => RoaringTreemap::deserialize_from(rest) + .map_err(|e| GeneralError(format!("Invalid portable deletion vector bitmap: {e}"))), + NATIVE_MAGIC => { + if rest.len() < 4 { + return Err(GeneralError( + "Native deletion vector bitmap missing count".to_string(), + )); + } + let count = i32::from_le_bytes(rest[0..4].try_into().unwrap()); + if count < 0 { + return Err(GeneralError(format!( + "Invalid RoaringBitmapArray length ({count} < 0)" + ))); + } + let mut pos = 4usize; + let mut treemap = RoaringTreemap::new(); + for key in 0..count as u64 { + if rest.len() < pos + 4 { + return Err(GeneralError( + "Native deletion vector bitmap truncated".to_string(), + )); + } + let size = i32::from_le_bytes(rest[pos..pos + 4].try_into().unwrap()); + pos += 4; + if size < 0 || rest.len() < pos + size as usize { + return Err(GeneralError( + "Native deletion vector bitmap truncated".to_string(), + )); + } + let bitmap = RoaringBitmap::deserialize_from(&rest[pos..pos + size as usize]) + .map_err(|e| { + GeneralError(format!("Invalid deletion vector sub-bitmap: {e}")) + })?; + pos += size as usize; + for value in bitmap { + treemap.insert((key << 32) | value as u64); + } + } + Ok(treemap) + } + other => Err(GeneralError(format!( + "Unexpected RoaringBitmapArray magic number {other}" + ))), + } +} + +/// Translate deleted row indexes into a [`ParquetAccessPlan`]: fully-deleted +/// row groups become `Skip`, untouched groups stay `Scan`, and partially +/// deleted groups get a `RowSelection` selecting the complement of the deleted +/// rows. Page-index pruning later INTERSECTS with these selections, so DV +/// skips and page skips compose. +pub fn build_access_plan( + row_group_row_counts: &[i64], + deleted: &RoaringTreemap, +) -> Result<ParquetAccessPlan, ExecutionError> { + let mut plan = ParquetAccessPlan::new_all(row_group_row_counts.len()); + // Single sweep over the (sorted) deleted row indexes, bucketing by row group. + let mut deleted_iter = deleted.iter().peekable(); + let mut group_start = 0u64; + for (idx, &num_rows) in row_group_row_counts.iter().enumerate() { + let num_rows = num_rows as u64; + let group_end = group_start + num_rows; + let mut selectors: Vec<RowSelector> = Vec::new(); + let mut cursor = group_start; + let mut deleted_in_group = 0u64; + while let Some(&row) = deleted_iter.peek() { + if row >= group_end { + break; + } + deleted_iter.next(); + deleted_in_group += 1; + if row > cursor { + selectors.push(RowSelector::select((row - cursor) as usize)); + } + // Merge runs of consecutive deleted rows into one skip. + match selectors.last_mut() { + Some(last) if last.skip => last.row_count += 1, + _ => selectors.push(RowSelector::skip(1)), Review Comment: **[P2] Reserve construction memory and account for the reader's copy** The cardinality cap and retained-plan reservations improve the original issue, but two allocations still bypass the pool at `7e09e04f`. `build_access_plan` finishes before `try_grow`, and Parquet's `RowSelection::from(Vec)` allocates a second vector during construction. Later, DataFusion 54.1's `create_initial_plan` deep-clones the attached access plan while the original remains in the file extensions, without an additional DV reservation. A counting-allocator probe using the unchanged `build_access_plan`/`total_selectors` functions and the locked dependencies tested 2,000,000 rows with exactly 1,000,000 alternating deletions, which the default cap admits. It measured 65,554,457 bytes of peak construction allocation before a 1-byte pool rejected the reservation. With a 32,000,000-byte reservation accepted, the opener-equivalent clone allocated another 32,000,025 bytes while the pool stayed at 32,000,000. These are allocator-requested bytes, not RSS or a reproduced executor OOM; reservation release on drop works correctly. Could we pre-reserve a conservative construction bound and then shrink it, and either transfer ownership or account for the active reader's copy? The current check can reject after memory pressure has already occurred. I would treat this as a remaining P2 under the new cap, rather than the original unbounded P1. ########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,438 @@ +/* + * 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.net.URI +import java.util.Locale + +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 +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 (not a further subclass)? Compared by class name, + * not `classOf`, deliberately: this is the first gate on every V1 scan, and it must stay inert + * when the contrib jar is deployed without delta-spark on the classpath: + * `classOf[DeltaParquetFileFormat]` here raises NoClassDefFoundError inside CometScanRule and + * takes down every parquet scan in the session. When the name matches, delta-spark is + * necessarily present (the instance exists), so the Delta types 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 it is claimable. Only + * called when [[isDeltaScan]] is true. `scanHelper` is the same [[CometScanExec]] the caller + * builds to drive [[CometDeltaNativeScan.convert]] on a claim, reused here (rather than listed + * separately) 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 + + if (format.isCDCRead) { + return Some("Native Delta scan does not support Change Data Feed reads") + } + + // Delta's DML machinery (findTouchedFiles with useMetadataRowIndex=false) injects a + // generated row-index column directly into the data schema and disables reader + // optimizations; its values must come from Spark's reader. Claiming such a scan would + // feed NULL row indexes into deletion-vector construction, silently corrupting DML. + 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 by serializing physical-name schemas (the parquet reader then + // matches file columns by name natively). 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; decline any column defaults under column mapping rather than + // return nulls where a default belongs. + 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 (not just top-level column + // names) to their physical, column-mapped form, and the shared native builder uses the + // required schema verbatim as the scan's output schema: struct fields below the top level + // would carry physical names. Ordinal access (GetStructField) is unaffected, but + // name-sensitive native expressions (e.g. to_json) read the Arrow struct field names + // directly and would leak physical names into query results. Restoring logical names + // natively needs a rename adapter/proto field for the logical schema (follow-up); 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 not + // supported, except Delta's DV bookkeeping columns, which the native path emits as + // constants (correct by construction once the DV is applied in the reader). + 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 not a DV read: it is Delta DML + // bookkeeping (findTouchedFiles building deletion bitmaps from REAL row indexes). + // Claiming it with constant row indexes 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") + } + // The 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") + } + // The row-index column's real values are consumed inside the reader when Spark applies + // the DV; native applies the DV itself and emits a dead constant instead, so the value + // must be provably unused above the scan (beyond the _metadata reassembly that gets + // discarded). + 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; decline rather + // than silently return nulls for backfilled columns in old files. + if (getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with deletion vectors") + } + // Bound the memory the native side will retain for expanded DV row selectors before + // committing to native execution: applying a deletion vector expands it into per-row + // RowSelectors that are reserved against the execution memory pool at scan time (see + // delta_dv.rs), and an alternating deleted/retained bitmap produces one non-coalescing + // selector per row. A row group's selector count is bounded above by + // 2*cardinality + #row-groups (each deleted row splits at most one run into a + // select/skip pair, plus one selector per row-group boundary), so the descriptor's + // cardinality -- deserialized at planning time via selectedDvDescriptors, no bitmap + // decode needed -- is a sound, pessimistic upper bound on the native reservation. + // Pessimistic by design: a large but CONTIGUOUS deletion is declined the same as a + // large alternating one, even though it would retain far fewer selectors natively; the + // conf below makes that recoverable. + val tableRoot = scanExec.relation.location.rootPaths.head.toString + val maxDeletedRowsPerFile = DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.get() + val oversizedCardinalities = selectedDvDescriptors(scanHelper, tableRoot) + .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 InputFileBlockHolder, a thread-local set by Spark's + // FileScanRDD; the native scan does not populate it. Delta's own DELETE/UPDATE/MERGE + // find-touched-files scans use input_file_name, so this gate is load-bearing for DML + // correctness (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: the gates above already proved its + // values are dead and the native path emits a constant for it. + 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, so claiming even a + // supported-encryption scan would fail at execution. + 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 (schema-evolution backfill of map/struct/array columns) + // cannot be serialized; a silently-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; otherwise a custom Hadoop FileSystem would fail at execution instead of + // falling back gracefully. Mirrors core's unsupportedFsSchemes gate. + val 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") + } + val unsupportedFsSchemes = scanExec.relation.location.rootPaths + .map(_.toUri) + .filter { uri => + val sch = uri.getScheme + sch != null && { + val sl = sch.toLowerCase(Locale.ROOT) + !libhdfsSchemes.contains(sl) && !CometScanRule.isNativelyReadableScheme(uri) + } + } + .map(_.getScheme.toLowerCase(Locale.ROOT)) + .toSet + if (unsupportedFsSchemes.nonEmpty) { + return Some( + "Native Delta scan does not support filesystem scheme(s) " + + s"${unsupportedFsSchemes.mkString(", ")}") + } + + // A Delta shallow clone across buckets followed by an append is a valid table whose data + // files span multiple object-store authorities. The shared native scan builder resolves + // the whole scan's ObjectStoreUrl from the FIRST selected file only and then strips every + // other file down to its bare object-store path, so a later file under a different store + // would silently read through the first file's store handle -- normally a NoSuchKey, but + // the wrong data if a same-named key happens to exist in both stores. Force file listing + // here (scanHelper is already built for the claim path, so this is not extra work) and + // decline rather than risk it. + val dataFileUris = + scanHelper.selectedPartitions.iterator.flatMap(_.files).map(_.getPath.toUri).toSeq + val multiStore = multiStoreReason(dataFileUris) + if (multiStore.isDefined) { + return multiStore + } + + // Reuse core's generic native-scan gates (ignoreCorruptFiles/ignoreMissingFiles, + // AQE DPP on Spark 3.4, exec enabled). This 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 at + * planning time and normalized to absolute on-disk paths via `copyWithAbsolutePath` (a no-op + * for inline and already-absolute descriptors), so callers never need `tableRoot` again to + * resolve a UUID-relative sidecar. Returns `Seq.empty` for the plain shape + * ([[CometDeltaNativeScan.isDvShape]] false on the wrapped scan): only DV reads carry the + * row-index-filter metadata this deserializes. + * + * Shared plumbing: finding 8's cross-authority object-store option merge + * ([[CometDeltaNativeScan.convert]]) and finding 3's DV cardinality decline gate both need + * every selected file's descriptor; this is the one planning-time deserialization pass for both + * consumers. + */ + 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 + } + + /** + * Returns a decline reason when `uris` span more than one object-store authority (scheme plus + * the URI's raw authority component -- userinfo, host, and port together -- all lowercased so + * e.g. `S3A://Bucket:1234` and `s3a://bucket:1234` collapse to the same authority), or `None` + * when every URI shares a single authority. `file://` paths never carry an authority, so purely + * local scans across any number of distinct directories are unaffected. Factored out of + * [[declineReason]] so it is directly unit-testable without a Spark session. + */ + 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 URI's raw + * `getAuthority` rather than the individually-parsed host/port/userinfo fields. Two pitfalls + * that motivate this: + * - `getAuthority` already includes userinfo (e.g. the container in + * `abfss://[email protected]`), so two containers on the same storage + * account no longer collapse into one authority the way `getHost` alone would. + * - `getHost` (and `getUserInfo`/`getPort`) return `null` for the *entire* 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 the same empty-host + * key. `getAuthority` returns the raw authority text regardless of RFC conformance, so it + * stays accurate for exactly the URIs where the structured getters fail. + * + * A `null` authority (schemes with no authority component, e.g. `file:///tmp/x`) 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" + } + + /** + * True when `dataType` is, or structurally contains (through array elements or map keys/ + * values), a [[StructType]]. Array and map are structural container types whose own + * "element"/"key"/"value" labels are never column-mapped; only the [[StructType]] fields + * reachable through them carry Delta's physical, column-mapped names. + */ + private def containsNestedStruct(dataType: DataType): Boolean = dataType match { + case _: StructType => true + case ArrayType(elementType, _) => containsNestedStruct(elementType) + case MapType(keyType, valueType, _) => + containsNestedStruct(keyType) || containsNestedStruct(valueType) + case _ => false + } + + /** + * True when the scan's row-index column value is provably dead above the scan. The standard DV + * plan shape routes it only into a `named_struct(... row_index ...) AS _metadata` projection + * whose result the final projection discards; anything else (a query actually selecting + * `_metadata.row_index`) makes the value live and must decline. Conservative: any unrecognized + * consumption pattern returns false. + */ + private def rowIndexUnusedAbove(plan: SparkPlan, scanExec: FileSourceScanExec): Boolean = { + val rowIndexAttrs = scanExec.output + .filter(_.name == CometDeltaNativeScan.RowIndexColumn) + .map(_.exprId) + .toSet + if (rowIndexAttrs.isEmpty) { + return true + } + // Transitive taint analysis: everything derived (via Project aliases) from the + // row-index attribute within the VISIBLE plan. The plan handed to this rule may be + // an AQE stage fragment, so anything tainted that reaches the fragment's own output + // escapes to invisible consumers and must decline. Non-Project consumption of any + // tainted attribute (a Filter, Aggregate, Join key, ...) declines outright. + var tainted = rowIndexAttrs + var changed = true + while (changed) { + changed = false + plan.foreach { + case p: ProjectExec => + p.projectList.foreach { + case a: Alias + if !tainted.contains(a.exprId) && + a.references.exists(r => tainted.contains(r.exprId)) => + tainted += a.exprId + changed = true + case _ => + } + case _ => + } + } + val nonProjectConsumer = plan.exists { + case _: ProjectExec => false + case n if n ne scanExec => + n.expressions.exists(_.references.exists(r => tainted.contains(r.exprId))) Review Comment: **[P2] Track live row-index values across Union output remapping** Could we propagate row-index dependencies through `UnionExec`'s positional outputs, or conservatively decline this shape? The current analysis only propagates through `ProjectExec` aliases. `UnionExec.expressions` is empty and its output uses the first child's expression IDs, so a live row-index alias in the second branch disappears from both checks here. With native Delta scans enabled and AQE disabled, this valid query over DV-backed tables reaches the gap: ```sql SELECT id, _metadata.row_index AS ri FROM delta.`t1` UNION ALL SELECT id, _metadata.row_index AS ri FROM delta.`t2` ``` I ran the unchanged `rowIndexUnusedAbove` method on actual Spark 4.0.2 / Delta 4.0.0 plans. Each table had one file containing IDs 0–4, with ID 2 deleted from the first and ID 3 from the second. The first scan correctly returns `false`, but the second returns `true`; its other admission inputs pass. The DV serializer then supplies `0L` for that supposedly unused row index. Modeling exactly that substitution changes the second branch's row indexes to zero and `SUM(ri)` over the union from 15 to 8. This is an exact-method/physical-plan probe, not a full Comet/JNI execution. Please add second-branch selection and aggregate-over-union regression coverage. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
