andygrove commented on code in PR #5365:
URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3917353053
##########
native/proto/src/proto/operator.proto:
##########
@@ -77,6 +77,8 @@ message Operator {
// zero runtime cost.
ContribScan contrib_scan = 200;
}
+
+ reserved "delta_scan";
Review Comment:
`delta_scan` was never on `main`. Field 119 is and always was
`window_group_limit`, and `delta_scan` only existed in earlier revisions of
this branch. Since this reserves the name rather than a field number it does
not buy wire compatibility either, and it reads as though a released field was
removed. Could it just be dropped?
##########
contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/JvmLowercaseParitySuite.scala:
##########
@@ -0,0 +1,343 @@
+/*
+ * 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.util.Locale
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.comet.serde.operator.JvmCaseTables
+
+/**
+ * Self-validating proof that `JvmCaseTables.mirrorLowercase` -- a
line-for-line mirror of the
+ * native `JvmCaseTables::lowercase` in
`native/core/src/parquet/schema_adapter.rs`, run over the
+ * tables `JvmCaseTables` generates from the RUNNING JVM -- reproduces this
JVM's
+ * `String.toLowerCase(Locale.ROOT)`. Because both the tables and the
expectations come from the
+ * same running JVM, this suite passes on ANY supported JDK by construction.
+ *
+ * Needs no `SparkSession`, so it extends `AnyFunSuite` directly rather than
`CometDeltaTestBase`.
+ */
+class JvmLowercaseParitySuite extends AnyFunSuite {
Review Comment:
This suite proves that `JvmCaseTables.mirrorLowercase` equals
`String.toLowerCase(Locale.ROOT)`, but `mirrorLowercase` is not the code that
runs. The Rust `JvmCaseTables::lowercase` is, and its only coverage is the
hand-written 27-range `jdk17_test_tables` in `schema_adapter.rs`. Two
hand-written ports of an algorithm this intricate will drift eventually, and
the drift would surface as silently wrong column matching rather than a failure.
I checked the Rust side directly and it passes cleanly, as described in my
top-level comment, so this is not a bug report. It is that the check does not
exist in the repo. Could it live here? Writing the JVM-generated tables plus
the expected outputs to a test resource and reading them from a Rust test would
verify the real implementation and would make the 539-line Scala mirror
unnecessary.
Separately, since what this validates is core behavior rather than Delta
behavior, could the suite move to `spark/src/test`? The `delta` filter in
`compute-changes.py` does include `spark/src/main/**` and `native/**/src/**`,
so it fires today, but anyone building or testing Comet without `-Pdelta` never
runs it, and if the Delta contrib is ever renamed or dropped the verification
goes with it.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala:
##########
@@ -125,166 +150,233 @@ object CometNativeScan extends
CometOperatorSerde[CometScanExec] with CometTypeS
scan: CometScanExec,
builder: Operator.Builder,
childOp: OperatorOuterClass.Operator*):
Option[OperatorOuterClass.Operator] = {
- val nativeScanBuilder = OperatorOuterClass.NativeScan.newBuilder()
+ // Extract object store options from first file (S3 configs apply to all
files in scan).
+ // Use selectedPartitions (static) instead of getFilePartitions() because
at planning time
+ // DPP subqueries haven't been resolved yet. Object store options don't
depend on DPP.
+ val firstFileUri = scan.selectedPartitions
+ .flatMap(_.files.headOption)
+ .headOption
+ .map(_.getPath.toUri)
+
+ // Collect S3/cloud storage configurations
+ val hadoopConf = scan.relation.sparkSession.sessionState
+ .newHadoopConfWithOptions(scan.relation.options)
+
+ buildNativeScanCommon(
+ source = scan.simpleStringWithNodeId(),
+ output = scan.output,
+ requiredSchema = scan.requiredSchema,
+ dataSchema = scan.relation.dataSchema,
+ partitionSchema = scan.relation.partitionSchema,
+ fileConstantMetadataColumns = scan.wrapped.fileConstantMetadataColumns,
+ dataFilters = scan.supportedDataFilters,
+ firstFileUri = firstFileUri,
+ hadoopConf = hadoopConf,
+ conf = scan.conf) match {
+ case Some(commonBuilder) =>
+ // Sink operators don't have children
+ builder.clearChildren()
+ val nativeScanBuilder = OperatorOuterClass.NativeScan.newBuilder()
+ // Set common data in NativeScan (file_partition will be populated at
execution time)
+ nativeScanBuilder.setCommon(commonBuilder.build())
+ Some(builder.setNativeScan(nativeScanBuilder).build())
+ case None =>
+ // There are unsupported scan type
+ withFallbackReason(
+ scan,
+ s"unsupported Comet operator: ${scan.nodeName}, due to unsupported
data types above")
+ None
+ }
+ }
+
+ /**
+ * Build the `NativeScanCommon` proto shared by the core parquet scan and
contrib scans that
+ * delegate to the same native parquet machinery (e.g. a Delta scan contrib,
which passes
+ * physical-name schemas under column mapping). Returns `None` when an
output data type cannot
+ * be serialized; the caller is responsible for tagging a fallback reason.
+ *
+ * Visibility note: `private[comet]` means a contrib caller must live under
an
+ * `org.apache.comet.*` package (the same constraint `PlanDataInjector`
implementers have).
+ */
+ private[comet] def buildNativeScanCommon(
+ source: String,
+ output: Seq[Attribute],
+ requiredSchema: StructType,
+ dataSchema: StructType,
+ partitionSchema: StructType,
+ fileConstantMetadataColumns: Seq[AttributeReference],
+ dataFilters: Seq[Expression],
+ firstFileUri: Option[URI],
+ hadoopConf: Configuration,
+ conf: SQLConf): Option[OperatorOuterClass.NativeScanCommon.Builder] = {
val commonBuilder = OperatorOuterClass.NativeScanCommon.newBuilder()
// Set source in common (used as part of injection key)
- commonBuilder.setSource(scan.simpleStringWithNodeId())
+ commonBuilder.setSource(source)
- val scanTypes = scan.output.flatten { attr =>
+ val scanTypes = output.flatten { attr =>
serializeDataType(attr.dataType)
}
- if (scanTypes.length == scan.output.length) {
- commonBuilder.addAllFields(scanTypes.asJava)
-
- // Sink operators don't have children
- builder.clearChildren()
-
- if (scan.conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) {
- val dataFilters = new ListBuffer[Expr]()
- for (filter <- scan.supportedDataFilters) {
- exprToProto(filter, scan.output) match {
- case Some(proto) => dataFilters += proto
- case _ =>
- logWarning(s"Unsupported data filter $filter")
- }
+ if (scanTypes.length != output.length) {
+ // There are unsupported scan types
+ return None
+ }
+ commonBuilder.addAllFields(scanTypes.asJava)
+
+ if (conf.getConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED)) {
+ val filterProtos = new ListBuffer[Expr]()
+ for (filter <- dataFilters) {
+ exprToProto(filter, output) match {
+ case Some(proto) => filterProtos += proto
+ case _ =>
+ logWarning(s"Unsupported data filter $filter")
}
- commonBuilder.addAllDataFilters(dataFilters.asJava)
- }
-
- val possibleDefaultValues =
getExistenceDefaultValues(scan.requiredSchema)
- if (possibleDefaultValues.exists(_ != null)) {
- // Our schema has default values. Serialize two lists, one with the
default values
- // and another with the indexes in the schema so the native side can
map missing
- // columns to these default values.
- val (defaultValues, indexes) =
possibleDefaultValues.iterator.zipWithIndex
- .filter { case (expr, _) => expr != null }
- .map { case (expr, index) =>
- // ResolveDefaultColumnsUtil.getExistenceDefaultValues has
evaluated these
- // expressions and they should now just be literals.
- (Literal(expr), index.toLong.asInstanceOf[java.lang.Long])
- }
- .toList
- .unzip
- commonBuilder.addAllDefaultValues(
- defaultValues.flatMap(exprToProto(_, scan.output)).asJava)
- commonBuilder.addAllDefaultValuesIndexes(indexes.asJava)
}
+ commonBuilder.addAllDataFilters(filterProtos.asJava)
+ }
- // Extract object store options from first file (S3 configs apply to all
files in scan).
- // Use selectedPartitions (static) instead of getFilePartitions()
because at planning time
- // DPP subqueries haven't been resolved yet. Object store options don't
depend on DPP.
- val firstFileUri = scan.selectedPartitions
- .flatMap(_.files.headOption)
- .headOption
- .map(_.getPath.toUri)
-
- // Constant metadata columns (file_path, file_name, file_size,
file_block_start,
- // file_block_length, file_modification_time) are known before opening
the file and
- // constant for every row read from it, exactly like partition columns.
Spark places
- // them immediately after partition columns in `scan.output`
- // (FileSourceStrategy.scala: readDataColumns ++
generatedMetadataColumns ++
- // partitionColumns ++ constantMetadataColumns), so appending them after
the real
- // partition schema here keeps the two in lockstep.
- val constantMetadataFields =
scan.wrapped.fileConstantMetadataColumns.map(attr =>
- StructField(s"$constantMetadataFieldPrefix${attr.name}",
attr.dataType, attr.nullable))
- val partitionSchemaFields = scan.relation.partitionSchema.fields.toSeq ++
- constantMetadataFields
- val partitionSchema = schema2Proto(partitionSchemaFields)
- val requiredSchema = schema2Proto(scan.requiredSchema)
-
- // Spark's required schema can prune a Variant column, including a
Variant nested under an
- // unrequested struct. The complete relation schema still contains that
unsupported type,
- // and serializing it would throw even though the native reader never
needs those bytes.
- // Keep ordinary fields unchanged and replace a requested
Variant-bearing root with its
- // already-validated, pruned required field. A requested actual Variant
never reaches this
- // point because CometScanRule keeps those scans on Spark.
- val nativeDataSchema =
StructType(scan.relation.dataSchema.fields.flatMap { field =>
- if (containsVariantType(field.dataType)) {
- scan.requiredSchema.fields.find(requiredField =>
- scan.conf.resolver(requiredField.name, field.name))
- } else {
- Some(field)
- }
- })
- val dataSchema = schema2Proto(nativeDataSchema)
-
- val dataSchemaIndexes = scan.requiredSchema.map(field => {
- nativeDataSchema.fieldIndex(field.name)
- })
- val partitionSchemaIndexes = nativeDataSchema.fields.length until
- (nativeDataSchema.length + partitionSchemaFields.length)
-
- val projectionVector = (dataSchemaIndexes ++
partitionSchemaIndexes).map(idx =>
- idx.toLong.asInstanceOf[java.lang.Long])
-
- commonBuilder.addAllProjectionVector(projectionVector.asJava)
-
- // In `CometScanRule`, we ensure partitionSchema (including constant
metadata columns)
- // is supported.
- assert(partitionSchema.length == partitionSchemaFields.length)
-
- commonBuilder.addAllDataSchema(dataSchema.asJava)
- commonBuilder.addAllRequiredSchema(requiredSchema.asJava)
- commonBuilder.addAllPartitionSchema(partitionSchema.asJava)
-
commonBuilder.setSessionTimezone(scan.conf.getConfString("spark.sql.session.timeZone"))
-
commonBuilder.setCaseSensitive(scan.conf.getConf[Boolean](SQLConf.CASE_SENSITIVE))
-
- // SPARK-53535 (Spark 4.1+): when reading a struct whose requested
fields are all
- // missing in the Parquet file, the new default preserves the parent
struct's
- // nullness from the file (so non-null parents materialize as a struct
of all-null
- // fields). Pre-4.1 Spark hardcodes the legacy behavior (whole struct
null), which
- // matches the Comet default we use as fallback.
- val returnNullStructConfKey =
- "spark.sql.legacy.parquet.returnNullStructIfAllFieldsMissing"
- val returnNullStructDefault = if (isSpark41Plus) "false" else "true"
- commonBuilder.setReturnNullStructIfAllFieldsMissing(
- scan.conf.getConfString(returnNullStructConfKey,
returnNullStructDefault).toBoolean)
-
- // Field-ID matching: only ask the native side to do extra work when the
conf is on AND
- // the requested schema actually carries IDs. Spark's ParquetReadSupport
applies the same
- // gate before invoking matchIdField.
- val useFieldId =
- scan.conf.getConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED) &&
- ParquetUtils.hasFieldIds(scan.requiredSchema)
- commonBuilder.setUseFieldId(useFieldId)
- commonBuilder.setIgnoreMissingFieldId(
- scan.conf.getConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID))
-
-
commonBuilder.setAllowTypePromotion(CometConf.COMET_SCHEMA_EVOLUTION_ENABLED)
-
commonBuilder.setAllowTimestampLtzToNtz(CometConf.COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ)
-
- // Collect S3/cloud storage configurations
- val hadoopConf = scan.relation.sparkSession.sessionState
- .newHadoopConfWithOptions(scan.relation.options)
-
-
commonBuilder.setEncryptionEnabled(CometParquetUtils.encryptionEnabled(hadoopConf))
-
- firstFileUri.foreach { uri =>
- val objectStoreOptions =
- NativeConfig.extractObjectStoreOptions(hadoopConf, uri)
- objectStoreOptions.foreach { case (key, value) =>
- commonBuilder.putObjectStoreOptions(key, value)
+ val possibleDefaultValues = getExistenceDefaultValues(requiredSchema)
+ if (possibleDefaultValues.exists(_ != null)) {
+ // Our schema has default values. Serialize two lists, one with the
default values
+ // and another with the indexes in the schema so the native side can map
missing
+ // columns to these default values.
+ val (defaultValues, indexes) =
possibleDefaultValues.iterator.zipWithIndex
+ .filter { case (expr, _) => expr != null }
+ .map { case (expr, index) =>
+ // ResolveDefaultColumnsUtil.getExistenceDefaultValues has evaluated
these
+ // expressions and they should now just be literals.
+ (Literal(expr), index.toLong.asInstanceOf[java.lang.Long])
}
+ .toList
+ .unzip
+ commonBuilder.addAllDefaultValues(defaultValues.flatMap(exprToProto(_,
output)).asJava)
+ commonBuilder.addAllDefaultValuesIndexes(indexes.asJava)
+ }
+
+ // Constant metadata columns (file_path, file_name, file_size,
file_block_start,
+ // file_block_length, file_modification_time) are known before opening the
file and
+ // constant for every row read from it, exactly like partition columns.
Spark places
+ // them immediately after partition columns in the scan output
+ // (FileSourceStrategy.scala: readDataColumns ++ generatedMetadataColumns
++
+ // partitionColumns ++ constantMetadataColumns), so appending them after
the real
+ // partition schema here keeps the two in lockstep.
+ val constantMetadataFields = uniqueConstantMetadataFields(
+ fileConstantMetadataColumns,
+ dataSchema.fields.map(_.name).toSet ++
partitionSchema.fields.map(_.name).toSet)
+ val partitionSchemaFields = partitionSchema.fields.toSeq ++
constantMetadataFields
+ val partitionSchemaProto = schema2Proto(partitionSchemaFields)
+ val requiredSchemaProto = schema2Proto(requiredSchema)
+
+ // Spark's required schema can prune a Variant column, including a Variant
nested under an
+ // unrequested struct. The complete relation schema still contains that
unsupported type,
+ // and serializing it would throw even though the native reader never
needs those bytes.
+ // Keep ordinary fields unchanged and replace a requested Variant-bearing
root with its
+ // already-validated, pruned required field. A requested actual Variant
never reaches this
+ // point because CometScanRule keeps those scans on Spark.
+ val nativeDataSchema = StructType(dataSchema.fields.flatMap { field =>
+ if (containsVariantType(field.dataType)) {
+ requiredSchema.fields.find(requiredField =>
conf.resolver(requiredField.name, field.name))
+ } else {
+ Some(field)
}
+ })
+ val dataSchemaProto = schema2Proto(nativeDataSchema)
- // Set common data in NativeScan (file_partition will be populated at
execution time)
- nativeScanBuilder.setCommon(commonBuilder.build())
+ val dataSchemaIndexes = requiredSchema.map(field => {
+ nativeDataSchema.fieldIndex(field.name)
+ })
+ val partitionSchemaIndexes = nativeDataSchema.fields.length until
+ (nativeDataSchema.length + partitionSchemaFields.length)
- Some(builder.setNativeScan(nativeScanBuilder).build())
+ val projectionVector = (dataSchemaIndexes ++
partitionSchemaIndexes).map(idx =>
+ idx.toLong.asInstanceOf[java.lang.Long])
- } else {
- // There are unsupported scan type
- withFallbackReason(
- scan,
- s"unsupported Comet operator: ${scan.nodeName}, due to unsupported
data types above")
- None
+ commonBuilder.addAllProjectionVector(projectionVector.asJava)
+
+ // In `CometScanRule`, we ensure partitionSchema (including constant
metadata columns)
+ // is supported.
+ assert(partitionSchemaProto.length == partitionSchemaFields.length)
+
+ commonBuilder.addAllDataSchema(dataSchemaProto.asJava)
+ commonBuilder.addAllRequiredSchema(requiredSchemaProto.asJava)
+ commonBuilder.addAllPartitionSchema(partitionSchemaProto.asJava)
+
+ populateScanConfFlags(commonBuilder, requiredSchema, firstFileUri,
hadoopConf, conf)
+
+ Some(commonBuilder)
+ }
+
+ /**
+ * Populate the configuration-derived flags of a `NativeScanCommon`: session
timezone, case
+ * sensitivity, struct-nullness legacy flag, field-ID matching, type
promotion, encryption, and
+ * object-store options. Shared with contrib scans that assemble their own
schemas/projection
+ * (e.g. the Delta contrib's deletion-vector shape) so new flags added here
reach them without
+ * drift.
+ */
+ private[comet] def populateScanConfFlags(
+ commonBuilder: OperatorOuterClass.NativeScanCommon.Builder,
+ requiredSchema: StructType,
+ firstFileUri: Option[URI],
+ hadoopConf: Configuration,
+ conf: SQLConf): Unit = {
+
commonBuilder.setSessionTimezone(conf.getConfString("spark.sql.session.timeZone"))
+ val caseSensitive = conf.getConf[Boolean](SQLConf.CASE_SENSITIVE)
+ commonBuilder.setCaseSensitive(caseSensitive)
+ if (!caseSensitive) {
+ // Ship THIS JVM's case data so native's case-insensitive footer
matching reproduces
+ // this JVM's `toLowerCase(Locale.ROOT)` exactly, whatever Unicode
version it bundles.
+ JvmCaseTables.populate(commonBuilder)
Review Comment:
Attaching the tables here means every `NativeScanCommon` carries its own
copy with no dedup across scans in a plan. I measured 18,801 bytes per scan.
Running `select k1, d1 from t1` repeated under `UNION ALL` and summing the
serialized native blocks:
| scans | `caseSensitive=true` | `caseSensitive=false` (the default) |
|---|---|---|
| 1 | 165 B | 18,966 B |
| 2 | 330 B | 37,932 B |
| 3 | 495 B | 56,898 B |
That payload rides in the Spark task binary for every stage, so a 20-table
query would ship around 380 KB of identical tables.
The change to `QueryContextInternerSuite` in this PR is what led me here.
That suite asserts `before > after * 4` to protect the SQL-text interning
optimization, and pinning it to `spark.sql.caseSensitive=true` is what keeps it
passing. On the default config it fails: I get `before=33316 after=21767
ratio=1.53` against `before=14513 after=2964 ratio=4.90` with case sensitivity
on. In other words the plan is now larger than it was before interning landed,
and the suite no longer covers the default configuration.
The tables are a pure function of the JDK and identical for every scan in a
plan. Could they be interned at the root `Operator` the same way
`sql_text_pool` already is, so a plan pays once rather than once per scan?
Better still, sending them to each executor once at plugin init keyed by the
fingerprint would take the per-plan cost to zero and let
`QueryContextInternerSuite` go back to testing the default config.
##########
contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala:
##########
@@ -0,0 +1,1707 @@
+/*
+ * 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.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")
+ }
+
+ // 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 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
+ }
+
+ // 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]],
+ * 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"
+
+ /**
+ *
`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(
+ "fs.s3a.access.key" -> LookupPasswordConsumer,
+ "fs.s3a.secret.key" -> LookupPasswordConsumer,
+ "fs.s3a.session.token" -> LookupPasswordConsumer,
+ "fs.s3a.aws.credentials.provider" -> PropagatedOptionConsumer,
+ "fs.s3a.assumed.role.arn" -> PropagatedOptionConsumer,
+ "fs.s3a.assumed.role.session.name" -> PropagatedOptionConsumer,
+ "fs.s3a.assumed.role.credentials.provider" -> PropagatedOptionConsumer,
+ "fs.s3a.endpoint" -> PropagatedOptionConsumer,
+ "fs.s3a.endpoint.region" -> PropagatedOptionConsumer,
+ "fs.s3a.path.style.access" -> PropagatedOptionConsumer,
+ "fs.s3a.requester.pays.enabled" -> PropagatedOptionConsumer,
+ "fs.s3a.comet.credential.provider.class" -> PropagatedOptionConsumer)
+
+ /** The compared keys alone, in [[S3ConfigKeyConsumers]] order
(discovery-harness surface). */
+ private[delta] val AllS3ConfigKeys: Seq[String] =
S3ConfigKeyConsumers.map(_._1)
+
+ /**
+ * The short-bucket-then-global value resolved for `baseKey` under `bucket`
from `hadoopConf`,
+ * skipping an empty value at either alias exactly like [[plainValue]]. NOT
used by
+ * [[s3ConfigDivergenceReason]]/[[s3KeyDivergenceReason]] any more -- every
key checked there
+ * resolves per its declared [[S3ConfigKeyConsumers]] tier (see
[[s3KeyDivergenceReason]]),
+ * neither of which matches this function's read. This function's one
remaining caller is
+ * [[shortThenGlobalOrReason]], which reads provider-CLASS strings (from the
ORIGINAL,
+ * unpropagated conf) for name-support validation in [[providerClassReason]]/
+ * [[assumedRoleProviderClassReason]] -- by the time those run,
[[s3ConfigDivergenceReason]] has
+ * already proven Hadoop's and native's effective values agree for the same
key, so whichever of
+ * the two (equal) values this narrower read returns does not affect
correctness there. NEVER
+ * used to compute native's own effective value -- 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
+ * [[hadoopLookupPasswordEffective]]), 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 exact strings `S3AEncryptionMethods#getMethod` accepts, verified via
`javap`/CFR against
+ * `hadoop-aws` 3.3.4's `S3AEncryptionMethods` enum: `NONE("")`,
`SSE_S3("AES256", serverSide =
+ * true, requiresSecret = false)`, `SSE_KMS("SSE-KMS", serverSide = true,
requiresSecret =
+ * false)`, `SSE_C("SSE-C", serverSide = true, requiresSecret = true)`,
`CSE_KMS("CSE-KMS",
+ * serverSide = false, requiresSecret = true)`, `CSE_CUSTOM("CSE-CUSTOM",
serverSide = false,
+ * requiresSecret = true)`. `getMethod` parses case-insensitively
+ * (`values().find(_.getMethod.equalsIgnoreCase(algorithm))`), matched below
the same way.
+ *
+ * ALLOWLIST, not a blocklist (replaces the former SSE-C-only blocklist):
only the algorithms S3
+ * decrypts transparently on GET/HEAD given read permission alone, with NO
extra request header
+ * and NO client-side step, are safe for a native scan that forwards none of
Hadoop's
+ * `fs.s3a.encryption.*`/`fs.s3a.server-side-encryption*` options --
+ * - `AES256` (SSE_S3, `serverSide = true`): plain server-side encryption,
transparent on GET.
+ * - `SSE-KMS` (SSE_KMS, `serverSide = true`): server-side, KMS-managed
key, transparent on
+ * GET given KMS decrypt permission (no header).
+ * - `DSSE-KMS`: NOT present in this enum on `hadoop-aws` 3.3.4 (confirmed
by the six values
+ * listed above) -- `S3AEncryptionMethods.getMethod("DSSE-KMS")` throws
+ * `IOException("Unknown encryption algorithm DSSE-KMS")` on this
version, so
+ * `S3AUtils#buildEncryptionSecrets` (and therefore Hadoop's own reader)
already fails
+ * before ever reading such a table under 3.3.4, meaning this string can
never actually be
+ * the resolved value on the declared target version -- admitting it
here is inert there.
+ * Included anyway, forward-compatible, for a newer `hadoop-aws` on the
runtime classpath (a
+ * later Hadoop release; this module has no compile-time `hadoop-aws`
dependency, see the
+ * string-literal-keys note above) where DSSE-KMS is a real, dual-layer,
server-side
+ * algorithm decrypted transparently on GET the same way SSE-KMS is.
Every other value
+ * declines: `SSE-C` (SSE_C is `serverSide = true` in Hadoop's own enum,
but `requiresSecret
+ * \= true` -- 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 --
whose request factory
+ * attaches the key -- would), `CSE-KMS`/`CSE-CUSTOM` (`serverSide =
false`: client-side
+ * encryption decrypts object bytes locally in the SDK layer, which the
native Parquet
+ * reader has no equivalent of -- it would read raw ciphertext), and any
future/unknown
+ * value (a value `S3AEncryptionMethods.getMethod` itself would reject
is certainly not one
+ * of the three confirmed-transparent algorithms above; declining is the
only safe default
+ * for anything this gate cannot positively confirm).
+ */
+ private val AllowedEncryptionAlgorithms: Set[String] = Set("AES256",
"SSE-KMS", "DSSE-KMS")
+
+ /**
+ * `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`/CFR
+ * decompilation of `hadoop-aws` 3.3.4's `S3AUtils.class`:
+ * {{{
+ * String algorithm = lookupBucketSecret(bucket, conf,
"fs.s3a.encryption.algorithm");
+ * if (algorithm == null)
+ * algorithm = lookupBucketSecret(bucket, conf,
"fs.s3a.server-side-encryption-algorithm");
+ * if (algorithm == null)
+ * algorithm = lookupPassword(null, conf, "fs.s3a.encryption.algorithm");
+ * if (algorithm == null)
+ * algorithm = lookupPassword(null, conf,
"fs.s3a.server-side-encryption-algorithm");
+ * }}}
+ * i.e. bucket-tier (canonical, then deprecated), THEN global-tier
(canonical, then deprecated)
+ * -- the two tiers are never interleaved key-by-key, so this must stay two
explicit bucket-tier
+ * lookups followed by two explicit global-tier lookups, not a single
+ * [[hadoopLookupPasswordEffective]] call per key (which would let an unset
canonical bucket key
+ * fall through straight to the canonical GLOBAL value ahead of a SET
deprecated bucket key, the
+ * wrong answer).
+ *
+ * THE FIX for the SSE-C long-bucket-alias gap is entirely inside the bucket
tier:
+ * `lookupBucketSecret` itself is long-then-short, decompiled from
`hadoop-aws` 3.3.4's
+ * `S3AUtils.class`:
+ * {{{
+ * // longBucketKey = fs.s3a.bucket.B.fs.s3a.<key>
+ * String longBucketKey = String.format(BUCKET_PATTERN, bucket, baseKey);
+ * String initialVal = getPassword(conf, longBucketKey, null, null);
+ * // shortBucketKey = fs.s3a.bucket.B.<key>
+ * String shortBucketKey = String.format(BUCKET_PATTERN, bucket, subkey);
+ * // keeps initialVal (the LONG value) if non-empty
+ * return getPassword(conf, shortBucketKey, initialVal, null);
+ * }}}
+ * i.e. the SAME long-bucket-key construction and long-wins-if-nonempty
semantics as
+ * `S3AUtils#lookupPassword` (see
[[LookupPasswordConsumer]]/[[hadoopLookupPasswordEffective]])
+ * -- the encryption algorithm is NOT one of the keys that flows through
+ * `S3AUtils#propagateBucketOptions` (which folds an unrelated per-bucket
LONG form into an
+ * unread key). An earlier version of this function modeled the bucket tier
as SHORT-only,
+ * documented as "the LONG bucket form is genuinely never consulted for this
key" -- that
+ * documentation was wrong (this decompilation supersedes it): a bucket
configured only via
+ * `fs.s3a.bucket.B.fs.s3a.encryption.algorithm=SSE-C` bypassed the SSE-C
gate entirely, because
+ * Hadoop's own reader DOES read that long form (and picks SSE-C), while
this function reported
+ * `None` (nothing set) and the allowlist check below never even ran.
+ *
+ * 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-key cascade below redundant (but
harmless) for that case;
+ * it remains the operative path only when nothing else in the process has
loaded
+ * `S3AFileSystem` yet.
+ *
+ * ALSO walks the Hadoop-credential-provider (JCEKS) path via
[[resolveViaCredentialAliases]]
+ * for each of the four lookups below, matching
`lookupBucketSecret`/`lookupPassword`'s real
+ * per-alias `getPassword` calls (quoted above) exactly: both are
`getPassword`, not plain
+ * `Configuration#get`, so a bucket storing the algorithm name ONLY in a
JCEKS keystore is
+ * exactly as real a Hadoop deployment shape for this key as it is for the
credential keys
+ * [[hadoopLookupPasswordEffective]] already covers -- there is nothing
algorithm-specific that
+ * makes JCEKS storage implausible here, so an earlier version of this
function skipping it
+ * (documented at the time as "the algorithm NAME is not
credential-sensitive data, so storing
+ * it in a keystore is not a realistic Hadoop deployment pattern") was an
unjustified, narrower
+ * read than Hadoop's own resolver actually performs, under-declining a
bucket whose algorithm
+ * is keystore-only. [[resolveViaCredentialAliases]]'s Arm B/C split still
means this is zero
+ * extra I/O for the common case: keystore I/O only happens when a Hadoop
credential-provider
+ * path is actually configured for the bucket, contained in that function's
own try/catch.
+ * `bucketTier`/`globalTier` return `Left` (propagated straight through by
[[orElseTier]]) when
+ * [[resolveViaCredentialAliases]] cannot safely verify a tier at all (an
S3A-scoped provider
+ * path, or a corrupt/unreadable global keystore) -- correctly
short-circuiting the whole
+ * cascade with a decline rather than silently falling through to a later
tier that might look
+ * unset only because the true value was unverifiable.
+ */
+ private def effectiveEncryptionAlgorithm(
+ hadoopConf: Configuration,
+ bucket: String): Either[String, Option[(String, String)]] = {
+ def bucketTier(baseKey: String): Either[String, Option[(String, String)]]
= {
+ val longKey = s"fs.s3a.bucket.$bucket.$baseKey"
+ val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.")
+ resolveViaCredentialAliases(hadoopConf, bucket, Seq(longKey, shortKey))
+ .map(_.map(baseKey -> _))
+ }
+ def globalTier(baseKey: String): Either[String, Option[(String, String)]] =
+ resolveViaCredentialAliases(hadoopConf, bucket, Seq(baseKey))
+ .map(_.map(baseKey -> _))
+
+ // Short-circuits on Left (unverifiable tier) or Right(Some(_))
(resolved); only Right(None)
+ // (tier definitively unset) falls through to `next`, mirroring
buildEncryptionSecrets's
+ // sequential `if (algorithm == null) algorithm = ...` cascade exactly.
+ def orElseTier(
+ current: Either[String, Option[(String, String)]],
+ next: => Either[String, Option[(String, String)]])
+ : Either[String, Option[(String, String)]] =
+ current match {
+ case Left(reason) => Left(reason)
+ case Right(Some(value)) => Right(Some(value))
+ case Right(None) => next
+ }
+
+ orElseTier(
+ bucketTier(S3EncryptionAlgorithmKey),
+ orElseTier(
+ bucketTier(DeprecatedS3EncryptionAlgorithmKey),
+ orElseTier(
+ globalTier(S3EncryptionAlgorithmKey),
+ globalTier(DeprecatedS3EncryptionAlgorithmKey))))
+ }
+
+ private def unsupportedEncryptionAlgorithmDeclineReason(
+ bucket: String,
+ algorithmKey: String,
+ algorithm: String): String =
+ s"Native Delta scan does not support $algorithmKey=$algorithm for $bucket
" +
+ "(the native S3 client only supports unencrypted objects and S3's
transparent " +
+ "server-side algorithms -- AES256/SSE-S3, SSE-KMS, and DSSE-KMS decrypt
on GET/HEAD given " +
+ "read permission alone, with no extra request header; SSE-C additionally
requires the " +
+ "customer-provided key resent as a header on every GET/HEAD request,
which the native S3 " +
+ "client's extract_s3_config_options never forwards, and
CSE-KMS/CSE-CUSTOM decrypt object " +
+ "bytes client-side, a layer the native Parquet reader does not have --
any of these would " +
+ "fail outright or silently read ciphertext where Hadoop's own reader
succeeds)"
+
+ /**
+ * First reason any bucket among `uris` is configured for an encryption
algorithm the native S3
+ * client cannot safely read, or `None` when claimable. Allowlist-based (see
+ * [[AllowedEncryptionAlgorithms]]): only `AES256`/`SSE-KMS`/`DSSE-KMS` (and
unset/empty) pass;
+ * every other resolved value -- `SSE-C`, `CSE-KMS`, `CSE-CUSTOM`, or any
unrecognized future
+ * algorithm string -- declines. Deliberately NOT a blocklist keyed on
`SSE-C` alone: an
+ * allowlist is safe by construction against a Hadoop release adding a new
encryption method
+ * this gate has never heard of, where a blocklist would silently admit it.
Never interpolates a
+ * resolved key value, only key names, the bucket, and the (non-secret)
algorithm name. Declines
+ * on a `Left` from [[effectiveEncryptionAlgorithm]] too (an unverifiable
credential-provider
+ * arm, e.g. an S3A-scoped provider path or a corrupt/unreadable global
keystore) -- the
+ * algorithm cannot be ruled safe when it cannot be read at all.
+ */
+ private[delta] def unsupportedEncryptionAlgorithmReason(
+ 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) match {
+ case Left(reason) => Some(reason)
+ case Right(None) => None
+ case Right(Some((key, value))) =>
+ if
(!AllowedEncryptionAlgorithms.exists(_.equalsIgnoreCase(value))) {
+ Some(unsupportedEncryptionAlgorithmDeclineReason(bucket, key,
value))
+ } else {
+ None
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Canonical Hadoop S3A HTTP-proxy host config key, verified via CFR
decompilation of
+ * `hadoop-aws` 3.3.4's `S3AUtils.class` (`initProxySupport`):
+ * {{{
+ * String proxyHost = conf.getTrimmed("fs.s3a.proxy.host", "");
+ * int proxyPort = conf.getInt("fs.s3a.proxy.port", -1);
+ * if (!proxyHost.isEmpty()) {
+ * ...
+ * String proxyUsername =
+ * S3AUtils.lookupPassword(bucket, conf, "fs.s3a.proxy.username",
null, null);
+ * String proxyPassword =
+ * S3AUtils.lookupPassword(bucket, conf, "fs.s3a.proxy.password",
null, null);
+ * ...
+ * }
+ * }}}
+ * `fs.s3a.proxy.host`/`fs.s3a.proxy.port` resolve via a PLAIN,
non-bucket-scoped, non-JCEKS
+ * `Configuration#getTrimmed`/`getInt` call -- NOT `lookupPassword` --
against whatever conf
+ * `S3AFileSystem#initialize` already ran through `propagateBucketOptions`
before
+ * `createAwsConf`/`initProxySupport` ever runs; only the SIBLING
`fs.s3a.proxy.username`/
+ * `fs.s3a.proxy.password` keys go through `lookupPassword` (bucket
long/short/global,
+ * JCEKS-aware). So the host is bucket-aware only via
`propagateBucketOptions`'s short-bucket-
+ * form folding, never the long-bucket form, and never a credential-provider
read -- the SAME
+ * shape as `endpoint`/`path.style.access` ([[PropagatedOptionConsumer]], see
+ * [[S3ConfigKeyConsumers]]'s doc), not the credential family. `hadoop-aws`
3.4.x (the Spark 4.x
+ * profiles' version) moves this code to
`AWSClientConfig#createProxyConfiguration`/
+ * `#createAsyncProxyConfiguration` but keeps the exact same reads, verified
via `javap` against
+ * 3.4.2: `conf.getTrimmed("fs.s3a.proxy.host", "")` for the host,
`S3AUtils.lookupPassword` for
+ * username/password only.
+ *
+ * [[proxyGateReason]] therefore resolves the host EXACTLY like its real
consumer -- plain
+ * `Configuration#getTrimmed` on the [[propagateBucketOptions]] result, no
provider arms, no
+ * [[ClearTextFallbackKey]] handling -- rather than through the wider
`lookupPassword` cascade
+ * an earlier version reused here. The wider cascade was wrong in BOTH
directions for this key:
+ * with only the GLOBAL Hadoop provider path set and
[[ClearTextFallbackKey]] false,
+ * `getPassword` hides a plaintext host that `getTrimmed` serves to Hadoop
anyway (a missed
+ * decline, the exact bypass this gate exists to close -- native has NO
HTTP-proxy support of
+ * any kind, no `fs.s3a.proxy.*` key is read anywhere in `s3.rs`); and an
S3A-scoped provider
+ * path or a lone long-form bucket alias declined a bucket whose real
consumer can never see a
+ * host from either source (pure over-refusal -- no keystore can supply the
host to a plain
+ * `getTrimmed`, and the long form folds into the unread
`fs.s3a.fs.s3a.proxy.host`).
+ */
+ private val S3ProxyHostKey = "fs.s3a.proxy.host"
+
+ private def unsupportedProxyReason(bucket: String, key: String): String =
+ s"Native Delta scan does not support $key configured for $bucket (the
native S3 client has " +
+ "no HTTP proxy support at all -- no fs.s3a.proxy.* key is read anywhere
in its object " +
+ "store layer -- so a claimed scan would connect to S3 directly instead
of routing through " +
+ "the configured proxy, either bypassing an egress/network-segmentation
policy or simply " +
+ "failing to reach the endpoint)"
+
+ /**
+ * First reason any bucket among `uris` has an HTTP proxy configured via
[[S3ProxyHostKey]], or
+ * `None` when claimable. Reads the host EXACTLY like its real consumer (see
+ * [[S3ProxyHostKey]]'s doc): plain `Configuration#getTrimmed` on the
[[propagateBucketOptions]]
+ * result, so this gate is always zero-I/O -- no credential provider is ever
consulted for the
+ * host, because none ever supplies it to Hadoop either. Ordered alongside
+ * [[unsupportedEncryptionAlgorithmReason]] among the conf-only gates, ahead
of
+ * [[s3ConfigDivergenceReason]]. The try/catch guards `Configuration#get`'s
+ * `IllegalStateException` on a `${...}` substitution cycle, same as
[[s3KeyDivergenceReason]].
+ * Never interpolates a resolved value: the proxy HOST is not secret, but
naming it here would
+ * be a strange place to first surface it, and proxy CREDENTIALS
+ * (`fs.s3a.proxy.username`/`fs.s3a.proxy.password`, not read by this gate
at all -- the whole
+ * point of gating on the host is that a non-empty host declines before any
proxy credential
+ * would ever need to be forwarded) must never appear in a decline reason
regardless.
+ */
+ private[delta] def proxyGateReason(
+ hadoopConf: Configuration,
+ uris: Seq[URI],
+ propagatedConfCache: MutableMap[String, Configuration] =
MutableMap.empty)
+ : Option[String] = {
+ val buckets = uris.flatMap(s3Bucket).distinct
+ buckets.foldLeft(Option.empty[String]) { (declined, bucket) =>
+ if (declined.isDefined) {
+ declined
+ } else {
+ try {
+ val propagatedConf =
+ propagatedConfCache.getOrElseUpdate(
+ bucket,
+ propagateBucketOptions(hadoopConf, bucket))
+ if (propagatedConf.getTrimmed(S3ProxyHostKey, "").nonEmpty) {
+ Some(unsupportedProxyReason(bucket, S3ProxyHostKey))
+ } else {
+ None
+ }
+ } catch {
+ case e @ (_: IOException | _: RuntimeException) =>
+ Some(unverifiableValueReason(bucket, S3ProxyHostKey, e))
+ }
+ }
+ }
+ }
+
+ /**
+ * `baseKey`'s long, short, then global per-bucket aliases, in Hadoop's own
resolution order.
+ */
+ private def longThenShortThenGlobalAliases(bucket: String, baseKey: String):
Seq[String] = {
+ val suffix = baseKey.stripPrefix("fs.s3a.")
+ Seq(s"fs.s3a.bucket.$bucket.fs.s3a.$suffix",
s"fs.s3a.bucket.$bucket.$suffix", baseKey)
+ }
+
+ 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 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"
+
+ /**
+ * The three-way Hadoop credential-provider-path precheck shared by every
`getPassword`-based
+ * resolution below, factored out of what used to be
[[hadoopLookupPasswordEffective]]'s own
Review Comment:
A few of the doc comments in this file describe how the code got here rather
than what it does. This one refers to `hadoopLookupPasswordEffective`, and
there are similar ones at 1139 ("Generalizes what used to be"), 1415
("previously relied on `s3ConfigDivergenceReason` running first"), and 1489
("the variable-reference precheck this used to run ahead of the class-support
check is gone"). `DeltaScanContribSuite.scala:37` has one too, referring to
"core's now-deleted extension-SPI".
Some of those symbols are not in the tree, so a reader who greps for them
comes up empty, and the commits are squashed so the history is not recoverable
there either. Could these be trimmed down to what the code does now, with any
load-bearing context moved to the PR description?
##########
native/core/src/parquet/schema_adapter.rs:
##########
@@ -76,6 +77,583 @@ fn schema_has_field_ids(schema: &SchemaRef) -> bool {
schema.fields().iter().any(|f| parse_field_id(f).is_some())
}
+//
---------------------------------------------------------------------------------------------
+// JVM-shipped case tables: reproduce the PLANNING JVM's
`String.toLowerCase(Locale.ROOT)`,
+// which Spark's Parquet footer field matching is built on.
+//
+// The data arrives on `NativeScanCommon` (populated by `JvmCaseTables.scala`
when
+// case_sensitive = false), generated from the very JVM that plans the query,
so the native
+// matcher is correct BY CONSTRUCTION for whatever JDK runs Spark. The one
contextual mapping
+// (`Locale.ROOT` has exactly one: Greek capital sigma U+03A3) cannot be a
per-codepoint table
+// entry, so its condition is ported as an algorithm over a shipped
per-codepoint
+// classification -- see `JvmCaseTables::lowercase`.
+//
---------------------------------------------------------------------------------------------
+
+// Sigma-scan classes: the wire contract shared with `JvmCaseTables.scala`.
The classes are
+// the UAX#29-style word-break classes (ALetter, Numeric, MidLetter, MidNum,
MidNumLet,
+// Extend, Format) as the PLANNING JVM's legacy break iterator actually
realizes them --
+// probed per codepoint from `BreakIterator.isBoundary` on the JVM side --
plus classes for
+// its pre-UAX#29 extensions (the danda and the supplementary-plane behaviors
of its UTF-16
+// DFA), with cased variants split out via the JDK's `isCased`. Any codepoint
outside every
+// shipped range -- and any class value this build does not know -- is a word
boundary: the
+// sigma context scan stops there, which is also the safe reading for class
values added by a
+// NEWER JVM-side generator.
+const CLASS_ALETTER_CASED: u8 = 1;
+const CLASS_ALETTER: u8 = 2;
+const CLASS_NUMERIC: u8 = 3;
+const CLASS_MID_LETTER: u8 = 4;
+const CLASS_MID_NUM: u8 = 5;
+const CLASS_MID_NUM_LET: u8 = 6;
+/// Cased supplementary char: attaches to the preceding word and closes it
(and forms a word
+/// of its own at raw text start).
+const CLASS_SUPP_CASED: u8 = 7;
+/// U+0964/U+0965: word-terminal, chains only into digits.
+const CLASS_DANDA: u8 = 8;
+/// U+0345, the one cased combining mark: cased only when its run is attached
to a word.
+const CLASS_EXTEND_CASED: u8 = 9;
+/// Cased digit-base (Nl Roman numerals): joins like CLASS_ALETTER_CASED when
reached
+/// directly, but bridges only mid-num punctuation, never mid-letter.
+const CLASS_NUMERIC_CASED: u8 = 10;
+/// Non-cased Mn/Me marks: riders that attach only to genuine letter/digit
bases.
+const CLASS_EXTEND: u8 = 11;
+/// Word-forming non-cased supplementary letter: a genuine letter-base that
closes the word
+/// immediately after itself.
+const CLASS_SUPP_LETTER: u8 = 12;
+/// Cf format characters: fully transparent (WB4-style) -- deleted from the
sequence before
+/// the scans run, so a pure-format rider chain bridges mid punctuation
("AΣ-<ZWJ>b" is one
+/// word exactly like "AΣ-b").
+const CLASS_FORMAT: u8 = 13;
+/// Supplementary chars that attach to the preceding word but never form one
themselves
+/// (supplementary combining marks, tag characters): a cased mark riding on
one belongs to
+/// the sigma's word only when the run hangs off a real base
(`supp_mn_anchor`).
+const CLASS_SUPP_MN: u8 = 14;
+/// Word-forming supplementary digit: like CLASS_SUPP_LETTER except a riding
cased mark
+/// carries only across mid-num (digit-context) punctuation, never mid-letter.
+const CLASS_SUPP_NUM: u8 = 15;
+/// Not on the wire: the absence of a class.
+const CLASS_BOUNDARY: u8 = 0;
+
+const CAPITAL_SIGMA: char = '\u{03A3}';
+const SMALL_SIGMA: char = '\u{03C3}';
+const SMALL_FINAL_SIGMA: char = '\u{03C2}';
+
+/// What a supplementary-mark run ultimately hangs off (see `supp_mn_anchor`).
+#[derive(PartialEq, Eq, Clone, Copy)]
+enum SuppMnAnchor {
+ None,
+ Letter,
+ Digit,
+}
+
+/// The planning JVM's case data, parsed once per scan from `NativeScanCommon`
and attached to
+/// [`SparkParquetOptions`]. Two tables:
+///
+/// - `lower`: every codepoint the JVM lowercases non-identically, with its
full (possibly
+/// multi-char, e.g. U+0130 -> "i" + U+0307) replacement; codepoints
absent here lowercase
+/// to themselves;
+/// - `class_ranges`: sorted, disjoint `(start, end, class)` codepoint
ranges holding the
+/// word-break classification the JVM probed from its own `BreakIterator`.
+///
+/// `lowercase` applies Java's algorithm over that data: per codepoint, U+03A3
takes its
+/// contextual final/non-final form via the ported `isFinalCased` condition --
word-boundary
+/// based (the JDK's legacy break-iterator word rules), NOT the
Unicode-standard Final_Sigma
+/// case-ignorable skip, so e.g. "A1Σ" lowers to "a1ς" -- and every other
codepoint takes its
+/// table replacement. `JvmCaseTables.mirrorLowercase` on the Scala side is
the line-for-line
+/// mirror of this function over the same generated data; the JVM-side parity
suite proves the
+/// pair equal to the running JDK's `String.toLowerCase(Locale.ROOT)` across
the full codepoint
+/// space (calibrated to zero mismatches on JDK 17, 21, and 25).
+#[derive(Debug)]
+pub struct JvmCaseTables {
+ /// Non-identity lowercase mappings: codepoint -> full replacement string.
+ lower: HashMap<char, String>,
+ /// Sorted, disjoint (start, end, class) inclusive codepoint ranges for
the sigma scan.
+ class_ranges: Vec<(u32, u32, u8)>,
+ /// Precomputed content hash so `SparkParquetOptions`'s derived `Hash`
stays cheap.
+ fingerprint: u64,
+}
+
+impl PartialEq for JvmCaseTables {
+ fn eq(&self, other: &Self) -> bool {
+ self.fingerprint == other.fingerprint
+ && self.class_ranges == other.class_ranges
+ && self.lower == other.lower
+ }
+}
+
+impl Eq for JvmCaseTables {}
+
+impl Hash for JvmCaseTables {
+ fn hash<H: Hasher>(&self, state: &mut H) {
+ // Equal contents always produce the equal (deterministically
computed) fingerprint,
+ // so hashing only the fingerprint is consistent with `PartialEq`.
+ state.write_u64(self.fingerprint);
+ }
+}
+
+fn is_letter_base(cls: u8) -> bool {
+ cls == CLASS_ALETTER_CASED
+ || cls == CLASS_ALETTER
+ || cls == CLASS_SUPP_CASED
+ || cls == CLASS_SUPP_LETTER
+}
+
+fn is_digit_base(cls: u8) -> bool {
+ cls == CLASS_NUMERIC || cls == CLASS_NUMERIC_CASED
+}
+
+impl JvmCaseTables {
+ /// Parse the proto representation: `lower_cp`/`lower_repl` are
index-aligned, and
+ /// `class_ranges` holds (start, end, class) triples. Malformed input
(length mismatch,
+ /// trailing partial triple, out-of-range codepoints) is dropped
entry-by-entry rather
+ /// than rejected: every dropped entry degrades one codepoint to
identity/boundary
+ /// behavior instead of failing the scan.
+ pub fn from_proto(lower_cp: &[u32], lower_repl: &[String], class_ranges:
&[u32]) -> Self {
+ let mut lower = HashMap::with_capacity(lower_cp.len());
+ for (cp, repl) in lower_cp.iter().zip(lower_repl.iter()) {
+ if let Some(c) = char::from_u32(*cp) {
+ lower.insert(c, repl.clone());
+ }
+ }
+ let ranges: Vec<(u32, u32, u8)> = class_ranges
+ .as_chunks::<3>()
+ .0
+ .iter()
+ .filter(|t| t[0] <= t[1] && t[1] <= 0x10FFFF &&
u8::try_from(t[2]).is_ok())
+ .map(|t| (t[0], t[1], t[2] as u8))
+ .collect();
+
+ let mut hasher = DefaultHasher::new();
+ for (start, end, class) in &ranges {
+ hasher.write_u32(*start);
+ hasher.write_u32(*end);
+ hasher.write_u8(*class);
+ }
+ let mut lower_sorted: Vec<(&char, &String)> = lower.iter().collect();
+ lower_sorted.sort_by_key(|(c, _)| **c);
+ for (c, repl) in lower_sorted {
+ hasher.write_u32(*c as u32);
+ hasher.write(repl.as_bytes());
+ }
+
+ Self {
+ lower,
+ class_ranges: ranges,
+ fingerprint: hasher.finish(),
+ }
+ }
+
+ /// Sigma-scan class of `c`; `CLASS_BOUNDARY` when no shipped range covers
it.
+ fn class_of(&self, c: char) -> u8 {
+ let cp = c as u32;
+ let mut lo = 0usize;
+ let mut hi = self.class_ranges.len();
+ while lo < hi {
+ let mid = (lo + hi) / 2;
+ let (start, end, class) = self.class_ranges[mid];
+ if cp < start {
+ hi = mid;
+ } else if cp > end {
+ lo = mid + 1;
+ } else {
+ return class;
+ }
+ }
+ CLASS_BOUNDARY
+ }
+
+ /// First position at or beyond `start` (stepping by `step`, i.e. -1
backward / +1
+ /// forward) whose class is not CLASS_EXTEND. Returns `None` if the scan
runs off the
+ /// array without finding one.
+ fn skip_extends(&self, cps: &[char], start: isize, step: isize) ->
Option<usize> {
+ let mut k = start;
+ while k >= 0 && (k as usize) < cps.len() {
+ if self.class_of(cps[k as usize]) != CLASS_EXTEND {
+ return Some(k as usize);
+ }
+ k += step;
+ }
+ None
+ }
+
+ /// As [`Self::skip_extends`], but also skips CLASS_EXTEND_CASED,
reporting whether one
+ /// was walked: a cased mark (U+0345) crossed while looking for a base is
itself cased
+ /// whenever the landing validates the run.
+ fn skip_extends_tracking_cased(
+ &self,
+ cps: &[char],
+ start: isize,
+ step: isize,
+ ) -> (Option<usize>, bool) {
+ let mut k = start;
+ let mut saw_cased = false;
+ while k >= 0 && (k as usize) < cps.len() {
+ let cls = self.class_of(cps[k as usize]);
+ if cls != CLASS_EXTEND && cls != CLASS_EXTEND_CASED {
+ return (Some(k as usize), saw_cased);
+ }
+ if cls == CLASS_EXTEND_CASED {
+ saw_cased = true;
+ }
+ k += step;
+ }
+ (None, saw_cased)
+ }
+
+ /// What the supplementary-mark run at `k` (CLASS_SUPP_MN) ultimately
hangs off, walking
+ /// down through further marks and supplementary chars: a letter-flavored
base, a
+ /// digit-flavored base, or nothing word-forming. A cased mark riding the
run belongs to
+ /// the sigma's word only per this anchor.
+ fn supp_mn_anchor(&self, cps: &[char], k: usize) -> SuppMnAnchor {
+ let mut m = k as isize - 1;
+ while m >= 0 {
+ let cls = self.class_of(cps[m as usize]);
+ if cls != CLASS_SUPP_MN && cls != CLASS_EXTEND && cls !=
CLASS_EXTEND_CASED {
+ break;
+ }
+ m -= 1;
+ }
+ if m < 0 {
+ return SuppMnAnchor::None;
+ }
+ match self.class_of(cps[m as usize]) {
+ CLASS_ALETTER_CASED | CLASS_ALETTER | CLASS_SUPP_CASED |
CLASS_SUPP_LETTER => {
+ SuppMnAnchor::Letter
+ }
+ CLASS_NUMERIC | CLASS_NUMERIC_CASED | CLASS_SUPP_NUM =>
SuppMnAnchor::Digit,
+ _ => SuppMnAnchor::None,
+ }
+ }
+
+ /// Backward half of the ported `isFinalCased`: is there a cased letter
before position
+ /// `i` within the sigma's word? Runs over the FORMAT-FILTERED sequence;
`leading_format`
+ /// says whether format chars were filtered off the raw text start.
+ fn scan_back_finds_cased(&self, cps: &[char], i: usize, leading_format:
bool) -> bool {
+ let mut last_letter = true; // the sigma itself is a letter
+ let mut j = i as isize - 1;
+ while j >= 0 {
+ match self.class_of(cps[j as usize]) {
+ CLASS_ALETTER_CASED | CLASS_NUMERIC_CASED => return true,
+ CLASS_ALETTER => {
+ last_letter = true;
+ j -= 1;
+ }
+ CLASS_NUMERIC => {
+ last_letter = false;
+ j -= 1;
+ }
+ CLASS_EXTEND => {
+ // Non-cased marks attach only to a real base below them;
anything else
+ // (mid punctuation, danda, boundary, text start) leaves
the run
+ // unattached.
+ let Some(k) = self.skip_extends(cps, j, -1) else {
+ return false;
+ };
+ let b = self.class_of(cps[k]);
+ let is_continuer = b == CLASS_ALETTER_CASED
+ || b == CLASS_NUMERIC_CASED
+ || b == CLASS_NUMERIC
+ || b == CLASS_EXTEND_CASED
+ || b == CLASS_ALETTER
+ || b == CLASS_SUPP_CASED
+ || b == CLASS_SUPP_LETTER
+ || b == CLASS_SUPP_NUM;
+ if !is_continuer {
+ return false;
+ }
+ j = k as isize;
+ }
+ CLASS_SUPP_CASED => {
+ // Closes the preceding word, so the scan stops -- except
at RAW text
+ // start (no filtered-out leading format chars), where the
DFA keeps it
+ // joined to what follows.
+ return j == 0 && !leading_format;
+ }
+ CLASS_SUPP_LETTER | CLASS_SUPP_MN | CLASS_SUPP_NUM => {
+ // Attach/close and never themselves cased; nothing beyond
is reachable.
+ return false;
+ }
+ CLASS_EXTEND_CASED => {
+ // Cased combining mark (U+0345): cased when its run hangs
off a base --
+ // a BMP letter/digit, a word-forming supplementary char
(which closes a
+ // word right below the mark, merging the mark into the
sigma's
+ // segment), or an ANCHORED supplementary mark.
+ let (Some(k), _) = self.skip_extends_tracking_cased(cps, j
- 1, -1) else {
+ return false;
+ };
+ let b = self.class_of(cps[k]);
+ if b == CLASS_ALETTER_CASED
+ || b == CLASS_NUMERIC
+ || b == CLASS_NUMERIC_CASED
+ || b == CLASS_ALETTER
+ || b == CLASS_SUPP_CASED
+ || b == CLASS_SUPP_LETTER
+ || b == CLASS_SUPP_NUM
+ {
+ return true;
+ }
+ if b == CLASS_SUPP_MN {
+ return self.supp_mn_anchor(cps, k) !=
SuppMnAnchor::None;
+ }
+ return false;
+ }
+ CLASS_DANDA => {
+ // Backward across a danda: the word part before it must
end in letters
+ // (grammar: letters, optional danda, then number+word
chains) -- or
+ // carry a riding cased mark on a word-forming base, or be
a cased
+ // supplementary char at text start -- and the danda
itself chains only
+ // into digits after it.
+ if last_letter {
+ return false;
+ }
+ let (Some(k), saw_cased_mark) =
+ self.skip_extends_tracking_cased(cps, j - 1, -1)
+ else {
+ return false;
+ };
+ let b = self.class_of(cps[k]);
+ if b == CLASS_ALETTER_CASED {
+ return true;
+ }
+ if b == CLASS_SUPP_CASED {
+ return saw_cased_mark || (k == 0 && !leading_format);
+ }
+ if b == CLASS_SUPP_LETTER {
+ return saw_cased_mark;
+ }
+ if b == CLASS_SUPP_MN {
+ return saw_cased_mark
+ && self.supp_mn_anchor(cps, k) ==
SuppMnAnchor::Letter;
+ }
+ if b != CLASS_ALETTER {
+ return false;
+ }
+ if saw_cased_mark {
+ return true;
+ }
+ last_letter = true;
+ j = k as isize;
+ }
+ cls @ (CLASS_MID_LETTER | CLASS_MID_NUM | CLASS_MID_NUM_LET)
=> {
+ // `<mid-letter><let>` / `<mid-num><digit>` require a
genuine
+ // letter/digit base before the punctuation; scanning
backward
+ // legitimately walks marks-then-base (marks trail their
base). A cased
+ // mark walked over rides whatever the punctuation hangs
off, including
+ // a context-matching anchored supplementary mark or
supplementary
+ // digit.
+ let mw_ok = cls == CLASS_MID_LETTER || cls ==
CLASS_MID_NUM_LET;
+ let mn_ok = cls == CLASS_MID_NUM || cls ==
CLASS_MID_NUM_LET;
+ let (Some(real_pos), saw_cased_mark) =
+ self.skip_extends_tracking_cased(cps, j - 1, -1)
+ else {
+ return false;
+ };
+ let b = self.class_of(cps[real_pos]);
+ if last_letter
+ && mw_ok
+ && saw_cased_mark
+ && b == CLASS_SUPP_MN
+ && self.supp_mn_anchor(cps, real_pos) ==
SuppMnAnchor::Letter
+ {
+ return true;
+ }
+ if !last_letter
+ && mn_ok
+ && saw_cased_mark
+ && (b == CLASS_SUPP_NUM
+ || (b == CLASS_SUPP_MN
+ && self.supp_mn_anchor(cps, real_pos) ==
SuppMnAnchor::Digit))
+ {
+ return true;
+ }
+ let bridge_valid = (last_letter && mw_ok &&
is_letter_base(b))
+ || (!last_letter && mn_ok && is_digit_base(b));
+ if !bridge_valid {
+ return false;
+ }
+ if saw_cased_mark {
+ return true;
+ }
+ j = real_pos as isize;
+ }
+ _ => return false,
+ }
+ }
+ false
+ }
+
+ /// Forward half of the ported `isFinalCased`: is there a cased letter
after position `i`
+ /// within the sigma's word? Runs over the FORMAT-FILTERED sequence.
+ fn scan_fwd_finds_cased(&self, cps: &[char], i: usize) -> bool {
+ let mut last_letter = true;
+ let mut j = i + 1;
+ while j < cps.len() {
+ match self.class_of(cps[j]) {
+ CLASS_ALETTER_CASED | CLASS_NUMERIC_CASED => return true,
+ CLASS_ALETTER => {
+ last_letter = true;
+ j += 1;
+ }
+ CLASS_NUMERIC => {
+ last_letter = false;
+ j += 1;
+ }
+ CLASS_EXTEND => {
+ // A mark run trailing the anchor is properly attached in
text order, so
+ // the run stays open past it, including onto mid
punctuation on its far
+ // side.
+ let Some(k) = self.skip_extends(cps, j as isize, 1) else {
+ return false;
+ };
+ let b = self.class_of(cps[k]);
+ let is_continuer = b == CLASS_ALETTER_CASED
+ || b == CLASS_NUMERIC_CASED
+ || b == CLASS_NUMERIC
+ || b == CLASS_EXTEND_CASED
+ || b == CLASS_ALETTER
+ || b == CLASS_SUPP_CASED
+ || b == CLASS_SUPP_LETTER
+ || b == CLASS_SUPP_MN
+ || b == CLASS_SUPP_NUM
+ || b == CLASS_DANDA
+ || b == CLASS_MID_LETTER
+ || b == CLASS_MID_NUM
+ || b == CLASS_MID_NUM_LET;
+ if !is_continuer {
+ return false;
+ }
+ j = k;
+ }
+ CLASS_SUPP_CASED | CLASS_EXTEND_CASED => {
+ // Attaches to the current word, so the scan sees it
(cased).
+ return true;
+ }
+ CLASS_SUPP_LETTER | CLASS_SUPP_MN | CLASS_SUPP_NUM => {
+ // Attach to the current word and close it; never
themselves cased, and
+ // nothing beyond is reachable.
+ return false;
+ }
+ // The danda attaches only to a word part that ends in letters
(reached
+ // after digits the word is already closed) and continues only
into a digit
+ // -- unless that digit is itself cased (a Roman numeral),
which resolves
+ // the scan immediately.
+ CLASS_DANDA if !last_letter => return false,
+ CLASS_DANDA
+ if j + 1 < cps.len() && self.class_of(cps[j + 1]) ==
CLASS_NUMERIC_CASED =>
+ {
+ return true;
+ }
+ CLASS_DANDA if j + 1 < cps.len() && self.class_of(cps[j + 1])
== CLASS_NUMERIC => {
+ last_letter = false;
+ j += 2;
+ }
+ cls @ (CLASS_MID_LETTER | CLASS_MID_NUM | CLASS_MID_NUM_LET)
=> {
+ // `<mid-letter><let>` / `<mid-num><digit>` require a
genuine
+ // letter/digit base IMMEDIATELY after the punctuation --
unlike the
+ // backward scan, marks here are never skipped past: a
mark directly
+ // after the punctuation is attached to the punctuation,
not a base, so
+ // it blocks the bridge. (Format chars are already
filtered out, which
+ // is what lets "AΣ-<ZWJ>b" bridge exactly like "AΣ-b".)
+ let mw_ok = cls == CLASS_MID_LETTER || cls ==
CLASS_MID_NUM_LET;
+ let mn_ok = cls == CLASS_MID_NUM || cls ==
CLASS_MID_NUM_LET;
+ if j + 1 >= cps.len() {
+ return false;
+ }
+ let b = self.class_of(cps[j + 1]);
+ if (last_letter && mw_ok && is_letter_base(b))
+ || (!last_letter && mn_ok && is_digit_base(b))
+ {
+ j += 1;
+ } else {
+ return false;
+ }
+ }
+ _ => return false,
+ }
+ }
+ false
+ }
+
+ /// Lowercase `s` the way the planning JVM's
`String.toLowerCase(Locale.ROOT)` does.
+ pub fn lowercase(&self, s: &str) -> String {
+ let raw: Vec<char> = s.chars().collect();
+ // Built lazily on the first sigma: the format-filtered sequence the
scans run over
+ // (WB4-style: the legacy break iterator's `<ignore>` class loops on
every DFA
+ // state), the raw->filtered index map, and whether format chars led
the raw text.
+ let mut filtered: Option<(Vec<char>, Vec<usize>, bool)> = None;
+ let mut out = String::with_capacity(s.len());
+ for (i, &c) in raw.iter().enumerate() {
+ if c == CAPITAL_SIGMA {
+ // The condition consults the ORIGINAL neighbors, exactly as
the JDK scans
+ // `src`, not the partially-lowered output. The
final/non-final target chars
+ // are Unicode-stable (pinned in `ConditionalSpecialCasing`'s
entry table).
+ let (f, idx, leading_format) = filtered.get_or_insert_with(|| {
+ let mut f = Vec::with_capacity(raw.len());
+ let mut idx = vec![0usize; raw.len()];
+ for (k, &rc) in raw.iter().enumerate() {
+ idx[k] = f.len();
+ if self.class_of(rc) != CLASS_FORMAT {
+ f.push(rc);
+ }
+ }
+ let leading_format = self.class_of(raw[0]) == CLASS_FORMAT;
+ (f, idx, leading_format)
+ });
+ let fi = idx[i];
+ let is_final = self.scan_back_finds_cased(f, fi,
*leading_format)
+ && !self.scan_fwd_finds_cased(f, fi);
+ out.push(if is_final {
+ SMALL_FINAL_SIGMA
+ } else {
+ SMALL_SIGMA
+ });
+ } else if let Some(repl) = self.lower.get(&c) {
+ out.push_str(repl);
+ } else {
+ out.push(c);
+ }
+ }
+ out
+ }
+}
+
+/// Lowercase `s` for case-insensitive field matching. With tables (populated
whenever
+/// `case_sensitive = false`, shared by the core Parquet scan and the Delta
contrib scan) this
+/// reproduces the planning JVM's `String.toLowerCase(Locale.ROOT)` exactly.
+///
+/// Without tables, fall back to Rust's `str::to_lowercase`, which agrees with
Java on all
+/// simple mappings and differs only where the two Unicode snapshots or the
sigma context
+/// diverge. This is a real, live path, not just a defensive default: the
Iceberg native scan
+/// (`SparkPhysicalExprAdapterFactory::new(_, None)`) defaults `case_sensitive
= false` and
+/// always reaches this fallback for its schema name remap, alongside
+/// `parquet_convert_struct_to_struct`'s general struct-cast matching and
Rust-only unit tests
+/// that construct `SparkParquetOptions` directly.
+pub(crate) fn java_lowercase(s: &str, tables: Option<&JvmCaseTables>) ->
String {
Review Comment:
This is the one I would most like to see addressed before merge.
`names_equal_ignore_case_java` replaces `eq_ignore_ascii_case` throughout
the scan path, and it costs two `String` allocations plus one `HashMap<char,
String>` lookup per character. `remap_physical_schema` is called once per
Parquet file opened (DataFusion `opener/mod.rs:844`, guarded by
`needs_rewrite`, which is true whenever there is a predicate or the projection
prunes columns, so in practice always), and it does O(physical x logical)
comparisons. `spark.sql.caseSensitive` defaults to `false`, so this is the
default path for every Comet user, Delta or not.
I measured it against a worktree at this PR's base commit, release profile,
best of 3, using identical already-lowercase ASCII column names, which is the
friendliest case for the old code. Per Parquet file opened:
| columns | base | this PR | |
|---|---|---|---|
| 23 | 7.2 us | 138 us | 19x |
| 64 | 24.2 us | 912 us | 38x |
| 200 | 141 us | 8.57 ms | 61x |
| 1024 | 3.83 ms | 230 ms | 60x |
End to end through a real `DataSourceExec`, including Parquet decode:
| shape | base | this PR | |
|---|---|---|---|
| 500 files x 24 cols, project 10 | 70.4 ms | 162 ms | 2.3x |
| 100 files x 200 cols, project 20 | 28.2 ms | 338 ms | 12.0x |
A 10,000-file scan of a 200-column table would pay roughly 84 seconds of
extra CPU.
There is an exact fast path available. I dumped the real table your
generator produces on JDK 17 and checked it: the only ASCII entries are the 26
plain `A-Z` to `a-z` single-character mappings, and no ASCII codepoint is
U+03A3. So for two pure-ASCII names, `java_lowercase(a) == java_lowercase(b)`
is equivalent to `a.eq_ignore_ascii_case(b)`. Would you add that guard at the
top of `names_equal_ignore_case_java`?
```rust
if a.is_ascii() && b.is_ascii() {
return a.eq_ignore_ascii_case(b);
}
```
That keeps the Kelvin sign and dotted-I cases working, since neither is
ASCII, while making the overwhelmingly common case free. Hoisting the
lowercased forms out of the O(P x L) loop in `remap_physical_schema` (lowercase
each schema once, then compare precomputed strings) would cover the remaining
non-ASCII case, and `check_column_duplicate` would benefit from the same
treatment.
--
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]
