andygrove commented on issue #5010:
URL:
https://github.com/apache/datafusion-comet/issues/5010#issuecomment-5159469674
## Findings from attempting a fix (#5202, closed)
I closed #5202 ("fail rather than silently return unrebased datetimes from
Parquet") because the
problem is substantially more complex than the issue text implies. #5221 now
only removes the dead
`spark.comet.exceptionOnDatetimeRebase` config for 1.0.0, so the correctness
gap tracked here is
untouched and still open.
Recording what the attempt turned up, so the next person picking this up (or
reviewing #5048)
starts from it rather than rediscovering it. All of the below is verified
against Spark's own
sources, not inferred.
---
### A. Detection is much narrower than "does the file carry a legacy marker"
**A1. The footer rule is "provably Proleptic Gregorian", not "provably
legacy".** Spark's
[`DataSourceUtils.getRebaseSpec`](https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceUtils.scala)
resolves the policy as: if `org.apache.spark.version` is present, LEGACY when
`version < minVersion || <marker key> != null`, else CORRECTED; **if the
version key is absent
entirely, fall through to the read-mode config.**
The absent-version case is not hypothetical and is easy to get wrong:
**Spark 2.4.5 and earlier
wrote no `org.apache.spark.version` key at all**, so the canonical legacy
files carry no metadata
hint whatsoever. Any detector keyed on "marker present" or "version < 3.0.0"
clears exactly the
files that need rebasing most. Credit to @peterxcli — #5048 got this right
and it is what caught
the same bug in my first commit.
Note Spark compares with `version < minVersion` as a **string**, not a
semantic version compare. A
native or JVM reimplementation should replicate the string comparison rather
than "fixing" it, or
it will diverge on odd version strings.
**A2. Two markers, two thresholds, tracked independently.**
`org.apache.spark.legacyDateTime` with `minVersion` 3.0.0 for
date/`TIMESTAMP_MICROS`/
`TIMESTAMP_MILLIS`, and `org.apache.spark.legacyINT96` with `minVersion`
**3.1.0** for INT96.
Collapsing them into one flag or one version threshold is wrong in both
directions.
**A3. Read modes come from `ParquetOptions`, not `SQLConf`.** Spark resolves
`datetimeRebaseModeInRead` / `int96RebaseModeInRead` through
`new ParquetOptions(options, conf)`, so a per-read
`.option("datetimeRebaseMode", "CORRECTED")`
overrides the session conf. Reading `SQLConf` directly silently ignores it.
(Again @peterxcli's
catch, from reviewing #5202.)
**A4. The file-level marker says nothing about whether any value is actually
affected.** Spark
stamps `legacyDateTime` on a *whole file* whenever the write mode was
LEGACY, regardless of the
values — and dates from 1582-10-15 onward rebase to themselves
(`RebaseDateTime.lastSwitchJulianDay == -141427`, and `julianGregDiffs.last
== 0`). So a large
fraction of marked files return perfectly correct results under Comet today.
Marker-only detection
over-triggers heavily: whichever remedy is chosen (raise, fall back, or
rebase), it fires on reads
that need nothing.
Row-group min/max statistics are the only cheap discriminator, and they are
only available where
the footer is already in hand.
**A5. Statistics cannot save INT96.** The Parquet spec gives INT96's 12
bytes no meaningful
ordering, so writers emit no usable min/max. INT96 columns therefore have to
be handled
conservatively regardless of statistics. Same for any row group with no
statistics. An all-null row
group *is* clear (no value to rebase), and so is an empty one.
**A6. Thresholds must come from Spark, not from native constants.**
`RebaseDateTime.lastSwitchJulianTs` is not a constant anyone should
re-derive: it is
`rebaseMap.values.map(_.switches.last).max` over Spark's **per-timezone**
rebase tables, with a
`require` that all diffs after it are zero. Read `lastSwitchJulianDay` and
`lastSwitchJulianTs` on
the JVM and send them through the proto, so the thresholds stay exact for
whichever Spark version
is in use. Scaling them to millis/nanos has to round toward the *unsafe*
direction and saturate
rather than overflow.
**A7. Timestamp rebasing is timezone-dependent, and the timezone is per
file.** When the resolved
policy is LEGACY, Spark builds `RebaseSpec(LEGACY,
Option(lookupFileMeta(SPARK_TIMEZONE_METADATA_KEY)))`
— i.e. it rebases using the writer's `org.apache.spark.timeZone`, falling
back to the session
timezone. Any actual native rebase implementation (option 1 below) needs the
full per-timezone
tables and per-file timezone resolution, not a single global offset. This is
the bulk of the work
in option 1 and is why it is not a small change.
**A8. The affected-column test must walk nested types.** Date/timestamp
columns hide inside
structs, lists, maps, dictionaries and unions, so the walk has to recurse —
and it must consider
only the *requested* top-level columns (case-insensitively), or reading a
modern date column out of
a file that also holds an ancient one fails for no reason.
---
### B. Where the check runs decides which remedies are even reachable
**B1. Plan-time detection costs real planning latency.** #5048's
`requiresDatetimeRebase` opens
every selected file's footer on the driver, per query, for any query
touching a date or timestamp
column. Three things compound: it uses `selectedPartitions` filtered with
`partitionFilters.filterNot(isDynamicPruningFilter)`, so it is 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 is serial. On
a table with thousands
of files that is significant added planning time and driver heap, paid even
when every file is
fine.
**B2. Read-time detection is free but cannot fall back.** In the native
reader the footer has
already been fetched and cached for the read itself, so inspection costs no
extra I/O, and it is the
only point where row-group statistics (A4) are available. But by then the
plan is fixed: the only
available outcome is to raise. **This is the core tension** —
correct-results-via-fallback requires
paying B1; free detection can only fail the query.
**B3. If done natively, the enforcement point must be the reader factory's
`get_metadata`.** Not the
expression/schema adapter: DataFusion only constructs that when the logical
and physical schemas
differ or a predicate is pushed down, so a plain `SELECT d FROM t` skips it
entirely.
---
### C. If the remedy is to raise, the error plumbing is not trivial
**C1. It has to surface as Spark's own exception.** `SparkUpgradeException`
with
`INCONSISTENT_BEHAVIOR_CROSS_VERSION.READ_ANCIENT_DATETIME`. This matters
mechanically, not just
cosmetically: `FileScanRDD` deliberately rethrows `SparkUpgradeException`
(two sites) instead of
wrapping it in `FAILED_READ_FILE`/`Encountered error while reading file`,
because the file is not
corrupt. Anything that loses the type gets misreported as a corrupt-file
error.
**C2. Getting it out of the native reader needs a downcast, not message
matching.**
`AsyncFileReader::get_metadata` can only return a `ParquetError`, so the
typed error has to travel
boxed in `ParquetError::External` and be recovered by downcast in the error
classifier.
**C3. Unresolved: the message conflicts with Comet's actual remedy.**
Spark's templated
`READ_ANCIENT_DATETIME` message advises setting `datetimeRebaseModeInRead`
to `LEGACY`/`CORRECTED`,
which does not help under Comet — the remedy is to disable Comet for the
query. Matching Spark's
exception type means Comet's real guidance can only travel as the cause; a
Comet-specific exception
makes the guidance primary but is no longer a `SparkUpgradeException` (see
C1). No good answer found.
---
### D. A raise-only remedy leaves a real divergence
Under read mode `LEGACY` on a version-less file, Spark rebases the values
and returns them
correctly. Comet has no rebasing to do it with, so it can only refuse a read
Spark completes
successfully. Only option 1 closes this.
---
### E. Test assets that already exist, and one test that asserts the bug
The 15 `before_1582_*` fixtures are already checked in under
`spark/src/test/resources/test-data/` — date / `TIMESTAMP_MICROS` /
`TIMESTAMP_MILLIS` / INT96 plain
/ INT96 dict, each written by Spark 2.4.5, 2.4.6 and 3.2.0. That covers
every physical encoding and
both footer paths: the 2.4.x files are the version-less case (A1), the 3.2.0
files the
marker-stamped case. Any fix should be validated against all 15.
`ParquetReadSuite.scala:1714` (`"reading ancient dates before 1582"`)
currently **asserts the wrong
behaviour** — its comment states "no rebase, no exception". It has to change
with any fix here.
---
### F. Also still outstanding after #5221
#5221 removes the config from `CometConf.scala` and the `scans.md` sentence,
but not the dead native
field `SparkParquetOptions::use_legacy_date_timestamp_or_ntz`
(`native/core/src/parquet/parquet_support.rs:78`, only ever initialized to
`false` at lines 111 and
127, never consulted). Both #5048 and #5202 removed it; neither has merged,
so it survives. Worth
removing independently.
---
### Where that leaves the options
With the costs now measured, restating the choices in the issue body:
1. **Rebase natively.** The only option with no divergence (D) and no
spurious failures (A4), but
it needs Spark's per-timezone rebase tables and per-file timezone
resolution ported natively
(A7). Largest change by a wide margin. #5047 (draft) is the existing
attempt.
2. **Fall back to Spark.** Always correct, but requires plan-time footer
reads (B1) and cannot use
row-group statistics, so it over-triggers (A4). The plan-time cost is the
objection to #5048;
it could be reduced (`SKIP_ROW_GROUPS`, post-DPP file set, parallelized)
but not eliminated.
3. **Raise.** Free to detect and precise thanks to statistics (B2, A4), but
fails queries Spark
answers (D) and needs the error plumbing in section C. This was #5202.
My own read is that option 3 is not worth shipping on its own — turning
silently-wrong results into
a hard failure is an improvement, but a narrower one than it looks once D is
accounted for, and it
spends the error-plumbing complexity without moving toward option 1. Option
2 with the plan-time
cost reduced is the pragmatic interim, and option 1 is the real fix. Other
opinions welcome —
cc @peterxcli.
--
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]
