fix(spec): decode pre-1.0 deletion-vector metas via the writer schema

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<Option<i64>> {
+    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<Vec<u8>> {
     if nullable {
         let idx = cursor.read_union_index()?;

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<Option<IndexMap<String, DeletionVectorMeta>>> {
     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 {

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<O
     Ok(Some(cursor.read_int()?))
 }
 
-fn read_optional_long(cursor: &mut AvroCursor, nullable: bool) -> crate::Result<Option<i64>> {
-    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,

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::<IndexManifestEntry>(&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::<IndexManifestEntry>(&bytes).is_err());
+    }
 }