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 b35453f0..986b27b9 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> { 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 e985bb01..ff1e63c0 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>, + + #[serde(default, rename = "_SOURCE_META", with = "serde_bytes")] + pub source_meta: Option>, } /// Metadata of index file. diff --git a/crates/paimon/src/spec/index_manifest.rs b/crates/paimon/src/spec/index_manifest.rs index b8ee2b52..a0430b2f 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>) -> 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 9e8b07b8..4614e89a 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 00000000..2685f45e --- /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) -> 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 { + 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, +} + +impl PkVectorSourceMeta { + pub fn new(source_files: Vec) -> crate::Result { + 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 { + 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 { + 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 { + let b = self.read_exact(2)?; + Ok(u16::from_be_bytes([b[0], b[1]])) + } + + fn read_i32_be(&mut self) -> crate::Result { + 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 { + 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 { + let len = cursor.read_u16_be()? as usize; + let bytes = cursor.read_exact(len)?; + let mut units: Vec = 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 { + 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 { + 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 cc87c70f..0b759b1d 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 e9ab86cd..cc86ceb0 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 57cce1c2..adfc628a 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 4a2980a6..cbff6e7e 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 dad551f4..2a81047e 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 c439bc67..c5bdc964 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 5803af82..0f0d8c25 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, }), };