fix: fall back for Parquet datetime rebasing

[peterxcli]

Which issue does this PR close?

Closes #5010.

Rationale for this change

Native Parquet scans cannot currently apply Spark's legacy Julian/Gregorian rebasing safely before Parquet pruning, filters, and aggregates consume datetime values. This can produce incorrect results for files written by Spark 2.x or with legacy datetime or INT96 rebasing.

I want to merge this conservative fallback first to restore correctness with a small, low-risk change. The native implementation in #5047 should remain a follow-up and can restore native acceleration after its broader per-file rebasing behavior is reviewed.

What changes are included in this PR?

• Inspect Parquet footer metadata and configured read modes before converting a scan to Comet.
• Fall back to Spark when any input file may require legacy datetime or INT96 rebasing, or when metadata inspection fails.
• Remove the dead datetime-rebase configuration and the unused SparkParquetOptions.use_legacy_date_timestamp_or_ntz field.
• Update scan compatibility documentation and add regression coverage for the fallback.
• Cache per-file footer facts (keyed by path, length, and modification time) so AQE re-planning and repeated queries over the same files pay no additional I/O, and add spark.comet.scan.parquet.checkDatetimeRebase (default true) to skip the check for data known to be corrected.
• Fall back to Spark for native Parquet writes when the write rebase mode is LEGACY and the schema contains dates or timestamps, since the native writer always writes corrected values and stamps corrected metadata.

Planning-time cost

The footer check reads each input file's footer (with SKIP_ROW_GROUPS) on the driver the first time the file is planned, 8 files at a time, stopping at the first legacy footer. Measured with 2,000 single-row files on a local SSD: ~10.9s cold (~5.4ms/file), 16-57ms once cached. Object storage is latency-bound (roughly 1-2 footer GETs per file across 8 threads); the per-file cache and the opt-out config exist for that case.

How are these changes tested?

• make core
• git diff --check
• ./mvnw test -Dtest=none -Dsuites="org.apache.comet.parquet.ParquetReadV1Suite fallback for Parquet datetime rebasing" -Dscalastyle.skip=true
• ./mvnw test -Dtest=none -Dsuites="org.apache.comet.parquet.CometParquetWriterSuite" -Dscalastyle.skip=true (34/34, including the new LEGACY write-mode fallback test)
• cargo test -p datafusion-comet parquet_writer

[peterxcli]

copied spec from https://github.com/apache/spark/blob/v4.1.2/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala#L130-L177

[andygrove]

Thanks for this @peterxcli — the diagnosis is right, and the Spark 2.4.5 "no org.apache.spark.version key at all" case is a good catch.

My concern is where the check runs. requiresDatetimeRebase opens every selected file's footer on the driver at plan time, for every query with a date or timestamp column. Three things compound: selectedPartitions lists with partitionFilters.filterNot(isDynamicPruningFilter), so it's the pre-DPP file set; ParquetFileReader.open reads the full footer including all row-group metadata rather than ParquetFooterReader.readFooter(..., SKIP_ROW_GROUPS); and it's serial. On a table with thousands of files that is significant added planning latency and driver heap, paid per query, even when every file turns out to be fine.

I opened #5202 as an alternative that does the same detection in the native Parquet reader factory, where the footer has already been fetched and cached for the read itself — so it costs no additional I/O. Being at that point also lets it consult row-group statistics, which matters because Spark stamps legacyDateTime on a whole file whenever the write mode was LEGACY, whether or not any value is actually old enough to rebase.

The tradeoff is the outcome: yours falls back to Spark and always returns correct results, mine raises. I don't think fallback is reachable without paying the plan-time cost, but that's worth other opinions.

One thing #5202 should take from here regardless: you read the rebase modes from new ParquetOptions(r.options, conf), which honors .option("datetimeRebaseMode", ...). #5202 reads the session conf only and misses it. I'll fix that and credit you.

[andygrove]

Note on this review: this was generated by an LLM (Claude Code) at my request while I worked through a review backlog. I have not verified the individual findings myself. Please treat everything below as suggestions to evaluate rather than as authoritative review feedback, and push back on anything that is wrong or already handled.

