sunchao commented on code in PR #5365:
URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3883807851
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -193,11 +688,16 @@ fn remap_physical_schema(
}
// Block accidental name match for ID-bearing logical fields whose
ID is missing
- // from the file. Mirrors Spark's `generateFakeColumnName` in
`matchIdField`.
+ // from the file. Mirrors Spark's `generateFakeColumnName` in
`matchIdField`. Runs
+ // regardless of `case_sensitive` (field-ID matching is
independent of Spark's
+ // casing config), so `case_tables` may be absent here even when
it would be
+ // populated for a genuinely case-insensitive scan --
`names_equal_ignore_case_java`
+ // degrades to `str::to_lowercase` in that case, which is fine for
this defensive,
+ // over-approximating check (see `java_lowercase`).
if should_match_by_id
&& unmatched_id_logical_names
.iter()
- .any(|name| name.eq_ignore_ascii_case(field.name()))
+ .any(|name| names_equal_ignore_case_java(name,
field.name(), case_tables))
Review Comment:
[P2] Respect case sensitivity when shielding unmatched field IDs
For an ordinary case-sensitive V1 Parquet scan with field-ID reading
enabled, request nullable BIGINT fields `Κ` (U+039A, ID 1) and `κ` (U+03BA, no
ID), and read a file containing only `κ` with ID 2 and value 7. Spark
null-fills the missing ID 1 field but reads the ID-less field by its exact
name. This unconditional case-folded check instead matches physical `κ` to
unmatched logical `Κ` and replaces the real column with a fake name; the final
nullable projection yields `(NULL, NULL)` instead of `(NULL, 7)`. File ID 2
satisfies the file-ID presence guard, and the names are distinct with case
sensitivity enabled. Requested BASE's ASCII comparison does not equate this
Greek pair. Please avoid hiding a physical field needed by a legitimate
exact-name match. This regression exists at the previous reviewed HEAD too, so
it is new to the discussion, not newly introduced by the latest increment.
##########
contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala:
##########
@@ -0,0 +1,1304 @@
+/*
+ * 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 scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.catalyst.expressions.{Alias, GenericInternalRow,
InputFileBlockLength, InputFileBlockStart, InputFileName}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import
org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
+import org.apache.spark.sql.comet.CometScanExec
+import org.apache.spark.sql.delta.DeltaParquetFileFormat
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
+import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec,
SparkPlan}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES
+import org.apache.comet.parquet.CometParquetUtils
+import org.apache.comet.rules.{CometScanRule, CometScanTypeChecker}
+import org.apache.comet.serde.operator.CometNativeScan
+import org.apache.comet.shims.ShimFileFormat
+
+/**
+ * Claim/decline gates for the native Delta scan. Correctness rule: when in
doubt, decline,
+ * Spark's Delta reader handles the scan and results stay correct, just
unaccelerated.
+ */
+object DeltaScanSupport {
+
+ /**
+ * Reader features the native path understands; anything else on the
protocol declines the
+ * table. `deletionVectors`/`columnMapping` are declined separately below
for specific reasons.
+ */
+ private val understoodReaderFeatures: Set[String] =
+ Set("columnMapping", "deletionVectors", "timestampNtz", "v2Checkpoint",
"vacuumProtocolCheck")
+
+ /**
+ * Is this exactly Delta's DSv1 parquet format? Compared by class name, not
`classOf`: a
+ * `classOf` reference would raise `NoClassDefFoundError` and break every
parquet scan when
+ * delta-spark is absent from the classpath.
+ */
+ def isDeltaScan(scanExec: FileSourceScanExec): Boolean =
+ scanExec.relation.fileFormat.getClass.getName ==
+ "org.apache.spark.sql.delta.DeltaParquetFileFormat"
+
+ /**
+ * First reason this Delta scan cannot go native, or None when claimable.
Only called when
+ * [[isDeltaScan]] is true. `scanHelper` is the [[CometScanExec]] built to
drive
+ * [[CometDeltaNativeScan.convert]] on a claim, reused 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
+ // Name mode is supported via physical-name schemas; id mode needs the
field-id path and
+ // stays declined until validated. Hoisted here since several gates below
reuse it.
+ val cmMode = metadata.columnMappingMode.name
+ // Descriptor deserialization is expensive, so hoist it into a `lazy val`:
it runs at most
+ // once per claim attempt, and not at all for the common non-DV-shape case.
+ val tableRoot = scanExec.relation.location.rootPaths.head.toString
+ lazy val dvDescriptors: Seq[DeletionVectorDescriptor] =
+ selectedDvDescriptors(scanHelper, tableRoot)
+
+ // Mirrors core's CometScanRule.isSchemaSupported so scan-time type gates
(unsigned-small-int
+ // fallback, collation, shredded-variant-struct) apply identically here.
Pure in-memory check,
+ // so it runs first, ahead of every I/O-bearing gate below.
+ val schemaFallbackReasons = new ListBuffer[String]()
+ val typeChecker = CometScanTypeChecker()
+ val requiredSchemaSupported =
+ typeChecker.isSchemaSupported(scanExec.requiredSchema,
schemaFallbackReasons)
+ val partitionSchemaSupported =
+ typeChecker.isSchemaSupported(scanExec.relation.partitionSchema,
schemaFallbackReasons)
+ if (!requiredSchemaSupported || !partitionSchemaSupported) {
+ return Some(
+ "Native Delta scan does not support the schema: " +
schemaFallbackReasons.mkString(", "))
+ }
+
+ if (format.isCDCRead) {
+ return Some("Native Delta scan does not support Change Data Feed reads")
+ }
+
+ // Delta's DML machinery (findTouchedFiles) disables reader optimizations
and needs real
+ // row indexes from Spark's reader; claiming here would feed NULL indexes
into DV construction.
+ if (!format.optimizationsEnabled) {
+ return Some("Native Delta scan does not support reads with reader
optimizations disabled")
+ }
+ if (scanExec.requiredSchema.exists(_.name ==
DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME) ||
+ scanExec.relation.dataSchema.exists(
+ _.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME)) {
+ return Some("Native Delta scan does not support Delta's generated
row-index column")
+ }
+
+ if (cmMode != "none" && cmMode != "name") {
+ return Some(s"Native Delta scan does not support column mapping mode
$cmMode")
+ }
+ // createPhysicalSchema wholesale-replaces field metadata, silently
dropping EXISTS_DEFAULT.
+ if (cmMode == "name" &&
+ getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) {
+ return Some(
+ "Native Delta scan does not support column defaults together with
column mapping")
+ }
+ // createPhysicalSchema rewrites nested StructField names too, and the
native builder emits the
+ // required schema verbatim as output, so name-sensitive expressions (e.g.
to_json) would leak
+ // physical names. Decline until a rename adapter exists.
+ if (cmMode == "name" &&
+ scanExec.requiredSchema.exists(f => containsNestedStruct(f.dataType))) {
+ return Some("Native Delta scan does not support column mapping with
nested struct fields")
+ }
+
+ val readerFeatures = protocol.readerFeatureNames
+ val unknownFeatures = readerFeatures -- understoodReaderFeatures
+ if (unknownFeatures.nonEmpty) {
+ return Some(
+ s"Native Delta scan does not support reader feature(s)
${unknownFeatures.mkString(", ")}")
+ }
+
+ // Non-constant metadata columns are generated per-row by Spark's reader
and unsupported,
+ // except Delta's DV bookkeeping columns, which the native path emits as
constants.
+ val knownColNames =
+ scanExec.relation.dataSchema.map(_.name).toSet ++
+ scanExec.relation.partitionSchema.map(_.name).toSet ++
+ scanExec.fileConstantMetadataColumns.map(_.name).toSet ++
+ CometDeltaNativeScan.internalColumnNames
+ val unknownOutput =
scanExec.output.map(_.name).filterNot(knownColNames.contains)
+ if (unknownOutput.nonEmpty) {
+ return Some(
+ s"Native Delta scan does not support generated column(s)
${unknownOutput.mkString(", ")}")
+ }
+
+ // Deletion-vector shape invariants (see
CometDeltaNativeScan.buildDvScanCommon).
+ if (CometDeltaNativeScan.isDvShape(scanExec)) {
+ // A row-index column WITHOUT is_row_deleted is Delta DML bookkeeping
(real row indexes),
+ // not a DV read; claiming it with a constant would corrupt the DVs
being written.
+ val hasIsRowDeleted =
+ scanExec.requiredSchema.exists(_.name ==
CometDeltaNativeScan.IsRowDeletedColumn)
+ val hasRowIndex =
+ scanExec.requiredSchema.exists(_.name ==
CometDeltaNativeScan.RowIndexColumn)
+ if (hasRowIndex && !hasIsRowDeleted) {
+ return Some(
+ "Native Delta scan does not support row-index reads outside a
deletion-vector scan")
+ }
+ // Internal columns must form a suffix of the read schema so data-column
positions agree
+ // between Spark's output and the stripped native schema.
+ val names = scanExec.requiredSchema.fields.map(_.name)
+ val firstInternal =
names.indexWhere(CometDeltaNativeScan.internalColumnNames.contains)
+ if
(!names.drop(firstInternal).forall(CometDeltaNativeScan.internalColumnNames.contains))
{
+ return Some("Native Delta scan requires DV bookkeeping columns to
trail the read schema")
+ }
+ // Native applies the DV itself and emits a dead constant for row-index,
so the real value
+ // must be provably unused above the scan.
+ if (!rowIndexUnusedAbove(plan, scanExec)) {
+ return Some(
+ "Native Delta scan cannot supply _metadata.row_index values consumed
by the query")
+ }
+ // The DV common builder does not serialize existence defaults yet.
+ if (getExistenceDefaultValues(scanExec.requiredSchema).exists(_ !=
null)) {
+ return Some(
+ "Native Delta scan does not support column defaults together with
deletion vectors")
+ }
+ // Bounds native's memory for expanded DV row selectors (delta_dv.rs),
pessimistically
+ // bounded by 2*cardinality + #row-groups; the conf below makes an
over-pessimistic decline
+ // recoverable.
+ val maxDeletedRowsPerFile =
DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.get()
+ val oversizedCardinalities = dvDescriptors
+ .map(_.cardinality)
+ .filter(_ > maxDeletedRowsPerFile)
+ if (oversizedCardinalities.nonEmpty) {
+ return Some(
+ "Native Delta scan does not support a deletion vector deleting " +
+ s"${oversizedCardinalities.max} rows in a single file, exceeding "
+
+
s"${DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.key}=$maxDeletedRowsPerFile")
+ }
+ }
+
+ // input_file_name & friends read from a thread-local Spark's FileScanRDD
sets; the native scan
+ // 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, but the
native builder resolves
+ // ObjectStoreUrl from only the FIRST selected file; force file listing
and decline rather than
+ // risk reading a later file through the wrong handle.
+ 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
+ }
+
+ // GCS's zero-I/O, conf-only credential-forwarding gate; ordered alongside
the S3 credential
+ // gates below since all presume a single, well-formed store identity per
URI.
+ val gcsAuthReason = gcsHadoopOnlyAuthReason(hadoopConf, dataFileUris ++
dvUris)
+ if (gcsAuthReason.isDefined) {
+ return gcsAuthReason
+ }
+
+ // Zero-I/O, conf-only, like the GCS gate above: decline any bucket
configured for SSE-C
+ // before the credential-divergence gates below, which do not otherwise
notice this table is
+ // readable through Hadoop only because Hadoop's request factory attaches
the customer key
+ // native never learns about.
+ val sseCReason = sseCustomerKeyReason(hadoopConf, dataFileUris ++ dvUris)
+ if (sseCReason.isDefined) {
+ return sseCReason
+ }
+
+ // Every fs.s3a.* option native's get_config (s3.rs) resolves must agree
between what Hadoop
+ // itself would use and what native would read from the forwarded,
substituted conf (covers
+ // long-form bucket credentials, JCEKS/credential-provider shadowing, and
any other
+ // short-vs-effective divergence in one mechanism); reuses hadoopConf from
the encryption gate
+ // above.
+ val s3Reason = s3ConfigDivergenceReason(hadoopConf, dataFileUris ++ dvUris)
+ if (s3Reason.isDefined) {
+ return s3Reason
+ }
+
+ // 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,
normalized to
+ * absolute on-disk paths. Returns `Seq.empty` for the plain shape. 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.
+ */
+ 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, since such a URI cannot come from a Hadoop-backed
source.
+ */
+ 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).
+ */
+ 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 + lowercased raw
+ * authority, so e.g. `S3A://Bucket` and `s3a://bucket` 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 parsed host/port/userinfo fields: `getHost` (and
`getUserInfo`/`getPort`)
+ * return `null` for the whole authority when it fails RFC 3986 `reg-name`
syntax (e.g. an
+ * underscore in a GCS bucket name, `gs://my_bucket`), which would silently
collapse 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`, which (like [[uriAuthority]]'s
getters) returns `null`
+ * for the whole authority on an RFC 3986 `reg-name` violation. 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`, then `://`, then a literal `***`
masking userinfo,
+ * then `@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 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.
+ */
+ 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 below. `hadoop-aws` is NOT on
this module's runtime
+ * classpath, so `org.apache.hadoop.fs.s3a.Constants` must never be
referenced here (would raise
+ * `NoClassDefFoundError` for sessions with no S3 dependency).
+ */
+ 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; both
+ * must be covered here too.
+ */
+ private def s3aBucketLongProviderPathKey(bucket: String): String =
+ s"fs.s3a.bucket.$bucket.fs.s3a.security.credential.provider.path"
+
+ 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,
or `None` when
+ * `uri`'s scheme is not `s3`/`s3a`. Parses the raw authority manually
rather than
+ * `URI#getHost`, avoiding the same RFC 3986 `reg-name` pitfall as
[[uriAuthority]].
+ */
+ 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
+ }
+ }
+
+ private def plainValue(hadoopConf: Configuration, key: String):
Option[String] =
+ Option(hadoopConf.get(key)).filter(_.nonEmpty)
+
+ /**
+ * The three per-bucket base keys where Hadoop's own effective value CAN
differ from
+ * short-then-global: `S3AUtils#lookupPassword` additionally honors the long
bucket form
+ * (`fs.s3a.bucket.B.fs.s3a.<key>`) and Hadoop credential providers (JCEKS
et al.) for exactly
+ * these three, and only these three -- every other `fs.s3a` option instead
flows through
+ * `S3AUtils#propagateBucketOptions`, which folds the long form into an
unread key
+ * (`fs.s3a.fs.s3a.<key>`) and leaves Hadoop's own effective value at
short-then-global already,
+ * identical to native's.
+ */
+ private val CredentialValueKeys: Seq[String] =
+ Seq("fs.s3a.access.key", "fs.s3a.secret.key", "fs.s3a.session.token")
+
+ /**
+ * Every OTHER per-bucket `fs.s3a.*` base key native's S3 client's
`get_config` (s3.rs)
+ * resolves, verified directly against its call sites:
`extract_s3_config_options`
+ * (endpoint.region, path.style.access, endpoint, requester.pays.enabled),
+ * `lookup_provider_class` (the Comet-specific credential-provider-class
activation key), and
+ * `build_credential_provider`/`build_aws_credential_provider_metadata`/
+ * `build_assume_role_credential_provider_metadata`
(aws.credentials.provider,
+ * assumed.role.credentials.provider, assumed.role.arn,
assumed.role.session.name). Every one of
+ * these resolves short-bucket-then-global inside `get_config` -- s3.rs
never reads the long
+ * bucket form. Literal strings, not the [[AwsCredentialsProviderKey]] /
+ * [[AssumedRoleCredentialsProviderKey]] vals declared below, purely to
avoid a forward
+ * reference inside this `object` body; kept textually identical to those
two constants.
+ */
+ private val OtherS3ConfigKeys: Seq[String] = Seq(
+ "fs.s3a.aws.credentials.provider",
+ "fs.s3a.assumed.role.arn",
+ "fs.s3a.assumed.role.session.name",
+ "fs.s3a.assumed.role.credentials.provider",
+ "fs.s3a.endpoint",
+ "fs.s3a.endpoint.region",
+ "fs.s3a.path.style.access",
+ "fs.s3a.requester.pays.enabled",
+ "fs.s3a.comet.credential.provider.class")
+
+ private val AllS3ConfigKeys: Seq[String] = CredentialValueKeys ++
OtherS3ConfigKeys
+
+ /**
+ * The short-bucket-then-global value resolved for `baseKey` under `bucket`
from `hadoopConf`,
+ * skipping an empty value at either alias exactly like [[plainValue]]. Used
on a
+ * [[propagateBucketOptions]] result, this is (for the non-credential keys in
+ * [[OtherS3ConfigKeys]]) Hadoop's own effective value, since real Hadoop
reads those options as
+ * plain `Configuration#get(baseKey)` calls against its propagated conf,
which already folded
+ * any per-bucket override into `baseKey` itself -- so re-checking the
(untouched) short bucket
+ * key here first is a no-op, not a second resolution mechanism, and gives
the identical value
+ * `Configuration#get(baseKey)` alone would. Also used by
[[shortThenGlobalOrReason]] to read
+ * provider-CLASS strings from the original conf for name-support
validation; the earlier
+ * [[s3ConfigDivergenceReason]] gate (which uses [[nativeShortThenGlobal]],
not this function,
+ * to model native's actual resolution) already runs ahead of that check and
would catch any
+ * divergence this skip-empty read models imprecisely. NEVER used to compute
native's own
+ * effective value any more -- see [[nativeShortThenGlobal]] for that.
+ */
+ private def shortThenGlobal(
+ 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))
+ }
+
+ /**
+ * The short-bucket-then-global value native's `get_config` (s3.rs) resolves
for `baseKey` under
+ * `bucket` from the ORIGINAL, unpropagated `hadoopConf` -- `NativeConfig
+ * .extractObjectStoreOptions` forwards `Configuration#get`'s substituted
value for every
+ * `fs.s3a.*` entry with no bucket-option propagation step of its own, so
the original conf is
+ * the right input here. Unlike [[shortThenGlobal]]/[[plainValue]], this
mirrors `get_config`
+ * faithfully: PRESENCE of the short-bucket key alone -- never its emptiness
-- decides whether
+ * native falls back to the global key (`get_config` is a plain
`HashMap::get`, which returns
+ * `Some` for a key explicitly set to `""`), so an explicitly empty or
whitespace-only
+ * short-bucket value resolves to `Some("")` here and never falls through to
global -- the
+ * OPPOSITE of Hadoop's own `getPassword`/`lookupPassword` semantics (see
+ * [[hadoopCredentialEffective]]), which treat empty as absent and keep
trying the next alias.
+ * The ONLY function used to compute native's effective value in
[[s3KeyDivergenceReason]].
+ *
+ * Deliberately does NOT apply `get_config_trimmed`'s `.trim()` here:
[[s3KeyDivergenceReason]]
+ * trims both this value and Hadoop's effective value together,
symmetrically, at the point they
+ * are compared, rather than one-sidedly here -- trimming only the native
side would flag a
+ * spurious divergence for a value neither side's whitespace actually
changes the behavior of
+ * once each side's own downstream parsing normalizes it (e.g. Hadoop's own
multi-line
+ * `fs.s3a.aws.credentials.provider` default, which both Hadoop and native
additionally trim per
+ * comma-separated entry after splitting), while a one-sided trim would make
an
+ * otherwise-identical default value look diverged for every bucket, never
claiming natively at
+ * all.
+ */
+ private def nativeShortThenGlobal(
+ hadoopConf: Configuration,
+ bucket: String,
+ baseKey: String): Option[String] = {
+ val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.")
+ Option(hadoopConf.get(shortKey)).orElse(Option(hadoopConf.get(baseKey)))
+ }
+
+ /**
+ * Faithful in-memory replica of `S3AUtils#propagateBucketOptions`
(`hadoop-aws`), which
+ * `S3AFileSystem#initialize` calls FIRST, before any option or credential
is read:
+ * `Configuration conf = propagateBucketOptions(originalConf, bucket); ...;
setConf(conf);` --
+ * every subsequent `conf.get`/`getPassword` call in that filesystem
instance, including
+ * `${...}` variable substitution, resolves against this propagated view,
not the original conf.
+ * `hadoop-aws` is not on this module's runtime classpath (see the
string-literal-keys note
+ * above), so `S3AUtils#propagateBucketOptions` cannot be called directly;
this reproduces its
+ * logic verbatim using only `hadoop-common`'s `Configuration`:
+ * {{{
+ * public static Configuration propagateBucketOptions(Configuration source,
String bucket) {
+ * final String bucketPrefix = FS_S3A_BUCKET_PREFIX + bucket + '.';
+ * final Configuration dest = new Configuration(source);
+ * for (Map.Entry<String, String> entry : source) {
+ * final String key = entry.getKey();
+ * final String value = entry.getValue(); // the (unexpanded) value
+ * if (!key.startsWith(bucketPrefix) || bucketPrefix.equals(key))
continue;
+ * final String stripped = key.substring(bucketPrefix.length());
+ * if (stripped.startsWith("bucket.") || "impl".equals(stripped)) {
+ * // ignored
+ * } else {
+ * final String generic = FS_S3A_PREFIX + stripped;
+ * dest.set(generic, value, ...); // overwrites any existing global
value
+ * }
+ * }
+ * return dest;
+ * }
+ * }}}
+ * Note the LONG bucket form (`fs.s3a.bucket.B.fs.s3a.<key>`) folds to an
unread
+ * `fs.s3a.fs.s3a.<key>` key here too, exactly like the real method --
`stripped` already starts
+ * with `fs.s3a.` in that case, so prepending `fs.s3a.` again produces a key
nothing ever reads.
+ */
+ private def propagateBucketOptions(hadoopConf: Configuration, bucket:
String): Configuration = {
+ val bucketPrefix = s"fs.s3a.bucket.$bucket."
+ val dest = new Configuration(hadoopConf)
+ hadoopConf.iterator().asScala.foreach { entry =>
+ val key = entry.getKey
+ if (key.startsWith(bucketPrefix) && key != bucketPrefix) {
+ val stripped = key.substring(bucketPrefix.length)
+ if (!stripped.startsWith("bucket.") && stripped != "impl") {
+ dest.set(s"fs.s3a.$stripped", entry.getValue)
+ }
+ }
+ }
+ dest
+ }
+
+ /**
+ * Canonical and deprecated Hadoop S3A encryption-algorithm config keys,
verified via `javap`
+ * against `hadoop-aws` 3.3.4's `org.apache.hadoop.fs.s3a.Constants`:
`S3_ENCRYPTION_ALGORITHM =
+ * "fs.s3a.encryption.algorithm"` (canonical) and
`SERVER_SIDE_ENCRYPTION_ALGORITHM =
+ * "fs.s3a.server-side-encryption-algorithm"` (DEPRECATED -- note the hyphen
before "algorithm",
+ * unlike the corresponding `*.key` constants below, which both use a `.key`
suffix).
+ * `hadoop-aws` is NOT on this module's runtime classpath, so these stay
string literals, same
+ * rationale as [[HadoopCredentialProviderPathKey]].
+ */
+ private val S3EncryptionAlgorithmKey = "fs.s3a.encryption.algorithm"
+ private val DeprecatedS3EncryptionAlgorithmKey =
"fs.s3a.server-side-encryption-algorithm"
+
+ /**
+ * The literal string `S3AEncryptionMethods.SSE_C.getMethod()` returns,
verified via `javap`
+ * against `hadoop-aws` 3.3.4: `S3AEncryptionMethods#getMethod(String)`
parses the configured
+ * algorithm with `values().find(_.getMethod.equalsIgnoreCase(algorithm))`,
so this gate's
+ * comparison is case-insensitive too, matching Hadoop's own parsing rather
than being stricter
+ * or looser than it.
+ */
+ private val SseCustomerKeyAlgorithm = "SSE-C"
+
+ /**
+ * `bucket`'s effective encryption-algorithm key and value under
`hadoopConf`, or `None` when
+ * neither the canonical nor deprecated key is set anywhere consulted.
Mirrors
+ * `S3AUtils#buildEncryptionSecrets`'s real resolution order (verified via
`javap`): it first
+ * tries `lookupBucketSecret`, which checks ONLY the bucket-scoped SHORT
form of each key
+ * (canonical before deprecated), then falls back to a BUCKET-AGNOSTIC
global `lookupPassword`
+ * call for each key (canonical before deprecated) -- confirmed from the
bytecode, which passes
+ * a literal `null` bucket argument at that second tier. Unlike
[[s3KeyDivergenceReason]]'s
+ * [[CredentialValueKeys]] handling, the LONG bucket form
+ * (`fs.s3a.bucket.B.fs.s3a.encryption.algorithm`) is genuinely never
consulted for this key:
+ * running it through [[propagateBucketOptions]] would only fold it into the
unread
+ * `fs.s3a.fs.s3a.encryption.algorithm` key documented there, matching
Hadoop's own behavior, so
+ * the global tier here reads `hadoopConf` directly rather than a propagated
replica.
+ *
+ * The canonical-vs-deprecated distinction below is frequently moot in
practice: `hadoop-aws`'s
+ * `S3AFileSystem.addDeprecatedKeys()` statically registers
`fs.s3a.server-side-encryption-*` as
+ * `Configuration`-level deprecated aliases of `fs.s3a.encryption.*`
(verified via `javap`), a
+ * registration that lives in a static field on Hadoop's `Configuration`
class -- process-wide
+ * once `S3AFileSystem`'s class has loaded anywhere in the JVM, which a real
scan has always
+ * already done by the time this gate runs, since reading the S3 table at
all requires loading
+ * that class. Once active, `Configuration#get` resolves either literal key
to the identical
+ * value transparently, making the two `plainValue`/`bucketShort` reads
below redundant (but
+ * harmless) for that case; they remain the operative path only when nothing
else in the process
+ * has loaded `S3AFileSystem` yet.
+ *
+ * Deliberately does NOT walk the Hadoop-credential-provider (JCEKS) path
+ * `buildEncryptionSecrets` falls back to when neither key is set in plain
conf: unlike the
+ * actual key MATERIAL
(`fs.s3a.encryption.key`/`fs.s3a.server-side-encryption.key`, never read
+ * here at all -- this gate only needs the ALGORITHM to decide), the
algorithm NAME is not
+ * credential-sensitive data, so storing it in a keystore is not a realistic
Hadoop deployment
+ * pattern. Documented, narrow limitation: a configuration that stores the
algorithm name ONLY
+ * in a credential provider would under-decline here.
+ */
+ private def effectiveEncryptionAlgorithm(
+ hadoopConf: Configuration,
+ bucket: String): Option[(String, String)] = {
+ def bucketShort(baseKey: String): Option[(String, String)] =
+ plainValue(hadoopConf, s"fs.s3a.bucket.$bucket." +
baseKey.stripPrefix("fs.s3a."))
+ .map(baseKey -> _)
+
+ bucketShort(S3EncryptionAlgorithmKey)
+ .orElse(bucketShort(DeprecatedS3EncryptionAlgorithmKey))
+ .orElse(plainValue(hadoopConf,
S3EncryptionAlgorithmKey).map(S3EncryptionAlgorithmKey -> _))
+ .orElse(plainValue(hadoopConf, DeprecatedS3EncryptionAlgorithmKey)
+ .map(DeprecatedS3EncryptionAlgorithmKey -> _))
+ }
+
+ private def sseCustomerKeyDeclineReason(bucket: String, algorithmKey:
String): String =
+ s"Native Delta scan does not support
$algorithmKey=$SseCustomerKeyAlgorithm for $bucket " +
+ "(SSE-C requires the customer-provided encryption key on every S3
GET/HEAD request; " +
+ "the native S3 client's extract_s3_config_options forwards none of
Hadoop's " +
+ "fs.s3a.encryption.*/fs.s3a.server-side-encryption* options, so requests
would go out " +
+ "without the required x-amz-server-side-encryption-customer-key* headers
and fail, where " +
+ "Hadoop's own reader -- whose request factory attaches the key --
succeeds)"
+
+ /**
+ * First reason any bucket among `uris` is configured for SSE-C (Server-Side
Encryption with
+ * Customer-Provided Keys), or `None` when claimable. Scoped to SSE-C only:
SSE-S3, SSE-KMS, and
+ * DSSE-KMS are all decrypted transparently by S3 on GET/HEAD once the
caller has read/decrypt
+ * permission -- no request header is required, so native's silent failure
to forward those
+ * options costs nothing on the read path this gate protects. SSE-C is
different in kind, not
+ * degree: S3 rejects a GET/HEAD for an SSE-C object outright (400 Bad
Request) unless the
+ * customer key is resent as a request header on every call, so a native
scan that never learns
+ * the key cannot succeed at all, where Hadoop's own reader (which does
attach it) would.
+ * Client-side encryption (CSE-KMS/CSE-CUSTOM) is a different mechanism
entirely -- it
+ * encrypts/decrypts object bytes locally rather than gating the S3 request
itself -- and is out
+ * of scope for this HTTP-header gate. Never interpolates a resolved key
value, only key names,
+ * the bucket, and the (non-secret) algorithm name.
+ */
+ private[delta] def sseCustomerKeyReason(
+ 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 {
+ effectiveEncryptionAlgorithm(hadoopConf, bucket).collect {
+ case (key, value) if value.equalsIgnoreCase(SseCustomerKeyAlgorithm)
=>
Review Comment:
[P2] Decline Hadoop CSE-KMS before claiming the native scan
This gate admits `fs.s3a.encryption.algorithm=CSE-KMS`, but the native S3
client has no client-side decryption layer. On the declared Spark 3.5.9/Hadoop
3.3.4 route, Hadoop's default S3 client factory selects an encryption client
with KMS materials and range-read decryption. An existing, uniformly
CSE-KMS-encrypted Delta table with valid credentials/key permissions, no
S3Guard, and ordinary primitive NoMapping/no-DV data is therefore readable
through Spark; enabling this contribution sends ciphertext to native Parquet
and fails the scan. SSE-KMS is transparent server-side encryption and is not
this case. Please retain Spark fallback for CSE-KMS until native decryption is
supported. The clean requested BASE has no bundled scan provider and leaves
this Delta scan with Spark; this finding predates the latest increment.
--
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]
