[core] Add LSM maintenance for sorted primary-key indexes

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).

docs/generated/core_configuration.html

@@ -1223,18 +1223,6 @@
  <td>String</td>
  <td>Comma-separated VECTOR columns indexed by primary-key vector indexes. Each column owns one index and must define fields.&lt;column&gt;.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>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2755,20 +2755,6 @@ public String toString() {
  .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 boolean primaryKeyVectorIndexEnabled() {
  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 Options primaryKeyBitmapIndexOptions(String column) {
  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 @@ private Options primaryKeySortedIndexOptions(
  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);

paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java

@@ -41,9 +41,7 @@
 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 @@ private static List<DictionaryBlockMeta> readIndexBlock(
  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 @@ SerializedKey firstKey() {
  }
  }
 
- 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;