From 26ccfdb6e21332f0d4ede5d53371835f1c3151c2 Mon Sep 17 00:00:00 2001 From: Niels Basjes Date: Tue, 12 Jan 2016 16:44:12 +0100 Subject: [PATCH 01/58] PARQUET-423: Make writing Avro to Parquet less noisy --- .../hadoop/ColumnChunkPageWriteStore.java | 11 +++++---- .../hadoop/InternalParquetRecordWriter.java | 7 ++++-- .../apache/parquet/hadoop/ParquetWriter.java | 24 +++++++++++++++---- 3 files changed, 32 insertions(+), 10 deletions(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java index 2eab54a1a0..6d54f9c719 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java @@ -62,6 +62,7 @@ private static final class ColumnChunkPageWriter implements PageWriter { private long compressedLength; private long totalValueCount; private int pageCount; + private boolean columnInfoLogging; private Set encodings = new HashSet(); @@ -70,10 +71,12 @@ private static final class ColumnChunkPageWriter implements PageWriter { private ColumnChunkPageWriter(ColumnDescriptor path, BytesCompressor compressor, - ByteBufferAllocator allocator) { + ByteBufferAllocator allocator, + boolean columnInfoLogging) { this.path = path; this.compressor = compressor; this.allocator = allocator; + this.columnInfoLogging = columnInfoLogging; this.buf = new ConcatenatingByteArrayCollector(); this.totalStatistics = getStatsBasedOnType(this.path.getType()); } @@ -186,7 +189,7 @@ public void writeToFileWriter(ParquetFileWriter writer) throws IOException { } writer.writeDataPages(buf, uncompressedLength, compressedLength, totalStatistics, new ArrayList(encodings)); writer.endColumn(); - if (INFO) { + if (INFO && columnInfoLogging) { LOG.info( String.format( "written %,dB for %s: %,d values, %,dB raw, %,dB comp, %d pages, encodings: %s", @@ -226,10 +229,10 @@ public String memUsageString(String prefix) { private final Map writers = new HashMap(); private final MessageType schema; - public ColumnChunkPageWriteStore(BytesCompressor compressor, MessageType schema, ByteBufferAllocator allocator) { + public ColumnChunkPageWriteStore(BytesCompressor compressor, MessageType schema, ByteBufferAllocator allocator, boolean columnInfoLogging) { this.schema = schema; for (ColumnDescriptor path : schema.getColumns()) { - writers.put(path, new ColumnChunkPageWriter(path, compressor, allocator)); + writers.put(path, new ColumnChunkPageWriter(path, compressor, allocator, columnInfoLogging)); } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index 74feb39033..dda39443ae 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -55,6 +55,7 @@ class InternalParquetRecordWriter { private final BytesCompressor compressor; private final boolean validating; private final ParquetProperties props; + private final boolean columnInfoLogging; private long recordCount = 0; private long recordCountForNextMemCheck = MINIMUM_RECORD_COUNT_FOR_CHECK; @@ -80,7 +81,8 @@ public InternalParquetRecordWriter( long rowGroupSize, BytesCompressor compressor, boolean validating, - ParquetProperties props) { + ParquetProperties props, + boolean columnInfoLogging) { this.parquetFileWriter = parquetFileWriter; this.writeSupport = checkNotNull(writeSupport, "writeSupport"); this.schema = schema; @@ -91,11 +93,12 @@ public InternalParquetRecordWriter( this.compressor = compressor; this.validating = validating; this.props = props; + this.columnInfoLogging = columnInfoLogging; initStore(); } private void initStore() { - pageStore = new ColumnChunkPageWriteStore(compressor, schema, props.getAllocator()); + pageStore = new ColumnChunkPageWriteStore(compressor, schema, props.getAllocator(), columnInfoLogging); columnStore = props.newColumnWriteStore(schema, pageStore); MessageColumnIO columnIO = new ColumnIOFactory(validating).getColumnIO(schema); this.recordConsumer = columnIO.getRecordWriter(columnStore); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java index 58cbe957d0..63333227f7 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java @@ -45,6 +45,7 @@ public class ParquetWriter implements Closeable { public static final boolean DEFAULT_IS_VALIDATING_ENABLED = false; public static final WriterVersion DEFAULT_WRITER_VERSION = ParquetProperties.DEFAULT_WRITER_VERSION; + public static final boolean DEFAULT_COLUMN_INFO_LOGGING = true; public static final String OBJECT_MODEL_NAME_PROP = "writer.model.name"; @@ -226,7 +227,8 @@ public ParquetWriter( .withDictionaryPageSize(dictionaryPageSize) .withDictionaryEncoding(enableDictionary) .withWriterVersion(writerVersion) - .build()); + .build(), + DEFAULT_COLUMN_INFO_LOGGING); } /** @@ -265,7 +267,8 @@ public ParquetWriter(Path file, Configuration conf, WriteSupport writeSupport boolean validating, Configuration conf, int maxPaddingSize, - ParquetProperties encodingProps) throws IOException { + ParquetProperties encodingProps, + boolean columnInfoLogging) throws IOException { WriteSupport.WriteContext writeContext = writeSupport.init(conf); MessageType schema = writeContext.getSchema(); @@ -284,7 +287,8 @@ public ParquetWriter(Path file, Configuration conf, WriteSupport writeSupport blockSize, compressor, validating, - encodingProps); + encodingProps, + columnInfoLogging); } public void write(T object) throws IOException { @@ -333,6 +337,7 @@ public abstract static class Builder> { private boolean enableValidation = DEFAULT_IS_VALIDATING_ENABLED; private ParquetProperties.Builder encodingPropsBuilder = ParquetProperties.builder(); + private boolean columnInfoLogging = DEFAULT_COLUMN_INFO_LOGGING; protected Builder(Path file) { this.file = file; @@ -485,6 +490,17 @@ public SELF withWriterVersion(WriterVersion version) { } /** + * Enable or disable logging information about the columns that have been written by the writer. + * + * @param columnInfoLogging whether this type of logging should be enabled + * @return this builder for method chaining. + */ + public SELF withColumnInfoLogging(boolean columnInfoLogging) { + this.columnInfoLogging = columnInfoLogging; + return self(); + } + + /** * Build a {@link ParquetWriter} with the accumulated configuration. * * @return a configured {@code ParquetWriter} instance. @@ -493,7 +509,7 @@ public SELF withWriterVersion(WriterVersion version) { public ParquetWriter build() throws IOException { return new ParquetWriter(file, mode, getWriteSupport(conf), codecName, rowGroupSize, enableValidation, conf, maxPaddingSize, - encodingPropsBuilder.build()); + encodingPropsBuilder.build(), columnInfoLogging); } } } From f65ae993c184c939f068a416246160b10949726e Mon Sep 17 00:00:00 2001 From: Niels Basjes Date: Tue, 12 Jan 2016 17:28:47 +0100 Subject: [PATCH 02/58] PARQUET-423: Fix various missing bits --- .../apache/parquet/hadoop/ParquetOutputFormat.java | 4 +++- .../apache/parquet/hadoop/ParquetRecordWriter.java | 13 ++++++++----- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index 31cc96ba10..88b3342f24 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -21,6 +21,7 @@ import static org.apache.parquet.Log.INFO; import static org.apache.parquet.Preconditions.checkNotNull; import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE; +import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_COLUMN_INFO_LOGGING; import static org.apache.parquet.hadoop.util.ContextUtil.getConfiguration; import java.io.IOException; @@ -411,7 +412,8 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp validating, props, memoryManager, - conf); + conf, + DEFAULT_COLUMN_INFO_LOGGING); } /** diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java index a9ade96b00..31c036674d 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java @@ -33,6 +33,7 @@ import org.apache.parquet.schema.MessageType; import static org.apache.parquet.Preconditions.checkNotNull; +import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_COLUMN_INFO_LOGGING; /** * Writes records to a Parquet file @@ -72,7 +73,8 @@ public ParquetRecordWriter( int dictionaryPageSize, boolean enableDictionary, boolean validating, - WriterVersion writerVersion) { + WriterVersion writerVersion, + boolean columnInfoLogging) { ParquetProperties props = ParquetProperties.builder() .withPageSize(pageSize) .withDictionaryPageSize(dictionaryPageSize) @@ -80,7 +82,7 @@ public ParquetRecordWriter( .withWriterVersion(writerVersion) .build(); internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, compressor, validating, props); + extraMetaData, blockSize, compressor, validating, props, columnInfoLogging); this.memoryManager = null; this.codecFactory = null; } @@ -117,7 +119,7 @@ public ParquetRecordWriter( .withWriterVersion(writerVersion) .build(); internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, compressor, validating, props); + extraMetaData, blockSize, compressor, validating, props, DEFAULT_COLUMN_INFO_LOGGING); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); this.codecFactory = null; @@ -144,11 +146,12 @@ public ParquetRecordWriter( boolean validating, ParquetProperties props, MemoryManager memoryManager, - Configuration conf) { + Configuration conf, + boolean columnInfoLogging) { this.codecFactory = new CodecFactory(conf, props.getPageSizeThreshold()); internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, extraMetaData, blockSize, codecFactory.getCompressor(codec), validating, - props); + props, columnInfoLogging); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); } From d206a6f0845be2211c004005c932029686eccf94 Mon Sep 17 00:00:00 2001 From: Niels Basjes Date: Fri, 15 Jan 2016 16:00:37 +0100 Subject: [PATCH 03/58] Undo most of the experiment --- .../hadoop/ColumnChunkPageWriteStore.java | 9 +++---- .../hadoop/InternalParquetRecordWriter.java | 7 ++---- .../parquet/hadoop/ParquetOutputFormat.java | 4 +--- .../parquet/hadoop/ParquetRecordWriter.java | 13 ++++------ .../apache/parquet/hadoop/ParquetWriter.java | 24 ++++--------------- 5 files changed, 15 insertions(+), 42 deletions(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java index 6d54f9c719..e1453dab6e 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java @@ -62,7 +62,6 @@ private static final class ColumnChunkPageWriter implements PageWriter { private long compressedLength; private long totalValueCount; private int pageCount; - private boolean columnInfoLogging; private Set encodings = new HashSet(); @@ -71,12 +70,10 @@ private static final class ColumnChunkPageWriter implements PageWriter { private ColumnChunkPageWriter(ColumnDescriptor path, BytesCompressor compressor, - ByteBufferAllocator allocator, - boolean columnInfoLogging) { + ByteBufferAllocator allocator) { this.path = path; this.compressor = compressor; this.allocator = allocator; - this.columnInfoLogging = columnInfoLogging; this.buf = new ConcatenatingByteArrayCollector(); this.totalStatistics = getStatsBasedOnType(this.path.getType()); } @@ -229,10 +226,10 @@ public String memUsageString(String prefix) { private final Map writers = new HashMap(); private final MessageType schema; - public ColumnChunkPageWriteStore(BytesCompressor compressor, MessageType schema, ByteBufferAllocator allocator, boolean columnInfoLogging) { + public ColumnChunkPageWriteStore(BytesCompressor compressor, MessageType schema, ByteBufferAllocator allocator) { this.schema = schema; for (ColumnDescriptor path : schema.getColumns()) { - writers.put(path, new ColumnChunkPageWriter(path, compressor, allocator, columnInfoLogging)); + writers.put(path, new ColumnChunkPageWriter(path, compressor, allocator)); } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index dda39443ae..74feb39033 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -55,7 +55,6 @@ class InternalParquetRecordWriter { private final BytesCompressor compressor; private final boolean validating; private final ParquetProperties props; - private final boolean columnInfoLogging; private long recordCount = 0; private long recordCountForNextMemCheck = MINIMUM_RECORD_COUNT_FOR_CHECK; @@ -81,8 +80,7 @@ public InternalParquetRecordWriter( long rowGroupSize, BytesCompressor compressor, boolean validating, - ParquetProperties props, - boolean columnInfoLogging) { + ParquetProperties props) { this.parquetFileWriter = parquetFileWriter; this.writeSupport = checkNotNull(writeSupport, "writeSupport"); this.schema = schema; @@ -93,12 +91,11 @@ public InternalParquetRecordWriter( this.compressor = compressor; this.validating = validating; this.props = props; - this.columnInfoLogging = columnInfoLogging; initStore(); } private void initStore() { - pageStore = new ColumnChunkPageWriteStore(compressor, schema, props.getAllocator(), columnInfoLogging); + pageStore = new ColumnChunkPageWriteStore(compressor, schema, props.getAllocator()); columnStore = props.newColumnWriteStore(schema, pageStore); MessageColumnIO columnIO = new ColumnIOFactory(validating).getColumnIO(schema); this.recordConsumer = columnIO.getRecordWriter(columnStore); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index 88b3342f24..31cc96ba10 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -21,7 +21,6 @@ import static org.apache.parquet.Log.INFO; import static org.apache.parquet.Preconditions.checkNotNull; import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE; -import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_COLUMN_INFO_LOGGING; import static org.apache.parquet.hadoop.util.ContextUtil.getConfiguration; import java.io.IOException; @@ -412,8 +411,7 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp validating, props, memoryManager, - conf, - DEFAULT_COLUMN_INFO_LOGGING); + conf); } /** diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java index 31c036674d..a9ade96b00 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java @@ -33,7 +33,6 @@ import org.apache.parquet.schema.MessageType; import static org.apache.parquet.Preconditions.checkNotNull; -import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_COLUMN_INFO_LOGGING; /** * Writes records to a Parquet file @@ -73,8 +72,7 @@ public ParquetRecordWriter( int dictionaryPageSize, boolean enableDictionary, boolean validating, - WriterVersion writerVersion, - boolean columnInfoLogging) { + WriterVersion writerVersion) { ParquetProperties props = ParquetProperties.builder() .withPageSize(pageSize) .withDictionaryPageSize(dictionaryPageSize) @@ -82,7 +80,7 @@ public ParquetRecordWriter( .withWriterVersion(writerVersion) .build(); internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, compressor, validating, props, columnInfoLogging); + extraMetaData, blockSize, compressor, validating, props); this.memoryManager = null; this.codecFactory = null; } @@ -119,7 +117,7 @@ public ParquetRecordWriter( .withWriterVersion(writerVersion) .build(); internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, compressor, validating, props, DEFAULT_COLUMN_INFO_LOGGING); + extraMetaData, blockSize, compressor, validating, props); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); this.codecFactory = null; @@ -146,12 +144,11 @@ public ParquetRecordWriter( boolean validating, ParquetProperties props, MemoryManager memoryManager, - Configuration conf, - boolean columnInfoLogging) { + Configuration conf) { this.codecFactory = new CodecFactory(conf, props.getPageSizeThreshold()); internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, extraMetaData, blockSize, codecFactory.getCompressor(codec), validating, - props, columnInfoLogging); + props); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java index 63333227f7..58cbe957d0 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java @@ -45,7 +45,6 @@ public class ParquetWriter implements Closeable { public static final boolean DEFAULT_IS_VALIDATING_ENABLED = false; public static final WriterVersion DEFAULT_WRITER_VERSION = ParquetProperties.DEFAULT_WRITER_VERSION; - public static final boolean DEFAULT_COLUMN_INFO_LOGGING = true; public static final String OBJECT_MODEL_NAME_PROP = "writer.model.name"; @@ -227,8 +226,7 @@ public ParquetWriter( .withDictionaryPageSize(dictionaryPageSize) .withDictionaryEncoding(enableDictionary) .withWriterVersion(writerVersion) - .build(), - DEFAULT_COLUMN_INFO_LOGGING); + .build()); } /** @@ -267,8 +265,7 @@ public ParquetWriter(Path file, Configuration conf, WriteSupport writeSupport boolean validating, Configuration conf, int maxPaddingSize, - ParquetProperties encodingProps, - boolean columnInfoLogging) throws IOException { + ParquetProperties encodingProps) throws IOException { WriteSupport.WriteContext writeContext = writeSupport.init(conf); MessageType schema = writeContext.getSchema(); @@ -287,8 +284,7 @@ public ParquetWriter(Path file, Configuration conf, WriteSupport writeSupport blockSize, compressor, validating, - encodingProps, - columnInfoLogging); + encodingProps); } public void write(T object) throws IOException { @@ -337,7 +333,6 @@ public abstract static class Builder> { private boolean enableValidation = DEFAULT_IS_VALIDATING_ENABLED; private ParquetProperties.Builder encodingPropsBuilder = ParquetProperties.builder(); - private boolean columnInfoLogging = DEFAULT_COLUMN_INFO_LOGGING; protected Builder(Path file) { this.file = file; @@ -490,17 +485,6 @@ public SELF withWriterVersion(WriterVersion version) { } /** - * Enable or disable logging information about the columns that have been written by the writer. - * - * @param columnInfoLogging whether this type of logging should be enabled - * @return this builder for method chaining. - */ - public SELF withColumnInfoLogging(boolean columnInfoLogging) { - this.columnInfoLogging = columnInfoLogging; - return self(); - } - - /** * Build a {@link ParquetWriter} with the accumulated configuration. * * @return a configured {@code ParquetWriter} instance. @@ -509,7 +493,7 @@ public SELF withColumnInfoLogging(boolean columnInfoLogging) { public ParquetWriter build() throws IOException { return new ParquetWriter(file, mode, getWriteSupport(conf), codecName, rowGroupSize, enableValidation, conf, maxPaddingSize, - encodingPropsBuilder.build(), columnInfoLogging); + encodingPropsBuilder.build()); } } } From 225bc33637746c5e331c37e4be53db1308cf4f8b Mon Sep 17 00:00:00 2001 From: Niels Basjes Date: Fri, 15 Jan 2016 16:03:24 +0100 Subject: [PATCH 04/58] Implement a few of the the isXxxxEnabled methods in the Log class and use it for the Hadoop message --- .../src/main/java/org/apache/parquet/Log.java | 32 +++++++++++++++++++ .../hadoop/ColumnChunkPageWriteStore.java | 4 +-- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/parquet-common/src/main/java/org/apache/parquet/Log.java b/parquet-common/src/main/java/org/apache/parquet/Log.java index e05465b559..c89f92de2b 100644 --- a/parquet-common/src/main/java/org/apache/parquet/Log.java +++ b/parquet-common/src/main/java/org/apache/parquet/Log.java @@ -62,6 +62,14 @@ public Log(Class c) { this.logger = LoggerFactory.getLogger(c); } + /** + * Do we want the debug messages? + * @return If we want the debug messages. + */ + public boolean isDebugEnabled() { + return logger.isDebugEnabled(); + } + /** * prints a debug message * @param m @@ -83,6 +91,14 @@ public void debug(Object m, Throwable t) { logger.debug(String.valueOf(m), t); } + /** + * Do we want the info messages? + * @return If we want the info messages. + */ + public boolean isInfoEnabled() { + return logger.isInfoEnabled(); + } + /** * prints an info message * @param m @@ -104,6 +120,14 @@ public void info(Object m, Throwable t) { logger.info(String.valueOf(m), t); } + /** + * Do we want the warning messages? + * @return If we want the warning messages. + */ + public boolean isWarnEnabled() { + return logger.isWarnEnabled(); + } + /** * prints a warn message * @param m @@ -125,6 +149,14 @@ public void warn(Object m, Throwable t) { logger.warn(String.valueOf(m), t); } + /** + * Do we want the error messages? + * @return If we want the error messages. + */ + public boolean isErrorEnabled() { + return logger.isErrorEnabled(); + } + /** * prints an error message * @param m diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java index e1453dab6e..6afd54962f 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java @@ -186,8 +186,8 @@ public void writeToFileWriter(ParquetFileWriter writer) throws IOException { } writer.writeDataPages(buf, uncompressedLength, compressedLength, totalStatistics, new ArrayList(encodings)); writer.endColumn(); - if (INFO && columnInfoLogging) { - LOG.info( + if (LOG.isDebugEnabled()) { + LOG.debug( String.format( "written %,dB for %s: %,d values, %,dB raw, %,dB comp, %d pages, encodings: %s", buf.size(), path, totalValueCount, uncompressedLength, compressedLength, pageCount, encodings) From 73da74a264dd81d880cad3d89dc075efcfeb3ea3 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Tue, 12 Jan 2016 14:15:40 -0800 Subject: [PATCH 05/58] PARQUET-212: Implement LIST read compatibility rules in Thrift This implements the read-side compatibility rules for 2-level and 3-level lists in Thrift. Thrift doesn't allow null elements inside lists, but 3-level lists may have optional elements. This PR adds a property, parquet.thrift.ignore-null-elements, that allows thrift to read lists with optional elements by ignoring nulls. This is off by default, but is provided as an opt-in for compatibility with data written by Hive. Thrift's schema conversion does not change because a Thrift class (or Scrooge etc.) must be set in a file's metadata or provided when constructing a reader. This replaces and closes #144. Author: Ryan Blue Closes #300 from rdblue/PARQUET-212-fix-thrift-3-level-lists and squashes the following commits: ac7c405 [Ryan Blue] PARQUET-212: Add tests for list of list cases from PARQUET-364. 356fdb7 [Ryan Blue] PARQUET-212: Rename isElementType => isListElementType. 5d3b094 [Ryan Blue] PARQUET-212: Fix list handling with projection. b5f207f [Ryan Blue] PARQUET-212: Add Configuration to the ThriftRecordConverter ctor. b87eb65 [Ryan Blue] PARQUET-212: Add property to ignore nulls in lists. 3d1e92f [Ryan Blue] PARQUET-212: Update thrift reads for LIST compatibility rules. 0bf2b45 [Ryan Blue] PARQUET-212: Read non-thrift files if a Thrift class is supplied. 4e148dc [Ryan Blue] PARQUET-212: Add DirectWriterTest base class. --- parquet-avro/pom.xml | 7 + .../parquet/avro/AvroRecordConverter.java | 13 +- .../parquet/avro/TestArrayCompatibility.java | 76 +- .../org/apache/parquet/DirectWriterTest.java | 102 +++ .../scrooge/ScroogeRecordConverter.java | 13 +- parquet-thrift/pom.xml | 7 + .../hadoop/thrift/ThriftReadSupport.java | 63 +- .../parquet/thrift/TBaseRecordConverter.java | 12 +- .../apache/parquet/thrift/ThriftMetaData.java | 21 +- .../parquet/thrift/ThriftRecordConverter.java | 123 ++- .../parquet/thrift/ThriftSchemaConverter.java | 38 + .../hadoop/thrift/TestArrayCompatibility.java | 779 ++++++++++++++++++ .../src/test/thrift/array_compat.thrift | 53 ++ 13 files changed, 1193 insertions(+), 114 deletions(-) create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/DirectWriterTest.java 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-avro/pom.xml b/parquet-avro/pom.xml index 94343438df..aad197d47b 100644 --- a/parquet-avro/pom.xml +++ b/parquet-avro/pom.xml @@ -87,6 +87,13 @@ ${slf4j.version} test + + org.apache.parquet + parquet-hadoop + ${project.version} + test-jar + test + 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 61d7d8ef55..38a761c40a 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 @@ -32,8 +32,10 @@ import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import org.apache.avro.Schema; import org.apache.avro.generic.GenericData; import org.apache.avro.reflect.ReflectData; @@ -753,13 +755,14 @@ static boolean isElementType(Type repeatedType, Schema elementSchema) { // synthetic wrapper. Must be a group with one optional or required 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-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java b/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java index 9c29e5030f..29264f06a5 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(); @@ -1045,68 +1035,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/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-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/pom.xml b/parquet-thrift/pom.xml index 2583870c9e..66594ca0ff 100644 --- a/parquet-thrift/pom.xml +++ b/parquet-thrift/pom.xml @@ -121,6 +121,13 @@ ${slf4j.version} test + + org.apache.parquet + parquet-hadoop + ${project.version} + test-jar + test + 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 a859128a74..1c020ae4cf 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; @@ -203,17 +204,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,17 +225,56 @@ 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); + } 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); + } + + 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()); - 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/ThriftMetaData.java b/parquet-thrift/src/main/java/org/apache/parquet/thrift/ThriftMetaData.java index a89f8d97c1..a7628cc6e6 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); 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 e18b0e6d17..3160d5fce2 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,7 @@ import java.util.List; import java.util.Map; -import org.apache.parquet.io.ParquetDecodingException; +import org.apache.hadoop.conf.Configuration; import org.apache.thrift.TException; import org.apache.thrift.protocol.TField; import org.apache.thrift.protocol.TList; @@ -34,6 +34,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; @@ -63,6 +66,12 @@ */ public class ThriftRecordConverter extends RecordMaterializer { + 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 static ParquetProtocol readFieldEnd = new ParquetProtocol("readFieldEnd()") { @Override public void readFieldEnd() throws TException { @@ -472,7 +481,7 @@ public int readI32() throws TException { * @author Julien Le Dem * */ - static class MapConverter extends GroupConverter { + class MapConverter extends GroupConverter { private final GroupCounter child; private final List mapEvents = new ArrayList(); @@ -534,7 +543,7 @@ public TMap readMapBegin() throws TException { * @author Julien Le Dem * */ - static class MapKeyValueConverter extends GroupConverter { + class MapKeyValueConverter extends GroupConverter { private Converter keyConverter; private Converter valueConverter; @@ -572,7 +581,7 @@ public void end() { * @author Julien Le Dem * */ - static class SetConverter extends CollectionConverter { + class SetConverter extends CollectionConverter { final ParquetProtocol readSetEnd = new ParquetProtocol("readSetEnd()") { @Override @@ -609,7 +618,7 @@ void collectionEnd() { * @author Julien Le Dem * */ - static class ListConverter extends CollectionConverter { + class ListConverter extends CollectionConverter { final ParquetProtocol readListEnd = new ParquetProtocol("readListEnd()") { @Override @@ -646,28 +655,36 @@ void collectionEnd() { * @author Julien Le Dem * */ - static abstract class CollectionConverter extends GroupConverter { + 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.isListElementType(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; } @@ -689,7 +706,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(); @@ -702,12 +722,63 @@ 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)) { + 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); + } + + @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 * */ - static class StructConverter extends GroupConverter { + class StructConverter extends GroupConverter { private final int schemaSize; @@ -789,9 +860,19 @@ public void end() { } private final ThriftReader thriftReader; private final ParquetReadProtocol protocol; - private final GroupConverter structConverter; + private GroupConverter structConverter; private List rootEvents = new ArrayList(); private boolean missingRequiredFieldsInProjection = false; + 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); + } /** * @@ -799,12 +880,18 @@ public void end() { * @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.protocol = new ParquetReadProtocol(); this.thriftType = thriftType; + if (conf != null) { + this.ignoreNullElements = conf.getBoolean( + IGNORE_NULL_LIST_ELEMENTS, + IGNORE_NULL_LIST_ELEMENTS_DEFAULT); + } MessageType fullSchema = ThriftSchemaConverter.convertWithoutProjection(thriftType); missingRequiredFieldsInProjection = hasMissingRequiredFieldInGroupType(requestedParquetSchema, fullSchema); this.structConverter = new StructConverter(rootEvents, requestedParquetSchema, new ThriftField(name, (short)0, Requirement.REQUIRED, thriftType)); @@ -874,7 +961,7 @@ public GroupConverter getRootConverter() { return structConverter; } - private static Converter newConverter(List events, Type type, ThriftField field) { + private Converter newConverter(List events, Type type, ThriftField field) { switch (field.getType().getType()) { case LIST: return new ListConverter(events, type.asGroupType(), field); 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 98820c37ee..b72f605ac3 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,10 +20,13 @@ import com.twitter.elephantbird.thrift.TStructDescriptor; import com.twitter.elephantbird.thrift.TStructDescriptor.Field; +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; @@ -37,6 +40,8 @@ import java.util.Collection; import java.util.List; +import static org.apache.parquet.schema.Type.Repetition.REPEATED; + /** * Given a thrift class, this class converts it to parquet schema, * a {@link FieldProjectionFilter} can be specified for projection pushdown. @@ -98,6 +103,39 @@ 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 isListElementType(Type repeatedType, + ThriftField thriftElement) { + if (repeatedType.isPrimitive() || + (repeatedType.asGroupType().getFieldCount() != 1) || + (repeatedType.asGroupType().getType(0).isRepetition(REPEATED))) { + // The repeated type must be the element type because it is an invalid + // synthetic wrapper. Must be a group with one optional or required field + return true; + } else if (thriftElement != null && thriftElement.getType() instanceof StructType) { + 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; + } + 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..df6154a464 --- /dev/null +++ b/parquet-thrift/src/test/java/org/apache/parquet/hadoop/thrift/TestArrayCompatibility.java @@ -0,0 +1,779 @@ +/** + * 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.Arrays; +import java.util.List; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericRecord; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.thrift.test.compat.ListOfLists; +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.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; +import org.apache.parquet.thrift.test.compat.ListOfSingleElementGroups; +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 + @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); + } + + @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)); + + 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().getCause().getMessage().contains("locations")); + } + + assertReaderContains(readerIgnoreNulls(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 testAvroCompatListInList() throws Exception { + Path test = writeDirect( + "message AvroCompatListInList {" + + " optional group listOfLists (LIST) {" + + " repeated group array (LIST) {" + + " repeated int32 array;" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("array", 0); // start writing array contents + + rc.startGroup(); + rc.startField("array", 0); // start writing inner array contents + + // write [34, 35, 36] + rc.addInteger(34); + rc.addInteger(35); + rc.addInteger(36); + + rc.endField("array", 0); // finished writing inner array contents + rc.endGroup(); + + // write an empty list + rc.startGroup(); + rc.endGroup(); + + rc.startGroup(); + rc.startField("array", 0); // start writing inner array contents + + // write [32, 33, 34] + rc.addInteger(32); + rc.addInteger(33); + rc.addInteger(34); + + rc.endField("array", 0); // finished writing inner array contents + rc.endGroup(); + + rc.endField("array", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLists expected = new ListOfLists(); + expected.addToListOfLists(Arrays.asList(34, 35, 36)); + expected.addToListOfLists(Arrays.asList()); + expected.addToListOfLists(Arrays.asList(32, 33, 34)); + + // should detect the "array" name + assertReaderContains(reader(test, ListOfLists.class), expected); + } + + @Test + public void testThriftCompatListInList() throws Exception { + Path test = writeDirect( + "message ThriftCompatListInList {" + + " optional group listOfLists (LIST) {" + + " repeated group listOfLists_tuple (LIST) {" + + " repeated int32 listOfLists_tuple_tuple;" + + " }" + + " }" + + "}", + new DirectWriter() { + @Override + public void write(RecordConsumer rc) { + rc.startMessage(); + rc.startField("locations", 0); + + rc.startGroup(); + rc.startField("listOfLists_tuple", 0); // start writing array contents + + rc.startGroup(); + rc.startField("listOfLists_tuple_tuple", 0); // start writing inner array contents + + // write [34, 35, 36] + rc.addInteger(34); + rc.addInteger(35); + rc.addInteger(36); + + rc.endField("listOfLists_tuple_tuple", 0); // finished writing inner array contents + rc.endGroup(); + + // write an empty list + rc.startGroup(); + rc.endGroup(); + + rc.startGroup(); + rc.startField("listOfLists_tuple_tuple", 0); // start writing inner array contents + + // write [32, 33, 34] + rc.addInteger(32); + rc.addInteger(33); + rc.addInteger(34); + + rc.endField("listOfLists_tuple_tuple", 0); // finished writing inner array contents + rc.endGroup(); + + rc.endField("listOfLists_tuple", 0); // finished writing array contents + rc.endGroup(); + + rc.endField("locations", 0); + rc.endMessage(); + } + }); + + ListOfLists expected = new ListOfLists(); + expected.addToListOfLists(Arrays.asList(34, 35, 36)); + expected.addToListOfLists(Arrays.asList()); + expected.addToListOfLists(Arrays.asList(32, 33, 34)); + + // should detect the "_tuple" names + assertReaderContains(reader(test, ListOfLists.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)); + + 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().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); + } +} 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..ce03a59fde --- /dev/null +++ b/parquet-thrift/src/test/thrift/array_compat.thrift @@ -0,0 +1,53 @@ +/** + * 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; +} + +struct ListOfLists { + 1: optional list> listOfLists; +} From 2aa830c9a5ecd397279777ed96e47adfc23ea385 Mon Sep 17 00:00:00 2001 From: proflin Date: Tue, 12 Jan 2016 14:21:32 -0800 Subject: [PATCH 06/58] PARQUET-421: Fix mismatch of javadoc names and method parameters in m... MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit …odule encoding, column, and hadoop Codes change now and then, but some corresponding doc comments are left out. This PR fixes only the doc comments that should have been changed. It should be OK, since none codes are touched. @rdblue could you take a look please? Cheers. Author: proflin Closes #307 from proflin/Minor--Fix-the-mismatch-of-the-parameters-and-their-doc-comments-in-module-encoding,-column,-and-hadoop and squashes the following commits: 34c7b01 [proflin] Minor: Fix the mismatch of the parameters and their doc comments in module encoding, column, and hadoop --- .../org/apache/parquet/column/page/DataPageV1.java | 4 ++-- .../apache/parquet/column/values/ValuesWriter.java | 14 +++++++------- .../values/plain/BooleanPlainValuesReader.java | 4 ++-- .../org/apache/parquet/schema/TypeConverter.java | 2 +- .../column/values/bitpacking/BitPacking.java | 2 +- .../hadoop/example/ExampleOutputFormat.java | 6 +++--- .../parquet/hadoop/metadata/ParquetMetadata.java | 1 - 7 files changed, 16 insertions(+), 17 deletions(-) mode change 100644 => 100755 parquet-column/src/main/java/org/apache/parquet/column/page/DataPageV1.java mode change 100644 => 100755 parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java mode change 100644 => 100755 parquet-column/src/main/java/org/apache/parquet/column/values/plain/BooleanPlainValuesReader.java mode change 100644 => 100755 parquet-column/src/main/java/org/apache/parquet/schema/TypeConverter.java mode change 100644 => 100755 parquet-encoding/src/main/java/org/apache/parquet/column/values/bitpacking/BitPacking.java mode change 100644 => 100755 parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java mode change 100644 => 100755 parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java diff --git a/parquet-column/src/main/java/org/apache/parquet/column/page/DataPageV1.java b/parquet-column/src/main/java/org/apache/parquet/column/page/DataPageV1.java old mode 100644 new mode 100755 index 22065178b4..6fbfe0caf7 --- a/parquet-column/src/main/java/org/apache/parquet/column/page/DataPageV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/page/DataPageV1.java @@ -41,10 +41,10 @@ public class DataPageV1 extends DataPage { * @param valuesEncoding the values encoding for this page * @param dlEncoding */ - public DataPageV1(BytesInput bytes, int valueCount, int uncompressedSize, Statistics stats, Encoding rlEncoding, Encoding dlEncoding, Encoding valuesEncoding) { + public DataPageV1(BytesInput bytes, int valueCount, int uncompressedSize, Statistics statistics, Encoding rlEncoding, Encoding dlEncoding, Encoding valuesEncoding) { super(Ints.checkedCast(bytes.size()), uncompressedSize, valueCount); this.bytes = bytes; - this.statistics = stats; + this.statistics = statistics; this.rlEncoding = rlEncoding; this.dlEncoding = dlEncoding; this.valuesEncoding = valuesEncoding; diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java b/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java old mode 100644 new mode 100755 index e5cf8e9441..3862b02eef --- a/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/values/ValuesWriter.java @@ -80,7 +80,7 @@ public void resetDictionary() { } /** - * ( > {@link #getBufferedMemorySize} ) + * ( > {@link #getBufferedSize} ) * @return the allocated size of the buffer */ abstract public long getAllocatedSize(); @@ -93,42 +93,42 @@ public void writeByte(int value) { } /** - * @param value the value to encode + * @param v the value to encode */ public void writeBoolean(boolean v) { throw new UnsupportedOperationException(getClass().getName()); } /** - * @param value the value to encode + * @param v the value to encode */ public void writeBytes(Binary v) { throw new UnsupportedOperationException(getClass().getName()); } /** - * @param value the value to encode + * @param v the value to encode */ public void writeInteger(int v) { throw new UnsupportedOperationException(getClass().getName()); } /** - * @param value the value to encode + * @param v the value to encode */ public void writeLong(long v) { throw new UnsupportedOperationException(getClass().getName()); } /** - * @param value the value to encode + * @param v the value to encode */ public void writeDouble(double v) { throw new UnsupportedOperationException(getClass().getName()); } /** - * @param value the value to encode + * @param v the value to encode */ public void writeFloat(float v) { throw new UnsupportedOperationException(getClass().getName()); diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/plain/BooleanPlainValuesReader.java b/parquet-column/src/main/java/org/apache/parquet/column/values/plain/BooleanPlainValuesReader.java old mode 100644 new mode 100755 index c3304907e2..a279938c69 --- a/parquet-column/src/main/java/org/apache/parquet/column/values/plain/BooleanPlainValuesReader.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/values/plain/BooleanPlainValuesReader.java @@ -51,7 +51,7 @@ public boolean readBoolean() { /** * {@inheritDoc} - * @see org.apache.parquet.column.values.ValuesReader#skipBoolean() + * @see org.apache.parquet.column.values.ValuesReader#skip() */ @Override public void skip() { @@ -60,7 +60,7 @@ public void skip() { /** * {@inheritDoc} - * @see org.apache.parquet.column.values.ValuesReader#initFromPage(byte[], int) + * @see org.apache.parquet.column.values.ValuesReader#initFromPage(int valueCount, ByteBuffer page, int offset) */ @Override public void initFromPage(int valueCount, ByteBuffer in, int offset) throws IOException { diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/TypeConverter.java b/parquet-column/src/main/java/org/apache/parquet/schema/TypeConverter.java old mode 100644 new mode 100755 index c22a877542..8be896fac0 --- a/parquet-column/src/main/java/org/apache/parquet/schema/TypeConverter.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/TypeConverter.java @@ -22,7 +22,7 @@ /** * to convert a MessageType tree - * @see Type#convert(TypeConverter) + * @see Type#convert(List, TypeConverter) * * @author Julien Le Dem * diff --git a/parquet-encoding/src/main/java/org/apache/parquet/column/values/bitpacking/BitPacking.java b/parquet-encoding/src/main/java/org/apache/parquet/column/values/bitpacking/BitPacking.java old mode 100644 new mode 100755 index c0acd11d80..7b557c7def --- a/parquet-encoding/src/main/java/org/apache/parquet/column/values/bitpacking/BitPacking.java +++ b/parquet-encoding/src/main/java/org/apache/parquet/column/values/bitpacking/BitPacking.java @@ -109,7 +109,7 @@ public static BitPackingWriter getBitPackingWriter(int bitLength, OutputStream o /** * * @param bitLength the width in bits of the integers to read - * @param inthe stream to read the bytes from + * @param in the stream to read the bytes from * @return the correct implementation for the width */ public static BitPackingReader createBitPackingReader(int bitLength, InputStream in, long valueCount) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java old mode 100644 new mode 100755 index 08184da187..d503e0d3e0 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/ExampleOutputFormat.java @@ -30,7 +30,7 @@ * An example output format * * must be provided the schema up front - * @see ExampleOutputFormat#setSchema(Configuration, MessageType) + * @see ExampleOutputFormat#setSchema(Job, MessageType) * @see GroupWriteSupport#PARQUET_EXAMPLE_SCHEMA * * @author Julien Le Dem @@ -40,8 +40,8 @@ public class ExampleOutputFormat extends ParquetOutputFormat { /** * set the schema being written to the job conf + * @param job * @param schema the schema of the data - * @param configuration the job configuration */ public static void setSchema(Job job, MessageType schema) { GroupWriteSupport.setSchema(schema, ContextUtil.getConfiguration(job)); @@ -49,7 +49,7 @@ public static void setSchema(Job job, MessageType schema) { /** * retrieve the schema from the conf - * @param configuration the job conf + * @param job * @return the schema */ public static MessageType getSchema(Job job) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java old mode 100644 new mode 100755 index a645117505..cb6af54ebb --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/metadata/ParquetMetadata.java @@ -101,7 +101,6 @@ public static ParquetMetadata fromJSON(String json) { * * @param fileMetaData file level metadata * @param blocks block level metadata - * @param keyValueMetaData */ public ParquetMetadata(FileMetaData fileMetaData, List blocks) { this.fileMetaData = fileMetaData; From 96c862ebaf4f447cab0ad0f303a4164fa720a9b5 Mon Sep 17 00:00:00 2001 From: proflin Date: Tue, 12 Jan 2016 14:45:24 -0800 Subject: [PATCH 07/58] =?UTF-8?q?PARQUET-422:=20Fix=20a=20potential=20bug?= =?UTF-8?q?=20in=20MessageTypeParser=20where=20we=20ignore=E2=80=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit … and overwrite the initial value of a method parameter In org.apache.parquet.schema.MessageTypeParser, for addGroupType() and addPrimitiveType(), the initial value of this parameter t is ignored, and t is overwritten here. This often indicates a mistaken belief that the write to the parameter will be conveyed back to the caller. This is a bug found by FindBugs™. Author: proflin Closes #308 from proflin/PARQUET-422 and squashes the following commits: df1f908 [proflin] PARQUET-422: Fix a potential bug in MessageTypeParser where we ignore and overwrite the initial value of a method parameter --- .../org/apache/parquet/schema/MessageTypeParser.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java b/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java index 3603c7995f..49813987df 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java @@ -106,14 +106,15 @@ private static void addType(String t, Tokenizer st, Types.GroupBuilder builder) // Read type. String type = st.nextToken(); if ("group".equalsIgnoreCase(type)) { - addGroupType(t, st, repetition, builder); + addGroupType(st, repetition, builder); } else { - addPrimitiveType(t, st, asPrimitive(type, st), repetition, builder); + addPrimitiveType(st, asPrimitive(type, st), repetition, builder); } } - private static void addGroupType(String t, Tokenizer st, Repetition r, GroupBuilder builder) { + private static void addGroupType(Tokenizer st, Repetition r, GroupBuilder builder) { GroupBuilder childBuilder = builder.group(r); + String t; String name = st.nextToken(); // Read annotation, if any. @@ -138,8 +139,9 @@ private static void addGroupType(String t, Tokenizer st, Repetition r, GroupBuil childBuilder.named(name); } - private static void addPrimitiveType(String t, Tokenizer st, PrimitiveTypeName type, Repetition r, Types.GroupBuilder builder) { + private static void addPrimitiveType(Tokenizer st, PrimitiveTypeName type, Repetition r, Types.GroupBuilder builder) { PrimitiveBuilder childBuilder = builder.primitive(type, r); + String t; if (type == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) { t = st.nextToken(); From fd9b49b02399ae72b0d9f72c57734d5f4db30959 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Thu, 28 Jan 2016 17:33:08 -0800 Subject: [PATCH 08/58] PARQUET-393: Update to parquet-format 2.3.1. Author: Ryan Blue Closes #303 from rdblue/PARQUET-393-update-parquet-format-version and squashes the following commits: 0e4c798 [Ryan Blue] PARQUET-393: Add TIME_MICROS and TIMESTAMP_MICROS. ca4a741 [Ryan Blue] PARQUET-393: Update to parquet-format 2.3.1. --- .../java/org/apache/parquet/schema/OriginalType.java | 2 ++ .../src/main/java/org/apache/parquet/schema/Types.java | 2 ++ .../java/org/apache/parquet/schema/TestTypeBuilders.java | 4 ++-- .../format/converter/ParquetMetadataConverter.java | 9 ++++++++- pom.xml | 2 +- 5 files changed, 15 insertions(+), 4 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/OriginalType.java b/parquet-column/src/main/java/org/apache/parquet/schema/OriginalType.java index 0ea89ba8ba..77acc5452b 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/OriginalType.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/OriginalType.java @@ -27,7 +27,9 @@ public enum OriginalType { DECIMAL, DATE, TIME_MILLIS, + TIME_MICROS, TIMESTAMP_MILLIS, + TIMESTAMP_MICROS, UINT_8, UINT_16, UINT_32, diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/Types.java b/parquet-column/src/main/java/org/apache/parquet/schema/Types.java index 947a84565d..4e7e8815c2 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/Types.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/Types.java @@ -424,7 +424,9 @@ protected PrimitiveType build(String name) { Preconditions.checkState(primitiveType == PrimitiveTypeName.INT32, originalType.toString() + " can only annotate INT32"); break; + case TIME_MICROS: case TIMESTAMP_MILLIS: + case TIMESTAMP_MICROS: case UINT_64: case INT_64: Preconditions.checkState(primitiveType == PrimitiveTypeName.INT64, diff --git a/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuilders.java b/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuilders.java index 5fac7bc892..0c39ef2ba1 100644 --- a/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuilders.java +++ b/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuilders.java @@ -509,7 +509,7 @@ public Type call() throws Exception { @Test public void testInt64Annotations() { OriginalType[] types = new OriginalType[] { - TIMESTAMP_MILLIS, UINT_64, INT_64}; + TIME_MICROS, TIMESTAMP_MILLIS, TIMESTAMP_MICROS, UINT_64, INT_64}; for (OriginalType logicalType : types) { PrimitiveType expected = new PrimitiveType(REQUIRED, INT64, "col", logicalType); PrimitiveType date = Types.required(INT64).as(logicalType).named("col"); @@ -520,7 +520,7 @@ public void testInt64Annotations() { @Test public void testInt64AnnotationsRejectNonInt64() { OriginalType[] types = new OriginalType[] { - TIMESTAMP_MILLIS, UINT_64, INT_64}; + TIME_MICROS, TIMESTAMP_MILLIS, TIMESTAMP_MICROS, UINT_64, INT_64}; for (final OriginalType logicalType : types) { PrimitiveTypeName[] nonInt64 = new PrimitiveTypeName[]{ BOOLEAN, INT32, INT96, DOUBLE, FLOAT, BINARY diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java index 6821bbfff4..b373bfb1fd 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java @@ -36,7 +36,6 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import org.apache.hadoop.fs.FSDataInputStream; import org.apache.parquet.CorruptStatistics; import org.apache.parquet.Log; import org.apache.parquet.hadoop.metadata.ColumnPath; @@ -336,8 +335,12 @@ OriginalType getOriginalType(ConvertedType type) { return OriginalType.DATE; case TIME_MILLIS: return OriginalType.TIME_MILLIS; + case TIME_MICROS: + return OriginalType.TIME_MICROS; case TIMESTAMP_MILLIS: return OriginalType.TIMESTAMP_MILLIS; + case TIMESTAMP_MICROS: + return OriginalType.TIMESTAMP_MICROS; case INTERVAL: return OriginalType.INTERVAL; case INT_8: @@ -384,8 +387,12 @@ ConvertedType getConvertedType(OriginalType type) { return ConvertedType.DATE; case TIME_MILLIS: return ConvertedType.TIME_MILLIS; + case TIME_MICROS: + return ConvertedType.TIME_MICROS; case TIMESTAMP_MILLIS: return ConvertedType.TIMESTAMP_MILLIS; + case TIMESTAMP_MICROS: + return ConvertedType.TIMESTAMP_MICROS; case INTERVAL: return ConvertedType.INTERVAL; case INT_8: diff --git a/pom.xml b/pom.xml index 3588de567d..94d7a02589 100644 --- a/pom.xml +++ b/pom.xml @@ -79,7 +79,7 @@ shaded.parquet 1.1.0 2.5.3 - 2.3.0-incubating + 2.3.1 1.7.0 thrift 2.10.4 From 84faceeeff79e4bde2a32e70fbfdf32cb6d891c2 Mon Sep 17 00:00:00 2001 From: proflin Date: Fri, 29 Jan 2016 11:38:34 -0800 Subject: [PATCH 09/58] PARQUET-432: Complete a todo for method ColumnDescriptor.compareTo() The ticket proposes to consider the case *path.length < o.path.length* in, for method ColumnDescriptor.compareTo(). Author: proflin Closes #314 from proflin/PARQUET-432 and squashes the following commits: 80ba94b [proflin] Addresses PR comments 6ccd00f [proflin] Revert Updates a4d2a4a [proflin] PARQUET-432: Complete a todo in method ColumnDescriptor.compareTo() 694b76b [proflin] Updates --- .../parquet/column/ColumnDescriptor.java | 6 +-- .../parquet/column/TestColumnDescriptor.java | 52 +++++++++++++++++++ 2 files changed, 55 insertions(+), 3 deletions(-) create mode 100644 parquet-column/src/test/java/org/apache/parquet/column/TestColumnDescriptor.java diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ColumnDescriptor.java b/parquet-column/src/main/java/org/apache/parquet/column/ColumnDescriptor.java index 8244913df5..61f13a2740 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/ColumnDescriptor.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/ColumnDescriptor.java @@ -117,14 +117,14 @@ public boolean equals(Object other) { @Override public int compareTo(ColumnDescriptor o) { - // TODO(julien): this will fail if o.path.length < this.path.length - for (int i = 0; i < path.length; i++) { + int length = path.length < o.path.length ? path.length : o.path.length; + for (int i = 0; i < length; i++) { int compareTo = path[i].compareTo(o.path[i]); if (compareTo != 0) { return compareTo; } } - return 0; + return path.length - o.path.length; } @Override diff --git a/parquet-column/src/test/java/org/apache/parquet/column/TestColumnDescriptor.java b/parquet-column/src/test/java/org/apache/parquet/column/TestColumnDescriptor.java new file mode 100644 index 0000000000..33d9cea43d --- /dev/null +++ b/parquet-column/src/test/java/org/apache/parquet/column/TestColumnDescriptor.java @@ -0,0 +1,52 @@ +/* + * 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.column; + +import org.apache.parquet.schema.PrimitiveType; +import org.junit.Test; + +import static junit.framework.Assert.assertEquals; + +public class TestColumnDescriptor { + + private ColumnDescriptor column(String... path) { + return new ColumnDescriptor(path, PrimitiveType.PrimitiveTypeName.INT32, 0, 0); + } + + @Test + public void testComparesTo() throws Exception { + assertEquals(column("a").compareTo(column("a")), 0); + assertEquals(column("a", "b").compareTo(column("a", "b")), 0); + + assertEquals(column("a").compareTo(column("b")), -1); + assertEquals(column("b").compareTo(column("a")), 1); + assertEquals(column("a", "a").compareTo(column("a", "b")), -1); + assertEquals(column("b", "a").compareTo(column("a", "a")), 1); + + assertEquals(column("a").compareTo(column("a", "b")), -1); + assertEquals(column("b").compareTo(column("a", "b")), 1); + + assertEquals(column("a", "b").compareTo(column("a")), 1); + assertEquals(column("a", "b").compareTo(column("b")), -1); + + assertEquals(column("").compareTo(column("")), 0); + assertEquals(column("").compareTo(column("a")), -1); + assertEquals(column("a").compareTo(column("")), 1); + } +} From 0c054b02d20071306bcfc26278a9d3debe9d2627 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cyrille=20Ch=C3=A9p=C3=A9lov=20=28TP12=29?= Date: Sun, 31 Jan 2016 19:21:48 -0800 Subject: [PATCH 10/58] PARQUET-480: Update for Cascading 3.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The code in parquet-cascading is adapted to the API as of Cascading 2.5.3 Some incompatible changes were introduced in Cascading 3.0. This patch forks the parquet-cascading module to also provide a parquet-cascading3 module, which is about identical save for overloads which changed from requiring a Foo to requiring a Foo Author: Cyrille Chépélov (TP12) Closes #284 from cchepelov/try_cascading3 and squashes the following commits: e7d1304 [Cyrille Chépélov (TP12)] Adding a @Deprecated notice on parquet-cascading's remaining classes 05a417d [Cyrille Chépélov (TP12)] cascading2/3: share back TupleWriteSupport.java (accidentally unmerged) 7fff2d4 [Cyrille Chépélov (TP12)] cascading/cascading3: remove duplicates, push common files into parquet-cascading-common23 338a416 [Cyrille Chépélov (TP12)] Removing unwanted file (what?!) + .gitignoring this kind of files d9f0455 [Cyrille Chépélov (TP12)] TupleEntry#get is now TupleEntry#getObject a7f490a [Cyrille Chépélov (TP12)] Revert "Missing test conversion to Cascading 3.0" cc8b870 [Cyrille Chépélov (TP12)] Missing test conversion to Cascading 3.0 2d73512 [Cyrille Chépélov (TP12)] conflicting values can come in one order or the other. Accept both. 33355d5 [Cyrille Chépélov (TP12)] Fix version mismatch (duh!) 7128639 [Cyrille Chépélov (TP12)] non-C locale can break tests implementation (decimal formats) 53aa2f9 [Cyrille Chépélov (TP12)] Adding a parquet-cascading3 module (forking the parquet-cascading module and accounting for API changes) --- .gitignore | 3 + README.md | 2 +- .../parquet/cascading/SchemaIntersection.java | 0 .../parquet/cascading/TupleReadSupport.java | 0 .../parquet/cascading/TupleWriteSupport.java | 0 .../cascading/convert/TupleConverter.java | 0 .../convert/TupleRecordMaterializer.java | 0 .../cascading/TestParquetTupleScheme.java | 0 .../src/test/resources/names.txt | 0 .../src/test/thrift/test.thrift | 0 parquet-cascading/pom.xml | 47 +++++ .../parquet/cascading/ParquetTBaseScheme.java | 1 + .../parquet/cascading/ParquetTupleScheme.java | 1 + .../parquet/cascading/ParquetValueScheme.java | 1 + .../cascading/TestParquetTBaseScheme.java | 3 +- parquet-cascading3/REVIEWERS.md | 27 +++ parquet-cascading3/pom.xml | 178 ++++++++++++++++ .../parquet/cascading/ParquetTBaseScheme.java | 80 ++++++++ .../parquet/cascading/ParquetTupleScheme.java | 191 ++++++++++++++++++ .../parquet/cascading/ParquetValueScheme.java | 184 +++++++++++++++++ .../cascading/TestParquetTBaseScheme.java | 186 +++++++++++++++++ .../hadoop/TestMergeMetadataFiles.java | 10 +- parquet_cascading.md | 13 ++ pom.xml | 26 +++ 24 files changed, 949 insertions(+), 4 deletions(-) rename {parquet-cascading => parquet-cascading-common23}/src/main/java/org/apache/parquet/cascading/SchemaIntersection.java (100%) rename {parquet-cascading => parquet-cascading-common23}/src/main/java/org/apache/parquet/cascading/TupleReadSupport.java (100%) rename {parquet-cascading => parquet-cascading-common23}/src/main/java/org/apache/parquet/cascading/TupleWriteSupport.java (100%) rename {parquet-cascading => parquet-cascading-common23}/src/main/java/org/apache/parquet/cascading/convert/TupleConverter.java (100%) rename {parquet-cascading => parquet-cascading-common23}/src/main/java/org/apache/parquet/cascading/convert/TupleRecordMaterializer.java (100%) rename {parquet-cascading => parquet-cascading-common23}/src/test/java/org/apache/parquet/cascading/TestParquetTupleScheme.java (100%) rename {parquet-cascading => parquet-cascading-common23}/src/test/resources/names.txt (100%) rename {parquet-cascading => parquet-cascading-common23}/src/test/thrift/test.thrift (100%) create mode 100644 parquet-cascading3/REVIEWERS.md create mode 100644 parquet-cascading3/pom.xml create mode 100644 parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java create mode 100644 parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java create mode 100644 parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetValueScheme.java create mode 100644 parquet-cascading3/src/test/java/org/apache/parquet/cascading/TestParquetTBaseScheme.java diff --git a/.gitignore b/.gitignore index cd3c0669ca..aa67d3d37a 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,6 @@ dependency-reduced-pom.xml parquet-scrooge/.cache .idea/* target/ +.cache +*~ +mvn_install.log diff --git a/README.md b/README.md index 2d9a50a124..9bb0be661c 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,7 @@ sudo make install Once protobuf and thrift are available in your path, you can build the project by running: ``` -mvn clean install +LC_ALL=C mvn clean install ``` ## Features diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/SchemaIntersection.java b/parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/SchemaIntersection.java similarity index 100% rename from parquet-cascading/src/main/java/org/apache/parquet/cascading/SchemaIntersection.java rename to parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/SchemaIntersection.java diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/TupleReadSupport.java b/parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/TupleReadSupport.java similarity index 100% rename from parquet-cascading/src/main/java/org/apache/parquet/cascading/TupleReadSupport.java rename to parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/TupleReadSupport.java diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/TupleWriteSupport.java b/parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/TupleWriteSupport.java similarity index 100% rename from parquet-cascading/src/main/java/org/apache/parquet/cascading/TupleWriteSupport.java rename to parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/TupleWriteSupport.java diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/convert/TupleConverter.java b/parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/convert/TupleConverter.java similarity index 100% rename from parquet-cascading/src/main/java/org/apache/parquet/cascading/convert/TupleConverter.java rename to parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/convert/TupleConverter.java diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/convert/TupleRecordMaterializer.java b/parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/convert/TupleRecordMaterializer.java similarity index 100% rename from parquet-cascading/src/main/java/org/apache/parquet/cascading/convert/TupleRecordMaterializer.java rename to parquet-cascading-common23/src/main/java/org/apache/parquet/cascading/convert/TupleRecordMaterializer.java diff --git a/parquet-cascading/src/test/java/org/apache/parquet/cascading/TestParquetTupleScheme.java b/parquet-cascading-common23/src/test/java/org/apache/parquet/cascading/TestParquetTupleScheme.java similarity index 100% rename from parquet-cascading/src/test/java/org/apache/parquet/cascading/TestParquetTupleScheme.java rename to parquet-cascading-common23/src/test/java/org/apache/parquet/cascading/TestParquetTupleScheme.java diff --git a/parquet-cascading/src/test/resources/names.txt b/parquet-cascading-common23/src/test/resources/names.txt similarity index 100% rename from parquet-cascading/src/test/resources/names.txt rename to parquet-cascading-common23/src/test/resources/names.txt diff --git a/parquet-cascading/src/test/thrift/test.thrift b/parquet-cascading-common23/src/test/thrift/test.thrift similarity index 100% rename from parquet-cascading/src/test/thrift/test.thrift rename to parquet-cascading-common23/src/test/thrift/test.thrift diff --git a/parquet-cascading/pom.xml b/parquet-cascading/pom.xml index 0cd858886e..cabb0038f7 100644 --- a/parquet-cascading/pom.xml +++ b/parquet-cascading/pom.xml @@ -102,6 +102,51 @@ + + org.codehaus.mojo + build-helper-maven-plugin + 1.7 + + + add-source + generate-sources + + add-source + + + + ../parquet-cascading-common23/src/main/java + + + + + add-test-source + generate-test-sources + + add-test-source + + + + ../parquet-cascading-common23/src/test/java + + + + + add-test-resource + generate-test-resources + + add-test-resource + + + + + ../parquet-cascading-common23/src/test/resources + + + + + + maven-enforcer-plugin @@ -115,6 +160,8 @@ 0.1.10 ${thrift.executable} + ../parquet-cascading-common23/src/main/thrift + ../parquet-cascading-common23/src/test/thrift diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java b/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java index ea70d43f8e..b34ee7d24a 100644 --- a/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java +++ b/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java @@ -33,6 +33,7 @@ import org.apache.parquet.hadoop.thrift.TBaseWriteSupport; import org.apache.parquet.thrift.TBaseRecordConverter; +@Deprecated // The parquet-cascading module depends on Cascading 2.x, and is being superseded with parquet-cascading3 for Cascading 3.x public class ParquetTBaseScheme> extends ParquetValueScheme { // In the case of reads, we can read the thrift class from the file metadata diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java b/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java index 41b56d0fcb..3b7d715273 100644 --- a/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java +++ b/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java @@ -59,6 +59,7 @@ * @author Avi Bryant */ +@Deprecated // The parquet-cascading module depends on Cascading 2.x, and is being superseded with parquet-cascading3 for Cascading 3.x public class ParquetTupleScheme extends Scheme{ private static final long serialVersionUID = 0L; diff --git a/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetValueScheme.java b/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetValueScheme.java index 9549ef43f6..6c34a8494b 100644 --- a/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetValueScheme.java +++ b/parquet-cascading/src/main/java/org/apache/parquet/cascading/ParquetValueScheme.java @@ -47,6 +47,7 @@ * This is an abstract class; implementations are expected to set up their Input/Output Formats * correctly in the respective Init methods. */ +@Deprecated // The parquet-cascading module depends on Cascading 2.x, and is being superseded with parquet-cascading3 for Cascading 3.x public abstract class ParquetValueScheme extends Scheme{ public static final class Config implements Serializable { diff --git a/parquet-cascading/src/test/java/org/apache/parquet/cascading/TestParquetTBaseScheme.java b/parquet-cascading/src/test/java/org/apache/parquet/cascading/TestParquetTBaseScheme.java index 841314ca7c..e0f33e1161 100644 --- a/parquet-cascading/src/test/java/org/apache/parquet/cascading/TestParquetTBaseScheme.java +++ b/parquet-cascading/src/test/java/org/apache/parquet/cascading/TestParquetTBaseScheme.java @@ -58,8 +58,9 @@ import java.util.HashMap; import java.util.Map; +@Deprecated // The parquet-cascading module depends on Cascading 2.x, and is being superseded with parquet-cascading3 for Cascading 3.x public class TestParquetTBaseScheme { - final String txtInputPath = "src/test/resources/names.txt"; + final String txtInputPath = "target/test-classes/names.txt"; final String parquetInputPath = "target/test/ParquetTBaseScheme/names-parquet-in"; final String parquetOutputPath = "target/test/ParquetTBaseScheme/names-parquet-out"; final String txtOutputPath = "target/test/ParquetTBaseScheme/names-txt-out"; diff --git a/parquet-cascading3/REVIEWERS.md b/parquet-cascading3/REVIEWERS.md new file mode 100644 index 0000000000..f7972357e9 --- /dev/null +++ b/parquet-cascading3/REVIEWERS.md @@ -0,0 +1,27 @@ + + +The following reviewers had reviewed the parquet-cascading (pre-Cascading 3.0) project: + +| Name | Apache Id | github id | +|--------------------|------------|-------------| +| Dmitriy Ryaboy | dvryaboy | dvryaboy | +| Tianshuo Deng | tianshuo | tsdeng | + + diff --git a/parquet-cascading3/pom.xml b/parquet-cascading3/pom.xml new file mode 100644 index 0000000000..ea552ad004 --- /dev/null +++ b/parquet-cascading3/pom.xml @@ -0,0 +1,178 @@ + + + + org.apache.parquet + parquet + ../pom.xml + 1.8.2-SNAPSHOT + + + 4.0.0 + + parquet-cascading3 + jar + + Apache Parquet Cascading (for Cascading 3.0 onwards) + https://parquet.apache.org + + + + conjars.org + http://conjars.org/repo + + + + + + org.apache.parquet + parquet-column + ${project.version} + + + org.apache.parquet + parquet-hadoop + ${project.version} + + + org.apache.parquet + parquet-thrift + ${project.version} + + + org.apache.hadoop + hadoop-client + ${hadoop.version} + provided + + + org.apache.parquet + parquet-column + ${project.version} + test-jar + test + + + org.mockito + mockito-all + 1.9.5 + test + + + cascading + cascading-hadoop + ${cascading3.version} + provided + + + + + + + + maven-enforcer-plugin + + + none + + + + true + + + + + + + org.codehaus.mojo + build-helper-maven-plugin + 1.7 + + + add-source + generate-sources + + add-source + + + + ../parquet-cascading-common23/src/main/java + + + + + add-test-source + generate-test-sources + + add-test-source + + + + ../parquet-cascading-common23/src/test/java + + + + + add-test-resource + generate-test-resources + + add-test-resource + + + + + ../parquet-cascading-common23/src/test/resources + + + + + + + + maven-enforcer-plugin + + + org.apache.maven.plugins + maven-jar-plugin + + + org.apache.thrift.tools + maven-thrift-plugin + 0.1.10 + + ${thrift.executable} + ../parquet-cascading-common23/src/main/thrift + ../parquet-cascading-common23/src/test/thrift + + + + thrift-sources + generate-test-sources + + testCompile + + + + + + + diff --git a/parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java b/parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java new file mode 100644 index 0000000000..af04b47c8e --- /dev/null +++ b/parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTBaseScheme.java @@ -0,0 +1,80 @@ +/* + * 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.cascading; + +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.OutputCollector; +import org.apache.hadoop.mapred.RecordReader; +import org.apache.thrift.TBase; + +import cascading.flow.FlowProcess; +import cascading.tap.Tap; +import org.apache.parquet.filter2.predicate.FilterPredicate; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.mapred.DeprecatedParquetInputFormat; +import org.apache.parquet.hadoop.mapred.DeprecatedParquetOutputFormat; +import org.apache.parquet.hadoop.thrift.ThriftReadSupport; +import org.apache.parquet.hadoop.thrift.TBaseWriteSupport; +import org.apache.parquet.thrift.TBaseRecordConverter; + +public class ParquetTBaseScheme> extends ParquetValueScheme { + + // In the case of reads, we can read the thrift class from the file metadata + public ParquetTBaseScheme() { + this(new Config()); + } + + public ParquetTBaseScheme(Class thriftClass) { + this(new Config().withRecordClass(thriftClass)); + } + + public ParquetTBaseScheme(FilterPredicate filterPredicate) { + this(new Config().withFilterPredicate(filterPredicate)); + } + + public ParquetTBaseScheme(FilterPredicate filterPredicate, Class thriftClass) { + this(new Config().withRecordClass(thriftClass).withFilterPredicate(filterPredicate)); + } + + public ParquetTBaseScheme(Config config) { + super(config); + } + + @Override + public void sourceConfInit(FlowProcess fp, + Tap tap, JobConf jobConf) { + super.sourceConfInit(fp, tap, jobConf); + jobConf.setInputFormat(DeprecatedParquetInputFormat.class); + ParquetInputFormat.setReadSupportClass(jobConf, ThriftReadSupport.class); + ThriftReadSupport.setRecordConverterClass(jobConf, TBaseRecordConverter.class); + } + + @Override + public void sinkConfInit(FlowProcess fp, + Tap tap, JobConf jobConf) { + + if (this.config.getKlass() == null) { + throw new IllegalArgumentException("To use ParquetTBaseScheme as a sink, you must specify a thrift class in the constructor"); + } + + DeprecatedParquetOutputFormat.setAsOutputFormat(jobConf); + DeprecatedParquetOutputFormat.setWriteSupportClass(jobConf, TBaseWriteSupport.class); + TBaseWriteSupport.setThriftClass(jobConf, this.config.getKlass()); + } +} diff --git a/parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java b/parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java new file mode 100644 index 0000000000..4532d3b3f8 --- /dev/null +++ b/parquet-cascading3/src/main/java/org/apache/parquet/cascading/ParquetTupleScheme.java @@ -0,0 +1,191 @@ +/* + * 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.cascading; + +import java.io.IOException; +import java.util.List; + +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.OutputCollector; +import org.apache.hadoop.mapred.RecordReader; + +import cascading.flow.FlowProcess; +import cascading.scheme.Scheme; +import cascading.scheme.SinkCall; +import cascading.scheme.SourceCall; +import cascading.tap.CompositeTap; +import cascading.tap.Tap; +import cascading.tap.TapException; +import cascading.tap.hadoop.Hfs; +import cascading.tuple.Fields; +import cascading.tuple.Tuple; +import cascading.tuple.TupleEntry; +import org.apache.parquet.filter2.predicate.FilterPredicate; +import org.apache.parquet.hadoop.Footer; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.ParquetOutputFormat; +import org.apache.parquet.hadoop.mapred.Container; +import org.apache.parquet.hadoop.mapred.DeprecatedParquetInputFormat; +import org.apache.parquet.hadoop.mapred.DeprecatedParquetOutputFormat; +import org.apache.parquet.schema.MessageType; + +import static org.apache.parquet.Preconditions.checkNotNull; + +/** + * A Cascading Scheme that converts Parquet groups into Cascading tuples. + * If you provide it with sourceFields, it will selectively materialize only the columns for those fields. + * The names must match the names in the Parquet schema. + * If you do not provide sourceFields, or use Fields.ALL or Fields.UNKNOWN, it will create one from the + * Parquet schema. + * Currently, only primitive types are supported. TODO: allow nested fields in the Parquet schema to be + * flattened to a top-level field in the Cascading tuple. + * + * @author Avi Bryant + */ + +public class ParquetTupleScheme extends Scheme{ + + private static final long serialVersionUID = 0L; + private String parquetSchema; + private final FilterPredicate filterPredicate; + + public ParquetTupleScheme() { + super(); + this.filterPredicate = null; + } + + public ParquetTupleScheme(Fields sourceFields) { + super(sourceFields); + this.filterPredicate = null; + } + + public ParquetTupleScheme(FilterPredicate filterPredicate) { + this.filterPredicate = checkNotNull(filterPredicate, "filterPredicate"); + } + + public ParquetTupleScheme(FilterPredicate filterPredicate, Fields sourceFields) { + super(sourceFields); + this.filterPredicate = checkNotNull(filterPredicate, "filterPredicate"); + } + + /** + * ParquetTupleScheme constructor used a sink need to be implemented + * + * @param sourceFields used for the reading step + * @param sinkFields used for the writing step + * @param schema is mandatory if you add sinkFields and needs to be the + * toString() from a MessageType. This value is going to be parsed when the + * parquet file will be created. + */ + public ParquetTupleScheme(Fields sourceFields, Fields sinkFields, final String schema) { + super(sourceFields, sinkFields); + parquetSchema = schema; + this.filterPredicate = null; + } + + @SuppressWarnings("rawtypes") + @Override + public void sourceConfInit(FlowProcess fp, + Tap tap, JobConf jobConf) { + + if (filterPredicate != null) { + ParquetInputFormat.setFilterPredicate(jobConf, filterPredicate); + } + + jobConf.setInputFormat(DeprecatedParquetInputFormat.class); + ParquetInputFormat.setReadSupportClass(jobConf, TupleReadSupport.class); + TupleReadSupport.setRequestedFields(jobConf, getSourceFields()); + } + + @Override + public Fields retrieveSourceFields(FlowProcess flowProcess, Tap tap) { + MessageType schema = readSchema(flowProcess, tap); + SchemaIntersection intersection = new SchemaIntersection(schema, getSourceFields()); + + setSourceFields(intersection.getSourceFields()); + + return getSourceFields(); + } + + private MessageType readSchema(FlowProcess flowProcess, Tap tap) { + try { + Hfs hfs; + + if( tap instanceof CompositeTap ) + hfs = (Hfs) ( (CompositeTap) tap ).getChildTaps().next(); + else + hfs = (Hfs) tap; + + List