Skip to content

PARQUET-651: Improve Avro's isElementType check. #352

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

Closed
rdblue wants to merge 2 commits into from
Closed

PARQUET-651: Improve Avro's isElementType check. #352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1efa248
PARQUET-651: Improve Avro's isElementType check.
rdblue Jul 11, 2016
ad9c1ee
PARQUET-651: Undo accidental default setting change.
rdblue Jul 13, 2016
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
23 changes: 17 additions & 6 deletions parquet-avro/src/main/java/org/apache/parquet/avro/AvroRecordConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,23 +42,28 @@
import org.apache.avro.Conversion;
import org.apache.avro.LogicalType;
import org.apache.avro.Schema;
import org.apache.avro.SchemaCompatibility;
import org.apache.avro.generic.GenericData;
import org.apache.avro.reflect.AvroIgnore;
import org.apache.avro.reflect.AvroName;
import org.apache.avro.reflect.AvroSchema;
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.Stringable;
import org.apache.avro.specific.SpecificData;
import org.apache.avro.util.ClassUtils;
import org.apache.parquet.Preconditions;
import org.apache.parquet.avro.AvroConverters.FieldStringConverter;
import org.apache.parquet.avro.AvroConverters.FieldStringableConverter;
import org.apache.parquet.filter2.predicate.SchemaCompatibilityValidator;
import org.apache.parquet.io.InvalidRecordException;
import org.apache.parquet.io.api.Converter;
import org.apache.parquet.io.api.GroupConverter;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.Type;

import static org.apache.avro.SchemaCompatibility.SchemaCompatibilityType.COMPATIBLE;
import static org.apache.avro.SchemaCompatibility.checkReaderWriterCompatibility;
import static org.apache.parquet.schema.Type.Repetition.REPEATED;
import static org.apache.parquet.schema.Type.Repetition.REQUIRED;

Expand Down Expand Up @@ -827,6 +832,14 @@ public void end() {
}
}

// Converter used to test whether a requested schema is a 2-level schema.
// This is used to convert the file's type assuming that the file uses
// 2-level lists and the result is checked to see if it matches the requested
// element type. This should always convert assuming 2-level lists because
// 2-level and 3-level can't be mixed.
private static final AvroSchemaConverter CONVERTER =
new AvroSchemaConverter(true);

@liancheng liancheng Jul 12, 2016

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.

Instead of always set AvroScehmaConverter.assumeRepeatedIsListElement to true, shall we pass the properly constructed AvroScehmaConverter from AvroReadSupport to AvroRecordMaterializer, and then here?

@liancheng liancheng Jul 12, 2016

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.

Oh I see your point, there's no need to do so.

/**
* Returns whether the given type is the element type of a list or is a
* synthetic group with one field that is the element type. This is
Expand All @@ -849,13 +862,11 @@ static boolean isElementType(Type repeatedType, Schema elementSchema) {
return true;
} else if (elementSchema != null &&
elementSchema.getType() == Schema.Type.RECORD) {
Set<String> fieldNames = new HashSet<String>();
for (Schema.Field field : elementSchema.getFields()) {
fieldNames.add(field.name());
Schema schemaFromRepeated = CONVERTER.convert(repeatedType.asGroupType());
if (checkReaderWriterCompatibility(elementSchema, schemaFromRepeated)
.getType() == COMPATIBLE) {
return true;
}
// The repeated type must be the element type because it matches the
// structure of the Avro element's schema.
return fieldNames.contains(repeatedType.asGroupType().getFieldName(0));
}
return false;
}
Expand Down
15 changes: 15 additions & 0 deletions parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,17 @@ public AvroSchemaConverter() {
this.writeOldListStructure = WRITE_OLD_LIST_STRUCTURE_DEFAULT;
}

/**
* Constructor used by {@link AvroRecordConverter#isElementType}, which always
* uses the 2-level list conversion.
*
* @param assumeRepeatedIsListElement whether to assume 2-level lists
*/
AvroSchemaConverter(boolean assumeRepeatedIsListElement) {
this.assumeRepeatedIsListElement = assumeRepeatedIsListElement;
this.writeOldListStructure = WRITE_OLD_LIST_STRUCTURE_DEFAULT;
}

