viirya commented on code in PR #7053:
URL: https://github.com/apache/datafusion-comet/pull/7053#discussion_r4304265598
##########
native/proto/src/proto/operator.proto:
##########
@@ -1016,6 +1016,11 @@ 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;
+ // The Spark version (`SPARK_VERSION_SHORT`) written into the footer's
key-value metadata as
+ // `org.apache.spark.version`, as Spark's own ParquetWriteSupport does.
Spark's datetime rebase
+ // logic reads it back to tell files written by Spark 3.0+ from older ones.
Absent in plans
Review Comment:
"Absent in plans serialized before the field was added" suggests plan
compatibility across Comet versions. But the JVM serde and the native planner
always ship together, and both serde paths now always set the field. Could the
comment just say the field is optional so native callers such as tests can omit
it? Otherwise a future reader may think cross-version plans are a contract to
preserve.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometWriteFiles.scala:
##########
@@ -135,6 +136,7 @@ object CometWriteFiles extends
CometOperatorSerde[WriteFilesExec] {
val writerOpBuilder = OperatorOuterClass.ParquetWriter
.newBuilder()
.setCompression(codec)
+ .setSparkVersion(SPARK_VERSION_SHORT)
Review Comment:
Same question as on `CometDataWritingCommand` for the 4.0+ path. The default
there is `CORRECTED`, but `LEGACY` set by the session or the write option still
needs the fallback.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometDataWritingCommand.scala:
##########
@@ -126,6 +127,7 @@ object CometDataWritingCommand extends
CometOperatorSerde[DataWritingCommandExec
.newBuilder()
.setOutputPath(outputPath)
.setCompression(codec)
+ .setSparkVersion(SPARK_VERSION_SHORT)
Review Comment:
Spark's writer only omits `org.apache.spark.legacyDateTime` when the
effective `datetimeRebaseModeInWrite` is not `LEGACY`. With `LEGACY`, Spark
writes rebased values plus that key and `org.apache.spark.timeZone`. The native
writer writes Gregorian values plus only the version key. Could
`getSupportLevel` here and in `CometWriteFiles` fall back when the effective
write mode (session conf or the `datetimeRebaseMode` write option) is `LEGACY`
and the schema has a date or timestamp column? Then the footer always means
what Spark's would.
On Spark 3.4 and 3.5 the default is `EXCEPTION`, where Spark refuses to
write a date before 1582-10-15 or a timestamp before 1900-01-01. The native
writer writes them. Could we check for ancient values natively, as the read
side already does for `EXCEPTION`, or fall back for date and timestamp columns
on that mode? If that belongs in a follow-up, please open an issue and link it
here.
##########
native/core/src/execution/operators/parquet_writer.rs:
##########
@@ -514,8 +530,15 @@ impl ExecutionPlan for ParquetWriterExec {
let part_file = self.output_path.clone();
// Configure writer properties
+ let key_value_metadata = self.spark_version.as_ref().map(|version| {
Review Comment:
Writing this key tells Spark's `getRebaseSpec` and Comet's
`datetime_rebase.rs` that the file's dates and INT64 timestamps need no rebase.
That holds only because this writer never rebases. Could you add a short
comment here stating that invariant? If the writer ever learns to honor
`LEGACY`, it then has to write `org.apache.spark.legacyDateTime` and
`org.apache.spark.timeZone` too, or readers will silently read shifted values.
##########
native/core/src/execution/operators/parquet_writer.rs:
##########
@@ -53,10 +53,13 @@ use futures::TryStreamExt;
use parquet::{
arrow::ArrowWriter,
basic::{Compression, GzipLevel, ZstdLevel},
- file::properties::WriterProperties,
+ file::{metadata::KeyValue, properties::WriterProperties},
};
use url::Url;
+/// Footer key under which Spark's `ParquetWriteSupport` records the Spark
version.
+const SPARK_VERSION_METADATA_KEY: &str = "org.apache.spark.version";
Review Comment:
`native/core/src/parquet/datetime_rebase.rs` already defines
`SPARK_VERSION_METADATA_KEY`. Could the writer reuse that one, made
`pub(crate)`, so the writer and the rebase reader cannot drift apart?
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -651,6 +652,31 @@ class CometParquetWriterSuite extends
CometParquetWriterTestBase {
}
}
+ test("parquet write records the Spark version in the footer") {
Review Comment:
This checks the footer string, but the reason for the key is the rebase
behavior. Would you consider adding a round trip that writes a pre-1582 date
natively and reads it back with
`spark.sql.parquet.datetimeRebaseModeInRead=LEGACY`, once with Comet's scan and
once with Spark's reader? Without this PR that read silently shifts the value.
So the test shows the user-facing benefit and also pins down the invariant from
the comment on `parquet_writer.rs`.
--
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]
