diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index ff69a2cc68d..821ea36a36c 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -307,6 +307,7 @@ jobs: org.apache.comet.parquet.ParquetReadV2Suite org.apache.comet.parquet.ParquetReadFromFakeHadoopFsSuite org.apache.comet.parquet.ParquetTimestampLtzAsNtzSuite + org.apache.comet.parquet.ParquetDatetimeRebaseSuite org.apache.spark.sql.comet.ParquetDatetimeRebaseV1Suite org.apache.spark.sql.comet.ParquetDatetimeRebaseV2Suite org.apache.spark.sql.comet.ParquetEncryptionITCase diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index f3f3f082695..0f4f828e534 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -123,6 +123,7 @@ jobs: org.apache.comet.parquet.ParquetReadV2Suite org.apache.comet.parquet.ParquetReadFromFakeHadoopFsSuite org.apache.comet.parquet.ParquetTimestampLtzAsNtzSuite + org.apache.comet.parquet.ParquetDatetimeRebaseSuite org.apache.spark.sql.comet.ParquetDatetimeRebaseV1Suite org.apache.spark.sql.comet.ParquetDatetimeRebaseV2Suite org.apache.spark.sql.comet.ParquetEncryptionITCase diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md index 0c07b5c2370..6909fd45a5b 100644 --- a/docs/source/user-guide/latest/compatibility/scans.md +++ b/docs/source/user-guide/latest/compatibility/scans.md @@ -49,20 +49,48 @@ The following features are not supported and cause Comet to fall back to Spark: `spark.comet.scan.allowDisabledParquetVectorizedReader=true` to opt in to running the Comet Parquet scan regardless. -The following limitation may produce incorrect results without falling back to Spark: - -- No support for datetime rebasing. When reading Parquet files containing dates or timestamps - written with `spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY` (which is Spark's default for - data written before Spark 3.0, using the hybrid Julian/Gregorian calendar), Comet reads them as - if they were written using the Proleptic Gregorian calendar. This produces silently-wrong - values for dates before October 15, 1582 in both projections and predicates. Comet also - ignores `spark.sql.parquet.datetimeRebaseModeInRead` and the file-level - `org.apache.spark.legacyDateTime` metadata that would tell it to rebase. The - `spark.comet.exceptionOnDatetimeRebase` config is currently dead code and does not raise on - legacy-calendar data. Tracked by +The following limitations raise an error at scan time rather than falling back to Spark: + +- No support for datetime rebasing. Spark writes dates/timestamps in the legacy hybrid + Julian/Gregorian calendar when `spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY`, and every + Spark before 3.0 did so unconditionally. Spark rebases those values to the Proleptic Gregorian + calendar on read; Comet does not implement rebasing, so reading them would return values shifted + by up to ten days. Rather than return wrong answers, Comet fails the read. Tracked by [#5010](https://github.com/apache/datafusion-comet/issues/5010). -The following limitations raise an error at scan time rather than falling back to Spark: + Comet only fails a read that would actually be affected, so this does not reject legacy-written + data wholesale. A read is refused when all of the following hold: + + 1. The scan reads a date or timestamp column (including nested ones). Reading only other columns + out of an affected file is fine. + 2. The Parquet footer does not prove those values are already Proleptic Gregorian. It proves it + for an `org.apache.spark.version` of 3.0 or later (3.1 for `INT96`) with no + `org.apache.spark.legacyDateTime` / `org.apache.spark.legacyINT96` marker, and also when + `spark.sql.parquet.datetimeRebaseModeInRead` / `spark.sql.parquet.int96RebaseModeInRead` is set + to `CORRECTED`, which Comet honors for version-less files exactly as Spark does. + 3. Parquet row-group statistics show the column actually holds a date before October 15, 1582 or a + timestamp before 1900-01-01T00:00:00Z. A legacy-written file whose values are all newer than + that rebases to itself and is read normally. + + How a column with no usable statistics is treated depends on what the footer said. If the + footer positively marks the file as legacy — an `org.apache.spark.legacyDateTime` / + `org.apache.spark.legacyINT96` marker, or a Spark older than the switch version — the read is + refused, and that includes every `INT96` column, whose bytes the Parquet spec gives no + ordering. If the footer records no writer version at all, the read goes ahead. + + That last distinction matters for files Spark did not write. Hive, Impala, Trino and plain + parquet-mr leave `org.apache.spark.version` unset, and Hive writes `TIMESTAMP` as `INT96`. + Refusing every unprovable column would make all such timestamp data unreadable through Comet + whatever its values, so Comet refuses only what it can positively show is affected. The + consequence is one gap: an `INT96` column, or a column written with statistics disabled, in a file + with no recorded writer version _and_ holding genuinely ancient values is read unrebased, where + Spark would have raised. Comet's `spark.comet.exceptionOnDatetimeRebase` cannot detect that case; + disable Comet for the query if you need Spark's behaviour there. + + The failure is a `SparkUpgradeException`, matching what Spark raises for the same data. Set + `spark.comet.exceptionOnDatetimeRebase=false` to read affected values as-is, without rebasing; + this reproduces the silently-incorrect results of earlier Comet versions. To get correct values, + disable Comet for the query so that Spark performs the rebasing. - Invalid UTF-8 bytes in `STRING` columns. Spark permits arbitrary byte sequences in a `STRING` column (for example from `CAST(X'C1' AS STRING)`), but Comet's native execution path is built on diff --git a/native/common/src/error.rs b/native/common/src/error.rs index baeb3a119ea..8feba10b04f 100644 --- a/native/common/src/error.rs +++ b/native/common/src/error.rs @@ -202,6 +202,16 @@ pub enum SparkError { #[error("Spark read schema expects field Ids, but Parquet file schema doesn't contain any field Ids. Please remove the field ids from Spark schema or ignore missing ids by setting `spark.sql.parquet.fieldId.read.ignoreMissing = true`")] ParquetMissingFieldIds, + /// A Parquet file holding dates/timestamps written in the legacy hybrid (Julian + Gregorian) + /// calendar, which Comet's native scan cannot rebase (#5010). Translated by the JVM shim into + /// Spark's `SparkUpgradeException` -- the same exception type Spark itself raises for this + /// data, and one Spark's `FileScanRDD` deliberately rethrows rather than wrapping in + /// `FAILED_READ_FILE`, since the file is perfectly readable and only Comet's calendar + /// handling is at fault. The `file_path` may be empty, in which case the JVM side fills it + /// from the per-task file list. + #[error("{message}")] + LegacyDatetimeRebase { file_path: String, message: String }, + /// Schema mismatch when reading a Parquet column under a requested schema /// that's incompatible with the physical column type. Translated by the JVM /// shim into Spark's `SchemaColumnConvertNotSupportedException`. The @@ -299,6 +309,7 @@ impl SparkError { SparkError::DuplicateFieldCaseInsensitive { .. } => "DuplicateFieldCaseInsensitive", SparkError::DuplicateFieldByFieldId { .. } => "DuplicateFieldByFieldId", SparkError::ParquetMissingFieldIds => "ParquetMissingFieldIds", + SparkError::LegacyDatetimeRebase { .. } => "LegacyDatetimeRebase", SparkError::ParquetSchemaConvert { .. } => "ParquetSchemaConvert", SparkError::CannotReadFile { .. } => "CannotReadFile", SparkError::Arrow(_) => "Arrow", @@ -544,6 +555,12 @@ impl SparkError { "message": message, }) } + SparkError::LegacyDatetimeRebase { file_path, message } => { + serde_json::json!({ + "filePath": file_path, + "message": message, + }) + } SparkError::Arrow(e) => { serde_json::json!({ "message": e.to_string(), @@ -628,6 +645,10 @@ impl SparkError { // file lacks field ids and `spark.sql.parquet.fieldId.read.ignoreMissing=false`. SparkError::ParquetMissingFieldIds => "java/lang/RuntimeException", + // LegacyDatetimeRebase - converted to SparkUpgradeException by the shim, matching + // what Spark raises for hybrid-calendar data it is asked not to rebase. + SparkError::LegacyDatetimeRebase { .. } => "org/apache/spark/SparkUpgradeException", + // ParquetSchemaConvert - converted to SchemaColumnConvertNotSupportedException by the shim SparkError::ParquetSchemaConvert { .. } => { "org/apache/spark/sql/execution/datasources/SchemaColumnConvertNotSupportedException" @@ -722,6 +743,11 @@ impl SparkError { // ParquetMissingFieldIds is a plain RuntimeException with no error class. SparkError::ParquetMissingFieldIds => None, + // LegacyDatetimeRebase - the JVM shim supplies Spark's + // INCONSISTENT_BEHAVIOR_CROSS_VERSION.READ_ANCIENT_DATETIME class when it builds the + // SparkUpgradeException, so none is exposed here. + SparkError::LegacyDatetimeRebase { .. } => None, + // Parquet schema mismatch — translated to SchemaColumnConvertNotSupportedException // by the JVM shim. The shim wraps it in the version-appropriate // SparkException error class, so no error class is exposed here. diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 5accd316096..f812df6c568 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -101,6 +101,7 @@ use datafusion::physical_expr::expressions::{Literal, StatsType}; use datafusion::physical_expr::window::WindowExpr; use datafusion::physical_expr::LexOrdering; +use crate::parquet::legacy_datetime::{LegacyCalendarGuard, ReadModes}; use crate::parquet::parquet_exec::init_datasource_exec; use arrow::array::{ new_empty_array, Array, ArrayRef, BinaryBuilder, BooleanArray, Date32Array, Decimal128Array, @@ -1535,6 +1536,16 @@ impl PhysicalPlanner { let files = self.get_partitioned_files(partition_files)?; let file_groups: Vec> = vec![files]; + let legacy_calendar_guard = LegacyCalendarGuard::for_scan( + common.exception_on_legacy_datetime, + &required_schema, + common.case_sensitive, + ReadModes { + datetime_corrected: common.legacy_datetime_read_mode_corrected, + int96_corrected: common.legacy_int96_read_mode_corrected, + }, + ); + let scan = init_datasource_exec( required_schema, Some(data_schema), @@ -1553,6 +1564,7 @@ impl PhysicalPlanner { common.encryption_enabled, common.use_field_id, common.ignore_missing_field_id, + legacy_calendar_guard, )?; Ok(( vec![], diff --git a/native/core/src/parquet/eager_page_index_reader_factory.rs b/native/core/src/parquet/comet_parquet_reader_factory.rs similarity index 76% rename from native/core/src/parquet/eager_page_index_reader_factory.rs rename to native/core/src/parquet/comet_parquet_reader_factory.rs index 278814c4bf9..aea39896293 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/comet_parquet_reader_factory.rs @@ -15,9 +15,11 @@ // specific language governing permissions and limitations // under the License. -//! A `ParquetFileReaderFactory` that always loads the Parquet page index into the shared -//! `FileMetadataCache` on the first metadata fetch for a file, instead of deferring to -//! DataFusion's opener. +//! Comet's `ParquetFileReaderFactory`. Every native Parquet scan installs it, so it is the one +//! place guaranteed to see each file's footer exactly once per open. It has two jobs, described +//! below: eager page-index loading, and legacy-calendar rejection. +//! +//! # Eager page-index loading //! //! DataFusion's opener requests `PageIndexPolicy::Skip` on the initial metadata load and defers //! loading the page index until row-group pruning shows it is still needed @@ -43,9 +45,19 @@ //! fetch to encrypted scans that have no pruning predicate at all. Encrypted opens get exactly //! the caller's requested policy, unchanged from stock behavior. //! -//! Filed upstream as apache/datafusion#23978. Revert this once the opener merges its deferred -//! page-index load back into `FileMetadataCache` instead of bypassing it. +//! Filed upstream as apache/datafusion#23978. Revert this behavior once the opener merges its +//! deferred page-index load back into `FileMetadataCache` instead of bypassing it. +//! +//! # Legacy-calendar rejection +//! +//! When `spark.comet.exceptionOnDatetimeRebase` is enabled and the scan reads a date or timestamp +//! column, a file whose footer says its values were written in the legacy hybrid calendar fails +//! the scan rather than returning silently-unrebased values. See [`crate::parquet:: +//! legacy_datetime`]. The check lives here, rather than in the schema/expression adapter, because +//! the adapter is only created when the logical and physical schemas differ or a predicate is +//! pushed down, whereas `get_metadata` runs for every file. +use crate::parquet::legacy_datetime::LegacyCalendarGuard; use bytes::Bytes; use datafusion::common::Result as DFResult; use datafusion::datasource::physical_plan::parquet::metadata::DFParquetMetadata; @@ -66,21 +78,29 @@ use std::ops::Range; use std::sync::Arc; #[derive(Debug)] -pub struct EagerPageIndexReaderFactory { +pub struct CometParquetFileReaderFactory { store: Arc, metadata_cache: Arc, + /// `Some` only when `spark.comet.exceptionOnDatetimeRebase` is on *and* this scan reads a + /// calendar-sensitive column, so a disarmed guard costs one `Option` check per file. + legacy_calendar_guard: Option, } -impl EagerPageIndexReaderFactory { - pub fn new(store: Arc, metadata_cache: Arc) -> Self { +impl CometParquetFileReaderFactory { + pub fn new( + store: Arc, + metadata_cache: Arc, + legacy_calendar_guard: Option, + ) -> Self { Self { store, metadata_cache, + legacy_calendar_guard, } } } -impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { +impl ParquetFileReaderFactory for CometParquetFileReaderFactory { fn create_reader( &self, partition_index: usize, @@ -102,27 +122,29 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { inner = inner.with_footer_size_hint(hint); } - Ok(Box::new(EagerPageIndexReader { + Ok(Box::new(CometParquetFileReader { file_metrics, store: Arc::clone(&self.store), inner, partitioned_file, metadata_cache: Arc::clone(&self.metadata_cache), metadata_size_hint, + legacy_calendar_guard: self.legacy_calendar_guard.clone(), })) } } -struct EagerPageIndexReader { +struct CometParquetFileReader { file_metrics: ParquetFileMetrics, store: Arc, inner: ParquetObjectReader, partitioned_file: PartitionedFile, metadata_cache: Arc, metadata_size_hint: Option, + legacy_calendar_guard: Option, } -impl AsyncFileReader for EagerPageIndexReader { +impl AsyncFileReader for CometParquetFileReader { fn get_bytes(&mut self, range: Range) -> BoxFuture<'_, parquet::errors::Result> { let bytes_scanned = range.end - range.start; self.file_metrics.bytes_scanned.add(bytes_scanned as usize); @@ -151,6 +173,7 @@ impl AsyncFileReader for EagerPageIndexReader { let metadata_cache = Arc::clone(&self.metadata_cache); let store = Arc::clone(&self.store); let metadata_size_hint = self.metadata_size_hint; + let legacy_calendar_guard = self.legacy_calendar_guard.clone(); async move { let file_decryption_properties = options .and_then(|o| o.file_decryption_properties()) @@ -161,7 +184,7 @@ impl AsyncFileReader for EagerPageIndexReader { options.map(|o| o.column_index_policy()) }; - DFParquetMetadata::new(store.as_ref(), &object_meta) + let metadata = DFParquetMetadata::new(store.as_ref(), &object_meta) .with_decryption_properties(file_decryption_properties) .with_file_metadata_cache(Some(metadata_cache)) .with_metadata_size_hint(metadata_size_hint) @@ -173,13 +196,19 @@ impl AsyncFileReader for EagerPageIndexReader { "Failed to fetch metadata for file {}: {e}", object_meta.location, )) - }) + })?; + + if let Some(guard) = &legacy_calendar_guard { + guard.check(&metadata)?; + } + + Ok(metadata) } .boxed() } } -impl Drop for EagerPageIndexReader { +impl Drop for CometParquetFileReader { fn drop(&mut self) { self.file_metrics .scan_efficiency_ratio diff --git a/native/core/src/parquet/legacy_datetime.rs b/native/core/src/parquet/legacy_datetime.rs new file mode 100644 index 00000000000..04299978bf0 --- /dev/null +++ b/native/core/src/parquet/legacy_datetime.rs @@ -0,0 +1,1079 @@ +// 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. + +//! Rejection of Parquet reads that would silently return unrebased dates/timestamps. +//! +//! Spark writes dates/timestamps in the legacy hybrid (Julian + Gregorian) calendar when +//! `spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY`, and every Spark before 3.0 did so +//! unconditionally. Spark rebases those values back to the Proleptic Gregorian calendar on read. +//! Comet's native scan does not implement rebasing (#5010), so it would return values shifted by +//! up to ten days. `spark.comet.exceptionOnDatetimeRebase` (on by default) fails such a read +//! instead. +//! +//! # Why the decision is two-stage +//! +//! Footer metadata alone is too coarse in both directions, so the footer narrows the question and +//! row-group statistics answer it. +//! +//! Too strict: Spark stamps `org.apache.spark.legacyDateTime` on a whole file whenever the write +//! mode was LEGACY, whether or not any value is old enough to rebase -- and dates from 1582-10-15 +//! onward rebase to themselves. Rejecting on the marker alone would fail a large number of reads +//! that return perfectly correct results. +//! +//! Too lax: Spark 2.4.5 and earlier wrote no `org.apache.spark.version` key at all, so the +//! canonical legacy files carry no marker whatsoever. Spark handles those through +//! `spark.sql.parquet.datetimeRebaseModeInRead`, whose default of EXCEPTION makes Spark raise when +//! it decodes an actually-ancient value. +//! +//! So the footer answers a three-way question per column -- see [`Calendar`] -- and how much +//! Parquet row-group statistics then have to prove depends on the answer: +//! +//! - [`Calendar::Gregorian`]: no rebasing would apply. Read as-is, no statistics consulted. +//! - [`Calendar::Legacy`]: the file says these values *are* hybrid-calendar, so anything +//! statistics cannot rule out is assumed affected. A column with no statistics is refused, and +//! so is any INT96 column, whose 12 bytes the Parquet spec gives no meaningful ordering. +//! - [`Calendar::Unknown`]: nothing says they are. Refuse only on positive proof -- a row-group +//! minimum below Spark's threshold. Missing statistics and INT96 prove nothing and are read. +//! +//! That last asymmetry is what keeps the default-on guard usable. Every non-Spark writer -- Hive, +//! Impala, Trino, plain parquet-mr -- leaves the version key unset, and Hive in particular writes +//! its timestamps as INT96. Treating unknown provenance as conservatively as a legacy marker would +//! refuse every Hive-written timestamp column ever, whatever its values, where Spark reads them +//! without complaint. Spark under its own EXCEPTION default raises only for a value it actually +//! decodes below the switch point, so refusing only on proof is what agrees with Spark. +//! +//! The residual gap, which [`ReadModes`] also notes: an unknown-provenance file whose affected +//! values statistics cannot expose -- an INT96 column, or one written with statistics off -- is +//! read unrebased, where Spark would have raised. Closing that needs a per-value check in the +//! decoder, or the rebasing itself (#5010). + +use arrow::datatypes::{DataType, Schema}; +use datafusion_comet_common::SparkError; +use parquet::basic::{ + ConvertedType, LogicalType, TimeUnit as ParquetTimeUnit, Type as PhysicalType, +}; +use parquet::errors::ParquetError; +use parquet::file::metadata::{KeyValue, ParquetMetaData}; +use parquet::file::statistics::Statistics; +use parquet::schema::types::ColumnDescriptor; + +/// Spark's Parquet footer key-value metadata keys, from `org.apache.spark.sql.package`. +const SPARK_VERSION_METADATA_KEY: &str = "org.apache.spark.version"; +const SPARK_LEGACY_DATETIME_METADATA_KEY: &str = "org.apache.spark.legacyDateTime"; +const SPARK_LEGACY_INT96_METADATA_KEY: &str = "org.apache.spark.legacyINT96"; + +/// The Spark version that switched each encoding to the Proleptic Gregorian calendar. Compared as +/// plain strings, exactly as Spark's `DataSourceUtils.getRebaseSpec` does -- matching Spark matters +/// more here than being right about version ordering, and the two only diverge for a hypothetical +/// major version of 10 or above. +const DATETIME_GREGORIAN_SINCE: &str = "3.0.0"; +const INT96_GREGORIAN_SINCE: &str = "3.1.0"; + +/// Days since the epoch at and above which julian-to-gregorian date rebasing is the identity: +/// 1582-10-15, the first Gregorian day. Mirrors Spark's `RebaseDateTime.lastSwitchJulianDay`. +const LAST_SWITCH_JULIAN_DAY: i32 = -141_427; + +/// Micros since the epoch at and above which julian-to-gregorian timestamp rebasing is the +/// identity for every time zone: 1900-01-01T00:00:00Z. Mirrors Spark's +/// `RebaseDateTime.lastSwitchJulianTs`, which Spark derives as the maximum switch point across its +/// per-timezone rebase tables. +/// +/// Both constants are asserted against the running Spark's own values by +/// `ParquetDatetimeRebaseSuite`, so a change on Spark's side fails a test rather than silently +/// shifting the threshold. They are duplicated here rather than sent from the JVM because reading +/// `lastSwitchJulianTs` forces `RebaseDateTime`'s static initializer, which parses ~590 KB of +/// bundled JSON and retains several MB -- a cost every driver would otherwise pay on its first +/// native scan, whether or not the guard is armed. +const LAST_SWITCH_JULIAN_MICROS: i64 = -2_208_988_800_000_000; + +/// Whether the user has told Spark to assume Proleptic Gregorian for files whose provenance the +/// footer does not record, via `spark.sql.parquet.datetimeRebaseModeInRead=CORRECTED` and its INT96 +/// counterpart. Both default to EXCEPTION, under which such a file is read only as far as +/// statistics can show it holds nothing affected. +/// +/// Where Comet still diverges from Spark for a version-less file: Spark decides per decoded value, +/// so it raises (under EXCEPTION) or rebases (under LEGACY) exactly the values that need it, while +/// Comet decides per column from statistics. So Comet reads an affected value unrebased when +/// statistics cannot expose it -- an INT96 column, or one written with statistics off -- and +/// refuses a whole column under LEGACY, which Spark would have rebased and returned correctly, +/// because Comet has no rebasing to do it with. +#[derive(Debug, Clone, Copy)] +pub struct ReadModes { + /// `spark.sql.parquet.datetimeRebaseModeInRead == CORRECTED`. + pub datetime_corrected: bool, + /// `spark.sql.parquet.int96RebaseModeInRead == CORRECTED`. + pub int96_corrected: bool, +} + +/// The armed legacy-calendar guard for one scan. Built once per scan; consulted once per file open. +#[derive(Debug, Clone)] +pub struct LegacyCalendarGuard { + /// Top-level requested field names that can decode a date or timestamp, used to ignore + /// calendar-sensitive columns the scan never reads. + requested_roots: Vec, + case_sensitive: bool, + read_modes: ReadModes, +} + +impl LegacyCalendarGuard { + /// `None` when the guard cannot ever fire for this scan, either because the config is off or + /// because nothing calendar-sensitive is being read. Callers keep that as `None` so a disarmed + /// guard costs a single `Option` check per file rather than any footer inspection. + pub fn for_scan( + enabled: bool, + required_schema: &Schema, + case_sensitive: bool, + read_modes: ReadModes, + ) -> Option { + if !enabled { + return None; + } + let requested_roots: Vec = required_schema + .fields() + .iter() + .filter(|field| data_type_has_date_or_timestamp(field.data_type())) + .map(|field| field.name().clone()) + .collect(); + if requested_roots.is_empty() { + return None; + } + Some(Self { + requested_roots, + case_sensitive, + read_modes, + }) + } + + /// `Err` if this file must not be read: some calendar-sensitive column the scan reads is not + /// provably Proleptic Gregorian, and either holds a value that would need rebasing or cannot be + /// shown not to. + pub fn check(&self, metadata: &ParquetMetaData) -> Result<(), ParquetError> { + if self.reads_unrebasable_values(metadata) { + return Err(legacy_calendar_error()); + } + Ok(()) + } + + fn reads_unrebasable_values(&self, metadata: &ParquetMetaData) -> bool { + let file_metadata = metadata.file_metadata(); + // Footer facts are per file, so resolve them once rather than per column. + let provenance = Provenance::from_footer(file_metadata.key_value_metadata()); + let descr = file_metadata.schema_descr(); + + for (leaf_index, column) in descr.columns().iter().enumerate() { + // `calendar_kind` is the cheaper of the two filters and prunes far more columns on a + // wide schema, so it goes first. + let Some(kind) = calendar_kind(column) else { + continue; + }; + if !self.reads(column) { + continue; + } + // How to treat a value statistics cannot decide on. See the module docs: a file that + // declares itself legacy gets the benefit of the doubt, one that declares nothing does + // not, because assuming the worst there would refuse every Hive-written timestamp. + let undecidable_is_affected = match provenance.calendar(kind, &self.read_modes) { + Calendar::Gregorian => continue, + Calendar::Legacy => true, + Calendar::Unknown => false, + }; + let Some(threshold) = kind.threshold() else { + // INT96 carries no usable statistics -- the Parquet spec gives its 12 bytes no + // meaningful ordering, so writers either omit min/max or write values that must + // not be compared. There is nothing to prove or disprove safety with. + if undecidable_is_affected { + return true; + } + continue; + }; + if row_groups_hold_values_below( + metadata, + leaf_index, + threshold, + undecidable_is_affected, + ) { + return true; + } + } + false + } + + /// Whether this leaf column sits under a top-level field the scan actually reads. Comparing + /// the root of the column path (rather than the full path) keeps nested date/timestamp fields + /// covered without having to reconstruct Parquet's list/map path encodings. + fn reads(&self, column: &ColumnDescriptor) -> bool { + let Some(root) = column.path().parts().first() else { + return false; + }; + self.requested_roots.iter().any(|name| { + if self.case_sensitive { + name == root + } else { + // Matches how the schema adapter resolves the same names. + name.eq_ignore_ascii_case(root) + } + }) + } +} + +/// What a file's footer says about the calendar its values were written in, resolved once per file. +struct Provenance<'a> { + /// The Spark version that wrote the file, absent for non-Spark writers and for Spark 2.4.5 and + /// earlier, which did not stamp the key. + writer_version: Option<&'a str>, + has_legacy_datetime_marker: bool, + has_legacy_int96_marker: bool, +} + +impl<'a> Provenance<'a> { + fn from_footer(key_value_metadata: Option<&'a Vec>) -> Self { + let has_key = |key: &str| { + key_value_metadata.is_some_and(|kv| kv.iter().any(|entry| entry.key == key)) + }; + Self { + writer_version: key_value_metadata.and_then(|kv| { + kv.iter() + .find(|entry| entry.key == SPARK_VERSION_METADATA_KEY) + .and_then(|entry| entry.value.as_deref()) + }), + has_legacy_datetime_marker: has_key(SPARK_LEGACY_DATETIME_METADATA_KEY), + has_legacy_int96_marker: has_key(SPARK_LEGACY_INT96_METADATA_KEY), + } + } + + /// What the footer says about the calendar this column's values were written in. + /// + /// Mirrors Spark's `DataSourceUtils.getRebaseSpec`, which resolves a legacy marker first, then + /// the writer version, then -- when the file records no version -- the read-mode config. The + /// only difference is that Comet keeps "the config did not say CORRECTED" as its own + /// [`Calendar::Unknown`] answer rather than collapsing it into legacy, because what it can + /// prove from statistics differs. See the module docs. + fn calendar(&self, kind: CalendarKind, read_modes: &ReadModes) -> Calendar { + let (gregorian_since, has_legacy_marker, versionless_is_corrected) = match kind { + CalendarKind::Int96 => ( + INT96_GREGORIAN_SINCE, + self.has_legacy_int96_marker, + read_modes.int96_corrected, + ), + CalendarKind::Date | CalendarKind::Timestamp(_) => ( + DATETIME_GREGORIAN_SINCE, + self.has_legacy_datetime_marker, + read_modes.datetime_corrected, + ), + }; + // An explicit marker is authoritative, and outranks both the writer version (a modern Spark + // writing with rebaseModeInWrite=LEGACY stamps both) and the read mode. + if has_legacy_marker { + return Calendar::Legacy; + } + match self.writer_version { + Some(version) if version >= gregorian_since => Calendar::Gregorian, + Some(_) => Calendar::Legacy, + None if versionless_is_corrected => Calendar::Gregorian, + None => Calendar::Unknown, + } + } +} + +/// What a file's footer establishes about the calendar a column's values were written in. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Calendar { + /// Provably already Proleptic Gregorian: a writer at or after the switch version that did not + /// opt back into LEGACY, or a version-less file the read mode declares CORRECTED. No rebasing + /// would apply, so Comet reads the values as-is. + Gregorian, + /// Provably the legacy hybrid calendar: an explicit legacy marker, or a Spark older than the + /// switch version. Spark would rebase these, so Comet refuses anything statistics cannot clear. + Legacy, + /// The footer records no writer version -- Spark 2.4.5 and earlier, and every non-Spark writer + /// -- and no read mode declared it corrected. The values may or may not be hybrid-calendar, so + /// Comet refuses only what statistics positively expose. + Unknown, +} + +/// Which of Spark's two rebase regimes a calendar-sensitive leaf column falls under. Spark tracks +/// INT96 separately from the other encodings: it switched calendars a release later and has its own +/// footer marker and read-mode config. +#[derive(Debug, Clone, Copy)] +enum CalendarKind { + /// A DATE column, in days. + Date, + /// TIMESTAMP_MILLIS or TIMESTAMP_MICROS, in an INT64 column whose statistics are ordinary + /// signed integers. + Timestamp(ParquetTimeUnit), + /// An INT96 timestamp. + Int96, +} + +impl CalendarKind { + /// The rebase threshold in this column's own physical units, or `None` when statistics cannot + /// decide the question. + fn threshold(self) -> Option { + match self { + CalendarKind::Date => Some(LAST_SWITCH_JULIAN_DAY as i64), + CalendarKind::Timestamp(unit) => Some(timestamp_threshold(unit)), + CalendarKind::Int96 => None, + } + } +} + +/// Classify a leaf column, or `None` if its values are not calendar-sensitive. +/// +/// Both `LogicalType` and the deprecated `ConvertedType` are consulted: legacy-calendar files are +/// by definition old, and files written before Parquet's logical-type rework carry only the +/// converted type. +/// +/// A `TIMESTAMP` that is not adjusted to UTC is Spark's TIMESTAMP_NTZ, which Spark never rebases in +/// either direction -- "TIMESTAMP_NTZ is a new data type and has no legacy files that need to do +/// rebase", as its `ParquetVectorUpdaterFactory` puts it. Spark still stamps +/// `org.apache.spark.legacyDateTime` on any file written under a LEGACY rebase mode whether or not +/// the schema has a column the mode could apply to, so without this an NTZ column in such a file +/// would be refused over values Spark reads back exactly as written. +/// +/// The classification is deliberately taken from the Parquet type rather than the requested Arrow +/// type: `spark.comet.allowTimestampLtzAsNtz` can request an NTZ Arrow type over a UTC-adjusted +/// Parquet column, whose values do need rebasing, and the file is what knows which it holds. +fn calendar_kind(column: &ColumnDescriptor) -> Option { + if column.physical_type() == PhysicalType::INT96 { + return Some(CalendarKind::Int96); + } + match column.logical_type_ref() { + Some(LogicalType::Date) => Some(CalendarKind::Date), + Some(LogicalType::Timestamp { + unit, + is_adjusted_to_u_t_c, + }) => is_adjusted_to_u_t_c.then(|| CalendarKind::Timestamp(*unit)), + Some(_) => None, + // Pre-logical-type files. Both of these converted types are UTC-normalised by definition, + // so there is no NTZ case to exclude here. + None => match column.converted_type() { + ConvertedType::DATE => Some(CalendarKind::Date), + ConvertedType::TIMESTAMP_MILLIS => { + Some(CalendarKind::Timestamp(ParquetTimeUnit::MILLIS)) + } + ConvertedType::TIMESTAMP_MICROS => { + Some(CalendarKind::Timestamp(ParquetTimeUnit::MICROS)) + } + _ => None, + }, + } +} + +/// Scale the micros threshold into `unit`. +/// +/// Scaling down to millis rounds toward negative infinity, which is the safe direction: it can +/// only make the threshold earlier, so a value in the truncated sub-millisecond window is treated +/// as affected rather than cleared. +fn timestamp_threshold(unit: ParquetTimeUnit) -> i64 { + match unit { + ParquetTimeUnit::MILLIS => LAST_SWITCH_JULIAN_MICROS.div_euclid(1_000), + ParquetTimeUnit::MICROS => LAST_SWITCH_JULIAN_MICROS, + ParquetTimeUnit::NANOS => LAST_SWITCH_JULIAN_MICROS.saturating_mul(1_000), + } +} + +/// Whether any row group's minimum for this column is below `threshold`. +/// +/// `undecidable_is_affected` is the answer for a row group whose statistics cannot settle the +/// question, either because the writer wrote none or because it omitted the minimum: `true` assumes +/// the worst, `false` requires positive proof. An all-null row group is cleared either way -- it has +/// no value to rebase -- and so is an empty one. +fn row_groups_hold_values_below( + metadata: &ParquetMetaData, + leaf_index: usize, + threshold: i64, + undecidable_is_affected: bool, +) -> bool { + metadata.row_groups().iter().any(|row_group| { + // An empty row group decodes nothing. + if row_group.num_rows() == 0 { + return false; + } + let Some(statistics) = row_group.column(leaf_index).statistics() else { + return undecidable_is_affected; + }; + match statistics_min(statistics) { + Some(min) => min < threshold, + // No minimum. Either the column is entirely null in this row group, in which case + // there is nothing to rebase, or the writer omitted the bound and we cannot tell. + None if statistics.null_count_opt() == Some(row_group.num_rows() as u64) => false, + None => undecidable_is_affected, + } + }) +} + +/// The minimum of a date/timestamp column's statistics, widened to `i64`. Only the two physical +/// types those logical types can use are handled; anything else is treated as no minimum. +fn statistics_min(statistics: &Statistics) -> Option { + match statistics { + Statistics::Int32(value) => value.min_opt().map(|min| *min as i64), + Statistics::Int64(value) => value.min_opt().copied(), + _ => None, + } +} + +fn data_type_has_date_or_timestamp(data_type: &DataType) -> bool { + match data_type { + DataType::Date32 | DataType::Date64 | DataType::Timestamp(_, _) => true, + DataType::List(field) + | DataType::LargeList(field) + | DataType::ListView(field) + | DataType::LargeListView(field) + | DataType::FixedSizeList(field, _) + | DataType::Map(field, _) + | DataType::RunEndEncoded(_, field) => data_type_has_date_or_timestamp(field.data_type()), + DataType::Struct(fields) => fields + .iter() + .any(|field| data_type_has_date_or_timestamp(field.data_type())), + DataType::Union(fields, _) => fields + .iter() + .any(|(_, field)| data_type_has_date_or_timestamp(field.data_type())), + DataType::Dictionary(_, value_type) => data_type_has_date_or_timestamp(value_type), + _ => false, + } +} + +/// The failure raised for a file that would need rebasing. +/// +/// Boxed inside `ParquetError::External` so the JNI error layer can recover it by downcast rather +/// than by matching on message text: `get_metadata` can only return a `ParquetError`, and a bare +/// `ParquetError::General` would be classified as a corrupt-file read (`FAILED_READ_FILE`), which +/// this is not. `file_path` is left empty for the JVM side to fill from the per-task file list, +/// which knows the file's real URI -- the object-store location available here has had its scheme +/// and leading slash normalised away. +pub fn legacy_calendar_error() -> ParquetError { + ParquetError::External(Box::new(SparkError::LegacyDatetimeRebase { + file_path: String::new(), + message: LEGACY_CALENDAR_MESSAGE.to_string(), + })) +} + +/// The user-facing explanation, defined once here and carried to the JVM in the error payload so +/// the shim that builds the exception does not restate it. +const LEGACY_CALENDAR_MESSAGE: &str = + "Comet cannot read this Parquet file: it holds dates or timestamps written in the legacy \ + hybrid (Julian + Gregorian) calendar, which Comet's native scan does not rebase to the \ + Proleptic Gregorian calendar. Reading it would return values shifted by up to ten days. \ + Set spark.comet.exceptionOnDatetimeRebase=false to read these values as-is without \ + rebasing, or disable Comet for this query so that Spark rebases them."; + +#[cfg(test)] +mod tests { + use super::*; + use arrow::datatypes::{Field, TimeUnit}; + use std::sync::Arc; + + /// Spark's defaults: both read modes EXCEPTION, so neither clears a version-less file. + const DEFAULT_MODES: ReadModes = ReadModes { + datetime_corrected: false, + int96_corrected: false, + }; + + const DATE: CalendarKind = CalendarKind::Date; + const INT96: CalendarKind = CalendarKind::Int96; + + fn kv(pairs: &[(&str, Option<&str>)]) -> Vec { + pairs + .iter() + .map(|(key, value)| KeyValue { + key: key.to_string(), + value: value.map(|v| v.to_string()), + }) + .collect() + } + + /// The calendar the footer establishes for `kind`, under Spark's default read modes. + fn calendar_of(kind: CalendarKind, pairs: &[(&str, Option<&str>)]) -> Calendar { + let kv = kv(pairs); + Provenance::from_footer(Some(&kv)).calendar(kind, &DEFAULT_MODES) + } + + /// Whether the footer proves `kind` needs no rebasing, under Spark's default read modes. + fn cleared(kind: CalendarKind, pairs: &[(&str, Option<&str>)]) -> bool { + calendar_of(kind, pairs) == Calendar::Gregorian + } + + #[test] + fn a_modern_corrected_writer_is_cleared() { + // The overwhelmingly common case: Spark 3.0+ with the default CORRECTED write mode stamps + // its version and no legacy marker, which positively proves the values are Gregorian. + for version in ["3.0.0", "3.5.9", "4.1.3"] { + assert!( + cleared(DATE, &[(SPARK_VERSION_METADATA_KEY, Some(version))]), + "date should be cleared for writer {version}" + ); + } + for version in ["3.1.0", "3.5.9", "4.1.3"] { + assert!( + cleared(INT96, &[(SPARK_VERSION_METADATA_KEY, Some(version))]), + "INT96 should be cleared for writer {version}" + ); + } + } + + #[test] + fn a_pre_switch_writer_is_provably_legacy() { + assert_eq!( + calendar_of(DATE, &[(SPARK_VERSION_METADATA_KEY, Some("2.4.6"))]), + Calendar::Legacy + ); + // INT96 switched a release later than the other encodings, so Spark 3.0 wrote hybrid INT96 + // while its DATE values were already Gregorian. + assert_eq!( + calendar_of(INT96, &[(SPARK_VERSION_METADATA_KEY, Some("3.0.3"))]), + Calendar::Legacy + ); + assert!(cleared( + DATE, + &[(SPARK_VERSION_METADATA_KEY, Some("3.0.3"))] + )); + } + + #[test] + fn an_explicit_legacy_marker_is_provably_legacy() { + // Written by a modern Spark with rebaseModeInWrite=LEGACY. The value Spark stamps is the + // empty string; Spark tests for key presence, not for a value. + for value in [Some(""), None] { + assert_eq!( + calendar_of( + DATE, + &[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_DATETIME_METADATA_KEY, value), + ] + ), + Calendar::Legacy + ); + assert_eq!( + calendar_of( + INT96, + &[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_INT96_METADATA_KEY, value), + ] + ), + Calendar::Legacy + ); + } + // The marker outranks a missing writer version too, rather than falling through to the + // read mode as an unmarked version-less file does. + assert_eq!( + calendar_of(DATE, &[(SPARK_LEGACY_DATETIME_METADATA_KEY, Some(""))]), + Calendar::Legacy + ); + } + + #[test] + fn the_two_legacy_markers_are_tracked_separately() { + // A modern writer that opted INT96 back into LEGACY says nothing about its DATE columns. + let footer = [ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_INT96_METADATA_KEY, Some("")), + ]; + assert!(cleared(DATE, &footer)); + assert_eq!(calendar_of(INT96, &footer), Calendar::Legacy); + } + + #[test] + fn a_file_with_no_writer_version_is_unknown_by_default() { + // Spark 2.4.5 and earlier stamped no version, and neither do Hive, Impala or plain + // parquet-mr. Nothing here proves either calendar, so the read is refused only where + // statistics positively expose an affected value -- never merely because a bound is + // missing, which would make every Hive timestamp column unreadable. + for footer in [ + &[][..], + &[("parquet-mr version", Some("1.13.1"))][..], + // A version key with no value is treated as no version at all. + &[(SPARK_VERSION_METADATA_KEY, None)][..], + ] { + assert_eq!(calendar_of(DATE, footer), Calendar::Unknown); + assert_eq!(calendar_of(INT96, footer), Calendar::Unknown); + } + assert_eq!( + Provenance::from_footer(None).calendar(DATE, &DEFAULT_MODES), + Calendar::Unknown + ); + } + + #[test] + fn corrected_read_mode_clears_a_version_less_file() { + // The user has asserted the values are already Gregorian, which is exactly what + // datetimeRebaseModeInRead=CORRECTED means to Spark. Comet honors it per encoding. + let modes = ReadModes { + datetime_corrected: true, + int96_corrected: false, + }; + let none = Provenance::from_footer(None); + assert_eq!(none.calendar(DATE, &modes), Calendar::Gregorian); + assert_eq!(none.calendar(INT96, &modes), Calendar::Unknown); + + let modes = ReadModes { + datetime_corrected: false, + int96_corrected: true, + }; + assert_eq!(none.calendar(DATE, &modes), Calendar::Unknown); + assert_eq!(none.calendar(INT96, &modes), Calendar::Gregorian); + } + + #[test] + fn corrected_read_mode_does_not_override_an_explicit_legacy_marker() { + // The footer is authoritative when it records provenance; the read mode only fills the gap + // when it does not. This mirrors Spark, which ignores the config for stamped files. + let modes = ReadModes { + datetime_corrected: true, + int96_corrected: true, + }; + let kv = kv(&[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_DATETIME_METADATA_KEY, Some("")), + ]); + assert_eq!( + Provenance::from_footer(Some(&kv)).calendar(DATE, &modes), + Calendar::Legacy + ); + } + + #[test] + fn int96_statistics_can_never_clear_a_column() { + assert_eq!(CalendarKind::Int96.threshold(), None); + } + + #[test] + fn date_threshold_is_the_switch_day() { + // 1582-10-15, the first Gregorian day. + assert_eq!(CalendarKind::Date.threshold(), Some(-141_427)); + } + + #[test] + fn millis_threshold_rounds_toward_the_unsafe_direction() { + // Truncating a negative micros threshold must not move it later in time, which would + // clear values that actually need rebasing. + let scaled = timestamp_threshold(ParquetTimeUnit::MILLIS); + assert!(scaled * 1_000 <= LAST_SWITCH_JULIAN_MICROS); + } + + #[test] + fn micros_and_nanos_thresholds_scale_exactly() { + assert_eq!( + timestamp_threshold(ParquetTimeUnit::MICROS), + LAST_SWITCH_JULIAN_MICROS + ); + assert_eq!( + timestamp_threshold(ParquetTimeUnit::NANOS), + LAST_SWITCH_JULIAN_MICROS * 1_000 + ); + } + + #[test] + fn statistics_min_reads_the_two_physical_types_dates_and_timestamps_use() { + let int32 = Statistics::int32(Some(-141_428), Some(0), None, Some(0), false); + assert_eq!(statistics_min(&int32), Some(-141_428)); + let int64 = Statistics::int64(Some(-2_208_988_800_000_001), Some(0), None, Some(0), false); + assert_eq!(statistics_min(&int64), Some(-2_208_988_800_000_001)); + let float = Statistics::float(Some(1.0), Some(2.0), None, Some(0), false); + assert_eq!(statistics_min(&float), None); + } + + fn date_field() -> Field { + Field::new("d", DataType::Date32, true) + } + + fn string_field() -> Field { + Field::new("s", DataType::Utf8, true) + } + + fn guard_for(schema: &Schema, enabled: bool) -> Option { + LegacyCalendarGuard::for_scan(enabled, schema, true, DEFAULT_MODES) + } + + #[test] + fn guard_is_disarmed_when_the_config_is_off() { + let schema = Schema::new(vec![date_field()]); + assert!(guard_for(&schema, false).is_none()); + } + + #[test] + fn guard_is_disarmed_when_no_calendar_sensitive_column_is_read() { + let schema = Schema::new(vec![ + string_field(), + Field::new("i", DataType::Int64, true), + Field::new( + "nested", + DataType::Struct( + vec![ + Field::new("b", DataType::Binary, true), + Field::new( + "l", + DataType::List(Arc::new(Field::new("e", DataType::Float64, true))), + true, + ), + ] + .into(), + ), + true, + ), + ]); + assert!(guard_for(&schema, true).is_none()); + } + + #[test] + fn guard_arms_on_top_level_and_nested_date_or_timestamp_columns() { + let ts = Field::new( + "t", + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + true, + ); + for field in [ + date_field(), + ts.clone(), + Field::new("l", DataType::List(Arc::new(date_field())), true), + Field::new( + "st", + DataType::Struct(vec![string_field(), date_field()].into()), + true, + ), + Field::new( + "m", + DataType::Map( + Arc::new(Field::new( + "entries", + DataType::Struct(vec![string_field(), ts].into()), + false, + )), + false, + ), + true, + ), + Field::new( + "dict", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Date32)), + true, + ), + ] { + let name = field.name().clone(); + let schema = Schema::new(vec![field]); + let guard = guard_for(&schema, true); + assert!(guard.is_some(), "expected {name} to arm the guard"); + assert_eq!(guard.unwrap().requested_roots, vec![name]); + } + } + + #[test] + fn guard_only_tracks_the_calendar_sensitive_roots() { + let schema = Schema::new(vec![ + string_field(), + date_field(), + Field::new("i", DataType::Int64, true), + ]); + let guard = guard_for(&schema, true).unwrap(); + assert_eq!(guard.requested_roots, vec!["d".to_string()]); + } + + /// A DATE leaf named `name`, as a Parquet schema descriptor would report it. + fn date_leaf(name: &str) -> ColumnDescriptor { + use parquet::basic::Repetition; + use parquet::schema::types::{ColumnPath, Type}; + let primitive = Type::primitive_type_builder(name, PhysicalType::INT32) + .with_logical_type(Some(LogicalType::Date)) + .with_repetition(Repetition::OPTIONAL) + .build() + .unwrap(); + ColumnDescriptor::new( + Arc::new(primitive), + 1, + 0, + ColumnPath::new(vec![name.to_string()]), + ) + } + + #[test] + fn case_sensitive_scans_require_an_exact_root_match() { + let schema = Schema::new(vec![Field::new("MyDate", DataType::Date32, true)]); + let guard = LegacyCalendarGuard::for_scan(true, &schema, true, DEFAULT_MODES).unwrap(); + assert!(guard.reads(&date_leaf("MyDate"))); + assert!(!guard.reads(&date_leaf("mydate"))); + } + + #[test] + fn case_insensitive_scans_match_a_root_in_any_case() { + let schema = Schema::new(vec![Field::new("MyDate", DataType::Date32, true)]); + let guard = LegacyCalendarGuard::for_scan(true, &schema, false, DEFAULT_MODES).unwrap(); + assert!(guard.reads(&date_leaf("MyDate"))); + assert!(guard.reads(&date_leaf("mydate"))); + assert!(guard.reads(&date_leaf("MYDATE"))); + assert!(!guard.reads(&date_leaf("other"))); + } + + /// Footer of a file no Spark wrote: parquet-mr stamps its own version and nothing else. Hive, + /// Impala and Trino all look like this. + const NON_SPARK_FOOTER: &[(&str, Option<&str>)] = &[("parquet-mr version", Some("1.10.1"))]; + + /// A DATE primitive, as the schema descriptor of a real file would hold it. + fn date_primitive() -> parquet::schema::types::Type { + use parquet::basic::Repetition; + use parquet::schema::types::Type; + Type::primitive_type_builder("d", PhysicalType::INT32) + .with_logical_type(Some(LogicalType::Date)) + .with_repetition(Repetition::OPTIONAL) + .build() + .unwrap() + } + + /// An INT96 timestamp primitive: what Hive writes for a TIMESTAMP column. + fn int96_primitive() -> parquet::schema::types::Type { + use parquet::basic::Repetition; + use parquet::schema::types::Type; + Type::primitive_type_builder("t", PhysicalType::INT96) + .with_repetition(Repetition::OPTIONAL) + .build() + .unwrap() + } + + /// An INT64 TIMESTAMP primitive. `is_adjusted_to_u_t_c` false is Spark's TIMESTAMP_NTZ. + fn timestamp_primitive(is_adjusted_to_u_t_c: bool) -> parquet::schema::types::Type { + use parquet::basic::Repetition; + use parquet::schema::types::Type; + Type::primitive_type_builder("t", PhysicalType::INT64) + .with_logical_type(Some(LogicalType::Timestamp { + unit: ParquetTimeUnit::MICROS, + is_adjusted_to_u_t_c, + })) + .with_repetition(Repetition::OPTIONAL) + .build() + .unwrap() + } + + /// Metadata for a single-column file with one row group -- everything the guard consults. + fn one_column_metadata( + leaf: parquet::schema::types::Type, + num_rows: i64, + statistics: Option, + footer: &[(&str, Option<&str>)], + ) -> ParquetMetaData { + use parquet::file::metadata::{ + ColumnChunkMetaData, FileMetaData, ParquetMetaDataBuilder, RowGroupMetaData, + }; + use parquet::schema::types::{SchemaDescPtr, SchemaDescriptor, Type}; + + let root = Type::group_type_builder("spark_schema") + .with_fields(vec![Arc::new(leaf)]) + .build() + .unwrap(); + let descr: SchemaDescPtr = Arc::new(SchemaDescriptor::new(Arc::new(root))); + let mut column = ColumnChunkMetaData::builder(descr.column(0)); + if let Some(statistics) = statistics { + column = column.set_statistics(statistics); + } + let row_group = RowGroupMetaData::builder(Arc::clone(&descr)) + .set_num_rows(num_rows) + .set_column_metadata(vec![column.build().unwrap()]) + .build() + .unwrap(); + let file_metadata = FileMetaData::new( + 1, + num_rows, + None, + Some(kv(footer)), + Arc::clone(&descr), + None, + ); + ParquetMetaDataBuilder::new(file_metadata) + .set_row_groups(vec![row_group]) + .build() + } + + /// A guard for a scan that reads the single named column, at Spark's default read modes. + fn guard_reading(name: &str, data_type: DataType) -> LegacyCalendarGuard { + let schema = Schema::new(vec![Field::new(name, data_type, true)]); + LegacyCalendarGuard::for_scan(true, &schema, true, DEFAULT_MODES).unwrap() + } + + fn timestamp_guard() -> LegacyCalendarGuard { + guard_reading("t", DataType::Timestamp(TimeUnit::Microsecond, None)) + } + + fn date_guard() -> LegacyCalendarGuard { + guard_reading("d", DataType::Date32) + } + + /// Statistics whose minimum is `min`, with no nulls. + fn date_statistics(min: i32) -> Statistics { + Statistics::int32(Some(min), Some(0), None, Some(0), false) + } + + #[test] + fn a_hive_written_int96_column_is_read() { + // The regression this guard must not cause: Hive writes TIMESTAMP as INT96 and stamps no + // Spark version, and INT96 statistics can never clear a column. Refusing on that + // combination would make every Hive-written timestamp column in existence unreadable, + // whatever its values, where Spark reads them without complaint. + let metadata = one_column_metadata(int96_primitive(), 10, None, NON_SPARK_FOOTER); + assert!(timestamp_guard().check(&metadata).is_ok()); + } + + #[test] + fn an_int96_column_a_footer_marks_legacy_is_refused() { + // Here the file itself says the values are hybrid-calendar, and nothing can narrow that to + // the affected rows, so the read is refused. + for footer in [ + &[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_INT96_METADATA_KEY, Some("")), + ][..], + // Spark 3.0 predates the INT96 switch. + &[(SPARK_VERSION_METADATA_KEY, Some("3.0.3"))][..], + ] { + let metadata = one_column_metadata(int96_primitive(), 10, None, footer); + assert!(timestamp_guard().check(&metadata).is_err()); + } + } + + #[test] + fn a_modern_spark_int96_column_is_read() { + let metadata = one_column_metadata( + int96_primitive(), + 10, + None, + &[(SPARK_VERSION_METADATA_KEY, Some("3.5.9"))], + ); + assert!(timestamp_guard().check(&metadata).is_ok()); + } + + #[test] + fn an_unknown_provenance_column_is_refused_only_on_proof() { + // Statistics expose an ancient value: Spark under its EXCEPTION default raises here too. + let ancient = one_column_metadata( + date_primitive(), + 10, + Some(date_statistics(LAST_SWITCH_JULIAN_DAY - 1)), + NON_SPARK_FOOTER, + ); + assert!(date_guard().check(&ancient).is_err()); + + // Statistics prove the values are all at or after the switch day. + let modern = one_column_metadata( + date_primitive(), + 10, + Some(date_statistics(LAST_SWITCH_JULIAN_DAY)), + NON_SPARK_FOOTER, + ); + assert!(date_guard().check(&modern).is_ok()); + + // No statistics at all proves nothing either way, so the read goes ahead. Writers with + // statistics disabled are common, and refusing them all is not worth the narrow class of + // genuinely ancient values it would catch. + let unknown = one_column_metadata(date_primitive(), 10, None, NON_SPARK_FOOTER); + assert!(date_guard().check(&unknown).is_ok()); + } + + #[test] + fn a_legacy_marked_column_is_refused_unless_statistics_clear_it() { + let legacy_footer = &[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_DATETIME_METADATA_KEY, Some("")), + ]; + + // The case that makes a default-on guard tolerable: Spark stamps the marker on the whole + // file whenever the write mode was LEGACY, and dates from 1582-10-15 on rebase to + // themselves, so a legacy-marked file of modern dates still reads. + let modern = one_column_metadata( + date_primitive(), + 10, + Some(date_statistics(LAST_SWITCH_JULIAN_DAY)), + legacy_footer, + ); + assert!(date_guard().check(&modern).is_ok()); + + let ancient = one_column_metadata( + date_primitive(), + 10, + Some(date_statistics(LAST_SWITCH_JULIAN_DAY - 1)), + legacy_footer, + ); + assert!(date_guard().check(&ancient).is_err()); + + // Unlike the unknown-provenance case, a missing bound here is assumed to be affected. + let no_statistics = one_column_metadata(date_primitive(), 10, None, legacy_footer); + assert!(date_guard().check(&no_statistics).is_err()); + } + + #[test] + fn an_all_null_legacy_marked_row_group_is_read() { + // No minimum, but the null count accounts for every row, so there is no value to rebase. + let metadata = one_column_metadata( + date_primitive(), + 10, + Some(Statistics::int32(None, None, None, Some(10), false)), + &[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_DATETIME_METADATA_KEY, Some("")), + ], + ); + assert!(date_guard().check(&metadata).is_ok()); + } + + #[test] + fn a_calendar_sensitive_column_the_scan_does_not_read_is_ignored() { + // A legacy-marked INT96 column, which would otherwise be refused outright, in a file the + // scan only reads a string column out of. + let metadata = one_column_metadata( + int96_primitive(), + 10, + None, + &[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_INT96_METADATA_KEY, Some("")), + ], + ); + // The guard would not even arm for a string-only scan, so arm it on an unrelated date + // column to prove the per-column check, not just `for_scan`, does the filtering. + assert!(date_guard().check(&metadata).is_ok()); + } + + #[test] + fn a_timestamp_ntz_column_is_never_calendar_sensitive() { + // Spark stamps the legacy marker from the write-mode conf alone, without regard to whether + // the schema holds a column the mode could apply to, and it never rebases NTZ in either + // direction. So a legacy-marked file of ancient NTZ values reads back exactly as written, + // and refusing it would fail a read Spark answers correctly. + let ancient = Statistics::int64( + Some(LAST_SWITCH_JULIAN_MICROS - 1), + Some(0), + None, + Some(0), + false, + ); + let legacy_footer = &[ + (SPARK_VERSION_METADATA_KEY, Some("4.1.3")), + (SPARK_LEGACY_DATETIME_METADATA_KEY, Some("")), + ]; + + let ntz = one_column_metadata( + timestamp_primitive(false), + 10, + Some(ancient.clone()), + legacy_footer, + ); + assert!(timestamp_guard().check(&ntz).is_ok()); + assert!(calendar_kind(&ntz.file_metadata().schema_descr().column(0)).is_none()); + + // The UTC-adjusted counterpart, which Spark does rebase, is still refused. The two differ + // only in `isAdjustedToUTC`, so this is what proves the exclusion is not too broad. + let ltz = one_column_metadata(timestamp_primitive(true), 10, Some(ancient), legacy_footer); + assert!(timestamp_guard().check(<z).is_err()); + } +} diff --git a/native/core/src/parquet/mod.rs b/native/core/src/parquet/mod.rs index cfa03220c10..0617d604d80 100644 --- a/native/core/src/parquet/mod.rs +++ b/native/core/src/parquet/mod.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -pub mod eager_page_index_reader_factory; +pub mod comet_parquet_reader_factory; pub mod encryption_support; pub mod parquet_exec; @@ -24,6 +24,7 @@ pub mod schema_adapter; pub mod util; mod cast_column; +pub mod legacy_datetime; pub(crate) mod objectstore; use std::collections::HashMap; @@ -228,6 +229,10 @@ pub unsafe extern "system" fn Java_org_apache_comet_parquet_Native_initRecordBat // so the native side does not need to do field-ID matching here. false, false, + // The legacy-calendar guard is not plumbed through this JNI entry point, which is + // driven by the external Iceberg integration rather than by CometNativeScan's proto. + // See the note in compatibility/scans.md. + None, )?; let partition_index: usize = 0; diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 1308ce97fca..f15fdacbfbe 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -16,8 +16,9 @@ // under the License. use crate::execution::operators::ExecutionError; -use crate::parquet::eager_page_index_reader_factory::EagerPageIndexReaderFactory; +use crate::parquet::comet_parquet_reader_factory::CometParquetFileReaderFactory; use crate::parquet::encryption_support::{CometEncryptionConfig, ENCRYPTION_FACTORY_ID}; +use crate::parquet::legacy_datetime::LegacyCalendarGuard; use crate::parquet::parquet_support::SparkParquetOptions; use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; use arrow::datatypes::{Field, SchemaRef}; @@ -56,6 +57,9 @@ use std::sync::Arc; /// /// `data_filters`: Any predicate that must be applied to the data returned by the scan. If /// specified, then `data_schema` must also be specified. +/// +/// `legacy_calendar_guard`: `Some` only when `spark.comet.exceptionOnDatetimeRebase` is set and +/// the scan reads a calendar-sensitive column; see `LegacyCalendarGuard::for_scan`. #[allow(clippy::too_many_arguments)] pub(crate) fn init_datasource_exec( required_schema: SchemaRef, @@ -75,6 +79,7 @@ pub(crate) fn init_datasource_exec( encryption_enabled: bool, use_field_id: bool, ignore_missing_field_id: bool, + legacy_calendar_guard: Option, ) -> Result, ExecutionError> { // Computed once and reused below for `try_pushdown_filters`. `copied_config()` clones only // `SessionConfig` (an `Arc` plus a small extensions map); `SessionContext:: @@ -154,10 +159,15 @@ pub(crate) fn init_datasource_exec( // a predicate that never resolves to fully-matched by row-group statistics alone (e.g. `IS // NOT NULL` on a column whose row groups don't carry `null_count` stats, as with TPC-DS // `store_sales`), the page index is re-fetched, uncached, on every open (comet#3978). - // `EagerPageIndexReaderFactory` forces the page index to load on the first fetch and be + // `CometParquetFileReaderFactory` forces the page index to load on the first fetch and be // cached with the footer, at the cost of losing the skip's benefit when it would have // applied. Filed upstream as apache/datafusion#23978; revert this once that's fixed. // + // The same factory is the enforcement point for `spark.comet.exceptionOnDatetimeRebase`: it + // sees each file's footer and row-group statistics, which together are what distinguish a + // legacy-written file that actually holds values needing rebasing from one whose values all + // rebase to themselves. A `None` guard disarms the check entirely. + // // TODO: metadata I/O is invisible in metrics. `fetch_metadata` reads via `ObjectStore::get_ranges`, // bypassing the `get_bytes` path where `bytes_scanned` is counted. A byte-counting ObjectStore // wrapper would surface it. @@ -165,7 +175,7 @@ pub(crate) fn init_datasource_exec( let store = runtime_env.object_store(&object_store_url)?; let metadata_cache = runtime_env.cache_manager.get_file_metadata_cache(); parquet_source = parquet_source.with_parquet_file_reader_factory(Arc::new( - EagerPageIndexReaderFactory::new(store, metadata_cache), + CometParquetFileReaderFactory::new(store, metadata_cache, legacy_calendar_guard), )); // Route data filters through `try_pushdown_filters` rather than calling @@ -254,7 +264,7 @@ fn get_options( session_parquet_options.max_predicate_cache_size; // Only gates whether the page index is used for row-group/page pruning // (datafusion's `enable_page_index` check around `page_pruning_predicate`). It does not - // stop the index from being fetched: `EagerPageIndexReaderFactory` always forces + // stop the index from being fetched: `CometParquetFileReaderFactory` always forces // `PageIndexPolicy::Optional` on unencrypted files regardless of the requested policy, so // disabling this reduces pruning, not read I/O. table_parquet_options.global.enable_page_index = session_parquet_options.enable_page_index; @@ -352,7 +362,7 @@ mod tests { // pruning shows it is still needed (apache/datafusion#22857). That on-demand load bypasses // `FileMetadataCache` entirely, so on a predicate that never resolves to fully-matched (e.g. // non-null `IS NOT NULL` join keys without row-group `null_count` stats), the page index is - // re-fetched, uncached, on every open. `EagerPageIndexReaderFactory` ignores the requested + // re-fetched, uncached, on every open. `CometParquetFileReaderFactory` ignores the requested // policy and always loads the page index into the cache on the first fetch, so it asserts // present here even though this scan has no pruning predicate to request it. #[tokio::test] @@ -403,6 +413,7 @@ mod tests { false, false, false, + None, ) .unwrap(); diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 2ee1230ed87..4191d876573 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -74,8 +74,6 @@ pub struct SparkParquetOptions { pub allow_incompat: bool, /// Support casting unsigned ints to signed ints (used by Parquet SchemaAdapter) pub allow_cast_unsigned_ints: bool, - /// Whether to read dates/timestamps that were written in the legacy hybrid Julian + Gregorian calendar as it is. If false, throw exceptions instead. If the spark type is TimestampNTZ, this should be true. - pub use_legacy_date_timestamp_or_ntz: bool, // Whether schema field names are case sensitive pub case_sensitive: bool, /// SPARK-53535 (Spark 4.1+): when reading a struct whose requested fields are all @@ -108,7 +106,6 @@ impl SparkParquetOptions { timezone: timezone.to_string(), allow_incompat, allow_cast_unsigned_ints: false, - use_legacy_date_timestamp_or_ntz: false, case_sensitive: false, return_null_struct_if_all_fields_missing: true, use_field_id: false, @@ -124,7 +121,6 @@ impl SparkParquetOptions { timezone: "".to_string(), allow_incompat, allow_cast_unsigned_ints: false, - use_legacy_date_timestamp_or_ntz: false, case_sensitive: false, return_null_struct_if_all_fields_missing: true, use_field_id: false, diff --git a/native/jni-bridge/src/errors.rs b/native/jni-bridge/src/errors.rs index d7d4005d7f7..66e6cf4466e 100644 --- a/native/jni-bridge/src/errors.rs +++ b/native/jni-bridge/src/errors.rs @@ -627,6 +627,14 @@ fn try_classify_file_read_error(error: &DataFusionError) -> Option { // error-message text, which DataFusion produces via `{:?}`) tells the two apart. Bail so the // underlying error surfaces through the normal native-exception path. DFE::ParquetError(pe) if parquet_external_wraps_arrow_error(pe) => None, + // A `SparkError` Comet's Parquet reader raised deliberately (the legacy-calendar + // rejection) travels as `ParquetError::External`, because `AsyncFileReader::get_metadata` + // can only return a `ParquetError`. Recover it by downcast and surface it as itself: it is + // a Comet limitation, not a corrupt/truncated file, and must not be relabelled + // FAILED_READ_FILE. + DFE::ParquetError(pe) if parquet_external_wraps_spark_error(pe) => { + spark_error_in_parquet_error(pe) + } // A genuinely-missing file (object_store NotFound) is distinct from a corrupt/truncated // one: Spark surfaces it as `readCurrentFileNotFoundError` ("It is possible the underlying // files have been updated."), not `cannotReadFilesError`. The NotFound may arrive directly @@ -678,6 +686,24 @@ fn try_classify_file_read_error(error: &DataFusionError) -> Option { } } +/// True if `pe` is a `ParquetError::External` wrapping a `SparkError` -- i.e. a failure Comet's +/// Parquet reader raised deliberately, such as the legacy-calendar rejection. Companion to +/// [`spark_error_in_parquet_error`], which extracts it; kept separate so the match guard does not +/// have to clone. +fn parquet_external_wraps_spark_error(pe: &ParquetError) -> bool { + matches!(pe, ParquetError::External(inner) if inner.downcast_ref::().is_some()) +} + +/// Recover a `SparkError` that Comet boxed into `ParquetError::External`. Typed rather than +/// message-based, so it cannot be confused with a genuine parquet failure whose prose happens to +/// look similar. +fn spark_error_in_parquet_error(pe: &ParquetError) -> Option { + match pe { + ParquetError::External(inner) => inner.downcast_ref::().cloned(), + _ => None, + } +} + /// True if `pe` is a `ParquetError::External` wrapping an `ArrowError`. DataFusion's parquet row /// filter returns a pushed-down predicate's evaluation failure as an `ArrowError` (e.g. /// `ComputeError` for an ANSI divide-by-zero), which the parquet reader then surfaces as diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index d9515edc991..3414b71ecf4 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -133,6 +133,24 @@ message NativeScanCommon { // SchemaColumnConvertNotSupportedException (Spark 3.x, SPARK-36182). Set // from Comet's per-Spark-version constant in ShimCometConf. bool allow_timestamp_ltz_to_ntz = 18; + // spark.comet.exceptionOnDatetimeRebase. When true (the default), the native + // scan fails rather than silently returning unrebased values for a Parquet + // file that both carries a legacy hybrid (Julian + Gregorian) calendar marker + // in its footer and actually holds a value old enough to be rebased. Comet + // does not implement rebasing. + bool exception_on_legacy_datetime = 19; + // Field numbers 20 and 21 previously carried Spark's rebase switch points. They + // are hardcoded natively now (reading them forced RebaseDateTime's static + // initializer, which parses ~590 KB of JSON on the driver) and asserted against + // Spark's values by ParquetDatetimeRebaseSuite. Do not reuse the numbers. + reserved 20, 21; + // spark.sql.parquet.datetimeRebaseModeInRead == CORRECTED. Spark applies the read mode only to + // files whose footer does not record a writer version (Spark 2.4.5 and earlier, and non-Spark + // writers); CORRECTED asserts their values are already Proleptic Gregorian. + bool legacy_datetime_read_mode_corrected = 22; + // spark.sql.parquet.int96RebaseModeInRead == CORRECTED. Tracked separately because Spark gives + // INT96 its own read mode and its own footer marker. + bool legacy_int96_read_mode_corrected = 23; } message NativeScan { diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index cdbc4308296..a3085c2f8fe 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -717,15 +717,22 @@ object CometConf extends ShimCometConf { val COMET_EXCEPTION_ON_LEGACY_DATE_TIMESTAMP: ConfigEntry[Boolean] = conf("spark.comet.exceptionOnDatetimeRebase") - .category(CATEGORY_EXEC) - .doc("Whether to throw exception when seeing dates/timestamps from the legacy hybrid " + - "(Julian + Gregorian) calendar. Since Spark 3, dates/timestamps were written according " + - "to the Proleptic Gregorian calendar. When this is true, Comet will " + - "throw exceptions when seeing these dates/timestamps that were written by Spark version " + - "before 3.0. If this is false, these dates/timestamps will be read as if they were " + - "written to the Proleptic Gregorian calendar and will not be rebased.") + .category(CATEGORY_SCAN) + .doc("Whether to fail rather than return incorrect values for dates/timestamps from the " + + "legacy hybrid (Julian + Gregorian) calendar. Since Spark 3, dates/timestamps were " + + "written according to the Proleptic Gregorian calendar, and Spark rebases older values " + + "on read. Comet's native scan does not implement rebasing, so when this is true (the " + + "default) it fails any scan that would read an affected value: one from a Parquet file " + + "that both carries a legacy-calendar marker in its footer (written by Spark before 3.0, " + + s"or with ${SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key}=LEGACY) and actually holds a date " + + "before 1582-10-15 or a timestamp before 1900-01-01T00:00:00Z. Legacy-written files " + + "whose values are all newer than that rebase to themselves and are read normally, as are " + + "files that record no writer version at all (every non-Spark writer) unless row-group " + + "statistics positively expose an affected value. Set this to false to read affected " + + "values as-is, without rebasing, which reproduces the silently-incorrect results of " + + s"earlier Comet versions. $COMPAT_GUIDE.") .booleanConf - .createWithDefault(false) + .createWithDefault(true) val COMET_ENABLE_PARTIAL_HASH_AGGREGATE: ConfigEntry[Boolean] = conf("spark.comet.testing.aggregate.partialMode.enabled") diff --git a/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala b/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala index a6bc21aca71..4fb0c7be3b2 100644 --- a/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala +++ b/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala @@ -86,11 +86,14 @@ object SparkErrorConverter extends ShimSparkErrorConverter { val json = parse(e.getMessage) val errorJson = json.extract[ErrorJson] val rawParams = errorJson.params.getOrElse(Map.empty) - // CannotReadFile carries the offending file path natively only for the object_store NotFound - // case; for corrupt/truncated parquet the native error has no path, so fall back to the - // per-task file list threaded in from CometExecIterator. + // Any error that declares a `filePath` but left it empty gets the per-task file list threaded + // in from CometExecIterator. The native side knows the path only in some cases (e.g. + // CannotReadFile for an object_store NotFound); for corrupt/truncated parquet, a + // legacy-calendar rejection, or a schema-convert mismatch it does not, and its object-store + // location would be scheme- and slash-normalised anyway. Keyed off the payload shape rather + // than a list of error-type names, so a new error carrying `filePath` needs no edit here. val params = - if (errorJson.errorType == "CannotReadFile" + if (rawParams.contains("filePath") && rawParams.get("filePath").forall(p => p == null || p.toString.isEmpty) && taskFilePaths.nonEmpty) { rawParams + ("filePath" -> taskFilePaths.mkString(",")) diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala index f017acfcd20..f4fa81fe216 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala @@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.{Expression, Literal} import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues import org.apache.spark.sql.comet.{CometNativeExec, CometNativeScanExec, CometScanExec} import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SubqueryAdaptiveBroadcastExec} -import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils +import org.apache.spark.sql.execution.datasources.parquet.{ParquetOptions, ParquetUtils} import org.apache.spark.sql.internal.SQLConf import org.apache.comet.{CometConf, ConfigEntry} @@ -91,6 +91,14 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging { !hasFallbackReason(scanExec) } + /** + * Whether a `*RebaseModeInRead` setting is CORRECTED. Takes the rendered value rather than the + * declared type, which is a plain String on Spark 3.x and a `LegacyBehaviorPolicy.Value` on + * 4.x. + */ + private def isCorrected(rebaseMode: String): Boolean = + rebaseMode.equalsIgnoreCase("CORRECTED") + /** Detects AQE DPP (SubqueryAdaptiveBroadcastExec), as opposed to non-AQE DPP. */ private def isAqeDynamicPruningFilter(e: Expression): Boolean = e.exists { @@ -213,6 +221,25 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging { commonBuilder.setAllowTypePromotion(CometConf.COMET_SCHEMA_EVOLUTION_ENABLED) commonBuilder.setAllowTimestampLtzToNtz(CometConf.COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ) + // Comet does not implement datetime rebasing, so the native scan fails on a file that + // would need it rather than returning silently-shifted values. The rebase switch points the + // native side compares row-group statistics against are hardcoded there and asserted against + // Spark's `RebaseDateTime` by ParquetDatetimeRebaseSuite -- reading them here would force + // that object's static initializer, which parses ~590 KB of bundled JSON on the driver. + commonBuilder.setExceptionOnLegacyDatetime( + CometConf.COMET_EXCEPTION_ON_LEGACY_DATE_TIMESTAMP.get(scan.conf)) + // Spark consults these read modes only for files whose footer records no writer version + // (Spark 2.4.5 and earlier, plus every non-Spark writer). CORRECTED asserts such a file's + // values are already Proleptic Gregorian, which lets Comet read them; the EXCEPTION default + // does not, and Spark raises on an ancient value there too. Read them through + // `ParquetOptions`, the same source Spark's own `ParquetFileFormat` uses, so a per-read + // `.option("datetimeRebaseMode", ...)` is honored and not just the session conf. + val parquetOptions = new ParquetOptions(scan.relation.options, scan.conf) + commonBuilder.setLegacyDatetimeReadModeCorrected( + isCorrected(parquetOptions.datetimeRebaseModeInRead.toString)) + commonBuilder.setLegacyInt96ReadModeCorrected( + isCorrected(parquetOptions.int96RebaseModeInRead.toString)) + // Collect S3/cloud storage configurations val hadoopConf = scan.relation.sparkSession.sessionState .newHadoopConfWithOptions(scan.relation.options) diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/shims/CometExceptions.scala b/spark/src/main/scala/org/apache/spark/sql/comet/shims/CometExceptions.scala new file mode 100644 index 00000000000..14732bbb1fc --- /dev/null +++ b/spark/src/main/scala/org/apache/spark/sql/comet/shims/CometExceptions.scala @@ -0,0 +1,58 @@ +/* + * 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.spark.sql.comet.shims + +import org.apache.spark.SparkUpgradeException +import org.apache.spark.sql.errors.QueryExecutionErrors +import org.apache.spark.sql.internal.SQLConf + +/** + * Builders for Comet exceptions that need Spark internals but no per-version shimming. + * + * This lives in `org.apache.spark.sql.comet.shims` rather than in `org.apache.comet` because the + * exception types it constructs are `private[spark]`; it is deliberately outside the per-version + * source roots, since every API it touches resolves identically on Spark 3.4 through 4.x. + */ +object CometExceptions { + + /** + * The failure Comet raises for a Parquet file whose dates/timestamps were written in the legacy + * hybrid (Julian + Gregorian) calendar, which Comet's native scan cannot rebase. + * + * Classified as Spark's own `SparkUpgradeException` with the `READ_ANCIENT_DATETIME` condition: + * that is the exception Spark raises for the same data, and one its `FileScanRDD` deliberately + * rethrows rather than wrapping in `FAILED_READ_FILE`, since the file is not corrupt. The + * templated message explains the calendar ambiguity accurately but advises setting Spark's + * rebase mode, which does not apply to Comet, so Comet's own remedy travels as the cause. + */ + def legacyDatetimeRebase(params: Map[String, Any]): SparkUpgradeException = { + val filePath = params.get("filePath").map(_.toString).filter(_.nonEmpty) + val cause = new RuntimeException( + params.get("message").map(_.toString).getOrElse("") + + filePath.map(p => s" File: $p").getOrElse("")) + new SparkUpgradeException( + "INCONSISTENT_BEHAVIOR_CROSS_VERSION.READ_ANCIENT_DATETIME", + Map( + "format" -> "Parquet", + "config" -> QueryExecutionErrors.toSQLConf(SQLConf.PARQUET_REBASE_MODE_IN_READ.key), + "option" -> QueryExecutionErrors.toDSOption("datetimeRebaseMode")), + cause) + } +} diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala index 09ac063cd22..75de28181d0 100644 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala @@ -298,6 +298,9 @@ trait ShimSparkErrorConverter { params("requiredId").toString.toInt, params("matchedFields").toString)) + case "LegacyDatetimeRebase" => + Some(CometExceptions.legacyDatetimeRebase(params)) + case "ParquetMissingFieldIds" => // Mirror Spark's `ParquetReadSupport.inferSchema`, which throws a plain // `RuntimeException` (not a SparkException) when the read schema requests field diff --git a/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala index c502e4d55d1..8aa1d601892 100644 --- a/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ b/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala @@ -293,6 +293,9 @@ trait ShimSparkErrorConverter { params("requiredId").toString.toInt, params("matchedFields").toString)) + case "LegacyDatetimeRebase" => + Some(CometExceptions.legacyDatetimeRebase(params)) + case "ParquetMissingFieldIds" => // Mirror Spark's `ParquetReadSupport.inferSchema`, which throws a plain // `RuntimeException` (not a SparkException) when the read schema requests field diff --git a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala index 874a6af97c0..2a4ceee4e20 100644 --- a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala @@ -313,6 +313,9 @@ trait ShimSparkErrorConverter { val filePath = params.get("filePath").map(_.toString).getOrElse("") Some(QueryExecutionErrors.cannotReadFilesError(dupCause, filePath)) + case "LegacyDatetimeRebase" => + Some(CometExceptions.legacyDatetimeRebase(params)) + case "ParquetMissingFieldIds" => // Mirror Spark's `ParquetReadSupport.inferSchema`. Same wrapping rationale as // `DuplicateFieldByFieldId`: wrap the RuntimeException in FAILED_READ_FILE.NO_HINT diff --git a/spark/src/test/scala/org/apache/comet/parquet/ParquetDatetimeRebaseSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/ParquetDatetimeRebaseSuite.scala new file mode 100644 index 00000000000..ab8caf9cf09 --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetDatetimeRebaseSuite.scala @@ -0,0 +1,409 @@ +/* + * 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.parquet + +import org.apache.spark.sql.CometTestBase +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.catalyst.util.RebaseDateTime +import org.apache.spark.sql.comet.CometNativeScanExec +import org.apache.spark.sql.internal.SQLConf + +import org.apache.comet.CometConf + +/** + * Tests for `spark.comet.exceptionOnDatetimeRebase`. + * + * Comet's native scan does not rebase dates/timestamps written in the legacy hybrid (Julian + + * Gregorian) calendar, so reading them unrebased returns values shifted by up to ten days. + * Implementing rebasing is tracked by https://github.com/apache/datafusion-comet/issues/5010; + * until then Comet fails such a read by default rather than returning wrong answers. + * + * The interesting cases here are the ones that must NOT fail: Spark stamps the legacy-calendar + * marker on a whole file whenever the write mode was LEGACY, but dates from 1582-10-15 onward + * rebase to themselves, so a marked file whose values are all modern is read normally. + * + * See https://github.com/apache/datafusion-comet/issues/5195 + */ +class ParquetDatetimeRebaseSuite extends CometTestBase { + + /** A date before the 1582-10-15 switch, so its Julian and Gregorian forms differ. */ + private val ancientDate = "1000-01-01" + + /** Dates from the switch onward rebase to themselves. */ + private val modernDate = "1990-01-01" + + /** What a legacy-written `ancientDate` reads back as when nobody rebases it. */ + private val ancientDateUnrebased = "1000-01-06" + + /** Writes the rows `select` produces to `path`, under the given SQL confs. */ + private def writeParquet(path: String, select: String, confs: (String, String)*): Unit = + withSQLConf(confs: _*) { + spark.sql(select).write.mode("overwrite").parquet(path) + } + + /** Writes `dates` (plus each date's text as a `label` column) in the given rebase mode. */ + private def writeDates(path: String, mode: String, dates: String*): Unit = { + val values = dates.map(d => s"('$d')").mkString(", ") + writeParquet( + path, + s"SELECT cast(s as date) AS d, s AS label FROM VALUES $values AS v(s)", + SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key -> mode) + } + + private def writeLegacyDates(path: String): Unit = + writeDates(path, "LEGACY", ancientDate, modernDate) + + /** `checkSparkAnswerAndOperator`, additionally requiring Comet's native Parquet scan. */ + private def checkNativeScanAnswer(df: => DataFrame): Unit = + checkSparkAnswerAndOperator(df, includeClasses = Seq(classOf[CometNativeScanExec])) + + /** + * Asserts a native scan without comparing to Spark, for the tests that deliberately expect + * Comet to differ from Spark. `stripAQEPlan` mirrors what `checkSparkAnswerAndOperator` does. + */ + private def assertNativeScan(df: DataFrame): Unit = { + val plan = stripAQEPlan(df.queryExecution.executedPlan) + assert( + plan.collectLeaves().exists(_.isInstanceOf[CometNativeScanExec]), + s"expected a CometNativeScanExec, got:\n$plan") + } + + /** A throwable and everything in its cause chain. */ + private def causeChain(t: Throwable): List[Throwable] = + Iterator.iterate(t)(_.getCause).takeWhile(_ != null).toList + + private def render(chain: List[Throwable]): String = + chain.map(e => s"${e.getClass.getName}: ${e.getMessage}").mkString("\n ") + + private def assertRaisesOnRebase(df: => DataFrame): Unit = { + val chain = causeChain(intercept[Throwable](df.collect())) + val messages = chain.map(_.getMessage).filter(_ != null) + assert( + messages.exists(_.contains("legacy hybrid (Julian + Gregorian) calendar")), + s"expected the legacy-calendar rejection in the cause chain, got:\n ${render(chain)}") + assert( + messages.exists(_.contains(CometConf.COMET_EXCEPTION_ON_LEGACY_DATE_TIMESTAMP.key)), + s"expected the message to name the config that raised, got:\n ${render(chain)}") + } + + test("the native rebase thresholds match Spark's own") { + // The native guard hardcodes these rather than receiving them from the JVM, because touching + // RebaseDateTime forces a static initializer that parses ~590 KB of bundled JSON and retains + // several MB -- a cost every driver would pay on its first native scan. This test is what keeps + // the copies honest: if Spark ever moves a switch point, it fails here rather than silently + // shifting the threshold. Keep in sync with LAST_SWITCH_JULIAN_DAY / LAST_SWITCH_JULIAN_MICROS + // in native/core/src/parquet/legacy_datetime.rs. + assert(RebaseDateTime.lastSwitchJulianDay == -141427) // 1582-10-15 + assert(RebaseDateTime.lastSwitchJulianTs == -2208988800000000L) // 1900-01-01T00:00:00Z + } + + test("ancient legacy-calendar dates raise by default") { + // No config set: the guard is on out of the box, which is the point of the default. Comet + // must not silently return shifted values. + assert(CometConf.COMET_EXCEPTION_ON_LEGACY_DATE_TIMESTAMP.defaultValue.contains(true)) + withTempPath { dir => + val path = dir.getCanonicalPath + writeLegacyDates(path) + assertRaisesOnRebase(spark.read.parquet(path)) + // A projection that only reads the date column still raises. + assertRaisesOnRebase(spark.read.parquet(path).select("d")) + // So does one that only filters on it: pushed-down filter columns are part of the required + // schema, so they are covered too. + assertRaisesOnRebase( + spark.read.parquet(path).where(s"d > date'$modernDate'").select("label")) + } + } + + test("legacy-calendar dates are read unrebased when exceptionOnDatetimeRebase is disabled") { + // The opt-out. Pins the pre-guard behaviour of #5010 so it stays reachable for anyone who + // needs it, and documents exactly how wrong it is. + withSQLConf( + CometConf.COMET_EXCEPTION_ON_LEGACY_DATE_TIMESTAMP.key -> "false", + // Return java.time.LocalDate rather than java.sql.Date, so `toString` renders the Proleptic + // Gregorian value and does not depend on JDK hybrid-calendar conversion. + SQLConf.DATETIME_JAVA8API_ENABLED.key -> "true") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeLegacyDates(path) + val df = spark.read.parquet(path).select("d") + assertNativeScan(df) + assert( + df.collect().map(_.get(0).toString).sorted === + Array(ancientDateUnrebased, modernDate).sorted) + } + } + } + + test("a projection with no date or timestamp column does not raise") { + // Nothing else in a Parquet file is calendar-sensitive, and Spark only raises when it + // decodes an affected value, so a scan that reads no date/timestamp must not be failed. + withTempPath { dir => + val path = dir.getCanonicalPath + writeLegacyDates(path) + checkNativeScanAnswer(spark.read.parquet(path).select("label")) + } + } + + test("proleptic-Gregorian files do not raise") { + // The overwhelmingly common case: written by Spark 3.0+ with the default CORRECTED mode, so + // the footer carries no legacy marker and the guard must cost nothing. + withTempPath { dir => + val path = dir.getCanonicalPath + writeDates(path, "CORRECTED", ancientDate, modernDate) + checkNativeScanAnswer(spark.read.parquet(path)) + } + } + + test("legacy-calendar INT96 timestamps raise even when their values look modern") { + // The footer marks this file legacy, and INT96 has no meaningful byte ordering, so writers + // produce no usable min/max and nothing can narrow the refusal to the affected rows. A file + // that declares itself legacy gets refused whole. (A file that declares no writer at all is + // read instead -- see the version-less INT96 fixtures below.) + withTempPath { dir => + val path = dir.getCanonicalPath + writeParquet( + path, + "SELECT cast('2020-06-30 12:00:00' as timestamp) AS ts", + SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> "INT96", + SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key -> "LEGACY") + assertRaisesOnRebase(spark.read.parquet(path)) + } + } + + test("a legacy-marked file holding only modern dates is read normally") { + // The case that makes a default-on guard tolerable. Spark stamps the legacy marker on the + // whole file whenever the write mode was LEGACY, but 1990-01-01 rebases to itself, so the + // values Comet reads are correct and the scan must not be failed. Row-group statistics are + // what let Comet tell this file apart from one holding ancient values. + withTempPath { dir => + val path = dir.getCanonicalPath + writeDates(path, "LEGACY", modernDate, "2020-06-30") + checkNativeScanAnswer(spark.read.parquet(path)) + } + } + + test("a legacy-marked file holding only modern timestamps is read normally") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeParquet( + path, + "SELECT cast('2020-06-30 12:00:00' as timestamp) AS ts", + SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> "TIMESTAMP_MICROS", + SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key -> "LEGACY") + checkNativeScanAnswer(spark.read.parquet(path)) + } + } + + test("a legacy-marked file holding an ancient timestamp raises") { + withTempPath { dir => + val path = dir.getCanonicalPath + writeParquet( + path, + "SELECT cast('1800-01-01 12:00:00' as timestamp) AS ts", + SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> "TIMESTAMP_MICROS", + SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key -> "LEGACY") + assertRaisesOnRebase(spark.read.parquet(path)) + } + } + + test("a legacy-marked TIMESTAMP_NTZ column is read even when its values are ancient") { + // Spark stamps the legacy marker from the write-mode conf alone, without regard to whether the + // schema holds a column the mode could apply to -- and it never rebases NTZ, in either + // direction: "TIMESTAMP_NTZ is a new data type and has no legacy files that need to do rebase". + // So these values read back exactly as written and Comet must agree with Spark, marker or not. + // Mirrors Spark's own "SPARK-46466: write and read TimestampNTZ with legacy rebase mode". + withTempPath { dir => + val path = dir.getCanonicalPath + writeParquet( + path, + s"SELECT cast('$ancientDate 01:10:10' as timestamp_ntz) AS ts", + SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key -> "LEGACY") + checkNativeScanAnswer(spark.read.parquet(path)) + } + } + + test("the rejection surfaces as SparkUpgradeException, not FAILED_READ_FILE") { + // Spark raises SparkUpgradeException for this data and its FileScanRDD deliberately rethrows + // that type rather than wrapping it in FAILED_READ_FILE, because the file is not corrupt. + // Comet classifies it the same way. + withTempPath { dir => + val path = dir.getCanonicalPath + writeLegacyDates(path) + val chain = causeChain(intercept[Throwable](spark.read.parquet(path).collect())) + val rendered = render(chain) + // Matched by name: SparkUpgradeException is private[spark], so this package cannot name + // the type. + assert( + chain.exists(_.getClass.getName == "org.apache.spark.SparkUpgradeException"), + s"expected a SparkUpgradeException in the cause chain, got:\n $rendered") + assert( + !rendered.contains("Encountered error while reading file"), + s"the rejection must not be relabelled FAILED_READ_FILE, got:\n $rendered") + } + } + + test("only the requested columns arm the guard") { + // Two date columns, only one ancient. Reading the modern one must not fail even though the + // file as a whole does hold an affected value. + withTempPath { dir => + val path = dir.getCanonicalPath + withSQLConf(SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key -> "LEGACY") { + spark + .sql( + s"SELECT cast('$ancientDate' as date) AS old_d, cast('$modernDate' as date) AS new_d") + .write + .mode("overwrite") + .parquet(path) + } + checkNativeScanAnswer(spark.read.parquet(path).select("new_d")) + + assertRaisesOnRebase(spark.read.parquet(path).select("old_d")) + } + } + + /** Sets both `*RebaseModeInRead` settings, which Spark applies only to version-less files. */ + private def withReadMode(mode: String)(f: => Unit): Unit = + withSQLConf( + SQLConf.PARQUET_REBASE_MODE_IN_READ.key -> mode, + SQLConf.PARQUET_INT96_REBASE_MODE_IN_READ.key -> mode)(f) + + private def fixture(name: String): String = + getResourceParquetFilePath(s"test-data/$name.snappy.parquet") + + /** + * Checked-in files holding pre-1582 dates / pre-1900 timestamps, in every physical encoding a + * calendar-sensitive column can use. + * + * Spark 2.4.5 stamped no `org.apache.spark.version`, so those files record no provenance and + * Spark resolves them through the `*RebaseModeInRead` settings. Spark 2.4.6 added the version + * key, and the v3_2_0 files were written with LEGACY rebase mode, so both of those are + * identified from the footer alone and the read modes do not apply to them. + * + * For a version-less file the guard refuses only what row-group statistics positively expose, + * which splits these two ways. + */ + private val versionlessProvableFixtures = Seq( + "before_1582_date_v2_4_5", + "before_1582_timestamp_micros_v2_4_5", + "before_1582_timestamp_millis_v2_4_5") + + /** + * The version-less INT96 fixtures, where nothing can be proven: the footer names no writer, and + * the Parquet spec gives INT96's 12 bytes no meaningful ordering, so there is no usable + * min/max. The guard reads these rather than refusing them -- see the test below for why. + */ + private val versionlessInt96Fixtures = + Seq("before_1582_timestamp_int96_plain_v2_4_5", "before_1582_timestamp_int96_dict_v2_4_5") + + private val versionlessFixtures = versionlessProvableFixtures ++ versionlessInt96Fixtures + + private val markedFixtures = Seq( + "before_1582_date_v2_4_6", + "before_1582_date_v3_2_0", + "before_1582_timestamp_micros_v2_4_6", + "before_1582_timestamp_micros_v3_2_0", + "before_1582_timestamp_millis_v2_4_6", + "before_1582_timestamp_millis_v3_2_0", + "before_1582_timestamp_int96_plain_v2_4_6", + "before_1582_timestamp_int96_plain_v3_2_0", + "before_1582_timestamp_int96_dict_v2_4_6", + "before_1582_timestamp_int96_dict_v3_2_0") + + private val ancientFixtures = versionlessFixtures ++ markedFixtures + + (versionlessProvableFixtures ++ markedFixtures).foreach { name => + test(s"$name raises under EXCEPTION read mode") { + // EXCEPTION is Spark's own default. For a version-less file Spark raises here too; for a + // footer-marked one Spark rebases and returns correct values, which Comet cannot do. Either + // way Comet must not return the shifted values. + withReadMode("EXCEPTION") { + assertRaisesOnRebase(spark.read.parquet(fixture(name))) + } + } + } + + versionlessInt96Fixtures.foreach { name => + test(s"$name is read unrebased under EXCEPTION read mode") { + // The guard's one blind spot, and a deliberate trade rather than an oversight. Spark decides + // per decoded value, so it raises for these; Comet decides per column, and for a file that + // names no writer it refuses only what statistics expose. INT96 has none. + // + // Assuming the worst instead would refuse every INT96 column in every file no Spark wrote -- + // which is how Hive writes TIMESTAMP, and Hive, Impala, Trino and plain parquet-mr all leave + // the version key unset. Those reads are overwhelmingly of modern values that Spark returns + // without complaint, so refusing them all to catch this fixture is the worse trade. Closing + // the gap properly needs a per-value check in the decoder, or the rebasing itself (#5010). + withReadMode("EXCEPTION") { + val df = spark.read.parquet(fixture(name)) + assertNativeScan(df) + assert(df.collect().length == 8) + } + } + } + + versionlessFixtures.foreach { name => + test(s"$name honors a per-read datetimeRebaseMode option") { + // Spark resolves the rebase mode through `ParquetOptions`, so `.option(...)` on the reader + // overrides the session conf. Comet reads it from the same place; previously it looked at + // the session conf only and ignored the option. Spotted by @peterxcli in #5048. + withReadMode("EXCEPTION") { + val df = spark.read + .option("datetimeRebaseMode", "CORRECTED") + .option("int96RebaseMode", "CORRECTED") + .parquet(fixture(name)) + checkNativeScanAnswer(df) + } + } + } + + versionlessFixtures.foreach { name => + test(s"$name matches Spark under CORRECTED read mode") { + // CORRECTED asserts the values are already Proleptic Gregorian, so Spark reads them as-is + // and no rebasing applies. Comet honors that and must agree with Spark exactly. + withReadMode("CORRECTED") { + checkNativeScanAnswer(spark.read.parquet(fixture(name))) + } + } + } + + markedFixtures.foreach { name => + test(s"$name still raises under CORRECTED read mode") { + // The footer records these files' provenance, so Spark ignores the read mode for them and + // rebases. Comet cannot, so it keeps refusing regardless of the setting. + withReadMode("CORRECTED") { + assertRaisesOnRebase(spark.read.parquet(fixture(name))) + } + } + } + + ancientFixtures.foreach { name => + test(s"$name is readable with exceptionOnDatetimeRebase disabled") { + // The opt-out has to keep working for every encoding, not just the ones the guard was + // easiest to write for. + withSQLConf(CometConf.COMET_EXCEPTION_ON_LEGACY_DATE_TIMESTAMP.key -> "false") { + withReadMode("EXCEPTION") { + val df = spark.read.parquet(fixture(name)) + assertNativeScan(df) + assert(df.collect().length == 8) + } + } + } + } +} diff --git a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala index a5fc92a5ebb..807d9b5b7c9 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala +++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala @@ -1711,26 +1711,4 @@ class ParquetReadV1Suite extends ParquetReadSuite with AdaptiveSparkPlanHelper { } } - test("reading ancient dates before 1582") { - // Verify that legacy dates (before 1582-10-15) are read without error. - // Comet does not support datetime rebasing, so these dates are read as if they were - // written using the Proleptic Gregorian calendar (no rebase, no exception). - val file = - getResourceParquetFilePath("test-data/before_1582_date_v3_2_0.snappy.parquet") - - val df = spark.read.parquet(file) - - // Verify Comet scan is in the plan - val plan = df.queryExecution.executedPlan - checkCometOperators(plan) - - // Verify all 8 rows are read and contain dates before 1582 - val rows = df.collect() - assert(rows.length == 8, s"Expected 8 rows, got ${rows.length}") - rows.foreach { row => - val date = row.getDate(0) - assert(date.toLocalDate.getYear < 1582, s"Expected date before 1582, got $date") - } - } - }