dwsmith1983 commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3831229371
########## 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: Took the positional-propagation option, since a blanket decline would reject every DV-backed UNION ALL (Delta appends the row-index column to requiredSchema on all DV reads). The taint fixed point now maps child.output(i) to union.output(i) for every branch, covering UnionExec and CometUnionExec; a CometUnionExec whose frozen output ever diverges in arity from its re-parented children declines outright rather than zipping silently. There's also a generic safety net now: any node with two or more children that isn't a positional union, where a tainted child attribute neither appears in the output by exprId nor in the node's expressions, declines, so this gap class can't silently recur (joins pass since they carry child exprIds; semi/anti joins trip only on eliminated-side taint, where declining is correct). Your exact query plus second-branch selection and aggregate-over-union are regression tests asserting values, SUM(ri) is 15, and an anti-regression proves DV unions without _met adata still claim both branches natively. One residual documented in the code: ReusedExchangeExec has the same positional shape but is unreachable at claim time because ReuseExchangeAndSubquery runs after the columnar rules. -- 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]
