Propagate delete file replication factor for Hadoop parquet files

[shanthoosh]

Problem Summary

PRs #219 and #229 added support for configuring the replication factor of delete files (delete-file-replication write option, spark.sql.iceberg.delete-file-replication sessionconf, write.delete-file-replication table property). The configured value is resolved correctly by SparkWriteConf and carried all the way into Iceberg's HadoopOutputFile, but for Parquet delete files it never reaches HDFS — and Parquet is the default delete file format, so in practice the setting had no effect.

Root cause: ParquetIO.file(...) short-circuits Hadoop-backed output files by rebuilding a parquet-hadoop OutputFile from just the path and Configuration:

if (file instanceof HadoopOutputFile) {
  HadoopOutputFile hfile = (HadoopOutputFile) file;
  return org.apache.parquet.hadoop.util.HadoopOutputFile.fromPath(hfile.getPath(), hfile.getConf());
}

parquet-hadoop then opens the output stream itself via fs.create(path, overwrite, bufferSize, fs.getDefaultReplication(path), blockSize) — using the filesystem default replication and discarding the replication factor carried by the Iceberg HadoopOutputFile. Iceberg's HadoopOutputFile.create(), the only place the configured replication is applied, is never invoked. As a result, merge-on-read DELETE/UPDATE/MERGE always produces Parquet delete files with the cluster default replication regardless of configuration.

Solution

• Expose the configured replication on HadoopOutputFile via a replication() accessor (non-positive means "not configured").
• In both ParquetIO.file(...) overloads, take the parquet-hadoop shortcut only when no custom replication is configured. When a replication factor is set, route the file through the stream-based ParquetOutputFile — the same path already used for non-Hadoop FileIO implementations — so the output stream is created by Iceberg's HadoopOutputFile.create(), which passes the configured replication to fs.create(...).
• Fix a latent bug this exposed: ParquetIO.stream(...) unwrapped Iceberg's position output stream to the raw FSDataOutputStream and abandoned the Iceberg wrapper. The wrapper's finalizer closes the underlying stream when garbage collected, which truncated files mid-write (ClosedChannelException) once Hadoop files started flowing through this path. The stream is now always wrapped in ParquetOutputStreamAdapter, which keeps the Iceberg stream referenced and closes it properly. This branch was previously unreachable for Hadoop files, so existing behavior is unaffected.

Behavior is unchanged when no replication is configured: the parquet-hadoop shortcut is still used and files inherit the filesystem default.

Testing Done

Unit tests — parquet:TestParquetReplication (new), using a RawLocalFileSystem subclass that records the replication factor passed to every FileSystem.create(...) call:

• routing: both ParquetIO.file(...) overloads use the parquet-hadoop shortcut without custom replication and skip it when one is set
• configured replication (5) reaches stream creation through all three writer paths: the default parquet-mr appender (with a read-back proving the file is valid Parquet), the Iceberg-native ParquetWriter, and the position-delete writer
• with nothing configured, the filesystem default replication is used (shortcut preserved)

Integration tests (SQL-only) — spark 3.5:TestParquetDeleteFileReplication (new): the catalog warehouse lives on a capturing filesystem, and the whole flow is driven through SQL —
CREATE TABLE (format-version 2, merge-on-read, Parquet) → INSERT → DELETE →
SELECT file_path FROM t.delete_files:

• 'write.delete-file-replication'='5' table property → all position delete files created with replication 5; data files keep the filesystem default; SELECT after DELETE returns correct rows (deletes apply on read)
• SET spark.sql.iceberg.delete-file-replication=7 session conf → delete files created with replication 7 (note: the hyphenated key must be backquoted in Spark SQL SET)