dwsmith1983 commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3861298389
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,981 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.contrib.delta + +import java.io.IOException +import java.net.URI +import java.util.Locale + +import scala.collection.mutable.ListBuffer + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.Path +import org.apache.spark.sql.catalyst.expressions.{Alias, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName} +import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData} +import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues +import org.apache.spark.sql.comet.CometScanExec +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec, SparkPlan} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType} + +import org.apache.comet.CometConf +import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES +import org.apache.comet.parquet.CometParquetUtils +import org.apache.comet.rules.{CometScanRule, CometScanTypeChecker} +import org.apache.comet.serde.operator.CometNativeScan +import org.apache.comet.shims.ShimFileFormat + +/** + * Claim/decline gates for the native Delta scan. Correctness rule: when in doubt, decline, + * Spark's Delta reader handles the scan and results stay correct, just unaccelerated. + */ +object DeltaScanSupport { + + /** + * Reader features the native path understands. Anything else on the protocol declines the + * table. Note `deletionVectors` and `columnMapping` are declined separately (below) so their + * fallback reasons are specific. + */ + private val understoodReaderFeatures: Set[String] = + Set("columnMapping", "deletionVectors", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck") + + /** + * Is this exactly Delta's DSv1 parquet format? Compared by class name, not `classOf`: this is + * the first gate on every V1 scan and must stay inert when delta-spark is absent from the + * classpath, where `classOf[DeltaParquetFileFormat]` would raise NoClassDefFoundError inside + * CometScanRule and take down every parquet scan in the session. A name match proves + * delta-spark is present, so Delta types used past this gate are safe. + */ + def isDeltaScan(scanExec: FileSourceScanExec): Boolean = + scanExec.relation.fileFormat.getClass.getName == + "org.apache.spark.sql.delta.DeltaParquetFileFormat" + + /** + * Returns the first reason this Delta scan cannot go native, or None when claimable. Only + * called when [[isDeltaScan]] is true. `scanHelper` is the same [[CometScanExec]] the caller + * builds to drive [[CometDeltaNativeScan.convert]] on a claim, reused here to resolve the + * scan's selected files for the multi-store gate below. + */ + def declineReason( + plan: SparkPlan, + scanExec: FileSourceScanExec, + scanHelper: CometScanExec): Option[String] = { + val format = scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + val protocol = format.protocol + val metadata = format.metadata + // Descriptor deserialization is expensive; hoisted once so it runs at most once per claim + // attempt. `lazy` since most scans are not DV-shaped and gates that return earlier should not + // pay for it. + val tableRoot = scanExec.relation.location.rootPaths.head.toString + lazy val dvDescriptors: Seq[DeletionVectorDescriptor] = + selectedDvDescriptors(scanHelper, tableRoot) + + // Mirrors core's CometScanRule.isSchemaSupported (the same type checker core's own + // FileSourceScanExec path runs) so scan-time type gates -- notably the default-on + // COMET_PARQUET_UNSIGNED_SMALL_INT_CHECK safety fallback for ShortType, plus the + // collation and shredded-variant-struct gates -- apply identically whether the scan is + // claimed through core or through this contrib. A pure in-memory schema check with no + // file or store I/O, so it runs first, ahead of every other gate below. + val schemaFallbackReasons = new ListBuffer[String]() + val typeChecker = CometScanTypeChecker() + val requiredSchemaSupported = + typeChecker.isSchemaSupported(scanExec.requiredSchema, schemaFallbackReasons) + val partitionSchemaSupported = + typeChecker.isSchemaSupported(scanExec.relation.partitionSchema, schemaFallbackReasons) + if (!requiredSchemaSupported || !partitionSchemaSupported) { + return Some( + "Native Delta scan does not support the schema: " + schemaFallbackReasons.mkString(", ")) + } + + if (format.isCDCRead) { + return Some("Native Delta scan does not support Change Data Feed reads") + } + + // Delta's DML machinery (findTouchedFiles) disables reader optimizations and needs real + // row indexes from Spark's reader; claiming here would feed NULL indexes into DV construction. + if (!format.optimizationsEnabled) { + return Some("Native Delta scan does not support reads with reader optimizations disabled") + } + if (scanExec.requiredSchema.exists(_.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME) || + scanExec.relation.dataSchema.exists( + _.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME)) { + return Some("Native Delta scan does not support Delta's generated row-index column") + } + + // Name mode is supported via physical-name schemas; id mode needs the field-id path and + // stays declined until validated. + val cmMode = metadata.columnMappingMode.name + if (cmMode != "none" && cmMode != "name") { + return Some(s"Native Delta scan does not support column mapping mode $cmMode") + } + // createPhysicalSchema wholesale-replaces field metadata, silently dropping EXISTS_DEFAULT. + if (cmMode == "name" && + getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with column mapping") + } + // createPhysicalSchema rewrites nested StructField names too, and the native builder uses the + // required schema verbatim as output: name-sensitive expressions (e.g. to_json) would leak + // physical names into results. Needs a rename adapter for the logical schema; decline until + // then. + if (cmMode == "name" && + scanExec.requiredSchema.exists(f => containsNestedStruct(f.dataType))) { + return Some("Native Delta scan does not support column mapping with nested struct fields") + } + + val readerFeatures = protocol.readerFeatureNames + val unknownFeatures = readerFeatures -- understoodReaderFeatures + if (unknownFeatures.nonEmpty) { + return Some( + s"Native Delta scan does not support reader feature(s) ${unknownFeatures.mkString(", ")}") + } + + // Non-constant metadata columns are generated per-row by Spark's reader and unsupported, + // except Delta's DV bookkeeping columns, which the native path emits as constants. + val knownColNames = + scanExec.relation.dataSchema.map(_.name).toSet ++ + scanExec.relation.partitionSchema.map(_.name).toSet ++ + scanExec.fileConstantMetadataColumns.map(_.name).toSet ++ + CometDeltaNativeScan.internalColumnNames + val unknownOutput = scanExec.output.map(_.name).filterNot(knownColNames.contains) + if (unknownOutput.nonEmpty) { + return Some( + s"Native Delta scan does not support generated column(s) ${unknownOutput.mkString(", ")}") + } + + // Deletion-vector shape invariants (see CometDeltaNativeScan.buildDvScanCommon). + if (CometDeltaNativeScan.isDvShape(scanExec)) { + // A row-index column WITHOUT is_row_deleted is Delta DML bookkeeping (real row indexes), + // not a DV read; claiming it with a constant would corrupt the DVs being written. + val hasIsRowDeleted = + scanExec.requiredSchema.exists(_.name == CometDeltaNativeScan.IsRowDeletedColumn) + val hasRowIndex = + scanExec.requiredSchema.exists(_.name == CometDeltaNativeScan.RowIndexColumn) + if (hasRowIndex && !hasIsRowDeleted) { + return Some( + "Native Delta scan does not support row-index reads outside a deletion-vector scan") + } + // Internal columns must form a suffix of the read schema so data-column positions agree + // between Spark's output and the stripped native schema. + val names = scanExec.requiredSchema.fields.map(_.name) + val firstInternal = names.indexWhere(CometDeltaNativeScan.internalColumnNames.contains) + if (!names.drop(firstInternal).forall(CometDeltaNativeScan.internalColumnNames.contains)) { + return Some("Native Delta scan requires DV bookkeeping columns to trail the read schema") + } + // Native applies the DV itself and emits a dead constant for row-index, so the real value + // must be provably unused above the scan. + if (!rowIndexUnusedAbove(plan, scanExec)) { + return Some( + "Native Delta scan cannot supply _metadata.row_index values consumed by the query") + } + // The DV common builder does not serialize existence defaults yet. + if (getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with deletion vectors") + } + // Bound native's memory for expanded DV row selectors (delta_dv.rs), bounded above by + // 2*cardinality + #row-groups; the descriptor's cardinality is a sound, pessimistic upper + // bound. The conf below makes an over-pessimistic decline recoverable. + val maxDeletedRowsPerFile = DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.get() + val oversizedCardinalities = dvDescriptors + .map(_.cardinality) + .filter(_ > maxDeletedRowsPerFile) + if (oversizedCardinalities.nonEmpty) { + return Some( + "Native Delta scan does not support a deletion vector deleting " + + s"${oversizedCardinalities.max} rows in a single file, exceeding " + + s"${DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.key}=$maxDeletedRowsPerFile") + } + } + + // input_file_name & friends read from a thread-local Spark's FileScanRDD sets; the native scan + // does not populate it, and Delta's DML find-touched-files scans use it (mirrors core's check + // in CometScanRule.nativeScan). + if (plan.exists(node => + node.expressions.exists(_.exists { + case _: InputFileName | _: InputFileBlockStart | _: InputFileBlockLength => true + case _ => false + }))) { + return Some( + "Native Delta scan is not compatible with input_file_name, " + + "input_file_block_start, or input_file_block_length") + } + + // Row-index metadata columns are generated per-row by Spark's reader (mirrors core); the DV + // shape's trailing row-index column is exempt since the gates above already proved it dead. + if (!CometDeltaNativeScan.isDvShape(scanExec) && + ShimFileFormat.findRowIndexColumnIndexInSchema(scanExec.requiredSchema) >= 0) { + return Some("Native Delta scan does not support row index generation") + } + + // Mirror core's vectorized-reader compatibility gate. + if (!SQLConf.get.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED) && + !CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.get()) { + return Some( + "Native Delta scan is incompatible with " + + s"${SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key}=false") + } + + // Decline ALL encrypted-parquet configurations (stricter than core): the exec node does not + // yet wire the decryption-key broadcast to executors. + val hadoopConf = scanExec.relation.sparkSession.sessionState + .newHadoopConfWithOptions(scanExec.relation.options) + if (CometParquetUtils.encryptionEnabled(hadoopConf)) { + return Some("Native Delta scan does not support encrypted parquet") + } + + // Nested-type column defaults cannot be serialized; a dropped default would misalign the + // value/index lists consumed positionally on the native side. Mirrors core's + // transformV1Scan gate. + val possibleDefaultValues = getExistenceDefaultValues(scanExec.requiredSchema) + if (possibleDefaultValues.exists(d => + d != null && (d.isInstanceOf[ArrayBasedMapData] || d + .isInstanceOf[GenericInternalRow] || d.isInstanceOf[GenericArrayData]))) { + return Some("Native Delta scan does not support default values for nested types") + } + + // Only claim scans whose root paths object_store (or the configured libhdfs schemes) can + // actually read (mirrors core's unsupportedFsSchemes gate). + val libhdfs = libhdfsSchemes + val unsupportedRootSchemes = + unsupportedSchemes(scanExec.relation.location.rootPaths.map(_.toUri), libhdfs) + if (unsupportedRootSchemes.nonEmpty) { + return Some( + "Native Delta scan does not support filesystem scheme(s) " + + s"${unsupportedRootSchemes.mkString(", ")}") + } + + // A shallow clone can span multiple object-store authorities; the native builder resolves the + // scan's ObjectStoreUrl from the FIRST selected file only, so a later file under a different + // store would silently read through the wrong handle. Force file listing (scanHelper is + // already built for the claim path) and decline rather than risk it. + val dataFileUris = + scanHelper.selectedPartitions.iterator.flatMap(_.files).map(_.getPath.toUri).toSeq + + // Both gates below need the DV absolute-path URIs; dvDescriptors is already memoized. + val dvUris = dvDescriptors + .filter(_.storageType != DeletionVectorDescriptor.INLINE_DV_MARKER) + .map(_.absolutePath(new Path(tableRoot)).toUri) + + // The root-path gate above only inspects the table root(s); selected files can resolve + // through a different scheme (e.g. `viewfs:`). Checked before the authority gates below, + // which presume every URI is natively resolvable. + val unsupportedSelected = unsupportedSelectedSchemeReason(dataFileUris ++ dvUris, libhdfs) + if (unsupportedSelected.isDefined) { + return unsupportedSelected + } + + // Checked before multiStoreReason, which presumes every URI resolves to a single store + // identity -- a userinfo-bearing authority provably does not (store keying drops userinfo). + val userInfoReason = userInfoBearingAuthorityReason(dataFileUris ++ dvUris) + if (userInfoReason.isDefined) { + return userInfoReason + } + + val multiStore = multiStoreReason(dataFileUris) + if (multiStore.isDefined) { + return multiStore + } + + // Credentials that resolve ONLY through a Hadoop credential provider (JCEKS et al.) are + // invisible to the plain-conf extraction forwarded to the native S3 client; reuses hadoopConf + // from the encryption gate above. + val credentialReason = credentialAliasReason(hadoopConf, dataFileUris ++ dvUris) + if (credentialReason.isDefined) { + return credentialReason + } + + // A credential-provider class native's build_aws_credential_provider_metadata (s3.rs) does + // not recognize errors at scan EXECUTION time, after the scan was already claimed; decline + // eagerly instead. + val providerReason = providerClassGateReason(hadoopConf, dataFileUris ++ dvUris) + if (providerReason.isDefined) { + return providerReason + } + + // Reuse core's generic native-scan gates (ignoreCorruptFiles/ignoreMissingFiles, AQE DPP on + // Spark 3.4, exec enabled); tags its own fallback reasons. + if (!CometNativeScan.isSupported(scanExec)) { + return Some("Core native scan gates rejected the scan (see reasons above)") + } + + None + } + + /** + * Deletion-vector descriptors for every file this DV-shape scan selected, deserialized once and + * normalized to absolute on-disk paths via `copyWithAbsolutePath`. Returns `Seq.empty` for the + * plain shape ([[CometDeltaNativeScan.isDvShape]] false). Shared by the DV cardinality gate and + * [[CometDeltaNativeScan.convert]]'s object-store option merge. + */ + private[delta] def selectedDvDescriptors( + scanHelper: CometScanExec, + tableRoot: String): Seq[DeletionVectorDescriptor] = { + if (!CometDeltaNativeScan.isDvShape(scanHelper.wrapped)) { + return Seq.empty + } + val tableRootPath = new Path(tableRoot) + scanHelper.selectedPartitions.iterator + .flatMap(_.files) + .flatMap { file => + file.metadata + .get(DeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED) + .map(enc => DeletionVectorDescriptor.deserializeFromBase64(enc.asInstanceOf[String])) + } + .map(_.copyWithAbsolutePath(tableRootPath)) + .toSeq + } + + /** + * The libhdfs scheme exemption set from [[org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES]], + * lowercased and defaulting to `Set("hdfs")` when unset. Hoisted so [[declineReason]]'s + * root-path and selected-file scheme gates share one parsed set. + */ + private[delta] def libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get() match { + case Some(s) => + s.split(",").map(_.trim.toLowerCase(Locale.ROOT)).filter(_.nonEmpty).toSet + case None => Set("hdfs") + } + + /** + * The lowercased, deduplicated schemes among `uris` that neither `libhdfs` nor Comet's native + * object_store layer ([[CometScanRule.isNativelyReadableScheme]]) can read. A `null` scheme is + * tolerated, not flagged: such a URI cannot come from a Hadoop-backed source. Factored out of + * [[declineReason]]'s two scheme gates so both share one predicate, unit-testable without a + * Spark session. + */ + private[delta] def unsupportedSchemes(uris: Seq[URI], libhdfs: Set[String]): Set[String] = { + uris + .filter { uri => + val sch = uri.getScheme + sch != null && { + val sl = sch.toLowerCase(Locale.ROOT) + !libhdfs.contains(sl) && !CometScanRule.isNativelyReadableScheme(uri) + } + } + .map(_.getScheme.toLowerCase(Locale.ROOT)) + .toSet + } + + /** + * Decline reason when any of `uris` -- the scan's selected data-file and deletion-vector URIs + * -- use a scheme [[unsupportedSchemes]] flags, or `None` when every URI is natively readable + * (or libhdfs-exempt). See [[declineReason]]'s call site for ordering vs. the authority gates + * below. + */ + private[delta] def unsupportedSelectedSchemeReason( + uris: Seq[URI], + libhdfs: Set[String]): Option[String] = { + val schemes = unsupportedSchemes(uris, libhdfs) + if (schemes.isEmpty) { + None + } else { + Some( + "Native Delta scan does not support selected data file or deletion vector filesystem " + + s"scheme(s) ${schemes.mkString(", ")}") + } + } + + /** + * Decline reason when `uris` span more than one object-store authority (scheme plus the raw + * authority -- userinfo, host, port -- lowercased so `S3A://Bucket:1234` and + * `s3a://bucket:1234` collapse), or `None` when they share one. `file://` paths carry no + * authority, so local scans across many directories are unaffected. + */ + private[delta] def multiStoreReason(uris: Seq[URI]): Option[String] = { + val authorities = uris.map(uriAuthority).distinct + if (authorities.size > 1) { + Some( + "Native Delta scan does not support data files spanning multiple object stores " + + s"(found: ${authorities.sorted.mkString(", ")})") + } else { + None + } + } + + /** + * Normalizes `uri` to a lowercased `scheme://authority` string, keyed on the raw `getAuthority` + * rather than the individually-parsed host/port/userinfo fields: `getAuthority` already + * includes userinfo, so containers on the same storage account don't collapse together; and + * `getHost` (and `getUserInfo`/`getPort`) return `null` for the whole authority when it doesn't + * conform to RFC 3986's `reg-name` syntax (e.g. an underscore in a GCS bucket name, + * `gs://my_bucket`), silently collapsing distinct buckets into one empty-host key. A `null` + * authority normalizes to the empty string. + */ + private[delta] def uriAuthority(uri: URI): String = { + val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("") + val authority = Option(uri.getAuthority).map(_.toLowerCase(Locale.ROOT)).getOrElse("") + s"$scheme://$authority" + } + + /** + * The raw userinfo component of `uri`'s authority, or empty when none. Splits at the LAST `@` + * rather than using `URI#getUserInfo`: as with [[uriAuthority]], the structured getters return + * `null` for the whole authority when it fails RFC 3986 `reg-name` syntax (e.g. an underscore + * in a GCS bucket name), hiding a real userinfo component. Never lowercased: userinfo is + * case-sensitive. + */ + private[delta] def uriUserInfo(uri: URI): String = { + val authority = Option(uri.getAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + if (at >= 0) authority.substring(0, at) else "" + } + + /** + * Redacts `uri`'s authority to `scheme` + `://` + a literal `***` + `@host[:port]`, for + * embedding in a decline reason. NEVER interpolate `uri.getAuthority` or [[uriUserInfo]] + * directly into a reason string: doing so would leak credentials embedded as URI userinfo (e.g. + * `s3a://AKIA...:secret@bucket`) into the SQL plan's explain output, fallback-reason logging, + * or the Spark UI. + */ + private[delta] def redactedAuthority(uri: URI): String = { + val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("") + val authority = Option(uri.getAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + val hostPort = if (at >= 0) authority.substring(at + 1) else authority + s"$scheme://***@$hostPort" + } + + /** + * Decline reason when any of `uris` carries userinfo in its authority (e.g. the container in an + * abfss:// path), or `None` when none do. The native store cache, `ObjectStoreUrl`, and + * DataFusion registry all key on scheme/host/port only, dropping userinfo, so two authorities + * differing only in userinfo collide onto the same store handle -- normally a missing-object + * error, but silently wrong data if a same-named object exists under both. + */ + private[delta] def userInfoBearingAuthorityReason(uris: Seq[URI]): Option[String] = { + val offending = uris.filter(uri => uriUserInfo(uri).nonEmpty).map(redactedAuthority).distinct + if (offending.isEmpty) { + None + } else { + Some("Native Delta scan does not support object-store paths whose authority carries " + + "userinfo (e.g. the container in an abfss:// path): the native object-store cache, " + + "ObjectStoreUrl and DataFusion registry all key on scheme, host and port only, so two " + + "containers on one storage account share a single store handle " + + s"(found: ${offending.sorted.mkString(", ")})") + } + } + + /** + * String-literal Hadoop conf keys consulted by [[credentialAliasReason]] below. `hadoop-aws` is + * NOT on this module's runtime classpath (`spark-hadoop-cloud` is test-scope only), so + * `org.apache.hadoop.fs.s3a.Constants` must never be referenced here -- doing so would raise + * `NoClassDefFoundError` and take down every Delta scan in a session with no S3 dependency at + * all, not just S3 ones. + */ + private val HadoopCredentialProviderPathKey = "hadoop.security.credential.provider.path" + private val S3aCredentialProviderPathKey = "fs.s3a.security.credential.provider.path" + + private def s3aBucketProviderPathKey(bucket: String): String = + s"fs.s3a.bucket.$bucket.security.credential.provider.path" + + /** + * The LONG form of [[s3aBucketProviderPathKey]]: `S3AUtils#lookupPassword` resolves per-bucket + * overrides through both a long key (`fs.s3a.bucket.B.<full base key>`) and a short key + * (`fs.s3a.bucket.B.<base key minus its "fs.s3a." prefix>`) -- see [[s3aCredentialAliases]] for + * why both must be covered here too. + */ + private def s3aBucketLongProviderPathKey(bucket: String): String = + s"fs.s3a.bucket.$bucket.fs.s3a.security.credential.provider.path" + + /** + * The aliases the native S3 credentials provider chain tries per bucket (short bucket form, + * then global), PLUS the LONG bucket form Hadoop's `S3AUtils#lookupPassword` also consults. + * Hadoop resolves long before short before global, keeping a non-empty long value over the + * short/global ones (`S3AUtils#getPassword` returns a non-empty `val` unchanged); native reads + * only short+global. So a JCEKS entry set ONLY under the long alias is resolved by Hadoop but + * invisible to native -- a shadowed value under ANY alias here means the caller must decline. + * List order only affects which alias name appears in the reason. + */ + private def s3aCredentialAliases(bucket: String): Seq[String] = + Seq( + s"fs.s3a.bucket.$bucket.fs.s3a.access.key", + s"fs.s3a.bucket.$bucket.fs.s3a.secret.key", + s"fs.s3a.bucket.$bucket.fs.s3a.session.token", + s"fs.s3a.bucket.$bucket.access.key", + s"fs.s3a.bucket.$bucket.secret.key", + s"fs.s3a.bucket.$bucket.session.token", + "fs.s3a.access.key", + "fs.s3a.secret.key", + "fs.s3a.session.token") + + private def nonEmptyConf(hadoopConf: Configuration, key: String): Boolean = + Option(hadoopConf.get(key)).exists(_.nonEmpty) + + /** + * The lowercase-scheme-checked S3/S3A bucket name from `uri`'s authority (host, minus any + * userinfo or port), or `None` when `uri`'s scheme is not `s3`/`s3a`. Parses the raw authority + * manually (last `@`, then last `:`) rather than using `URI#getHost`: the same RFC 3986 + * `reg-name` pitfall as [[uriAuthority]] applies (an underscore, valid in an S3 bucket name, + * makes `getHost` return `null` for the whole authority). + */ + private def s3Bucket(uri: URI): Option[String] = { + val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)) + if (scheme.contains("s3") || scheme.contains("s3a")) { + val authority = Option(uri.getAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + val hostAndPort = if (at >= 0) authority.substring(at + 1) else authority + val colon = hostAndPort.lastIndexOf(':') + val host = if (colon >= 0) hostAndPort.substring(0, colon) else hostAndPort + if (host.isEmpty) None else Some(host) + } else { + None + } + } + + /** + * The three per-bucket S3A credential base keys the native S3 client's `get_config` (`s3.rs`) + * resolves: short bucket key first, then global (native never reads the long bucket key). + */ + private val PlainCredentialBaseKeys: Seq[String] = + Seq("fs.s3a.access.key", "fs.s3a.secret.key", "fs.s3a.session.token") + + private def plainValue(hadoopConf: Configuration, key: String): Option[String] = + Option(hadoopConf.get(key)).filter(_.nonEmpty) + + /** + * The short-bucket-then-global value native's `get_config` (s3.rs) resolves for `baseKey` under + * `bucket`. Used by the credential-provider-class gates below, which read general S3A options + * the same way native does. + */ + private def effectiveOptionValue( + hadoopConf: Configuration, + bucket: String, + baseKey: String): Option[String] = { + val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.") + plainValue(hadoopConf, shortKey).orElse(plainValue(hadoopConf, baseKey)) + } + + private def plainLongFormCredentialDivergenceReason(bucket: String, longKey: String): String = + "Native Delta scan cannot forward long-form bucket credentials for " + + s"$bucket ($longKey is set but the native S3 client only reads the short bucket and " + + "global keys, so its credentials would differ from Hadoop's)" + + /** + * Zero-I/O plain-value divergence check (see [[s3aCredentialAliases]] for the long-before-short + * precedence). Native's `get_config` never reads the long form, so it can silently diverge from + * Hadoop's effective value. Declines when the long form is set and the two sides disagree + * (short set to the SAME value as long passes). Never interpolates a resolved value, only the + * key name. + * + * This applies to credentials ONLY. Hadoop resolves them via `S3AUtils#lookupPassword`, where + * the long bucket form (`fs.s3a.bucket.B.fs.s3a.<key>`) WINS when set -- a real, silent + * divergence from native, which never reads it. Every OTHER `fs.s3a` option instead flows + * through `S3AUtils#propagateBucketOptions`, which strips only ONE `fs.s3a.bucket.B.` prefix + * layer: a long-form key there folds into `fs.s3a.fs.s3a.<key>`, a key nothing else in Hadoop + * ever reads, so the long form is inert for those options -- Hadoop's own effective value + * already reduces to short.orElse(global), matching native. No gate is needed (or present) for + * that wider set; see DeltaScanContribSuite's pinned control test for the long-form endpoint + * case. + */ + private def plainLongFormCredentialReason( + hadoopConf: Configuration, + bucket: String): Option[String] = { + PlainCredentialBaseKeys.foldLeft(Option.empty[String]) { (declined, baseKey) => + if (declined.isDefined) { + declined + } else { + val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.") + val longKey = s"fs.s3a.bucket.$bucket.$baseKey" + val long = plainValue(hadoopConf, longKey) + if (long.isEmpty) { + None + } else { + val short = plainValue(hadoopConf, shortKey) + val global = plainValue(hadoopConf, baseKey) + val hadoopEffective = long.orElse(short).orElse(global) + val nativeEffective = short.orElse(global) + if (hadoopEffective != nativeEffective) { + Some(plainLongFormCredentialDivergenceReason(bucket, longKey)) + } else { + None + } + } + } + } + } + + /** + * String-literal mirror of every credential-provider class name s3.rs's + * `build_aws_credential_provider_metadata` and `is_anonymous_credential_provider` recognize + * (Hadoop S3A names plus AWS SDK v1/v2 names). `hadoop-aws` is NOT on this module's runtime + * classpath (see the note above [[HadoopCredentialProviderPathKey]]), so these are string + * literals, never `org.apache.hadoop.fs.s3a.auth.*` or AWS SDK provider `classOf` references. + */ + private val SupportedCredentialProviderClasses: Set[String] = Set( + "org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider", + "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider", + "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider", + "org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider", + "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider", + "software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider", + "com.amazonaws.auth.ContainerCredentialsProvider", + "com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper", + "software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider", + "com.amazonaws.auth.InstanceProfileCredentialsProvider", + "software.amazon.awssdk.auth.credentials.EnvironmentVariableCredentialsProvider", + "com.amazonaws.auth.EnvironmentVariableCredentialsProvider", + "software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider", + "com.amazonaws.auth.WebIdentityTokenCredentialsProvider", + "software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider", + "com.amazonaws.auth.profile.ProfileCredentialsProvider", + "software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider", + "com.amazonaws.auth.AnonymousAWSCredentials") + + private val AnonymousCredentialProviderClasses: Set[String] = Set( + "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider", + "software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider", + "com.amazonaws.auth.AnonymousAWSCredentials") + + private val HadoopAssumedRoleProviderClass = + "org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider" + + private val AwsCredentialsProviderKey = "fs.s3a.aws.credentials.provider" + private val AssumedRoleCredentialsProviderKey = "fs.s3a.assumed.role.credentials.provider" + + /** + * Splits a Hadoop-style comma-separated credential-provider-class list the same way s3.rs's + * `parse_credential_provider_names` does: split on comma, trim each entry, drop empties. + */ + private def parseProviderClassNames(value: String): Seq[String] = + value.split(",").map(_.trim).filter(_.nonEmpty).toSeq + + private def unsupportedProviderReason(bucket: String, key: String, className: String): String = + s"Native Delta scan does not support the credential provider class $className " + + s"configured via $key for $bucket (the native S3 client only supports a fixed set of " + + "provider classes; an unsupported class would fail at scan execution time, after the " + + "scan was already claimed, rather than at planning time)" + + private def mixedAnonymousProviderReason(bucket: String, key: String): String = + s"Native Delta scan does not support $key for $bucket naming an anonymous credential " + + "provider together with any other provider (the native S3 client rejects this " + + "combination at scan execution time)" + + private def anonymousAssumedRoleProviderReason(bucket: String, key: String): String = + s"Native Delta scan does not support an anonymous credential provider in $key for " + + s"$bucket (the native S3 client does not allow an anonymous provider as the base " + + "credentials for an assumed-role chain)" + + private def unsupportedProviderNameReason( + bucket: String, + key: String, + names: Seq[String]): Option[String] = + names + .find(name => !SupportedCredentialProviderClasses.contains(name)) + .map(unsupportedProviderReason(bucket, key, _)) + + /** + * Decline reason when `bucket`'s effective (short-bucket-then-global; see + * [[effectiveOptionValue]]) `assumed.role.credentials.provider` names an unsupported class, or + * an anonymous one (native's `build_assume_role_credential_provider_metadata` rejects ANY + * anonymous entry here, not just a mix). Unset defaults to native's own hardcoded + * `[SimpleAWSCredentialsProvider, EnvironmentVariableCredentialsProvider]` fallback, both + * always supported, so `None` is safe. + */ + private def assumedRoleProviderClassReason( + hadoopConf: Configuration, + bucket: String): Option[String] = { + effectiveOptionValue(hadoopConf, bucket, AssumedRoleCredentialsProviderKey) match { + case None => None + case Some(value) => + val names = parseProviderClassNames(value) + unsupportedProviderNameReason(bucket, AssumedRoleCredentialsProviderKey, names).orElse { + if (names.exists(AnonymousCredentialProviderClasses.contains)) { + Some(anonymousAssumedRoleProviderReason(bucket, AssumedRoleCredentialsProviderKey)) + } else { + None + } + } + } + } + + /** + * Decline reason when `bucket`'s effective `aws.credentials.provider` names a class native's + * `build_aws_credential_provider_metadata` (s3.rs) does not recognize, mixes an anonymous + * provider with any other provider (native's `build_credential_provider` rejects this + * combination), or -- when `AssumedRoleCredentialProvider` is among the names -- its + * `assumed.role.credentials.provider` sub-chain has the same problem. An unset/empty value is + * fine: native falls back to its own default AWS SDK provider chain. Checked so an unsupported + * class or invalid combination declines at planning time instead of erroring during scan + * execution, after the scan was already claimed. + */ + private def providerClassReason(hadoopConf: Configuration, bucket: String): Option[String] = { + effectiveOptionValue(hadoopConf, bucket, AwsCredentialsProviderKey).flatMap { value => + val names = parseProviderClassNames(value) + unsupportedProviderNameReason(bucket, AwsCredentialsProviderKey, names) + .orElse { + if (names.length > 1 && names.exists(AnonymousCredentialProviderClasses.contains)) { + Some(mixedAnonymousProviderReason(bucket, AwsCredentialsProviderKey)) + } else { + None + } + } + .orElse { + if (names.contains(HadoopAssumedRoleProviderClass)) { + assumedRoleProviderClassReason(hadoopConf, bucket) + } else { + None + } + } + } + } + + /** + * Returns the first reason any bucket among `uris` names an unsupported (or invalidly combined) + * credential-provider class, or `None` when every bucket's provider configuration is one native + * can build. See [[providerClassReason]]. + */ + private[delta] def providerClassGateReason( + hadoopConf: Configuration, + uris: Seq[URI]): Option[String] = { + val buckets = uris.flatMap(s3Bucket).distinct + buckets.foldLeft(Option.empty[String]) { (declined, bucket) => + if (declined.isDefined) declined else providerClassReason(hadoopConf, bucket) + } + } + + private def s3aScopedProviderPathReason(bucket: String, providerPathKey: String): String = + "Native Delta scan cannot forward Hadoop credential-provider aliases for " + + s"$bucket ($providerPathKey configures an S3A-scoped Hadoop credential provider that " + + "Configuration#getPassword does not consult, so the native S3 client's credentials " + + "cannot be verified)" + + private def shadowedCredentialAliasReason(bucket: String, alias: String): String = + "Native Delta scan cannot forward Hadoop credential-provider aliases for " + + s"$bucket ($alias resolves through $HadoopCredentialProviderPathKey but is not present " + + "as a plain configuration value, so the native S3 client would have no credentials)" + + private def unverifiableCredentialProviderReason(bucket: String, error: Throwable): String = + "Native Delta scan cannot verify Hadoop credential-provider aliases for " + + s"$bucket (reading $HadoopCredentialProviderPathKey raised " + + s"${error.getClass.getName}), declining rather than risk missing credentials" + + /** + * Compares each [[s3aCredentialAliases]] alias's plain `Configuration#get` value against + * `Configuration#getPassword` (providers first, plain conf as fallback); a resolved value that + * differs means the alias is invisible or wrong to native's plain-conf extraction, so decline. + * Only called once the GLOBAL provider path is established as the sole path for `bucket`. + * + * Runs inside try/catch: `getPassword` performs real keystore I/O, and a corrupt or unreadable + * store must decline this bucket rather than abort planning for the whole session. + */ + private def verifyGlobalProviderAliases( + hadoopConf: Configuration, + bucket: String): Option[String] = { + try { + s3aCredentialAliases(bucket).foldLeft(Option.empty[String]) { (declined, alias) => + if (declined.isDefined) { + declined + } else { + val resolved = + Option(hadoopConf.getPassword(alias)).map(new String(_)).filter(_.nonEmpty) + resolved match { + case Some(value) if !Option(hadoopConf.get(alias)).contains(value) => + Some(shadowedCredentialAliasReason(bucket, alias)) + case _ => None + } + } + } + } catch { + case e @ (_: IOException | _: RuntimeException) => + Some(unverifiableCredentialProviderReason(bucket, e)) + } + } + + /** + * The decline reason for `bucket` alone, or `None` when its credentials are safe to forward. + * `globalPathSet`/`s3aPathSet` are hoisted by the caller since they don't vary per bucket. + * + * Checked FIRST, regardless of provider-path config: [[plainLongFormCredentialReason]] is a + * zero-I/O, provider-unrelated check that is decisive on its own. + * + * Zero-I/O precheck: `None` immediately when none of the four provider-path keys are set -- a + * config-map lookup only, no keystore access. + * + * Arm A: the S3A-scoped provider-path keys point S3A's own resolution at a provider + * `Configuration#getPassword` does not consult, so any being set declines with no keystore + * read. + * + * Arm B: only the global path is set, which `getPassword` DOES consult; delegates to + * [[verifyGlobalProviderAliases]]. + */ + private def bucketCredentialAliasReason( + hadoopConf: Configuration, + bucket: String, + globalPathSet: Boolean, + s3aPathSet: Boolean): Option[String] = { + plainLongFormCredentialReason(hadoopConf, bucket).orElse { + val bucketPathKey = s3aBucketProviderPathKey(bucket) + val bucketLongPathKey = s3aBucketLongProviderPathKey(bucket) + val bucketPathSet = nonEmptyConf(hadoopConf, bucketPathKey) + val bucketLongPathSet = nonEmptyConf(hadoopConf, bucketLongPathKey) + if (!globalPathSet && !s3aPathSet && !bucketPathSet && !bucketLongPathSet) { + None + } else if (s3aPathSet || bucketPathSet || bucketLongPathSet) { + val offendingKey = + if (s3aPathSet) S3aCredentialProviderPathKey + else if (bucketPathSet) bucketPathKey + else bucketLongPathKey + Some(s3aScopedProviderPathReason(bucket, offendingKey)) + } else { + verifyGlobalProviderAliases(hadoopConf, bucket) + } + } + } + + /** + * Returns the first reason a native S3 scan cannot faithfully forward this table's Hadoop + * credentials, or `None` when claimable. `uris` is the scan's data-file and DV URIs; only + * `s3`/`s3a` authorities matter here (ABFS/WASB mooted by the userinfo gate, GCS out of scope). + * `Configuration#getPassword` checks every provider FIRST, falling back to plain conf only when + * unset, so a keystore-only or shadowing alias is invisible/wrong to native's plain-conf + * extraction; when in doubt, decline. Never interpolates a resolved value, only key names. + */ + private[delta] def credentialAliasReason( + hadoopConf: Configuration, + uris: Seq[URI]): Option[String] = { + val buckets = uris.flatMap(s3Bucket).distinct + if (buckets.isEmpty) { + return None Review Comment: Added the conservative gate: any scan whose data files or DVs touch gs:// declines when a fs.gs.auth.* key is set, since the native resolver forwards no GCS options. ADC-only configs still claim because both engines resolve credentials the same way there. Tests cover your mixed local-data plus gs DV sidecar shape and the scheme scoping. -- 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]
