fix: fall back for Parquet datetime rebasing

docs/source/user-guide/latest/compatibility/scans.md

@@ -43,6 +43,15 @@ The following features are not supported and cause Comet to fall back to Spark:
 - No support for `input_file_name()`, `input_file_block_start()`, or `input_file_block_length()` SQL functions.
   Comet's Parquet scan does not use Spark's `FileScanRDD`, so these functions cannot populate their values.
 - No support for `ignoreMissingFiles` or `ignoreCorruptFiles` being set to `true`
+- Files that require datetime rebasing. Comet falls back to Spark when Parquet metadata or
+  `spark.sql.parquet.datetimeRebaseModeInRead` /
+  `spark.sql.parquet.int96RebaseModeInRead` requires legacy-calendar handling. This includes
+  files written by any Spark version with a corresponding legacy rebase mode, not only files
+  written before Spark 3.0. Detecting this requires reading each input file's footer on the
+  driver during planning (results are cached per file); users whose data is known to be free
+  of legacy-calendar values can skip the check by setting
+  `spark.comet.scan.parquet.checkDatetimeRebase=false`.
+  See [#5010](https://github.com/apache/datafusion-comet/issues/5010).
 - `spark.sql.parquet.enableVectorizedReader=false`. Disabling the vectorized reader opts into
   Spark's parquet-mr semantics (silent overflow, null-on-narrowing), which Comet's native reader
   does not replicate. By default Comet falls back to Spark in this case. Set
@@ -51,17 +60,6 @@ The following features are not supported and cause Comet to fall back to Spark:
 - A read schema that repeats a Parquet field id, at the top level or within a struct, when
   `spark.sql.parquet.fieldId.read.enabled=true`.
 
-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. 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:
 
 - Selecting a field by name when multiple physical siblings match, including inside structs,

native/core/src/execution/operators/parquet_writer.rs

@@ -53,7 +53,7 @@ use futures::TryStreamExt;
 use parquet::{
     arrow::ArrowWriter,
     basic::{Compression, GzipLevel, ZstdLevel},
-    file::properties::WriterProperties,
+    file::{metadata::KeyValue, properties::WriterProperties},
 };
 use url::Url;
 
@@ -240,6 +240,8 @@ pub struct ParquetWriterExec {
     column_names: Vec<String>,
     /// Catalyst's target schema, including nullability and Parquet field metadata.
     output_schema: Option<SchemaRef>,
+    /// Runtime Spark version to record in the Parquet metadata
+    spark_version: String,
     /// Object store configuration options
     object_store_options: HashMap<String, String>,
     /// Metrics
@@ -261,6 +263,7 @@ impl ParquetWriterExec {
         partition_id: i32,
         column_names: Vec<String>,
         output_schema: Option<SchemaRef>,
+        spark_version: String,
         object_store_options: HashMap<String, String>,
     ) -> Result<Self> {
         // Preserve the input's partitioning so each partition writes its own file
@@ -283,6 +286,7 @@ impl ParquetWriterExec {
             partition_id,
             column_names,
             output_schema,
+            spark_version,
             object_store_options,
             metrics: ExecutionPlanMetricsSet::new(),
             cache,
@@ -471,6 +475,7 @@ impl ExecutionPlan for ParquetWriterExec {
                 self.partition_id,
                 self.column_names.clone(),
                 self.output_schema.clone(),
+                self.spark_version.clone(),
                 self.object_store_options.clone(),
             )?)),
             _ => Err(DataFusionError::Internal(
@@ -531,6 +536,13 @@ impl ExecutionPlan for ParquetWriterExec {
         // Configure writer properties
         let props = WriterProperties::builder()
             .set_compression(compression)
+            // Spark identifies corrected datetime files by its writer version and the absence of
+            // legacy markers. Comet always writes corrected values, so use the same metadata:
+            // https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetWriteSupport.scala#L126-L146
+            .set_key_value_metadata(Some(vec![KeyValue::new(
+                "org.apache.spark.version".to_string(),
+                Some(self.spark_version.clone()),
+            )]))
             .build();
 
         let object_store_options = self.object_store_options.clone();
@@ -683,6 +695,7 @@ mod tests {
             3,
             vec!["id".to_string()],
             None,
+            "4.2.0".to_string(),
             HashMap::new(),
         )?;
 
@@ -753,6 +766,7 @@ mod tests {
             0,
             vec!["required_id".to_string(), "values".to_string()],
             Some(output_schema),
+            "4.2.0".to_string(),
             HashMap::new(),
         )?;
 
@@ -822,6 +836,7 @@ mod tests {
             0,
             vec!["values".to_string()],
             Some(output_schema),
+            "4.2.0".to_string(),
             HashMap::new(),
         )?;
 
@@ -1081,8 +1096,9 @@ mod tests {
             ParquetCompression::None,
             0, // partition_id
             column_names,
-            None,           // output_schema
-            HashMap::new(), // object_store_options
+            None,                // output_schema
+            "4.2.0".to_string(), // spark_version
+            HashMap::new(),      // object_store_options
         )?;
 
         // Create a session context and execute the plan

native/core/src/execution/planner.rs

@@ -2006,6 +2006,7 @@ impl PhysicalPlanner {
                     writer.column_names.clone(),
                     (!writer.output_schema.is_empty())
                         .then(|| convert_spark_types_to_arrow_schema(&writer.output_schema)),
+                    writer.spark_version.clone(),
                     object_store_options,
                 )?);
 

native/core/src/parquet/parquet_support.rs

@@ -91,8 +91,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
@@ -131,7 +129,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,
@@ -148,7 +145,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,

native/proto/src/proto/operator.proto

@@ -928,6 +928,8 @@ message ParquetWriter {
   // Catalyst's target schema, including top-level nullability and Parquet field IDs.
   // Nested collection field IDs are carried by the corresponding DataType messages.
   repeated SparkStructField output_schema = 9;
+  // Runtime Spark version written to Parquet metadata for Spark reader compatibility.
+  string spark_version = 10;
 }
 
 enum AggregateMode {

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -991,6 +991,22 @@ object CometConf extends ShimCometConf {
       .booleanConf
       .createWithDefault(true)
 
+  val COMET_SCAN_PARQUET_CHECK_DATETIME_REBASE: ConfigEntry[Boolean] =
+    conf("spark.comet.scan.parquet.checkDatetimeRebase")
+      .category(CATEGORY_SCAN)
+      .doc(
+        "Whether to inspect Parquet footer metadata during planning to detect files whose " +
+          "dates/timestamps may require legacy (hybrid Julian/Gregorian) datetime rebasing, " +
+          "and fall back to Spark for those scans. The check reads each input file's footer " +
+          "on the driver the first time the file is planned; results are cached per file. " +
+          "Disable only when all input files are known to contain datetime values written " +
+          "with the proleptic Gregorian calendar (for example, written by Spark 3.x or later " +
+          "with corrected rebase modes). When disabled, Comet reads legacy files without " +
+          "rebasing, which produces results that differ from Spark for dates and timestamps " +
+          s"before 1582-10-15. $COMPAT_GUIDE.")
+      .booleanConf
+      .createWithDefault(true)
+
   val COMET_SCAN_ALLOW_DISABLED_PARQUET_VECTORIZED_READER: ConfigEntry[Boolean] =
     conf("spark.comet.scan.allowDisabledParquetVectorizedReader")
       .category(CATEGORY_SCAN)

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -27,6 +27,7 @@ import java.util.concurrent.ConcurrentHashMap
 import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
 import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.internal.Logging
@@ -35,9 +36,10 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpre
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.util.{sideBySide, ArrayBasedMapData, GenericArrayData, MetadataColumnHelper}
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
-import org.apache.spark.sql.comet.{CometBatchScanExec, CometScanExec}
+import org.apache.spark.sql.comet.{CometBatchScanExec, CometScanExec, CometScanUtils}
 import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SparkPlan, SubqueryAdaptiveBroadcastExec}
 import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.execution.datasources.parquet.ParquetOptions
 import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
 import org.apache.spark.sql.internal.SQLConf
@@ -49,6 +51,7 @@ import org.apache.comet.CometSparkSessionExtensions.{isCometLoaded, isSpark35Plu
 import org.apache.comet.iceberg.{CometIcebergNativeScanMetadata, IcebergReflection}
 import org.apache.comet.objectstore.NativeConfig
 import org.apache.comet.parquet.CometParquetUtils.{encryptionEnabled, isEncryptionConfigSupported, readFieldId}
+import org.apache.comet.serde.SupportLevel
 import org.apache.comet.serde.operator.{CometIcebergNativeScan, CometNativeScan}
 import org.apache.comet.shims.{CometTypeShim, ShimCometStreaming, ShimFileFormat, ShimSubqueryBroadcast}
 
@@ -365,7 +368,45 @@ case class CometScanRule(session: SparkSession)
     if (!isSchemaSupported(scanExec, r)) {
       return None
     }
-    Some(CometScanExec(scanExec, session))
+    val cometScan = CometScanExec(scanExec, session)
+    val hasDate = SupportLevel.containsType(scanExec.requiredSchema, classOf[DateType])
+    // TIMESTAMP_NTZ values are never rebased by Spark, on write or on read (Spark's
+    // ParquetVectorUpdaterFactory: "TIMESTAMP_NTZ is a new data type and has no legacy files
+    // that need to do rebase"). The rebase question arises for a requested NTZ column only
+    // when the underlying Parquet column is a TIMESTAMP (LTZ or INT96) that may carry
+    // legacy-calendar values, and Comet permits that read only when
+    // COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ is true (Spark 4.x, SPARK-47447).
+    val hasTimestamp =
+      SupportLevel.containsType(scanExec.requiredSchema, classOf[TimestampType]) ||
+        (COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ &&
+          SupportLevel.containsType(scanExec.requiredSchema, classOf[TimestampNTZType]))
+    if ((hasDate || hasTimestamp) && COMET_SCAN_PARQUET_CHECK_DATETIME_REBASE.get()) {
+      val options = new ParquetOptions(r.options, conf)
+      val files = cometScan.selectedPartitions.iterator
+        .flatMap(_.files.iterator.map(f =>
+          CometScanUtils.ParquetFileInfo(f.getPath, f.getLen, f.getModificationTime)))
+        .toSeq
+      try {
+        if (CometScanUtils.requiresDatetimeRebase(
+            files,
+            hadoopConf,
+            options.datetimeRebaseModeInRead,
+            options.int96RebaseModeInRead,
+            hasDate,
+            hasTimestamp)) {
+          withFallbackReason(scanExec, "Native Parquet scan does not support datetime rebasing")
+          return None
+        }
+      } catch {
+        case NonFatal(e) =>
+          logWarning("Unable to inspect Parquet datetime rebase metadata", e)
+          withFallbackReason(
+            scanExec,
+            "Native Parquet scan could not verify datetime rebase metadata")
+          return None
+      }
+    }
+    Some(cometScan)
   }
 
   private def transformV2Scan(scanExec: BatchScanExec): SparkPlan = {

spark/src/main/scala/org/apache/comet/serde/operator/CometDataWritingCommand.scala

@@ -21,7 +21,7 @@ package org.apache.comet.serde.operator
 
 import scala.jdk.CollectionConverters._
 
-import org.apache.spark.SparkException
+import org.apache.spark.{SPARK_VERSION_SHORT, SparkException}
 import org.apache.spark.sql.comet.{CometEmptyRelationExec, CometNativeExec, CometNativeWriteExec, CometScanWrapper}
 import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.execution.adaptive.QueryStageExec
@@ -89,6 +89,10 @@ object CometDataWritingCommand extends CometOperatorSerde[DataWritingCommandExec
               return Unsupported(Some(s"Unsupported compression codec: $codec"))
             }
 
+            NativeWriteUtils
+              .legacyDatetimeRebaseWriteReason(cmd.query.output)
+              .foreach(reason => return Unsupported(Some(reason)))
+
             Incompatible(Some("Parquet write support is highly experimental"))
           case _ =>
             Unsupported(Some("Only Parquet writes are supported"))
@@ -135,6 +139,7 @@ object CometDataWritingCommand extends CometOperatorSerde[DataWritingCommandExec
         .newBuilder()
         .setOutputPath(outputPath)
         .setCompression(codec)
+        .setSparkVersion(SPARK_VERSION_SHORT)
         .addAllColumnNames(cmd.query.output.map(_.name).asJava)
         .addAllOutputSchema(schema2Proto(
           cmd.query.schema.fields.toIndexedSeq,

spark/src/main/scala/org/apache/comet/serde/operator/CometWriteFiles.scala

@@ -21,6 +21,7 @@ package org.apache.comet.serde.operator
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
+import org.apache.spark.SPARK_VERSION_SHORT
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.comet.{CometNativeExec, CometWriteFilesExec}
 import org.apache.spark.sql.execution.datasources.WriteFilesExec
@@ -102,6 +103,10 @@ object CometWriteFiles extends CometOperatorSerde[WriteFilesExec] {
       return Unsupported(Some(s"Unsupported compression codec: $codec"))
     }
 
+    NativeWriteUtils
+      .legacyDatetimeRebaseWriteReason(op.child.output)
+      .foreach(reason => return Unsupported(Some(reason)))
+
     Incompatible(Some("Parquet write support is highly experimental"))
   }
 
@@ -135,6 +140,7 @@ object CometWriteFiles extends CometOperatorSerde[WriteFilesExec] {
     val writerOpBuilder = OperatorOuterClass.ParquetWriter
       .newBuilder()
       .setCompression(codec)
+      .setSparkVersion(SPARK_VERSION_SHORT)
 
     // getSupportLevel already declined the write if the tag is absent, so this cannot be empty.
     outputPathOf(op).foreach { outputPath =>

spark/src/main/scala/org/apache/comet/serde/operator/NativeWriteUtils.scala

@@ -23,11 +23,13 @@ import java.util.Locale
 
 import org.apache.hadoop.fs.Path
 import org.apache.parquet.hadoop.ParquetOutputFormat
+import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DateType, TimestampType}
 
-import org.apache.comet.serde.OperatorOuterClass
+import org.apache.comet.serde.{OperatorOuterClass, SupportLevel}
 import org.apache.comet.serde.QueryPlanSerde.serializeDataType
 
 /**
@@ -186,6 +188,37 @@ object NativeWriteUtils {
           "spark.comet.parquet.write.enabled=false to write this table with Spark.")
     }
 
+  /**
+   * A fallback reason when the session asks for a LEGACY datetime rebase on write, or `None`.
+   *
+   * The native writer always writes proleptic Gregorian (corrected) datetime values and stamps
+   * `org.apache.spark.version` with no legacy markers. Honoring a LEGACY write rebase mode would
+   * require rebasing the values and stamping `org.apache.spark.legacyDateTime` /
+   * `org.apache.spark.legacyINT96`, so fall back to Spark rather than silently ignoring the
+   * requested mode and letting readers trust a "corrected" marker over legacy-intent data.
+   * TIMESTAMP_NTZ is exempt because Spark never rebases NTZ values on write.
+   */
+  def legacyDatetimeRebaseWriteReason(output: Seq[Attribute]): Option[String] = {
+    val hasDate = output.exists(a => SupportLevel.containsType(a.dataType, classOf[DateType]))
+    val hasTimestamp =
+      output.exists(a => SupportLevel.containsType(a.dataType, classOf[TimestampType]))
+    // Both write rebase mode configs default to EXCEPTION in all supported Spark versions.
+    def isLegacyWriteMode(key: String): Boolean =
+      SQLConf.get.getConfString(key, "EXCEPTION").toUpperCase(Locale.ROOT) == "LEGACY"
+    val legacyModeKeys =
+      ((if (hasDate || hasTimestamp) Seq(SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key)
+        else Seq.empty) ++
+        (if (hasTimestamp) Seq(SQLConf.PARQUET_INT96_REBASE_MODE_IN_WRITE.key)
+         else Seq.empty)).filter(isLegacyWriteMode)
+    if (legacyModeKeys.isEmpty) {
+      None
+    } else {
+      Some(
+        "Native Parquet write always writes corrected (proleptic Gregorian) datetime values " +
+          s"and does not support LEGACY rebase mode (${legacyModeKeys.mkString(", ")})")
+    }
+  }
+
   /** Compression codecs Comet's native Parquet writer can produce. */
   val supportedCompressionCodecs: Set[String] =
     Set("none", "uncompressed", "snappy", "lz4", "zstd", "gzip")