Skip to content

fix: preserve Catalyst nullability and field IDs in native Parquet writes - #5369

Open
sunchao wants to merge 1 commit into
apache:mainfrom
sunchao:dev/chao/codex/comet-5305-parquet-schema-field-ids
Open

fix: preserve Catalyst nullability and field IDs in native Parquet writes#5369
sunchao wants to merge 1 commit into
apache:mainfrom
sunchao:dev/chao/codex/comet-5305-parquet-schema-field-ids

Conversation

@sunchao

@sunchao sunchao commented Aug 15, 2026

Copy link
Copy Markdown
Member

Why are the changes needed?

Closes #5305.

A Parquet write has two different schemas: the schema of the Arrow batches arriving at the native writer, and the schema Spark intends to persist in the Parquet file. The first describes how data is transported through execution. The second is the durable contract every future reader sees in the file footer. Comet currently treats them as the same schema, even though they are not.

The native writer receives data through a placeholder scan that marks top-level fields nullable and does not retain their Catalyst field metadata. It then builds the Parquet schema by renaming those incoming Arrow fields. The values are written successfully, but Spark's intended nullability and Parquet field IDs are already gone.

For example, suppose the Catalyst write schema is:

required_number: LONG NOT NULL   [field ID = 11]
optional_text:   STRING NULL     [field ID = 22]

Today, Comet writes a footer equivalent to:

message schema {
  optional int64 required_number;
  optional binary optional_text (STRING);
}

Spark expects:

message schema {
  required int64 required_number = 11;
  optional binary optional_text (STRING) = 22;
}

The file may still look correct when read back immediately by column name, which makes the problem easy to miss. However, required_number has silently become nullable, and both columns have lost the stable identities needed for schema evolution. For example, an evolved read schema should still identify the same columns after both a rename and a reorder:

Written schema:   original_number [11], original_text [22]
Requested schema: renamed_text [22],    renamed_number [11]

Without field IDs in the footer, Spark cannot recover that mapping. Its field-ID-aware reader fails by default with:

Spark read schema expects field Ids, but Parquet file schema doesn't contain any field Ids.

Setting spark.sql.parquet.fieldId.read.ignoreMissing=true avoids the exception but can return nulls instead, turning a visible compatibility failure into incorrect results.

Nested collections introduce another, less obvious version of the same problem. A schema using Delta-style column-mapping metadata can assign separate IDs to the synthetic fields inside lists and maps:

tags [200]
  element [201]

attrs [300]
  key   [301]
  value [302]

Because element, key, and value are not Catalyst StructFields, their IDs are stored on the nearest parent field under parquet.field.nested.ids. Copying only ordinary struct-field IDs would still produce incomplete Parquet schemas for these nested cases.

What changes were proposed in this PR?

This PR makes the Catalyst write schema, rather than the intermediate Arrow input schema, the source of truth for the Parquet file. At the Spark/native boundary, the existing writer operator now carries the complete target schema alongside the incoming data. The native side reconstructs that schema and uses it to initialize the Parquet writer, so the footer receives Spark's actual column names, required/optional annotations, and field IDs.

The conceptual change is:

Before: Arrow input schema -> rename columns -> Parquet file schema

After:  Catalyst target schema -> native writer schema -> Parquet file schema
        Arrow input batches ---------------------------> row data

The schema travels recursively rather than stopping at top-level columns. Ordinary nested struct fields retain their own IDs and nullability, while list-element and map-key/value IDs are recovered from the nearest parent field's Delta-style metadata. Relative collection paths are preserved through nesting and restart at each nested struct, matching how those IDs are represented in Catalyst.

Incoming Arrow batches can still use their existing execution-oriented field names and metadata. They are aligned with the target write schema without weakening validation of physical data types or nullability. Spark's spark.sql.parquet.fieldId.write.enabled setting remains authoritative: its default-enabled behavior preserves IDs, and explicitly disabling it suppresses IDs consistently at every nesting level. Older serialized plans without a target write schema retain their existing behavior.

The change applies directly to Comet's existing Spark 3.5/4.0 native Parquet writer. It preserves compatibility with Delta-shaped and Iceberg-style field-ID schemas, but does not claim to replace or accelerate Delta's separate transactional writer. It is independent of the writer-seam refactor in #5293.

How was this PR tested?

The new Scala regressions execute the actual CometNativeWriteExec path and inspect the footer of every generated Parquet part file. They verify required versus optional fields, top-level and nested IDs, list/map synthetic IDs, the default/explicitly enabled/disabled field-ID settings, and a real Spark read with both renamed and reordered columns. The read-back test disables Comet so the resulting file is interpreted by Spark's own field-ID reader.

The nullability regression was first reproduced against the unfixed writer, where it failed with OPTIONAL did not equal REQUIRED.

The complete Spark 3.5 native writer suite passes, including all three new regressions and existing coverage for complex types, compression, and save modes:

./mvnw -o -Pspark-3.5 test -Dtest=none \
  -Dsuites=org.apache.comet.parquet.CometParquetWriterSuite

Result: 33 tests passed.

The focused native Parquet regressions also pass under Spark 4.0:

./mvnw -o -Pspark-4.0 test -Dtest=none \
  '-Dsuites=org.apache.comet.parquet.CometParquetWriterSuite native parquet'

Result: 3 tests passed.

Rust coverage validates absent versus zero-valued field IDs, nested list/map metadata, preserved nullability, and a real written Parquet footer:

cargo test -p datafusion-comet execution::serde::tests --lib
cargo test -p datafusion-comet execution::operators::parquet_writer::tests --lib
cargo fmt --all -- --check

Results: 3 serde tests passed and 2 writer tests passed; four pre-existing HDFS integration tests remain ignored because they require a running HDFS cluster. Maven Spotless and Scalastyle checks also passed under both Spark profiles.

@sunchao
sunchao marked this pull request as ready for review August 15, 2026 20:44
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Native Parquet writer derives schema nullability and field IDs from Arrow rather than Catalyst

1 participant