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/07/14 13:13:26 UTC
(paimon) branch master updated: [core] Add LSM maintenance for sorted primary-key indexes (#8613)
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 b524bae13a [core] Add LSM maintenance for sorted primary-key indexes (#8613)
b524bae13a is described below
commit b524bae13a9e05d94b4161cefd2bcb4929e1613a
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue Jul 14 21:13:19 2026 +0800
[core] Add LSM maintenance for sorted primary-key indexes (#8613)
- add field-level `pk-index.compaction.level-fanout` and
`pk-index.compaction.stale-ratio-threshold` options for primary-key
indexes
- share the LSM level planner between sorted and vector primary-key
indexes, replacing the vector-only configuration
- maintain sorted primary-key indexes incrementally with independent
multi-source index compaction
- merge stale or over-fanout sorted index sources without coupling the
work to data compaction
- fail fast when a selected index build plan fails instead of silently
skipping it
---
docs/docs/primary-key-table/vector-index.md | 4 +-
docs/generated/core_configuration.html | 12 -
.../main/java/org/apache/paimon/CoreOptions.java | 35 +-
.../bitmap/BitmapGlobalIndexFormat.java | 130 +++--
.../bitmap/BitmapGlobalIndexWriter.java | 87 +++-
.../globalindex/bitmap/BitmapIndexReader.java | 37 +-
.../bitmap/LazyFilteredBitmapIndexReaderTest.java | 141 +++++-
.../sorted/SortedGlobalIndexBuilder.java | 24 +-
.../sorted/SortedSingleColumnIndexWriter.java | 20 +-
.../pk/BucketedPrimaryKeyIndexMaintainer.java | 31 +-
.../paimon/index/pk/PrimaryKeyIndexDefinition.java | 20 +-
.../index/pk/PrimaryKeyIndexDefinitions.java | 12 +-
.../PrimaryKeyIndexLevels.java} | 87 ++--
.../pksorted/BucketedSortedIndexMaintainer.java | 270 ++++++-----
.../index/pksorted/PkSortedBucketIndexState.java | 52 +-
.../index/pksorted/PkSortedIndexBuilder.java | 62 ++-
.../paimon/index/pksorted/PkSortedIndexFile.java | 118 +++--
.../paimon/index/pksorted/PkSortedIndexGroup.java | 74 ++-
.../pkvector/BucketedVectorIndexMaintainer.java | 18 +-
.../org/apache/paimon/schema/SchemaValidation.java | 22 +-
.../table/source/PrimaryKeySortedIndexScan.java | 537 +++++++++++++++++++--
.../sorted/SortedGlobalIndexBuilderTest.java | 51 ++
.../pk/BucketedPrimaryKeyIndexMaintainerTest.java | 93 +++-
.../index/pk/PrimaryKeyIndexDefinitionsTest.java | 35 ++
.../paimon/index/pk/PrimaryKeyIndexLevelsTest.java | 171 +++++++
.../BucketedSortedIndexMaintainerTest.java | 433 ++++++++++++++---
.../pksorted/PkSortedBucketIndexStateTest.java | 129 ++++-
.../index/pksorted/PkSortedIndexBuilderTest.java | 107 +++-
.../index/pksorted/PkSortedIndexFileTest.java | 190 ++++++--
.../PrimaryKeySortedIndexMaintenanceTest.java | 24 +-
.../pksorted/PrimaryKeySortedIndexOptionsTest.java | 25 +-
.../BucketedVectorIndexMaintainerTest.java | 6 +-
.../index/pkvector/PkVectorAnnLevelsTest.java | 141 ------
.../paimon/operation/PrimaryKeyIndexWriteTest.java | 11 +
.../PrimaryKeySortedIndexValidationTest.java | 11 +
.../PrimaryKeyVectorIndexValidationTest.java | 8 +-
.../paimon/table/BitmapGlobalIndexTableTest.java | 19 +-
.../source/PrimaryKeySortedIndexReadTest.java | 61 ++-
.../source/PrimaryKeySortedIndexResultTest.java | 12 +-
.../source/PrimaryKeySortedIndexScanTest.java | 151 +++++-
.../flink/globalindex/SortedIndexTopoBuilder.java | 14 +
.../globalindex/SortedIndexTopoBuilderTest.java | 56 +++
.../DefaultGlobalIndexTopoBuilderTest.java | 7 +
43 files changed, 2714 insertions(+), 834 deletions(-)
diff --git a/docs/docs/primary-key-table/vector-index.md b/docs/docs/primary-key-table/vector-index.md
index 899e6f0d32..54e33325d5 100644
--- a/docs/docs/primary-key-table/vector-index.md
+++ b/docs/docs/primary-key-table/vector-index.md
@@ -103,8 +103,8 @@ The vector comment directive converts the SQL `ARRAY<FLOAT>` column to Paimon's
| `fields.<column>.pk-vector.index.type` | Yes | ANN implementation, such as `ivf-flat`, `ivf-pq`, `ivf-hnsw-flat`, `ivf-hnsw-sq`, or `lumina`. |
| `fields.<column>.pk-vector.distance.metric` | No | `l2`, `cosine`, or `inner_product`. The default is `inner_product`. |
| `fields.<column>.pk-vector.index.options` | No | JSON object containing build options for the selected ANN implementation. Unqualified keys are scoped to that implementation. |
-| `pk-vector.index.compaction.level-fanout` | No | Number of similarly sized ANN segments which triggers a rebuild and maximum row-count ratio within one size tier. Default: `5`. |
-| `pk-vector.index.compaction.stale-ratio-threshold` | No | Ratio of rows from inactive source files which triggers an ANN rebuild. Default: `0.2`. |
+| `fields.<column>.pk-index.compaction.level-fanout` | No | Number of similarly sized index groups which triggers a rebuild and maximum row-count ratio within one size tier. Shared by vector, BTree, and Bitmap primary-key indexes. Default: `5`. |
+| `fields.<column>.pk-index.compaction.stale-ratio-threshold` | No | Ratio of rows from inactive source files which triggers an index rebuild. Shared by vector, BTree, and Bitmap primary-key indexes. Default: `0.2`. |
For algorithm-specific build and search options, see
[Vector Index](../multimodal-table/global-index/vector).
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index cc8ceaaec3..a0db693b2f 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1223,18 +1223,6 @@ This config option does not affect the default filesystem metastore.</td>
<td>String</td>
<td>Comma-separated VECTOR columns indexed by primary-key vector indexes. Each column owns one index and must define fields.<column>.pk-vector.index.type. Index options and distance metric are also field-scoped. The first release supports exactly one column.</td>
</tr>
- <tr>
- <td><h5>pk-vector.index.compaction.level-fanout</h5></td>
- <td style="word-wrap: break-word;">5</td>
- <td>Integer</td>
- <td>Number of similarly sized ANN segments that triggers a rebuild and the maximum row-count ratio within one size tier.</td>
- </tr>
- <tr>
- <td><h5>pk-vector.index.compaction.stale-ratio-threshold</h5></td>
- <td style="word-wrap: break-word;">0.2</td>
- <td>Double</td>
- <td>Ratio of rows belonging to inactive source files that triggers an ANN segment rebuild.</td>
- </tr>
<tr>
<td><h5>postpone.batch-write-fixed-bucket</h5></td>
<td style="word-wrap: break-word;">true</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 9130e4bc03..2537fbb793 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2755,20 +2755,6 @@ public class CoreOptions implements Serializable {
.withDescription(
"Comma-separated columns indexed by primary-key Bitmap indexes.");
- public static final ConfigOption<Integer> PK_VECTOR_INDEX_COMPACTION_LEVEL_FANOUT =
- key("pk-vector.index.compaction.level-fanout")
- .intType()
- .defaultValue(5)
- .withDescription(
- "Number of similarly sized ANN segments that triggers a rebuild and the maximum row-count ratio within one size tier.");
-
- public static final ConfigOption<Double> PK_VECTOR_INDEX_COMPACTION_STALE_RATIO_THRESHOLD =
- key("pk-vector.index.compaction.stale-ratio-threshold")
- .doubleType()
- .defaultValue(0.2)
- .withDescription(
- "Ratio of rows belonging to inactive source files that triggers an ANN segment rebuild.");
-
@Immutable
public static final ConfigOption<Boolean> PK_CLUSTERING_OVERRIDE =
key("pk-clustering-override")
@@ -4301,12 +4287,20 @@ public class CoreOptions implements Serializable {
return options.getOptional(PK_VECTOR_INDEX_COLUMNS).isPresent();
}
- public int primaryKeyVectorIndexCompactionLevelFanout() {
- return options.get(PK_VECTOR_INDEX_COMPACTION_LEVEL_FANOUT);
+ public int primaryKeyIndexCompactionLevelFanout(String column) {
+ return options.getInteger(primaryKeyIndexCompactionLevelFanoutKey(column), 5);
+ }
+
+ public double primaryKeyIndexCompactionStaleRatioThreshold(String column) {
+ return options.getDouble(primaryKeyIndexCompactionStaleRatioThresholdKey(column), 0.2);
+ }
+
+ public static String primaryKeyIndexCompactionLevelFanoutKey(String column) {
+ return "fields." + column + ".pk-index.compaction.level-fanout";
}
- public double primaryKeyVectorIndexCompactionStaleRatioThreshold() {
- return options.get(PK_VECTOR_INDEX_COMPACTION_STALE_RATIO_THRESHOLD);
+ public static String primaryKeyIndexCompactionStaleRatioThresholdKey(String column) {
+ return "fields." + column + ".pk-index.compaction.stale-ratio-threshold";
}
public List<String> primaryKeyVectorIndexColumns() {
@@ -4340,6 +4334,7 @@ public class CoreOptions implements Serializable {
private Options primaryKeySortedIndexOptions(
String column, String optionFamily, String algorithmPrefix) {
Options resolved = new Options(toConfiguration().toMap());
+ resolved.remove("sorted-index.records-per-range");
String optionKey = "fields." + column + "." + optionFamily + ".index.options";
String serialized = options.get(optionKey);
if (serialized == null || serialized.trim().isEmpty()) {
@@ -4363,9 +4358,7 @@ public class CoreOptions implements Serializable {
optionKey);
checkArgument(value != null, "%s value for key %s must not be null.", optionKey, key);
String qualifiedKey =
- key.startsWith(algorithmPrefix)
- || key.startsWith("sorted-index.")
- || key.startsWith("fields.")
+ key.startsWith(algorithmPrefix) || key.startsWith("fields.")
? key
: algorithmPrefix + key;
String previous = resolved.get(qualifiedKey);
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java
index 2e355b8cde..0d5c608268 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java
@@ -41,9 +41,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
-import java.util.Map;
import static org.apache.paimon.sst.SstFileUtils.crc32c;
@@ -56,71 +54,83 @@ class BitmapGlobalIndexFormat {
private BitmapGlobalIndexFormat() {}
- static void write(
- PositionOutputStream outputStream,
- RoaringNavigableMap64 nullRows,
- RoaringNavigableMap64 nonNullRows,
- Map<SerializedKey, RoaringNavigableMap64> bitmaps,
- int dictionaryBlockSize,
- @Nullable BlockCompressionFactory compressionFactory)
- throws IOException {
- Preconditions.checkArgument(
- dictionaryBlockSize > 0, "Bitmap dictionary block size must be greater than 0.");
+ static class StreamingWriter {
+
+ private final PositionOutputStream outputStream;
+ private final DataOutputStream out;
+ private final int dictionaryBlockSize;
+ @Nullable private final BlockCompressionFactory compressionFactory;
+ private final List<DictionaryBlockMeta> dictionaryBlockMetas = new ArrayList<>();
+
+ private DictionaryBlockBuilder currentDictionaryBlock = new DictionaryBlockBuilder();
+ private int valueCount;
+
+ StreamingWriter(
+ PositionOutputStream outputStream,
+ int dictionaryBlockSize,
+ @Nullable BlockCompressionFactory compressionFactory) {
+ Preconditions.checkArgument(
+ dictionaryBlockSize > 0,
+ "Bitmap dictionary block size must be greater than 0.");
+ this.outputStream = outputStream;
+ this.out = new DataOutputStream(outputStream);
+ this.dictionaryBlockSize = dictionaryBlockSize;
+ this.compressionFactory = compressionFactory;
+ }
+
+ void write(SerializedKey key, RoaringNavigableMap64 bitmap) throws IOException {
+ BlockInfo bitmapBlock = writeBitmapBlock(outputStream, out, bitmap);
+ DictionaryEntry dictionaryEntry = new DictionaryEntry(key, bitmapBlock);
+ if (currentDictionaryBlock.hasEntries()
+ && currentDictionaryBlock.estimatedSizeAfter(dictionaryEntry)
+ > dictionaryBlockSize) {
+ flushDictionaryBlock();
+ }
+ currentDictionaryBlock.add(dictionaryEntry);
+ valueCount++;
+ }
+
+ void finish(RoaringNavigableMap64 nullRows, RoaringNavigableMap64 nonNullRows)
+ throws IOException {
+ flushDictionaryBlock();
+ BlockInfo nullRowsBlock = writeBitmapBlock(outputStream, out, nullRows);
+ BlockInfo nonNullRowsBlock = writeBitmapBlock(outputStream, out, nonNullRows);
+ BlockInfo indexBlock =
+ writeIndexBlock(outputStream, out, dictionaryBlockMetas, compressionFactory);
- DataOutputStream out = new DataOutputStream(outputStream);
- BlockInfo nullRowsBlock = writeBitmapBlock(outputStream, out, nullRows);
- BlockInfo nonNullRowsBlock = writeBitmapBlock(outputStream, out, nonNullRows);
- DictionaryBlocks dictionaryBlocks =
- writeDictionaryAndBitmapBlocks(
- outputStream, out, bitmaps, dictionaryBlockSize, compressionFactory);
- BlockInfo indexBlock =
- writeIndexBlock(outputStream, out, dictionaryBlocks.blocks, compressionFactory);
+ writeFooter(out, nullRowsBlock, nonNullRowsBlock, indexBlock, valueCount);
+ }
+
+ private void flushDictionaryBlock() throws IOException {
+ if (!currentDictionaryBlock.hasEntries()) {
+ return;
+ }
+ dictionaryBlockMetas.add(
+ writeDictionaryBlock(
+ outputStream, out, currentDictionaryBlock, compressionFactory));
+ currentDictionaryBlock = new DictionaryBlockBuilder();
+ }
+ }
+ private static void writeFooter(
+ DataOutputStream out,
+ BlockInfo nullRowsBlock,
+ BlockInfo nonNullRowsBlock,
+ BlockInfo indexBlock,
+ int valueCount)
+ throws IOException {
out.writeLong(nullRowsBlock.offset);
out.writeInt(nullRowsBlock.length);
out.writeLong(nonNullRowsBlock.offset);
out.writeInt(nonNullRowsBlock.length);
out.writeLong(indexBlock.offset);
out.writeInt(indexBlock.length);
- out.writeInt(dictionaryBlocks.valueCount);
+ out.writeInt(valueCount);
out.writeInt(VERSION);
out.writeInt(MAGIC);
out.flush();
}
- private static DictionaryBlocks writeDictionaryAndBitmapBlocks(
- PositionOutputStream outputStream,
- DataOutputStream out,
- Map<SerializedKey, RoaringNavigableMap64> bitmaps,
- int dictionaryBlockSize,
- @Nullable BlockCompressionFactory compressionFactory)
- throws IOException {
- List<Map.Entry<SerializedKey, RoaringNavigableMap64>> entries =
- new ArrayList<>(bitmaps.entrySet());
- Collections.sort(entries, (o1, o2) -> o1.getKey().compareTo(o2.getKey()));
-
- List<DictionaryBlockMeta> dictionaryBlockMetas = new ArrayList<>();
- DictionaryBlockBuilder current = new DictionaryBlockBuilder();
- int valueCount = 0;
- for (Map.Entry<SerializedKey, RoaringNavigableMap64> entry : entries) {
- BlockInfo bitmapBlock = writeBitmapBlock(outputStream, out, entry.getValue());
- DictionaryEntry dictionaryEntry = new DictionaryEntry(entry.getKey(), bitmapBlock);
- if (current.hasEntries()
- && current.estimatedSizeAfter(dictionaryEntry) > dictionaryBlockSize) {
- dictionaryBlockMetas.add(
- writeDictionaryBlock(outputStream, out, current, compressionFactory));
- current = new DictionaryBlockBuilder();
- }
- current.add(dictionaryEntry);
- valueCount++;
- }
- if (current.hasEntries()) {
- dictionaryBlockMetas.add(
- writeDictionaryBlock(outputStream, out, current, compressionFactory));
- }
- return new DictionaryBlocks(dictionaryBlockMetas, valueCount);
- }
-
private static BlockInfo writeBitmapBlock(
PositionOutputStream outputStream, DataOutputStream out, RoaringNavigableMap64 bitmap)
throws IOException {
@@ -206,7 +216,6 @@ class BitmapGlobalIndexFormat {
int length = readVarLenInt(input);
blocks.add(new DictionaryBlockMeta(new SerializedKey(keyBytes), offset, length));
}
- Collections.sort(blocks, (o1, o2) -> o1.firstKey.compareTo(o2.firstKey));
return blocks;
}
@@ -522,17 +531,6 @@ class BitmapGlobalIndexFormat {
}
}
- private static class DictionaryBlocks {
-
- private final List<DictionaryBlockMeta> blocks;
- private final int valueCount;
-
- private DictionaryBlocks(List<DictionaryBlockMeta> blocks, int valueCount) {
- this.blocks = blocks;
- this.valueCount = valueCount;
- }
- }
-
private static class BlockEncoding {
private final byte[] bytes;
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexWriter.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexWriter.java
index 25469d962d..f13bf00555 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexWriter.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexWriter.java
@@ -29,25 +29,31 @@ import org.apache.paimon.utils.RoaringNavigableMap64;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.IOException;
import java.util.Collections;
import java.util.Comparator;
-import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
-/** The {@link GlobalIndexSingleColumnWriter} implementation for bitmap index. */
-public class BitmapGlobalIndexWriter implements GlobalIndexSingleColumnWriter {
+/**
+ * The {@link GlobalIndexSingleColumnWriter} implementation for bitmap index. Non-null keys must be
+ * written in monotonically increasing order so completed bitmaps can be streamed to the output file
+ * instead of retained until {@link #finish()}.
+ */
+public class BitmapGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Closeable {
private final GlobalIndexFileWriter fileWriter;
private final KeySerializer keySerializer;
private final Comparator<Object> comparator;
private final int dictionaryBlockSize;
@Nullable private final BlockCompressionFactory compressionFactory;
- private final Map<BitmapGlobalIndexFormat.SerializedKey, RoaringNavigableMap64> bitmaps;
+ private final RoaringNavigableMap64 currentBitmap;
private final RoaringNavigableMap64 nullRows;
private final RoaringNavigableMap64 nonNullRows;
+ private String fileName;
+ private PositionOutputStream outputStream;
+ private BitmapGlobalIndexFormat.StreamingWriter streamingWriter;
private long rowCount;
private Object firstKey;
private Object lastKey;
@@ -62,7 +68,7 @@ public class BitmapGlobalIndexWriter implements GlobalIndexSingleColumnWriter {
this.comparator = keySerializer.createComparator();
this.dictionaryBlockSize = dictionaryBlockSize;
this.compressionFactory = compressionFactory;
- this.bitmaps = new LinkedHashMap<>();
+ this.currentBitmap = new RoaringNavigableMap64();
this.nullRows = new RoaringNavigableMap64();
this.nonNullRows = new RoaringNavigableMap64();
}
@@ -76,10 +82,21 @@ public class BitmapGlobalIndexWriter implements GlobalIndexSingleColumnWriter {
}
nonNullRows.add(relativeRowId);
- updateMinMax(key);
- BitmapGlobalIndexFormat.SerializedKey serializedKey =
- BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, key);
- bitmaps.computeIfAbsent(serializedKey, k -> new RoaringNavigableMap64()).add(relativeRowId);
+ if (lastKey != null) {
+ int comparison = comparator.compare(key, lastKey);
+ if (comparison < 0) {
+ throw new IllegalArgumentException(
+ "Bitmap index keys must be written in monotonically increasing order.");
+ }
+ if (comparison > 0) {
+ flushCurrentBitmap();
+ }
+ }
+ if (firstKey == null) {
+ firstKey = key;
+ }
+ lastKey = key;
+ currentBitmap.add(relativeRowId);
}
@Override
@@ -88,15 +105,10 @@ public class BitmapGlobalIndexWriter implements GlobalIndexSingleColumnWriter {
return Collections.emptyList();
}
- String fileName = fileWriter.newFileName(BitmapGlobalIndexerFactory.IDENTIFIER);
- try (PositionOutputStream outputStream = fileWriter.newOutputStream(fileName)) {
- BitmapGlobalIndexFormat.write(
- outputStream,
- nullRows,
- nonNullRows,
- bitmaps,
- dictionaryBlockSize,
- compressionFactory);
+ try {
+ flushCurrentBitmap();
+ streamingWriter().finish(nullRows, nonNullRows);
+ close();
} catch (IOException e) {
throw new RuntimeException("Error in closing bitmap index writer.", e);
}
@@ -110,12 +122,39 @@ public class BitmapGlobalIndexWriter implements GlobalIndexSingleColumnWriter {
return Collections.singletonList(new ResultEntry(fileName, rowCount, meta));
}
- private void updateMinMax(Object key) {
- if (firstKey == null || comparator.compare(key, firstKey) < 0) {
- firstKey = key;
+ @Override
+ public void close() throws IOException {
+ PositionOutputStream stream = outputStream;
+ outputStream = null;
+ if (stream != null) {
+ stream.close();
+ }
+ }
+
+ private void flushCurrentBitmap() {
+ if (currentBitmap.isEmpty()) {
+ return;
}
- if (lastKey == null || comparator.compare(key, lastKey) > 0) {
- lastKey = key;
+ try {
+ streamingWriter()
+ .write(
+ BitmapGlobalIndexFormat.SerializedKey.fromObject(
+ keySerializer, lastKey),
+ currentBitmap);
+ currentBitmap.clear();
+ } catch (IOException e) {
+ throw new RuntimeException("Error in writing bitmap index files.", e);
+ }
+ }
+
+ private BitmapGlobalIndexFormat.StreamingWriter streamingWriter() throws IOException {
+ if (streamingWriter == null) {
+ fileName = fileWriter.newFileName(BitmapGlobalIndexerFactory.IDENTIFIER);
+ outputStream = fileWriter.newOutputStream(fileName);
+ streamingWriter =
+ new BitmapGlobalIndexFormat.StreamingWriter(
+ outputStream, dictionaryBlockSize, compressionFactory);
}
+ return streamingWriter;
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapIndexReader.java
index c5c46394a7..2828f28852 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapIndexReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapIndexReader.java
@@ -188,9 +188,7 @@ class BitmapIndexReader implements BitmapGlobalIndexFormat.SeekableReader, Close
return new RoaringNavigableMap64();
}
- BitmapGlobalIndexFormat.BlockInfo bitmapBlock =
- findBitmapBlock(
- BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, literal));
+ BitmapGlobalIndexFormat.BlockInfo bitmapBlock = findBitmapBlock(literal);
if (bitmapBlock == null) {
return new RoaringNavigableMap64();
}
@@ -212,7 +210,7 @@ class BitmapIndexReader implements BitmapGlobalIndexFormat.SeekableReader, Close
continue;
}
- BitmapGlobalIndexFormat.BlockInfo bitmapBlock = findBitmapBlock(key);
+ BitmapGlobalIndexFormat.BlockInfo bitmapBlock = findBitmapBlock(literal);
if (bitmapBlock != null) {
result.or(readBitmap(bitmapBlock));
}
@@ -346,18 +344,17 @@ class BitmapIndexReader implements BitmapGlobalIndexFormat.SeekableReader, Close
private int firstPossibleDictionaryBlock(
List<BitmapGlobalIndexFormat.DictionaryBlockMeta> blocks,
BitmapGlobalIndexFormat.SerializedKey key) {
- int index = findDictionaryBlockIndex(blocks, key);
+ int index = findSerializedDictionaryBlockIndex(blocks, key);
return Math.max(index, 0);
}
- private BitmapGlobalIndexFormat.BlockInfo findBitmapBlock(
- BitmapGlobalIndexFormat.SerializedKey key) {
+ private BitmapGlobalIndexFormat.BlockInfo findBitmapBlock(Object key) {
List<BitmapGlobalIndexFormat.DictionaryBlockMeta> blocks = dictionaryBlocks.get();
if (blocks.isEmpty()) {
return null;
}
- int index = findDictionaryBlockIndex(blocks, key);
+ int index = findLogicalDictionaryBlockIndex(blocks, key);
if (index < 0) {
return null;
}
@@ -365,7 +362,9 @@ class BitmapIndexReader implements BitmapGlobalIndexFormat.SeekableReader, Close
BitmapGlobalIndexFormat.DictionaryBlock dictionaryBlock =
dictionaryBlock(blocks.get(index));
for (BitmapGlobalIndexFormat.DictionaryEntry entry : dictionaryBlock.entries) {
- int compare = entry.key.compareTo(key);
+ int compare =
+ comparator.compare(
+ keySerializer.deserialize(MemorySlice.wrap(entry.key.bytes())), key);
if (compare == 0) {
return entry.bitmapBlock;
} else if (compare > 0) {
@@ -383,7 +382,7 @@ class BitmapIndexReader implements BitmapGlobalIndexFormat.SeekableReader, Close
}
}
- private int findDictionaryBlockIndex(
+ private int findSerializedDictionaryBlockIndex(
List<BitmapGlobalIndexFormat.DictionaryBlockMeta> blocks,
BitmapGlobalIndexFormat.SerializedKey key) {
int low = 0;
@@ -400,6 +399,24 @@ class BitmapIndexReader implements BitmapGlobalIndexFormat.SeekableReader, Close
return high;
}
+ private int findLogicalDictionaryBlockIndex(
+ List<BitmapGlobalIndexFormat.DictionaryBlockMeta> blocks, Object key) {
+ int low = 0;
+ int high = blocks.size() - 1;
+ while (low <= high) {
+ int mid = (low + high) >>> 1;
+ Object firstKey =
+ keySerializer.deserialize(MemorySlice.wrap(blocks.get(mid).firstKey.bytes()));
+ int compare = comparator.compare(firstKey, key);
+ if (compare <= 0) {
+ low = mid + 1;
+ } else {
+ high = mid - 1;
+ }
+ }
+ return high;
+ }
+
private BitmapGlobalIndexFormat.DictionaryBlock dictionaryBlock(
BitmapGlobalIndexFormat.DictionaryBlockMeta blockMeta) {
return dictionaryBlockCache.computeIfAbsent(blockMeta, this::readDictionaryBlock);
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java
index 0a1439ed86..bf0b36f667 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java
@@ -35,6 +35,7 @@ import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.VarCharType;
import org.apache.paimon.utils.Pair;
@@ -42,6 +43,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -51,9 +53,11 @@ import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link LazyFilteredBitmapReader}. */
public class LazyFilteredBitmapIndexReaderTest {
@@ -133,6 +137,96 @@ public class LazyFilteredBitmapIndexReaderTest {
}
}
+ @Test
+ public void testFlushesCompletedBitmapBeforeFinish() throws Exception {
+ AtomicReference<ByteArrayPositionOutputStream> output = new AtomicReference<>();
+ GlobalIndexFileWriter streamingFileWriter =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + ".index";
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) {
+ ByteArrayPositionOutputStream stream = new ByteArrayPositionOutputStream();
+ output.set(stream);
+ return stream;
+ }
+ };
+ GlobalIndexSingleColumnWriter writer = globalIndexer.createWriter(streamingFileWriter);
+
+ writer.write(str("A"), 0);
+ writer.write(str("B"), 1);
+
+ assertThat(output.get()).isNotNull();
+ assertThat(output.get().getPos()).isPositive();
+ writer.finish();
+ }
+
+ @Test
+ public void testClosesWriterAfterWriteFailure() throws Exception {
+ AtomicReference<ByteArrayPositionOutputStream> output = new AtomicReference<>();
+ GlobalIndexFileWriter streamingFileWriter =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + ".index";
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) {
+ ByteArrayPositionOutputStream stream = new ByteArrayPositionOutputStream();
+ output.set(stream);
+ return stream;
+ }
+ };
+ GlobalIndexSingleColumnWriter writer = globalIndexer.createWriter(streamingFileWriter);
+ writer.write(str("A"), 0);
+ writer.write(str("B"), 1);
+
+ assertThatThrownBy(() -> writer.write(str("A"), 2))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThat(writer).isInstanceOf(AutoCloseable.class);
+ ((AutoCloseable) writer).close();
+
+ assertThat(output.get().closed).isTrue();
+ }
+
+ @Test
+ public void testRejectsUnsortedKeys() throws Exception {
+ GlobalIndexSingleColumnWriter writer = globalIndexer.createWriter(fileWriter);
+ writer.write(str("B"), 0);
+
+ assertThatThrownBy(() -> writer.write(str("A"), 1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("monotonically increasing");
+ }
+
+ @Test
+ public void testLogicalOrderForNumericKeys() throws Exception {
+ DataField intField = new DataField(2, "number", DataTypes.INT());
+ FieldRef intFieldRef = new FieldRef(2, "number", DataTypes.INT());
+ Options options = new Options();
+ options.set(
+ BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE,
+ org.apache.paimon.options.MemorySize.ofBytes(1));
+ BitmapGlobalIndexer intIndexer = new BitmapGlobalIndexer(intField, options);
+ GlobalIndexSingleColumnWriter writer = intIndexer.createWriter(fileWriter);
+ writer.write(-1, 0);
+ writer.write(0, 1);
+ ResultEntry result = writer.finish().get(0);
+ Path filePath = new Path(basePath, result.fileName());
+ GlobalIndexIOMeta meta =
+ new GlobalIndexIOMeta(filePath, fileIO.getFileSize(filePath), result.meta());
+
+ try (GlobalIndexReader reader =
+ intIndexer.createReader(
+ fileReader, Collections.singletonList(meta), newDirectExecutorService())) {
+ assertRows(reader.visitEqual(intFieldRef, 0).join(), 1L);
+ }
+ }
+
@Test
public void testFallbackScanPredicates() throws Exception {
GlobalIndexIOMeta meta =
@@ -416,7 +510,15 @@ public class LazyFilteredBitmapIndexReaderTest {
private GlobalIndexIOMeta writeData(List<Pair<BinaryString, Long>> data) throws IOException {
GlobalIndexSingleColumnWriter writer = globalIndexer.createWriter(fileWriter);
- for (Pair<BinaryString, Long> pair : data) {
+ List<Pair<BinaryString, Long>> sortedData = new ArrayList<>(data);
+ sortedData.sort(
+ (left, right) -> {
+ if (left.getKey() == null) {
+ return right.getKey() == null ? 0 : -1;
+ }
+ return right.getKey() == null ? 1 : left.getKey().compareTo(right.getKey());
+ });
+ for (Pair<BinaryString, Long> pair : sortedData) {
writer.write(pair.getKey(), pair.getValue());
}
@@ -487,4 +589,41 @@ public class LazyFilteredBitmapIndexReaderTest {
super.seek(desired);
}
}
+
+ private static class ByteArrayPositionOutputStream extends PositionOutputStream {
+
+ private final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ private boolean closed;
+
+ @Override
+ public long getPos() {
+ return output.size();
+ }
+
+ @Override
+ public void write(int b) {
+ output.write(b);
+ }
+
+ @Override
+ public void write(byte[] bytes) throws IOException {
+ output.write(bytes);
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length) {
+ output.write(bytes, offset, length);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ output.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ output.close();
+ }
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
index 6173c070f9..f97e80fca3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilder.java
@@ -245,20 +245,20 @@ public class SortedGlobalIndexBuilder implements Serializable {
public List<CommitMessage> buildForSinglePartition(
Range rowRange, BinaryRow partition, Iterator<InternalRow> data) throws IOException {
FieldGetter indexFieldGetter = InternalRow.createFieldGetter(indexField.type(), 0);
- SortedSingleColumnIndexWriter writer =
- new SortedSingleColumnIndexWriter(recordsPerRange, this::createWriter);
-
- while (data.hasNext()) {
- InternalRow row = data.next();
- long localRowId = row.getLong(1) - rowRange.from;
- writer.write(indexFieldGetter.getFieldOrNull(row), localRowId);
- }
+ try (SortedSingleColumnIndexWriter writer =
+ new SortedSingleColumnIndexWriter(recordsPerRange, this::createWriter)) {
+ while (data.hasNext()) {
+ InternalRow row = data.next();
+ long localRowId = row.getLong(1) - rowRange.from;
+ writer.write(indexFieldGetter.getFieldOrNull(row), localRowId);
+ }
- List<CommitMessage> commitMessages = new ArrayList<>();
- for (List<ResultEntry> resultEntries : writer.finish()) {
- commitMessages.add(flushIndex(rowRange, resultEntries, partition));
+ List<CommitMessage> commitMessages = new ArrayList<>();
+ for (List<ResultEntry> resultEntries : writer.finish()) {
+ commitMessages.add(flushIndex(rowRange, resultEntries, partition));
+ }
+ return commitMessages;
}
- return commitMessages;
}
public GlobalIndexSingleColumnWriter createWriter() throws IOException {
diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
index 6bc37a0f7d..80a8a17ff4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/sorted/SortedSingleColumnIndexWriter.java
@@ -23,6 +23,7 @@ import org.apache.paimon.globalindex.ResultEntry;
import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -32,7 +33,7 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
/** Rotates single-column index writers after a configured number of logical records. */
-public final class SortedSingleColumnIndexWriter {
+public final class SortedSingleColumnIndexWriter implements Closeable {
private final long recordsPerRange;
private final Factory factory;
@@ -73,6 +74,23 @@ public final class SortedSingleColumnIndexWriter {
return Collections.unmodifiableList(copy);
}
+ @Override
+ public void close() throws IOException {
+ finished = true;
+ currentRecordCount = 0;
+ GlobalIndexSingleColumnWriter writer = currentWriter;
+ currentWriter = null;
+ if (writer instanceof AutoCloseable) {
+ try {
+ ((AutoCloseable) writer).close();
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException("Failed to close the active sorted index writer.", e);
+ }
+ }
+ }
+
private void finishCurrentWriter() {
resultGroups.add(currentWriter.finish());
currentWriter = null;
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
index 5a8403c410..938c3ed10e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
@@ -51,6 +51,7 @@ public final class BucketedPrimaryKeyIndexMaintainer {
@Nullable private final BucketedVectorIndexMaintainer vectorMaintainer;
private final List<BucketedSortedIndexMaintainer> sortedMaintainers;
+ private int nextSortedMaintainerIndex;
private BucketedPrimaryKeyIndexMaintainer(
@Nullable BucketedVectorIndexMaintainer vectorMaintainer,
@@ -144,23 +145,21 @@ public final class BucketedPrimaryKeyIndexMaintainer {
CompactIncrement compactIncrement,
List<BucketedSortedIndexMaintainer.SortedIndexCommit> commits)
throws Exception {
- BucketedSortedIndexMaintainer active = activeSortedMaintainer();
for (BucketedSortedIndexMaintainer maintainer : sortedMaintainers) {
- commits.add(
- maintainer.prepareCommit(
- appendIncrement, compactIncrement, false, maintainer == active));
+ commits.add(maintainer.prepareCommit(appendIncrement, compactIncrement, false, false));
}
if (activeSortedMaintainer() != null) {
return;
}
- for (BucketedSortedIndexMaintainer maintainer : sortedMaintainers) {
- if (maintainer == active && !maintainer.state().uncoveredSourceFiles().isEmpty()) {
- continue;
- }
- if (!maintainer.state().uncoveredSourceFiles().isEmpty()) {
+ int count = sortedMaintainers.size();
+ for (int offset = 0; offset < count; offset++) {
+ int index = (nextSortedMaintainerIndex + offset) % count;
+ BucketedSortedIndexMaintainer maintainer = sortedMaintainers.get(index);
+ if (maintainer.hasPendingMaintenance()) {
commits.add(
maintainer.prepareCommit(appendIncrement, compactIncrement, false, true));
+ nextSortedMaintainerIndex = (index + 1) % count;
break;
}
}
@@ -312,7 +311,9 @@ public final class BucketedPrimaryKeyIndexMaintainer {
readerFactoryBuilder,
field,
definition.indexType(),
- definition.options()));
+ definition.options(),
+ definition.compactionLevelFanout(),
+ definition.compactionStaleRatioThreshold()));
break;
default:
throw new IllegalArgumentException(
@@ -370,16 +371,22 @@ public final class BucketedPrimaryKeyIndexMaintainer {
private final DataField field;
private final String indexType;
private final org.apache.paimon.options.Options options;
+ private final int compactionLevelFanout;
+ private final double compactionStaleRatioThreshold;
private SortedDefinitionFactory(
KeyValueFileReaderFactory.Builder readerFactoryBuilder,
DataField field,
String indexType,
- org.apache.paimon.options.Options options) {
+ org.apache.paimon.options.Options options,
+ int compactionLevelFanout,
+ double compactionStaleRatioThreshold) {
this.readerFactoryBuilder = readerFactoryBuilder;
this.field = field;
this.indexType = indexType;
this.options = options;
+ this.compactionLevelFanout = compactionLevelFanout;
+ this.compactionStaleRatioThreshold = compactionStaleRatioThreshold;
}
private BucketedSortedIndexMaintainer create(
@@ -405,6 +412,8 @@ public final class BucketedPrimaryKeyIndexMaintainer {
indexType,
indexFile,
builder::build,
+ compactionLevelFanout,
+ compactionStaleRatioThreshold,
restoredDataFiles,
restoredPayloads,
executor);
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
index 76f7b3a2ef..266f529051 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
@@ -35,14 +35,24 @@ public class PrimaryKeyIndexDefinition {
private final String indexType;
private final Options options;
private final Family family;
+ private final int compactionLevelFanout;
+ private final double compactionStaleRatioThreshold;
public PrimaryKeyIndexDefinition(
- String column, int fieldId, String indexType, Options options, Family family) {
+ String column,
+ int fieldId,
+ String indexType,
+ Options options,
+ Family family,
+ int compactionLevelFanout,
+ double compactionStaleRatioThreshold) {
this.column = column;
this.fieldId = fieldId;
this.indexType = indexType;
this.options = options;
this.family = family;
+ this.compactionLevelFanout = compactionLevelFanout;
+ this.compactionStaleRatioThreshold = compactionStaleRatioThreshold;
}
public String column() {
@@ -64,4 +74,12 @@ public class PrimaryKeyIndexDefinition {
public Family family() {
return family;
}
+
+ public int compactionLevelFanout() {
+ return compactionLevelFanout;
+ }
+
+ public double compactionStaleRatioThreshold() {
+ return compactionStaleRatioThreshold;
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
index 16f2fdaa3a..9d56db1e67 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
@@ -61,7 +61,9 @@ public class PrimaryKeyIndexDefinitions {
field.id(),
BTreeGlobalIndexerFactory.IDENTIFIER,
options.primaryKeyBTreeIndexOptions(column),
- PrimaryKeyIndexDefinition.Family.BTREE));
+ PrimaryKeyIndexDefinition.Family.BTREE,
+ options.primaryKeyIndexCompactionLevelFanout(column),
+ options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
} else if (bitmapColumns.contains(column)) {
definitions.add(
new PrimaryKeyIndexDefinition(
@@ -69,7 +71,9 @@ public class PrimaryKeyIndexDefinitions {
field.id(),
BitmapGlobalIndexerFactory.IDENTIFIER,
options.primaryKeyBitmapIndexOptions(column),
- PrimaryKeyIndexDefinition.Family.BITMAP));
+ PrimaryKeyIndexDefinition.Family.BITMAP,
+ options.primaryKeyIndexCompactionLevelFanout(column),
+ options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
} else if (vectorColumns.contains(column)) {
definitions.add(
new PrimaryKeyIndexDefinition(
@@ -77,7 +81,9 @@ public class PrimaryKeyIndexDefinitions {
field.id(),
options.primaryKeyVectorIndexType(column),
options.primaryKeyVectorIndexOptions(column),
- PrimaryKeyIndexDefinition.Family.VECTOR));
+ PrimaryKeyIndexDefinition.Family.VECTOR,
+ options.primaryKeyIndexCompactionLevelFanout(column),
+ options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnLevels.java b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexLevels.java
similarity index 61%
rename from paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnLevels.java
rename to paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexLevels.java
index bd028b529e..ff3dae74b0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnLevels.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexLevels.java
@@ -16,11 +16,8 @@
* limitations under the License.
*/
-package org.apache.paimon.index.pkvector;
+package org.apache.paimon.index.pk;
-import org.apache.paimon.index.IndexFileMeta;
-import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
-import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.io.DataFileMeta;
import java.util.ArrayList;
@@ -30,37 +27,47 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
+import java.util.function.Function;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Derives logical ANN compaction levels from immutable segment source metadata. */
-final class PkVectorAnnLevels {
+/** Derives logical compaction levels from immutable primary-key index source metadata. */
+public final class PrimaryKeyIndexLevels<T> {
private final int fanout;
private final double staleRatioThreshold;
-
- PkVectorAnnLevels(int fanout, double staleRatioThreshold) {
- checkArgument(fanout > 1, "ANN level fanout must be greater than one.");
+ private final Function<T, String> identity;
+ private final Function<T, List<PrimaryKeyIndexSourceFile>> sources;
+
+ public PrimaryKeyIndexLevels(
+ int fanout,
+ double staleRatioThreshold,
+ Function<T, String> identity,
+ Function<T, List<PrimaryKeyIndexSourceFile>> sources) {
+ checkArgument(fanout > 1, "Primary-key index level fanout must be greater than one.");
checkArgument(
staleRatioThreshold > 0 && staleRatioThreshold <= 1,
- "ANN stale ratio threshold must be in (0, 1].");
+ "Primary-key index stale ratio threshold must be in (0, 1].");
this.fanout = fanout;
this.staleRatioThreshold = staleRatioThreshold;
+ this.identity = identity;
+ this.sources = sources;
}
- Optional<Plan> pick(List<IndexFileMeta> segments, Map<String, DataFileMeta> activeSourceFiles) {
- IndexFileMeta staleCandidate = null;
+ public Optional<Plan<T>> pick(List<T> units, Map<String, DataFileMeta> activeSourceFiles) {
+ T staleCandidate = null;
double highestStaleRatio = -1;
- for (IndexFileMeta segment : segments) {
- double staleRatio = staleRatio(segment, activeSourceFiles);
+ for (T unit : units) {
+ double staleRatio = staleRatio(unit, activeSourceFiles);
if (staleRatio >= staleRatioThreshold
&& (staleRatio > highestStaleRatio
|| (staleRatio == highestStaleRatio
&& (staleCandidate == null
- || segment.fileName()
- .compareTo(staleCandidate.fileName())
+ || identity.apply(unit)
+ .compareTo(
+ identity.apply(staleCandidate))
< 0)))) {
- staleCandidate = segment;
+ staleCandidate = unit;
highestStaleRatio = staleRatio;
}
}
@@ -69,10 +76,9 @@ final class PkVectorAnnLevels {
createPlan(Collections.singletonList(staleCandidate), activeSourceFiles));
}
- List<IndexFileMeta> candidates = new ArrayList<>(segments);
+ List<T> candidates = new ArrayList<>(units);
candidates.sort(
- Comparator.comparingLong(PkVectorAnnLevels::buildRowCount)
- .thenComparing(IndexFileMeta::fileName));
+ Comparator.comparingLong(this::buildRowCount).thenComparing(identity::apply));
for (int start = 0; start + fanout <= candidates.size(); start++) {
long smallest = buildRowCount(candidates.get(start));
long largest = buildRowCount(candidates.get(start + fanout - 1));
@@ -86,12 +92,10 @@ final class PkVectorAnnLevels {
return Optional.empty();
}
- private static double staleRatio(
- IndexFileMeta segment, Map<String, DataFileMeta> activeSourceFiles) {
+ private double staleRatio(T unit, Map<String, DataFileMeta> activeSourceFiles) {
long totalRows = 0;
long staleRows = 0;
- for (PrimaryKeyIndexSourceFile source :
- PrimaryKeyIndexSourceMeta.fromIndexFile(segment).sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile source : sources.apply(unit)) {
totalRows = Math.addExact(totalRows, source.rowCount());
DataFileMeta active = activeSourceFiles.get(source.fileName());
if (active == null) {
@@ -99,36 +103,33 @@ final class PkVectorAnnLevels {
} else {
checkArgument(
active.rowCount() == source.rowCount(),
- "ANN source %s row count does not match active data file.",
+ "Primary-key index source %s row count does not match active data file.",
source.fileName());
}
}
return totalRows == 0 ? 0 : ((double) staleRows) / totalRows;
}
- private static Plan createPlan(
- List<IndexFileMeta> inputSegments, Map<String, DataFileMeta> activeSourceFiles) {
+ private Plan<T> createPlan(List<T> inputUnits, Map<String, DataFileMeta> activeSourceFiles) {
Map<String, DataFileMeta> selectedSources = new TreeMap<>();
- for (IndexFileMeta segment : inputSegments) {
- for (PrimaryKeyIndexSourceFile source :
- PrimaryKeyIndexSourceMeta.fromIndexFile(segment).sourceFiles()) {
+ for (T unit : inputUnits) {
+ for (PrimaryKeyIndexSourceFile source : sources.apply(unit)) {
DataFileMeta active = activeSourceFiles.get(source.fileName());
if (active != null) {
checkArgument(
active.rowCount() == source.rowCount(),
- "ANN source %s row count does not match active data file.",
+ "Primary-key index source %s row count does not match active data file.",
source.fileName());
selectedSources.put(active.fileName(), active);
}
}
}
- return new Plan(inputSegments, new ArrayList<>(selectedSources.values()));
+ return new Plan<>(inputUnits, new ArrayList<>(selectedSources.values()));
}
- private static long buildRowCount(IndexFileMeta segment) {
+ private long buildRowCount(T unit) {
long rowCount = 0;
- for (PrimaryKeyIndexSourceFile source :
- PrimaryKeyIndexSourceMeta.fromIndexFile(segment).sourceFiles()) {
+ for (PrimaryKeyIndexSourceFile source : sources.apply(unit)) {
rowCount = Math.addExact(rowCount, source.rowCount());
}
return rowCount;
@@ -141,22 +142,22 @@ final class PkVectorAnnLevels {
return value * multiplier;
}
- /** A deterministic ANN rebuild selection. */
- static final class Plan {
+ /** A deterministic primary-key index rebuild selection. */
+ public static final class Plan<T> {
- private final List<IndexFileMeta> inputSegments;
+ private final List<T> inputUnits;
private final List<DataFileMeta> sourceFiles;
- private Plan(List<IndexFileMeta> inputSegments, List<DataFileMeta> sourceFiles) {
- this.inputSegments = Collections.unmodifiableList(new ArrayList<>(inputSegments));
+ private Plan(List<T> inputUnits, List<DataFileMeta> sourceFiles) {
+ this.inputUnits = Collections.unmodifiableList(new ArrayList<>(inputUnits));
this.sourceFiles = Collections.unmodifiableList(new ArrayList<>(sourceFiles));
}
- List<IndexFileMeta> inputSegments() {
- return inputSegments;
+ public List<T> inputUnits() {
+ return inputUnits;
}
- List<DataFileMeta> sourceFiles() {
+ public List<DataFileMeta> sourceFiles() {
return sourceFiles;
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java
index 44695b5f54..a0216da28b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainer.java
@@ -19,6 +19,7 @@
package org.apache.paimon.index.pksorted;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexLevels;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourcePolicy;
import org.apache.paimon.io.CompactIncrement;
@@ -33,18 +34,14 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
-import java.util.HashSet;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -59,6 +56,7 @@ public class BucketedSortedIndexMaintainer {
private final String indexType;
private final PkSortedIndexFile indexFile;
private final BuildFunction buildFunction;
+ private final PrimaryKeyIndexLevels<PkSortedIndexGroup> levels;
private final Map<String, DataFileMeta> activeSourceFiles = new LinkedHashMap<>();
private final List<PkSortedIndexGroup> groups = new ArrayList<>();
private final List<IndexFileMeta> pendingRestoredDeletions = new ArrayList<>();
@@ -70,6 +68,8 @@ public class BucketedSortedIndexMaintainer {
String indexType,
PkSortedIndexFile indexFile,
BuildFunction buildFunction,
+ int levelFanout,
+ double staleRatioThreshold,
List<DataFileMeta> restoredDataFiles,
List<IndexFileMeta> restoredPayloads,
ExecutorService executor) {
@@ -77,6 +77,12 @@ public class BucketedSortedIndexMaintainer {
this.indexType = indexType;
this.indexFile = indexFile;
this.buildFunction = buildFunction;
+ this.levels =
+ new PrimaryKeyIndexLevels<>(
+ levelFanout,
+ staleRatioThreshold,
+ PkSortedIndexGroup::identity,
+ PkSortedIndexGroup::sourceFiles);
this.executor = executor;
for (DataFileMeta dataFile : restoredDataFiles) {
if (PrimaryKeyIndexSourcePolicy.shouldRead(dataFile)) {
@@ -129,29 +135,29 @@ public class BucketedSortedIndexMaintainer {
List<IndexFileMeta> removed = new ArrayList<>(pendingRestoredDeletions);
pendingRestoredDeletions.clear();
- removed.addAll(removeInactiveGroups());
- Set<String> failedSources = new HashSet<>();
while (true) {
- Optional<CompletedBuild> completed =
- finishPendingBuild(waitCompaction, failedSources);
+ Optional<CompletedBuild> completed = finishPendingBuild(waitCompaction);
if (completed.isPresent()) {
- acceptOrDelete(completed.get(), created, failedSources);
+ acceptOrDelete(completed.get(), created, removed);
}
if (pendingBuild == null && allowBuildStart) {
- DataFileMeta uncovered = firstUncoveredSource(failedSources);
+ DataFileMeta uncovered = firstUncoveredSource();
if (uncovered != null) {
- PendingBuild next = new PendingBuild(uncovered);
- try {
- next.start();
- pendingBuild = next;
- } catch (RejectedExecutionException e) {
- failedSources.add(sourceIdentity(uncovered));
- LOG.warn(
- "Primary-key {} index build for source file {} was rejected.",
- indexType,
- uncovered.fileName(),
- e);
+ startBuild(Collections.singletonList(uncovered), Collections.emptyList());
+ } else {
+ Optional<PrimaryKeyIndexLevels.Plan<PkSortedIndexGroup>> plan =
+ levels.pick(groups, activeSourceFiles);
+ if (plan.isPresent()) {
+ if (plan.get().sourceFiles().isEmpty()) {
+ replaceInputGroups(
+ plan.get().inputUnits(),
+ Optional.empty(),
+ created,
+ removed);
+ continue;
+ }
+ startBuild(plan.get().sourceFiles(), plan.get().inputUnits());
}
}
}
@@ -240,115 +246,150 @@ public class BucketedSortedIndexMaintainer {
}
}
- private List<IndexFileMeta> removeInactiveGroups() {
- List<IndexFileMeta> removed = new ArrayList<>();
- Iterator<PkSortedIndexGroup> iterator = groups.iterator();
- while (iterator.hasNext()) {
- PkSortedIndexGroup group = iterator.next();
- DataFileMeta active = activeSourceFiles.get(group.sourceFile().fileName());
- if (active == null || active.rowCount() != group.sourceFile().rowCount()) {
- iterator.remove();
- removed.addAll(group.payloads());
- }
- }
- return removed;
- }
-
@Nullable
- private DataFileMeta firstUncoveredSource(Set<String> failedSources) {
+ private DataFileMeta firstUncoveredSource() {
List<DataFileMeta> candidates = new ArrayList<>(activeSourceFiles.values());
candidates.sort(Comparator.comparing(DataFileMeta::fileName));
for (DataFileMeta candidate : candidates) {
- if (!isCovered(candidate) && !failedSources.contains(sourceIdentity(candidate))) {
+ if (!isCovered(candidate, Collections.emptyList())) {
return candidate;
}
}
return null;
}
- private boolean isCovered(DataFileMeta candidate) {
+ private boolean isCovered(DataFileMeta candidate, List<PkSortedIndexGroup> excludedGroups) {
for (PkSortedIndexGroup group : groups) {
- if (group.sourceFile().fileName().equals(candidate.fileName())
- && group.sourceFile().rowCount() == candidate.rowCount()) {
- return true;
+ if (excludedGroups.contains(group)) {
+ continue;
+ }
+ for (PrimaryKeyIndexSourceFile source : group.sourceFiles()) {
+ if (source.fileName().equals(candidate.fileName())
+ && source.rowCount() == candidate.rowCount()) {
+ return true;
+ }
}
}
return false;
}
- private Optional<CompletedBuild> finishPendingBuild(boolean blocking, Set<String> failedSources)
- throws InterruptedException {
+ private void startBuild(List<DataFileMeta> sourceFiles, List<PkSortedIndexGroup> inputGroups) {
+ PendingBuild next = new PendingBuild(sourceFiles, inputGroups);
+ next.start();
+ pendingBuild = next;
+ }
+
+ private Optional<CompletedBuild> finishPendingBuild(boolean blocking) throws Exception {
if (pendingBuild == null || (!blocking && !pendingBuild.isDone())) {
return Optional.empty();
}
PendingBuild completed = pendingBuild;
try {
- List<IndexFileMeta> payloads = completed.get();
+ IndexFileMeta payload = completed.get();
pendingBuild = null;
- return Optional.of(new CompletedBuild(completed.sourceFile, payloads));
+ return Optional.of(
+ new CompletedBuild(completed.sourceFiles, completed.inputGroups, payload));
} catch (CancellationException e) {
pendingBuild = null;
- failedSources.add(sourceIdentity(completed.sourceFile));
- return Optional.empty();
+ throw e;
} catch (ExecutionException e) {
pendingBuild = null;
- failedSources.add(sourceIdentity(completed.sourceFile));
- LOG.warn(
- "Primary-key {} index build for source file {} failed after {} attempts; "
- + "the source remains uncovered.",
- indexType,
- completed.sourceFile.fileName(),
- MAX_BUILD_ATTEMPTS,
- e.getCause());
- return Optional.empty();
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new RuntimeException(cause);
}
}
private void acceptOrDelete(
- CompletedBuild completed, List<IndexFileMeta> created, Set<String> failedSources) {
- DataFileMeta active = activeSourceFiles.get(completed.sourceFile.fileName());
- PrimaryKeyIndexSourceFile source =
- new PrimaryKeyIndexSourceFile(
- completed.sourceFile.fileName(), completed.sourceFile.rowCount());
+ CompletedBuild completed, List<IndexFileMeta> created, List<IndexFileMeta> removed) {
+ List<PrimaryKeyIndexSourceFile> sources = new ArrayList<>();
+ boolean sourcesStillActive = true;
+ for (DataFileMeta sourceFile : completed.sourceFiles) {
+ sources.add(
+ new PrimaryKeyIndexSourceFile(sourceFile.fileName(), sourceFile.rowCount()));
+ DataFileMeta active = activeSourceFiles.get(sourceFile.fileName());
+ if (active == null || active.rowCount() != sourceFile.rowCount()) {
+ sourcesStillActive = false;
+ }
+ }
+ boolean inputsStillPresent = groups.containsAll(completed.inputGroups);
+ boolean outputOverlapsRetainedGroup = false;
+ for (DataFileMeta sourceFile : completed.sourceFiles) {
+ if (isCovered(sourceFile, completed.inputGroups)) {
+ outputOverlapsRetainedGroup = true;
+ break;
+ }
+ }
+ if (!sourcesStillActive || !inputsStillPresent || outputOverlapsRetainedGroup) {
+ deleteGenerated(completed.payload);
+ return;
+ }
Optional<PkSortedIndexGroup> group;
try {
- group = PkSortedIndexGroup.create(fieldId, indexType, source, completed.payloads);
+ group =
+ PkSortedIndexGroup.create(
+ fieldId,
+ indexType,
+ sources,
+ Collections.singletonList(completed.payload));
} catch (RuntimeException e) {
- failedSources.add(sourceIdentity(completed.sourceFile));
- deleteGenerated(completed.payloads);
- LOG.warn(
- "Primary-key {} index build for source file {} produced invalid metadata.",
- indexType,
- completed.sourceFile.fileName(),
- e);
- return;
+ deleteGenerated(completed.payload);
+ throw new IllegalStateException(
+ "Primary-key " + indexType + " index build produced invalid metadata.", e);
}
- if (active == null
- || active.rowCount() != completed.sourceFile.rowCount()
- || isCovered(active)
- || !group.isPresent()) {
- deleteGenerated(completed.payloads);
- failedSources.add(sourceIdentity(completed.sourceFile));
- return;
+ if (!group.isPresent()) {
+ deleteGenerated(completed.payload);
+ throw new IllegalStateException(
+ "Primary-key " + indexType + " index build produced an incomplete group.");
}
- groups.add(group.get());
- created.addAll(completed.payloads);
+ replaceInputGroups(completed.inputGroups, group, created, removed);
}
- private void deleteGenerated(List<IndexFileMeta> payloads) {
- for (IndexFileMeta payload : payloads) {
- try {
- indexFile.delete(payload);
- } catch (RuntimeException e) {
- LOG.warn("Failed to delete unpublished primary-key sorted index payload.", e);
+ private void replaceInputGroups(
+ List<PkSortedIndexGroup> inputGroups,
+ Optional<PkSortedIndexGroup> outputGroup,
+ List<IndexFileMeta> created,
+ List<IndexFileMeta> removed) {
+ for (PkSortedIndexGroup inputGroup : inputGroups) {
+ groups.remove(inputGroup);
+ for (IndexFileMeta payload : inputGroup.payloads()) {
+ if (created.remove(payload)) {
+ indexFile.delete(payload);
+ } else {
+ removed.add(payload);
+ }
}
}
+ if (outputGroup.isPresent()) {
+ groups.add(outputGroup.get());
+ created.addAll(outputGroup.get().payloads());
+ }
+ }
+
+ private void deleteGenerated(IndexFileMeta payload) {
+ try {
+ indexFile.delete(payload);
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to delete unpublished primary-key sorted index payload.", e);
+ }
}
public synchronized boolean buildNotCompleted() {
return pendingBuild != null;
}
+ public synchronized boolean hasPendingMaintenance() {
+ return pendingBuild != null
+ || !pendingRestoredDeletions.isEmpty()
+ || firstUncoveredSource() != null
+ || levels.pick(groups, activeSourceFiles).isPresent();
+ }
+
public int fieldId() {
return fieldId;
}
@@ -389,35 +430,37 @@ public class BucketedSortedIndexMaintainer {
private final class PendingBuild {
- private final DataFileMeta sourceFile;
- @Nullable private List<IndexFileMeta> result;
- @Nullable private Future<List<IndexFileMeta>> future;
+ private final List<DataFileMeta> sourceFiles;
+ private final List<PkSortedIndexGroup> inputGroups;
+ @Nullable private IndexFileMeta result;
+ @Nullable private Future<IndexFileMeta> future;
private boolean cancelled;
- private PendingBuild(DataFileMeta sourceFile) {
- this.sourceFile = sourceFile;
+ private PendingBuild(List<DataFileMeta> sourceFiles, List<PkSortedIndexGroup> inputGroups) {
+ this.sourceFiles = Collections.unmodifiableList(new ArrayList<>(sourceFiles));
+ this.inputGroups = Collections.unmodifiableList(new ArrayList<>(inputGroups));
}
private void start() {
future =
executor.submit(
() -> {
- List<IndexFileMeta> payloads = buildWithRetries();
+ IndexFileMeta payload = buildWithRetries();
synchronized (PendingBuild.this) {
if (!cancelled) {
- result = payloads;
- return payloads;
+ result = payload;
+ return payload;
}
}
- deleteGenerated(payloads);
+ deleteGenerated(payload);
throw new CancellationException();
});
}
- private List<IndexFileMeta> buildWithRetries() throws Exception {
+ private IndexFileMeta buildWithRetries() throws Exception {
for (int attempt = 1; ; attempt++) {
try {
- return buildFunction.build(sourceFile);
+ return buildFunction.build(sourceFiles);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new CancellationException();
@@ -439,36 +482,41 @@ public class BucketedSortedIndexMaintainer {
return future.isDone();
}
- private List<IndexFileMeta> get() throws InterruptedException, ExecutionException {
+ private IndexFileMeta get() throws InterruptedException, ExecutionException {
return future.get();
}
private void cancel() {
- Future<List<IndexFileMeta>> buildFuture;
- List<IndexFileMeta> payloads;
+ Future<IndexFileMeta> buildFuture;
+ IndexFileMeta payload;
synchronized (this) {
cancelled = true;
buildFuture = future;
- payloads = result;
+ payload = result;
result = null;
}
if (buildFuture != null) {
buildFuture.cancel(true);
}
- if (payloads != null) {
- deleteGenerated(payloads);
+ if (payload != null) {
+ deleteGenerated(payload);
}
}
}
private static final class CompletedBuild {
- private final DataFileMeta sourceFile;
- private final List<IndexFileMeta> payloads;
+ private final List<DataFileMeta> sourceFiles;
+ private final List<PkSortedIndexGroup> inputGroups;
+ private final IndexFileMeta payload;
- private CompletedBuild(DataFileMeta sourceFile, List<IndexFileMeta> payloads) {
- this.sourceFile = sourceFile;
- this.payloads = payloads;
+ private CompletedBuild(
+ List<DataFileMeta> sourceFiles,
+ List<PkSortedIndexGroup> inputGroups,
+ IndexFileMeta payload) {
+ this.sourceFiles = sourceFiles;
+ this.inputGroups = inputGroups;
+ this.payload = payload;
}
}
@@ -491,15 +539,11 @@ public class BucketedSortedIndexMaintainer {
return false;
}
- private static String sourceIdentity(DataFileMeta file) {
- return file.fileName() + '\0' + file.rowCount();
- }
-
- /** Builds all rotated payloads for one physical source file. */
+ /** Builds one payload for ordered physical source files. */
@FunctionalInterface
public interface BuildFunction {
- List<IndexFileMeta> build(DataFileMeta sourceFile) throws Exception;
+ IndexFileMeta build(List<DataFileMeta> sourceFiles) throws Exception;
}
/** Sorted-index changes for append and compact snapshot routing. */
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexState.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexState.java
index 8fd6729522..078c719680 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexState.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexState.java
@@ -24,10 +24,12 @@ import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
/** Immutable sorted-index state for one field and bucket. */
public final class PkSortedBucketIndexState {
@@ -53,14 +55,15 @@ public final class PkSortedBucketIndexState {
String indexType,
List<PrimaryKeyIndexSourceFile> activeSourceFiles,
List<IndexFileMeta> activePayloads) {
- Map<String, List<IndexFileMeta>> payloadsBySource = new LinkedHashMap<>();
+ Map<List<PrimaryKeyIndexSourceFile>, List<IndexFileMeta>> payloadsBySources =
+ new LinkedHashMap<>();
List<IndexFileMeta> rejected = new ArrayList<>();
for (IndexFileMeta payload : activePayloads) {
try {
- PrimaryKeyIndexSourceFile sourceFile =
- PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFile();
- payloadsBySource
- .computeIfAbsent(sourceFile.fileName(), key -> new ArrayList<>())
+ List<PrimaryKeyIndexSourceFile> sourceFiles =
+ PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFiles();
+ payloadsBySources
+ .computeIfAbsent(sourceFiles, key -> new ArrayList<>())
.add(payload);
} catch (RuntimeException ignored) {
rejected.add(payload);
@@ -68,27 +71,36 @@ public final class PkSortedBucketIndexState {
}
List<PkSortedIndexGroup> groups = new ArrayList<>();
+ Set<PrimaryKeyIndexSourceFile> activeSet = new HashSet<>(activeSourceFiles);
+ Set<PrimaryKeyIndexSourceFile> coveredSet = new HashSet<>();
+ for (Map.Entry<List<PrimaryKeyIndexSourceFile>, List<IndexFileMeta>> entry :
+ payloadsBySources.entrySet()) {
+ Optional<PkSortedIndexGroup> group =
+ PkSortedIndexGroup.create(fieldId, indexType, entry.getKey(), entry.getValue());
+ boolean overlapsActiveSource = false;
+ for (PrimaryKeyIndexSourceFile sourceFile : entry.getKey()) {
+ if (activeSet.contains(sourceFile) && coveredSet.contains(sourceFile)) {
+ overlapsActiveSource = true;
+ break;
+ }
+ }
+ if (group.isPresent() && !overlapsActiveSource) {
+ groups.add(group.get());
+ coveredSet.addAll(entry.getKey());
+ } else {
+ rejected.addAll(entry.getValue());
+ }
+ }
+
List<PrimaryKeyIndexSourceFile> covered = new ArrayList<>();
List<PrimaryKeyIndexSourceFile> uncovered = new ArrayList<>();
for (PrimaryKeyIndexSourceFile sourceFile : activeSourceFiles) {
- List<IndexFileMeta> payloads = payloadsBySource.remove(sourceFile.fileName());
- if (payloads == null || payloads.isEmpty()) {
- uncovered.add(sourceFile);
+ if (coveredSet.contains(sourceFile)) {
+ covered.add(sourceFile);
} else {
- Optional<PkSortedIndexGroup> group =
- PkSortedIndexGroup.create(fieldId, indexType, sourceFile, payloads);
- if (group.isPresent()) {
- groups.add(group.get());
- covered.add(sourceFile);
- } else {
- uncovered.add(sourceFile);
- rejected.addAll(payloads);
- }
+ uncovered.add(sourceFile);
}
}
- for (List<IndexFileMeta> inactivePayloads : payloadsBySource.values()) {
- rejected.addAll(inactivePayloads);
- }
return new PkSortedBucketIndexState(groups, covered, uncovered, rejected);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
index 7a7604abe9..ff78d72a6a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilder.java
@@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.options.Options;
@@ -37,15 +38,17 @@ import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Sorts one physical data file and builds its source-backed scalar index payloads. */
+/** Sorts physical data files and builds their source-backed scalar index payloads. */
public class PkSortedIndexBuilder {
- private static final int LOCAL_ROW_ID_FIELD_ID = Integer.MAX_VALUE;
+ private static final int ROW_ID_FIELD_ID = Integer.MAX_VALUE;
private final ReaderFactory readerFactory;
private final PkSortedIndexFile indexFile;
@@ -79,8 +82,11 @@ public class PkSortedIndexBuilder {
this.ioManager = ioManager;
}
- public List<org.apache.paimon.index.IndexFileMeta> build(DataFileMeta dataFile)
- throws IOException {
+ public IndexFileMeta build(List<DataFileMeta> dataFiles) throws IOException {
+ checkArgument(!dataFiles.isEmpty(), "A sorted index build requires source files.");
+ List<DataFileMeta> orderedDataFiles = new ArrayList<>(dataFiles);
+ orderedDataFiles.sort(Comparator.comparing(DataFileMeta::fileName));
+
IOManager actualIOManager = ioManager;
boolean ownsIOManager = false;
if (actualIOManager == null) {
@@ -95,9 +101,7 @@ public class PkSortedIndexBuilder {
RowType.of(
indexField,
new DataField(
- LOCAL_ROW_ID_FIELD_ID,
- "_LOCAL_ROW_ID",
- DataTypes.BIGINT().notNull()));
+ ROW_ID_FIELD_ID, "_ROW_ID", DataTypes.BIGINT().notNull()));
sortBuffer =
BinaryExternalSortBuffer.create(
actualIOManager,
@@ -109,17 +113,34 @@ public class PkSortedIndexBuilder {
coreOptions.spillCompressOptions(),
coreOptions.writeBufferSpillDiskSize());
- try (Reader reader = readerFactory.create(dataFile)) {
- checkArgument(
- reader.rowCount() == dataFile.rowCount(),
- "Sorted reader row count %s does not match data file %s row count %s.",
- reader.rowCount(),
- dataFile.fileName(),
- dataFile.rowCount());
- PkSortedDataFileReader.Entry entry;
- while ((entry = reader.readNext()) != null) {
- sortBuffer.write(GenericRow.of(entry.value(), entry.rowPosition()));
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new ArrayList<>();
+ long sourceOffset = 0;
+ for (DataFileMeta dataFile : orderedDataFiles) {
+ sourceFiles.add(
+ new PrimaryKeyIndexSourceFile(dataFile.fileName(), dataFile.rowCount()));
+ try (Reader reader = readerFactory.create(dataFile)) {
+ checkArgument(
+ reader.rowCount() == dataFile.rowCount(),
+ "Sorted reader row count %s does not match data file %s row count %s.",
+ reader.rowCount(),
+ dataFile.fileName(),
+ dataFile.rowCount());
+ PkSortedDataFileReader.Entry entry;
+ while ((entry = reader.readNext()) != null) {
+ checkArgument(
+ entry.rowPosition() >= 0
+ && entry.rowPosition() < dataFile.rowCount(),
+ "Row position %s is outside data file %s row range [0, %s).",
+ entry.rowPosition(),
+ dataFile.fileName(),
+ dataFile.rowCount());
+ sortBuffer.write(
+ GenericRow.of(
+ entry.value(),
+ Math.addExact(sourceOffset, entry.rowPosition())));
+ }
}
+ sourceOffset = Math.addExact(sourceOffset, dataFile.rowCount());
}
Iterator<InternalRow> sortedRows =
@@ -142,12 +163,7 @@ public class PkSortedIndexBuilder {
valueGetter.getFieldOrNull(row), row.getLong(1));
}
};
- return indexFile.build(
- new PrimaryKeyIndexSourceFile(dataFile.fileName(), dataFile.rowCount()),
- indexField,
- indexType,
- options,
- sortedEntries);
+ return indexFile.build(sourceFiles, indexField, indexType, options, sortedEntries);
} finally {
if (sortBuffer != null) {
sortBuffer.clear();
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
index f6ab090e48..7737393d25 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java
@@ -26,8 +26,6 @@ import org.apache.paimon.globalindex.GlobalIndexWriter;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
-import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
-import org.apache.paimon.globalindex.sorted.SortedSingleColumnIndexWriter;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFile;
import org.apache.paimon.index.IndexFileMeta;
@@ -36,12 +34,11 @@ import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.IOUtils;
import javax.annotation.Nullable;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -49,86 +46,83 @@ import java.util.Map;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Builds source-backed BTree or Bitmap payloads for one physical data file. */
+/** Builds source-backed BTree or Bitmap payloads for ordered physical data files. */
public class PkSortedIndexFile extends IndexFile {
public PkSortedIndexFile(FileIO fileIO, IndexPathFactory pathFactory) {
super(fileIO, pathFactory);
}
- public List<IndexFileMeta> build(
- PrimaryKeyIndexSourceFile sourceFile,
+ public IndexFileMeta build(
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
DataField indexField,
String indexType,
Options indexOptions,
Iterator<Entry> sortedEntries)
throws IOException {
+ long sourceRowCount = 0;
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
+ sourceRowCount = Math.addExact(sourceRowCount, sourceFile.rowCount());
+ }
checkArgument(
- sourceFile.rowCount() > 0,
- "A sorted index group must reference at least one source row.");
+ sourceRowCount > 0, "A sorted index group must reference at least one source row.");
TrackingFileWriter fileWriter = new TrackingFileWriter();
+ GlobalIndexSingleColumnWriter writer = null;
boolean success = false;
try {
- long recordsPerRange =
- indexOptions.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE);
- SortedSingleColumnIndexWriter writer =
- new SortedSingleColumnIndexWriter(
- recordsPerRange,
- () -> createWriter(indexType, indexField, indexOptions, fileWriter));
+ writer = createWriter(indexType, indexField, indexOptions, fileWriter);
long writtenRows = 0;
while (sortedEntries.hasNext()) {
Entry entry = sortedEntries.next();
checkArgument(
- entry.localRowId >= 0 && entry.localRowId < sourceFile.rowCount(),
- "Local row id %s is outside source file %s row range [0, %s).",
- entry.localRowId,
- sourceFile.fileName(),
- sourceFile.rowCount());
- writer.write(entry.value, entry.localRowId);
+ entry.rowId >= 0 && entry.rowId < sourceRowCount,
+ "Row id %s is outside sorted index group row range [0, %s).",
+ entry.rowId,
+ sourceRowCount);
+ writer.write(entry.value, entry.rowId);
writtenRows++;
}
checkArgument(
- writtenRows == sourceFile.rowCount(),
- "Sorted index input row count %s does not match source file %s row count %s.",
+ writtenRows == sourceRowCount,
+ "Sorted index input row count %s does not match source row count %s.",
writtenRows,
- sourceFile.fileName(),
- sourceFile.rowCount());
-
- List<List<ResultEntry>> resultGroups = writer.finish();
- List<IndexFileMeta> payloads = new ArrayList<>();
- long payloadRows = 0;
- byte[] sourceMeta = new PrimaryKeyIndexSourceMeta(sourceFile).serialize();
- for (List<ResultEntry> resultGroup : resultGroups) {
- for (ResultEntry result : resultGroup) {
- payloadRows = Math.addExact(payloadRows, result.rowCount());
- Path payloadPath = fileWriter.path(result.fileName());
- payloads.add(
- new IndexFileMeta(
- indexType,
- result.fileName(),
- fileIO.getFileSize(payloadPath),
- result.rowCount(),
- new GlobalIndexMeta(
- 0,
- sourceFile.rowCount() - 1,
- indexField.id(),
- null,
- result.meta(),
- sourceMeta),
- pathFactory.isExternalPath() ? payloadPath.toString() : null));
- }
- }
+ sourceRowCount);
+
+ List<ResultEntry> results = writer.finish();
checkArgument(
- payloadRows == sourceFile.rowCount(),
- "Sorted payload row count %s does not match source file %s row count %s.",
- payloadRows,
- sourceFile.fileName(),
- sourceFile.rowCount());
+ results.size() == 1,
+ "Sorted index build must produce exactly one payload file, but produced %s.",
+ results.size());
+ ResultEntry result = results.get(0);
+ checkArgument(
+ result.rowCount() == sourceRowCount,
+ "Sorted payload row count %s does not match source row count %s.",
+ result.rowCount(),
+ sourceRowCount);
+ byte[] sourceMeta = new PrimaryKeyIndexSourceMeta(sourceFiles).serialize();
+ Path payloadPath = fileWriter.path(result.fileName());
+ IndexFileMeta payload =
+ new IndexFileMeta(
+ indexType,
+ result.fileName(),
+ fileIO.getFileSize(payloadPath),
+ result.rowCount(),
+ new GlobalIndexMeta(
+ 0,
+ sourceRowCount - 1,
+ indexField.id(),
+ null,
+ result.meta(),
+ sourceMeta),
+ pathFactory.isExternalPath() ? payloadPath.toString() : null);
success = true;
- return Collections.unmodifiableList(payloads);
+ return payload;
} finally {
+ if (writer instanceof AutoCloseable) {
+ IOUtils.closeQuietly((AutoCloseable) writer);
+ }
if (!success) {
fileWriter.deleteCreatedFiles();
}
@@ -150,15 +144,15 @@ public class PkSortedIndexFile extends IndexFile {
return (GlobalIndexSingleColumnWriter) writer;
}
- /** One sorted scalar value and its zero-based position in the source data file. */
+ /** One sorted scalar value and its zero-based ordinal in the ordered source group. */
public static final class Entry {
@Nullable private final Object value;
- private final long localRowId;
+ private final long rowId;
- public Entry(@Nullable Object value, long localRowId) {
+ public Entry(@Nullable Object value, long rowId) {
this.value = value;
- this.localRowId = localRowId;
+ this.rowId = rowId;
}
@Nullable
@@ -166,8 +160,8 @@ public class PkSortedIndexFile extends IndexFile {
return value;
}
- public long localRowId() {
- return localRowId;
+ public long rowId() {
+ return rowId;
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexGroup.java b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexGroup.java
index 46f161eadb..bd0acfe8c9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexGroup.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexGroup.java
@@ -30,14 +30,14 @@ import java.util.List;
import java.util.Optional;
import java.util.Set;
-/** All rotated payloads that index one physical source data file. */
+/** All rotated payloads that index the same ordered source data files. */
public final class PkSortedIndexGroup {
- private final PrimaryKeyIndexSourceFile sourceFile;
+ private final List<PrimaryKeyIndexSourceFile> sourceFiles;
private final List<IndexFileMeta> payloads;
- PkSortedIndexGroup(PrimaryKeyIndexSourceFile sourceFile, List<IndexFileMeta> payloads) {
- this.sourceFile = sourceFile;
+ PkSortedIndexGroup(List<PrimaryKeyIndexSourceFile> sourceFiles, List<IndexFileMeta> payloads) {
+ this.sourceFiles = Collections.unmodifiableList(new ArrayList<>(sourceFiles));
this.payloads = Collections.unmodifiableList(new ArrayList<>(payloads));
}
@@ -46,34 +46,84 @@ public final class PkSortedIndexGroup {
String indexType,
PrimaryKeyIndexSourceFile sourceFile,
List<IndexFileMeta> payloads) {
+ return create(fieldId, indexType, Collections.singletonList(sourceFile), payloads);
+ }
+
+ static Optional<PkSortedIndexGroup> create(
+ int fieldId,
+ String indexType,
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
+ List<IndexFileMeta> payloads) {
+ long sourceRowCount = 0;
+ Set<String> sourceNames = new HashSet<>();
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
+ if (!sourceNames.add(sourceFile.fileName())) {
+ return Optional.empty();
+ }
+ try {
+ sourceRowCount = Math.addExact(sourceRowCount, sourceFile.rowCount());
+ } catch (ArithmeticException e) {
+ return Optional.empty();
+ }
+ }
+ if (sourceFiles.isEmpty()) {
+ return Optional.empty();
+ }
+
long payloadRowCount = 0;
Set<String> payloadNames = new HashSet<>();
for (IndexFileMeta payload : payloads) {
GlobalIndexMeta meta = payload.globalIndexMeta();
- PrimaryKeyIndexSourceFile payloadSource =
- PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFile();
+ List<PrimaryKeyIndexSourceFile> payloadSources =
+ PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFiles();
if (!payloadNames.add(payload.fileName())
- || !sourceFile.equals(payloadSource)
+ || !sourceFiles.equals(payloadSources)
|| !indexType.equals(payload.indexType())
|| meta == null
|| meta.indexFieldId() != fieldId
|| meta.rowRangeStart() != 0
- || meta.rowRangeEnd() != sourceFile.rowCount() - 1) {
+ || meta.rowRangeEnd() != sourceRowCount - 1) {
+ return Optional.empty();
+ }
+ try {
+ payloadRowCount = Math.addExact(payloadRowCount, payload.rowCount());
+ } catch (ArithmeticException e) {
return Optional.empty();
}
- payloadRowCount += payload.rowCount();
}
- if (payloadRowCount != sourceFile.rowCount()) {
+ if (payloadRowCount != sourceRowCount) {
return Optional.empty();
}
- return Optional.of(new PkSortedIndexGroup(sourceFile, payloads));
+ return Optional.of(new PkSortedIndexGroup(sourceFiles, payloads));
}
public PrimaryKeyIndexSourceFile sourceFile() {
- return sourceFile;
+ if (sourceFiles.size() != 1) {
+ throw new IllegalStateException(
+ String.format(
+ "Expected exactly one source file, but found %s.", sourceFiles.size()));
+ }
+ return sourceFiles.get(0);
+ }
+
+ public List<PrimaryKeyIndexSourceFile> sourceFiles() {
+ return sourceFiles;
}
public List<IndexFileMeta> payloads() {
return payloads;
}
+
+ public String identity() {
+ List<String> names = new ArrayList<>();
+ for (IndexFileMeta payload : payloads) {
+ names.add(payload.fileName());
+ }
+ Collections.sort(names);
+ StringBuilder identity = new StringBuilder();
+ for (String name : names) {
+ identity.append(name.length()).append(':').append(name);
+ }
+ return identity.toString();
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java
index 1e002aa787..016b0d1320 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainer.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexLevels;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourcePolicy;
@@ -63,7 +64,7 @@ public class BucketedVectorIndexMaintainer {
private final String metric;
private final String algorithm;
private final PkVectorDataFileReader.Factory vectorReaderFactory;
- private final PkVectorAnnLevels annLevels;
+ private final PrimaryKeyIndexLevels<IndexFileMeta> annLevels;
private ExecutorService executor;
private final List<IndexFileMeta> annSegments;
private final Map<String, DataFileMeta> activeSourceFiles;
@@ -89,9 +90,12 @@ public class BucketedVectorIndexMaintainer {
this.vectorReaderFactory = vectorReaderFactory;
CoreOptions coreOptions = new CoreOptions(indexOptions);
this.annLevels =
- new PkVectorAnnLevels(
- coreOptions.primaryKeyVectorIndexCompactionLevelFanout(),
- coreOptions.primaryKeyVectorIndexCompactionStaleRatioThreshold());
+ new PrimaryKeyIndexLevels<>(
+ coreOptions.primaryKeyIndexCompactionLevelFanout(vectorField.name()),
+ coreOptions.primaryKeyIndexCompactionStaleRatioThreshold(
+ vectorField.name()),
+ IndexFileMeta::fileName,
+ segment -> sourceMeta(segment).sourceFiles());
this.executor = executor;
List<IndexFileMeta> definitionPayloads = new ArrayList<>();
@@ -163,14 +167,14 @@ public class BucketedVectorIndexMaintainer {
if (!uncovered.isEmpty()) {
startPendingBuild(uncovered, Collections.<IndexFileMeta>emptyList());
} else {
- Optional<PkVectorAnnLevels.Plan> plan =
+ Optional<PrimaryKeyIndexLevels.Plan<IndexFileMeta>> plan =
annLevels.pick(annSegments, activeSourceFiles);
if (plan.isPresent()) {
if (plan.get().sourceFiles().isEmpty()) {
- removeSegments(plan.get().inputSegments(), created, removed);
+ removeSegments(plan.get().inputUnits(), created, removed);
continue;
}
- startPendingBuild(plan.get().sourceFiles(), plan.get().inputSegments());
+ startPendingBuild(plan.get().sourceFiles(), plan.get().inputUnits());
}
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index f89f8ee071..f9dd882b70 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -896,16 +896,6 @@ public class SchemaValidation {
return;
}
- checkArgument(
- options.primaryKeyVectorIndexCompactionLevelFanout() > 1,
- "%s must be greater than 1.",
- CoreOptions.PK_VECTOR_INDEX_COMPACTION_LEVEL_FANOUT.key());
- double staleRatio = options.primaryKeyVectorIndexCompactionStaleRatioThreshold();
- checkArgument(
- staleRatio > 0 && staleRatio <= 1,
- "%s must be in (0, 1].",
- CoreOptions.PK_VECTOR_INDEX_COMPACTION_STALE_RATIO_THRESHOLD.key());
-
List<String> indexColumns = options.primaryKeyVectorIndexColumns();
checkArgument(
new HashSet<>(indexColumns).size() == indexColumns.size(),
@@ -981,6 +971,18 @@ public class SchemaValidation {
validateUniquePrimaryKeyIndexColumns(indexedColumns, vectorColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, btreeColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, bitmapColumns);
+ for (String column : indexedColumns) {
+ String fanoutKey = CoreOptions.primaryKeyIndexCompactionLevelFanoutKey(column);
+ checkArgument(
+ options.primaryKeyIndexCompactionLevelFanout(column) > 1,
+ "%s must be greater than 1.",
+ fanoutKey);
+ String staleRatioKey =
+ CoreOptions.primaryKeyIndexCompactionStaleRatioThresholdKey(column);
+ double staleRatio = options.primaryKeyIndexCompactionStaleRatioThreshold(column);
+ checkArgument(
+ staleRatio > 0 && staleRatio <= 1, "%s must be in (0, 1].", staleRatioKey);
+ }
}
private static void validateNoDuplicatePrimaryKeyIndexColumns(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
index 9e8c842fd2..bcc98207b0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
@@ -38,22 +38,34 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.IndexFilePathFactories;
import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.RoaringNavigableMap64;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -156,6 +168,8 @@ public final class PrimaryKeySortedIndexScan {
Pair<BinaryRow, Integer> bucket = bucketEntry.getKey();
List<IndexFileMeta> bucketPayloads =
payloadsByBucket.getOrDefault(bucket, Collections.emptyList());
+ Set<PrimaryKeyIndexSourceFile> activeSourceFiles =
+ new HashSet<>(bucketEntry.getValue());
Map<String, Map<Integer, PkSortedIndexGroup>> groupsBySource = new LinkedHashMap<>();
for (PrimaryKeyIndexDefinition definition : scalarDefinitions) {
List<IndexFileMeta> definitionPayloads = new ArrayList<>();
@@ -175,11 +189,15 @@ public final class PrimaryKeySortedIndexScan {
bucketEntry.getValue(),
definitionPayloads);
for (PkSortedIndexGroup group : state.groups()) {
- groupsBySource
- .computeIfAbsent(
- group.sourceFile().fileName(),
- ignored -> new LinkedHashMap<>())
- .put(definition.fieldId(), group);
+ for (PrimaryKeyIndexSourceFile sourceFile : group.sourceFiles()) {
+ if (!activeSourceFiles.contains(sourceFile)) {
+ continue;
+ }
+ groupsBySource
+ .computeIfAbsent(
+ sourceFile.fileName(), ignored -> new LinkedHashMap<>())
+ .put(definition.fieldId(), group);
+ }
}
} catch (RuntimeException e) {
rethrowIfInterrupted(e);
@@ -224,47 +242,498 @@ public final class PrimaryKeySortedIndexScan {
}
}
+ Map<PkSortedIndexGroup, SharedGlobalIndexReader> sharedReaders = new IdentityHashMap<>();
List<EvaluatedFile> files = new ArrayList<>();
- for (FilePlan file : plan.files()) {
- GlobalIndexEvaluator evaluator =
- new GlobalIndexEvaluator(
- rowType,
- fieldId -> {
- PrimaryKeyIndexDefinition definition =
- definitionsByField.get(fieldId);
- Optional<PkSortedIndexGroup> group = file.group(fieldId);
- if (definition == null || !group.isPresent()) {
- return Collections.emptyList();
- }
- return Collections.singletonList(
- readerFactory.create(
- file, definition, group.get().payloads()));
- });
- Optional<GlobalIndexResult> result;
- try {
- result = evaluator.evaluate(predicate);
- } catch (RuntimeException e) {
- rethrowIfInterrupted(e);
- LOG.warn(
- "Failed to evaluate primary-key sorted index for data file {}; "
- + "falling back to a raw scan for this file.",
- file.dataFile().fileName(),
- e);
- result = Optional.empty();
- } finally {
- evaluator.close();
+ try {
+ for (FilePlan file : plan.files()) {
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(
+ rowType,
+ fieldId -> {
+ PrimaryKeyIndexDefinition definition =
+ definitionsByField.get(fieldId);
+ Optional<PkSortedIndexGroup> group = file.group(fieldId);
+ if (definition == null || !group.isPresent()) {
+ return Collections.emptyList();
+ }
+ SharedGlobalIndexReader reader = sharedReaders.get(group.get());
+ if (reader == null) {
+ reader =
+ new SharedGlobalIndexReader(
+ group.get().sourceFiles(),
+ () ->
+ readerFactory.create(
+ file,
+ definition,
+ group.get().payloads()));
+ sharedReaders.put(group.get(), reader);
+ }
+ return Collections.singletonList(
+ fileLocalReader(file, group.get(), reader));
+ });
+ Optional<GlobalIndexResult> result;
+ try {
+ result = evaluator.evaluate(predicate);
+ } catch (RuntimeException e) {
+ rethrowIfInterrupted(e);
+ LOG.warn(
+ "Failed to evaluate primary-key sorted index for data file {}; "
+ + "falling back to a raw scan for this file.",
+ file.dataFile().fileName(),
+ e);
+ result = Optional.empty();
+ } finally {
+ evaluator.close();
+ }
+ files.add(new EvaluatedFile(file, result));
}
- files.add(new EvaluatedFile(file, result));
+ } finally {
+ IOUtils.closeAllQuietly(sharedReaders.values());
}
return new EvaluatedPlan(plan.snapshotId(), files);
}
+ private static GlobalIndexReader fileLocalReader(
+ FilePlan file, PkSortedIndexGroup group, SharedGlobalIndexReader reader) {
+ List<PrimaryKeyIndexSourceFile> sourceFiles = group.sourceFiles();
+ PrimaryKeyIndexSourceFile target =
+ new PrimaryKeyIndexSourceFile(
+ file.dataFile().fileName(), file.dataFile().rowCount());
+ int sourceIndex = -1;
+ for (int i = 0; i < sourceFiles.size(); i++) {
+ if (sourceFiles.get(i).equals(target)) {
+ sourceIndex = i;
+ break;
+ }
+ }
+ checkArgument(
+ sourceIndex >= 0,
+ "Data file %s is not covered by its sorted-index source group.",
+ file.dataFile().fileName());
+ return new FileLocalGlobalIndexReader(reader, sourceIndex);
+ }
+
private static void rethrowIfInterrupted(RuntimeException exception) {
if (Thread.currentThread().isInterrupted()) {
throw exception;
}
}
+ /** Shares one source-group reader and its group-global query results across source files. */
+ private static final class SharedGlobalIndexReader implements GlobalIndexReader {
+
+ private final Supplier<GlobalIndexReader> readerFactory;
+ private final Map<QueryKey, CompletableFuture<Optional<GlobalIndexResult>>> results;
+ private final Map<
+ CompletableFuture<Optional<GlobalIndexResult>>,
+ CompletableFuture<List<Optional<GlobalIndexResult>>>>
+ localizedResults;
+ private final long[] sourceOffsets;
+
+ private GlobalIndexReader reader;
+ private RuntimeException readerFailure;
+
+ private SharedGlobalIndexReader(
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
+ Supplier<GlobalIndexReader> readerFactory) {
+ this.readerFactory = readerFactory;
+ this.results = new ConcurrentHashMap<>();
+ this.localizedResults = new ConcurrentHashMap<>();
+ this.sourceOffsets = new long[sourceFiles.size() + 1];
+ for (int i = 0; i < sourceFiles.size(); i++) {
+ sourceOffsets[i + 1] =
+ Math.addExact(sourceOffsets[i], sourceFiles.get(i).rowCount());
+ }
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIsNotNull(FieldRef fieldRef) {
+ return query(
+ QueryKey.of(QueryOperation.IS_NOT_NULL, fieldRef),
+ () -> reader().visitIsNotNull(fieldRef));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIsNull(FieldRef fieldRef) {
+ return query(
+ QueryKey.of(QueryOperation.IS_NULL, fieldRef),
+ () -> reader().visitIsNull(fieldRef));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.STARTS_WITH, fieldRef, literal),
+ () -> reader().visitStartsWith(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.ENDS_WITH, fieldRef, literal),
+ () -> reader().visitEndsWith(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.CONTAINS, fieldRef, literal),
+ () -> reader().visitContains(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.LIKE, fieldRef, literal),
+ () -> reader().visitLike(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.LESS_THAN, fieldRef, literal),
+ () -> reader().visitLessThan(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.GREATER_OR_EQUAL, fieldRef, literal),
+ () -> reader().visitGreaterOrEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.NOT_EQUAL, fieldRef, literal),
+ () -> reader().visitNotEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.LESS_OR_EQUAL, fieldRef, literal),
+ () -> reader().visitLessOrEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.EQUAL, fieldRef, literal),
+ () -> reader().visitEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.GREATER_THAN, fieldRef, literal),
+ () -> reader().visitGreaterThan(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return query(
+ QueryKey.ofLiterals(QueryOperation.IN, fieldRef, literals),
+ () -> reader().visitIn(fieldRef, literals));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return query(
+ QueryKey.ofLiterals(QueryOperation.NOT_IN, fieldRef, literals),
+ () -> reader().visitNotIn(fieldRef, literals));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return query(
+ QueryKey.of(QueryOperation.BETWEEN, fieldRef, from, to),
+ () -> reader().visitBetween(fieldRef, from, to));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return query(
+ QueryKey.of(QueryOperation.NOT_BETWEEN, fieldRef, from, to),
+ () -> reader().visitNotBetween(fieldRef, from, to));
+ }
+
+ private CompletableFuture<Optional<GlobalIndexResult>> query(
+ QueryKey key,
+ Supplier<CompletableFuture<Optional<GlobalIndexResult>>> querySupplier) {
+ return results.computeIfAbsent(
+ key,
+ ignored -> {
+ try {
+ return checkNotNull(querySupplier.get());
+ } catch (RuntimeException e) {
+ CompletableFuture<Optional<GlobalIndexResult>> failed =
+ new CompletableFuture<>();
+ failed.completeExceptionally(e);
+ return failed;
+ }
+ });
+ }
+
+ private CompletableFuture<Optional<GlobalIndexResult>> localize(
+ CompletableFuture<Optional<GlobalIndexResult>> result, int sourceIndex) {
+ return localizedResults
+ .computeIfAbsent(result, future -> future.thenApply(this::partitionBySource))
+ .thenApply(partitions -> partitions.get(sourceIndex));
+ }
+
+ private List<Optional<GlobalIndexResult>> partitionBySource(
+ Optional<GlobalIndexResult> result) {
+ int sourceCount = sourceOffsets.length - 1;
+ if (!result.isPresent()) {
+ return Collections.nCopies(sourceCount, Optional.empty());
+ }
+ if (sourceCount == 1) {
+ return Collections.singletonList(result);
+ }
+
+ List<RoaringNavigableMap64> partitions = new ArrayList<>(sourceCount);
+ for (int i = 0; i < sourceCount; i++) {
+ partitions.add(new RoaringNavigableMap64());
+ }
+
+ long totalRowCount = sourceOffsets[sourceCount];
+ int sourceIndex = 0;
+ for (long position : result.get().results()) {
+ if (position < 0 || position >= totalRowCount) {
+ for (int i = 0; i < sourceCount; i++) {
+ partitions.get(i).add(sourceOffsets[i + 1] - sourceOffsets[i]);
+ }
+ continue;
+ }
+ while (position >= sourceOffsets[sourceIndex + 1]) {
+ sourceIndex++;
+ }
+ partitions.get(sourceIndex).add(position - sourceOffsets[sourceIndex]);
+ }
+
+ List<Optional<GlobalIndexResult>> localized = new ArrayList<>(sourceCount);
+ for (RoaringNavigableMap64 partition : partitions) {
+ localized.add(Optional.of(GlobalIndexResult.create(partition)));
+ }
+ return localized;
+ }
+
+ private synchronized GlobalIndexReader reader() {
+ if (reader != null) {
+ return reader;
+ }
+ if (readerFailure != null) {
+ throw readerFailure;
+ }
+ try {
+ reader = checkNotNull(readerFactory.get());
+ return reader;
+ } catch (RuntimeException e) {
+ readerFailure = e;
+ throw e;
+ }
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ if (reader != null) {
+ GlobalIndexReader readerToClose = reader;
+ reader = null;
+ readerToClose.close();
+ }
+ }
+ }
+
+ private enum QueryOperation {
+ IS_NOT_NULL,
+ IS_NULL,
+ STARTS_WITH,
+ ENDS_WITH,
+ CONTAINS,
+ LIKE,
+ LESS_THAN,
+ GREATER_OR_EQUAL,
+ NOT_EQUAL,
+ LESS_OR_EQUAL,
+ EQUAL,
+ GREATER_THAN,
+ IN,
+ NOT_IN,
+ BETWEEN,
+ NOT_BETWEEN
+ }
+
+ private static final class QueryKey {
+
+ private final QueryOperation operation;
+ private final FieldRef fieldRef;
+ private final List<Object> literals;
+
+ private QueryKey(QueryOperation operation, FieldRef fieldRef, List<Object> literals) {
+ this.operation = operation;
+ this.fieldRef = fieldRef;
+ this.literals = Collections.unmodifiableList(new ArrayList<>(literals));
+ }
+
+ private static QueryKey of(
+ QueryOperation operation, FieldRef fieldRef, Object... literals) {
+ return new QueryKey(operation, fieldRef, Arrays.asList(literals));
+ }
+
+ private static QueryKey ofLiterals(
+ QueryOperation operation, FieldRef fieldRef, List<Object> literals) {
+ return new QueryKey(operation, fieldRef, literals);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof QueryKey)) {
+ return false;
+ }
+ QueryKey queryKey = (QueryKey) o;
+ return operation == queryKey.operation
+ && Objects.equals(fieldRef, queryKey.fieldRef)
+ && Objects.equals(literals, queryKey.literals);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(operation, fieldRef, literals);
+ }
+ }
+
+ /** Restricts merged source-group ordinals to one source file's local row positions. */
+ private static final class FileLocalGlobalIndexReader implements GlobalIndexReader {
+
+ private final SharedGlobalIndexReader wrapped;
+ private final int sourceIndex;
+
+ private FileLocalGlobalIndexReader(SharedGlobalIndexReader wrapped, int sourceIndex) {
+ this.wrapped = wrapped;
+ this.sourceIndex = sourceIndex;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIsNotNull(FieldRef fieldRef) {
+ return localize(wrapped.visitIsNotNull(fieldRef));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIsNull(FieldRef fieldRef) {
+ return localize(wrapped.visitIsNull(fieldRef));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitStartsWith(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitEndsWith(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitContains(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitLike(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitLessThan(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitGreaterOrEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitNotEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitLessOrEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+ FieldRef fieldRef, Object literal) {
+ return localize(wrapped.visitGreaterThan(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return localize(wrapped.visitIn(fieldRef, literals));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return localize(wrapped.visitNotIn(fieldRef, literals));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return localize(wrapped.visitBetween(fieldRef, from, to));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return localize(wrapped.visitNotBetween(fieldRef, from, to));
+ }
+
+ private CompletableFuture<Optional<GlobalIndexResult>> localize(
+ CompletableFuture<Optional<GlobalIndexResult>> result) {
+ return wrapped.localize(result, sourceIndex);
+ }
+
+ @Override
+ public void close() {}
+ }
+
/** Immutable groups for all source files in one captured snapshot. */
public static final class Plan {
diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
index cb3e55d1a2..84aca44201 100644
--- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexBuilderTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.KeySerializer;
import org.apache.paimon.globalindex.ResultEntry;
@@ -52,6 +53,7 @@ import org.apache.paimon.utils.Pair;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.Closeable;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
@@ -101,6 +103,55 @@ public class SortedGlobalIndexBuilderTest extends TableTestBase {
assertThat(results.get(1)).extracting(ResultEntry::fileName).containsExactly("index-2");
}
+ @Test
+ public void testSingleColumnWriterClosesActiveWriter() throws Exception {
+ GlobalIndexSingleColumnWriter activeWriter =
+ mock(
+ GlobalIndexSingleColumnWriter.class,
+ org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
+ SortedSingleColumnIndexWriter rotatingWriter =
+ new SortedSingleColumnIndexWriter(2, () -> activeWriter);
+ rotatingWriter.write(10, 0);
+
+ assertThat(rotatingWriter).isInstanceOf(AutoCloseable.class);
+ ((AutoCloseable) rotatingWriter).close();
+
+ verify((Closeable) activeWriter).close();
+ }
+
+ @Test
+ public void testBuildForSinglePartitionClosesWriterAfterFailure() throws Exception {
+ createTableDefault();
+ GlobalIndexSingleColumnWriter activeWriter =
+ mock(
+ GlobalIndexSingleColumnWriter.class,
+ org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
+ org.mockito.Mockito.doThrow(new RuntimeException("write failed"))
+ .when(activeWriter)
+ .write(10, 0);
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(getTableDefault(), "btree") {
+ @Override
+ public GlobalIndexSingleColumnWriter createWriter() {
+ return activeWriter;
+ }
+ };
+ builder.withIndexField("f0");
+
+ org.assertj.core.api.Assertions.assertThatThrownBy(
+ () ->
+ builder.buildForSinglePartition(
+ new org.apache.paimon.utils.Range(0, 0),
+ null,
+ Collections.<InternalRow>singletonList(
+ GenericRow.of(10, 0L))
+ .iterator()))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("write failed");
+
+ verify((Closeable) activeWriter).close();
+ }
+
@Override
public Schema schemaDefault() {
Schema.Builder schemaBuilder = Schema.newBuilder();
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java
index c2acfda21b..b5d7e8d6ea 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java
@@ -50,6 +50,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
@@ -84,7 +85,7 @@ class BucketedPrimaryKeyIndexMaintainerTest {
}
@Test
- void testScalarFailureDoesNotSuppressOtherDefinitionsOrVector() throws Exception {
+ void testScalarFailureAbortsOtherDefinitionsAndVector() throws Exception {
DataFileMeta source = dataFile("data-1", 3);
IndexFileMeta vectorPayload =
new IndexFileMeta("vector", "vector", 1, 3, (GlobalIndexMeta) null, null);
@@ -117,18 +118,24 @@ class BucketedPrimaryKeyIndexMaintainerTest {
source,
dataFile -> {
buildOrder.add("bitmap");
- return Collections.singletonList(bitmapPayload);
+ return bitmapPayload;
});
BucketedPrimaryKeyIndexMaintainer maintainer =
BucketedPrimaryKeyIndexMaintainer.of(vector, Arrays.asList(bitmap, btree));
CompactIncrement compactIncrement = compactAfter(source);
- maintainer.prepareCommit(DataIncrement.emptyIncrement(), compactIncrement, true);
+ assertThatThrownBy(
+ () ->
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(), compactIncrement, true))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("expected BTree failure");
- assertThat(buildOrder).containsExactly("btree", "btree", "btree", "bitmap");
- assertThat(compactIncrement.newIndexFiles()).containsExactly(vectorPayload, bitmapPayload);
+ assertThat(buildOrder).containsExactly("btree", "btree", "btree");
+ assertThat(compactIncrement.newIndexFiles()).isEmpty();
assertThat(compactIncrement.deletedIndexFiles()).isEmpty();
assertThat(maintainer.buildNotCompleted()).isFalse();
+ verify(vectorCommit).abort(any());
}
@Test
@@ -149,8 +156,7 @@ class BucketedPrimaryKeyIndexMaintainerTest {
firstStarted.countDown();
releaseFirst.await();
activeBuilds.decrementAndGet();
- return Collections.singletonList(
- payload("btree", source, 3, 7, "btree"));
+ return payload("btree", source, 3, 7, "btree");
});
BucketedSortedIndexMaintainer second =
sortedMaintainer(
@@ -161,8 +167,7 @@ class BucketedPrimaryKeyIndexMaintainerTest {
enterBuild(activeBuilds, peakBuilds);
secondStarted.countDown();
activeBuilds.decrementAndGet();
- return Collections.singletonList(
- payload("bitmap", source, 3, 8, "bitmap"));
+ return payload("bitmap", source, 3, 8, "bitmap");
});
BucketedPrimaryKeyIndexMaintainer maintainer =
BucketedPrimaryKeyIndexMaintainer.ofSorted(Arrays.asList(second, first));
@@ -187,6 +192,41 @@ class BucketedPrimaryKeyIndexMaintainerTest {
assertThat(maintainer.buildNotCompleted()).isFalse();
}
+ @Test
+ void testNonBlockingCoordinatorStartsCoveredFanoutMaintenance() throws Exception {
+ DataFileMeta sourceA = dataFile("data-a", 3);
+ DataFileMeta sourceB = dataFile("data-b", 3);
+ IndexFileMeta payloadA = payload("index-a", sourceA, 3, 7, "btree");
+ IndexFileMeta payloadB = payload("index-b", sourceB, 3, 7, "btree");
+ IndexFileMeta merged = payload("index-ab", Arrays.asList(sourceA, sourceB), 6, 7, "btree");
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ BucketedSortedIndexMaintainer sorted =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
+ sourceFiles -> {
+ started.countDown();
+ release.await();
+ return merged;
+ },
+ 2,
+ 1.0,
+ Arrays.asList(sourceA, sourceB),
+ Arrays.asList(payloadA, payloadB),
+ buildExecutor);
+ BucketedPrimaryKeyIndexMaintainer maintainer =
+ BucketedPrimaryKeyIndexMaintainer.ofSorted(Collections.singletonList(sorted));
+
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(), CompactIncrement.emptyIncrement(), false);
+ boolean buildStarted = started.await(1, TimeUnit.SECONDS);
+ release.countDown();
+
+ assertThat(buildStarted).isTrue();
+ }
+
@Test
void testBlockingPrepareRollsBackAllDefinitionsAfterInterruption() throws Exception {
DataFileMeta source = dataFile("data-1", 3);
@@ -201,8 +241,7 @@ class BucketedPrimaryKeyIndexMaintainerTest {
source,
dataFile -> {
int attempt = firstBuilds.incrementAndGet();
- return Collections.singletonList(
- payload("btree-" + attempt, source, 3, 7, "btree"));
+ return payload("btree-" + attempt, source, 3, 7, "btree");
});
BucketedSortedIndexMaintainer second =
sortedMaintainer(
@@ -215,8 +254,7 @@ class BucketedPrimaryKeyIndexMaintainerTest {
secondStarted.countDown();
blockSecond.await();
}
- return Collections.singletonList(
- payload("bitmap-" + attempt, source, 3, 8, "bitmap"));
+ return payload("bitmap-" + attempt, source, 3, 8, "bitmap");
});
BucketedPrimaryKeyIndexMaintainer maintainer =
BucketedPrimaryKeyIndexMaintainer.ofSorted(Arrays.asList(second, first));
@@ -265,6 +303,8 @@ class BucketedPrimaryKeyIndexMaintainerTest {
indexType,
new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
buildFunction,
+ 5,
+ 0.2,
Collections.emptyList(),
Collections.emptyList(),
buildExecutor);
@@ -293,8 +333,27 @@ class BucketedPrimaryKeyIndexMaintainerTest {
long payloadRowCount,
int fieldId,
String indexType) {
- PrimaryKeyIndexSourceFile source =
- new PrimaryKeyIndexSourceFile(sourceFile.fileName(), sourceFile.rowCount());
+ return payload(
+ fileName,
+ Collections.singletonList(sourceFile),
+ payloadRowCount,
+ fieldId,
+ indexType);
+ }
+
+ private static IndexFileMeta payload(
+ String fileName,
+ List<DataFileMeta> sourceFiles,
+ long payloadRowCount,
+ int fieldId,
+ String indexType) {
+ List<PrimaryKeyIndexSourceFile> sources = new ArrayList<>();
+ long rowCount = 0;
+ for (DataFileMeta sourceFile : sourceFiles) {
+ sources.add(
+ new PrimaryKeyIndexSourceFile(sourceFile.fileName(), sourceFile.rowCount()));
+ rowCount = Math.addExact(rowCount, sourceFile.rowCount());
+ }
return new IndexFileMeta(
indexType,
fileName,
@@ -302,11 +361,11 @@ class BucketedPrimaryKeyIndexMaintainerTest {
payloadRowCount,
new GlobalIndexMeta(
0,
- source.rowCount() - 1,
+ rowCount - 1,
fieldId,
null,
new byte[] {1},
- new PrimaryKeyIndexSourceMeta(source).serialize()),
+ new PrimaryKeyIndexSourceMeta(sources).serialize()),
null);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
index b5d7875adc..ab8ca6d2bf 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
@@ -63,6 +63,41 @@ class PrimaryKeyIndexDefinitionsTest {
assertThat(definitions)
.extracting(PrimaryKeyIndexDefinition::indexType)
.containsExactly("btree", "bitmap", "ivf-pq");
+ assertThat(definitions)
+ .extracting(PrimaryKeyIndexDefinition::compactionLevelFanout)
+ .containsOnly(5);
+ assertThat(definitions)
+ .extracting(PrimaryKeyIndexDefinition::compactionStaleRatioThreshold)
+ .containsOnly(0.2);
+ }
+
+ @Test
+ void testResolvesFieldScopedCompactionOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "name");
+ options.put("fields.name.pk-index.compaction.level-fanout", "7");
+ options.put("fields.name.pk-index.compaction.stale-ratio-threshold", "0.4");
+
+ PrimaryKeyIndexDefinition definition =
+ PrimaryKeyIndexDefinitions.create(schema(options)).definitions().get(0);
+
+ assertThat(definition.compactionLevelFanout()).isEqualTo(7);
+ assertThat(definition.compactionStaleRatioThreshold()).isEqualTo(0.4);
+ }
+
+ @Test
+ void testLegacyVectorCompactionOptionsAreIgnored() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_VECTOR_INDEX_COLUMNS.key(), "embedding");
+ options.put("fields.embedding.pk-vector.index.type", "ivf-pq");
+ options.put("pk-vector.index.compaction.level-fanout", "9");
+ options.put("pk-vector.index.compaction.stale-ratio-threshold", "0.8");
+
+ PrimaryKeyIndexDefinition definition =
+ PrimaryKeyIndexDefinitions.create(schema(options)).definitions().get(0);
+
+ assertThat(definition.compactionLevelFanout()).isEqualTo(5);
+ assertThat(definition.compactionStaleRatioThreshold()).isEqualTo(0.2);
}
@Test
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexLevelsTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexLevelsTest.java
new file mode 100644
index 0000000000..989bb3857c
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexLevelsTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.index.pk;
+
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link PrimaryKeyIndexLevels}. */
+class PrimaryKeyIndexLevelsTest {
+
+ @Test
+ void testPicksSimilarLogicalUnitsAtFanout() {
+ PrimaryKeyIndexLevels<TestUnit> levels =
+ new PrimaryKeyIndexLevels<>(3, 0.2, TestUnit::id, TestUnit::sources);
+ DataFileMeta dataA = dataFile("data-a", 30);
+ DataFileMeta dataB = dataFile("data-b", 40);
+ DataFileMeta dataC = dataFile("data-c", 50);
+ TestUnit unitA = unit("unit-a", dataA);
+ TestUnit unitB = unit("unit-b", dataB);
+ TestUnit unitC = unit("unit-c", dataC);
+ Map<String, DataFileMeta> active = active(dataA, dataB, dataC);
+
+ PrimaryKeyIndexLevels.Plan<TestUnit> plan =
+ levels.pick(Arrays.asList(unitC, unitA, unitB), active).get();
+
+ assertThat(plan.inputUnits()).containsExactly(unitA, unitB, unitC);
+ assertThat(plan.sourceFiles()).containsExactly(dataA, dataB, dataC);
+ }
+
+ @Test
+ void testPicksUnitAtStaleRatioThreshold() {
+ PrimaryKeyIndexLevels<TestUnit> levels =
+ new PrimaryKeyIndexLevels<>(5, 0.4, TestUnit::id, TestUnit::sources);
+ DataFileMeta retired = dataFile("retired", 40);
+ DataFileMeta activeData = dataFile("active", 60);
+ TestUnit unit = unit("unit", retired, activeData);
+
+ PrimaryKeyIndexLevels.Plan<TestUnit> plan =
+ levels.pick(Collections.singletonList(unit), active(activeData)).get();
+
+ assertThat(plan.inputUnits()).containsExactly(unit);
+ assertThat(plan.sourceFiles()).containsExactly(activeData);
+ }
+
+ @Test
+ void testPicksUnitWithHighestStaleRatio() {
+ PrimaryKeyIndexLevels<TestUnit> levels =
+ new PrimaryKeyIndexLevels<>(5, 0.2, TestUnit::id, TestUnit::sources);
+ DataFileMeta activeA = dataFile("active-a", 50);
+ DataFileMeta activeB = dataFile("active-b", 20);
+ TestUnit halfStale = unit("unit-a", dataFile("retired-a", 50), activeA);
+ TestUnit mostlyStale = unit("unit-b", dataFile("retired-b", 80), activeB);
+
+ PrimaryKeyIndexLevels.Plan<TestUnit> plan =
+ levels.pick(Arrays.asList(halfStale, mostlyStale), active(activeA, activeB)).get();
+
+ assertThat(plan.inputUnits()).containsExactly(mostlyStale);
+ }
+
+ @Test
+ void testBreaksEqualStaleRatioByIdentity() {
+ PrimaryKeyIndexLevels<TestUnit> levels =
+ new PrimaryKeyIndexLevels<>(5, 0.2, TestUnit::id, TestUnit::sources);
+ TestUnit unitA = unit("unit-a", dataFile("retired-a", 10));
+ TestUnit unitB = unit("unit-b", dataFile("retired-b", 10));
+
+ PrimaryKeyIndexLevels.Plan<TestUnit> plan =
+ levels.pick(Arrays.asList(unitB, unitA), Collections.emptyMap()).get();
+
+ assertThat(plan.inputUnits()).containsExactly(unitA);
+ assertThat(plan.sourceFiles()).isEmpty();
+ }
+
+ @Test
+ void testSaturatesFanoutSizeComparison() {
+ PrimaryKeyIndexLevels<TestUnit> levels =
+ new PrimaryKeyIndexLevels<>(2, 1.0, TestUnit::id, TestUnit::sources);
+ DataFileMeta smaller = dataFile("data-a", Long.MAX_VALUE / 2 + 1);
+ DataFileMeta larger = dataFile("data-b", Long.MAX_VALUE - 1);
+ TestUnit unitA = unit("unit-a", smaller);
+ TestUnit unitB = unit("unit-b", larger);
+
+ PrimaryKeyIndexLevels.Plan<TestUnit> plan =
+ levels.pick(Arrays.asList(unitB, unitA), active(smaller, larger)).get();
+
+ assertThat(plan.inputUnits()).containsExactly(unitA, unitB);
+ }
+
+ private static TestUnit unit(String id, DataFileMeta... files) {
+ return new TestUnit(
+ id,
+ Arrays.stream(files)
+ .map(
+ file ->
+ new PrimaryKeyIndexSourceFile(
+ file.fileName(), file.rowCount()))
+ .collect(java.util.stream.Collectors.toList()));
+ }
+
+ private static Map<String, DataFileMeta> active(DataFileMeta... files) {
+ Map<String, DataFileMeta> active = new LinkedHashMap<>();
+ for (DataFileMeta file : files) {
+ active.put(file.fileName(), file);
+ }
+ return active;
+ }
+
+ private static DataFileMeta dataFile(String fileName, long rowCount) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 100,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static final class TestUnit {
+
+ private final String id;
+ private final List<PrimaryKeyIndexSourceFile> sources;
+
+ private TestUnit(String id, List<PrimaryKeyIndexSourceFile> sources) {
+ this.id = id;
+ this.sources = sources;
+ }
+
+ private String id() {
+ return id;
+ }
+
+ private List<PrimaryKeyIndexSourceFile> sources() {
+ return sources;
+ }
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java
index 835c686b36..3dcbb1bc19 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/BucketedSortedIndexMaintainerTest.java
@@ -48,6 +48,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests bucket-local BTree/Bitmap source maintenance. */
class BucketedSortedIndexMaintainerTest {
@@ -55,6 +56,11 @@ class BucketedSortedIndexMaintainerTest {
@TempDir java.nio.file.Path tempPath;
private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ @Test
+ void testHasSingleConstructor() {
+ assertThat(BucketedSortedIndexMaintainer.class.getDeclaredConstructors()).hasSize(1);
+ }
+
@AfterEach
void shutdownExecutor() {
executor.shutdownNow();
@@ -66,18 +72,19 @@ class BucketedSortedIndexMaintainerTest {
DataFileMeta newSource = dataFile("data-2", 3);
List<IndexFileMeta> oldPayloads =
Arrays.asList(payload("old-1", oldSource, 2), payload("old-2", oldSource, 1));
- List<IndexFileMeta> newPayloads =
- Arrays.asList(payload("new-1", newSource, 2), payload("new-2", newSource, 1));
+ IndexFileMeta newPayload = payload("new", newSource, 3);
PkSortedIndexFile indexFile = new PkSortedIndexFile(LocalFileIO.create(), pathFactory());
BucketedSortedIndexMaintainer maintainer =
new BucketedSortedIndexMaintainer(
7,
"btree",
indexFile,
- dataFile -> {
- assertThat(dataFile).isEqualTo(newSource);
- return newPayloads;
+ sourceFiles -> {
+ assertThat(sourceFiles).containsExactly(newSource);
+ return newPayload;
},
+ 5,
+ 0.2,
Collections.singletonList(oldSource),
oldPayloads,
executor);
@@ -92,8 +99,7 @@ class BucketedSortedIndexMaintainerTest {
true);
assertThat(commit.compactIncrement()).isPresent();
- assertThat(commit.compactIncrement().get().newIndexFiles())
- .containsExactlyElementsOf(newPayloads);
+ assertThat(commit.compactIncrement().get().newIndexFiles()).containsExactly(newPayload);
assertThat(commit.compactIncrement().get().deletedIndexFiles())
.containsExactlyElementsOf(oldPayloads);
assertThat(maintainer.state().coveredSourceFiles())
@@ -101,20 +107,203 @@ class BucketedSortedIndexMaintainerTest {
assertThat(maintainer.state().uncoveredSourceFiles()).isEmpty();
}
+ @Test
+ void testFanoutCompactionReplacesCompleteGroups() throws Exception {
+ DataFileMeta sourceA = dataFile("data-a", 3);
+ DataFileMeta sourceB = dataFile("data-b", 3);
+ IndexFileMeta payloadA = payload("index-a", sourceA, 3);
+ IndexFileMeta payloadB = payload("index-b", sourceB, 3);
+ IndexFileMeta merged = payload("index-ab", Arrays.asList(sourceA, sourceB), 6);
+ BucketedSortedIndexMaintainer maintainer =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
+ sources -> {
+ assertThat(sources).containsExactly(sourceA, sourceB);
+ return merged;
+ },
+ 2,
+ 1.0,
+ Arrays.asList(sourceA, sourceB),
+ Arrays.asList(payloadA, payloadB),
+ executor);
+
+ BucketedSortedIndexMaintainer.SortedIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(), CompactIncrement.emptyIncrement(), true);
+
+ assertThat(commit.appendIncrement()).isPresent();
+ assertThat(commit.appendIncrement().get().newIndexFiles()).containsExactly(merged);
+ assertThat(commit.appendIncrement().get().deletedIndexFiles())
+ .containsExactly(payloadA, payloadB);
+ assertThat(maintainer.state().groups()).hasSize(1);
+ assertThat(maintainer.state().groups().get(0).sourceFiles())
+ .containsExactly(
+ new PrimaryKeyIndexSourceFile("data-a", 3),
+ new PrimaryKeyIndexSourceFile("data-b", 3));
+ }
+
+ @Test
+ void testCoveredFanoutIsReportedAsPendingMaintenance() {
+ DataFileMeta sourceA = dataFile("data-a", 3);
+ DataFileMeta sourceB = dataFile("data-b", 3);
+ BucketedSortedIndexMaintainer maintainer =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
+ sources -> {
+ throw new AssertionError("Pending build must not start.");
+ },
+ 2,
+ 1.0,
+ Arrays.asList(sourceA, sourceB),
+ Arrays.asList(
+ payload("index-a", sourceA, 3), payload("index-b", sourceB, 3)),
+ executor);
+
+ assertThat(maintainer.hasPendingMaintenance()).isTrue();
+ }
+
+ @Test
+ void testPartiallyStaleGroupRebuildsOnlyActiveSources() throws Exception {
+ DataFileMeta staleSource = dataFile("data-a", 3);
+ DataFileMeta activeSource = dataFile("data-b", 7);
+ IndexFileMeta oldPayload =
+ payload("index-ab", Arrays.asList(staleSource, activeSource), 10);
+ IndexFileMeta rebuiltPayload = payload("index-b", activeSource, 7);
+ BucketedSortedIndexMaintainer maintainer =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
+ sourceFiles -> {
+ assertThat(sourceFiles).containsExactly(activeSource);
+ return rebuiltPayload;
+ },
+ 5,
+ 0.3,
+ Arrays.asList(staleSource, activeSource),
+ Collections.singletonList(oldPayload),
+ executor);
+
+ BucketedSortedIndexMaintainer.SortedIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.singletonList(staleSource),
+ Collections.emptyList(),
+ Collections.emptyList()),
+ true);
+
+ assertThat(commit.compactIncrement()).isPresent();
+ assertThat(commit.compactIncrement().get().newIndexFiles()).containsExactly(rebuiltPayload);
+ assertThat(commit.compactIncrement().get().deletedIndexFiles()).containsExactly(oldPayload);
+ assertThat(maintainer.state().groups()).hasSize(1);
+ assertThat(maintainer.state().groups().get(0).sourceFiles())
+ .containsExactly(new PrimaryKeyIndexSourceFile("data-b", 7));
+ }
+
+ @Test
+ void testAllStaleGroupIsDeletedWithoutBuild() throws Exception {
+ DataFileMeta source = dataFile("data-a", 3);
+ IndexFileMeta oldPayload = payload("index-a", source, 3);
+ AtomicInteger builds = new AtomicInteger();
+ BucketedSortedIndexMaintainer maintainer =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
+ sourceFiles -> {
+ builds.incrementAndGet();
+ return oldPayload;
+ },
+ 5,
+ 0.2,
+ Collections.singletonList(source),
+ Collections.singletonList(oldPayload),
+ executor);
+
+ BucketedSortedIndexMaintainer.SortedIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.singletonList(source),
+ Collections.emptyList(),
+ Collections.emptyList()),
+ true);
+
+ assertThat(builds).hasValue(0);
+ assertThat(commit.compactIncrement()).isPresent();
+ assertThat(commit.compactIncrement().get().newIndexFiles()).isEmpty();
+ assertThat(commit.compactIncrement().get().deletedIndexFiles()).containsExactly(oldPayload);
+ assertThat(maintainer.state().groups()).isEmpty();
+ }
+
+ @Test
+ void testBlockingFanoutDeletesIntermediateGeneratedGroups() throws Exception {
+ List<DataFileMeta> sources =
+ Arrays.asList(
+ dataFile("data-a", 3),
+ dataFile("data-b", 3),
+ dataFile("data-c", 3),
+ dataFile("data-d", 3));
+ AtomicInteger generation = new AtomicInteger();
+ List<IndexFileMeta> generated = new java.util.ArrayList<>();
+ TrackingPkSortedIndexFile indexFile =
+ new TrackingPkSortedIndexFile(LocalFileIO.create(), pathFactory());
+ BucketedSortedIndexMaintainer maintainer =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ indexFile,
+ sourceFiles -> {
+ IndexFileMeta payload =
+ payload(
+ "index-" + generation.incrementAndGet(),
+ sourceFiles,
+ sourceFiles.size() * 3L);
+ generated.add(payload);
+ return payload;
+ },
+ 2,
+ 1.0,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ executor);
+
+ BucketedSortedIndexMaintainer.SortedIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.emptyList(), sources, Collections.emptyList()),
+ true);
+
+ assertThat(generated).hasSize(7);
+ assertThat(commit.compactIncrement()).isPresent();
+ assertThat(commit.compactIncrement().get().newIndexFiles())
+ .containsExactly(generated.get(6));
+ assertThat(commit.compactIncrement().get().deletedIndexFiles()).isEmpty();
+ assertThat(indexFile.deleted())
+ .containsExactlyInAnyOrderElementsOf(generated.subList(0, 6));
+ }
+
@Test
void testRestoreDeletesInvalidPayloadsBeforePublishingReplacement() throws Exception {
DataFileMeta source = dataFile("data-1", 3);
List<IndexFileMeta> invalidPayloads =
Collections.singletonList(payload("invalid", source, 2));
- List<IndexFileMeta> replacementPayloads =
- Arrays.asList(payload("new-1", source, 2), payload("new-2", source, 1));
+ IndexFileMeta replacementPayload = payload("new", source, 3);
PkSortedIndexFile indexFile = new PkSortedIndexFile(LocalFileIO.create(), pathFactory());
BucketedSortedIndexMaintainer maintainer =
new BucketedSortedIndexMaintainer(
7,
"btree",
indexFile,
- dataFile -> replacementPayloads,
+ sourceFiles -> replacementPayload,
+ 5,
+ 0.2,
Collections.singletonList(source),
invalidPayloads,
executor);
@@ -127,18 +316,20 @@ class BucketedSortedIndexMaintainerTest {
assertThat(repair.appendIncrement().get().deletedIndexFiles())
.containsExactlyElementsOf(invalidPayloads);
assertThat(repair.appendIncrement().get().newIndexFiles())
- .containsExactlyElementsOf(replacementPayloads);
+ .containsExactly(replacementPayload);
BucketedSortedIndexMaintainer restored =
new BucketedSortedIndexMaintainer(
7,
"btree",
indexFile,
- dataFile -> {
+ sourceFiles -> {
throw new AssertionError("Covered source must not be rebuilt.");
},
+ 5,
+ 0.2,
Collections.singletonList(source),
- replacementPayloads,
+ Collections.singletonList(replacementPayload),
executor);
BucketedSortedIndexMaintainer.SortedIndexCommit stable =
restored.prepareCommit(
@@ -154,7 +345,7 @@ class BucketedSortedIndexMaintainerTest {
}
@Test
- void testBuildFailureDoesNotBlockSourceRemoval() throws Exception {
+ void testFinalBuildFailureThrowsAndRollsBackSourceTransition() {
DataFileMeta oldSource = dataFile("data-1", 3);
DataFileMeta newSource = dataFile("data-2", 3);
List<IndexFileMeta> oldPayloads =
@@ -165,50 +356,52 @@ class BucketedSortedIndexMaintainerTest {
7,
"btree",
new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
- dataFile -> {
+ sourceFiles -> {
attempts.incrementAndGet();
throw new IllegalStateException("expected build failure");
},
+ 5,
+ 0.2,
Collections.singletonList(oldSource),
oldPayloads,
executor);
- BucketedSortedIndexMaintainer.SortedIndexCommit commit =
- maintainer.prepareCommit(
- DataIncrement.emptyIncrement(),
- new CompactIncrement(
- Collections.singletonList(oldSource),
- Collections.singletonList(newSource),
- Collections.emptyList()),
- true);
+ assertThatThrownBy(
+ () ->
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.singletonList(oldSource),
+ Collections.singletonList(newSource),
+ Collections.emptyList()),
+ true))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("expected build failure");
assertThat(attempts).hasValue(3);
- assertThat(commit.compactIncrement()).isPresent();
- assertThat(commit.compactIncrement().get().newIndexFiles()).isEmpty();
- assertThat(commit.compactIncrement().get().deletedIndexFiles())
- .containsExactlyElementsOf(oldPayloads);
- assertThat(maintainer.state().coveredSourceFiles()).isEmpty();
- assertThat(maintainer.state().uncoveredSourceFiles())
- .containsExactly(new PrimaryKeyIndexSourceFile("data-2", 3));
+ assertThat(maintainer.state().coveredSourceFiles())
+ .containsExactly(new PrimaryKeyIndexSourceFile("data-1", 3));
+ assertThat(maintainer.state().uncoveredSourceFiles()).isEmpty();
}
@Test
void testTransientFailureRetriesAndPublishesWholeGroup() throws Exception {
DataFileMeta source = dataFile("data-1", 3);
- List<IndexFileMeta> payloads =
- Arrays.asList(payload("new-1", source, 2), payload("new-2", source, 1));
+ IndexFileMeta payload = payload("new", source, 3);
AtomicInteger attempts = new AtomicInteger();
BucketedSortedIndexMaintainer maintainer =
new BucketedSortedIndexMaintainer(
7,
"btree",
new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
- dataFile -> {
+ sourceFiles -> {
if (attempts.incrementAndGet() < 3) {
throw new IllegalStateException("expected transient failure");
}
- return payloads;
+ return payload;
},
+ 5,
+ 0.2,
Collections.emptyList(),
Collections.emptyList(),
executor);
@@ -219,8 +412,7 @@ class BucketedSortedIndexMaintainerTest {
assertThat(attempts).hasValue(3);
assertThat(commit.compactIncrement()).isPresent();
- assertThat(commit.compactIncrement().get().newIndexFiles())
- .containsExactlyElementsOf(payloads);
+ assertThat(commit.compactIncrement().get().newIndexFiles()).containsExactly(payload);
assertThat(maintainer.state().coveredSourceFiles())
.containsExactly(new PrimaryKeyIndexSourceFile("data-1", 3));
}
@@ -228,8 +420,7 @@ class BucketedSortedIndexMaintainerTest {
@Test
void testNonBlockingBuildPublishesOnLaterCommit() throws Exception {
DataFileMeta source = dataFile("data-1", 3);
- List<IndexFileMeta> payloads =
- Arrays.asList(payload("new-1", source, 2), payload("new-2", source, 1));
+ IndexFileMeta payload = payload("new", source, 3);
CountDownLatch started = new CountDownLatch(1);
CountDownLatch release = new CountDownLatch(1);
BucketedSortedIndexMaintainer maintainer =
@@ -237,11 +428,13 @@ class BucketedSortedIndexMaintainerTest {
7,
"btree",
new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
- dataFile -> {
+ sourceFiles -> {
started.countDown();
release.await();
- return payloads;
+ return payload;
},
+ 5,
+ 0.2,
Collections.emptyList(),
Collections.emptyList(),
executor);
@@ -260,8 +453,7 @@ class BucketedSortedIndexMaintainerTest {
DataIncrement.emptyIncrement(), CompactIncrement.emptyIncrement(), false);
assertThat(second.appendIncrement()).isPresent();
- assertThat(second.appendIncrement().get().newIndexFiles())
- .containsExactlyElementsOf(payloads);
+ assertThat(second.appendIncrement().get().newIndexFiles()).containsExactly(payload);
assertThat(maintainer.buildNotCompleted()).isFalse();
}
@@ -269,12 +461,8 @@ class BucketedSortedIndexMaintainerTest {
void testStaleCompletionIsDeletedBeforeReplacementPublishes() throws Exception {
DataFileMeta staleSource = dataFile("data-1", 3);
DataFileMeta activeSource = dataFile("data-2", 3);
- List<IndexFileMeta> stalePayloads =
- Arrays.asList(
- payload("stale-1", staleSource, 2), payload("stale-2", staleSource, 1));
- List<IndexFileMeta> activePayloads =
- Arrays.asList(
- payload("active-1", activeSource, 2), payload("active-2", activeSource, 1));
+ IndexFileMeta stalePayload = payload("stale", staleSource, 3);
+ IndexFileMeta activePayload = payload("active", activeSource, 3);
CountDownLatch staleStarted = new CountDownLatch(1);
CountDownLatch releaseStale = new CountDownLatch(1);
CountDownLatch staleCompleted = new CountDownLatch(1);
@@ -285,16 +473,19 @@ class BucketedSortedIndexMaintainerTest {
7,
"btree",
indexFile,
- dataFile -> {
- if (dataFile.fileName().equals(staleSource.fileName())) {
+ sourceFiles -> {
+ DataFileMeta sourceFile = sourceFiles.get(0);
+ if (sourceFile.fileName().equals(staleSource.fileName())) {
staleStarted.countDown();
releaseStale.await();
staleCompleted.countDown();
- return stalePayloads;
+ return stalePayload;
}
- assertThat(dataFile).isEqualTo(activeSource);
- return activePayloads;
+ assertThat(sourceFile).isEqualTo(activeSource);
+ return activePayload;
},
+ 5,
+ 0.2,
Collections.emptyList(),
Collections.emptyList(),
executor);
@@ -315,16 +506,67 @@ class BucketedSortedIndexMaintainerTest {
maintainer.prepareCommit(
DataIncrement.emptyIncrement(), CompactIncrement.emptyIncrement(), true);
- assertThat(indexFile.deleted()).containsExactlyElementsOf(stalePayloads);
+ assertThat(indexFile.deleted()).containsExactly(stalePayload);
assertThat(commit.appendIncrement()).isPresent();
- assertThat(commit.appendIncrement().get().newIndexFiles())
- .containsExactlyElementsOf(activePayloads);
+ assertThat(commit.appendIncrement().get().newIndexFiles()).containsExactly(activePayload);
assertThat(maintainer.state().coveredSourceFiles())
.containsExactly(new PrimaryKeyIndexSourceFile("data-2", 3));
}
@Test
- void testRejectedSubmissionLeavesSourceUncovered() throws Exception {
+ void testFanoutCompletionIsDiscardedWhenPlannedSourceRetires() throws Exception {
+ DataFileMeta sourceA = dataFile("data-a", 3);
+ DataFileMeta sourceB = dataFile("data-b", 3);
+ IndexFileMeta payloadA = payload("index-a", sourceA, 3);
+ IndexFileMeta payloadB = payload("index-b", sourceB, 3);
+ IndexFileMeta staleMerged = payload("index-ab", Arrays.asList(sourceA, sourceB), 6);
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ TrackingPkSortedIndexFile indexFile =
+ new TrackingPkSortedIndexFile(LocalFileIO.create(), pathFactory());
+ BucketedSortedIndexMaintainer maintainer =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ indexFile,
+ sourceFiles -> {
+ started.countDown();
+ release.await();
+ return staleMerged;
+ },
+ 2,
+ 1.0,
+ Arrays.asList(sourceA, sourceB),
+ Arrays.asList(payloadA, payloadB),
+ executor);
+
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(), CompactIncrement.emptyIncrement(), false);
+ assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.singletonList(sourceB),
+ Collections.emptyList(),
+ Collections.emptyList()),
+ false);
+ release.countDown();
+ executor.submit(() -> {}).get(5, TimeUnit.SECONDS);
+
+ BucketedSortedIndexMaintainer.SortedIndexCommit cleanup =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(), CompactIncrement.emptyIncrement(), true);
+
+ assertThat(indexFile.deleted()).contains(staleMerged);
+ assertThat(cleanup.appendIncrement()).isPresent();
+ assertThat(cleanup.appendIncrement().get().newIndexFiles()).isEmpty();
+ assertThat(cleanup.appendIncrement().get().deletedIndexFiles()).containsExactly(payloadB);
+ assertThat(maintainer.state().groups()).hasSize(1);
+ assertThat(maintainer.state().groups().get(0).payloads()).containsExactly(payloadA);
+ }
+
+ @Test
+ void testRejectedSubmissionThrowsAndRollsBackSourceTransition() {
DataFileMeta source = dataFile("data-1", 3);
ExecutorService rejectedExecutor = Executors.newSingleThreadExecutor();
rejectedExecutor.shutdownNow();
@@ -333,20 +575,53 @@ class BucketedSortedIndexMaintainerTest {
7,
"btree",
new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
- dataFile -> Collections.singletonList(payload("new", source, 3)),
+ sourceFiles -> payload("new", source, 3),
+ 5,
+ 0.2,
Collections.emptyList(),
Collections.emptyList(),
rejectedExecutor);
- BucketedSortedIndexMaintainer.SortedIndexCommit commit =
- maintainer.prepareCommit(
- DataIncrement.emptyIncrement(), compactAfter(source), false);
+ assertThatThrownBy(
+ () ->
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ compactAfter(source),
+ false))
+ .isInstanceOf(java.util.concurrent.RejectedExecutionException.class);
- assertThat(commit.appendIncrement()).isEmpty();
- assertThat(commit.compactIncrement()).isEmpty();
assertThat(maintainer.buildNotCompleted()).isFalse();
- assertThat(maintainer.state().uncoveredSourceFiles())
- .containsExactly(new PrimaryKeyIndexSourceFile("data-1", 3));
+ assertThat(maintainer.state().uncoveredSourceFiles()).isEmpty();
+ }
+
+ @Test
+ void testMalformedOutputThrowsAndRollsBackSourceTransition() {
+ DataFileMeta source = dataFile("data-1", 3);
+ IndexFileMeta malformed = payload("malformed", source, 2);
+ TrackingPkSortedIndexFile indexFile =
+ new TrackingPkSortedIndexFile(LocalFileIO.create(), pathFactory());
+ BucketedSortedIndexMaintainer maintainer =
+ new BucketedSortedIndexMaintainer(
+ 7,
+ "btree",
+ indexFile,
+ sourceFiles -> malformed,
+ 5,
+ 0.2,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ executor);
+
+ assertThatThrownBy(
+ () ->
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(), compactAfter(source), true))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("incomplete group");
+
+ assertThat(indexFile.deleted()).containsExactly(malformed);
+ assertThat(maintainer.state().groups()).isEmpty();
+ assertThat(maintainer.state().uncoveredSourceFiles()).isEmpty();
}
@Test
@@ -362,7 +637,7 @@ class BucketedSortedIndexMaintainerTest {
7,
"btree",
indexFile,
- dataFile -> {
+ sourceFiles -> {
started.countDown();
boolean released = false;
while (!released) {
@@ -373,8 +648,10 @@ class BucketedSortedIndexMaintainerTest {
// Simulate a builder completing despite cancellation.
}
}
- return Collections.singletonList(generated);
+ return generated;
},
+ 5,
+ 0.2,
Collections.emptyList(),
Collections.emptyList(),
executor);
@@ -402,7 +679,7 @@ class BucketedSortedIndexMaintainerTest {
7,
"btree",
new PkSortedIndexFile(LocalFileIO.create(), pathFactory()),
- dataFile -> {
+ sourceFiles -> {
started.countDown();
try {
new CountDownLatch(1).await();
@@ -412,6 +689,8 @@ class BucketedSortedIndexMaintainerTest {
throw e;
}
},
+ 5,
+ 0.2,
Collections.singletonList(oldSource),
oldPayloads,
executor);
@@ -455,8 +734,18 @@ class BucketedSortedIndexMaintainerTest {
private static IndexFileMeta payload(
String fileName, DataFileMeta sourceFile, long payloadRowCount) {
- PrimaryKeyIndexSourceFile source =
- new PrimaryKeyIndexSourceFile(sourceFile.fileName(), sourceFile.rowCount());
+ return payload(fileName, Collections.singletonList(sourceFile), payloadRowCount);
+ }
+
+ private static IndexFileMeta payload(
+ String fileName, List<DataFileMeta> sourceFiles, long payloadRowCount) {
+ List<PrimaryKeyIndexSourceFile> sources = new java.util.ArrayList<>();
+ long rowCount = 0;
+ for (DataFileMeta sourceFile : sourceFiles) {
+ sources.add(
+ new PrimaryKeyIndexSourceFile(sourceFile.fileName(), sourceFile.rowCount()));
+ rowCount = Math.addExact(rowCount, sourceFile.rowCount());
+ }
return new IndexFileMeta(
"btree",
fileName,
@@ -464,11 +753,11 @@ class BucketedSortedIndexMaintainerTest {
payloadRowCount,
new GlobalIndexMeta(
0,
- source.rowCount() - 1,
+ rowCount - 1,
7,
null,
new byte[] {1},
- new PrimaryKeyIndexSourceMeta(source).serialize()),
+ new PrimaryKeyIndexSourceMeta(sources).serialize()),
null);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexStateTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexStateTest.java
index 4235bf7456..758937b40e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexStateTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedBucketIndexStateTest.java
@@ -27,6 +27,7 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
@@ -54,6 +55,114 @@ class PkSortedBucketIndexStateTest {
assertThat(state.rejectedPayloads()).isEmpty();
}
+ @Test
+ void testMultiSourcePayloadsFormOneCoveredGroup() {
+ PrimaryKeyIndexSourceFile sourceA = new PrimaryKeyIndexSourceFile("data-a", 3);
+ PrimaryKeyIndexSourceFile sourceB = new PrimaryKeyIndexSourceFile("data-b", 7);
+ List<PrimaryKeyIndexSourceFile> sources = Arrays.asList(sourceA, sourceB);
+ PkSortedBucketIndexState state =
+ PkSortedBucketIndexState.fromActivePayloads(
+ 7,
+ "btree",
+ sources,
+ Arrays.asList(
+ payload("index-1", sources, "btree", 7, 0, 9, 4),
+ payload("index-2", sources, "btree", 7, 0, 9, 6)));
+
+ assertThat(state.groups()).hasSize(1);
+ assertThat(state.groups().get(0).sourceFiles()).containsExactly(sourceA, sourceB);
+ assertThat(state.coveredSourceFiles()).containsExactly(sourceA, sourceB);
+ assertThat(state.uncoveredSourceFiles()).isEmpty();
+ assertThat(state.rejectedPayloads()).isEmpty();
+ }
+
+ @Test
+ void testPartiallyStaleGroupRemainsAndCoversItsActiveSource() {
+ PrimaryKeyIndexSourceFile stale = new PrimaryKeyIndexSourceFile("data-a", 3);
+ PrimaryKeyIndexSourceFile active = new PrimaryKeyIndexSourceFile("data-b", 7);
+ List<PrimaryKeyIndexSourceFile> sources = Arrays.asList(stale, active);
+
+ PkSortedBucketIndexState state =
+ PkSortedBucketIndexState.fromActivePayloads(
+ 7,
+ "btree",
+ Collections.singletonList(active),
+ Collections.singletonList(
+ payload("index-1", sources, "btree", 7, 0, 9, 10)));
+
+ assertThat(state.groups()).hasSize(1);
+ assertThat(state.groups().get(0).sourceFiles()).containsExactly(stale, active);
+ assertThat(state.coveredSourceFiles()).containsExactly(active);
+ assertThat(state.uncoveredSourceFiles()).isEmpty();
+ assertThat(state.rejectedPayloads()).isEmpty();
+ }
+
+ @Test
+ void testOverlappingActiveSourceRejectsLaterGroup() {
+ PrimaryKeyIndexSourceFile sourceA = new PrimaryKeyIndexSourceFile("data-a", 3);
+ PrimaryKeyIndexSourceFile sourceB = new PrimaryKeyIndexSourceFile("data-b", 7);
+
+ IndexFileMeta first =
+ payload("index-ab", Arrays.asList(sourceA, sourceB), "btree", 7, 0, 9, 10);
+ IndexFileMeta overlapping = payload("index-b", sourceB, "btree", 7, 0, 6, 7);
+ PkSortedBucketIndexState state =
+ PkSortedBucketIndexState.fromActivePayloads(
+ 7,
+ "btree",
+ Arrays.asList(sourceA, sourceB),
+ Arrays.asList(first, overlapping));
+
+ assertThat(state.groups()).hasSize(1);
+ assertThat(state.groups().get(0).payloads()).containsExactly(first);
+ assertThat(state.coveredSourceFiles()).containsExactly(sourceA, sourceB);
+ assertThat(state.uncoveredSourceFiles()).isEmpty();
+ assertThat(state.rejectedPayloads()).containsExactly(overlapping);
+ }
+
+ @Test
+ void testDuplicateSourcesRejectWholeGroup() {
+ PrimaryKeyIndexSourceFile source = new PrimaryKeyIndexSourceFile("data-a", 3);
+ List<PrimaryKeyIndexSourceFile> duplicateSources = Arrays.asList(source, source);
+ IndexFileMeta duplicated = payload("index-a", duplicateSources, "btree", 7, 0, 5, 6);
+
+ PkSortedBucketIndexState state =
+ PkSortedBucketIndexState.fromActivePayloads(
+ 7,
+ "btree",
+ Collections.singletonList(source),
+ Collections.singletonList(duplicated));
+
+ assertThat(state.groups()).isEmpty();
+ assertThat(state.uncoveredSourceFiles()).containsExactly(source);
+ assertThat(state.rejectedPayloads()).containsExactly(duplicated);
+ }
+
+ @Test
+ void testSourceRowCountOverflowRejectsWholeGroup() {
+ PrimaryKeyIndexSourceFile huge = new PrimaryKeyIndexSourceFile("data-a", Long.MAX_VALUE);
+ PrimaryKeyIndexSourceFile extra = new PrimaryKeyIndexSourceFile("data-b", 1);
+ IndexFileMeta overflowing =
+ payload(
+ "index-overflow",
+ Arrays.asList(huge, extra),
+ "btree",
+ 7,
+ 0,
+ Long.MAX_VALUE,
+ 1);
+
+ PkSortedBucketIndexState state =
+ PkSortedBucketIndexState.fromActivePayloads(
+ 7,
+ "btree",
+ Arrays.asList(huge, extra),
+ Collections.singletonList(overflowing));
+
+ assertThat(state.groups()).isEmpty();
+ assertThat(state.uncoveredSourceFiles()).containsExactly(huge, extra);
+ assertThat(state.rejectedPayloads()).containsExactly(overflowing);
+ }
+
@Test
void testIncompletePayloadRowCountLeavesSourceUncovered() {
PrimaryKeyIndexSourceFile source = new PrimaryKeyIndexSourceFile("data-1", 10);
@@ -193,7 +302,25 @@ class PkSortedBucketIndexStateTest {
long rangeStart,
long rangeEnd,
long rowCount) {
- byte[] sourceMeta = new PrimaryKeyIndexSourceMeta(source).serialize();
+ return payload(
+ fileName,
+ Collections.singletonList(source),
+ indexType,
+ fieldId,
+ rangeStart,
+ rangeEnd,
+ rowCount);
+ }
+
+ private static IndexFileMeta payload(
+ String fileName,
+ java.util.List<PrimaryKeyIndexSourceFile> sources,
+ String indexType,
+ int fieldId,
+ long rangeStart,
+ long rangeEnd,
+ long rowCount) {
+ byte[] sourceMeta = new PrimaryKeyIndexSourceMeta(sources).serialize();
return new IndexFileMeta(
indexType,
fileName,
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
index 1c14998503..917f013da7 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
@@ -30,7 +30,7 @@ import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexer;
-import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
+import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
@@ -46,6 +46,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -64,6 +65,16 @@ class PkSortedIndexBuilderTest {
@TempDir java.nio.file.Path tempPath;
+ @Test
+ void testHasSingleBuildReturningIndexFileMeta() {
+ List<Method> buildMethods =
+ Arrays.stream(PkSortedIndexBuilder.class.getDeclaredMethods())
+ .filter(method -> method.getName().equals("build"))
+ .collect(Collectors.toList());
+ assertThat(buildMethods).hasSize(1);
+ assertThat(buildMethods.get(0).getReturnType()).isEqualTo(IndexFileMeta.class);
+ }
+
@Test
void testBuildsQueryableBTreeAndBitmapFromUnsortedPhysicalRows() throws Exception {
List<PkSortedDataFileReader.Entry> entries =
@@ -76,9 +87,9 @@ class PkSortedIndexBuilderTest {
IndexPathFactory pathFactory = pathFactory(directory);
Options options = options();
IOManager ioManager = IOManager.create(directory.resolve("spill").toString());
- List<IndexFileMeta> payloads;
+ IndexFileMeta payload;
try {
- payloads =
+ payload =
new PkSortedIndexBuilder(
ignored -> new ArrayReader(entries),
new PkSortedIndexFile(fileIO, pathFactory),
@@ -86,15 +97,65 @@ class PkSortedIndexBuilderTest {
indexType,
options,
ioManager)
- .build(source);
+ .build(Collections.singletonList(source));
} finally {
ioManager.close();
}
- assertThat(payloads).hasSize(2);
- assertQuery(fileIO, pathFactory, indexType, options, payloads, false, 20, 0L, 3L);
- assertQuery(fileIO, pathFactory, indexType, options, payloads, true, null, 1L);
+ assertQuery(fileIO, pathFactory, indexType, options, payload, false, 20, 0L, 3L);
+ assertQuery(fileIO, pathFactory, indexType, options, payload, true, null, 1L);
+ }
+ }
+
+ @Test
+ void testBuildsSeveralSourcesInDeterministicOrdinalOrder() throws Exception {
+ DataFileMeta sourceB = dataFile("data-b", 3);
+ DataFileMeta sourceA = dataFile("data-a", 2);
+ List<PrimaryKeyIndexSourceFile> capturedSources = new ArrayList<>();
+ List<PkSortedIndexFile.Entry> capturedEntries = new ArrayList<>();
+ PkSortedIndexFile capturingFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath)) {
+ @Override
+ public IndexFileMeta build(
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
+ DataField indexField,
+ String indexType,
+ Options indexOptions,
+ Iterator<Entry> sortedEntries) {
+ capturedSources.addAll(sourceFiles);
+ sortedEntries.forEachRemaining(capturedEntries::add);
+ return ignoredPayload();
+ }
+ };
+
+ IOManager ioManager = IOManager.create(tempPath.resolve("multi-spill").toString());
+ try {
+ new PkSortedIndexBuilder(
+ dataFile ->
+ new ArrayReader(
+ dataFile.fileName().equals("data-a")
+ ? Arrays.asList(entry(3, 0), entry(0, 1))
+ : Arrays.asList(
+ entry(1, 0), entry(4, 1), entry(2, 2))),
+ capturingFile,
+ field(),
+ "btree",
+ options(),
+ ioManager)
+ .build(Arrays.asList(sourceB, sourceA));
+ } finally {
+ ioManager.close();
}
+
+ assertThat(capturedSources)
+ .extracting(PrimaryKeyIndexSourceFile::fileName)
+ .containsExactly("data-a", "data-b");
+ assertThat(capturedEntries)
+ .extracting(PkSortedIndexFile.Entry::value)
+ .containsExactly(0, 1, 2, 3, 4);
+ assertThat(capturedEntries)
+ .extracting(PkSortedIndexFile.Entry::rowId)
+ .containsExactly(1L, 2L, 4L, 0L, 3L);
}
@Test
@@ -110,8 +171,8 @@ class PkSortedIndexBuilderTest {
PkSortedIndexFile capturingFile =
new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath)) {
@Override
- public List<IndexFileMeta> build(
- PrimaryKeyIndexSourceFile sourceFile,
+ public IndexFileMeta build(
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
DataField indexField,
String indexType,
Options indexOptions,
@@ -119,7 +180,7 @@ class PkSortedIndexBuilderTest {
while (sortedEntries.hasNext()) {
sortedValues.add((Integer) sortedEntries.next().value());
}
- return Collections.emptyList();
+ return ignoredPayload();
}
};
Options options = options();
@@ -139,7 +200,7 @@ class PkSortedIndexBuilderTest {
protected IOManager createTemporaryIOManager() {
return ioManager;
}
- }.build(dataFile("large-data-file", rowCount));
+ }.build(Collections.singletonList(dataFile("large-data-file", rowCount)));
assertThat(sortedValues)
.containsExactlyElementsOf(
@@ -155,19 +216,17 @@ class PkSortedIndexBuilderTest {
IndexPathFactory pathFactory,
String indexType,
Options options,
- List<IndexFileMeta> payloads,
+ IndexFileMeta payload,
boolean isNull,
Object literal,
Long... expected)
throws Exception {
- List<GlobalIndexIOMeta> ioMetas = new ArrayList<>();
- for (IndexFileMeta payload : payloads) {
- ioMetas.add(
- new GlobalIndexIOMeta(
- pathFactory.toPath(payload.fileName()),
- payload.fileSize(),
- payload.globalIndexMeta().indexMeta()));
- }
+ List<GlobalIndexIOMeta> ioMetas =
+ Collections.singletonList(
+ new GlobalIndexIOMeta(
+ pathFactory.toPath(payload.fileName()),
+ payload.fileSize(),
+ payload.globalIndexMeta().indexMeta()));
ExecutorService executor = newDirectExecutorService();
try (GlobalIndexReader reader =
GlobalIndexer.create(indexType, field(), options)
@@ -190,14 +249,16 @@ class PkSortedIndexBuilderTest {
return new PkSortedDataFileReader.Entry(value, position);
}
+ private static IndexFileMeta ignoredPayload() {
+ return new IndexFileMeta("test", "test", 0, 0, (GlobalIndexMeta) null, null);
+ }
+
private static DataField field() {
return new DataField(7, "indexed", DataTypes.INT());
}
private static Options options() {
- Options options = new Options();
- options.set(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE, 2L);
- return options;
+ return new Options();
}
private static DataFileMeta dataFile(String fileName, long rowCount) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
index adc8ba93d2..4720ade85b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexFileTest.java
@@ -24,7 +24,6 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
-import org.apache.paimon.globalindex.sorted.SortedIndexOptions;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
@@ -38,12 +37,15 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.lang.reflect.Method;
import java.nio.file.Files;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
@@ -55,16 +57,51 @@ class PkSortedIndexFileTest {
@TempDir java.nio.file.Path tempPath;
@Test
- void testBuildsRotatedBTreeAndBitmapPayloadGroups() throws Exception {
+ void testHasSingleBuildEntryPoint() {
+ List<Method> buildMethods =
+ Arrays.stream(PkSortedIndexFile.class.getDeclaredMethods())
+ .filter(method -> method.getName().equals("build"))
+ .collect(Collectors.toList());
+ assertThat(buildMethods).hasSize(1);
+ assertThat(buildMethods.get(0).getReturnType()).isEqualTo(IndexFileMeta.class);
+ }
+
+ @Test
+ void testUsesCloseableStreamingBitmapWriter() throws Exception {
+ PkSortedIndexFile indexFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath));
+ GlobalIndexSingleColumnWriter writer =
+ indexFile.createWriter(
+ "bitmap",
+ field(),
+ options(),
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + ".index";
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) {
+ throw new AssertionError("Writer must open output lazily.");
+ }
+ });
+
+ assertThat(writer).isInstanceOf(AutoCloseable.class);
+ ((AutoCloseable) writer).close();
+ }
+
+ @Test
+ void testBuildsSingleBTreeAndBitmapPayload() throws Exception {
for (String indexType : Arrays.asList("btree", "bitmap")) {
java.nio.file.Path indexDirectory = Files.createDirectory(tempPath.resolve(indexType));
PkSortedIndexFile indexFile =
new PkSortedIndexFile(LocalFileIO.create(), pathFactory(indexDirectory));
PrimaryKeyIndexSourceFile source = new PrimaryKeyIndexSourceFile("data-file", 3);
- List<IndexFileMeta> payloads =
+ IndexFileMeta payload =
indexFile.build(
- source,
+ Collections.singletonList(source),
field(),
indexType,
options(),
@@ -74,30 +111,52 @@ class PkSortedIndexFileTest {
new PkSortedIndexFile.Entry(20, 0))
.iterator());
- assertThat(payloads).hasSize(2);
- assertThat(payloads).extracting(IndexFileMeta::indexType).containsOnly(indexType);
- assertThat(payloads).extracting(IndexFileMeta::rowCount).containsExactly(2L, 1L);
- assertThat(payloads)
- .allSatisfy(
- payload -> {
- GlobalIndexMeta meta = payload.globalIndexMeta();
- assertThat(meta.rowRangeStart()).isZero();
- assertThat(meta.rowRangeEnd()).isEqualTo(2);
- assertThat(meta.indexFieldId()).isEqualTo(7);
- assertThat(meta.indexMeta()).isNotEmpty();
- assertThat(
- PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
- .sourceFile())
- .isEqualTo(source);
- assertThat(indexFile.exists(payload)).isTrue();
- });
+ assertThat(payload.indexType()).isEqualTo(indexType);
+ assertThat(payload.rowCount()).isEqualTo(3L);
+ GlobalIndexMeta meta = payload.globalIndexMeta();
+ assertThat(meta.rowRangeStart()).isZero();
+ assertThat(meta.rowRangeEnd()).isEqualTo(2);
+ assertThat(meta.indexFieldId()).isEqualTo(7);
+ assertThat(meta.indexMeta()).isNotEmpty();
+ assertThat(PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFile())
+ .isEqualTo(source);
+ assertThat(indexFile.exists(payload)).isTrue();
}
}
@Test
- void testFailedSecondPayloadDeletesWholeGroup() throws Exception {
+ void testBuildsMultiSourcePayloadsInOneOrdinalDomain() throws Exception {
+ PkSortedIndexFile indexFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath));
+ List<PrimaryKeyIndexSourceFile> sources =
+ Arrays.asList(
+ new PrimaryKeyIndexSourceFile("data-a", 2),
+ new PrimaryKeyIndexSourceFile("data-b", 3));
+
+ IndexFileMeta payload =
+ indexFile.build(
+ sources,
+ field(),
+ "btree",
+ options(),
+ Arrays.asList(
+ new PkSortedIndexFile.Entry(null, 3),
+ new PkSortedIndexFile.Entry(10, 1),
+ new PkSortedIndexFile.Entry(20, 4),
+ new PkSortedIndexFile.Entry(30, 0),
+ new PkSortedIndexFile.Entry(40, 2))
+ .iterator());
+
+ assertThat(payload.rowCount()).isEqualTo(5L);
+ assertThat(payload.globalIndexMeta().rowRangeStart()).isZero();
+ assertThat(payload.globalIndexMeta().rowRangeEnd()).isEqualTo(4);
+ assertThat(PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFiles())
+ .containsExactlyElementsOf(sources);
+ }
+
+ @Test
+ void testRejectsMultiplePayloadsAndDeletesWholeGroup() throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
- AtomicInteger writerNumber = new AtomicInteger();
PkSortedIndexFile indexFile =
new PkSortedIndexFile(fileIO, pathFactory(tempPath)) {
@Override
@@ -106,62 +165,105 @@ class PkSortedIndexFileTest {
DataField indexField,
Options indexOptions,
GlobalIndexFileWriter fileWriter) {
- int currentWriter = writerNumber.incrementAndGet();
return new GlobalIndexSingleColumnWriter() {
private long rowCount;
@Override
public void write(Object key, long relativeRowId) {
- if (currentWriter == 2) {
- throw new RuntimeException("injected second payload failure");
- }
rowCount++;
}
@Override
public List<ResultEntry> finish() {
- String fileName = fileWriter.newFileName("test");
- try (PositionOutputStream output =
- fileWriter.newOutputStream(fileName)) {
- output.write(new byte[] {1});
- } catch (IOException e) {
- throw new RuntimeException(e);
+ List<ResultEntry> results = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ String fileName = fileWriter.newFileName("test");
+ try (PositionOutputStream output =
+ fileWriter.newOutputStream(fileName)) {
+ output.write(new byte[] {1});
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ results.add(
+ new ResultEntry(fileName, rowCount, new byte[] {2}));
}
- return Collections.singletonList(
- new ResultEntry(fileName, rowCount, new byte[] {2}));
+ return results;
}
};
}
};
- Options options = options();
- options.set(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE, 1L);
assertThatThrownBy(
() ->
indexFile.build(
- new PrimaryKeyIndexSourceFile("data-file", 2),
+ Collections.singletonList(
+ new PrimaryKeyIndexSourceFile("data-file", 2)),
field(),
"btree",
- options,
+ options(),
Arrays.asList(
new PkSortedIndexFile.Entry(10, 0),
new PkSortedIndexFile.Entry(20, 1))
.iterator()))
- .hasMessageContaining("injected second payload failure");
+ .hasMessageContaining("must produce exactly one payload file");
try (Stream<java.nio.file.Path> files = Files.list(tempPath)) {
assertThat(files).isEmpty();
}
}
+ @Test
+ void testClosesWriterAfterFailedBuild() throws Exception {
+ AtomicBoolean closed = new AtomicBoolean();
+ class CloseableWriter implements GlobalIndexSingleColumnWriter, AutoCloseable {
+
+ @Override
+ public void write(Object key, long relativeRowId) {}
+
+ @Override
+ public List<ResultEntry> finish() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ }
+
+ PkSortedIndexFile indexFile =
+ new PkSortedIndexFile(LocalFileIO.create(), pathFactory(tempPath)) {
+ @Override
+ protected GlobalIndexSingleColumnWriter createWriter(
+ String indexType,
+ DataField indexField,
+ Options indexOptions,
+ GlobalIndexFileWriter fileWriter) {
+ return new CloseableWriter();
+ }
+ };
+
+ assertThatThrownBy(
+ () ->
+ indexFile.build(
+ Collections.singletonList(
+ new PrimaryKeyIndexSourceFile("data-file", 1)),
+ field(),
+ "bitmap",
+ options(),
+ Collections.singletonList(
+ new PkSortedIndexFile.Entry(10, 1))
+ .iterator()))
+ .hasMessageContaining("outside sorted index group row range");
+ assertThat(closed).isTrue();
+ }
+
private static DataField field() {
return new DataField(7, "indexed", DataTypes.INT());
}
private static Options options() {
- Options options = new Options();
- options.set(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE, 2L);
- return options;
+ return new Options();
}
private static IndexPathFactory pathFactory(java.nio.file.Path directory) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java
index 8ed3b4234f..473ffe7963 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexMaintenanceTest.java
@@ -260,17 +260,20 @@ class PrimaryKeySortedIndexMaintenanceTest {
assertThat(typePayloads.stream().mapToLong(IndexFileMeta::rowCount).sum())
.as("total row count of %s source groups", typePayloads.get(0).indexType())
.isEqualTo(rowCount);
- Map<PrimaryKeyIndexSourceFile, List<IndexFileMeta>> bySource =
+ Map<List<PrimaryKeyIndexSourceFile>, List<IndexFileMeta>> bySources =
typePayloads.stream()
.collect(
Collectors.groupingBy(
payload ->
PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
- .sourceFile()));
- for (Map.Entry<PrimaryKeyIndexSourceFile, List<IndexFileMeta>> sourceGroup :
- bySource.entrySet()) {
+ .sourceFiles()));
+ for (Map.Entry<List<PrimaryKeyIndexSourceFile>, List<IndexFileMeta>> sourceGroup :
+ bySources.entrySet()) {
assertThat(sourceGroup.getValue().stream().mapToLong(IndexFileMeta::rowCount).sum())
- .isEqualTo(sourceGroup.getKey().rowCount());
+ .isEqualTo(
+ sourceGroup.getKey().stream()
+ .mapToLong(PrimaryKeyIndexSourceFile::rowCount)
+ .sum());
}
}
}
@@ -278,7 +281,10 @@ class PrimaryKeySortedIndexMaintenanceTest {
private static Set<String> sourceNames(List<IndexFileMeta> payloads) {
Set<String> result = new HashSet<>();
for (IndexFileMeta payload : payloads) {
- result.add(PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFile().fileName());
+ for (PrimaryKeyIndexSourceFile source :
+ PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFiles()) {
+ result.add(source.fileName());
+ }
}
return result;
}
@@ -317,12 +323,6 @@ class PrimaryKeySortedIndexMaintenanceTest {
options.put(CoreOptions.COMPACTION_FORCE_REWRITE_ALL_FILES.key(), "true");
options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "itemId");
options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "comment");
- options.put(
- "fields.itemId.pk-btree.index.options",
- "{\"sorted-index.records-per-range\":\"2\"}");
- options.put(
- "fields.comment.pk-bitmap.index.options",
- "{\"sorted-index.records-per-range\":\"2\"}");
SchemaManager schemaManager =
new SchemaManager(LocalFileIO.create(), new Path(tempDir.toUri()));
TableSchema schema =
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java
index 1822598773..7bb2018e0e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PrimaryKeySortedIndexOptionsTest.java
@@ -50,29 +50,34 @@ class PrimaryKeySortedIndexOptionsTest {
}
@Test
- void testResolvesBTreeIndexOptions() {
+ void testResolvesBTreeIndexAndSortOptions() {
Map<String, String> values = new HashMap<>();
- values.put(
- "fields.name.pk-btree.index.options",
- "{\"block-size\":\"4 kb\",\"sorted-index.records-per-range\":\"10\"}");
+ values.put("sorted-index.records-per-range", "10");
+ values.put("write-buffer-size", "8 mb");
+ values.put("page-size", "32 kb");
+ values.put("local-sort.max-num-file-handles", "16");
+ values.put("spill-compression", "lz4");
+ values.put("write-buffer-spill.max-disk-size", "1 gb");
+ values.put("fields.name.pk-btree.index.options", "{\"block-size\":\"4 kb\"}");
Options options = new CoreOptions(values).primaryKeyBTreeIndexOptions("name");
assertThat(options.get("btree-index.block-size")).isEqualTo("4 kb");
- assertThat(options.get("sorted-index.records-per-range")).isEqualTo("10");
+ assertThat(options.get("write-buffer-size")).isEqualTo("8 mb");
+ assertThat(options.get("page-size")).isEqualTo("32 kb");
+ assertThat(options.get("local-sort.max-num-file-handles")).isEqualTo("16");
+ assertThat(options.get("spill-compression")).isEqualTo("lz4");
+ assertThat(options.get("write-buffer-spill.max-disk-size")).isEqualTo("1 gb");
+ assertThat(options.get("sorted-index.records-per-range")).isNull();
}
@Test
void testResolvesBitmapIndexOptions() {
Map<String, String> values = new HashMap<>();
- values.put(
- "fields.status.pk-bitmap.index.options",
- "{\"dictionary-block-size\":\"8 kb\","
- + "\"sorted-index.records-per-range\":\"20\"}");
+ values.put("fields.status.pk-bitmap.index.options", "{\"dictionary-block-size\":\"8 kb\"}");
Options options = new CoreOptions(values).primaryKeyBitmapIndexOptions("status");
assertThat(options.get("bitmap-index.dictionary-block-size")).isEqualTo("8 kb");
- assertThat(options.get("sorted-index.records-per-range")).isEqualTo("20");
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
index 8cd89bb59c..35196debc0 100644
--- a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/BucketedVectorIndexMaintainerTest.java
@@ -274,7 +274,7 @@ class BucketedVectorIndexMaintainerTest {
DataField vectorField =
new DataField(7, "embedding", DataTypes.VECTOR(2, DataTypes.FLOAT()));
Options options = indexOptions();
- options.setString("pk-vector.index.compaction.stale-ratio-threshold", "1.0");
+ options.setString("fields.embedding.pk-index.compaction.stale-ratio-threshold", "1.0");
DataFileMeta data1 = dataFile("data-1");
DataFileMeta data2 = dataFile("data-2");
DataFileMeta data3 = dataFile("data-3");
@@ -371,7 +371,7 @@ class BucketedVectorIndexMaintainerTest {
DataField vectorField =
new DataField(7, "embedding", DataTypes.VECTOR(2, DataTypes.FLOAT()));
Options options = indexOptions();
- options.setString("pk-vector.index.compaction.level-fanout", "3");
+ options.setString("fields.embedding.pk-index.compaction.level-fanout", "3");
DataFileMeta data1 = dataFile("data-1");
DataFileMeta data2 = dataFile("data-2");
DataFileMeta data3 = dataFile("data-3");
@@ -444,7 +444,7 @@ class BucketedVectorIndexMaintainerTest {
DataField vectorField =
new DataField(7, "embedding", DataTypes.VECTOR(2, DataTypes.FLOAT()));
Options options = indexOptions();
- options.setString("pk-vector.index.compaction.level-fanout", "3");
+ options.setString("fields.embedding.pk-index.compaction.level-fanout", "3");
List<DataFileMeta> dataFiles = new ArrayList<>();
List<IndexFileMeta> initialSegments = new ArrayList<>();
for (int i = 1; i <= 6; i++) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnLevelsTest.java b/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnLevelsTest.java
deleted file mode 100644
index 14c8648fea..0000000000
--- a/paimon-core/src/test/java/org/apache/paimon/index/pkvector/PkVectorAnnLevelsTest.java
+++ /dev/null
@@ -1,141 +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.index.pkvector;
-
-import org.apache.paimon.index.GlobalIndexMeta;
-import org.apache.paimon.index.IndexFileMeta;
-import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
-import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
-import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.manifest.FileSource;
-import org.apache.paimon.stats.SimpleStats;
-
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.Map;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for derived primary-key vector ANN levels. */
-class PkVectorAnnLevelsTest {
-
- @Test
- void testPicksSimilarSegmentsAcrossAbsoluteBoundary() {
- PkVectorAnnLevels levels = new PkVectorAnnLevels(3, 0.2);
- DataFileMeta dataA = dataFile("data-a", 99);
- DataFileMeta dataB = dataFile("data-b", 100);
- DataFileMeta dataC = dataFile("data-c", 101);
- IndexFileMeta annA = segment("ann-a", dataA);
- IndexFileMeta annB = segment("ann-b", dataB);
- IndexFileMeta annC = segment("ann-c", dataC);
- Map<String, DataFileMeta> active = new LinkedHashMap<>();
- active.put(dataA.fileName(), dataA);
- active.put(dataB.fileName(), dataB);
- active.put(dataC.fileName(), dataC);
-
- PkVectorAnnLevels.Plan plan = levels.pick(Arrays.asList(annC, annA, annB), active).get();
-
- assertThat(plan.inputSegments()).containsExactly(annA, annB, annC);
- }
-
- @Test
- void testPicksLevelZeroSegmentsAtFanout() {
- PkVectorAnnLevels levels = new PkVectorAnnLevels(3, 0.2);
- DataFileMeta dataA = dataFile("data-a", 30);
- DataFileMeta dataB = dataFile("data-b", 40);
- DataFileMeta dataC = dataFile("data-c", 50);
- IndexFileMeta annA = segment("ann-a", dataA);
- IndexFileMeta annB = segment("ann-b", dataB);
- IndexFileMeta annC = segment("ann-c", dataC);
- Map<String, DataFileMeta> active = new LinkedHashMap<>();
- active.put(dataA.fileName(), dataA);
- active.put(dataB.fileName(), dataB);
- active.put(dataC.fileName(), dataC);
-
- PkVectorAnnLevels.Plan plan = levels.pick(Arrays.asList(annC, annA, annB), active).get();
-
- assertThat(plan.inputSegments()).containsExactly(annA, annB, annC);
- assertThat(plan.sourceFiles()).containsExactly(dataA, dataB, dataC);
- }
-
- @Test
- void testPicksSegmentAboveStaleRatio() {
- PkVectorAnnLevels levels = new PkVectorAnnLevels(5, 0.2);
- DataFileMeta retired = dataFile("retired", 60);
- DataFileMeta activeData = dataFile("active", 60);
- IndexFileMeta ann = segment("ann", retired, activeData);
- Map<String, DataFileMeta> active =
- Collections.singletonMap(activeData.fileName(), activeData);
-
- PkVectorAnnLevels.Plan plan = levels.pick(Collections.singletonList(ann), active).get();
-
- assertThat(plan.inputSegments()).containsExactly(ann);
- assertThat(plan.sourceFiles()).containsExactly(activeData);
- }
-
- private static IndexFileMeta segment(String fileName, long rowCount) {
- return segment(fileName, dataFile(fileName + "-data", rowCount));
- }
-
- private static IndexFileMeta segment(String fileName, DataFileMeta source) {
- return segment(fileName, new DataFileMeta[] {source});
- }
-
- private static IndexFileMeta segment(String fileName, DataFileMeta... sources) {
- long rowCount = Arrays.stream(sources).mapToLong(DataFileMeta::rowCount).sum();
- byte[] sourceMeta =
- new PrimaryKeyIndexSourceMeta(
- Arrays.stream(sources)
- .map(
- source ->
- new PrimaryKeyIndexSourceFile(
- source.fileName(),
- source.rowCount()))
- .collect(java.util.stream.Collectors.toList()))
- .serialize();
- return new IndexFileMeta(
- "test-vector-ann",
- fileName,
- 100,
- rowCount,
- new GlobalIndexMeta(0, rowCount - 1, 7, null, new byte[] {1}, sourceMeta),
- null);
- }
-
- private static DataFileMeta dataFile(String fileName, long rowCount) {
- return DataFileMeta.forAppend(
- fileName,
- 100,
- rowCount,
- SimpleStats.EMPTY_STATS,
- 0,
- 0,
- 1,
- Collections.emptyList(),
- null,
- FileSource.COMPACT,
- null,
- null,
- null,
- null);
- }
-}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
index f864f61526..afd56c6207 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
@@ -73,6 +73,10 @@ class PrimaryKeyIndexWriteTest {
options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "itemId");
options.put(CoreOptions.PK_BITMAP_INDEX_COLUMNS.key(), "comment");
+ options.put("fields.itemId.pk-index.compaction.level-fanout", "7");
+ options.put("fields.itemId.pk-index.compaction.stale-ratio-threshold", "0.4");
+ options.put("fields.comment.pk-index.compaction.level-fanout", "7");
+ options.put("fields.comment.pk-index.compaction.stale-ratio-threshold", "0.4");
TestFileStore store = createStore(options);
KeyValueFileStoreWrite write = (KeyValueFileStoreWrite) store.newWrite();
write.withIOManager(ioManager);
@@ -84,6 +88,13 @@ class PrimaryKeyIndexWriteTest {
assertThat(container.primaryKeyIndexMaintainer).isNotNull();
assertThat(container.primaryKeyIndexMaintainer.buildNotCompleted()).isFalse();
+ List<?> sortedMaintainers =
+ (List<?>) readField(container.primaryKeyIndexMaintainer, "sortedMaintainers");
+ for (Object sortedMaintainer : sortedMaintainers) {
+ Object levels = readField(sortedMaintainer, "levels");
+ assertThat(readField(levels, "fanout")).isEqualTo(7);
+ assertThat(readField(levels, "staleRatioThreshold")).isEqualTo(0.4);
+ }
write.close();
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java
index d55dd836ef..6da0672ada 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeySortedIndexValidationTest.java
@@ -164,6 +164,17 @@ class PrimaryKeySortedIndexValidationTest {
.hasMessageContaining("not supported by global index");
}
+ @Test
+ void testRejectsInvalidCompactionLevelFanout() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "payload");
+ options.put("fields.payload.pk-index.compaction.level-fanout", "1");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("fields.payload.pk-index.compaction.level-fanout")
+ .hasMessageContaining("greater than 1");
+ }
+
private static Map<String, String> enabledOptions() {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.BUCKET.key(), "1");
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyVectorIndexValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyVectorIndexValidationTest.java
index af6155411b..99136a5099 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyVectorIndexValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyVectorIndexValidationTest.java
@@ -245,20 +245,20 @@ class PrimaryKeyVectorIndexValidationTest {
@Test
void testRejectsInvalidAnnCompactionLevelFanout() {
Map<String, String> options = enabledOptions();
- options.put(CoreOptions.PK_VECTOR_INDEX_COMPACTION_LEVEL_FANOUT.key(), "1");
+ options.put("fields.embedding.pk-index.compaction.level-fanout", "1");
assertThatThrownBy(() -> validateTableSchema(schema(options)))
- .hasMessageContaining("pk-vector.index.compaction.level-fanout")
+ .hasMessageContaining("fields.embedding.pk-index.compaction.level-fanout")
.hasMessageContaining("greater than 1");
}
@Test
void testRejectsInvalidAnnCompactionStaleRatio() {
Map<String, String> options = enabledOptions();
- options.put(CoreOptions.PK_VECTOR_INDEX_COMPACTION_STALE_RATIO_THRESHOLD.key(), "1.1");
+ options.put("fields.embedding.pk-index.compaction.stale-ratio-threshold", "1.1");
assertThatThrownBy(() -> validateTableSchema(schema(options)))
- .hasMessageContaining("pk-vector.index.compaction.stale-ratio-threshold")
+ .hasMessageContaining("fields.embedding.pk-index.compaction.stale-ratio-threshold")
.hasMessageContaining("(0, 1]");
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java
index d7318e0eb5..dc276b8d98 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java
@@ -27,6 +27,7 @@ import org.apache.paimon.globalindex.GlobalIndexScanner;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexWriter;
import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.globalindex.KeySerializer;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
@@ -49,6 +50,7 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
@@ -218,6 +220,7 @@ public class BitmapGlobalIndexTableTest extends DataEvolutionTestBase {
InternalRow.createFieldGetter(
indexField.type(), readRowType.getFieldIndex(indexField.name()));
int rowIdIndex = readRowType.getFieldIndex(SpecialFields.ROW_ID.name());
+ List<Pair<Object, Long>> entries = new ArrayList<>();
try (RecordReader<InternalRow> reader =
table.newReadBuilder()
@@ -229,10 +232,24 @@ public class BitmapGlobalIndexTableTest extends DataEvolutionTestBase {
InternalRow row = iterator.next();
long rowId = row.getLong(rowIdIndex);
if (rowId >= rowRange.from && rowId <= rowRange.to) {
- writer.write(fieldGetter.getFieldOrNull(row), rowId - rowRange.from);
+ entries.add(Pair.of(fieldGetter.getFieldOrNull(row), rowId - rowRange.from));
}
}
}
+ Comparator<Object> keyComparator =
+ KeySerializer.create(indexField.type()).createComparator();
+ entries.sort(
+ (left, right) -> {
+ if (left.getKey() == null) {
+ return right.getKey() == null ? 0 : -1;
+ }
+ return right.getKey() == null
+ ? 1
+ : keyComparator.compare(left.getKey(), right.getKey());
+ });
+ for (Pair<Object, Long> entry : entries) {
+ writer.write(entry.getKey(), entry.getValue());
+ }
List<ResultEntry> resultEntries = writer.finish();
List<IndexFileMeta> indexFileMetas =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
index 0cca7fb5e2..8987fe268d 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexReadTest.java
@@ -56,9 +56,6 @@ class PrimaryKeySortedIndexReadTest extends TableTestBase {
.option(CoreOptions.BUCKET.key(), "1")
.option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true")
.option(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "score")
- .option(
- "fields.score.pk-btree.index.options",
- "{\"sorted-index.records-per-range\":\"2\"}")
.build();
}
@@ -146,4 +143,62 @@ class PrimaryKeySortedIndexReadTest extends TableTestBase {
}
assertThat(historicIds).containsExactlyInAnyOrder(1, 3);
}
+
+ @Test
+ void testReadAfterIndexCompaction() throws Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("score", DataTypes.INT())
+ .column("tag", DataTypes.STRING())
+ .primaryKey("id")
+ .option(CoreOptions.BUCKET.key(), "1")
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true")
+ .option(CoreOptions.TARGET_FILE_SIZE.key(), "1 b")
+ .option(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "score")
+ .option(CoreOptions.primaryKeyIndexCompactionLevelFanoutKey("score"), "2")
+ .build();
+ catalog.createTable(identifier(), schema, false);
+ FileStoreTable table = getTableDefault();
+ BinaryString tag = BinaryString.fromString("tag");
+ List<InternalRow> firstBatch = new ArrayList<>();
+ List<InternalRow> secondBatch = new ArrayList<>();
+ List<Integer> expectedIds = new ArrayList<>();
+ for (int id = 1; id <= 2_000; id++) {
+ int score = id % 5;
+ (id % 2 == 0 ? secondBatch : firstBatch).add(GenericRow.of(id, score, tag));
+ if (score == 0) {
+ expectedIds.add(id);
+ }
+ }
+ write(table, ioManager, firstBatch.toArray(new InternalRow[0]));
+ write(table, ioManager, secondBatch.toArray(new InternalRow[0]));
+ compact(table, BinaryRow.EMPTY_ROW, 0, ioManager, true);
+
+ Snapshot snapshot = table.store().snapshotManager().latestSnapshot();
+ List<IndexFileMeta> payloads =
+ table.store()
+ .newIndexFileHandler()
+ .scanSourceIndexes(snapshot, BinaryRow.EMPTY_ROW, 0);
+ assertThat(payloads)
+ .singleElement()
+ .satisfies(
+ payload ->
+ assertThat(
+ PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
+ .sourceFiles())
+ .hasSize(2));
+
+ Predicate predicate = new PredicateBuilder(table.rowType()).equal(1, 0);
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ assertThat(plan.splits()).hasSize(2).allMatch(IndexedSplit.class::isInstance);
+
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ readBuilder.newRead().executeFilter().createReader(plan)) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ assertThat(ids).containsExactlyInAnyOrderElementsOf(expectedIds);
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java
index 3b7089f776..6055e3b500 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java
@@ -77,7 +77,9 @@ class PrimaryKeySortedIndexResultTest {
7,
BTreeGlobalIndexerFactory.IDENTIFIER,
new Options(),
- PrimaryKeyIndexDefinition.Family.BTREE);
+ PrimaryKeyIndexDefinition.Family.BTREE,
+ 5,
+ 0.2);
PrimaryKeySortedIndexScan.Plan plan =
PrimaryKeySortedIndexScan.plan(
11,
@@ -135,7 +137,9 @@ class PrimaryKeySortedIndexResultTest {
7,
BTreeGlobalIndexerFactory.IDENTIFIER,
new Options(),
- PrimaryKeyIndexDefinition.Family.BTREE);
+ PrimaryKeyIndexDefinition.Family.BTREE,
+ 5,
+ 0.2);
PrimaryKeySortedIndexScan.Plan plan =
PrimaryKeySortedIndexScan.plan(
11,
@@ -175,7 +179,9 @@ class PrimaryKeySortedIndexResultTest {
7,
BTreeGlobalIndexerFactory.IDENTIFIER,
new Options(),
- PrimaryKeyIndexDefinition.Family.BTREE);
+ PrimaryKeyIndexDefinition.Family.BTREE,
+ 5,
+ 0.2);
PrimaryKeySortedIndexScan.Plan plan =
PrimaryKeySortedIndexScan.plan(
11,
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
index c0343638cf..e7b72db82e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table.source;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.index.GlobalIndexMeta;
@@ -40,13 +41,16 @@ import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -59,6 +63,7 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/** Tests source-backed BTree and Bitmap planning in file-local row-position space. */
@@ -198,6 +203,74 @@ class PrimaryKeySortedIndexScanTest {
assertThat(evaluated.files().get(0).result().get().results()).containsExactly(3L);
}
+ @Test
+ void testReadMergedSourceGroupInFileLocalPositions() throws IOException {
+ DataFileMeta first = dataFile("data-1", 2);
+ DataFileMeta second = dataFile("data-2", 3);
+ DataSplit split = dataSplit(11, 0, true, second, first);
+ PrimaryKeyIndexDefinition definition =
+ definition(
+ 7,
+ BTreeGlobalIndexerFactory.IDENTIFIER,
+ PrimaryKeyIndexDefinition.Family.BTREE);
+ IndexFileMeta mergedPayload =
+ payload(
+ "btree-merged",
+ Arrays.asList(
+ new PrimaryKeyIndexSourceFile("data-1", 2),
+ new PrimaryKeyIndexSourceFile("data-2", 3)),
+ "btree",
+ 7,
+ 5);
+ PrimaryKeySortedIndexScan.Plan plan =
+ PrimaryKeySortedIndexScan.plan(
+ 11,
+ Collections.singletonList(split),
+ Collections.singletonList(definition),
+ Collections.singletonList(payloadEntry(0, mergedPayload)));
+ RowType rowType = RowType.of(new DataField(7, "f7", DataTypes.INT()));
+ Predicate predicate = new PredicateBuilder(rowType).equal(0, 42);
+ AtomicInteger readersCreated = new AtomicInteger();
+ AtomicInteger queries = new AtomicInteger();
+ CountingRoaringNavigableMap64 groupPositions = new CountingRoaringNavigableMap64();
+ groupPositions.add(1);
+ groupPositions.add(3);
+ groupPositions.add(4);
+ GlobalIndexReader reader = mock(GlobalIndexReader.class);
+ when(reader.visitEqual(any(), eq(42)))
+ .thenAnswer(
+ ignored -> {
+ queries.incrementAndGet();
+ return completedResult(groupPositions);
+ });
+
+ PrimaryKeySortedIndexScan.EvaluatedPlan evaluated =
+ PrimaryKeySortedIndexScan.evaluate(
+ plan,
+ rowType,
+ predicate,
+ Collections.singletonList(definition),
+ (ignoredFile, ignoredDefinition, payloads) -> {
+ readersCreated.incrementAndGet();
+ assertThat(payloads).containsExactly(mergedPayload);
+ return reader;
+ });
+ PrimaryKeySortedIndexResult result = new PrimaryKeySortedIndexResult(evaluated);
+
+ assertThat(readersCreated).hasValue(1);
+ assertThat(queries).hasValue(1);
+ assertThat(groupPositions.iteratedPositions()).isEqualTo(3);
+ verify(reader, times(1)).close();
+ assertThat(result.splits()).hasSize(2);
+ assertThat(result.splits()).allMatch(IndexedSplit.class::isInstance);
+ IndexedSplit secondSplit = (IndexedSplit) result.splits().get(0);
+ assertThat(secondSplit.dataSplit().dataFiles()).containsExactly(second);
+ assertThat(secondSplit.rowRanges()).containsExactly(new Range(1, 2));
+ IndexedSplit firstSplit = (IndexedSplit) result.splits().get(1);
+ assertThat(firstSplit.dataSplit().dataFiles()).containsExactly(first);
+ assertThat(firstSplit.rowRanges()).containsExactly(new Range(1, 1));
+ }
+
@Test
void testPerFileBooleanFallbackSemantics() {
DataSplit split = dataSplit(11, 0, dataFile("data-1", 4));
@@ -283,23 +356,61 @@ class PrimaryKeySortedIndexScanTest {
private static PrimaryKeyIndexDefinition definition(
int fieldId, String indexType, PrimaryKeyIndexDefinition.Family family) {
return new PrimaryKeyIndexDefinition(
- "f" + fieldId, fieldId, indexType, new Options(), family);
+ "f" + fieldId, fieldId, indexType, new Options(), family, 5, 0.2);
}
private static GlobalIndexReader readerWithPositions(long... rowPositions) {
+ GlobalIndexReader reader = mock(GlobalIndexReader.class);
+ when(reader.visitEqual(any(), any())).thenReturn(completedResult(rowPositions));
+ return reader;
+ }
+
+ private static CompletableFuture<Optional<GlobalIndexResult>> completedResult(
+ long... rowPositions) {
RoaringNavigableMap64 positions = new RoaringNavigableMap64();
for (long rowPosition : rowPositions) {
positions.add(rowPosition);
}
- GlobalIndexReader reader = mock(GlobalIndexReader.class);
- when(reader.visitEqual(any(), any()))
- .thenReturn(
- CompletableFuture.completedFuture(
- Optional.of(GlobalIndexResult.create(positions))));
- return reader;
+ return completedResult(positions);
+ }
+
+ private static CompletableFuture<Optional<GlobalIndexResult>> completedResult(
+ RoaringNavigableMap64 positions) {
+ return CompletableFuture.completedFuture(Optional.of(GlobalIndexResult.create(positions)));
+ }
+
+ private static class CountingRoaringNavigableMap64 extends RoaringNavigableMap64 {
+
+ private final AtomicInteger iteratedPositions = new AtomicInteger();
+
+ @Override
+ public Iterator<Long> iterator() {
+ Iterator<Long> wrapped = super.iterator();
+ return new Iterator<Long>() {
+ @Override
+ public boolean hasNext() {
+ return wrapped.hasNext();
+ }
+
+ @Override
+ public Long next() {
+ iteratedPositions.incrementAndGet();
+ return wrapped.next();
+ }
+ };
+ }
+
+ int iteratedPositions() {
+ return iteratedPositions.get();
+ }
}
private static DataSplit dataSplit(long snapshotId, int bucket, DataFileMeta... files) {
+ return dataSplit(snapshotId, bucket, false, files);
+ }
+
+ private static DataSplit dataSplit(
+ long snapshotId, int bucket, boolean rawConvertible, DataFileMeta... files) {
return DataSplit.builder()
.withSnapshot(snapshotId)
.withPartition(BinaryRow.EMPTY_ROW)
@@ -308,7 +419,7 @@ class PrimaryKeySortedIndexScanTest {
.withTotalBuckets(2)
.withDataFiles(Arrays.asList(files))
.isStreaming(false)
- .rawConvertible(false)
+ .rawConvertible(rawConvertible)
.build();
}
@@ -341,10 +452,26 @@ class PrimaryKeySortedIndexScanTest {
String indexType,
int fieldId,
long payloadRowCount) {
- byte[] sourceMeta =
- new PrimaryKeyIndexSourceMeta(
- new PrimaryKeyIndexSourceFile(sourceName, sourceRowCount))
- .serialize();
+ return payload(
+ fileName,
+ Collections.singletonList(
+ new PrimaryKeyIndexSourceFile(sourceName, sourceRowCount)),
+ indexType,
+ fieldId,
+ payloadRowCount);
+ }
+
+ private static IndexFileMeta payload(
+ String fileName,
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
+ String indexType,
+ int fieldId,
+ long payloadRowCount) {
+ byte[] sourceMeta = new PrimaryKeyIndexSourceMeta(sourceFiles).serialize();
+ long sourceRowCount = 0;
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
+ sourceRowCount = Math.addExact(sourceRowCount, sourceFile.rowCount());
+ }
return new IndexFileMeta(
indexType,
fileName,
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
index b68611eb31..2b627913ef 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilder.java
@@ -462,6 +462,20 @@ public class SortedIndexTopoBuilder {
commitMessages.clear();
}
+ @Override
+ public void close() throws Exception {
+ try {
+ GlobalIndexSingleColumnWriter writer = currentWriter;
+ currentWriter = null;
+ counter = 0;
+ if (writer instanceof AutoCloseable) {
+ ((AutoCloseable) writer).close();
+ }
+ } finally {
+ super.close();
+ }
+ }
+
private void flushCurrentWriter() throws IOException {
if (counter > 0 && currentWriter != null) {
commitMessages.add(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
index 21941e35e6..5c47ce05ba 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/globalindex/SortedIndexTopoBuilderTest.java
@@ -19,14 +19,20 @@
package org.apache.paimon.flink.globalindex;
import org.apache.paimon.flink.globalindex.SortedIndexTopoBuilder.SortedBuildTask;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.sorted.SortedGlobalIndexBuilder;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.Range;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.jupiter.api.Test;
+import java.io.Closeable;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -41,6 +47,56 @@ import static org.mockito.Mockito.when;
/** Tests for {@link SortedIndexTopoBuilder}. */
public class SortedIndexTopoBuilderTest {
+ @Test
+ public void testSupportsBitmapAndBTree() {
+ assertThat(SortedIndexTopoBuilder.supports("bitmap")).isTrue();
+ assertThat(SortedIndexTopoBuilder.supports("btree")).isTrue();
+ assertThat(SortedIndexTopoBuilder.supports("inverted")).isFalse();
+ }
+
+ @Test
+ public void testWriteIndexOperatorClosesActiveWriter() throws Exception {
+ Class<?> operatorClass = null;
+ for (Class<?> candidate : SortedIndexTopoBuilder.class.getDeclaredClasses()) {
+ if (candidate.getSimpleName().equals("WriteIndexOperator")) {
+ operatorClass = candidate;
+ break;
+ }
+ }
+ assertThat(operatorClass).isNotNull();
+ Constructor<?> constructor =
+ operatorClass.getDeclaredConstructor(
+ List.class,
+ int.class,
+ SortedGlobalIndexBuilder.class,
+ int.class,
+ int.class,
+ int.class,
+ org.apache.paimon.types.DataType.class);
+ constructor.setAccessible(true);
+ Object operator =
+ constructor.newInstance(
+ Collections.emptyList(),
+ 0,
+ mock(SortedGlobalIndexBuilder.class),
+ 0,
+ 0,
+ 0,
+ DataTypes.INT());
+ GlobalIndexSingleColumnWriter activeWriter =
+ mock(
+ GlobalIndexSingleColumnWriter.class,
+ org.mockito.Mockito.withSettings().extraInterfaces(Closeable.class));
+ Field currentWriter = operatorClass.getDeclaredField("currentWriter");
+ currentWriter.setAccessible(true);
+ currentWriter.set(operator, activeWriter);
+
+ Method close = operatorClass.getMethod("close");
+ close.invoke(operator);
+
+ verify((Closeable) activeWriter).close();
+ }
+
@Test
public void testBuildIndexReturnsFalseWhenNoBuildTask() throws Exception {
SortedGlobalIndexBuilder indexBuilder = mock(SortedGlobalIndexBuilder.class);
diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilderTest.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilderTest.java
index ffb826b6c2..d27a8de8b8 100644
--- a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilderTest.java
+++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexTopoBuilderTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.globalindex;
import org.apache.paimon.options.Options;
+import org.apache.paimon.spark.globalindex.sorted.SortedIndexTopoBuilder;
import org.junit.jupiter.api.Test;
@@ -34,6 +35,12 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link DefaultGlobalIndexTopoBuilder}. */
public class DefaultGlobalIndexTopoBuilderTest {
+ @Test
+ void testBitmapUsesSortedTopologyBuilder() {
+ assertThat(GlobalIndexTopologyBuilderUtils.createTopoBuilder("bitmap"))
+ .isInstanceOf(SortedIndexTopoBuilder.class);
+ }
+
@Test
void testRowsPerShardUsesMergedBuildOptions() {
Map<String, String> tableOptions = new HashMap<>();