sunchao commented on code in PR #5365: URL: https://github.com/apache/datafusion-comet/pull/5365#discussion_r4162650461
########## contrib/delta-spark/src/main/scala/org/apache/comet/contrib/delta/DeltaScanSupport.scala: ########## @@ -0,0 +1,1913 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.contrib.delta + +import java.io.IOException +import java.net.URI +import java.util.Locale + +import scala.collection.mutable.{ListBuffer, Map => MutableMap} +import scala.jdk.CollectionConverters._ + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.Path +import org.apache.spark.sql.catalyst.expressions.{Alias, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName} +import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, GenericArrayData} +import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues +import org.apache.spark.sql.comet.CometScanExec +import org.apache.spark.sql.delta.DeltaParquetFileFormat +import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor +import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec, SparkPlan} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType} + +import org.apache.comet.CometConf +import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES +import org.apache.comet.objectstore.NativeConfig +import org.apache.comet.parquet.CometParquetUtils +import org.apache.comet.rules.{CometScanRule, CometScanTypeChecker} +import org.apache.comet.serde.operator.CometNativeScan +import org.apache.comet.shims.ShimFileFormat + +/** + * Claim/decline gates for the native Delta scan. Correctness rule: when in doubt, decline, + * Spark's Delta reader handles the scan and results stay correct, just unaccelerated. + */ +object DeltaScanSupport { + + /** + * Reader features the native path understands; anything else on the protocol declines the + * table. `deletionVectors`/`columnMapping` are declined separately below for specific reasons. + */ + private val understoodReaderFeatures: Set[String] = + Set("columnMapping", "deletionVectors", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck") + + /** + * Is this exactly Delta's DSv1 parquet format? Compared by class name, not `classOf`: a + * `classOf` reference would raise `NoClassDefFoundError` and break every parquet scan when + * delta-spark is absent from the classpath. + */ + def isDeltaScan(scanExec: FileSourceScanExec): Boolean = + scanExec.relation.fileFormat.getClass.getName == + "org.apache.spark.sql.delta.DeltaParquetFileFormat" + + /** + * Claim-time artifacts [[declineReason]] already computes but [[CometDeltaNativeScan.convert]] + * also needs -- threaded through by reference (populated only on the claimable path, right + * before `declineReason` returns `None`) so a claimed scan does not pay to recompute either: + * the Hadoop conf ([[org.apache.spark.sql.internal.SessionState#newHadoopConfWithOptions]] is + * not cheap) and the deletion-vector descriptors (base64-decoded, non-trivial only for DV-shape + * scans). One instance is created per claim attempt in `DeltaScanContrib` and passed to both + * `declineReason` and `convert`. + */ + private[delta] final class DeltaClaimMemo { + var hadoopConf: Configuration = _ + var dvDescriptors: Seq[DeletionVectorDescriptor] = Seq.empty + } + + /** + * First reason this Delta scan cannot go native, or None when claimable (in which case `memo` + * is populated for [[CometDeltaNativeScan.convert]] to reuse). Only called when [[isDeltaScan]] + * is true. `scanHelper` is the [[CometScanExec]] built to drive `convert` on a claim, reused + * for the multi-store gate below. + */ + def declineReason( + plan: SparkPlan, + scanExec: FileSourceScanExec, + scanHelper: CometScanExec, + memo: DeltaClaimMemo): Option[String] = { + val format = scanExec.relation.fileFormat.asInstanceOf[DeltaParquetFileFormat] + val protocol = format.protocol + val metadata = format.metadata + // Name mode is supported via physical-name schemas; id mode needs the field-id path and + // stays declined until validated. Hoisted here since several gates below reuse it. + val cmMode = metadata.columnMappingMode.name + // Descriptor deserialization is expensive, so hoist it into a `lazy val`, forced at most + // once in this method; on the claimable path the result is handed to `convert` through + // `memo` below, so a claimed scan deserializes the descriptors exactly once end to end. + val tableRoot = scanExec.relation.location.rootPaths.head.toString + lazy val dvDescriptors: Seq[DeletionVectorDescriptor] = + selectedDvDescriptors(scanHelper, tableRoot) + + // Mirrors core's CometScanRule.isSchemaSupported so scan-time type gates (unsigned-small-int + // fallback, collation, shredded-variant-struct) apply identically here. Pure in-memory check, + // so it runs first, ahead of every I/O-bearing gate below. + // Unlike core, a required Variant root stays declined: this path lacks core's Variant gates. + val schemaFallbackReasons = new ListBuffer[String]() + val typeChecker = CometScanTypeChecker() + val requiredSchemaSupported = + typeChecker.isSchemaSupported(scanExec.requiredSchema, schemaFallbackReasons) + val partitionSchemaSupported = + typeChecker.isSchemaSupported(scanExec.relation.partitionSchema, schemaFallbackReasons) + if (!requiredSchemaSupported || !partitionSchemaSupported) { + return Some( + "Native Delta scan does not support the schema: " + schemaFallbackReasons.mkString(", ")) + } + + if (format.isCDCRead) { + return Some("Native Delta scan does not support Change Data Feed reads") + } + + // Delta's DML machinery (findTouchedFiles) disables reader optimizations and needs real + // row indexes from Spark's reader; claiming here would feed NULL indexes into DV construction. + if (!format.optimizationsEnabled) { + return Some("Native Delta scan does not support reads with reader optimizations disabled") + } + if (scanExec.requiredSchema.exists(_.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME) || + scanExec.relation.dataSchema.exists( + _.name == DeltaParquetFileFormat.ROW_INDEX_COLUMN_NAME)) { + return Some("Native Delta scan does not support Delta's generated row-index column") + } + + if (cmMode != "none" && cmMode != "name") { + return Some(s"Native Delta scan does not support column mapping mode $cmMode") + } + // createPhysicalSchema wholesale-replaces field metadata, silently dropping EXISTS_DEFAULT. + if (cmMode == "name" && + getExistenceDefaultValues(scanExec.requiredSchema).exists(_ != null)) { + return Some( + "Native Delta scan does not support column defaults together with column mapping") + } + // createPhysicalSchema rewrites nested StructField names too, and the native builder emits the + // required schema verbatim as output, so name-sensitive expressions (e.g. to_json) would leak + // physical names. Decline until a rename adapter exists. + if (cmMode == "name" && + scanExec.requiredSchema.exists(f => containsNestedStruct(f.dataType))) { + return Some("Native Delta scan does not support column mapping with nested struct fields") + } + + val readerFeatures = protocol.readerFeatureNames + val unknownFeatures = readerFeatures -- understoodReaderFeatures + if (unknownFeatures.nonEmpty) { + return Some( + s"Native Delta scan does not support reader feature(s) ${unknownFeatures.mkString(", ")}") + } + + // Non-constant metadata columns are generated per-row by Spark's reader and unsupported, + // except Delta's DV bookkeeping columns, which the native path emits as constants. + val knownColNames = + scanExec.relation.dataSchema.map(_.name).toSet ++ + scanExec.relation.partitionSchema.map(_.name).toSet ++ + scanExec.fileConstantMetadataColumns.map(_.name).toSet ++ Review Comment: [P2] Preserve the fallback for split-dependent metadata. This allowlist accepts `file_block_start` and `file_block_length`, and Delta claims the scan before core’s metadata guard runs. For a split Delta file, Spark assigns a row group to the split containing its midpoint, whereas DataFusion uses its first page. Consequently, selecting `_metadata.file_block_start` returns incorrect values: the reproduced 5,000-row table reports `20480` through Spark/Delta and `0` through the current native reader. The base kept these Delta reads on Spark. Please explicitly decline these two metadata columns, as core already does, until native split attribution matches Spark. Evidence: With Spark 3.5.9/Delta 3.3.2, write `spark.range(0,5000).selectExpr('id', "concat('value_', cast(id as string)) as s").coalesce(1)` as Delta, set `spark.sql.files.maxPartitionBytes=4096`, and select `id, _metadata.file_block_start, _metadata.file_block_length`. Spark produced 12 partitions and `(start=20480, length=4096, count=5000)`. A disposable test calling the exact-head `init_datasource_exec` on the same file and 4096-byte ranges, with the partition constants supplied by `partition2Proto`, produced `(start=0, length=4096, rows=5000)`. Logs: `/tmp/5365-a2df-block-spark.log` and `/tmp/5365-a2df-current-block-native.log`. Delta admission and serialization were source-traced; full Comet/JNI execution was not run. ########## native/core/src/parquet/datetime_rebase.rs: ########## @@ -0,0 +1,3192 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Per-file datetime calendar-rebase handling for the parquet scan. +//! +//! Spark 2.4 and earlier wrote dates and timestamps in the hybrid Julian + Gregorian calendar; +//! Spark 3.0+ uses the proleptic Gregorian calendar and records the calendar policy of every +//! file it writes in the parquet footer's key-value metadata (`org.apache.spark.version`, +//! `org.apache.spark.legacyDateTime`, `org.apache.spark.legacyINT96`, +//! `org.apache.spark.timeZone`). Spark's reader resolves the rebase policy from EACH FILE's +//! writer metadata (`DataSourceUtils.datetimeRebaseSpec` / `int96RebaseSpec`) -- the session's +//! `spark.sql.parquet.datetimeRebaseModeInRead` conf only applies to files whose metadata does +//! not decide the policy on its own -- so a reader that ignores the metadata silently returns +//! values shifted by up to ten days for dates before 1582-10-15 (e.g. `1500-01-01` reads as +//! `1500-01-10`). +//! +//! This module mirrors that per-file resolution: [`resolve_file_rebase_policies`] computes the +//! date / INT64-timestamp / INT96-timestamp policies from a file's arrow schema metadata (the +//! parquet key-value pairs survive the parquet -> arrow schema conversion), and +//! [`wrap_datetime_rebase`] wraps the per-file rewritten expressions' column references in a +//! [`SparkDatetimeRebaseExpr`] that rebases values exactly where that is possible without the +//! JVM's historical timezone tables (dates always; timestamps for a fixed UTC writer zone) and +//! refuses -- rather than silently corrupting -- ancient values it cannot rebase. Nested +//! columns are rebuilt leaf by leaf (struct / list / map / fixed-size list / dictionary), each +//! leaf under its own policy, with nulls and offsets preserved. Modern values are always the +//! identity under every policy: from 1582-10-15 onward for dates, and from +//! [`LAST_SWITCH_JULIAN_TS_SECONDS`] (1900-01-01T00:00:00Z, Spark's +//! `RebaseDateTime.lastSwitchJulianTs`) onward for timestamps. +//! +//! Spark applies `datetimeRebaseSpec` to INT64 `TIMESTAMP_MICROS` / `TIMESTAMP_MILLIS` columns +//! and `int96RebaseSpec` to INT96 columns. The two physical types are indistinguishable in the +//! arrow schema DataFusion hands the expression adapter (both surface as `Timestamp(us, "UTC")` +//! after INT96 coercion), so Comet's parquet reader factory stamps the file's INT96 leaf +//! ordinals -- taken from the parquet footer's own `SchemaDescriptor` -- into the key-value +//! metadata under [`INT96_LEAVES_METADATA_KEY`] before the arrow schema is derived (see +//! [`stamp_int96_leaves`] and `eager_page_index_reader_factory.rs`), and the adapter attributes +//! every timestamp leaf to its spec from that stamp. Without a stamp, the two specs are merged: +//! agreement decides, disagreement degrades to [`RebasePolicy::CheckAncient`]. +//! +//! The wrapper sits BENEATH the schema adapter's nested narrowing (the struct -> struct convert +//! that keeps only the requested children), which is what keeps those ordinals physical -- but +//! it means the wrapper sees every physical child, requested or not. Spark only ever decodes +//! the requested nested schema, so [`FileRebasePolicies::restrict_to_requested`] marks the +//! physical leaves the narrowing drops as the identity: an unrequested ancient `s.ts` never +//! blocks `select s.d`, exactly as in Spark. +//! +//! The same pairing decides a timestamp leaf's policy by the type the query READS it as, since +//! `ParquetVectorUpdaterFactory.getUpdater` keys on the requested Spark type, not the parquet +//! annotation: a leaf read as `TIMESTAMP_NTZ` never rebases (INT96 or INT64; Spark 4.x's +//! `BinaryToSQLTimestampUpdater` / `LongUpdater` consult no mode, and Spark 3.x refuses the +//! INT96 and adjusted-INT64 pairings outright, which Comet's `allow_timestamp_ltz_to_ntz` gate +//! reproduces); a leaf read as `TIMESTAMP` rebases under the datetime spec even when the file +//! declares it `isAdjustedToUTC=false` (`isTimestampTypeMatched` checks the unit only); and a +//! `DATE` leaf keeps the date policy whether read as `DATE` or, on Spark 4.x, as +//! `TIMESTAMP_NTZ` (`DateToTimestampNTZWithRebaseUpdater`). +//! +//! Currently only enabled by the Delta scan arms via +//! `SparkParquetOptions::rebase_from_file_metadata`, which also carries the session read modes +//! ([`SessionRebaseModes`], forwarded from the JVM) that decide the policy for files without +//! Spark writer metadata; the plain NativeScan keeps its documented no-rebase behavior (see +//! the compatibility guide and issue #5010). + +use std::collections::HashMap; +use std::fmt::{self, Display}; +use std::hash::{Hash, Hasher}; +use std::sync::Arc; + +use arrow::array::{ + Array, ArrayRef, AsArray, Date32Array, FixedSizeListArray, GenericListArray, MapArray, + OffsetSizeTrait, PrimitiveArray, RecordBatch, StructArray, +}; +use arrow::datatypes::{ + ArrowPrimitiveType, ArrowTimestampType, DataType, Date32Type, FieldRef, Schema, SchemaRef, + TimeUnit, TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType, + TimestampSecondType, +}; +use arrow::error::ArrowError; +use datafusion::common::tree_node::{Transformed, TreeNode}; +use datafusion::common::{DataFusionError, Result as DataFusionResult}; +use datafusion::physical_expr::expressions::Column; +use datafusion::physical_expr::PhysicalExpr; +use datafusion::physical_plan::ColumnarValue; +use parquet::basic::Type as ParquetPhysicalType; +use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaData}; +use parquet::schema::types::SchemaDescriptor; + +use super::name_fold::fold_names; +use super::parquet_support::field_id; + +/// Footer key naming the Spark release that wrote the file; absent for non-Spark writers. +const SPARK_VERSION_METADATA_KEY: &str = "org.apache.spark.version"; +/// Present (empty value) when the file's dates and INT64 timestamps were written with +/// `spark.sql.parquet.datetimeRebaseModeInWrite=LEGACY`. +const SPARK_LEGACY_DATETIME_KEY: &str = "org.apache.spark.legacyDateTime"; +/// Present (empty value) when the file's INT96 timestamps were written with +/// `spark.sql.parquet.int96RebaseModeInWrite=LEGACY`. +const SPARK_LEGACY_INT96_KEY: &str = "org.apache.spark.legacyINT96"; +/// The writer session's time zone, stamped alongside either legacy flag. +const SPARK_TIMEZONE_KEY: &str = "org.apache.spark.timeZone"; + +/// Key-value metadata entry Comet's parquet reader factory adds to a file's footer metadata +/// (in memory only, never written back) so the expression adapter can tell INT96 timestamp +/// columns from INT64 ones after both have been coerced to the same arrow type. Value: +/// `"<leaf count>:<comma-separated INT96 leaf ordinals>"`, where leaves are the file's +/// primitive columns in `SchemaDescriptor::columns()` order -- the same depth-first order +/// parquet-rs assigns arrow leaves, so an arrow-side depth-first walk lines up with it. The +/// leaf count lets the reader detect a stamp that does not describe the schema it is paired +/// with (see [`Int96Attribution::from_schema`]). +pub(crate) const INT96_LEAVES_METADATA_KEY: &str = "comet.int96_leaf_columns"; + +/// Day of the Gregorian cutover (1582-10-15) as days since the epoch; rebasing is the identity +/// from this day onward. Same value as Spark's `RebaseDateTime.lastSwitchJulianDay`. +const LAST_SWITCH_JULIAN_DAY: i32 = -141427; + +/// Spark's `RebaseDateTime.lastSwitchJulianTs` (and `lastSwitchGregorianTs`) in seconds since +/// the epoch: 1900-01-01T00:00:00Z. Spark derives it as the latest switch instant across every +/// zone in its `julian-gregorian-rebase-micros.json` table (`getLastSwitchTs`, which also +/// asserts the calendars' difference is zero for every zone from then on): most zones ran on +/// local mean time before 1900, so the last instant at which rebasing changes a value in ANY +/// zone is 1900-01-01T00:00:00Z, not the 1582 cutover. `createTimestampRebaseFuncInRead` +/// under `EXCEPTION` throws exactly for `micros < lastSwitchJulianTs` (after converting +/// `TIMESTAMP_MILLIS` to micros), and `rebaseJulianToGregorianMicros` is the identity from it +/// onward in every zone. The value is in seconds so it scales exactly to any timestamp unit. +pub(crate) const LAST_SWITCH_JULIAN_TS_SECONDS: i64 = -2_208_988_800; + +/// The per-century differences between the Julian and proleptic Gregorian calendars, and the +/// Julian-calendar switch days at which each difference starts to apply. Copied verbatim from +/// Spark's `RebaseDateTime.julianGregDiffs` / `julianGregDiffSwitchDay` (which Spark generated +/// from `localRebaseJulianToGregorianDays`); `rebase_julian_to_gregorian_days` must stay +/// value-for-value equal to Spark's `rebaseJulianToGregorianDays`. +const JULIAN_GREG_DIFFS: [i32; 14] = [2, 1, 0, -1, -2, -3, -4, -5, -6, -7, -8, -9, -10, 0]; +const JULIAN_GREG_DIFF_SWITCH_DAY: [i32; 14] = [ + -719164, -682945, -646420, -609895, -536845, -500320, -463795, -390745, -354220, -317695, + -244645, -208120, -171595, -141427, +]; + +/// Proleptic-Gregorian days since 1970-01-01 for a nominal civil date, via Howard Hinnant's +/// `days_from_civil`. `d` may exceed the month's length; the excess rolls into the following +/// month exactly like `LocalDate.of(y, m, 1).plusDays(d - 1)` in Spark's +/// `localRebaseJulianToGregorianDays` (how the non-existent proleptic date `1000-02-29`, +/// valid in the Julian calendar, lands on `1000-03-01`). +fn days_from_civil(y: i64, m: i64, d: i64) -> i64 { + let y = if m <= 2 { y - 1 } else { y }; + let era = y.div_euclid(400); + let yoe = y - era * 400; // [0, 399] + let mp = (m + 9) % 12; // [0, 11], March = 0 + let doy = (153 * mp + 2) / 5 + d - 1; + let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; + era * 146097 + doe - 719468 +} + +/// Julian-calendar civil date `(year, month, day)` for a day count since 1970-01-01 that labels +/// days in the Julian calendar (astronomical year numbering: 1 BCE is year 0). Standard +/// Julian-day-number conversion (E.G. Richards' algorithm), exact for any day. +fn julian_day_to_civil(days: i64) -> (i64, i64, i64) { + // Integer (noon) Julian Day Number of this civil day: 1970-01-01 is JDN 2440588. + let jdn = days + 2_440_588; + let f = jdn + 1401; + let e = 4 * f + 3; + let g = e.rem_euclid(1461) / 4; + let h = 5 * g + 2; + let day = h.rem_euclid(153) / 5 + 1; + let month = (h / 153 + 2).rem_euclid(12) + 1; + let year = e.div_euclid(1461) - 4716 + (14 - month) / 12; + (year, month, day) +} + +/// Exact port of Spark's `RebaseDateTime.rebaseJulianToGregorianDays`: reinterprets a day count +/// written in the hybrid Julian + Gregorian calendar as the proleptic Gregorian day count of the +/// same nominal civil date. Identity for days from 1582-10-15 onward. Days before the tables' +/// range (before Julian `0001-01-01`) take the calendar-arithmetic path, mirroring Spark's +/// `localRebaseJulianToGregorianDays` fallback. +pub(crate) fn rebase_julian_to_gregorian_days(days: i32) -> i32 { + if days < JULIAN_GREG_DIFF_SWITCH_DAY[0] { + let (y, m, d) = julian_day_to_civil(days as i64); + (days_from_civil(y, m, 1) + (d - 1)) as i32 + } else { + // Spark's rebaseDays: linear search from the most recent switch day. + let mut i = JULIAN_GREG_DIFF_SWITCH_DAY.len(); + loop { + i -= 1; + if i == 0 || days >= JULIAN_GREG_DIFF_SWITCH_DAY[i] { + break; + } + } + days + JULIAN_GREG_DIFFS[i] + } +} + +/// Timezone strings from `org.apache.spark.timeZone` that denote a fixed zero-offset zone in +/// both `java.util.TimeZone` and `java.time`. Only for these is timestamp rebasing the pure +/// nominal-date shift [`SparkDatetimeRebaseExpr::rebase_timestamp_utc`] computes; any other (or +/// absent) zone needs the JVM's historical timezone tables and stays on the +/// refuse-ancient-values path. +const UTC_EQUIVALENT_TIMEZONES: [&str; 6] = ["UTC", "Etc/UTC", "GMT", "Etc/GMT", "Z", "+00:00"]; + +/// How the writer's session time zone (if recorded) affects timestamp rebasing. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum WriterTimeZone { + /// A fixed zero-offset zone: rebasing reduces to the exact nominal-date shift. + Utc, + /// Any other zone, or none recorded (pre-3.0 files): ancient values cannot be rebased + /// without the JVM's historical timezone data. + OtherOrUnknown, +} + +/// One session-level datetime rebase read mode (a `LegacyBehaviorPolicy` value of +/// `spark.sql.parquet.datetimeRebaseModeInRead` / `int96RebaseModeInRead`), consulted by +/// [`resolve_file_rebase_policies`] ONLY for files whose footer metadata does not decide the +/// policy on its own -- exactly the `getOrElse` fallback in Spark's +/// `DataSourceUtils.getRebaseSpec`. Files that carry `org.apache.spark.version` ignore these +/// modes entirely, on every Spark version. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub(crate) enum RebaseReadMode { + /// Refuse ancient values (Spark raises `SparkUpgradeException`); maps to + /// [`RebasePolicy::CheckAncient`]. The default mirrors the conservative posture used + /// before the conf was plumbed through (and Spark 3.x's own conf default). + #[default] + Exception, + /// Read values as proleptic Gregorian without rebasing. + Corrected, + /// Rebase from the hybrid Julian + Gregorian calendar. + Legacy, +} + +impl RebaseReadMode { + /// Parses a `LegacyBehaviorPolicy` conf value. `SQLConf` validates and upper-cases the + /// session conf, but a per-relation `datetimeRebaseMode` option arrives verbatim, so the + /// match is case-insensitive. Anything unrecognized -- including the empty string a proto + /// producer that predates the field sends -- falls back to [`RebaseReadMode::Exception`], + /// which refuses ancient values rather than silently corrupting them. + pub(crate) fn from_conf_value(value: &str) -> Self { + match value.to_ascii_uppercase().as_str() { + "CORRECTED" => RebaseReadMode::Corrected, + "LEGACY" => RebaseReadMode::Legacy, + _ => RebaseReadMode::Exception, + } + } +} + +/// The session's effective datetime rebase read modes, one per spec class (INT64 +/// dates/timestamps vs INT96 timestamps), forwarded from the JVM at planning time. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] +pub(crate) struct SessionRebaseModes { + /// `spark.sql.parquet.datetimeRebaseModeInRead` (or the relation's `datetimeRebaseMode`). + pub datetime: RebaseReadMode, + /// `spark.sql.parquet.int96RebaseModeInRead` (or the relation's `int96RebaseMode`). + pub int96: RebaseReadMode, +} + +/// Calendar policy of one file's date or timestamp columns, resolved from writer metadata the +/// same way Spark's `DataSourceUtils.getRebaseSpec` resolves it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum RebasePolicy { + /// Written in the proleptic Gregorian calendar; values pass through untouched. + Corrected, + /// Written in the hybrid Julian + Gregorian calendar; values must be rebased. + Legacy(WriterTimeZone), + /// Policy could not be pinned down (contradictory flags, or a non-Spark writer under the + /// `EXCEPTION` read mode): modern values -- identical under either calendar -- pass, + /// ancient values raise. Mirrors Spark's `EXCEPTION` behavior (`SparkUpgradeException`). + CheckAncient, +} + +/// Which of a file's leaf columns are physically INT96, from the stamp the parquet reader +/// factory adds under [`INT96_LEAVES_METADATA_KEY`]. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) enum Int96Attribution { + /// No stamp, or a stamp whose leaf count does not match the schema it arrived with: the + /// INT64 and INT96 timestamp specs cannot be told apart per column and are merged. + Unknown, + /// Sorted leaf ordinals (depth-first over the file schema's primitive columns) that are + /// INT96; every other timestamp leaf is INT64. + Known(Vec<usize>), +} + +impl Int96Attribution { + /// Parses the stamp out of `schema`'s metadata and validates its leaf count against the + /// schema's own depth-first leaf count, so a stamp that does not describe this schema (a + /// crafted footer key, or a cached-metadata mismatch) degrades to [`Self::Unknown`]. + fn from_schema(schema: &Schema) -> Self { + let Some(stamp) = schema.metadata().get(INT96_LEAVES_METADATA_KEY) else { + return Int96Attribution::Unknown; + }; + let Some((count, ordinals)) = stamp.split_once(':') else { + return Int96Attribution::Unknown; + }; + let schema_leaves: usize = schema + .fields() + .iter() + .map(|f| leaf_count(f.data_type())) + .sum(); + if count.parse::<usize>().ok() != Some(schema_leaves) { + return Int96Attribution::Unknown; + } + let parsed: Option<Vec<usize>> = if ordinals.is_empty() { + Some(Vec::new()) + } else { + ordinals + .split(',') + .map(|o| o.parse::<usize>().ok().filter(|o| *o < schema_leaves)) + .collect() + }; + match parsed { + Some(mut leaves) => { + leaves.sort_unstable(); + Int96Attribution::Known(leaves) + } + None => Int96Attribution::Unknown, + } + } + + /// `Some(true)` / `Some(false)` when the leaf is known to be INT96 / INT64, `None` when + /// the attribution is unknown. + fn is_int96(&self, leaf: usize) -> Option<bool> { + match self { + Int96Attribution::Unknown => None, + Int96Attribution::Known(leaves) => Some(leaves.binary_search(&leaf).is_ok()), + } + } +} + +/// The [`INT96_LEAVES_METADATA_KEY`] value describing `schema`: its leaf count and the +/// ordinals of its INT96 primitive columns. +pub(crate) fn int96_leaf_stamp(schema: &SchemaDescriptor) -> String { + let ordinals: Vec<String> = schema + .columns() + .iter() + .enumerate() + .filter(|(_, column)| column.physical_type() == ParquetPhysicalType::INT96) + .map(|(ordinal, _)| ordinal.to_string()) + .collect(); + format!("{}:{}", schema.num_columns(), ordinals.join(",")) +} + +/// Returns a copy of `metadata` whose key-value metadata carries the [`int96_leaf_stamp`] of +/// its own schema, or `None` when it already does (the common case after the first open of a +/// file, since the caller caches the stamped copy). Any pre-existing entry under the key -- +/// a file cannot legitimately carry one -- is replaced, never trusted. Only the file-level +/// key-value list changes; row groups and page indexes are carried over as-is. The parquet +/// API cannot carry a file decryptor, nor `FileMetaData`'s crate-private encryption fields +/// (encryption algorithm, footer signing key metadata), across this rebuild, so callers must +/// not stamp opens that supply decryption properties -- and the only consumer, the Delta +/// scan, declines every encrypted-parquet configuration before planning, so a parquet +/// modular encryption file never reaches this path with or without those properties. +pub(crate) fn stamp_int96_leaves(metadata: &ParquetMetaData) -> Option<ParquetMetaData> { + let file_metadata = metadata.file_metadata(); + let stamp = int96_leaf_stamp(file_metadata.schema_descr()); + let existing = file_metadata + .key_value_metadata() + .and_then(|kvs| kvs.iter().find(|kv| kv.key == INT96_LEAVES_METADATA_KEY)) + .and_then(|kv| kv.value.as_deref()); + if existing == Some(stamp.as_str()) { + return None; + } + let mut key_values: Vec<KeyValue> = file_metadata + .key_value_metadata() + .map(|kvs| { + kvs.iter() + .filter(|kv| kv.key != INT96_LEAVES_METADATA_KEY) + .cloned() + .collect() + }) + .unwrap_or_default(); + key_values.push(KeyValue::new(INT96_LEAVES_METADATA_KEY.to_string(), stamp)); + let stamped_file_metadata = FileMetaData::new( + file_metadata.version(), + file_metadata.num_rows(), + file_metadata.created_by().map(str::to_string), + Some(key_values), + file_metadata.schema_descr_ptr(), + file_metadata.column_orders().cloned(), + ); + Some( + ParquetMetaData::new(stamped_file_metadata, metadata.row_groups().to_vec()) + .into_builder() + .set_column_index(metadata.column_index().cloned()) + .set_offset_index(metadata.offset_index().cloned()) + .build(), + ) +} + +/// Per-file rebase policies for the three affected column classes, plus the INT96 +/// attribution that selects between the two timestamp specs per leaf. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct FileRebasePolicies { + /// `DATE` columns, governed by `org.apache.spark.legacyDateTime` alone. + pub date: RebasePolicy, + /// INT64 `TIMESTAMP_MICROS` / `TIMESTAMP_MILLIS` columns, adjusted to UTC or not: the + /// datetime spec (same resolution as `date`), as Spark's `ParquetVectorUpdaterFactory` + /// selects for INT64 read as `TIMESTAMP`. + pub int64_timestamp: RebasePolicy, + /// INT96 columns: the INT96 spec (`org.apache.spark.legacyINT96`, min version 3.1.0). + pub int96_timestamp: RebasePolicy, + /// Which timestamp leaves are INT96. See [`Int96Attribution`]. + pub int96_leaves: Int96Attribution, + /// Sorted depth-first leaf ordinals -- over the physical file schema, the same ordinals + /// `int96_leaves` uses -- that the query does not read: nested children the schema + /// adapter's struct narrowing drops before any value leaves the scan. Spark never decodes + /// them either, so their policy is the identity whatever the file's calendar. Empty until + /// [`Self::restrict_to_requested`] runs (every leaf requested). + pub unrequested_leaves: Vec<usize>, + /// Sorted physical leaf ordinals of timezone-carrying timestamps (INT96, or INT64 with + /// `isAdjustedToUTC=true`) the query reads as `TIMESTAMP_NTZ`. Spark decodes those with + /// `BinaryToSQLTimestampUpdater` / `LongUpdater`, which never rebase, so their policy is + /// the identity whatever the file's calendar. Filled by [`Self::restrict_to_requested`]. + pub ntz_requested_leaves: Vec<usize>, + /// Sorted physical leaf ordinals of timezone-free INT64 timestamps + /// (`isAdjustedToUTC=false`) the query reads as `TIMESTAMP`. Spark's INT64 branch checks + /// only the unit (`isTimestampTypeMatched`) and hands a `TimestampType` request to + /// `LongWithRebaseUpdater` under the datetime spec, adjusted or not, so these leaves take + /// the same policy as adjusted INT64 leaves. Filled by [`Self::restrict_to_requested`]. + pub ltz_requested_leaves: Vec<usize>, +} + +/// The leaf ordinals [`push_unrequested_leaves`] records while pairing a physical type with +/// the type the query reads it as. Each list is emitted in depth-first order, so it is already +/// sorted for the binary searches in [`FileRebasePolicies`]. +#[derive(Debug, Default)] +struct LeafPairing { + unrequested: Vec<usize>, + ntz_requested: Vec<usize>, + ltz_requested: Vec<usize>, +} + +impl FileRebasePolicies { + /// True when some policy is not the plain proleptic-Gregorian pass-through, i.e. when the + /// per-column wrap in [`wrap_datetime_rebase`] can install anything at all. + pub(crate) fn any_rebase_needed(&self) -> bool { + self.date != RebasePolicy::Corrected + || self.int64_timestamp != RebasePolicy::Corrected + || self.int96_timestamp != RebasePolicy::Corrected + } + + fn is_requested(&self, leaf: usize) -> bool { + self.unrequested_leaves.binary_search(&leaf).is_err() + } + + /// The policy of the `Date32` leaf at depth-first ordinal `leaf`: the file's date policy, + /// or the identity when the query does not read that leaf. + fn date_policy(&self, leaf: usize) -> RebasePolicy { + if self.is_requested(leaf) { + self.date + } else { + RebasePolicy::Corrected + } + } + + /// The policy of the timezone-carrying timestamp leaf at depth-first ordinal `leaf`: the + /// identity when the query does not read it or reads it as `TIMESTAMP_NTZ` (Spark's NTZ + /// updaters never rebase; on Spark 3.x the pairing is refused before any rebase decision, + /// which Comet's `allow_timestamp_ltz_to_ntz` gate reproduces); otherwise its physical + /// type's spec when the attribution is known, or else the two specs merged -- agreement + /// decides, disagreement degrades to [`RebasePolicy::CheckAncient`], which still passes + /// every modern value and refuses only ancient ones. + fn timestamp_policy(&self, leaf: usize) -> RebasePolicy { + if !self.is_requested(leaf) || self.ntz_requested_leaves.binary_search(&leaf).is_ok() { + return RebasePolicy::Corrected; + } + match self.int96_leaves.is_int96(leaf) { + Some(true) => self.int96_timestamp, + Some(false) => self.int64_timestamp, + None if self.int64_timestamp == self.int96_timestamp => self.int64_timestamp, + None => RebasePolicy::CheckAncient, + } + } + + /// The policy of the timezone-free timestamp leaf at depth-first ordinal `leaf` (INT64 with + /// `isAdjustedToUTC=false`): the identity unless the query reads it as `TIMESTAMP`, which + /// Spark decodes with `LongWithRebaseUpdater` under the datetime spec exactly like an + /// adjusted INT64 leaf. The stamp is still consulted so a leaf it names INT96 follows the + /// INT96 spec; without a stamp the physical type itself proves INT64, so the two specs are + /// not merged. + fn tz_free_timestamp_policy(&self, leaf: usize) -> RebasePolicy { + if !self.is_requested(leaf) || self.ltz_requested_leaves.binary_search(&leaf).is_err() { + return RebasePolicy::Corrected; + } + match self.int96_leaves.is_int96(leaf) { + Some(true) => self.int96_timestamp, + _ => self.int64_timestamp, + } + } + + /// These policies with every physical leaf the query does not read marked the identity, + /// and every timestamp leaf whose requested type differs from its physical one in timezone + /// presence recorded, so [`leaf_policies`] can pick the policy Spark's + /// `ParquetVectorUpdaterFactory.getUpdater` picks for the REQUESTED type. + /// `requested` pairs each top-level field of `physical_schema` (by position) with the type + /// of the logical field the schema adapter narrows it to -- `None` for a column without a + /// logical counterpart, whose leaves are left as they are (no expression reads it anyway). + /// Nested children pair the way the adapter's struct convert selects them (see + /// [`push_unrequested_leaves`]); the INT96 attribution is untouched, since the ordinals + /// stay physical. `requested` is parallel to the schema's fields; should a caller pass a + /// shorter slice, the trailing columns simply keep every leaf (the safe direction). + pub(crate) fn restrict_to_requested( + mut self, + physical_schema: &Schema, + requested: &[Option<&DataType>], + case_sensitive: bool, + use_field_id: bool, + ) -> Self { + debug_assert_eq!(requested.len(), physical_schema.fields().len()); + let matching = FieldMatching { + case_sensitive, + use_field_id, + }; + let mut next_leaf = 0; + let mut pairing = LeafPairing::default(); + for (field, requested) in physical_schema.fields().iter().zip(requested) { + match requested { + Some(logical) => push_unrequested_leaves( + field.data_type(), + logical, + &mut next_leaf, + matching, + &mut pairing, + ), + None => next_leaf += leaf_count(field.data_type()), + } + } + // Emitted in depth-first order, so already sorted for the binary searches. + self.unrequested_leaves = pairing.unrequested; + self.ntz_requested_leaves = pairing.ntz_requested; + self.ltz_requested_leaves = pairing.ltz_requested; + self + } +} + +/// The field-matching rules of the schema adapter's nested narrowing +/// (`parquet_convert_struct_to_struct`): names fold per `case_sensitive`, and Parquet field ids +/// select fields when `use_field_id` is set. +#[derive(Debug, Clone, Copy)] +struct FieldMatching { + case_sensitive: bool, + use_field_id: bool, +} + +/// Appends to `out.unrequested` the depth-first leaf ordinals of `physical` (counting from +/// `next_leaf`, which advances past every leaf of `physical`) that reading it as `requested` +/// drops, and records in `out.ntz_requested` / `out.ltz_requested` the timestamp leaves whose +/// requested type has the opposite timezone presence (a timezone-carrying leaf read as +/// `TIMESTAMP_NTZ`, a timezone-free leaf read as `TIMESTAMP`); the unit is irrelevant to +/// either, as it is to Spark's `isTimestampTypeMatched`. +/// +/// Recurses through exactly the pairings `parquet_convert_array` narrows, and no others: a +/// struct child is dropped only when NO requested child selects it by either rule the struct +/// convert uses -- folded name, or Parquet field id when ids are in play -- and an ambiguous +/// child (several requested children select it) is kept; `List` pairs with `List` by element +/// type, and `Map` with a `Map` of the same key ordering by its entries, positionally. Any +/// other pairing -- a `LargeList` / `FixedSizeList` / dictionary, a map whose ordering +/// differs, or a shape mismatch -- is handed to arrow's cast or passed through whole by the +/// convert, so it keeps every leaf under its physical type's policy. Keeping a superset of +/// what the narrowing reads is always safe (a spurious check at worst); dropping a leaf the +/// narrowing reads would skip its rebase, so every doubt resolves to "requested". Timestamp +/// leaves inside those pass-through shapes are never recorded either, so they keep the +/// physical rule (a spurious check for an NTZ request, no rebase for a `TIMESTAMP` request of +/// a timezone-free leaf); Spark's requested schemas never take those arrow shapes. +fn push_unrequested_leaves( + physical: &DataType, + requested: &DataType, + next_leaf: &mut usize, + matching: FieldMatching, + out: &mut LeafPairing, +) { + match (physical, requested) { + (DataType::Timestamp(_, Some(_)), DataType::Timestamp(_, None)) => { + out.ntz_requested.push(*next_leaf); + *next_leaf += 1; + } + (DataType::Timestamp(_, None), DataType::Timestamp(_, Some(_))) => { + out.ltz_requested.push(*next_leaf); + *next_leaf += 1; + } + (DataType::Struct(physical_fields), DataType::Struct(requested_fields)) => { + let names: Vec<&str> = physical_fields + .iter() + .chain(requested_fields.iter()) + .map(|f| f.name().as_str()) + .collect(); + // A fold failure means the names could not be compared at all; keeping every leaf + // requested is the safe superset, the same as the pass-through pairings below. + let Ok(folded) = fold_names(&names, matching.case_sensitive) else { + *next_leaf += leaf_count(physical); + return; + }; + let (physical_folded, requested_folded) = folded.split_at(physical_fields.len()); + for (i, child) in physical_fields.iter().enumerate() { + let child_id = if matching.use_field_id { + field_id(child) + } else { + None + }; + let mut selectors = requested_fields.iter().enumerate().filter(|(j, r)| { + requested_folded[*j] == physical_folded[i] + || (child_id.is_some() && field_id(r) == child_id) + }); + match (selectors.next(), selectors.next()) { + (None, _) => { + let n = leaf_count(child.data_type()); + out.unrequested.extend(*next_leaf..*next_leaf + n); + *next_leaf += n; + } + (Some((_, requested_child)), None) => push_unrequested_leaves( + child.data_type(), + requested_child.data_type(), + next_leaf, + matching, + out, + ), + (Some(_), Some(_)) => *next_leaf += leaf_count(child.data_type()), + } + } + } + (DataType::List(physical_item), DataType::List(requested_item)) => push_unrequested_leaves( + physical_item.data_type(), + requested_item.data_type(), + next_leaf, + matching, + out, + ), + ( + DataType::Map(physical_entries, physical_sorted), + DataType::Map(requested_entries, requested_sorted), + ) if physical_sorted == requested_sorted => { + match (physical_entries.data_type(), requested_entries.data_type()) { + (DataType::Struct(physical_kv), DataType::Struct(requested_kv)) + if physical_kv.len() == requested_kv.len() => + { + for (p, r) in physical_kv.iter().zip(requested_kv.iter()) { + push_unrequested_leaves( + p.data_type(), + r.data_type(), + next_leaf, + matching, + out, + ); + } + } + _ => *next_leaf += leaf_count(physical), + } + } + _ => *next_leaf += leaf_count(physical), Review Comment: [P2] Suppress temporal rebasing when Spark requests a supported numeric type. The catch-all pairing leaves timestamp/date leaves governed by their physical temporal policy even when requested as `BIGINT` or `INT`. A concrete supported case is metadata-free `TIMESTAMP(NANOS,true)` Parquet inferred as `BIGINT` with `spark.sql.legacy.parquet.nanosAsLong=true`, then converted to Delta. Under `datetimeRebaseModeInRead=EXCEPTION`, Spark returns the stored integer, but the new Delta rebase wrapper rejects an 1800-era value and fails the query. Spark’s numeric updaters bypass calendar conversion. Please account for numeric requested types in the leaf policy, preserving genuine temporal rebasing, and add a converted-Parquet regression. Evidence: Wrote metadata-free `message m { required int64 v (TIMESTAMP(NANOS,true)); }` containing `-5364662400000000000`. Spark 3.5.9/Delta 3.3.2 inferred `BIGINT` and returned that integer before and after `CONVERT TO DELTA ... NO STATISTICS` under EXCEPTION mode. The exact-head native scan builder with requested `Int64` returned it with rebasing disabled or CORRECTED, but metadata rebasing plus EXCEPTION failed with `Native scan cannot rebase ancient values in column 'v'`. MICROS-to-BIGINT and DATE-to-INT controls reproduced the same discrepancy. Logs: `/tmp/5365-a2df-current-numeric-reproduction.log` and `/tmp/5365-a2df-current-spark35-delta-reproduction.log`. Spark 3.5.9/4.0.4/4.1.3 `ParquetVectorUpdaterFactory` selects numeric updaters before temporal handling. Full Comet/Delta/JNI execution was not run. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
