Skip to content

[Parquet] Add support to read Parquet files written with old 2-level list structures. #3774

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 10 commits into from
Feb 1, 2022
Merged

[Parquet] Add support to read Parquet files written with old 2-level list structures. #3774

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
222f45e
Add support to read Parquet files written with old 2-level list struc…
SinghAsDev Dec 14, 2021
a84a545
Add tests sepcifically for old list written by parquet-avro and parqu…
SinghAsDev Jan 17, 2022
5ee2e4e
Updates to add support for reading 2 level lists to flink and pig
SinghAsDev Jan 25, 2022
d8f0fe8
Move getting list element to ParquetSchemaUtils
SinghAsDev Jan 25, 2022
b4f9c6a
Make ParquetTypeVisitor handle 2 level list in same way as TypeWithSc…
SinghAsDev Jan 25, 2022
65d624b
Remove leftover import
SinghAsDev Jan 25, 2022
9240f6e
Pull null checks up to fix NPE
SinghAsDev Jan 25, 2022
bf3d40b
Minor refactor
SinghAsDev Jan 30, 2022
78378de
Add test for two level parquet list read in iceberg generics
SinghAsDev Jan 31, 2022
c44c909
Revert back to previous version of visitThreeLevelList as repeatedEle…
SinghAsDev Jan 31, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.apache.flink.table.data.TimestampData;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.Schema;
import org.apache.iceberg.parquet.ParquetSchemaUtil;
import org.apache.iceberg.parquet.ParquetValueReader;
import org.apache.iceberg.parquet.ParquetValueReaders;
import org.apache.iceberg.parquet.TypeWithSchemaVisitor;
Expand Down Expand Up @@ -143,13 +144,12 @@ public ParquetValueReader<?> list(Types.ListType expectedList, GroupType array,
return null;
}

GroupType repeated = array.getFields().get(0).asGroupType();
String[] repeatedPath = currentPath();

int repeatedD = type.getMaxDefinitionLevel(repeatedPath) - 1;
int repeatedR = type.getMaxRepetitionLevel(repeatedPath) - 1;

Type elementType = repeated.getType(0);
Type elementType = ParquetSchemaUtil.getListElementType(array);
int elementD = type.getMaxDefinitionLevel(path(elementType.getName())) - 1;

return new ArrayReader<>(repeatedD, repeatedR, ParquetValueReaders.option(elementType, elementD, elementReader));
Expand Down
57 changes: 57 additions & 0 deletions flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,19 @@

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.Files;
import org.apache.iceberg.Schema;
import org.apache.iceberg.avro.AvroSchemaUtil;
import org.apache.iceberg.data.DataTest;
import org.apache.iceberg.data.RandomGenericData;
import org.apache.iceberg.data.Record;
Expand All @@ -35,11 +43,60 @@
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.types.Types;
import org.apache.parquet.avro.AvroParquetWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.junit.Assert;
import org.junit.Test;

import static org.apache.iceberg.types.Types.NestedField.optional;