Trading a silent wrong answer for a fallback is the right call, and I like that the fallback decision mirrors Spark's own DataSourceUtils.datetimeRebaseSpec closely, including the string comparison against "3.0.0" and the treatment of a missing version key. Stamping org.apache.spark.version from the native writer is a necessary companion change and it is good that it is here rather than a follow-up.

Three concerns.

Every Parquet footer is now read on the driver at planning time

requiresDatetimeRebase opens every selected file's footer before the scan is converted, on any query whose required schema contains a DATE or TIMESTAMP. The short circuit helps when a legacy file is found early, but the common case is that no file needs rebasing, and then all of them are read. On a table with tens of thousands of files in object storage, eight threads doing a footer read each is going to add a lot of planning latency to queries that previously did no driver-side I/O at all.

Some things that would help:

• How long does this add for a realistically large table? A number in the description would let us judge whether this is fine or not.
• Is the result cached anywhere? CometScanRule runs per query execution, and AQE re-planning can invoke it again. Paying this repeatedly for the same file set would be worse than paying it once.
• Should there be a config to skip the check for users who know their data is corrected, accepting the previous behavior? Given the fallback is conservative, an opt-out seems reasonable.

What happens when the write rebase mode is LEGACY?

The native writer now unconditionally stamps org.apache.spark.version and writes corrected values. If a user has spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY, Spark would write rebased values and mark the file with org.apache.spark.legacyDateTime. As far as I can see from this diff, Comet's native write path ignores that config, writes corrected values, and stamps a version that says "corrected". A later reader would then trust the marker and read the wrong values with no error.

Does CometDataWritingCommand already fall back when the write rebase mode is not CORRECTED? If it does, pointing at that check would settle it. If it does not, that is a data-corruption path and it should either be fixed here or filed with a clear issue, because this PR is what makes the version marker load-bearing.

The TimestampNTZType condition

val hasTimestamp =
  SupportLevel.containsType(scanExec.requiredSchema, classOf[TimestampType]) ||
    (COMET_SCHEMA_EVOLUTION_ENABLED &&
      SupportLevel.containsType(scanExec.requiredSchema, classOf[TimestampNTZType]))

Tying the NTZ case to schema evolution is not obvious. Is the reasoning that NTZ can only come from an INT96 physical column when schema evolution lets a TIMESTAMP be read as TIMESTAMP_NTZ? If so, a comment saying that would help a lot, because on its face the rebase question and the schema-evolution config are unrelated and a future reader is likely to "simplify" this.

One smaller thing

The version comparison v < minVersion is a lexicographic string compare, which would misclassify a hypothetical Spark 10.x. That is exactly what Spark does, so matching it is correct. Worth a one-line comment saying it deliberately mirrors DataSourceUtils, otherwise it looks like a bug someone will "fix".

[peterxcli]

