dwsmith1983 commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3886368860
########## 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: Covered by the allowlist change: CSE-KMS, CSE-CUSTOM, and any unknown algorithm decline, server-side ones stay claimable. Two adjacent things got picked up in the same pass: the algorithm can live in a JCEKS keystore so it resolves through the provider machinery now, and there's a proxy gate since native has no proxy support at all. -- 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]
