diff --git a/crates/paimon/src/spec/avro/decode_helpers.rs b/crates/paimon/src/spec/avro/decode_helpers.rs index 38165329b..698603150 100644 --- a/crates/paimon/src/spec/avro/decode_helpers.rs +++ b/crates/paimon/src/spec/avro/decode_helpers.rs @@ -47,6 +47,22 @@ pub(crate) fn read_long_field(cursor: &mut AvroCursor, nullable: bool) -> crate: cursor.read_long() } +/// Reads a nullable long, preserving the null/present distinction. +/// Returns `None` for the null branch of a `["null", "long"]` union (a +/// non-nullable field is always `Some`). +pub(crate) fn read_optional_long( + cursor: &mut AvroCursor, + nullable: bool, +) -> crate::Result> { + if nullable { + let idx = cursor.read_union_index()?; + if idx == 0 { + return Ok(None); + } + } + Ok(Some(cursor.read_long()?)) +} + pub(crate) fn read_bytes_field(cursor: &mut AvroCursor, nullable: bool) -> crate::Result> { if nullable { let idx = cursor.read_union_index()?; 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 50cfece1c..f0fb0c696 100644 --- a/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs +++ b/crates/paimon/src/spec/avro/index_manifest_entry_decode.rs @@ -19,9 +19,9 @@ use super::cursor::AvroCursor; use super::decode::{neg_count_to_usize, AvroRecordDecode}; use super::decode_helpers::{ extract_record_schema, normalize_partition, read_bytes_field, read_int_field, read_long_field, - read_nullable_string_field, read_string_field, + read_nullable_string_field, read_optional_long, read_string_field, }; -use super::schema::{skip_nullable_field, WriterSchema}; +use super::schema::{skip_nullable_field, FieldSchema, WriterSchema}; use crate::spec::index_manifest::IndexManifestEntry; use crate::spec::manifest_common::FileKind; use crate::spec::{DeletionVectorMeta, GlobalIndexMeta, IndexFileMeta}; @@ -64,7 +64,12 @@ impl AvroRecordDecode for IndexManifestEntry { "_FILE_SIZE" => file_size = Some(read_long_field(cursor, field.nullable)?), "_ROW_COUNT" => row_count = Some(read_long_field(cursor, field.nullable)?), "_DELETIONS_VECTORS_RANGES" | "_DELETION_VECTORS_RANGES" => { - deletion_vectors_ranges = decode_nullable_dv_ranges(cursor, field.nullable)?; + deletion_vectors_ranges = decode_nullable_dv_ranges( + cursor, + &field.name, + &field.schema, + field.nullable, + )?; } "_EXTERNAL_PATH" => { external_path = read_nullable_string_field(cursor, field.nullable)?; @@ -95,8 +100,46 @@ impl AvroRecordDecode for IndexManifestEntry { } } +/// Peel `array<["null", record]>` down to the item record's writer schema. +/// +/// `WriterSchema::parse` unwraps a *field*'s nullable union, so the outer +/// `["null", array]` is already gone, but array items keep theirs — Java builds +/// the element with `RowType.of`, which is nullable, so the schema here is +/// `Array(Union([Null, Record]))` and `extract_record_schema` alone cannot reach +/// the record. +/// +/// `f0`, `f1` and `f2` hold the data file name, offset and length, and Java has +/// always declared all three non-null. `f0` is the map key, so defaulting it +/// would silently collapse every item of an entry onto one key; reject the +/// schema instead, which is also what the serde reader does with such a file. +fn dv_item_record_schema<'a>( + field_name: &str, + schema: &'a FieldSchema, +) -> crate::Result<&'a WriterSchema> { + let err = |detail: String| crate::Error::UnexpectedError { + message: format!("avro decode: {field_name} {detail}"), + source: None, + }; + let record = match schema { + FieldSchema::Array(item) => match item.as_ref() { + FieldSchema::Union(branches) => branches.iter().find_map(extract_record_schema), + _ => None, + }, + _ => None, + } + .ok_or_else(|| err("is not an array of nullable records".to_owned()))?; + for required in ["f0", "f1", "f2"] { + if !record.fields.iter().any(|f| f.name == required) { + return Err(err(format!("item record has no `{required}` field"))); + } + } + Ok(record) +} + fn decode_nullable_dv_ranges( cursor: &mut AvroCursor, + field_name: &str, + schema: &FieldSchema, nullable: bool, ) -> crate::Result>> { if nullable { @@ -105,7 +148,13 @@ fn decode_nullable_dv_ranges( return Ok(None); } } - // Array of nullable records + // `_CARDINALITY` only exists in writer schemas from 1.0.0 on, where Java + // pointed this field at `DeletionVectorMeta.SCHEMA` (#4699); 0.8.0 through + // 0.9.x declared the item as `RowType.of(STRING, INT, INT)`, i.e. `f0`/`f1`/ + // `f2` and nothing else. Walk the writer's own field list so an older record + // does not consume the next item's bytes, and so a future field appended to + // `DeletionVectorMeta.SCHEMA` is skipped rather than misread. + let item_schema = dv_item_record_schema(field_name, schema)?; let mut map = IndexMap::new(); loop { let count = cursor.read_long()?; @@ -124,18 +173,21 @@ fn decode_nullable_dv_ranges( if item_idx == 0 { continue; } - // Record fields: f0 (string), f1 (int), f2 (int), _CARDINALITY (nullable long) - let f0 = cursor.read_string()?.to_string(); - let f1 = cursor.read_int()?; - let f2 = cursor.read_int()?; - let cardinality = { - let c_idx = cursor.read_union_index()?; - if c_idx == 0 { - None - } else { - Some(cursor.read_long()?) + // `f0`/`f1`/`f2` are known present — `dv_item_record_schema` rejects + // an item record missing any of them — so these defaults never survive. + let mut f0 = String::new(); + let mut f1 = 0; + let mut f2 = 0; + let mut cardinality = None; + for field in &item_schema.fields { + match field.name.as_str() { + "f0" => f0 = read_string_field(cursor, field.nullable)?, + "f1" => f1 = read_int_field(cursor, field.nullable)?, + "f2" => f2 = read_int_field(cursor, field.nullable)?, + "_CARDINALITY" => cardinality = read_optional_long(cursor, field.nullable)?, + _ => skip_nullable_field(cursor, &field.schema, field.nullable)?, } - }; + } map.insert( f0, DeletionVectorMeta { diff --git a/crates/paimon/src/spec/avro/manifest_file_meta_decode.rs b/crates/paimon/src/spec/avro/manifest_file_meta_decode.rs index fa5cd09eb..3151e5557 100644 --- a/crates/paimon/src/spec/avro/manifest_file_meta_decode.rs +++ b/crates/paimon/src/spec/avro/manifest_file_meta_decode.rs @@ -18,7 +18,8 @@ use super::cursor::AvroCursor; use super::decode::{neg_count_to_usize, AvroRecordDecode}; use super::decode_helpers::{ - extract_record_schema, read_bytes_field, read_int_field, read_long_field, read_string_field, + extract_record_schema, read_bytes_field, read_int_field, read_long_field, read_optional_long, + read_string_field, }; use super::schema::{skip_nullable_field, FieldSchema, WriterSchema}; use crate::spec::stats::BinaryTableStats; @@ -94,16 +95,6 @@ fn read_optional_int(cursor: &mut AvroCursor, nullable: bool) -> crate::Result crate::Result> { - if nullable { - let idx = cursor.read_union_index()?; - if idx == 0 { - return Ok(None); - } - } - Ok(Some(cursor.read_long()?)) -} - /// Decode a nullable BinaryTableStats: union ["null", record] or direct record. pub(crate) fn decode_nullable_binary_table_stats( cursor: &mut AvroCursor, diff --git a/crates/paimon/src/spec/index_manifest.rs b/crates/paimon/src/spec/index_manifest.rs index 4050ece53..c6a7f011f 100644 --- a/crates/paimon/src/spec/index_manifest.rs +++ b/crates/paimon/src/spec/index_manifest.rs @@ -596,4 +596,143 @@ mod tests { ); assert_eq!(decoded[0].index_file.external_path, None); } + + fn dv_entry(file_name: &str, ranges: &[(&str, i32, i32)]) -> IndexManifestEntry { + IndexManifestEntry { + version: 1, + kind: FileKind::Add, + partition: vec![0, 0, 0, 0], + bucket: 0, + index_file: IndexFileMeta { + index_type: "DELETION_VECTORS".into(), + file_name: file_name.into(), + file_size: 35, + row_count: 1, + deletion_vectors_ranges: Some( + ranges + .iter() + .map(|(data_file, offset, length)| { + ( + (*data_file).to_string(), + DeletionVectorMeta { + offset: *offset, + length: *length, + cardinality: None, + }, + ) + }) + .collect(), + ), + external_path: None, + global_index_meta: None, + }, + } + } + + /// `INDEX_MANIFEST_ENTRY_SCHEMA` with one field removed from the deletion-vector + /// item record. Removing `_CARDINALITY` reproduces the shape 0.8.0 through 0.9.x + /// wrote, where Java declared the item as `RowType.of(STRING, INT, INT)`. + /// Derived from the current schema rather than hand-written so the two cannot + /// drift apart. + fn schema_without_dv_item_field(removed: &str) -> String { + let mut schema: serde_json::Value = + serde_json::from_str(INDEX_MANIFEST_ENTRY_SCHEMA).unwrap(); + let item_fields = schema + .get_mut("fields") + .unwrap() + .as_array_mut() + .unwrap() + .iter_mut() + .find(|field| { + field.get("name").and_then(|name| name.as_str()) + == Some("_DELETIONS_VECTORS_RANGES") + }) + .unwrap() + .get_mut("type") + .unwrap()[1] + .get_mut("items") + .unwrap()[1] + .get_mut("fields") + .unwrap() + .as_array_mut() + .unwrap(); + let before = item_fields.len(); + item_fields + .retain(|field| field.get("name").and_then(|name| name.as_str()) != Some(removed)); + assert_eq!( + item_fields.len(), + before - 1, + "field must exist to be removed" + ); + serde_json::to_string(&schema).unwrap() + } + + #[test] + fn dv_item_record_without_cardinality_field_decodes_as_none() { + // Two entries carrying two deletion-vector items each. Reading a + // `_CARDINALITY` the writer never wrote steals the *next* item's bytes, so a + // single item would only ever run off the end of the block and would prove + // nothing about the misalignment. + let entries = vec![ + dv_entry( + "idx-0", + &[("data-0.parquet", 1, 26), ("data-1.parquet", 27, 30)], + ), + dv_entry( + "idx-1", + &[("data-2.parquet", 3, 11), ("data-3.parquet", 14, 19)], + ), + ]; + let bytes = crate::spec::to_avro_bytes_with_compression( + &schema_without_dv_item_field("_CARDINALITY"), + &entries, + crate::spec::DEFAULT_AVRO_COMPRESSION, + ) + .unwrap(); + + let decoded = IndexManifest::read_from_bytes(&bytes).unwrap(); + assert_eq!(decoded, entries); + // `IndexMap` compares order-insensitively, but order is exactly what a + // misaligned cursor destroys, so pin it separately. + let keys: Vec<&str> = decoded[1] + .index_file + .deletion_vectors_ranges + .as_ref() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + assert_eq!(keys, ["data-2.parquet", "data-3.parquet"]); + // The serde reader already read these bytes correctly, because the + // deletion-vector helper struct marks `_CARDINALITY` `#[serde(default)]`. + // The two readers of one file must not disagree. + assert_eq!( + crate::spec::from_avro_bytes::(&bytes).unwrap(), + entries + ); + } + + #[test] + fn dv_item_record_without_f0_field_is_rejected() { + // `f0` is the map key and Java has always declared it non-null, so a writer + // schema without it is not something to guess a default for: every item of + // an entry would collapse onto one key. The serde reader rejects such a file + // too, because its `f0` has no `#[serde(default)]`. + let entries = vec![dv_entry("idx-0", &[("data-0.parquet", 1, 26)])]; + let bytes = crate::spec::to_avro_bytes_with_compression( + &schema_without_dv_item_field("f0"), + &entries, + crate::spec::DEFAULT_AVRO_COMPRESSION, + ) + .unwrap(); + + let err = IndexManifest::read_from_bytes(&bytes) + .unwrap_err() + .to_string(); + assert!( + err.contains("_DELETIONS_VECTORS_RANGES item record has no `f0` field"), + "{err}" + ); + assert!(crate::spec::from_avro_bytes::(&bytes).is_err()); + } }