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 2546a0620f [format] Close the row-file input stream when the footer or
index fails to parse (#9258)
2546a0620f is described below
commit 2546a0620fa6ee80064de1dc651fea3f9871d44a
Author: ZIHAN DAI <[email protected]>
AuthorDate: Thu Aug 20 12:22:48 2026 +1000
[format] Close the row-file input stream when the footer or index fails to
parse (#9258)
---
.../paimon/format/row/RowFormatReaderFactory.java | 44 +++---
.../format/row/RowFormatReaderFactoryLeakTest.java | 150 +++++++++++++++++++++
2 files changed, 176 insertions(+), 18 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
index 1cab7626be..bda5a428bb 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/row/RowFormatReaderFactory.java
@@ -53,26 +53,34 @@ public class RowFormatReaderFactory implements
FormatReaderFactory {
SeekableInputStream in = fileIO.newInputStream(path);
- int tailSize = (int) Math.min(TAIL_PREFETCH_SIZE, fileSize);
- long tailOffset = fileSize - tailSize;
- in.seek(tailOffset);
- byte[] tailBuf = new byte[tailSize];
- IOUtils.readFully(in, tailBuf);
+ // Ownership of the stream passes to RowFormatReader only on the last
line. Everything
+ // before it parses lengths and offsets taken from the file itself, so
a truncated or
+ // corrupt file can throw anywhere in between and would otherwise leak
the stream.
+ try {
+ int tailSize = (int) Math.min(TAIL_PREFETCH_SIZE, fileSize);
+ long tailOffset = fileSize - tailSize;
+ in.seek(tailOffset);
+ byte[] tailBuf = new byte[tailSize];
+ IOUtils.readFully(in, tailBuf);
- RowFileFooter footer =
- RowFileFooter.readFrom(tailBuf, tailSize -
RowFileFooter.FOOTER_SIZE);
+ RowFileFooter footer =
+ RowFileFooter.readFrom(tailBuf, tailSize -
RowFileFooter.FOOTER_SIZE);
- RowBlockIndex blockIndex;
- if (footer.indexOffset >= tailOffset) {
- int indexOffsetInBuf = (int) (footer.indexOffset - tailOffset);
- byte[] indexData = new byte[footer.indexLength];
- System.arraycopy(tailBuf, indexOffsetInBuf, indexData, 0,
footer.indexLength);
- blockIndex = RowBlockIndex.readFrom(indexData);
- } else {
- blockIndex = RowBlockIndex.readFrom(in, footer.indexOffset,
footer.indexLength);
- }
+ RowBlockIndex blockIndex;
+ if (footer.indexOffset >= tailOffset) {
+ int indexOffsetInBuf = (int) (footer.indexOffset - tailOffset);
+ byte[] indexData = new byte[footer.indexLength];
+ System.arraycopy(tailBuf, indexOffsetInBuf, indexData, 0,
footer.indexLength);
+ blockIndex = RowBlockIndex.readFrom(indexData);
+ } else {
+ blockIndex = RowBlockIndex.readFrom(in, footer.indexOffset,
footer.indexLength);
+ }
- return new RowFormatReader(
- in, path, footer, blockIndex, rowType, projection,
context.selection());
+ return new RowFormatReader(
+ in, path, footer, blockIndex, rowType, projection,
context.selection());
+ } catch (Throwable t) {
+ IOUtils.closeQuietly(in);
+ throw t;
+ }
}
}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/row/RowFormatReaderFactoryLeakTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/row/RowFormatReaderFactoryLeakTest.java
new file mode 100644
index 0000000000..a6f447034a
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/row/RowFormatReaderFactoryLeakTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.row;
+
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatReaderFactory;
+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.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * {@link RowFormatReaderFactory#createReader} opens the input stream before
parsing a footer and a
+ * block index whose offsets and lengths come from the file itself. Ownership
only passes to {@link
+ * RowFormatReader} on the last line, so a corrupt file must not leave the
stream open.
+ *
+ * <p>This matters because {@code DataFileRecordReader.createReader} treats an
{@code IOException}
+ * or {@code RuntimeException} from here as an ignorable corrupt file when
{@code
+ * ignore-corrupt-files} is set: it logs, returns null, and the scan moves to
the next file. One
+ * leaked descriptor per corrupt file, for as long as the scan runs.
+ */
+class RowFormatReaderFactoryLeakTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void aCorruptFooterDoesNotLeakTheInputStream() throws Exception {
+ Path path = new Path(new Path(tempDir.toString()), "corrupt.row");
+ CountingFileIO fileIO = new CountingFileIO();
+
+ // A file long enough to be read as a footer, but with none of the
expected content.
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+ out.write(new byte[128]);
+ }
+
+ RowType rowType = RowType.of(DataTypes.INT());
+ FormatReaderFactory readerFactory =
+ FileFormat.fromIdentifier("row", new Options())
+ .createReaderFactory(rowType, rowType, new
ArrayList<>());
+
+ assertThatThrownBy(
+ () ->
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO,
+ path,
+ fileIO.getFileSize(path),
+ null,
+ null)))
+ .hasMessageContaining("Invalid row file magic");
+
+ assertThat(fileIO.openInputStreams()).isZero();
+ }
+
+ @Test
+ void anEmptyFileDoesNotLeakTheInputStream() throws Exception {
+ Path path = new Path(new Path(tempDir.toString()), "empty.row");
+ CountingFileIO fileIO = new CountingFileIO();
+ try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
+ // nothing: an aborted write leaves a zero-length object behind
+ }
+
+ RowType rowType = RowType.of(DataTypes.INT());
+ FormatReaderFactory readerFactory =
+ FileFormat.fromIdentifier("row", new Options())
+ .createReaderFactory(rowType, rowType, new
ArrayList<>());
+
+ assertThatThrownBy(
+ () ->
+ readerFactory.createReader(
+ new FormatReaderContext(
+ fileIO, path,
fileIO.getFileSize(path), null, null)));
+
+ assertThat(fileIO.openInputStreams()).isZero();
+ }
+
+ /** Counts input streams that have been opened and not yet closed. */
+ private static class CountingFileIO extends LocalFileIO {
+
+ private final AtomicInteger open = new AtomicInteger();
+
+ int openInputStreams() {
+ return open.get();
+ }
+
+ @Override
+ public SeekableInputStream newInputStream(Path f) throws IOException {
+ SeekableInputStream delegate = super.newInputStream(f);
+ open.incrementAndGet();
+ return new SeekableInputStream() {
+
+ @Override
+ public void seek(long desired) throws IOException {
+ delegate.seek(desired);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return delegate.getPos();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return delegate.read();
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException
{
+ return delegate.read(b, off, len);
+ }
+
+ @Override
+ public void close() throws IOException {
+ open.decrementAndGet();
+ delegate.close();
+ }
+ };
+ }
+ }
+}
