From 1da507ecc6b0471098ebb4a24037734a727ec985 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Fri, 14 Oct 2016 10:02:18 -0700 Subject: [PATCH 1/2] PARQUET-751: Add setRequestedSchema to ParquetFileReader. This fixes a bug introduced by dictionary filters, which reused an existing file reader to avoid opening multiple input streams. Before that commit, a new file reader was opened and passed the projection columns from the read context. The fix is to set the requested schema on the file reader instead of creating a new instance. This also adds a test to ensure that column projection works to catch bugs like this in the future. --- .../hadoop/InternalParquetRecordReader.java | 1 + .../parquet/hadoop/ParquetFileReader.java | 8 + .../TestInputFormatColumnProjection.java | 168 ++++++++++++++++++ 3 files changed, 177 insertions(+) create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordReader.java index d43fd7d840..85b669175a 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordReader.java @@ -179,6 +179,7 @@ public void initialize(ParquetFileReader reader, Configuration configuration) this.unmaterializableRecordCounter = new UnmaterializableRecordCounter(configuration, total); this.filterRecords = configuration.getBoolean( RECORD_FILTERING_ENABLED, RECORD_FILTERING_ENABLED_DEFAULT); + reader.setRequestedSchema(requestedSchema); LOG.info("RecordReader initialized will read a total of " + total + " records."); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index 9e95535b7c..4af26d0a0e 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -95,6 +95,7 @@ import org.apache.parquet.hadoop.util.counters.BenchmarkCounter; import org.apache.parquet.io.ParquetDecodingException; import org.apache.parquet.io.InputFile; +import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; /** @@ -674,6 +675,13 @@ public List getRowGroups() { return blocks; } + public void setRequestedSchema(MessageType projection) { + paths.clear(); + for (ColumnDescriptor col : projection.getColumns()) { + paths.put(ColumnPath.get(col.getPath()), col); + } + } + public void appendTo(ParquetFileWriter writer) throws IOException { writer.appendRowGroups(f, blocks, true); } diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java new file mode 100644 index 0000000000..a3da664993 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java @@ -0,0 +1,168 @@ +/* + * 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; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapreduce.Counters; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.Mapper; +import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; +import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.example.ExampleInputFormat; +import org.apache.parquet.hadoop.example.ExampleOutputFormat; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.Types; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.UUID; + +import static java.lang.Thread.sleep; +import static org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.Counter.BYTES_WRITTEN; +import static org.apache.parquet.schema.OriginalType.UTF8; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; + +public class TestInputFormatColumnProjection { + public static final String FILE_CONTENT = "" + + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ," + + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ," + + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"; + + public static MessageType PARQUET_TYPE = Types.buildMessage() + .required(BINARY).as(UTF8).named("uuid") + .required(BINARY).as(UTF8).named("char") + .named("FormatTestObject"); + + public static class Writer extends Mapper { + public static final SimpleGroupFactory GROUP_FACTORY = new SimpleGroupFactory(PARQUET_TYPE); + @Override + protected void map(LongWritable key, Text value, Context context) + throws IOException, InterruptedException { + // writes each character of the line with a UUID + String line = value.toString(); + for (int i = 0; i < line.length(); i += 1) { + Group group = GROUP_FACTORY.newGroup(); + group.add(0, Binary.fromString(UUID.randomUUID().toString())); + group.add(1, Binary.fromString(line.substring(i, i+1))); + context.write(null, group); + } + } + } + + public static class Reader extends Mapper { + @Override + protected void map(Void key, Group value, Context context) + throws IOException, InterruptedException { + // Do nothing. The test uses Hadoop FS counters for verification. + } + } + + @Rule + public TemporaryFolder temp = new TemporaryFolder(); + + @Test + public void testProjectionSize() throws Exception { + File inputFile = temp.newFile(); + FileOutputStream out = new FileOutputStream(inputFile); + out.write(FILE_CONTENT.getBytes("UTF-8")); + out.close(); + + File tempFolder = temp.newFolder(); + tempFolder.delete(); + Path tempPath = new Path(tempFolder.toURI()); + + File outputFolder = temp.newFile(); + outputFolder.delete(); + + Configuration conf = new Configuration(); + // set the projection schema + conf.set("parquet.read.schema", Types.buildMessage() + .required(BINARY).as(UTF8).named("char") + .named("FormatTestObject").toString()); + + // disable summary metadata, it isn't needed + conf.set("parquet.enable.summary-metadata", "false"); + conf.set("parquet.example.schema", PARQUET_TYPE.toString()); + + long bytesWritten; + long bytesRead; + + { + Job writeJob = new Job(conf, "write"); + writeJob.setInputFormatClass(TextInputFormat.class); + TextInputFormat.addInputPath(writeJob, new Path(inputFile.toString())); + + writeJob.setOutputFormatClass(ExampleOutputFormat.class); + writeJob.setMapperClass(Writer.class); + writeJob.setNumReduceTasks(0); // write directly to Parquet without reduce + ParquetOutputFormat.setBlockSize(writeJob, 10240); + ParquetOutputFormat.setPageSize(writeJob, 512); + ParquetOutputFormat.setDictionaryPageSize(writeJob, 1024); + ParquetOutputFormat.setEnableDictionary(writeJob, true); + ParquetOutputFormat.setMaxPaddingSize(writeJob, 1023); // always pad + ParquetOutputFormat.setOutputPath(writeJob, tempPath); + + waitForJob(writeJob); + + Counters counters = writeJob.getCounters(); + bytesWritten = counters.findCounter(BYTES_WRITTEN).getValue(); + } + + { + Job readJob = new Job(conf, "read"); + readJob.setInputFormatClass(ExampleInputFormat.class); + TextInputFormat.addInputPath(readJob, tempPath); + + readJob.setOutputFormatClass(TextOutputFormat.class); + readJob.setMapperClass(Reader.class); + readJob.setNumReduceTasks(0); // no reduce phase + TextOutputFormat.setOutputPath(readJob, new Path(outputFolder.toString())); + + waitForJob(readJob); + + Counters counters = readJob.getCounters(); + bytesRead = counters.getGroup("parquet") + .findCounter("bytesread") + .getValue(); + } + + Assert.assertTrue("Should read less than 10% of the input file size", + bytesRead < (bytesWritten / 10)); + } + + private void waitForJob(Job job) throws Exception { + job.submit(); + while (!job.isComplete()) { + sleep(100); + } + if (!job.isSuccessful()) { + throw new RuntimeException("job failed " + job.getJobName()); + } + } +} From 7ea0c163cb33f1b11d77672c9e0603cc7424e88e Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Tue, 18 Oct 2016 14:39:13 -0700 Subject: [PATCH 2/2] PARQUET-751: Fix column projection test. This updates the test to only use the Parquet counters, and to not run for Hadoop 1 because it would require more reflection code that isn't worth adding for Hadoop 1. --- .../TestInputFormatColumnProjection.java | 34 +++++++++++++------ 1 file changed, 23 insertions(+), 11 deletions(-) diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java index a3da664993..a6d2732d1b 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatColumnProjection.java @@ -19,10 +19,12 @@ package org.apache.parquet.hadoop; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; -import org.apache.hadoop.mapreduce.Counters; +import org.apache.hadoop.mapreduce.Counter; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; @@ -31,10 +33,12 @@ import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.hadoop.example.ExampleInputFormat; import org.apache.parquet.hadoop.example.ExampleOutputFormat; +import org.apache.parquet.hadoop.util.ContextUtil; import org.apache.parquet.io.api.Binary; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.Types; import org.junit.Assert; +import org.junit.Assume; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; @@ -44,7 +48,6 @@ import java.util.UUID; import static java.lang.Thread.sleep; -import static org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.Counter.BYTES_WRITTEN; import static org.apache.parquet.schema.OriginalType.UTF8; import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; @@ -76,10 +79,18 @@ protected void map(LongWritable key, Text value, Context context) } public static class Reader extends Mapper { + + public static Counter bytesReadCounter = null; + public static void setBytesReadCounter(Counter bytesRead) { + bytesReadCounter = bytesRead; + } + @Override protected void map(Void key, Group value, Context context) throws IOException, InterruptedException { // Do nothing. The test uses Hadoop FS counters for verification. + setBytesReadCounter(ContextUtil.getCounter( + context, "parquet", "bytesread")); } } @@ -88,6 +99,9 @@ protected void map(Void key, Group value, Context context) @Test public void testProjectionSize() throws Exception { + Assume.assumeTrue( // only run this test for Hadoop 2 + org.apache.hadoop.mapreduce.JobContext.class.isInterface()); + File inputFile = temp.newFile(); FileOutputStream out = new FileOutputStream(inputFile); out.write(FILE_CONTENT.getBytes("UTF-8")); @@ -110,9 +124,6 @@ public void testProjectionSize() throws Exception { conf.set("parquet.enable.summary-metadata", "false"); conf.set("parquet.example.schema", PARQUET_TYPE.toString()); - long bytesWritten; - long bytesRead; - { Job writeJob = new Job(conf, "write"); writeJob.setInputFormatClass(TextInputFormat.class); @@ -129,11 +140,15 @@ public void testProjectionSize() throws Exception { ParquetOutputFormat.setOutputPath(writeJob, tempPath); waitForJob(writeJob); + } - Counters counters = writeJob.getCounters(); - bytesWritten = counters.findCounter(BYTES_WRITTEN).getValue(); + long bytesWritten = 0; + FileSystem fs = FileSystem.getLocal(conf); + for (FileStatus file : fs.listStatus(tempPath)) { + bytesWritten += file.getLen(); } + long bytesRead; { Job readJob = new Job(conf, "read"); readJob.setInputFormatClass(ExampleInputFormat.class); @@ -146,10 +161,7 @@ public void testProjectionSize() throws Exception { waitForJob(readJob); - Counters counters = readJob.getCounters(); - bytesRead = counters.getGroup("parquet") - .findCounter("bytesread") - .getValue(); + bytesRead = Reader.bytesReadCounter.getValue(); } Assert.assertTrue("Should read less than 10% of the input file size",