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 c2673f63d3 [core] Support row ID rewriting in scalar global indexes
(#10678)
c2673f63d3 is described below
commit c2673f63d3a22fe21bbea4d0fb1aef140f4e9cfe
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 12:30:16 2026 +0800
[core] Support row ID rewriting in scalar global indexes (#10678)
---
.../apache/paimon/globalindex/GlobalIndexer.java | 21 ++
.../globalindex/bitmap/BitmapGlobalIndexer.java | 111 +++++++++++
.../globalindex/btree/BTreeGlobalIndexer.java | 82 ++++++++
.../globalindex/GlobalIndexRowIdRewriteTest.java | 213 +++++++++++++++++++++
4 files changed, 427 insertions(+)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
index 9792602393..beb08f07c0 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
@@ -29,6 +29,7 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
+import java.util.function.LongUnaryOperator;
/** Abstract base class for global indexers. */
public interface GlobalIndexer {
@@ -49,6 +50,26 @@ public interface GlobalIndexer {
@Nullable List<Range> rowRanges,
ExecutorService executor);
+ /**
+ * Rewrites stored index keys with mapped, relative row IDs without
reading source data.
+ *
+ * <p>The input files share the relative row-ID domain {@code [0,
sourceRowCount)}. The mapping
+ * returns a row ID in {@code [0, targetRowCount)}, or {@code -1} to
exclude a source row.
+ * Output files share the target domain and may need to be read together,
just like the input
+ * files.
+ */
+ default List<ResultEntry> rewriteRowIds(
+ GlobalIndexFileReader reader,
+ List<GlobalIndexIOMeta> files,
+ long sourceRowCount,
+ long targetRowCount,
+ LongUnaryOperator mapping,
+ GlobalIndexFileWriter writer)
+ throws IOException {
+ throw new UnsupportedOperationException(
+ "This global index does not support rewriting stored row
IDs.");
+ }
+
static GlobalIndexer create(String type, List<DataField> indexFields,
Options options) {
return GlobalIndexerFactoryUtils.load(type).create(indexFields,
options);
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
index c63ded522f..a5711abcfc 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
@@ -20,23 +20,33 @@ package org.apache.paimon.globalindex.bitmap;
import org.apache.paimon.compression.BlockCompressionFactory;
import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.SortedGlobalIndexer;
+import org.apache.paimon.globalindex.SortedIndexFileMeta;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringNavigableMap64;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
+import java.util.function.LongUnaryOperator;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/** The {@link GlobalIndexer} for bitmap index. */
public class BitmapGlobalIndexer implements SortedGlobalIndexer {
@@ -86,4 +96,105 @@ public class BitmapGlobalIndexer implements
SortedGlobalIndexer {
return new LazyFilteredBitmapReader(
fileReader, files, keySerializer, fallbackScanMaxSize,
totalRowCount, executor);
}
+
+ @Override
+ public List<ResultEntry> rewriteRowIds(
+ GlobalIndexFileReader fileReader,
+ List<GlobalIndexIOMeta> files,
+ long sourceRowCount,
+ long targetRowCount,
+ LongUnaryOperator mapping,
+ GlobalIndexFileWriter fileWriter)
+ throws IOException {
+ checkArgument(sourceRowCount >= 0 && targetRowCount >= 0, "Invalid row
count.");
+ List<ResultEntry> results = new ArrayList<>();
+ if (targetRowCount == 0) {
+ return results;
+ }
+ for (GlobalIndexIOMeta file : files) {
+ try (SeekableInputStream input = fileReader.getInputStream(file)) {
+ BitmapGlobalIndexFormat.SeekableReader reader =
+ (offset, length) -> {
+ input.seek(offset);
+ byte[] bytes = new byte[length];
+ IOUtils.readFully(input, bytes);
+ return bytes;
+ };
+ BitmapGlobalIndexFormat.Footer footer =
+ BitmapGlobalIndexFormat.readFooter(reader,
file.fileSize());
+ RoaringNavigableMap64 nullRows =
+ mapRowIds(
+ BitmapGlobalIndexFormat.readBitmap(reader,
footer.nullRowsBlock),
+ sourceRowCount,
+ targetRowCount,
+ mapping);
+ RoaringNavigableMap64 nonNullRows =
+ mapRowIds(
+ BitmapGlobalIndexFormat.readBitmap(reader,
footer.nonNullRowsBlock),
+ sourceRowCount,
+ targetRowCount,
+ mapping);
+ if (nullRows.isEmpty() && nonNullRows.isEmpty()) {
+ continue;
+ }
+ String fileName =
fileWriter.newFileName(BitmapGlobalIndexerFactory.IDENTIFIER);
+ byte[] firstKey = null;
+ byte[] lastKey = null;
+ try (PositionOutputStream output =
fileWriter.newOutputStream(fileName)) {
+ BitmapGlobalIndexFormat.StreamingWriter writer =
+ new BitmapGlobalIndexFormat.StreamingWriter(
+ output, dictionaryBlockSize,
compressionFactory);
+ for (BitmapGlobalIndexFormat.DictionaryBlockMeta block :
+ BitmapGlobalIndexFormat.readIndexBlock(reader,
footer.indexBlock)) {
+ for (BitmapGlobalIndexFormat.DictionaryEntry entry :
+
BitmapGlobalIndexFormat.readDictionaryBlock(reader, block)
+ .entries()) {
+ RoaringNavigableMap64 rowIds =
+ mapRowIds(
+ BitmapGlobalIndexFormat.readBitmap(
+ reader,
entry.bitmapBlock()),
+ sourceRowCount,
+ targetRowCount,
+ mapping);
+ if (!rowIds.isEmpty()) {
+ // Dictionary order is unchanged, so encoded
keys need no decoding.
+ writer.write(entry.key(), rowIds);
+ if (firstKey == null) {
+ firstKey = entry.key().bytes();
+ }
+ lastKey = entry.key().bytes();
+ }
+ }
+ }
+ writer.finish(nullRows, nonNullRows);
+ }
+ results.add(
+ new ResultEntry(
+ fileName,
+ nullRows.getLongCardinality() +
nonNullRows.getLongCardinality(),
+ new SortedIndexFileMeta(firstKey, lastKey,
!nullRows.isEmpty())
+ .serialize()));
+ }
+ }
+ return results;
+ }
+
+ private static RoaringNavigableMap64 mapRowIds(
+ RoaringNavigableMap64 rows,
+ long sourceRowCount,
+ long targetRowCount,
+ LongUnaryOperator mapping) {
+ RoaringNavigableMap64 result = new RoaringNavigableMap64();
+ for (long source : rows) {
+ checkArgument(
+ source >= 0 && source < sourceRowCount, "Invalid source
row ID: %s", source);
+ long target = mapping.applyAsLong(source);
+ checkArgument(
+ target >= -1 && target < targetRowCount, "Invalid target
row ID: %s", target);
+ if (target >= 0) {
+ result.add(target);
+ }
+ }
+ return result;
+ }
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
index e8db64056d..8733bc2e10 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
@@ -26,6 +26,7 @@ import org.apache.paimon.globalindex.GlobalIndexKeyExtractor;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
@@ -37,6 +38,7 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BloomFilter;
import org.apache.paimon.utils.LazyField;
import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringNavigableMap64;
import javax.annotation.Nullable;
@@ -44,6 +46,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
+import java.util.function.LongUnaryOperator;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -152,4 +155,83 @@ public class BTreeGlobalIndexer implements
SortedGlobalIndexer {
rowRanges,
executor);
}
+
+ @Override
+ public List<ResultEntry> rewriteRowIds(
+ GlobalIndexFileReader fileReader,
+ List<GlobalIndexIOMeta> files,
+ long sourceRowCount,
+ long targetRowCount,
+ LongUnaryOperator mapping,
+ GlobalIndexFileWriter fileWriter)
+ throws IOException {
+ checkArgument(sourceRowCount >= 0 && targetRowCount >= 0, "Invalid row
count.");
+ List<ResultEntry> results = new ArrayList<>();
+ if (targetRowCount == 0) {
+ return results;
+ }
+ for (GlobalIndexIOMeta file : files) {
+ try (BTreeIndexReader reader =
+ new BTreeIndexReader(
+ keySerializer, fileReader, file,
cacheManager.get(), null)) {
+ RoaringNavigableMap64 nullRows = new RoaringNavigableMap64();
+ reader.scanNullRowIds(
+ rowId ->
+ addMappedRowId(
+ nullRows, rowId, sourceRowCount,
targetRowCount, mapping));
+ BTreeIndexWriter writer = null;
+ try {
+ if (!nullRows.isEmpty()) {
+ writer = createWriter(fileWriter);
+ for (long rowId : nullRows) {
+ writer.write(null, rowId);
+ }
+ }
+ BTreeIndexReader.EntryIterator entries =
reader.entryIterator();
+ while (entries.hasNext()) {
+ BTreeIndexReader.KeyRowIds entry = entries.next();
+ RoaringNavigableMap64 rowIds = new
RoaringNavigableMap64();
+ for (long rowId : entry.rowIds()) {
+ addMappedRowId(rowIds, rowId, sourceRowCount,
targetRowCount, mapping);
+ }
+ if (!rowIds.isEmpty()) {
+ if (writer == null) {
+ writer = createWriter(fileWriter);
+ }
+ // A permutation can reverse a posting list; v2
requires sorted,
+ // unique IDs.
+ for (long rowId : rowIds) {
+ writer.write(entry.key(), rowId);
+ }
+ }
+ }
+ if (writer != null) {
+ results.addAll(writer.finish());
+ }
+ } finally {
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+ }
+ return results;
+ }
+
+ private static void addMappedRowId(
+ RoaringNavigableMap64 result,
+ long sourceRowId,
+ long sourceRowCount,
+ long targetRowCount,
+ LongUnaryOperator mapping) {
+ checkArgument(
+ sourceRowId >= 0 && sourceRowId < sourceRowCount,
+ "Invalid source row ID: %s",
+ sourceRowId);
+ long target = mapping.applyAsLong(sourceRowId);
+ checkArgument(target >= -1 && target < targetRowCount, "Invalid target
row ID: %s", target);
+ if (target >= 0) {
+ result.add(target);
+ }
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexRowIdRewriteTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexRowIdRewriteTest.java
new file mode 100644
index 0000000000..84cf6b9963
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexRowIdRewriteTest.java
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexer;
+import org.apache.paimon.globalindex.btree.BTreeGlobalIndexer;
+import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+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.IntType;
+import org.apache.paimon.utils.Pair;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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 scalar index payload rewrites through query results. */
+class GlobalIndexRowIdRewriteTest {
+
+ private static final DataField FIELD = new DataField(0, "value", new
IntType());
+ private static final FieldRef REF = new FieldRef(0, "value", new
IntType());
+
+ @TempDir java.nio.file.Path tempPath;
+
+ private FileIO fileIO;
+ private Path basePath;
+ private GlobalIndexFileReader fileReader;
+ private GlobalIndexFileWriter fileWriter;
+ private AtomicInteger outputFiles;
+
+ @BeforeEach
+ void setUp() {
+ fileIO = LocalFileIO.create();
+ basePath = new Path(tempPath.toUri());
+ outputFiles = new AtomicInteger();
+ fileReader = meta -> fileIO.newInputStream(meta.filePath());
+ fileWriter =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + "-" + UUID.randomUUID();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String
fileName)
+ throws IOException {
+ outputFiles.incrementAndGet();
+ return fileIO.newOutputStream(new Path(basePath,
fileName), false);
+ }
+ };
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree-v1", "btree-v2", "bitmap"})
+ void testPermutationAndRemovalAcrossPayloadFiles(String type) throws
Exception {
+ GlobalIndexer indexer = indexer(type);
+ List<GlobalIndexIOMeta> source = new ArrayList<>();
+ source.addAll(write(indexer, 0, 2, null, 1, 2));
+ source.addAll(write(indexer, 4, -3, 4, null, 2));
+ long[] mapping = {3, 1, -1, 0, 4, -1, 2, 5};
+ List<ResultEntry> rewritten =
+ indexer.rewriteRowIds(
+ fileReader, source, 8, 6, id -> mapping[(int) id],
fileWriter);
+ assertThat(rewritten).hasSize(2);
+
assertThat(rewritten.stream().mapToLong(ResultEntry::rowCount).sum()).isEqualTo(6);
+ try (GlobalIndexReader reader = reader(indexer, rewritten, 6)) {
+ assertRows(reader.visitEqual(REF, 2).join(), 0, 3, 5);
+ assertRows(reader.visitEqual(REF, -3).join(), 4);
+ assertRows(reader.visitEqual(REF, 1).join());
+ assertRows(reader.visitEqual(REF, 4).join());
+ assertRows(reader.visitIsNull(REF).join(), 1, 2);
+ assertRows(reader.visitIsNotNull(REF).join(), 0, 3, 4, 5);
+ assertRows(reader.visitNotEqual(REF, 2).join(), 4);
+ assertRows(reader.visitLessThan(REF, 0).join(), 4);
+ }
+ // A rewrite must leave the old payload usable by readers pinned to
its snapshot.
+ try (GlobalIndexReader reader =
+ indexer.createReader(fileReader, source, 8, null,
newDirectExecutorService())) {
+ assertRows(reader.visitEqual(REF, 2).join(), 0, 3, 7);
+ assertRows(reader.visitIsNull(REF).join(), 1, 6);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree-v1", "btree-v2", "bitmap"})
+ void testMappedPostingIdsAreSortedAndUnique(String type) throws Exception {
+ GlobalIndexer indexer = indexer(type);
+ List<GlobalIndexIOMeta> source = write(indexer, 0, 7, 7, null, 7);
+ long[] mapping = {2, 1, 0, 1};
+ List<ResultEntry> rewritten =
+ indexer.rewriteRowIds(
+ fileReader, source, 4, 3, id -> mapping[(int) id],
fileWriter);
+ assertThat(rewritten).hasSize(1);
+ assertThat(rewritten.get(0).rowCount()).isEqualTo(3);
+ try (GlobalIndexReader reader = reader(indexer, rewritten, 3)) {
+ assertRows(reader.visitEqual(REF, 7).join(), 1, 2);
+ assertRows(reader.visitIsNull(REF).join(), 0);
+ assertRows(reader.visitNotEqual(REF, 8).join(), 1, 2);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree-v1", "btree-v2", "bitmap"})
+ void testNullOnlyAndFullyExcludedPayloads(String type) throws Exception {
+ GlobalIndexer indexer = indexer(type);
+ List<GlobalIndexIOMeta> source = write(indexer, 0, null, null);
+ List<ResultEntry> rewritten =
+ indexer.rewriteRowIds(fileReader, source, 2, 2, id -> 1 - id,
fileWriter);
+ try (GlobalIndexReader reader = reader(indexer, rewritten, 2)) {
+ assertRows(reader.visitIsNull(REF).join(), 0, 1);
+ assertRows(reader.visitIsNotNull(REF).join());
+ }
+ int before = outputFiles.get();
+ assertThat(indexer.rewriteRowIds(fileReader, source, 2, 1, id -> -1,
fileWriter)).isEmpty();
+ assertThat(indexer.rewriteRowIds(fileReader, source, 2, 0, id -> -1,
fileWriter)).isEmpty();
+ assertThat(outputFiles).hasValue(before);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"btree-v1", "btree-v2", "bitmap"})
+ void testRejectsMappingOutsideTargetDomain(String type) throws Exception {
+ GlobalIndexer indexer = indexer(type);
+ List<GlobalIndexIOMeta> source = write(indexer, 0, 7);
+ assertThatThrownBy(
+ () -> indexer.rewriteRowIds(fileReader, source, 1, 1,
id -> 1, fileWriter))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid target row ID");
+ }
+
+ private GlobalIndexer indexer(String type) {
+ Options options = new Options();
+ if (type.equals("bitmap")) {
+ return new BitmapGlobalIndexer(FIELD, options);
+ }
+ options.set(BTreeIndexOptions.BTREE_INDEX_FILE_VERSION,
type.equals("btree-v1") ? 1 : 2);
+ return new BTreeGlobalIndexer(Collections.singletonList(FIELD),
options);
+ }
+
+ private List<GlobalIndexIOMeta> write(GlobalIndexer indexer, long start,
Integer... values)
+ throws IOException {
+ List<Pair<Integer, Long>> rows = new ArrayList<>();
+ for (int i = 0; i < values.length; i++) {
+ rows.add(Pair.of(values[i], start + i));
+ }
+ Comparator<Integer> comparator =
Comparator.nullsFirst(Integer::compareTo);
+ rows.sort((left, right) -> comparator.compare(left.getKey(),
right.getKey()));
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
indexer.createWriter(fileWriter);
+ for (Pair<Integer, Long> row : rows) {
+ writer.write(row.getKey(), row.getValue());
+ }
+ return toMetas(writer.finish());
+ }
+
+ private GlobalIndexReader reader(
+ GlobalIndexer indexer, List<ResultEntry> entries, long rowCount)
throws IOException {
+ return indexer.createReader(
+ fileReader, toMetas(entries), rowCount, null,
newDirectExecutorService());
+ }
+
+ private List<GlobalIndexIOMeta> toMetas(List<ResultEntry> entries) throws
IOException {
+ List<GlobalIndexIOMeta> result = new ArrayList<>();
+ for (ResultEntry entry : entries) {
+ Path path = new Path(basePath, entry.fileName());
+ result.add(
+ new GlobalIndexIOMeta(
+ path, fileIO.getFileSize(path), entry.rowCount(),
entry.meta()));
+ }
+ return result;
+ }
+
+ private void assertRows(Optional<GlobalIndexResult> result, long...
expected) {
+ assertThat(result).isPresent();
+
assertThat(result.get().results().toArray(Integer.MAX_VALUE)).containsExactly(expected);
+ }
+}
