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 088d4880ff [core] Add public binary serialization for
BucketVectorSearchSplit (#9386)
088d4880ff is described below
commit 088d4880ffa3e46566dd6c94d46a2bd1a03f87b0
Author: Junrui Lee <[email protected]>
AuthorDate: Wed Aug 26 17:29:30 2026 +0800
[core] Add public binary serialization for BucketVectorSearchSplit (#9386)
---
.../table/source/BucketVectorSearchSplit.java | 91 +++++--
.../table/source/BucketVectorSearchSplitTest.java | 272 +++++++++++++++++++++
2 files changed, 345 insertions(+), 18 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java
index 31c2f7b0cb..75ca1f58a1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/BucketVectorSearchSplit.java
@@ -20,7 +20,9 @@ package org.apache.paimon.table.source;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexFileMetaSerializer;
+import org.apache.paimon.io.DataInputView;
import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputView;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
import org.apache.paimon.utils.Range;
@@ -33,6 +35,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.TreeMap;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -40,10 +43,12 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
public class BucketVectorSearchSplit extends VectorSearchSplit {
private static final long serialVersionUID = 1L;
+
+ private static final long MAGIC = 0x504B5653504C4954L;
private static final int VERSION = 1;
private DataSplit dataSplit;
- private transient List<IndexFileMeta> payloadFiles;
+ private List<IndexFileMeta> payloadFiles;
private Map<String, List<Range>> rowRangesByFile;
public BucketVectorSearchSplit(DataSplit dataSplit, List<IndexFileMeta>
payloadFiles) {
@@ -84,32 +89,70 @@ public class BucketVectorSearchSplit extends
VectorSearchSplit {
return rowRangesByFile;
}
- private void writeObject(ObjectOutputStream out) throws IOException {
- out.defaultWriteObject();
+ /**
+ * Serialize to the byte form a reader outside the JVM consumes, following
{@code IndexedSplit}:
+ * magic and version, then the nested {@link DataSplit}, then this split's
own state.
+ *
+ * <p>Row-range entries are written sorted by file name, so two splits
that compare equal
+ * serialize to the same bytes even though the map's iteration order is
its construction order.
+ *
+ * <p>{@link #VERSION} pins this envelope and the layout of what it nests,
not the nested bytes
+ * themselves: {@link DataSplit#serialize} carries its own version, while
{@link
+ * IndexFileMetaSerializer} carries none, so a change to {@code
IndexFileMeta.SCHEMA} has to
+ * bump this version too.
+ */
+ public void serialize(DataOutputView out) throws IOException {
+ out.writeLong(MAGIC);
out.writeInt(VERSION);
- out.writeInt(payloadFiles.size());
- IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
- for (IndexFileMeta payloadFile : payloadFiles) {
- serializer.serialize(payloadFile, new
DataOutputViewStreamWrapper(out));
+ dataSplit.serialize(out);
+ new IndexFileMetaSerializer().serializeList(payloadFiles, out);
+ out.writeInt(rowRangesByFile.size());
+ for (Map.Entry<String, List<Range>> entry : new
TreeMap<>(rowRangesByFile).entrySet()) {
+ out.writeUTF(entry.getKey());
+ out.writeInt(entry.getValue().size());
+ for (Range range : entry.getValue()) {
+ out.writeLong(range.from);
+ out.writeLong(range.to);
+ }
}
}
- private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
- in.defaultReadObject();
+ /** Reverse of {@link #serialize(DataOutputView)}. */
+ public static BucketVectorSearchSplit deserialize(DataInputView in) throws
IOException {
+ long magic = in.readLong();
+ if (magic != MAGIC) {
+ throw new IOException("Corrupted BucketVectorSearchSplit: wrong
magic number " + magic);
+ }
int version = in.readInt();
if (version != VERSION) {
throw new IOException("Unsupported BucketVectorSearchSplit
version: " + version);
}
- int payloadFileCount = in.readInt();
- if (payloadFileCount < 0) {
- throw new IOException("Negative primary-key vector payload file
count.");
+ DataSplit dataSplit = DataSplit.deserialize(in);
+ List<IndexFileMeta> payloadFiles = new
IndexFileMetaSerializer().deserializeList(in);
+ int rangeFileCount = in.readInt();
+ Map<String, List<Range>> rowRangesByFile = new LinkedHashMap<>();
+ for (int i = 0; i < rangeFileCount; i++) {
+ String fileName = in.readUTF();
+ int rangeCount = in.readInt();
+ List<Range> ranges = new ArrayList<>(rangeCount);
+ for (int j = 0; j < rangeCount; j++) {
+ ranges.add(new Range(in.readLong(), in.readLong()));
+ }
+ rowRangesByFile.put(fileName, ranges);
}
- List<IndexFileMeta> payloadFiles = new ArrayList<>(payloadFileCount);
- IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
- for (int i = 0; i < payloadFileCount; i++) {
- payloadFiles.add(serializer.deserialize(new
DataInputViewStreamWrapper(in)));
- }
- this.payloadFiles = Collections.unmodifiableList(payloadFiles);
+ return new BucketVectorSearchSplit(dataSplit, payloadFiles,
rowRangesByFile);
+ }
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ serialize(new DataOutputViewStreamWrapper(out));
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
+ BucketVectorSearchSplit other = deserialize(new
DataInputViewStreamWrapper(in));
+
+ this.dataSplit = other.dataSplit;
+ this.payloadFiles = other.payloadFiles;
+ this.rowRangesByFile = other.rowRangesByFile;
}
@Override
@@ -130,4 +173,16 @@ public class BucketVectorSearchSplit extends
VectorSearchSplit {
public int hashCode() {
return Objects.hash(dataSplit, payloadFiles, rowRangesByFile);
}
+
+ @Override
+ public String toString() {
+ return "BucketVectorSearchSplit{"
+ + "dataSplit="
+ + dataSplit
+ + ", payloadFiles="
+ + payloadFiles
+ + ", rowRangesByFile="
+ + rowRangesByFile
+ + '}';
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/BucketVectorSearchSplitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/BucketVectorSearchSplitTest.java
new file mode 100644
index 0000000000..c2f5c78261
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/BucketVectorSearchSplitTest.java
@@ -0,0 +1,272 @@
+/*
+ * 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.table.source;
+
+import org.apache.paimon.data.BinaryArray;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.index.DeletionVectorMeta;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataInputDeserializer;
+import org.apache.paimon.io.DataOutputSerializer;
+import org.apache.paimon.io.PojoDataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.utils.InstantiationUtil;
+import org.apache.paimon.utils.Range;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests the byte form of {@link BucketVectorSearchSplit}, which a reader
outside the JVM reads. */
+class BucketVectorSearchSplitTest {
+
+ @Test
+ void testRoundTrip() throws Exception {
+ BucketVectorSearchSplit expected = split();
+
+ // Compared whole rather than field by field: only that catches a
field the format forgets
+ // to carry.
+ BucketVectorSearchSplit actual = deserialize(serialize(expected));
+ assertThat(actual).isEqualTo(expected);
+ assertPayload(actual.payloadFiles().get(0));
+ }
+
+ /** Java engines ship this split through object serialization, which uses
the same form. */
+ @Test
+ void testJavaSerializationRoundTrip() throws Exception {
+ BucketVectorSearchSplit expected = split();
+ byte[] bytes = InstantiationUtil.serializeObject(expected);
+ BucketVectorSearchSplit actual =
+ InstantiationUtil.deserializeObject(
+ bytes, BucketVectorSearchSplit.class.getClassLoader());
+ assertThat(actual).isEqualTo(expected);
+ assertPayload(actual.payloadFiles().get(0));
+ }
+
+ /** Payloads go through {@code IndexFileMetaSerializer}, so every field it
carries survives. */
+ @Test
+ void testRoundTripKeepsDeletionVectorRanges() throws Exception {
+ LinkedHashMap<String, DeletionVectorMeta> dvRanges = new
LinkedHashMap<>();
+ dvRanges.put("data-1.orc", new DeletionVectorMeta("data-1.orc", 0, 8,
2L));
+ BucketVectorSearchSplit expected = withDvRanges(dvRanges);
+
+ assertThat(deserialize(serialize(expected))).isEqualTo(expected);
+ assertThat(
+
InstantiationUtil.<BucketVectorSearchSplit>deserializeObject(
+ InstantiationUtil.serializeObject(expected),
+
BucketVectorSearchSplit.class.getClassLoader()))
+ .isEqualTo(expected);
+ }
+
+ /** Two splits that compare equal have to serialize to the same bytes. */
+ @Test
+ void testSerializationIsCanonical() throws Exception {
+ DataSplit dataSplit =
+ dataSplit(Arrays.asList(dataFile("data-1.orc"),
dataFile("data-2.orc")));
+ List<Range> ranges = Collections.singletonList(new Range(0, 1));
+ Map<String, List<Range>> ascending = new LinkedHashMap<>();
+ ascending.put("data-1.orc", ranges);
+ ascending.put("data-2.orc", ranges);
+ Map<String, List<Range>> descending = new LinkedHashMap<>();
+ descending.put("data-2.orc", ranges);
+ descending.put("data-1.orc", ranges);
+
+ List<IndexFileMeta> payloadFiles = split().payloadFiles();
+ BucketVectorSearchSplit first =
+ new BucketVectorSearchSplit(dataSplit, payloadFiles,
ascending);
+ BucketVectorSearchSplit second =
+ new BucketVectorSearchSplit(dataSplit, payloadFiles,
descending);
+ assertThat(first).isEqualTo(second);
+ assertThat(serialize(first)).isEqualTo(serialize(second));
+ }
+
+ /** A bucket with no vector payload still plans a split, so both
collections can be empty. */
+ @Test
+ void testRoundTripWithoutPayloadsOrRowRanges() throws Exception {
+ BucketVectorSearchSplit expected =
+ new BucketVectorSearchSplit(
+ split().dataSplit(), Collections.emptyList(),
Collections.emptyMap());
+ assertThat(deserialize(serialize(expected))).isEqualTo(expected);
+ }
+
+ /** Names carry more than ASCII, so they have to survive unchanged. */
+ @Test
+ void testRoundTripNonAsciiNames() throws Exception {
+ String indexType = "向量-ivf🚀";
+ BucketVectorSearchSplit actual =
deserialize(serialize(withPayloadIndexType(indexType)));
+
assertThat(actual.payloadFiles().get(0).indexType()).isEqualTo(indexType);
+ }
+
+ @Test
+ void testRejectsForeignBytes() throws Exception {
+ byte[] wrongMagic = serialize(split());
+ ByteBuffer.wrap(wrongMagic).putLong(0, 1);
+ assertThatThrownBy(() -> deserialize(wrongMagic))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("wrong magic number");
+
+ byte[] unsupportedVersion = serialize(split());
+ ByteBuffer.wrap(unsupportedVersion).putInt(Long.BYTES, 2);
+ assertThatThrownBy(() -> deserialize(unsupportedVersion))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported BucketVectorSearchSplit
version: 2");
+ }
+
+ private static byte[] serialize(BucketVectorSearchSplit split) throws
IOException {
+ DataOutputSerializer out = new DataOutputSerializer(256);
+ split.serialize(out);
+ return out.getCopyOfBuffer();
+ }
+
+ private static BucketVectorSearchSplit deserialize(byte[] bytes) throws
IOException {
+ return BucketVectorSearchSplit.deserialize(new
DataInputDeserializer(bytes));
+ }
+
+ /**
+ * Non-empty rows and stats on purpose: the nested BinaryRow layout is the
part of this message
+ * a reader is most likely to get wrong, and an empty row exercises none
of it.
+ */
+ private static DataFileMeta dataFile(String fileName) {
+ SimpleStats stats =
+ new SimpleStats(
+ BinaryRow.singleColumn("min_value"),
+ BinaryRow.singleColumn("max_value"),
+ BinaryArray.fromLongArray(new Long[] {0L}));
+ return new PojoDataFileMeta(
+ fileName,
+ 1_234,
+ 6,
+ BinaryRow.singleColumn("min_key"),
+ BinaryRow.singleColumn("max_key"),
+ stats,
+ stats,
+ 3,
+ 9,
+ 7,
+ 1,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(1_700_000_000_000L),
+ 0L,
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ 40L,
+ Arrays.asList("k", "v"),
+ new long[] {3, 9});
+ }
+
+ private static DataSplit dataSplit(List<DataFileMeta> dataFiles) {
+ return DataSplit.builder()
+ .withSnapshot(11)
+ .withPartition(BinaryRow.singleColumn(20250826))
+ .withBucket(2)
+ .withBucketPath("bucket-2")
+ .withTotalBuckets(8)
+ .withDataFiles(dataFiles)
+ .build();
+ }
+
+ private static BucketVectorSearchSplit split() {
+ DataSplit dataSplit =
dataSplit(Collections.singletonList(dataFile("data-1.orc")));
+
+ byte[] sourceMeta =
+ new PrimaryKeyIndexSourceMeta(
+ 1,
+ Collections.singletonList(
+ new
PrimaryKeyIndexSourceFile("data-1.orc", 6)))
+ .serialize();
+ IndexFileMeta payload =
+ new IndexFileMeta(
+ "ivf-pq",
+ "ann-0.idx",
+ 5_000_000_000L,
+ 6,
+ new GlobalIndexMeta(
+ 40, 45, 7, new int[] {3, 5}, new byte[] {1, 2,
3}, sourceMeta),
+ "s3://vector-bucket/ann-0.idx");
+
+ Map<String, List<Range>> ranges = new LinkedHashMap<>();
+ ranges.put("data-1.orc", Arrays.asList(new Range(0, 1), new Range(4,
5)));
+ return new BucketVectorSearchSplit(dataSplit,
Collections.singletonList(payload), ranges);
+ }
+
+ private static BucketVectorSearchSplit withPayloadIndexType(String
indexType) {
+ BucketVectorSearchSplit split = split();
+ IndexFileMeta payload = split.payloadFiles().get(0);
+ IndexFileMeta renamed =
+ new IndexFileMeta(
+ indexType,
+ payload.fileName(),
+ payload.fileSize(),
+ payload.rowCount(),
+ payload.globalIndexMeta(),
+ payload.externalPath());
+ return new BucketVectorSearchSplit(
+ split.dataSplit(), Collections.singletonList(renamed),
split.rowRangesByFile());
+ }
+
+ private static BucketVectorSearchSplit withDvRanges(
+ LinkedHashMap<String, DeletionVectorMeta> dvRanges) {
+ BucketVectorSearchSplit base = split();
+ IndexFileMeta payload = base.payloadFiles().get(0);
+ IndexFileMeta withDvRanges =
+ new IndexFileMeta(
+ payload.indexType(),
+ payload.fileName(),
+ payload.fileSize(),
+ payload.rowCount(),
+ dvRanges,
+ payload.externalPath(),
+ payload.globalIndexMeta());
+ return new BucketVectorSearchSplit(
+ base.dataSplit(), Collections.singletonList(withDvRanges),
base.rowRangesByFile());
+ }
+
+ private static void assertPayload(IndexFileMeta payload) {
+ assertThat(payload.indexType()).isEqualTo("ivf-pq");
+ assertThat(payload.fileName()).isEqualTo("ann-0.idx");
+ assertThat(payload.fileSize()).isEqualTo(5_000_000_000L);
+ assertThat(payload.rowCount()).isEqualTo(6);
+
assertThat(payload.externalPath()).isEqualTo("s3://vector-bucket/ann-0.idx");
+ assertThat(payload.globalIndexMeta()).isNotNull();
+ assertThat(payload.globalIndexMeta().rowRangeStart()).isEqualTo(40);
+ assertThat(payload.globalIndexMeta().rowRangeEnd()).isEqualTo(45);
+ assertThat(payload.globalIndexMeta().indexFieldId()).isEqualTo(7);
+
assertThat(payload.globalIndexMeta().extraFieldIds()).containsExactly(3, 5);
+ assertThat(payload.globalIndexMeta().indexMeta()).containsExactly(1,
2, 3);
+
assertThat(PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFile().fileName())
+ .isEqualTo("data-1.orc");
+ }
+}
