sunchao commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3831646182
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,613 @@ +/* + * 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 + // Descriptor deserialization is the expensive part of both consumers below (the DV + // cardinality gate and the store-identity collision gate); hoisted once here so it runs at + // most once per claim attempt regardless of how many gates end up needing it. `lazy` because + // most scans are not DV-shaped and selectedDvDescriptors short-circuits to Seq.empty for + // them, but paying even that check is unnecessary work for gates that return earlier. + val tableRoot = scanExec.relation.location.rootPaths.head.toString + lazy val dvDescriptors: Seq[DeletionVectorDescriptor] = + selectedDvDescriptors(scanHelper, tableRoot) + + 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 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 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 Review Comment: **[P2] Check selected-file schemes before claiming a shallow clone** Could we apply this filesystem gate to the selected data-file URIs, not just the table's `rootPaths`? A valid Delta shallow clone can have a supported `file:` table root while its data files still reference `viewfs://review-mount/source/table/...`. With the default libhdfs scheme set (`hdfs` only), both authority checks accept these same-authority files, and the ordinary `LongType` scan serializes successfully, so the contrib claims it. Native store preparation then fails with `Generic URL error: Unable to recognise URL "viewfs://..."` instead of leaving the scan with Spark. At `bc98657f`, a local-only Spark 4.0.2 / Delta 4.0.0 probe wrote a Delta table through Hadoop's built-in viewfs mount, shallow-cloned it to a local directory, and successfully read `[0, 1, 2]`. Its actual scan had a `file:` root and `viewfs:` selected files. The exact-current authority helpers accepted those files, while the exact native store-preparation helper rejected their URI. This was a stock-engine/exact-helper probe, not a full Comet/JNI run. Checking the schemes of the files actually selected before claiming would preserve Spark fallback for this valid table. -- 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]
