You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/10/15 03:57:07 UTC
(paimon) branch master updated: [file-index] Support stream file index containers (#10101)
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 22ea78d2b6 [file-index] Support stream file index containers (#10101)
22ea78d2b6 is described below
commit 22ea78d2b6457f939233f351b2d48c55e2b237bd
Author: Zhang Jiawei <30...@users.noreply.github.com>
AuthorDate: Thu Oct 15 11:57:00 2026 +0800
[file-index] Support stream file index containers (#10101)
---
docs/docs/concepts/spec/fileindex.mdx | 69 +++-
docs/generated/core_configuration.html | 8 +-
.../main/java/org/apache/paimon/CoreOptions.java | 27 +-
.../apache/paimon/fileindex/FileIndexOptions.java | 6 +
.../apache/paimon/fileindex/FileIndexFormat.java | 382 +++++++++------------
.../paimon/fileindex/FileIndexFormatUtils.java | 148 ++++++++
.../apache/paimon/fileindex/FileIndexFormatV1.java | 189 ++++++++++
.../apache/paimon/fileindex/FileIndexFormatV2.java | 158 +++++++++
.../paimon/fileindex/FileIndexPredicate.java | 43 ++-
.../apache/paimon/fileindex/FileIndexWriter.java | 13 +-
.../org/apache/paimon/fileindex/FileIndexer.java | 15 +
.../paimon/fileindex/bitmap/BitmapFileIndex.java | 10 +-
.../bloomfilter/BloomFilterFileIndex.java | 16 +-
.../bsi/BitSliceIndexBitmapFileIndex.java | 5 +
.../fileindex/rangebitmap/BitSliceIndexBitmap.java | 4 +-
.../paimon/fileindex/rangebitmap/RangeBitmap.java | 6 +-
.../rangebitmap/RangeBitmapFileIndex.java | 7 +-
.../dictionary/chunked/ChunkedDictionary.java | 4 +-
.../dictionary/chunked/FixedLengthChunk.java | 4 +-
.../rangebitmap/dictionary/chunked/KeyFactory.java | 6 +-
.../dictionary/chunked/VariableLengthChunk.java | 4 +-
.../FileIndexFormatCompatibilityTest.java | 309 +++++++++++++++++
.../fileindex/FileIndexFormatFormatTest.java | 322 +++++++++++++++--
.../fileindex/FileIndexPredicateCloseTest.java | 68 +++-
.../paimon/fileindex/FileIndexPredicateTest.java | 17 +-
.../fileindex/FileIndexerCompatibilityTest.java | 187 ++++++++++
.../apache/paimon/index/FileIndexProcessor.java | 182 +++++-----
.../org/apache/paimon/io/DataFileIndexWriter.java | 114 +++---
.../paimon/io/SpillableIndexOutputStream.java | 113 ++++++
.../paimon/table/system/FileIndexesTable.java | 6 +-
.../java/org/apache/paimon/CoreOptionsTest.java | 18 +
...DedicatedFormatRollingFileWriterVectorTest.java | 4 +-
.../paimon/index/FileIndexProcessorTest.java | 79 ++++-
.../apache/paimon/io/DataFileIndexWriterTest.java | 138 +++++++-
.../apache/paimon/io/FileIndexEvaluatorTest.java | 13 +-
.../org/apache/paimon/io/SparseFileIndexIO.java | 84 +++++
.../paimon/io/StreamingTestFileIndexFactory.java | 90 +++++
.../paimon/table/system/FileIndexesTableTest.java | 30 +-
.../org.apache.paimon.fileindex.FileIndexerFactory | 16 +
.../flink/action/RewriteFileIndexActionITCase.java | 4 +-
.../procedure/RewriteFileIndexProcedureITCase.java | 20 +-
.../apache/paimon/spark/SparkFileIndexITCase.java | 8 +-
42 files changed, 2473 insertions(+), 473 deletions(-)
diff --git a/docs/docs/concepts/spec/fileindex.mdx b/docs/docs/concepts/spec/fileindex.mdx
index e782df6f47..6fde4d722b 100644
--- a/docs/docs/concepts/spec/fileindex.mdx
+++ b/docs/docs/concepts/spec/fileindex.mdx
@@ -37,7 +37,7 @@ deletion vectors, see [Table Index](./tableindex).
| Encoding | Section |
| --- | --- |
-| Shared header and column offsets | [Index File](#index-file) |
+| File index container | [Index File](#index-file) |
| Bloom filter | [BloomFilter](#index-bloomfilter) |
| Bitmap | [Bitmap](#index-bitmap) |
| Range bitmap | [Range Bitmap](#index-range-bitmap) |
@@ -45,7 +45,13 @@ deletion vectors, see [Table Index](./tableindex).
## Index File
-File index file format. Put all column and offset in the header.
+The container version is selected by `file-index.format.version` (default: `1`). Readers support
+both versions. Enable version `2` after upgrading the readers of the table. The container
+version is independent of each index type's payload version.
+
+### V1
+
+V1 puts the index entries in the header before the payloads.
<pre>
______________________________________ _____________________
@@ -95,6 +101,65 @@ redundant bytes: var bytes (for compatibility with later versio
BODY: column index bytes + column index bytes + column index bytes + .......
</pre>
+### V2
+
+V2 writes the payloads first, followed by the footer containing the index entries and a fixed
+12-byte trailer.
+All integers are signed and big-endian; strings use Java `writeUTF` encoding.
+
+<pre>
+ ______________________________________ _____________________
+| magic | version | PREFIX
+|--------------------------------------| ---------------------
+| BODY |
+| BODY | BODY
+| BODY |
+|--------------------------------------| ---------------------
+| column number |
+|--------------------------------------|
+| column 1 | index number |
+|--------------------------------------|
+| index name 1 | start pos | length |
+|--------------------------------------|
+| index name 2 | start pos | length |
+|--------------------------------------|
+| column 2 | index number |
+|--------------------------------------|
+| index name 1 | start pos | length |
+|--------------------------------------|
+| index name 2 | start pos | length |
+|--------------------------------------|
+| ... | FOOTER
+|--------------------------------------|
+| ... |
+|--------------------------------------| ---------------------
+| footer length | tail magic | TRAILER
+|______________________________________| _____________________
+
+magic: 8 bytes long, value is 1493475289347502L, BIG_ENDIAN
+version: 4 bytes int, BIG_ENDIAN
+BODY: column index bytes + column index bytes + .......
+column number: 4 bytes int, BIG_ENDIAN
+column x name: var bytes, Java modified-utf-8
+index number: 4 bytes int (how many column items below), BIG_ENDIAN
+index name x: var bytes, Java modified-utf-8
+start pos: 8 bytes long, BIG_ENDIAN
+length: 8 bytes long, BIG_ENDIAN
+footer length: 4 bytes int, BIG_ENDIAN
+tail magic: 8 bytes long, value is PAFIDX02
+</pre>
+
+`start pos` is relative to the container start. For a container of length `F`, the footer starts at
+`F - 12 - footer length`.
+`FileIndexFormat.createReader` receives the container length so it can locate the trailer and
+footer.
+
+V2 supports a container larger than 2 GiB by storing payload positions and lengths as 64-bit
+values. Payloads are written through `FileIndexWriter.writeTo(OutputStream)`; existing byte-array
+writers use its default implementation. `FileIndexer.createReader` receives both the start position
+and length as 64-bit values. Index implementations that load a payload into a `byte[]`, such as
+Bloom filter, may reject lengths greater than `Integer.MAX_VALUE`.
+
## Index: BloomFilter
Options are:
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 4c30f37c5a..d0ad00368d 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -716,11 +716,17 @@ under the License.
<td>String</td>
<td>Default aggregate function of all fields for partial-update and aggregate merge function.</td>
</tr>
+ <tr>
+ <td><h5>file-index.format.version</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>File index container version to write (1 or 2). Version 2 supports 64-bit payload positions and lengths.</td>
+ </tr>
<tr>
<td><h5>file-index.in-manifest-threshold</h5></td>
<td style="word-wrap: break-word;">500 bytes</td>
<td>MemorySize</td>
- <td>The threshold to store file index bytes in manifest.</td>
+ <td>The threshold to store file index bytes in manifest. Valid range: 0 to 2,147,483,647 bytes.</td>
</tr>
<tr>
<td><h5>file-index.read.enabled</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index f9cb0914f6..b350e3b2b2 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -433,7 +433,15 @@ public class CoreOptions implements Serializable {
key("file-index.in-manifest-threshold")
.memoryType()
.defaultValue(MemorySize.parse("500 B"))
- .withDescription("The threshold to store file index bytes in manifest.");
+ .withDescription(
+ "The threshold to store file index bytes in manifest. Valid range: 0 to 2,147,483,647 bytes.");
+
+ public static final ConfigOption<Integer> FILE_INDEX_FORMAT_VERSION =
+ key("file-index.format.version")
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "File index container version to write (1 or 2). Version 2 supports 64-bit payload positions and lengths.");
public static final ConfigOption<Boolean> FILE_INDEX_READ_ENABLED =
key("file-index.read.enabled")
@@ -4564,8 +4572,23 @@ public class CoreOptions implements Serializable {
return new FileIndexOptions(this);
}
+ public int fileIndexFormatVersion() {
+ int version = options.get(FILE_INDEX_FORMAT_VERSION);
+ if (version != 1 && version != 2) {
+ throw new IllegalArgumentException("file-index.format.version must be 1 or 2");
+ }
+ return version;
+ }
+
public long fileIndexInManifestThreshold() {
- return options.get(FILE_INDEX_IN_MANIFEST_THRESHOLD).getBytes();
+ long bytes = options.get(FILE_INDEX_IN_MANIFEST_THRESHOLD).getBytes();
+ checkArgument(
+ bytes >= 0 && bytes <= Integer.MAX_VALUE,
+ "'%s' must be between 0 and %s bytes, but was %s bytes.",
+ FILE_INDEX_IN_MANIFEST_THRESHOLD.key(),
+ Integer.MAX_VALUE,
+ bytes);
+ return bytes;
}
public boolean fileIndexReadEnabled() {
diff --git a/paimon-api/src/main/java/org/apache/paimon/fileindex/FileIndexOptions.java b/paimon-api/src/main/java/org/apache/paimon/fileindex/FileIndexOptions.java
index 4dd34dbadc..3dbfc8c621 100644
--- a/paimon-api/src/main/java/org/apache/paimon/fileindex/FileIndexOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/fileindex/FileIndexOptions.java
@@ -40,6 +40,7 @@ public class FileIndexOptions {
// if the filter size greater than fileIndexInManifestThreshold, we put it in file
private final long fileIndexInManifestThreshold;
+ private final int formatVersion;
private final Map<Column, Map<String, Options>> indexTypeOptions;
private final Map<Column, Map<String, Options>> topLevelMapColumnOptions;
@@ -52,6 +53,7 @@ public class FileIndexOptions {
this.indexTypeOptions = new HashMap<>();
this.topLevelMapColumnOptions = new HashMap<>();
this.fileIndexInManifestThreshold = coreOptions.fileIndexInManifestThreshold();
+ this.formatVersion = coreOptions.fileIndexFormatVersion();
setupOptions(coreOptions);
}
@@ -186,6 +188,10 @@ public class FileIndexOptions {
return indexTypeOptions.isEmpty();
}
+ public int formatVersion() {
+ return formatVersion;
+ }
+
public long fileIndexInManifestThreshold() {
return fileIndexInManifestThreshold;
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
index c2d27ec9f0..0359d6073f 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormat.java
@@ -18,7 +18,6 @@
package org.apache.paimon.fileindex;
-import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.fileindex.empty.EmptyFileIndexReader;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.options.Options;
@@ -27,84 +26,38 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.Pair;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
+import javax.annotation.Nullable;
+
import java.io.Closeable;
import java.io.DataInputStream;
-import java.io.DataOutput;
-import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
-/**
- * File index file format. Put all column and offset in the header.
- *
- * <pre>
- * ______________________________________ _____________________
- * | magic ｜version｜head length |
- * |--------------------------------------|
- * | column number |
- * |--------------------------------------|
- * | column 1 ｜ index number |
- * |--------------------------------------|
- * | index name 1 ｜start pos ｜length |
- * |--------------------------------------|
- * | index name 2 ｜start pos ｜length |
- * |--------------------------------------|
- * | index name 3 ｜start pos ｜length |
- * |--------------------------------------| HEAD
- * | column 2 ｜ index number |
- * |--------------------------------------|
- * | index name 1 ｜start pos ｜length |
- * |--------------------------------------|
- * | index name 2 ｜start pos ｜length |
- * |--------------------------------------|
- * | index name 3 ｜start pos ｜length |
- * |--------------------------------------|
- * | ... |
- * |--------------------------------------|
- * | ... |
- * |--------------------------------------|
- * | redundant length ｜redundant bytes |
- * |--------------------------------------| ---------------------
- * | BODY |
- * | BODY |
- * | BODY | BODY
- * | BODY |
- * |______________________________________| _____________________
- *
- * magic: 8 bytes long
- * version: 4 bytes int
- * head length: 4 bytes int
- * column number: 4 bytes int
- * column x: var bytes utf (length + bytes)
- * index number: 4 bytes int (how many column items below)
- * index name x: var bytes utf
- * start pos: 4 bytes int
- * length: 4 bytes int
- * redundant length: 4 bytes int (for compatibility with later versions, in this version, content is zero)
- * redundant bytes: var bytes (for compatibility with later version, in this version, is empty)
- * BODY: column index bytes + column index bytes + column index bytes + .......
- *
- * </pre>
- */
+import static org.apache.paimon.fileindex.FileIndexFormatUtils.EMPTY_INDEX_FLAG;
+import static org.apache.paimon.fileindex.FileIndexFormatUtils.VERSION_1;
+import static org.apache.paimon.fileindex.FileIndexFormatUtils.VERSION_2;
+
+/** Version-dispatching entry point and shared payload access for file index containers. */
public final class FileIndexFormat {
- private static final long MAGIC = 1493475289347502L;
- private static final int EMPTY_INDEX_FLAG = -1;
+ /** Writes a single payload to the container output. */
+ @FunctionalInterface
+ public interface Payload {
+ void writeTo(OutputStream output) throws IOException;
+ }
enum Version {
- V_1(1);
+ V_1(VERSION_1),
+ V_2(VERSION_2);
private final int version;
@@ -117,15 +70,45 @@ public final class FileIndexFormat {
}
}
+ public static Writer createWriter(OutputStream outputStream, int version) throws IOException {
+ if (version == Version.V_1.version()) {
+ return new FileIndexFormatV1.Writer(outputStream);
+ } else if (version == Version.V_2.version()) {
+ return new FileIndexFormatV2.Writer(outputStream);
+ } else {
+ throw new IllegalArgumentException("Unsupported file index version: " + version);
+ }
+ }
+
+ /** @deprecated Use {@link #createWriter(OutputStream, int)}. */
+ @Deprecated
public static Writer createWriter(OutputStream outputStream) {
- return new Writer(outputStream);
+ return new FileIndexFormatV1.Writer(outputStream);
+ }
+
+ public static Reader createReader(
+ SeekableInputStream inputStream, RowType fileRowType, long length) {
+ return new Reader(inputStream, fileRowType, length);
+ }
+
+ static Reader createReaderAfterVersion(
+ SeekableInputStream inputStream, RowType fileRowType, long length, int version) {
+ return new Reader(inputStream, fileRowType, length, version);
}
+ /** @deprecated Use {@link #createReader(SeekableInputStream, RowType, long)}. */
+ @Deprecated
public static Reader createReader(SeekableInputStream inputStream, RowType fileRowType) {
return new Reader(inputStream, fileRowType);
}
- /** Creates a reader for accessing header metadata without reading index payloads. */
+ /** Creates a reader for accessing index metadata without reading index payloads. */
+ public static Reader createMetadataReader(SeekableInputStream inputStream, long length) {
+ return createReader(inputStream, RowType.builder().build(), length);
+ }
+
+ /** @deprecated Use {@link #createMetadataReader(SeekableInputStream, long)}. */
+ @Deprecated
public static Reader createMetadataReader(SeekableInputStream inputStream) {
return new Reader(inputStream, RowType.builder().build());
}
@@ -135,10 +118,11 @@ public final class FileIndexFormat {
private final String columnName;
private final String indexType;
- private final int sizeInBytes;
+ private final long sizeInBytes;
private final boolean empty;
- private FileIndexMeta(String columnName, String indexType, int sizeInBytes, boolean empty) {
+ private FileIndexMeta(
+ String columnName, String indexType, long sizeInBytes, boolean empty) {
this.columnName = columnName;
this.indexType = indexType;
this.sizeInBytes = sizeInBytes;
@@ -153,7 +137,13 @@ public final class FileIndexFormat {
return indexType;
}
+ /** @deprecated Use {@link #sizeInBytesLong()}. */
+ @Deprecated
public int sizeInBytes() {
+ return Math.toIntExact(sizeInBytes);
+ }
+
+ public long sizeInBytesLong() {
return sizeInBytes;
}
@@ -165,109 +155,51 @@ public final class FileIndexFormat {
/** Writer for file index file. */
public static class Writer implements Closeable {
- private final DataOutputStream dataOutputStream;
+ @Nullable private final Writer delegate;
- // for version compatible
- private static final int REDUNDANT_LENGTH = 0;
+ protected Writer() {
+ this.delegate = null;
+ }
+ /** @deprecated Use {@link FileIndexFormat#createWriter(OutputStream, int)}. */
+ @Deprecated
public Writer(OutputStream outputStream) {
- this.dataOutputStream = new DataOutputStream(outputStream);
+ this.delegate = new FileIndexFormatV1.Writer(outputStream);
}
+ /**
+ * @deprecated Use {@link #writeIndex(String, String, Payload)} and {@link #finish()} to
+ * stream payloads into the container.
+ */
+ @Deprecated
public void writeColumnIndexes(Map<String, Map<String, byte[]>> indexes)
throws IOException {
-
- Map<String, Map<String, Pair<Integer, Integer>>> bodyInfo = new LinkedHashMap<>();
-
- // construct body
- ByteArrayOutputStream baos = new ByteArrayOutputStream(256);
- for (Map.Entry<String, Map<String, byte[]>> columnMap : indexes.entrySet()) {
- Map<String, Pair<Integer, Integer>> innerMap =
- bodyInfo.computeIfAbsent(columnMap.getKey(), k -> new LinkedHashMap<>());
- Map<String, byte[]> bytesMap = columnMap.getValue();
- for (Map.Entry<String, byte[]> entry : bytesMap.entrySet()) {
- int startPosition = baos.size();
- byte[] v = entry.getValue();
- if (v == null) {
- innerMap.put(entry.getKey(), Pair.of(EMPTY_INDEX_FLAG, 0));
- } else {
- baos.write(entry.getValue());
- innerMap.put(
- entry.getKey(),
- Pair.of(startPosition, baos.size() - startPosition));
- }
+ for (Map.Entry<String, Map<String, byte[]>> column : indexes.entrySet()) {
+ for (Map.Entry<String, byte[]> index : column.getValue().entrySet()) {
+ byte[] payload = index.getValue();
+ writeIndex(
+ column.getKey(),
+ index.getKey(),
+ payload == null ? null : output -> output.write(payload));
}
}
- byte[] body = baos.toByteArray();
- writeHead(bodyInfo);
-
- // writeBody
- dataOutputStream.write(body);
+ finish();
}
- private void writeHead(Map<String, Map<String, Pair<Integer, Integer>>> bodyInfo)
+ /** Writes one payload to the container. A null payload is empty. */
+ public void writeIndex(String columnName, String indexType, @Nullable Payload payload)
throws IOException {
-
- int headLength = calculateHeadLength(bodyInfo);
-
- // writeMagic
- dataOutputStream.writeLong(MAGIC);
- // writeVersion
- dataOutputStream.writeInt(Version.V_1.version());
- // writeHeadLength
- dataOutputStream.writeInt(headLength);
- // writeColumnSize
- dataOutputStream.writeInt(bodyInfo.size());
- for (Map.Entry<String, Map<String, Pair<Integer, Integer>>> entry :
- bodyInfo.entrySet()) {
- // writeColumnName
- dataOutputStream.writeUTF(entry.getKey());
- // writeIndexTypeSize
- dataOutputStream.writeInt(entry.getValue().size());
- // writeColumnInfo, offset = headLength
- for (Map.Entry<String, Pair<Integer, Integer>> indexEntry :
- entry.getValue().entrySet()) {
- dataOutputStream.writeUTF(indexEntry.getKey());
- int start = indexEntry.getValue().getLeft();
- dataOutputStream.writeInt(
- start == EMPTY_INDEX_FLAG ? EMPTY_INDEX_FLAG : start + headLength);
- dataOutputStream.writeInt(indexEntry.getValue().getRight());
- }
- }
- // writeRedundantLength
- dataOutputStream.writeInt(REDUNDANT_LENGTH);
+ delegate.writeIndex(columnName, indexType, payload);
}
- private int calculateHeadLength(Map<String, Map<String, Pair<Integer, Integer>>> bodyInfo)
- throws IOException {
- // magic 8 bytes, version 4 bytes, head length 4 bytes,
- // column number 4 bytes, body info start&length 8 bytes per
- // column-index, index number size 4 bytes per column, redundant length 4 bytes;
- int baseLength =
- 8
- + 4
- + 4
- + 4
- + bodyInfo.values().stream().mapToInt(Map::size).sum() * 8
- + bodyInfo.size() * 4
- + 4;
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutput dataOutput = new DataOutputStream(baos);
- for (Map.Entry<String, Map<String, Pair<Integer, Integer>>> entry :
- bodyInfo.entrySet()) {
- dataOutput.writeUTF(entry.getKey());
- for (String s : entry.getValue().keySet()) {
- dataOutput.writeUTF(s);
- }
- }
-
- return baseLength + baos.size();
+ /** Completes the container after all payloads have been written. */
+ public void finish() throws IOException {
+ delegate.finish();
}
@Override
public void close() throws IOException {
- IOUtils.closeQuietly(dataOutputStream);
+ delegate.close();
}
}
@@ -275,49 +207,54 @@ public final class FileIndexFormat {
public static class Reader implements Closeable {
private final SeekableInputStream seekableInputStream;
- // get header and cache it.
- private final Map<String, Map<String, Pair<Integer, Integer>>> header = new HashMap<>();
+ // Cache the index entries.
+ private final Map<String, Map<String, Pair<Long, Long>>> indexEntries = new HashMap<>();
private final Map<String, DataField> fields = new HashMap<>();
+ /**
+ * @deprecated Use {@link FileIndexFormat#createReader(SeekableInputStream, RowType, long)}.
+ */
+ @Deprecated
public Reader(SeekableInputStream seekableInputStream, RowType fileRowType) {
+ this(seekableInputStream, fileRowType, -1L);
+ }
+
+ private Reader(SeekableInputStream seekableInputStream, RowType fileRowType, long length) {
+ this(seekableInputStream, fileRowType, length, null);
+ }
+
+ private Reader(
+ SeekableInputStream seekableInputStream,
+ RowType fileRowType,
+ long length,
+ @Nullable Integer version) {
this.seekableInputStream = seekableInputStream;
DataInputStream dataInputStream = new DataInputStream(seekableInputStream);
fileRowType.getFields().forEach(field -> this.fields.put(field.name(), field));
try {
- long magic = dataInputStream.readLong();
- if (magic != MAGIC) {
- throw new RuntimeException("This file is not file index file.");
- }
-
- int version = dataInputStream.readInt();
- if (version != Version.V_1.version()) {
+ int actualVersion =
+ version == null
+ ? FileIndexFormatUtils.readVersion(dataInputStream)
+ : version;
+ if (actualVersion == Version.V_2.version()) {
+ if (length < 0) {
+ throw new RuntimeException(
+ "File index length is required to read version " + actualVersion);
+ }
+ indexEntries.putAll(
+ FileIndexFormatV2.readIndexEntries(seekableInputStream, length));
+ } else if (actualVersion == Version.V_1.version()) {
+ indexEntries.putAll(FileIndexFormatV1.readIndexEntries(dataInputStream));
+ } else {
throw new RuntimeException(
"This index file is version of "
- + version
+ + actualVersion
+ ", not in supported version list ["
+ Version.V_1.version()
+ + ", "
+ + Version.V_2.version()
+ "]");
}
-
- int headLength = dataInputStream.readInt();
- byte[] head = new byte[headLength - 8 - 4 - 4];
- dataInputStream.readFully(head);
-
- try (DataInputStream dataInput =
- new DataInputStream(new ByteArrayInputStream(head))) {
- int columnSize = dataInput.readInt();
- for (int i = 0; i < columnSize; i++) {
- String columnName = dataInput.readUTF();
- int indexSize = dataInput.readInt();
- Map<String, Pair<Integer, Integer>> indexMap =
- this.header.computeIfAbsent(columnName, n -> new HashMap<>());
- for (int j = 0; j < indexSize; j++) {
- indexMap.put(
- dataInput.readUTF(),
- Pair.of(dataInput.readInt(), dataInput.readInt()));
- }
- }
- }
} catch (IOException | RuntimeException e) {
// Callers wrap the constructor in try-with-resources on the stream,
// but a throwing constructor never assigns the resource, so both
@@ -329,7 +266,7 @@ public final class FileIndexFormat {
}
public Set<FileIndexReader> readColumnIndex(String columnName) {
- return Optional.ofNullable(header.getOrDefault(columnName, null))
+ return Optional.ofNullable(indexEntries.getOrDefault(columnName, null))
.map(
f ->
f.entrySet().stream()
@@ -343,14 +280,14 @@ public final class FileIndexFormat {
.orElse(Collections.emptySet());
}
- /** Returns the index metadata parsed from the header without reading index payloads. */
+ /** Returns the parsed index metadata without reading index payloads. */
public List<FileIndexMeta> indexMetas() {
List<FileIndexMeta> metas = new ArrayList<>();
- for (Map.Entry<String, Map<String, Pair<Integer, Integer>>> columnEntry :
- header.entrySet()) {
- for (Map.Entry<String, Pair<Integer, Integer>> indexEntry :
+ for (Map.Entry<String, Map<String, Pair<Long, Long>>> columnEntry :
+ indexEntries.entrySet()) {
+ for (Map.Entry<String, Pair<Long, Long>> indexEntry :
columnEntry.getValue().entrySet()) {
- Pair<Integer, Integer> startAndLength = indexEntry.getValue();
+ Pair<Long, Long> startAndLength = indexEntry.getValue();
metas.add(
new FileIndexMeta(
columnEntry.getKey(),
@@ -363,7 +300,7 @@ public final class FileIndexFormat {
}
private FileIndexReader getFileIndexReader(
- String columnName, String indexType, Pair<Integer, Integer> startAndLength) {
+ String columnName, String indexType, Pair<Long, Long> startAndLength) {
if (startAndLength.getLeft() == EMPTY_INDEX_FLAG) {
return EmptyFileIndexReader.INSTANCE;
}
@@ -377,47 +314,58 @@ public final class FileIndexFormat {
startAndLength.getRight());
}
- private byte[] getBytesWithStartAndLength(Pair<Integer, Integer> startAndLength) {
- byte[] b = new byte[startAndLength.getRight()];
- try {
- seekableInputStream.seek(startAndLength.getLeft());
- int n = 0;
- int len = b.length;
- // read fully until b is full else throw.
- while (n < len) {
- int count = seekableInputStream.read(b, n, len - n);
- if (count < 0) {
- throw new EOFException();
- }
- n += count;
+ /**
+ * Copies a stored payload without loading it into a byte array or decoding its plugin
+ * format.
+ */
+ public void copyPayload(String columnName, String indexType, OutputStream output)
+ throws IOException {
+ Pair<Long, Long> startAndLength = indexEntries.get(columnName).get(indexType);
+ seekableInputStream.seek(startAndLength.getLeft());
+ byte[] buffer = new byte[8192];
+ long remaining = startAndLength.getRight();
+ while (remaining > 0) {
+ int count =
+ seekableInputStream.read(
+ buffer, 0, (int) Math.min(remaining, buffer.length));
+ if (count < 0) {
+ throw new EOFException();
}
- } catch (IOException e) {
- throw new RuntimeException(e);
+ output.write(buffer, 0, count);
+ remaining -= count;
}
- return b;
}
+ /** @deprecated Read payloads through {@link #copyPayload(String, String, OutputStream)}. */
+ @Deprecated
public Map<String, Map<String, byte[]>> readAll() {
Map<String, Map<String, byte[]>> result = new HashMap<>();
- for (Map.Entry<String, Map<String, Pair<Integer, Integer>>> entryOuter :
- header.entrySet()) {
- for (Map.Entry<String, Pair<Integer, Integer>> entryInner :
- entryOuter.getValue().entrySet()) {
- result.computeIfAbsent(entryOuter.getKey(), key -> new HashMap<>())
- .put(
- entryInner.getKey(),
- getBytesWithStartAndLength(entryInner.getValue()));
+ for (Map.Entry<String, Map<String, Pair<Long, Long>>> column :
+ indexEntries.entrySet()) {
+ for (Map.Entry<String, Pair<Long, Long>> index : column.getValue().entrySet()) {
+ result.computeIfAbsent(column.getKey(), key -> new HashMap<>())
+ .put(index.getKey(), readPayload(index.getValue()));
}
}
return result;
}
- @VisibleForTesting
- // only for test yet
- Optional<byte[]> getBytesWithNameAndType(String columnName, String indexType) {
- return Optional.ofNullable(header.getOrDefault(columnName, null))
- .map(i -> i.getOrDefault(indexType, null))
- .map(this::getBytesWithStartAndLength);
+ private byte[] readPayload(Pair<Long, Long> startAndLength) {
+ byte[] payload = new byte[Math.toIntExact(startAndLength.getRight())];
+ try {
+ seekableInputStream.seek(startAndLength.getLeft());
+ int read = 0;
+ while (read < payload.length) {
+ int count = seekableInputStream.read(payload, read, payload.length - read);
+ if (count < 0) {
+ throw new EOFException();
+ }
+ read += count;
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return payload;
}
@Override
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatUtils.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatUtils.java
new file mode 100644
index 0000000000..79c427b296
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatUtils.java
@@ -0,0 +1,148 @@
+/*
+ * 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.fileindex;
+
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.utils.Pair;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Utilities shared by file index container versions. */
+final class FileIndexFormatUtils {
+
+ static final long MAGIC = 1493475289347502L;
+ static final int EMPTY_INDEX_FLAG = -1;
+ static final int VERSION_1 = 1;
+ static final int VERSION_2 = 2;
+
+ private FileIndexFormatUtils() {}
+
+ static void writeMagicAndVersion(DataOutput output, int version) throws IOException {
+ // writeMagic
+ output.writeLong(MAGIC);
+ // writeVersion
+ output.writeInt(version);
+ }
+
+ static int readVersion(DataInput input) throws IOException {
+ if (input.readLong() != MAGIC) {
+ throw new RuntimeException("This file is not file index file.");
+ }
+ return input.readInt();
+ }
+
+ static Map<String, Map<String, Pair<Long, Long>>> readIndexEntries(DataInput input, int version)
+ throws IOException {
+ Map<String, Map<String, Pair<Long, Long>>> indexEntries = new HashMap<>();
+ int columnSize = input.readInt();
+ for (int i = 0; i < columnSize; i++) {
+ String columnName = input.readUTF();
+ int indexSize = input.readInt();
+ Map<String, Pair<Long, Long>> entries =
+ indexEntries.computeIfAbsent(columnName, ignored -> new HashMap<>());
+ for (int j = 0; j < indexSize; j++) {
+ String indexType = input.readUTF();
+ Pair<Long, Long> startAndLength =
+ version == VERSION_1
+ ? Pair.of((long) input.readInt(), (long) input.readInt())
+ : Pair.of(input.readLong(), input.readLong());
+ entries.put(indexType, startAndLength);
+ }
+ }
+ return indexEntries;
+ }
+
+ static void writeIndexEntries(
+ DataOutput output,
+ Map<String, Map<String, Pair<Long, Long>>> indexEntries,
+ int version,
+ long startOffset)
+ throws IOException {
+ // writeColumnSize
+ output.writeInt(indexEntries.size());
+ for (Map.Entry<String, Map<String, Pair<Long, Long>>> column : indexEntries.entrySet()) {
+ // writeColumnName
+ output.writeUTF(column.getKey());
+ // writeIndexTypeSize
+ output.writeInt(column.getValue().size());
+ for (Map.Entry<String, Pair<Long, Long>> index : column.getValue().entrySet()) {
+ output.writeUTF(index.getKey());
+ long start = index.getValue().getLeft();
+ // Keep EMPTY_INDEX_FLAG unchanged when converting a relative start position.
+ start = start == EMPTY_INDEX_FLAG ? EMPTY_INDEX_FLAG : start + startOffset;
+ if (version == VERSION_1) {
+ output.writeInt((int) start);
+ output.writeInt(index.getValue().getRight().intValue());
+ } else {
+ output.writeLong(start);
+ output.writeLong(index.getValue().getRight());
+ }
+ }
+ }
+ }
+
+ static class CountingPositionOutputStream extends PositionOutputStream {
+
+ private final OutputStream output;
+ private long position;
+
+ CountingPositionOutputStream(OutputStream output) {
+ this.output = output;
+ }
+
+ @Override
+ public long getPos() {
+ return position;
+ }
+
+ @Override
+ public void write(int value) throws IOException {
+ long next = Math.addExact(position, 1L);
+ output.write(value);
+ position = next;
+ }
+
+ @Override
+ public void write(byte[] bytes) throws IOException {
+ write(bytes, 0, bytes.length);
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length) throws IOException {
+ long next = Math.addExact(position, length);
+ output.write(bytes, offset, length);
+ position = next;
+ }
+
+ @Override
+ public void flush() throws IOException {
+ output.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ output.close();
+ }
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatV1.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatV1.java
new file mode 100644
index 0000000000..f8a288b36c
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatV1.java
@@ -0,0 +1,189 @@
+/*
+ * 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.fileindex;
+
+import org.apache.paimon.utils.Pair;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.paimon.fileindex.FileIndexFormatUtils.VERSION_1;
+
+/**
+ * V1 file index container: header followed by payloads.
+ *
+ * <pre>
+ * ______________________________________ _____________________
+ * | magic | version| head length |
+ * |--------------------------------------|
+ * | column number |
+ * |--------------------------------------|
+ * | column 1 | index number |
+ * |--------------------------------------|
+ * | index name 1 | start pos | length |
+ * |--------------------------------------|
+ * | index name 2 | start pos | length |
+ * |--------------------------------------|
+ * | index name 3 | start pos | length |
+ * |--------------------------------------| HEAD
+ * | column 2 | index number |
+ * |--------------------------------------|
+ * | index name 1 | start pos | length |
+ * |--------------------------------------|
+ * | index name 2 | start pos | length |
+ * |--------------------------------------|
+ * | index name 3 | start pos | length |
+ * |--------------------------------------|
+ * | ... |
+ * |--------------------------------------|
+ * | ... |
+ * |--------------------------------------|
+ * | redundant length | redundant bytes |
+ * |--------------------------------------| ---------------------
+ * | BODY |
+ * | BODY |
+ * | BODY | BODY
+ * | BODY |
+ * |______________________________________| _____________________
+ *
+ * magic: 8 bytes long
+ * version: 4 bytes int
+ * head length: 4 bytes int
+ * column number: 4 bytes int
+ * column x: var bytes utf (length + bytes)
+ * index number: 4 bytes int (how many column items below)
+ * index name x: var bytes utf
+ * start pos: 4 bytes int
+ * length: 4 bytes int
+ * redundant length: 4 bytes int (for compatibility with later versions, in this version, content is zero)
+ * redundant bytes: var bytes (for compatibility with later version, in this version, is empty)
+ * BODY: column index bytes + column index bytes + column index bytes + .......
+ *
+ * </pre>
+ */
+final class FileIndexFormatV1 {
+
+ private FileIndexFormatV1() {}
+
+ /** Reads the header after the common magic and version have been consumed. */
+ static Map<String, Map<String, Pair<Long, Long>>> readIndexEntries(
+ DataInputStream dataInputStream) throws IOException {
+ int headLength = dataInputStream.readInt();
+ byte[] head = new byte[headLength - 8 - 4 - 4];
+ dataInputStream.readFully(head);
+ try (DataInputStream dataInput = new DataInputStream(new ByteArrayInputStream(head))) {
+ return FileIndexFormatUtils.readIndexEntries(dataInput, VERSION_1);
+ }
+ }
+
+ /** V1 writer buffers payloads to compute the preceding header. */
+ static final class Writer extends FileIndexFormat.Writer {
+
+ private final DataOutputStream dataOutputStream;
+ private final ByteArrayOutputStream body = new ByteArrayOutputStream(256);
+ private final Map<String, Map<String, Pair<Long, Long>>> indexEntries =
+ new LinkedHashMap<>();
+
+ // for version compatible
+ private static final int REDUNDANT_LENGTH = 0;
+
+ Writer(OutputStream outputStream) {
+ this.dataOutputStream = new DataOutputStream(outputStream);
+ }
+
+ @Override
+ public void writeIndex(
+ String columnName, String indexType, @Nullable FileIndexFormat.Payload payload)
+ throws IOException {
+ Map<String, Pair<Long, Long>> column =
+ indexEntries.computeIfAbsent(columnName, ignored -> new LinkedHashMap<>());
+ if (payload == null) {
+ // Empty indexes have no payload and use EMPTY_INDEX_FLAG as their start position.
+ column.put(indexType, Pair.of((long) FileIndexFormatUtils.EMPTY_INDEX_FLAG, 0L));
+ } else {
+ int start = body.size();
+ payload.writeTo(body);
+ column.put(indexType, Pair.of((long) start, (long) body.size() - start));
+ }
+ }
+
+ @Override
+ public void finish() throws IOException {
+ writeHead(indexEntries);
+
+ // writeBody
+ body.writeTo(dataOutputStream);
+ }
+
+ private void writeHead(Map<String, Map<String, Pair<Long, Long>>> indexEntries)
+ throws IOException {
+
+ int headLength = calculateHeadLength(indexEntries);
+
+ FileIndexFormatUtils.writeMagicAndVersion(dataOutputStream, VERSION_1);
+ // writeHeadLength
+ dataOutputStream.writeInt(headLength);
+ // writeColumnInfo, offset = headLength
+ FileIndexFormatUtils.writeIndexEntries(
+ dataOutputStream, indexEntries, VERSION_1, headLength);
+ // writeRedundantLength
+ dataOutputStream.writeInt(REDUNDANT_LENGTH);
+ }
+
+ private int calculateHeadLength(Map<String, Map<String, Pair<Long, Long>>> indexEntries)
+ throws IOException {
+ // magic 8 bytes, version 4 bytes, head length 4 bytes,
+ // column number 4 bytes, body info start&length 8 bytes per
+ // column-index, index number size 4 bytes per column, redundant length 4 bytes;
+ int baseLength =
+ 8
+ + 4
+ + 4
+ + 4
+ + indexEntries.values().stream().mapToInt(Map::size).sum() * 8
+ + indexEntries.size() * 4
+ + 4;
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutput dataOutput = new DataOutputStream(baos);
+ for (Map.Entry<String, Map<String, Pair<Long, Long>>> entry : indexEntries.entrySet()) {
+ dataOutput.writeUTF(entry.getKey());
+ for (String s : entry.getValue().keySet()) {
+ dataOutput.writeUTF(s);
+ }
+ }
+
+ return baseLength + baos.size();
+ }
+
+ @Override
+ public void close() throws IOException {
+ dataOutputStream.close();
+ }
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatV2.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatV2.java
new file mode 100644
index 0000000000..1dfb6505be
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexFormatV2.java
@@ -0,0 +1,158 @@
+/*
+ * 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.fileindex;
+
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.utils.Pair;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.paimon.fileindex.FileIndexFormatUtils.VERSION_2;
+
+/**
+ * V2 file index container: prefix and body followed by footer and trailer.
+ *
+ * <pre>
+ * ______________________________________ _____________________
+ * | magic | version | PREFIX
+ * |--------------------------------------| ---------------------
+ * | BODY |
+ * | BODY | BODY
+ * | BODY |
+ * |--------------------------------------| ---------------------
+ * | column number |
+ * |--------------------------------------|
+ * | column 1 | index number |
+ * |--------------------------------------|
+ * | index name 1 | start pos | length |
+ * |--------------------------------------|
+ * | index name 2 | start pos | length |
+ * |--------------------------------------| FOOTER
+ * | column 2 | index number |
+ * |--------------------------------------|
+ * | index name 1 | start pos | length |
+ * |--------------------------------------|
+ * | index name 2 | start pos | length |
+ * |--------------------------------------|
+ * | ... |
+ * |--------------------------------------|
+ * | ... |
+ * |--------------------------------------| ---------------------
+ * | footer length | tail magic | TRAILER
+ * |______________________________________| _____________________
+ *
+ * magic: 8 bytes long
+ * version: 4 bytes int
+ * BODY: column index bytes + column index bytes + .......
+ * column number: 4 bytes int
+ * column x: var bytes utf (length + bytes)
+ * index number: 4 bytes int (how many column items below)
+ * index name x: var bytes utf
+ * start pos: 8 bytes long
+ * length: 8 bytes long
+ * footer length: 4 bytes int
+ * tail magic: 8 bytes long
+ *
+ * </pre>
+ */
+final class FileIndexFormatV2 {
+
+ // ASCII PAFIDX02.
+ private static final long TAIL_MAGIC = 0x5041464944583032L;
+ private static final int TRAILER_LENGTH = 12;
+
+ private FileIndexFormatV2() {}
+
+ static Map<String, Map<String, Pair<Long, Long>>> readIndexEntries(
+ SeekableInputStream seekableInputStream, long length) throws IOException {
+ seekableInputStream.seek(length - TRAILER_LENGTH);
+ DataInputStream tail = new DataInputStream(seekableInputStream);
+ int footerLength = tail.readInt();
+ if (tail.readLong() != TAIL_MAGIC) {
+ throw new IOException("Invalid file index tail magic");
+ }
+ long footerStart = length - TRAILER_LENGTH - footerLength;
+ byte[] footer = new byte[footerLength];
+ seekableInputStream.seek(footerStart);
+ new DataInputStream(seekableInputStream).readFully(footer);
+ try (DataInputStream in = new DataInputStream(new ByteArrayInputStream(footer))) {
+ return FileIndexFormatUtils.readIndexEntries(in, VERSION_2);
+ }
+ }
+
+ /** V2 writer records streamed payload ranges in the footer. */
+ static final class Writer extends FileIndexFormat.Writer {
+
+ private final DataOutputStream dataOutputStream;
+ private final FileIndexFormatUtils.CountingPositionOutputStream positionOutputStream;
+ private final Map<String, Map<String, Pair<Long, Long>>> indexEntries =
+ new LinkedHashMap<>();
+
+ Writer(OutputStream outputStream) throws IOException {
+ this.positionOutputStream =
+ new FileIndexFormatUtils.CountingPositionOutputStream(outputStream);
+ this.dataOutputStream = new DataOutputStream(positionOutputStream);
+ FileIndexFormatUtils.writeMagicAndVersion(dataOutputStream, VERSION_2);
+ }
+
+ @Override
+ public void writeIndex(
+ String columnName, String indexType, @Nullable FileIndexFormat.Payload payload)
+ throws IOException {
+ Map<String, Pair<Long, Long>> column =
+ indexEntries.computeIfAbsent(columnName, ignored -> new LinkedHashMap<>());
+ if (payload == null) {
+ // Empty indexes have no payload and use EMPTY_INDEX_FLAG as their start position.
+ column.put(indexType, Pair.of((long) FileIndexFormatUtils.EMPTY_INDEX_FLAG, 0L));
+ } else {
+ long start = positionOutputStream.getPos();
+ payload.writeTo(dataOutputStream);
+ column.put(indexType, Pair.of(start, positionOutputStream.getPos() - start));
+ }
+ }
+
+ @Override
+ public void finish() throws IOException {
+ writeFooter(indexEntries);
+ }
+
+ private void writeFooter(Map<String, Map<String, Pair<Long, Long>>> indexEntries)
+ throws IOException {
+ long start = positionOutputStream.getPos();
+ FileIndexFormatUtils.writeIndexEntries(dataOutputStream, indexEntries, VERSION_2, 0L);
+ // writeFooterLength
+ dataOutputStream.writeInt(Math.toIntExact(positionOutputStream.getPos() - start));
+ // writeTailMagic
+ dataOutputStream.writeLong(TAIL_MAGIC);
+ }
+
+ @Override
+ public void close() throws IOException {
+ dataOutputStream.close();
+ }
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java
index 82d14cf7ca..6874f8116d 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java
@@ -32,6 +32,7 @@ import org.apache.paimon.predicate.PredicateVisitor;
import org.apache.paimon.predicate.SortValue;
import org.apache.paimon.predicate.TopN;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,6 +40,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.Closeable;
+import java.io.DataInputStream;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
@@ -59,20 +61,53 @@ public class FileIndexPredicate implements Closeable {
@Nullable private Path path;
public FileIndexPredicate(Path path, FileIO fileIO, RowType fileRowType) throws IOException {
- this(fileIO.newInputStream(path), fileRowType);
+ this.reader = createReader(path, fileIO, fileRowType);
this.path = path;
}
public FileIndexPredicate(byte[] serializedBytes, RowType fileRowType) {
- this(new ByteArraySeekableStream(serializedBytes), fileRowType);
+ this(new ByteArraySeekableStream(serializedBytes), fileRowType, serializedBytes.length);
}
+ /**
+ * @deprecated Use {@link #FileIndexPredicate(SeekableInputStream, RowType, long)} for V2
+ * containers.
+ */
+ @Deprecated
public FileIndexPredicate(SeekableInputStream inputStream, RowType fileRowType) {
- // createReader itself closes the stream when the header fails validation, so
- // there is no stream to release here anymore.
this.reader = FileIndexFormat.createReader(inputStream, fileRowType);
}
+ public FileIndexPredicate(
+ SeekableInputStream inputStream, RowType fileRowType, long containerLength) {
+ // createReader itself closes the stream when container validation fails, so
+ // there is no stream to release here anymore.
+ this.reader = FileIndexFormat.createReader(inputStream, fileRowType, containerLength);
+ }
+
+ private static FileIndexFormat.Reader createReader(
+ Path path, FileIO fileIO, RowType fileRowType) throws IOException {
+ SeekableInputStream inputStream = fileIO.newInputStream(path);
+ int version;
+ try {
+ version = FileIndexFormatUtils.readVersion(new DataInputStream(inputStream));
+ } catch (IOException | RuntimeException e) {
+ IOUtils.closeQuietly(inputStream);
+ throw new RuntimeException("Exception happens while construct file index reader.", e);
+ }
+
+ long length = -1;
+ if (version == FileIndexFormatUtils.VERSION_2) {
+ try {
+ length = fileIO.getFileStatus(path).getLen();
+ } catch (IOException | RuntimeException e) {
+ IOUtils.closeQuietly(inputStream);
+ throw e;
+ }
+ }
+ return FileIndexFormat.createReaderAfterVersion(inputStream, fileRowType, length, version);
+ }
+
public FileIndexResult evaluate(@Nullable Predicate predicate) {
if (predicate == null) {
return REMAIN;
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexWriter.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexWriter.java
index ede678353a..ad506b3799 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexWriter.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexWriter.java
@@ -18,6 +18,9 @@
package org.apache.paimon.fileindex;
+import java.io.IOException;
+import java.io.OutputStream;
+
/** To write file index. */
public abstract class FileIndexWriter {
@@ -31,7 +34,15 @@ public abstract class FileIndexWriter {
/** The key object may be reused, if saved in memory, please be sure to manually copy it. */
public abstract void write(Object key);
- public abstract byte[] serializedBytes();
+ public byte[] serializedBytes() {
+ throw new UnsupportedOperationException(
+ "This file index writer requires streaming output.");
+ }
+
+ /** Writes one payload. Existing byte-array writers can use the default implementation. */
+ public void writeTo(OutputStream output) throws IOException {
+ output.write(serializedBytes());
+ }
public boolean empty() {
return empty;
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexer.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexer.java
index d6c51286a5..d1549d86a6 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexer.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexer.java
@@ -32,8 +32,23 @@ public interface FileIndexer {
FileIndexWriter createWriter();
+ /** @deprecated Override the long overload to support positions and lengths beyond int32. */
+ @Deprecated
FileIndexReader createReader(SeekableInputStream inputStream, int start, int length);
+ default FileIndexReader createReader(SeekableInputStream inputStream, long start, long length) {
+ if (start < Integer.MIN_VALUE
+ || start > Integer.MAX_VALUE
+ || length < Integer.MIN_VALUE
+ || length > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException(
+ String.format(
+ "File index payload range exceeds int32: start = %s, length = %s.",
+ start, length));
+ }
+ return createReader(inputStream, (int) start, (int) length);
+ }
+
static FileIndexer create(String type, DataType dataType, Options options) {
FileIndexerFactory fileIndexerFactory = FileIndexerFactoryUtils.load(type);
return fileIndexerFactory.create(dataType, options);
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapFileIndex.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapFileIndex.java
index d74b0a5f2d..fab7b247ec 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapFileIndex.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapFileIndex.java
@@ -70,6 +70,12 @@ public class BitmapFileIndex implements FileIndexer {
@Override
public FileIndexReader createReader(
SeekableInputStream seekableInputStream, int start, int length) {
+ return createReader(seekableInputStream, (long) start, (long) length);
+ }
+
+ @Override
+ public FileIndexReader createReader(
+ SeekableInputStream seekableInputStream, long start, long length) {
try {
Reader reader = new Reader(seekableInputStream, start, options);
return valuesAreTruncated(dataType) ? new TruncatedValueReader(reader) : reader;
@@ -209,7 +215,7 @@ public class BitmapFileIndex implements FileIndexer {
private static class Reader extends FileIndexReader {
private final SeekableInputStream seekableInputStream;
- private final int headStart;
+ private final long headStart;
private final Map<Object, RoaringBitmap32> bitmaps = new LinkedHashMap<>();
private BitmapFileIndexMeta bitmapFileIndexMeta;
@@ -217,7 +223,7 @@ public class BitmapFileIndex implements FileIndexer {
private final Options options;
- public Reader(SeekableInputStream seekableInputStream, int start, Options options) {
+ public Reader(SeekableInputStream seekableInputStream, long start, Options options) {
this.seekableInputStream = seekableInputStream;
this.headStart = start;
this.options = options;
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java
index 2364466dc8..458b9f6f91 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java
@@ -71,9 +71,15 @@ public class BloomFilterFileIndex implements FileIndexer {
@Override
public FileIndexReader createReader(SeekableInputStream inputStream, int start, int length) {
+ return createReader(inputStream, (long) start, (long) length);
+ }
+
+ @Override
+ public FileIndexReader createReader(SeekableInputStream inputStream, long start, long length) {
+ int payloadLength = checkedPayloadLength(length);
try {
inputStream.seek(start);
- byte[] serializedBytes = new byte[length];
+ byte[] serializedBytes = new byte[payloadLength];
IOUtils.readFully(inputStream, serializedBytes);
return new Reader(dataType, serializedBytes);
} catch (IOException e) {
@@ -81,6 +87,14 @@ public class BloomFilterFileIndex implements FileIndexer {
}
}
+ private static int checkedPayloadLength(long length) {
+ if (length > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException(
+ "bloom-filter file index payload length exceeds int32: " + length);
+ }
+ return (int) length;
+ }
+
private static class Writer extends FileIndexWriter {
private final BloomFilter64 filter;
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/bsi/BitSliceIndexBitmapFileIndex.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/bsi/BitSliceIndexBitmapFileIndex.java
index e48e0934fc..91e0911d3e 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/bsi/BitSliceIndexBitmapFileIndex.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/bsi/BitSliceIndexBitmapFileIndex.java
@@ -69,6 +69,11 @@ public class BitSliceIndexBitmapFileIndex implements FileIndexer {
@Override
public FileIndexReader createReader(SeekableInputStream inputStream, int start, int length) {
+ return createReader(inputStream, (long) start, (long) length);
+ }
+
+ @Override
+ public FileIndexReader createReader(SeekableInputStream inputStream, long start, long length) {
try {
inputStream.seek(start);
DataInput input = new DataInputStream(inputStream);
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/BitSliceIndexBitmap.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/BitSliceIndexBitmap.java
index deb65ce00f..27acb96df0 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/BitSliceIndexBitmap.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/BitSliceIndexBitmap.java
@@ -41,12 +41,12 @@ public class BitSliceIndexBitmap {
private final ByteBuffer indexes;
private final RoaringBitmap32[] slices;
private final SeekableInputStream in;
- private final int bodyOffset;
+ private final long bodyOffset;
private RoaringBitmap32 ebm;
private boolean initialized = false;
- public BitSliceIndexBitmap(SeekableInputStream in, int offset) throws IOException {
+ public BitSliceIndexBitmap(SeekableInputStream in, long offset) throws IOException {
in.seek(offset);
byte[] headerLengthInBytes = new byte[Integer.BYTES];
readFully(in, headerLengthInBytes);
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmap.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmap.java
index 784eb1fca2..6e8f86e3e0 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmap.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmap.java
@@ -49,8 +49,8 @@ public class RangeBitmap {
@Nullable private final Object min;
@Nullable private final Object max;
private final int cardinality;
- private final int dictionaryOffset;
- private final int bsiOffset;
+ private final long dictionaryOffset;
+ private final long bsiOffset;
private final SeekableInputStream in;
private final KeyFactory factory;
@@ -59,7 +59,7 @@ public class RangeBitmap {
private Dictionary dictionary;
private BitSliceIndexBitmap bsi;
- public RangeBitmap(SeekableInputStream in, int offset, KeyFactory factory) {
+ public RangeBitmap(SeekableInputStream in, long offset, KeyFactory factory) {
ByteBuffer headers;
int headerLength;
try {
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmapFileIndex.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmapFileIndex.java
index 0cfa5fa56f..1e1b4e7ff2 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmapFileIndex.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/RangeBitmapFileIndex.java
@@ -61,6 +61,11 @@ public class RangeBitmapFileIndex implements FileIndexer {
@Override
public FileIndexReader createReader(SeekableInputStream in, int start, int length) {
+ return createReader(in, (long) start, (long) length);
+ }
+
+ @Override
+ public FileIndexReader createReader(SeekableInputStream in, long start, long length) {
return new Reader(dataType, in, start);
}
@@ -102,7 +107,7 @@ public class RangeBitmapFileIndex implements FileIndexer {
private final Function<Object, Object> converter;
private final RangeBitmap bitmap;
- public Reader(DataType dataType, SeekableInputStream in, int start) {
+ public Reader(DataType dataType, SeekableInputStream in, long start) {
KeyFactory factory = KeyFactory.create(dataType);
this.converter = factory.createConverter();
this.bitmap = new RangeBitmap(in, start, factory);
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/ChunkedDictionary.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/ChunkedDictionary.java
index 6a719fdad0..b7ec86fb5c 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/ChunkedDictionary.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/ChunkedDictionary.java
@@ -37,7 +37,7 @@ public class ChunkedDictionary implements Dictionary {
private final int size;
private final int offsetsLength;
private final int chunksLength;
- private final int bodyOffset;
+ private final long bodyOffset;
private final SeekableInputStream in;
private final KeyFactory factory;
private final Comparator<Object> comparator;
@@ -46,7 +46,7 @@ public class ChunkedDictionary implements Dictionary {
private ByteBuffer offsets;
private ByteBuffer chunks;
- public ChunkedDictionary(SeekableInputStream in, int offset, KeyFactory factory)
+ public ChunkedDictionary(SeekableInputStream in, long offset, KeyFactory factory)
throws IOException {
in.seek(offset);
byte[] headerLengthInBytes = new byte[Integer.BYTES];
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/FixedLengthChunk.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/FixedLengthChunk.java
index 90504d1ad2..d2c3ab970e 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/FixedLengthChunk.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/FixedLengthChunk.java
@@ -38,7 +38,7 @@ public class FixedLengthChunk extends AbstractChunk {
private final int fixedLength;
private int size;
- private int keysBaseOffset;
+ private long keysBaseOffset;
private int keysLength;
private SeekableInputStream in;
private ByteBuffer keys;
@@ -65,7 +65,7 @@ public class FixedLengthChunk extends AbstractChunk {
public FixedLengthChunk(
ByteBuffer headers,
- int keysBaseOffset,
+ long keysBaseOffset,
SeekableInputStream in,
KeyFactory.KeyDeserializer deserializer,
Comparator<Object> comparator) {
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/KeyFactory.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/KeyFactory.java
index e8f4e2897e..f0e985528b 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/KeyFactory.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/KeyFactory.java
@@ -146,7 +146,7 @@ public interface KeyFactory {
Chunk createChunk(Object key, int code, int offset, int limitedSerializedSizeInBytes);
- Chunk mmapChunk(ByteBuffer buffer, int offset, SeekableInputStream in);
+ Chunk mmapChunk(ByteBuffer buffer, long offset, SeekableInputStream in);
/** The key serializer. */
interface KeySerializer {
@@ -176,7 +176,7 @@ public interface KeyFactory {
}
@Override
- public Chunk mmapChunk(ByteBuffer buffer, int offset, SeekableInputStream in) {
+ public Chunk mmapChunk(ByteBuffer buffer, long offset, SeekableInputStream in) {
return new FixedLengthChunk(
buffer, offset, in, createDeserializer(), createComparator());
}
@@ -198,7 +198,7 @@ public interface KeyFactory {
}
@Override
- public Chunk mmapChunk(ByteBuffer buffer, int offset, SeekableInputStream in) {
+ public Chunk mmapChunk(ByteBuffer buffer, long offset, SeekableInputStream in) {
return new VariableLengthChunk(
buffer, offset, in, createDeserializer(), createComparator());
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/VariableLengthChunk.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/VariableLengthChunk.java
index ddec1e446b..1ca17a38f2 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/VariableLengthChunk.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/rangebitmap/dictionary/chunked/VariableLengthChunk.java
@@ -38,7 +38,7 @@ public class VariableLengthChunk extends AbstractChunk {
private int size;
private int currentOffset;
- private int keysBaseOffset;
+ private long keysBaseOffset;
private int offsetsLength;
private int keysLength;
private SeekableInputStream in;
@@ -69,7 +69,7 @@ public class VariableLengthChunk extends AbstractChunk {
public VariableLengthChunk(
ByteBuffer headers,
- int keysBaseOffset,
+ long keysBaseOffset,
SeekableInputStream in,
KeyFactory.KeyDeserializer deserializer,
Comparator<Object> comparator) {
diff --git a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatCompatibilityTest.java b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatCompatibilityTest.java
new file mode 100644
index 0000000000..3511768eb2
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatCompatibilityTest.java
@@ -0,0 +1,309 @@
+/*
+ * 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.fileindex;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.fileindex.bitmap.BitmapFileIndex;
+import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.RoaringBitmap32;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.tools.JavaCompiler;
+import javax.tools.StandardJavaFileManager;
+import javax.tools.ToolProvider;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory.BITMAP_INDEX;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Binary compatibility tests for {@link FileIndexFormat}. */
+public class FileIndexFormatCompatibilityTest {
+
+ @TempDir private Path temporaryDirectory;
+
+ @Test
+ public void testLegacyApi() throws Exception {
+ Class<?> legacyClient = compileLegacyClient();
+ Map<String, Map<String, byte[]>> indexes = new LinkedHashMap<>();
+ indexes.put("f0", Collections.singletonMap("legacy", new byte[] {42}));
+
+ byte[] factoryContainer = write(legacyClient, "writeWithFactory", indexes);
+ assertThat(read(legacyClient, "readWithFactory", factoryContainer)).isEqualTo(42);
+ assertThat(readMetadata(legacyClient, factoryContainer)).isEqualTo(1);
+
+ byte[] constructorContainer = write(legacyClient, "writeWithConstructor", indexes);
+ assertThat(read(legacyClient, "readWithConstructor", constructorContainer)).isEqualTo(42);
+
+ Method createPredicate =
+ legacyClient.getDeclaredMethod(
+ "createPredicate", SeekableInputStream.class, RowType.class);
+ RowType rowType = RowType.builder().field("f0", DataTypes.INT()).build();
+ try (FileIndexPredicate ignored =
+ (FileIndexPredicate)
+ createPredicate.invoke(
+ null, new ByteArraySeekableStream(v1BitmapContainer()), rowType)) {
+ PredicateBuilder predicateBuilder = new PredicateBuilder(rowType);
+ assertThat(((BitmapIndexResult) ignored.evaluate(predicateBuilder.equal(0, 42))).get())
+ .isEqualTo(RoaringBitmap32.bitmapOf(0));
+ assertThat(ignored.evaluate(predicateBuilder.equal(0, 43)).remain()).isFalse();
+ }
+
+ Method coreThreshold = legacyClient.getDeclaredMethod("coreThreshold", CoreOptions.class);
+ assertThat((long) coreThreshold.invoke(null, new CoreOptions(new Options())))
+ .isEqualTo(500L);
+ Method fileIndexThreshold =
+ legacyClient.getDeclaredMethod("fileIndexThreshold", FileIndexOptions.class);
+ assertThat((long) fileIndexThreshold.invoke(null, new FileIndexOptions())).isEqualTo(500L);
+ }
+
+ private byte[] write(
+ Class<?> legacyClient, String methodName, Map<String, Map<String, byte[]>> indexes)
+ throws Exception {
+ Method write = legacyClient.getDeclaredMethod(methodName, OutputStream.class, Map.class);
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ write.invoke(null, output, indexes);
+ return output.toByteArray();
+ }
+
+ private int read(Class<?> legacyClient, String methodName, byte[] container) throws Exception {
+ Method read =
+ legacyClient.getDeclaredMethod(
+ methodName, SeekableInputStream.class, RowType.class);
+ return (int)
+ read.invoke(
+ null, new ByteArraySeekableStream(container), RowType.builder().build());
+ }
+
+ private int readMetadata(Class<?> legacyClient, byte[] container) throws Exception {
+ Method readMetadata =
+ legacyClient.getDeclaredMethod("readMetadata", SeekableInputStream.class);
+ return (int) readMetadata.invoke(null, new ByteArraySeekableStream(container));
+ }
+
+ private byte[] v1BitmapContainer() throws IOException {
+ FileIndexWriter indexWriter =
+ new BitmapFileIndex(DataTypes.INT(), new Options()).createWriter();
+ indexWriter.writeRecord(42);
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, 1)) {
+ writer.writeIndex("f0", BITMAP_INDEX, indexWriter::writeTo);
+ writer.finish();
+ }
+ return output.toByteArray();
+ }
+
+ private Class<?> compileLegacyClient() throws Exception {
+ Path sourceDirectory = temporaryDirectory.resolve("source");
+ Path classesDirectory = temporaryDirectory.resolve("classes");
+ Path oldFormat =
+ sourceDirectory.resolve("org/apache/paimon/fileindex/FileIndexFormat.java");
+ Path oldPredicate =
+ sourceDirectory.resolve("org/apache/paimon/fileindex/FileIndexPredicate.java");
+ Path oldCoreOptions = sourceDirectory.resolve("org/apache/paimon/CoreOptions.java");
+ Path oldFileIndexOptions =
+ sourceDirectory.resolve("org/apache/paimon/fileindex/FileIndexOptions.java");
+ Path legacyClient =
+ sourceDirectory.resolve(
+ "org/apache/paimon/fileindex/compatible/LegacyFileIndexFormatClient.java");
+ Files.createDirectories(oldFormat.getParent());
+ Files.createDirectories(legacyClient.getParent());
+ Files.createDirectories(classesDirectory);
+
+ Files.write(
+ oldFormat,
+ Arrays.asList(
+ "package org.apache.paimon.fileindex;",
+ "import java.io.Closeable;",
+ "import java.io.IOException;",
+ "import java.io.OutputStream;",
+ "import java.util.List;",
+ "import java.util.Map;",
+ "import org.apache.paimon.fs.SeekableInputStream;",
+ "import org.apache.paimon.types.RowType;",
+ "public final class FileIndexFormat {",
+ " public static Writer createWriter(OutputStream output) { return null; }",
+ " public static Reader createReader(SeekableInputStream input, RowType type) { return null; }",
+ " public static Reader createMetadataReader(SeekableInputStream input) { return null; }",
+ " public static class Writer implements Closeable {",
+ " public Writer(OutputStream output) {}",
+ " public void writeColumnIndexes(Map<String, Map<String, byte[]>> indexes)",
+ " throws IOException {}",
+ " public void close() throws IOException {}",
+ " }",
+ " public static class Reader implements Closeable {",
+ " public Reader(SeekableInputStream input, RowType type) {}",
+ " public Map<String, Map<String, byte[]>> readAll() { return null; }",
+ " public List<FileIndexMeta> indexMetas() { return null; }",
+ " public void close() throws IOException {}",
+ " }",
+ " public static class FileIndexMeta {",
+ " public int sizeInBytes() { return 0; }",
+ " }",
+ "}"),
+ StandardCharsets.UTF_8);
+ Files.write(
+ oldPredicate,
+ Arrays.asList(
+ "package org.apache.paimon.fileindex;",
+ "import java.io.Closeable;",
+ "import java.io.IOException;",
+ "import org.apache.paimon.fs.SeekableInputStream;",
+ "import org.apache.paimon.types.RowType;",
+ "public class FileIndexPredicate implements Closeable {",
+ " public FileIndexPredicate(SeekableInputStream input, RowType type) {}",
+ " public void close() throws IOException {}",
+ "}"),
+ StandardCharsets.UTF_8);
+ Files.write(
+ oldCoreOptions,
+ Arrays.asList(
+ "package org.apache.paimon;",
+ "public class CoreOptions {",
+ " public long fileIndexInManifestThreshold() { return 0; }",
+ "}"),
+ StandardCharsets.UTF_8);
+ Files.write(
+ oldFileIndexOptions,
+ Arrays.asList(
+ "package org.apache.paimon.fileindex;",
+ "public class FileIndexOptions {",
+ " public long fileIndexInManifestThreshold() { return 0; }",
+ "}"),
+ StandardCharsets.UTF_8);
+ Files.write(
+ legacyClient,
+ Arrays.asList(
+ "package org.apache.paimon.fileindex.compatible;",
+ "import java.io.IOException;",
+ "import java.io.OutputStream;",
+ "import java.util.Map;",
+ "import org.apache.paimon.CoreOptions;",
+ "import org.apache.paimon.fileindex.FileIndexFormat;",
+ "import org.apache.paimon.fileindex.FileIndexOptions;",
+ "import org.apache.paimon.fileindex.FileIndexPredicate;",
+ "import org.apache.paimon.fs.SeekableInputStream;",
+ "import org.apache.paimon.types.RowType;",
+ "public class LegacyFileIndexFormatClient {",
+ " public static void writeWithFactory(",
+ " OutputStream output, Map<String, Map<String, byte[]>> indexes)",
+ " throws IOException {",
+ " try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output)) {",
+ " writer.writeColumnIndexes(indexes);",
+ " }",
+ " }",
+ " public static void writeWithConstructor(",
+ " OutputStream output, Map<String, Map<String, byte[]>> indexes)",
+ " throws IOException {",
+ " try (FileIndexFormat.Writer writer = new FileIndexFormat.Writer(output)) {",
+ " writer.writeColumnIndexes(indexes);",
+ " }",
+ " }",
+ " public static int readWithFactory(SeekableInputStream input, RowType type)",
+ " throws IOException {",
+ " try (FileIndexFormat.Reader reader = FileIndexFormat.createReader(input, type)) {",
+ " return read(reader);",
+ " }",
+ " }",
+ " public static int readWithConstructor(SeekableInputStream input, RowType type)",
+ " throws IOException {",
+ " try (FileIndexFormat.Reader reader = new FileIndexFormat.Reader(input, type)) {",
+ " return read(reader);",
+ " }",
+ " }",
+ " public static int readMetadata(SeekableInputStream input) throws IOException {",
+ " try (FileIndexFormat.Reader reader = FileIndexFormat.createMetadataReader(input)) {",
+ " return reader.indexMetas().get(0).sizeInBytes();",
+ " }",
+ " }",
+ " private static int read(FileIndexFormat.Reader reader) {",
+ " if (reader.indexMetas().get(0).sizeInBytes() != 1) {",
+ " throw new AssertionError();",
+ " }",
+ " return reader.readAll().get(\"f0\").get(\"legacy\")[0];",
+ " }",
+ " public static FileIndexPredicate createPredicate(",
+ " SeekableInputStream input, RowType type) {",
+ " return new FileIndexPredicate(input, type);",
+ " }",
+ " public static long coreThreshold(CoreOptions options) {",
+ " return options.fileIndexInManifestThreshold();",
+ " }",
+ " public static long fileIndexThreshold(FileIndexOptions options) {",
+ " return options.fileIndexInManifestThreshold();",
+ " }",
+ "}"),
+ StandardCharsets.UTF_8);
+
+ JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
+ assertThat(compiler).as("A JDK compiler is required").isNotNull();
+ try (StandardJavaFileManager fileManager =
+ compiler.getStandardFileManager(null, null, StandardCharsets.UTF_8)) {
+ boolean compiled =
+ compiler.getTask(
+ null,
+ fileManager,
+ null,
+ Arrays.asList(
+ "-d",
+ classesDirectory.toString(),
+ "-source",
+ "8",
+ "-target",
+ "8",
+ "-Xlint:-options"),
+ null,
+ fileManager.getJavaFileObjects(
+ oldFormat.toFile(),
+ oldPredicate.toFile(),
+ oldCoreOptions.toFile(),
+ oldFileIndexOptions.toFile(),
+ legacyClient.toFile()))
+ .call();
+ assertThat(compiled).isTrue();
+ }
+
+ URL[] urls = {classesDirectory.toUri().toURL()};
+ try (URLClassLoader classLoader =
+ new URLClassLoader(urls, FileIndexFormat.class.getClassLoader())) {
+ return classLoader.loadClass(
+ "org.apache.paimon.fileindex.compatible.LegacyFileIndexFormatClient");
+ }
+ }
+}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java
index c6ce94c89a..fdea151ad1 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexFormatFormatTest.java
@@ -18,34 +18,46 @@
package org.apache.paimon.fileindex;
+import org.apache.paimon.fileindex.bitmap.BitmapFileIndex;
+import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
import org.apache.paimon.fileindex.empty.EmptyFileIndexReader;
import org.apache.paimon.fs.ByteArraySeekableStream;
import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.RoaringBitmap32;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.OutputStream;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.paimon.utils.RandomUtil.randomBytes;
import static org.apache.paimon.utils.RandomUtil.randomString;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.tuple;
/** Test for {@link FileIndexFormat}. */
public class FileIndexFormatFormatTest {
private static final Random RANDOM = new Random();
+ private static final RowType ROW_TYPE = RowType.of(DataTypes.INT());
@Test
public void testCreateReaderClosesStreamOnBadMagic() {
@@ -61,7 +73,9 @@ public class FileIndexFormatFormatTest {
};
Throwable thrown =
Assertions.catchThrowable(
- () -> FileIndexFormat.createReader(counting, RowType.builder().build()));
+ () ->
+ FileIndexFormat.createReader(
+ counting, RowType.builder().build(), notIndexFile.length));
// A throwing constructor never assigns the caller's try-with-resources resource,
// so closing the stream is the constructor's job.
Assertions.assertThat(closed.get()).isTrue();
@@ -73,8 +87,6 @@ public class FileIndexFormatFormatTest {
@Test
public void testWriteRead() throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos);
-
Map<String, Map<String, byte[]>> indexes = new HashMap<>();
for (int j = 0; j < RANDOM.nextInt(1000); j++) {
String type = randomString(RANDOM.nextInt(100));
@@ -85,22 +97,33 @@ public class FileIndexFormatFormatTest {
}
}
- writer.writeColumnIndexes(indexes);
- writer.close();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos, 1)) {
+ for (Map.Entry<String, Map<String, byte[]>> column : indexes.entrySet()) {
+ for (Map.Entry<String, byte[]> index : column.getValue().entrySet()) {
+ writer.writeIndex(
+ column.getKey(),
+ index.getKey(),
+ output -> output.write(index.getValue()));
+ }
+ }
+ writer.finish();
+ }
byte[] indexBytes = baos.toByteArray();
FileIndexFormat.Reader reader =
FileIndexFormat.createReader(
- new ByteArraySeekableStream(indexBytes), RowType.builder().build());
+ new ByteArraySeekableStream(indexBytes),
+ RowType.builder().build(),
+ indexBytes.length);
for (Map.Entry<String, Map<String, byte[]>> entry : indexes.entrySet()) {
String column = entry.getKey();
for (String type : entry.getValue().keySet()) {
- byte[] b =
- reader.getBytesWithNameAndType(column, type)
- .orElseThrow(RuntimeException::new);
- Assertions.assertThat(b).containsExactly(indexes.get(column).get(type));
+ ByteArrayOutputStream copied = new ByteArrayOutputStream();
+ reader.copyPayload(column, type, copied);
+ Assertions.assertThat(copied.toByteArray())
+ .containsExactly(indexes.get(column).get(type));
}
}
}
@@ -108,15 +131,11 @@ public class FileIndexFormatFormatTest {
@Test
public void testEmptyFileIndex() throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos);
-
- Map<String, Map<String, byte[]>> indexes = new HashMap<>();
-
- indexes.computeIfAbsent("a", a -> new HashMap<>()).put("b", null);
- indexes.computeIfAbsent("a", a -> new HashMap<>()).put("c", null);
-
- writer.writeColumnIndexes(indexes);
- writer.close();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos, 1)) {
+ writer.writeIndex("a", "b", null);
+ writer.writeIndex("a", "c", null);
+ writer.finish();
+ }
byte[] indexBytes = baos.toByteArray();
@@ -126,7 +145,8 @@ public class FileIndexFormatFormatTest {
RowType.builder()
.field("a", DataTypes.BYTES())
.field("b", DataTypes.STRING())
- .build());
+ .build(),
+ indexBytes.length);
Collection<FileIndexReader> fileIndexFormatList = reader.readColumnIndex("a");
Assertions.assertThat(fileIndexFormatList.size()).isEqualTo(1);
@@ -137,21 +157,17 @@ public class FileIndexFormatFormatTest {
@Test
public void testIndexMetas() throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- Map<String, Map<String, byte[]>> indexes = new LinkedHashMap<>();
- indexes.computeIfAbsent("user_id", key -> new LinkedHashMap<>())
- .put("bitmap", new byte[] {1, 2, 3});
- indexes.computeIfAbsent("user_id", key -> new LinkedHashMap<>())
- .put("bloom-filter", new byte[] {4, 5});
- indexes.computeIfAbsent("region", key -> new LinkedHashMap<>()).put("bitmap", null);
-
- try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos)) {
- writer.writeColumnIndexes(indexes);
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos, 1)) {
+ writer.writeIndex("user_id", "bitmap", output -> output.write(new byte[] {1, 2, 3}));
+ writer.writeIndex("user_id", "bloom-filter", output -> output.write(new byte[] {4, 5}));
+ writer.writeIndex("region", "bitmap", null);
+ writer.finish();
}
List<FileIndexFormat.FileIndexMeta> metas;
try (FileIndexFormat.Reader reader =
FileIndexFormat.createMetadataReader(
- new ByteArraySeekableStream(baos.toByteArray()))) {
+ new ByteArraySeekableStream(baos.toByteArray()), baos.size())) {
metas = reader.indexMetas();
}
@@ -159,13 +175,249 @@ public class FileIndexFormatFormatTest {
.extracting(
FileIndexFormat.FileIndexMeta::columnName,
FileIndexFormat.FileIndexMeta::indexType,
- FileIndexFormat.FileIndexMeta::sizeInBytes,
+ FileIndexFormat.FileIndexMeta::sizeInBytesLong,
FileIndexFormat.FileIndexMeta::empty)
.containsExactlyInAnyOrder(
- tuple("user_id", "bitmap", 3, false),
- tuple("user_id", "bloom-filter", 2, false),
- tuple("region", "bitmap", 0, true));
+ tuple("user_id", "bitmap", 3L, false),
+ tuple("user_id", "bloom-filter", 2L, false),
+ tuple("region", "bitmap", 0L, true));
Assertions.assertThatThrownBy(() -> metas.clear())
.isInstanceOf(UnsupportedOperationException.class);
}
+
+ @Test
+ public void testV2TotalSizeOverTwoGiBAndLegacyBackendAtLargePosition() throws Exception {
+ SparseOutput output = new SparseOutput();
+ byte[] payload = new byte[1024 * 1024];
+ BitmapFileIndex indexer = new BitmapFileIndex(DataTypes.INT(), new Options());
+ FileIndexWriter bitmap = indexer.createWriter();
+ bitmap.writeRecord(42);
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, 2)) {
+ for (int i = 0; i < 2049; i++) {
+ writer.writeIndex(
+ "large", String.format("part%04d", i), stream -> stream.write(payload));
+ }
+ writer.writeIndex("f0", "bitmap", bitmap::writeTo);
+ writer.finish();
+ }
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createReader(output.input(), ROW_TYPE, output.position)) {
+ assertThat(output.position).isGreaterThan(Integer.MAX_VALUE);
+ FileIndexReader bitmapReader = reader.readColumnIndex("f0").iterator().next();
+ assertThat(
+ ((BitmapIndexResult)
+ bitmapReader.visitEqual(
+ new FieldRef(0, "f0", DataTypes.INT()), 42))
+ .get())
+ .isEqualTo(RoaringBitmap32.bitmapOf(0));
+ }
+ }
+
+ @Test
+ public void testV2StreamedPayloadOverInt32() throws Exception {
+ SparseOutput output = new SparseOutput();
+ byte[] block = new byte[1024 * 1024];
+ BitmapFileIndex indexer = new BitmapFileIndex(DataTypes.INT(), new Options());
+ FileIndexWriter bitmap = indexer.createWriter();
+ bitmap.writeRecord(42);
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, 2)) {
+ writer.writeIndex(
+ "large",
+ "bitmap",
+ stream -> {
+ for (int i = 0; i < 2049; i++) {
+ stream.write(block);
+ }
+ });
+ writer.writeIndex("f0", "bitmap", bitmap::writeTo);
+ writer.finish();
+ }
+
+ RowType rowType =
+ RowType.builder()
+ .field("large", DataTypes.INT())
+ .field("f0", DataTypes.INT())
+ .build();
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createReader(output.input(), rowType, output.position)) {
+ assertThat(reader.indexMetas())
+ .extracting(FileIndexFormat.FileIndexMeta::sizeInBytesLong)
+ .contains((long) block.length * 2049);
+ FileIndexReader bitmapReader = reader.readColumnIndex("f0").iterator().next();
+ assertThat(
+ ((BitmapIndexResult)
+ bitmapReader.visitEqual(
+ new FieldRef(1, "f0", DataTypes.INT()), 42))
+ .get())
+ .isEqualTo(RoaringBitmap32.bitmapOf(0));
+ }
+ }
+
+ @Test
+ public void testV2CopyPayloadOverInt32() throws Exception {
+ SparseOutput source = new SparseOutput();
+ byte[] block = new byte[1024 * 1024];
+ long payloadLength = (long) block.length * 2049 + 1;
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(source, 2)) {
+ writer.writeIndex(
+ "large",
+ "bitmap",
+ stream -> {
+ for (int i = 0; i < 2049; i++) {
+ stream.write(block);
+ }
+ stream.write(7);
+ });
+ writer.finish();
+ }
+
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createMetadataReader(source.input(), source.position)) {
+ final long[] copied = {0};
+ final int[] lastByte = {-1};
+ reader.copyPayload(
+ "large",
+ "bitmap",
+ new OutputStream() {
+ @Override
+ public void write(int value) {
+ copied[0]++;
+ lastByte[0] = value;
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length) {
+ copied[0] += length;
+ lastByte[0] = bytes[offset + length - 1] & 0xff;
+ }
+ });
+ assertThat(copied[0]).isEqualTo(payloadLength);
+ assertThat(lastByte[0]).isEqualTo(7);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void testIncrementalWriteAndCopyPayload(int version) throws Exception {
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, version)) {
+ writer.writeIndex("a", "bitmap", stream -> stream.write(new byte[] {1, 2, 3}));
+ writer.writeIndex("a", "bsi", null);
+ writer.writeIndex("b", "bitmap", stream -> {});
+ writer.finish();
+ }
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createMetadataReader(
+ new ByteArraySeekableStream(output.toByteArray()), output.size())) {
+ ByteArrayOutputStream copied = new ByteArrayOutputStream();
+ reader.copyPayload("a", "bitmap", copied);
+ assertThat(copied.toByteArray()).containsExactly(1, 2, 3);
+ assertThat(reader.indexMetas())
+ .extracting(FileIndexFormat.FileIndexMeta::empty)
+ .containsExactlyInAnyOrder(false, true, false);
+ }
+ }
+
+ @Test
+ public void testV2RejectInvalidTailMagic() throws Exception {
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, 2)) {
+ writer.writeIndex("f0", "bitmap", stream -> stream.write(new byte[] {1, 2, 3}));
+ writer.finish();
+ }
+ byte[] bytes = output.toByteArray();
+ bytes[bytes.length - 1] = 0;
+ assertThatThrownBy(
+ () ->
+ FileIndexFormat.createReader(
+ new ByteArraySeekableStream(bytes), ROW_TYPE, bytes.length))
+ .rootCause()
+ .hasMessageContaining("Invalid file index tail magic");
+ }
+
+ /**
+ * Retains small metadata writes while large zero-filled payload writes only advance position.
+ */
+ private static class SparseOutput extends OutputStream {
+ private final TreeMap<Long, byte[]> segments = new TreeMap<>();
+ private long position;
+
+ @Override
+ public void write(int value) {
+ write(new byte[] {(byte) value}, 0, 1);
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length) {
+ if (length < 1024) {
+ segments.put(position, Arrays.copyOfRange(bytes, offset, offset + length));
+ }
+ position += length;
+ }
+
+ private SeekableInputStream input() {
+ return new SeekableInputStream() {
+ private long cursor;
+
+ @Override
+ public void seek(long desired) {
+ cursor = desired;
+ }
+
+ @Override
+ public long getPos() {
+ return cursor;
+ }
+
+ @Override
+ public int read() {
+ if (cursor >= position) {
+ return -1;
+ }
+ Map.Entry<Long, byte[]> entry = segments.floorEntry(cursor);
+ int value =
+ entry != null && cursor - entry.getKey() < entry.getValue().length
+ ? entry.getValue()[(int) (cursor - entry.getKey())] & 0xff
+ : 0;
+ cursor++;
+ return value;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ if (len == 0) {
+ return 0;
+ }
+ if (cursor >= position) {
+ return -1;
+ }
+ int n = (int) Math.min(len, position - cursor);
+ long end = cursor + n;
+ Arrays.fill(b, off, off + n, (byte) 0);
+ Map.Entry<Long, byte[]> entry = segments.floorEntry(cursor);
+ if (entry == null) {
+ entry = segments.ceilingEntry(cursor);
+ }
+ while (entry != null && entry.getKey() < end) {
+ long overlapStart = Math.max(cursor, entry.getKey());
+ long overlapEnd = Math.min(end, entry.getKey() + entry.getValue().length);
+ if (overlapStart < overlapEnd) {
+ System.arraycopy(
+ entry.getValue(),
+ (int) (overlapStart - entry.getKey()),
+ b,
+ off + (int) (overlapStart - cursor),
+ (int) (overlapEnd - overlapStart));
+ }
+ entry = segments.higherEntry(entry.getKey());
+ }
+ cursor = end;
+ return n;
+ }
+
+ @Override
+ public void close() {}
+ };
+ }
+ }
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateCloseTest.java b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateCloseTest.java
index 9e84dc7d62..31d4ca2103 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateCloseTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateCloseTest.java
@@ -19,23 +19,76 @@
package org.apache.paimon.fileindex;
import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
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;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Tests that {@link FileIndexPredicate} always releases the stream it is handed. */
public class FileIndexPredicateCloseTest {
private static final RowType ROW_TYPE = RowType.of(DataTypes.INT());
+ @Test
+ public void testV1DoesNotLoadFileStatus() throws IOException {
+ FileIO fileIO = mock(FileIO.class);
+ Path path = new Path("file:/index");
+ when(fileIO.newInputStream(path)).thenReturn(new ByteArraySeekableStream(container(1)));
+
+ try (FileIndexPredicate ignored = new FileIndexPredicate(path, fileIO, ROW_TYPE)) {
+ verify(fileIO).newInputStream(path);
+ verify(fileIO, never()).getFileStatus(path);
+ }
+ }
+
+ @Test
+ public void testV2LoadsFileStatus() throws IOException {
+ FileIO fileIO = mock(FileIO.class);
+ Path path = new Path("file:/index");
+ byte[] container = container(2);
+ FileStatus fileStatus = mock(FileStatus.class);
+ when(fileIO.newInputStream(path)).thenReturn(new ByteArraySeekableStream(container));
+ when(fileIO.getFileStatus(path)).thenReturn(fileStatus);
+ when(fileStatus.getLen()).thenReturn((long) container.length);
+
+ try (FileIndexPredicate ignored = new FileIndexPredicate(path, fileIO, ROW_TYPE)) {
+ verify(fileIO).newInputStream(path);
+ verify(fileIO).getFileStatus(path);
+ }
+ }
+
+ @Test
+ public void testV2FileStatusFailureReleasesStream() throws IOException {
+ FileIO fileIO = mock(FileIO.class);
+ Path path = new Path("file:/index");
+ AtomicInteger closed = new AtomicInteger();
+ when(fileIO.newInputStream(path)).thenReturn(tracking(container(2), closed));
+ IOException exception = new IOException("Failed to get file status");
+ when(fileIO.getFileStatus(path)).thenThrow(exception);
+
+ assertThatThrownBy(() -> new FileIndexPredicate(path, fileIO, ROW_TYPE))
+ .isSameAs(exception);
+
+ verify(fileIO).newInputStream(path);
+ verify(fileIO).getFileStatus(path);
+ assertThat(closed).hasValue(1);
+ }
+
/**
* The reader rejects a file whose magic does not match, and at that point nothing else holds a
* reference to the stream, so the constructor has to release it itself.
@@ -45,7 +98,12 @@ public class FileIndexPredicateCloseTest {
AtomicInteger closed = new AtomicInteger();
byte[] notAnIndexFile = new byte[64];
- assertThatThrownBy(() -> new FileIndexPredicate(tracking(notAnIndexFile, closed), ROW_TYPE))
+ assertThatThrownBy(
+ () ->
+ new FileIndexPredicate(
+ tracking(notAnIndexFile, closed),
+ ROW_TYPE,
+ notAnIndexFile.length))
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("Exception happens while construct file index reader.")
.hasRootCauseMessage("This file is not file index file.");
@@ -84,4 +142,12 @@ public class FileIndexPredicateCloseTest {
}
};
}
+
+ private static byte[] container(int version) throws IOException {
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, version)) {
+ writer.finish();
+ }
+ return output.toByteArray();
+ }
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateTest.java b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateTest.java
index e599c354ab..35e42ad14f 100644
--- a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexPredicateTest.java
@@ -36,8 +36,6 @@ import java.io.ByteArrayOutputStream;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
import java.util.Set;
import static org.apache.paimon.fileindex.FileIndexResult.REMAIN;
@@ -67,13 +65,10 @@ public class FileIndexPredicateTest {
new BloomFilterFileIndex(DataTypes.DOUBLE(), new Options()).createWriter();
indexWriter.writeRecord(Double.NaN);
- Map<String, Map<String, byte[]>> indexes = new HashMap<>();
- indexes.computeIfAbsent("d", column -> new HashMap<>())
- .put(BLOOM_FILTER, indexWriter.serializedBytes());
-
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos)) {
- writer.writeColumnIndexes(indexes);
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos, 1)) {
+ writer.writeIndex("d", BLOOM_FILTER, indexWriter::writeTo);
+ writer.finish();
}
try (FileIndexPredicate predicate = new FileIndexPredicate(baos.toByteArray(), rowType)) {
@@ -91,9 +86,9 @@ public class FileIndexPredicateTest {
private static FileIndexPredicate emptyFileIndexPredicate() throws Exception {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
- FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos);
- writer.writeColumnIndexes(new HashMap<String, java.util.Map<String, byte[]>>());
- writer.close();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(baos, 1)) {
+ writer.finish();
+ }
return new FileIndexPredicate(baos.toByteArray(), RowType.builder().build());
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexerCompatibilityTest.java b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexerCompatibilityTest.java
new file mode 100644
index 0000000000..83045a8fb7
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fileindex/FileIndexerCompatibilityTest.java
@@ -0,0 +1,187 @@
+/*
+ * 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.fileindex;
+
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataType;
+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 javax.tools.JavaCompiler;
+import javax.tools.StandardJavaFileManager;
+import javax.tools.ToolProvider;
+
+import java.io.ByteArrayOutputStream;
+import java.lang.reflect.Field;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Binary compatibility tests for {@link FileIndexer}. */
+public class FileIndexerCompatibilityTest {
+
+ private static final String LEGACY_INDEX = "legacy-binary";
+
+ @TempDir private Path temporaryDirectory;
+
+ @Test
+ public void testLegacyBinaryReadsV1Container() throws Exception {
+ FileIndexer legacyIndexer = compileLegacyIndexer();
+ FileIndexerFactory legacyFactory =
+ new FileIndexerFactory() {
+ @Override
+ public String identifier() {
+ return LEGACY_INDEX;
+ }
+
+ @Override
+ public FileIndexer create(DataType type, Options options) {
+ return legacyIndexer;
+ }
+ };
+
+ Field factoriesField = FileIndexerFactoryUtils.class.getDeclaredField("factories");
+ factoriesField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<String, FileIndexerFactory> factories =
+ (Map<String, FileIndexerFactory>) factoriesField.get(null);
+ FileIndexerFactory previous = factories.put(LEGACY_INDEX, legacyFactory);
+ try {
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, 1)) {
+ writer.writeIndex("f0", LEGACY_INDEX, stream -> stream.write(42));
+ writer.finish();
+ }
+
+ byte[] bytes = output.toByteArray();
+ RowType rowType = RowType.builder().field("f0", DataTypes.INT()).build();
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createReader(
+ new ByteArraySeekableStream(bytes), rowType, bytes.length)) {
+ assertThat(reader.readColumnIndex("f0")).hasSize(1);
+ }
+ } finally {
+ if (previous == null) {
+ factories.remove(LEGACY_INDEX);
+ } else {
+ factories.put(LEGACY_INDEX, previous);
+ }
+ }
+ }
+
+ private FileIndexer compileLegacyIndexer() throws Exception {
+ Path sourceDirectory = temporaryDirectory.resolve("source");
+ Path classesDirectory = temporaryDirectory.resolve("classes");
+ Path oldInterface = sourceDirectory.resolve("org/apache/paimon/fileindex/FileIndexer.java");
+ Path legacyImplementation =
+ sourceDirectory.resolve(
+ "org/apache/paimon/fileindex/compatible/LegacyFileIndexer.java");
+ Files.createDirectories(oldInterface.getParent());
+ Files.createDirectories(legacyImplementation.getParent());
+ Files.createDirectories(classesDirectory);
+
+ Files.write(
+ oldInterface,
+ Arrays.asList(
+ "package org.apache.paimon.fileindex;",
+ "import org.apache.paimon.fs.SeekableInputStream;",
+ "public interface FileIndexer {",
+ " FileIndexWriter createWriter();",
+ " FileIndexReader createReader(",
+ " SeekableInputStream inputStream, int start, int length);",
+ "}"),
+ StandardCharsets.UTF_8);
+ Files.write(
+ legacyImplementation,
+ Arrays.asList(
+ "package org.apache.paimon.fileindex.compatible;",
+ "import org.apache.paimon.fileindex.FileIndexReader;",
+ "import org.apache.paimon.fileindex.FileIndexer;",
+ "import org.apache.paimon.fileindex.FileIndexWriter;",
+ "import org.apache.paimon.fs.SeekableInputStream;",
+ "public class LegacyFileIndexer implements FileIndexer {",
+ " public FileIndexWriter createWriter() { return null; }",
+ " public FileIndexReader createReader(",
+ " SeekableInputStream input, int start, int length) {",
+ " try {",
+ " input.seek(start);",
+ " if (length != 1 || input.read() != 42) {",
+ " throw new RuntimeException(\"Unexpected legacy payload\");",
+ " }",
+ " return new FileIndexReader() {};",
+ " } catch (java.io.IOException e) {",
+ " throw new RuntimeException(e);",
+ " }",
+ " }",
+ "}"),
+ StandardCharsets.UTF_8);
+
+ JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
+ assertThat(compiler).as("A JDK compiler is required").isNotNull();
+ try (StandardJavaFileManager fileManager =
+ compiler.getStandardFileManager(null, null, StandardCharsets.UTF_8)) {
+ boolean compiled =
+ compiler.getTask(
+ null,
+ fileManager,
+ null,
+ Arrays.asList(
+ "-classpath",
+ System.getProperty("java.class.path"),
+ "-d",
+ classesDirectory.toString(),
+ "-source",
+ "8",
+ "-target",
+ "8",
+ "-Xlint:-options"),
+ null,
+ fileManager.getJavaFileObjects(
+ oldInterface.toFile(), legacyImplementation.toFile()))
+ .call();
+ assertThat(compiled).isTrue();
+ }
+
+ URL[] urls = {classesDirectory.toUri().toURL()};
+ try (URLClassLoader classLoader =
+ new URLClassLoader(urls, FileIndexer.class.getClassLoader())) {
+ Class<?> legacyClass =
+ classLoader.loadClass(
+ "org.apache.paimon.fileindex.compatible.LegacyFileIndexer");
+ assertThat(
+ legacyClass.getDeclaredMethod(
+ "createReader",
+ org.apache.paimon.fs.SeekableInputStream.class,
+ int.class,
+ int.class))
+ .isNotNull();
+ return (FileIndexer) legacyClass.getDeclaredConstructor().newInstance();
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java b/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java
index 3030ac3bb2..21dfd98300 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/FileIndexProcessor.java
@@ -28,6 +28,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileIndexWriter;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.SpillableIndexOutputStream;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.reader.RecordReader;
@@ -42,9 +43,7 @@ import org.apache.paimon.utils.FileStorePathFactory;
import javax.annotation.Nullable;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
-import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -67,7 +66,7 @@ public class FileIndexProcessor {
private final FileStorePathFactory pathFactory;
private final DataFilePathFactories pathFactories;
private final SchemaCache schemaInfoCache;
- private final long sizeInMeta;
+ private final int sizeInMeta;
public FileIndexProcessor(FileStoreTable table) {
this.table = table;
@@ -76,7 +75,7 @@ public class FileIndexProcessor {
this.pathFactory = table.store().pathFactory();
this.pathFactories = new DataFilePathFactories(pathFactory);
this.schemaInfoCache = new SchemaCache(fileIndexOptions, table.schemaManager());
- this.sizeInMeta = table.coreOptions().fileIndexInManifestThreshold();
+ this.sizeInMeta = Math.toIntExact(table.coreOptions().fileIndexInManifestThreshold());
}
public DataFileMeta process(BinaryRow partition, int bucket, ManifestEntry manifestEntry)
@@ -90,102 +89,113 @@ public class FileIndexProcessor {
.filter(name -> name.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX))
.collect(Collectors.toList());
extras.removeAll(indexFiles);
-
Path newIndexPath;
- Map<String, Map<String, byte[]>> maintainers;
- // load
+ FileIndexFormat.Reader sourceReader = null;
if (!indexFiles.isEmpty()) {
- String indexFile = indexFiles.get(0);
- try (FileIndexFormat.Reader indexReader =
+ Path sourcePath = dataFilePathFactory.toAlignedPath(indexFiles.get(0), dataFileMeta);
+ long sourceLength = fileIO.getFileStatus(sourcePath).getLen();
+ sourceReader =
FileIndexFormat.createReader(
- fileIO.newInputStream(
- dataFilePathFactory.toAlignedPath(indexFile, dataFileMeta)),
- schemaInfo.fileSchema)) {
- maintainers = indexReader.readAll();
- }
- newIndexPath =
- createNewFileIndexFilePath(
- dataFilePathFactory.toAlignedPath(indexFile, dataFileMeta));
+ fileIO.newInputStream(sourcePath), schemaInfo.fileSchema, sourceLength);
+ newIndexPath = createNewFileIndexFilePath(sourcePath);
} else {
- maintainers = new HashMap<>();
newIndexPath = dataFileToFileIndexPath(dataFilePathFactory.toPath(dataFileMeta));
}
- // remove unnecessary
- for (Map.Entry<String, Map<String, byte[]>> entry : new HashSet<>(maintainers.entrySet())) {
- String name = entry.getKey();
- if (!schemaInfo.projectedColFullNames.contains(name)) {
- maintainers.remove(name);
- } else {
- Map<String, byte[]> indexTypeBytes = maintainers.get(name);
- Set<String> configuredIndexTypes =
- schemaInfo.projectedIndexTypes.getOrDefault(name, Collections.emptySet());
- indexTypeBytes
- .keySet()
- .removeIf(indexType -> !configuredIndexTypes.contains(indexType));
+ try (FileIndexFormat.Reader oldReader = sourceReader) {
+ Map<String, Map<String, FileIndexFormat.Payload>> entries = new HashMap<>();
+ if (oldReader != null) {
+ for (FileIndexFormat.FileIndexMeta meta : oldReader.indexMetas()) {
+ String column = meta.columnName();
+ String type = meta.indexType();
+ if (schemaInfo.projectedColFullNames.contains(column)
+ && schemaInfo
+ .projectedIndexTypes
+ .getOrDefault(column, Collections.emptySet())
+ .contains(type)) {
+ entries.computeIfAbsent(column, ignored -> new HashMap<>())
+ .put(
+ type,
+ meta.empty()
+ ? null
+ : output ->
+ oldReader.copyPayload(
+ column, type, output));
+ }
+ }
}
- }
- // ignore close, do not close to write file, only collect serialized maintainers
- @SuppressWarnings("resource")
- DataFileIndexWriter dataFileIndexWriter =
- DataFileIndexWriter.create(
- fileIO,
- newIndexPath,
- schemaInfo.fileSchema.project(schemaInfo.projectedIndexCols),
- fileIndexOptions,
- schemaInfo.colNameMapping);
- if (dataFileIndexWriter != null) {
- // projectedIndexCols index into the file schema. withProjection would re-interpret
- // them against the current table schema, so a schema change that shifts columns (drop
- // a middle column, add another) would read the wrong column and rebuild the index over
- // it. Read with the same file-schema projection the writer above uses.
- RowType indexReadType = schemaInfo.fileSchema.project(schemaInfo.projectedIndexCols);
- try (RecordReader<InternalRow> reader =
- table.newReadBuilder()
- .withReadType(indexReadType)
- .newRead()
- .createReader(
- DataSplit.builder()
- .withPartition(partition)
- .withBucket(bucket)
- .withBucketPath(
- pathFactory
- .bucketPath(partition, bucket)
- .toString())
- .withTotalBuckets(manifestEntry.totalBuckets())
- .withDataFiles(Collections.singletonList(dataFileMeta))
- .rawConvertible(true)
- .build())) {
- reader.forEachRemaining(dataFileIndexWriter::write);
+ // Collect the new writers, not their serialized payloads.
+ @SuppressWarnings("resource")
+ DataFileIndexWriter newIndexes =
+ DataFileIndexWriter.create(
+ fileIO,
+ newIndexPath,
+ schemaInfo.fileSchema.project(schemaInfo.projectedIndexCols),
+ fileIndexOptions,
+ schemaInfo.colNameMapping);
+ if (newIndexes != null) {
+ // projectedIndexCols index into the file schema. withProjection would re-interpret
+ // them against the current table schema, so a schema change that shifts columns
+ // would rebuild the index over the wrong column.
+ RowType indexReadType =
+ schemaInfo.fileSchema.project(schemaInfo.projectedIndexCols);
+ try (RecordReader<InternalRow> reader =
+ table.newReadBuilder()
+ .withReadType(indexReadType)
+ .newRead()
+ .createReader(
+ DataSplit.builder()
+ .withPartition(partition)
+ .withBucket(bucket)
+ .withBucketPath(
+ pathFactory
+ .bucketPath(partition, bucket)
+ .toString())
+ .withTotalBuckets(manifestEntry.totalBuckets())
+ .withDataFiles(
+ Collections.singletonList(dataFileMeta))
+ .rawConvertible(true)
+ .build())) {
+ reader.forEachRemaining(newIndexes::write);
+ }
+ newIndexes.forEachIndex(
+ (column, type, writer) ->
+ entries.computeIfAbsent(column, ignored -> new HashMap<>())
+ .put(type, writer == null ? null : writer::writeTo));
}
- dataFileIndexWriter
- .serializeMaintainers()
- .forEach(
- (key, value) ->
- maintainers
- .computeIfAbsent(key, k -> new HashMap<>())
- .putAll(value));
- }
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- try (FileIndexFormat.Writer indexWriter = FileIndexFormat.createWriter(baos)) {
- if (!maintainers.isEmpty()) {
- indexWriter.writeColumnIndexes(maintainers);
+ if (entries.isEmpty()) {
+ return dataFileMeta.copy(extras).copy((byte[]) null);
}
- }
- if (baos.size() > sizeInMeta) {
- try (OutputStream outputStream = fileIO.newOutputStream(newIndexPath, true)) {
- outputStream.write(baos.toByteArray());
+ SpillableIndexOutputStream output =
+ new SpillableIndexOutputStream(fileIO, newIndexPath, sizeInMeta);
+ try {
+ try (FileIndexFormat.Writer writer =
+ FileIndexFormat.createWriter(output, fileIndexOptions.formatVersion())) {
+ for (Map.Entry<String, Map<String, FileIndexFormat.Payload>> column :
+ entries.entrySet()) {
+ for (Map.Entry<String, FileIndexFormat.Payload> index :
+ column.getValue().entrySet()) {
+ writer.writeIndex(column.getKey(), index.getKey(), index.getValue());
+ }
+ }
+ writer.finish();
+ }
+ if (output.spilled()) {
+ extras.add(newIndexPath.getName());
+ return dataFileMeta.copy(extras).copy((byte[]) null);
+ }
+ return dataFileMeta.copy(extras).copy(output.embeddedBytes());
+ } catch (IOException | RuntimeException e) {
+ try {
+ output.abort();
+ } catch (IOException cleanupFailure) {
+ e.addSuppressed(cleanupFailure);
+ }
+ throw e;
}
- extras.add(newIndexPath.getName());
- return dataFileMeta.copy(extras).copy((byte[]) null);
- } else if (baos.size() == 0) {
- return dataFileMeta.copy(extras).copy((byte[]) null);
- } else {
- return dataFileMeta.copy(extras).copy(baos.toByteArray());
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/DataFileIndexWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/DataFileIndexWriter.java
index 2c620dd143..6c122b1a44 100644
--- a/paimon-core/src/main/java/org/apache/paimon/io/DataFileIndexWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/io/DataFileIndexWriter.java
@@ -37,11 +37,8 @@ import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
-import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
-import java.io.OutputStream;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -58,7 +55,8 @@ public final class DataFileIndexWriter implements Closeable {
private final Path path;
// if the filter size greater than fileIndexInManifestThreshold, we put it in file
- private final long inManifestThreshold;
+ private final int inManifestThreshold;
+ private final int formatVersion;
// index type, column name -> index maintainer
private final Map<String, Map<String, IndexMaintainer>> indexMaintainers = new HashMap<>();
@@ -145,7 +143,8 @@ public final class DataFileIndexWriter implements Closeable {
}
}
}
- this.inManifestThreshold = fileIndexOptions.fileIndexInManifestThreshold();
+ this.inManifestThreshold = Math.toIntExact(fileIndexOptions.fileIndexInManifestThreshold());
+ this.formatVersion = fileIndexOptions.formatVersion();
}
public void write(InternalRow row) {
@@ -158,36 +157,57 @@ public final class DataFileIndexWriter implements Closeable {
@Override
public void close() throws IOException {
- Map<String, Map<String, byte[]>> indexMaps = serializeMaintainers();
-
- ByteArrayOutputStream out = new ByteArrayOutputStream();
- try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(out)) {
- writer.writeColumnIndexes(indexMaps);
- }
-
- if (out.size() > inManifestThreshold) {
- try (OutputStream outputStream = fileIO.newOutputStream(path, true)) {
- outputStream.write(out.toByteArray());
+ SpillableIndexOutputStream output =
+ new SpillableIndexOutputStream(fileIO, path, inManifestThreshold);
+ try {
+ try (FileIndexFormat.Writer writer =
+ FileIndexFormat.createWriter(output, formatVersion)) {
+ forEachIndex(
+ (column, type, indexWriter) ->
+ writer.writeIndex(
+ column,
+ type,
+ indexWriter == null ? null : indexWriter::writeTo));
+ writer.finish();
+ }
+ if (output.spilled()) {
+ resultFileName = path.getName();
+ } else {
+ embeddedIndexBytes = output.embeddedBytes();
}
- resultFileName = path.getName();
- } else {
- embeddedIndexBytes = out.toByteArray();
+ } catch (IOException | RuntimeException e) {
+ try {
+ output.abort();
+ } catch (IOException cleanupFailure) {
+ e.addSuppressed(cleanupFailure);
+ }
+ throw e;
}
}
- public Map<String, Map<String, byte[]>> serializeMaintainers() {
- Map<String, Map<String, byte[]>> indexMaps = new HashMap<>();
- for (Map<String, IndexMaintainer> columnIndexMaintainers : indexMaintainers.values()) {
- for (IndexMaintainer indexMaintainer : columnIndexMaintainers.values()) {
- Map<String, byte[]> mapBytes = indexMaintainer.serializedBytes();
- for (Map.Entry<String, byte[]> entry : mapBytes.entrySet()) {
- indexMaps
- .computeIfAbsent(entry.getKey(), k -> new HashMap<>())
- .put(indexMaintainer.getIndexType(), entry.getValue());
- }
+ /** Visits the index writers without serializing their payloads. */
+ public void forEachIndex(IndexConsumer consumer) throws IOException {
+ Map<String, Map<String, FileIndexWriter>> entries = new HashMap<>();
+ for (Map<String, IndexMaintainer> columnMaintainers : indexMaintainers.values()) {
+ for (IndexMaintainer maintainer : columnMaintainers.values()) {
+ maintainer.forEachIndex(
+ (column, type, writer) ->
+ entries.computeIfAbsent(column, ignored -> new HashMap<>())
+ .put(type, writer));
+ }
+ }
+ for (Map.Entry<String, Map<String, FileIndexWriter>> column : entries.entrySet()) {
+ for (Map.Entry<String, FileIndexWriter> index : column.getValue().entrySet()) {
+ consumer.accept(column.getKey(), index.getKey(), index.getValue());
}
}
- return indexMaps;
+ }
+
+ /** Consumes the writer for one column and index type. */
+ @FunctionalInterface
+ public interface IndexConsumer {
+ void accept(String columnName, String indexType, @Nullable FileIndexWriter writer)
+ throws IOException;
}
public FileIndexResult result() {
@@ -241,9 +261,7 @@ public final class DataFileIndexWriter implements Closeable {
void write(InternalRow row);
- String getIndexType();
-
- Map<String, byte[]> serializedBytes();
+ void forEachIndex(IndexConsumer consumer) throws IOException;
}
/** One index maintainer for one column. */
@@ -269,12 +287,8 @@ public final class DataFileIndexWriter implements Closeable {
fileIndexWriter.writeRecord(getter.getFieldOrNull(row));
}
- public String getIndexType() {
- return indexType;
- }
-
- public Map<String, byte[]> serializedBytes() {
- return Collections.singletonMap(columnName, fileIndexWriter.serializedBytes());
+ public void forEachIndex(IndexConsumer consumer) throws IOException {
+ consumer.accept(columnName, indexType, fileIndexWriter);
}
}
@@ -348,22 +362,14 @@ public final class DataFileIndexWriter implements Closeable {
.createWriter());
}
- public String getIndexType() {
- return indexType;
- }
-
- public Map<String, byte[]> serializedBytes() {
- Map<String, byte[]> result = new HashMap<>();
- indexWritersMap.forEach(
- (k, v) -> {
- if (!v.empty()) {
- result.put(
- FileIndexCommon.toMapKey(columnName, k), v.serializedBytes());
- } else {
- result.put(FileIndexCommon.toMapKey(columnName, k), null);
- }
- });
- return result;
+ public void forEachIndex(IndexConsumer consumer) throws IOException {
+ for (Map.Entry<String, FileIndexWriter> entry : indexWritersMap.entrySet()) {
+ FileIndexWriter writer = entry.getValue();
+ consumer.accept(
+ FileIndexCommon.toMapKey(columnName, entry.getKey()),
+ indexType,
+ writer.empty() ? null : writer);
+ }
}
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/SpillableIndexOutputStream.java b/paimon-core/src/main/java/org/apache/paimon/io/SpillableIndexOutputStream.java
new file mode 100644
index 0000000000..72a7b19231
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/io/SpillableIndexOutputStream.java
@@ -0,0 +1,113 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+
+/** Keeps a small file index in memory and switches to an independent file at the threshold. */
+public final class SpillableIndexOutputStream extends OutputStream {
+
+ private final FileIO fileIO;
+ private final Path path;
+ private final int threshold;
+ private ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ private OutputStream fileOutput;
+ private boolean fileCreated;
+ private boolean closed;
+
+ public SpillableIndexOutputStream(FileIO fileIO, Path path, int threshold) {
+ this.fileIO = fileIO;
+ this.path = path;
+ this.threshold = threshold;
+ }
+
+ @Override
+ public void write(int value) throws IOException {
+ if (fileOutput == null && buffer.size() >= threshold) {
+ spill();
+ }
+ if (fileOutput == null) {
+ buffer.write(value);
+ } else {
+ fileOutput.write(value);
+ }
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length) throws IOException {
+ if (fileOutput == null && (long) buffer.size() + length > threshold) {
+ spill();
+ }
+ if (fileOutput == null) {
+ buffer.write(bytes, offset, length);
+ } else {
+ fileOutput.write(bytes, offset, length);
+ }
+ }
+
+ private void spill() throws IOException {
+ fileOutput = fileIO.newOutputStream(path, true);
+ fileCreated = true;
+ buffer.writeTo(fileOutput);
+ buffer = new ByteArrayOutputStream();
+ }
+
+ public boolean spilled() {
+ return fileCreated;
+ }
+
+ public byte[] embeddedBytes() {
+ return buffer.toByteArray();
+ }
+
+ @Override
+ public void flush() throws IOException {
+ if (fileOutput != null) {
+ fileOutput.flush();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!closed) {
+ closed = true;
+ if (fileOutput != null) {
+ fileOutput.close();
+ }
+ }
+ }
+
+ /** Removes an unpublished independent file after a failed write. */
+ public void abort() throws IOException {
+ try {
+ close();
+ } finally {
+ if (fileCreated) {
+ if (!fileIO.delete(path, false) && fileIO.exists(path)) {
+ throw new IOException("Failed to delete partial file index file " + path);
+ }
+ }
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/FileIndexesTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/FileIndexesTable.java
index b13ccd7bb9..a56b014211 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/FileIndexesTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/FileIndexesTable.java
@@ -327,7 +327,7 @@ public class FileIndexesTable implements ReadonlyTable {
if (embeddedIndex != null) {
try (FileIndexFormat.Reader reader =
FileIndexFormat.createMetadataReader(
- new ByteArraySeekableStream(embeddedIndex))) {
+ new ByteArraySeekableStream(embeddedIndex), embeddedIndex.length)) {
return toRows(
dataSplit,
file,
@@ -365,7 +365,7 @@ public class FileIndexesTable implements ReadonlyTable {
long containerSize = storeTable.fileIO().getFileStatus(indexPath).getLen();
try (FileIndexFormat.Reader reader =
FileIndexFormat.createMetadataReader(
- storeTable.fileIO().newInputStream(indexPath))) {
+ storeTable.fileIO().newInputStream(indexPath), containerSize)) {
return toRows(
dataSplit,
file,
@@ -420,7 +420,7 @@ public class FileIndexesTable implements ReadonlyTable {
BinaryString.fromString(indexMeta.indexType()),
BinaryString.fromString(storageType),
indexFilePath,
- (long) indexMeta.sizeInBytes(),
+ indexMeta.sizeInBytesLong(),
containerSize,
indexMeta.empty()));
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
index 585dd718dd..513331f755 100644
--- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
@@ -353,6 +353,24 @@ public class CoreOptionsTest {
.hasMessageContaining("local-kv-db.block-size");
}
+ @Test
+ public void testFileIndexInManifestThreshold() {
+ Options conf = new Options();
+ conf.set(CoreOptions.FILE_INDEX_IN_MANIFEST_THRESHOLD, MemorySize.parse("0 bytes"));
+ assertThat(new CoreOptions(conf).fileIndexInManifestThreshold()).isZero();
+
+ conf.set(
+ CoreOptions.FILE_INDEX_IN_MANIFEST_THRESHOLD, MemorySize.parse("2147483647 bytes"));
+ assertThat(new CoreOptions(conf).fileIndexInManifestThreshold())
+ .isEqualTo((long) Integer.MAX_VALUE);
+
+ conf.set(
+ CoreOptions.FILE_INDEX_IN_MANIFEST_THRESHOLD, MemorySize.parse("2147483648 bytes"));
+ assertThatThrownBy(() -> new CoreOptions(conf).fileIndexInManifestThreshold())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(CoreOptions.FILE_INDEX_IN_MANIFEST_THRESHOLD.key());
+ }
+
@Test
public void testFormatTableCommitCleanupThreadNumDefaultsTo64AndAcceptsBounds() {
Options conf = new Options();
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
index 9f1234a4da..cfdc7cae75 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/append/DedicatedFormatRollingFileWriterVectorTest.java
@@ -333,7 +333,9 @@ public class DedicatedFormatRollingFileWriterVectorTest {
PredicateBuilder predicateBuilder = new PredicateBuilder(normalFileSchema);
try (FileIndexPredicate index =
new FileIndexPredicate(
- new ByteArraySeekableStream(mainFile.embeddedIndex()), normalFileSchema)) {
+ new ByteArraySeekableStream(mainFile.embeddedIndex()),
+ normalFileSchema,
+ mainFile.embeddedIndex().length)) {
assertThat(((BitmapIndexResult) index.evaluate(predicateBuilder.equal(0, 2))).get())
.isEqualTo(RoaringBitmap32.bitmapOf(2));
assertThat(index.evaluate(predicateBuilder.equal(0, 99)).remain()).isFalse();
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java b/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java
index bfdf0f45b2..2dbc54cea8 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/FileIndexProcessorTest.java
@@ -31,6 +31,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.io.SparseFileIndexIO;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.schema.Schema;
@@ -44,6 +45,8 @@ import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Collections;
import java.util.HashMap;
@@ -60,12 +63,74 @@ public class FileIndexProcessorTest {
@TempDir java.nio.file.Path tempDir;
@Test
- public void testProcessIndexesTwoKeysOfOneMapColumn() throws Exception {
+ public void testV2RewritesContainerOverTwoGiB() throws Exception {
+ SparseFileIndexIO fileIO = new SparseFileIndexIO();
+ Path warehouse = new Path(tempDir.toUri());
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.FILE_FORMAT.key(), "parquet");
+ options.put(CoreOptions.FILE_INDEX_FORMAT_VERSION.key(), "2");
+ options.put("file-index.in-manifest-threshold", "1B");
+ options.put("file-index.stream-test.columns", "large,small");
+ options.put("file-index.stream-test.large.large", "true");
+ RowType rowType =
+ RowType.builder()
+ .field("large", DataTypes.INT())
+ .field("small", DataTypes.INT())
+ .build();
+ Identifier identifier = Identifier.create("mydb", "t");
+ FileStoreTable table;
+ try (FileSystemCatalog catalog = new FileSystemCatalog(fileIO, warehouse)) {
+ catalog.createDatabase("mydb", false);
+ catalog.createTable(
+ identifier,
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.singletonList("large"),
+ options,
+ ""),
+ false);
+ table = (FileStoreTable) catalog.getTable(identifier);
+ }
+
+ String commitUser = UUID.randomUUID().toString();
+ try (TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser)) {
+ write.write(GenericRow.of(1, 2));
+ commit.commit(1, write.prepareCommit(false, 1));
+ }
+
+ ManifestEntry entry = table.store().newScan().plan().files().get(0);
+ DataFileMeta rewritten =
+ new FileIndexProcessor(table).process(entry.partition(), entry.bucket(), entry);
+ String indexFile =
+ rewritten.extraFiles().stream()
+ .filter(name -> name.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("no rewritten file index"));
+ Path indexPath =
+ new Path(
+ table.store().pathFactory().bucketPath(entry.partition(), entry.bucket()),
+ indexFile);
+ long length = fileIO.getFileStatus(indexPath).getLen();
+ assertThat(length).isGreaterThan(Integer.MAX_VALUE);
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createReader(fileIO.newInputStream(indexPath), rowType, length)) {
+ assertThat(reader.readColumnIndex("large")).hasSize(1);
+ assertThat(reader.readColumnIndex("small")).hasSize(1);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void testProcessIndexesTwoKeysOfOneMapColumn(int version) throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
Path warehouse = new Path(tempDir.toString());
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.BUCKET.key(), "1");
options.put(CoreOptions.FILE_FORMAT.key(), "parquet");
+ options.put(CoreOptions.FILE_INDEX_FORMAT_VERSION.key(), Integer.toString(version));
// both entries share the top level column "m"
options.put(CoreOptions.FILE_INDEX + ".bloom-filter.columns", "m[k1],m[k2]");
RowType rowType =
@@ -122,8 +187,13 @@ public class FileIndexProcessorTest {
table.store().pathFactory().bucketPath(entry.partition(), entry.bucket()),
indexFile);
try (FileIndexFormat.Reader reader =
- FileIndexFormat.createReader(fileIO.newInputStream(indexPath), rowType)) {
- assertThat(reader.readAll().keySet()).containsExactlyInAnyOrder("m[k1]", "m[k2]");
+ FileIndexFormat.createReader(
+ fileIO.newInputStream(indexPath),
+ rowType,
+ fileIO.getFileStatus(indexPath).getLen())) {
+ assertThat(reader.indexMetas())
+ .extracting(FileIndexFormat.FileIndexMeta::columnName)
+ .containsExactlyInAnyOrder("m[k1]", "m[k2]");
}
}
@@ -239,7 +309,8 @@ public class FileIndexProcessorTest {
// file), so the index was rebuilt over nulls and 100 would be absent.
FieldRef vRef = new FieldRef(0, "v", DataTypes.INT());
try (FileIndexFormat.Reader reader =
- FileIndexFormat.createReader(new ByteArraySeekableStream(embedded), rowType)) {
+ FileIndexFormat.createReader(
+ new ByteArraySeekableStream(embedded), rowType, embedded.length)) {
Set<FileIndexReader> vReaders = reader.readColumnIndex("v");
assertThat(vReaders).isNotEmpty();
for (FileIndexReader vReader : vReaders) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/DataFileIndexWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/DataFileIndexWriterTest.java
index df892f3447..c5c917d4dc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/DataFileIndexWriterTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/DataFileIndexWriterTest.java
@@ -28,6 +28,7 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fileindex.FileIndexFormat;
+import org.apache.paimon.fileindex.FileIndexOptions;
import org.apache.paimon.fileindex.FileIndexReader;
import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
import org.apache.paimon.fs.FileIO;
@@ -54,7 +55,10 @@ import org.apache.paimon.utils.RoaringBitmap32;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -63,6 +67,9 @@ import java.util.function.Consumer;
import java.util.stream.Collectors;
import static org.apache.paimon.options.CatalogOptions.CACHE_ENABLED;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.catchThrowable;
/** Tests for {@link DataFileIndexWriter}. */
public class DataFileIndexWriterTest {
@@ -76,7 +83,127 @@ public class DataFileIndexWriterTest {
boolean bloomExists = false;
@Test
- public void testCreatingMultipleIndexesOnOneColumn() throws Exception {
+ public void testSpillableIndexOutputStream() throws Exception {
+ Path path = new Path(tempFile.resolve("index.idx").toUri());
+ SpillableIndexOutputStream embedded = new SpillableIndexOutputStream(fileIO, path, 4);
+ embedded.write(new byte[] {1, 2, 3, 4});
+ embedded.close();
+ assertThat(embedded.spilled()).isFalse();
+ assertThat(embedded.embeddedBytes()).containsExactly(1, 2, 3, 4);
+
+ SpillableIndexOutputStream external = new SpillableIndexOutputStream(fileIO, path, 4);
+ external.write(new byte[] {1, 2, 3, 4});
+ external.write(5);
+ external.close();
+ assertThat(external.spilled()).isTrue();
+ try (org.apache.paimon.fs.SeekableInputStream input = fileIO.newInputStream(path)) {
+ byte[] bytes = new byte[5];
+ input.read(bytes);
+ assertThat(bytes).containsExactly(1, 2, 3, 4, 5);
+ }
+ external.abort();
+ assertThat(fileIO.exists(path)).isFalse();
+ }
+
+ @Test
+ public void testV2TableWriterStreamsPayloadOverTwoGiB() throws Exception {
+ Options options = new Options();
+ options.setString("file-index.format.version", "2");
+ options.setString("file-index.in-manifest-threshold", "1B");
+ options.setString("file-index.stream-test.columns", "large,small");
+ options.setString("file-index.stream-test.large.large", "true");
+ Path path = new Path(tempFile.resolve("large.index").toUri());
+ FileIO sparseFileIO = new SparseFileIndexIO();
+ RowType rowType =
+ RowType.builder()
+ .field("large", DataTypes.INT())
+ .field("small", DataTypes.INT())
+ .build();
+ DataFileIndexWriter writer =
+ new DataFileIndexWriter(
+ sparseFileIO,
+ path,
+ rowType,
+ new FileIndexOptions(new CoreOptions(options)),
+ null);
+ writer.write(GenericRow.of(1, 2));
+ writer.close();
+ assertThat(writer.result().independentIndexFile()).isEqualTo(path.getName());
+
+ long length = sparseFileIO.getFileStatus(path).getLen();
+ assertThat(length).isGreaterThan(Integer.MAX_VALUE);
+ try (FileIndexFormat.Reader reader =
+ FileIndexFormat.createReader(sparseFileIO.newInputStream(path), rowType, length)) {
+ assertThat(reader.indexMetas())
+ .filteredOn(meta -> meta.columnName().equals("large"))
+ .extracting(FileIndexFormat.FileIndexMeta::sizeInBytesLong)
+ .containsExactly(2049L * 1024 * 1024 + 1);
+ assertThat(reader.readColumnIndex("large")).hasSize(1);
+ assertThat(reader.readColumnIndex("small")).hasSize(1);
+ }
+ }
+
+ @Test
+ public void testV2FailedWriteDeletesPartialIndexFile() throws Exception {
+ Options options = new Options();
+ options.setString("file-index.format.version", "2");
+ options.setString("file-index.in-manifest-threshold", "1B");
+ options.setString("file-index.stream-test.columns", "a");
+ options.setString("file-index.stream-test.a.fail", "true");
+ Path path = new Path(tempFile.resolve("failed.index").toUri());
+ DataFileIndexWriter writer =
+ new DataFileIndexWriter(
+ fileIO,
+ path,
+ RowType.builder().field("a", DataTypes.INT()).build(),
+ new FileIndexOptions(new CoreOptions(options)),
+ null);
+ writer.write(GenericRow.of(1));
+ assertThatThrownBy(writer::close)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Test index write failure");
+ assertThat(fileIO.exists(path)).isFalse();
+ assertThat(writer.result().independentIndexFile()).isNull();
+ }
+
+ @Test
+ public void testV2FailedWriteReportsDeleteFailure() throws Exception {
+ Options options = new Options();
+ options.setString("file-index.format.version", "2");
+ options.setString("file-index.in-manifest-threshold", "1B");
+ options.setString("file-index.stream-test.columns", "a");
+ options.setString("file-index.stream-test.a.fail", "true");
+ Path path = new Path(tempFile.resolve("undeleted.index").toUri());
+ FileIO deleteFailingFileIO =
+ new LocalFileIO() {
+ @Override
+ public boolean delete(Path path, boolean recursive) {
+ return false;
+ }
+ };
+ DataFileIndexWriter writer =
+ new DataFileIndexWriter(
+ deleteFailingFileIO,
+ path,
+ RowType.builder().field("a", DataTypes.INT()).build(),
+ new FileIndexOptions(new CoreOptions(options)),
+ null);
+ writer.write(GenericRow.of(1));
+
+ Throwable failure = catchThrowable(writer::close);
+ assertThat(failure)
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Test index write failure");
+ assertThat(failure.getSuppressed()).hasSize(1);
+ assertThat(failure.getSuppressed()[0])
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Failed to delete partial file index file");
+ assertThat(deleteFailingFileIO.exists(path)).isTrue();
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void testCreatingMultipleIndexesOnOneColumn(int version) throws Exception {
String tableName = "test";
String col1 = "f0";
@@ -84,6 +211,7 @@ public class DataFileIndexWriterTest {
Identifier identifier = Identifier.create(tableName, tableName);
Map<String, String> optionsMap = new HashMap<>();
+ optionsMap.put("file-index.format.version", Integer.toString(version));
optionsMap.put("file-index.bitmap.columns", col1);
optionsMap.put("file-index.bsi.columns", col1);
optionsMap.put("file-index.bloom-filter.columns", col2);
@@ -197,12 +325,12 @@ public class DataFileIndexWriterTest {
.collect(Collectors.toList());
// assert index file exist and only one index file
assert indexFiles.size() == 1;
+ Path indexPath = dataFilePathFactory.toAlignedPath(indexFiles.get(0), dataFileMeta);
try (FileIndexFormat.Reader reader =
FileIndexFormat.createReader(
- fileIO.newInputStream(
- dataFilePathFactory.toAlignedPath(
- indexFiles.get(0), dataFileMeta)),
- tableSchema.logicalRowType())) {
+ fileIO.newInputStream(indexPath),
+ tableSchema.logicalRowType(),
+ fileIO.getFileStatus(indexPath).getLen())) {
Set<FileIndexReader> fileIndexReaders = reader.readColumnIndex(columnName);
for (FileIndexReader fileIndexReader : fileIndexReaders) {
consumer.accept(fileIndexReader);
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
index bd7f781855..5c1c4c79da 100644
--- a/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/io/FileIndexEvaluatorTest.java
@@ -41,8 +41,6 @@ import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
@@ -203,15 +201,10 @@ public class FileIndexEvaluatorTest {
indexWriter.write(1);
indexWriter.write(2);
- Map<String, Map<String, byte[]>> indexes = new HashMap<>();
- indexes.put(
- "a",
- Collections.singletonMap(
- BitmapFileIndexFactory.BITMAP_INDEX, indexWriter.serializedBytes()));
-
ByteArrayOutputStream output = new ByteArrayOutputStream();
- try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output)) {
- writer.writeColumnIndexes(indexes);
+ try (FileIndexFormat.Writer writer = FileIndexFormat.createWriter(output, 1)) {
+ writer.writeIndex("a", BitmapFileIndexFactory.BITMAP_INDEX, indexWriter::writeTo);
+ writer.finish();
}
return output.toByteArray();
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/SparseFileIndexIO.java b/paimon-core/src/test/java/org/apache/paimon/io/SparseFileIndexIO.java
new file mode 100644
index 0000000000..7d5fd354e1
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/io/SparseFileIndexIO.java
@@ -0,0 +1,84 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+
+/** Uses sparse local files for large zero-filled test index payloads. */
+public class SparseFileIndexIO extends LocalFileIO {
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
+ if (!path.getName().endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX)) {
+ return super.newOutputStream(path, overwrite);
+ }
+ File file = new File(path.toUri());
+ file.getParentFile().mkdirs();
+ RandomAccessFile output = new RandomAccessFile(file, "rw");
+ output.setLength(0);
+ return new PositionOutputStream() {
+ private long position;
+
+ @Override
+ public long getPos() {
+ return position;
+ }
+
+ @Override
+ public void write(int value) throws IOException {
+ output.write(value);
+ position++;
+ }
+
+ @Override
+ public void write(byte[] bytes) throws IOException {
+ write(bytes, 0, bytes.length);
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length) throws IOException {
+ boolean zeros = length >= 8192;
+ for (int i = offset; zeros && i < offset + length; i++) {
+ zeros = bytes[i] == 0;
+ }
+ if (zeros) {
+ output.seek(position + length);
+ } else {
+ output.write(bytes, offset, length);
+ }
+ position += length;
+ }
+
+ @Override
+ public void flush() {}
+
+ @Override
+ public void close() throws IOException {
+ output.setLength(position);
+ output.close();
+ }
+ };
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/StreamingTestFileIndexFactory.java b/paimon-core/src/test/java/org/apache/paimon/io/StreamingTestFileIndexFactory.java
new file mode 100644
index 0000000000..fb9fd146c6
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/io/StreamingTestFileIndexFactory.java
@@ -0,0 +1,90 @@
+/*
+ * 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.io;
+
+import org.apache.paimon.fileindex.FileIndexReader;
+import org.apache.paimon.fileindex.FileIndexWriter;
+import org.apache.paimon.fileindex.FileIndexer;
+import org.apache.paimon.fileindex.FileIndexerFactory;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataType;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/** Test-only index whose streaming payload can exceed the byte-array limit. */
+public class StreamingTestFileIndexFactory implements FileIndexerFactory {
+
+ private static final byte MARKER = 7;
+
+ @Override
+ public String identifier() {
+ return "stream-test";
+ }
+
+ @Override
+ public FileIndexer create(DataType type, Options options) {
+ boolean large = options.getBoolean("large", false);
+ boolean fail = options.getBoolean("fail", false);
+ return new FileIndexer() {
+ @Override
+ public FileIndexWriter createWriter() {
+ return new FileIndexWriter() {
+ @Override
+ public void write(Object key) {}
+
+ @Override
+ public void writeTo(OutputStream output) throws IOException {
+ if (large) {
+ byte[] block = new byte[1024 * 1024];
+ for (int i = 0; i < 2049; i++) {
+ output.write(block);
+ }
+ }
+ output.write(MARKER);
+ if (fail) {
+ throw new IOException("Test index write failure");
+ }
+ }
+ };
+ }
+
+ @Override
+ public FileIndexReader createReader(
+ SeekableInputStream inputStream, int start, int length) {
+ return createReader(inputStream, (long) start, (long) length);
+ }
+
+ @Override
+ public FileIndexReader createReader(
+ SeekableInputStream inputStream, long start, long length) {
+ try {
+ inputStream.seek(start + length - 1);
+ if (inputStream.read() != MARKER) {
+ throw new IOException("Missing streaming index marker");
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return new FileIndexReader() {};
+ }
+ };
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/FileIndexesTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/FileIndexesTableTest.java
index 133d07238c..5ff507a1b5 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/system/FileIndexesTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/system/FileIndexesTableTest.java
@@ -39,6 +39,8 @@ import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataTypes;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.UncheckedIOException;
import java.util.ArrayList;
@@ -52,9 +54,10 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FileIndexesTable}. */
public class FileIndexesTableTest extends TableTestBase {
- @Test
- public void testEmbeddedFileIndexes() throws Exception {
- FileIndexesTable fileIndexesTable = createTable("EmbeddedIndexes", "1 MB");
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void testEmbeddedFileIndexes(int version) throws Exception {
+ FileIndexesTable fileIndexesTable = createTable("EmbeddedIndexes", "1 MB", version);
List<InternalRow> rows = read(fileIndexesTable);
assertIndexRows(rows, "EMBEDDED");
@@ -67,9 +70,10 @@ public class FileIndexesTableTest extends TableTestBase {
&& !row.getBoolean(12));
}
- @Test
- public void testExternalFileIndexes() throws Exception {
- FileIndexesTable fileIndexesTable = createTable("ExternalIndexes", "1 B");
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ public void testExternalFileIndexes(int version) throws Exception {
+ FileIndexesTable fileIndexesTable = createTable("ExternalIndexes", "1 B", version);
List<InternalRow> rows = read(fileIndexesTable);
assertIndexRows(rows, "FILE");
@@ -213,9 +217,20 @@ public class FileIndexesTableTest extends TableTestBase {
return createTable(tableName, inManifestThreshold, false);
}
+ private FileIndexesTable createTable(String tableName, String inManifestThreshold, int version)
+ throws Exception {
+ return createTable(tableName, inManifestThreshold, false, version);
+ }
+
private FileIndexesTable createTable(
String tableName, String inManifestThreshold, boolean writeSeparateFiles)
throws Exception {
+ return createTable(tableName, inManifestThreshold, writeSeparateFiles, 1);
+ }
+
+ private FileIndexesTable createTable(
+ String tableName, String inManifestThreshold, boolean writeSeparateFiles, int version)
+ throws Exception {
Identifier identifier = identifier(tableName);
catalog.createTable(
identifier,
@@ -225,6 +240,9 @@ public class FileIndexesTableTest extends TableTestBase {
.partitionKeys("pt")
.option(CoreOptions.BUCKET.key(), "1")
.option(CoreOptions.BUCKET_KEY.key(), "id")
+ .option(
+ CoreOptions.FILE_INDEX_FORMAT_VERSION.key(),
+ Integer.toString(version))
.option("file-index.bitmap.columns", "id")
.option("file-index.bloom-filter.columns", "id")
.option(
diff --git a/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.fileindex.FileIndexerFactory b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.fileindex.FileIndexerFactory
new file mode 100644
index 0000000000..7c5ada7a43
--- /dev/null
+++ b/paimon-core/src/test/resources/META-INF/services/org.apache.paimon.fileindex.FileIndexerFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.paimon.io.StreamingTestFileIndexFactory
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RewriteFileIndexActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RewriteFileIndexActionITCase.java
index fbfd56b000..ad9dd7783e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RewriteFileIndexActionITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/RewriteFileIndexActionITCase.java
@@ -194,7 +194,9 @@ public class RewriteFileIndexActionITCase extends ActionITCaseBase {
.toAlignedPath(file, entry.file());
try (FileIndexFormat.Reader reader =
FileIndexFormat.createReader(
- table.fileIO().newInputStream(indexFilePath), table.rowType())) {
+ table.fileIO().newInputStream(indexFilePath),
+ table.rowType(),
+ table.fileIO().getFileStatus(indexFilePath).getLen())) {
Set<FileIndexReader> readerSetK = reader.readColumnIndex("k");
Assertions.assertThat(readerSetK.size()).isEqualTo(1);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java
index ad83304d16..33ca046d20 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RewriteFileIndexProcedureITCase.java
@@ -39,7 +39,6 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -242,7 +241,9 @@ public class RewriteFileIndexProcedureITCase extends CatalogITCaseBase {
.noneMatch(s -> s.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX));
reader =
FileIndexFormat.createReader(
- new ByteArraySeekableStream(embeddedIndex), table.rowType());
+ new ByteArraySeekableStream(embeddedIndex),
+ table.rowType(),
+ embeddedIndex.length);
} else {
Assertions.assertThat(embeddedIndex).isNull();
String indexFile =
@@ -261,12 +262,15 @@ public class RewriteFileIndexProcedureITCase extends CatalogITCaseBase {
.toAlignedPath(indexFile, entry.file());
reader =
FileIndexFormat.createReader(
- table.fileIO().newInputStream(indexFilePath), table.rowType());
+ table.fileIO().newInputStream(indexFilePath),
+ table.rowType(),
+ table.fileIO().getFileStatus(indexFilePath).getLen());
}
try (FileIndexFormat.Reader indexReader = reader) {
- Map<String, Map<String, byte[]>> indexes = indexReader.readAll();
- Assertions.assertThat(indexes).containsKey("k");
- Assertions.assertThat(indexes.get("k").keySet()).containsExactly(expectedIndexType);
+ Assertions.assertThat(indexReader.indexMetas())
+ .filteredOn(meta -> meta.columnName().equals("k"))
+ .extracting(FileIndexFormat.FileIndexMeta::indexType)
+ .containsExactly(expectedIndexType);
}
}
}
@@ -329,7 +333,9 @@ public class RewriteFileIndexProcedureITCase extends CatalogITCaseBase {
.toAlignedPath(file, entry.file());
try (FileIndexFormat.Reader reader =
FileIndexFormat.createReader(
- table.fileIO().newInputStream(indexFilePath), table.rowType())) {
+ table.fileIO().newInputStream(indexFilePath),
+ table.rowType(),
+ table.fileIO().getFileStatus(indexFilePath).getLen())) {
Set<FileIndexReader> readerSetK = reader.readColumnIndex("v");
Assertions.assertThat(readerSetK.size()).isEqualTo(0);
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java
index 9e9d839edd..c0f5c664b9 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFileIndexITCase.java
@@ -235,12 +235,12 @@ public class SparkFileIndexITCase extends SparkWriteITCase {
.collect(Collectors.toList());
// assert index file exist and only one index file
assert indexFiles.size() == 1;
+ Path indexPath = dataFilePathFactory.toAlignedPath(indexFiles.get(0), dataFileMeta);
try (FileIndexFormat.Reader reader =
FileIndexFormat.createReader(
- fileIO.newInputStream(
- dataFilePathFactory.toAlignedPath(
- indexFiles.get(0), dataFileMeta)),
- tableSchema.logicalRowType())) {
+ fileIO.newInputStream(indexPath),
+ tableSchema.logicalRowType(),
+ fileIO.getFileStatus(indexPath).getLen())) {
Optional<FileIndexReader> fileIndexReader =
reader.readColumnIndex("a").stream().findFirst();
// assert index reader exist