fix: reject duplicate Parquet field names before decoding - #5786
Conversation
github-actions
Bot
added
bug
|
@andygrove created this pr to address your recent issue. When you have the time could you please check the provided solution? |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Reviewed 23efb2437d868916b313c4c2405bb98d26ce293d against 4eeb1f80f0541f72389a11e6e2d0ee269d648c23. I found no actionable correctness issue in this change.
The prior reader could decode byte-identical sibling names into multiplied rows or fail after its column readers lost synchronization, as reported in #5783. This adds a recursive physical-schema check in EagerPageIndexReader::get_metadata, after metadata retrieval and before Arrow schema construction and decoding. The locked DataFusion 55.0.0 / Parquet 59.3.0 sources confirm that cache hits still pass through this check. Decryption options and the existing page-index policy remain intact. Filter pushdown retains the factory. Files eliminated before metadata loading are never decoded.
On the maintained Spark 3.5 and 4.0 branches, case-sensitive name lookup selects the last identical sibling, case-insensitive lookup rejects multiple matches, and enabled field-ID lookup can resolve fields independently of names. This PR deliberately chooses the clear-error option accepted in #5783: it rejects duplicate physical names even if they are unprojected or have distinct IDs. The compatibility guide states this narrower behavior and the option to disable Comet. Unique sibling names are unaffected by this check. Case-distinct names are allowed here and remain subject to the existing case-insensitive ambiguity checks. Each group has its own name set, including nested LIST/MAP groups, so names in separate structs do not collide. Since validation precedes values, nulls, batch boundaries and numeric conversions cannot bypass it. Maintained Spark 3.4/4.1 source branches were unavailable. No source-level compatibility claim is made for those versions.
Validation
The 12 added cases cover two/three identical children, an additional distinct sibling, array elements and map values at batch sizes 1 and 4096, plus repeated reads, unprojected duplicates in both case modes, and a valid separate-group/case-distinct control. The failure cases assert a native scan and the specific new error. Repeated reads exercise the path but do not independently prove a cache hit. The cache guarantee follows from the inspected call chain.
The author reports 139 Scala tests and 18 encryption tests passing at 513d6fc26, plus native reader/cache and structural-narrowing checks. The reader factory, scan setup, regression suite and Cargo lock are unchanged between that commit and this head, but inherited timestamp-conversion changes make the overall trees different. Those reports are historical evidence. At the September 9, 10:30 UTC refresh, CI, CodeQL and the Delta gate were action_required. Only labeling had succeeded. Current product compilation/execution is therefore unverified. I ran source/whitespace checks, not a local product build or test.
Performance
The new work is an expected linear walk over physical schema nodes for each metadata request, using one HashSet per group and borrowed names. It adds no per-row or per-batch work, column copies, or object-store reads. Cache hits repeat this walk intentionally so cached metadata cannot bypass validation. Allocation depends on schema width and nesting. No benchmark was supplied or run, so this review does not claim a measured throughput improvement or quantify the cost for very wide schemas.
Design
The metadata boundary is the appropriate place to prevent this decoder failure: resolving names later in the schema adapter cannot undo rows already combined by decoding. Checking the entire physical schema also keeps the safety rule independent of projection and field-ID adaptation. This is a conservative compatibility tradeoff, explicitly documented, rather than an implementation of Spark's duplicate selection. The existing page-index factory already owns this metadata path, and both its module documentation and installation site now require preserving validation when that workaround is replaced. Future Spark-compatible selection would need safe duplicate handling before decoder construction. No additional abstraction is needed for this error-based fix.
Abstraction & complexity
The change adds one private recursive helper and reuses the existing Parquet error channel. A separate set per group directly expresses sibling uniqueness, without normalization or cross-group state. Tests extend the existing native-reader suite, and the two preservation comments explain the otherwise easy-to-miss lifetime of the guard. I found no actionable complexity or abstraction issue.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for picking this up. I checked the branch out locally, built it, and ran CometNativeReaderSuite (74 passed, plus the one pre-existing NullType cancel). To get a baseline I commented out the single validate_field_names call and rebuilt, which reproduces main's behavior for this path exactly, then ran the same probes against both builds.
The thing I keep coming back to is that the guard rejects the whole file regardless of what the query projects, so a query that returns the right answer today starts failing. On a file written as spark.range(3).selectExpr("id", "named_struct('dup', id, 'dup', id + 100) as s"):
| query | Spark | main | this branch |
|---|---|---|---|
spark.read.schema("id bigint") |
3 rows | 3 rows, correct | error |
same plus where id > 1000, so every row is pruned |
empty | empty | error |
Since an explicit read schema is the only way to read one of these files at all, one bad struct makes the entire file unreadable by Comet, and the only escape is turning Comet off for the query. I don't think that follows from #5783. I said a clear error was acceptable for the case that returns wrong results, not for queries that are correct today.
Would you consider scoping the walk to the subtree reachable from the required schema? The required schema is right there in init_datasource_exec, so the factory could be constructed with the folded top-level names and skip root children outside that set while still recursing fully into the selected ones. When use_field_id is set names don't identify the projection, so that case would keep the current whole-schema behavior. That still closes #5783 and leaves the currently-correct queries working.
Second thing. validate_field_names runs on root_schema(), so duplicates in the root group are one of the two branches it guards, but every new test builds its duplicate with named_struct and can only reach the nested branch. I said in the issue that top-level duplicates were unreachable, which is true of Spark's writer but not of Parquet, and this suite already has writeDirect at line 1036 for writing an arbitrary MessageType through a raw RecordConsumer. I tried it with
message spark_schema {
optional int64 a;
optional int64 a;
optional int64 b;
}
and a single row a=1, a=2, b=3. On main, reading schema("a bigint") returns two rows from a one-row file where Spark returns [1], and reading schema("b bigint") is correct on main but errors here. So this PR is also fixing a root-level wrong-results case that nothing currently asserts. Could you add it? A handful of Rust unit tests directly on validate_field_names would be cheap too, and would cover shapes Scala can't write: a LIST element group, a MAP key_value group, and same-name-in-separate-groups. I wrote six against this branch and they all pass in under a millisecond.
On the batch-size dimension, that was clearly load-bearing for your RED run, where 1 vs 4096 decided whether you got multiplied rows or a desync error. Now that the check fires in get_metadata before any decoder exists, both arms run identical code and assert the identical message. Would you swap those five duplicates for the root-group case above? Same test count, more of the function covered.
Dropping the #5783 link from the docs makes sense since this closes it, but could you file a follow-up for the Spark-compatible resolution and link that instead? The datetime rebasing entry just above links #5010 the same way, and as written the limitation reads as permanent with nowhere to track it. Worth capturing in that follow-up: matching Spark isn't one rule. On the two-a file above Spark resolved the root-level duplicate to the first child, while #5783 found last-wins for the nested case through caseSensitiveParquetFieldMap. That's a good argument for erroring first, which is what you've done.
Last, this needs a rebase and eager_page_index_reader_factory.rs has moved a lot underneath it, from 224 lines to about 1050 on main via the scan I/O metrics work (#5453) and the Variant projection work (#5794). get_metadata now binds the fetch as a Result, records metrics off it, unwraps with let metadata = metadata?;, and ends in if spark_variant_schema { with_spark_arrow_schema(metadata) } else { Ok(metadata) }. The validation wants to go straight after that unwrap and before the branch so both arms are covered. Please re-run the new suite after the merge, that placement is easy to get subtly wrong in a conflict resolution.
A few things I checked that are fine, so you don't have to. The factory is installed at the only production ParquetSource::new site, so every native scan is covered. Encrypted opens go through the same get_metadata. The error reaches the user with the file path attached, since Spark wraps it in FAILED_READ_FILE.NO_HINT, so there's no need to add the location to the message. And I measured the cost of the walk on a wide schema (1000 leaf fields, 20 files, every open a metadata cache hit): median 56.3ms without the validation against 57.2 to 59.3ms across three runs with it, which is inside the run-to-run noise. No perf concern.
23efb24 to
b696cc3
Compare
|
Thanks for the detailed review and reproductions. Addressed the requests in b696cc3 and rebased onto main, preserving the updated reader metrics and Variant handling.
Validation: reproduced both valid-projection failures before the fix. Afterward, the full Spark 4.1 native-reader suite passed 70 tests (one existing NullType cancellation), and all 8 focused cases passed on Spark 3.5. Rust Parquet tests: 188 passed, one existing ignored benchmark. Native build, whole-reactor packaging, all-target workspace Clippy with warnings denied, semantic/syntactic Scalafix, Spotless, formatting, and whitespace checks passed. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed b696cc31951c223d9d68a0768fb3958970d77753 against de1eb4f86c12af0895784c93e1f152c705f6ef0e. No new or remaining P1/P2 findings.
The update addresses the unprojected-column regression in the earlier review: name-based reads check required top-level roots with the existing case-folding rules, recurse through each selected subtree, and skip empty projections. Field-ID reads retain the documented whole-schema check. Validation runs after metadata retrieval and metrics recording, before the Variant branch and Arrow decoding; cached metadata follows the same path.
The new coverage includes raw root duplicates, unrelated columns, repeated reads, pruning, count-only reads, renamed field IDs, and Rust LIST/MAP/separate-group cases. The compatibility guide links #5884 for reader-dependent Spark resolution. The added work remains per metadata request; I did not run a performance benchmark.
At the September 12, 22:29 UTC refresh, CI had 56 successful and 10 skipped checks. I inspected logs confirming all eight duplicate-name cases passed on Spark 3.5 and Spark 4.1, plus all six new Rust cases. These jobs checked out merge commit 68fc20a8d75e524b3e5c80e550e7d5497692a02e; all four changed files and inspected supporting sources match the reviewed head. Five inherited base files make the complete trees different. No local product build was run. Canonical Spark source checks covered maintained 3.5/4.0 branches; maintained 3.4/4.1 branches were unavailable.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
The scoping to projected roots fixes the case I raised, but recursing into each selected root with projected_fields set to None still rejects a read that works today.
Take a file whose s group is dup, dup, other, read as spark.read.schema("s struct<other: bigint>"). Spark's clipParquetGroupFields iterates the requested fields only, so other resolves to a single child and the clipped read schema never mentions dup. I checked ParquetReadSupport on 3.4, 3.5, 4.0 and 4.1 and the matcher is the same on all of them. Comet gets there the same way today. is_pure_structural_narrowing returns true because other has exactly one folded match, replace_with_spark_cast leaves DataFusion's CastExpr in place, and build_projection_read_plan clips that cast down to the single other leaf, so the duplicate leaves are never decoded. That is the same leaf pruning the two issue #4859 tests in this suite assert. On this branch validate_field_names errors before any of it runs, and it holds in both case-sensitivity modes, so it is reachable under the default spark.sql.caseSensitive=false.
The comment on validate_field_names says nested projection does not safely separate duplicate leaves. That is true when the duplicate name is itself requested, because resolver_matches is then 2, is_pure_structural_narrowing returns false, and the whole root is decoded. It is not true when the duplicate is only a sibling of what was asked for. Would it make sense to carry the required schema down the recursion instead of dropping it at the root, and reject only when a requested field name matches more than one physical sibling? That is the rule Spark applies, and #5884 already describes the guard as covering selected ambiguous groups. A test for the shape that should keep working would be worth having too. A file written as named_struct('dup', id, 'dup', id + 100, 'other', id + 900) read back as s struct<other: bigint> returns the right answer on main and errors here, and nothing in the suite catches it.
This also lands on top of #5654, which is open and takes the opposite position on the same files. resolve_struct_mapping there resolves byte-identical siblings last-wins in case-sensitive mode, and shadowed_by_later_duplicate extends that to the root group. If this merges first none of that is reachable for a native Parquet scan, because get_metadata errors before the adapter runs. The two conflict textually as well. git merge-tree reports eight hunks in eager_page_index_reader_factory.rs and one in parquet_exec.rs, because #5654 hangs its own with_field_id_check validator off the same builder and the same get_metadata call site. Both are clean against main on their own, so neither CI run shows it. Your #5884 records that Spark 4.1.3 with the vectorized reader returned {0, 100, 1} for the nested fixture rather than last-wins, which argues against #5654's resolution as written. Could you and @dwsmith1983 settle an order, and note on #5654 whether its duplicate-name resolution should give way to the error here?
One more ordering point. fold_name and fold_schema_names become fallible in #5845, and this PR adds the only two new callers outside name_fold.rs. Both sit inside closures with nowhere to put an error. with_required_schema is a -> Self builder, and the fold_name call in validate_field_names is inside an is_some_and closure whose error type is ParquetError rather than DataFusionError. Whichever lands second will need to thread the Result through rather than reach for an unwrap.
|
The validator now preserves nested projection information and reuses the reader's structural-narrowing check. The regression reads the unique The nested-projection regression failed before the fix. A second regression with a real dictionary-encoded Arrow schema hint failed before the conservative hint guard. Final Orion verification passed: four focused Rust tests and the full For merge order, I suggest landing this decoder safety guard before #5654, then rebasing #5654 and preserving rejection until its last-wins path has evidence that ambiguous leaves decode correctly. #5845 is also still open; whichever lands second needs to propagate fallible name folding through the validator, projection decision, and shared structural-narrowing helper without |
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for the rework. The scoping does fix the case I raised, and I confirmed the pruned reads work: s struct<other>, a struct nested one level deeper, and s array<struct<other>> all return the right rows against main's answer, with and without a pushed filter and with rowFilterPushdown on. I also reproduced your run on the head: CometNativeReaderSuite 71 passed with the one NullType cancel, 190 Rust parquet tests, clippy and fmt clean.
Before anything else, eager_page_index_reader_factory.rs:596 starts with // ponytail:. That looks like a tooling marker rather than something meant for the file. It's the only occurrence in the repo.
The thing I can't get past is cost. The walk in the previous revision was cheap, but reconstructing the projection on every get_metadata is not. Both the physical.fields().iter().find(...) in the selected map and the projected.iter().find(...) inside validate_field_names are O(n²) with a fresh fold_name allocation on both sides of every comparison. Measured on a release build over a flat file with every column projected:
| leaf columns | previous revision | this revision |
|---|---|---|
| 100 | 1.6 µs | 363 µs |
| 250 | 3.3 µs | 2.2 ms |
| 500 | 6.7 µs | 8.8 ms |
| 1000 | 13.6 µs | 36.4 ms |
That is 36 ms per file open on a thousand-column table, on every open including metadata cache hits, which is the exact workload the factory's own doc comment is about. parquet_to_arrow_schema is only about 270 µs of it, so the conversion is fine and the matching loops are the problem. Could the cheap walk run first? validate_field_names(root, None, ..) is strictly stricter than the projection-aware call, since the projected version only ever skips sibling pairs the full walk also checks. So if the walk passes you can return immediately and never build selected at all, which puts the common path back at 13.6 µs and confines the expensive analysis to the rare file that actually has a duplicate. When you do need it, is_pure_structural_narrowing right next door already folds each name once and says why: "O(sources), not O(targets x sources), matching this file's bulk-fold convention."
Second, I want to revisit the field-ID case. I accepted whole-schema validation there last round, but that was on the premise that names can't identify the projection. Field IDs can, and Comet already resolves them in remap_physical_schema via id_to_phys_names. On your own root-duplicate fixture, reading renamed_b by field id 3 returns [3] in Spark and on main, and errors on this branch. Your test asserts that error two lines after asserting that the same b read by name works, so the same column in the same file succeeds or fails depending only on whether the conf is on. Could the walk be restricted to the IDs the required schema resolves, the same way it's restricted by name?
On validate_field_type, the Map and FixedSizeList arms permit pruning that DataFusion never performs. nested_schema_pruning::clip_type clips (Struct, Struct), (List, List) and (LargeList, LargeList), and its own comment says "maps, dictionaries, fixed-size lists, views, is kept wholesale". Those arms are unreachable today only because is_pure_structural_narrowing returns false for Map and requires exact equality for FixedSizeList, but projected_fields_skip_unselected_nested_duplicates asserts a pruned DataType::Map is fine, so the helper's contract now records map pruning as safe. If someone extends is_pure_structural_narrowing to maps later, which is a natural follow-up to #4859, the guard silently starts skipping duplicates the decoder will read. Dropping both arms costs nothing, since they fall through to _ => validate_field_names(schema, None, ..), which is what happens today anyway. While you're there, could you add a line at is_pure_structural_narrowing's definition noting the second caller? Its doc comment reads as a pure optimization allow list, and it's now load-bearing for correctness.
A few test and doc points. The array shape the guard newly permits has no end-to-end coverage. I checked and it does work, but the Rust unit test exercises the validator in isolation and would keep passing even if clip_type stopped clipping through a list. The Scala case would catch that. In duplicate Parquet field names outside a nested projection remain readable, val name = "other" is fixed inside the Seq(true, false) loop, so both iterations are identical and the case-sensitivity dimension isn't exercised. The shape that would exercise it fails: S struct<OTHER: bigint> under caseSensitive=false gives the duplicate error while Spark returns three rows, because is_pure_structural_narrowing needs an exact name match. That's not a regression, main gives StructArrayReader out of sync, but the loop reads as coverage it doesn't provide. Same for maps: a pruned s map<string, struct<other: bigint>> read errors while Spark returns rows, and nothing asserts it. That matters for the scans.md wording, which says the check covers "structs, arrays, and maps" and that "safely pruned nested fields are skipped" and "applies in both case-sensitivity modes". A user reads that as "select only the unique sibling and you're fine", and maps, case-differing read schemas, and field-ID reads all contradict it.
Last, and I realize this is late to raise, but could we talk about the layer? #5783 traces back to #5602, which replaced the unconditional assert_eq!(field_name_to_index_map.len(), from_fields.len()) in parquet_support.rs with a duplicate error gated on !parquet_options.case_sensitive, so the byte-identical case now falls through to indices[0]. I raised that on #5602 itself.
I don't think a revert is the answer. #5602 is the Unicode fold fix for #5495, so reverting brings back silent NULLs for a file column like MÜNCHEN read as münchen, and it introduced name_fold.rs, which is now used throughout parquet_exec.rs, parquet_support.rs and schema_adapter.rs, including the two call sites this PR adds. #5845 exists only to make those folds fallible, so a revert takes it with it, and six commits have landed on schema_adapter.rs since. It also wouldn't give us what we want, because the assert was a panic rather than an error and, living inside parquet_convert_struct_to_struct, it never covered the root group. No 1.0.x exposure either, since #5602 isn't on branch-1.0.
What #5602 really did was split one blunt unconditional check into a proper Spark-matching error for the case-insensitive half and nothing for the byte-identical half. So the narrow fix is to give the other half an error too. I tried exactly that: change the guard so a collision also errors when case-sensitive, worded so it doesn't claim case-insensitive mode, and disable this PR's validate_field_names call so the metadata layer behaves like main. All five nested shapes in #5783 error cleanly, including the array and map ones. Every pruned read keeps working with no projection reconstruction at all, including the field-ID read of renamed_b. CometNativeReaderSuite gives 64 passed, with the only failures being this PR's seven new tests. The appeal is that checking the struct the decoder actually produced gets projection-awareness for free, so there's no parquet_to_arrow_schema per open, no coupling to is_pure_structural_narrowing, no Arrow-schema-hint conservatism, and no field-ID false rejection.
The gap is the root group, and that one isn't #5602's doing. message spark_schema { optional int64 a=1; optional int64 a=2; optional int64 b=3; } read as a bigint still returns two rows where Spark returns [1], so it needs its own check, and remap_physical_schema already folds every root name so it would be O(n) there. I'd also want to confirm the adapter path is reached when the required type happens to equal the physical type and no cast is inserted. The PR description says rejecting after decoding is too late because the decoder has already combined the leaves, and that's right for resolving, but in every shape I tried the rejection fired before any row came back. Would you be willing to try that direction before we spend more rounds on the projection reconstruction?
On ordering, #5654 is still open and still resolves byte-identical siblings last-wins in resolve_struct_mapping, and git merge-tree still reports conflicts in eager_page_index_reader_factory.rs and parquet_exec.rs. #5845 is also still open. Worth settling with @dwsmith1983 before either lands. And CI hasn't run at this head, only the label job, so the green run in the earlier review was the previous commit.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 38b8fd162d4a21424ca9fcc22df90ae0b2baa18e against de1eb4f86c12af0895784c93e1f152c705f6ef0e, including the changes since b696cc31951c223d9d68a0768fb3958970d77753. The nested exact-name projection case is fixed: selecting s.other can skip unrequested duplicate siblings when the decoder performs the same structural narrowing. Schema hints and conversions retain the full-subtree check before decoding.
Two P2 points from Andy's current-head review remain:
- P2 — Quadratic projection matching on ordinary schemas. The selected-schema construction and validation each search one field list for every field in the other, folding both names on every comparison. An isolated probe of the exact Rust bodies counted 2,002,000
fold_namecalls for 1,000 unique ASCII columns, versus 1,000 per metadata request in the previous head, in both case modes. These calls return owned strings, and metadata-cache hits still run the validation. A cheap duplicate-free path and pre-folded/indexed lookups would avoid this per-file cost. - P2 — Unrelated duplicates still reject an unambiguous field-ID read. With
a(id=1), a(id=2), b(id=3), selectingrenamed_bby ID 3 disables projection filtering and fails on the unusedafields. The existing Scala test explicitly expects that failure. Maintained Spark 3.5/4.0 source selects the unique requested ID; Comet's ID remapping and the plain-column projection path can select that physical root without the duplicate roots. I no longer consider the documented whole-schema restriction sufficient justification for rejecting this case.
Both concerns are already covered in that review; I have no new inline findings. The local probe used exact function bodies with lightweight type doubles and ASCII names. It verifies operation counts and guard decisions, not decoder/JNI/Spark execution or wall-clock scan performance.
At the September 15, 03:29 UTC snapshot, only labeling had passed; four workflows were action_required. There is no current-head product-test execution evidence. The author's reported test results and the previous head's green CI are not independent validation of this head. Canonical Spark source checks covered maintained 3.5/4.0 branches; maintained 3.4/4.1 branches were unavailable.
#5654 now carries the narrow fix Andy sketched above: a requested nested field with identical siblings is refused in the resolver, an unrequested duplicate sibling stays readable, and exact root duplicates read the first column like Spark's reader, each with tests. #5845 is merged and #5654 threads its fallible folding through the adapter and the shared helpers. Given that, I would land #5654 first and rebase this one onto it for what remains, the map and list shapes and the metadata-time check if it is still wanted. Ordering thread is on #5654. |
|
Agreed on landing #5654 first. The remaining nested-only patch is verified in testing: 231 native Parquet tests, 6 native Iceberg tests, and 140 Scala tests passed (0 failures; existing skips remain); native build, root Maven verify, formatting, and all-target workspace Clippy also passed. Publication is awaiting #5654 landing so this PR does not import another author's unmerged history. These results describe the isolated candidate, not a newly pushed revision of this PR. The separate mixed INT64/INT32 duplicate-root finding is now tracked in #5964, with fixture source and Spark 4.1.3 reproduction. With Spark filter pushdown disabled, separate single-ID reads produce paired |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked unchanged head 38b8fd16 against de1eb4f8 after the new discussion. Both P2 findings from the previous review remain in the published code: quadratic projection matching on ordinary schemas, and rejection of an unambiguous field-ID read because unrelated roots have duplicate names. There are no new findings or duplicate inline comments.
David proposes landing #5654 first, and Erik agrees. That is a reasonable coordination plan. Afterward, publish the rebased #5786 so its remaining map/list and metadata-time checks can be assessed against the actual decoder path, including projection, field IDs and schema hints. This review does not verify #5654's implementation or treat the proposed sequencing as resolving either P2.
Erik explicitly identifies the reported 231 native Parquet, 6 native Iceberg and 140 Scala passes as results for an unpublished isolated candidate. They do not validate the current PR head. His separate mixed-type duplicate-root investigation also does not resolve the unrelated-column field-ID rejection above.
At September 15, 16:42 UTC, only labeling had run successfully. Four workflows remain action_required, with no product-test jobs. Source and canonical Spark 3.5/4.0 checks confirm the existing findings remain applicable. Maintained 3.4/4.1 sources are unavailable. No new runtime test or benchmark was run.
|
Dependency follow-up: the separate mixed-type investigation now has parquet-java PR apache/parquet-java#3796, with 112 targeted tests and style/package checks passing on its current master base. The nested-only candidate for this PR remains separate and unpublished: 231 native Parquet, 6 Iceberg, 140 Scala and 18 encryption tests passed. I will integrate it after #5654 lands, rerun checks on the actual merged tree, and then update this branch. These results are not CI results for the published PR head. |
|
Triage note: the "Spark-compatible duplicate-name resolution" you name as separate work is already open as #5654. It makes case-sensitive duplicate names resolve last-wins like Spark's Since one PR rejects reads and the other resolves them, it matters which cases each of you is claiming — there is at least one shape, case-sensitive duplicate top-level names, where rejecting here would pre-empt resolving there. Could you and @dwsmith1983 agree the split and the landing order? |
Under `spark.sql.parquet.fieldId.read.enabled` Spark resolves each requested field to the one Parquet field carrying its id, and raises FOUND_DUPLICATE_FIELD_IN_FIELD_ID_LOOKUP_MODE when more than one answers. Comet never looked at field ids, so a requested struct that repeats an id was read positionally and returned rows where Spark raises. The ids ride in the requested schema's `StructField.metadata`, so this is decidable from the plan. `DataTypeSupport` gains a `hasDuplicateFieldIds` predicate beside the existing duplicate-name one, and `CometScanTypeChecker` declines a struct that trips it when field id matching is on. The trait's own recursion carries the check into nested structs, arrays and maps, so the checker stays a shallow predicate. `ArrowCachedBatchSerializer.supportsType` also accepted a struct with duplicate child names, so caching such a relation stored it in Comet's Arrow format, which Java Arrow cannot import back because it keys struct children by name. One more schema check delegates it to Spark's default cache format, alongside the interval types already excluded there. No Parquet decoding changes; see apache#5786 for that path. Closes apache#5801.
|
checking this |
Under `spark.sql.parquet.fieldId.read.enabled` Spark resolves each requested field to the one Parquet field carrying its id, and raises FOUND_DUPLICATE_FIELD_IN_FIELD_ID_LOOKUP_MODE when more than one answers. Comet never looked at field ids, so a requested struct that repeats an id was read positionally and returned rows where Spark raises. The ids ride in the requested schema's `StructField.metadata`, so this is decidable from the plan. `DataTypeSupport` gains a `hasDuplicateFieldIds` predicate beside the existing duplicate-name one, and `CometScanTypeChecker` declines a struct that trips it when field id matching is on. The trait's own recursion carries the check into nested structs, arrays and maps, so the checker stays a shallow predicate. `ArrowCachedBatchSerializer.supportsType` also accepted a struct with duplicate child names, so caching such a relation stored it in Comet's Arrow format, which Java Arrow cannot import back because it keys struct children by name. One more schema check delegates it to Spark's default cache format, alongside the interval types already excluded there. No Parquet decoding changes; see apache#5786 for that path. Closes apache#5801.
Under `spark.sql.parquet.fieldId.read.enabled` Spark resolves each requested field to the one Parquet field carrying its id, and raises FOUND_DUPLICATE_FIELD_IN_FIELD_ID_LOOKUP_MODE when more than one answers. Comet never looked at field ids, so a requested schema that repeats one was read positionally and returned rows where Spark raises. The ids ride in the requested schema's `StructField.metadata`, so this is decidable from the plan. `DataTypeSupport` gains a `hasDuplicateFieldIds` predicate beside the existing duplicate-name one, and `CometScanTypeChecker` declines a field list that trips it. Both entry points are overridden, because they see different things. `isTypeSupported` is handed each field's data type, so it sees nested structs as the trait's recursion reaches them but can never compare two top-level fields; `isSchemaSupported` is where the schema's own field list is available. Each list is inspected exactly once, so nothing is re-walked at an enclosing level. `ArrowCachedBatchSerializer.supportsType` also accepted a struct with duplicate child names, so caching such a relation stored it in Comet's Arrow format, which Java Arrow cannot import back because it keys struct children by name. One more schema check delegates it to Spark's default cache format, alongside the interval types already excluded there, and the two copies of that predicate in the shuffle gate now call the shared one rather than spelling it out inline. No Parquet decoding changes; see apache#5786 for that path. Closes apache#5801.
|
@ErikBPF I didn't check tests, but I would expect the Comet falls back or at least fails in following scenarios:
I think most of cases following amazing work that @dwsmith1983 made in #5654 |
Reuse structural narrowing before pruning duplicate siblings. Keep full subtree validation when casts or schema hints change what the decoder reads.
Also make the Iceberg adapter test exercise case-sensitive mode, cover multi-file reads with and without mergeSchema, correct the compatibility note, and drop the ignored file-open timing harness.
235a16c to
f5e1dc5
Compare
|
Rebased onto Review points
Test matrix you listed
Compatibility note ( Validation (28-thread host, On ordering with #5654: the overlap is the referenced-column guard only, so I am fine landing after it and rebasing on top — happy to follow whatever order you prefer. |
|
Thanks for the rework. This is a much better shape than the metadata-time validator, and moving the byte-identical half of #5602's split into CI still has not run here. On Related to that, the new line on One cost question. A few smaller things. The message On the docs, "In case-sensitive mode Spark instead silently picks one sibling, so disable Comet for the query with an explicit read schema to use that resolution" is hard to follow, and I am not sure we want to point people at that resolution with much confidence. #5783 found last-wins for the nested case, the root-level probe I ran gave first, and your own #5964 has mixed-type roots returning anomalous values. Could the entry link #5964 alongside #5884 so a reader knows the fallback is not reliably one sibling either? Last, the description still describes the previous architecture. It says validation happens when loading native-reader metadata including cache hits, that the required schema is preserved through validation, and that field-ID reads validate the entire file schema. None of that is in this head, and that last one is the P2 this head fixed. The test section's 71 also predates the fourteen new cases, since the suite is at 85 now. The body lands as the commit message, so could you rewrite it against the current diff? |
Make the Spark inference assertion deterministic and exercise both Variant adapter paths. Centralize decoded-field checks and diagnostics, and avoid per-column index vectors when names do not collide. Clarify Spark fallback behavior and link the related issues. Refs apache#5783
|
Thanks for the detailed review. I pushed the follow-up in
Local native Parquet tests passed (202, with 5 ignored), as did the full Spark reader suite and formatting checks. New-head CI is pending. I could not add the test labels: GitHub denied |
|
Follow-up on |
| | DataType::FixedSizeList(field, _) | ||
| | DataType::ListView(field) | ||
| | DataType::LargeListView(field) | ||
| | DataType::Map(field, _) => { |
There was a problem hiding this comment.
why map.value is not checked?
There was a problem hiding this comment.
It is checked, one level down. The Map arm recurses into the entries field, which is a Struct of key and value, so the Struct arm walks both and recurses into the value's children. I confirmed it matters by dropping DataType::Map from that arm and rerunning. duplicate Parquet field names - non-pruning map fails clearly fails and nothing else does. The Rust map shape in issue_5783_physical_duplicates_survive_arrow_and_fail_before_output still passes without it, because it requests the duplicated name and match_struct_fields rejects it first.
andygrove
added
run-spark-4.1-tests
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for working through all of this. I rechecked everything from my last comment against 2d082c05. The schema-merge test passed on two more full-suite runs here after failing four times in five before, and both Variant sites are covered now. I also disabled each guard in turn, including the decoded-subtree check, its map arm, both Variant sites, the nested and root name checks, and the field-id root check. Every one has a test that fails without it. I've added the run-spark-4.1-tests, run-iceberg-tests and run-benchmark-check labels you asked for, and I'm happy with this once those come back green.
6 participants
Which issue does this PR close?
Closes #5783.
Rationale for this change
A Parquet struct can contain byte-identical sibling names. Decoding an ambiguous subtree before resolving those names can multiply rows. Reject selected ambiguity before decoding while preserving safe reads of unrelated columns and pruned children.
What changes are included in this PR?
match_struct_fieldsresolver to reject selected nested ambiguity in both plan-time checks and runtime conversion. Exact-name projections of unique children in structs and arrays of structs keep Parquet leaf pruning.How are these changes tested?
Local verification on Spark 4.1 / Scala 2.13 / JDK 17:
CometNativeReaderSuite: 85 succeeded, 0 failed, 1 existing cancellation (the NullType limitation tracked by Spark 4.1 NullType parquet: parquet-rs rejects BOOLEAN + Unknown logical type #4199 / SPARK-54220). The schema-merge case passed three times, including the full-suite run.issue_5783tests and the Variant adapter regression passed.git diff --checkpassed.These are local results for the follow-up diff. A real Parquet Variant file with duplicate physical children was not tested. The new PR head still needs CI, including Spark SQL 4.1, Iceberg and the benchmark check that runs all-target Clippy, before merge. No measured speedup is claimed.