andygrove commented on code in PR #5365:
URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r3996879797
##########
pom.xml:
##########
@@ -95,6 +95,11 @@ under the License.
<guava.version>33.2.1-jre</guava.version>
<testcontainers.version>1.21.4</testcontainers.version>
<amazon-awssdk-v2.version>2.31.51</amazon-awssdk-v2.version>
+ <!-- Delta Lake pairing for the contrib/delta-spark module (-Pdelta). Each
Spark profile
+ overrides these with its matching Delta release; the defaults match
the default
+ spark-4.1 profile. Delta 2.x ships as artifact delta-core, 3.x/4.x as
delta-spark. -->
+ <delta.artifact>delta-spark</delta.artifact>
+ <delta.version>4.3.1</delta.version>
Review Comment:
`delta.version` is now declared twice in this `<properties>` block: `4.1.0`
at line 54 (for the kernel `contrib-delta` profile) and `4.3.1` here. Maven
takes the last one so the build is right, but the comment above line 54 now
describes a pairing that no longer applies (`spark-4.1 -> 4.1.0`). Could the
first declaration and its comment go, or the two contribs use distinct property
names so a reader does not have to work out which one wins?
##########
spark/src/main/scala/org/apache/comet/rules/CometScanContrib.scala:
##########
@@ -147,6 +154,21 @@ object CometScanContrib extends Logging {
"declining it and continuing with Comet's built-in handling",
e)
None
+ case e: LinkageError =>
+ // A version-skewed contrib jar (compiled against a Comet internal
that has since
Review Comment:
This arm is the right idea, but the containment does not reach discovery.
`ContribServices.loadFrom` (`ContribServices.scala:97-99`, not touched by this
PR) catches only `NonFatal`, and `ServiceLoader` raises `NoClassDefFoundError`
straight from `Class.forName` when a provider's superclass or interface is
missing, which is exactly the version-skewed-jar case this comment describes.
Because `contribs` is a `lazy val`, the failed initializer is re-run on every
access, so every V1 and V2 scan would throw rather than fall back.
Could the discovery loop get the same `LinkageError` arm (log and skip),
with a test alongside `FatalScanContrib` that drives discovery against a
provider whose interface cannot load?
##########
contrib/delta-spark/README.md:
##########
@@ -0,0 +1,60 @@
+<!--
+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.
+-->
+
+# Comet Delta Lake Contrib (experimental)
+
+Native Delta Lake reads for Comet. Delta tables are scanned through Comet's
+existing native Parquet reader, so they get row-group pruning, page-index
+pruning, and filter pushdown, with deletion vectors applied inside the scan.
+
+Support is experimental and explicitly opt-in. Two things are required:
+
+1. This module's jar (`comet-contrib-delta-spark`) on the classpath, alongside
+ `delta-spark`. It is never bundled into `comet-spark`; without it, Comet
+ has no Delta surface at all.
+2. `spark.comet.scan.delta.enabled=true`. The default is `false`, so the jar
+ alone does nothing.
+
+Unsupported tables and features fall back to Spark's reader. See the
+[user guide](https://datafusion.apache.org/comet/user-guide/delta.html)
Review Comment:
Two things in this README:
- This link resolves to `user-guide/delta.html`, but the page lives under
`user-guide/latest/`, and `docs/source/conf.py` only has redirects for the
pre-existing pages, so it will 404. `latest/delta.html` or a redirect entry
would fix it.
- Line 53 builds with `-pl contrib/delta-spark`, which resolves
`comet-spark` from the local Maven repo. That is the stale-sibling trap the
contributor guide warns about. CI is fine because the workflow installs
`common,spark` immediately before. Could the README say the same, or run the
full reactor?
##########
spark/pom.xml:
##########
@@ -585,6 +585,19 @@ under the License.
<groupId>org.scalatest</groupId>
<artifactId>scalatest-maven-plugin</artifactId>
</plugin>
+ <plugin>
Review Comment:
This execution now runs on every profile, not only under `-Pdelta`: I
measured a 6.7 MB `-tests.jar` in `spark/target`, `install` puts it in the
local repo, and `dev/release/publish-to-maven.sh` uploads every jar it finds,
so each release would ship six of them. There is partial precedent (the
`-test-sources.jar`), but it should be a deliberate choice. Binding the
execution inside a `delta` profile in this pom would keep it to the builds that
need it. Also the comment says `contrib/delta`; the consumer is
`contrib/delta-spark`.
Related maintainer question I am raising here so it gets decided before
merge: `dev/release/build-release-comet.sh` never passes `-Pdelta`, so the
contrib jar the docs tell users to put on the classpath is never built or
published, and `maven.deploy.skip=false` in the contrib pom is moot today.
Either the release build adds `-Pdelta` (and then the artifact name
`comet-contrib-delta-spark4.1_2.13` deserves a look against the
`comet-spark-spark4.1_2.13` convention), or the docs should say
build-from-source is the only route for now.
##########
dev/ci/compute-changes.py:
##########
@@ -306,6 +306,23 @@
".mvn/**",
"mvnw",
],
+ "delta": [
+ "contrib/delta/**",
Review Comment:
A few things about this filter and the workflow it triggers, since every
queue run touching native or spark sources will now run three container jobs:
- `contrib/delta-spark/README.md` alone triggers the full suite (no `!**.md`
exclude, unlike `build_linux`).
- `contrib/delta/**` is the kernel crate, which this workflow never builds
(`delta_build_gate.yml` covers it). It can be dropped here.
- `spark/src/test/**` is excluded, but the contrib suites extend
`CometTestBase` and `CometS3TestBase` through the new test-jar, so a change to
those bases can break the contrib without running it. Adding those two files
would close that.
- In the workflow itself: `feature-off-build` restores no cargo cache, so it
is a cold debug build to run one test, and `dev-scripts-python` spends four
runners byte-compiling two scripts. One version, or folding it into preflight,
seems proportionate. The cache-save step gated on `refs/heads/main` never runs
under this policy (the restore does fall back warm to the `build_linux` cache,
and `--features delta` is a no-op since it is in the default set).
##########
docs/source/contributor-guide/ci.md:
##########
@@ -69,6 +69,7 @@ Each queue-only suite has a label that runs it on a pull
request:
| `run-spark-3.4-tests` | Spark SQL tests against Spark 3.4
|
| `run-spark-3.5-tests` | Spark SQL tests against Spark 3.5
|
| `run-spark-4.0-tests` | Spark SQL tests against Spark 4.0
|
+| `run-delta-tests` | Delta contrib tests against Spark 3.5
|
Review Comment:
The matrix in `delta_contrib_test.yml` runs Spark 3.5, 4.0 and 4.1, but this
row, the `POLICY` comment in `dev/ci/compute-changes.py:376`, and the
reusable-workflow list and label sentence in `.github/workflows/README.md`
still describe it as Spark 3.5 only. I confirmed the 4.1 leg passes locally, so
the description could claim it as well.
##########
.github/workflows/delta_contrib_test.yml:
##########
@@ -0,0 +1,169 @@
+# 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.
+
+name: Delta Contrib Tests
+
+# Reusable: invoked by ci.yml. Triggering, path filters, and concurrency
+# live in the umbrella workflow.
+on:
+ workflow_call:
+
+permissions:
+ contents: read
+
+env:
+ RUST_VERSION: stable
+ RUST_BACKTRACE: 1
+ # Force GNU ld on Linux: rust-lld cannot resolve -ljvm against the Zulu JDK
+ # layout installed by setup-java (same rationale as pr_build_linux.yml).
+ RUSTFLAGS: "-Clink-arg=-fuse-ld=bfd"
+ # The container's default locale is POSIX, which makes the JVM's file-path
encoder
+ # reject non-ASCII partition directory names the suites create.
+ LANG: "C.UTF-8"
+ LC_ALL: "C.UTF-8"
+
+jobs:
+
+ contrib-delta:
+ name: Delta contrib (Spark ${{ matrix.profile.spark }})
+ runs-on: ubuntu-24.04
+ container:
Review Comment:
This container has no Docker socket, and `CometDeltaS3Suite` `assume()`s out
when `DockerClientFactory.isDockerAvailable` is false, which scalatest reports
as canceled and the build treats as green. So the MinIO suite contributes no
coverage in CI even though the description lists it as live, and the S3 gate is
the logic I would most like exercised end to end.
Could you either mount `/var/run/docker.sock` into this job (or run that one
suite outside the container), or state in the workflow that the S3 suite is
manual-only and drop it from the description's CI coverage claim?
##########
contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala:
##########
@@ -0,0 +1,1838 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.contrib.delta
+
+import java.io.IOException
+import java.net.URI
+import java.util.Locale
+
+import scala.collection.mutable.{ListBuffer, Map => MutableMap}
+import scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.catalyst.expressions.{Alias, GenericInternalRow,
InputFileBlockLength, InputFileBlockStart, InputFileName}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData}
+import
org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
+import org.apache.spark.sql.comet.CometScanExec
+import org.apache.spark.sql.delta.DeltaParquetFileFormat
+import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
+import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec,
SparkPlan}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
+
+import org.apache.comet.CometConf
+import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES
+import org.apache.comet.objectstore.NativeConfig
+import org.apache.comet.parquet.CometParquetUtils
+import org.apache.comet.rules.{CometScanRule, CometScanTypeChecker}
+import org.apache.comet.serde.operator.CometNativeScan
+import org.apache.comet.shims.ShimFileFormat
+
+/**
+ * Claim/decline gates for the native Delta scan. Correctness rule: when in
doubt, decline,
+ * Spark's Delta reader handles the scan and results stay correct, just
unaccelerated.
+ */
+object DeltaScanSupport {
+
+ /**
+ * Reader features the native path understands; anything else on the
protocol declines the
+ * table. `deletionVectors`/`columnMapping` are declined separately below
for specific reasons.
+ */
+ private val understoodReaderFeatures: Set[String] =
+ Set("columnMapping", "deletionVectors", "timestampNtz", "v2Checkpoint",
"vacuumProtocolCheck")
+
+ /**
+ * Is this exactly Delta's DSv1 parquet format? Compared by class name, not
`classOf`: a
+ * `classOf` reference would raise `NoClassDefFoundError` and break every
parquet scan when
+ * delta-spark is absent from the classpath.
+ */
+ def isDeltaScan(scanExec: FileSourceScanExec): Boolean =
+ scanExec.relation.fileFormat.getClass.getName ==
+ "org.apache.spark.sql.delta.DeltaParquetFileFormat"
+
+ /**
+ * Claim-time artifacts [[declineReason]] already computes but
[[CometDeltaNativeScan.convert]]
+ * also needs -- threaded through by reference (populated only on the
claimable path, right
+ * before `declineReason` returns `None`) so a claimed scan does not pay to
recompute either:
+ * the Hadoop conf
([[org.apache.spark.sql.internal.SessionState#newHadoopConfWithOptions]] is
+ * not cheap) and the deletion-vector descriptors (base64-decoded,
non-trivial only for DV-shape
+ * scans). One instance is created per claim attempt in `DeltaScanContrib`
and passed to both
+ * `declineReason` and `convert`.
+ */
+ private[delta] final class DeltaClaimMemo {
+ var hadoopConf: Configuration = _
+ var dvDescriptors: Seq[DeletionVectorDescriptor] = Seq.empty
+ }
+
+ /**
+ * First reason this Delta scan cannot go native, or None when claimable (in
which case `memo`
+ * is populated for [[CometDeltaNativeScan.convert]] to reuse). Only called
when [[isDeltaScan]]
+ * is true. `scanHelper` is the [[CometScanExec]] built to drive `convert`
on a claim, reused
+ * for the multi-store gate below.
+ */
+ def declineReason(
+ plan: SparkPlan,
+ scanExec: FileSourceScanExec,
+ scanHelper: CometScanExec,
+ memo: DeltaClaimMemo): Option[String] = {
+ val format =
scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat]
+ val protocol = format.protocol
+ val metadata = format.metadata
+ // Name mode is supported via physical-name schemas; id mode needs the
field-id path and
+ // stays declined until validated. Hoisted here since several gates below
reuse it.
+ val cmMode = metadata.columnMappingMode.name
+ // Descriptor deserialization is expensive, so hoist it into a `lazy val`,
forced at most
+ // once in this method; on the claimable path the result is handed to
`convert` through
+ // `memo` below, so a claimed scan deserializes the descriptors exactly
once end to end.
+ val tableRoot = scanExec.relation.location.rootPaths.head.toString
+ lazy val dvDescriptors: Seq[DeletionVectorDescriptor] =
+ selectedDvDescriptors(scanHelper, tableRoot)
+
+ // Mirrors core's CometScanRule.isSchemaSupported so scan-time type gates
(unsigned-small-int
+ // fallback, collation, shredded-variant-struct) apply identically here.
Pure in-memory check,
+ // so it runs first, ahead of every I/O-bearing gate below.
+ val schemaFallbackReasons = new ListBuffer[String]()
+ val typeChecker = CometScanTypeChecker()
+ val requiredSchemaSupported =
+ typeChecker.isSchemaSupported(scanExec.requiredSchema,
schemaFallbackReasons)
+ val partitionSchemaSupported =
+ typeChecker.isSchemaSupported(scanExec.relation.partitionSchema,
schemaFallbackReasons)
+ if (!requiredSchemaSupported || !partitionSchemaSupported) {
+ return Some(
+ "Native Delta scan does not support the schema: " +
schemaFallbackReasons.mkString(", "))
+ }
+
+ if (format.isCDCRead) {
+ return Some("Native Delta scan does not support Change Data Feed reads")
+ }
+
+ // Delta's DML machinery (findTouchedFiles) disables reader optimizations
and needs real
+ // row indexes from Spark's reader; claiming here would feed NULL indexes
into DV construction.
+ if (!format.optimizationsEnabled) {
+ return Some("Native Delta scan does not support reads with reader
optimizations disabled")
+ }
+ if (scanExec.requiredSchema.exists(_.name ==
DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME) ||
+ scanExec.relation.dataSchema.exists(
+ _.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME)) {
+ return Some("Native Delta scan does not support Delta's generated
row-index column")
+ }
+
+ if (cmMode != "none" && cmMode != "name") {
+ return Some(s"Native Delta scan does not support column mapping mode
$cmMode")
+ }
+ // createPhysicalSchema wholesale-replaces field metadata, silently
dropping EXISTS_DEFAULT.
+ if (cmMode == "name" &&
+ getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) {
+ return Some(
+ "Native Delta scan does not support column defaults together with
column mapping")
+ }
+ // createPhysicalSchema rewrites nested StructField names too, and the
native builder emits the
+ // required schema verbatim as output, so name-sensitive expressions (e.g.
to_json) would leak
+ // physical names. Decline until a rename adapter exists.
+ if (cmMode == "name" &&
+ scanExec.requiredSchema.exists(f => containsNestedStruct(f.dataType))) {
+ return Some("Native Delta scan does not support column mapping with
nested struct fields")
+ }
+
+ val readerFeatures = protocol.readerFeatureNames
+ val unknownFeatures = readerFeatures -- understoodReaderFeatures
+ if (unknownFeatures.nonEmpty) {
+ return Some(
+ s"Native Delta scan does not support reader feature(s)
${unknownFeatures.mkString(", ")}")
+ }
+
+ // Non-constant metadata columns are generated per-row by Spark's reader
and unsupported,
+ // except Delta's DV bookkeeping columns, which the native path emits as
constants.
+ val knownColNames =
+ scanExec.relation.dataSchema.map(_.name).toSet ++
+ scanExec.relation.partitionSchema.map(_.name).toSet ++
+ scanExec.fileConstantMetadataColumns.map(_.name).toSet ++
+ CometDeltaNativeScan.internalColumnNames
+ val unknownOutput =
scanExec.output.map(_.name).filterNot(knownColNames.contains)
+ if (unknownOutput.nonEmpty) {
+ return Some(
+ s"Native Delta scan does not support generated column(s)
${unknownOutput.mkString(", ")}")
+ }
+
+ // Deletion-vector shape invariants (see
CometDeltaNativeScan.buildDvScanCommon).
+ if (CometDeltaNativeScan.isDvShape(scanExec)) {
+ // A row-index column WITHOUT is_row_deleted is Delta DML bookkeeping
(real row indexes),
+ // not a DV read; claiming it with a constant would corrupt the DVs
being written.
+ val hasIsRowDeleted =
+ scanExec.requiredSchema.exists(_.name ==
CometDeltaNativeScan.IsRowDeletedColumn)
+ val hasRowIndex =
+ scanExec.requiredSchema.exists(_.name ==
CometDeltaNativeScan.RowIndexColumn)
+ if (hasRowIndex && !hasIsRowDeleted) {
+ return Some(
+ "Native Delta scan does not support row-index reads outside a
deletion-vector scan")
+ }
+ // Internal columns must form a suffix of the read schema so data-column
positions agree
+ // between Spark's output and the stripped native schema.
+ val names = scanExec.requiredSchema.fields.map(_.name)
+ val firstInternal =
names.indexWhere(CometDeltaNativeScan.internalColumnNames.contains)
+ if
(!names.drop(firstInternal).forall(CometDeltaNativeScan.internalColumnNames.contains))
{
+ return Some("Native Delta scan requires DV bookkeeping columns to
trail the read schema")
+ }
+ // Native applies the DV itself and emits a dead constant for row-index,
so the real value
+ // must be provably unused above the scan.
+ if (!rowIndexUnusedAbove(plan, scanExec)) {
+ return Some(
+ "Native Delta scan cannot supply _metadata.row_index values consumed
by the query")
+ }
+ // The DV common builder does not serialize existence defaults yet.
+ if (getExistenceDefaultValues(scanExec.requiredSchema).exists(_ !=
null)) {
+ return Some(
+ "Native Delta scan does not support column defaults together with
deletion vectors")
+ }
+ // Bounds native's memory for expanded DV row selectors (delta_dv.rs),
pessimistically
+ // bounded by 2*cardinality + #row-groups; the conf below makes an
over-pessimistic decline
+ // recoverable.
+ val maxDeletedRowsPerFile =
DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.get()
+ val oversizedCardinalities = dvDescriptors
+ .map(_.cardinality)
+ .filter(_ > maxDeletedRowsPerFile)
+ if (oversizedCardinalities.nonEmpty) {
+ return Some(
+ "Native Delta scan does not support a deletion vector deleting " +
+ s"${oversizedCardinalities.max} rows in a single file, exceeding "
+
+
s"${DeltaScanConf.COMET_DELTA_MAX_DELETED_ROWS_PER_FILE.key}=$maxDeletedRowsPerFile")
+ }
+ }
+
+ // input_file_name & friends read from a thread-local Spark's FileScanRDD
sets; the native scan
+ // does not populate it, and Delta's DML find-touched-files scans use it
(mirrors core's check
+ // in CometScanRule.nativeScan).
+ if (plan.exists(node =>
+ node.expressions.exists(_.exists {
+ case _: InputFileName | _: InputFileBlockStart | _:
InputFileBlockLength => true
+ case _ => false
+ }))) {
+ return Some(
+ "Native Delta scan is not compatible with input_file_name, " +
+ "input_file_block_start, or input_file_block_length")
+ }
+
+ // Row-index metadata columns are generated per-row by Spark's reader
(mirrors core); the DV
+ // shape's trailing row-index column is exempt since the gates above
already proved it dead.
+ if (!CometDeltaNativeScan.isDvShape(scanExec) &&
+ ShimFileFormat.findRowIndexColumnIndexInSchema(scanExec.requiredSchema)
>= 0) {
+ return Some("Native Delta scan does not support row index generation")
+ }
+
+ // Mirror core's vectorized-reader compatibility gate.
+ if (!SQLConf.get.getConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED) &&
+ !CometConf.COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER.get()) {
+ return Some(
+ "Native Delta scan is incompatible with " +
+ s"${SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key}=false")
+ }
+
+ // Decline ALL encrypted-parquet configurations (stricter than core): the
exec node does not
+ // yet wire the decryption-key broadcast to executors.
+ val hadoopConf = scanExec.relation.sparkSession.sessionState
+ .newHadoopConfWithOptions(scanExec.relation.options)
+ // Populated now (rather than only at the very end) so it is available
even though several
+ // early-return gates below still lie ahead: cheap to set, and every one
of those gates
+ // declines the scan anyway, so `memo` is simply never read by `convert`
in that case.
+ memo.hadoopConf = hadoopConf
+ if (CometParquetUtils.encryptionEnabled(hadoopConf)) {
+ return Some("Native Delta scan does not support encrypted parquet")
+ }
+
+ // Nested-type column defaults cannot be serialized; a dropped default
would misalign the
+ // value/index lists consumed positionally on the native side. Mirrors
core's
+ // transformV1Scan gate.
+ val possibleDefaultValues =
getExistenceDefaultValues(scanExec.requiredSchema)
+ if (possibleDefaultValues.exists(d =>
+ d != null && (d.isInstanceOf[ArrayBasedMapData] || d
+ .isInstanceOf[GenericInternalRow] ||
d.isInstanceOf[GenericArrayData]))) {
+ return Some("Native Delta scan does not support default values for
nested types")
+ }
+
+ // An opted-in S3-compliant alias scheme (fs.comet.s3Compliant.schemes) is
declined before the
+ // generic scheme gate below so the reason says why: core's native scan
reads it through the
+ // S3 client, but the S3 divergence gates further down model Hadoop's
S3AFileSystem only.
+ val rootUris = scanExec.relation.location.rootPaths.map(_.toUri)
+ val aliasReason = s3CompliantAliasSchemeReason(hadoopConf, rootUris)
+ if (aliasReason.isDefined) {
+ return aliasReason
+ }
+
+ // Only claim scans whose root paths object_store (or the configured
libhdfs schemes) can
+ // actually read (mirrors core's unsupportedFsSchemes gate).
+ val libhdfs = libhdfsSchemes
+ val unsupportedRootSchemes = unsupportedSchemes(rootUris, libhdfs)
+ if (unsupportedRootSchemes.nonEmpty) {
+ return Some(
+ "Native Delta scan does not support filesystem scheme(s) " +
+ s"${unsupportedRootSchemes.mkString(", ")}")
+ }
+
+ // A recognized scheme can still carry a path object_store rejects (a
directory name with a
+ // newline surfaces as `%0A`), which native planning hard-fails on while
Spark's reader opens
+ // it. Mirrors core's root-path gate; the complete selected paths are
probed below.
+ val rejectedRoot = objectStoreRejectedPathReason(rootUris, libhdfs)
+ if (rejectedRoot.isDefined) {
+ return rejectedRoot
+ }
+
+ // A shallow clone can span multiple object-store authorities, but the
native builder resolves
+ // ObjectStoreUrl from only the FIRST selected file; force file listing
and decline rather than
+ // risk reading a later file through the wrong handle.
+ val dataFileUris =
+
scanHelper.selectedPartitions.iterator.flatMap(_.files).map(_.getPath.toUri).toSeq
+
+ // Both gates below need the DV absolute-path URIs; dvDescriptors is
already memoized.
+ val dvUris = dvDescriptors
+ .filter(_.storageType != DeletionVectorDescriptor.INLINE_DV_MARKER)
+ .map(_.absolutePath(new Path(tableRoot)).toUri)
+
+ // The root-path gate above only inspects the table root(s); selected
files can resolve
+ // through a different scheme (e.g. `viewfs:`). Checked before the
authority gates below,
+ // which presume every URI is natively resolvable.
+ val unsupportedSelected = unsupportedSelectedSchemeReason(dataFileUris ++
dvUris, libhdfs)
+ if (unsupportedSelected.isDefined) {
+ return unsupportedSelected
+ }
+
+ // Same path probe for every complete selected path, not just its
directory: a shallow
+ // clone's source can sit outside this root, and CONVERT TO DELTA keeps
the source Parquet
+ // basenames, so the rejected character can be in the file name itself.
The probe is a
+ // native URL parse with no I/O, so once per distinct URI costs less than
the scan's own
+ // per-file parse.
+ val rejectedSelected = objectStoreRejectedPathReason(dataFileUris ++
dvUris, libhdfs)
+ if (rejectedSelected.isDefined) {
+ return rejectedSelected
+ }
+
+ // Checked before multiStoreReason, which presumes every URI resolves to a
single store
+ // identity -- a userinfo-bearing authority provably does not (store
keying drops userinfo).
+ val userInfoReason = userInfoBearingAuthorityReason(dataFileUris ++ dvUris)
+ if (userInfoReason.isDefined) {
+ return userInfoReason
+ }
+
+ val multiStore = multiStoreReason(dataFileUris)
+ if (multiStore.isDefined) {
+ return multiStore
+ }
+
+ // GCS's zero-I/O, conf-only credential-forwarding gate; ordered alongside
the S3 credential
+ // gates below since all presume a single, well-formed store identity per
URI.
+ val gcsAuthReason = gcsHadoopOnlyAuthReason(hadoopConf, dataFileUris ++
dvUris)
+ if (gcsAuthReason.isDefined) {
+ return gcsAuthReason
+ }
+
+ // Zero-I/O, conf-only, like the GCS gate above: decline any bucket
configured for an
+ // encryption algorithm outside the allowlist (SSE-C, CSE-KMS, CSE-CUSTOM,
or unknown) before
+ // the credential-divergence gates below, which do not otherwise notice
this table is readable
+ // through Hadoop only because Hadoop's request factory (SSE-C) or
SDK-level decryption layer
+ // (CSE-*) does something native never learns about.
+ val encryptionReason =
+ unsupportedEncryptionAlgorithmReason(hadoopConf, dataFileUris ++ dvUris)
+ if (encryptionReason.isDefined) {
+ return encryptionReason
+ }
+
+ // Shared across the two gates below: propagateBucketOptions is a full
Configuration deep
+ // copy, and both gates would otherwise recompute it independently for the
same bucket(s)
+ // (once here, then again per-key inside s3ConfigDivergenceReason). One
cache, populated
+ // lazily per bucket on first use, makes it a single copy total per bucket
across both gates.
+ val propagatedConfCache = MutableMap.empty[String, Configuration]
+
+ // Always zero-I/O (plain propagated-conf read, no keystore): native's S3
client has no
+ // HTTP proxy support at all (no fs.s3a.proxy.* key is read anywhere in
s3.rs), so a bucket
+ // requiring a proxy for S3 egress must decline here rather than claim and
then connect
+ // directly, bypassing whatever network-segmentation/firewall policy
required the proxy.
+ val proxyReason = proxyGateReason(hadoopConf, dataFileUris ++ dvUris,
propagatedConfCache)
+ if (proxyReason.isDefined) {
+ return proxyReason
+ }
+
+ // Zero-I/O, conf-only, like the proxy gate above: Hadoop's
AssumedRoleCredentialProvider
+ // sends fs.s3a.assumed.role.policy as the session policy of its STS
AssumeRole request,
+ // while native's assumed-role provider never reads the key -- a claimed
scan would assume
+ // the role WITHOUT the configured session restriction, silently widening
permissions.
+ val rolePolicyReason =
+ assumedRolePolicyGateReason(hadoopConf, dataFileUris ++ dvUris,
propagatedConfCache)
+ if (rolePolicyReason.isDefined) {
+ return rolePolicyReason
+ }
+
+ // Every fs.s3a.* option native's get_config (s3.rs) resolves must agree
between what Hadoop
+ // itself would use and what native would read from the forwarded,
substituted conf (covers
+ // long-form bucket credentials, JCEKS/credential-provider shadowing, and
any other
+ // short-vs-effective divergence in one mechanism); reuses hadoopConf from
the encryption gate
+ // above.
+ val s3Reason =
+ s3ConfigDivergenceReason(hadoopConf, dataFileUris ++ dvUris,
propagatedConfCache)
+ if (s3Reason.isDefined) {
+ return s3Reason
+ }
+
+ // A credential-provider class native's
build_aws_credential_provider_metadata (s3.rs) does
+ // not recognize errors at scan EXECUTION time, after the scan was already
claimed; decline
+ // eagerly instead.
+ val providerReason = providerClassGateReason(hadoopConf, dataFileUris ++
dvUris)
+ if (providerReason.isDefined) {
+ return providerReason
+ }
+
+ // Reuse core's generic native-scan gates
(ignoreCorruptFiles/ignoreMissingFiles, AQE DPP on
+ // Spark 3.4, exec enabled); tags its own fallback reasons.
+ if (!CometNativeScan.isSupported(scanExec)) {
+ return Some("Core native scan gates rejected the scan (see reasons
above)")
+ }
+
+ // Claimable: hand the already-forced descriptors to `convert` via `memo`
so it does not
+ // deserialize them a second time.
+ memo.dvDescriptors = dvDescriptors
+ None
+ }
+
+ /**
+ * Deletion-vector descriptors for every file this DV-shape scan selected,
normalized to
+ * absolute on-disk paths. Returns `Seq.empty` for the plain shape. Shared
by the DV cardinality
+ * gate and [[CometDeltaNativeScan.convert]]'s object-store option merge.
+ */
+ private[delta] def selectedDvDescriptors(
+ scanHelper: CometScanExec,
+ tableRoot: String): Seq[DeletionVectorDescriptor] = {
+ if (!CometDeltaNativeScan.isDvShape(scanHelper.wrapped)) {
+ return Seq.empty
+ }
+ val tableRootPath = new Path(tableRoot)
+ scanHelper.selectedPartitions.iterator
+ .flatMap(_.files)
+ .flatMap { file =>
+ file.metadata
+ .get(DeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED)
+ .map(enc =>
DeletionVectorDescriptor.deserializeFromBase64(enc.asInstanceOf[String]))
+ }
+ .map(_.copyWithAbsolutePath(tableRootPath))
+ .toSeq
+ }
+
+ /**
+ * The libhdfs scheme exemption set from
[[org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES]],
+ * parsed exactly like core's scan gate (`NativeConfig.parseSchemeSet`:
split on commas,
+ * trimmed, lowercased) and defaulting to `Set("hdfs")` when unset.
+ */
+ private[delta] def libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get()
match {
+ case Some(s) => NativeConfig.parseSchemeSet(s)
+ case None => Set("hdfs")
+ }
+
+ /**
+ * Decline reason when any of `uris` uses a scheme opted in as an
S3-compliant alias through
+ * `fs.comet.s3Compliant.schemes` (e.g. `blob`), or `None`. Core's native
Parquet scan admits
+ * such a scheme and reads it through its S3 client, with `NativeConfig`
translating the vendor
+ * `fs.<scheme>.<authority>.*` keys into `fs.s3a.bucket.*` options. Spark,
however, reads the
+ * same table through the vendor's own Hadoop FileSystem, not
`S3AFileSystem`, and every S3
+ * divergence gate in this object ([[s3ConfigDivergenceReason]] and its
siblings) is verified
+ * against `S3AFileSystem`'s consumers only. With no model of how the vendor
filesystem resolves
+ * its configuration, whether native and Spark would agree cannot be
decided, so the scan is
+ * declined rather than claimed on a guess. Selected data-file and
deletion-vector URIs under an
+ * alias scheme are declined by the generic scheme gates, which never admit
an alias (see
+ * [[unsupportedSchemes]]).
+ */
+ private[delta] def s3CompliantAliasSchemeReason(
+ hadoopConf: Configuration,
+ uris: Seq[URI]): Option[String] = {
+ val aliases = NativeConfig.resolveS3CompliantSchemes(hadoopConf)
+ if (aliases.isEmpty) {
+ return None
+ }
+ val found = uris
+ .flatMap(uri => Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)))
+ .filter(aliases.contains)
+ .distinct
+ if (found.isEmpty) {
+ None
+ } else {
+ Some(
+ "Native Delta scan does not support S3-compliant alias filesystem
scheme(s) " +
+ s"${found.sorted.mkString(", ")}
(${CometConf.COMET_S3_COMPLIANT_SCHEMES_KEY}): " +
+ "Spark reads them through a vendor filesystem whose S3 configuration
resolution the " +
+ "native scan's S3AFileSystem divergence model cannot verify")
+ }
+ }
+
+ /**
+ * The lowercased, deduplicated schemes among `uris` that neither `libhdfs`
nor Comet's native
+ * object_store layer ([[CometScanRule.isNativelyReadableScheme]]) can read.
A `null` scheme is
+ * tolerated, not flagged, since such a URI cannot come from a Hadoop-backed
source. The alias
+ * set handed to core's gate is deliberately empty: an
`fs.comet.s3Compliant.schemes` alias is
+ * never admitted here (see [[s3CompliantAliasSchemeReason]]), even though
core admits it.
+ */
+ private[delta] def unsupportedSchemes(uris: Seq[URI], libhdfs: Set[String]):
Set[String] = {
+ uris
+ .filter { uri =>
+ val sch = uri.getScheme
+ sch != null && {
+ val sl = sch.toLowerCase(Locale.ROOT)
+ !libhdfs.contains(sl) &&
!CometScanRule.isNativelyReadableScheme(uri, Set.empty)
+ }
+ }
+ .map(_.getScheme.toLowerCase(Locale.ROOT))
+ .toSet
+ }
+
+ /**
+ * Decline reason naming the first of `uris` whose path object_store rejects
even though it
+ * recognizes the scheme ([[CometScanRule.objectStoreAcceptsPath]], e.g. a
directory name
+ * containing a newline, `%0A` in the URI), or `None`. Schemes in `libhdfs`
never reach
+ * object_store's path parser and are skipped, as is a `null` scheme (see
+ * [[unsupportedSchemes]]); an S3-compliant alias is declined before this
gate runs. The probe
+ * is uncached but is a plain native URL parse with no I/O
+ * ([[CometScanRule.objectStoreAcceptsPath]]), so callers pass every
complete selected path
+ * (root paths, data files and deletion vectors) once per distinct URI; a
converted table can
+ * carry the rejected character in a file basename. The reason masks any
userinfo in the named
+ * URI ([[redactedAuthority]]).
+ */
+ private[delta] def objectStoreRejectedPathReason(
+ uris: Seq[URI],
+ libhdfs: Set[String]): Option[String] = {
+ uris.distinct
+ .find { uri =>
+ val sch = uri.getScheme
+ sch != null && !libhdfs.contains(sch.toLowerCase(Locale.ROOT)) &&
+ !CometScanRule.objectStoreAcceptsPath(uri)
+ }
+ .map { uri =>
+ // Mask userinfo (see redactedAuthority); the raw path keeps its
percent encoding so the
+ // reason shows the rejected sequence as written.
+ val shown =
+ if (uriUserInfo(uri).isEmpty) uri.toString
+ else
s"${redactedAuthority(uri)}${Option(uri.getRawPath).getOrElse("")}"
+ s"Native Delta scan cannot open path '$shown': object_store rejects it
" +
+ "(e.g. an unsupported character in the path)"
+ }
+ }
+
+ /**
+ * Decline reason when any of `uris` -- the scan's selected data-file and
deletion-vector URIs
+ * -- use a scheme [[unsupportedSchemes]] flags, or `None` when every URI is
natively readable
+ * (or libhdfs-exempt).
+ */
+ private[delta] def unsupportedSelectedSchemeReason(
+ uris: Seq[URI],
+ libhdfs: Set[String]): Option[String] = {
+ val schemes = unsupportedSchemes(uris, libhdfs)
+ if (schemes.isEmpty) {
+ None
+ } else {
+ Some(
+ "Native Delta scan does not support selected data file or deletion
vector filesystem " +
+ s"scheme(s) ${schemes.mkString(", ")}")
+ }
+ }
+
+ /**
+ * Decline reason when `uris` span more than one object-store authority
(scheme + lowercased raw
+ * authority, so e.g. `S3A://Bucket` and `s3a://bucket` collapse), or `None`
when they share
+ * one. `file://` paths carry no authority, so local scans across many
directories are
+ * unaffected.
+ */
+ private[delta] def multiStoreReason(uris: Seq[URI]): Option[String] = {
+ val authorities = uris.map(uriAuthority).distinct
+ if (authorities.size > 1) {
+ Some(
+ "Native Delta scan does not support data files spanning multiple
object stores " +
+ s"(found: ${authorities.sorted.mkString(", ")})")
+ } else {
+ None
+ }
+ }
+
+ /**
+ * Normalizes `uri` to a lowercased `scheme://authority` string, keyed on
the raw `getAuthority`
+ * rather than the parsed host/port/userinfo fields: `getHost` (and
`getUserInfo`/`getPort`)
+ * return `null` for the whole authority when it fails RFC 3986 `reg-name`
syntax (e.g. an
+ * underscore in a GCS bucket name, `gs://my_bucket`), which would silently
collapse distinct
+ * buckets into one empty-host key. A `null` authority normalizes to the
empty string.
+ */
+ private[delta] def uriAuthority(uri: URI): String = {
+ val scheme =
Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("")
+ val authority =
Option(uri.getAuthority).map(_.toLowerCase(Locale.ROOT)).getOrElse("")
+ s"$scheme://$authority"
+ }
+
+ /**
+ * The raw userinfo component of `uri`'s authority, or empty when none.
Splits at the LAST `@`
+ * rather than using `URI#getUserInfo`, which (like [[uriAuthority]]'s
getters) returns `null`
+ * for the whole authority on an RFC 3986 `reg-name` violation. Never
lowercased: userinfo is
+ * case-sensitive.
+ */
+ private[delta] def uriUserInfo(uri: URI): String = {
+ val authority = Option(uri.getAuthority).getOrElse("")
+ val at = authority.lastIndexOf('@')
+ if (at >= 0) authority.substring(0, at) else ""
+ }
+
+ /**
+ * Redacts `uri`'s authority to `scheme`, then `://`, then a literal `***`
masking userinfo,
+ * then `@host[:port]`, for embedding in a decline reason. NEVER interpolate
`uri.getAuthority`
+ * or [[uriUserInfo]] directly into a reason string: doing so would leak
credentials embedded as
+ * URI userinfo into the SQL plan's explain output, fallback-reason logging,
or the Spark UI.
+ */
+ private[delta] def redactedAuthority(uri: URI): String = {
+ val scheme =
Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("")
+ val authority = Option(uri.getAuthority).getOrElse("")
+ val at = authority.lastIndexOf('@')
+ val hostPort = if (at >= 0) authority.substring(at + 1) else authority
+ s"$scheme://***@$hostPort"
+ }
+
+ /**
+ * Decline reason when any of `uris` carries userinfo in its authority (e.g.
the container in an
+ * abfss:// path), or `None` when none do. The native store cache,
`ObjectStoreUrl`, and
+ * DataFusion registry all key on scheme/host/port only, dropping userinfo,
so two authorities
+ * differing only in userinfo collide onto the same store handle.
+ */
+ private[delta] def userInfoBearingAuthorityReason(uris: Seq[URI]):
Option[String] = {
+ val offending = uris.filter(uri =>
uriUserInfo(uri).nonEmpty).map(redactedAuthority).distinct
+ if (offending.isEmpty) {
+ None
+ } else {
+ Some("Native Delta scan does not support object-store paths whose
authority carries " +
+ "userinfo (e.g. the container in an abfss:// path): the native
object-store cache, " +
+ "ObjectStoreUrl and DataFusion registry all key on scheme, host and
port only, so two " +
+ "containers on one storage account share a single store handle " +
+ s"(found: ${offending.sorted.mkString(", ")})")
+ }
+ }
+
+ /**
+ * String-literal Hadoop conf keys consulted below. `hadoop-aws` is NOT on
this module's runtime
+ * classpath, so `org.apache.hadoop.fs.s3a.Constants` must never be
referenced here (would raise
+ * `NoClassDefFoundError` for sessions with no S3 dependency).
+ */
+ private val HadoopCredentialProviderPathKey =
"hadoop.security.credential.provider.path"
+ private val S3aCredentialProviderPathKey =
"fs.s3a.security.credential.provider.path"
+
+ /**
+ *
`CommonConfigurationKeysPublic.HADOOP_SECURITY_CREDENTIAL_CLEAR_TEXT_FALLBACK`,
default
+ * `true`, verified via `javap` against `hadoop-common` 3.3.4's
+ * `Configuration#getPasswordFromConfig`: `getPassword` only falls back to
reading a plaintext
+ * conf value once `getBoolean(<this key>, true)` holds -- with the flag
off, a plaintext value
+ * is invisible to every `getPassword`-based resolver, even when no
credential provider is
+ * configured at all.
+ */
+ private val ClearTextFallbackKey =
"hadoop.security.credential.clear-text-fallback"
+
+ private def s3aBucketProviderPathKey(bucket: String): String =
+ s"fs.s3a.bucket.$bucket.security.credential.provider.path"
+
+ /**
+ * The LONG form of [[s3aBucketProviderPathKey]]: `S3AUtils#lookupPassword`
resolves per-bucket
+ * overrides through both a long key (`fs.s3a.bucket.B.<full base key>`) and
a short key; both
+ * must be covered here too.
+ */
+ private def s3aBucketLongProviderPathKey(bucket: String): String =
+ s"fs.s3a.bucket.$bucket.fs.s3a.security.credential.provider.path"
+
+ private def nonEmptyConf(hadoopConf: Configuration, key: String): Boolean =
+ Option(hadoopConf.get(key)).exists(_.nonEmpty)
+
+ /**
+ * The lowercase-scheme-checked S3/S3A bucket name from `uri`'s authority,
or `None` when
+ * `uri`'s scheme is not `s3`/`s3a`. Parses the raw authority manually
rather than
+ * `URI#getHost`, avoiding the same RFC 3986 `reg-name` pitfall as
[[uriAuthority]].
+ */
+ private def s3Bucket(uri: URI): Option[String] = {
+ val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT))
+ if (scheme.contains("s3") || scheme.contains("s3a")) {
+ val authority = Option(uri.getAuthority).getOrElse("")
+ val at = authority.lastIndexOf('@')
+ val hostAndPort = if (at >= 0) authority.substring(at + 1) else authority
+ val colon = hostAndPort.lastIndexOf(':')
+ val host = if (colon >= 0) hostAndPort.substring(0, colon) else
hostAndPort
+ if (host.isEmpty) None else Some(host)
+ } else {
+ None
+ }
+ }
+
+ private def plainValue(hadoopConf: Configuration, key: String):
Option[String] =
+ Option(hadoopConf.get(key)).filter(_.nonEmpty)
+
+ /**
+ * How Hadoop's OWN consumer reads one of the keys compared by
[[s3ConfigDivergenceReason]],
+ * which decides how [[s3KeyDivergenceReason]] computes the Hadoop-effective
side of its
+ * equality check. Exactly two consumer families exist among
[[AllS3ConfigKeys]] in `hadoop-aws`
+ * 3.3.4, each verified via `javap`/CFR against the real call sites (cited
per key on
+ * [[S3ConfigKeyConsumers]]). The tier must mirror the key's ACTUAL
consumer: resolving a
+ * [[PropagatedOptionConsumer]] key through the wider `lookupPassword`
cascade is NOT fail-safe
+ * for a value-EQUALITY comparator -- a long-form alias value Hadoop itself
never reads can
+ * EQUAL native's resolution while Hadoop's true propagate-then-plain-get
value differs, turning
+ * a real divergence into a wrongly-claimed scan (the endpoint
`${...}`-redirect shape pinned in
+ * `DeltaScanContribSuite`).
+ */
+ private[delta] sealed trait S3ConfigConsumer
+
+ /**
+ * Read via `S3AUtils#lookupPassword(bucket, conf, baseKey)`, verified via
`javap` against
+ * `hadoop-aws` 3.3.4: builds `longBucketKey = "fs.s3a.bucket." + bucket +
"." + baseKey` (the
+ * FULL, already-`fs.s3a`-prefixed base key appended after the bucket
segment) and reads it via
+ * `Configuration#getPassword` BEFORE the short-bucket key, keeping the long
value whenever
+ * `getPassword` returns non-empty and only falling through to
short-then-global otherwise.
+ * `getPassword` is Hadoop-credential-provider-aware and skips plaintext
conf entirely when
+ * [[ClearTextFallbackKey]] is false. Modeled by
[[hadoopLookupPasswordEffective]].
+ */
+ private[delta] case object LookupPasswordConsumer extends S3ConfigConsumer
+
+ /**
+ * Read via `S3AUtils#propagateBucketOptions` followed by a plain
`Configuration#get`-family
+ * call (`getTrimmed`/`getBoolean`/`getClasses`) against the propagated
view: the short bucket
+ * form wins only by having overwritten the global key during propagation,
the long bucket form
+ * folds into an unread `fs.s3a.fs.s3a.*` key, and neither a credential
provider nor
+ * [[ClearTextFallbackKey]] is ever consulted. Modeled as a plain
`Configuration#get` on the
+ * [[propagateBucketOptions]] result, which also expands `${...}` references
under that
+ * propagated view exactly like the real consumer.
+ */
+ private[delta] case object PropagatedOptionConsumer extends S3ConfigConsumer
+
+ /**
+ * Every `fs.s3a.*` base key that governs whether a claimed native scan
actually behaves like
+ * Hadoop's own reader would, paired with the consumer family Hadoop
resolves it through -- ONE
+ * list, with each key's resolution tier declared beside it, so a key can
never sit in the
+ * comparator without a deliberate classification (adding one without
picking a tier does not
+ * compile). The entries are every per-bucket `fs.s3a.*` base key native's
S3 client's
+ * `get_config` (s3.rs) resolves, verified directly against its call sites:
+ * `extract_s3_config_options` (endpoint.region, path.style.access, endpoint,
+ * requester.pays.enabled), `lookup_provider_class` (the Comet-specific
+ * credential-provider-class activation key), and
+ * `build_credential_provider`/`build_aws_credential_provider_metadata`/
+ * `build_assume_role_credential_provider_metadata`
(aws.credentials.provider,
+ * assumed.role.credentials.provider, assumed.role.arn,
assumed.role.session.name).
+ *
+ * Tier assignments, each verified via `javap`/CFR against `hadoop-aws`
3.3.4:
+ * - access.key/secret.key/session.token: `S3AUtils#getAWSAccessKeys` and
+ * `MarshalledCredentialBinding#fromFileSystem` (reached from
+ * `TemporaryAWSCredentialsProvider`) resolve all three via
`S3AUtils#lookupPassword` --
+ * [[LookupPasswordConsumer]].
+ * - aws.credentials.provider and assumed.role.credentials.provider:
+ * `S3AUtils#buildAWSProviderList` -> `loadAWSProviderClasses` -> plain
+ * `Configuration#getClasses` -- [[PropagatedOptionConsumer]].
+ * - assumed.role.arn/session.name: `AssumedRoleCredentialProvider`'s
constructor reads both
+ * via plain `Configuration#getTrimmed` -- [[PropagatedOptionConsumer]].
+ * - endpoint (`S3AFileSystem`: `getTrimmed`), endpoint.region
(`DefaultS3ClientFactory`:
+ * `getTrimmed`), path.style.access (`S3AFileSystem`: `getBoolean`) --
+ * [[PropagatedOptionConsumer]].
+ * - requester.pays.enabled: not read anywhere in `hadoop-aws` 3.3.4 (the
constant does not
+ * even exist in its `Constants` class); later releases read it via
plain `getBoolean`
+ * against the propagated conf, so the plain tier is both the faithful
forward model and
+ * inert on 3.3.4 -- [[PropagatedOptionConsumer]].
+ * - comet.credential.provider.class: Comet's own activation key, plain
conf read on both
+ * sides, never a Hadoop key at all -- [[PropagatedOptionConsumer]].
+ *
+ * SYNC NOTE: the key list must stay a superset of native's
`NATIVE_S3A_CONFIG_PROPERTIES`
+ * constant (`native/core/src/parquet/objectstore/s3.rs`, property suffixes
without the
+ * `fs.s3a.` prefix) -- `DeltaScanContribSuite`'s discovery-harness test
asserts this
+ * mechanically against [[AllS3ConfigKeys]]. Literal strings, not the
+ * [[AwsCredentialsProviderKey]] / [[AssumedRoleCredentialsProviderKey]]
vals declared below,
+ * purely to avoid a forward reference inside this `object` body; kept
textually identical to
+ * those two constants.
+ */
+ private[delta] val S3ConfigKeyConsumers: Seq[(String, S3ConfigConsumer)] =
Seq(
Review Comment:
Two settings that change which endpoint native talks to do not appear in
this model:
- `fs.s3a.connection.ssl.enabled`: Hadoop prefixes a scheme-less
`fs.s3a.endpoint` with `http://` when this is false, while native
`normalize_endpoint` (`s3.rs:289-293`) always prefixes `https://`. An on-prem
MinIO or Ceph table with `fs.s3a.endpoint=minio:9000` and SSL off claims
natively and then fails at execution where Spark reads fine. A zero-I/O decline
like the proxy gate would cover it (endpoint has no `://` and the effective
flag is false), and this one is MinIO-testable.
- `fs.s3a.assumed.role.sts.endpoint` (and `.sts.endpoint.region`): Hadoop
sends AssumeRole to the configured STS endpoint, while native builds
`AssumeRoleProvider` with SDK defaults (`s3.rs:879-882`). Same shape as the
session-policy gate: decline when it is set.
Does the discovery harness in `DeltaScanContribSuite` catch either of these?
It looks like it only flags key names containing
key/secret/token/password/encryption, so it would miss both.
##########
native/core/src/execution/delta_dv.rs:
##########
@@ -0,0 +1,2096 @@
+// 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.
+
+//! Delta Lake deletion-vector decoding and translation into DataFusion
+//! [`ParquetAccessPlan`]s (feature = "delta").
+//!
+//! Wire formats implemented here (from delta-spark's `DeletionVectorStore` /
+//! `RoaringBitmapArray`, v3.3.2):
+//! - On-disk DV file: 1 version byte at the start of the file; at
+//! `descriptor.offset`: `[i32 BE size][data: size bytes][i32 BE
CRC32(data)]`.
+//! - `data`: `[i32 LE magic]` then either
+//! - magic 1681511376 ("native"): `[i32 LE count]`, then per bitmap
+//! `[i32 LE size][standard 32-bit RoaringBitmap]`, keys implicit (index);
+//! - magic 1681511377 ("portable", the spec's 64-bit extension): `[i64 LE
+//! count]`, then per bitmap `[i32 LE key][standard 32-bit RoaringBitmap]`
+//! with keys ascending -- exactly [`RoaringTreemap`]'s serialized form.
+
+use std::mem::size_of;
+use std::sync::Arc;
+
+use datafusion::datasource::listing::PartitionedFile;
+use
datafusion::datasource::physical_plan::parquet::metadata::DFParquetMetadata;
+use datafusion::datasource::physical_plan::parquet::{ParquetAccessPlan,
RowGroupAccess};
+use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use datafusion::execution::runtime_env::RuntimeEnv;
+use futures::{StreamExt, TryStreamExt};
+use object_store::path::Path;
+use object_store::{ObjectStore, ObjectStoreExt};
+use parquet::arrow::arrow_reader::{RowSelection, RowSelector};
+use parquet::file::metadata::{PageIndexPolicy, ParquetMetaData};
+use roaring::{RoaringBitmap, RoaringTreemap};
+
+use crate::execution::operators::ExecutionError;
+use crate::execution::operators::ExecutionError::GeneralError;
+use datafusion_comet_proto::spark_operator::DeltaSparkDvDescriptor;
+
+const NATIVE_MAGIC: i32 = 1681511376;
+const PORTABLE_MAGIC: i32 = 1681511377;
+
+/// Unframe a DV blob read from `descriptor.offset` of a DV file:
+/// `[i32 BE size][data][i32 BE crc]`. Verifies both the size against the
+/// descriptor's `size_in_bytes` and the CRC32 checksum.
+pub fn unframe_dv_blob(blob: &[u8], expected_size: usize) -> Result<&[u8],
ExecutionError> {
+ if blob.len() < 8 {
+ return Err(GeneralError(format!(
+ "Deletion vector blob too short: {} bytes",
+ blob.len()
+ )));
+ }
+ let size = i32::from_be_bytes(blob[0..4].try_into().unwrap());
+ if size < 0 || size as usize != expected_size {
+ return Err(GeneralError(format!(
+ "Deletion vector size mismatch: file says {size}, descriptor says
{expected_size}"
+ )));
+ }
+ let end = 4 + size as usize;
+ if blob.len() < end + 4 {
+ return Err(GeneralError(format!(
+ "Deletion vector blob truncated: need {} bytes, have {}",
+ end + 4,
+ blob.len()
+ )));
+ }
+ let data = &blob[4..end];
+ let expected_crc = i32::from_be_bytes(blob[end..end +
4].try_into().unwrap());
+ let actual_crc = crc32fast::hash(data) as i32;
+ if expected_crc != actual_crc {
+ return Err(GeneralError(
+ "Deletion vector checksum mismatch".to_string(),
+ ));
+ }
+ Ok(data)
+}
+
+/// Deserialize the magic-prefixed RoaringBitmapArray into a 64-bit treemap of
+/// deleted row indexes.
+pub fn deserialize_dv_bitmap(data: &[u8]) -> Result<RoaringTreemap,
ExecutionError> {
+ if data.len() < 4 {
+ return Err(GeneralError(
+ "Deletion vector bitmap too short for magic number".to_string(),
+ ));
+ }
+ let magic = i32::from_le_bytes(data[0..4].try_into().unwrap());
+ let rest = &data[4..];
+ match magic {
+ PORTABLE_MAGIC => RoaringTreemap::deserialize_from(rest)
+ .map_err(|e| GeneralError(format!("Invalid portable deletion
vector bitmap: {e}"))),
+ NATIVE_MAGIC => {
+ if rest.len() < 4 {
+ return Err(GeneralError(
+ "Native deletion vector bitmap missing count".to_string(),
+ ));
+ }
+ let count = i32::from_le_bytes(rest[0..4].try_into().unwrap());
+ if count < 0 {
+ return Err(GeneralError(format!(
+ "Invalid RoaringBitmapArray length ({count} < 0)"
+ )));
+ }
+ let mut pos = 4usize;
+ let mut treemap = RoaringTreemap::new();
+ for key in 0..count as u64 {
+ if rest.len() < pos + 4 {
+ return Err(GeneralError(
+ "Native deletion vector bitmap truncated".to_string(),
+ ));
+ }
+ let size = i32::from_le_bytes(rest[pos..pos +
4].try_into().unwrap());
+ pos += 4;
+ if size < 0 || rest.len() < pos + size as usize {
+ return Err(GeneralError(
+ "Native deletion vector bitmap truncated".to_string(),
+ ));
+ }
+ let bitmap = RoaringBitmap::deserialize_from(&rest[pos..pos +
size as usize])
+ .map_err(|e| {
+ GeneralError(format!("Invalid deletion vector
sub-bitmap: {e}"))
+ })?;
+ pos += size as usize;
+ for value in bitmap {
+ treemap.insert((key << 32) | value as u64);
+ }
+ }
+ Ok(treemap)
+ }
+ other => Err(GeneralError(format!(
+ "Unexpected RoaringBitmapArray magic number {other}"
+ ))),
+ }
+}
+
+/// Translate deleted row indexes into a [`ParquetAccessPlan`]: fully-deleted
+/// row groups become `Skip`, untouched groups stay `Scan`, and partially
+/// deleted groups get a `RowSelection` selecting the complement of the deleted
+/// rows. Page-index pruning later INTERSECTS with these selections, so DV
+/// skips and page skips compose.
+pub fn build_access_plan(
+ row_group_row_counts: &[i64],
+ deleted: &RoaringTreemap,
+) -> Result<ParquetAccessPlan, ExecutionError> {
+ let mut plan = ParquetAccessPlan::new_all(row_group_row_counts.len());
+ // Single sweep over the (sorted) deleted row indexes, bucketing by row
group.
+ let mut deleted_iter = deleted.iter().peekable();
+ let mut group_start = 0u64;
+ for (idx, &num_rows) in row_group_row_counts.iter().enumerate() {
+ // A corrupt footer can report a negative row count. `num_rows as u64`
would otherwise
+ // wrap it into a huge positive value, silently corrupting every
row-group boundary
+ // computed from `group_start`/`group_end` below (and therefore which
deleted row indexes
+ // land in which row group) instead of failing loudly.
+ if num_rows < 0 {
+ return Err(GeneralError(format!(
+ "Parquet footer reports a negative row count ({num_rows}) for
row group {idx}"
+ )));
+ }
+ let num_rows = num_rows as u64;
+ let group_end = group_start + num_rows;
+ let mut selectors: Vec<RowSelector> = Vec::new();
+ let mut cursor = group_start;
+ let mut deleted_in_group = 0u64;
+ while let Some(&row) = deleted_iter.peek() {
+ if row >= group_end {
+ break;
+ }
+ deleted_iter.next();
+ deleted_in_group += 1;
+ if row > cursor {
+ selectors.push(RowSelector::select((row - cursor) as usize));
+ }
+ // Merge runs of consecutive deleted rows into one skip.
+ match selectors.last_mut() {
+ Some(last) if last.skip => last.row_count += 1,
+ _ => selectors.push(RowSelector::skip(1)),
+ }
+ cursor = row + 1;
+ }
+ if deleted_in_group == num_rows && num_rows > 0 {
+ plan.skip(idx);
+ } else if deleted_in_group > 0 {
+ if group_end > cursor {
+ selectors.push(RowSelector::select((group_end - cursor) as
usize));
+ }
+ plan.scan_selection(idx, RowSelection::from(selectors));
+ }
+ group_start = group_end;
+ }
+ // A deleted index beyond the file's total row count means the DV does not
+ // belong to this file (stale or corrupted metadata); silently dropping it
+ // would under-apply deletions.
+ if let Some(&row) = deleted_iter.peek() {
+ return Err(GeneralError(format!(
+ "Deletion vector marks row {row} but the file only has
{group_start} rows"
+ )));
+ }
+ Ok(plan)
+}
+
+/// Verify a decoded deletion vector's row count matches the descriptor's
+/// declared `cardinality`, mirroring Delta's JVM reader
+/// (`StoredBitmap.validateCardinality`). The CRC and framing checks catch
+/// corruption but not a stale, otherwise well-formed bitmap whose row count
+/// no longer matches the descriptor -- that would silently under- or
+/// over-delete rows.
+fn validate_cardinality(
+ file_path: &str,
+ expected: i64,
+ deleted: &RoaringTreemap,
+) -> Result<(), ExecutionError> {
+ let actual = deleted.len();
+ if actual != expected as u64 {
+ return Err(GeneralError(format!(
+ "Deletion vector for {file_path} has cardinality mismatch:
descriptor says {expected}, decoded bitmap has {actual} deleted rows"
+ )));
+ }
+ Ok(())
+}
+
+/// One data file plus everything needed to apply its deletion vector. The
+/// file's size comes from `file.object_meta.size` (built by the planner from
+/// the proto's `file_size`).
+///
+/// `data_store` and `dv_store` are resolved by the caller *before* entering
+/// the async `attach_access_plans` runtime (see its doc comment): building an
+/// object store is sync I/O that, for a cold S3 authority, internally issues
+/// its own `Handle::block_on` calls, which panics if nested inside another
+/// `block_on`. Resolving up front means this module never constructs a
+/// store itself.
+pub struct DvScanFile {
+ pub file: PartitionedFile,
+ /// Full URL of the data file (proto `file_path`).
+ pub file_path: String,
+ pub dv: Option<DeltaSparkDvDescriptor>,
+ /// Object store for `file_path`, pre-resolved by the caller. Only read
+ /// when `dv` is `Some` (files without a deletion vector never open their
+ /// footer here), but every file carries one so the struct's shape
+ /// doesn't depend on whether a deletion vector is present.
+ pub data_store: Arc<dyn ObjectStore>,
+ /// Store and within-store path for an on-disk deletion vector's absolute
+ /// path, pre-resolved by the caller. `None` when the file has no
+ /// deletion vector or the deletion vector is stored inline.
+ pub dv_store: Option<(Arc<dyn ObjectStore>, Path)>,
+}
+
+/// Execution-memory-pool reservation covering one file's expanded DV row
selectors across
+/// their *entire* lifetime attached to a scan -- from `build_access_plan`'s
construction
+/// through DataFusion 54.1's reader normalizing the attached
[`ParquetAccessPlan`]
+/// (`create_initial_plan`'s deep clone plus `into_overall_row_selection`'s
combined
+/// `RowSelection`; see [`reader_peak_bytes`]) -- attached to the file's
[`PartitionedFile`]
+/// extensions alongside its [`ParquetAccessPlan`]. The reservation's lifetime
is tied to the
+/// `PartitionedFile` it is attached to, so it is released back to the pool
exactly when the
+/// plan is dropped (query completion or an early-terminated scan), never held
open longer.
+/// Newtype-wrapped so it occupies its own slot in the multi-slot, type-keyed
`extensions` map
+/// (`datafusion_common::extensions::Extensions`) alongside the plan, rather
than a bare
+/// `MemoryReservation` colliding with one some other extension might attach.
+pub struct DvAccessPlanReservation(pub MemoryReservation);
+
+/// Total number of [`RowSelector`]s materialized across `plan`'s per-row-group
+/// selections (`RowGroupAccess::Selection`); `Scan`/`Skip` row groups
+/// contribute none. An alternating deleted/retained bitmap produces one
+/// non-coalescing selector per row (see [`reader_peak_bytes`]'s doc comment
+/// for the worst-case accounting), so this count -- not the deletion
+/// vector's cardinality -- is the thing that must be bounded and reserved
+/// against the execution memory pool.
+fn total_selectors(plan: &ParquetAccessPlan) -> usize {
+ plan.inner()
+ .iter()
+ .map(|access| match access {
+ RowGroupAccess::Selection(selection) => selection.iter().count(),
+ _ => 0,
+ })
+ .sum()
+}
+
+/// Multiplier bounding the peak allocation live *during construction* of one
+/// file's [`RowSelection`]s, relative to the conservative selector-count
+/// bound `S = 2 * cardinality + num_row_groups` (one non-coalescing selector
+/// per deleted row in the worst-case alternating pattern, doubled, plus up to
+/// one extra boundary selector per row group). Split `S` into `r`, the
+/// selectors already retained from row groups `build_access_plan` has
+/// finished, and `c`, the selectors accumulated so far in the current row
+/// group's source `Vec`; `r` and `c` partition the selectors counted toward
+/// `S`, so `r + c <= S` always. While the current group is being built, the
+/// `Vec`'s doubling growth strategy can leave its backing allocation at up to
+/// `2 * c` (the next power-of-two capacity above `c`). Once the group
+/// finishes, `RowSelection::from(Vec)` (parquet's `FromIterator` impl,
+/// `with_capacity` + copy) builds a second, separate `Vec` of size `c` from
+/// that source while the source is still alive, so at the moment the copy
+/// begins, the retained selectors, the current group's doubled source `Vec`,
+/// and the copy are all live simultaneously: `r + 2c + c = r + 3c`. Since
+/// `r >= 0`, `r + 3c <= 3r + 3c = 3(r + c) <= 3S`. 3x covers that peak.
+const CONSTRUCTION_PEAK_FACTOR: usize = 3;
+
+/// Upper bound on how much larger a `Vec`'s backing allocation can be than
its element count
+/// after being built by repeated pushes: `std`'s doubling growth strategy
never leaves a `Vec`
+/// of `n` elements with a backing allocation larger than the next power of
two above `n`, which
+/// is at most `2 * n` for any `n >= 1`.
+const VEC_GROWTH_CAPACITY_FACTOR: usize = 2;
+
+/// `RawVec`'s minimum non-zero capacity for element sizes `<= 1024` bytes
([`RowSelector`] is
+/// 16 bytes on 64-bit platforms: a `usize` row count plus a padded `bool`).
Applied once per
+/// row group (or per contiguous run of row groups) a fresh
`from_fn`/`FlatMap`-driven `Vec`
+/// gets built for (see [`reader_peak_bytes`]), so even a group or run whose
true selector count
+/// is tiny still pays this floor.
+const MIN_VEC_CAPACITY_SELECTORS: usize = 4;
+
+/// Conservative upper bound, in bytes, on the peak allocation live while
DataFusion 54.1's
+/// reader normalizes one file's attached [`ParquetAccessPlan`] -- the
allocation this module's
+/// steady-state reservation must cover, not merely the plan's own retained
selector bytes.
+/// THREE allocations can be live simultaneously by the time
`into_overall_row_selection`
+/// returns, not two -- the clone is only exact when page-index pruning never
touches it:
+///
+/// 1. **Attached original** (`selectors`, exact): `create_initial_plan`
deep-clones the
+/// attached plan while the original remains reachable from the file's
`extensions` until
+/// the scan consumes it. The ORIGINAL's own selector `Vec`s are exact -- a
coalesced
+/// [`RowSelection`] built via `RowSelection::from(Vec<RowSelector>)` (what
+/// `build_access_plan` uses) has no excess capacity, because that
conversion is a plain
+/// `with_capacity(len)` copy, not a `size_hint`-blind fold.
+/// 2. **The clone, possibly capacity-inflated** (`<=
VEC_GROWTH_CAPACITY_FACTOR * selectors +
+/// MIN_VEC_CAPACITY_SELECTORS * num_row_groups`): if page-index pruning
fires
+/// (`PagePruningAccessPlanFilter`; `access_plan.rs`'s `scan_selection` on
a row group that
+/// already carries a `RowGroupAccess::Selection` calls
`existing.intersection(&page_derived)`
+/// -- `RowSelection::intersection` -> `intersect_row_selections`), it
replaces the CLONE's
+/// per-row-group selection with that intersection's output.
`intersect_row_selections` is
+/// ANOTHER `from_fn` generator with `size_hint() == (0, None)`, so each
intersected row
+/// group's backing `Vec` starts at `with_capacity(0)` and doubles as it
grows, independent
+/// of whatever capacity the pre-intersection selection had. This inflated
clone is still
+/// live when `into_overall_row_selection` later moves its buffer. Term 1's
exactness
+/// guarantee holds for the ORIGINAL always, and for the clone only when
page-index pruning
+/// never fires against it -- once it does, the clone must be charged at
the SAME
+/// growth-capped bound as a fresh combined-selection `Vec` (term 3),
summed once per row
+/// group rather than once per run, since each row group's `Selection` is
intersected
+/// independently.
+/// 3. **Per-run combined-selection allocation** (`<=
VEC_GROWTH_CAPACITY_FACTOR * (selectors +
+/// num_row_groups) + MIN_VEC_CAPACITY_SELECTORS * num_row_groups`):
`into_overall_row_selection`
+/// collects each contiguous run of row groups' selectors into a *new*
`RowSelection` via a
+/// `FlatMap` whose `size_hint().0 == 0`, so that run's `Vec` starts at
`with_capacity(0)`
+/// and doubles as it grows -- capping its backing allocation at
+/// `max(MIN_VEC_CAPACITY_SELECTORS, next_power_of_two(len))`, which is at
most
+/// `MIN_VEC_CAPACITY_SELECTORS + VEC_GROWTH_CAPACITY_FACTOR * len` for a
run of `len`
+/// selectors. `len` is at most that run's share of `selectors` plus one
boundary selector
+/// per `RowGroupAccess::Scan` row group in the run (`Scan` always
contributes exactly one
+/// `RowSelector::select(num_rows)`; see `access_plan.rs`'s
`into_overall_row_selection`).
+/// Summing across at most `num_row_groups` runs (each spans >= 1 row
group) bounds the total
+/// at `VEC_GROWTH_CAPACITY_FACTOR * selectors +
(MIN_VEC_CAPACITY_SELECTORS +
+/// VEC_GROWTH_CAPACITY_FACTOR) * num_row_groups`.
+///
+/// Summing all three terms and converting to bytes: `((1 + 2 *
VEC_GROWTH_CAPACITY_FACTOR) *
+/// selectors + (2 * MIN_VEC_CAPACITY_SELECTORS + VEC_GROWTH_CAPACITY_FACTOR)
* num_row_groups)
+/// * size_of::<RowSelector>()` -- with the constants above, `(5 * selectors +
10 *
+/// num_row_groups) * size_of::<RowSelector>()`. Checked against two
measured worst cases:
+///
+/// - No page-index pruning (the original P2 report; term 2 stays exact): one
2,000,000-row
+/// group, 1,000,000 alternating deletions, `selectors = 2,000,000`.
Measured allocator peak
+/// 97,554,457 B; the byte-for-byte accounting for the attached original
plus the (here,
+/// exact) clone plus the inflated combined selection explains 97,554,432 B
of that, a 25 B
+/// residue we did not attribute. This bound gives 160,000,160 B -- much
looser here because
+/// it must also cover the next case, where the clone is NOT exact.
+/// - Page-index pruning fires against the clone: one 1,048,577-row group,
`selectors =
+/// 1,048,577`. Measured peak 83,886,096 B; this bound gives 83,886,320 B (a
224 B, <1%
+/// margin -- deliberately tight, since this is the case that drives the
bound).
+///
+/// Uses checked arithmetic throughout: a selector or row-group count large
enough to overflow
+/// `usize` indicates a corrupted or malicious input, reported as a clean
error rather than
+/// panicking.
+fn reader_peak_bytes(selectors: usize, num_row_groups: usize) -> Result<usize,
ExecutionError> {
+ let overflow = || {
+ GeneralError(format!(
+ "Deletion vector reader-peak bound overflowed for {selectors}
selectors and \
+ {num_row_groups} row groups"
+ ))
+ };
+ // Term 1: the attached original -- exact, untouched by page-index pruning
(only the clone
+ // is ever intersected; see the doc comment above).
+ let attached_term = selectors;
+ // Term 2: the clone, bounded as if page-index pruning DID fire against
every row group
+ // (safe even when it doesn't: term 2's bound is always >= `selectors`, so
it never
+ // undershoots the exact case either).
+ let clone_growth = selectors
+ .checked_mul(VEC_GROWTH_CAPACITY_FACTOR)
+ .ok_or_else(overflow)?;
+ let clone_floor = num_row_groups
+ .checked_mul(MIN_VEC_CAPACITY_SELECTORS)
+ .ok_or_else(overflow)?;
+ let clone_term =
clone_growth.checked_add(clone_floor).ok_or_else(overflow)?;
+ // Term 3: into_overall_row_selection's per-run combined-selection
allocation.
+ let combined_growth = selectors
+ .checked_mul(VEC_GROWTH_CAPACITY_FACTOR)
+ .ok_or_else(overflow)?;
+ let combined_floor = num_row_groups
+ .checked_mul(MIN_VEC_CAPACITY_SELECTORS + VEC_GROWTH_CAPACITY_FACTOR)
+ .ok_or_else(overflow)?;
+ let combined_term = combined_growth
+ .checked_add(combined_floor)
+ .ok_or_else(overflow)?;
+
+ let selector_bound = attached_term
+ .checked_add(clone_term)
+ .and_then(|sum| sum.checked_add(combined_term))
+ .ok_or_else(overflow)?;
+ selector_bound
+ .checked_mul(size_of::<RowSelector>())
+ .ok_or_else(overflow)
+}
+
+/// Upper bound, in [`RowSelector`]s, on how many extra selectors the parquet
reader's
+/// page-index pruning can add on top of the deletion vector's own selection
when normalizing
+/// one file, from that file's already-fetched [`ParquetMetaData`].
+///
+/// `intersect_row_selections` (parquet's `selection.rs`), which combines a
page-pruning
+/// selection with the deletion vector's selection, is a `from_fn` generator
whose
+/// `size_hint()` is `(0, None)`: for inputs of length `a` and `b`, its output
can have up to
+/// `a + b` selectors -- longer than either input. Bounding the page-pruning
side of that sum
+/// requires knowing how many selectors a page-index-derived selection could
produce: at most
+/// two per data page (one skip, one select, in the worst case of alternating
page-level
+/// pruning decisions), summed over every column of every row group.
+///
+/// Returns `0` when `metadata` carries no offset index
(`metadata.offset_index()` is `None`).
+/// This is provably safe, not merely a convenient default: page-index pruning
cannot produce a
+/// page-level selection without the offset index to locate pages by, so there
are no
+/// page-pruning selectors to bound. The offset index is fetched with
+/// `PageIndexPolicy::Optional` from the same `FileMetadataCache` entry the
scan's reader later
+/// reopens (see [`attach_access_plan`]'s footer-fetch comment), so this
function observes
+/// exactly what the reader will see.
+///
+/// Uses checked arithmetic throughout for the same reason as
[`admission_bound_bytes`].
+fn page_selection_bound_selectors(metadata: &ParquetMetaData) -> Result<usize,
ExecutionError> {
+ let Some(offset_index) = metadata.offset_index() else {
+ return Ok(0);
+ };
+ let overflow = || {
+ GeneralError(
+ "Deletion vector page-selection bound overflowed while summing
offset-index page \
+ locations"
+ .to_string(),
+ )
+ };
+ let mut total_page_locations = 0usize;
+ for row_group in offset_index {
+ for column in row_group {
+ total_page_locations = total_page_locations
+ .checked_add(column.page_locations().len())
+ .ok_or_else(overflow)?;
+ }
+ }
+ total_page_locations.checked_mul(2).ok_or_else(overflow)
+}
+
+/// Execution-memory-pool admission bound, in bytes, for one file's
deletion-vector access
+/// plan -- reserved *before* calling `build_access_plan` (see
[`attach_access_plan`]'s
+/// pre-reserve call site) to cover the larger of two peaks live at different
points in the
+/// plan's lifetime. In practice the reader-normalization peak below dominates
the construction
+/// peak unconditionally for any non-trivial input (`reader_peak_bytes(S, G) =
(5S + 10G) *
+/// size_of::<RowSelector>()` always exceeds `CONSTRUCTION_PEAK_FACTOR * S *
+/// size_of::<RowSelector>() = 3S * size_of::<RowSelector>()` once `S >= 1`,
since the `5S` term
+/// alone already exceeds `3S`); the construction term is retained as a
documented floor rather
+/// than dropped, since it is cheap to compute and keeps this bound correct
even if the reader's
+/// growth factors ever shrink below construction's.
+///
+/// - **Construction peak** (`CONSTRUCTION_PEAK_FACTOR * S`, see that
constant's doc comment):
+/// live while `build_access_plan` builds the plan's `RowSelection`s.
Construction's
+/// transient allocations fully unwind before `build_access_plan` returns,
so this peak never
+/// overlaps the reader-normalization peak below.
+/// - **Reader-normalization peak** (`reader_peak_bytes(S +
page_bound_selectors,
+/// num_row_groups)`, see that function): live later, once DataFusion's
reader normalizes the
+/// attached plan. `S = 2 * cardinality + num_row_groups` is the same
conservative bound on
+/// the plan's final retained selector count used for the construction peak
-- it provably
+/// bounds `R = total_selectors(&plan)` (`R <= S`, from `build_access_plan`'s
+/// one-non-coalescing-selector-per-deleted-row worst case plus one boundary
selector per row
+/// group), so `S + page_bound_selectors` bounds `R` after page-index
inflation the same way
+/// `S` bounds `R` before it.
+///
+/// These two peaks never overlap in time, so `max` -- not `sum` -- is the
correct combinator:
+/// reserving their sum would over-reserve for no safety benefit.
+///
+/// Deliberately not clamped by the file's total row count here, unlike the
reader-peak target
+/// `attach_access_plan` resizes down to after construction (see that call
site): `S`'s
+/// `+ num_row_groups` boundary term is a worst-case padding margin that can
legitimately exceed
+/// the total row count for a small, heavily-deleted file, and admission
sizing has no actual
+/// retained-selector count yet to clamp against -- only after construction,
once `R` is known,
+/// is clamping to the total row count both meaningful and strictly tighter.
Leaving this bound
+/// unclamped only ever makes admission more conservative, never less safe.
+///
+/// Uses checked arithmetic throughout: a cardinality, row-group count, or
page bound large
+/// enough to overflow `usize` while computing this bound indicates a
corrupted or malicious
+/// descriptor, reported as a clean error rather than panicking.
+fn admission_bound_bytes(
+ cardinality: i64,
+ num_row_groups: usize,
+ page_bound_selectors: usize,
+) -> Result<usize, ExecutionError> {
+ let overflow = || {
+ GeneralError(format!(
+ "Deletion vector admission bound overflowed for cardinality
{cardinality}, \
+ {num_row_groups} row groups, and page bound
{page_bound_selectors} selectors"
+ ))
+ };
+ let cardinality_usize = usize::try_from(cardinality).map_err(|_|
overflow())?;
+ // S: the conservative bound on the plan's final *retained* selector count
(what
+ // `total_selectors(&plan)` cannot exceed) -- unchanged from the
pre-existing
+ // construction-only bound this function replaces.
+ let s = cardinality_usize
+ .checked_mul(2)
+ .and_then(|doubled| doubled.checked_add(num_row_groups))
+ .ok_or_else(overflow)?;
+
+ let construction_bytes = s
+ .checked_mul(size_of::<RowSelector>())
+ .and_then(|bytes| bytes.checked_mul(CONSTRUCTION_PEAK_FACTOR))
+ .ok_or_else(overflow)?;
+
+ let s_plus_page =
s.checked_add(page_bound_selectors).ok_or_else(overflow)?;
+ let reader_bytes = reader_peak_bytes(s_plus_page, num_row_groups)?;
+
+ Ok(construction_bytes.max(reader_bytes))
+}
+
+/// Upper bound on concurrent DV-blob and footer fetches per partition. Both
+/// are small ranged reads, so a modest fan-out hides object-store latency
+/// without flooding the store client.
+const DV_FETCH_CONCURRENCY: usize = 8;
+
+/// Called via `block_on` at plan-creation time on the executor task: DV blobs
+/// are small ranged reads and footers are needed to learn row-group
+/// boundaries. Files are fetched concurrently (bounded by
+/// [`DV_FETCH_CONCURRENCY`]) with input order preserved. Footer fetches go
+/// through the scan's shared FileMetadataCache, so the scan's subsequent open
+/// of the same file is served from cache. That reuse relies on each input
+/// [`PartitionedFile`] being returned as-is (only `with_extension` applied),
+/// never rebuilt: the cache entry is keyed by this exact `object_meta` and the
+/// scan later looks it up through the same struct.
+///
+/// Deliberately takes no object-store options map and imports no
+/// store-construction helper: every [`DvScanFile`] arrives with its stores
+/// already resolved by the caller (see its doc comment), so this async path
+/// structurally cannot build an object store -- only `runtime_env` is still
+/// threaded through, for the shared `FileMetadataCache` and (per file) the
+/// execution `MemoryPool` each expanded access plan's row selectors are
+/// reserved against -- see [`DvAccessPlanReservation`].
+pub async fn attach_access_plans(
+ runtime_env: Arc<RuntimeEnv>,
+ files: Vec<DvScanFile>,
+) -> Result<Vec<PartitionedFile>, ExecutionError> {
+ futures::stream::iter(files)
+ .map(|scan_file| attach_access_plan(Arc::clone(&runtime_env),
scan_file))
+ .buffered(DV_FETCH_CONCURRENCY)
+ .try_collect()
+ .await
+}
+
+/// Resolve one file's deletion vector into an attached [`ParquetAccessPlan`];
+/// files without a DV pass through untouched.
+async fn attach_access_plan(
+ runtime_env: Arc<RuntimeEnv>,
+ scan_file: DvScanFile,
+) -> Result<PartitionedFile, ExecutionError> {
+ let DvScanFile {
+ file,
+ file_path,
+ dv,
+ data_store,
+ dv_store,
+ } = scan_file;
+ let dv = match dv {
+ Some(dv) => dv,
+ None => return Ok(file),
+ };
+ // Delta's canonical `DeletionVectorDescriptor.EMPTY`: inline storage,
empty
+ // payload, size 0, cardinality 0. Spark's reader returns all rows for it;
+ // decoding would fail (the empty payload is too short for a magic
+ // number), so pass the file through unchanged before attempting to read
it.
+ if dv.cardinality == 0 && dv.size_in_bytes == 0 {
+ return Ok(file);
+ }
+ if dv.size_in_bytes < 0 {
+ return Err(GeneralError(format!(
+ "Deletion vector for {file_path} has negative size {}",
+ dv.size_in_bytes
+ )));
+ }
+ if dv.cardinality < 0 {
+ return Err(GeneralError(format!(
+ "Deletion vector for {file_path} has negative cardinality {}",
+ dv.cardinality
+ )));
+ }
+
+ let data: Vec<u8> = if let Some(inline) = dv.inline_data {
Review Comment:
The on-disk branch verifies the payload against `size_in_bytes` in
`unframe_dv_blob`, but the inline branch takes `inline_data` as-is, so an
inline payload whose length disagrees with the descriptor decodes silently.
Since the JVM does the z85 decode, this is the only native check point for
inline DVs. Could it compare the length before `deserialize_dv_bitmap`?
##########
native/core/src/execution/delta_dv.rs:
##########
@@ -0,0 +1,2096 @@
+// 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.
+
+//! Delta Lake deletion-vector decoding and translation into DataFusion
+//! [`ParquetAccessPlan`]s (feature = "delta").
+//!
+//! Wire formats implemented here (from delta-spark's `DeletionVectorStore` /
+//! `RoaringBitmapArray`, v3.3.2):
+//! - On-disk DV file: 1 version byte at the start of the file; at
+//! `descriptor.offset`: `[i32 BE size][data: size bytes][i32 BE
CRC32(data)]`.
+//! - `data`: `[i32 LE magic]` then either
+//! - magic 1681511376 ("native"): `[i32 LE count]`, then per bitmap
+//! `[i32 LE size][standard 32-bit RoaringBitmap]`, keys implicit (index);
+//! - magic 1681511377 ("portable", the spec's 64-bit extension): `[i64 LE
+//! count]`, then per bitmap `[i32 LE key][standard 32-bit RoaringBitmap]`
+//! with keys ascending -- exactly [`RoaringTreemap`]'s serialized form.
+
+use std::mem::size_of;
+use std::sync::Arc;
+
+use datafusion::datasource::listing::PartitionedFile;
+use
datafusion::datasource::physical_plan::parquet::metadata::DFParquetMetadata;
+use datafusion::datasource::physical_plan::parquet::{ParquetAccessPlan,
RowGroupAccess};
+use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use datafusion::execution::runtime_env::RuntimeEnv;
+use futures::{StreamExt, TryStreamExt};
+use object_store::path::Path;
+use object_store::{ObjectStore, ObjectStoreExt};
+use parquet::arrow::arrow_reader::{RowSelection, RowSelector};
+use parquet::file::metadata::{PageIndexPolicy, ParquetMetaData};
+use roaring::{RoaringBitmap, RoaringTreemap};
+
+use crate::execution::operators::ExecutionError;
+use crate::execution::operators::ExecutionError::GeneralError;
+use datafusion_comet_proto::spark_operator::DeltaSparkDvDescriptor;
+
+const NATIVE_MAGIC: i32 = 1681511376;
+const PORTABLE_MAGIC: i32 = 1681511377;
+
+/// Unframe a DV blob read from `descriptor.offset` of a DV file:
+/// `[i32 BE size][data][i32 BE crc]`. Verifies both the size against the
+/// descriptor's `size_in_bytes` and the CRC32 checksum.
+pub fn unframe_dv_blob(blob: &[u8], expected_size: usize) -> Result<&[u8],
ExecutionError> {
+ if blob.len() < 8 {
+ return Err(GeneralError(format!(
+ "Deletion vector blob too short: {} bytes",
+ blob.len()
+ )));
+ }
+ let size = i32::from_be_bytes(blob[0..4].try_into().unwrap());
+ if size < 0 || size as usize != expected_size {
+ return Err(GeneralError(format!(
+ "Deletion vector size mismatch: file says {size}, descriptor says
{expected_size}"
+ )));
+ }
+ let end = 4 + size as usize;
+ if blob.len() < end + 4 {
+ return Err(GeneralError(format!(
+ "Deletion vector blob truncated: need {} bytes, have {}",
+ end + 4,
+ blob.len()
+ )));
+ }
+ let data = &blob[4..end];
+ let expected_crc = i32::from_be_bytes(blob[end..end +
4].try_into().unwrap());
+ let actual_crc = crc32fast::hash(data) as i32;
+ if expected_crc != actual_crc {
+ return Err(GeneralError(
+ "Deletion vector checksum mismatch".to_string(),
+ ));
+ }
+ Ok(data)
+}
+
+/// Deserialize the magic-prefixed RoaringBitmapArray into a 64-bit treemap of
+/// deleted row indexes.
+pub fn deserialize_dv_bitmap(data: &[u8]) -> Result<RoaringTreemap,
ExecutionError> {
+ if data.len() < 4 {
+ return Err(GeneralError(
+ "Deletion vector bitmap too short for magic number".to_string(),
+ ));
+ }
+ let magic = i32::from_le_bytes(data[0..4].try_into().unwrap());
+ let rest = &data[4..];
+ match magic {
+ PORTABLE_MAGIC => RoaringTreemap::deserialize_from(rest)
+ .map_err(|e| GeneralError(format!("Invalid portable deletion
vector bitmap: {e}"))),
+ NATIVE_MAGIC => {
+ if rest.len() < 4 {
+ return Err(GeneralError(
+ "Native deletion vector bitmap missing count".to_string(),
+ ));
+ }
+ let count = i32::from_le_bytes(rest[0..4].try_into().unwrap());
+ if count < 0 {
+ return Err(GeneralError(format!(
+ "Invalid RoaringBitmapArray length ({count} < 0)"
+ )));
+ }
+ let mut pos = 4usize;
+ let mut treemap = RoaringTreemap::new();
+ for key in 0..count as u64 {
+ if rest.len() < pos + 4 {
+ return Err(GeneralError(
+ "Native deletion vector bitmap truncated".to_string(),
+ ));
+ }
+ let size = i32::from_le_bytes(rest[pos..pos +
4].try_into().unwrap());
+ pos += 4;
+ if size < 0 || rest.len() < pos + size as usize {
+ return Err(GeneralError(
+ "Native deletion vector bitmap truncated".to_string(),
+ ));
+ }
+ let bitmap = RoaringBitmap::deserialize_from(&rest[pos..pos +
size as usize])
+ .map_err(|e| {
+ GeneralError(format!("Invalid deletion vector
sub-bitmap: {e}"))
+ })?;
+ pos += size as usize;
+ for value in bitmap {
+ treemap.insert((key << 32) | value as u64);
+ }
+ }
+ Ok(treemap)
+ }
+ other => Err(GeneralError(format!(
+ "Unexpected RoaringBitmapArray magic number {other}"
+ ))),
+ }
+}
+
+/// Translate deleted row indexes into a [`ParquetAccessPlan`]: fully-deleted
+/// row groups become `Skip`, untouched groups stay `Scan`, and partially
+/// deleted groups get a `RowSelection` selecting the complement of the deleted
+/// rows. Page-index pruning later INTERSECTS with these selections, so DV
+/// skips and page skips compose.
+pub fn build_access_plan(
+ row_group_row_counts: &[i64],
+ deleted: &RoaringTreemap,
+) -> Result<ParquetAccessPlan, ExecutionError> {
+ let mut plan = ParquetAccessPlan::new_all(row_group_row_counts.len());
+ // Single sweep over the (sorted) deleted row indexes, bucketing by row
group.
+ let mut deleted_iter = deleted.iter().peekable();
+ let mut group_start = 0u64;
+ for (idx, &num_rows) in row_group_row_counts.iter().enumerate() {
+ // A corrupt footer can report a negative row count. `num_rows as u64`
would otherwise
+ // wrap it into a huge positive value, silently corrupting every
row-group boundary
+ // computed from `group_start`/`group_end` below (and therefore which
deleted row indexes
+ // land in which row group) instead of failing loudly.
+ if num_rows < 0 {
+ return Err(GeneralError(format!(
+ "Parquet footer reports a negative row count ({num_rows}) for
row group {idx}"
+ )));
+ }
+ let num_rows = num_rows as u64;
+ let group_end = group_start + num_rows;
Review Comment:
This is the one unchecked add in an otherwise fully checked path; a corrupt
footer with two row groups near `i64::MAX` panics here in debug and wraps in
release, which then misfires the "beyond total rows" check below. `checked_add`
with the existing error style would match the rest.
On tests: the on-disk fixture writes a single blob at offset 1, and the
access-plan test deletes rows in the middle of a group. Two DVs in one on-disk
file (exercising the `offset..offset+framed_len` slicing) and a deleted row on
a row-group boundary (last row of group k and first row of k+1) would cover
paths that are currently untested.
##########
native/core/Cargo.toml:
##########
@@ -103,12 +106,25 @@ datafusion-functions-nested = { version = "55.1.0" }
[features]
backtrace = ["datafusion/backtrace"]
-default = ["hdfs-opendal"]
+default = ["hdfs-opendal", "delta"]
hdfs-opendal = ["opendal", "object_store_opendal", "hdfs-sys"]
jemalloc = ["tikv-jemallocator", "tikv-jemalloc-ctl"]
-# Delta Lake integration. When enabled, links the `comet-contrib-delta` crate
-# into `libcomet` and activates the `OpStruct::DeltaScan` dispatcher arm.
-# Default builds carry zero Delta surface.
+# Native Delta Lake scan support for the JVM-planned path
(contrib/delta-spark).
+# In the default set: inert at runtime unless the contrib jar is on the
+# classpath (ServiceLoader) AND spark.comet.scan.delta.enabled is set, so it
+# cannot affect non-Delta scans. Opt out with --no-default-features for slim
Review Comment:
Could this state the rationale directly rather than pointing at the PR
review? Something like: inert without both the contrib jar and the config, and
no new crates since `roaring` and `crc32fast` are already in the tree. The git
history already records the discussion. The footprint figure here (82 KB) also
differs from the one in `dev/verify-contrib-delta-gate.sh` (84 KB); one number
in one place would be enough.
##########
native/core/src/parquet/datetime_rebase.rs:
##########
@@ -0,0 +1,2662 @@
+// 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.
+
+//! Per-file datetime calendar-rebase handling for the parquet scan.
+//!
+//! Spark 2.4 and earlier wrote dates and timestamps in the hybrid Julian +
Gregorian calendar;
+//! Spark 3.0+ uses the proleptic Gregorian calendar and records the calendar
policy of every
+//! file it writes in the parquet footer's key-value metadata
(`org.apache.spark.version`,
+//! `org.apache.spark.legacyDateTime`, `org.apache.spark.legacyINT96`,
+//! `org.apache.spark.timeZone`). Spark's reader resolves the rebase policy
from EACH FILE's
+//! writer metadata (`DataSourceUtils.datetimeRebaseSpec` / `int96RebaseSpec`)
-- the session's
+//! `spark.sql.parquet.datetimeRebaseModeInRead` conf only applies to files
whose metadata does
+//! not decide the policy on its own -- so a reader that ignores the metadata
silently returns
+//! values shifted by up to ten days for dates before 1582-10-15 (e.g.
`1500-01-01` reads as
+//! `1500-01-10`).
+//!
+//! This module mirrors that per-file resolution:
[`resolve_file_rebase_policies`] computes the
+//! date / INT64-timestamp / INT96-timestamp policies from a file's arrow
schema metadata (the
+//! parquet key-value pairs survive the parquet -> arrow schema conversion),
and
+//! [`wrap_datetime_rebase`] wraps the per-file rewritten expressions' column
references in a
+//! [`SparkDatetimeRebaseExpr`] that rebases values exactly where that is
possible without the
+//! JVM's historical timezone tables (dates always; timestamps for a fixed UTC
writer zone) and
+//! refuses -- rather than silently corrupting -- ancient values it cannot
rebase. Nested
+//! columns are rebuilt leaf by leaf (struct / list / map / fixed-size list /
dictionary), each
+//! leaf under its own policy, with nulls and offsets preserved. Modern values
are always the
+//! identity under every policy: from 1582-10-15 onward for dates, and from
+//! [`LAST_SWITCH_JULIAN_TS_SECONDS`] (1900-01-01T00:00:00Z, Spark's
+//! `RebaseDateTime.lastSwitchJulianTs`) onward for timestamps.
+//!
+//! Spark applies `datetimeRebaseSpec` to INT64 `TIMESTAMP_MICROS` /
`TIMESTAMP_MILLIS` columns
+//! and `int96RebaseSpec` to INT96 columns. The two physical types are
indistinguishable in the
+//! arrow schema DataFusion hands the expression adapter (both surface as
`Timestamp(us, "UTC")`
+//! after INT96 coercion), so Comet's parquet reader factory stamps the file's
INT96 leaf
+//! ordinals -- taken from the parquet footer's own `SchemaDescriptor` -- into
the key-value
+//! metadata under [`INT96_LEAVES_METADATA_KEY`] before the arrow schema is
derived (see
+//! [`stamp_int96_leaves`] and `eager_page_index_reader_factory.rs`), and the
adapter attributes
+//! every timestamp leaf to its spec from that stamp. Without a stamp, the two
specs are merged:
+//! agreement decides, disagreement degrades to [`RebasePolicy::CheckAncient`].
+//!
+//! The wrapper sits BENEATH the schema adapter's nested narrowing (the struct
-> struct convert
+//! that keeps only the requested children), which is what keeps those
ordinals physical -- but
+//! it means the wrapper sees every physical child, requested or not. Spark
only ever decodes
+//! the requested nested schema, so
[`FileRebasePolicies::restrict_to_requested`] marks the
+//! physical leaves the narrowing drops as the identity: an unrequested
ancient `s.ts` never
+//! blocks `select s.d`, exactly as in Spark.
+//!
+//! Currently only enabled by the Delta scan arms via
+//! `SparkParquetOptions::rebase_from_file_metadata`, which also carries the
session read modes
+//! ([`SessionRebaseModes`], forwarded from the JVM) that decide the policy
for files without
+//! Spark writer metadata; the plain NativeScan keeps its documented no-rebase
behavior (see
+//! the compatibility guide and issue #5010).
+
+use std::collections::HashMap;
+use std::fmt::{self, Display};
+use std::hash::{Hash, Hasher};
+use std::sync::Arc;
+
+use arrow::array::{
+ Array, ArrayRef, AsArray, Date32Array, FixedSizeListArray,
GenericListArray, MapArray,
+ OffsetSizeTrait, PrimitiveArray, RecordBatch, StructArray,
+};
+use arrow::datatypes::{
+ ArrowTimestampType, DataType, Date32Type, FieldRef, Schema, SchemaRef,
TimeUnit,
+ TimestampMicrosecondType, TimestampMillisecondType,
TimestampNanosecondType,
+ TimestampSecondType,
+};
+use arrow::error::ArrowError;
+use datafusion::common::tree_node::{Transformed, TreeNode};
+use datafusion::common::{DataFusionError, Result as DataFusionResult};
+use datafusion::physical_expr::expressions::Column;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ColumnarValue;
+use parquet::basic::Type as ParquetPhysicalType;
+use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaData};
+use parquet::schema::types::SchemaDescriptor;
+
+use super::name_fold::fold_names;
+use super::schema_adapter::parse_field_id;
+
+/// Footer key naming the Spark release that wrote the file; absent for
non-Spark writers.
+const SPARK_VERSION_METADATA_KEY: &str = "org.apache.spark.version";
+/// Present (empty value) when the file's dates and INT64 timestamps were
written with
+/// `spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY`.
+const SPARK_LEGACY_DATETIME_KEY: &str = "org.apache.spark.legacyDateTime";
+/// Present (empty value) when the file's INT96 timestamps were written with
+/// `spark.sql.parquet.int96RebaseModeInWrite=LEGACY`.
+const SPARK_LEGACY_INT96_KEY: &str = "org.apache.spark.legacyINT96";
+/// The writer session's time zone, stamped alongside either legacy flag.
+const SPARK_TIMEZONE_KEY: &str = "org.apache.spark.timeZone";
+
+/// Key-value metadata entry Comet's parquet reader factory adds to a file's
footer metadata
+/// (in memory only, never written back) so the expression adapter can tell
INT96 timestamp
+/// columns from INT64 ones after both have been coerced to the same arrow
type. Value:
+/// `"<leaf count>:<comma-separated INT96 leaf ordinals>"`, where leaves are
the file's
+/// primitive columns in `SchemaDescriptor::columns()` order -- the same
depth-first order
+/// parquet-rs assigns arrow leaves, so an arrow-side depth-first walk lines
up with it. The
+/// leaf count lets the reader detect a stamp that does not describe the
schema it is paired
+/// with (see [`Int96Attribution::from_schema`]).
+pub(crate) const INT96_LEAVES_METADATA_KEY: &str = "comet.int96_leaf_columns";
+
+/// Day of the Gregorian cutover (1582-10-15) as days since the epoch;
rebasing is the identity
+/// from this day onward. Same value as Spark's
`RebaseDateTime.lastSwitchJulianDay`.
+const LAST_SWITCH_JULIAN_DAY: i32 = -141427;
+
+/// Spark's `RebaseDateTime.lastSwitchJulianTs` (and `lastSwitchGregorianTs`)
in seconds since
+/// the epoch: 1900-01-01T00:00:00Z. Spark derives it as the latest switch
instant across every
+/// zone in its `julian-gregorian-rebase-micros.json` table
(`getLastSwitchTs`, which also
+/// asserts the calendars' difference is zero for every zone from then on):
most zones ran on
+/// local mean time before 1900, so the last instant at which rebasing changes
a value in ANY
+/// zone is 1900-01-01T00:00:00Z, not the 1582 cutover.
`createTimestampRebaseFuncInRead`
+/// under `EXCEPTION` throws exactly for `micros < lastSwitchJulianTs` (after
converting
+/// `TIMESTAMP_MILLIS` to micros), and `rebaseJulianToGregorianMicros` is the
identity from it
+/// onward in every zone. The value is in seconds so it scales exactly to any
timestamp unit.
+pub(crate) const LAST_SWITCH_JULIAN_TS_SECONDS: i64 = -2_208_988_800;
+
+/// The per-century differences between the Julian and proleptic Gregorian
calendars, and the
+/// Julian-calendar switch days at which each difference starts to apply.
Copied verbatim from
+/// Spark's `RebaseDateTime.julianGregDiffs` / `julianGregDiffSwitchDay`
(which Spark generated
+/// from `localRebaseJulianToGregorianDays`);
`rebase_julian_to_gregorian_days` must stay
+/// value-for-value equal to Spark's `rebaseJulianToGregorianDays`.
+const JULIAN_GREG_DIFFS: [i32; 14] = [2, 1, 0, -1, -2, -3, -4, -5, -6, -7, -8,
-9, -10, 0];
+const JULIAN_GREG_DIFF_SWITCH_DAY: [i32; 14] = [
+ -719164, -682945, -646420, -609895, -536845, -500320, -463795, -390745,
-354220, -317695,
+ -244645, -208120, -171595, -141427,
+];
+
+/// Proleptic-Gregorian days since 1970-01-01 for a nominal civil date, via
Howard Hinnant's
+/// `days_from_civil`. `d` may exceed the month's length; the excess rolls
into the following
+/// month exactly like `LocalDate.of(y, m, 1).plusDays(d - 1)` in Spark's
+/// `localRebaseJulianToGregorianDays` (how the non-existent proleptic date
`1000-02-29`,
+/// valid in the Julian calendar, lands on `1000-03-01`).
+fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
+ let y = if m <= 2 { y - 1 } else { y };
+ let era = y.div_euclid(400);
+ let yoe = y - era * 400; // [0, 399]
+ let mp = (m + 9) % 12; // [0, 11], March = 0
+ let doy = (153 * mp + 2) / 5 + d - 1;
+ let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
+ era * 146097 + doe - 719468
+}
+
+/// Julian-calendar civil date `(year, month, day)` for a day count since
1970-01-01 that labels
+/// days in the Julian calendar (astronomical year numbering: 1 BCE is year
0). Standard
+/// Julian-day-number conversion (E.G. Richards' algorithm), exact for any day.
+fn julian_day_to_civil(days: i64) -> (i64, i64, i64) {
+ // Integer (noon) Julian Day Number of this civil day: 1970-01-01 is JDN
2440588.
+ let jdn = days + 2_440_588;
+ let f = jdn + 1401;
+ let e = 4 * f + 3;
+ let g = e.rem_euclid(1461) / 4;
+ let h = 5 * g + 2;
+ let day = h.rem_euclid(153) / 5 + 1;
+ let month = (h / 153 + 2).rem_euclid(12) + 1;
+ let year = e.div_euclid(1461) - 4716 + (14 - month) / 12;
+ (year, month, day)
+}
+
+/// Exact port of Spark's `RebaseDateTime.rebaseJulianToGregorianDays`:
reinterprets a day count
+/// written in the hybrid Julian + Gregorian calendar as the proleptic
Gregorian day count of the
+/// same nominal civil date. Identity for days from 1582-10-15 onward. Days
before the tables'
+/// range (before Julian `0001-01-01`) take the calendar-arithmetic path,
mirroring Spark's
+/// `localRebaseJulianToGregorianDays` fallback.
+pub(crate) fn rebase_julian_to_gregorian_days(days: i32) -> i32 {
+ if days < JULIAN_GREG_DIFF_SWITCH_DAY[0] {
+ let (y, m, d) = julian_day_to_civil(days as i64);
+ (days_from_civil(y, m, 1) + (d - 1)) as i32
+ } else {
+ // Spark's rebaseDays: linear search from the most recent switch day.
+ let mut i = JULIAN_GREG_DIFF_SWITCH_DAY.len();
+ loop {
+ i -= 1;
+ if i == 0 || days >= JULIAN_GREG_DIFF_SWITCH_DAY[i] {
+ break;
+ }
+ }
+ days + JULIAN_GREG_DIFFS[i]
+ }
+}
+
+/// Timezone strings from `org.apache.spark.timeZone` that denote a fixed
zero-offset zone in
+/// both `java.util.TimeZone` and `java.time`. Only for these is timestamp
rebasing the pure
+/// nominal-date shift [`SparkDatetimeRebaseExpr::rebase_timestamp_utc`]
computes; any other (or
+/// absent) zone needs the JVM's historical timezone tables and stays on the
+/// refuse-ancient-values path.
+const UTC_EQUIVALENT_TIMEZONES: [&str; 6] = ["UTC", "Etc/UTC", "GMT",
"Etc/GMT", "Z", "+00:00"];
+
+/// How the writer's session time zone (if recorded) affects timestamp
rebasing.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub(crate) enum WriterTimeZone {
+ /// A fixed zero-offset zone: rebasing reduces to the exact nominal-date
shift.
+ Utc,
+ /// Any other zone, or none recorded (pre-3.0 files): ancient values
cannot be rebased
+ /// without the JVM's historical timezone data.
+ OtherOrUnknown,
+}
+
+/// One session-level datetime rebase read mode (a `LegacyBehaviorPolicy`
value of
+/// `spark.sql.parquet.datetimeRebaseModeInRead` / `int96RebaseModeInRead`),
consulted by
+/// [`resolve_file_rebase_policies`] ONLY for files whose footer metadata does
not decide the
+/// policy on its own -- exactly the `getOrElse` fallback in Spark's
+/// `DataSourceUtils.getRebaseSpec`. Files that carry
`org.apache.spark.version` ignore these
+/// modes entirely, on every Spark version.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
+pub(crate) enum RebaseReadMode {
+ /// Refuse ancient values (Spark raises `SparkUpgradeException`); maps to
+ /// [`RebasePolicy::CheckAncient`]. The default mirrors the conservative
posture used
+ /// before the conf was plumbed through (and Spark 3.x's own conf default).
+ #[default]
+ Exception,
+ /// Read values as proleptic Gregorian without rebasing.
+ Corrected,
+ /// Rebase from the hybrid Julian + Gregorian calendar.
+ Legacy,
+}
+
+impl RebaseReadMode {
+ /// Parses a `LegacyBehaviorPolicy` conf value. `SQLConf` validates and
upper-cases the
+ /// session conf, but a per-relation `datetimeRebaseMode` option arrives
verbatim, so the
+ /// match is case-insensitive. Anything unrecognized -- including the
empty string a proto
+ /// producer that predates the field sends -- falls back to
[`RebaseReadMode::Exception`],
+ /// which refuses ancient values rather than silently corrupting them.
+ pub(crate) fn from_conf_value(value: &str) -> Self {
+ match value.to_ascii_uppercase().as_str() {
+ "CORRECTED" => RebaseReadMode::Corrected,
+ "LEGACY" => RebaseReadMode::Legacy,
+ _ => RebaseReadMode::Exception,
+ }
+ }
+}
+
+/// The session's effective datetime rebase read modes, one per spec class
(INT64
+/// dates/timestamps vs INT96 timestamps), forwarded from the JVM at planning
time.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
+pub(crate) struct SessionRebaseModes {
+ /// `spark.sql.parquet.datetimeRebaseModeInRead` (or the relation's
`datetimeRebaseMode`).
+ pub datetime: RebaseReadMode,
+ /// `spark.sql.parquet.int96RebaseModeInRead` (or the relation's
`int96RebaseMode`).
+ pub int96: RebaseReadMode,
+}
+
+/// Calendar policy of one file's date or timestamp columns, resolved from
writer metadata the
+/// same way Spark's `DataSourceUtils.getRebaseSpec` resolves it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub(crate) enum RebasePolicy {
+ /// Written in the proleptic Gregorian calendar; values pass through
untouched.
+ Corrected,
+ /// Written in the hybrid Julian + Gregorian calendar; values must be
rebased.
+ Legacy(WriterTimeZone),
+ /// Policy could not be pinned down (contradictory flags, or a non-Spark
writer under the
+ /// `EXCEPTION` read mode): modern values -- identical under either
calendar -- pass,
+ /// ancient values raise. Mirrors Spark's `EXCEPTION` behavior
(`SparkUpgradeException`).
+ CheckAncient,
+}
+
+/// Which of a file's leaf columns are physically INT96, from the stamp the
parquet reader
+/// factory adds under [`INT96_LEAVES_METADATA_KEY`].
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) enum Int96Attribution {
+ /// No stamp, or a stamp whose leaf count does not match the schema it
arrived with: the
+ /// INT64 and INT96 timestamp specs cannot be told apart per column and
are merged.
+ Unknown,
+ /// Sorted leaf ordinals (depth-first over the file schema's primitive
columns) that are
+ /// INT96; every other timestamp leaf is INT64.
+ Known(Vec<usize>),
+}
+
+impl Int96Attribution {
+ /// Parses the stamp out of `schema`'s metadata and validates its leaf
count against the
+ /// schema's own depth-first leaf count, so a stamp that does not describe
this schema (a
+ /// crafted footer key, or a cached-metadata mismatch) degrades to
[`Self::Unknown`].
+ fn from_schema(schema: &Schema) -> Self {
+ let Some(stamp) = schema.metadata().get(INT96_LEAVES_METADATA_KEY)
else {
+ return Int96Attribution::Unknown;
+ };
+ let Some((count, ordinals)) = stamp.split_once(':') else {
+ return Int96Attribution::Unknown;
+ };
+ let schema_leaves: usize = schema
+ .fields()
+ .iter()
+ .map(|f| leaf_count(f.data_type()))
+ .sum();
+ if count.parse::<usize>().ok() != Some(schema_leaves) {
+ return Int96Attribution::Unknown;
+ }
+ let parsed: Option<Vec<usize>> = if ordinals.is_empty() {
+ Some(Vec::new())
+ } else {
+ ordinals
+ .split(',')
+ .map(|o| o.parse::<usize>().ok().filter(|o| *o <
schema_leaves))
+ .collect()
+ };
+ match parsed {
+ Some(mut leaves) => {
+ leaves.sort_unstable();
+ Int96Attribution::Known(leaves)
+ }
+ None => Int96Attribution::Unknown,
+ }
+ }
+
+ /// `Some(true)` / `Some(false)` when the leaf is known to be INT96 /
INT64, `None` when
+ /// the attribution is unknown.
+ fn is_int96(&self, leaf: usize) -> Option<bool> {
+ match self {
+ Int96Attribution::Unknown => None,
+ Int96Attribution::Known(leaves) =>
Some(leaves.binary_search(&leaf).is_ok()),
+ }
+ }
+}
+
+/// The [`INT96_LEAVES_METADATA_KEY`] value describing `schema`: its leaf
count and the
+/// ordinals of its INT96 primitive columns.
+pub(crate) fn int96_leaf_stamp(schema: &SchemaDescriptor) -> String {
+ let ordinals: Vec<String> = schema
+ .columns()
+ .iter()
+ .enumerate()
+ .filter(|(_, column)| column.physical_type() ==
ParquetPhysicalType::INT96)
+ .map(|(ordinal, _)| ordinal.to_string())
+ .collect();
+ format!("{}:{}", schema.num_columns(), ordinals.join(","))
+}
+
+/// Returns a copy of `metadata` whose key-value metadata carries the
[`int96_leaf_stamp`] of
+/// its own schema, or `None` when it already does (the common case after the
first open of a
+/// file, since the caller caches the stamped copy). Any pre-existing entry
under the key --
+/// a file cannot legitimately carry one -- is replaced, never trusted. Only
the file-level
+/// key-value list changes; row groups and page indexes are carried over
as-is. The parquet
+/// API cannot carry a file decryptor, nor `FileMetaData`'s crate-private
encryption fields
+/// (encryption algorithm, footer signing key metadata), across this rebuild,
so callers must
+/// not stamp opens that supply decryption properties -- and the only
consumer, the Delta
+/// scan, declines every encrypted-parquet configuration before planning, so a
parquet
+/// modular encryption file never reaches this path with or without those
properties.
+pub(crate) fn stamp_int96_leaves(metadata: &ParquetMetaData) ->
Option<ParquetMetaData> {
+ let file_metadata = metadata.file_metadata();
+ let stamp = int96_leaf_stamp(file_metadata.schema_descr());
+ let existing = file_metadata
+ .key_value_metadata()
+ .and_then(|kvs| kvs.iter().find(|kv| kv.key ==
INT96_LEAVES_METADATA_KEY))
+ .and_then(|kv| kv.value.as_deref());
+ if existing == Some(stamp.as_str()) {
+ return None;
+ }
+ let mut key_values: Vec<KeyValue> = file_metadata
+ .key_value_metadata()
+ .map(|kvs| {
+ kvs.iter()
+ .filter(|kv| kv.key != INT96_LEAVES_METADATA_KEY)
+ .cloned()
+ .collect()
+ })
+ .unwrap_or_default();
+ key_values.push(KeyValue::new(INT96_LEAVES_METADATA_KEY.to_string(),
stamp));
+ let stamped_file_metadata = FileMetaData::new(
+ file_metadata.version(),
+ file_metadata.num_rows(),
+ file_metadata.created_by().map(str::to_string),
+ Some(key_values),
+ file_metadata.schema_descr_ptr(),
+ file_metadata.column_orders().cloned(),
+ );
+ Some(
+ ParquetMetaData::new(stamped_file_metadata,
metadata.row_groups().to_vec())
+ .into_builder()
+ .set_column_index(metadata.column_index().cloned())
+ .set_offset_index(metadata.offset_index().cloned())
+ .build(),
+ )
+}
+
+/// Per-file rebase policies for the three affected column classes, plus the
INT96
+/// attribution that selects between the two timestamp specs per leaf.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub(crate) struct FileRebasePolicies {
+ /// `DATE` columns, governed by `org.apache.spark.legacyDateTime` alone.
+ pub date: RebasePolicy,
+ /// INT64 `TIMESTAMP_MICROS` / `TIMESTAMP_MILLIS` columns: the datetime
spec (same
+ /// resolution as `date`), as Spark's `ParquetVectorUpdaterFactory`
selects for INT64.
+ pub int64_timestamp: RebasePolicy,
+ /// INT96 columns: the INT96 spec (`org.apache.spark.legacyINT96`, min
version 3.1.0).
+ pub int96_timestamp: RebasePolicy,
+ /// Which timestamp leaves are INT96. See [`Int96Attribution`].
+ pub int96_leaves: Int96Attribution,
+ /// Sorted depth-first leaf ordinals -- over the physical file schema, the
same ordinals
+ /// `int96_leaves` uses -- that the query does not read: nested children
the schema
+ /// adapter's struct narrowing drops before any value leaves the scan.
Spark never decodes
+ /// them either, so their policy is the identity whatever the file's
calendar. Empty until
+ /// [`Self::restrict_to_requested`] runs (every leaf requested).
+ pub unrequested_leaves: Vec<usize>,
+}
+
+impl FileRebasePolicies {
+ /// True when some policy is not the plain proleptic-Gregorian
pass-through, i.e. when the
+ /// per-column wrap in [`wrap_datetime_rebase`] can install anything at
all.
+ pub(crate) fn any_rebase_needed(&self) -> bool {
+ self.date != RebasePolicy::Corrected
+ || self.int64_timestamp != RebasePolicy::Corrected
+ || self.int96_timestamp != RebasePolicy::Corrected
+ }
+
+ fn is_requested(&self, leaf: usize) -> bool {
+ self.unrequested_leaves.binary_search(&leaf).is_err()
+ }
+
+ /// The policy of the `Date32` leaf at depth-first ordinal `leaf`: the
file's date policy,
+ /// or the identity when the query does not read that leaf.
+ fn date_policy(&self, leaf: usize) -> RebasePolicy {
+ if self.is_requested(leaf) {
+ self.date
+ } else {
+ RebasePolicy::Corrected
+ }
+ }
+
+ /// The policy of the timezone-carrying timestamp leaf at depth-first
ordinal `leaf`: the
+ /// identity when the query does not read it; otherwise its physical
type's spec when the
+ /// attribution is known, or else the two specs merged -- agreement
decides, disagreement
+ /// degrades to [`RebasePolicy::CheckAncient`], which still passes every
modern value and
+ /// refuses only ancient ones.
+ fn timestamp_policy(&self, leaf: usize) -> RebasePolicy {
+ if !self.is_requested(leaf) {
+ return RebasePolicy::Corrected;
+ }
+ match self.int96_leaves.is_int96(leaf) {
+ Some(true) => self.int96_timestamp,
+ Some(false) => self.int64_timestamp,
+ None if self.int64_timestamp == self.int96_timestamp =>
self.int64_timestamp,
+ None => RebasePolicy::CheckAncient,
+ }
+ }
+
+ /// These policies with every physical leaf the query does not read marked
the identity.
+ /// `requested` pairs each top-level field of `physical_schema` (by
position) with the type
+ /// of the logical field the schema adapter narrows it to -- `None` for a
column without a
+ /// logical counterpart, whose leaves are left as they are (no expression
reads it anyway).
+ /// Nested children pair the way the adapter's struct convert selects them
(see
+ /// [`push_unrequested_leaves`]); the INT96 attribution is untouched,
since the ordinals
+ /// stay physical. `requested` is parallel to the schema's fields; should
a caller pass a
+ /// shorter slice, the trailing columns simply keep every leaf (the safe
direction).
+ pub(crate) fn restrict_to_requested(
+ mut self,
+ physical_schema: &Schema,
+ requested: &[Option<&DataType>],
+ case_sensitive: bool,
+ use_field_id: bool,
+ ) -> Self {
+ debug_assert_eq!(requested.len(), physical_schema.fields().len());
+ let matching = FieldMatching {
+ case_sensitive,
+ use_field_id,
+ };
+ let mut next_leaf = 0;
+ let mut unrequested = Vec::new();
+ for (field, requested) in
physical_schema.fields().iter().zip(requested) {
+ match requested {
+ Some(logical) => push_unrequested_leaves(
+ field.data_type(),
+ logical,
+ &mut next_leaf,
+ matching,
+ &mut unrequested,
+ ),
+ None => next_leaf += leaf_count(field.data_type()),
+ }
+ }
+ // Emitted in depth-first order, so already sorted for
`is_requested`'s binary search.
+ self.unrequested_leaves = unrequested;
+ self
+ }
+}
+
+/// The field-matching rules of the schema adapter's nested narrowing
+/// (`parquet_convert_struct_to_struct`): names fold per `case_sensitive`, and
Parquet field ids
+/// select fields when `use_field_id` is set.
+#[derive(Debug, Clone, Copy)]
+struct FieldMatching {
+ case_sensitive: bool,
+ use_field_id: bool,
+}
+
+/// Appends to `out` the depth-first leaf ordinals of `physical` (counting
from `next_leaf`,
+/// which advances past every leaf of `physical`) that reading it as
`requested` drops.
+///
+/// Recurses through exactly the pairings `parquet_convert_array` narrows, and
no others: a
+/// struct child is dropped only when NO requested child selects it by either
rule the struct
+/// convert uses -- folded name, or Parquet field id when ids are in play --
and an ambiguous
+/// child (several requested children select it) is kept; `List` pairs with
`List` by element
+/// type, and `Map` with a `Map` of the same key ordering by its entries,
positionally. Any
+/// other pairing -- a leaf, a `LargeList` / `FixedSizeList` / dictionary, a
map whose ordering
+/// differs, or a shape mismatch -- is handed to arrow's cast or passed
through whole by the
+/// convert, so it keeps every leaf. Keeping a superset of what the narrowing
reads is always
+/// safe (a spurious check at worst); dropping a leaf the narrowing reads
would skip its
+/// rebase, so every doubt resolves to "requested".
+fn push_unrequested_leaves(
+ physical: &DataType,
+ requested: &DataType,
+ next_leaf: &mut usize,
+ matching: FieldMatching,
+ out: &mut Vec<usize>,
+) {
+ match (physical, requested) {
+ (DataType::Struct(physical_fields),
DataType::Struct(requested_fields)) => {
+ let names: Vec<&str> = physical_fields
+ .iter()
+ .chain(requested_fields.iter())
+ .map(|f| f.name().as_str())
+ .collect();
+ let folded = fold_names(&names, matching.case_sensitive);
+ let (physical_folded, requested_folded) =
folded.split_at(physical_fields.len());
+ for (i, child) in physical_fields.iter().enumerate() {
+ let child_id = if matching.use_field_id {
+ parse_field_id(child)
+ } else {
+ None
+ };
+ let mut selectors =
requested_fields.iter().enumerate().filter(|(j, r)| {
+ requested_folded[*j] == physical_folded[i]
+ || (child_id.is_some() && parse_field_id(r) ==
child_id)
+ });
+ match (selectors.next(), selectors.next()) {
+ (None, _) => {
+ let n = leaf_count(child.data_type());
+ out.extend(*next_leaf..*next_leaf + n);
+ *next_leaf += n;
+ }
+ (Some((_, requested_child)), None) =>
push_unrequested_leaves(
+ child.data_type(),
+ requested_child.data_type(),
+ next_leaf,
+ matching,
+ out,
+ ),
+ (Some(_), Some(_)) => *next_leaf +=
leaf_count(child.data_type()),
+ }
+ }
+ }
+ (DataType::List(physical_item), DataType::List(requested_item)) =>
push_unrequested_leaves(
+ physical_item.data_type(),
+ requested_item.data_type(),
+ next_leaf,
+ matching,
+ out,
+ ),
+ (
+ DataType::Map(physical_entries, physical_sorted),
+ DataType::Map(requested_entries, requested_sorted),
+ ) if physical_sorted == requested_sorted => {
+ match (physical_entries.data_type(),
requested_entries.data_type()) {
+ (DataType::Struct(physical_kv), DataType::Struct(requested_kv))
+ if physical_kv.len() == requested_kv.len() =>
+ {
+ for (p, r) in physical_kv.iter().zip(requested_kv.iter()) {
+ push_unrequested_leaves(
+ p.data_type(),
+ r.data_type(),
+ next_leaf,
+ matching,
+ out,
+ );
+ }
+ }
+ _ => *next_leaf += leaf_count(physical),
+ }
+ }
+ _ => *next_leaf += leaf_count(physical),
+ }
+}
+
+/// The writer time zone recorded in `metadata`, classified for timestamp
rebasing. Mirrors the
+/// `Option(lookupFileMeta(SPARK_TIMEZONE_METADATA_KEY))` lookup Spark's
`getRebaseSpec` performs
+/// for every LEGACY resolution, conf-fallback included; Spark substitutes the
JVM default zone
+/// when the key is absent (`RebaseSpec.timeZone`), which is unavailable
natively, so an absent or
+/// non-UTC zone classifies as [`WriterTimeZone::OtherOrUnknown`] (dates still
rebase fully --
+/// the day rebase is zone-free -- while ancient timestamps refuse rather than
guess).
+fn writer_time_zone(metadata: &HashMap<String, String>) -> WriterTimeZone {
+ match metadata.get(SPARK_TIMEZONE_KEY) {
+ Some(tz) if UTC_EQUIVALENT_TIMEZONES.contains(&tz.as_str()) =>
WriterTimeZone::Utc,
+ _ => WriterTimeZone::OtherOrUnknown,
+ }
+}
+
+/// One spec resolution, mirroring Spark's `DataSourceUtils.getRebaseSpec`
exactly: a Spark
+/// version below `min_version` (lexicographic comparison, same as the Scala
`String.<`) or a
+/// present legacy flag means LEGACY; a Spark version at/after `min_version`
without the flag
+/// means CORRECTED; no Spark version at all falls back to `conf_mode`, the
session read conf
+/// forwarded from the JVM (`getRebaseSpec`'s `modeByConfig` fallback, its
ONLY use of the
+/// conf): CORRECTED passes values through, LEGACY rebases (with the writer
zone from the
+/// file's `org.apache.spark.timeZone` key, same lookup as the metadata-driven
LEGACY path),
+/// and EXCEPTION refuses ancient values as [`RebasePolicy::CheckAncient`].
+fn resolve_spec(
+ metadata: &HashMap<String, String>,
+ min_version: &str,
+ legacy_key: &str,
+ conf_mode: RebaseReadMode,
+) -> RebasePolicy {
+ match metadata.get(SPARK_VERSION_METADATA_KEY) {
+ None => match conf_mode {
+ RebaseReadMode::Corrected => RebasePolicy::Corrected,
+ RebaseReadMode::Legacy =>
RebasePolicy::Legacy(writer_time_zone(metadata)),
+ RebaseReadMode::Exception => RebasePolicy::CheckAncient,
+ },
+ Some(version) => {
+ if version.as_str() < min_version ||
metadata.contains_key(legacy_key) {
+ RebasePolicy::Legacy(writer_time_zone(metadata))
+ } else {
+ RebasePolicy::Corrected
+ }
+ }
+ }
+}
+
+/// Resolves the per-file rebase policies from a file's arrow schema: the
parquet footer's
+/// key-value pairs in its metadata decide the specs (the datetime spec uses
min version
+/// `3.0.0` and the INT96 spec `3.1.0`, matching
`DataSourceUtils.datetimeRebaseSpec` /
+/// `int96RebaseSpec`; `session_modes` supplies the per-spec conf fallback for
files without
+/// Spark writer metadata), and the reader factory's INT96 stamp -- validated
against the
+/// schema's leaf structure -- attributes each timestamp leaf to its spec.
+pub(crate) fn resolve_file_rebase_policies(
+ physical_file_schema: &Schema,
+ session_modes: SessionRebaseModes,
+) -> FileRebasePolicies {
+ let metadata = physical_file_schema.metadata();
+ let datetime_spec = resolve_spec(
+ metadata,
+ "3.0.0",
+ SPARK_LEGACY_DATETIME_KEY,
+ session_modes.datetime,
+ );
+ let int96_spec = resolve_spec(
+ metadata,
+ "3.1.0",
+ SPARK_LEGACY_INT96_KEY,
+ session_modes.int96,
+ );
+ FileRebasePolicies {
+ date: datetime_spec,
+ int64_timestamp: datetime_spec,
+ int96_timestamp: int96_spec,
+ int96_leaves: Int96Attribution::from_schema(physical_file_schema),
+ unrequested_leaves: Vec::new(),
+ }
+}
+
+/// Number of primitive leaves `dt` contains in a depth-first walk -- the same
count and order
+/// parquet-rs uses when it maps the file's `SchemaDescriptor` columns onto
the arrow schema, so
+/// arrow-side leaf ordinals line up with [`int96_leaf_stamp`]'s.
+fn leaf_count(dt: &DataType) -> usize {
+ match dt {
+ DataType::Struct(fields) => fields.iter().map(|f|
leaf_count(f.data_type())).sum(),
+ DataType::List(f)
+ | DataType::LargeList(f)
+ | DataType::FixedSizeList(f, _)
+ | DataType::ListView(f)
+ | DataType::LargeListView(f)
+ | DataType::Map(f, _) => leaf_count(f.data_type()),
+ DataType::Dictionary(_, value) => leaf_count(value),
+ DataType::RunEndEncoded(_, value) => leaf_count(value.data_type()),
+ DataType::Union(fields, _) => fields.iter().map(|(_, f)|
leaf_count(f.data_type())).sum(),
+ _ => 1,
+ }
+}
+
+/// Appends the policy of every leaf of `dt`, in depth-first order, to `out`,
consuming leaf
+/// ordinals from `next_leaf` (exactly [`leaf_count`] of them). Only `Date32`
and
+/// timezone-carrying timestamps have a policy to apply, and only when the
query reads the
+/// leaf; timezone-free timestamps are `TIMESTAMP_NTZ`, which Spark never
rebases, and every
+/// other leaf is the identity ([`RebasePolicy::Corrected`]).
+fn leaf_policies(
+ dt: &DataType,
+ next_leaf: &mut usize,
+ policies: &FileRebasePolicies,
+ out: &mut Vec<RebasePolicy>,
+) {
+ match dt {
+ DataType::Date32 => {
+ out.push(policies.date_policy(*next_leaf));
+ *next_leaf += 1;
+ }
+ DataType::Timestamp(_, Some(_)) => {
+ out.push(policies.timestamp_policy(*next_leaf));
+ *next_leaf += 1;
+ }
+ DataType::Struct(fields) => {
+ for f in fields {
+ leaf_policies(f.data_type(), next_leaf, policies, out);
+ }
+ }
+ // Mirrors `leaf_count` variant for variant, so a rebase-affected leaf
inside a nested
+ // type `rebase_array` cannot rebuild (views, run-end, union -- never
produced from a
+ // parquet schema) still gets its real policy and makes `rebase_array`
refuse loudly
+ // instead of being stamped the identity.
+ DataType::List(f)
+ | DataType::LargeList(f)
+ | DataType::FixedSizeList(f, _)
+ | DataType::ListView(f)
+ | DataType::LargeListView(f)
+ | DataType::Map(f, _) => leaf_policies(f.data_type(), next_leaf,
policies, out),
+ DataType::Dictionary(_, value) => leaf_policies(value, next_leaf,
policies, out),
+ DataType::RunEndEncoded(_, value) => {
+ leaf_policies(value.data_type(), next_leaf, policies, out)
+ }
+ DataType::Union(fields, _) => {
+ for (_, f) in fields.iter() {
+ leaf_policies(f.data_type(), next_leaf, policies, out);
+ }
+ }
+ _ => {
+ *next_leaf += 1;
+ out.push(RebasePolicy::Corrected);
+ }
+ }
+}
+
+/// Wraps every column reference in `expr` whose physical file type contains a
rebase-affected
+/// leaf under a policy that needs handling with a [`SparkDatetimeRebaseExpr`]
carrying that
+/// column's per-leaf policies, so both the per-file projection and the
pushed-down predicate
+/// evaluate rebased values. Columns whose leaves are all the identity --
unaffected types,
+/// affected types under [`RebasePolicy::Corrected`], or leaves the query does
not read (see
+/// [`FileRebasePolicies::restrict_to_requested`]) -- pass through unwrapped.
(The pruning
+/// predicates derived from the wrapped predicate treat the wrapper as an
opaque expression and
+/// skip pruning on those columns -- conservative, since file-level statistics
are in the
+/// file's own calendar.)
+pub(crate) fn wrap_datetime_rebase(
+ expr: Arc<dyn PhysicalExpr>,
+ physical_schema: &SchemaRef,
+ policies: &FileRebasePolicies,
+) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
+ expr.transform(|e| {
+ let Some(col) = e.downcast_ref::<Column>() else {
+ return Ok(Transformed::no(e));
+ };
+ // Missing columns were already replaced with literals; any surviving
reference is
+ // physical-schema-indexed. Out-of-range means a non-file column
(defensive): skip.
+ let Some(field) = physical_schema.fields().get(col.index()) else {
+ return Ok(Transformed::no(e));
+ };
+ // This column's first leaf ordinal: the leaves of every preceding
top-level field.
+ let mut next_leaf: usize = physical_schema.fields()[..col.index()]
+ .iter()
+ .map(|f| leaf_count(f.data_type()))
+ .sum();
+ let mut column_leaf_policies =
Vec::with_capacity(leaf_count(field.data_type()));
+ leaf_policies(
+ field.data_type(),
+ &mut next_leaf,
+ policies,
+ &mut column_leaf_policies,
+ );
+ if column_leaf_policies
+ .iter()
+ .all(|p| *p == RebasePolicy::Corrected)
+ {
+ return Ok(Transformed::no(e));
+ }
+ Ok(Transformed::yes(Arc::new(SparkDatetimeRebaseExpr {
+ child: e,
+ field: Arc::clone(field),
+ leaf_policies: column_leaf_policies,
+ }) as Arc<dyn PhysicalExpr>))
+ })
+ .map(|t| t.data)
+}
+
+/// Applies a file's calendar-rebase policies to one column: rebases exactly
where possible,
+/// raises on ancient values it cannot rebase, and passes modern values (the
identity under
+/// every policy) through untouched. Nested columns are rebuilt leaf by leaf
with nulls and
+/// offsets preserved. See the module doc for the policy table.
+#[derive(Debug, Eq)]
+struct SparkDatetimeRebaseExpr {
+ child: Arc<dyn PhysicalExpr>,
+ /// The physical file field this expression reads (type preserved by the
rebase).
+ field: FieldRef,
+ /// One policy per primitive leaf of `field`'s type, in depth-first order
(a single entry
+ /// for a flat column). At least one is not [`RebasePolicy::Corrected`].
+ leaf_policies: Vec<RebasePolicy>,
+}
+
+impl SparkDatetimeRebaseExpr {
+ /// The refusal error, as an [`ArrowError`] so `try_unary` closures can
raise it directly;
+ /// it converts into a `DataFusionError` at the `?` in `evaluate`.
+ fn rebase_error(&self, detail: &str) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "Native scan cannot rebase ancient values in column '{}': the file
was written \
+ with the legacy (hybrid Julian/Gregorian) calendar, or does not
declare which \
+ calendar it used, and {detail}. Reading it natively would return
silently \
+ shifted values; disable the native Delta scan \
+ (spark.comet.scan.delta.enabled=false) to let Spark read this
table",
+ self.field.name(),
+ ))
+ }
+
+ fn internal_error(&self, detail: impl Display) -> DataFusionError {
+ DataFusionError::Internal(format!(
+ "SparkDatetimeRebaseExpr on column '{}': {detail}",
+ self.field.name()
+ ))
+ }
+
+ /// Rebases a timestamp column written at a fixed zero-offset zone: shift
the nominal day
+ /// with the exact date table, keep the time of day. Matches Spark's
+ /// `rebaseJulianToGregorianMicros` for UTC, where the hybrid calendar's
day boundaries sit
+ /// exactly on multiples of a day and no timezone transition can apply
(UTC's last switch
+ /// instant in Spark's rebase table is the 1582-10-15 cutover itself).
+ fn rebase_timestamp_utc(&self, v: i64, units_per_day: i64) -> Result<i64,
ArrowError> {
+ // Compare in days, not units: the cutover day times a nanosecond day
does not fit i64.
+ let day = v.div_euclid(units_per_day);
+ if day >= LAST_SWITCH_JULIAN_DAY as i64 {
+ return Ok(v);
+ }
+ let time_of_day = v - day * units_per_day;
+ let day = i32::try_from(day).map_err(|_| {
+ self.rebase_error("the value is outside the rebaseable timestamp
range")
+ })?;
+ let rebased = rebase_julian_to_gregorian_days(day) as i64;
+ rebased
+ .checked_mul(units_per_day)
+ .and_then(|d| d.checked_add(time_of_day))
+ .ok_or_else(|| self.rebase_error("the rebased value overflows the
timestamp range"))
+ }
+
+ /// The refuse-ancient-values policy for timestamps: values from
+ /// [`LAST_SWITCH_JULIAN_TS_SECONDS`] onward are identical under both
calendars in every
+ /// zone (Spark's `createTimestampRebaseFuncInRead` under `EXCEPTION`
accepts exactly
+ /// these, and `rebaseJulianToGregorianMicros` is the identity on them for
any zone);
+ /// older values raise.
+ fn check_ancient_timestamp(
+ &self,
+ v: i64,
+ units_per_second: i64,
+ detail: &str,
+ ) -> Result<i64, ArrowError> {
+ if v >= LAST_SWITCH_JULIAN_TS_SECONDS * units_per_second {
+ Ok(v)
+ } else {
+ Err(self.rebase_error(detail))
+ }
+ }
+
+ fn rebase_timestamp_array<T: ArrowTimestampType>(
+ &self,
+ array: &PrimitiveArray<T>,
+ policy: RebasePolicy,
+ units_per_second: i64,
+ ) -> DataFusionResult<ArrayRef> {
+ let tz = array.timezone().map(Arc::<str>::from);
+ let rebased: PrimitiveArray<T> = match policy {
+ RebasePolicy::Corrected => return Ok(Arc::new(array.clone())),
+ RebasePolicy::Legacy(WriterTimeZone::Utc) =>
arrow::compute::try_unary(array, |v| {
+ self.rebase_timestamp_utc(v, units_per_second * 86_400)
+ })?,
+ RebasePolicy::Legacy(WriterTimeZone::OtherOrUnknown) => {
Review Comment:
This arm and the `CheckAncient` arms below (timestamps here, dates at ~901)
go through `try_unary`, which allocates a fresh values buffer and writes every
value back unchanged. These are the policies a metadata-free file under
EXCEPTION mode hits on every batch. A validity-aware `all(v >= cutoff)` over
`values()` followed by `Ok(Arc::clone(array))` would make them allocation-free,
and `Legacy(Utc)` could short-circuit the same way when the batch minimum is at
or after the cutover.
--
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]
