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 4dba8f84f1 [format] Validate blob file footer and index metadata
(#9223)
4dba8f84f1 is described below
commit 4dba8f84f1ec87bfecfb308d3afb927acab10af0
Author: Wenchao Wu <[email protected]>
AuthorDate: Sat Aug 15 20:15:18 2026 +0800
[format] Validate blob file footer and index metadata (#9223)
---
.../apache/paimon/format/blob/BlobFileMeta.java | 72 ++++++-
.../paimon/format/blob/BlobFileMetaTest.java | 209 +++++++++++++++++++++
2 files changed, 273 insertions(+), 8 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileMeta.java
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileMeta.java
index 24fe3a764d..7de594fc0a 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileMeta.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileMeta.java
@@ -32,46 +32,102 @@ import java.util.Iterator;
/** Metadata of blob file. */
public class BlobFileMeta {
+ private static final int FILE_FOOTER_LENGTH = Integer.BYTES + Byte.BYTES;
+ private static final int MIN_RECORD_LENGTH = Integer.BYTES + Long.BYTES +
Integer.BYTES;
+
private final long[] blobLengths;
private final long[] blobOffsets;
private final @Nullable int[] returnedPositions;
public BlobFileMeta(SeekableInputStream in, long fileSize, @Nullable
RoaringBitmap32 selection)
throws IOException {
- in.seek(fileSize - 5);
- byte[] header = new byte[5];
+ if (fileSize < FILE_FOOTER_LENGTH) {
+ throw new IOException(
+ String.format(
+ "Corrupt blob file: file size %s is smaller than
footer size %s.",
+ fileSize, FILE_FOOTER_LENGTH));
+ }
+
+ in.seek(fileSize - FILE_FOOTER_LENGTH);
+ byte[] header = new byte[FILE_FOOTER_LENGTH];
IOUtils.readFully(in, header);
- byte version = header[4];
+ byte version = header[Integer.BYTES];
if (version != BlobFormatWriter.VERSION) {
throw new IOException("Unsupported version: " + version);
}
int indexLength = BytesUtils.getInt(header, 0);
+ long maximumIndexLength = fileSize - FILE_FOOTER_LENGTH;
+ if (indexLength < 0 || indexLength > maximumIndexLength) {
+ throw new IOException(
+ String.format(
+ "Corrupt blob file: invalid index length %s for
file size %s.",
+ indexLength, fileSize));
+ }
- in.seek(fileSize - 5 - indexLength);
+ long indexStart = maximumIndexLength - indexLength;
+ in.seek(indexStart);
byte[] indexBytes = new byte[indexLength];
IOUtils.readFully(in, indexBytes);
- long[] blobLengths = DeltaVarintCompressor.decompress(indexBytes);
+ long[] blobLengths;
+ try {
+ blobLengths = DeltaVarintCompressor.decompress(indexBytes);
+ } catch (RuntimeException e) {
+ throw new IOException("Corrupt blob file: invalid index.", e);
+ }
long[] blobOffsets = new long[blobLengths.length];
long offset = 0;
for (int i = 0; i < blobLengths.length; i++) {
- if (blobLengths[i] < 0) {
+ long blobLength = blobLengths[i];
+ if (blobLength == BlobFormatWriter.NULL_LENGTH
+ || blobLength == BlobFormatWriter.PLACE_HOLDER_LENGTH) {
blobOffsets[i] = -1;
} else {
+ if (blobLength < MIN_RECORD_LENGTH) {
+ throw new IOException(
+ String.format(
+ "Corrupt blob file: invalid record length
%s at position %s.",
+ blobLength, i));
+ }
+ if (blobLength > indexStart - offset) {
+ throw new IOException(
+ String.format(
+ "Corrupt blob file: record length %s at
position %s exceeds the data region.",
+ blobLength, i));
+ }
blobOffsets[i] = offset;
- offset += blobLengths[i];
+ offset += blobLength;
}
}
+ if (offset != indexStart) {
+ throw new IOException(
+ String.format(
+ "Corrupt blob file: indexed records use %s bytes,
but data region contains %s bytes.",
+ offset, indexStart));
+ }
int[] returnedPositions = null;
if (selection != null) {
- int cardinality = (int) selection.getCardinality();
+ long selectionCardinality = selection.getCardinality();
+ if (selectionCardinality > blobLengths.length) {
+ throw new IOException(
+ String.format(
+ "Invalid blob selection: cardinality %s
exceeds record count %s.",
+ selectionCardinality, blobLengths.length));
+ }
+ int cardinality = (int) selectionCardinality;
returnedPositions = new int[cardinality];
long[] newLengths = new long[cardinality];
long[] newOffsets = new long[cardinality];
Iterator<Integer> iterator = selection.iterator();
for (int i = 0; i < cardinality; i++) {
Integer next = iterator.next();
+ if (next < 0 || next >= blobLengths.length) {
+ throw new IOException(
+ String.format(
+ "Invalid blob selection: position %s is
outside record count %s.",
+ next, blobLengths.length));
+ }
newLengths[i] = blobLengths[next];
newOffsets[i] = blobOffsets[next];
returnedPositions[i] = next;
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileMetaTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileMetaTest.java
new file mode 100644
index 0000000000..3b3f7a1cc8
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileMetaTest.java
@@ -0,0 +1,209 @@
+/*
+ * 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.blob;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.utils.DeltaVarintCompressor;
+import org.apache.paimon.utils.RoaringBitmap32;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.UUID;
+
+import static org.apache.paimon.utils.StreamUtils.intToLittleEndian;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link BlobFileMeta}. */
+public class BlobFileMetaTest {
+
+ private static final int RECORD_LENGTH = Integer.BYTES + Long.BYTES +
Integer.BYTES;
+
+ @TempDir java.nio.file.Path tempPath;
+
+ private FileIO fileIO;
+ private Path file;
+
+ @BeforeEach
+ public void beforeEach() {
+ fileIO = LocalFileIO.create();
+ file = new Path(new Path(tempPath.toUri()),
UUID.randomUUID().toString());
+ }
+
+ @Test
+ public void testRejectsFileSmallerThanFooter() throws IOException {
+ for (int size = 0; size < Integer.BYTES + Byte.BYTES; size++) {
+ writeFile(new byte[size]);
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("smaller than footer");
+ }
+ }
+
+ @Test
+ public void testRejectsInvalidIndexLength() throws IOException {
+ writeFile(blobFileBytes(0, new byte[0], -1, BlobFormatWriter.VERSION));
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid index length -1");
+
+ writeFile(blobFileBytes(0, new byte[0], 1, BlobFormatWriter.VERSION));
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid index length 1");
+ }
+
+ @Test
+ public void testRejectsUnsupportedVersionAndMalformedIndex() throws
IOException {
+ writeFile(blobFileBytes(0, new byte[0], 0, (byte)
(BlobFormatWriter.VERSION + 1)));
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported version");
+
+ byte[] malformedIndex = new byte[] {(byte) 0x80};
+ writeFile(
+ blobFileBytes(0, malformedIndex, malformedIndex.length,
BlobFormatWriter.VERSION));
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid index")
+ .hasCauseInstanceOf(RuntimeException.class);
+ }
+
+ @Test
+ public void testRejectsInvalidRecordLength() throws IOException {
+ assertInvalidRecordLength(-3, 0);
+ assertInvalidRecordLength(0, 0);
+ assertInvalidRecordLength(RECORD_LENGTH - 1, RECORD_LENGTH - 1);
+ }
+
+ @Test
+ public void testRejectsRecordLengthSumMismatch() throws IOException {
+ writeFile(blobFileBytes(RECORD_LENGTH - 1, RECORD_LENGTH));
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("exceeds the data region");
+
+ writeFile(blobFileBytes(RECORD_LENGTH + 1, RECORD_LENGTH));
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("indexed records use")
+ .hasMessageContaining("data region contains");
+ }
+
+ @Test
+ public void testAcceptsEmptyFileAndKnownSentinels() throws IOException {
+ writeFile(blobFileBytes(0));
+ assertThat(readMeta(null).recordNumber()).isZero();
+
+ writeFile(
+ blobFileBytes(
+ RECORD_LENGTH,
+ BlobFormatWriter.NULL_LENGTH,
+ RECORD_LENGTH,
+ BlobFormatWriter.PLACE_HOLDER_LENGTH));
+ BlobFileMeta meta = readMeta(null);
+ assertThat(meta.recordNumber()).isEqualTo(3);
+ assertThat(meta.isNull(0)).isTrue();
+ assertThat(meta.blobOffset(0)).isEqualTo(-1);
+ assertThat(meta.blobOffset(1)).isZero();
+ assertThat(meta.isPlaceHolder(2)).isTrue();
+ assertThat(meta.blobOffset(2)).isEqualTo(-1);
+ }
+
+ @Test
+ public void testValidatesSelectionPositions() throws IOException {
+ writeFile(blobFileBytes(RECORD_LENGTH, RECORD_LENGTH));
+
+ RoaringBitmap32 tooMany = new RoaringBitmap32();
+ tooMany.add(0);
+ tooMany.add(1);
+ assertThatThrownBy(() -> readMeta(tooMany))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("cardinality 2 exceeds record count 1");
+
+ RoaringBitmap32 outOfRange = new RoaringBitmap32();
+ outOfRange.add(1);
+ assertThatThrownBy(() -> readMeta(outOfRange))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("position 1 is outside record count 1");
+ }
+
+ @Test
+ public void testNonContiguousAndEmptySelection() throws IOException {
+ writeFile(blobFileBytes(RECORD_LENGTH * 3, RECORD_LENGTH,
RECORD_LENGTH, RECORD_LENGTH));
+
+ RoaringBitmap32 selection = new RoaringBitmap32();
+ selection.add(0);
+ selection.add(2);
+ BlobFileMeta selected = readMeta(selection);
+ assertThat(selected.recordNumber()).isEqualTo(2);
+ assertThat(selected.blobOffset(0)).isZero();
+ assertThat(selected.blobOffset(1)).isEqualTo(RECORD_LENGTH * 2L);
+ assertThat(selected.returnedPosition(1)).isZero();
+ assertThat(selected.returnedPosition(2)).isEqualTo(2);
+
+ BlobFileMeta empty = readMeta(new RoaringBitmap32());
+ assertThat(empty.recordNumber()).isZero();
+ }
+
+ private void assertInvalidRecordLength(long recordLength, int dataLength)
throws IOException {
+ writeFile(blobFileBytes(dataLength, recordLength));
+ assertThatThrownBy(() -> readMeta(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("invalid record length " + recordLength);
+ }
+
+ private BlobFileMeta readMeta(RoaringBitmap32 selection) throws
IOException {
+ try (SeekableInputStream in = fileIO.newInputStream(file)) {
+ return new BlobFileMeta(in, fileIO.getFileSize(file), selection);
+ }
+ }
+
+ private void writeFile(byte[] bytes) throws IOException {
+ Files.write(Paths.get(file.toUri()), bytes);
+ }
+
+ private static byte[] blobFileBytes(int dataLength, long... recordLengths)
{
+ byte[] index = DeltaVarintCompressor.compress(recordLengths);
+ return blobFileBytes(dataLength, index, index.length,
BlobFormatWriter.VERSION);
+ }
+
+ private static byte[] blobFileBytes(
+ int dataLength, byte[] index, int declaredIndexLength, byte
version) {
+ byte[] bytes = new byte[dataLength + index.length + Integer.BYTES +
Byte.BYTES];
+ System.arraycopy(index, 0, bytes, dataLength, index.length);
+ byte[] indexLengthBytes = intToLittleEndian(declaredIndexLength);
+ System.arraycopy(
+ indexLengthBytes,
+ 0,
+ bytes,
+ bytes.length - Integer.BYTES - Byte.BYTES,
+ indexLengthBytes.length);
+ bytes[bytes.length - 1] = version;
+ return bytes;
+ }
+}