public AvroSchemaConverter(Configuration conf) {
this.assumeRepeatedIsListElement = conf.getBoolean(
ADD_LIST_ELEMENT_RECORDS, ADD_LIST_ELEMENT_RECORDS_DEFAULT);
Expand Down Expand Up @@ -220,6 +231,10 @@ public Schema convert(MessageType parquetSchema) {
return convertFields(parquetSchema.getName(), parquetSchema.getFields());
}

Schema convert(GroupType parquetSchema) {
return convertFields(parquetSchema.getName(), parquetSchema.getFields());
}

private Schema convertFields(String name, List<Type> parquetFields) {
List<Schema.Field> fields = new ArrayList<Schema.Field>();
for (Type parquetType : parquetFields) {
Expand Down
148 changes: 147 additions & 1 deletion parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
import org.apache.avro.generic.IndexedRecord;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.schema.MessageType;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Ignore;
Expand All @@ -44,10 +46,13 @@

public class TestArrayCompatibility extends DirectWriterTest {

public static final Configuration OLD_BEHAVIOR_CONF = new Configuration();
public static final Configuration NEW_BEHAVIOR_CONF = new Configuration();

@BeforeClass
public static void setupNewBehaviorConfiguration() {
OLD_BEHAVIOR_CONF.setBoolean(
AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS, true);
NEW_BEHAVIOR_CONF.setBoolean(
AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS, false);
}
Expand Down Expand Up @@ -1035,16 +1040,157 @@ public void write(RecordConsumer rc) {
assertReaderContains(newBehaviorReader(test), newSchema, newRecord);
}

@Test
public void testListOfSingleElementStructsWithElementField()
throws Exception {
Path test = writeDirect(
"message ListOfSingleElementStructsWithElementField {" +
" optional group list_of_structs (LIST) {" +
" repeated group list {" +
" required group element {" +
" required float element;" +
" }" +
" }" +
" }" +
"}",
new DirectWriter() {
@Override
public void write(RecordConsumer rc) {
rc.startMessage();
rc.startField("list_of_structs", 0);

rc.startGroup();
rc.startField("list", 0); // start writing array contents

// write a non-null element
rc.startGroup(); // array level
rc.startField("element", 0);

// the inner element field
rc.startGroup();
rc.startField("element", 0);
rc.addFloat(33.0F);
rc.endField("element", 0);
rc.endGroup();

rc.endField("element", 0);
rc.endGroup(); // array level

// write a second non-null element
rc.startGroup(); // array level
rc.startField("element", 0);

// the inner element field
rc.startGroup();
rc.startField("element", 0);
rc.addFloat(34.0F);
rc.endField("element", 0);
rc.endGroup();

rc.endField("element", 0);
rc.endGroup(); // array level

rc.endField("list", 0); // finished writing array contents
rc.endGroup();

rc.endField("list_of_structs", 0);
rc.endMessage();
}
});

Schema structWithElementField = record("element",
field("element", primitive(Schema.Type.FLOAT)));

// old behavior - assume that the repeated type is the element type
Schema elementRecord = record("list",
field("element", structWithElementField));
Schema oldSchema = record("ListOfSingleElementStructsWithElementField",
optionalField("list_of_structs", array(elementRecord)));
GenericRecord oldRecord = instance(oldSchema,
"list_of_structs", Arrays.asList(
instance(elementRecord, "element",
instance(structWithElementField, "element", 33.0F)),
instance(elementRecord, "element",
instance(structWithElementField, "element", 34.0F))));

// check the schema
ParquetFileReader reader = ParquetFileReader
.open(new Configuration(), test);
MessageType fileSchema = reader.getFileMetaData().getSchema();
Assert.assertEquals("Converted schema should assume 2-layer structure",
oldSchema,
new AvroSchemaConverter(OLD_BEHAVIOR_CONF).convert(fileSchema));

// both should default to the 2-layer structure
assertReaderContains(oldBehaviorReader(test), oldSchema, oldRecord);

Schema newSchema = record("ListOfSingleElementStructsWithElementField",
optionalField("list_of_structs", array(structWithElementField)));
GenericRecord newRecord = instance(newSchema,
"list_of_structs", Arrays.asList(
instance(structWithElementField, "element", 33.0F),
instance(structWithElementField, "element", 34.0F)));

// check the schema
Assert.assertEquals("Converted schema should assume 3-layer structure",
newSchema,
new AvroSchemaConverter(NEW_BEHAVIOR_CONF).convert(fileSchema));
assertReaderContains(newBehaviorReader(test), newSchema, newRecord);

// check that this works with compatible nested schemas

Schema structWithDoubleElementField = record("element",
field("element", primitive(Schema.Type.DOUBLE)));

Schema doubleElementRecord = record("list",
field("element", structWithDoubleElementField));
Schema oldDoubleSchema = record(
"ListOfSingleElementStructsWithElementField",
optionalField("list_of_structs", array(doubleElementRecord)));
GenericRecord oldDoubleRecord = instance(oldDoubleSchema,
"list_of_structs", Arrays.asList(
instance(doubleElementRecord, "element",
instance(structWithDoubleElementField, "element", 33.0)),
instance(doubleElementRecord, "element",
instance(structWithDoubleElementField, "element", 34.0))));
assertReaderContains(oldBehaviorReader(test, oldDoubleSchema),
oldDoubleSchema, oldDoubleRecord);

Schema newDoubleSchema = record(
"ListOfSingleElementStructsWithElementField",
optionalField("list_of_structs", array(structWithDoubleElementField)));
GenericRecord newDoubleRecord = instance(newDoubleSchema,
"list_of_structs", Arrays.asList(
instance(structWithDoubleElementField, "element", 33.0),
instance(structWithDoubleElementField, "element", 34.0)));
assertReaderContains(newBehaviorReader(test, newDoubleSchema),
newDoubleSchema, newDoubleRecord);
}

public <T extends IndexedRecord> AvroParquetReader<T> oldBehaviorReader(
Path path) throws IOException {
return new AvroParquetReader<T>(path);
return new AvroParquetReader<T>(OLD_BEHAVIOR_CONF, path);
}

public <T extends IndexedRecord> AvroParquetReader<T> oldBehaviorReader(
Path path, Schema expectedSchema) throws IOException {
Configuration conf = new Configuration(OLD_BEHAVIOR_CONF);
AvroReadSupport.setAvroReadSchema(conf, expectedSchema);
return new AvroParquetReader<T>(conf, path);
}

public <T extends IndexedRecord> AvroParquetReader<T> newBehaviorReader(
Path path) throws IOException {
return new AvroParquetReader<T>(NEW_BEHAVIOR_CONF, path);
}

public <T extends IndexedRecord> AvroParquetReader<T> newBehaviorReader(
Path path, Schema expectedSchema) throws IOException {
Configuration conf = new Configuration(NEW_BEHAVIOR_CONF);
AvroReadSupport.setAvroReadSchema(conf, expectedSchema);
return new AvroParquetReader<T>(conf, path);
}

public <T extends IndexedRecord> void assertReaderContains(
AvroParquetReader<T> reader, Schema expectedSchema, T... expectedRecords)
throws IOException {
Expand Down