andygrove commented on code in PR #7053:
URL: https://github.com/apache/datafusion-comet/pull/7053#discussion_r4455771397
##########
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:
Added a comment at the footer write in 9e402088f stating the invariant,
including that supporting LEGACY would also need to write
`org.apache.spark.legacyDateTime` and `org.apache.spark.timeZone`. The Scala
serde now also declines a LEGACY write rebase mode, so the invariant holds on
every path.
##########
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:
The writer now imports `SPARK_VERSION_METADATA_KEY` from
`datetime_rebase.rs` (made `pub(crate)`), and the Rust test uses it too. Fixed
in 9e402088f.
##########
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:
Moved the LEGACY check into this PR in 9e402088f:
`NativeWriteUtils.legacyDatetimeRebaseWriteReason`, copied from #5048 unchanged
so the two merge cleanly. `getSupportLevel` returns `Unsupported` when
`datetimeRebaseModeInWrite` (or `int96RebaseModeInWrite` for a timestamp
column) is `LEGACY` and the schema has a date or timestamp column. Only the
session conf is checked, per your correction. `EXCEPTION` stays with #7047.
Test: `parquet write falls back to Spark when the datetime write rebase mode is
LEGACY`.
##########
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 fix for the 4.0+ path, in 9e402088f.
##########
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:
Kept `optional` and the native side now skips an empty value, so the key is
not written. The proto comment says the field is optional so native callers can
omit it, with no cross-version wording. Because `jvm_output_stream` took field
10 on main, `spark_version` is now field 11. Fixed in 9e402088f.
##########
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:
Added `parquet write of a pre-1582 date reads back unchanged under LEGACY
read rebase` in 9e402088f. It writes dates starting at 1500-01-01 natively and
reads them back under `datetimeRebaseModeInRead=LEGACY`, once with Comet and
once with `COMET_ENABLED=false`.
--
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]