public class TestFlinkParquetReader extends DataTest {
private static final int NUM_RECORDS = 100;

@Test
public void testTwoLevelList() throws IOException {
Schema schema = new Schema(
optional(1, "arraybytes", Types.ListType.ofRequired(3, Types.BinaryType.get())),
optional(2, "topbytes", Types.BinaryType.get())
);
org.apache.avro.Schema avroSchema = AvroSchemaUtil.convert(schema.asStruct());

File testFile = temp.newFile();
Assert.assertTrue(testFile.delete());

ParquetWriter<GenericRecord> writer = AvroParquetWriter.<GenericRecord>builder(new Path(testFile.toURI()))
.withDataModel(GenericData.get())
.withSchema(avroSchema)
.config("parquet.avro.add-list-element-records", "true")
.config("parquet.avro.write-old-list-structure", "true")
.build();

GenericRecordBuilder recordBuilder = new GenericRecordBuilder(avroSchema);
List<ByteBuffer> expectedByteList = new ArrayList();
byte[] expectedByte = {0x00, 0x01};
ByteBuffer expectedBinary = ByteBuffer.wrap(expectedByte);
expectedByteList.add(expectedBinary);
recordBuilder.set("arraybytes", expectedByteList);
recordBuilder.set("topbytes", expectedBinary);
GenericData.Record expectedRecord = recordBuilder.build();

writer.write(expectedRecord);
writer.close();

try (CloseableIterable<RowData> reader = Parquet.read(Files.localInput(testFile))
.project(schema)
.createReaderFunc(type -> FlinkParquetReaders.buildReader(schema, type))
.build()) {
Iterator<RowData> rows = reader.iterator();
Assert.assertTrue("Should have at least one row", rows.hasNext());
RowData rowData = rows.next();
Assert.assertArrayEquals(rowData.getArray(0).getBinary(0), expectedByte);
Assert.assertArrayEquals(rowData.getBinary(1), expectedByte);
Assert.assertFalse("Should not have more than one row", rows.hasNext());
}
}

private void writeAndValidate(Iterable<Record> iterable, Schema schema) throws IOException {
File testFile = temp.newFile();
Assert.assertTrue("Delete should succeed", testFile.delete());
Expand Down
3 changes: 1 addition & 2 deletions parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,13 +183,12 @@ public ParquetValueReader<?> list(Types.ListType expectedList, GroupType array,
return null;
}

GroupType repeated = array.getFields().get(0).asGroupType();
String[] repeatedPath = currentPath();

int repeatedD = type.getMaxDefinitionLevel(repeatedPath) - 1;
int repeatedR = type.getMaxRepetitionLevel(repeatedPath) - 1;

Type elementType = repeated.getType(0);
Type elementType = ParquetSchemaUtil.getListElementType(array);
int elementD = type.getMaxDefinitionLevel(path(elementType.getName())) - 1;

return new ParquetValueReaders.ListReader<>(repeatedD, repeatedR,
Expand Down
14 changes: 10 additions & 4 deletions parquet/src/main/java/org/apache/iceberg/parquet/ApplyNameMapping.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,11 +67,17 @@ public Type list(GroupType list, Type elementType) {
Preconditions.checkArgument(elementType != null,
"List type must have element field");

boolean isOldListElementType = ParquetSchemaUtil.isOldListElementType(list);
MappedField field = nameMapping.find(currentPath());
Type listType = Types.buildGroup(list.getRepetition())
.as(LogicalTypeAnnotation.listType())
.repeatedGroup().addFields(elementType).named(list.getFieldName(0))
.named(list.getName());

Types.GroupBuilder<GroupType> listBuilder = Types.buildGroup(list.getRepetition())
.as(LogicalTypeAnnotation.listType());
if (isOldListElementType) {
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated
listBuilder.addFields(elementType);
Comment thread
SinghAsDev marked this conversation as resolved.
} else {
listBuilder.repeatedGroup().addFields(elementType).named(list.getFieldName(0));
}
Type listType = listBuilder.named(list.getName());

return field == null ? listType : listType.withId(field.id());
}
Expand Down
7 changes: 1 addition & 6 deletions parquet/src/main/java/org/apache/iceberg/parquet/MessageTypeToType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,12 +96,7 @@ public Type struct(GroupType struct, List<Type> fieldTypes) {

@Override
public Type list(GroupType array, Type elementType) {
GroupType repeated = array.getType(0).asGroupType();
org.apache.parquet.schema.Type element = repeated.getType(0);

Preconditions.checkArgument(
!element.isRepetition(Repetition.REPEATED),
"Elements cannot have repetition REPEATED: %s", element);
org.apache.parquet.schema.Type element = ParquetSchemaUtil.getListElementType(array);

Integer elementFieldId = getId(element);

Expand Down
3 changes: 1 addition & 2 deletions parquet/src/main/java/org/apache/iceberg/parquet/ParquetAvroValueReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,13 +121,12 @@ public ParquetValueReader<?> struct(Types.StructType expected, GroupType struct,
@Override
public ParquetValueReader<?> list(Types.ListType expectedList, GroupType array,
ParquetValueReader<?> elementReader) {
GroupType repeated = array.getFields().get(0).asGroupType();
String[] repeatedPath = currentPath();

int repeatedD = type.getMaxDefinitionLevel(repeatedPath) - 1;
int repeatedR = type.getMaxRepetitionLevel(repeatedPath) - 1;

Type elementType = repeated.getType(0);
Type elementType = ParquetSchemaUtil.getListElementType(array);
int elementD = type.getMaxDefinitionLevel(path(elementType.getName())) - 1;

return new ListReader<>(repeatedD, repeatedR, ParquetValueReaders.option(elementType, elementD, elementReader));
Expand Down
59 changes: 59 additions & 0 deletions parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import java.util.function.Function;
import org.apache.iceberg.Schema;
import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
Expand Down Expand Up @@ -164,4 +165,62 @@ public Boolean primitive(PrimitiveType primitive) {
}
}

public static Type getListElementType(GroupType array) {
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated
Type repeated = array.getFields().get(0);
boolean isOldListElementType = ParquetSchemaUtil.isOldListElementType(array);
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated

Preconditions.checkArgument(isOldListElementType ||
repeated.asGroupType().getFieldCount() <= 1,
"Invalid list: repeated group is not a single field: %s", array);
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated

return isOldListElementType ? repeated : repeated.asGroupType().getType(0);
Comment thread
SinghAsDev marked this conversation as resolved.
}

// Parquet LIST backwards-compatibility rules.
// https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#backward-compatibility-rules
static boolean isOldListElementType(GroupType list) {
Type repeatedType = list.getFields().get(0);
String parentName = list.getName();

return
// For legacy 2-level list types with primitive element type, e.g.:
//
// // ARRAY<INT> (nullable list, non-null elements)
// optional group my_list (LIST) {
// repeated int32 element;
// }
//
repeatedType.isPrimitive() ||
// For legacy 2-level list types whose element type is a group type with 2 or more fields,
// e.g.:
//
// // ARRAY<STRUCT<str: STRING, num: INT>> (nullable list, non-null elements)
// optional group my_list (LIST) {
// repeated group element {
// required binary str (UTF8);
// required int32 num;
// };
// }
//
repeatedType.asGroupType().getFieldCount() > 1 ||
// For legacy 2-level list types generated by parquet-avro (Parquet version < 1.6.0), e.g.:
//
// // ARRAY<STRUCT<str: STRING>> (nullable list, non-null elements)
// optional group my_list (LIST) {
// repeated group array {
// required binary str (UTF8);
// };
// }
repeatedType.getName().equals("array") ||
// For Parquet data generated by parquet-thrift, e.g.:
//
// // ARRAY<STRUCT<str: STRING>> (nullable list, non-null elements)
// optional group my_list (LIST) {
// repeated group my_list_tuple {
// required binary str (UTF8);
// };
// }
//
repeatedType.getName().equals(parentName + "_tuple");
}
}
9 changes: 4 additions & 5 deletions parquet/src/main/java/org/apache/iceberg/parquet/ParquetTypeVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,17 +66,16 @@ private static <T> T visitList(GroupType list, ParquetTypeVisitor<T> visitor) {
Preconditions.checkArgument(list.getFieldCount() == 1,
"Invalid list: does not contain single repeated field: %s", list);

GroupType repeatedElement = list.getFields().get(0).asGroupType();
Type repeatedElement = list.getFields().get(0);
Comment thread
SinghAsDev marked this conversation as resolved.
Preconditions.checkArgument(repeatedElement.isRepetition(Type.Repetition.REPEATED),
"Invalid list: inner group is not repeated");
Preconditions.checkArgument(repeatedElement.getFieldCount() <= 1,
"Invalid list: repeated group is not a single field: %s", list);

Type elementField = ParquetSchemaUtil.getListElementType(list);

visitor.beforeRepeatedElement(repeatedElement);
try {
T elementResult = null;
if (repeatedElement.getFieldCount() > 0) {
Type elementField = repeatedElement.getType(0);
if (repeatedElement.isPrimitive() || repeatedElement.asGroupType().getFieldCount() > 0) {

@rdblue rdblue Jan 25, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The changes to this class are inconsistent with the changes to the TypeWithSchemaVisitor. Here, the repeated element is always visited (beforeRepeatedElement call above) and may be processed again as the element. The other avoids pushing the name on the stack. If beforeRepeatedElement were used to track names, I think it would get a duplicate name for the repeated group.

@SinghAsDev SinghAsDev Jan 25, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, addressed this. It would be nice to have some tests to check this behavior. But, I don't think we need to block on that, unless you disagree.

visitor.beforeElementField(elementField);
try {
elementResult = visit(elementField, visitor);
Expand Down
11 changes: 8 additions & 3 deletions parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,15 +108,20 @@ public Type struct(GroupType struct, List<Type> fields) {

@Override
public Type list(GroupType list, Type element) {
GroupType repeated = list.getType(0).asGroupType();
Type originalElement = repeated.getType(0);
Type repeated = list.getType(0);
boolean isOldListElementType = ParquetSchemaUtil.isOldListElementType(list);
Type originalElement = ParquetSchemaUtil.getListElementType(list);
Integer elementId = getId(originalElement);

if (elementId != null && selectedIds.contains(elementId)) {
return list;
} else if (element != null) {
if (!Objects.equal(element, originalElement)) {
return list.withNewFields(repeated.withNewFields(element));
if (isOldListElementType) {
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated
return list.withNewFields(element);
} else {
return list.withNewFields(repeated.asGroupType().withNewFields(element));
}
}
return list;
}
Expand Down
69 changes: 49 additions & 20 deletions parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,29 +63,14 @@ public static <T> T visit(org.apache.iceberg.types.Type iType, Type type, TypeWi
Preconditions.checkArgument(group.getFieldCount() == 1,
"Invalid list: does not contain single repeated field: %s", group);

GroupType repeatedElement = group.getFields().get(0).asGroupType();
Type repeatedElement = group.getFields().get(0);
Preconditions.checkArgument(repeatedElement.isRepetition(Type.Repetition.REPEATED),
"Invalid list: inner group is not repeated");
Preconditions.checkArgument(repeatedElement.getFieldCount() <= 1,
"Invalid list: repeated group is not a single field: %s", group);

Types.ListType list = null;
Types.NestedField element = null;
if (iType != null) {
list = iType.asListType();
element = list.fields().get(0);
}

visitor.fieldNames.push(repeatedElement.getName());
try {
T elementResult = null;
if (repeatedElement.getFieldCount() > 0) {
elementResult = visitField(element, repeatedElement.getType(0), visitor);
}

return visitor.list(list, group, elementResult);
} finally {
visitor.fieldNames.pop();
if (ParquetSchemaUtil.isOldListElementType(group)) {
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated
return visitTwoLevelList(iType, visitor, group, repeatedElement);
} else {
return visitThreeLevelList(iType, visitor, group, repeatedElement);
}

case MAP:
Expand Down Expand Up @@ -149,6 +134,50 @@ public static <T> T visit(org.apache.iceberg.types.Type iType, Type type, TypeWi
}
}

private static <T> T visitTwoLevelList(
org.apache.iceberg.types.Type iType,
TypeWithSchemaVisitor<T> visitor,
GroupType group,
Type repeatedElement) {
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated
Types.ListType list = null;
Types.NestedField element = null;
if (iType != null) {
list = iType.asListType();
Comment thread
SinghAsDev marked this conversation as resolved.
Outdated
element = list.fields().get(0);
}

@rdblue rdblue Jan 28, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Style: extra newline. Also, we either wrap argument lists at the same level or start all arguments on the next line at 2 indents.

@SinghAsDev SinghAsDev Jan 30, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the style used in various parts of code are different. For example, IIUC ParquetReadSupport.prepareForRead is different than what you are saying. Earlier you also had mentioned Iceberg does not use new param at new line pattern. Updating this part to keep the same level (align params start with previous line) and wrap.

Let me know which style we should try to follow and I can try to update the intellij-style that we provide with Iceberg repo accordingly. I don't know if it is possible, but I can try.

@rdblue rdblue Jan 30, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Don't worry about where still is incorrect in other places. We'll eventually track those down and fix them, but we do want to keep style from diverging when it is caught by a review. So please do fix this.

@SinghAsDev SinghAsDev Jan 30, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It should already be fixed. Does the update still have style issue?

T elementResult = visitField(element, repeatedElement, visitor);
return visitor.list(list, group, elementResult);
}

private static <T> T visitThreeLevelList(
org.apache.iceberg.types.Type iType,
TypeWithSchemaVisitor<T> visitor,
GroupType group,
Type repeatedElement) {
Preconditions.checkArgument(repeatedElement.asGroupType().getFieldCount() <= 1,
"Invalid list: repeated group is not a single field: %s", group);

Types.ListType list = null;
Types.NestedField element = null;
if (iType != null) {
list = iType.asListType();
element = list.fields().get(0);
}

visitor.fieldNames.push(repeatedElement.getName());
try {
T elementResult = null;
if (repeatedElement.asGroupType().getFieldCount() > 0) {
elementResult = visitField(element, repeatedElement.asGroupType().getType(0), visitor);
}

return visitor.list(list, group, elementResult);
} finally {
visitor.fieldNames.pop();
}
}

private static <T> T visitField(Types.NestedField iField, Type field, TypeWithSchemaVisitor<T> visitor) {
visitor.fieldNames.push(field.getName());
try {
Expand Down
Loading