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); + } + } +}