You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2026/08/15 17:47:24 UTC
[PR] fix: preserve Catalyst nullability and field IDs in native Parquet writes [datafusion-comet]
sunchao opened a new pull request, #5369:
URL: https://github.com/apache/datafusion-comet/pull/5369
## Why are the changes needed?
Closes #5305.
Comet's existing native Parquet writer derives its output schema from the placeholder scan rather than the Catalyst write schema. The placeholder marks top-level columns nullable and drops their field metadata, so the generated Parquet footer records required columns as optional and omits the field IDs required by Spark, Delta column mapping, and Iceberg-compatible schemas. Delta also stores IDs for synthetic list elements and map keys/values separately on the nearest Catalyst struct field; losing those IDs breaks nested column mapping even when ordinary struct-field IDs are preserved.
## What changes were proposed in this PR?
- Carry the complete Catalyst target schema through the existing native Parquet writer protobuf and construct the Arrow/Parquet writer from that schema instead of the placeholder scan schema.
- Respect Spark's `spark.sql.parquet.fieldId.write.enabled` setting, including its default-enabled behavior and explicit disablement, for top-level and nested struct fields.
- Preserve Delta-style `parquet.field.nested.ids` metadata on synthetic list elements and map key/value fields, including nested collections and the nearest-struct-field path reset.
- Allow metadata-only differences between incoming Arrow batches and the target schema without weakening data-type or nullability validation.
- Add Rust footer/schema regression coverage and Spark write/read footer regressions that assert the native writer actually runs, verify required/optional repetitions and field IDs, and read renamed/reordered columns by ID.
This changes the existing Spark 3.5/4.0 writer path directly and does not depend on or replace the writer-seam refactor in #5293.
## How was this PR tested?
- Reproduced the pre-fix failure through the real native writer: `OPTIONAL did not equal REQUIRED` in the new Catalyst-schema footer regression.
- `cargo fmt --all -- --check`
- `cargo test -p datafusion-comet execution::serde::tests --lib` — 3 passing tests for absent/zero IDs, list/map metadata, and nested collections.
- `cargo test -p datafusion-comet execution::operators::parquet_writer::tests --lib` — 2 passing tests, including a real Parquet footer round trip; 4 existing HDFS-cluster tests remain ignored.
- `./mvnw -o -Pspark-3.5 test -Dtest=none -Dsuites=org.apache.comet.parquet.CometParquetWriterSuite` — all 33 native Parquet writer tests passed.
- `./mvnw -o -Pspark-4.0 test -Dtest=none '-Dsuites=org.apache.comet.parquet.CometParquetWriterSuite native parquet'` — all 3 new write/footer/read regressions passed.
- Maven Spotless and Scalastyle checks passed under both Spark profiles.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: preserve Catalyst nullability and field IDs in native Parquet writes [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #5369:
URL: https://github.com/apache/datafusion-comet/pull/5369#discussion_r3799830847
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -143,6 +145,168 @@ class CometParquetWriterSuite extends CometTestBase {
}
}
+ test(
+ "native parquet writer preserves Catalyst nullability and honors field ID write settings") {
+ val requiredMetadata = parquetFieldMetadata(11)
+ val optionalMetadata = parquetFieldMetadata(22)
+ val data = spark
+ .range(0, 3)
+ .select(
+ $"id".as("required_number", requiredMetadata),
+ when($"id" === 1L, $"id".cast(StringType)).as("optional_text", optionalMetadata),
+ $"id".as("unmapped_number"))
+
+ assert(!data.schema("required_number").nullable)
+ assert(data.schema("optional_text").nullable)
+
+ Seq(None, Some(true), Some(false)).foreach { configuredValue =>
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ withNativeWriter {
+ def writeAndVerify(): Unit = {
+ val plan = captureWritePlan(path => data.write.parquet(path), outputPath)
+ assertHasCometNativeWriteExec(plan)
+
+ val expectedIds = configuredValue.getOrElse(true)
+ assertParquetSchemas(outputPath) { schema =>
+ val root = schema.asGroupType()
+ val required = root.getType("required_number")
+ val optional = root.getType("optional_text")
+ val unmapped = root.getType("unmapped_number")
+
+ assert(required.getRepetition == Type.Repetition.REQUIRED)
+ assert(optional.getRepetition == Type.Repetition.OPTIONAL)
+ assert(
+ Option(required.getId).map(_.intValue()) ==
+ (if (expectedIds) Some(11) else None))
+ assert(
+ Option(optional.getId).map(_.intValue()) ==
+ (if (expectedIds) Some(22) else None))
+ assert(unmapped.getId == null)
+ }
+ }
+
+ configuredValue match {
+ case Some(enabled) =>
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> enabled.toString) {
+ writeAndVerify()
+ }
+ case None =>
+ assert(spark.conf.get(SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key).toBoolean)
+ writeAndVerify()
+ }
+ }
+ }
+ }
+ }
+
+ test("native parquet writer preserves nested and Delta collection field IDs") {
+ val detailsMetadata = parquetFieldMetadata(100)
+ val requiredChildMetadata = parquetFieldMetadata(101)
+ val optionalChildMetadata = parquetFieldMetadata(102)
+ val innerMetadata = parquetFieldMetadata(130, "inner.element" -> 131L)
+ val tagsMetadata = parquetFieldMetadata(200, "tags.element" -> 201L)
+ val attrsMetadata =
+ parquetFieldMetadata(300, "attrs.key" -> 301L, "attrs.value" -> 302L)
+
+ val data = spark
+ .range(0, 2)
+ .select(
+ struct(
+ $"id".as("required_child", requiredChildMetadata),
+ when($"id" === 1L, $"id").as("optional_child", optionalChildMetadata),
+ array($"id").as("inner", innerMetadata)).as("details", detailsMetadata),
+ array(when($"id" === 1L, $"id")).as("tags", tagsMetadata),
+ map($"id".cast(StringType), when($"id" === 1L, $"id")).as("attrs", attrsMetadata))
+
+ Seq(true, false).foreach { writeFieldIds =>
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ withNativeWriter {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> writeFieldIds.toString) {
+ val plan = captureWritePlan(path => data.write.parquet(path), outputPath)
+ assertHasCometNativeWriteExec(plan)
+
+ assertParquetSchemas(outputPath) { schema =>
+ val root = schema.asGroupType()
+
+ def assertField(field: Type, id: Int, nullable: Boolean): Unit = {
+ val expectedRepetition =
+ if (nullable) Type.Repetition.OPTIONAL else Type.Repetition.REQUIRED
+ assert(field.getRepetition == expectedRepetition)
+ assert(Option(field.getId).map(_.intValue()) ==
+ (if (writeFieldIds) Some(id) else None))
+ }
+
+ val details = root.getType("details")
+ assertField(details, 100, nullable = false)
+ val detailsGroup = details.asGroupType()
+ assertField(detailsGroup.getType("required_child"), 101, nullable = false)
+ assertField(detailsGroup.getType("optional_child"), 102, nullable = true)
+
+ val inner = detailsGroup.getType("inner")
+ assertField(inner, 130, nullable = false)
+ val innerList = inner.asGroupType().getType(0)
+ assert(innerList.getRepetition == Type.Repetition.REPEATED)
+ assertField(innerList.asGroupType().getType(0), 131, nullable = false)
+
+ val tags = root.getType("tags")
+ assertField(tags, 200, nullable = false)
+ val tagsList = tags.asGroupType().getType(0)
+ assert(tagsList.getRepetition == Type.Repetition.REPEATED)
+ assertField(tagsList.asGroupType().getType(0), 201, nullable = true)
+
+ val attrs = root.getType("attrs")
+ assertField(attrs, 300, nullable = false)
+ val entries = attrs.asGroupType().getType(0)
+ assert(entries.getRepetition == Type.Repetition.REPEATED)
+ val entriesGroup = entries.asGroupType()
+ assertField(entriesGroup.getType("key"), 301, nullable = false)
+ assertField(entriesGroup.getType("value"), 302, nullable = true)
+ }
+ }
+ }
+
+ checkAnswer(spark.read.parquet(outputPath), data)
+ }
+ }
+ }
+
+ test("Spark reads native parquet output by field ID after columns are renamed and reordered") {
Review Comment:
Added a field-ID read/write round trip covering a renamed nested struct field, a renamed field inside an `array<struct>`, and a renamed map alongside reordered top-level columns. The array element and map key/value IDs are preserved as well. Spark treats `element`, `key`, and `value` as structural names, so those fields cannot be renamed independently; their IDs are verified directly in the existing Parquet-footer regression.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: preserve Catalyst nullability and field IDs in native Parquet writes [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #5369:
URL: https://github.com/apache/datafusion-comet/pull/5369#discussion_r3799827567
##########
native/core/src/execution/operators/parquet_writer.rs:
##########
@@ -533,13 +542,17 @@ impl ExecutionPlan for ParquetWriterExec {
// Rename columns in the batch to match output schema
let renamed_batch = if !column_names.is_empty() {
- RecordBatch::try_new(Arc::clone(&schema_for_write), batch.columns().to_vec())
- .map_err(|e| {
- DataFusionError::Execution(format!(
- "Failed to rename batch columns: {}",
- e
- ))
- })?
+ // Collection field IDs exist on the target schema, not on arrays produced by
+ // the placeholder Scan. Ignore nested field names/metadata during validation
+ // while still checking data types and nested/top-level nullability.
Review Comment:
Yes. The placeholder scan and target schema both use the same Catalyst data types, and the scan normalizes incoming arrays before they reach the writer. `with_match_field_names(false)` ignores nested field names and metadata, but Arrow still recursively validates nullability for list elements, struct fields, and map entries. I clarified the comment and added `test_parquet_writer_rejects_mismatched_nested_nullability`, which verifies the writer rejects a nullable list element against a non-nullable target even when the actual data contains no nulls.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: preserve Catalyst nullability and field IDs in native Parquet writes [datafusion-comet]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #5369:
URL: https://github.com/apache/datafusion-comet/pull/5369#discussion_r3799521316
##########
native/core/src/execution/operators/parquet_writer.rs:
##########
@@ -533,13 +542,17 @@ impl ExecutionPlan for ParquetWriterExec {
// Rename columns in the batch to match output schema
let renamed_batch = if !column_names.is_empty() {
- RecordBatch::try_new(Arc::clone(&schema_for_write), batch.columns().to_vec())
- .map_err(|e| {
- DataFusionError::Execution(format!(
- "Failed to rename batch columns: {}",
- e
- ))
- })?
+ // Collection field IDs exist on the target schema, not on arrays produced by
+ // the placeholder Scan. Ignore nested field names/metadata during validation
+ // while still checking data types and nested/top-level nullability.
Review Comment:
Is the nested nullability guaranteed to match with the target schema here?
##########
spark/src/test/scala/org/apache/comet/parquet/CometParquetWriterSuite.scala:
##########
@@ -143,6 +145,168 @@ class CometParquetWriterSuite extends CometTestBase {
}
}
+ test(
+ "native parquet writer preserves Catalyst nullability and honors field ID write settings") {
+ val requiredMetadata = parquetFieldMetadata(11)
+ val optionalMetadata = parquetFieldMetadata(22)
+ val data = spark
+ .range(0, 3)
+ .select(
+ $"id".as("required_number", requiredMetadata),
+ when($"id" === 1L, $"id".cast(StringType)).as("optional_text", optionalMetadata),
+ $"id".as("unmapped_number"))
+
+ assert(!data.schema("required_number").nullable)
+ assert(data.schema("optional_text").nullable)
+
+ Seq(None, Some(true), Some(false)).foreach { configuredValue =>
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ withNativeWriter {
+ def writeAndVerify(): Unit = {
+ val plan = captureWritePlan(path => data.write.parquet(path), outputPath)
+ assertHasCometNativeWriteExec(plan)
+
+ val expectedIds = configuredValue.getOrElse(true)
+ assertParquetSchemas(outputPath) { schema =>
+ val root = schema.asGroupType()
+ val required = root.getType("required_number")
+ val optional = root.getType("optional_text")
+ val unmapped = root.getType("unmapped_number")
+
+ assert(required.getRepetition == Type.Repetition.REQUIRED)
+ assert(optional.getRepetition == Type.Repetition.OPTIONAL)
+ assert(
+ Option(required.getId).map(_.intValue()) ==
+ (if (expectedIds) Some(11) else None))
+ assert(
+ Option(optional.getId).map(_.intValue()) ==
+ (if (expectedIds) Some(22) else None))
+ assert(unmapped.getId == null)
+ }
+ }
+
+ configuredValue match {
+ case Some(enabled) =>
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> enabled.toString) {
+ writeAndVerify()
+ }
+ case None =>
+ assert(spark.conf.get(SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key).toBoolean)
+ writeAndVerify()
+ }
+ }
+ }
+ }
+ }
+
+ test("native parquet writer preserves nested and Delta collection field IDs") {
+ val detailsMetadata = parquetFieldMetadata(100)
+ val requiredChildMetadata = parquetFieldMetadata(101)
+ val optionalChildMetadata = parquetFieldMetadata(102)
+ val innerMetadata = parquetFieldMetadata(130, "inner.element" -> 131L)
+ val tagsMetadata = parquetFieldMetadata(200, "tags.element" -> 201L)
+ val attrsMetadata =
+ parquetFieldMetadata(300, "attrs.key" -> 301L, "attrs.value" -> 302L)
+
+ val data = spark
+ .range(0, 2)
+ .select(
+ struct(
+ $"id".as("required_child", requiredChildMetadata),
+ when($"id" === 1L, $"id").as("optional_child", optionalChildMetadata),
+ array($"id").as("inner", innerMetadata)).as("details", detailsMetadata),
+ array(when($"id" === 1L, $"id")).as("tags", tagsMetadata),
+ map($"id".cast(StringType), when($"id" === 1L, $"id")).as("attrs", attrsMetadata))
+
+ Seq(true, false).foreach { writeFieldIds =>
+ withTempPath { dir =>
+ val outputPath = new File(dir, "output.parquet").getAbsolutePath
+
+ withNativeWriter {
+ withSQLConf(SQLConf.PARQUET_FIELD_ID_WRITE_ENABLED.key -> writeFieldIds.toString) {
+ val plan = captureWritePlan(path => data.write.parquet(path), outputPath)
+ assertHasCometNativeWriteExec(plan)
+
+ assertParquetSchemas(outputPath) { schema =>
+ val root = schema.asGroupType()
+
+ def assertField(field: Type, id: Int, nullable: Boolean): Unit = {
+ val expectedRepetition =
+ if (nullable) Type.Repetition.OPTIONAL else Type.Repetition.REQUIRED
+ assert(field.getRepetition == expectedRepetition)
+ assert(Option(field.getId).map(_.intValue()) ==
+ (if (writeFieldIds) Some(id) else None))
+ }
+
+ val details = root.getType("details")
+ assertField(details, 100, nullable = false)
+ val detailsGroup = details.asGroupType()
+ assertField(detailsGroup.getType("required_child"), 101, nullable = false)
+ assertField(detailsGroup.getType("optional_child"), 102, nullable = true)
+
+ val inner = detailsGroup.getType("inner")
+ assertField(inner, 130, nullable = false)
+ val innerList = inner.asGroupType().getType(0)
+ assert(innerList.getRepetition == Type.Repetition.REPEATED)
+ assertField(innerList.asGroupType().getType(0), 131, nullable = false)
+
+ val tags = root.getType("tags")
+ assertField(tags, 200, nullable = false)
+ val tagsList = tags.asGroupType().getType(0)
+ assert(tagsList.getRepetition == Type.Repetition.REPEATED)
+ assertField(tagsList.asGroupType().getType(0), 201, nullable = true)
+
+ val attrs = root.getType("attrs")
+ assertField(attrs, 300, nullable = false)
+ val entries = attrs.asGroupType().getType(0)
+ assert(entries.getRepetition == Type.Repetition.REPEATED)
+ val entriesGroup = entries.asGroupType()
+ assertField(entriesGroup.getType("key"), 301, nullable = false)
+ assertField(entriesGroup.getType("value"), 302, nullable = true)
+ }
+ }
+ }
+
+ checkAnswer(spark.read.parquet(outputPath), data)
+ }
+ }
+ }
+
+ test("Spark reads native parquet output by field ID after columns are renamed and reordered") {
Review Comment:
Maybe also add coverage for nested types. Maybe a test that renames a struct field and a list element or map key/value?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: preserve Catalyst nullability and field IDs in native Parquet writes [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao merged PR #5369:
URL: https://github.com/apache/datafusion-comet/pull/5369
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org