This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5a702aca45 [format] Close the ParquetFileReader when post-construction
setup fails (#9576)
5a702aca45 is described below
commit 5a702aca45cf747834abbd199bfac18f06d85900
Author: YangJie <[email protected]>
AuthorDate: Fri Sep 4 03:24:12 2026 -0400
[format] Close the ParquetFileReader when post-construction setup fails
(#9576)
---
.../format/parquet/ParquetReaderFactory.java | 108 ++++++++--------
.../parquet/ParquetReaderFactoryLeakTest.java | 139 +++++++++++++++++++++
2 files changed, 198 insertions(+), 49 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
index 2874849bb5..ab7d277a62 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
@@ -151,57 +151,67 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
throw t;
}
- ShreddingReadPlan readPlan =
- ShreddingReadPlanFactories.createReadPlan(
- readType,
- readFieldMetadata(reader),
- fileSchema,
- shreddingReadPlanFactories(readType));
- DataField[] physicalReadFields =
readFields(readPlan.physicalRowType());
- RequestedSchema requestedSchema =
- readPlan.isIdentity()
- ? getOrCreateRequestedSchema(fileSchema)
- : createRequestedSchema(fileSchema,
physicalReadFields);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Create reader of the parquet file {}, the fileSchema is
{}, the requestedSchema is {}.",
- context.filePath(),
- fileSchema,
- requestedSchema.messageType);
- }
-
- int configuredBatchSize = computeBatchSize(reader,
requestedSchema.messageType);
- Preconditions.checkArgument(
- configuredBatchSize > 0,
- "Parquet read batch size should be positive: %s",
- configuredBatchSize);
- ReadBatchSizer readBatchSizer = context.readBatchSizer();
- int initialBatchSize =
- readBatchSizer == null
- ? configuredBatchSize
- :
readBatchSizer.batchSize().orElse(configuredBatchSize);
- reader.setRequestedSchema(requestedSchema.messageType);
- WritableColumnVector[] writableVectors =
- createWritableVectors(initialBatchSize, physicalReadFields);
- IntFunction<WritableColumnVector[]> vectorFactory =
- size -> createWritableVectors(size, physicalReadFields);
-
- VectorizedParquetRecordReader parquetReader =
- new VectorizedParquetRecordReader(
+ // The reader owns the open stream from here, so close it if the setup
below fails.
+ try {
+ ShreddingReadPlan readPlan =
+ ShreddingReadPlanFactories.createReadPlan(
+ readType,
+ readFieldMetadata(reader),
+ fileSchema,
+ shreddingReadPlanFactories(readType));
+ DataField[] physicalReadFields =
readFields(readPlan.physicalRowType());
+ RequestedSchema requestedSchema =
+ readPlan.isIdentity()
+ ? getOrCreateRequestedSchema(fileSchema)
+ : createRequestedSchema(fileSchema,
physicalReadFields);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Create reader of the parquet file {}, the fileSchema
is {}, the requestedSchema is {}.",
context.filePath(),
- reader,
fileSchema,
- requestedSchema.fields,
- writableVectors,
- initialBatchSize,
- configuredBatchSize,
- context.fileIO(),
- readBatchSizer,
- vectorFactory);
- return readPlan.isIdentity()
- ? parquetReader
- : new ShreddingFormatReader(parquetReader, readPlan);
+ requestedSchema.messageType);
+ }
+
+ int configuredBatchSize = computeBatchSize(reader,
requestedSchema.messageType);
+ Preconditions.checkArgument(
+ configuredBatchSize > 0,
+ "Parquet read batch size should be positive: %s",
+ configuredBatchSize);
+ ReadBatchSizer readBatchSizer = context.readBatchSizer();
+ int initialBatchSize =
+ readBatchSizer == null
+ ? configuredBatchSize
+ :
readBatchSizer.batchSize().orElse(configuredBatchSize);
+ reader.setRequestedSchema(requestedSchema.messageType);
+ WritableColumnVector[] writableVectors =
+ createWritableVectors(initialBatchSize,
physicalReadFields);
+ IntFunction<WritableColumnVector[]> vectorFactory =
+ size -> createWritableVectors(size, physicalReadFields);
+
+ VectorizedParquetRecordReader parquetReader =
+ new VectorizedParquetRecordReader(
+ context.filePath(),
+ reader,
+ fileSchema,
+ requestedSchema.fields,
+ writableVectors,
+ initialBatchSize,
+ configuredBatchSize,
+ context.fileIO(),
+ readBatchSizer,
+ vectorFactory);
+ return readPlan.isIdentity()
+ ? parquetReader
+ : new ShreddingFormatReader(parquetReader, readPlan);
+ } catch (Throwable t) {
+ try {
+ reader.close();
+ } catch (Throwable closeFailure) {
+ t.addSuppressed(closeFailure);
+ }
+ throw t;
+ }
}
private RequestedSchema getOrCreateRequestedSchema(MessageType fileSchema)
{
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReaderFactoryLeakTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReaderFactoryLeakTest.java
new file mode 100644
index 0000000000..6898203ba6
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReaderFactoryLeakTest.java
@@ -0,0 +1,139 @@
+/*
+ * 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.paimon.format.parquet;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.schema.MessageType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests that {@link ParquetReaderFactory#createReader} closes the file
exactly once. */
+class ParquetReaderFactoryLeakTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private final RowType rowType = RowType.of(DataTypes.INT());
+
+ @Test
+ void testSetupFailureAfterReaderExistsClosesFile() throws IOException {
+ Path file = writeParquet();
+ CountingFileIO fileIO = new CountingFileIO();
+
+ // computeBatchSize is called after the ParquetFileReader has been
constructed, so a
+ // zero batch size fails the check inside the region this test is
about.
+ ParquetReaderFactory factory =
+ new ParquetReaderFactory(new Options(), rowType, 1024, null) {
+ @Override
+ protected int computeBatchSize(
+ ParquetFileReader reader, MessageType
requestedSchema) {
+ return 0;
+ }
+ };
+
+ assertThatThrownBy(() -> factory.createReader(context(fileIO, file)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Parquet read batch size should be
positive");
+ assertThat(fileIO.closed).hasValue(1);
+ }
+
+ @Test
+ void testSuccessfulReadClosesFileOnce() throws IOException {
+ Path file = writeParquet();
+ CountingFileIO fileIO = new CountingFileIO();
+
+ ParquetReaderFactory factory = new ParquetReaderFactory(new Options(),
rowType, 1024, null);
+ try (FileRecordReader<InternalRow> reader =
factory.createReader(context(fileIO, file))) {
+ assertThat(reader.readBatch().next().getInt(0)).isEqualTo(1);
+ }
+ assertThat(fileIO.closed).hasValue(1);
+ }
+
+ private FormatReaderContext context(CountingFileIO fileIO, Path file)
throws IOException {
+ return new FormatReaderContext(fileIO, file, fileIO.getFileSize(file),
null, null);
+ }
+
+ private Path writeParquet() throws IOException {
+ Path file = new Path(tempDir.toUri().toString(), "a.parquet");
+ FileFormat format = FileFormat.fromIdentifier("parquet", new
Options());
+ try (PositionOutputStream out =
LocalFileIO.create().newOutputStream(file, false)) {
+ FormatWriter writer =
format.createWriterFactory(rowType).create(out, "zstd");
+ writer.addElement(GenericRow.of(1));
+ writer.close();
+ }
+ return file;
+ }
+
+ private static class CountingFileIO extends LocalFileIO {
+
+ private final AtomicInteger closed = new AtomicInteger();
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws
IOException {
+ SeekableInputStream inner = super.newInputStream(path);
+ return new SeekableInputStream() {
+
+ @Override
+ public void seek(long desired) throws IOException {
+ inner.seek(desired);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return inner.getPos();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return inner.read();
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException
{
+ return inner.read(b, off, len);
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed.incrementAndGet();
+ inner.close();
+ }
+ };
+ }
+ }
+}
