From b4e4cc01332b39a902c67b5c0dd7db686c8bd202 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Tue, 10 Mar 2015 12:04:07 -0700 Subject: [PATCH 1/8] PARQUET-212: Add DirectWriterTest base class. This adds convenience methods for writing to files using the RecordConsumer API directly. This is useful for mimicing files from other writers for compatibility tests. --- parquet-avro/pom.xml | 7 ++ .../parquet/avro/TestArrayCompatibility.java | 76 +------------ parquet-hadoop/pom.xml | 11 ++ .../org/apache/parquet/DirectWriterTest.java | 102 ++++++++++++++++++ parquet-thrift/pom.xml | 7 ++ 5 files changed, 129 insertions(+), 74 deletions(-) create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/DirectWriterTest.java diff --git a/parquet-avro/pom.xml b/parquet-avro/pom.xml index 535bb85546..99764362dd 100644 --- a/parquet-avro/pom.xml +++ b/parquet-avro/pom.xml @@ -81,6 +81,13 @@ test-jar test + + org.apache.parquet + parquet-hadoop + ${project.version} + test-jar + test + 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 c4585a710e..2676f71e52 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 @@ -18,12 +18,10 @@ */ package org.apache.parquet.avro; -import java.io.File; import java.io.IOException; import java.util.Arrays; import java.util.HashMap; import java.util.Map; -import java.util.UUID; import org.apache.avro.Schema; import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.IndexedRecord; @@ -32,14 +30,9 @@ import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Ignore; -import org.junit.Rule; import org.junit.Test; -import org.junit.rules.TemporaryFolder; -import org.apache.parquet.hadoop.ParquetWriter; -import org.apache.parquet.hadoop.api.WriteSupport; +import org.apache.parquet.DirectWriterTest; import org.apache.parquet.io.api.RecordConsumer; -import org.apache.parquet.schema.MessageType; -import org.apache.parquet.schema.MessageTypeParser; import static org.apache.parquet.avro.AvroTestUtil.array; import static org.apache.parquet.avro.AvroTestUtil.field; @@ -49,10 +42,7 @@ import static org.apache.parquet.avro.AvroTestUtil.primitive; import static org.apache.parquet.avro.AvroTestUtil.record; -public class TestArrayCompatibility { - - @Rule - public final TemporaryFolder tempDir = new TemporaryFolder(); +public class TestArrayCompatibility extends DirectWriterTest { public static final Configuration NEW_BEHAVIOR_CONF = new Configuration(); @@ -909,68 +899,6 @@ public void write(RecordConsumer rc) { assertReaderContains(newBehaviorReader(test), newSchema, newRecord); } - private interface DirectWriter { - public void write(RecordConsumer consumer); - } - - private static class DirectWriteSupport extends WriteSupport { - private RecordConsumer recordConsumer; - private final MessageType type; - private final DirectWriter writer; - private final Map metadata; - - private DirectWriteSupport(MessageType type, DirectWriter writer, - Map metadata) { - this.type = type; - this.writer = writer; - this.metadata = metadata; - } - - @Override - public WriteContext init(Configuration configuration) { - return new WriteContext(type, metadata); - } - - @Override - public void prepareForWrite(RecordConsumer recordConsumer) { - this.recordConsumer = recordConsumer; - } - - @Override - public void write(Void record) { - writer.write(recordConsumer); - } - } - - private Path writeDirect(String type, DirectWriter writer) throws IOException { - return writeDirect(MessageTypeParser.parseMessageType(type), writer); - } - - private Path writeDirect(String type, DirectWriter writer, - Map metadata) throws IOException { - return writeDirect(MessageTypeParser.parseMessageType(type), writer, metadata); - } - - private Path writeDirect(MessageType type, DirectWriter writer) throws IOException { - return writeDirect(type, writer, new HashMap()); - } - - private Path writeDirect(MessageType type, DirectWriter writer, - Map metadata) throws IOException { - File temp = tempDir.newFile(UUID.randomUUID().toString()); - temp.deleteOnExit(); - temp.delete(); - - Path path = new Path(temp.getPath()); - - ParquetWriter parquetWriter = new ParquetWriter( - path, new DirectWriteSupport(type, writer, metadata)); - parquetWriter.write(null); - parquetWriter.close(); - - return path; - } - public AvroParquetReader oldBehaviorReader( Path path) throws IOException { return new AvroParquetReader(path); diff --git a/parquet-hadoop/pom.xml b/parquet-hadoop/pom.xml index 38271dfc6c..e7ef53f766 100644 --- a/parquet-hadoop/pom.xml +++ b/parquet-hadoop/pom.xml @@ -103,6 +103,17 @@ org.apache.maven.plugins maven-jar-plugin + + org.apache.maven.plugins + maven-jar-plugin + + + + test-jar + + + + diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/DirectWriterTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/DirectWriterTest.java new file mode 100644 index 0000000000..074d2e8b66 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/DirectWriterTest.java @@ -0,0 +1,102 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.junit.Rule; +import org.junit.rules.TemporaryFolder; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.api.WriteSupport; +import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; + +public class DirectWriterTest { + + @Rule + public final TemporaryFolder tempDir = new TemporaryFolder(); + + protected interface DirectWriter { + public void write(RecordConsumer consumer); + } + + protected Path writeDirect(String type, DirectWriter writer) throws IOException { + return writeDirect(MessageTypeParser.parseMessageType(type), writer); + } + + protected Path writeDirect(String type, DirectWriter writer, + Map metadata) throws IOException { + return writeDirect(MessageTypeParser.parseMessageType(type), writer, metadata); + } + + protected Path writeDirect(MessageType type, DirectWriter writer) throws IOException { + return writeDirect(type, writer, new HashMap()); + } + + protected Path writeDirect(MessageType type, DirectWriter writer, + Map metadata) throws IOException { + File temp = tempDir.newFile(UUID.randomUUID().toString()); + temp.deleteOnExit(); + temp.delete(); + + Path path = new Path(temp.getPath()); + + ParquetWriter parquetWriter = new ParquetWriter( + path, new DirectWriteSupport(type, writer, metadata)); + parquetWriter.write(null); + parquetWriter.close(); + + return path; + } + + protected static class DirectWriteSupport extends WriteSupport { + private RecordConsumer recordConsumer; + private final MessageType type; + private final DirectWriter writer; + private final Map metadata; + + protected DirectWriteSupport(MessageType type, DirectWriter writer, + Map metadata) { + this.type = type; + this.writer = writer; + this.metadata = metadata; + } + + @Override + public WriteContext init(Configuration configuration) { + return new WriteContext(type, metadata); + } + + @Override + public void prepareForWrite(RecordConsumer recordConsumer) { + this.recordConsumer = recordConsumer; + } + + @Override + public void write(Void record) { + writer.write(recordConsumer); + } + } +} diff --git a/parquet-thrift/pom.xml b/parquet-thrift/pom.xml index 6e86dd8e9e..1b1508f24a 100644 --- a/parquet-thrift/pom.xml +++ b/parquet-thrift/pom.xml @@ -121,6 +121,13 @@ ${thrift.version} provided + + org.apache.parquet + parquet-hadoop + ${project.version} + test-jar + test + From ca475f0ab688aaa141a5e6319e0a02687075b39c Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Tue, 10 Mar 2015 12:07:02 -0700 Subject: [PATCH 2/8] PARQUET-212: Read non-thrift files if a Thrift class is supplied. Parquet-thrift can now read files not written by parquet-thrift if an appropriate Thrift class is supplied. This adds a check to derive the necessary StructType from a class. Previously, attempting to read a file without Thrift metadata in its properties would return a null ThriftMetaData and throw NPE. This also updates the logic in ThriftMetaData.fromExtraMetaData to avoid NPE when the class is present by the descriptor property is not. --- .../hadoop/thrift/ThriftReadSupport.java | 14 ++++++++----- .../apache/parquet/thrift/ThriftMetaData.java | 21 +++++++++++++++++-- 2 files changed, 28 insertions(+), 7 deletions(-) diff --git a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java index cb9bf661cf..b9684e986a 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java @@ -203,17 +203,16 @@ private void initThriftClassFromMultipleFiles(Map> fileMetad } @SuppressWarnings("unchecked") - private void initThriftClass(Map fileMetadata, Configuration conf) throws ClassNotFoundException { + private void initThriftClass(ThriftMetaData metadata, Configuration conf) throws ClassNotFoundException { if (thriftClass != null) { return; } String className = conf.get(THRIFT_READ_CLASS_KEY, null); if (className == null) { - final ThriftMetaData metaData = ThriftMetaData.fromExtraMetaData(fileMetadata); - if (metaData == null) { + if (metadata == null) { throw new ParquetDecodingException("Could not read file as the Thrift class is not provided and could not be resolved from the file"); } - thriftClass = (Class)metaData.getThriftClass(); + thriftClass = (Class)metadata.getThriftClass(); } else { thriftClass = (Class)Class.forName(className); } @@ -225,7 +224,12 @@ public RecordMaterializer prepareForRead(Configuration configuration, org.apache.parquet.hadoop.api.ReadSupport.ReadContext readContext) { ThriftMetaData thriftMetaData = ThriftMetaData.fromExtraMetaData(keyValueMetaData); try { - initThriftClass(keyValueMetaData, configuration); + initThriftClass(thriftMetaData, configuration); + + // if there was not metadata in the file, get it from requested class + if (thriftMetaData == null) { + thriftMetaData = ThriftMetaData.fromThriftClass(thriftClass); + } String converterClassName = configuration.get(RECORD_CONVERTER_CLASS_KEY, RECORD_CONVERTER_DEFAULT); @SuppressWarnings("unchecked") diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftMetaData.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftMetaData.java index f0a9624669..8b1565abb9 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftMetaData.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftMetaData.java @@ -23,6 +23,7 @@ import org.apache.parquet.hadoop.BadConfigurationException; import org.apache.parquet.thrift.struct.ThriftType; import org.apache.parquet.thrift.struct.ThriftType.StructType; +import org.apache.thrift.TBase; /** * @@ -86,19 +87,35 @@ public StructType getDescriptor() { * Reads ThriftMetadata from the parquet file footer. * * @param extraMetaData extraMetaData field of the parquet footer - * @return + * @return the ThriftMetaData used to write a data file */ public static ThriftMetaData fromExtraMetaData( Map extraMetaData) { final String thriftClassName = extraMetaData.get(THRIFT_CLASS); final String thriftDescriptorString = extraMetaData.get(THRIFT_DESCRIPTOR); - if (thriftClassName == null && thriftDescriptorString == null) { + if (thriftClassName == null || thriftDescriptorString == null) { return null; } final StructType descriptor = parseDescriptor(thriftDescriptorString); return new ThriftMetaData(thriftClassName, descriptor); } + /** + * Creates ThriftMetaData from a Thrift-generated class. + * + * @param thriftClass a Thrift-generated class + * @return ThriftMetaData for the given class + */ + @SuppressWarnings("unchecked") + public static ThriftMetaData fromThriftClass(Class thriftClass) { + if (thriftClass != null && TBase.class.isAssignableFrom(thriftClass)) { + Class> tClass = (Class>) thriftClass; + StructType descriptor = new ThriftSchemaConverter().toStructType(tClass); + return new ThriftMetaData(thriftClass.getName(), descriptor); + } + return null; + } + private static StructType parseDescriptor(String json) { try { return (StructType)ThriftType.fromJSON(json); From a3d490e416e23de19464de788fd2f744c5d92e3f Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Tue, 10 Mar 2015 12:11:38 -0700 Subject: [PATCH 3/8] PARQUET-212: Update thrift reads for LIST compatibility rules. This mirrors the support in parquet-avro and uses an isElementType check in the schema converter. Thrift classes are compiled and must always be present, so there is no ambiguous case because the expected structure is always known. This initial version suppresses nulls when list elements are optional. There is no way to pass a null list element to thrift when it constructs records. This does not change how parquet-thrift writes or converts schemas, so there are no projection changes needed. --- .../parquet/thrift/ThriftRecordConverter.java | 76 ++- .../parquet/thrift/ThriftSchemaConverter.java | 30 + .../hadoop/thrift/TestArrayCompatibility.java | 620 ++++++++++++++++++ .../src/test/thrift/array_compat.thrift | 49 ++ 4 files changed, 767 insertions(+), 8 deletions(-) create mode 100644 parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java create mode 100644 parquet-thrift/src/test/thrift/array_compat.thrift diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java index ec0f4ff245..cd156c64ef 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java @@ -24,6 +24,8 @@ import java.util.List; import java.util.Map; +import org.apache.parquet.Log; +import org.apache.parquet.Preconditions; import org.apache.thrift.TException; import org.apache.thrift.protocol.TField; import org.apache.thrift.protocol.TList; @@ -62,6 +64,8 @@ */ public class ThriftRecordConverter extends RecordMaterializer { + private static final Log LOG = Log.getLog(ThriftRecordConverter.class); + final ParquetProtocol readFieldEnd = new ParquetProtocol("readFieldEnd()") { @Override public void readFieldEnd() throws TException { @@ -637,26 +641,34 @@ void collectionEnd() { */ abstract class CollectionConverter extends GroupConverter { + private ElementConverter elementConverter = null; private final Converter child; private final Counter childCounter; private List listEvents = new ArrayList(); private final List parentEvents; private ThriftTypeID valuesType; - private final Type nestedType; CollectionConverter(List parentEvents, GroupType parquetSchema, ThriftField values) { this.parentEvents = parentEvents; if (parquetSchema.getFieldCount() != 1) { throw new IllegalArgumentException("lists have only one field. " + parquetSchema + " size = " + parquetSchema.getFieldCount()); } - nestedType = parquetSchema.getType(0); + Type repeatedType = parquetSchema.getType(0); valuesType = values.getType().getType(); - if (nestedType.isPrimitive()) { - PrimitiveCounter counter = new PrimitiveCounter(newConverter(listEvents, nestedType, values).asPrimitiveConverter()); - child = counter; - childCounter = counter; + if (ThriftSchemaConverter.isElementType(repeatedType, values)) { + if (repeatedType.isPrimitive()) { + PrimitiveCounter counter = new PrimitiveCounter(newConverter(listEvents, repeatedType, values).asPrimitiveConverter()); + child = counter; + childCounter = counter; + } else { + GroupCounter counter = new GroupCounter(newConverter(listEvents, repeatedType, values).asGroupConverter()); + child = counter; + childCounter = counter; + } } else { - GroupCounter counter = new GroupCounter(newConverter(listEvents, nestedType, values).asGroupConverter()); + this.elementConverter = new ElementConverter(parquetSchema.getName(), + listEvents, repeatedType.asGroupType(), values); + GroupCounter counter = new GroupCounter(elementConverter); child = counter; childCounter = counter; } @@ -678,7 +690,10 @@ public void start() { @Override public void end() { - final int count = childCounter.getCount(); + int count = childCounter.getCount(); + if (elementConverter != null) { + count -= elementConverter.getNullElementCount(); + } collectionStart(count, valuesType.getThriftType()); parentEvents.addAll(listEvents); listEvents.clear(); @@ -691,6 +706,51 @@ public void end() { } + class ElementConverter extends GroupConverter { + + private Converter elementConverter; + private List listEvents; + private List elementEvents; + private int nullElementCount; + + public ElementConverter(String listName, List listEvents, + GroupType repeatedType, ThriftField thriftElement) { + this.listEvents = listEvents; + this.elementEvents = new ArrayList(); + Type elementType = repeatedType.getType(0); + if (elementType.isRepetition(Type.Repetition.OPTIONAL)) { + LOG.warn("List " + listName + + " has optional elements: null elements are ignored."); + } + elementConverter = newConverter(elementEvents, elementType, thriftElement); + } + + @Override + public Converter getConverter(int fieldIndex) { + Preconditions.checkArgument( + fieldIndex == 0, "Illegal field index: %s", fieldIndex); + return elementConverter; + } + + @Override + public void start() { + elementEvents.clear(); + } + + @Override + public void end() { + if (elementEvents.size() > 0) { + listEvents.addAll(elementEvents); + } else { + nullElementCount += 1; + } + } + + public int getNullElementCount() { + return nullElementCount; + } + } + /** * converts to Struct * @author Julien Le Dem diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java index 4ce1e910a6..3a569ace22 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java @@ -20,6 +20,7 @@ import com.twitter.elephantbird.thrift.TStructDescriptor; import com.twitter.elephantbird.thrift.TStructDescriptor.Field; +import org.apache.parquet.schema.Type; import org.apache.thrift.TBase; import org.apache.thrift.TEnum; import org.apache.thrift.TUnion; @@ -84,6 +85,35 @@ private static StructType toStructType(TStructDescriptor struct) { return new StructType(children, structOrUnionType(struct.getThriftClass())); } + /** + * 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 + * determined by checking whether the type can be a synthetic group and by + * checking whether a potential synthetic group matches the expected + * ThriftField. + *

+ * This method never guesses because the expected ThriftField is known. + * + * @param repeatedType a type that may be the element type + * @param thriftElement the expected Schema for list elements + * @return {@code true} if the repeatedType is the element schema + */ + static boolean isElementType(Type repeatedType, ThriftField thriftElement) { + if (repeatedType.isPrimitive() || + (repeatedType.asGroupType().getFieldCount() != 1)) { + // The repeated type must be the element type because it is an invalid + // synthetic wrapper (must be a group with one field). + return true; + } else if (thriftElement != null && thriftElement.getType() instanceof StructType) { + List fields = ((StructType) thriftElement.getType()).getChildren(); + // If the repeated type matches the structure of the ThriftField, then it + // must be the element type. + return (fields.size() == 1 && + fields.get(0).getName().equals(repeatedType.asGroupType().getFieldName(0))); + } + return false; + } + private static ThriftField toThriftField(String name, Field field, ThriftField.Requirement requirement) { ThriftType type; switch (ThriftTypeID.fromByte(field.getType())) { diff --git a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java new file mode 100644 index 0000000000..f637f764fb --- /dev/null +++ b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java @@ -0,0 +1,620 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet.hadoop.thrift; + +import com.google.common.collect.Lists; +import java.io.IOException; +import java.util.List; +import org.apache.hadoop.fs.Path; +import org.apache.thrift.TBase; +import org.junit.Assert; +import org.junit.Ignore; +import org.junit.Test; +import org.apache.parquet.DirectWriterTest; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.thrift.ThriftParquetReader; +import org.apache.parquet.thrift.test.compat.ListOfCounts; +import org.apache.parquet.thrift.test.compat.ListOfInts; +import org.apache.parquet.thrift.test.compat.ListOfLocations; +import org.apache.parquet.thrift.test.compat.ListOfSingleElementGroups; +import org.apache.parquet.thrift.test.compat.Location; +import org.apache.parquet.thrift.test.compat.SingleElementGroup; + +public class TestArrayCompatibility extends DirectWriterTest { + + @Test + @Ignore("Not yet supported") + public void testUnannotatedListOfPrimitives() throws Exception { + Path test = writeDirect( + "message UnannotatedListOfPrimitives {" + + " repeated int32 list_of_ints;" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("list_of_ints", 0); + + rc.addInteger(34); + rc.addInteger(35); + rc.addInteger(36); + + rc.endField("list_of_ints", 0); + rc.endMessage(); + } + }); + } + + @Test + @Ignore("Not yet supported") + public void testUnannotatedListOfGroups() throws Exception { + Path test = writeDirect( + "message UnannotatedListOfGroups {" + + " repeated group list_of_points {" + + " required float x;" + + " required float y;" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("list_of_points", 0); + + rc.startGroup(); + rc.startField("x", 0); + rc.addFloat(1.0f); + rc.endField("x", 0); + rc.startField("y", 1); + rc.addFloat(1.0f); + rc.endField("y", 1); + rc.endGroup(); + + rc.startGroup(); + rc.startField("x", 0); + rc.addFloat(2.0f); + rc.endField("x", 0); + rc.startField("y", 1); + rc.addFloat(2.0f); + rc.endField("y", 1); + rc.endGroup(); + + rc.endField("list_of_points", 0); + rc.endMessage(); + } + }); + } + + @Test + public void testRepeatedPrimitiveInList() throws Exception { + Path test = writeDirect( + "message RepeatedPrimitiveInList {" + + " required group list_of_ints (LIST) {" + + " repeated int32 array;" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("list_of_ints", 0); + + rc.startGroup(); + rc.startField("array", 0); + + rc.addInteger(34); + rc.addInteger(35); + rc.addInteger(36); + + rc.endField("array", 0); + rc.endGroup(); + + rc.endField("list_of_ints", 0); + rc.endMessage(); + } + }); + + ListOfInts expected = new ListOfInts(Lists.newArrayList(34, 35,36)); + ListOfInts actual = reader(test, ListOfInts.class).read(); + Assert.assertEquals("Should read record correctly", expected, actual); + } + + public > ParquetReader reader( + Path file, Class thriftClass) throws IOException { + return ThriftParquetReader.build(file) + .withThriftClass(thriftClass) + .build(); + } + + public void assertReaderContains(ParquetReader reader, T... expected) + throws IOException { + T record; + List actual = Lists.newArrayList(); + while ((record = reader.read()) != null) { + actual.add(record); + } + Assert.assertEquals("Should match exepected records", + Lists.newArrayList(expected), actual); + } + + @Test + public void testMultiFieldGroupInList() throws Exception { + // tests the missing element layer, detected by a multi-field group + Path test = writeDirect( + "message MultiFieldGroupInList {" + + " optional group locations (LIST) {" + + " repeated group element {" + + " required double latitude;" + + " required double longitude;" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(0.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(180.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLocations expected = new ListOfLocations(); + expected.addToLocations(new Location(0.0, 0.0)); + expected.addToLocations(new Location(0.0, 180.0)); + + assertReaderContains(reader(test, ListOfLocations.class), expected); + } + + @Test + public void testSingleFieldGroupInList() throws Exception { + // this tests the case where older data has an ambiguous structure, but the + // correct interpretation can be determined from the thrift class + + Path test = writeDirect( + "message SingleFieldGroupInList {" + + " optional group single_element_groups (LIST) {" + + " repeated group single_element_group {" + + " required int64 count;" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("single_element_groups", 0); + + rc.startGroup(); + rc.startField("single_element_group", 0); // start writing array contents + + rc.startGroup(); + rc.startField("count", 0); + rc.addLong(1234L); + rc.endField("count", 0); + rc.endGroup(); + + rc.startGroup(); + rc.startField("count", 0); + rc.addLong(2345L); + rc.endField("count", 0); + rc.endGroup(); + + rc.endField("single_element_group", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("single_element_groups", 0); + rc.endMessage(); + } + }); + + // the behavior in this case depends on the thrift class used to read + + // test a class with the extra single_element_group level + ListOfSingleElementGroups expectedOldBehavior = new ListOfSingleElementGroups(); + expectedOldBehavior.addToSingle_element_groups(new SingleElementGroup(1234L)); + expectedOldBehavior.addToSingle_element_groups(new SingleElementGroup(2345L)); + + assertReaderContains(reader(test, ListOfSingleElementGroups.class), expectedOldBehavior); + + // test a class without the extra level + ListOfCounts expectedNewBehavior = new ListOfCounts(); + expectedNewBehavior.addToSingle_element_groups(1234L); + expectedNewBehavior.addToSingle_element_groups(2345L); + + assertReaderContains(reader(test, ListOfCounts.class), expectedNewBehavior); + } + + @Test + public void testNewOptionalGroupInList() throws Exception { + Path test = writeDirect( + "message NewOptionalGroupInList {" + + " optional group locations (LIST) {" + + " repeated group list {" + + " optional group element {" + + " required double latitude;" + + " required double longitude;" + + " }" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("list", 0); // start writing array contents + + // write a non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(0.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + // write a null element (element field is omitted) + rc.startGroup(); // array level + rc.endGroup(); // array level + + // write a second non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(180.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + rc.endField("list", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLocations expected = new ListOfLocations(); + expected.addToLocations(new Location(0.0, 0.0)); + // null is not included because thrift does not allow null in lists + //expected.addToLocations(null); + expected.addToLocations(new Location(0.0, 180.0)); + + assertReaderContains(reader(test, ListOfLocations.class), expected); + } + + @Test + public void testNewRequiredGroupInList() throws Exception { + Path test = writeDirect( + "message NewRequiredGroupInList {" + + " optional group locations (LIST) {" + + " repeated group list {" + + " required group element {" + + " required double latitude;" + + " required double longitude;" + + " }" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("list", 0); // start writing array contents + + // write a non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(180.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + // write a second non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(0.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + rc.endField("list", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLocations expected = new ListOfLocations(); + expected.addToLocations(new Location(0.0, 180.0)); + expected.addToLocations(new Location(0.0, 0.0)); + + assertReaderContains(reader(test, ListOfLocations.class), expected); + } + + @Test + public void testAvroCompatRequiredGroupInList() throws Exception { + Path test = writeDirect( + "message AvroCompatRequiredGroupInList {" + + " optional group locations (LIST) {" + + " repeated group array {" + + " required group element {" + + " required double latitude;" + + " required double longitude;" + + " }" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("array", 0); // start writing array contents + + // write a non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(90.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(180.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + // write a second non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(-90.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(0.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + rc.endField("array", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLocations expected = new ListOfLocations(); + expected.addToLocations(new Location(90.0, 180.0)); + expected.addToLocations(new Location(-90.0, 0.0)); + + assertReaderContains(reader(test, ListOfLocations.class), expected); + } + + @Test + public void testOldThriftCompatRequiredGroupInList() throws Exception { + Path test = writeDirect( + "message OldThriftCompatRequiredGroupInList {" + + " optional group locations (LIST) {" + + " repeated group locations_tuple {" + + " required group element {" + + " required double latitude;" + + " required double longitude;" + + " }" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("locations_tuple", 0); // start writing array contents + + // write a non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(180.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + // write a second non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(0.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + rc.endField("locations_tuple", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLocations expected = new ListOfLocations(); + expected.addToLocations(new Location(0.0, 180.0)); + expected.addToLocations(new Location(0.0, 0.0)); + + assertReaderContains(reader(test, ListOfLocations.class), expected); + } + + @Test + public void testHiveCompatOptionalGroupInList() throws Exception { + Path test = writeDirect( + "message HiveCompatOptionalGroupInList {" + + " optional group locations (LIST) {" + + " repeated group bag {" + + " optional group element {" + + " required double latitude;" + + " required double longitude;" + + " }" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("bag", 0); // start writing array contents + + // write a non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(180.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + // write a second non-null element + rc.startGroup(); // array level + rc.startField("element", 0); + + rc.startGroup(); + rc.startField("latitude", 0); + rc.addDouble(0.0); + rc.endField("latitude", 0); + rc.startField("longitude", 1); + rc.addDouble(0.0); + rc.endField("longitude", 1); + rc.endGroup(); + + rc.endField("element", 0); + rc.endGroup(); // array level + + rc.endField("bag", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLocations expected = new ListOfLocations(); + expected.addToLocations(new Location(0.0, 180.0)); + expected.addToLocations(new Location(0.0, 0.0)); + + assertReaderContains(reader(test, ListOfLocations.class), expected); + } +} diff --git a/parquet-thrift/src/test/thrift/array_compat.thrift b/parquet-thrift/src/test/thrift/array_compat.thrift new file mode 100644 index 0000000000..0065485dc7 --- /dev/null +++ b/parquet-thrift/src/test/thrift/array_compat.thrift @@ -0,0 +1,49 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +namespace java org.apache.parquet.thrift.test.compat + +struct ListOfInts { + 1: required list list_of_ints; +} + +struct Location { + 1: required double latitude; + 2: required double longitude; +} + +struct ListOfLocations { + 1: optional list locations; +} + +struct SingleElementGroup { + 1: required i64 count; +} + +struct SingleElementGroupDifferentName { + 1: required i64 differentFieldName; +} + +struct ListOfSingleElementGroups { + 1: optional list single_element_groups; +} + +struct ListOfCounts { + 1: optional list single_element_groups; +} From ee8d154afcd3b05c3b478f1eccd483faf6e46774 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Tue, 10 Mar 2015 12:55:52 -0700 Subject: [PATCH 4/8] PARQUET-212: Add property to ignore nulls in lists. This adds parquet.thrift.ignore-null-elements to suppress an exception thrown when reading a list with optional elements. This makes reading Hive lists an opt-in so that the caller must be aware that null values are ignored. --- .../hadoop/thrift/ThriftReadSupport.java | 2 + .../parquet/thrift/ThriftRecordConverter.java | 49 +++++++++++-- .../hadoop/thrift/TestArrayCompatibility.java | 71 +++++++++++++------ 3 files changed, 96 insertions(+), 26 deletions(-) diff --git a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java index b9684e986a..30dd5e1a51 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java @@ -237,6 +237,8 @@ public RecordMaterializer prepareForRead(Configuration configuration, Constructor> constructor = converterClass.getConstructor(Class.class, MessageType.class, StructType.class); ThriftRecordConverter converter = constructor.newInstance(thriftClass, readContext.getRequestedSchema(), thriftMetaData.getDescriptor()); + converter.setConf(configuration); + converter.initialize(); return converter; } catch (Exception t) { throw new RuntimeException("Unable to create Thrift Converter for Thrift metadata " + thriftMetaData, t); diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java index cd156c64ef..aae0114159 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java @@ -24,8 +24,8 @@ import java.util.List; import java.util.Map; -import org.apache.parquet.Log; -import org.apache.parquet.Preconditions; +import org.apache.hadoop.conf.Configurable; +import org.apache.hadoop.conf.Configuration; import org.apache.thrift.TException; import org.apache.thrift.protocol.TField; import org.apache.thrift.protocol.TList; @@ -35,6 +35,9 @@ import org.apache.thrift.protocol.TStruct; import org.apache.thrift.protocol.TType; +import org.apache.parquet.Log; +import org.apache.parquet.Preconditions; +import org.apache.parquet.io.ParquetDecodingException; import org.apache.parquet.io.api.Binary; import org.apache.parquet.io.api.Converter; import org.apache.parquet.io.api.GroupConverter; @@ -62,10 +65,14 @@ * * @param */ -public class ThriftRecordConverter extends RecordMaterializer { +public class ThriftRecordConverter extends RecordMaterializer implements Configurable { private static final Log LOG = Log.getLog(ThriftRecordConverter.class); + public static final String IGNORE_NULL_LIST_ELEMENTS = + "parquet.thrift.ignore-null-elements"; + private static final boolean IGNORE_NULL_LIST_ELEMENTS_DEFAULT = false; + final ParquetProtocol readFieldEnd = new ParquetProtocol("readFieldEnd()") { @Override public void readFieldEnd() throws TException { @@ -719,8 +726,14 @@ public ElementConverter(String listName, List listEvents, this.elementEvents = new ArrayList(); Type elementType = repeatedType.getType(0); if (elementType.isRepetition(Type.Repetition.OPTIONAL)) { - LOG.warn("List " + listName + - " has optional elements: null elements are ignored."); + if (ignoreNullElements) { + LOG.warn("List " + listName + + " has optional elements: null elements are ignored."); + } else { + throw new ParquetDecodingException("Cannot read list " + listName + + " with optional elements: set " + IGNORE_NULL_LIST_ELEMENTS + + " to ignore nulls."); + } } elementConverter = newConverter(elementEvents, elementType, thriftElement); } @@ -838,9 +851,13 @@ public void end() { } private final ThriftReader thriftReader; private final ParquetReadProtocol protocol; - private final GroupConverter structConverter; + private final MessageType requestedParquetSchema; + private final String name; + private GroupConverter structConverter; private List rootEvents = new ArrayList(); private boolean missingRequiredFieldsInProjection = false; + private Configuration conf = null; + private boolean ignoreNullElements = IGNORE_NULL_LIST_ELEMENTS_DEFAULT; /** * @@ -852,13 +869,33 @@ public void end() { public ThriftRecordConverter(ThriftReader thriftReader, String name, MessageType requestedParquetSchema, ThriftType.StructType thriftType) { super(); this.thriftReader = thriftReader; + this.name = name; + this.requestedParquetSchema = requestedParquetSchema; this.protocol = new ParquetReadProtocol(); this.thriftType = thriftType; + } + + public void initialize() { MessageType fullSchema = new ThriftSchemaConverter().convert(thriftType); missingRequiredFieldsInProjection = hasMissingRequiredFieldInGroupType(requestedParquetSchema, fullSchema); this.structConverter = new StructConverter(rootEvents, requestedParquetSchema, new ThriftField(name, (short)0, Requirement.REQUIRED, thriftType)); } + @Override + public void setConf(Configuration configuration) { + this.conf = configuration; + if (conf != null) { + this.ignoreNullElements = conf.getBoolean( + IGNORE_NULL_LIST_ELEMENTS, + IGNORE_NULL_LIST_ELEMENTS_DEFAULT); + } + } + + @Override + public Configuration getConf() { + return conf; + } + private boolean hasMissingRequiredFieldInGroupType(GroupType requested, GroupType fullSchema) { for (Type field : fullSchema.getFields()) { diff --git a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java index f637f764fb..d602710af9 100644 --- a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java +++ b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java @@ -21,6 +21,7 @@ import com.google.common.collect.Lists; import java.io.IOException; import java.util.List; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.thrift.TBase; import org.junit.Assert; @@ -30,6 +31,7 @@ import org.apache.parquet.hadoop.ParquetReader; import org.apache.parquet.io.api.RecordConsumer; import org.apache.parquet.thrift.ThriftParquetReader; +import org.apache.parquet.thrift.ThriftRecordConverter; import org.apache.parquet.thrift.test.compat.ListOfCounts; import org.apache.parquet.thrift.test.compat.ListOfInts; import org.apache.parquet.thrift.test.compat.ListOfLocations; @@ -37,6 +39,9 @@ import org.apache.parquet.thrift.test.compat.Location; import org.apache.parquet.thrift.test.compat.SingleElementGroup; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + public class TestArrayCompatibility extends DirectWriterTest { @Test @@ -136,24 +141,6 @@ public void write(RecordConsumer rc) { Assert.assertEquals("Should read record correctly", expected, actual); } - public > ParquetReader reader( - Path file, Class thriftClass) throws IOException { - return ThriftParquetReader.build(file) - .withThriftClass(thriftClass) - .build(); - } - - public void assertReaderContains(ParquetReader reader, T... expected) - throws IOException { - T record; - List actual = Lists.newArrayList(); - while ((record = reader.read()) != null) { - actual.add(record); - } - Assert.assertEquals("Should match exepected records", - Lists.newArrayList(expected), actual); - } - @Test public void testMultiFieldGroupInList() throws Exception { // tests the missing element layer, detected by a multi-field group @@ -339,7 +326,15 @@ public void write(RecordConsumer rc) { //expected.addToLocations(null); expected.addToLocations(new Location(0.0, 180.0)); - assertReaderContains(reader(test, ListOfLocations.class), expected); + try { + assertReaderContains(reader(test, ListOfLocations.class), expected); + fail("Should fail: locations are optional and not ignored"); + } catch (RuntimeException e) { + // e is a RuntimeException wrapping the decoding exception + assertTrue(e.getCause().getMessage().contains("locations")); + } + + assertReaderContains(readerIgnoreNulls(test, ListOfLocations.class), expected); } @Test @@ -615,6 +610,42 @@ public void write(RecordConsumer rc) { expected.addToLocations(new Location(0.0, 180.0)); expected.addToLocations(new Location(0.0, 0.0)); - assertReaderContains(reader(test, ListOfLocations.class), expected); + try { + assertReaderContains(reader(test, ListOfLocations.class), expected); + fail("Should fail: locations are optional and not ignored"); + } catch (RuntimeException e) { + // e is a RuntimeException wrapping the decoding exception + assertTrue(e.getCause().getMessage().contains("locations")); + } + + assertReaderContains(readerIgnoreNulls(test, ListOfLocations.class), expected); + } + + public > ParquetReader reader( + Path file, Class thriftClass) throws IOException { + return ThriftParquetReader.build(file) + .withThriftClass(thriftClass) + .build(); + } + + public > ParquetReader readerIgnoreNulls( + Path file, Class thriftClass) throws IOException { + Configuration conf = new Configuration(); + conf.setBoolean(ThriftRecordConverter.IGNORE_NULL_LIST_ELEMENTS, true); + return ThriftParquetReader.build(file) + .withThriftClass(thriftClass) + .withConf(conf) + .build(); + } + + public void assertReaderContains(ParquetReader reader, T... expected) + throws IOException { + T record; + List actual = Lists.newArrayList(); + while ((record = reader.read()) != null) { + actual.add(record); + } + Assert.assertEquals("Should match exepected records", + Lists.newArrayList(expected), actual); } } From 4c71d2853a40cb855bd6cbf3127048bb888594ee Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Wed, 11 Mar 2015 11:34:26 -0700 Subject: [PATCH 5/8] PARQUET-212: Exclude ThriftRecordConverter from semver check. This should be a temporary change, until 1.6.0 is released. The Semver check is catching that the ThriftRecordConverter now implements Configurable. This is a binary-compatible change that should not require a major version update. --- pom.xml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pom.xml b/pom.xml index d27456c80c..4df38d706b 100644 --- a/pom.xml +++ b/pom.xml @@ -242,6 +242,8 @@ org/apache/parquet/hadoop/ParquetInputSplit shaded/** parquet/** + + org/apache/parquet/thrift/ThriftRecordConverter From 223eabf483d3f6dc0a74617790b19e88f427a757 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Wed, 11 Mar 2015 14:27:11 -0700 Subject: [PATCH 6/8] PARQUET-212: Add Configuration to the ThriftRecordConverter ctor. Previously, this added Configurable to ThriftRecordConverter, but that caused problems with semver and test failures because a separate initialize method had to be called. This is brittle because callers might not know to call initialize and the class is part of the API because subclassing is allowed. To avoid the issue, this replaces the Configurable interface and initialize method with a new constructor that takes a Configuration. --- .../scrooge/ScroogeRecordConverter.java | 13 ++++- .../hadoop/thrift/ThriftReadSupport.java | 57 +++++++++++++++---- .../parquet/thrift/TBaseRecordConverter.java | 12 +++- .../parquet/thrift/ThriftRecordConverter.java | 38 +++++-------- .../hadoop/thrift/TestArrayCompatibility.java | 4 +- 5 files changed, 84 insertions(+), 40 deletions(-) diff --git a/parquet-scrooge/src/main/java/org/apache/parquet/scrooge/ScroogeRecordConverter.java b/parquet-scrooge/src/main/java/org/apache/parquet/scrooge/ScroogeRecordConverter.java index d385999abf..9c4faa0d1f 100644 --- a/parquet-scrooge/src/main/java/org/apache/parquet/scrooge/ScroogeRecordConverter.java +++ b/parquet-scrooge/src/main/java/org/apache/parquet/scrooge/ScroogeRecordConverter.java @@ -18,6 +18,7 @@ */ package org.apache.parquet.scrooge; +import org.apache.hadoop.conf.Configuration; import org.apache.thrift.TException; import org.apache.thrift.protocol.TProtocol; @@ -31,8 +32,16 @@ public class ScroogeRecordConverter extends ThriftRecordConverter { - + /** + * This is for compatibility only. + * @deprecated will be removed in 2.x + */ + @Deprecated public ScroogeRecordConverter(final Class thriftClass, MessageType parquetSchema, StructType thriftType) { + this(thriftClass, parquetSchema, thriftType, null); + } + + public ScroogeRecordConverter(final Class thriftClass, MessageType parquetSchema, StructType thriftType, Configuration conf) { super(new ThriftReader() { @SuppressWarnings("unchecked") ThriftStructCodec codec = (ThriftStructCodec) getCodec(thriftClass); @@ -40,7 +49,7 @@ public ScroogeRecordConverter(final Class thriftClass, MessageType parquetSch public T readOneRecord(TProtocol protocol) throws TException { return codec.decode(protocol); } - }, thriftClass.getSimpleName(), parquetSchema, thriftType); + }, thriftClass.getSimpleName(), parquetSchema, thriftType, conf); } private static ThriftStructCodec getCodec(Class klass) { diff --git a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java index 30dd5e1a51..3fc0fcff42 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftReadSupport.java @@ -19,6 +19,7 @@ package org.apache.parquet.hadoop.thrift; import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; import java.util.Map; import java.util.Set; @@ -225,23 +226,55 @@ public RecordMaterializer prepareForRead(Configuration configuration, ThriftMetaData thriftMetaData = ThriftMetaData.fromExtraMetaData(keyValueMetaData); try { initThriftClass(thriftMetaData, configuration); + } catch (ClassNotFoundException e) { + throw new RuntimeException("Cannot find Thrift object class for metadata: " + thriftMetaData, e); + } + + // if there was not metadata in the file, get it from requested class + if (thriftMetaData == null) { + thriftMetaData = ThriftMetaData.fromThriftClass(thriftClass); + } - // if there was not metadata in the file, get it from requested class - if (thriftMetaData == null) { - thriftMetaData = ThriftMetaData.fromThriftClass(thriftClass); + String converterClassName = configuration.get(RECORD_CONVERTER_CLASS_KEY, RECORD_CONVERTER_DEFAULT); + return getRecordConverterInstance(converterClassName, thriftClass, + readContext.getRequestedSchema(), thriftMetaData.getDescriptor(), + configuration); + } + + @SuppressWarnings("unchecked") + private static ThriftRecordConverter getRecordConverterInstance( + String converterClassName, Class thriftClass, + MessageType requestedSchema, StructType descriptor, Configuration conf) { + Class> converterClass; + try { + converterClass = (Class>) Class.forName(converterClassName); + } catch (ClassNotFoundException e) { + throw new RuntimeException("Cannot find Thrift converter class: " + converterClassName, e); + } + + try { + // first try the new version that accepts a Configuration + try { + Constructor> constructor = + converterClass.getConstructor(Class.class, MessageType.class, StructType.class, Configuration.class); + return constructor.newInstance(thriftClass, requestedSchema, descriptor, conf); + } catch (IllegalAccessException e) { + // try the other constructor pattern + } catch (NoSuchMethodException e) { + // try to find the other constructor pattern } - String converterClassName = configuration.get(RECORD_CONVERTER_CLASS_KEY, RECORD_CONVERTER_DEFAULT); - @SuppressWarnings("unchecked") - Class> converterClass = (Class>) Class.forName(converterClassName); Constructor> constructor = converterClass.getConstructor(Class.class, MessageType.class, StructType.class); - ThriftRecordConverter converter = constructor.newInstance(thriftClass, readContext.getRequestedSchema(), thriftMetaData.getDescriptor()); - converter.setConf(configuration); - converter.initialize(); - return converter; - } catch (Exception t) { - throw new RuntimeException("Unable to create Thrift Converter for Thrift metadata " + thriftMetaData, t); + return constructor.newInstance(thriftClass, requestedSchema, descriptor); + } catch (InstantiationException e) { + throw new RuntimeException("Failed to construct Thrift converter class: " + converterClassName, e); + } catch (InvocationTargetException e) { + throw new RuntimeException("Failed to construct Thrift converter class: " + converterClassName, e); + } catch (IllegalAccessException e) { + throw new RuntimeException("Cannot access constructor for Thrift converter class: " + converterClassName, e); + } catch (NoSuchMethodException e) { + throw new RuntimeException("Cannot find constructor for Thrift converter class: " + converterClassName, e); } } } diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/TBaseRecordConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/TBaseRecordConverter.java index 17a68d678a..6483e5919a 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/TBaseRecordConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/TBaseRecordConverter.java @@ -18,6 +18,7 @@ */ package org.apache.parquet.thrift; +import org.apache.hadoop.conf.Configuration; import org.apache.thrift.TBase; import org.apache.thrift.TException; import org.apache.thrift.protocol.TProtocol; @@ -28,7 +29,16 @@ public class TBaseRecordConverter> extends ThriftRecordConverter { + /** + * This is for compatibility only. + * @deprecated will be removed in 2.x + */ + @Deprecated public TBaseRecordConverter(final Class thriftClass, MessageType requestedParquetSchema, StructType thriftType) { + this(thriftClass, requestedParquetSchema, thriftType, null); + } + + public TBaseRecordConverter(final Class thriftClass, MessageType requestedParquetSchema, StructType thriftType, Configuration conf) { super(new ThriftReader() { @Override public T readOneRecord(TProtocol protocol) throws TException { @@ -42,7 +52,7 @@ public T readOneRecord(TProtocol protocol) throws TException { throw new ParquetDecodingException("Thrift class or constructor not public " + thriftClass, e); } } - }, thriftClass.getSimpleName(), requestedParquetSchema, thriftType); + }, thriftClass.getSimpleName(), requestedParquetSchema, thriftType, conf); } } diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java index aae0114159..d9bdb7b1c6 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java @@ -65,7 +65,7 @@ * * @param */ -public class ThriftRecordConverter extends RecordMaterializer implements Configurable { +public class ThriftRecordConverter extends RecordMaterializer { private static final Log LOG = Log.getLog(ThriftRecordConverter.class); @@ -851,49 +851,41 @@ public void end() { } private final ThriftReader thriftReader; private final ParquetReadProtocol protocol; - private final MessageType requestedParquetSchema; - private final String name; private GroupConverter structConverter; private List rootEvents = new ArrayList(); private boolean missingRequiredFieldsInProjection = false; - private Configuration conf = null; private boolean ignoreNullElements = IGNORE_NULL_LIST_ELEMENTS_DEFAULT; + /** + * This is for compatibility only. + * @deprecated will be removed in 2.x + */ + @Deprecated + public ThriftRecordConverter(ThriftReader thriftReader, String name, MessageType requestedParquetSchema, ThriftType.StructType thriftType) { + this(thriftReader, name, requestedParquetSchema, thriftType, null); + } + /** * * @param thriftReader the class responsible for instantiating the final object and read from the protocol * @param name the name of that type ( the thrift class simple name) * @param requestedParquetSchema the schema for the incoming columnar events * @param thriftType the thrift type descriptor + * @param conf a Configuration */ - public ThriftRecordConverter(ThriftReader thriftReader, String name, MessageType requestedParquetSchema, ThriftType.StructType thriftType) { + public ThriftRecordConverter(ThriftReader thriftReader, String name, MessageType requestedParquetSchema, ThriftType.StructType thriftType, Configuration conf) { super(); this.thriftReader = thriftReader; - this.name = name; - this.requestedParquetSchema = requestedParquetSchema; this.protocol = new ParquetReadProtocol(); this.thriftType = thriftType; - } - - public void initialize() { - MessageType fullSchema = new ThriftSchemaConverter().convert(thriftType); - missingRequiredFieldsInProjection = hasMissingRequiredFieldInGroupType(requestedParquetSchema, fullSchema); - this.structConverter = new StructConverter(rootEvents, requestedParquetSchema, new ThriftField(name, (short)0, Requirement.REQUIRED, thriftType)); - } - - @Override - public void setConf(Configuration configuration) { - this.conf = configuration; if (conf != null) { this.ignoreNullElements = conf.getBoolean( IGNORE_NULL_LIST_ELEMENTS, IGNORE_NULL_LIST_ELEMENTS_DEFAULT); } - } - - @Override - public Configuration getConf() { - return conf; + MessageType fullSchema = new ThriftSchemaConverter().convert(thriftType); + missingRequiredFieldsInProjection = hasMissingRequiredFieldInGroupType(requestedParquetSchema, fullSchema); + this.structConverter = new StructConverter(rootEvents, requestedParquetSchema, new ThriftField(name, (short)0, Requirement.REQUIRED, thriftType)); } private boolean hasMissingRequiredFieldInGroupType(GroupType requested, GroupType fullSchema) { diff --git a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java index d602710af9..02da9c492b 100644 --- a/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java +++ b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java @@ -331,7 +331,7 @@ public void write(RecordConsumer rc) { fail("Should fail: locations are optional and not ignored"); } catch (RuntimeException e) { // e is a RuntimeException wrapping the decoding exception - assertTrue(e.getCause().getMessage().contains("locations")); + assertTrue(e.getCause().getCause().getMessage().contains("locations")); } assertReaderContains(readerIgnoreNulls(test, ListOfLocations.class), expected); @@ -615,7 +615,7 @@ public void write(RecordConsumer rc) { fail("Should fail: locations are optional and not ignored"); } catch (RuntimeException e) { // e is a RuntimeException wrapping the decoding exception - assertTrue(e.getCause().getMessage().contains("locations")); + assertTrue(e.getCause().getCause().getMessage().contains("locations")); } assertReaderContains(readerIgnoreNulls(test, ListOfLocations.class), expected); From 4d9afd17ac0082483c69d207048259a061422a24 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Wed, 11 Mar 2015 14:28:02 -0700 Subject: [PATCH 7/8] PARQUET-212: Fix list handling with projection. This fixes parquet-avro and parquet-thrift list handling for the case where the Parquet schema columns are projected and only one remains. This appears to not match the element schema because the element schema has more than one field. The solution is to match the element schema if the Parquet schema is a subset. --- .../parquet/avro/AvroIndexedRecordConverter.java | 13 ++++++++----- .../parquet/thrift/ThriftSchemaConverter.java | 16 ++++++++++------ 2 files changed, 18 insertions(+), 11 deletions(-) diff --git a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroIndexedRecordConverter.java b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroIndexedRecordConverter.java index 262c4235aa..bae2ee6194 100644 --- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroIndexedRecordConverter.java +++ b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroIndexedRecordConverter.java @@ -20,7 +20,9 @@ import java.lang.reflect.Constructor; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import org.apache.avro.Schema; import org.apache.avro.generic.GenericArray; import org.apache.avro.generic.GenericData; @@ -362,13 +364,14 @@ static boolean isElementType(Type repeatedType, Schema elementSchema) { // synthetic wrapper (must be a group with one field). return true; } else if (elementSchema != null && - elementSchema.getType() == Schema.Type.RECORD && - elementSchema.getFields().size() == 1 && - elementSchema.getFields().get(0).name().equals( - repeatedType.asGroupType().getFieldName(0))) { + elementSchema.getType() == Schema.Type.RECORD) { + Set fieldNames = new HashSet(); + for (Schema.Field field : elementSchema.getFields()) { + fieldNames.add(field.name()); + } // The repeated type must be the element type because it matches the // structure of the Avro element's schema. - return true; + return fieldNames.contains(repeatedType.asGroupType().getFieldName(0)); } return false; } diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java index 3a569ace22..a2ec5eb6bb 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java @@ -20,11 +20,13 @@ import com.twitter.elephantbird.thrift.TStructDescriptor; import com.twitter.elephantbird.thrift.TStructDescriptor.Field; -import org.apache.parquet.schema.Type; +import java.util.HashSet; +import java.util.Set; import org.apache.thrift.TBase; import org.apache.thrift.TEnum; import org.apache.thrift.TUnion; +import org.apache.parquet.schema.Type; import org.apache.parquet.schema.MessageType; import org.apache.parquet.thrift.projection.FieldProjectionFilter; import org.apache.parquet.thrift.struct.ThriftField; @@ -105,11 +107,13 @@ static boolean isElementType(Type repeatedType, ThriftField thriftElement) { // synthetic wrapper (must be a group with one field). return true; } else if (thriftElement != null && thriftElement.getType() instanceof StructType) { - List fields = ((StructType) thriftElement.getType()).getChildren(); - // If the repeated type matches the structure of the ThriftField, then it - // must be the element type. - return (fields.size() == 1 && - fields.get(0).getName().equals(repeatedType.asGroupType().getFieldName(0))); + Set fieldNames = new HashSet(); + for (ThriftField field : ((StructType) thriftElement.getType()).getChildren()) { + fieldNames.add(field.getName()); + } + // If the repeated type is a subset of the structure of the ThriftField, + // then it must be the element type. + return fieldNames.contains(repeatedType.asGroupType().getFieldName(0)); } return false; } From d6e77ad566043adaca5730b2e3a294ca845a697d Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Tue, 7 Apr 2015 16:23:45 -0700 Subject: [PATCH 8/8] PARQUET-212: Rename isElementType => isListElementType. --- .../java/org/apache/parquet/thrift/ThriftRecordConverter.java | 3 +-- .../java/org/apache/parquet/thrift/ThriftSchemaConverter.java | 3 ++- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java index d9bdb7b1c6..c526891f55 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftRecordConverter.java @@ -24,7 +24,6 @@ import java.util.List; import java.util.Map; -import org.apache.hadoop.conf.Configurable; import org.apache.hadoop.conf.Configuration; import org.apache.thrift.TException; import org.apache.thrift.protocol.TField; @@ -662,7 +661,7 @@ abstract class CollectionConverter extends GroupConverter { } Type repeatedType = parquetSchema.getType(0); valuesType = values.getType().getType(); - if (ThriftSchemaConverter.isElementType(repeatedType, values)) { + if (ThriftSchemaConverter.isListElementType(repeatedType, values)) { if (repeatedType.isPrimitive()) { PrimitiveCounter counter = new PrimitiveCounter(newConverter(listEvents, repeatedType, values).asPrimitiveConverter()); child = counter; diff --git a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java index a2ec5eb6bb..2a10abb566 100644 --- a/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java +++ b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftSchemaConverter.java @@ -100,7 +100,8 @@ private static StructType toStructType(TStructDescriptor struct) { * @param thriftElement the expected Schema for list elements * @return {@code true} if the repeatedType is the element schema */ - static boolean isElementType(Type repeatedType, ThriftField thriftElement) { + static boolean isListElementType(Type repeatedType, + ThriftField thriftElement) { if (repeatedType.isPrimitive() || (repeatedType.asGroupType().getFieldCount() != 1)) { // The repeated type must be the element type because it is an invalid