Repository navigation
Parquet: Fix nested initial default applied when ancestor struct is null - #17320
amogh-jahagirdar merged 7 commits into
Conversation
|
Spark/Flink tests will fail because their row oriented readers still need to be updated to have a similar fix as to what's done in BaseParquetReader. But I wnated to talk through the solution first with others. |
9a0a571 to
5adf95b
Compare
|
Maybe we should test based on the spec example:
If I understand correctly, the |
|
1048894 to
936efed
Compare
8e8ef58 to
4c6c948
Compare
| @@ -182,4 +184,154 @@ public void testTwoLevelList() throws IOException { | |||
| assertThat(Lists.newArrayList(reader)).hasSize(1); | |||
| } | |||
| } | |||
|
|
|||
| @Test | |||
There was a problem hiding this comment.
i'm a little skeptical this test is needed now, feels a bit duplicative
There was a problem hiding this comment.
I did keep a more "representative" test but dropped the other one; the rationale for keeping the more realistic test for generic reader was that even though all of the other tests share the same Parquet reader logic, there's still some logic in the different implementations which adapt to the engine record type (e.g. FlinkParquetReaders/ SparkParquetReaders) which compute things like readersById and invoke the newly added structFieldReader.
The Generic case is another implementation of that and we'd want to catch any bugs in that middle layer that may only apply for generic reader but don't impact the Spark/Flink ones.
77d74b3 to
cdba2fb
Compare
| @@ -129,12 +131,94 @@ public static Type fieldType(GroupType group, String name) { | |||
|
|
|||
| public static MessageType pruneColumns(MessageType fileSchema, Schema expectedSchema) { | |||
There was a problem hiding this comment.
This method is called from ParquetReadSupport.init which is used by parquet-mr. There the returned reader context returns the newly projected extra columns and these columns are returned if no createReaderFunc has been set.
If this is a valid, supported scenario (parquet-mr, no createReaderFunc), then we should fix it as the returned values will change.
This could help to repro in TestGenericData:
private static final Schema NESTED_MULTI_LEAF_SCHEMA =
new Schema(
Types.NestedField.required(1, "id", Types.LongType.get()),
Types.NestedField.optional(
2,
"nested",
Types.StructType.of(
Types.NestedField.required(3, "big", Types.StringType.get()),
Types.NestedField.required(4, "inner", Types.StringType.get()))));
private static final Schema NESTED_DEFAULT_ONLY_READ_SCHEMA =
new Schema(
Types.NestedField.required(1, "id", Types.LongType.get()),
Types.NestedField.optional("nested")
.withId(2)
.ofType(
Types.StructType.of(
Types.NestedField.optional("added")
.withId(9)
.ofType(Types.StringType.get())
.withInitialDefault(Literal.of("US"))
.build()))
.build());
@Test
public void testGenericReadDoesNotMaterializeUnprojectedColumns() throws IOException {
Types.StructType nestedType =
NESTED_MULTI_LEAF_SCHEMA.findField("nested").type().asStructType();
Record present = GenericRecord.create(NESTED_MULTI_LEAF_SCHEMA);
present.setField("id", 1L);
Record presentNested = GenericRecord.create(nestedType);
presentNested.setField("big", "b");
presentNested.setField("inner", "a");
present.setField("nested", presentNested);
Record nullNested = GenericRecord.create(NESTED_MULTI_LEAF_SCHEMA);
nullNested.setField("id", 2L);
nullNested.setField("nested", null);
OutputFile output = new InMemoryOutputFile();
try (FileAppender<Record> appender =
Parquet.write(output)
.schema(NESTED_MULTI_LEAF_SCHEMA)
.createWriterFunc(GenericParquetWriter::create)
.build()) {
appender.add(present);
appender.add(nullNested);
}
// the generic read path has no reader function and materializes the pruned projection directly
List<org.apache.avro.generic.GenericRecord> rows;
try (CloseableIterable<org.apache.avro.generic.GenericRecord> reader =
Parquet.read(output.toInputFile()).project(NESTED_DEFAULT_ONLY_READ_SCHEMA).build()) {
rows = Lists.newArrayList(reader);
}
assertThat(rows).hasSize(2);
for (org.apache.avro.generic.GenericRecord row : rows) {
org.apache.avro.generic.GenericRecord nested =
(org.apache.avro.generic.GenericRecord) row.get("nested");
if (nested != null) {
assertThat(nested.getSchema().getFields())
.extracting(org.apache.avro.Schema.Field::name)
.as("read records must not expose columns outside the projection")
.doesNotContain("big", "inner");
}
}
}
There was a problem hiding this comment.
I'm not sure this is a valid case, or at least a case we need to concern ourselves with in my opinion. The parquet read path where a reader function is not passed in, isn't invoked anywhere within the codebase and all the engines pass through a reader function. That path never supported default values let alone this specific handling of a default value case. Though it maybe a good idea in a separate PR to block reads through that path if there are initial default values and a reader function isn't specified, but considering it's been there and hasn't really been a practical issue I'd prefer to leave it for a separate PR>
There was a problem hiding this comment.
I think eventually we will have to make sure that the prune side and the read side use the same predicates. Otherwise we will just messing around and hoping 😄
There was a problem hiding this comment.
Yeah, the logic to choose the leaf is already shared but the conditions for does a given struct need a presence column wasn't shared. We'll need to plumb through readersById through structReader everywhere to unify that latter check. let me see how invasive that is first, if it is, maybe best to do that in a follow on
8858ea7 to
c4a182d
Compare
When a query projects only a nested field that has an initial default (or a metadata/partition constant) and drops the struct's real fields, PruneColumns reduces the struct to an empty group. That discards the file leaf columns whose definition levels carry the struct's per-row null-ness, so a null struct was incorrectly materialized as a struct of default values. To address this we solve this in two parts: 1. Retain 1 leaf that we probe for (even if it's not being projected) just so we have a definition level to work with 2. The default value constant reader uses that real leafs definition level to derive if the parent struct is null or not so that we surface the spec compliant results.
pvary caught this: the leaf we borrow to signal a constant-only struct's null-ness could sit behind a LIST/MAP, so it doesn't produce one value per struct occurrence and desyncs the reader. Share the selection logic between schema pruning and reader construction so they agree on the same leaf, restrict it to repetition-safe leaves, and fail loudly instead of silently falling back when none exists.
… the tracking of struct presence rather than individual field readers, addressing case where there isn't a default value
The presence column tracking whether a struct is present can now be a leaf nested inside a list or map: PresenceReader drains the row's repeated values, and selectPresenceColumn picks the shallowest leaf rather than failing when the struct has no flat leaf.
d7a198a to
7d7a95e
Compare
| return null; | ||
| } | ||
|
|
||
| return leafColumns(fileSchema, path).stream() |
There was a problem hiding this comment.
should we get the leafColumn from the caller?
There was a problem hiding this comment.
We still calculate the leafColumns(fileSchema, path) twice. Here and in struct(expected, fields)
There was a problem hiding this comment.
Ah sorry, I had AI try and address these nits and I think it missed this :( . Latest update should avoid recomputing the leaf columns, we just pass it through selectPresenceColumns
Inline PresenceReader construction instead of the single-caller withPresence helper, and make leafColumns private since it is only used within ParquetSchemaUtil. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01NmF9i2p9FfujkAogdhcDKc
…tReader SparkParquetReaders for v4.2 built struct readers inline instead of calling ParquetValueReaders.structReader, so it never got the presence column that makes a null struct's definition level visible when its only remaining leaf reads a constant (default or metadata). Every struct/list/map default and null-struct test failed against Parquet on 4.2 as a result; other engines and Spark versions were unaffected. Fix it to match v4.1 and route through the shared struct reader. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01NmF9i2p9FfujkAogdhcDKc
pvary
left a comment
pvary
left a comment
There was a problem hiding this comment.
Well done @amogh-jahagirdar!
Small nits, and +1.
We can get the others in follow-ups
Compute the leaf list once in struct() and pass it to selectPresenceColumn instead of recomputing it there. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01NmF9i2p9FfujkAogdhcDKc
|
Thanks for the great reviews @pvary , really helpful in catching subtle issues! I think this should be good to go in for the 1.12 release, so i'll merge. |
Drop apache#6264 and apache#6837, which are fixed on main. Scope the datetime rebase entry to the native Parquet scan and say the Delta scan rebases. Add the ignored Parquet writer properties from apache#5304 and note that the row group failure did not reproduce. Explain how apache/iceberg#17320 covers the identity partition source case. Reword several entries, break the list of Spark SQL tests into one bullet each, and note that spark.comet.scan.enabled is a testing setting.
…it log (apache#7052) * docs: list known Parquet and Iceberg correctness issues and add a data source audit log Add a compatibility page that lists the open issues in which a Parquet or Iceberg read or write gives a different result from Spark without falling back, with links to the issues, and point to it from the compatibility overview, the Parquet scan page, the Iceberg guide and the Iceberg writes page. The datetime rebasing item moves from scans.md to the new page. Correct the Iceberg writes page's claim that the float/double null-count difference cannot change IS NULL pruning: iceberg-java's counts are the wrong ones (apache/iceberg#17561). Start a Data Source Audits section in the contributor guide, next to the expression audits, recording the nested-type audits of the native Parquet and Iceberg scans and writers. * docs: add Parquet and Iceberg queries that fail where Spark succeeds Add a section to the known issues page for the open issues in which a Parquet or Iceberg read or write fails with an error where Spark, or iceberg-java, succeeds: Parquet and Iceberg reads, object store and credential settings that the native clients resolve differently from Hadoop, and memory failures in the native writers. The page becomes "Known Issues in Parquet and Iceberg Reads and Writes". The existing sections move under "Results that differ from Spark" and keep their heading anchors, and the links from the other pages use the new title. * docs: drop the Spark 4.1 NullType read limitation parquet-rs accepts the UNKNOWN logical type on any physical type since apache/arrow-rs#9855, which the parquet 59 crate includes, so reading a Spark 4.1 Parquet file with a NullType column no longer fails (apache#4199). * docs: address review feedback on the known issues page Drop apache#6264 and apache#6837, which are fixed on main. Scope the datetime rebase entry to the native Parquet scan and say the Delta scan rebases. Add the ignored Parquet writer properties from apache#5304 and note that the row group failure did not reproduce. Explain how apache/iceberg#17320 covers the identity partition source case. Reword several entries, break the list of Spark SQL tests into one bullet each, and note that spark.comet.scan.enabled is a testing setting.
When a query projects only a nested field that has an initial default and doesn't include the struct's real fields, PruneColumns reduces the struct to an empty group (which I think is correct, expected behavior for that abstraction). That discards the file leaf columns whose definition levels carry if the struct itself is null or not, so a null struct was incorrectly materialized as a struct of default values. in SQL
where struct is null but struct.field has a default value would surface the default value instead of null, which is not spec compliant.
To address this we solve this in two parts:
Retain 1 leaf that we probe for (even if it's not being projected) just so we have a definition level to work with. I don't think there's any other way to get definition levels because those are carried at the very leaf level of a struct.
The default value constant reader uses that real leafs definition level to derive if the parent struct is null or not so that we surface the spec compliant results.