Thanks @andygrove — all four points were worth acting on. Addressed in 2323bae (and the branch is now merged up to main, which also resolves the conflict with #5490's output_schema proto field).

1. Plan-time footer reads

The earlier round had already moved the check to SKIP_ROW_GROUPS footer reads with a bounded 8-way pool and early exit on the first legacy footer. This round adds the three things you asked about:

• Caching: footer facts (the Spark version key and the two legacy markers) are now cached per (path, length, modificationTime) in a bounded LRU (32K entries). The facts are independent of read modes and requested types, so one entry answers the rebase question for any query — AQE re-planning and repeated queries over the same file set no longer pay any I/O. Only never-before-seen files read a footer.
• Opt-out: new config spark.comet.scan.parquet.checkDatetimeRebase (default true). Users who know their data is corrected can disable the check and get the previous behavior.
• Numbers: 2,000 single-row files on a local SSD: cold check ~10.9s (~5.4ms/file at 8-way parallelism), fully-cached re-check 16–57ms. Object storage will be latency-bound (roughly one to two footer GETs per file across 8 threads), which is exactly why the cache and the opt-out exist. I've added the numbers to the PR description.

On the pre-DPP file set (from your earlier comment): selectedPartitions is indeed the superset before dynamic pruning, but plan time is the only place a fallback decision can be made, and checking a superset errs in the conservative direction; with the cache the extra files are read once, ever.

2. LEGACY write rebase mode

Correct — there was no check, and the write path silently ignored spark.sql.parquet.datetimeRebaseModeInWrite. One correction to the failure scenario as stated: since the native writer writes corrected values and stamps a corrected version with no legacy markers, the file it produces is self-consistent — a Spark 3.x+ reader gets correct values back. The real problems are silently ignoring an explicit user config, and downstream legacy readers (Spark 2.x, Hive) that LEGACY exists to serve reading shifted values. Either way the fix is the same: CometDataWritingCommand.getSupportLevel now returns Unsupported when the effective write rebase mode is LEGACY (datetimeRebaseModeInWrite, plus int96RebaseModeInWrite when the schema has timestamps) and the output schema contains dates or timestamps. TIMESTAMP_NTZ is exempt because Spark never rebases NTZ on write.

Deliberately not gated: EXCEPTION, the default. Spark under EXCEPTION writes the same corrected bytes as CORRECTED and only raises when it encounters a pre-Gregorian value; falling back on EXCEPTION would disable native datetime writes under default configs entirely, and the raise is data-dependent so it can't be detected at plan time. The remaining divergence is narrow — Comet writes a correct, correctly-stamped value where Spark would raise on an ancient one. I can file a follow-up for a native-side ancient-value check if you want EXCEPTION parity.

3. The TimestampNTZ condition

Your guess is exactly right, and the constant I used made it look worse than it is. TIMESTAMP_NTZ never needs rebasing by itself: Spark neither rebases NTZ on write (ParquetWriteSupport emits the raw micros) nor on read (ParquetVectorUpdaterFactory: "TIMESTAMP_NTZ is a new data type and has no legacy files that need to do rebase"). The rebase question only arises when a Parquet TIMESTAMP (LTZ or INT96) column is read as NTZ, which Comet permits only on Spark 4.x. The precise gate for that permission is COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ (SPARK-47447), not COMET_SCHEMA_EVOLUTION_ENABLED — they happen to carry the same per-version values, which is why the tests passed, but the schema-evolution constant is about widening primitive promotions and was the wrong name for this. Switched to COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ with a comment explaining the above.

4. The version comparison

Added a comment noting that v < minVersion deliberately mirrors the lexicographic string comparison in Spark's DataSourceUtils.datetimeRebaseSpec.

Tests: the three ParquetReadV1Suite rebase tests (now also covering the cache, the early exit, and the opt-out config), a new CometParquetWriterSuite test asserting LEGACY write mode falls back while CORRECTED stays native, the full CometParquetWriterSuite (34/34) as a regression check on the write gate, and the native parquet_writer cargo tests all pass locally.

[andygrove]

Thanks for working through the last round. SKIP_ROW_GROUPS, the bounded pool with the early exit, the per-file cache and the opt-out cover what I asked about the planning cost, and gating LEGACY writes plus stamping org.apache.spark.version closes the write-side hole.

The branch conflicts with main now, and one of the changes it needs to absorb won't show up as a conflict. #5763 moved native writes on Spark 4.0+ to CometWriteFiles and CometWriteFilesExec, so after the rebase the LEGACY rebase-mode gate and setSparkVersion in CometDataWritingCommand only cover 3.x. Without setSparkVersion, the 4.x path stamps org.apache.spark.version with an empty string. Spark compares that against "3.0.0" as a string and treats the file as legacy, so it rebases Comet-written ancient dates and timestamps on read, and this PR's scan check falls back on every Comet-written file. Could the gate and the version go into a helper both serdes use, and could ParquetWriterExec leave the key out when the version is empty?

The other thing is the read check on Spark 3.4 and 3.5, where datetimeRebaseModeInRead and int96RebaseModeInRead default to EXCEPTION. A file without org.apache.spark.version falls through to the mode, so every Parquet file written by something other than Spark, such as Trino, Hive, Flink, pyarrow or DuckDB, now falls back whenever the query reads a date or timestamp. That's the right call for matching Spark, since under EXCEPTION Spark raises on an ancient value we would return. But the version stamp this PR adds to createParquetWriter in CometTestBase makes every helper-written fixture look Spark-written, so no test exercises it, and scans.md doesn't mention it. Could scans.md spell out the 3.x case and point those users at setting the two Spark read modes to CORRECTED, the 4.0 default, rather than at checkDatetimeRebase=false? And could a test write a file without the version key and pin both outcomes, falling back under EXCEPTION and staying native under CORRECTED?