Handle primitive REPEATED field not contained in LIST annotated group #6649

zeevm · 2024-10-29T23:41:15Z

Which issue does this PR close?

Closes #6648

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

tustvold

Can we please get a test for this

tustvold · 2024-10-30T07:37:53Z

parquet/src/record/reader.rs

+            let reader = Reader::PrimitiveReader(field.clone(), Box::new(column));
+            if repetition == Repetition::REPEATED && path.len() == 1 {
+                if curr_def_level != 1 || curr_rep_level != 1 {
+                    return Err(ParquetError::General(format!("Top level REPEATED primitve field {} should have definition and repetition levels of 1", field.name())));


I'm not very familiar with this code, but AFAICT there's no reason we're necessarily at the top-level here

This should handle these cases (from the format ):

A repeated field that is neither contained by a LIST- or MAP-annotated group nor annotated by LIST or MAP should be interpreted as a required list of required elements where the element type is the type of the field.

Implementations should use either LIST and MAP annotations or unannotated repeated fields, but not both. When using the annotations, no unannotated repeated types are allowed.

IIUC we're talking about top level fields not contained in LIST groups, this is a legacy schema for lists.

Right but this could be contained in a nullable group for example

Right but this could be contained in a nullable group for example

In fact there's a test for this in parquet-testing (which is used by the test_tree_reader_handle_repeated_fields_with_no_annotation test)...repeated_no_annotation.parquet.

created by: parquet-rs version 0.3.0 (build b45ce7cba2199f22d93269c150d8a83916c69b5e) message user { REQUIRED INT32 id; OPTIONAL group phoneNumbers { REPEATED group phone { REQUIRED INT64 number; OPTIONAL BYTE_ARRAY kind (UTF8); } }

But parquet-rs handles this case, so perhaps it's just the top level unannotated list that needs special handling.

I just did a test of

message spark_schema { REPEATED group my_struct { REQUIRED INT32 key; OPTIONAL INT32 value; } }

and this also worked without the patch in this PR. A second test with

message spark_schema { REPEATED INT32 key; REPEATED INT32 value; }

fails without the patch.

test code

#[test] fn test_repeated_struct_no_annotation() { let schema = " message spark_schema { REPEATED group my_struct { REQUIRED INT32 key; OPTIONAL INT32 value; } } "; let schema = Arc::new(parse_message_type(schema).unwrap()); // Write Parquet file to buffer //let mut file = std::fs::File::create("/Users/seidl/test_struct.pq").unwrap(); let mut buffer: Vec<u8> = Vec::new(); let mut file_writer = SerializedFileWriter::new(&mut buffer, schema, Default::default()).unwrap(); let mut row_group_writer = file_writer.next_row_group().unwrap(); // Write column my_struct.key let mut column_writer = row_group_writer.next_column().unwrap().unwrap(); column_writer .typed::<Int32Type>() .write_batch( &[1, 2, 3, 4, 5, 6, 7, 8, 9], Some(&[1, 1, 1, 1, 1, 1, 1, 1, 1]), Some(&[0, 1, 1, 0, 1, 1, 0, 1, 1]), ) .unwrap(); column_writer.close().unwrap(); // Write column my_struct.value let mut column_writer = row_group_writer.next_column().unwrap().unwrap(); column_writer .typed::<Int32Type>() .write_batch( &[42, 7, 6, 99, 100, 2, 7, 9], Some(&[2, 2, 2, 2, 2, 2, 2, 1, 2]), Some(&[0, 1, 1, 0, 1, 1, 0, 1, 1]), ) .unwrap(); column_writer.close().unwrap(); // Finalize Parquet file row_group_writer.close().unwrap(); file_writer.close().unwrap(); assert_eq!(&buffer[0..4], b"PAR1"); // Read Parquet file from buffer let file_reader = SerializedFileReader::new(Bytes::from(buffer)).unwrap(); let rows: Vec<_> = file_reader .get_row_iter(None) .unwrap() .map(|row| row.unwrap()) .collect(); let expected_rows = vec![ row![( "my_struct".to_string(), list![ group![ ("key".to_string(), Field::Int(1)), ("value".to_string(), Field::Int(42)) ], group![ ("key".to_string(), Field::Int(2)), ("value".to_string(), Field::Int(7)) ], group![ ("key".to_string(), Field::Int(3)), ("value".to_string(), Field::Int(6)) ] ] )], row![( "my_struct".to_string(), list![ group![ ("key".to_string(), Field::Int(4)), ("value".to_string(), Field::Int(99)) ], group![ ("key".to_string(), Field::Int(5)), ("value".to_string(), Field::Int(100)) ], group![ ("key".to_string(), Field::Int(6)), ("value".to_string(), Field::Int(2)) ] ] )], row![( "my_struct".to_string(), list![ group![ ("key".to_string(), Field::Int(7)), ("value".to_string(), Field::Int(7)) ], group![ ("key".to_string(), Field::Int(8)), ("value".to_string(), Field::Null) ], group![ ("key".to_string(), Field::Int(9)), ("value".to_string(), Field::Int(9)) ] ] )], ]; assert_eq!(rows, expected_rows); } #[test] fn test_repeated_primitive_no_annotation() { let schema = " message spark_schema { REPEATED INT32 key; REPEATED INT32 value; } "; let schema = Arc::new(parse_message_type(schema).unwrap()); // Write Parquet file to buffer //let mut file = std::fs::File::create("/Users/seidl/test_struct.pq").unwrap(); let mut buffer: Vec<u8> = Vec::new(); let mut file_writer = SerializedFileWriter::new(&mut buffer, schema, Default::default()).unwrap(); let mut row_group_writer = file_writer.next_row_group().unwrap(); // Write column my_struct.key let mut column_writer = row_group_writer.next_column().unwrap().unwrap(); column_writer .typed::<Int32Type>() .write_batch( &[1, 2, 3, 4, 5, 6, 7, 8, 9], Some(&[1, 1, 1, 1, 1, 1, 1, 1, 1]), Some(&[0, 1, 1, 0, 1, 1, 0, 1, 1]), ) .unwrap(); column_writer.close().unwrap(); // Write column my_struct.value let mut column_writer = row_group_writer.next_column().unwrap().unwrap(); column_writer .typed::<Int32Type>() .write_batch( &[42, 7, 6, 99, 100, 2], Some(&[1, 1, 1, 1, 1, 1, 0]), Some(&[0, 1, 1, 0, 1, 1, 0]), ) .unwrap(); column_writer.close().unwrap(); // Finalize Parquet file row_group_writer.close().unwrap(); file_writer.close().unwrap(); assert_eq!(&buffer[0..4], b"PAR1"); // Read Parquet file from buffer let file_reader = SerializedFileReader::new(Bytes::from(buffer)).unwrap(); let rows: Vec<_> = file_reader .get_row_iter(None) .unwrap() .map(|row| row.unwrap()) .collect(); let expected_rows = vec![ row![ ( "key".to_string(), list![Field::Int(1), Field::Int(2), Field::Int(3)] ), ( "value".to_string(), list![Field::Int(42), Field::Int(7), Field::Int(6)] ), ], row![ ( "key".to_string(), list![Field::Int(4), Field::Int(5), Field::Int(6)] ), ( "value".to_string(), list![Field::Int(99), Field::Int(100), Field::Int(2)] ), ], row![ ( "key".to_string(), list![Field::Int(7), Field::Int(8), Field::Int(9)] ), ( "value".to_string(), list![] ), ], ]; assert_eq!(rows, expected_rows); }

yes, your first case is explicitly accounted for in the current implementation , see the comment there

The second case isn't implemented at the moment and requires the patch

etseidl

Given some testing, I think this is a good fix for a very specific problem. Thanks!

zeevm · 2024-10-31T14:47:17Z

@tustvold / @etseidl How do I trigger the checks again without commiting new changes? The test file I added is not merged into parquet-testing so the new UT should pass now

etseidl · 2024-10-31T14:49:44Z

I lack permissions. Perhaps @alamb can kick off the tests again.

zeevm · 2024-10-31T15:12:03Z

@tustvold / @etseidl My new test file is merged into parquet-testing, my local branch runs the UT successfully, but the CI fails on the UT not finding the file, any ideas why that is?

Is there some step required after commiting a new test file to update the test environment?

etseidl · 2024-10-31T15:32:52Z

I'll take a look in a little bit. It could be the CI is pegged to a specific parquet-testing commit.

etseidl · 2024-10-31T16:14:04Z

@zeevm I admit to not being an expert here. Looking at the error logs, CI is pulling an older commit. Pure speculation, but based on this I wonder if you need to do git submodule update --remote in your branch, create a new commit and push it. That may cause CI to pick up the recent changes.

Anyone with more git/github/workflow knowledge please chime in if I'm just embarrassing myself here 😅

Update: that seemed to work. See 916572f (after the submodule update, I staged everything under arrow-rs and committed). I still don't know if this is the right thing to do, but CI passed.

zeevm · 2024-10-31T17:39:54Z

@etseidl you suggestion works, and the UT now passes, can you merge?

etseidl · 2024-10-31T17:47:52Z

can you merge?

I'm afraid I cannot as I'm not a committer. I know committer bandwidth is very constrained at the moment, so it may take little while.

zeevm · 2024-10-31T18:24:06Z

@tustvold is this good for merge?

alamb

Thank you @zeevm and @etseidl -- this makes sense to me

alamb · 2024-10-31T20:05:58Z

parquet/src/record/reader.rs

@@ -138,7 +138,17 @@ impl TreeBuilder {
                .column_descr_ptr();
            let col_reader = row_group_reader.get_column_reader(orig_index)?;
            let column = TripletIter::new(col_descr, col_reader, self.batch_size);
-            Reader::PrimitiveReader(field, Box::new(column))
+            let reader = Reader::PrimitiveReader(field.clone(), Box::new(column));


I believe the changes in parquet-testing come from apache/parquet-testing#61 👍

alamb · 2024-10-31T20:09:02Z

parquet/src/record/reader.rs

@@ -1688,6 +1698,131 @@ mod tests {
        assert_eq!(rows, expected_rows);
    }

+    #[test]
+    fn test_tree_reader_handle_primitive_repeated_fields_with_no_annotation() {


I ran these tests without the change in this PR and it failed like this. Thus I conclude this test covers the code change

assertion `left == right` failed left: [Row { fields: [("Int32_list", Int(0)), ("String_list", Str("foo")), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", Int(0)), ("String_list_in_group", Str("foo"))] }))] }, Row { fields: [("Int32_list", Int(1)), ("String_list", Str("zero")), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", Int(1)), ("String_list_in_group", Str("zero"))] }))] }, Row { fields: [("Int32_list", Int(2)), ("String_list", Str("one")), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", Int(2)), ("String_list_in_group", Str("one"))] }))] }, Row { fields: [("Int32_list", Int(3)), ("String_list", Str("two")), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", Int(3)), ("String_list_in_group", Str("two"))] }))] }] right: [Row { fields: [("Int32_list", ListInternal(List { elements: [Int(0), Int(1), Int(2), Int(3)] })), ("String_list", ListInternal(List { elements: [Str("foo"), Str("zero"), Str("one"), Str("two")] })), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", ListInternal(List { elements: [Int(0), Int(1), Int(2), Int(3)] })), ("String_list_in_group", ListInternal(List { elements: [Str("foo"), Str("zero"), Str("one"), Str("two")] }))] }))] }, Row { fields: [("Int32_list", ListInternal(List { elements: [] })), ("String_list", ListInternal(List { elements: [Str("three")] })), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", ListInternal(List { elements: [] })), ("String_list_in_group", ListInternal(List { elements: [Str("three")] }))] }))] }, Row { fields: [("Int32_list", ListInternal(List { elements: [Int(4)] })), ("String_list", ListInternal(List { elements: [Str("four")] })), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", ListInternal(List { elements: [Int(4)] })), ("String_list_in_group", ListInternal(List { elements: [Str("four")] }))] }))] }, Row { fields: [("Int32_list", ListInternal(List { elements: [Int(5), Int(6), Int(7), Int(8)] })), ("String_list", ListInternal(List { elements: [Str("five"), Str("six"), Str("seven"), Str("eight")] })), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", ListInternal(List { elements: [Int(5), Int(6), Int(7), Int(8)] })), ("String_list_in_group", ListInternal(List { elements: [Str("five"), Str("six"), Str("seven"), Str("eight")] }))] }))] }] Left: [Row { fields: [("Int32_list", Int(0)), ("String_list", Str("foo")), ("group_of_lists", Group(Row { fields: [("Int32_list_in_group", Int(0)), ("String_list_in_group", Str("foo"))] }))] }, Row { fields: [("Int32_list", Int(1)), ("String_list", Str("ze ... Right: [Row { fields: [("Int32_list", ListInternal(List { elements: [Int(0), Int(1), Int(2), Int(3)] })), ("String_list", ListInternal(List { elements: [Str("foo"), Str("zero"), Str("one"), Str("two")] })), ("group_of_lists", Group(Row { fields: [("Int32_li ...

correct @alamb , are you good to merge this?

alamb · 2024-10-31T22:35:43Z

Yes, I plan to merge this PR tomorrow unless there are any more comments

alamb · 2024-11-02T10:30:28Z

🚀

alamb · 2024-11-02T10:30:36Z

Thanks again @zeevm and @etseidl

wgtmac · 2024-11-04T02:16:42Z

Sorry for the late reply. I'm not sure whether we should fix reading or totally prohibit writing repeated primitive fields without LIST annotation as a list type. This is a gray area from the spec and should be discussed prior to the fix. The file added to parquet-testing is for interoperability across parquet implementations to have the same behavior. I have opened apache/parquet-format#466 to clarify it.

alamb · 2024-11-08T18:27:57Z

Sorry for the late reply. I'm not sure whether we should fix reading or totally prohibit writing repeated primitive fields without LIST annotation as a list type. This is a gray area from the spec and should be discussed prior to the fix. The file added to parquet-testing is for interoperability across parquet implementations to have the same behavior. I have opened apache/parquet-format#466 to clarify it.

Than you @wgtmac -- given there are parquet files (such as what is in parquet-format) that do have this structure, I think supporting reads makes sense. If we decide it would be good to prohibit writers fro writing such files that sounds like a good follow on to me, but it still adds value to parquet-rs to read such files from my perspective.

Handle primitive REPEATED field not contained in LIST annotated group

0a41e02

github-actions bot added the parquet Changes to the parquet crate label Oct 29, 2024

cargo fmt

a9bad06

tustvold reviewed Oct 30, 2024

View reviewed changes

etseidl approved these changes Oct 30, 2024

View reviewed changes

zeevm mentioned this pull request Oct 31, 2024

Add repeated_primitive_no_list.parquet apache/parquet-testing#61

Merged

Ze'ev Maor added 3 commits October 31, 2024 15:11

Add UT

80606f0

cargo fmt

1c2f3f6

comment

c27bff3

clippy

5a5a972

clippy

f441b26

etseidl mentioned this pull request Oct 31, 2024

[DO NOT MERGE] Update parquet-testing submodule #6663

Closed

Ze'ev Maor added 2 commits October 31, 2024 19:21

update parquet-testing module

752116f

cargo fmt

d71a776

alamb approved these changes Oct 31, 2024

View reviewed changes

alamb merged commit 6859c87 into apache:master Nov 2, 2024
16 checks passed

alamb mentioned this pull request Nov 16, 2024

Primitive REPEATED fields not contained in LIST annotated groups aren't read as lists by record reader #6648

Closed

etseidl mentioned this pull request Dec 16, 2024

Support Repeated fields in Record APIs #2394

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Handle primitive REPEATED field not contained in LIST annotated group #6649

Handle primitive REPEATED field not contained in LIST annotated group #6649

zeevm commented Oct 29, 2024 •

edited

Loading

tustvold left a comment

tustvold Oct 30, 2024

zeevm Oct 30, 2024 •

edited

Loading

tustvold Oct 30, 2024

etseidl Oct 30, 2024

etseidl Oct 30, 2024

zeevm Oct 30, 2024

etseidl left a comment

zeevm commented Oct 31, 2024

etseidl commented Oct 31, 2024

zeevm commented Oct 31, 2024

etseidl commented Oct 31, 2024

etseidl commented Oct 31, 2024 •

edited

Loading

zeevm commented Oct 31, 2024

etseidl commented Oct 31, 2024

zeevm commented Oct 31, 2024

alamb left a comment

alamb Oct 31, 2024

alamb Oct 31, 2024

zeevm Oct 31, 2024

alamb commented Oct 31, 2024

alamb commented Nov 2, 2024

alamb commented Nov 2, 2024

wgtmac commented Nov 4, 2024

alamb commented Nov 8, 2024

Handle primitive REPEATED field not contained in LIST annotated group #6649

Handle primitive REPEATED field not contained in LIST annotated group #6649

Conversation

zeevm commented Oct 29, 2024 • edited Loading

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

tustvold left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zeevm Oct 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

etseidl left a comment

Choose a reason for hiding this comment

zeevm commented Oct 31, 2024

etseidl commented Oct 31, 2024

zeevm commented Oct 31, 2024

etseidl commented Oct 31, 2024

etseidl commented Oct 31, 2024 • edited Loading

zeevm commented Oct 31, 2024

etseidl commented Oct 31, 2024

zeevm commented Oct 31, 2024

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Oct 31, 2024

alamb commented Nov 2, 2024

alamb commented Nov 2, 2024

wgtmac commented Nov 4, 2024

alamb commented Nov 8, 2024

zeevm commented Oct 29, 2024 •

edited

Loading

zeevm Oct 30, 2024 •

edited

Loading

etseidl commented Oct 31, 2024 •

edited

Loading