From b01576a4ff02e655bd70814cf735970fdf996bc9 Mon Sep 17 00:00:00 2001 From: Shanthoosh Pazhanjur Venkataraman Date: Mon, 13 Jul 2026 19:20:06 -0700 Subject: [PATCH] Parquet: Propagate delete file replication factor to Hadoop output streams ParquetIO's Hadoop shortcut rebuilds a parquet-hadoop OutputFile from just the path and Configuration, which discards the replication factor carried by Iceberg's HadoopOutputFile and opens the stream with the filesystem default replication. Route Hadoop output files with a custom replication factor through the stream-based ParquetOutputFile so the stream is created by Iceberg's HadoopOutputFile with the configured replication. Also stop unwrapping Iceberg's position output stream to the underlying FSDataOutputStream: the abandoned wrapper's finalizer closes the stream mid-write once it is garbage collected. Verified end-to-end through SQL-only Spark 3.5 tests: CREATE TABLE with merge-on-read Parquet, INSERT, DELETE, and the delete_files metadata table, with the catalog warehouse on a capturing filesystem that records the replication factor passed to FileSystem.create. Covers the write.delete-file-replication table property and the spark.sql.iceberg.delete-file-replication session conf set via SQL SET, and verifies data files keep the filesystem default. Co-Authored-By: Claude Fable 5 --- .../iceberg/hadoop/HadoopOutputFile.java | 5 + .../org/apache/iceberg/parquet/ParquetIO.java | 37 +-- .../parquet/TestParquetReplication.java | 213 ++++++++++++++++++ .../TestParquetDeleteFileReplication.java | 195 ++++++++++++++++ 4 files changed, 432 insertions(+), 18 deletions(-) create mode 100644 parquet/src/test/java/org/apache/iceberg/parquet/TestParquetReplication.java create mode 100644 spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestParquetDeleteFileReplication.java diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopOutputFile.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopOutputFile.java index 9224c4ef5c..69a4d72cd4 100644 --- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopOutputFile.java +++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopOutputFile.java @@ -157,6 +157,11 @@ public FileSystem getFileSystem() { return fs; } + /** Returns the configured replication factor, or a non-positive value when not set. */ + public short replication() { + return replication; + } + @Override public String location() { return path.toString(); 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 8b2a6e242b..758ce1995d 100644 --- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetIO.java +++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetIO.java @@ -20,15 +20,12 @@ import java.io.IOException; import java.io.InputStream; -import java.io.OutputStream; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; -import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.iceberg.exceptions.RuntimeIOException; import org.apache.iceberg.hadoop.HadoopInputFile; import org.apache.iceberg.hadoop.HadoopOutputFile; import org.apache.iceberg.io.DelegatingInputStream; -import org.apache.iceberg.io.DelegatingOutputStream; import org.apache.parquet.hadoop.util.HadoopStreams; import org.apache.parquet.io.DelegatingPositionOutputStream; import org.apache.parquet.io.DelegatingSeekableInputStream; @@ -59,11 +56,15 @@ static InputFile file(org.apache.iceberg.io.InputFile file) { static OutputFile file(org.apache.iceberg.io.OutputFile file) { if (file instanceof HadoopOutputFile) { HadoopOutputFile hfile = (HadoopOutputFile) file; - try { - return org.apache.parquet.hadoop.util.HadoopOutputFile.fromPath( - hfile.getPath(), hfile.getConf()); - } catch (IOException e) { - throw new RuntimeIOException(e, "Failed to create Parquet output file for %s", file); + // the Parquet Hadoop output file opens its own stream with the filesystem default + // replication, so it cannot be used when a custom replication factor is configured + if (hfile.replication() <= 0) { + try { + return org.apache.parquet.hadoop.util.HadoopOutputFile.fromPath( + hfile.getPath(), hfile.getConf()); + } catch (IOException e) { + throw new RuntimeIOException(e, "Failed to create Parquet output file for %s", file); + } } } return new ParquetOutputFile(file); @@ -72,10 +73,14 @@ static OutputFile file(org.apache.iceberg.io.OutputFile file) { static OutputFile file(org.apache.iceberg.io.OutputFile file, Configuration conf) { if (file instanceof HadoopOutputFile) { HadoopOutputFile hfile = (HadoopOutputFile) file; - try { - return org.apache.parquet.hadoop.util.HadoopOutputFile.fromPath(hfile.getPath(), conf); - } catch (IOException e) { - throw new RuntimeIOException(e, "Failed to create Parquet output file for %s", file); + // the Parquet Hadoop output file opens its own stream with the filesystem default + // replication, so it cannot be used when a custom replication factor is configured + if (hfile.replication() <= 0) { + try { + return org.apache.parquet.hadoop.util.HadoopOutputFile.fromPath(hfile.getPath(), conf); + } catch (IOException e) { + throw new RuntimeIOException(e, "Failed to create Parquet output file for %s", file); + } } } return new ParquetOutputFile(file); @@ -92,12 +97,8 @@ static SeekableInputStream stream(org.apache.iceberg.io.SeekableInputStream stre } static PositionOutputStream stream(org.apache.iceberg.io.PositionOutputStream stream) { - if (stream instanceof DelegatingOutputStream) { - OutputStream wrapped = ((DelegatingOutputStream) stream).getDelegate(); - if (wrapped instanceof FSDataOutputStream) { - return HadoopStreams.wrap((FSDataOutputStream) wrapped); - } - } + // do not unwrap to the underlying FSDataOutputStream: abandoning the Iceberg wrapper lets + // HadoopStreams' finalizer close the stream while Parquet is still writing to it return new ParquetOutputStreamAdapter(stream); } diff --git a/parquet/src/test/java/org/apache/iceberg/parquet/TestParquetReplication.java b/parquet/src/test/java/org/apache/iceberg/parquet/TestParquetReplication.java new file mode 100644 index 0000000000..99f0043a4e --- /dev/null +++ b/parquet/src/test/java/org/apache/iceberg/parquet/TestParquetReplication.java @@ -0,0 +1,213 @@ +/* + * 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.apache.iceberg.types.Types.NestedField.optional; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.File; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecordBuilder; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RawLocalFileSystem; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.util.Progressable; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.avro.AvroSchemaUtil; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.data.parquet.GenericParquetWriter; +import org.apache.iceberg.deletes.PositionDelete; +import org.apache.iceberg.deletes.PositionDeleteWriter; +import org.apache.iceberg.hadoop.HadoopOutputFile; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +public class TestParquetReplication { + + private static final Schema SCHEMA = new Schema(optional(1, "id", Types.IntegerType.get())); + private static final short CUSTOM_REPLICATION = 5; + + @TempDir private File tempDir; + + @Test + public void testHadoopShortcutUsedOnlyWithoutCustomReplication() { + Configuration conf = new Configuration(); + Path path = newPath("routing.parquet"); + + OutputFile withReplication = HadoopOutputFile.fromPath(path, conf, CUSTOM_REPLICATION); + assertThat(((HadoopOutputFile) withReplication).replication()).isEqualTo(CUSTOM_REPLICATION); + assertThat(ParquetIO.file(withReplication)) + .isNotInstanceOf(org.apache.parquet.hadoop.util.HadoopOutputFile.class); + assertThat(ParquetIO.file(withReplication, conf)) + .isNotInstanceOf(org.apache.parquet.hadoop.util.HadoopOutputFile.class); + + OutputFile withoutReplication = HadoopOutputFile.fromPath(path, conf); + assertThat(ParquetIO.file(withoutReplication)) + .isInstanceOf(org.apache.parquet.hadoop.util.HadoopOutputFile.class); + assertThat(ParquetIO.file(withoutReplication, conf)) + .isInstanceOf(org.apache.parquet.hadoop.util.HadoopOutputFile.class); + } + + @Test + public void testReplicationPropagatesThroughDefaultAppender() throws IOException { + Configuration conf = captureConf(); + Path path = newPath("default-appender.parquet"); + OutputFile out = HadoopOutputFile.fromPath(path, conf, CUSTOM_REPLICATION); + + try (FileAppender appender = + Parquet.write(out).schema(SCHEMA).named("test").build()) { + appender.add(record(1)); + } + + assertThat(CaptureReplicationFileSystem.capturedReplication(path)) + .isEqualTo(CUSTOM_REPLICATION); + + // the file written through the stream-based path must still be valid parquet + List rows = Lists.newArrayList(); + try (CloseableIterable reader = + Parquet.read(out.toInputFile()).project(SCHEMA).build()) { + reader.forEach(rows::add); + } + assertThat(rows).hasSize(1); + assertThat(rows.get(0).get("id")).isEqualTo(1); + } + + @Test + public void testReplicationPropagatesThroughIcebergParquetWriter() throws IOException { + Configuration conf = captureConf(); + Path path = newPath("iceberg-writer.parquet"); + OutputFile out = HadoopOutputFile.fromPath(path, conf, CUSTOM_REPLICATION); + + try (FileAppender appender = + Parquet.write(out) + .schema(SCHEMA) + .named("test") + .createWriterFunc(ParquetAvroWriter::buildWriter) + .build()) { + appender.add(record(2)); + } + + assertThat(CaptureReplicationFileSystem.capturedReplication(path)) + .isEqualTo(CUSTOM_REPLICATION); + } + + @Test + public void testReplicationPropagatesThroughPositionDeleteWriter() throws IOException { + Configuration conf = captureConf(); + Path path = newPath("position-deletes.parquet"); + OutputFile out = HadoopOutputFile.fromPath(path, conf, CUSTOM_REPLICATION); + + PositionDeleteWriter deleteWriter = + Parquet.writeDeletes(out) + .createWriterFunc(GenericParquetWriter::buildWriter) + .withSpec(PartitionSpec.unpartitioned()) + .buildPositionWriter(); + + PositionDelete delete = PositionDelete.create(); + try (PositionDeleteWriter writer = deleteWriter) { + writer.write(delete.set("file:/data/data-file.parquet", 0L, null)); + } + + assertThat(deleteWriter.toDeleteFile().recordCount()).isEqualTo(1L); + assertThat(CaptureReplicationFileSystem.capturedReplication(path)) + .isEqualTo(CUSTOM_REPLICATION); + } + + @Test + public void testFilesystemDefaultReplicationWhenNotConfigured() throws IOException { + Configuration conf = captureConf(); + Path path = newPath("default-replication.parquet"); + OutputFile out = HadoopOutputFile.fromPath(path, conf); + + try (FileAppender appender = + Parquet.write(out).schema(SCHEMA).named("test").build()) { + appender.add(record(3)); + } + + short fsDefault = path.getFileSystem(conf).getDefaultReplication(path); + assertThat(CaptureReplicationFileSystem.capturedReplication(path)).isEqualTo(fsDefault); + } + + private Path newPath(String name) { + return new Path(new File(tempDir, name).toURI()); + } + + private Configuration captureConf() { + Configuration conf = new Configuration(); + conf.setClass("fs.file.impl", CaptureReplicationFileSystem.class, FileSystem.class); + conf.setBoolean("fs.file.impl.disable.cache", true); + return conf; + } + + private GenericData.Record record(int id) { + return new GenericRecordBuilder(AvroSchemaUtil.convert(SCHEMA, "test")).set("id", id).build(); + } + + /** Local filesystem that records the replication factor passed to each stream creation. */ + public static class CaptureReplicationFileSystem extends RawLocalFileSystem { + private static final Map CAPTURED = Maps.newConcurrentMap(); + + static short capturedReplication(Path path) { + Short replication = CAPTURED.get(path.getName()); + assertThat(replication).as("No stream was created for %s", path).isNotNull(); + return replication; + } + + @Override + public FSDataOutputStream create( + Path path, + boolean overwrite, + int bufferSize, + short replication, + long blockSize, + Progressable progress) + throws IOException { + CAPTURED.put(path.getName(), replication); + return super.create(path, overwrite, bufferSize, replication, blockSize, progress); + } + + @Override + public FSDataOutputStream create( + Path path, + FsPermission permission, + boolean overwrite, + int bufferSize, + short replication, + long blockSize, + Progressable progress) + throws IOException { + CAPTURED.put(path.getName(), replication); + return super.create( + path, permission, overwrite, bufferSize, replication, blockSize, progress); + } + } +} diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestParquetDeleteFileReplication.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestParquetDeleteFileReplication.java new file mode 100644 index 0000000000..c626f126d6 --- /dev/null +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestParquetDeleteFileReplication.java @@ -0,0 +1,195 @@ +/* + * 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.spark; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.net.URI; +import java.nio.file.Files; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RawLocalFileSystem; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.util.Progressable; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** + * End-to-end SQL tests for delete file replication with Parquet. + * + *

The catalog warehouse lives on a dedicated {@code capturefs:} filesystem that records the + * replication factor passed to every {@link FileSystem#create} call, which is what HDFS would + * receive in production. The full write path is exercised through SQL only: {@code CREATE TABLE} + * with merge-on-read Parquet, {@code INSERT}, {@code DELETE}, and the {@code delete_files} metadata + * table for locating the produced position delete files. + */ +public class TestParquetDeleteFileReplication extends TestBase { + + private static final String CATALOG = "parqreplcat"; + private static final String TABLE = CATALOG + ".default.parquet_repl_test"; + + @BeforeAll + public static void setUpCatalog() throws IOException { + // session confs are copied into sessionState().newHadoopConf(), which SparkCatalog uses + spark.conf().set("fs.capturefs.impl", CaptureReplicationFileSystem.class.getName()); + spark.conf().set("fs.capturefs.impl.disable.cache", "true"); + + String warehouse = Files.createTempDirectory("parquet-repl-warehouse").toString(); + spark.conf().set("spark.sql.catalog." + CATALOG, SparkCatalog.class.getName()); + spark.conf().set("spark.sql.catalog." + CATALOG + ".type", "hadoop"); + spark.conf().set("spark.sql.catalog." + CATALOG + ".warehouse", "capturefs:" + warehouse); + } + + @AfterEach + public void cleanUp() { + spark.conf().unset(SparkSQLProperties.DELETE_FILE_REPLICATION); + sql("DROP TABLE IF EXISTS %s", TABLE); + } + + @Test + public void testDeleteFileReplicationFromTableProperty() { + sql( + "CREATE TABLE %s (id INT, data STRING) USING iceberg " + + "TBLPROPERTIES (" + + "'format-version'='2', " + + "'write.delete.mode'='merge-on-read', " + + "'write.format.default'='parquet', " + + "'write.delete-file-replication'='5')", + TABLE); + + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e')", TABLE); + sql("DELETE FROM %s WHERE id IN (1, 2)", TABLE); + + // the delete files must reach the filesystem with the configured replication factor + List deleteFiles = sql("SELECT file_path FROM %s.delete_files", TABLE); + assertThat(deleteFiles).isNotEmpty(); + for (Object[] row : deleteFiles) { + String location = (String) row[0]; + assertThat(location).endsWith(".parquet"); + assertThat(CaptureReplicationFileSystem.capturedReplication(location)) + .as("Delete file %s should be created with the configured replication", location) + .isEqualTo((short) 5); + } + + // data files are not covered by the delete file replication setting + List dataFiles = sql("SELECT file_path FROM %s.files WHERE content = 0", TABLE); + assertThat(dataFiles).isNotEmpty(); + for (Object[] row : dataFiles) { + String location = (String) row[0]; + assertThat(CaptureReplicationFileSystem.capturedReplication(location)) + .as("Data file %s should keep the filesystem default replication", location) + .isNotEqualTo((short) 5); + } + + // merge-on-read must still resolve the deletes on read + assertThat(sql("SELECT id FROM %s ORDER BY id", TABLE)) + .extracting(row -> row[0]) + .containsExactly(3, 4, 5); + } + + @Test + public void testDeleteFileReplicationFromSqlSessionConf() { + sql( + "CREATE TABLE %s (id INT, data STRING) USING iceberg " + + "TBLPROPERTIES (" + + "'format-version'='2', " + + "'write.delete.mode'='merge-on-read', " + + "'write.format.default'='parquet')", + TABLE); + + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e')", TABLE); + + // the key contains hyphens, so the Spark SQL parser requires backquotes + sql("SET `%s`=7", SparkSQLProperties.DELETE_FILE_REPLICATION); + sql("DELETE FROM %s WHERE id IN (1, 2)", TABLE); + + List deleteFiles = sql("SELECT file_path FROM %s.delete_files", TABLE); + assertThat(deleteFiles).isNotEmpty(); + for (Object[] row : deleteFiles) { + String location = (String) row[0]; + assertThat(location).endsWith(".parquet"); + assertThat(CaptureReplicationFileSystem.capturedReplication(location)) + .as("Delete file %s should be created with the session conf replication", location) + .isEqualTo((short) 7); + } + + assertThat(sql("SELECT id FROM %s ORDER BY id", TABLE)) + .extracting(row -> row[0]) + .containsExactly(3, 4, 5); + } + + /** + * Local filesystem under a dedicated {@code capturefs:} scheme that records the replication + * factor passed to each stream creation. + */ + public static class CaptureReplicationFileSystem extends RawLocalFileSystem { + private static final Map CAPTURED = Maps.newConcurrentMap(); + + static short capturedReplication(String location) { + String name = new Path(location).getName(); + Short replication = CAPTURED.get(name); + assertThat(replication).as("No stream was created for %s", location).isNotNull(); + return replication; + } + + @Override + public String getScheme() { + return "capturefs"; + } + + @Override + public URI getUri() { + return URI.create("capturefs:///"); + } + + @Override + public FSDataOutputStream create( + Path path, + boolean overwrite, + int bufferSize, + short replication, + long blockSize, + Progressable progress) + throws IOException { + CAPTURED.put(path.getName(), replication); + return super.create(path, overwrite, bufferSize, replication, blockSize, progress); + } + + @Override + public FSDataOutputStream create( + Path path, + FsPermission permission, + boolean overwrite, + int bufferSize, + short replication, + long blockSize, + Progressable progress) + throws IOException { + CAPTURED.put(path.getName(), replication); + return super.create( + path, permission, overwrite, bufferSize, replication, blockSize, progress); + } + } +}