[core] Unify single-column global index writer (#8275)

This PR merges the previous singleton and parallel single-column global
index writer APIs into one `GlobalIndexSingleColumnWriter` interface.
Single-column index writers now receive the caller-provided
shard-relative row id through `write(@nullable Object key, long
relativeRowId)`.

Author: JingsongLi

…obalindex/GlobalIndexParallelWriter.java …index/GlobalIndexSingleColumnWriter.javapaimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexParallelWriter.java renamed to paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexParallelWriter.java renamed to paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java

@@ -20,16 +20,14 @@
 
 import javax.annotation.Nullable;
 
-/** Parallel Index writer for global index with relative row id (from 0 to rowCnt - 1). */
-public interface GlobalIndexParallelWriter extends GlobalIndexWriter {
+/** Index writer for single-column global index with relative row id (from 0 to rowCnt - 1). */
+public interface GlobalIndexSingleColumnWriter extends GlobalIndexWriter {
 
     /**
-     * Write the indexed key and its related localRowId to the index File. The input row id is
-     * "local" which means it is calculated by the original row id minus the start row id of current
-     * index range.
+     * Write the indexed key and its related relative row id to the index file.
      *
      * @param key nullable index key
-     * @param relativeRowId local row id calculated by {@code rowId - rangeStart}.
+     * @param relativeRowId local row id calculated by {@code rowId - rangeStart}
      */
     void write(@Nullable Object key, long relativeRowId);
 }

paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingletonWriter.java

@@ -20,7 +20,7 @@
 
 import java.util.List;
 
-/** Parallel Index writer for global index with relative row id (from 0 to rowCnt - 1). */
+/** Index writer for global index. */
 public interface GlobalIndexWriter {
 
     List<ResultEntry> finish();

paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexWriter.java

@@ -20,8 +20,7 @@
 
 import org.apache.paimon.compression.BlockCompressionFactory;
 import org.apache.paimon.fs.PositionOutputStream;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
 import org.apache.paimon.globalindex.ResultEntry;
 import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
 import org.apache.paimon.memory.MemorySlice;
@@ -42,9 +41,9 @@
 import java.util.zip.CRC32;
 
 /**
- * The {@link GlobalIndexSingletonWriter} implementation for BTree index. Note that users must keep
- * written keys monotonically incremental. All null keys are stored in a separate bitmap, which will
- * be serialized and appended to the file end on close. The layout is as below:
+ * The {@link GlobalIndexSingleColumnWriter} implementation for BTree index. Note that users must
+ * keep written keys monotonically incremental. All null keys are stored in a separate bitmap, which
+ * will be serialized and appended to the file end on close. The layout is as below:
  *
  * <pre>
  *    +-----------------------------------+------+
@@ -67,7 +66,7 @@
  * <p>For efficiency, we combine entries with the same keys and store a compact list of row ids for
  * each key.
  */
-public class BTreeIndexWriter implements GlobalIndexParallelWriter {
+public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter {
 
     private final String fileName;
     private final PositionOutputStream out;

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java

@@ -26,9 +26,9 @@
 import org.apache.paimon.fs.PositionOutputStream;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
 import org.apache.paimon.globalindex.GlobalIndexReader;
 import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
 import org.apache.paimon.globalindex.ResultEntry;
 import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
 import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
@@ -277,7 +277,7 @@ public void testInPredicate() throws Exception {
     protected abstract GlobalIndexReader prepareDataAndCreateReader() throws Exception;
 
     protected GlobalIndexIOMeta writeData(List<Pair<Object, Long>> data) throws IOException {
-        GlobalIndexParallelWriter indexWriter = globalIndexer.createWriter(fileWriter);
+        GlobalIndexSingleColumnWriter indexWriter = globalIndexer.createWriter(fileWriter);
         for (Pair<Object, Long> pair : data) {
             indexWriter.write(pair.getKey(), pair.getValue());
         }

paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java

@@ -23,7 +23,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.PositionOutputStream;
 import org.apache.paimon.fs.local.LocalFileIO;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
 import org.apache.paimon.globalindex.ResultEntry;
 import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
 import org.apache.paimon.memory.MemorySliceOutput;
@@ -155,7 +155,7 @@ public PositionOutputStream newOutputStream(String fileName)
                                 new Path(new Path(tempPath.toUri()), fileName), true);
                     }
                 };
-        GlobalIndexParallelWriter indexWriter =
+        GlobalIndexSingleColumnWriter indexWriter =
                 new BTreeGlobalIndexer(
                                 new DataField(
                                         1, "testField", new VarCharType(VarCharType.MAX_LENGTH)),

paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java

@@ -23,9 +23,9 @@
 import org.apache.paimon.fs.PositionOutputStream;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
 import org.apache.paimon.globalindex.GlobalIndexReader;
 import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
 import org.apache.paimon.globalindex.ResultEntry;
 import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
 import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
@@ -353,7 +353,7 @@ private List<GlobalIndexIOMeta> writeMultipleFiles() throws IOException {
     }
 
     private GlobalIndexIOMeta writeData(List<Pair<Object, Long>> subData) throws IOException {
-        GlobalIndexParallelWriter indexWriter = globalIndexer.createWriter(fileWriter);
+        GlobalIndexSingleColumnWriter indexWriter = globalIndexer.createWriter(fileWriter);
         for (Pair<Object, Long> pair : subData) {
             indexWriter.write(pair.getKey(), pair.getValue());
         }

paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java

@@ -20,9 +20,9 @@
 
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
 import org.apache.paimon.globalindex.GlobalIndexReader;
 import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
 import org.apache.paimon.globalindex.ResultEntry;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
@@ -233,7 +233,7 @@ public void testConcurrentAccess() throws Exception {
 
     private GlobalIndexIOMeta writeDataWithIndexer(
             BTreeGlobalIndexer indexer, List<Pair<Object, Long>> subData) throws IOException {
-        GlobalIndexParallelWriter indexWriter = indexer.createWriter(fileWriter);
+        GlobalIndexSingleColumnWriter indexWriter = indexer.createWriter(fileWriter);
         for (Pair<Object, Long> pair : subData) {
             indexWriter.write(pair.getKey(), pair.getValue());
         }

paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java

@@ -54,6 +54,7 @@ public class TestFullTextGlobalIndexReader implements GlobalIndexReader {
     private final GlobalIndexIOMeta ioMeta;
 
     private String[] documents;
+    private long[] rowIds;
     private int count;
 
     public TestFullTextGlobalIndexReader(
@@ -90,10 +91,10 @@ public CompletableFuture<Optional<ScoredGlobalIndexResult>> visitFullTextSearch(
                 continue;
             }
             if (topK.size() < effectiveK) {
-                topK.offer(new ScoredRow(i, score));
+                topK.offer(new ScoredRow(rowIds[i], score));
             } else if (score > topK.peek().score) {
                 topK.poll();
-                topK.offer(new ScoredRow(i, score));
+                topK.offer(new ScoredRow(rowIds[i], score));
             }
         }
 
@@ -138,12 +139,14 @@ private void ensureLoaded() throws IOException {
 
             // Read documents
             documents = new String[count];
+            rowIds = new long[count];
             for (int i = 0; i < count; i++) {
-                byte[] lenBytes = new byte[4];
-                readFully(in, lenBytes);
-                ByteBuffer lenBuf = ByteBuffer.wrap(lenBytes);
-                lenBuf.order(ByteOrder.LITTLE_ENDIAN);
-                int textLen = lenBuf.getInt();
+                byte[] entryHeaderBytes = new byte[Long.BYTES + Integer.BYTES];
+                readFully(in, entryHeaderBytes);
+                ByteBuffer entryHeader = ByteBuffer.wrap(entryHeaderBytes);
+                entryHeader.order(ByteOrder.LITTLE_ENDIAN);
+                rowIds[i] = entryHeader.getLong();
+                int textLen = entryHeader.getInt();
 
                 byte[] textBytes = new byte[textLen];
                 readFully(in, textBytes);
@@ -170,6 +173,7 @@ private static void readFully(SeekableInputStream in, byte[] buf) throws IOExcep
     @Override
     public void close() throws IOException {
         documents = null;
+        rowIds = null;
     }
 
     // =================== unsupported predicate operations =====================

paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java

@@ -20,7 +20,7 @@
 
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.fs.PositionOutputStream;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
 import org.apache.paimon.globalindex.ResultEntry;
 import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
 
@@ -41,24 +41,27 @@
  * <pre>
  *   [4 bytes] count (int)
  *   For each document:
+ *     [8 bytes] relative row id (long)
  *     [4 bytes] text length in bytes (int)
  *     [N bytes] UTF-8 text
  * </pre>
  */
-public class TestFullTextGlobalIndexWriter implements GlobalIndexSingletonWriter {
+public class TestFullTextGlobalIndexWriter implements GlobalIndexSingleColumnWriter {
 
     private static final String FILE_NAME_PREFIX = "test-fulltext";
 
     private final GlobalIndexFileWriter fileWriter;
-    private final List<String> documents;
+    private final List<Document> documents;
+    private long rowCount;
 
     public TestFullTextGlobalIndexWriter(GlobalIndexFileWriter fileWriter) {
         this.fileWriter = fileWriter;
         this.documents = new ArrayList<>();
     }
 
     @Override
-    public void write(Object fieldData) {
+    public void write(Object fieldData, long relativeRowId) {
+        rowCount++;
         if (fieldData == null) {
             throw new IllegalArgumentException("Text field data must not be null");
         }
@@ -72,7 +75,7 @@ public void write(Object fieldData) {
             throw new IllegalArgumentException(
                     "Unsupported text type: " + fieldData.getClass().getName());
         }
-        documents.add(text);
+        documents.add(new Document(relativeRowId, text));
     }
 
     @Override
@@ -91,20 +94,31 @@ public List<ResultEntry> finish() {
                 out.write(header.array());
 
                 // Documents
-                for (String doc : documents) {
-                    byte[] textBytes = doc.getBytes(StandardCharsets.UTF_8);
-                    ByteBuffer lenBuf = ByteBuffer.allocate(4);
-                    lenBuf.order(ByteOrder.LITTLE_ENDIAN);
-                    lenBuf.putInt(textBytes.length);
-                    out.write(lenBuf.array());
+                for (Document doc : documents) {
+                    byte[] textBytes = doc.text.getBytes(StandardCharsets.UTF_8);
+                    ByteBuffer entryHeader = ByteBuffer.allocate(Long.BYTES + Integer.BYTES);
+                    entryHeader.order(ByteOrder.LITTLE_ENDIAN);
+                    entryHeader.putLong(doc.relativeRowId);
+                    entryHeader.putInt(textBytes.length);
+                    out.write(entryHeader.array());
                     out.write(textBytes);
                 }
                 out.flush();
             }
 
-            return Collections.singletonList(new ResultEntry(fileName, documents.size(), null));
+            return Collections.singletonList(new ResultEntry(fileName, rowCount, null));
         } catch (IOException e) {
             throw new RuntimeException("Failed to write test full-text index", e);
         }
     }
+
+    private static class Document {
+        private final long relativeRowId;
+        private final String text;
+
+        private Document(long relativeRowId, String text) {
+            this.relativeRowId = relativeRowId;
+            this.text = text;
+        }
+    }
 }