This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12368-924810d8730a3619cf478bb96ddc6217ed0be7ec in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 9fcff6e819ebed84f6d3e7e2e779aba4d5573cb0 Author: Ricky Makhija <[email protected]> AuthorDate: Sun Oct 18 05:56:10 2026 +0000 [Fix][Connector-File] Reuse the connector FileSystem for Parquet and ORC reads/writes (#12368) --- .../seatunnel/file/hadoop/FileSystemInputFile.java | 65 +++++++++++ .../file/hadoop/FileSystemOutputFile.java | 96 ++++++++++++++++ .../file/hadoop/HadoopFileSystemProxy.java | 8 ++ .../file/sink/writer/ParquetWriteStrategy.java | 8 +- .../file/source/reader/OrcReadStrategy.java | 6 +- .../file/source/reader/ParquetReadStrategy.java | 52 +++++++-- .../source/split/ParquetFileSplitStrategy.java | 6 +- .../file/hadoop/FileSystemParquetFileTest.java | 127 +++++++++++++++++++++ 8 files changed, 354 insertions(+), 14 deletions(-) diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemInputFile.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemInputFile.java new file mode 100644 index 00000000000..59fe9685656 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemInputFile.java @@ -0,0 +1,65 @@ +/* + * 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.seatunnel.connectors.seatunnel.file.hadoop; + +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.hadoop.util.HadoopStreams; +import org.apache.parquet.io.InputFile; +import org.apache.parquet.io.SeekableInputStream; + +import java.io.IOException; + +/** + * A Parquet {@link InputFile} bound to an already-open {@link FileSystem}. + * + * <p>Read-side counterpart of {@link FileSystemOutputFile}: {@code HadoopInputFile.fromPath} + * resolves its own FileSystem, which leaks one per file while the Hadoop FileSystem cache is + * disabled. + */ +public class FileSystemInputFile implements InputFile { + + private final FileSystem fileSystem; + private final FileStatus status; + + public FileSystemInputFile(FileSystem fileSystem, FileStatus status) { + this.fileSystem = fileSystem; + this.status = status; + } + + public static FileSystemInputFile fromPath(FileSystem fileSystem, Path path) + throws IOException { + return new FileSystemInputFile(fileSystem, fileSystem.getFileStatus(path)); + } + + @Override + public long getLength() { + return status.getLen(); + } + + @Override + public SeekableInputStream newStream() throws IOException { + return HadoopStreams.wrap(fileSystem.open(status.getPath())); + } + + @Override + public String toString() { + return status.getPath().toString(); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemOutputFile.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemOutputFile.java new file mode 100644 index 00000000000..043057d1e20 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemOutputFile.java @@ -0,0 +1,96 @@ +/* + * 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.seatunnel.connectors.seatunnel.file.hadoop; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.hadoop.util.HadoopStreams; +import org.apache.parquet.io.OutputFile; +import org.apache.parquet.io.PositionOutputStream; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +/** + * A Parquet {@link OutputFile} bound to an already-open {@link FileSystem}. + * + * <p>{@code HadoopOutputFile.fromPath} resolves its own FileSystem through {@code + * Path#getFileSystem}. The file connector disables the Hadoop FileSystem cache, so that call + * allocates a FileSystem per file which nobody closes; its metrics source stays registered in the + * static {@code DefaultMetricsSystem} for the life of the JVM. Reusing the caller's FileSystem + * avoids the allocation entirely. Stream creation mirrors {@code HadoopOutputFile} exactly. + */ +public class FileSystemOutputFile implements OutputFile { + + /** Mirrors {@code HadoopOutputFile}: a buffer size must be supplied alongside a block size. */ + private static final int DFS_BUFFER_SIZE_DEFAULT = 4096; + + private static final Set<String> BLOCK_FS_SCHEMES = + Collections.unmodifiableSet(new HashSet<>(Arrays.asList("hdfs", "webhdfs", "viewfs"))); + + private final FileSystem fileSystem; + private final Path path; + + public FileSystemOutputFile(FileSystem fileSystem, Path path) { + this.fileSystem = fileSystem; + this.path = fileSystem.makeQualified(path); + } + + @Override + public PositionOutputStream create(long blockSizeHint) throws IOException { + return create(false, blockSizeHint); + } + + @Override + public PositionOutputStream createOrOverwrite(long blockSizeHint) throws IOException { + return create(true, blockSizeHint); + } + + private PositionOutputStream create(boolean overwrite, long blockSizeHint) throws IOException { + return HadoopStreams.wrap( + fileSystem.create( + path, + overwrite, + DFS_BUFFER_SIZE_DEFAULT, + fileSystem.getDefaultReplication(path), + Math.max(fileSystem.getDefaultBlockSize(path), blockSizeHint))); + } + + @Override + public boolean supportsBlockSize() { + return BLOCK_FS_SCHEMES.contains(fileSystem.getUri().getScheme()); + } + + @Override + public long defaultBlockSize() { + return fileSystem.getDefaultBlockSize(path); + } + + @Override + public String getPath() { + return path.toString(); + } + + @Override + public String toString() { + return path.toString(); + } +} diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java index e1269482f26..a29147c0556 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java @@ -450,6 +450,14 @@ public class HadoopFileSystemProxy implements Serializable, Closeable { return fileSystem; } + /** The {@link Configuration} this proxy was built with, initializing it if necessary. */ + public Configuration getConfiguration() { + if (configuration == null) { + initialize(); + } + return configuration; + } + /** Returns the scheme of the initialized target or source file system. */ public String getScheme() { return getFileSystem().getScheme(); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java index 878c900e47a..9e81221ef29 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java @@ -28,6 +28,7 @@ import org.apache.seatunnel.common.exception.CommonError; import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException; +import org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileSystemOutputFile; import org.apache.seatunnel.connectors.seatunnel.file.sink.config.FileSinkConfig; import org.apache.avro.Conversions; @@ -45,7 +46,7 @@ import org.apache.parquet.column.ParquetProperties; import org.apache.parquet.example.data.simple.NanoTime; import org.apache.parquet.hadoop.ParquetFileWriter; import org.apache.parquet.hadoop.ParquetWriter; -import org.apache.parquet.hadoop.util.HadoopOutputFile; +import org.apache.parquet.io.OutputFile; import org.apache.parquet.schema.ConversionPatterns; import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; @@ -199,8 +200,9 @@ public class ParquetWriteStrategy extends AbstractWriteStrategy<ParquetWriter<Ge AvroWriteSupport.WRITE_FIXED_AS_INT96, String.join(",", writePathsAsInt96)); } - HadoopOutputFile outputFile = - HadoopOutputFile.fromPath(path, getConfiguration(hadoopConf)); + OutputFile outputFile = + new FileSystemOutputFile( + hadoopFileSystemProxy.getFileSystem(), path); ParquetWriter<GenericRecord> newWriter = AvroParquetWriter.<GenericRecord>builder(outputFile) .withWriteMode(ParquetFileWriter.Mode.OVERWRITE) diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/OrcReadStrategy.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/OrcReadStrategy.java index 38dc55a97fe..0f1dcfebcb4 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/OrcReadStrategy.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/OrcReadStrategy.java @@ -106,7 +106,8 @@ public class OrcReadStrategy extends AbstractReadStrategy { hadoopFileSystemProxy.doWithHadoopAuth( (configuration, userGroupInformation) -> { OrcFile.ReaderOptions readerOptions = - OrcFile.readerOptions(configuration); + OrcFile.readerOptions(configuration) + .filesystem(hadoopFileSystemProxy.getFileSystem()); return OrcFile.createReader(new Path(path), readerOptions); })) { TypeDescription schema = TypeDescription.createStruct(); @@ -167,7 +168,8 @@ public class OrcReadStrategy extends AbstractReadStrategy { hadoopFileSystemProxy.doWithHadoopAuth( ((configuration, userGroupInformation) -> { OrcFile.ReaderOptions readerOptions = - OrcFile.readerOptions(configuration); + OrcFile.readerOptions(configuration) + .filesystem(hadoopFileSystemProxy.getFileSystem()); return OrcFile.createReader(new Path(path), readerOptions); }))) { TypeDescription schema = reader.getSchema(); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java index de2bc75d69a..baa632495c1 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java @@ -33,6 +33,7 @@ import org.apache.seatunnel.common.exception.CommonError; import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode; import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException; +import org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileSystemInputFile; import org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplit; import org.apache.avro.Conversions; @@ -43,15 +44,17 @@ import org.apache.avro.generic.GenericRecord; import org.apache.avro.util.Utf8; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.Path; +import org.apache.parquet.HadoopReadOptions; import org.apache.parquet.avro.AvroParquetReader; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.NanoTime; import org.apache.parquet.hadoop.ParquetFileReader; import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.api.ReadSupport; import org.apache.parquet.hadoop.example.GroupReadSupport; import org.apache.parquet.hadoop.metadata.FileMetaData; import org.apache.parquet.hadoop.metadata.ParquetMetadata; -import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.io.InputFile; import org.apache.parquet.io.api.Binary; import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.LogicalTypeAnnotation; @@ -128,10 +131,11 @@ public class ParquetReadStrategy extends AbstractReadStrategy { } Path filePath = new Path(path); Map<String, String> partitionsMap = parsePartitionsByPath(path); - HadoopInputFile hadoopInputFile = + InputFile hadoopInputFile = hadoopFileSystemProxy.doWithHadoopAuth( (configuration, userGroupInformation) -> - HadoopInputFile.fromPath(filePath, configuration)); + FileSystemInputFile.fromPath( + hadoopFileSystemProxy.getFileSystem(), filePath)); int fieldsCount = seaTunnelRowType.getTotalFields(); GenericData dataModel = new GenericData(); dataModel.addLogicalTypeConversion(new Conversions.DecimalConversion()); @@ -143,6 +147,10 @@ public class ParquetReadStrategy extends AbstractReadStrategy { AvroParquetReader.Builder<GenericData.Record> builder = AvroParquetReader.<GenericData.Record>builder(hadoopInputFile) .withDataModel(dataModel); + // HadoopConf carries READ_INT96_AS_FIXED / ADD_LIST_ELEMENT_RECORDS. ParquetReader lifts + // those automatically only from a HadoopInputFile, so set them explicitly here. withConf() + // rebuilds the read options, so it must run before withFileRange() below. + builder.withConf(hadoopFileSystemProxy.getConfiguration()); if (useSplitRange) { long start = split.getStart(); long end = start + split.getLength(); @@ -184,9 +192,18 @@ public class ParquetReadStrategy extends AbstractReadStrategy { ParquetReader<Group> reader = hadoopFileSystemProxy.doWithHadoopAuth( (configuration, userGroupInformation) -> { + // ParquetReader.builder(ReadSupport, Path) resolves its own + // FileSystem inside build() (twice: path.getFileSystem plus + // HadoopInputFile.fromStatus), which leaks with the FS cache + // disabled. Bind to the connector's FileSystem instead. ParquetReader.Builder<Group> builder = - ParquetReader.builder(new GroupReadSupport(), filePath) - .withConf(configuration); + new GroupParquetReaderBuilder( + FileSystemInputFile.fromPath( + hadoopFileSystemProxy.getFileSystem(), + filePath)); + // withConf() rebuilds the read options, so it must precede + // withFileRange() below. + builder.withConf(hadoopFileSystemProxy.getConfiguration()); if (useSplitRange) { long start = split.getStart(); long end = start + split.getLength(); @@ -750,9 +767,10 @@ public class ParquetReadStrategy extends AbstractReadStrategy { try (ParquetFileReader reader = hadoopFileSystemProxy.doWithHadoopAuth( ((configuration, userGroupInformation) -> { - HadoopInputFile hadoopInputFile = - HadoopInputFile.fromPath(new Path(path), configuration); - return ParquetFileReader.open(hadoopInputFile); + return ParquetFileReader.open( + FileSystemInputFile.fromPath( + hadoopFileSystemProxy.getFileSystem(), new Path(path)), + HadoopReadOptions.builder(configuration).build()); }))) { metadata = reader.getFooter(); } catch (IOException e) { @@ -997,4 +1015,22 @@ public class ParquetReadStrategy extends AbstractReadStrategy { return fieldIndex == -1 ? null : configRowType.getFieldType(fieldIndex); } + + /** + * Binds the native (Group) Parquet reader to an already-open {@link + * org.apache.hadoop.fs.FileSystem}. {@code ParquetReader.Builder(InputFile)} is protected and + * there is no public {@code builder(ReadSupport, InputFile)}, so this mirrors what {@code + * AvroParquetReader.Builder} does for the Avro path. + */ + private static final class GroupParquetReaderBuilder extends ParquetReader.Builder<Group> { + + private GroupParquetReaderBuilder(InputFile file) { + super(file); + } + + @Override + protected ReadSupport<Group> getReadSupport() { + return new GroupReadSupport(); + } + } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/ParquetFileSplitStrategy.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/ParquetFileSplitStrategy.java index b89bddfa53f..5e4fd49b73a 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/ParquetFileSplitStrategy.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/ParquetFileSplitStrategy.java @@ -20,10 +20,12 @@ package org.apache.seatunnel.connectors.seatunnel.file.source.split; import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode; +import org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileSystemInputFile; import org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.parquet.HadoopReadOptions; import org.apache.parquet.hadoop.ParquetFileReader; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.util.HadoopInputFile; @@ -150,7 +152,9 @@ public class ParquetFileSplitStrategy implements FileSplitStrategy, Closeable { (configuration, userGroupInformation) -> { try (ParquetFileReader reader = ParquetFileReader.open( - HadoopInputFile.fromPath(path, configuration))) { + FileSystemInputFile.fromPath( + hadoopFileSystemProxy.getFileSystem(), path), + HadoopReadOptions.builder(configuration).build())) { return reader.getFooter().getBlocks(); } }); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemParquetFileTest.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemParquetFileTest.java new file mode 100644 index 00000000000..536a57c5743 --- /dev/null +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileSystemParquetFileTest.java @@ -0,0 +1,127 @@ +/* + * 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.seatunnel.connectors.seatunnel.file.hadoop; + +import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.io.PositionOutputStream; +import org.apache.parquet.io.SeekableInputStream; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledOnOs; +import org.junit.jupiter.api.condition.OS; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.charset.StandardCharsets; + +/** Verifies the {@link FileSystem}-bound Parquet {@code OutputFile}/{@code InputFile} adapters. */ +class FileSystemParquetFileTest { + + @TempDir private java.nio.file.Path tempDir; + + @Test + @DisabledOnOs( + value = OS.WINDOWS, + disabledReason = + "Hadoop local filesystem writes require winutils/HADOOP_HOME on Windows") + void testWriteThenReadThroughTheAdapters() throws Exception { + byte[] payload = "seatunnel".getBytes(StandardCharsets.UTF_8); + try (HadoopFileSystemProxy proxy = new HadoopFileSystemProxy(new HadoopConf("file:///"))) { + FileSystem fs = proxy.getFileSystem(); + Path path = new Path(tempDir.resolve("adapter.bin").toString()); + + try (PositionOutputStream out = new FileSystemOutputFile(fs, path).create(0L)) { + out.write(payload); + } + + FileSystemInputFile inputFile = FileSystemInputFile.fromPath(fs, path); + Assertions.assertEquals(payload.length, inputFile.getLength()); + byte[] read = new byte[payload.length]; + try (SeekableInputStream in = inputFile.newStream()) { + in.readFully(read); + } + Assertions.assertArrayEquals(payload, read); + } + } + + @Test + @DisabledOnOs( + value = OS.WINDOWS, + disabledReason = + "Hadoop local filesystem writes require winutils/HADOOP_HOME on Windows") + void testCreateOrOverwriteReplacesExistingFile() throws Exception { + try (HadoopFileSystemProxy proxy = new HadoopFileSystemProxy(new HadoopConf("file:///"))) { + FileSystem fs = proxy.getFileSystem(); + Path path = new Path(tempDir.resolve("overwrite.bin").toString()); + FileSystemOutputFile outputFile = new FileSystemOutputFile(fs, path); + + try (PositionOutputStream out = outputFile.create(0L)) { + out.write("first-write".getBytes(StandardCharsets.UTF_8)); + } + try (PositionOutputStream out = outputFile.createOrOverwrite(0L)) { + out.write("second".getBytes(StandardCharsets.UTF_8)); + } + + Assertions.assertEquals( + "second".length(), FileSystemInputFile.fromPath(fs, path).getLength()); + } + } + + @Test + @DisabledOnOs( + value = OS.WINDOWS, + disabledReason = + "Hadoop local filesystem writes require winutils/HADOOP_HOME on Windows") + void testProxyOwnsASingleFileSystemSharedByManyFiles() throws Exception { + try (HadoopFileSystemProxy proxy = new HadoopFileSystemProxy(new HadoopConf("file:///"))) { + FileSystem fs = proxy.getFileSystem(); + // The proxy is the single owner: repeated lookups must not allocate a new + // FileSystem. Per-file resolution (HadoopOutputFile/HadoopInputFile.fromPath, or + // ParquetReader.builder(ReadSupport, Path)) allocated one per file and leaked its + // metrics registration, which is what these adapters exist to avoid. + Assertions.assertSame(fs, proxy.getFileSystem()); + + // Many files, one FileSystem instance. + for (int i = 0; i < 3; i++) { + Path path = new Path(tempDir.resolve("shared-" + i + ".bin").toString()); + try (PositionOutputStream out = new FileSystemOutputFile(fs, path).create(0L)) { + out.write(("row" + i).getBytes(StandardCharsets.UTF_8)); + } + Assertions.assertEquals(4, FileSystemInputFile.fromPath(fs, path).getLength()); + Assertions.assertSame(fs, proxy.getFileSystem()); + } + } + } + + @Test + void testBlockSizeMatchesTheUnderlyingFileSystem() throws Exception { + try (HadoopFileSystemProxy proxy = new HadoopFileSystemProxy(new HadoopConf("file:///"))) { + FileSystem fs = proxy.getFileSystem(); + Path path = new Path(tempDir.resolve("blocksize.bin").toString()); + FileSystemOutputFile outputFile = new FileSystemOutputFile(fs, path); + + // Mirrors HadoopOutputFile: only hdfs/webhdfs/viewfs report block-size support. + Assertions.assertFalse(outputFile.supportsBlockSize()); + Assertions.assertEquals(fs.getDefaultBlockSize(path), outputFile.defaultBlockSize()); + Assertions.assertEquals(fs.makeQualified(path).toString(), outputFile.getPath()); + } + } +}
