This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new eed779e [core] Parse primary-key vector index source metadata
(_SOURCE_META) (#515)
eed779e is described below
commit eed779e16cd0bf12f81f92b9a115d6c52b86e9a1
Author: Junrui Lee <[email protected]>
AuthorDate: Mon Jul 13 13:04:34 2026 +0800
[core] Parse primary-key vector index source metadata (_SOURCE_META) (#515)
---
.../src/spec/avro/index_manifest_entry_decode.rs | 25 +-
crates/paimon/src/spec/index_file_meta.rs | 3 +
crates/paimon/src/spec/index_manifest.rs | 145 ++++++-
crates/paimon/src/spec/mod.rs | 3 +
crates/paimon/src/spec/pk_vector_source.rs | 452 +++++++++++++++++++++
.../src/table/btree_global_index_build_builder.rs | 2 +
.../paimon/src/table/global_index_drop_builder.rs | 1 +
crates/paimon/src/table/global_index_scanner.rs | 1 +
.../paimon/src/table/lumina_index_build_builder.rs | 2 +
crates/paimon/src/table/table_commit.rs | 1 +
crates/paimon/src/table/vector_search_builder.rs | 1 +
.../paimon/src/table/vindex_index_build_builder.rs | 3 +
12 files changed, 635 insertions(+), 4 deletions(-)
diff --git a/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
b/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
index b35453f..986b27b 100644
--- a/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
+++ b/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs
@@ -18,7 +18,8 @@
use super::cursor::AvroCursor;
use super::decode::{neg_count_to_usize, AvroRecordDecode};
use super::decode_helpers::{
- normalize_partition, read_bytes_field, read_int_field, read_long_field,
read_string_field,
+ extract_record_schema, normalize_partition, read_bytes_field,
read_int_field, read_long_field,
+ read_string_field,
};
use super::schema::{skip_nullable_field, WriterSchema};
use crate::spec::index_manifest::IndexManifestEntry;
@@ -65,7 +66,8 @@ impl AvroRecordDecode for IndexManifestEntry {
deletion_vectors_ranges =
decode_nullable_dv_ranges(cursor, field.nullable)?;
}
"_GLOBAL_INDEX" => {
- global_index_meta = decode_nullable_global_index(cursor,
field.nullable)?;
+ global_index_meta =
+ decode_nullable_global_index(cursor, field.nullable,
&field.schema)?;
}
_ => skip_nullable_field(cursor, &field.schema,
field.nullable)?,
}
@@ -145,6 +147,7 @@ fn decode_nullable_dv_ranges(
fn decode_nullable_global_index(
cursor: &mut AvroCursor,
nullable: bool,
+ schema: &super::schema::FieldSchema,
) -> crate::Result<Option<GlobalIndexMeta>> {
if nullable {
let idx = cursor.read_union_index()?;
@@ -192,11 +195,29 @@ fn decode_nullable_global_index(
}
};
+ // _SOURCE_META: nullable bytes — only present in >= #8549 writer schemas.
+ // Guard on the writer's nested field list so a legacy 5-field
_GLOBAL_INDEX
+ // record does not misalign the cursor into the next record.
+ let has_source_meta = extract_record_schema(schema)
+ .map(|s| s.fields.iter().any(|f| f.name == "_SOURCE_META"))
+ .unwrap_or(false);
+ let source_meta = if has_source_meta {
+ let u_idx = cursor.read_union_index()?;
+ if u_idx == 0 {
+ None
+ } else {
+ Some(cursor.read_bytes()?.to_vec())
+ }
+ } else {
+ None
+ };
+
Ok(Some(GlobalIndexMeta {
row_range_start,
row_range_end,
index_field_id,
extra_field_ids,
index_meta,
+ source_meta,
}))
}
diff --git a/crates/paimon/src/spec/index_file_meta.rs
b/crates/paimon/src/spec/index_file_meta.rs
index e985bb0..ff1e63c 100644
--- a/crates/paimon/src/spec/index_file_meta.rs
+++ b/crates/paimon/src/spec/index_file_meta.rs
@@ -46,6 +46,9 @@ pub struct GlobalIndexMeta {
#[serde(default, rename = "_INDEX_META", with = "serde_bytes")]
pub index_meta: Option<Vec<u8>>,
+
+ #[serde(default, rename = "_SOURCE_META", with = "serde_bytes")]
+ pub source_meta: Option<Vec<u8>>,
}
/// Metadata of index file.
diff --git a/crates/paimon/src/spec/index_manifest.rs
b/crates/paimon/src/spec/index_manifest.rs
index b8ee2b5..a0430b2 100644
--- a/crates/paimon/src/spec/index_manifest.rs
+++ b/crates/paimon/src/spec/index_manifest.rs
@@ -71,7 +71,8 @@ pub const INDEX_MANIFEST_ENTRY_SCHEMA: &str = r#"{
{"name": "_ROW_RANGE_END", "type": "long"},
{"name": "_INDEX_FIELD_ID", "type": "int"},
{"name": "_EXTRA_FIELD_IDS", "type": ["null", {"type":
"array", "items": "int"}], "default": null},
- {"name": "_INDEX_META", "type": ["null", "bytes"],
"default": null}
+ {"name": "_INDEX_META", "type": ["null", "bytes"],
"default": null},
+ {"name": "_SOURCE_META", "type": ["null", "bytes"],
"default": null}
]
}]
}
@@ -173,7 +174,7 @@ mod tests {
use indexmap::IndexMap;
use super::*;
- use crate::spec::DeletionVectorMeta;
+ use crate::spec::{DeletionVectorMeta, GlobalIndexMeta};
#[test]
fn test_read_index_manifest_file() {
@@ -276,4 +277,144 @@ mod tests {
};
assert_eq!(sample, decoded);
}
+
+ fn global_index_entry(source_meta: Option<Vec<u8>>) -> IndexManifestEntry {
+ IndexManifestEntry {
+ version: 1,
+ kind: FileKind::Add,
+ partition: vec![0, 1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6],
+ bucket: 0,
+ index_file: IndexFileMeta {
+ index_type: "GLOBAL_INDEX".into(),
+ file_name: "gi-1".into(),
+ file_size: 42,
+ row_count: 7,
+ deletion_vectors_ranges: None,
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start: 10,
+ row_range_end: 20,
+ index_field_id: 3,
+ extra_field_ids: Some(vec![4, 5]),
+ index_meta: Some(vec![9, 8, 7]),
+ source_meta,
+ }),
+ },
+ }
+ }
+
+ #[test]
+ fn source_meta_round_trips_through_index_manifest() {
+ // New-format writer schema carries _SOURCE_META; a Some(..) value
must round-trip.
+ let entry = global_index_entry(Some(vec![1, 2, 3]));
+ let bytes = crate::spec::to_avro_bytes_with_compression(
+ INDEX_MANIFEST_ENTRY_SCHEMA,
+ std::slice::from_ref(&entry),
+ crate::spec::DEFAULT_AVRO_COMPRESSION,
+ )
+ .unwrap();
+ let decoded = IndexManifest::read_from_bytes(&bytes).unwrap();
+ assert_eq!(decoded[0], entry);
+ assert_eq!(
+ decoded[0]
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .source_meta,
+ Some(vec![1, 2, 3])
+ );
+
+ // A None source_meta must also round-trip as None.
+ let entry_none = global_index_entry(None);
+ let bytes_none = crate::spec::to_avro_bytes_with_compression(
+ INDEX_MANIFEST_ENTRY_SCHEMA,
+ std::slice::from_ref(&entry_none),
+ crate::spec::DEFAULT_AVRO_COMPRESSION,
+ )
+ .unwrap();
+ let decoded_none =
IndexManifest::read_from_bytes(&bytes_none).unwrap();
+ assert_eq!(decoded_none[0], entry_none);
+ assert_eq!(
+ decoded_none[0]
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .source_meta,
+ None
+ );
+ }
+
+ #[test]
+ fn legacy_five_field_global_index_decodes_without_source_meta() {
+ // 5-field _GLOBAL_INDEX schema (pre-#8549): no _SOURCE_META.
Identical to
+ // INDEX_MANIFEST_ENTRY_SCHEMA with the trailing _SOURCE_META line
removed.
+ const LEGACY_SCHEMA: &str = r#"{
+ "type": "record",
+ "name": "org.apache.paimon.avro.generated.record",
+ "fields": [
+ {"name": "_VERSION", "type": "int"},
+ {"name": "_KIND", "type": "int"},
+ {"name": "_PARTITION", "type": "bytes"},
+ {"name": "_BUCKET", "type": "int"},
+ {"name": "_INDEX_TYPE", "type": "string"},
+ {"name": "_FILE_NAME", "type": "string"},
+ {"name": "_FILE_SIZE", "type": "long"},
+ {"name": "_ROW_COUNT", "type": "long"},
+ {
+ "default": null,
+ "name": "_DELETIONS_VECTORS_RANGES",
+ "type": ["null", {
+ "type": "array",
+ "items": ["null", {
+ "type": "record",
+ "name":
"org.apache.paimon.avro.generated.record__DELETIONS_VECTORS_RANGES",
+ "fields": [
+ {"name": "f0", "type": "string"},
+ {"name": "f1", "type": "int"},
+ {"name": "f2", "type": "int"},
+ {"name": "_CARDINALITY", "type": ["null", "long"],
"default": null}
+ ]
+ }]
+ }]
+ },
+ {
+ "default": null,
+ "name": "_GLOBAL_INDEX",
+ "type": ["null", {
+ "type": "record",
+ "name":
"org.apache.paimon.avro.generated.record__GLOBAL_INDEX",
+ "fields": [
+ {"name": "_ROW_RANGE_START", "type": "long"},
+ {"name": "_ROW_RANGE_END", "type": "long"},
+ {"name": "_INDEX_FIELD_ID", "type": "int"},
+ {"name": "_EXTRA_FIELD_IDS", "type": ["null", {"type":
"array", "items": "int"}], "default": null},
+ {"name": "_INDEX_META", "type": ["null", "bytes"],
"default": null}
+ ]
+ }]
+ }
+ ]
+}"#;
+
+ // Written by a pre-#8549 writer (5-field record, source_meta absent).
+ let entry = global_index_entry(None);
+ let bytes = crate::spec::to_avro_bytes_with_compression(
+ LEGACY_SCHEMA,
+ std::slice::from_ref(&entry),
+ crate::spec::DEFAULT_AVRO_COMPRESSION,
+ )
+ .unwrap();
+ // Decoding with the current 6-field reader must not misalign the
stream.
+ let decoded = IndexManifest::read_from_bytes(&bytes).unwrap();
+ assert_eq!(decoded[0], entry);
+ assert_eq!(
+ decoded[0]
+ .index_file
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .source_meta,
+ None
+ );
+ }
}
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 9e8b07b..4614e89 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -64,6 +64,9 @@ pub use manifest_file_meta::*;
mod index_file_meta;
pub use index_file_meta::*;
+mod pk_vector_source;
+pub use pk_vector_source::*;
+
mod index_manifest;
pub use index_manifest::{IndexManifest, IndexManifestEntry};
mod manifest;
diff --git a/crates/paimon/src/spec/pk_vector_source.rs
b/crates/paimon/src/spec/pk_vector_source.rs
new file mode 100644
index 0000000..2685f45
--- /dev/null
+++ b/crates/paimon/src/spec/pk_vector_source.rs
@@ -0,0 +1,452 @@
+// 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.
+
+//! Primary-key vector (bucket-local ANN) source metadata.
+//!
+//! Parses the `_SOURCE_META` blob embedded in
[`crate::spec::GlobalIndexMeta`],
+//! written by Java Paimon's `PkVectorSourceMeta` (apache/paimon#8549). The
blob
+//! lists the ordered source data files backing an ANN segment; a segment's
+//! ordinals concatenate those files in order, so an ordinal maps to a
+//! `(data file, physical row position)` pair.
+//!
+//! Two distinct encodings are involved and must not be conflated: Avro
extracts
+//! the opaque `_SOURCE_META` bytes (handled in the Avro decoder), and the
bytes
+//! *inside* use Java `DataOutput` (big-endian ints/longs, modified-UTF-8 via
+//! `writeUTF`) — parsed here, independent of Avro.
+
+use crate::spec::GlobalIndexMeta;
+
+fn data_invalid(message: impl Into<String>) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+/// The `_SOURCE_META` frame version written by Java `PkVectorSourceMeta`.
+const SOURCE_META_VERSION: i32 = 1;
+
+/// One source data file captured when a PK-vector ANN segment was built.
+///
+/// Mirrors Java `org.apache.paimon.index.pkvector.PkVectorSourceFile`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PkVectorSourceFile {
+ file_name: String,
+ row_count: i64,
+}
+
+impl PkVectorSourceFile {
+ pub fn new(file_name: String, row_count: i64) -> crate::Result<Self> {
+ if row_count < 0 {
+ return Err(data_invalid(format!(
+ "source file row count must not be negative: {row_count}"
+ )));
+ }
+ Ok(Self {
+ file_name,
+ row_count,
+ })
+ }
+
+ pub fn file_name(&self) -> &str {
+ &self.file_name
+ }
+
+ pub fn row_count(&self) -> i64 {
+ self.row_count
+ }
+}
+
+/// Ordered source data files for a primary-key vector index payload.
+///
+/// Mirrors Java `org.apache.paimon.index.pkvector.PkVectorSourceMeta`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PkVectorSourceMeta {
+ source_files: Vec<PkVectorSourceFile>,
+}
+
+impl PkVectorSourceMeta {
+ pub fn new(source_files: Vec<PkVectorSourceFile>) -> crate::Result<Self> {
+ if source_files.is_empty() {
+ return Err(data_invalid("a vector index must reference source
files"));
+ }
+ Ok(Self { source_files })
+ }
+
+ pub fn source_files(&self) -> &[PkVectorSourceFile] {
+ &self.source_files
+ }
+
+ /// Build from the `_SOURCE_META` bytes carried on a [`GlobalIndexMeta`].
+ /// Errors if the metadata carries no source blob.
+ pub fn from_global_index_meta(meta: &GlobalIndexMeta) ->
crate::Result<Self> {
+ let bytes = meta
+ .source_meta
+ .as_deref()
+ .ok_or_else(|| data_invalid("global index meta has no vector
source metadata"))?;
+ Self::deserialize(bytes)
+ }
+
+ /// Map an ANN segment ordinal to `(data file name, physical row
position)`.
+ ///
+ /// Ordinals concatenate source files in stored order: file `i` owns
+ /// `[sum(rows[..i]), sum(rows[..=i]))`. The returned position is local to
+ /// its file.
+ pub fn resolve(&self, ordinal: i64) -> crate::Result<(String, i64)> {
+ if ordinal < 0 {
+ return Err(data_invalid(format!(
+ "vector ordinal must not be negative: {ordinal}"
+ )));
+ }
+ let mut cumulative: i64 = 0;
+ for file in &self.source_files {
+ let next = cumulative
+ .checked_add(file.row_count)
+ .ok_or_else(|| data_invalid("vector source row counts overflow
i64"))?;
+ if ordinal < next {
+ return Ok((file.file_name.clone(), ordinal - cumulative));
+ }
+ cumulative = next;
+ }
+ Err(data_invalid(format!(
+ "vector ordinal {ordinal} is out of range (total rows
{cumulative})"
+ )))
+ }
+
+ /// Parse a Java `PkVectorSourceMeta`-serialized `_SOURCE_META` blob.
+ pub fn deserialize(bytes: &[u8]) -> crate::Result<Self> {
+ let mut cursor = DataInputCursor::new(bytes);
+ let version = cursor.read_i32_be()?;
+ if version != SOURCE_META_VERSION {
+ return Err(data_invalid(format!(
+ "unsupported vector source version: {version}"
+ )));
+ }
+ let count = cursor.read_i32_be()?;
+ if count <= 0 {
+ return Err(data_invalid("a vector index must reference source
files"));
+ }
+ // NOT Vec::with_capacity(count): count is untrusted and may be huge.
+ let mut source_files = Vec::new();
+ for _ in 0..count {
+ let file_name = read_java_utf(&mut cursor)?;
+ let row_count = cursor.read_i64_be()?;
+ source_files.push(PkVectorSourceFile::new(file_name, row_count)?);
+ }
+ if cursor.remaining() != 0 {
+ return Err(data_invalid(
+ "unexpected trailing bytes in vector source metadata",
+ ));
+ }
+ Self::new(source_files)
+ }
+}
+
+/// Minimal big-endian reader mirroring the Java `DataInput` primitives the
+/// `_SOURCE_META` frame uses. Module-private by design (see the PR1 spec: no
+/// shared `common/` abstraction until a second consumer exists).
+struct DataInputCursor<'a> {
+ bytes: &'a [u8],
+ position: usize,
+}
+
+impl<'a> DataInputCursor<'a> {
+ fn new(bytes: &'a [u8]) -> Self {
+ Self { bytes, position: 0 }
+ }
+
+ fn remaining(&self) -> usize {
+ self.bytes.len() - self.position
+ }
+
+ fn read_exact(&mut self, len: usize) -> crate::Result<&'a [u8]> {
+ if self.remaining() < len {
+ return Err(data_invalid(format!(
+ "unexpected end of vector source metadata: need {len} bytes,
{} remain",
+ self.remaining()
+ )));
+ }
+ let slice = &self.bytes[self.position..self.position + len];
+ self.position += len;
+ Ok(slice)
+ }
+
+ fn read_u16_be(&mut self) -> crate::Result<u16> {
+ let b = self.read_exact(2)?;
+ Ok(u16::from_be_bytes([b[0], b[1]]))
+ }
+
+ fn read_i32_be(&mut self) -> crate::Result<i32> {
+ let b = self.read_exact(4)?;
+ Ok(i32::from_be_bytes([b[0], b[1], b[2], b[3]]))
+ }
+
+ fn read_i64_be(&mut self) -> crate::Result<i64> {
+ let b = self.read_exact(8)?;
+ Ok(i64::from_be_bytes([
+ b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7],
+ ]))
+ }
+}
+
+/// Decode one Java `DataOutput#writeUTF` value: a u16-BE byte length followed
by
+/// modified UTF-8. Reassembles UTF-16 code units, then `String::from_utf16`.
+///
+/// A Java-written NUL is the two-byte sequence `C0 80`; a raw `0x00` is
accepted
+/// (Java `DataInputStream#readUTF` treats `0x00..=0x7F` as single-byte chars).
+fn read_java_utf(cursor: &mut DataInputCursor<'_>) -> crate::Result<String> {
+ let len = cursor.read_u16_be()? as usize;
+ let bytes = cursor.read_exact(len)?;
+ let mut units: Vec<u16> = Vec::new();
+ let mut i = 0;
+ while i < len {
+ let b0 = bytes[i];
+ if b0 & 0x80 == 0 {
+ // 1-byte: 0xxxxxxx (includes raw 0x00)
+ units.push(b0 as u16);
+ i += 1;
+ } else if b0 & 0xE0 == 0xC0 {
+ // 2-byte: 110xxxxx 10xxxxxx
+ if i + 1 >= len || bytes[i + 1] & 0xC0 != 0x80 {
+ return Err(data_invalid("malformed modified UTF-8 (2-byte)"));
+ }
+ let u = (((b0 & 0x1F) as u16) << 6) | ((bytes[i + 1] & 0x3F) as
u16);
+ units.push(u);
+ i += 2;
+ } else if b0 & 0xF0 == 0xE0 {
+ // 3-byte: 1110xxxx 10xxxxxx 10xxxxxx
+ if i + 2 >= len || bytes[i + 1] & 0xC0 != 0x80 || bytes[i + 2] &
0xC0 != 0x80 {
+ return Err(data_invalid("malformed modified UTF-8 (3-byte)"));
+ }
+ let u = (((b0 & 0x0F) as u16) << 12)
+ | (((bytes[i + 1] & 0x3F) as u16) << 6)
+ | ((bytes[i + 2] & 0x3F) as u16);
+ units.push(u);
+ i += 3;
+ } else {
+ return Err(data_invalid("malformed modified UTF-8 (lead byte)"));
+ }
+ }
+ String::from_utf16(&units)
+ .map_err(|_| data_invalid("modified UTF-8 did not decode to valid
UTF-16"))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ /// Build a Java `writeUTF` byte sequence (u16-BE length + modified UTF-8)
+ /// for ASCII/BMP test strings. Encodes via UTF-16 code units to match
Java.
+ fn java_write_utf(s: &str) -> Vec<u8> {
+ let mut body = Vec::new();
+ for c in s.encode_utf16() {
+ if (0x0001..=0x007F).contains(&c) {
+ body.push(c as u8);
+ } else if c > 0x07FF {
+ body.push(0xE0 | (c >> 12) as u8);
+ body.push(0x80 | ((c >> 6) & 0x3F) as u8);
+ body.push(0x80 | (c & 0x3F) as u8);
+ } else {
+ body.push(0xC0 | (c >> 6) as u8);
+ body.push(0x80 | (c & 0x3F) as u8);
+ }
+ }
+ let mut out = (body.len() as u16).to_be_bytes().to_vec();
+ out.extend_from_slice(&body);
+ out
+ }
+
+ fn frame(files: &[(&str, i64)]) -> Vec<u8> {
+ let mut out = Vec::new();
+ out.extend_from_slice(&1i32.to_be_bytes()); // version
+ out.extend_from_slice(&(files.len() as i32).to_be_bytes());
+ for (name, rows) in files {
+ out.extend_from_slice(&java_write_utf(name));
+ out.extend_from_slice(&rows.to_be_bytes());
+ }
+ out
+ }
+
+ #[test]
+ fn read_java_utf_round_trips_ascii_and_multibyte() {
+ for s in ["data-abc.parquet", "café", "ü", "日本語", "🚀"] {
+ let bytes = java_write_utf(s);
+ let mut cursor = DataInputCursor::new(&bytes);
+ assert_eq!(read_java_utf(&mut cursor).unwrap(), s);
+ assert_eq!(cursor.remaining(), 0);
+ }
+ }
+
+ #[test]
+ fn read_java_utf_decodes_c0_80_as_nul() {
+ // Java writeUTF encodes U+0000 as 0xC0 0x80.
+ let bytes = [0x00, 0x02, 0xC0, 0x80];
+ let mut cursor = DataInputCursor::new(&bytes);
+ assert_eq!(read_java_utf(&mut cursor).unwrap(), "\u{0000}");
+ }
+
+ #[test]
+ fn read_java_utf_accepts_raw_nul() {
+ // Raw 0x00 is a valid single-byte char for Java readUTF; must not
error.
+ let bytes = [0x00, 0x01, 0x00];
+ let mut cursor = DataInputCursor::new(&bytes);
+ assert_eq!(read_java_utf(&mut cursor).unwrap(), "\u{0000}");
+ }
+
+ #[test]
+ fn read_java_utf_rejects_truncated_and_bad_continuation() {
+ // 2-byte lead with no continuation.
+ let mut c1 = DataInputCursor::new(&[0x00, 0x01, 0xC0]);
+ assert!(read_java_utf(&mut c1).is_err());
+ // 3-byte lead with bad 2nd byte.
+ let mut c2 = DataInputCursor::new(&[0x00, 0x03, 0xE0, 0x00, 0x80]);
+ assert!(read_java_utf(&mut c2).is_err());
+ }
+
+ #[test]
+ fn deserialize_single_source_file() {
+ let bytes = frame(&[("data-abc.parquet", 100)]);
+ let meta = PkVectorSourceMeta::deserialize(&bytes).unwrap();
+ assert_eq!(meta.source_files().len(), 1);
+ assert_eq!(meta.source_files()[0].file_name(), "data-abc.parquet");
+ assert_eq!(meta.source_files()[0].row_count(), 100);
+ }
+
+ #[test]
+ fn deserialize_multi_source_files() {
+ let bytes = frame(&[("f0", 3), ("f1", 5)]);
+ let meta = PkVectorSourceMeta::deserialize(&bytes).unwrap();
+ assert_eq!(meta.source_files().len(), 2);
+ assert_eq!(meta.source_files()[1].row_count(), 5);
+ }
+
+ #[test]
+ fn deserialize_rejects_bad_version() {
+ let mut bytes = frame(&[("f0", 1)]);
+ bytes[0..4].copy_from_slice(&2i32.to_be_bytes());
+ assert!(PkVectorSourceMeta::deserialize(&bytes).is_err());
+ }
+
+ #[test]
+ fn deserialize_rejects_zero_count() {
+ let mut out = Vec::new();
+ out.extend_from_slice(&1i32.to_be_bytes());
+ out.extend_from_slice(&0i32.to_be_bytes());
+ assert!(PkVectorSourceMeta::deserialize(&out).is_err());
+ }
+
+ #[test]
+ fn deserialize_rejects_trailing_bytes() {
+ let mut bytes = frame(&[("f0", 1)]);
+ bytes.push(0xFF);
+ assert!(PkVectorSourceMeta::deserialize(&bytes).is_err());
+ }
+
+ #[test]
+ fn deserialize_rejects_truncated_input() {
+ let bytes = frame(&[("f0", 1)]);
+ assert!(PkVectorSourceMeta::deserialize(&bytes[..bytes.len() -
2]).is_err());
+ }
+
+ #[test]
+ fn deserialize_rejects_negative_row_count() {
+ let bytes = frame(&[("f0", -1)]);
+ assert!(PkVectorSourceMeta::deserialize(&bytes).is_err());
+ }
+
+ #[test]
+ fn new_rejects_empty() {
+ assert!(PkVectorSourceMeta::new(Vec::new()).is_err());
+ }
+
+ #[test]
+ fn resolve_single_file() {
+ let meta = PkVectorSourceMeta::deserialize(&frame(&[("f0",
3)])).unwrap();
+ assert_eq!(meta.resolve(0).unwrap(), ("f0".to_string(), 0));
+ assert_eq!(meta.resolve(2).unwrap(), ("f0".to_string(), 2));
+ }
+
+ #[test]
+ fn resolve_multi_file_prefix_sum_boundaries() {
+ // f0 owns ordinals 0..=2, f1 owns 3..=7.
+ let meta = PkVectorSourceMeta::deserialize(&frame(&[("f0", 3), ("f1",
5)])).unwrap();
+ assert_eq!(meta.resolve(2).unwrap(), ("f0".to_string(), 2)); // last
of f0
+ assert_eq!(meta.resolve(3).unwrap(), ("f1".to_string(), 0)); // first
of f1
+ assert_eq!(meta.resolve(7).unwrap(), ("f1".to_string(), 4)); // last
of f1
+ }
+
+ #[test]
+ fn resolve_rejects_negative_ordinal() {
+ let meta = PkVectorSourceMeta::deserialize(&frame(&[("f0",
3)])).unwrap();
+ assert!(meta.resolve(-1).is_err());
+ }
+
+ #[test]
+ fn resolve_rejects_ordinal_at_or_past_total() {
+ let meta = PkVectorSourceMeta::deserialize(&frame(&[("f0",
3)])).unwrap();
+ assert!(meta.resolve(3).is_err()); // total == 3, valid range 0..=2
+ }
+
+ #[test]
+ fn from_global_index_meta_errors_without_source_meta() {
+ let meta = GlobalIndexMeta {
+ row_range_start: 0,
+ row_range_end: 0,
+ index_field_id: 0,
+ extra_field_ids: None,
+ index_meta: None,
+ source_meta: None,
+ };
+ assert!(PkVectorSourceMeta::from_global_index_meta(&meta).is_err());
+ }
+
+ #[test]
+ fn from_global_index_meta_parses_source_meta() {
+ let meta = GlobalIndexMeta {
+ row_range_start: 0,
+ row_range_end: 0,
+ index_field_id: 0,
+ extra_field_ids: None,
+ index_meta: None,
+ source_meta: Some(frame(&[("f0", 3)])),
+ };
+ let parsed =
PkVectorSourceMeta::from_global_index_meta(&meta).unwrap();
+ assert_eq!(parsed.source_files()[0].file_name(), "f0");
+ }
+
+ #[test]
+ fn resolve_rejects_row_count_overflow() {
+ // Two individually-valid row counts whose prefix sum overflows i64.
+ // Resolving past the first file forces the checked_add on the second.
+ let meta = PkVectorSourceMeta::new(vec![
+ PkVectorSourceFile::new("f0".to_string(), i64::MAX).unwrap(),
+ PkVectorSourceFile::new("f1".to_string(), 1).unwrap(),
+ ])
+ .unwrap();
+ assert!(meta.resolve(i64::MAX).is_err());
+ }
+
+ #[test]
+ fn read_java_utf_rejects_lone_high_surrogate() {
+ // U+D800 (a lone high surrogate) as 3-byte modified UTF-8: ED A0 80.
+ // It decodes to a single UTF-16 code unit that is not valid on its
own,
+ // so String::from_utf16 must reject it (not panic).
+ let bytes = [0x00, 0x03, 0xED, 0xA0, 0x80];
+ let mut cursor = DataInputCursor::new(&bytes);
+ assert!(read_java_utf(&mut cursor).is_err());
+ }
+}
diff --git a/crates/paimon/src/table/btree_global_index_build_builder.rs
b/crates/paimon/src/table/btree_global_index_build_builder.rs
index cc87c70..0b759b1 100644
--- a/crates/paimon/src/table/btree_global_index_build_builder.rs
+++ b/crates/paimon/src/table/btree_global_index_build_builder.rs
@@ -286,6 +286,7 @@ impl<'a> BTreeGlobalIndexBuildBuilder<'a> {
row_range_end: shard.row_range_end,
index_field_id: index_field.id(),
extra_field_ids: None,
+ source_meta: None,
index_meta: Some(index_meta.serialize()),
}),
})
@@ -1989,6 +1990,7 @@ mod tests {
row_range_end: hole_end,
index_field_id: name_field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: None,
}),
};
diff --git a/crates/paimon/src/table/global_index_drop_builder.rs
b/crates/paimon/src/table/global_index_drop_builder.rs
index e9ab86c..cc86ceb 100644
--- a/crates/paimon/src/table/global_index_drop_builder.rs
+++ b/crates/paimon/src/table/global_index_drop_builder.rs
@@ -216,6 +216,7 @@ mod tests {
row_range_end,
index_field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: None,
}),
}
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index 57cce1c..adfc628 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -1439,6 +1439,7 @@ mod tests {
row_range_end,
index_field_id: field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: Some(meta.serialize()),
}),
},
diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs
b/crates/paimon/src/table/lumina_index_build_builder.rs
index 4a2980a..cbff6e7 100644
--- a/crates/paimon/src/table/lumina_index_build_builder.rs
+++ b/crates/paimon/src/table/lumina_index_build_builder.rs
@@ -241,6 +241,7 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
row_range_end: shard.row_range_end,
index_field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: Some(index_meta),
}),
})
@@ -1697,6 +1698,7 @@ mod tests {
row_range_end: end,
index_field_id: field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: None,
}),
};
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index dad551f..2a81047 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -2884,6 +2884,7 @@ mod tests {
row_range_end,
index_field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: None,
}),
}
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index c439bc6..c5bdc96 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -1897,6 +1897,7 @@ mod tests {
row_range_end: 9,
index_field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: None,
}),
},
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
index 5803af8..0f0d8c2 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -266,6 +266,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
row_range_end: shard.row_range_end,
index_field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: Some(index_meta),
}),
})
@@ -1091,6 +1092,7 @@ mod tests {
row_range_end: end,
index_field_id: field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: None,
}),
};
@@ -1333,6 +1335,7 @@ mod tests {
row_range_end: coverage[0].to(),
index_field_id: field_id,
extra_field_ids: None,
+ source_meta: None,
index_meta: None,
}),
};
