dwsmith1983 commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r4006924646
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,1838 @@ +/* + * 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, Map => MutableMap} +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.objectstore.NativeConfig +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" + + /** + * Claim-time artifacts [[declineReason]] already computes but [[CometDeltaNativeScan.convert]] + * also needs -- threaded through by reference (populated only on the claimable path, right + * before `declineReason` returns `None`) so a claimed scan does not pay to recompute either: + * the Hadoop conf ([[org.apache.spark.sql.internal.SessionState#newHadoopConfWithOptions]] is + * not cheap) and the deletion-vector descriptors (base64-decoded, non-trivial only for DV-shape + * scans). One instance is created per claim attempt in `DeltaScanContrib` and passed to both + * `declineReason` and `convert`. + */ + private[delta] final class DeltaClaimMemo { + var hadoopConf: Configuration = _ + var dvDescriptors: Seq[DeletionVectorDescriptor] = Seq.empty + } + + /** + * First reason this Delta scan cannot go native, or None when claimable (in which case `memo` + * is populated for [[CometDeltaNativeScan.convert]] to reuse). Only called when [[isDeltaScan]] + * is true. `scanHelper` is the [[CometScanExec]] built to drive `convert` on a claim, reused + * for the multi-store gate below. + */ + def declineReason( + plan: SparkPlan, + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + memo: DeltaClaimMemo): 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`, forced at most + // once in this method; on the claimable path the result is handed to `convert` through + // `memo` below, so a claimed scan deserializes the descriptors exactly once end to end. + 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) + // Populated now (rather than only at the very end) so it is available even though several + // early-return gates below still lie ahead: cheap to set, and every one of those gates + // declines the scan anyway, so `memo` is simply never read by `convert` in that case. + memo.hadoopConf = hadoopConf + 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") + } + + // An opted-in S3-compliant alias scheme (fs.comet.s3Compliant.schemes) is declined before the + // generic scheme gate below so the reason says why: core's native scan reads it through the + // S3 client, but the S3 divergence gates further down model Hadoop's S3AFileSystem only. + val rootUris = scanExec.relation.location.rootPaths.map(_.toUri) + val aliasReason = s3CompliantAliasSchemeReason(hadoopConf, rootUris) + if (aliasReason.isDefined) { + return aliasReason + } + + // 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(rootUris, libhdfs) + if (unsupportedRootSchemes.nonEmpty) { + return Some( + "Native Delta scan does not support filesystem scheme(s) " + + s"${unsupportedRootSchemes.mkString(", ")}") + } + + // A recognized scheme can still carry a path object_store rejects (a directory name with a + // newline surfaces as `%0A`), which native planning hard-fails on while Spark's reader opens + // it. Mirrors core's root-path gate; the complete selected paths are probed below. + val rejectedRoot = objectStoreRejectedPathReason(rootUris, libhdfs) + if (rejectedRoot.isDefined) { + return rejectedRoot + } + + // 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 + } + + // Same path probe for every complete selected path, not just its directory: a shallow + // clone's source can sit outside this root, and CONVERT TO DELTA keeps the source Parquet + // basenames, so the rejected character can be in the file name itself. The probe is a + // native URL parse with no I/O, so once per distinct URI costs less than the scan's own + // per-file parse. + val rejectedSelected = objectStoreRejectedPathReason(dataFileUris ++ dvUris, libhdfs) + if (rejectedSelected.isDefined) { + return rejectedSelected + } + + // 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 an + // encryption algorithm outside the allowlist (SSE-C, CSE-KMS, CSE-CUSTOM, or unknown) before + // the credential-divergence gates below, which do not otherwise notice this table is readable + // through Hadoop only because Hadoop's request factory (SSE-C) or SDK-level decryption layer + // (CSE-*) does something native never learns about. + val encryptionReason = + unsupportedEncryptionAlgorithmReason(hadoopConf, dataFileUris ++ dvUris) + if (encryptionReason.isDefined) { + return encryptionReason + } + + // Shared across the two gates below: propagateBucketOptions is a full Configuration deep + // copy, and both gates would otherwise recompute it independently for the same bucket(s) + // (once here, then again per-key inside s3ConfigDivergenceReason). One cache, populated + // lazily per bucket on first use, makes it a single copy total per bucket across both gates. + val propagatedConfCache = MutableMap.empty[String, Configuration] + + // Always zero-I/O (plain propagated-conf read, no keystore): native's S3 client has no + // HTTP proxy support at all (no fs.s3a.proxy.* key is read anywhere in s3.rs), so a bucket + // requiring a proxy for S3 egress must decline here rather than claim and then connect + // directly, bypassing whatever network-segmentation/firewall policy required the proxy. + val proxyReason = proxyGateReason(hadoopConf, dataFileUris ++ dvUris, propagatedConfCache) + if (proxyReason.isDefined) { + return proxyReason + } + + // Zero-I/O, conf-only, like the proxy gate above: Hadoop's AssumedRoleCredentialProvider + // sends fs.s3a.assumed.role.policy as the session policy of its STS AssumeRole request, + // while native's assumed-role provider never reads the key -- a claimed scan would assume + // the role WITHOUT the configured session restriction, silently widening permissions. + val rolePolicyReason = + assumedRolePolicyGateReason(hadoopConf, dataFileUris ++ dvUris, propagatedConfCache) + if (rolePolicyReason.isDefined) { + return rolePolicyReason + } + + // 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, propagatedConfCache) + 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)") + } + + // Claimable: hand the already-forced descriptors to `convert` via `memo` so it does not + // deserialize them a second time. + memo.dvDescriptors = dvDescriptors + 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]], + * parsed exactly like core's scan gate (`NativeConfig.parseSchemeSet`: split on commas, + * trimmed, lowercased) and defaulting to `Set("hdfs")` when unset. + */ + private[delta] def libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get() match { + case Some(s) => NativeConfig.parseSchemeSet(s) + case None => Set("hdfs") + } + + /** + * Decline reason when any of `uris` uses a scheme opted in as an S3-compliant alias through + * `fs.comet.s3Compliant.schemes` (e.g. `blob`), or `None`. Core's native Parquet scan admits + * such a scheme and reads it through its S3 client, with `NativeConfig` translating the vendor + * `fs.<scheme>.<authority>.*` keys into `fs.s3a.bucket.*` options. Spark, however, reads the + * same table through the vendor's own Hadoop FileSystem, not `S3AFileSystem`, and every S3 + * divergence gate in this object ([[s3ConfigDivergenceReason]] and its siblings) is verified + * against `S3AFileSystem`'s consumers only. With no model of how the vendor filesystem resolves + * its configuration, whether native and Spark would agree cannot be decided, so the scan is + * declined rather than claimed on a guess. Selected data-file and deletion-vector URIs under an + * alias scheme are declined by the generic scheme gates, which never admit an alias (see + * [[unsupportedSchemes]]). + */ + private[delta] def s3CompliantAliasSchemeReason( + hadoopConf: Configuration, + uris: Seq[URI]): Option[String] = { + val aliases = NativeConfig.resolveS3CompliantSchemes(hadoopConf) + if (aliases.isEmpty) { + return None + } + val found = uris + .flatMap(uri => Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT))) + .filter(aliases.contains) + .distinct + if (found.isEmpty) { + None + } else { + Some( + "Native Delta scan does not support S3-compliant alias filesystem scheme(s) " + + s"${found.sorted.mkString(", ")} (${CometConf.COMET_S3_COMPLIANT_SCHEMES_KEY}): " + + "Spark reads them through a vendor filesystem whose S3 configuration resolution the " + + "native scan's S3AFileSystem divergence model cannot verify") + } + } + + /** + * 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. The alias + * set handed to core's gate is deliberately empty: an `fs.comet.s3Compliant.schemes` alias is + * never admitted here (see [[s3CompliantAliasSchemeReason]]), even though core admits it. + */ + 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, Set.empty) + } + } + .map(_.getScheme.toLowerCase(Locale.ROOT)) + .toSet + } + + /** + * Decline reason naming the first of `uris` whose path object_store rejects even though it + * recognizes the scheme ([[CometScanRule.objectStoreAcceptsPath]], e.g. a directory name + * containing a newline, `%0A` in the URI), or `None`. Schemes in `libhdfs` never reach + * object_store's path parser and are skipped, as is a `null` scheme (see + * [[unsupportedSchemes]]); an S3-compliant alias is declined before this gate runs. The probe + * is uncached but is a plain native URL parse with no I/O + * ([[CometScanRule.objectStoreAcceptsPath]]), so callers pass every complete selected path + * (root paths, data files and deletion vectors) once per distinct URI; a converted table can + * carry the rejected character in a file basename. The reason masks any userinfo in the named + * URI ([[redactedAuthority]]). + */ + private[delta] def objectStoreRejectedPathReason( + uris: Seq[URI], + libhdfs: Set[String]): Option[String] = { + uris.distinct + .find { uri => + val sch = uri.getScheme + sch != null && !libhdfs.contains(sch.toLowerCase(Locale.ROOT)) && + !CometScanRule.objectStoreAcceptsPath(uri) + } + .map { uri => + // Mask userinfo (see redactedAuthority); the raw path keeps its percent encoding so the + // reason shows the rejected sequence as written. + val shown = + if (uriUserInfo(uri).isEmpty) uri.toString + else s"${redactedAuthority(uri)}${Option(uri.getRawPath).getOrElse("")}" + s"Native Delta scan cannot open path '$shown': object_store rejects it " + + "(e.g. an unsupported character in the path)" + } + } + + /** + * 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" + + /** + * `CommonConfigurationKeysPublic.HADOOP_SECURITY_CREDENTIAL_CLEAR_TEXT_FALLBACK`, default + * `true`, verified via `javap` against `hadoop-common` 3.3.4's + * `Configuration#getPasswordFromConfig`: `getPassword` only falls back to reading a plaintext + * conf value once `getBoolean(<this key>, true)` holds -- with the flag off, a plaintext value + * is invisible to every `getPassword`-based resolver, even when no credential provider is + * configured at all. + */ + private val ClearTextFallbackKey = "hadoop.security.credential.clear-text-fallback" + + 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) + + /** + * How Hadoop's OWN consumer reads one of the keys compared by [[s3ConfigDivergenceReason]], + * which decides how [[s3KeyDivergenceReason]] computes the Hadoop-effective side of its + * equality check. Exactly two consumer families exist among [[AllS3ConfigKeys]] in `hadoop-aws` + * 3.3.4, each verified via `javap`/CFR against the real call sites (cited per key on + * [[S3ConfigKeyConsumers]]). The tier must mirror the key's ACTUAL consumer: resolving a + * [[PropagatedOptionConsumer]] key through the wider `lookupPassword` cascade is NOT fail-safe + * for a value-EQUALITY comparator -- a long-form alias value Hadoop itself never reads can + * EQUAL native's resolution while Hadoop's true propagate-then-plain-get value differs, turning + * a real divergence into a wrongly-claimed scan (the endpoint `${...}`-redirect shape pinned in + * `DeltaScanContribSuite`). + */ + private[delta] sealed trait S3ConfigConsumer + + /** + * Read via `S3AUtils#lookupPassword(bucket, conf, baseKey)`, verified via `javap` against + * `hadoop-aws` 3.3.4: builds `longBucketKey = "fs.s3a.bucket." + bucket + "." + baseKey` (the + * FULL, already-`fs.s3a`-prefixed base key appended after the bucket segment) and reads it via + * `Configuration#getPassword` BEFORE the short-bucket key, keeping the long value whenever + * `getPassword` returns non-empty and only falling through to short-then-global otherwise. + * `getPassword` is Hadoop-credential-provider-aware and skips plaintext conf entirely when + * [[ClearTextFallbackKey]] is false. Modeled by [[hadoopLookupPasswordEffective]]. + */ + private[delta] case object LookupPasswordConsumer extends S3ConfigConsumer + + /** + * Read via `S3AUtils#propagateBucketOptions` followed by a plain `Configuration#get`-family + * call (`getTrimmed`/`getBoolean`/`getClasses`) against the propagated view: the short bucket + * form wins only by having overwritten the global key during propagation, the long bucket form + * folds into an unread `fs.s3a.fs.s3a.*` key, and neither a credential provider nor + * [[ClearTextFallbackKey]] is ever consulted. Modeled as a plain `Configuration#get` on the + * [[propagateBucketOptions]] result, which also expands `${...}` references under that + * propagated view exactly like the real consumer. + */ + private[delta] case object PropagatedOptionConsumer extends S3ConfigConsumer + + /** + * Every `fs.s3a.*` base key that governs whether a claimed native scan actually behaves like + * Hadoop's own reader would, paired with the consumer family Hadoop resolves it through -- ONE + * list, with each key's resolution tier declared beside it, so a key can never sit in the + * comparator without a deliberate classification (adding one without picking a tier does not + * compile). The entries are every 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). + * + * Tier assignments, each verified via `javap`/CFR against `hadoop-aws` 3.3.4: + * - access.key/secret.key/session.token: `S3AUtils#getAWSAccessKeys` and + * `MarshalledCredentialBinding#fromFileSystem` (reached from + * `TemporaryAWSCredentialsProvider`) resolve all three via `S3AUtils#lookupPassword` -- + * [[LookupPasswordConsumer]]. + * - aws.credentials.provider and assumed.role.credentials.provider: + * `S3AUtils#buildAWSProviderList` -> `loadAWSProviderClasses` -> plain + * `Configuration#getClasses` -- [[PropagatedOptionConsumer]]. + * - assumed.role.arn/session.name: `AssumedRoleCredentialProvider`'s constructor reads both + * via plain `Configuration#getTrimmed` -- [[PropagatedOptionConsumer]]. + * - endpoint (`S3AFileSystem`: `getTrimmed`), endpoint.region (`DefaultS3ClientFactory`: + * `getTrimmed`), path.style.access (`S3AFileSystem`: `getBoolean`) -- + * [[PropagatedOptionConsumer]]. + * - requester.pays.enabled: not read anywhere in `hadoop-aws` 3.3.4 (the constant does not + * even exist in its `Constants` class); later releases read it via plain `getBoolean` + * against the propagated conf, so the plain tier is both the faithful forward model and + * inert on 3.3.4 -- [[PropagatedOptionConsumer]]. + * - comet.credential.provider.class: Comet's own activation key, plain conf read on both + * sides, never a Hadoop key at all -- [[PropagatedOptionConsumer]]. + * + * SYNC NOTE: the key list must stay a superset of native's `NATIVE_S3A_CONFIG_PROPERTIES` + * constant (`native/core/src/parquet/objectstore/s3.rs`, property suffixes without the + * `fs.s3a.` prefix) -- `DeltaScanContribSuite`'s discovery-harness test asserts this + * mechanically against [[AllS3ConfigKeys]]. 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[delta] val S3ConfigKeyConsumers: Seq[(String, S3ConfigConsumer)] = Seq( Review Comment: > Does the discovery harness in `DeltaScanContribSuite` catch either of these? No, by design: the harness bounds the keys native reads, and both of these are keys Hadoop reads that native ignores. Added `hadoopOnlyEndpointGateReason` next to the proxy and session-policy gates: a scheme-less `fs.s3a.endpoint` with `fs.s3a.connection.ssl.enabled=false` declines, and either assumed-role STS endpoint key declines whenever set, both resolved on the propagated conf so per-bucket overrides apply. Six tests cover the decline, the SSL default, an endpoint carrying its own scheme, the per-bucket SSL override leaving another bucket claimable, both STS keys, and the empty-conf control. -- 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]
