diff --git a/bindings/python/tests/test_read.py b/bindings/python/tests/test_read.py index 49ed926d1..c53d656a1 100644 --- a/bindings/python/tests/test_read.py +++ b/bindings/python/tests/test_read.py @@ -736,10 +736,10 @@ def test_split_serialize_produces_split_v1_binary(): version, type_id = struct.unpack_from(">ii", data, 8) assert version == 1 assert type_id == 1 - # DataSplit v8 body follows: MAGIC + VERSION(8) + # DataSplit body follows: MAGIC + VERSION magic, body_version = struct.unpack_from(">qi", data, 16) assert magic == -2394839472490812314 - assert body_version == 8 + assert body_version == 9 assert splits[0].serialize() == data # deterministic diff --git a/crates/paimon/src/spec/avro/manifest_entry_decode.rs b/crates/paimon/src/spec/avro/manifest_entry_decode.rs index df251bd3a..0de57aaec 100644 --- a/crates/paimon/src/spec/avro/manifest_entry_decode.rs +++ b/crates/paimon/src/spec/avro/manifest_entry_decode.rs @@ -221,6 +221,7 @@ fn decode_data_file_meta( let mut external_path: Option = None; let mut first_row_id: Option = None; let mut write_cols: Option> = None; + let mut column_max_sequence_numbers: Option> = None; for field in &writer_schema.fields { match field.name.as_str() { @@ -260,6 +261,9 @@ fn decode_data_file_meta( "_EXTERNAL_PATH" => external_path = decode_nullable_string(cursor, field.nullable)?, "_FIRST_ROW_ID" => first_row_id = decode_nullable_long(cursor, field.nullable)?, "_WRITE_COLS" => write_cols = decode_nullable_string_array(cursor, field.nullable)?, + "_WRITE_COLS_SEQUENCES" => { + column_max_sequence_numbers = decode_nullable_long_array(cursor, field.nullable)? + } _ => skip_nullable_field(cursor, &field.schema, field.nullable)?, } } @@ -285,6 +289,7 @@ fn decode_data_file_meta( external_path, first_row_id, write_cols, + column_max_sequence_numbers, }) } @@ -309,6 +314,55 @@ fn decode_string_array(cursor: &mut AvroCursor) -> crate::Result> { Ok(result) } +/// Decode an Avro `array` whose items are plain (non-union) longs, mirroring +/// [`decode_string_array`]. Paimon declares `_WRITE_COLS_SEQUENCES` as +/// `array`, so items carry no per-element union index. +fn decode_long_array(cursor: &mut AvroCursor) -> crate::Result> { + let mut result = Vec::new(); + loop { + let count = cursor.read_long()?; + if count == 0 { + break; + } + let count = if count < 0 { + cursor.skip_long()?; + neg_count_to_usize(count)? + } else { + count as usize + }; + // A zigzag long is at least one byte, so the remaining input bounds how many + // elements can really follow. Reserving the declared count instead would let a + // corrupt manifest ask for an arbitrary allocation before the first read fails. + result.reserve(count.min(cursor.remaining())); + for _ in 0..count { + result.push(cursor.read_long()?); + } + } + Ok(result) +} + +fn decode_nullable_long_array( + cursor: &mut AvroCursor, + nullable: bool, +) -> crate::Result>> { + if nullable { + // A two-branch `["null", array]` union only ever encodes 0 or 1. Anything else + // means the stream is not where the schema says it is, so stop rather than read + // the following bytes as an array. + match cursor.read_union_index()? { + 0 => return Ok(None), + 1 => {} + other => { + return Err(crate::Error::DataInvalid { + message: format!("invalid union index {other} for _WRITE_COLS_SEQUENCES"), + source: None, + }) + } + } + } + Ok(Some(decode_long_array(cursor)?)) +} + fn decode_nullable_long(cursor: &mut AvroCursor, nullable: bool) -> crate::Result> { if nullable { let idx = cursor.read_union_index()?; @@ -406,5 +460,6 @@ fn default_data_file_meta() -> DataFileMeta { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/spec/data_file.rs b/crates/paimon/src/spec/data_file.rs index 314c6d15a..97e2d2d07 100644 --- a/crates/paimon/src/spec/data_file.rs +++ b/crates/paimon/src/spec/data_file.rs @@ -17,7 +17,8 @@ use crate::spec::stats::BinaryTableStats; use crate::spec::{ - extract_datum, serialize_binary_array_str, BinaryRow, BinaryRowBuilder, DataField, Datum, + extract_datum, serialize_binary_array_long, serialize_binary_array_str, BinaryRow, + BinaryRowBuilder, DataField, Datum, }; use chrono::serde::ts_milliseconds_option::deserialize as from_millis_opt; use chrono::serde::ts_milliseconds_option::serialize as to_millis_opt; @@ -122,6 +123,16 @@ pub struct DataFileMeta { skip_serializing_if = "Option::is_none" )] pub write_cols: Option>, + + /// Per-column maximum sequence numbers, positionally aligned with + /// [`Self::write_cols`] (used in data evolution mode). Absent on files + /// written before this field existed. + #[serde( + rename = "_WRITE_COLS_SEQUENCES", + default, + skip_serializing_if = "Option::is_none" + )] + pub column_max_sequence_numbers: Option>, } impl Display for DataFileMeta { @@ -132,7 +143,8 @@ impl Display for DataFileMeta { minKey={:?}, maxKey={:?}, keyStats={:?}, valueStats={:?}, \ minSequenceNumber={}, maxSequenceNumber={}, schemaId={}, level={}, \ extraFiles={:?}, creationTime={:?}, deleteRowCount={:?}, fileSource={:?}, \ - valueStatsCols={:?}, externalPath={:?}, firstRowId={:?}, writeCols={:?}}}", + valueStatsCols={:?}, externalPath={:?}, firstRowId={:?}, writeCols={:?}, \ + columnMaxSequenceNumbers={:?}}}", self.file_name, self.file_size, self.row_count, @@ -153,10 +165,38 @@ impl Display for DataFileMeta { self.external_path, self.first_row_id, self.write_cols, + self.column_max_sequence_numbers, ) } } +/// Which `DataFileMeta.SCHEMA` revision a serialized `BinaryRow` follows. +/// +/// The schema only ever grows by appending nullable fields, so the two layouts +/// share slots 0..=19 and differ solely in arity. Which one a `DataSplit` body +/// carries is decided by the split's own version, not by the row bytes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DataFileMetaRowLayout { + /// 20 fields, ending at `_WRITE_COLS`. Mirrors Java + /// `DataFileMetaWriteColsLegacySerializer`. + WriteCols, + /// 21 fields, ending at `_WRITE_COLS_SEQUENCES`. Mirrors Java + /// `DataFileMetaSerializer`, and is what this crate writes. + WriteColsSequences, +} + +impl DataFileMetaRowLayout { + /// The layout this crate serializes to, matching current Java. + pub const CURRENT: Self = Self::WriteColsSequences; + + fn arity(self) -> usize { + match self { + Self::WriteCols => 20, + Self::WriteColsSequences => 21, + } + } +} + fn opt_long(b: &mut BinaryRowBuilder, pos: usize, v: Option) { match v { Some(x) => b.write_long(pos, x), @@ -164,6 +204,26 @@ fn opt_long(b: &mut BinaryRowBuilder, pos: usize, v: Option) { } } +/// Write an `array` field, or a null at `pos`. +fn opt_long_array(b: &mut BinaryRowBuilder, pos: usize, v: &Option>) { + match v { + Some(a) => { + let elements: Vec> = a.iter().copied().map(Some).collect(); + b.write_bytes(pos, &serialize_binary_array_long(&elements)); + } + None => b.set_null_at(pos), + } +} + +/// Read an `array` field. Java declares the elements non-null, so a +/// null element is malformed rather than something to silently map to a placeholder. +fn read_non_null_long_array(row: &BinaryRow, pos: usize) -> crate::Result> { + crate::spec::deserialize_binary_array_long(row.get_binary(pos)?)? + .into_iter() + .map(|v| v.ok_or_else(|| data_file_err("array has a null element"))) + .collect() +} + fn opt_str_array(b: &mut BinaryRowBuilder, pos: usize, v: &Option>) { match v { Some(a) => b.write_bytes(pos, &serialize_binary_array_str(a)), @@ -282,11 +342,15 @@ impl DataFileMeta { Some((from, to)) } - /// Serialize as a `DataFileMeta.SCHEMA` (version 8) BinaryRow, raw data without the - /// arity prefix -- the form `DataSplit#serialize` writes per file as `writeInt(len) + data`. - /// Fields, order and nullability mirror Java `DataFileMetaSerializer#toRow`. + /// Serialize as a [`DataFileMetaRowLayout::CURRENT`] `DataFileMeta.SCHEMA` BinaryRow, + /// raw data without the arity prefix -- the form `DataSplit#serialize` writes per file + /// as `writeInt(len) + data`. Fields, order and nullability mirror Java + /// `DataFileMetaSerializer#toRow`. + /// + /// Only the current layout is written. Reading an older one stays supported because a + /// `DataSplit` on disk or on the wire may still carry it. pub fn to_serialized_row_data(&self) -> crate::Result> { - let mut b = BinaryRowBuilder::new(20); + let mut b = BinaryRowBuilder::new(DataFileMetaRowLayout::CURRENT.arity() as i32); b.write_bytes(0, self.file_name.as_bytes()); b.write_long(1, self.file_size); b.write_long(2, self.row_count); @@ -325,14 +389,29 @@ impl DataFileMeta { } opt_long(&mut b, 18, self.first_row_id); opt_str_array(&mut b, 19, &self.write_cols); + opt_long_array(&mut b, 20, &self.column_max_sequence_numbers); Ok(b.build_row_data()) } - /// Reverse of [`DataFileMeta::to_serialized_row_data`]: decode the fixed - /// 20-field `DataFileMeta` BinaryRow (version 8 layout). - pub fn from_serialized_row_data(data: &[u8]) -> crate::Result { + /// Reverse of [`DataFileMeta::to_serialized_row_data`]: decode a `DataFileMeta` + /// BinaryRow written in `layout`. + /// + /// The row bytes carry no arity of their own, so the caller must supply the layout + /// that the enclosing container declared. + pub fn from_serialized_row_data( + data: &[u8], + layout: DataFileMetaRowLayout, + ) -> crate::Result { use crate::spec::deserialize_binary_array_str; - let row = BinaryRow::from_bytes(20, data.to_vec()); + let fixed_part = BinaryRow::cal_fix_part_size_in_bytes(layout.arity() as i32) as usize; + if data.len() < fixed_part { + return Err(data_file_err(&format!( + "DataFileMeta row of {} bytes is shorter than the {fixed_part}-byte fixed part \ + of the {layout:?} layout", + data.len() + ))); + } + let row = BinaryRow::from_bytes(layout.arity() as i32, data.to_vec()); let file_name = String::from_utf8(row.get_binary(0)?.to_vec()) .map_err(|_| data_file_err("file_name is not valid UTF-8"))?; @@ -390,6 +469,11 @@ impl DataFileMeta { } else { Some(deserialize_binary_array_str(row.get_binary(19)?)?) }; + let column_max_sequence_numbers = match layout { + DataFileMetaRowLayout::WriteCols => None, + DataFileMetaRowLayout::WriteColsSequences if row.is_null_at(20) => None, + DataFileMetaRowLayout::WriteColsSequences => Some(read_non_null_long_array(&row, 20)?), + }; Ok(DataFileMeta { file_name, @@ -412,6 +496,7 @@ impl DataFileMeta { external_path, first_row_id, write_cols, + column_max_sequence_numbers, }) } @@ -524,6 +609,7 @@ mod tests { external_path: None, first_row_id: Some(100), write_cols: Some(vec!["k".to_string(), "v".to_string()]), + column_max_sequence_numbers: None, } } @@ -565,6 +651,7 @@ mod tests { external_path: Some("s3://bucket/data-full.parquet".to_string()), first_row_id: Some(1_000), write_cols: Some(vec!["a".to_string(), "b".to_string(), "c".to_string()]), + column_max_sequence_numbers: None, } } @@ -591,6 +678,7 @@ mod tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } @@ -601,11 +689,79 @@ mod tests { sample_minimal_data_file_meta(), ] { let bytes = meta.to_serialized_row_data().unwrap(); - let back = DataFileMeta::from_serialized_row_data(&bytes).unwrap(); + let back = + DataFileMeta::from_serialized_row_data(&bytes, DataFileMetaRowLayout::CURRENT) + .unwrap(); assert_eq!(back, meta); } } + #[test] + fn column_max_sequence_numbers_round_trip() { + let mut meta = sample_full_data_file_meta(); + meta.column_max_sequence_numbers = Some(vec![15, 100, 150, 200]); + let bytes = meta.to_serialized_row_data().unwrap(); + let back = + DataFileMeta::from_serialized_row_data(&bytes, DataFileMetaRowLayout::CURRENT).unwrap(); + assert_eq!( + back.column_max_sequence_numbers, + Some(vec![15, 100, 150, 200]) + ); + assert_eq!(back, meta); + } + + #[test] + fn empty_column_max_sequence_numbers_round_trips() { + let mut meta = sample_full_data_file_meta(); + meta.column_max_sequence_numbers = Some(Vec::new()); + let bytes = meta.to_serialized_row_data().unwrap(); + let back = + DataFileMeta::from_serialized_row_data(&bytes, DataFileMetaRowLayout::CURRENT).unwrap(); + // An empty array must stay distinguishable from an absent one. + assert_eq!(back.column_max_sequence_numbers, Some(Vec::new())); + assert_eq!(back, meta); + } + + #[test] + fn row_shorter_than_the_layout_fixed_part_is_rejected() { + let meta = sample_full_data_file_meta(); + let bytes = meta.to_serialized_row_data().unwrap(); + let fixed_part = + BinaryRow::cal_fix_part_size_in_bytes(DataFileMetaRowLayout::CURRENT.arity() as i32) + as usize; + + for truncated in [0, fixed_part - 1] { + assert!( + DataFileMeta::from_serialized_row_data( + &bytes[..truncated], + DataFileMetaRowLayout::CURRENT + ) + .is_err(), + "a {truncated}-byte row must not decode as the current layout" + ); + } + assert!( + DataFileMeta::from_serialized_row_data(&bytes, DataFileMetaRowLayout::CURRENT).is_ok() + ); + } + + /// The legacy layout has no slot for the field, so decoding a row as `WriteCols` + /// yields `None` rather than reading past the fields that layout declares. Java does + /// the same: `DataFileMetaSerializerTest#testLegacySerializerDropsColumnSequences` + /// round-trips through the legacy serializer and expects the field back as null. + #[test] + fn legacy_layout_decodes_without_column_max_sequence_numbers() { + let mut meta = sample_full_data_file_meta(); + meta.column_max_sequence_numbers = Some(vec![7]); + let bytes = meta.to_serialized_row_data().unwrap(); + + let back = DataFileMeta::from_serialized_row_data(&bytes, DataFileMetaRowLayout::WriteCols) + .unwrap(); + assert_eq!(back.column_max_sequence_numbers, None); + meta.column_max_sequence_numbers = None; + assert_eq!(back, meta); + } + #[test] fn row_id_range_rejects_non_positive_count_and_overflow() { let mut file = data_file("data.parquet"); @@ -707,6 +863,7 @@ mod tests { "externalPath=Some(\"s3://bucket/data-1.parquet\")", "firstRowId=Some(100)", "writeCols=Some([\"k\", \"v\"])", + "columnMaxSequenceNumbers=None", ] { assert!( display.contains(expected), diff --git a/crates/paimon/src/spec/manifest.rs b/crates/paimon/src/spec/manifest.rs index c25dba9ad..ed2fa162b 100644 --- a/crates/paimon/src/spec/manifest.rs +++ b/crates/paimon/src/spec/manifest.rs @@ -166,6 +166,7 @@ mod tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, }; ManifestEntry::new(kind, vec![], 0, 1, file, 2) } diff --git a/crates/paimon/src/spec/manifest_entry.rs b/crates/paimon/src/spec/manifest_entry.rs index 63473247f..484f2bd9f 100644 --- a/crates/paimon/src/spec/manifest_entry.rs +++ b/crates/paimon/src/spec/manifest_entry.rs @@ -231,7 +231,8 @@ pub const MANIFEST_ENTRY_SCHEMA: &str = r#"["null", { {"name": "_VALUE_STATS_COLS", "type": ["null", {"type": "array", "items": "string"}], "default": null}, {"name": "_EXTERNAL_PATH", "type": ["null", "string"], "default": null}, {"name": "_FIRST_ROW_ID", "type": ["null", "long"], "default": null}, - {"name": "_WRITE_COLS", "type": ["null", {"type": "array", "items": "string"}], "default": null} + {"name": "_WRITE_COLS", "type": ["null", {"type": "array", "items": "string"}], "default": null}, + {"name": "_WRITE_COLS_SEQUENCES", "type": ["null", {"type": "array", "items": "long"}], "default": null} ] }} ] diff --git a/crates/paimon/src/spec/objects_file.rs b/crates/paimon/src/spec/objects_file.rs index 7db066fad..d5d0d9fe1 100644 --- a/crates/paimon/src/spec/objects_file.rs +++ b/crates/paimon/src/spec/objects_file.rs @@ -89,6 +89,12 @@ mod tests { use chrono::{DateTime, Utc}; fn manifest_entry() -> ManifestEntry { + manifest_entry_with_sequences(None) + } + + fn manifest_entry_with_sequences( + column_max_sequence_numbers: Option>, + ) -> ManifestEntry { let value_bytes = vec![ 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 49, 0, 0, 0, 0, 0, 0, 129, 1, 0, 0, 0, 0, 0, 0, 0, ]; @@ -131,6 +137,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers, }, 2, ) @@ -189,6 +196,63 @@ mod tests { assert_eq!(original, decoded); } + /// The Avro manifest record carries `_WRITE_COLS_SEQUENCES`, so a populated value must + /// survive a write/read round trip rather than being dropped on either side. + #[test] + fn test_roundtrip_manifest_entry_with_column_max_sequence_numbers() { + for value in [Some(vec![15, 100, 150, 200]), Some(Vec::new()), None] { + let original = vec![manifest_entry_with_sequences(value.clone())]; + let bytes = to_avro_bytes(MANIFEST_ENTRY_SCHEMA, &original).unwrap(); + + let decoded = from_avro_bytes::(&bytes).unwrap(); + assert_eq!(decoded, original); + assert_eq!(decoded[0].file().column_max_sequence_numbers, value); + + // The hand-written decoder reads the same bytes by field name. + let fast = from_avro_bytes_fast::(&bytes).unwrap(); + assert_eq!(fast, original); + assert_eq!(fast[0].file().column_max_sequence_numbers, value); + } + } + + /// A writer that predates the field omits it entirely; it must read back as absent. + #[test] + fn test_read_manifest_entry_without_column_max_sequence_numbers_field() { + let mut schema: serde_json::Value = serde_json::from_str(MANIFEST_ENTRY_SCHEMA).unwrap(); + let fields = schema.as_array_mut().unwrap()[1] + .as_object_mut() + .unwrap() + .get_mut("fields") + .unwrap() + .as_array_mut() + .unwrap(); + let file_fields = fields + .iter_mut() + .find(|field| field.get("name").and_then(|name| name.as_str()) == Some("_FILE")) + .unwrap() + .get_mut("type") + .unwrap() + .get_mut("fields") + .unwrap() + .as_array_mut() + .unwrap(); + let before = file_fields.len(); + file_fields.retain(|field| { + field.get("name").and_then(|name| name.as_str()) != Some("_WRITE_COLS_SEQUENCES") + }); + assert_eq!( + file_fields.len(), + before - 1, + "field must exist to be removed" + ); + let legacy_schema = serde_json::to_string(&schema).unwrap(); + + let original = vec![manifest_entry()]; + let bytes = to_avro_bytes(&legacy_schema, &original).unwrap(); + let decoded = from_avro_bytes_fast::(&bytes).unwrap(); + assert_eq!(decoded[0].file().column_max_sequence_numbers, None); + } + #[test] fn test_read_manifest_entry_with_legacy_rust_schema() { let mut schema: serde_json::Value = serde_json::from_str(MANIFEST_ENTRY_SCHEMA).unwrap(); @@ -315,6 +379,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, }, 2 ), @@ -356,6 +421,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, }, 2 ), diff --git a/crates/paimon/src/table/bin_pack.rs b/crates/paimon/src/table/bin_pack.rs index b4dc178a4..8c5826953 100644 --- a/crates/paimon/src/table/bin_pack.rs +++ b/crates/paimon/src/table/bin_pack.rs @@ -101,6 +101,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/data_evolution_reader.rs b/crates/paimon/src/table/data_evolution_reader.rs index 11a4cb969..a456505aa 100644 --- a/crates/paimon/src/table/data_evolution_reader.rs +++ b/crates/paimon/src/table/data_evolution_reader.rs @@ -6389,6 +6389,7 @@ mod tests { external_path: None, first_row_id: Some(first_row_id), write_cols: write_cols.map(|cols| cols.into_iter().map(str::to_string).collect()), + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index 078a2a128..d8b66e86c 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/crates/paimon/src/table/data_evolution_writer.rs @@ -1218,6 +1218,7 @@ mod tests { external_path: None, first_row_id, write_cols, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs index c9e2e1a7b..526f8c7c9 100644 --- a/crates/paimon/src/table/data_file_reader.rs +++ b/crates/paimon/src/table/data_file_reader.rs @@ -1102,6 +1102,7 @@ mod row_tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } @@ -1543,6 +1544,7 @@ mod tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } @@ -1994,6 +1996,7 @@ mod vector_parquet_tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/data_file_writer.rs b/crates/paimon/src/table/data_file_writer.rs index 75f17c6fa..76f49a0d0 100644 --- a/crates/paimon/src/table/data_file_writer.rs +++ b/crates/paimon/src/table/data_file_writer.rs @@ -276,6 +276,7 @@ impl DataFileWriter { external_path: None, first_row_id, write_cols, + column_max_sequence_numbers: None, } } } diff --git a/crates/paimon/src/table/format_table_scan.rs b/crates/paimon/src/table/format_table_scan.rs index 5a6a00c83..1638c0a67 100644 --- a/crates/paimon/src/table/format_table_scan.rs +++ b/crates/paimon/src/table/format_table_scan.rs @@ -669,6 +669,7 @@ fn data_file_meta(file_name: String, file_size: i64, schema_id: i64) -> DataFile external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/global_index_drop_builder.rs b/crates/paimon/src/table/global_index_drop_builder.rs index 80a0ad7f8..6bc70e729 100644 --- a/crates/paimon/src/table/global_index_drop_builder.rs +++ b/crates/paimon/src/table/global_index_drop_builder.rs @@ -327,6 +327,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/goldens/datasplit_v9.bin b/crates/paimon/src/table/goldens/datasplit_v9.bin new file mode 100644 index 000000000..2bfd4ac19 Binary files /dev/null and b/crates/paimon/src/table/goldens/datasplit_v9.bin differ diff --git a/crates/paimon/src/table/goldens/split_v1_data.bin b/crates/paimon/src/table/goldens/split_v1_data.bin index 6cbade9c5..9d2f6c085 100644 Binary files a/crates/paimon/src/table/goldens/split_v1_data.bin and b/crates/paimon/src/table/goldens/split_v1_data.bin differ diff --git a/crates/paimon/src/table/goldens/split_v1_indexed.bin b/crates/paimon/src/table/goldens/split_v1_indexed.bin index 0d20df101..eab6fbe8f 100644 Binary files a/crates/paimon/src/table/goldens/split_v1_indexed.bin and b/crates/paimon/src/table/goldens/split_v1_indexed.bin differ diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index d922c267a..88b1bf806 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -1003,6 +1003,7 @@ mod tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/kv_file_writer.rs b/crates/paimon/src/table/kv_file_writer.rs index b5e94f543..bf1cd9aa1 100644 --- a/crates/paimon/src/table/kv_file_writer.rs +++ b/crates/paimon/src/table/kv_file_writer.rs @@ -533,6 +533,7 @@ impl KeyValueFileWriter { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, }) } diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs b/crates/paimon/src/table/lumina_index_build_builder.rs index e98aedbe8..7bed2cc36 100644 --- a/crates/paimon/src/table/lumina_index_build_builder.rs +++ b/crates/paimon/src/table/lumina_index_build_builder.rs @@ -906,6 +906,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/merge_tree_split_generator.rs b/crates/paimon/src/table/merge_tree_split_generator.rs index cceee451e..6f33d2be5 100644 --- a/crates/paimon/src/table/merge_tree_split_generator.rs +++ b/crates/paimon/src/table/merge_tree_split_generator.rs @@ -460,6 +460,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/partition_stat.rs b/crates/paimon/src/table/partition_stat.rs index ae5e55802..caa8c4a74 100644 --- a/crates/paimon/src/table/partition_stat.rs +++ b/crates/paimon/src/table/partition_stat.rs @@ -242,6 +242,7 @@ mod tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, }; ManifestEntry::new(kind, partition, 0, 1, file, 2) } diff --git a/crates/paimon/src/table/pk_full_text_bucket_search.rs b/crates/paimon/src/table/pk_full_text_bucket_search.rs index 298733f0a..d774e237f 100644 --- a/crates/paimon/src/table/pk_full_text_bucket_search.rs +++ b/crates/paimon/src/table/pk_full_text_bucket_search.rs @@ -397,6 +397,7 @@ mod tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_full_text_bucket_state.rs b/crates/paimon/src/table/pk_full_text_bucket_state.rs index f9122ea11..48ad4b21f 100644 --- a/crates/paimon/src/table/pk_full_text_bucket_state.rs +++ b/crates/paimon/src/table/pk_full_text_bucket_state.rs @@ -255,6 +255,7 @@ mod tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_full_text_read.rs b/crates/paimon/src/table/pk_full_text_read.rs index 3c5a34b1e..63b58f841 100644 --- a/crates/paimon/src/table/pk_full_text_read.rs +++ b/crates/paimon/src/table/pk_full_text_read.rs @@ -719,6 +719,7 @@ mod read_tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_full_text_scan.rs b/crates/paimon/src/table/pk_full_text_scan.rs index f60938582..5a677655f 100644 --- a/crates/paimon/src/table/pk_full_text_scan.rs +++ b/crates/paimon/src/table/pk_full_text_scan.rs @@ -457,6 +457,7 @@ mod tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_vector_data_file_reader.rs b/crates/paimon/src/table/pk_vector_data_file_reader.rs index 9fadaf2b5..c3087beb4 100644 --- a/crates/paimon/src/table/pk_vector_data_file_reader.rs +++ b/crates/paimon/src/table/pk_vector_data_file_reader.rs @@ -312,6 +312,7 @@ mod integration_tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_vector_indexed_split_read.rs b/crates/paimon/src/table/pk_vector_indexed_split_read.rs index c6158108f..d234ddcac 100644 --- a/crates/paimon/src/table/pk_vector_indexed_split_read.rs +++ b/crates/paimon/src/table/pk_vector_indexed_split_read.rs @@ -212,6 +212,7 @@ mod tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } @@ -433,6 +434,7 @@ mod e2e_tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_vector_orchestrator.rs b/crates/paimon/src/table/pk_vector_orchestrator.rs index 1a816bd9a..ac0d3ab35 100644 --- a/crates/paimon/src/table/pk_vector_orchestrator.rs +++ b/crates/paimon/src/table/pk_vector_orchestrator.rs @@ -641,6 +641,7 @@ mod tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } @@ -974,6 +975,7 @@ mod e2e_tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_vector_position_read.rs b/crates/paimon/src/table/pk_vector_position_read.rs index 294468228..067218162 100644 --- a/crates/paimon/src/table/pk_vector_position_read.rs +++ b/crates/paimon/src/table/pk_vector_position_read.rs @@ -338,6 +338,7 @@ mod tests { external_path: None, first_row_id, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/pk_vector_scan.rs b/crates/paimon/src/table/pk_vector_scan.rs index 00bec032b..cfc528b25 100644 --- a/crates/paimon/src/table/pk_vector_scan.rs +++ b/crates/paimon/src/table/pk_vector_scan.rs @@ -414,6 +414,7 @@ mod tests { external_path: None, first_row_id: Some(0), write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/postpone_file_writer.rs b/crates/paimon/src/table/postpone_file_writer.rs index 8ee83cba9..978649cd4 100644 --- a/crates/paimon/src/table/postpone_file_writer.rs +++ b/crates/paimon/src/table/postpone_file_writer.rs @@ -306,5 +306,6 @@ fn build_meta( external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/referenced_files.rs b/crates/paimon/src/table/referenced_files.rs index b602bbfdd..851e87198 100644 --- a/crates/paimon/src/table/referenced_files.rs +++ b/crates/paimon/src/table/referenced_files.rs @@ -911,6 +911,7 @@ mod tests { external_path: Some("s3://bucket/external/data-0.row".to_string()), first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, }; assert_eq!( @@ -968,6 +969,7 @@ mod tests { external_path: Some(format!("{external_dir}/data-0.row")), first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, }; let entry = ManifestEntry::new(FileKind::Add, vec![0u8; 12], 0, 1, data_file, 0); Manifest::write(&file_io, &manifest_path, &[entry]) @@ -1248,6 +1250,7 @@ mod tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, }; let entry = ManifestEntry::new(FileKind::Add, vec![0u8; 12], 0, 1, data_file, 2); Manifest::write(&file_io, &manifest_path, &[entry]) diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs b/crates/paimon/src/table/sorted_global_index_build_builder.rs index f06579195..94fc31650 100644 --- a/crates/paimon/src/table/sorted_global_index_build_builder.rs +++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs @@ -1084,6 +1084,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/source.rs b/crates/paimon/src/table/source.rs index c1e697ec0..38bfdb896 100644 --- a/crates/paimon/src/table/source.rs +++ b/crates/paimon/src/table/source.rs @@ -19,7 +19,7 @@ //! //! Reference: [org.apache.paimon.table.source](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/). -use crate::spec::{BinaryRow, DataFileMeta}; +use crate::spec::{BinaryRow, DataFileMeta, DataFileMetaRowLayout}; use crate::table::stats_filter::group_by_overlapping_row_id; use serde::{Deserialize, Serialize}; use std::sync::Arc; @@ -211,6 +211,7 @@ mod row_range_tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } @@ -682,8 +683,8 @@ impl DataSplit { DataSplitBuilder::new() } - /// Serialize the DataSplit fields to Java `DataSplit#serialize` (version 8) binary. - /// Byte-compatible with `compatibility/datasplit-v8`. Row ranges are not part of the v8 + /// Serialize the DataSplit fields to Java `DataSplit#serialize` (version 9) binary. + /// Byte-compatible with `compatibility/datasplit-v9`. Row ranges are not part of the /// format; `serialize_split_v1` wraps a row-range split as an `IndexedSplit` instead. pub fn serialize(&self) -> crate::Result> { let mut out = Vec::new(); @@ -711,11 +712,12 @@ impl DataSplit { Ok(out) } - /// Reverse of [`DataSplit::serialize`]: parse a raw v8 `DataSplit#serialize` body. + /// Reverse of [`DataSplit::serialize`]: parse a raw `DataSplit#serialize` body. + /// Both the current version and the legacy v8 layout are accepted. /// Consumes the entire buffer; trailing bytes are an error. pub fn deserialize(data: &[u8]) -> crate::Result { let mut cur = data; - let split = Self::read_v8_body(&mut cur)?; + let split = Self::read_body(&mut cur)?; if !cur.is_empty() { return Err(crate::Error::DataInvalid { message: format!("{} trailing bytes after DataSplit", cur.len()), @@ -725,9 +727,9 @@ impl DataSplit { Ok(split) } - /// Read a v8 `DataSplit` body from the cursor, leaving it positioned after the body + /// Read a `DataSplit` body from the cursor, leaving it positioned after the body /// (used both by `deserialize` and the SPLIT_V1 frame reader). - fn read_v8_body(cur: &mut &[u8]) -> crate::Result { + fn read_body(cur: &mut &[u8]) -> crate::Result { let magic = read_i64(cur)?; if magic != SPLIT_MAGIC { return Err(crate::Error::DataInvalid { @@ -736,17 +738,27 @@ impl DataSplit { }); } match read_i32(cur)? { - SPLIT_VERSION => Self::read_v8_body_after_version(cur), + SPLIT_VERSION => { + Self::read_body_after_version(cur, DataFileMetaRowLayout::WriteColsSequences) + } + SPLIT_VERSION_LEGACY_WRITE_COLS => { + Self::read_body_after_version(cur, DataFileMetaRowLayout::WriteCols) + } version => Err(crate::Error::Unsupported { message: format!( - "DataSplit version {version} not supported (only v{SPLIT_VERSION})" + "DataSplit version {version} not supported \ + (only v{SPLIT_VERSION_LEGACY_WRITE_COLS} and v{SPLIT_VERSION})" ), }), } } - /// Read the fields following the magic + version header of a v8 `DataSplit` body. - fn read_v8_body_after_version(cur: &mut &[u8]) -> crate::Result { + /// Read the fields following the magic + version header of a `DataSplit` body. Only the + /// per-file `DataFileMeta` row layout varies between the supported versions. + fn read_body_after_version( + cur: &mut &[u8], + file_layout: DataFileMetaRowLayout, + ) -> crate::Result { let snapshot_id = read_i64(cur)?; let part_len = read_i32(cur)?; @@ -803,7 +815,7 @@ impl DataSplit { }); } let row = take(cur, len as usize)?; - data_files.push(DataFileMeta::from_serialized_row_data(row)?); + data_files.push(DataFileMeta::from_serialized_row_data(row, file_layout)?); } let data_deletion_files = read_deletion_list(cur)?; @@ -881,7 +893,7 @@ impl DataSplit { } let type_id = read_i32(&mut cur)?; let split = match type_id { - SPLIT_SER_TYPE_DATA_SPLIT => Self::read_v8_body(&mut cur)?, + SPLIT_SER_TYPE_DATA_SPLIT => Self::read_body(&mut cur)?, SPLIT_SER_TYPE_INDEXED_SPLIT => { let im = read_i64(&mut cur)?; if im != INDEXED_SPLIT_MAGIC { @@ -896,7 +908,7 @@ impl DataSplit { message: format!("IndexedSplit version {iv} not supported"), }); } - let body = Self::read_v8_body(&mut cur)?; + let body = Self::read_body(&mut cur)?; let ranges_n = read_i32(&mut cur)?; if ranges_n < 0 { return Err(crate::Error::DataInvalid { @@ -946,7 +958,11 @@ impl DataSplit { /// Java `DataSplit#MAGIC` / `VERSION` for the serialize format. const SPLIT_MAGIC: i64 = -2394839472490812314; -const SPLIT_VERSION: i32 = 8; +const SPLIT_VERSION: i32 = 9; +/// Java bumped `DataSplit#VERSION` to 9 when it appended `_WRITE_COLS_SEQUENCES` to +/// `DataFileMeta.SCHEMA`. The split body is unchanged; only the per-file row grew, so +/// v8 bodies stay readable by decoding their files with the older row layout. +const SPLIT_VERSION_LEGACY_WRITE_COLS: i32 = 8; /// Java `SplitSerializer` frame: magic "SPLIT_V1", version, and the `DataSplit` type id. const SPLIT_SER_MAGIC: i64 = 0x53504C49545F5631; // "SPLIT_V1" @@ -1343,6 +1359,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } @@ -1444,7 +1461,11 @@ mod tests { // Generated by Apache Paimon Java 1.4.2 using BinaryRow.EMPTY_ROW and // DataSplit#serialize. The partition field is length 12: arity=0 plus // the 8-byte fixed part initialized by Java's BinaryRow static block. - let java_golden = [ + // + // The split carries no data files, so the only byte that the later version bump + // moved is the version int itself -- which makes this golden a check of the split + // body rather than of the per-file row layout. + let java_golden_v8 = [ 0xde, 0xc3, 0xd2, 0x30, 0x2c, 0x19, 0xec, 0x66, 0x00, 0x00, 0x00, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x0c, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x12, @@ -1462,8 +1483,14 @@ mod tests { .build() .unwrap(); + let mut expected = java_golden_v8; + expected[8..12].copy_from_slice(&SPLIT_VERSION.to_be_bytes()); + let bytes = split.serialize().expect("serialize"); - assert_eq!(bytes, java_golden); + assert_eq!(bytes, expected); + // A legacy body still reads, and re-serializing it normalizes to the current version. + let from_v8 = DataSplit::deserialize(&java_golden_v8).expect("deserialize v8"); + assert_eq!(from_v8.serialize().expect("reserialize v8"), expected); let restored = DataSplit::deserialize(&bytes).expect("deserialize"); assert_eq!(restored.snapshot_id(), split.snapshot_id()); assert_eq!(restored.partition().arity(), 0); @@ -1747,16 +1774,17 @@ mod tests { } #[test] - fn serialize_matches_datasplit_v8() { + fn serialize_matches_datasplit_v9() { // Golden generated by Java (paimon-core DataSplitCompatibleTest) for cross-language parity. - let expected = include_bytes!("goldens/datasplit_v8.bin"); - let split = sample_v8_split(); + let expected = include_bytes!("goldens/datasplit_v9.bin"); + let split = sample_v9_split(); assert_eq!(split.serialize().unwrap().as_slice(), &expected[..]); } - /// The fixture split whose Java-generated bytes live in `goldens/datasplit_v8.bin`. - /// Shared by the serialize and deserialize golden tests. - fn sample_v8_split() -> DataSplit { + /// The fixture split behind the Java-generated `goldens/datasplit_v*.bin`. The two + /// goldens describe the same split; only `_WRITE_COLS_SEQUENCES` differs, matching + /// Java's `testSerializerCompatibleV8` / `testSerializerCompatibleV9`. + fn sample_split(column_max_sequence_numbers: Option>) -> DataSplit { use chrono::DateTime; let mut pb = crate::spec::BinaryRowBuilder::new(1); @@ -1796,6 +1824,7 @@ mod tests { .map(|s| s.to_string()) .collect(), ), + column_max_sequence_numbers, }; DataSplitBuilder::new() @@ -1816,6 +1845,18 @@ mod tests { .unwrap() } + /// The v8 golden's split: its file predates `_WRITE_COLS_SEQUENCES`. + fn sample_v8_split() -> DataSplit { + sample_split(None) + } + + /// The v9 golden's split: the same file, carrying per-column maximum sequence numbers. + fn sample_v9_split() -> DataSplit { + sample_split(Some(vec![15, 100, 150, 200])) + } + + /// A v8 body stays readable: the split is identical apart from the field that + /// layout has no slot for. #[test] fn deserialize_matches_datasplit_v8_golden() { let golden = include_bytes!("goldens/datasplit_v8.bin"); @@ -1823,9 +1864,23 @@ mod tests { assert_eq!(split, sample_v8_split()); } + #[test] + fn deserialize_matches_datasplit_v9_golden() { + let golden = include_bytes!("goldens/datasplit_v9.bin"); + let split = DataSplit::deserialize(golden).unwrap(); + assert_eq!(split, sample_v9_split()); + } + + #[test] + fn deserialize_rejects_unsupported_version() { + let mut bytes = sample_v9_split().serialize().unwrap(); + bytes[8..12].copy_from_slice(&7i32.to_be_bytes()); + assert!(DataSplit::deserialize(&bytes).is_err()); + } + #[test] fn deserialize_round_trips_serialize() { - let split = sample_v8_split(); + let split = sample_v9_split(); assert_eq!( DataSplit::deserialize(&split.serialize().unwrap()).unwrap(), split @@ -1834,7 +1889,7 @@ mod tests { #[test] fn deserialize_rejects_trailing_bytes() { - let mut bytes = sample_v8_split().serialize().unwrap(); + let mut bytes = sample_v9_split().serialize().unwrap(); bytes.push(0xFF); assert!(DataSplit::deserialize(&bytes).is_err()); } @@ -2114,6 +2169,7 @@ mod tests { external_path: None, first_row_id: None, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index e5f09f9ea..7bf70936e 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -3265,6 +3265,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs index 4ff0c5a07..99c01f7d9 100644 --- a/crates/paimon/src/table/table_read.rs +++ b/crates/paimon/src/table/table_read.rs @@ -1509,6 +1509,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index b28915f65..601eeae90 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -2174,6 +2174,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } @@ -2556,6 +2557,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/vector_search_builder.rs b/crates/paimon/src/table/vector_search_builder.rs index 4692b6c12..497ccd986 100644 --- a/crates/paimon/src/table/vector_search_builder.rs +++ b/crates/paimon/src/table/vector_search_builder.rs @@ -4630,6 +4630,7 @@ mod tests { external_path: None, first_row_id, write_cols: None, + column_max_sequence_numbers: None, } } @@ -7019,6 +7020,7 @@ mod residual_positions_tests { external_path: None, first_row_id, write_cols: None, + column_max_sequence_numbers: None, } } diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs index a7ef0897e..572f4b3cd 100644 --- a/crates/paimon/src/table/vindex_index_build_builder.rs +++ b/crates/paimon/src/table/vindex_index_build_builder.rs @@ -1221,6 +1221,7 @@ mod tests { external_path: None, file_source: None, value_stats_cols: None, + column_max_sequence_numbers: None, } }