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 3bee09f21f [core] Unify single-column global index writer (#8275)
3bee09f21f is described below
commit 3bee09f21fbbee2e7f21852aa1a0800b22099d25
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jun 18 12:15:57 2026 +0800
[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)`.
---
...ter.java => GlobalIndexSingleColumnWriter.java} | 10 ++-
.../globalindex/GlobalIndexSingletonWriter.java | 26 -------
.../paimon/globalindex/GlobalIndexWriter.java | 2 +-
.../paimon/globalindex/btree/BTreeIndexWriter.java | 11 ++-
.../globalindex/btree/AbstractIndexReaderTest.java | 4 +-
.../globalindex/btree/BTreeIndexMetaTest.java | 4 +-
.../globalindex/btree/BTreeThreadSafetyTest.java | 4 +-
.../btree/LazyFilteredBTreeIndexReaderTest.java | 4 +-
.../TestFullTextGlobalIndexReader.java | 18 +++--
.../TestFullTextGlobalIndexWriter.java | 38 ++++++----
.../testvector/TestVectorGlobalIndexReader.java | 14 +++-
.../testvector/TestVectorGlobalIndexWriter.java | 41 ++++++++---
.../globalindex/btree/BTreeGlobalIndexBuilder.java | 14 ++--
.../table/source/FullTextSearchBuilderTest.java | 22 +++---
.../table/source/VectorSearchBuilderTest.java | 43 ++++++------
.../paimon/flink/btree/BTreeIndexTopoBuilder.java | 4 +-
.../flink/globalindex/GenericIndexTopoBuilder.java | 7 +-
.../procedure/VectorSearchProcedureITCase.java | 10 +--
.../index/LuminaVectorGlobalIndexWriter.java | 37 +++++-----
.../paimon/lumina/index/JavaPyLuminaE2ETest.java | 23 +++---
.../paimon/lumina/index/LuminaVectorBenchmark.java | 2 +-
.../index/LuminaVectorGlobalIndexScanTest.java | 8 +--
.../lumina/index/LuminaVectorGlobalIndexTest.java | 81 ++++++++++++----------
.../index/LuminaVectorGlobalIndexWriterTest.java | 8 +--
.../globalindex/DefaultGlobalIndexBuilder.java | 22 +++---
.../index/TantivyFullTextGlobalIndexWriter.java | 24 +++----
.../paimon/tantivy/index/JavaPyTantivyE2ETest.java | 10 +--
.../index/TantivyFullTextGlobalIndexTest.java | 32 ++++-----
.../vector/index/VectorGlobalIndexWriter.java | 36 +++++-----
.../paimon/vector/index/VectorGlobalIndexTest.java | 39 ++++++-----
30 files changed, 318 insertions(+), 280 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexParallelWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
similarity index 72%
rename from
paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexParallelWriter.java
rename to
paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
index 810998e6ed..67f5b95c07 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexParallelWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingleColumnWriter.java
@@ -20,16 +20,14 @@ package org.apache.paimon.globalindex;
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);
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingletonWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingletonWriter.java
deleted file mode 100644
index d8a06d874b..0000000000
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexSingletonWriter.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.globalindex;
-
-import javax.annotation.Nullable;
-
-/** Index writer for global index. */
-public interface GlobalIndexSingletonWriter extends GlobalIndexWriter {
- void write(@Nullable Object key);
-}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexWriter.java
index 7e97f89319..89ab394c9d 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexWriter.java
@@ -20,7 +20,7 @@ package org.apache.paimon.globalindex;
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();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
index fe4c14a449..5053b2dbb1 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java
@@ -20,8 +20,7 @@ package org.apache.paimon.globalindex.btree;
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.List;
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 @@ import java.util.zip.CRC32;
* <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;
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java
index 1e81b7c744..b31cc248e0 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java
@@ -26,9 +26,9 @@ import org.apache.paimon.fs.Path;
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 abstract class AbstractIndexReaderTest {
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());
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java
index 5b37aac50d..64467cdc63 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java
@@ -23,7 +23,7 @@ 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.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 @@ class BTreeIndexMetaTest {
new Path(new Path(tempPath.toUri()),
fileName), true);
}
};
- GlobalIndexParallelWriter indexWriter =
+ GlobalIndexSingleColumnWriter indexWriter =
new BTreeGlobalIndexer(
new DataField(
1, "testField", new
VarCharType(VarCharType.MAX_LENGTH)),
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
index af923d1f66..55d08432a6 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
@@ -23,9 +23,9 @@ import org.apache.paimon.fs.Path;
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 @@ public class BTreeThreadSafetyTest {
}
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());
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
index 86e9e8227e..cad9861f40 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
@@ -20,9 +20,9 @@ package org.apache.paimon.globalindex.btree;
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 class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
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());
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java
index ec5c73cd7c..7e9684f267 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.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 class TestFullTextGlobalIndexReader implements
GlobalIndexReader {
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 @@ public class TestFullTextGlobalIndexReader implements
GlobalIndexReader {
// 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 @@ public class TestFullTextGlobalIndexReader implements
GlobalIndexReader {
@Override
public void close() throws IOException {
documents = null;
+ rowIds = null;
}
// =================== unsupported predicate operations
=====================
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexWriter.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexWriter.java
index cf8d2de24a..71ce5a096b 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexWriter.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexWriter.java
@@ -20,7 +20,7 @@ package org.apache.paimon.globalindex.testfulltext;
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,16 +41,18 @@ import java.util.List;
* <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;
@@ -58,7 +60,8 @@ public class TestFullTextGlobalIndexWriter implements
GlobalIndexSingletonWriter
}
@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 class TestFullTextGlobalIndexWriter implements
GlobalIndexSingletonWriter
throw new IllegalArgumentException(
"Unsupported text type: " +
fieldData.getClass().getName());
}
- documents.add(text);
+ documents.add(new Document(relativeRowId, text));
}
@Override
@@ -91,20 +94,31 @@ public class TestFullTextGlobalIndexWriter implements
GlobalIndexSingletonWriter
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;
+ }
+ }
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexReader.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexReader.java
index fd233ba06b..1e8cc432f2 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexReader.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexReader.java
@@ -60,6 +60,7 @@ public class TestVectorGlobalIndexReader implements
GlobalIndexReader {
private final String requiredOptionValue;
private float[][] vectors;
+ private long[] rowIds;
private int dimension;
private int count;
@@ -123,15 +124,15 @@ public class TestVectorGlobalIndexReader implements
GlobalIndexReader {
new PriorityQueue<>(effectiveK + 1,
Comparator.comparingDouble(s -> s.score));
for (int i = 0; i < count; i++) {
- if (includeRowIds != null && !includeRowIds.contains(i)) {
+ if (includeRowIds != null && !includeRowIds.contains(rowIds[i])) {
continue;
}
float score = computeScore(queryVector, vectors[i]);
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));
}
}
@@ -206,8 +207,14 @@ public class TestVectorGlobalIndexReader implements
GlobalIndexReader {
// Read vectors
vectors = new float[count][dimension];
+ rowIds = new long[count];
+ byte[] rowIdBytes = new byte[Long.BYTES];
byte[] vectorBytes = new byte[dimension * Float.BYTES];
for (int i = 0; i < count; i++) {
+ readFully(in, rowIdBytes);
+ ByteBuffer rowIdBuf = ByteBuffer.wrap(rowIdBytes);
+ rowIdBuf.order(ByteOrder.LITTLE_ENDIAN);
+ rowIds[i] = rowIdBuf.getLong();
readFully(in, vectorBytes);
ByteBuffer vectorBuf = ByteBuffer.wrap(vectorBytes);
vectorBuf.order(ByteOrder.LITTLE_ENDIAN);
@@ -236,6 +243,7 @@ public class TestVectorGlobalIndexReader implements
GlobalIndexReader {
@Override
public void close() throws IOException {
vectors = null;
+ rowIds = null;
}
// =================== unsupported predicate operations
=====================
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexWriter.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexWriter.java
index ea8ad346cb..3bb3dde42f 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexWriter.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexWriter.java
@@ -20,7 +20,7 @@ package org.apache.paimon.globalindex.testvector;
import org.apache.paimon.data.InternalArray;
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;
@@ -40,16 +40,19 @@ import java.util.List;
* <pre>
* [4 bytes] dimension (int)
* [4 bytes] count (int)
- * [count * dim * 4 bytes] float vectors (row-major order)
+ * For each vector:
+ * [8 bytes] relative row id (long)
+ * [dim * 4 bytes] float vector
* </pre>
*/
-public class TestVectorGlobalIndexWriter implements GlobalIndexSingletonWriter
{
+public class TestVectorGlobalIndexWriter implements
GlobalIndexSingleColumnWriter {
private static final String FILE_NAME_PREFIX = "test-vector";
private final GlobalIndexFileWriter fileWriter;
private final int dimension;
- private final List<float[]> vectors;
+ private final List<VectorEntry> vectors;
+ private long rowCount;
public TestVectorGlobalIndexWriter(GlobalIndexFileWriter fileWriter, int
dimension) {
this.fileWriter = fileWriter;
@@ -58,7 +61,8 @@ public class TestVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter {
}
@Override
- public void write(Object fieldData) {
+ public void write(Object fieldData, long relativeRowId) {
+ rowCount++;
if (fieldData == null) {
throw new IllegalArgumentException("Vector field data must not be
null");
}
@@ -83,14 +87,14 @@ public class TestVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter {
int expectedDim =
dimension > 0
? dimension
- : (vectors.isEmpty() ? vector.length :
vectors.get(0).length);
+ : (vectors.isEmpty() ? vector.length :
vectors.get(0).vector.length);
if (vector.length != expectedDim) {
throw new IllegalArgumentException(
String.format(
"Vector dimension mismatch: expected %d, but got
%d",
expectedDim, vector.length));
}
- vectors.add(vector);
+ vectors.add(new VectorEntry(relativeRowId, vector));
}
@Override
@@ -99,7 +103,7 @@ public class TestVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter {
return Collections.emptyList();
}
- int dim = dimension > 0 ? dimension : vectors.get(0).length;
+ int dim = dimension > 0 ? dimension : vectors.get(0).vector.length;
int count = vectors.size();
try {
@@ -113,21 +117,36 @@ public class TestVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter {
out.write(header.array());
// Vector data
+ ByteBuffer rowIdBuf = ByteBuffer.allocate(Long.BYTES);
+ rowIdBuf.order(ByteOrder.LITTLE_ENDIAN);
ByteBuffer vectorBuf = ByteBuffer.allocate(dim * Float.BYTES);
vectorBuf.order(ByteOrder.LITTLE_ENDIAN);
- for (float[] vec : vectors) {
+ for (VectorEntry entry : vectors) {
+ rowIdBuf.clear();
+ rowIdBuf.putLong(entry.relativeRowId);
+ out.write(rowIdBuf.array());
vectorBuf.clear();
for (int i = 0; i < dim; i++) {
- vectorBuf.putFloat(vec[i]);
+ vectorBuf.putFloat(entry.vector[i]);
}
out.write(vectorBuf.array());
}
out.flush();
}
- return Collections.singletonList(new ResultEntry(fileName, count,
null));
+ return Collections.singletonList(new ResultEntry(fileName,
rowCount, null));
} catch (IOException e) {
throw new RuntimeException("Failed to write test vector index", e);
}
}
+
+ private static class VectorEntry {
+ private final long relativeRowId;
+ private final float[] vector;
+
+ private VectorEntry(long relativeRowId, float[] vector) {
+ this.relativeRowId = relativeRowId;
+ this.vector = vector;
+ }
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilder.java
index ad68d83eb3..0fcbbcbb43 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilder.java
@@ -26,7 +26,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.InternalRow.FieldGetter;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.globalindex.DataEvolutionBatchScan;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexWriter;
import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.ResultEntry;
@@ -256,7 +256,7 @@ public class BTreeGlobalIndexBuilder implements
Serializable {
public List<CommitMessage> buildForSinglePartition(
Range rowRange, BinaryRow partition, Iterator<InternalRow> data)
throws IOException {
long counter = 0;
- GlobalIndexParallelWriter currentWriter = null;
+ GlobalIndexSingleColumnWriter currentWriter = null;
List<CommitMessage> commitMessages = new ArrayList<>();
FieldGetter indexFieldGetter =
InternalRow.createFieldGetter(indexField.type(), 0);
@@ -284,15 +284,15 @@ public class BTreeGlobalIndexBuilder implements
Serializable {
return commitMessages;
}
- public GlobalIndexParallelWriter createWriter() throws IOException {
- GlobalIndexParallelWriter currentWriter;
+ public GlobalIndexSingleColumnWriter createWriter() throws IOException {
+ GlobalIndexSingleColumnWriter currentWriter;
GlobalIndexWriter indexWriter = createIndexWriter(table, INDEX_TYPE,
indexField, options);
- if (!(indexWriter instanceof GlobalIndexParallelWriter)) {
+ if (!(indexWriter instanceof GlobalIndexSingleColumnWriter)) {
throw new RuntimeException(
- "Unexpected implementation, the index writer of BTree
should be an instance of GlobalIndexParallelWriter, but found: "
+ "Unexpected implementation, the index writer of BTree
should be an instance of GlobalIndexSingleColumnWriter, but found: "
+ indexWriter.getClass().getName());
}
- currentWriter = (GlobalIndexParallelWriter) indexWriter;
+ currentWriter = (GlobalIndexSingleColumnWriter) indexWriter;
return currentWriter;
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
index 2d7b8c97fa..13f84d1b97 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
@@ -25,7 +25,7 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.GlobalIndexResult;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import
org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexerFactory;
@@ -361,15 +361,15 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
Options options = table.coreOptions().toConfiguration();
DataField textField = table.rowType().getField(TEXT_FIELD_NAME);
- GlobalIndexSingletonWriter writer =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestFullTextGlobalIndexerFactory.IDENTIFIER,
textField,
options);
- for (String doc : documents) {
- writer.write(doc);
+ for (int i = 0; i < documents.length; i++) {
+ writer.write(documents[i], i);
}
List<ResultEntry> entries = writer.finish();
@@ -404,15 +404,15 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
int mid = documents.length / 2;
// Build first index file covering rows [0, mid)
- GlobalIndexSingletonWriter writer1 =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer1 =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestFullTextGlobalIndexerFactory.IDENTIFIER,
textField,
options);
for (int i = 0; i < mid; i++) {
- writer1.write(documents[i]);
+ writer1.write(documents[i], i);
}
List<ResultEntry> entries1 = writer1.finish();
Range rowRange1 = new Range(0, mid - 1);
@@ -427,15 +427,15 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
entries1);
// Build second index file covering rows [mid, end)
- GlobalIndexSingletonWriter writer2 =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer2 =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestFullTextGlobalIndexerFactory.IDENTIFIER,
textField,
options);
for (int i = mid; i < documents.length; i++) {
- writer2.write(documents[i]);
+ writer2.write(documents[i], i - mid);
}
List<ResultEntry> entries2 = writer2.finish();
Range rowRange2 = new Range(mid, documents.length - 1);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
index 75b6897e8e..fb5ca42764 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
@@ -25,9 +25,8 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
import org.apache.paimon.globalindex.GlobalIndexResult;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
@@ -611,15 +610,15 @@ public class VectorSearchBuilderTest extends
TableTestBase {
Options options = table.coreOptions().toConfiguration();
DataField vectorField = table.rowType().getField(fieldName);
- GlobalIndexSingletonWriter writer =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestVectorGlobalIndexerFactory.IDENTIFIER,
vectorField,
options);
- for (float[] vec : vectors) {
- writer.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
}
List<ResultEntry> entries = writer.finish();
@@ -654,15 +653,15 @@ public class VectorSearchBuilderTest extends
TableTestBase {
int mid = vectors.length / 2;
// Build first index file covering rows [0, mid)
- GlobalIndexSingletonWriter writer1 =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer1 =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestVectorGlobalIndexerFactory.IDENTIFIER,
vectorField,
options);
for (int i = 0; i < mid; i++) {
- writer1.write(vectors[i]);
+ writer1.write(vectors[i], i);
}
List<ResultEntry> entries1 = writer1.finish();
Range rowRange1 = new Range(0, mid - 1);
@@ -677,15 +676,15 @@ public class VectorSearchBuilderTest extends
TableTestBase {
entries1);
// Build second index file covering rows [mid, end)
- GlobalIndexSingletonWriter writer2 =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer2 =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestVectorGlobalIndexerFactory.IDENTIFIER,
vectorField,
options);
for (int i = mid; i < vectors.length; i++) {
- writer2.write(vectors[i]);
+ writer2.write(vectors[i], i - mid);
}
List<ResultEntry> entries2 = writer2.finish();
Range rowRange2 = new Range(mid, vectors.length - 1);
@@ -819,15 +818,15 @@ public class VectorSearchBuilderTest extends
TableTestBase {
Options options = table.coreOptions().toConfiguration();
DataField vectorField = table.rowType().getField(VECTOR_FIELD_NAME);
- GlobalIndexSingletonWriter writer =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestVectorGlobalIndexerFactory.IDENTIFIER,
vectorField,
options);
- for (float[] vec : vectors) {
- writer.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
}
List<ResultEntry> entries = writer.finish();
@@ -859,8 +858,8 @@ public class VectorSearchBuilderTest extends TableTestBase {
Options options = table.coreOptions().toConfiguration();
DataField idField = table.rowType().getField("id");
- GlobalIndexParallelWriter writer =
- (GlobalIndexParallelWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table, BTreeGlobalIndexerFactory.IDENTIFIER,
idField, options);
for (int id : ids) {
@@ -898,15 +897,15 @@ public class VectorSearchBuilderTest extends
TableTestBase {
Options options = table.coreOptions().toConfiguration();
DataField vectorField = table.rowType().getField(VECTOR_FIELD_NAME);
- GlobalIndexSingletonWriter writer =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestVectorGlobalIndexerFactory.IDENTIFIER,
vectorField,
options);
- for (float[] vec : vectors) {
- writer.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
}
List<ResultEntry> entries = writer.finish();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
index 9542eea4a4..98697fae73 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
@@ -37,7 +37,7 @@ import org.apache.paimon.flink.sorter.TableSorter;
import org.apache.paimon.flink.utils.BoundedOneInputOperator;
import org.apache.paimon.flink.utils.JavaTypeInfo;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
-import org.apache.paimon.globalindex.GlobalIndexParallelWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexBuilder;
import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
import org.apache.paimon.options.Options;
@@ -377,7 +377,7 @@ public class BTreeIndexTopoBuilder {
private transient long counter;
private transient BTreeBuildTask currentTask;
private transient BinaryRow currentPartition;
- private transient GlobalIndexParallelWriter currentWriter;
+ private transient GlobalIndexSingleColumnWriter currentWriter;
private transient List<CommitMessage> commitMessages;
private transient Map<Integer, BTreeBuildTask> buildTasksById;
private transient InternalRow.FieldGetter indexFieldGetter;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
index af256da8ec..39e6db0cf8 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/globalindex/GenericIndexTopoBuilder.java
@@ -30,7 +30,7 @@ import org.apache.paimon.flink.utils.BoundedOneInputOperator;
import org.apache.paimon.flink.utils.JavaTypeInfo;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.index.IndexFileMeta;
@@ -684,13 +684,14 @@ public class GenericIndexTopoBuilder {
}
// Only write rows within this shard's range
if (currentRowId >= task.shardRange.from) {
+ long rowId = currentRowId - task.shardRange.from;
if (multiColumn) {
- long rowId = currentRowId -
task.shardRange.from;
((GlobalIndexMultiColumnWriter) indexWriter)
.write(rowId,
writerProjection.replaceRow(row));
} else {
Object fieldData =
indexFieldGetters[0].getFieldOrNull(row);
- ((GlobalIndexSingletonWriter)
indexWriter).write(fieldData);
+ ((GlobalIndexSingleColumnWriter) indexWriter)
+ .write(fieldData, rowId);
}
rowsSeen++;
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java
index 7beab44fdc..d71de22d2a 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/VectorSearchProcedureITCase.java
@@ -23,7 +23,7 @@ import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.CatalogITCaseBase;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexer;
import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory;
@@ -218,15 +218,15 @@ public class VectorSearchProcedureITCase extends
CatalogITCaseBase {
Options options = table.coreOptions().toConfiguration();
DataField vectorField = table.rowType().getField(VECTOR_FIELD);
- GlobalIndexSingletonWriter writer =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TestVectorGlobalIndexerFactory.IDENTIFIER,
vectorField,
options);
- for (float[] vec : vectors) {
- writer.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
}
List<ResultEntry> entries = writer.finish();
diff --git
a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
index 206da270a4..9d9ebeaad8 100644
---
a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
+++
b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriter.java
@@ -21,7 +21,7 @@ package org.apache.paimon.lumina.index;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalVector;
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;
import org.apache.paimon.types.ArrayType;
@@ -50,16 +50,17 @@ import java.util.concurrent.ConcurrentHashMap;
/**
* Vector global index writer using Lumina. Builds a single index file per
shard.
*
- * <p>Vectors are spilled to a temporary file on disk as they arrive via
{@link #write(Object)},
- * keeping Java heap usage constant (~8 MB buffer) regardless of dataset size.
During index build,
- * the Lumina builder streams vectors from the temp file via the {@link
LuminaDataset} callback API.
+ * <p>Vectors are spilled to a temporary file on disk as they arrive via
{@link #write(Object,
+ * long)}, keeping Java heap usage constant (~8 MB buffer) regardless of
dataset size. During index
+ * build, the Lumina builder streams vectors from the temp file via the {@link
LuminaDataset}
+ * callback API.
*
* <p><b>Thread safety:</b> This class is <b>not</b> thread-safe. The
underlying {@code
* LuminaBuilder} must be used from a single thread or the caller must provide
external
* synchronization. The internal Lumina executor thread pool size is
controlled globally by the
* {@code LUMINA_EXECUTOR_THREAD_COUNT} environment variable and cannot be
configured per-instance.
*/
-public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter, Closeable {
+public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Closeable {
private static final String FILE_NAME_PREFIX = "lumina";
@@ -92,7 +93,7 @@ public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter
private long count;
private boolean closed;
- private long logicalRowId;
+ private long rowCount;
public LuminaVectorGlobalIndexWriter(
GlobalIndexFileWriter fileWriter,
@@ -154,23 +155,23 @@ public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter
}
@Override
- public void write(Object fieldData) {
+ public void write(Object fieldData, long relativeRowId) {
if (fieldData == null) {
- logicalRowId++;
+ rowCount++;
return;
}
// Validation must complete before any buffer/state mutation below
- float[] src = materializeAndValidate(fieldData);
+ float[] src = materializeAndValidate(fieldData, relativeRowId);
if (writeBuf.remaining() < recordSizeInBytes) {
flushWriteBuffer();
}
- writeBuf.putLong(logicalRowId);
+ writeBuf.putLong(relativeRowId);
for (int i = 0; i < dim; i++) {
writeBuf.putFloat(src[i]);
}
- logicalRowId++;
+ rowCount++;
count++;
}
@@ -179,12 +180,12 @@ public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter
* input directly (zero-copy). For InternalVector/InternalArray, reads
into the reusable
* vectorBuf field.
*/
- private float[] materializeAndValidate(Object fieldData) {
+ private float[] materializeAndValidate(Object fieldData, long
relativeRowId) {
if (fieldData instanceof float[]) {
float[] vector = (float[]) fieldData;
checkDimension(vector.length);
for (int i = 0; i < dim; i++) {
- checkFinite(vector[i], i);
+ checkFinite(vector[i], relativeRowId, i);
}
return vector;
} else if (fieldData instanceof InternalVector) {
@@ -192,7 +193,7 @@ public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter
checkDimension(vector.size());
for (int i = 0; i < dim; i++) {
float v = vector.getFloat(i);
- checkFinite(v, i);
+ checkFinite(v, relativeRowId, i);
vectorBuf[i] = v;
}
return vectorBuf;
@@ -204,7 +205,7 @@ public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter
throw new IllegalArgumentException("Vector element at
index " + i + " is null");
}
float v = array.getFloat(i);
- checkFinite(v, i);
+ checkFinite(v, relativeRowId, i);
vectorBuf[i] = v;
}
return vectorBuf;
@@ -301,7 +302,7 @@ public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter
LuminaIndexMeta meta = new LuminaIndexMeta(luminaOptions);
// rowCount = logical rows including nulls (not just indexed
vectors)
- return new ResultEntry(fileName, logicalRowId, meta.serialize());
+ return new ResultEntry(fileName, rowCount, meta.serialize());
}
}
@@ -365,12 +366,12 @@ public class LuminaVectorGlobalIndexWriter implements
GlobalIndexSingletonWriter
}
}
- private void checkFinite(float value, int elementIndex) {
+ private void checkFinite(float value, long relativeRowId, int
elementIndex) {
if (!Float.isFinite(value)) {
throw new IllegalArgumentException(
String.format(
"Vector element at rowId=%d, index=%d is %s",
- logicalRowId, elementIndex,
Float.toString(value)));
+ relativeRowId, elementIndex,
Float.toString(value)));
}
}
diff --git
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java
index ce613dfd61..d402a40ac9 100644
---
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java
+++
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/JavaPyLuminaE2ETest.java
@@ -25,8 +25,7 @@ import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
-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.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.index.IndexFileMeta;
@@ -170,16 +169,16 @@ public class JavaPyLuminaE2ETest {
DataField embeddingField = table.rowType().getField("embedding");
Options indexOptions = table.coreOptions().toConfiguration();
- GlobalIndexSingletonWriter writer =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
LuminaVectorGlobalIndexerFactory.IDENTIFIER,
embeddingField,
indexOptions);
- for (float[] vec : vectors) {
- writer.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
}
List<ResultEntry> entries = writer.finish();
@@ -280,15 +279,15 @@ public class JavaPyLuminaE2ETest {
// Build Lumina vector index on "embedding".
DataField embeddingField = table.rowType().getField("embedding");
- GlobalIndexSingletonWriter vectorWriter =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter vectorWriter =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
LuminaVectorGlobalIndexerFactory.IDENTIFIER,
embeddingField,
indexOptions);
- for (float[] vec : vectors) {
- vectorWriter.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ vectorWriter.write(vectors[i], i);
}
List<ResultEntry> vectorEntries = vectorWriter.finish();
List<IndexFileMeta> vectorIndexFiles =
@@ -303,8 +302,8 @@ public class JavaPyLuminaE2ETest {
// Build BTree global index on "id".
DataField idField = table.rowType().getField("id");
- GlobalIndexParallelWriter idWriter =
- (GlobalIndexParallelWriter)
+ GlobalIndexSingleColumnWriter idWriter =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table, BTreeGlobalIndexerFactory.IDENTIFIER,
idField, indexOptions);
for (int i = 0; i < vectors.length; i++) {
diff --git
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorBenchmark.java
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorBenchmark.java
index 1458352bba..3511fba4a3 100644
---
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorBenchmark.java
+++
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorBenchmark.java
@@ -388,7 +388,7 @@ public class LuminaVectorBenchmark {
for (int d = 0; d < dimension; d++) {
vec[d] = (float) insertRandom.nextDouble() * 2 - 1;
}
- writer.write(vec);
+ writer.write(vec, i);
}
long writeEnd = System.currentTimeMillis();
System.out.printf(
diff --git
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java
index 52100fb2e1..85d4ebe6be 100644
---
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java
+++
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexScanTest.java
@@ -212,8 +212,8 @@ public class LuminaVectorGlobalIndexScanTest {
LuminaVectorGlobalIndexWriter indexWriter =
new LuminaVectorGlobalIndexWriter(
fileWriter, new ArrayType(DataTypes.FLOAT()),
indexOptions);
- for (float[] vec : vectors) {
- indexWriter.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ indexWriter.write(vectors[i], i);
}
List<ResultEntry> entries = indexWriter.finish();
@@ -302,8 +302,8 @@ public class LuminaVectorGlobalIndexScanTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(
fileWriter, new ArrayType(DataTypes.FLOAT()),
indexOptions);
- for (float[] vec : vectors) {
- writer.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
}
List<ResultEntry> entries = writer.finish();
diff --git
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexTest.java
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexTest.java
index ab37247295..b9ab7e1219 100644
---
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexTest.java
+++
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexTest.java
@@ -46,7 +46,6 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Random;
@@ -150,7 +149,7 @@ public class LuminaVectorGlobalIndexTest {
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
List<float[]> testVectors = generateRandomVectors(numVectors,
dimension);
- testVectors.forEach(writer::write);
+ writeVectors(writer, testVectors);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results,
metricIndexPath);
@@ -185,7 +184,7 @@ public class LuminaVectorGlobalIndexTest {
int numVectors = 10;
List<float[]> testVectors = generateRandomVectors(numVectors,
dimension);
- testVectors.forEach(writer::write);
+ writeVectors(writer, testVectors);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, dimIndexPath);
@@ -216,7 +215,7 @@ public class LuminaVectorGlobalIndexTest {
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
float[] wrongDimVector = new float[32];
- assertThatThrownBy(() -> writer.write(wrongDimVector))
+ assertThatThrownBy(() -> writer.write(wrongDimVector, 0))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("dimension mismatch");
}
@@ -236,7 +235,7 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorIndexOptions indexOptions = new
LuminaVectorIndexOptions(options);
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- Arrays.stream(vectors).forEach(writer::write);
+ writeVectors(writer, vectors);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -299,7 +298,7 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorIndexOptions indexOptions = new
LuminaVectorIndexOptions(options);
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- Arrays.stream(vectors).forEach(writer::write);
+ writeVectors(writer, vectors);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -366,7 +365,7 @@ public class LuminaVectorGlobalIndexTest {
int numVectors = 350;
List<float[]> testVectors = generateRandomVectors(numVectors,
dimension);
- testVectors.forEach(writer::write);
+ writeVectors(writer, testVectors);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -417,7 +416,7 @@ public class LuminaVectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
writeIndexOptions);
- Arrays.stream(vectors).forEach(writer::write);
+ writeVectors(writer, vectors);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -467,8 +466,8 @@ public class LuminaVectorGlobalIndexTest {
new LuminaVectorGlobalIndexWriter(fileWriter, vecFieldType,
indexOptions);
// Write using BinaryVector (InternalVector)
- for (float[] vec : vectors) {
- writer.write(BinaryVector.fromPrimitiveArray(vec));
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(BinaryVector.fromPrimitiveArray(vectors[i]), i);
}
List<ResultEntry> results = writer.finish();
@@ -507,8 +506,8 @@ public class LuminaVectorGlobalIndexTest {
new float[] {0.0f, 1.0f},
new float[] {0.7f, 0.7f}
};
- for (float[] vec : vectors) {
- writer.write(vec);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
}
List<ResultEntry> results = writer.finish();
@@ -548,12 +547,12 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- writer.write(vectors[0]); // row 0
- writer.write(null); // row 1 - null
- writer.write(vectors[1]); // row 2
- writer.write(null); // row 3 - null
- writer.write(null); // row 4 - null
- writer.write(vectors[2]); // row 5
+ writer.write(vectors[0], 0); // row 0
+ writer.write(null, 1); // row 1 - null
+ writer.write(vectors[1], 2); // row 2
+ writer.write(null, 3); // row 3 - null
+ writer.write(null, 4); // row 4 - null
+ writer.write(vectors[2], 5); // row 5
List<ResultEntry> results = writer.finish();
assertThat(results).hasSize(1);
@@ -591,9 +590,9 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- writer.write(null);
- writer.write(null);
- writer.write(null);
+ writer.write(null, 0);
+ writer.write(null, 1);
+ writer.write(null, 2);
List<ResultEntry> results = writer.finish();
assertThat(results).isEmpty();
@@ -618,12 +617,12 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- writer.write(vectors[0]); // row 0
- writer.write(null); // row 1 - null
- writer.write(vectors[1]); // row 2
- writer.write(vectors[2]); // row 3
- writer.write(null); // row 4 - null
- writer.write(vectors[3]); // row 5
+ writer.write(vectors[0], 0); // row 0
+ writer.write(null, 1); // row 1 - null
+ writer.write(vectors[1], 2); // row 2
+ writer.write(vectors[2], 3); // row 3
+ writer.write(null, 4); // row 4 - null
+ writer.write(vectors[3], 5); // row 5
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -659,8 +658,8 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- writer.write(null); // row 0 - null
- writer.write(new float[] {1.0f, 0.0f}); // row 1
+ writer.write(null, 0); // row 0 - null
+ writer.write(new float[] {1.0f, 0.0f}, 1); // row 1
List<ResultEntry> results = writer.finish();
assertThat(results).hasSize(1);
@@ -689,8 +688,8 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- writer.write(new float[] {1.0f, 0.0f}); // row 0
- writer.write(null); // row 1 - null
+ writer.write(new float[] {1.0f, 0.0f}, 0); // row 0
+ writer.write(null, 1); // row 1 - null
List<ResultEntry> results = writer.finish();
assertThat(results).hasSize(1);
@@ -719,7 +718,7 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- assertThatThrownBy(() -> writer.write(new float[] {1.0f, Float.NaN}))
+ assertThatThrownBy(() -> writer.write(new float[] {1.0f, Float.NaN},
0))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("rowId=0")
.hasMessageContaining("index=1")
@@ -736,14 +735,14 @@ public class LuminaVectorGlobalIndexTest {
LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(fileWriter, vectorType,
indexOptions);
- writer.write(null); // row 0 - null, advances logicalRowId
- assertThatThrownBy(() -> writer.write(new float[]
{Float.POSITIVE_INFINITY, 0.0f}))
+ writer.write(null, 0); // row 0 - null
+ assertThatThrownBy(() -> writer.write(new float[]
{Float.POSITIVE_INFINITY, 0.0f}, 1))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("rowId=1")
.hasMessageContaining("index=0")
.hasMessageContaining("Infinity");
- assertThatThrownBy(() -> writer.write(new float[] {0.0f,
Float.NEGATIVE_INFINITY}))
+ assertThatThrownBy(() -> writer.write(new float[] {0.0f,
Float.NEGATIVE_INFINITY}, 1))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("rowId=1")
.hasMessageContaining("index=1")
@@ -757,6 +756,18 @@ public class LuminaVectorGlobalIndexTest {
return options;
}
+ private void writeVectors(LuminaVectorGlobalIndexWriter writer,
List<float[]> vectors) {
+ for (int i = 0; i < vectors.size(); i++) {
+ writer.write(vectors.get(i), i);
+ }
+ }
+
+ private void writeVectors(LuminaVectorGlobalIndexWriter writer, float[][]
vectors) {
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
+ }
+ }
+
private List<float[]> generateRandomVectors(int count, int dimension) {
Random random = new Random(42);
List<float[]> vectors = new ArrayList<>();
diff --git
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriterTest.java
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriterTest.java
index b0b1996e73..ecae030045 100644
---
a/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriterTest.java
+++
b/paimon-lumina/src/test/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexWriterTest.java
@@ -54,9 +54,9 @@ public class LuminaVectorGlobalIndexWriterTest {
LuminaVectorIndexOptions.DIMENSION),
"2");
- writer.write(new float[] {1.0f, 0.0f});
+ writer.write(new float[] {1.0f, 0.0f}, 0);
- assertThatThrownBy(() -> writer.write(new float[] {1.0f, 0.0f,
0.0f}))
+ assertThatThrownBy(() -> writer.write(new float[] {1.0f, 0.0f,
0.0f}, 1))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("expected 2")
.hasMessageContaining("got 3");
@@ -94,9 +94,9 @@ public class LuminaVectorGlobalIndexWriterTest {
try (LuminaVectorGlobalIndexWriter writer =
new LuminaVectorGlobalIndexWriter(
createNoopFileWriter(), arrayFieldType, indexOptions))
{
- writer.write(new float[] {1.0f, 0.0f, 0.0f});
+ writer.write(new float[] {1.0f, 0.0f, 0.0f}, 0);
- assertThatThrownBy(() -> writer.write(new float[] {1.0f, 0.0f}))
+ assertThatThrownBy(() -> writer.write(new float[] {1.0f, 0.0f}, 1))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("expected 3")
.hasMessageContaining("got 2");
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
index ae87dc96a4..fd13f5ee0a 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/globalindex/DefaultGlobalIndexBuilder.java
@@ -21,7 +21,7 @@ package org.apache.paimon.spark.globalindex;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.GlobalIndexWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.index.IndexFileMeta;
@@ -165,16 +165,22 @@ public class DefaultGlobalIndexBuilder implements
Serializable {
rowCounter.add(1);
}
} else {
- GlobalIndexSingletonWriter singleWriter =
(GlobalIndexSingletonWriter) indexWriter;
+ GlobalIndexSingleColumnWriter singleWriter =
+ (GlobalIndexSingleColumnWriter) indexWriter;
InternalRow.FieldGetter getter =
InternalRow.createFieldGetter(
indexField.type(),
readType.getFieldIndex(indexField.name()));
- rows.forEachRemaining(
- row -> {
- Object indexO = getter.getFieldOrNull(row);
- singleWriter.write(indexO);
- rowCounter.add(1);
- });
+ int rowIdIndex =
readType.getFieldIndex(SpecialFields.ROW_ID.name());
+ while (rows.hasNext()) {
+ InternalRow row = rows.next();
+ long absRowId = row.getLong(rowIdIndex);
+ if (absRowId < rowRange.from || absRowId > rowRange.to) {
+ continue;
+ }
+ Object indexO = getter.getFieldOrNull(row);
+ singleWriter.write(indexO, absRowId - rowRange.from);
+ rowCounter.add(1);
+ }
}
return indexWriter.finish();
} finally {
diff --git
a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexWriter.java
b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexWriter.java
index 66b6c19f6d..052a8cf2a5 100644
---
a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexWriter.java
+++
b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexWriter.java
@@ -20,7 +20,7 @@ package org.apache.paimon.tantivy.index;
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;
import org.apache.paimon.tantivy.TantivyIndexWriter;
@@ -48,7 +48,7 @@ import java.util.List;
* <p>Text data is written to a local Tantivy index via JNI. On {@link
#finish()}, the index
* directory is packed into a single file and written to the global index file
system.
*/
-public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingletonWriter, Closeable {
+public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingleColumnWriter, Closeable {
private static final String FILE_NAME_PREFIX = "tantivy";
private static final Logger LOG =
@@ -58,7 +58,7 @@ public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingletonWri
private final TantivyFullTextIndexOptions indexOptions;
private File tempIndexDir;
private TantivyIndexWriter writer;
- private long rowId;
+ private long rowCount;
private boolean closed;
public TantivyFullTextGlobalIndexWriter(GlobalIndexFileWriter fileWriter) {
@@ -69,7 +69,7 @@ public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingletonWri
GlobalIndexFileWriter fileWriter, TantivyFullTextIndexOptions
indexOptions) {
this.fileWriter = fileWriter;
this.indexOptions = indexOptions;
- this.rowId = 0;
+ this.rowCount = 0;
this.closed = false;
try {
@@ -84,9 +84,9 @@ public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingletonWri
}
@Override
- public void write(Object fieldData) {
+ public void write(Object fieldData, long relativeRowId) {
if (fieldData == null) {
- rowId++;
+ rowCount++;
return;
}
@@ -100,14 +100,14 @@ public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingletonWri
"Unsupported field type: " +
fieldData.getClass().getName());
}
- writer.addDocument(rowId, text);
- rowId++;
+ writer.addDocument(relativeRowId, text);
+ rowCount++;
}
@Override
public List<ResultEntry> finish() {
try {
- if (rowId == 0) {
+ if (rowCount == 0) {
return Collections.emptyList();
}
@@ -129,7 +129,7 @@ public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingletonWri
}
private ResultEntry packIndex() throws IOException {
- LOG.info("Packing Tantivy index: {} documents", rowId);
+ LOG.info("Packing Tantivy index for {} rows", rowCount);
String fileName = fileWriter.newFileName(FILE_NAME_PREFIX);
try (PositionOutputStream out = fileWriter.newOutputStream(fileName)) {
@@ -170,8 +170,8 @@ public class TantivyFullTextGlobalIndexWriter implements
GlobalIndexSingletonWri
out.flush();
}
- LOG.info("Tantivy index packed: {} documents", rowId);
- return new ResultEntry(fileName, rowId, indexOptions.serialize());
+ LOG.info("Tantivy index packed for {} rows", rowCount);
+ return new ResultEntry(fileName, rowCount, indexOptions.serialize());
}
private static void writeInt(PositionOutputStream out, int value) throws
IOException {
diff --git
a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/JavaPyTantivyE2ETest.java
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/JavaPyTantivyE2ETest.java
index fcb2ef7acb..f249227973 100644
---
a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/JavaPyTantivyE2ETest.java
+++
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/JavaPyTantivyE2ETest.java
@@ -25,7 +25,7 @@ import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
-import org.apache.paimon.globalindex.GlobalIndexSingletonWriter;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
@@ -199,8 +199,8 @@ public class JavaPyTantivyE2ETest {
indexOptions.set(TantivyFullTextIndexOptions.REMOVE_STOP_WORDS,
true);
}
- GlobalIndexSingletonWriter writer =
- (GlobalIndexSingletonWriter)
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
GlobalIndexBuilderUtils.createIndexWriter(
table,
TantivyFullTextGlobalIndexerFactory.IDENTIFIER,
@@ -208,8 +208,8 @@ public class JavaPyTantivyE2ETest {
indexOptions);
// Write the same text data to the index.
- for (String content : contents) {
- writer.write(BinaryString.fromString(content));
+ for (int i = 0; i < contents.size(); i++) {
+ writer.write(BinaryString.fromString(contents.get(i)), i);
}
List<ResultEntry> entries = writer.finish();
diff --git
a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
index de13f51645..5986ce70f3 100644
---
a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
+++
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
@@ -125,9 +125,9 @@ public class TantivyFullTextGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
TantivyFullTextGlobalIndexWriter writer = new
TantivyFullTextGlobalIndexWriter(fileWriter);
- writer.write(BinaryString.fromString("Apache Paimon is a streaming
data lake platform"));
- writer.write(BinaryString.fromString("Tantivy is a full-text search
engine in Rust"));
- writer.write(BinaryString.fromString("Paimon supports real-time data
ingestion"));
+ writer.write(BinaryString.fromString("Apache Paimon is a streaming
data lake platform"), 0);
+ writer.write(BinaryString.fromString("Tantivy is a full-text search
engine in Rust"), 1);
+ writer.write(BinaryString.fromString("Paimon supports real-time data
ingestion"), 2);
List<ResultEntry> results = writer.finish();
assertThat(results).hasSize(1);
@@ -171,7 +171,7 @@ public class TantivyFullTextGlobalIndexTest {
new TantivyFullTextIndexOptions(
TantivyFullTextGlobalIndexerFactory.removeTantivyPrefix(options)));
- writer.write(BinaryString.fromString("Apache Paimon supports Chinese
text"));
+ writer.write(BinaryString.fromString("Apache Paimon supports Chinese
text"), 0);
List<ResultEntry> results = writer.finish();
assertThat(results).hasSize(1);
@@ -194,8 +194,8 @@ public class TantivyFullTextGlobalIndexTest {
new TantivyFullTextIndexOptions(
TantivyFullTextGlobalIndexerFactory.removeTantivyPrefix(options)));
- writer.write(BinaryString.fromString("张华在百货公司当售货员"));
- writer.write(BinaryString.fromString("Apache Paimon supports full text
search"));
+ writer.write(BinaryString.fromString("张华在百货公司当售货员"), 0);
+ writer.write(BinaryString.fromString("Apache Paimon supports full text
search"), 1);
List<ResultEntry> results = writer.finish();
TantivyFullTextIndexOptions indexOptions =
@@ -222,8 +222,8 @@ public class TantivyFullTextGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
TantivyFullTextGlobalIndexWriter writer = new
TantivyFullTextGlobalIndexWriter(fileWriter);
- writer.write(BinaryString.fromString("Hello world"));
- writer.write(BinaryString.fromString("Foo bar baz"));
+ writer.write(BinaryString.fromString("Hello world"), 0);
+ writer.write(BinaryString.fromString("Foo bar baz"), 1);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -245,9 +245,9 @@ public class TantivyFullTextGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
TantivyFullTextGlobalIndexWriter writer = new
TantivyFullTextGlobalIndexWriter(fileWriter);
- writer.write(BinaryString.fromString("Paimon data lake"));
- writer.write(null); // row 1 is null, should be skipped
- writer.write(BinaryString.fromString("Paimon streaming"));
+ writer.write(BinaryString.fromString("Paimon data lake"), 0);
+ writer.write(null, 1); // row 1 is null, should be skipped
+ writer.write(BinaryString.fromString("Paimon streaming"), 2);
List<ResultEntry> results = writer.finish();
assertThat(results.get(0).rowCount()).isEqualTo(3);
@@ -291,7 +291,7 @@ public class TantivyFullTextGlobalIndexTest {
if (i % 10 == 0) {
text += " special_keyword";
}
- writer.write(BinaryString.fromString(text));
+ writer.write(BinaryString.fromString(text), i);
}
List<ResultEntry> results = writer.finish();
@@ -323,7 +323,7 @@ public class TantivyFullTextGlobalIndexTest {
TantivyFullTextGlobalIndexWriter writer = new
TantivyFullTextGlobalIndexWriter(fileWriter);
for (int i = 0; i < 20; i++) {
- writer.write(BinaryString.fromString("paimon document " + i));
+ writer.write(BinaryString.fromString("paimon document " + i), i);
}
List<ResultEntry> results = writer.finish();
@@ -346,8 +346,8 @@ public class TantivyFullTextGlobalIndexTest {
public void testPoolReuse() throws IOException {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
TantivyFullTextGlobalIndexWriter writer = new
TantivyFullTextGlobalIndexWriter(fileWriter);
- writer.write(BinaryString.fromString("Apache Paimon streaming lake"));
- writer.write(BinaryString.fromString("Tantivy full-text search"));
+ writer.write(BinaryString.fromString("Apache Paimon streaming lake"),
0);
+ writer.write(BinaryString.fromString("Tantivy full-text search"), 1);
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -379,7 +379,7 @@ public class TantivyFullTextGlobalIndexTest {
TantivyFullTextGlobalIndexWriter writer =
(TantivyFullTextGlobalIndexWriter)
indexer.createWriter(fileWriter);
- writer.write(BinaryString.fromString("test via indexer factory"));
+ writer.write(BinaryString.fromString("test via indexer factory"), 0);
List<ResultEntry> results = writer.finish();
assertThat(results).hasSize(1);
diff --git
a/paimon-vector/paimon-vector-index/src/main/java/org/apache/paimon/vector/index/VectorGlobalIndexWriter.java
b/paimon-vector/paimon-vector-index/src/main/java/org/apache/paimon/vector/index/VectorGlobalIndexWriter.java
index 1958200958..0554bb7862 100644
---
a/paimon-vector/paimon-vector-index/src/main/java/org/apache/paimon/vector/index/VectorGlobalIndexWriter.java
+++
b/paimon-vector/paimon-vector-index/src/main/java/org/apache/paimon/vector/index/VectorGlobalIndexWriter.java
@@ -21,7 +21,7 @@ package org.apache.paimon.vector.index;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalVector;
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;
import org.apache.paimon.index.vector.VectorIndexWriter;
@@ -48,13 +48,13 @@ import java.util.Map;
/**
* Vector global index writer using paimon-vector-index.
*
- * <p>Vectors are spilled to a temporary file on disk as they arrive via
{@link #write(Object)},
- * keeping Java heap usage constant (~8 MB buffer). During index build,
vectors are read back for
- * training and batch insertion.
+ * <p>Vectors are spilled to a temporary file on disk as they arrive via
{@link #write(Object,
+ * long)}, keeping Java heap usage constant (~8 MB buffer). During index
build, vectors are read
+ * back for training and batch insertion.
*
* <p><b>Thread safety:</b> This class is <b>not</b> thread-safe.
*/
-public class VectorGlobalIndexWriter implements GlobalIndexSingletonWriter,
Closeable {
+public class VectorGlobalIndexWriter implements GlobalIndexSingleColumnWriter,
Closeable {
private static final String FILE_NAME_PREFIX = "vector";
@@ -77,7 +77,7 @@ public class VectorGlobalIndexWriter implements
GlobalIndexSingletonWriter, Clos
private long count;
private boolean closed;
- private long logicalRowId;
+ private long rowCount;
public VectorGlobalIndexWriter(
GlobalIndexFileWriter fileWriter,
@@ -129,31 +129,31 @@ public class VectorGlobalIndexWriter implements
GlobalIndexSingletonWriter, Clos
}
@Override
- public void write(Object fieldData) {
+ public void write(Object fieldData, long relativeRowId) {
if (fieldData == null) {
- logicalRowId++;
+ rowCount++;
return;
}
- float[] src = materializeAndValidate(fieldData);
+ float[] src = materializeAndValidate(fieldData, relativeRowId);
if (writeBuf.remaining() < recordSizeInBytes) {
flushWriteBuffer();
}
- writeBuf.putLong(logicalRowId);
+ writeBuf.putLong(relativeRowId);
for (int i = 0; i < dim; i++) {
writeBuf.putFloat(src[i]);
}
- logicalRowId++;
+ rowCount++;
count++;
}
- private float[] materializeAndValidate(Object fieldData) {
+ private float[] materializeAndValidate(Object fieldData, long
relativeRowId) {
if (fieldData instanceof float[]) {
float[] vector = (float[]) fieldData;
checkDimension(vector.length);
for (int i = 0; i < dim; i++) {
- checkFinite(vector[i], i);
+ checkFinite(vector[i], relativeRowId, i);
}
return vector;
} else if (fieldData instanceof InternalVector) {
@@ -161,7 +161,7 @@ public class VectorGlobalIndexWriter implements
GlobalIndexSingletonWriter, Clos
checkDimension(vector.size());
for (int i = 0; i < dim; i++) {
float v = vector.getFloat(i);
- checkFinite(v, i);
+ checkFinite(v, relativeRowId, i);
vectorBuf[i] = v;
}
return vectorBuf;
@@ -173,7 +173,7 @@ public class VectorGlobalIndexWriter implements
GlobalIndexSingletonWriter, Clos
throw new IllegalArgumentException("Vector element at
index " + i + " is null");
}
float v = array.getFloat(i);
- checkFinite(v, i);
+ checkFinite(v, relativeRowId, i);
vectorBuf[i] = v;
}
return vectorBuf;
@@ -262,7 +262,7 @@ public class VectorGlobalIndexWriter implements
GlobalIndexSingletonWriter, Clos
System.currentTimeMillis() - buildStart);
VectorIndexMeta meta = new VectorIndexMeta();
- return new ResultEntry(fileName, logicalRowId, meta.serialize());
+ return new ResultEntry(fileName, rowCount, meta.serialize());
}
}
@@ -364,12 +364,12 @@ public class VectorGlobalIndexWriter implements
GlobalIndexSingletonWriter, Clos
}
}
- private void checkFinite(float value, int elementIndex) {
+ private void checkFinite(float value, long relativeRowId, int
elementIndex) {
if (!Float.isFinite(value)) {
throw new IllegalArgumentException(
String.format(
"Vector element at rowId=%d, index=%d is %s",
- logicalRowId, elementIndex,
Float.toString(value)));
+ relativeRowId, elementIndex,
Float.toString(value)));
}
}
diff --git
a/paimon-vector/paimon-vector-index/src/test/java/org/apache/paimon/vector/index/VectorGlobalIndexTest.java
b/paimon-vector/paimon-vector-index/src/test/java/org/apache/paimon/vector/index/VectorGlobalIndexTest.java
index b10c843ef2..e82664f806 100644
---
a/paimon-vector/paimon-vector-index/src/test/java/org/apache/paimon/vector/index/VectorGlobalIndexTest.java
+++
b/paimon-vector/paimon-vector-index/src/test/java/org/apache/paimon/vector/index/VectorGlobalIndexTest.java
@@ -45,7 +45,6 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -106,7 +105,7 @@ public class VectorGlobalIndexTest {
VectorGlobalIndexWriter writer = createIvfPqWriter(fileWriter,
vectorType, options);
float[] wrongDimVector = new float[32];
- assertThatThrownBy(() -> writer.write(wrongDimVector))
+ assertThatThrownBy(() -> writer.write(wrongDimVector, 0))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("dimension mismatch");
}
@@ -130,7 +129,7 @@ public class VectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
VectorGlobalIndexWriter writer = createIvfPqWriter(fileWriter,
vectorType, options);
- assertThatThrownBy(() -> writer.write(new float[] {1.0f, Float.NaN}))
+ assertThatThrownBy(() -> writer.write(new float[] {1.0f, Float.NaN},
0))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("rowId=0")
.hasMessageContaining("index=1")
@@ -144,8 +143,8 @@ public class VectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
VectorGlobalIndexWriter writer = createIvfPqWriter(fileWriter,
vectorType, options);
- writer.write(null); // row 0 - null, advances logicalRowId
- assertThatThrownBy(() -> writer.write(new float[]
{Float.POSITIVE_INFINITY, 0.0f}))
+ writer.write(null, 0); // row 0 - null
+ assertThatThrownBy(() -> writer.write(new float[]
{Float.POSITIVE_INFINITY, 0.0f}, 1))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("rowId=1")
.hasMessageContaining("index=0")
@@ -159,9 +158,9 @@ public class VectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
VectorGlobalIndexWriter writer = createIvfPqWriter(fileWriter,
vectorType, options);
- writer.write(null);
- writer.write(null);
- writer.write(null);
+ writer.write(null, 0);
+ writer.write(null, 1);
+ writer.write(null, 2);
List<ResultEntry> results = writer.finish();
assertThat(results).isEmpty();
@@ -223,7 +222,9 @@ public class VectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
VectorGlobalIndexWriter writer = createIvfPqWriter(fileWriter,
vectorType, options);
- Arrays.stream(vectors).forEach(writer::write);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
+ }
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -260,7 +261,9 @@ public class VectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
VectorGlobalIndexWriter writer = createIvfPqWriter(fileWriter,
vectorType, options);
- Arrays.stream(vectors).forEach(writer::write);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
+ }
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
@@ -300,12 +303,12 @@ public class VectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
VectorGlobalIndexWriter writer = createIvfPqWriter(fileWriter,
vectorType, options);
- writer.write(vectors[0]); // row 0
- writer.write(null); // row 1 - null
- writer.write(vectors[1]); // row 2
- writer.write(null); // row 3 - null
- writer.write(null); // row 4 - null
- writer.write(vectors[2]); // row 5
+ writer.write(vectors[0], 0); // row 0
+ writer.write(null, 1); // row 1 - null
+ writer.write(vectors[1], 2); // row 2
+ writer.write(null, 3); // row 3 - null
+ writer.write(null, 4); // row 4 - null
+ writer.write(vectors[2], 5); // row 5
List<ResultEntry> results = writer.finish();
assertThat(results).hasSize(1);
@@ -352,7 +355,9 @@ public class VectorGlobalIndexTest {
GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
VectorGlobalIndexWriter writer = (VectorGlobalIndexWriter)
indexer.createWriter(fileWriter);
- Arrays.stream(vectors).forEach(writer::write);
+ for (int i = 0; i < vectors.length; i++) {
+ writer.write(vectors[i], i);
+ }
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
