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 052adc366f [core] Harden and slim FM index storage (#9450)
052adc366f is described below
commit 052adc366f63adc613dce01e2f1e7fec82f3c449
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Aug 28 23:35:18 2026 +0800
[core] Harden and slim FM index storage (#9450)
---
docs/docs/multimodal-table/global-index/fm.mdx | 9 +-
docs/generated/core_configuration.html | 6 +
.../paimon/globalindex/fmindex/FMBytePattern.java | 61 -----
.../globalindex/fmindex/FMGlobalIndexOptions.java | 2 +-
.../globalindex/fmindex/FMGlobalIndexReader.java | 293 +++------------------
.../globalindex/fmindex/FMGlobalIndexWriter.java | 89 +------
.../paimon/globalindex/fmindex/FMIndexFile.java | 182 ++++---------
.../globalindex/fmindex/FMIndexReadContext.java | 13 +
.../globalindex/fmindex/FMGlobalIndexTest.java | 279 ++++++++------------
.../src/test/resources/fmindex-v1-golden.base64 | 2 +-
.../spark/sql/PrimaryKeySortedIndexTest.scala | 3 +-
11 files changed, 231 insertions(+), 708 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/fm.mdx
b/docs/docs/multimodal-table/global-index/fm.mdx
index 6ab2bf4a99..87666483d7 100644
--- a/docs/docs/multimodal-table/global-index/fm.mdx
+++ b/docs/docs/multimodal-table/global-index/fm.mdx
@@ -30,9 +30,10 @@ values do not match; empty needles follow the normal Paimon
predicate semantics.
The writer divides source rows into independent partitions and appends them to
one checksummed
container file. Each partition stores a compressed wavelet matrix, sampled
suffix-array values,
-row boundaries, null rows, and exact verification pages. Reads demand-load
bounded blocks instead
-of downloading the complete container. If locating matches would cost more
than exact
-verification, the reader scans the relevant verification pages and still
returns an exact result.
+row boundaries, and null rows. It does not duplicate the source values. Reads
demand-load bounded
+blocks instead of downloading the complete container. If locating matches
would cost too much, the
+index declines the predicate so the normal data path can scan the source
values and preserve exact
+results.
## Create a Global FM Index
@@ -94,7 +95,7 @@ JSON object.
| `fm-index.compression-level` | `1` | Compression level for codecs which
support levels. |
| `fm-index.read-cache-size` | `64 mb` | Maximum decoded rank and sample block
cache per indexer. |
| `fm-index.demand-page-size` | `512 kb` | Target contiguous range size when
demand-loading blocks. |
-| `fm-index.locate-cost-ratio` | `0.001` | Maximum estimated suffix-array
locate work relative to exact stored-value scan bytes before verification
fallback. |
+| `fm-index.locate-cost-ratio` | `0.001` | Maximum estimated suffix-array
locate work relative to source text bytes before declining index evaluation. |
`fm-index.partition-size` and `fm-index.partition-row-count` bound build
memory and the unit of
independent reads. Smaller partitions reduce peak construction memory but
increase the number of
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 4dca9eb1d6..b8a48d5e8d 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1296,6 +1296,12 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<td>Boolean</td>
<td>Enables clustering by non-primary key fields. When set to
true, the physical sort order of data files is determined by the configured
'clustering.columns' instead of the primary key, optimizing query performance
for non-PK columns.</td>
</tr>
+ <tr>
+ <td><h5>pk-fm.index.columns</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Comma-separated character columns indexed by primary-key FM
indexes.</td>
+ </tr>
<tr>
<td><h5>pk-full-text.index.columns</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMBytePattern.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMBytePattern.java
deleted file mode 100644
index cbb49f1c7d..0000000000
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMBytePattern.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.globalindex.fmindex;
-
-/** Immutable linear-time byte pattern used by the exact dense-occurrence
fallback. */
-final class FMBytePattern {
-
- private final byte[] needle;
- private final int[] failure;
-
- FMBytePattern(byte[] needle) {
- this.needle = needle;
- this.failure = new int[needle.length];
- int matched = 0;
- for (int i = 1; i < needle.length; i++) {
- while (matched > 0 && needle[i] != needle[matched]) {
- matched = failure[matched - 1];
- }
- if (needle[i] == needle[matched]) {
- matched++;
- }
- failure[i] = matched;
- }
- }
-
- boolean contains(byte[] value, int offset, int length) {
- if (needle.length == 0) {
- return true;
- }
- int matched = 0;
- int end = offset + length;
- for (int i = offset; i < end; i++) {
- while (matched > 0 && value[i] != needle[matched]) {
- matched = failure[matched - 1];
- }
- if (value[i] == needle[matched]) {
- matched++;
- if (matched == needle.length) {
- return true;
- }
- }
- }
- return false;
- }
-}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexOptions.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexOptions.java
index 864d54dfe9..2cab94fce5 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexOptions.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexOptions.java
@@ -76,7 +76,7 @@ public final class FMGlobalIndexOptions {
.doubleType()
.defaultValue(0.001d)
.withDescription(
- "Maximum estimated SA-locate work divided by exact
stored-value scan bytes before falling back to verification.");
+ "Maximum estimated SA-locate work divided by
source text bytes before declining index evaluation.");
private FMGlobalIndexOptions() {}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java
index 8189e60cee..2e2da0f3ec 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexReader.java
@@ -19,10 +19,7 @@
package org.apache.paimon.globalindex.fmindex;
import org.apache.paimon.data.BinaryString;
-import org.apache.paimon.fs.FileRange;
import org.apache.paimon.fs.SeekableInputStream;
-import org.apache.paimon.fs.VectoredReadUtils;
-import org.apache.paimon.fs.VectoredReadable;
import org.apache.paimon.globalindex.ContainsRefiningGlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexResult;
@@ -47,11 +44,6 @@ import java.util.function.ToIntFunction;
/** Exact FM-index reader using blocked wavelet ranks and value-sampled suffix
locations. */
final class FMGlobalIndexReader implements ContainsRefiningGlobalIndexReader {
- private static final int MAX_VERIFICATION_RANGE_SIZE = 4 * 1024 * 1024;
- private static final int MAX_VERIFICATION_UNCOMPRESSED_RANGE_SIZE = 4 *
1024 * 1024;
- private static final int MAX_VERIFICATION_RANGE_BATCH_SIZE = 8;
- private static final int ESTIMATED_READ_REQUEST_BYTES = 64 * 1024;
-
@Nullable private final GlobalIndexFileReader fileReader;
@Nullable private final GlobalIndexIOMeta file;
private final ExecutorService executor;
@@ -181,8 +173,18 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
current.footer.firstRowId +
current.footer.rowCount)) {
return exactEmptyResult();
}
- // Enumerate selective intervals and use the stored-value path
when occurrence-level
- // SA location would cost more than Milvus' count-first guard
permits.
+ boolean hasNonEmptyNeedle = false;
+ for (byte[] needle : needles) {
+ if (needle.length > 0) {
+ hasNonEmptyNeedle = true;
+ break;
+ }
+ }
+ if (hasNonEmptyNeedle && !readContext.supportsLocate()) {
+ return Optional.empty();
+ }
+ // Like Milvus, decline index evaluation when occurrence-level SA
location is too
+ // expensive. The caller can then scan the source values without
duplicating them here.
List<SearchInterval> intervals = new ArrayList<>(needles.size());
for (byte[] needle : needles) {
if (needle.length > 0) {
@@ -196,13 +198,9 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
intervals.sort((left, right) -> Integer.compare(left.size(),
right.size()));
RoaringNavigableMap64 result = null;
- List<byte[]> denseNeedles = new ArrayList<>();
for (SearchInterval interval : intervals) {
- RoaringNavigableMap64 effectiveCandidates =
- result != null ? result : candidates == null ? null :
candidates.results();
- if (!shouldLocate(current, interval, effectiveCandidates)) {
- denseNeedles.add(interval.needle);
- continue;
+ if (!shouldLocate(current, interval)) {
+ return Optional.empty();
}
RoaringNavigableMap64 matches = locateRows(input, current,
interval);
if (result == null) {
@@ -217,11 +215,7 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
return exactEmptyResult();
}
}
- if (!denseNeedles.isEmpty()) {
- RoaringNavigableMap64 verificationCandidates =
- result != null ? result : candidates == null ? null :
candidates.results();
- result = verifyRows(input, current, denseNeedles,
verificationCandidates);
- } else if (result == null) {
+ if (result == null) {
result = nullRows(input, current, false);
if (candidates != null) {
result.and(candidates.results());
@@ -259,218 +253,26 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
for (int i = needle.length - 1; i >= 0 && lower < upper; i--) {
int symbol = directory.byteToSymbol[needle[i] & 0xFF];
if (symbol < 0) {
- return new SearchInterval(0, 0, needle);
+ return new SearchInterval(0, 0);
}
int cumulative = directory.cumulativeCounts[symbol];
RankPair ranks = rankPair(input, directory, symbol, lower, upper);
lower = cumulative + ranks.lower;
upper = cumulative + ranks.upper;
}
- return new SearchInterval(lower, upper, needle);
+ return new SearchInterval(lower, upper);
}
- private boolean shouldLocate(
- Metadata metadata,
- SearchInterval interval,
- @Nullable RoaringNavigableMap64 candidates) {
+ private boolean shouldLocate(Metadata metadata, SearchInterval interval) {
+ if (!readContext.supportsLocate()) {
+ return false;
+ }
FMIndexFile.Directory directory = metadata.directory;
double locateCost = (double) interval.size() * directory.sampleRate;
- if (candidates != null) {
- long verificationBytes = 0;
- for (FMIndexFile.VerificationPageMeta page :
directory.verificationPages) {
- if (pageSelected(metadata.footer.firstRowId, page,
candidates)) {
- verificationBytes += page.block.uncompressedLength;
- }
- }
- if (locateCost >= locateCostRatio * verificationBytes) {
- return false;
- }
- }
long textBytes = (long) directory.textLength - directory.rowCount - 1;
return locateCost < locateCostRatio * textBytes;
}
- private RoaringNavigableMap64 verifyRows(
- SeekableInputStream input,
- Metadata metadata,
- List<byte[]> needles,
- @Nullable RoaringNavigableMap64 candidates)
- throws IOException {
- List<FMBytePattern> patterns = new ArrayList<>(needles.size());
- for (byte[] needle : needles) {
- patterns.add(new FMBytePattern(needle));
- }
- RoaringNavigableMap64 result = new RoaringNavigableMap64();
- List<VerificationRange> ranges = verificationRanges(metadata,
candidates);
- for (int rangePosition = 0; rangePosition < ranges.size(); ) {
- int rangeEnd = rangePosition;
- long batchStoredBytes = 0;
- while (rangeEnd < ranges.size()
- && rangeEnd - rangePosition <
MAX_VERIFICATION_RANGE_BATCH_SIZE) {
- VerificationRange range = ranges.get(rangeEnd);
- if (rangeEnd > rangePosition
- && batchStoredBytes + range.storedLength >
MAX_VERIFICATION_RANGE_SIZE) {
- break;
- }
- batchStoredBytes += range.storedLength;
- rangeEnd++;
- }
- List<VerificationRange> batch = ranges.subList(rangePosition,
rangeEnd);
- byte[][] storedRanges = readVerificationRanges(input, batch);
- for (int rangeIndex = 0; rangeIndex < batch.size(); rangeIndex++) {
- VerificationRange range = batch.get(rangeIndex);
- List<byte[]> pages =
- FMIndexFile.decodeVerificationBlockRange(
- storedRanges[rangeIndex], range.blocks);
- for (int pageOffset = 0; pageOffset < pages.size();
pageOffset++) {
- verifyPage(
- metadata.footer.firstRowId,
-
metadata.directory.verificationPages.get(range.pageStart + pageOffset),
- pages.get(pageOffset),
- patterns,
- candidates,
- result);
- }
- }
- rangePosition = rangeEnd;
- }
- return result;
- }
-
- private List<VerificationRange> verificationRanges(
- Metadata metadata, @Nullable RoaringNavigableMap64 candidates) {
- List<FMIndexFile.VerificationPageMeta> pages =
metadata.directory.verificationPages;
- List<VerificationRange> ranges = new ArrayList<>();
- int pagePosition = 0;
- while (pagePosition < pages.size()) {
- if (!pageSelected(metadata.footer.firstRowId,
pages.get(pagePosition), candidates)) {
- pagePosition++;
- continue;
- }
- int pageStart = pagePosition;
- List<FMIndexFile.BlockInfo> blocks = new ArrayList<>();
- long storedBytes = 0;
- long uncompressedBytes = 0;
- long physicalEnd = -1;
- while (pagePosition < pages.size()
- && pageSelected(
- metadata.footer.firstRowId,
pages.get(pagePosition), candidates)) {
- FMIndexFile.BlockInfo block = pages.get(pagePosition).block;
- if (!blocks.isEmpty()
- && (block.offset != physicalEnd
- || storedBytes + block.storedLength >
MAX_VERIFICATION_RANGE_SIZE
- || uncompressedBytes + block.uncompressedLength
- >
MAX_VERIFICATION_UNCOMPRESSED_RANGE_SIZE)) {
- break;
- }
- FMIndexFile.validateVerificationBlock(block, fileSize());
- blocks.add(block);
- storedBytes += block.storedLength;
- uncompressedBytes += block.uncompressedLength;
- physicalEnd = block.offset + block.storedLength;
- pagePosition++;
- }
- ranges.add(new VerificationRange(pageStart, blocks, storedBytes));
- }
- return ranges;
- }
-
- private byte[][] readVerificationRanges(
- SeekableInputStream input, List<VerificationRange> ranges) throws
IOException {
- byte[][] result = new byte[ranges.size()][];
- if (ranges.size() == 1 || !(input instanceof VectoredReadable)) {
- for (int i = 0; i < ranges.size(); i++) {
- result[i] =
- FMIndexFile.readVerificationBlockRange(
- input, ranges.get(i).blocks, fileSize());
- }
- return result;
- }
-
- VectoredReadable readable = (VectoredReadable) input;
- List<FileRange> fileRanges = new ArrayList<>(ranges.size());
- for (VerificationRange range : ranges) {
- fileRanges.add(
- FileRange.createFileRange(
- range.blocks.get(0).offset, (int)
range.storedLength));
- }
- VectoredReadUtils.ReadOptions options =
- VectoredReadUtils.ReadOptions.from(readable)
-
.withMinSeekForVectorReads(ESTIMATED_READ_REQUEST_BYTES)
- .withSequentialReadFallback(false);
- VectoredReadUtils.readVectored(readable, fileRanges, options);
- for (int i = 0; i < fileRanges.size(); i++) {
- try {
- result[i] = fileRanges.get(i).getData().join();
- } catch (CompletionException e) {
- if (e.getCause() instanceof IOException) {
- throw (IOException) e.getCause();
- }
- throw e;
- }
- }
- return result;
- }
-
- private static void verifyPage(
- long firstRowId,
- FMIndexFile.VerificationPageMeta page,
- byte[] values,
- List<FMBytePattern> patterns,
- @Nullable RoaringNavigableMap64 candidates,
- RoaringNavigableMap64 result) {
- int offset = 0;
- for (int row = 0; row < page.rowCount; row++) {
- Preconditions.checkState(
- offset <= values.length - Integer.BYTES,
- "FM index verification page ended before its declared
rows.");
- int length = readInt(values, offset);
- offset += Integer.BYTES;
- Preconditions.checkState(
- length == -1 || (length >= 0 && length <= values.length -
offset),
- "Invalid FM index verification value length.");
- long rowId = firstRowId + page.firstRow + row;
- if (length >= 0
- && (candidates == null || candidates.contains(rowId))
- && matchesAll(patterns, values, offset, length)) {
- result.add(rowId);
- }
- if (length >= 0) {
- offset += length;
- }
- }
- Preconditions.checkState(
- offset == values.length, "FM index verification page has
trailing bytes.");
- }
-
- private static boolean pageSelected(
- long firstRowId,
- FMIndexFile.VerificationPageMeta page,
- @Nullable RoaringNavigableMap64 candidates) {
- if (candidates == null) {
- return true;
- }
- long first = firstRowId + page.firstRow;
- return candidates.intersects(first, first + page.rowCount);
- }
-
- private static boolean matchesAll(
- List<FMBytePattern> patterns, byte[] value, int offset, int
length) {
- for (FMBytePattern pattern : patterns) {
- if (!pattern.contains(value, offset, length)) {
- return false;
- }
- }
- return true;
- }
-
- private static int readInt(byte[] bytes, int offset) {
- return ((bytes[offset] & 0xFF) << 24)
- | ((bytes[offset + 1] & 0xFF) << 16)
- | ((bytes[offset + 2] & 0xFF) << 8)
- | (bytes[offset + 3] & 0xFF);
- }
-
private RoaringNavigableMap64 locateRows(
SeekableInputStream input, Metadata metadata, SearchInterval
interval)
throws IOException {
@@ -788,9 +590,13 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
private RoaringNavigableMap64 nullRows(
SeekableInputStream input, Metadata metadata, boolean selectNulls)
throws IOException {
RoaringNavigableMap64 result = new RoaringNavigableMap64();
- for (int row = 0; row < metadata.directory.rowCount; row++) {
- if (bit(input, metadata.directory.nullRows, row) == selectNulls) {
- result.add(metadata.footer.firstRowId + row);
+ FMIndexFile.BitVectorMeta nullRows = metadata.directory.nullRows;
+ for (FMIndexFile.BitBlockMeta meta : nullRows.blocks) {
+ FMIndexFile.BitBlock block = bitBlock(input, nullRows, meta);
+ for (int local = 0; local < meta.bitCount; local++) {
+ if (block.get(local) == selectNulls) {
+ result.add(metadata.footer.firstRowId + meta.firstBit +
local);
+ }
}
}
return result;
@@ -808,9 +614,12 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
Preconditions.checkState(
containerLoader != null && partition != null,
"Missing FM index container metadata.");
- containerLoader.validate(input);
+ FMIndexFile.ContainerFooter containerFooter =
containerLoader.validate(input);
FMIndexFile.Footer footer =
FMIndexFile.readFooter(input, partition,
file.fileSize());
+ Preconditions.checkState(
+ footer.featureFlags == containerFooter.featureFlags,
+ "FM index partition and container feature flags do not
match.");
FMIndexFile.Directory directory =
FMIndexFile.readDirectory(input, footer,
file.fileSize());
current = new Metadata(footer, directory);
@@ -825,11 +634,6 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
return fileReader.getInputStream(file);
}
- private long fileSize() {
- Preconditions.checkState(file != null, "Missing FM index file.");
- return file.fileSize();
- }
-
@Nullable
private static byte[] needle(@Nullable Object literal) {
if (literal == null) {
@@ -923,28 +727,13 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
@Override
public void close() {}
- private static final class VerificationRange {
- private final int pageStart;
- private final List<FMIndexFile.BlockInfo> blocks;
- private final long storedLength;
-
- private VerificationRange(
- int pageStart, List<FMIndexFile.BlockInfo> blocks, long
storedLength) {
- this.pageStart = pageStart;
- this.blocks = blocks;
- this.storedLength = storedLength;
- }
- }
-
private static final class SearchInterval {
private final int lower;
private final int upper;
- private final byte[] needle;
- private SearchInterval(int lower, int upper, byte[] needle) {
+ private SearchInterval(int lower, int upper) {
this.lower = lower;
this.upper = upper;
- this.needle = needle;
}
private boolean isEmpty() {
@@ -979,25 +768,27 @@ final class FMGlobalIndexReader implements
ContainsRefiningGlobalIndexReader {
static final class ContainerMetadataLoader {
private final GlobalIndexIOMeta file;
private final FMIndexFile.IndexMeta expected;
- private boolean validated;
+ @Nullable private FMIndexFile.ContainerFooter footer;
ContainerMetadataLoader(GlobalIndexIOMeta file, FMIndexFile.IndexMeta
expected) {
this.file = file;
this.expected = expected;
}
- synchronized void validate(SeekableInputStream input) throws
IOException {
- if (validated) {
- return;
+ synchronized FMIndexFile.ContainerFooter validate(SeekableInputStream
input)
+ throws IOException {
+ if (footer != null) {
+ return footer;
}
- FMIndexFile.ContainerFooter footer =
+ FMIndexFile.ContainerFooter current =
FMIndexFile.readContainerFooter(input, file.fileSize());
FMIndexFile.IndexMeta actual =
- FMIndexFile.readContainerDirectory(input, footer,
file.fileSize());
+ FMIndexFile.readContainerDirectory(input, current,
file.fileSize());
Preconditions.checkState(
expected.sameLayout(actual),
"FM index manifest metadata does not match the container
directory.");
- validated = true;
+ footer = current;
+ return current;
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java
index 01b5ca8449..c43c4c1979 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java
@@ -28,7 +28,6 @@ import org.apache.paimon.utils.Preconditions;
import javax.annotation.Nullable;
-import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -40,9 +39,6 @@ import java.util.List;
/** Streaming, bounded-partition writer for an exact byte-oriented FM index. */
public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter,
Closeable {
- private static final int VERIFICATION_PAGE_ROW_COUNT = 128;
- private static final int TARGET_VERIFICATION_PAGE_SIZE = 64 * 1024;
-
private final GlobalIndexFileWriter fileWriter;
private final int maxPartitionTextLength;
private final int maxPartitionRowCount;
@@ -93,12 +89,6 @@ public class FMGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Close
"FM index expects BinaryString values, but found %s.",
key.getClass().getName());
bytes = ((BinaryString) key).toBytes();
- Preconditions.checkArgument(
- bytes.length
- <=
FMIndexFile.MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH
- - Integer.BYTES,
- "A value exceeds the FM index exact-fallback block limit
(%s bytes).",
- FMIndexFile.MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH);
}
long encodedLength = (bytes == null ? 0L : bytes.length) + 1L;
Preconditions.checkArgument(
@@ -257,8 +247,6 @@ public class FMGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Close
FMIndexFile.BitVectorMeta rowBoundaries =
FMIndexFile.writeBitVector(
stream, output, boundaryWords, symbols.length,
compressionFactory);
- List<FMIndexFile.VerificationPageMeta> verificationPages =
- writeVerificationPages(stream, output, symbols,
alphabet.symbolToByte);
FMIndexFile.Directory directory =
new FMIndexFile.Directory(
partitionRowCount,
@@ -273,8 +261,7 @@ public class FMGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Close
sampled,
samples,
nullVector,
- rowBoundaries,
- verificationPages);
+ rowBoundaries);
FMIndexFile.BlockInfo directoryBlock =
FMIndexFile.writeDirectory(stream, output, directory,
compressionFactory);
FMIndexFile.writeFooter(
@@ -336,68 +323,6 @@ public class FMGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Close
return counts;
}
- private List<FMIndexFile.VerificationPageMeta> writeVerificationPages(
- PositionOutputStream stream,
- DataOutputStream output,
- char[] symbols,
- int[] symbolToByte)
- throws IOException {
- List<FMIndexFile.VerificationPageMeta> pages = new ArrayList<>();
- int row = 0;
- int symbolPosition = 0;
- while (row < partitionRowCount) {
- int firstRow = row;
- ByteArrayOutputStream bytes = new ByteArrayOutputStream();
- DataOutputStream values = new DataOutputStream(bytes);
- int pageRows = 0;
- while (row < partitionRowCount) {
- int separator = symbolPosition;
- while (separator < symbols.length && symbols[separator] !=
FMIndexFile.SEPARATOR) {
- separator++;
- }
- Preconditions.checkState(
- separator < symbols.length,
- "FM index encoded text is missing a row separator.");
- int valueLength = separator - symbolPosition;
- long recordLength = Integer.BYTES + (nullRows[row] ? 0L :
valueLength);
- if (pageRows > 0
- && (pageRows >= VERIFICATION_PAGE_ROW_COUNT
- || bytes.size() + recordLength >
TARGET_VERIFICATION_PAGE_SIZE)) {
- break;
- }
- Preconditions.checkState(
- bytes.size() + recordLength
- <=
FMIndexFile.MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH,
- "FM index verification value exceeds the supported
block size.");
- if (nullRows[row]) {
- Preconditions.checkState(
- valueLength == 0, "FM index null row contains
encoded bytes.");
- values.writeInt(-1);
- } else {
- values.writeInt(valueLength);
- for (int i = symbolPosition; i < separator; i++) {
- values.writeByte(symbolToByte[symbols[i]]);
- }
- }
- symbolPosition = separator + 1;
- row++;
- pageRows++;
- }
- values.flush();
- pages.add(
- new FMIndexFile.VerificationPageMeta(
- firstRow,
- pageRows,
- FMIndexFile.writeBlock(
- stream, output, bytes.toByteArray(),
compressionFactory)));
- }
- Preconditions.checkState(
- symbolPosition == symbols.length - 1
- && symbols[symbolPosition] == FMIndexFile.TERMINATOR,
- "FM index verification rows do not cover the encoded text.");
- return pages;
- }
-
private static DenseAlphabet densify(char[] symbols) {
boolean[] present = new boolean[256];
for (char symbol : symbols) {
@@ -413,18 +338,12 @@ public class FMGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Close
byteToSymbol[value] = alphabetSize++;
}
}
- int[] symbolToByte = new int[alphabetSize];
- for (int value = 0; value < byteToSymbol.length; value++) {
- if (byteToSymbol[value] >= 0) {
- symbolToByte[byteToSymbol[value]] = value;
- }
- }
for (int i = 0; i < symbols.length; i++) {
if (symbols[i] >= FMIndexFile.FIRST_BYTE_SYMBOL) {
symbols[i] = (char) byteToSymbol[symbols[i] -
FMIndexFile.FIRST_BYTE_SYMBOL];
}
}
- return new DenseAlphabet(alphabetSize, byteToSymbol, symbolToByte);
+ return new DenseAlphabet(alphabetSize, byteToSymbol);
}
private void ensureNullCapacity(int capacity) {
@@ -442,12 +361,10 @@ public class FMGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Close
private static final class DenseAlphabet {
private final int alphabetSize;
private final int[] byteToSymbol;
- private final int[] symbolToByte;
- private DenseAlphabet(int alphabetSize, int[] byteToSymbol, int[]
symbolToByte) {
+ private DenseAlphabet(int alphabetSize, int[] byteToSymbol) {
this.alphabetSize = alphabetSize;
this.byteToSymbol = byteToSymbol;
- this.symbolToByte = symbolToByte;
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java
index d02b68dc21..0ce01c600a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexFile.java
@@ -25,8 +25,6 @@ import org.apache.paimon.compression.BlockDecompressor;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.VectoredReadable;
-import org.apache.paimon.memory.MemorySegment;
-import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.utils.Preconditions;
import javax.annotation.Nullable;
@@ -40,19 +38,18 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-
-import static org.apache.paimon.sst.SstFileUtils.crc32c;
+import java.util.zip.CRC32;
/**
* Portable V1 container layout for partitioned, demand-paged FM indexes.
*
* <p>Each physical index file contains one or more canonical, contiguous
partitions followed by a
* checksummed container directory and fixed footer. A partition contains
dense-alphabet blocked
- * quaternary wavelet levels, sampled-SA mask and values, null mask,
row-boundary mask,
- * exact-verification value pages, its directory, and a fixed footer. Every
independently readable
- * block records its offset, stored and uncompressed lengths, compression ID
and CRC32C. The reader
- * validates all physical ranges before allocating decoded buffers and
verifies the stored checksum
- * before decompression.
+ * quaternary wavelet levels, sampled-SA mask and values, null mask,
row-boundary mask, its
+ * directory, and a fixed footer. Every independently readable block records
its offset, stored and
+ * uncompressed lengths, compression ID and CRC32 over the stored bytes
followed by the compression
+ * ID byte. The reader validates all physical ranges before allocating decoded
buffers and verifies
+ * the stored checksum before decompression.
*/
final class FMIndexFile {
@@ -72,12 +69,8 @@ final class FMIndexFile {
private static final int FEATURE_VALUE_SAMPLED_SA = 1;
private static final int FEATURE_DENSE_QUAD_WAVELET = 1 << 1;
private static final int FEATURE_SEPARATOR_ROW_IDS = 1 << 2;
- private static final int FEATURE_EXACT_DENSE_FALLBACK = 1 << 3;
private static final int FEATURE_FLAGS =
- FEATURE_VALUE_SAMPLED_SA
- | FEATURE_DENSE_QUAD_WAVELET
- | FEATURE_SEPARATOR_ROW_IDS
- | FEATURE_EXACT_DENSE_FALLBACK;
+ FEATURE_VALUE_SAMPLED_SA | FEATURE_DENSE_QUAD_WAVELET |
FEATURE_SEPARATOR_ROW_IDS;
static final int BLOCK_WORDS = 4096;
static final int BLOCK_BITS = BLOCK_WORDS * Long.SIZE;
@@ -90,10 +83,14 @@ final class FMIndexFile {
static final int FOOTER_CHECKSUM_OFFSET = 60;
static final int MAX_DIRECTORY_UNCOMPRESSED_LENGTH = 16 * 1024 * 1024;
static final int MAX_DATA_BLOCK_UNCOMPRESSED_LENGTH = 64 * 1024;
- static final int MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH = 64 * 1024 *
1024;
private FMIndexFile() {}
+ private static void validateFeatureFlags(int flags, String scope) {
+ Preconditions.checkState(
+ flags == FEATURE_FLAGS, "Unsupported FM index %s feature
flags: %s.", scope, flags);
+ }
+
static byte[] writeIndexMeta(long firstRowId, long rowCount,
List<PartitionMeta> partitions) {
Preconditions.checkArgument(firstRowId >= 0 && rowCount > 0, "Invalid
FM index row range.");
Preconditions.checkArgument(
@@ -211,8 +208,7 @@ final class FMIndexFile {
}
long offset = stream.getPos();
out.write(stored, 0, storedLength);
- int checksum =
- crc32c(new MemorySlice(MemorySegment.wrap(stored), 0,
storedLength), compression);
+ int checksum = crc32(stored, 0, storedLength, compression);
return new BlockInfo(
offset, storedLength, uncompressed.length,
compression.persistentId(), checksum);
}
@@ -342,12 +338,6 @@ final class FMIndexFile {
writeIntVectorMeta(data, directory.sampleValues);
writeBitVectorMeta(data, directory.nullRows);
writeBitVectorMeta(data, directory.rowBoundaries);
- data.writeInt(directory.verificationPages.size());
- for (VerificationPageMeta page : directory.verificationPages) {
- data.writeInt(page.firstRow);
- data.writeInt(page.rowCount);
- writeBlockInfo(data, page.block);
- }
data.flush();
byte[] uncompressed = bytes.toByteArray();
Preconditions.checkState(
@@ -429,10 +419,8 @@ final class FMIndexFile {
"FM index container footer checksum mismatch: expected=%s,
actual=%s.",
expectedChecksum,
actualChecksum);
- Preconditions.checkState(
- readInt(bytes, 44) == FEATURE_FLAGS,
- "Unsupported FM index container feature flags: %s.",
- readInt(bytes, 44));
+ int featureFlags = readInt(bytes, 44);
+ validateFeatureFlags(featureFlags, "container");
Preconditions.checkState(
readInt(bytes, 48) == 0, "Invalid FM index container reserved
field.");
@@ -456,7 +444,7 @@ final class FMIndexFile {
Preconditions.checkState(
directory.offset + directory.storedLength == fileSize -
CONTAINER_FOOTER_LENGTH,
"FM index container directory is not immediately before the
footer.");
- return new ContainerFooter(directory, firstRowId, rowCount,
partitionCount);
+ return new ContainerFooter(directory, firstRowId, rowCount,
partitionCount, featureFlags);
}
static IndexMeta readContainerDirectory(
@@ -497,10 +485,8 @@ final class FMIndexFile {
"FM index partition footer checksum mismatch: expected=%s,
actual=%s.",
expectedChecksum,
actualChecksum);
- Preconditions.checkState(
- readInt(bytes, 44) == FEATURE_FLAGS,
- "Unsupported FM index partition feature flags: %s.",
- readInt(bytes, 44));
+ int featureFlags = readInt(bytes, 44);
+ validateFeatureFlags(featureFlags, "partition");
Preconditions.checkState(
readInt(bytes, 48) == 0, "Invalid FM index partition reserved
field.");
@@ -532,7 +518,8 @@ final class FMIndexFile {
textLength,
sampleRate,
partition.startOffset,
- partition.endOffset);
+ partition.endOffset,
+ featureFlags);
}
static Footer readFooter(SeekableInputStream input, long fileSize) throws
IOException {
@@ -647,34 +634,6 @@ final class FMIndexFile {
Preconditions.checkState(
rowBoundaries.totalOnes == rowCount,
"FM index row-boundary cardinality does not match its row
count.");
- int verificationPageCount = data.readInt();
- Preconditions.checkState(
- verificationPageCount > 0 && verificationPageCount <= rowCount,
- "Invalid FM index verification page count.");
- List<VerificationPageMeta> verificationPages = new
ArrayList<>(verificationPageCount);
- int nextRow = 0;
- for (int i = 0; i < verificationPageCount; i++) {
- int firstRow = data.readInt();
- int pageRowCount = data.readInt();
- BlockInfo block = readBlockInfo(data);
- Preconditions.checkState(
- firstRow == nextRow
- && pageRowCount > 0
- && pageRowCount <= rowCount - firstRow
- && block.uncompressedLength >= pageRowCount *
Integer.BYTES
- && block.uncompressedLength
- <=
MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH,
- "Invalid FM index verification page metadata.");
- validateCanonicalBlock(
- block,
- expectedOffset,
- footer.directory.offset,
- MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH);
- verificationPages.add(new VerificationPageMeta(firstRow,
pageRowCount, block));
- nextRow += pageRowCount;
- }
- Preconditions.checkState(
- nextRow == rowCount, "FM index verification pages do not cover
all rows.");
Preconditions.checkState(
expectedOffset[0] == footer.directory.offset,
"FM index payload blocks are not canonical and contiguous.");
@@ -693,13 +652,12 @@ final class FMIndexFile {
sampledRows,
sampleValues,
nullRows,
- rowBoundaries,
- verificationPages);
+ rowBoundaries);
}
static byte[] readBlock(SeekableInputStream input, BlockInfo block, long
fileSize)
throws IOException {
- validateBlock(block, fileSize,
MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH, true);
+ validateBlock(block, fileSize, MAX_DIRECTORY_UNCOMPRESSED_LENGTH,
true);
byte[] stored = readAt(input, block.offset, block.storedLength);
return decodeStoredBlock(stored, 0, block);
}
@@ -731,52 +689,12 @@ final class FMIndexFile {
return result;
}
- static byte[] readVerificationBlockRange(
- SeekableInputStream input, List<BlockInfo> blocks, long fileSize)
throws IOException {
- Preconditions.checkArgument(
- !blocks.isEmpty(), "FM verification range must contain
blocks.");
- long firstOffset = blocks.get(0).offset;
- long nextOffset = firstOffset;
- long totalStoredLength = 0;
- for (BlockInfo block : blocks) {
- validateVerificationBlock(block, fileSize);
- Preconditions.checkState(
- block.offset == nextOffset,
- "FM verification blocks are not canonical and
contiguous.");
- totalStoredLength += block.storedLength;
- Preconditions.checkState(
- totalStoredLength <= Integer.MAX_VALUE,
- "FM verification range exceeds the supported read size.");
- nextOffset += block.storedLength;
- }
- return readAt(input, firstOffset, (int) totalStoredLength);
- }
-
- static List<byte[]> decodeVerificationBlockRange(byte[] stored,
List<BlockInfo> blocks) {
- List<byte[]> result = new ArrayList<>(blocks.size());
- int offset = 0;
- for (BlockInfo block : blocks) {
- result.add(decodeStoredBlock(stored, offset, block));
- offset += block.storedLength;
- }
- Preconditions.checkState(
- offset == stored.length, "FM verification range contains
trailing bytes.");
- return result;
- }
-
- static void validateVerificationBlock(BlockInfo block, long fileSize) {
- validateBlock(block, fileSize,
MAX_VERIFICATION_BLOCK_UNCOMPRESSED_LENGTH, false);
- }
-
private static byte[] decodeStoredBlock(byte[] stored, int offset,
BlockInfo block) {
Preconditions.checkState(
offset >= 0 && block.storedLength <= stored.length - offset,
"FM demand-page block exceeds its stored bytes.");
BlockCompressionType compression = compression(block.compressionId);
- int checksum =
- crc32c(
- new MemorySlice(MemorySegment.wrap(stored), offset,
block.storedLength),
- compression);
+ int checksum = crc32(stored, offset, block.storedLength, compression);
Preconditions.checkState(
checksum == block.checksum,
"FM index block checksum mismatch: expected=%s, actual=%s.",
@@ -1209,20 +1127,28 @@ final class FMIndexFile {
}
private static int footerChecksum(byte[] footer) {
- return crc32c(
- new MemorySlice(MemorySegment.wrap(footer), 0,
FOOTER_CHECKSUM_OFFSET),
- BlockCompressionType.NONE);
+ return crc32(footer, 0, FOOTER_CHECKSUM_OFFSET,
BlockCompressionType.NONE);
}
private static int indexMetaChecksum(byte[] metadata) {
- return crc32c(
- new MemorySlice(
- MemorySegment.wrap(metadata),
- 0,
- metadata.length - INDEX_META_CHECKSUM_LENGTH),
+ return crc32(
+ metadata,
+ 0,
+ metadata.length - INDEX_META_CHECKSUM_LENGTH,
BlockCompressionType.NONE);
}
+ /** V1 checksum contract: IEEE CRC32 over stored bytes followed by the
compression ID byte. */
+ static int crc32(byte[] stored, int offset, int length,
BlockCompressionType compressionType) {
+ Preconditions.checkArgument(
+ offset >= 0 && length >= 0 && length <= stored.length - offset,
+ "Invalid FM checksum byte range.");
+ CRC32 crc = new CRC32();
+ crc.update(stored, offset, length);
+ crc.update(compressionType.persistentId() & 0xFF);
+ return (int) crc.getValue();
+ }
+
private static byte[] readAt(SeekableInputStream input, long offset, int
length)
throws IOException {
byte[] result = new byte[length];
@@ -1482,7 +1408,6 @@ final class FMIndexFile {
final IntVectorMeta sampleValues;
final BitVectorMeta nullRows;
final BitVectorMeta rowBoundaries;
- final List<VerificationPageMeta> verificationPages;
Directory(
int rowCount,
@@ -1497,8 +1422,7 @@ final class FMIndexFile {
BitVectorMeta sampledRows,
IntVectorMeta sampleValues,
BitVectorMeta nullRows,
- BitVectorMeta rowBoundaries,
- List<VerificationPageMeta> verificationPages) {
+ BitVectorMeta rowBoundaries) {
this.rowCount = rowCount;
this.textLength = textLength;
this.sampleRate = sampleRate;
@@ -1513,19 +1437,6 @@ final class FMIndexFile {
this.sampleValues = sampleValues;
this.nullRows = nullRows;
this.rowBoundaries = rowBoundaries;
- this.verificationPages = verificationPages;
- }
- }
-
- static final class VerificationPageMeta {
- final int firstRow;
- final int rowCount;
- final BlockInfo block;
-
- VerificationPageMeta(int firstRow, int rowCount, BlockInfo block) {
- this.firstRow = firstRow;
- this.rowCount = rowCount;
- this.block = block;
}
}
@@ -1537,6 +1448,7 @@ final class FMIndexFile {
final int sampleRate;
final long partitionStartOffset;
final long partitionEndOffset;
+ final int featureFlags;
Footer(
BlockInfo directory,
@@ -1545,7 +1457,8 @@ final class FMIndexFile {
int textLength,
int sampleRate,
long partitionStartOffset,
- long partitionEndOffset) {
+ long partitionEndOffset,
+ int featureFlags) {
this.directory = directory;
this.firstRowId = firstRowId;
this.rowCount = rowCount;
@@ -1553,6 +1466,7 @@ final class FMIndexFile {
this.sampleRate = sampleRate;
this.partitionStartOffset = partitionStartOffset;
this.partitionEndOffset = partitionEndOffset;
+ this.featureFlags = featureFlags;
}
}
@@ -1561,13 +1475,19 @@ final class FMIndexFile {
final long firstRowId;
final long rowCount;
final int partitionCount;
+ final int featureFlags;
private ContainerFooter(
- BlockInfo directory, long firstRowId, long rowCount, int
partitionCount) {
+ BlockInfo directory,
+ long firstRowId,
+ long rowCount,
+ int partitionCount,
+ int featureFlags) {
this.directory = directory;
this.firstRowId = firstRowId;
this.rowCount = rowCount;
this.partitionCount = partitionCount;
+ this.featureFlags = featureFlags;
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexReadContext.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexReadContext.java
index 45f0df9865..13c8cd4192 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexReadContext.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMIndexReadContext.java
@@ -32,6 +32,15 @@ import java.util.function.Supplier;
final class FMIndexReadContext {
private static final int DEFAULT_MAX_CONCURRENT_FILE_READS = 8;
+ // A full decoded quaternary rank block contains packed words plus four
rank-prefix integers
+ // per 64 words and one terminal prefix. It is larger than a bit-rank or
sampled-value block.
+ private static final long QUAD_BLOCK_WORD_BYTES = FMIndexFile.BLOCK_WORDS
* Long.BYTES;
+ private static final long QUAD_BLOCK_PREFIX_BYTES =
+ ((FMIndexFile.BLOCK_WORDS + 63L) / 64L + 1L) * 4L * Integer.BYTES;
+ // A locate must at least retain its largest decoded block; otherwise
repeated access performs
+ // another physical read every time.
+ private static final long MIN_LOCATE_CACHE_BYTES =
+ QUAD_BLOCK_WORD_BYTES + QUAD_BLOCK_PREFIX_BYTES;
private final long cacheBudget;
private final Semaphore fileReadPermits = new
Semaphore(DEFAULT_MAX_CONCURRENT_FILE_READS);
@@ -46,6 +55,10 @@ final class FMIndexReadContext {
return (int) Math.min(configuredPageSize, Math.min(cacheBudget,
Integer.MAX_VALUE));
}
+ boolean supportsLocate() {
+ return cacheBudget >= MIN_LOCATE_CACHE_BYTES;
+ }
+
@Nullable
synchronized <T> T get(GlobalIndexIOMeta file, FMIndexFile.BlockInfo
block, Class<T> type) {
CacheEntry entry = cache.get(new BlockKey(file, block, type));
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
index de8b576df9..edc1c9f2d8 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
@@ -103,8 +103,9 @@ public class FMGlobalIndexTest {
Options options = new Options();
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 3);
- options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 4);
+ options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 1);
options.set(FMGlobalIndexOptions.COMPRESSION, "lz4");
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
indexer = new FMGlobalIndexer(dataField, options);
}
@@ -160,7 +161,9 @@ public class FMGlobalIndexTest {
Options options = new Options();
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 2);
options.set(FMGlobalIndexOptions.PARTITION_SIZE,
MemorySize.ofKibiBytes(1));
+ options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 1);
options.set(FMGlobalIndexOptions.COMPRESSION, "none");
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
indexer = new FMGlobalIndexer(dataField, options);
List<GlobalIndexIOMeta> files =
@@ -206,7 +209,8 @@ public class FMGlobalIndexTest {
public void testRandomizedExactnessAgainstByteScan() throws Exception {
Options options = new Options();
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 17);
- options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 8);
+ options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 1);
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
indexer = new FMGlobalIndexer(dataField, options);
Random random = new Random(99173);
List<BinaryString> values = new ArrayList<>();
@@ -306,6 +310,7 @@ public class FMGlobalIndexTest {
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 100);
options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 4);
options.set(FMGlobalIndexOptions.COMPRESSION, "none");
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
indexer = new FMGlobalIndexer(dataField, options);
List<GlobalIndexIOMeta> actualFiles =
writeData(
@@ -347,6 +352,13 @@ public class FMGlobalIndexTest {
}
}
+ @Test
+ public void testV1ChecksumUsesIeeeCrc32Contract() {
+ byte[] bytes = "123456789".getBytes(StandardCharsets.US_ASCII);
+ assertThat(FMIndexFile.crc32(bytes, 0, bytes.length,
BlockCompressionType.NONE))
+ .isEqualTo(0x00C49E49);
+ }
+
@Test
public void testDefaultLz4IsPersistedPerCompressibleBlock() throws
Exception {
String repeated = String.join("", Collections.nCopies(20_000,
"compressible-value-"));
@@ -417,7 +429,11 @@ public class FMGlobalIndexTest {
}
@Test
- public void testDenseOccurrenceGuardFallsBackToExactStoredValues() throws
Exception {
+ public void testDenseOccurrenceGuardDeclinesIndexEvaluation() throws
Exception {
+ Options options = new Options();
+ options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 4);
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
+ indexer = new FMGlobalIndexer(dataField, options);
String repeated = String.join("", Collections.nCopies(10_000, "a"));
List<GlobalIndexIOMeta> files =
writeData(Arrays.asList(str(repeated + "b"), str(repeated)),
0);
@@ -437,9 +453,8 @@ public class FMGlobalIndexTest {
corruptByte(file, sampleBlockOffset);
try (GlobalIndexReader reader = createReader(files, 2)) {
- // Locating every 'a' occurrence would read the corrupted SA
samples. The count-first
- // guard instead scans independently checksummed stored-value
pages and remains exact.
- assertRows(reader.visitContains(fieldRef, str("a")).join(), 0L,
1L);
+ // Dense intervals are left to the source scan, so the corrupted
SA sample is not read.
+ assertThat(reader.visitContains(fieldRef,
str("a")).join()).isEmpty();
assertThatThrownBy(() -> reader.visitContains(fieldRef,
str("b")).join())
.isInstanceOf(CompletionException.class)
.hasMessageContaining("block checksum mismatch");
@@ -450,7 +465,7 @@ public class FMGlobalIndexTest {
conservative.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 0.0001d);
indexer = new FMGlobalIndexer(dataField, conservative);
try (GlobalIndexReader reader = createReader(files, 2)) {
- assertRows(reader.visitContains(fieldRef, str("b")).join(), 0L);
+ assertThat(reader.visitContains(fieldRef,
str("b")).join()).isEmpty();
}
}
@@ -470,7 +485,80 @@ public class FMGlobalIndexTest {
}
@Test
- public void testCandidateAwareFallbackAvoidsOccurrenceLocation() throws
Exception {
+ public void testZeroReadCacheDeclinesLocate() throws Exception {
+ Options options = new Options();
+ options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 10);
+ options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 1024);
+ options.set(FMGlobalIndexOptions.COMPRESSION, "none");
+ options.set(FMGlobalIndexOptions.READ_CACHE_SIZE,
MemorySize.ofBytes(0));
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
+ indexer = new FMGlobalIndexer(dataField, options);
+ String value = String.join("", Collections.nCopies(3_000, "x")) +
"unique-needle";
+ List<GlobalIndexIOMeta> files =
writeData(Collections.singletonList(str(value)), 0);
+
+ try (GlobalIndexReader reader = createReader(files, 1)) {
+ assertThat(reader.visitContains(fieldRef,
str("unique-needle")).join()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testZeroReadCacheScansNullBitmapOncePerBlock() throws
Exception {
+ Options options = new Options();
+ options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 1_000);
+ options.set(FMGlobalIndexOptions.COMPRESSION, "none");
+ options.set(FMGlobalIndexOptions.READ_CACHE_SIZE,
MemorySize.ofBytes(0));
+ indexer = new FMGlobalIndexer(dataField, options);
+ List<BinaryString> values = new ArrayList<>();
+ for (int row = 0; row < 1_000; row++) {
+ values.add((row & 1) == 0 ? null : str("value-" + row));
+ }
+ List<GlobalIndexIOMeta> files = writeData(values, 0);
+
+ AtomicInteger preadCalls = new AtomicInteger();
+ fileReader =
+ meta ->
+ new CountingVectoredInput(
+ fileIO.newInputStream(meta.filePath()),
preadCalls);
+ try (GlobalIndexReader reader = createReader(files, values.size())) {
+ assertRows(
+ reader.visitIsNull(fieldRef).join(),
+ java.util.stream.LongStream.range(0, values.size())
+ .filter(row -> (row & 1) == 0)
+ .toArray());
+ }
+ assertThat(preadCalls.get()).isLessThan(10);
+ }
+
+ @Test
+ public void testDenseQueriesDeclineWithoutStoredValues() throws Exception {
+ Options options = new Options();
+ options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 10);
+ options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 4);
+ options.set(FMGlobalIndexOptions.COMPRESSION, "none");
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
+ indexer = new FMGlobalIndexer(dataField, options);
+ List<GlobalIndexIOMeta> files = writeData(Arrays.asList(str("aaaa"),
null, str("bbbb")), 0);
+
+ try (GlobalIndexReader reader = createReader(files, 3)) {
+ assertThat(reader.visitContains(fieldRef,
str("a")).join()).isEmpty();
+ assertRows(reader.visitContains(fieldRef, str("missing")).join());
+ assertRows(reader.visitContains(fieldRef, str("")).join(), 0L, 2L);
+ org.apache.paimon.utils.RoaringNavigableMap64 candidates =
+ new org.apache.paimon.utils.RoaringNavigableMap64();
+ candidates.add(2L);
+ assertRows(
+
((org.apache.paimon.globalindex.ContainsRefiningGlobalIndexReader) reader)
+ .visitContainsConjunction(
+ fieldRef,
+ Collections.singletonList(str("bbbb")),
+ GlobalIndexResult.create(candidates))
+ .join(),
+ 2L);
+ }
+ }
+
+ @Test
+ public void testDenseCandidateQueryDeclinesIndexEvaluation() throws
Exception {
Options options = new Options();
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 1_000);
options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 4);
@@ -478,20 +566,6 @@ public class FMGlobalIndexTest {
indexer = new FMGlobalIndexer(dataField, options);
List<GlobalIndexIOMeta> files =
writeData(new ArrayList<>(Collections.nCopies(200,
str("aaaaa"))), 0);
- GlobalIndexIOMeta file = files.get(0);
- long sampleBlockOffset;
- try (org.apache.paimon.fs.SeekableInputStream input =
- fileIO.newInputStream(file.filePath())) {
- FMIndexFile.Footer footer = FMIndexFile.readFooter(input,
file.fileSize());
- sampleBlockOffset =
- FMIndexFile.readDirectory(input, footer, file.fileSize())
- .sampleValues
- .blocks
- .get(0)
- .block
- .offset;
- }
- corruptByte(file, sampleBlockOffset);
org.apache.paimon.utils.RoaringNavigableMap64 candidates =
new org.apache.paimon.utils.RoaringNavigableMap64();
@@ -499,21 +573,18 @@ public class FMGlobalIndexTest {
try (GlobalIndexReader reader = createReader(files, 200)) {
org.apache.paimon.globalindex.ContainsRefiningGlobalIndexReader
refining =
(org.apache.paimon.globalindex.ContainsRefiningGlobalIndexReader) reader;
- // The FM interval has only 1,000 occurrences, below the normal
count-first guard. A
- // one-row ANN/sibling candidate makes one sequential verification
page cheaper than
- // locating all occurrences, so the corrupted (and unused) samples
must not be read.
- assertRows(
- refining.visitContainsConjunction(
- fieldRef,
- Collections.singletonList(str("a")),
- GlobalIndexResult.create(candidates))
- .join(),
- 17L);
+ assertThat(
+ refining.visitContainsConjunction(
+ fieldRef,
+
Collections.singletonList(str("a")),
+
GlobalIndexResult.create(candidates))
+ .join())
+ .isEmpty();
}
}
@Test
- public void testMediumOccurrenceGuardUsesExactStoredValues() throws
Exception {
+ public void testMediumOccurrenceGuardDeclinesIndexEvaluation() throws
Exception {
Options options = new Options();
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 1_000);
options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 32);
@@ -521,27 +592,9 @@ public class FMGlobalIndexTest {
indexer = new FMGlobalIndexer(dataField, options);
List<GlobalIndexIOMeta> files =
writeData(new ArrayList<>(Collections.nCopies(200,
str("aaaaa"))), 0);
- GlobalIndexIOMeta file = files.get(0);
- long sampleBlockOffset;
- try (org.apache.paimon.fs.SeekableInputStream input =
- fileIO.newInputStream(file.filePath())) {
- FMIndexFile.Footer footer = FMIndexFile.readFooter(input,
file.fileSize());
- sampleBlockOffset =
- FMIndexFile.readDirectory(input, footer, file.fileSize())
- .sampleValues
- .blocks
- .get(0)
- .block
- .offset;
- }
- corruptByte(file, sampleBlockOffset);
try (GlobalIndexReader reader = createReader(files, 200)) {
- // 1,000 occurrences are below the old unconditional 4,096 locate
threshold, but
- // locating them is much more expensive than scanning the
checksummed value pages.
- assertRows(
- reader.visitContains(fieldRef, str("a")).join(),
- java.util.stream.LongStream.range(0, 200).toArray());
+ assertThat(reader.visitContains(fieldRef,
str("a")).join()).isEmpty();
}
}
@@ -549,7 +602,9 @@ public class FMGlobalIndexTest {
public void testCandidatePartitionPruningSkipsUnrelatedWaveletData()
throws Exception {
Options options = new Options();
options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 2);
+ options.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 1);
options.set(FMGlobalIndexOptions.COMPRESSION, "none");
+ options.set(FMGlobalIndexOptions.LOCATE_COST_RATIO, 1d);
indexer = new FMGlobalIndexer(dataField, options);
List<GlobalIndexIOMeta> files =
writeData(
@@ -648,69 +703,6 @@ public class FMGlobalIndexTest {
.hasMessageContaining("metadata checksum mismatch");
}
- @Test
- public void testDenseFallbackCoalescesVerificationPages() throws Exception
{
- Options options = new Options();
- options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 1_000);
- options.set(FMGlobalIndexOptions.COMPRESSION, "none");
- indexer = new FMGlobalIndexer(dataField, options);
- List<BinaryString> values =
- new ArrayList<>(
- Collections.nCopies(
- 512, str(String.join("",
Collections.nCopies(2_048, "a")))));
- List<GlobalIndexIOMeta> files = writeData(values, 0);
-
- AtomicInteger preadCalls = new AtomicInteger();
- fileReader =
- meta ->
- new CountingVectoredInput(
- fileIO.newInputStream(meta.filePath()),
- preadCalls,
- new AtomicLong());
- try (GlobalIndexReader reader = createReader(files, values.size())) {
- assertRows(
- reader.visitContains(fieldRef, str("a")).join(),
- java.util.stream.LongStream.range(0,
values.size()).toArray());
- }
- assertThat(preadCalls.get()).isLessThanOrEqualTo(10);
- }
-
- @Test
- public void testSparseFallbackReadsVerificationRangesConcurrently() throws
Exception {
- Options options = new Options();
- options.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 10);
- options.set(FMGlobalIndexOptions.COMPRESSION, "none");
- indexer = new FMGlobalIndexer(dataField, options);
- BinaryString value = str(String.join("", Collections.nCopies(100_000,
"a")));
- List<GlobalIndexIOMeta> files =
- writeData(new ArrayList<>(Collections.nCopies(6, value)), 0);
-
- AtomicInteger concurrentPreads = new AtomicInteger();
- fileReader =
- meta ->
- new ConcurrentCountingVectoredInput(
- fileIO.newInputStream(meta.filePath()),
concurrentPreads);
- org.apache.paimon.utils.RoaringNavigableMap64 candidates =
- new org.apache.paimon.utils.RoaringNavigableMap64();
- candidates.add(0L);
- candidates.add(2L);
- candidates.add(4L);
- try (GlobalIndexReader reader = createReader(files, 6)) {
- org.apache.paimon.globalindex.ContainsRefiningGlobalIndexReader
refining =
-
(org.apache.paimon.globalindex.ContainsRefiningGlobalIndexReader) reader;
- assertRows(
- refining.visitContainsConjunction(
- fieldRef,
- Collections.singletonList(str("a")),
- GlobalIndexResult.create(candidates))
- .join(),
- 0L,
- 2L,
- 4L);
- }
- assertThat(concurrentPreads.get()).isGreaterThanOrEqualTo(2);
- }
-
@Test
public void testDemandPagingDoesNotOutgrowReadCache() throws Exception {
Options options = new Options();
@@ -739,31 +731,6 @@ public class FMGlobalIndexTest {
assertThat(maximumPread.get()).isLessThanOrEqualTo(MemorySize.ofKibiBytes(64).getBytes());
}
- @Test
- public void testDenseExactFallbackVerificationCorruptionFailsClosed()
throws Exception {
- String repeated = String.join("", Collections.nCopies(10_000, "a"));
- List<GlobalIndexIOMeta> files =
writeData(Collections.singletonList(str(repeated)), 0);
- GlobalIndexIOMeta file = files.get(0);
- long verificationOffset;
- try (org.apache.paimon.fs.SeekableInputStream input =
- fileIO.newInputStream(file.filePath())) {
- FMIndexFile.Footer footer = FMIndexFile.readFooter(input,
file.fileSize());
- verificationOffset =
- FMIndexFile.readDirectory(input, footer, file.fileSize())
- .verificationPages
- .get(0)
- .block
- .offset;
- }
- corruptByte(file, verificationOffset);
-
- try (GlobalIndexReader reader = createReader(files, 1)) {
- assertThatThrownBy(() -> reader.visitContains(fieldRef,
str("a")).join())
- .isInstanceOf(CompletionException.class)
- .hasMessageContaining("block checksum mismatch");
- }
- }
-
private List<GlobalIndexIOMeta> writeData(List<BinaryString> values, long
firstRowId)
throws Exception {
FMGlobalIndexWriter writer = indexer.createWriter(fileWriter);
@@ -889,36 +856,4 @@ public class FMGlobalIndexTest {
return vectored.pread(position, buffer, offset, length);
}
}
-
- private static final class ConcurrentCountingVectoredInput extends
SeekableInputStreamWrapper
- implements VectoredReadable {
-
- private final VectoredReadable vectored;
- private final AtomicInteger maximumConcurrentPreads;
- private final AtomicInteger currentPreads = new AtomicInteger();
-
- private ConcurrentCountingVectoredInput(
- SeekableInputStream input, AtomicInteger
maximumConcurrentPreads) {
- super(input);
- this.vectored = (VectoredReadable) input;
- this.maximumConcurrentPreads = maximumConcurrentPreads;
- }
-
- @Override
- public int pread(long position, byte[] buffer, int offset, int length)
throws IOException {
- int concurrent = currentPreads.incrementAndGet();
- maximumConcurrentPreads.accumulateAndGet(concurrent, Math::max);
- try {
- try {
- Thread.sleep(10L);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IOException("Interrupted while observing
vectored reads.", e);
- }
- return vectored.pread(position, buffer, offset, length);
- } finally {
- currentPreads.decrementAndGet();
- }
- }
- }
}
diff --git a/paimon-common/src/test/resources/fmindex-v1-golden.base64
b/paimon-common/src/test/resources/fmindex-v1-golden.base64
index f7efdf4dca..8b573075c0 100644
--- a/paimon-common/src/test/resources/fmindex-v1-golden.base64
+++ b/paimon-common/src/test/resources/fmindex-v1-golden.base64
@@ -1 +1 @@
-AAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAADAAAAAcAAAAAAAAAAAAAAAABVVAAAAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAgAAAACAAAABgAAACBVv8X1AAAAAQAAAAIAAAAAAAAABQAAAAAAAaAkAAAAEAAAAAgAAAAAAAAABAAAAAwAAAABAAAAAgAAAAAAAAABAAAAAAAAAAIAAAABAAAAAgAAAAAAAAAEAAAAAAADAMAAAAAGYmFuYW5h/////wAAAAgA/2JhbmFuYQAAAAAAAAAEAAAAEwAAAAQAAAACAAAABwAAEAAAAAAC////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
[...]
+AAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAADAAAAAcAAAAAAAAAAAAAAAABVVAAAAAAAQAAAAIAAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAgAAAACAAAABgAAACBVv8X1AAAAAQAAAAIAAAAAAAAABQAAAAAAAaAkAAAAEAAAAAgAAAAAAAAABAAAAAwAAAABAAAAAgAAAAAAAAABAAAAAAAAAAIAAAABAAAAAgAAAAAAAAAEAAAAAAADAMAAAAAEAAAAEwAAAAQAAAACAAAABwAAEAAAAAAC////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
[...]
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
index 228a925348..01beef921a 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeySortedIndexTest.scala
@@ -33,6 +33,7 @@ class PrimaryKeySortedIndexTest extends PaimonSparkTestBase {
test("primary-key FM index supports exact contains with partitioned
container") {
withTable("t") {
+ // Force this tiny fixture through FM locate; production defaults may
prefer a source scan.
spark.sql("""
|CREATE TABLE t (id INT, content STRING)
|TBLPROPERTIES (
@@ -41,7 +42,7 @@ class PrimaryKeySortedIndexTest extends PaimonSparkTestBase {
| 'deletion-vectors.enabled' = 'true',
| 'pk-fm.index.columns' = 'content',
| 'fields.content.pk-fm.index.options' =
- | '{"partition-row-count":"2"}'
+ |
'{"partition-row-count":"2","sa-sample-rate":"1","locate-cost-ratio":"1"}'
|)
|""".stripMargin)
spark.sql("""
