From 1efa248a34c64ab880520e6b79df693c36b310af Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Mon, 11 Jul 2016 10:42:42 -0700 Subject: [PATCH 1/2] PARQUET-651: Improve Avro's isElementType check. The Avro implementation needs to check whether the read schema that is passed by the user (or automatically converted from the file schema) expects an extra 1-field layer to be returned, which matches the previous behavior of Avro when reading a 3-level list. Before this commit, the check was done by testing the structure of the expected list element type against the repeated group's schema. If they matched, then Avro assumed that the user expected an extra layer. However, for records that happened to match (1-field records with a field named "element") the check could be wrong and would cause exceptions later. This commit updates the check to convert the file's element schema to Avro and compare the compatibility of that schema with what was passed by the user. This checks the entire tree from the element down and gets the answer right based on the element and its children, not just the field names on the element. --- .../parquet/avro/AvroRecordConverter.java | 23 ++- .../parquet/avro/AvroSchemaConverter.java | 17 +- .../parquet/avro/TestArrayCompatibility.java | 148 +++++++++++++++++- 3 files changed, 180 insertions(+), 8 deletions(-) diff --git a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroRecordConverter.java b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroRecordConverter.java index 10bb29bc50..c0d6dc2a25 100644 --- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroRecordConverter.java +++ b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroRecordConverter.java @@ -42,9 +42,11 @@ 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; @@ -52,6 +54,7 @@ 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; @@ -59,6 +62,8 @@ 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; @@ -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); + /** * 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 @@ -849,13 +862,11 @@ static boolean isElementType(Type repeatedType, Schema elementSchema) { return true; } else if (elementSchema != null && elementSchema.getType() == Schema.Type.RECORD) { - Set fieldNames = new HashSet(); - 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; } diff --git a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java index 6b9b94cb65..44455fd7f6 100644 --- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java +++ b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java @@ -54,7 +54,7 @@ public class AvroSchemaConverter { public static final String ADD_LIST_ELEMENT_RECORDS = "parquet.avro.add-list-element-records"; - private static final boolean ADD_LIST_ELEMENT_RECORDS_DEFAULT = true; + private static final boolean ADD_LIST_ELEMENT_RECORDS_DEFAULT = false; private final boolean assumeRepeatedIsListElement; private final boolean writeOldListStructure; @@ -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); @@ -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 parquetFields) { List fields = new ArrayList(); for (Type parquetType : parquetFields) { diff --git a/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java b/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java index 29264f06a5..aa577ab579 100644 --- a/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java +++ b/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java @@ -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; @@ -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); } @@ -1035,9 +1040,143 @@ 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 AvroParquetReader oldBehaviorReader( Path path) throws IOException { - return new AvroParquetReader(path); + return new AvroParquetReader(OLD_BEHAVIOR_CONF, path); + } + + public AvroParquetReader oldBehaviorReader( + Path path, Schema expectedSchema) throws IOException { + Configuration conf = new Configuration(OLD_BEHAVIOR_CONF); + AvroReadSupport.setAvroReadSchema(conf, expectedSchema); + return new AvroParquetReader(conf, path); } public AvroParquetReader newBehaviorReader( @@ -1045,6 +1184,13 @@ public AvroParquetReader newBehaviorReader( return new AvroParquetReader(NEW_BEHAVIOR_CONF, path); } + public AvroParquetReader newBehaviorReader( + Path path, Schema expectedSchema) throws IOException { + Configuration conf = new Configuration(NEW_BEHAVIOR_CONF); + AvroReadSupport.setAvroReadSchema(conf, expectedSchema); + return new AvroParquetReader(conf, path); + } + public void assertReaderContains( AvroParquetReader reader, Schema expectedSchema, T... expectedRecords) throws IOException { From ad9c1ee36ff2bbe3917db75ce18759f96abaa8a7 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Wed, 13 Jul 2016 15:11:26 -0700 Subject: [PATCH 2/2] PARQUET-651: Undo accidental default setting change. --- .../main/java/org/apache/parquet/avro/AvroSchemaConverter.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java index 44455fd7f6..70b6525f60 100644 --- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java +++ b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java @@ -54,7 +54,7 @@ public class AvroSchemaConverter { public static final String ADD_LIST_ELEMENT_RECORDS = "parquet.avro.add-list-element-records"; - private static final boolean ADD_LIST_ELEMENT_RECORDS_DEFAULT = false; + private static final boolean ADD_LIST_ELEMENT_RECORDS_DEFAULT = true; private final boolean assumeRepeatedIsListElement; private final boolean writeOldListStructure;