This is an automated email from the ASF dual-hosted git repository.
danielcweeks pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new cfc5504ad8 parquet: don't parse a full Hadoop config file for each
Parquet write if not needed (#18398)
cfc5504ad8 is described below
commit cfc5504ad8b806c253b2a5773c0078a7b1a7b79b
Author: Nicholas Barton <[email protected]>
AuthorDate: Thu Oct 8 17:46:21 2026 -0400
parquet: don't parse a full Hadoop config file for each Parquet write if
not needed (#18398)
* core: dont parse a full Hadoop config file for each write if not needed
---
.../java/org/apache/iceberg/parquet/Parquet.java | 9 ++-
.../iceberg/parquet/ParquetCodecFactory.java | 35 +++++++---
.../java/org/apache/iceberg/parquet/ParquetIO.java | 8 ++-
.../org/apache/iceberg/parquet/ParquetWriter.java | 6 +-
.../iceberg/parquet/TestParquetCodecFactory.java | 74 ++++++++++++++++++++++
.../apache/iceberg/parquet/TestVariantMetrics.java | 4 +-
6 files changed, 115 insertions(+), 21 deletions(-)
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
index d5a036e1f1..71d9e0ddb7 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
@@ -117,6 +117,8 @@ import org.apache.parquet.avro.AvroReadSupport;
import org.apache.parquet.avro.AvroWriteSupport;
import org.apache.parquet.column.ParquetProperties;
import org.apache.parquet.column.ParquetProperties.WriterVersion;
+import org.apache.parquet.conf.HadoopParquetConfiguration;
+import org.apache.parquet.conf.ParquetConfiguration;
import org.apache.parquet.conf.PlainParquetConfiguration;
import org.apache.parquet.crypto.FileDecryptionProperties;
import org.apache.parquet.crypto.FileEncryptionProperties;
@@ -169,7 +171,7 @@ public class Parquet {
public static class WriteBuilder implements InternalData.WriteBuilder {
private final OutputFile file;
- private final Configuration conf;
+ private final ParquetConfiguration conf;
private final Map<String, String> metadata = Maps.newLinkedHashMap();
private final Map<String, String> config = Maps.newLinkedHashMap();
private Schema schema = null;
@@ -186,9 +188,10 @@ public class Parquet {
private WriteBuilder(OutputFile file) {
this.file = file;
if (file instanceof HadoopOutputFile) {
- this.conf = new Configuration(((HadoopOutputFile) file).getConf());
+ this.conf =
+ new HadoopParquetConfiguration(new
Configuration(((HadoopOutputFile) file).getConf()));
} else {
- this.conf = new Configuration();
+ this.conf = new PlainParquetConfiguration();
}
}
diff --git
a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetCodecFactory.java
b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetCodecFactory.java
index bfe87f8e50..f0089b3897 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetCodecFactory.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetCodecFactory.java
@@ -21,18 +21,29 @@ package org.apache.iceberg.parquet;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.parquet.conf.HadoopParquetConfiguration;
+import org.apache.parquet.conf.ParquetConfiguration;
import org.apache.parquet.hadoop.BadConfigurationException;
import org.apache.parquet.hadoop.CodecFactory;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.hadoop.util.ConfigurationUtil;
/**
- * This class implements a codec factory that is used when reading from
Parquet. It adds a
- * workaround to cache codecs by name and level, not just by name. This can be
removed when this
- * change is made to Parquet.
+ * A codec factory that caches codecs by compression level as well as by name,
so that requests for
+ * the same codec at different levels do not share an instance.
*/
public class ParquetCodecFactory extends CodecFactory {
+ /**
+ * @deprecated since 1.13.0; this class is internal to {@link Parquet} and
should not be
+ * constructed directly.
+ */
+ @Deprecated
public ParquetCodecFactory(Configuration configuration, int pageSize) {
+ this(new HadoopParquetConfiguration(configuration), pageSize);
+ }
+
+ ParquetCodecFactory(ParquetConfiguration configuration, int pageSize) {
super(configuration, pageSize);
}
@@ -57,10 +68,14 @@ public class ParquetCodecFactory extends CodecFactory {
try {
codecClass = Class.forName(codecClassName);
} catch (ClassNotFoundException e) {
- // Try to load the class using the job classloader
- codecClass = configuration.getClassLoader().loadClass(codecClassName);
+ // Try to load the class using the configuration's classloader
+ codecClass = conf.getClassByName(codecClassName);
}
- codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass,
configuration);
+ // Hadoop codecs are Configurable, so instantiation needs a Hadoop
Configuration
+ codec =
+ (CompressionCodec)
+ ReflectionUtils.newInstance(
+ codecClass,
ConfigurationUtil.createHadoopConfiguration(conf));
CODEC_BY_NAME.put(cacheKey, codec);
return codec;
} catch (ClassNotFoundException e) {
@@ -73,16 +88,16 @@ public class ParquetCodecFactory extends CodecFactory {
String level = null;
switch (codecName) {
case GZIP:
- level = configuration.get("zlib.compress.level");
+ level = conf.get("zlib.compress.level");
break;
case BROTLI:
- level = configuration.get("compression.brotli.quality");
+ level = conf.get("compression.brotli.quality");
break;
case ZSTD:
- level = configuration.get("parquet.compression.codec.zstd.level");
+ level = conf.get("parquet.compression.codec.zstd.level");
if (level == null) {
// keep "io.compression.codec.zstd.level" for backwards compatibility
- level = configuration.get("io.compression.codec.zstd.level");
+ level = conf.get("io.compression.codec.zstd.level");
}
break;
default:
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetIO.java
b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetIO.java
index ecca4eb0d6..e22c813125 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetIO.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetIO.java
@@ -25,7 +25,6 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.iceberg.exceptions.RuntimeIOException;
@@ -37,6 +36,8 @@ import org.apache.iceberg.io.FileRange;
import org.apache.iceberg.io.RangeReadable;
import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.parquet.bytes.ByteBufferAllocator;
+import org.apache.parquet.conf.ParquetConfiguration;
+import org.apache.parquet.hadoop.util.ConfigurationUtil;
import org.apache.parquet.hadoop.util.HadoopStreams;
import org.apache.parquet.io.DelegatingPositionOutputStream;
import org.apache.parquet.io.DelegatingSeekableInputStream;
@@ -80,11 +81,12 @@ class ParquetIO {
return new ParquetOutputFile(file);
}
- static OutputFile file(org.apache.iceberg.io.OutputFile file, Configuration
conf) {
+ static OutputFile file(org.apache.iceberg.io.OutputFile file,
ParquetConfiguration conf) {
if (file instanceof HadoopOutputFile) {
HadoopOutputFile hfile = (HadoopOutputFile) file;
try {
- return
org.apache.parquet.hadoop.util.HadoopOutputFile.fromPath(hfile.getPath(), conf);
+ return org.apache.parquet.hadoop.util.HadoopOutputFile.fromPath(
+ hfile.getPath(),
ConfigurationUtil.createHadoopConfiguration(conf));
} catch (IOException e) {
throw new RuntimeIOException(
e, "Failed to create Parquet output file for %s", file.location());
diff --git
a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
index ee5e5c264a..72ed83da49 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
@@ -24,7 +24,6 @@ import java.io.UncheckedIOException;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
-import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.Metrics;
import org.apache.iceberg.MetricsConfig;
import org.apache.iceberg.Schema;
@@ -35,6 +34,7 @@ import
org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.parquet.column.ColumnWriteStore;
import org.apache.parquet.column.ParquetProperties;
import org.apache.parquet.compression.CompressionCodecFactory;
+import org.apache.parquet.conf.ParquetConfiguration;
import org.apache.parquet.crypto.FileEncryptionProperties;
import org.apache.parquet.crypto.InternalFileEncryptor;
import org.apache.parquet.hadoop.ColumnChunkPageWriteStore;
@@ -57,7 +57,7 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
private final int columnIndexTruncateLength;
private final ParquetFileWriter.Mode writeMode;
private final OutputFile output;
- private final Configuration conf;
+ private final ParquetConfiguration conf;
private final InternalFileEncryptor fileEncryptor;
private final boolean trackUncompressedSize;
@@ -75,7 +75,7 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
@SuppressWarnings("unchecked")
ParquetWriter(
- Configuration conf,
+ ParquetConfiguration conf,
OutputFile output,
Schema schema,
MessageType parquetSchema,
diff --git
a/parquet/src/test/java/org/apache/iceberg/parquet/TestParquetCodecFactory.java
b/parquet/src/test/java/org/apache/iceberg/parquet/TestParquetCodecFactory.java
new file mode 100644
index 0000000000..cdb95a742e
--- /dev/null
+++
b/parquet/src/test/java/org/apache/iceberg/parquet/TestParquetCodecFactory.java
@@ -0,0 +1,74 @@
+/*
+ * 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.iceberg.parquet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.parquet.conf.ParquetConfiguration;
+import org.apache.parquet.conf.PlainParquetConfiguration;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.junit.jupiter.api.Test;
+
+final class TestParquetCodecFactory {
+
+ private static final int PAGE_SIZE = 1024 * 1024;
+
+ @Test
+ void testCachesCodecsByLevelWithoutHadoopConfiguration() {
+ PlainParquetConfiguration level3 = new PlainParquetConfiguration();
+ level3.set("parquet.compression.codec.zstd.level", "3");
+ PlainParquetConfiguration level5 = new PlainParquetConfiguration();
+ level5.set("parquet.compression.codec.zstd.level", "5");
+
+ CompressionCodec codecAtLevel3 = codec(level3, CompressionCodecName.ZSTD);
+ CompressionCodec codecAtLevel5 = codec(level5, CompressionCodecName.ZSTD);
+
+ assertThat(codecAtLevel3).isNotNull();
+ assertThat(codecAtLevel5).isNotNull().isNotSameAs(codecAtLevel3);
+ assertThat(codec(level3,
CompressionCodecName.ZSTD)).isSameAs(codecAtLevel3);
+ }
+
+ @Test
+ void testLegacyZstdLevelPropertyIsPartOfTheCacheKey() {
+ PlainParquetConfiguration legacy = new PlainParquetConfiguration();
+ legacy.set("io.compression.codec.zstd.level", "7");
+ PlainParquetConfiguration current = new PlainParquetConfiguration();
+ current.set("parquet.compression.codec.zstd.level", "7");
+
+ assertThat(codec(legacy, CompressionCodecName.ZSTD))
+ .isSameAs(codec(current, CompressionCodecName.ZSTD));
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void testHadoopConfigurationStillAccepted() {
+ Configuration conf = new Configuration(false);
+ conf.set("zlib.compress.level", "BEST_COMPRESSION");
+ ParquetCodecFactory factory = new ParquetCodecFactory(conf, PAGE_SIZE);
+
+ assertThat(factory.getCompressor(CompressionCodecName.GZIP)).isNotNull();
+ assertThat(factory.getCodec(CompressionCodecName.UNCOMPRESSED)).isNull();
+ }
+
+ private static CompressionCodec codec(ParquetConfiguration conf,
CompressionCodecName name) {
+ return new ParquetCodecFactory(conf, PAGE_SIZE).getCodec(name);
+ }
+}
diff --git
a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java
b/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java
index e8fb8f1452..202e6362eb 100644
--- a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java
+++ b/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantMetrics.java
@@ -30,7 +30,6 @@ import java.util.Set;
import java.util.UUID;
import java.util.function.Consumer;
import java.util.stream.Stream;
-import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.Metrics;
import org.apache.iceberg.MetricsConfig;
import org.apache.iceberg.MetricsTestUtil;
@@ -60,6 +59,7 @@ import org.apache.iceberg.variants.VariantValue;
import org.apache.iceberg.variants.Variants;
import org.apache.parquet.column.ParquetProperties;
import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.conf.PlainParquetConfiguration;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.ParquetFileWriter;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
@@ -878,7 +878,7 @@ public class TestVariantMetrics {
// Parquet.write() cannot disable stats on variant sub-columns (no field
IDs)
ParquetWriter<Record> writer =
new ParquetWriter<>(
- new Configuration(),
+ new PlainParquetConfiguration(),
out,
SCHEMA,
parquetSchema,
