diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java b/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java index 895d0670fa..d971b909b9 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java @@ -19,17 +19,17 @@ package org.apache.parquet; +import static org.apache.parquet.conf.ParquetInputFilters.getFilter; +import static org.apache.parquet.conf.ParquetInputProperties.BLOOM_FILTERING_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.COLUMN_INDEX_FILTERING_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.DICTIONARY_FILTERING_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.HADOOP_VECTORED_IO_DEFAULT; +import static org.apache.parquet.conf.ParquetInputProperties.HADOOP_VECTORED_IO_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.OFF_HEAP_DECRYPT_BUFFER_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.RECORD_FILTERING_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.STATS_FILTERING_ENABLED; import static org.apache.parquet.format.converter.ParquetMetadataConverter.NO_FILTER; -import static org.apache.parquet.hadoop.ParquetInputFormat.BLOOM_FILTERING_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.COLUMN_INDEX_FILTERING_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.DICTIONARY_FILTERING_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.HADOOP_VECTORED_IO_DEFAULT; -import static org.apache.parquet.hadoop.ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.OFF_HEAP_DECRYPT_BUFFER_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.RECORD_FILTERING_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.STATS_FILTERING_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter; import static org.apache.parquet.hadoop.UnmaterializableRecordCounter.BAD_RECORD_THRESHOLD_CONF_KEY; import java.util.Collections; diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/conf/ParquetInputFilters.java b/parquet-hadoop/src/main/java/org/apache/parquet/conf/ParquetInputFilters.java new file mode 100644 index 0000000000..4de2a48e26 --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/conf/ParquetInputFilters.java @@ -0,0 +1,84 @@ +/* + * 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.conf; + +import java.io.IOException; +import org.apache.parquet.filter.UnboundRecordFilter; +import org.apache.parquet.filter2.compat.FilterCompat; +import org.apache.parquet.filter2.compat.FilterCompat.Filter; +import org.apache.parquet.filter2.predicate.FilterPredicate; +import org.apache.parquet.hadoop.BadConfigurationException; +import org.apache.parquet.hadoop.util.ConfigurationUtil; +import org.apache.parquet.hadoop.util.SerializationUtil; + +/** + * Resolves the filter used to read Parquet files from a {@link ParquetConfiguration}. + * This class is decoupled from the legacy Hadoop {@code Configuration} and + * {@code InputFormat} machinery, so that the filter resolution logic can be used + * without loading {@code FileInputFormat} and its transitive dependencies. + */ +public final class ParquetInputFilters { + + private ParquetInputFilters() {} + + /** + * @param configuration a configuration + * @return an unbound record filter class + */ + public static Class getUnboundRecordFilter(ParquetConfiguration configuration) { + return ConfigurationUtil.getClassFromConfig( + configuration, ParquetInputProperties.UNBOUND_RECORD_FILTER, UnboundRecordFilter.class); + } + + /** + * @param configuration a configuration + * @return the filter predicate stored in the configuration, or null if none is set + */ + public static FilterPredicate getFilterPredicate(ParquetConfiguration configuration) { + try { + return SerializationUtil.readObjectFromConfAsBase64(ParquetInputProperties.FILTER_PREDICATE, configuration); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private static UnboundRecordFilter getUnboundRecordFilterInstance(ParquetConfiguration configuration) { + Class clazz = getUnboundRecordFilter(configuration); + if (clazz == null) { + return null; + } + try { + return (UnboundRecordFilter) clazz.newInstance(); + } catch (InstantiationException | IllegalAccessException e) { + throw new BadConfigurationException("could not instantiate unbound record filter class", e); + } + } + + /** + * Returns a non-null Filter, which is a wrapper around either a + * FilterPredicate, an UnboundRecordFilter, or a no-op filter. + * + * @param conf a configuration + * @return a filter for the unbound record filter specified in conf + */ + public static Filter getFilter(ParquetConfiguration conf) { + return FilterCompat.get(getFilterPredicate(conf), getUnboundRecordFilterInstance(conf)); + } +} diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/conf/ParquetInputProperties.java b/parquet-hadoop/src/main/java/org/apache/parquet/conf/ParquetInputProperties.java new file mode 100644 index 0000000000..7b560bdc24 --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/conf/ParquetInputProperties.java @@ -0,0 +1,71 @@ +/* + * 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.conf; + +/** + * Stores the read configuration property keys used to configure how Parquet + * files are read. This class contains only constants and is decoupled from the + * legacy Hadoop {@code Configuration} and {@code InputFormat} machinery, so that + * the configuration properties can be used without loading {@code FileInputFormat} + * and its transitive dependencies. + */ +public final class ParquetInputProperties { + + /** key to configure the ReadSupport implementation */ + public static final String READ_SUPPORT_CLASS = "parquet.read.support.class"; + + /** key to configure the filter */ + public static final String UNBOUND_RECORD_FILTER = "parquet.read.filter"; + + /** key to configure type checking for conflicting schemas (default: true) */ + public static final String STRICT_TYPE_CHECKING = "parquet.strict.typing"; + + /** key to configure the filter predicate */ + public static final String FILTER_PREDICATE = "parquet.private.read.filter.predicate"; + + /** key to configure whether record-level filtering is enabled */ + public static final String RECORD_FILTERING_ENABLED = "parquet.filter.record-level.enabled"; + + /** key to configure whether row group stats filtering is enabled */ + public static final String STATS_FILTERING_ENABLED = "parquet.filter.stats.enabled"; + + /** key to configure whether row group dictionary filtering is enabled */ + public static final String DICTIONARY_FILTERING_ENABLED = "parquet.filter.dictionary.enabled"; + + /** key to configure whether column index filtering of pages is enabled */ + public static final String COLUMN_INDEX_FILTERING_ENABLED = "parquet.filter.columnindex.enabled"; + + /** key to configure whether page level checksum verification is enabled */ + public static final String PAGE_VERIFY_CHECKSUM_ENABLED = "parquet.page.verify-checksum.enabled"; + + /** key to configure whether row group bloom filtering is enabled */ + public static final String BLOOM_FILTERING_ENABLED = "parquet.filter.bloom.enabled"; + + /** Key to configure if off-heap buffer should be used for decryption */ + public static final String OFF_HEAP_DECRYPT_BUFFER_ENABLED = "parquet.decrypt.off-heap.buffer.enabled"; + + /** Key to enable/disable vectored io while reading parquet files: {@value}. */ + public static final String HADOOP_VECTORED_IO_ENABLED = "parquet.hadoop.vectored.io.enabled"; + + /** Default value of parquet.hadoop.vectored.io.enabled is {@value}. */ + public static final boolean HADOOP_VECTORED_IO_DEFAULT = true; + + private ParquetInputProperties() {} +} 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 19b1d5426e..642e51c3f9 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 @@ -19,8 +19,8 @@ package org.apache.parquet.hadoop; import static java.lang.String.format; -import static org.apache.parquet.hadoop.ParquetInputFormat.RECORD_FILTERING_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.STRICT_TYPE_CHECKING; +import static org.apache.parquet.conf.ParquetInputProperties.RECORD_FILTERING_ENABLED; +import static org.apache.parquet.conf.ParquetInputProperties.STRICT_TYPE_CHECKING; import java.io.IOException; import java.util.Collections; diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java index 8e05d49bd3..694502878e 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java @@ -51,8 +51,9 @@ import org.apache.parquet.Preconditions; import org.apache.parquet.conf.HadoopParquetConfiguration; import org.apache.parquet.conf.ParquetConfiguration; +import org.apache.parquet.conf.ParquetInputFilters; +import org.apache.parquet.conf.ParquetInputProperties; import org.apache.parquet.filter.UnboundRecordFilter; -import org.apache.parquet.filter2.compat.FilterCompat; import org.apache.parquet.filter2.compat.FilterCompat.Filter; import org.apache.parquet.filter2.compat.RowGroupFilter; import org.apache.parquet.filter2.predicate.FilterPredicate; @@ -82,10 +83,10 @@ * It must be a subset of the original schema. Only the columns needed to reconstruct the records with the requestedSchema will be scanned. * * @param the type of the materialized records - * @see #READ_SUPPORT_CLASS - * @see #UNBOUND_RECORD_FILTER - * @see #STRICT_TYPE_CHECKING - * @see #FILTER_PREDICATE + * @see ParquetInputProperties#READ_SUPPORT_CLASS + * @see ParquetInputProperties#UNBOUND_RECORD_FILTER + * @see ParquetInputProperties#STRICT_TYPE_CHECKING + * @see ParquetInputProperties#FILTER_PREDICATE * @see #TASK_SIDE_METADATA */ public class ParquetInputFormat extends FileInputFormat { @@ -94,58 +95,80 @@ public class ParquetInputFormat extends FileInputFormat { /** * key to configure the ReadSupport implementation + * @deprecated use {@link ParquetInputProperties#READ_SUPPORT_CLASS} */ - public static final String READ_SUPPORT_CLASS = "parquet.read.support.class"; + @Deprecated + public static final String READ_SUPPORT_CLASS = ParquetInputProperties.READ_SUPPORT_CLASS; /** * key to configure the filter + * @deprecated use {@link ParquetInputProperties#UNBOUND_RECORD_FILTER} */ - public static final String UNBOUND_RECORD_FILTER = "parquet.read.filter"; + @Deprecated + public static final String UNBOUND_RECORD_FILTER = ParquetInputProperties.UNBOUND_RECORD_FILTER; /** * key to configure type checking for conflicting schemas (default: true) + * @deprecated use {@link ParquetInputProperties#STRICT_TYPE_CHECKING} */ - public static final String STRICT_TYPE_CHECKING = "parquet.strict.typing"; + @Deprecated + public static final String STRICT_TYPE_CHECKING = ParquetInputProperties.STRICT_TYPE_CHECKING; /** * key to configure the filter predicate + * @deprecated use {@link ParquetInputProperties#FILTER_PREDICATE} */ - public static final String FILTER_PREDICATE = "parquet.private.read.filter.predicate"; + @Deprecated + public static final String FILTER_PREDICATE = ParquetInputProperties.FILTER_PREDICATE; /** * key to configure whether record-level filtering is enabled + * @deprecated use {@link ParquetInputProperties#RECORD_FILTERING_ENABLED} */ - public static final String RECORD_FILTERING_ENABLED = "parquet.filter.record-level.enabled"; + @Deprecated + public static final String RECORD_FILTERING_ENABLED = ParquetInputProperties.RECORD_FILTERING_ENABLED; /** * key to configure whether row group stats filtering is enabled + * @deprecated use {@link ParquetInputProperties#STATS_FILTERING_ENABLED} */ - public static final String STATS_FILTERING_ENABLED = "parquet.filter.stats.enabled"; + @Deprecated + public static final String STATS_FILTERING_ENABLED = ParquetInputProperties.STATS_FILTERING_ENABLED; /** * key to configure whether row group dictionary filtering is enabled + * @deprecated use {@link ParquetInputProperties#DICTIONARY_FILTERING_ENABLED} */ - public static final String DICTIONARY_FILTERING_ENABLED = "parquet.filter.dictionary.enabled"; + @Deprecated + public static final String DICTIONARY_FILTERING_ENABLED = ParquetInputProperties.DICTIONARY_FILTERING_ENABLED; /** * key to configure whether column index filtering of pages is enabled + * @deprecated use {@link ParquetInputProperties#COLUMN_INDEX_FILTERING_ENABLED} */ - public static final String COLUMN_INDEX_FILTERING_ENABLED = "parquet.filter.columnindex.enabled"; + @Deprecated + public static final String COLUMN_INDEX_FILTERING_ENABLED = ParquetInputProperties.COLUMN_INDEX_FILTERING_ENABLED; /** * key to configure whether page level checksum verification is enabled + * @deprecated use {@link ParquetInputProperties#PAGE_VERIFY_CHECKSUM_ENABLED} */ - public static final String PAGE_VERIFY_CHECKSUM_ENABLED = "parquet.page.verify-checksum.enabled"; + @Deprecated + public static final String PAGE_VERIFY_CHECKSUM_ENABLED = ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED; /** * key to configure whether row group bloom filtering is enabled + * @deprecated use {@link ParquetInputProperties#BLOOM_FILTERING_ENABLED} */ - public static final String BLOOM_FILTERING_ENABLED = "parquet.filter.bloom.enabled"; + @Deprecated + public static final String BLOOM_FILTERING_ENABLED = ParquetInputProperties.BLOOM_FILTERING_ENABLED; /** * Key to configure if off-heap buffer should be used for decryption + * @deprecated use {@link ParquetInputProperties#OFF_HEAP_DECRYPT_BUFFER_ENABLED} */ - public static final String OFF_HEAP_DECRYPT_BUFFER_ENABLED = "parquet.decrypt.off-heap.buffer.enabled"; + @Deprecated + public static final String OFF_HEAP_DECRYPT_BUFFER_ENABLED = ParquetInputProperties.OFF_HEAP_DECRYPT_BUFFER_ENABLED; /** * key to turn on or off task side metadata loading (default true) @@ -164,13 +187,17 @@ public class ParquetInputFormat extends FileInputFormat { /** * Key to enable/disable vectored io while reading parquet files: * {@value}. + * @deprecated use {@link ParquetInputProperties#HADOOP_VECTORED_IO_ENABLED} */ - public static final String HADOOP_VECTORED_IO_ENABLED = "parquet.hadoop.vectored.io.enabled"; + @Deprecated + public static final String HADOOP_VECTORED_IO_ENABLED = ParquetInputProperties.HADOOP_VECTORED_IO_ENABLED; /** * Default value of parquet.hadoop.vectored.io.enabled is {@value}. + * @deprecated use {@link ParquetInputProperties#HADOOP_VECTORED_IO_DEFAULT} */ - public static final boolean HADOOP_VECTORED_IO_DEFAULT = true; + @Deprecated + public static final boolean HADOOP_VECTORED_IO_DEFAULT = ParquetInputProperties.HADOOP_VECTORED_IO_DEFAULT; public static void setTaskSideMetaData(Job job, boolean taskSideMetadata) { ContextUtil.getConfiguration(job).setBoolean(TASK_SIDE_METADATA, taskSideMetadata); @@ -200,20 +227,7 @@ public static void setUnboundRecordFilter(Job job, Class getUnboundRecordFilter(Configuration configuration) { - return ConfigurationUtil.getClassFromConfig(configuration, UNBOUND_RECORD_FILTER, UnboundRecordFilter.class); - } - - private static UnboundRecordFilter getUnboundRecordFilterInstance(ParquetConfiguration configuration) { - Class clazz = - ConfigurationUtil.getClassFromConfig(configuration, UNBOUND_RECORD_FILTER, UnboundRecordFilter.class); - if (clazz == null) { - return null; - } - try { - return (UnboundRecordFilter) clazz.newInstance(); - } catch (InstantiationException | IllegalAccessException e) { - throw new BadConfigurationException("could not instantiate unbound record filter class", e); - } + return ParquetInputFilters.getUnboundRecordFilter(new HadoopParquetConfiguration(configuration)); } public static void setReadSupportClass(JobConf conf, Class readSupportClass) { @@ -238,15 +252,7 @@ public static void setFilterPredicate(Configuration configuration, FilterPredica } private static FilterPredicate getFilterPredicate(Configuration configuration) { - return getFilterPredicate(new HadoopParquetConfiguration(configuration)); - } - - private static FilterPredicate getFilterPredicate(ParquetConfiguration configuration) { - try { - return SerializationUtil.readObjectFromConfAsBase64(FILTER_PREDICATE, configuration); - } catch (IOException e) { - throw new RuntimeException(e); - } + return ParquetInputFilters.getFilterPredicate(new HadoopParquetConfiguration(configuration)); } /** @@ -260,8 +266,17 @@ public static Filter getFilter(Configuration conf) { return getFilter(new HadoopParquetConfiguration(conf)); } + /** + * Returns a non-null Filter, which is a wrapper around either a + * FilterPredicate, an UnboundRecordFilter, or a no-op filter. + * + * @param conf a configuration + * @return a filter for the unbound record filter specified in conf + * @deprecated use {@link ParquetInputFilters#getFilter(ParquetConfiguration)} + */ + @Deprecated public static Filter getFilter(ParquetConfiguration conf) { - return FilterCompat.get(getFilterPredicate(conf), getUnboundRecordFilterInstance(conf)); + return ParquetInputFilters.getFilter(conf); } private LruCache footersCache; diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestPropertiesDrivenEncryption.java b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestPropertiesDrivenEncryption.java index acccb566d4..02fed1dade 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestPropertiesDrivenEncryption.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestPropertiesDrivenEncryption.java @@ -18,11 +18,11 @@ */ package org.apache.parquet.crypto; +import static org.apache.parquet.conf.ParquetInputProperties.OFF_HEAP_DECRYPT_BUFFER_ENABLED; import static org.apache.parquet.hadoop.ParquetFileWriter.EFMAGIC; import static org.apache.parquet.hadoop.ParquetFileWriter.EF_MAGIC_STR; import static org.apache.parquet.hadoop.ParquetFileWriter.MAGIC; import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.OVERWRITE; -import static org.apache.parquet.hadoop.ParquetInputFormat.OFF_HEAP_DECRYPT_BUFFER_ENABLED; import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; import java.io.File; diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java index 0a1878d436..cf447654a4 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/DeprecatedInputFormatTest.java @@ -52,6 +52,7 @@ import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; +import org.apache.parquet.conf.ParquetInputProperties; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.hadoop.api.ReadSupport; @@ -116,7 +117,7 @@ private void runMapReduceJob(CompressionCodecName codec) } { jobConf.set(ReadSupport.PARQUET_READ_SCHEMA, readSchema); - jobConf.set(ParquetInputFormat.READ_SUPPORT_CLASS, GroupReadSupport.class.getCanonicalName()); + jobConf.set(ParquetInputProperties.READ_SUPPORT_CLASS, GroupReadSupport.class.getCanonicalName()); jobConf.setInputFormat(MyDeprecatedInputFormat.class); MyDeprecatedInputFormat.setInputPaths(jobConf, parquetPath); jobConf.setOutputFormat(org.apache.hadoop.mapred.TextOutputFormat.class); @@ -244,7 +245,7 @@ public void testCombineParquetInputFormat() throws Exception { conf.setMapperClass(DeprecatedWriteMapper.class); org.apache.hadoop.mapred.FileInputFormat.setInputPaths(conf, new Path(inputDir.toURI())); org.apache.hadoop.mapred.TextOutputFormat.setOutputPath(conf, new Path(outputDir.toURI())); - conf.set(ParquetInputFormat.READ_SUPPORT_CLASS, GroupReadSupport.class.getCanonicalName()); + conf.set(ParquetInputProperties.READ_SUPPORT_CLASS, GroupReadSupport.class.getCanonicalName()); JobClient.runJob(conf); File partFile = outputDir.listFiles(new PartFileFilter())[0]; try (BufferedReader br = new BufferedReader(new FileReader(partFile))) { diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestColumnChunkPageWriteStore.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestColumnChunkPageWriteStore.java index 5705197b5c..efc6e05444 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestColumnChunkPageWriteStore.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestColumnChunkPageWriteStore.java @@ -62,6 +62,7 @@ import org.apache.parquet.column.statistics.BinaryStatistics; import org.apache.parquet.column.statistics.Statistics; import org.apache.parquet.compression.CompressionCodecFactory.BytesInputCompressor; +import org.apache.parquet.conf.ParquetInputProperties; import org.apache.parquet.hadoop.ParquetFileWriter.Mode; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.CompressionCodecName; @@ -149,7 +150,7 @@ public void testWithDirectBuffers() throws Exception { Configuration config = new Configuration(conf); // we want to test the path with direct buffers so we need to enable this config as well // even though this file is not encrypted - config.set(ParquetInputFormat.OFF_HEAP_DECRYPT_BUFFER_ENABLED, "true"); + config.set(ParquetInputProperties.OFF_HEAP_DECRYPT_BUFFER_ENABLED, "true"); test(config, allocator = TrackingByteBufferAllocator.wrap(new DirectByteBufferAllocator())); } diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDataPageChecksums.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDataPageChecksums.java index da3f9248b2..1665298027 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDataPageChecksums.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDataPageChecksums.java @@ -49,6 +49,7 @@ import org.apache.parquet.column.page.PageWriter; import org.apache.parquet.column.statistics.Statistics; import org.apache.parquet.compression.CompressionCodecFactory.BytesInputCompressor; +import org.apache.parquet.conf.ParquetInputProperties; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.GroupFactory; import org.apache.parquet.example.data.simple.SimpleGroupFactory; @@ -309,7 +310,7 @@ private Path writeNestedWithNullsSampleParquetFile( private void testWriteOnVerifyOff(ParquetProperties.WriterVersion version) throws IOException { Configuration conf = new Configuration(); conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, true); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, false); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, false); Path path = writeSimpleParquetFile(conf, CompressionCodecName.UNCOMPRESSED, version); @@ -350,7 +351,7 @@ public void testWriteOnVerifyOffV2() throws IOException { private void testWriteOffVerifyOff(ParquetProperties.WriterVersion version) throws IOException { Configuration conf = new Configuration(); conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, false); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, false); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, false); Path path = writeSimpleParquetFile(conf, CompressionCodecName.UNCOMPRESSED, version); @@ -381,7 +382,7 @@ public void testWriteOffVerifyOffV2() throws IOException { private void testWriteOffVerifyOn(ParquetProperties.WriterVersion version) throws IOException { Configuration conf = new Configuration(); conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, false); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, true); Path path = writeSimpleParquetFile(conf, CompressionCodecName.UNCOMPRESSED, version); @@ -412,7 +413,7 @@ public void testWriteOffVerifyOnV2() throws IOException { private void testWriteOnVerifyOn(ParquetProperties.WriterVersion version) throws IOException { Configuration conf = new Configuration(); conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, true); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, true); Path path = writeSimpleParquetFile(conf, CompressionCodecName.UNCOMPRESSED, version); @@ -476,7 +477,7 @@ private void testCorruptedPage(ParquetProperties.WriterVersion version) throws I // First we disable checksum verification, the corruption will go undetected as it is in the // data section of the page - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, false); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, false); try (ParquetFileReader reader = getParquetFileReader(path, conf, List.of(colADesc, colBDesc))) { PageReadStore pageReadStore = reader.readNextRowGroup(); @@ -493,7 +494,7 @@ private void testCorruptedPage(ParquetProperties.WriterVersion version) throws I } // Now we enable checksum verification, the corruption should be detected - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, true); try (ParquetFileReader reader = getParquetFileReader(path, conf, List.of(colADesc, colBDesc))) { // We expect an exception on the first encountered corrupt page (in readAllPages) assertVerificationFailed(reader); @@ -519,7 +520,7 @@ public void testCorruptedPageV2() throws IOException { private void testCompression(ParquetProperties.WriterVersion version) throws IOException { Configuration conf = new Configuration(); conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, true); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, true); Path path = writeSimpleParquetFile(conf, CompressionCodecName.SNAPPY, version); @@ -566,7 +567,7 @@ private void testNestedWithNulls(ParquetProperties.WriterVersion version) throws // Write out sample file via the non-checksum code path, extract the raw bytes to calculate the // reference crc with conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, false); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, false); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, false); Path refPath = writeNestedWithNullsSampleParquetFile(conf, false, CompressionCodecName.SNAPPY, version); try (ParquetFileReader refReader = getParquetFileReader(refPath, conf, List.of(colCIdDesc, colDValDesc))) { @@ -576,7 +577,7 @@ private void testNestedWithNulls(ParquetProperties.WriterVersion version) throws // Write out sample file with checksums conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, true); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, true); Path path = writeNestedWithNullsSampleParquetFile(conf, false, CompressionCodecName.SNAPPY, version); try (ParquetFileReader reader = getParquetFileReader(path, conf, List.of(colCIdDesc, colDValDesc))) { @@ -609,7 +610,7 @@ private void testDictionaryEncoding(ParquetProperties.WriterVersion version) thr // Write out dictionary encoded sample file via the non-checksum code path, extract the raw // bytes to calculate the reference crc with conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, false); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, false); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, false); Path refPath = writeNestedWithNullsSampleParquetFile(conf, true, CompressionCodecName.SNAPPY, version); try (ParquetFileReader refReader = @@ -622,7 +623,7 @@ private void testDictionaryEncoding(ParquetProperties.WriterVersion version) thr // Write out sample file with checksums conf.setBoolean(ParquetOutputFormat.PAGE_WRITE_CHECKSUM_ENABLED, true); - conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true); + conf.setBoolean(ParquetInputProperties.PAGE_VERIFY_CHECKSUM_ENABLED, true); Path path = writeNestedWithNullsSampleParquetFile(conf, true, CompressionCodecName.SNAPPY, version); try (ParquetFileReader reader = getParquetFileReader(path, conf, Collections.singletonList(colDValDesc))) { 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 d4fb01022e..3c52f0ccd9 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 @@ -39,6 +39,7 @@ 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.conf.ParquetInputProperties; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.hadoop.example.ExampleInputFormat; @@ -116,7 +117,7 @@ public void testProjectionSize(boolean readType) throws Exception { Configuration conf = new Configuration(); // set the vector IO option - conf.setBoolean(ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED, readType); + conf.setBoolean(ParquetInputProperties.HADOOP_VECTORED_IO_ENABLED, readType); // set the projection schema conf.set( "parquet.read.schema", diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java index 69e6bc3f2a..58933013d3 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java @@ -22,10 +22,10 @@ import static org.apache.parquet.column.Encoding.BIT_PACKED; import static org.apache.parquet.column.Encoding.PLAIN; import static org.apache.parquet.column.Encoding.RLE_DICTIONARY; +import static org.apache.parquet.conf.ParquetInputProperties.READ_SUPPORT_CLASS; import static org.apache.parquet.format.converter.ParquetMetadataConverter.MAX_STATS_SIZE; import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.CREATE; import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.OVERWRITE; -import static org.apache.parquet.hadoop.ParquetInputFormat.READ_SUPPORT_CLASS; import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE; import static org.apache.parquet.hadoop.ParquetWriter.MAX_PADDING_SIZE_DEFAULT; import static org.apache.parquet.hadoop.TestUtils.enforceEmptyDir; @@ -76,6 +76,7 @@ import org.apache.parquet.column.statistics.LongStatistics; import org.apache.parquet.column.values.bloomfilter.BlockSplitBloomFilter; import org.apache.parquet.column.values.bloomfilter.BloomFilter; +import org.apache.parquet.conf.ParquetInputProperties; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.SimpleGroup; import org.apache.parquet.format.Statistics; @@ -160,7 +161,7 @@ public class TestParquetFileWriter { private Configuration getTestConfiguration(boolean vectoredRead) { Configuration conf = new Configuration(); // set the vector IO option - conf.setBoolean(ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED, vectoredRead); + conf.setBoolean(ParquetInputProperties.HADOOP_VECTORED_IO_ENABLED, vectoredRead); return conf; } diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java index 203e7ab91a..b8b85d6baf 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/example/TestInputOutputFormat.java @@ -45,6 +45,7 @@ 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.conf.ParquetInputProperties; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.filter2.predicate.FilterApi; @@ -96,7 +97,7 @@ public void setUp() { private Configuration createConf(boolean readType) { Configuration conf = new Configuration(); - conf.setBoolean(ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED, readType); + conf.setBoolean(ParquetInputProperties.HADOOP_VECTORED_IO_ENABLED, readType); return conf; } @@ -268,12 +269,12 @@ public void testReadWriteTaskSideMDAggressiveFilter(boolean readType) // this filter predicate should trigger row group filtering that drops all row-groups ParquetInputFormat.setFilterPredicate(conf, FilterApi.eq(FilterApi.intColumn("line"), -1000)); - final String fpString = conf.get(ParquetInputFormat.FILTER_PREDICATE); + final String fpString = conf.get(ParquetInputProperties.FILTER_PREDICATE); runMapReduceJob(conf, CompressionCodecName.UNCOMPRESSED, new HashMap() { { put("parquet.task.side.metadata", "true"); - put(ParquetInputFormat.FILTER_PREDICATE, fpString); + put(ParquetInputProperties.FILTER_PREDICATE, fpString); } }); @@ -290,12 +291,12 @@ public void testReadWriteFilter(boolean readType) throws IOException, ClassNotFo // this filter predicate should keep some records but not all (first 500 characters) // "line" is actually position in the file... ParquetInputFormat.setFilterPredicate(conf, FilterApi.lt(FilterApi.intColumn("line"), 500)); - final String fpString = conf.get(ParquetInputFormat.FILTER_PREDICATE); + final String fpString = conf.get(ParquetInputProperties.FILTER_PREDICATE); runMapReduceJob(conf, CompressionCodecName.UNCOMPRESSED, new HashMap() { { put("parquet.task.side.metadata", "true"); - put(ParquetInputFormat.FILTER_PREDICATE, fpString); + put(ParquetInputProperties.FILTER_PREDICATE, fpString); } });