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 b89aa34e fix(table): accept DataSplit version 9 and mirror
_WRITE_COLS_SEQUENCES (#745)
b89aa34e is described below
commit b89aa34e76e28b7fc15d42203fd935999ca1ce46
Author: Junrui Lee <[email protected]>
AuthorDate: Wed Aug 26 10:24:40 2026 +0800
fix(table): accept DataSplit version 9 and mirror _WRITE_COLS_SEQUENCES
(#745)
---
bindings/python/tests/test_read.py | 4 +-
.../paimon/src/spec/avro/manifest_entry_decode.rs | 55 +++++++
crates/paimon/src/spec/data_file.rs | 179 +++++++++++++++++++--
crates/paimon/src/spec/manifest.rs | 1 +
crates/paimon/src/spec/manifest_entry.rs | 3 +-
crates/paimon/src/spec/objects_file.rs | 66 ++++++++
crates/paimon/src/table/bin_pack.rs | 1 +
crates/paimon/src/table/data_evolution_reader.rs | 1 +
crates/paimon/src/table/data_evolution_writer.rs | 1 +
crates/paimon/src/table/data_file_reader.rs | 3 +
crates/paimon/src/table/data_file_writer.rs | 1 +
crates/paimon/src/table/format_table_scan.rs | 1 +
.../paimon/src/table/global_index_drop_builder.rs | 1 +
crates/paimon/src/table/goldens/datasplit_v9.bin | Bin 0 -> 738 bytes
crates/paimon/src/table/goldens/split_v1_data.bin | Bin 896 -> 912 bytes
.../paimon/src/table/goldens/split_v1_indexed.bin | Bin 961 -> 977 bytes
crates/paimon/src/table/kv_file_reader.rs | 1 +
crates/paimon/src/table/kv_file_writer.rs | 1 +
.../paimon/src/table/lumina_index_build_builder.rs | 1 +
.../paimon/src/table/merge_tree_split_generator.rs | 1 +
crates/paimon/src/table/partition_stat.rs | 1 +
.../paimon/src/table/pk_full_text_bucket_search.rs | 1 +
.../paimon/src/table/pk_full_text_bucket_state.rs | 1 +
crates/paimon/src/table/pk_full_text_read.rs | 1 +
crates/paimon/src/table/pk_full_text_scan.rs | 1 +
.../paimon/src/table/pk_vector_data_file_reader.rs | 1 +
.../src/table/pk_vector_indexed_split_read.rs | 2 +
crates/paimon/src/table/pk_vector_orchestrator.rs | 2 +
crates/paimon/src/table/pk_vector_position_read.rs | 1 +
crates/paimon/src/table/pk_vector_scan.rs | 1 +
crates/paimon/src/table/postpone_file_writer.rs | 1 +
crates/paimon/src/table/referenced_files.rs | 3 +
.../src/table/sorted_global_index_build_builder.rs | 1 +
crates/paimon/src/table/source.rs | 106 +++++++++---
crates/paimon/src/table/table_commit.rs | 1 +
crates/paimon/src/table/table_read.rs | 1 +
crates/paimon/src/table/table_scan.rs | 2 +
crates/paimon/src/table/vector_search_builder.rs | 2 +
.../paimon/src/table/vindex_index_build_builder.rs | 1 +
39 files changed, 412 insertions(+), 39 deletions(-)
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 49ed926d..c53d656a 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 df251bd3..0de57aae 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<String> = None;
let mut first_row_id: Option<i64> = None;
let mut write_cols: Option<Vec<String>> = None;
+ let mut column_max_sequence_numbers: Option<Vec<i64>> = 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<Vec<String>> {
Ok(result)
}
+/// Decode an Avro `array<long>` whose items are plain (non-union) longs,
mirroring
+/// [`decode_string_array`]. Paimon declares `_WRITE_COLS_SEQUENCES` as
+/// `array<bigint not null>`, so items carry no per-element union index.
+fn decode_long_array(cursor: &mut AvroCursor) -> crate::Result<Vec<i64>> {
+ 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<Option<Vec<i64>>> {
+ 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<Option<i64>> {
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 314c6d15..97e2d2d0 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<Vec<String>>,
+
+ /// 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<Vec<i64>>,
}
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<i64>) {
match v {
Some(x) => b.write_long(pos, x),
@@ -164,6 +204,26 @@ fn opt_long(b: &mut BinaryRowBuilder, pos: usize, v:
Option<i64>) {
}
}
+/// Write an `array<bigint not null>` field, or a null at `pos`.
+fn opt_long_array(b: &mut BinaryRowBuilder, pos: usize, v: &Option<Vec<i64>>) {
+ match v {
+ Some(a) => {
+ let elements: Vec<Option<i64>> =
a.iter().copied().map(Some).collect();
+ b.write_bytes(pos, &serialize_binary_array_long(&elements));
+ }
+ None => b.set_null_at(pos),
+ }
+}
+
+/// Read an `array<bigint not null>` 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<Vec<i64>> {
+ crate::spec::deserialize_binary_array_long(row.get_binary(pos)?)?
+ .into_iter()
+ .map(|v| v.ok_or_else(|| data_file_err("array<bigint not null> has a
null element")))
+ .collect()
+}
+
fn opt_str_array(b: &mut BinaryRowBuilder, pos: usize, v:
&Option<Vec<String>>) {
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<Vec<u8>> {
- 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<DataFileMeta> {
+ /// 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<DataFileMeta> {
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 c25dba9a..ed2fa162 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 63473247..484f2bd9 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 7db066fa..d5d0d9fe 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<Vec<i64>>,
+ ) -> 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::<ManifestEntry>(&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::<ManifestEntry>(&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::<ManifestEntry>(&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 b4dc178a..8c582695 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 11a4cb96..a456505a 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 078a2a12..d8b66e86 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 c9e2e1a7..526f8c7c 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 75f17c6f..76f49a0d 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 5a6a00c8..1638c0a6 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 80a0ad7f..6bc70e72 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 00000000..2bfd4ac1
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 6cbade9c..9d2f6c08 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 0d20df10..eab6fbe8 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 d922c267..88b1bf80 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 b5e94f54..bf1cd9aa 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 e98aedbe..7bed2cc3 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 cceee451..6f33d2be 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 ae5e5580..caa8c4a7 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 298733f0..d774e237 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 f9122ea1..48ad4b21 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 3c5a34b1..63b58f84 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 f6093858..5a677655 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 9fadaf2b..c3087beb 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 c6158108..d234ddca 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 1a816bd9..ac0d3ab3 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 29446822..06721816 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 00bec032..cfc528b2 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 8ee83cba..978649cd 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 b602bbfd..851e8719 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 f0657919..94fc3165 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 c1e697ec..38bfdb89 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<Vec<u8>> {
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<DataSplit> {
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<DataSplit> {
+ fn read_body(cur: &mut &[u8]) -> crate::Result<DataSplit> {
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<DataSplit>
{
+ /// 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<DataSplit> {
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<Vec<i64>>) ->
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 e5f09f9e..7bf70936 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 4ff0c5a0..99c01f7d 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 b28915f6..601eeae9 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 4692b6c1..497ccd98 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 a7ef0897..572f4b3c 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,
}
}
