fix(spec): preserve manifest extra files

crates/paimon/src/spec/avro/manifest_entry_decode.rs

@@ -409,7 +409,7 @@ fn decode_nullable_string(
     Ok(Some(cursor.read_string()?.to_string()))
 }
 
-fn decode_nullable_string_array(
+pub(super) fn decode_nullable_string_array(
     cursor: &mut AvroCursor,
     nullable: bool,
 ) -> crate::Result<Option<Vec<String>>> {

crates/paimon/src/spec/avro/manifest_file_meta_decode.rs

@@ -20,6 +20,7 @@ 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,
 };
+use super::manifest_entry_decode::decode_nullable_string_array;
 use super::schema::{skip_nullable_field, FieldSchema, WriterSchema};
 use crate::spec::stats::BinaryTableStats;
 use crate::spec::ManifestFileMeta;
@@ -39,6 +40,7 @@ impl AvroRecordDecode for ManifestFileMeta {
         let mut max_level: Option<i32> = None;
         let mut min_row_id: Option<i64> = None;
         let mut max_row_id: Option<i64> = None;
+        let mut extra_files: Option<Vec<String>> = None;
 
         for field in &writer_schema.fields {
             match field.name.as_str() {
@@ -62,6 +64,9 @@ impl AvroRecordDecode for ManifestFileMeta {
                 "_MAX_LEVEL" => max_level = read_optional_int(cursor, field.nullable)?,
                 "_MIN_ROW_ID" => min_row_id = read_optional_long(cursor, field.nullable)?,
                 "_MAX_ROW_ID" => max_row_id = read_optional_long(cursor, field.nullable)?,
+                "_EXTRA_FILES" => {
+                    extra_files = decode_nullable_string_array(cursor, field.nullable)?
+                }
                 _ => skip_nullable_field(cursor, &field.schema, field.nullable)?,
             }
         }
@@ -80,6 +85,7 @@ impl AvroRecordDecode for ManifestFileMeta {
             max_level,
             min_row_id,
             max_row_id,
+            extra_files,
         ))
     }
 }

crates/paimon/src/spec/manifest_file_meta.rs

@@ -103,6 +103,17 @@ pub struct ManifestFileMeta {
         skip_serializing_if = "Option::is_none"
     )]
     max_row_id: Option<i64>,
+
+    /// Files owned by this manifest and sharing its lifecycle.
+    ///
+    /// `None` preserves the distinction between legacy manifest lists (where the
+    /// field is absent) and an explicitly empty list written by a newer writer.
+    #[serde(
+        rename = "_EXTRA_FILES",
+        default,
+        skip_serializing_if = "Option::is_none"
+    )]
+    extra_files: Option<Vec<String>>,
 }
 
 impl ManifestFileMeta {
@@ -184,6 +195,12 @@ impl ManifestFileMeta {
         self.max_row_id
     }
 
+    /// Get files owned by this manifest, if the metadata was recorded.
+    #[inline]
+    pub fn extra_files(&self) -> Option<&[String]> {
+        self.extra_files.as_deref()
+    }
+
     /// Attach bucket / level statistics aggregated from manifest entries.
     ///
     /// Use this in writers that have access to the entries that the manifest covers.
@@ -214,6 +231,14 @@ impl ManifestFileMeta {
         self
     }
 
+    /// Attach files whose lifecycle is owned by this manifest.
+    #[inline]
+    #[must_use]
+    pub fn with_extra_files(mut self, extra_files: Option<Vec<String>>) -> Self {
+        self.extra_files = extra_files;
+        self
+    }
+
     #[inline]
     pub fn new(
         file_name: String,
@@ -237,6 +262,7 @@ impl ManifestFileMeta {
             max_level: None,
             min_row_id: None,
             max_row_id: None,
+            extra_files: None,
         }
     }
 
@@ -256,6 +282,7 @@ impl ManifestFileMeta {
         max_level: Option<i32>,
         min_row_id: Option<i64>,
         max_row_id: Option<i64>,
+        extra_files: Option<Vec<String>>,
     ) -> ManifestFileMeta {
         Self {
             version,
@@ -271,6 +298,7 @@ impl ManifestFileMeta {
             max_level,
             min_row_id,
             max_row_id,
+            extra_files,
         }
     }
 }
@@ -301,7 +329,8 @@ pub const MANIFEST_FILE_META_SCHEMA: &str = r#"["null", {
         {"name": "_MIN_LEVEL", "type": ["null", "int"], "default": null},
         {"name": "_MAX_LEVEL", "type": ["null", "int"], "default": null},
         {"name": "_MIN_ROW_ID", "type": ["null", "long"], "default": null},
-        {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": null}
+        {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": null},
+        {"name": "_EXTRA_FILES", "type": ["null", {"type": "array", "items": "string"}], "default": null}
     ]
 }]"#;
 

crates/paimon/src/spec/manifest_list.rs

@@ -97,7 +97,8 @@ mod tests {
                 0,
                 BinaryTableStats::new(value_bytes.clone(), value_bytes.clone(), vec![Some(3)]),
                 1,
-            ),
+            )
+            .with_extra_files(Some(Vec::new())),
         ];
 
         ManifestList::write(&file_io, path, &original)
@@ -270,5 +271,6 @@ mod tests {
         assert_eq!(decoded[0].max_bucket(), None);
         assert_eq!(decoded[0].min_level(), None);
         assert_eq!(decoded[0].max_level(), None);
+        assert_eq!(decoded[0].extra_files(), None);
     }
 }

crates/paimon/src/spec/objects_file.rs

@@ -199,6 +199,7 @@ mod tests {
                 "_MAX_LEVEL",
                 "_MIN_ROW_ID",
                 "_MAX_ROW_ID",
+                "_EXTRA_FILES",
             ],
         );
         assert_record_field_order(
@@ -228,12 +229,17 @@ mod tests {
             None,
             Some(100),
             Some(199),
+            Some(vec!["manifest-row-tracking-0.idx".to_string()]),
         )];
         let bytes = to_avro_bytes(MANIFEST_FILE_META_SCHEMA, &original).unwrap();
         let decoded = from_avro_bytes::<ManifestFileMeta>(&bytes).unwrap();
         assert_eq!(original, decoded);
         assert_eq!(decoded[0].min_row_id(), Some(100));
         assert_eq!(decoded[0].max_row_id(), Some(199));
+        assert_eq!(
+            decoded[0].extra_files(),
+            Some(["manifest-row-tracking-0.idx".to_string()].as_slice())
+        );
     }
 
     #[test]

crates/paimon/src/table/referenced_files.rs

@@ -19,7 +19,7 @@
 //!
 //! Reference: [LocalOrphanFilesClean](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/operation/LocalOrphanFilesClean.java)
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Mutex;
 #[cfg(test)]
 use std::sync::{
@@ -407,6 +407,34 @@ async fn collect_snapshot_files(
             .or_insert(meta.file_size());
     }
 
+    // Manifest-owned sidecars live in the manifest directory and must remain
+    // referenced for exactly as long as their owner manifest. Java Paimon may
+    // add these files even when this Rust process only reads and rewrites the
+    // manifest list metadata.
+    let manifest_extra_names = all_manifest_metas
+        .iter()
+        .flat_map(|meta| meta.extra_files().unwrap_or_default())
+        .cloned()
+        .collect::<HashSet<_>>();
+    if !manifest_extra_names.is_empty() {
+        let manifest_extra_names = manifest_extra_names.into_iter().collect::<Vec<_>>();
+        let manifest_extra_paths = manifest_extra_names
+            .iter()
+            .map(|name| manifest_sm.manifest_path(name))
+            .collect::<Vec<_>>();
+        let sizes = try_join_all(
+            manifest_extra_paths
+                .iter()
+                .map(|path| try_stat_file_size(file_io, path)),
+        )
+        .await?;
+        for (name, size) in manifest_extra_names.into_iter().zip(sizes) {
+            if size > 0 {
+                file_set.manifest_files.entry(name).or_insert(size);
+            }
+        }
+    }
+
     // Read manifest files to get data file entries, using cache by full path
     let manifest_paths: Vec<String> = all_manifest_metas
         .iter()
@@ -953,6 +981,7 @@ mod tests {
         let external_dir = "memory:/external_sidecar_references";
         let sidecar_name = "data-0.row.index";
         let sidecar_content = "sidecar-bytes";
+        let manifest_sidecar_name = "manifest-external-sidecar-0.index";
         let file_io = test_file_io();
 
         file_io
@@ -969,6 +998,12 @@ mod tests {
             sidecar_content,
         )
         .await;
+        write_test_file(
+            &file_io,
+            &format!("{table_path}/manifest/{manifest_sidecar_name}"),
+            "manifest-sidecar-bytes",
+        )
+        .await;
 
         let manifest_name = "manifest-external-sidecar-0";
         let manifest_path = format!("{table_path}/manifest/{manifest_name}");
@@ -1009,7 +1044,8 @@ mod tests {
             0,
             BinaryTableStats::empty(),
             0,
-        );
+        )
+        .with_extra_files(Some(vec![manifest_sidecar_name.to_string()]));
         ManifestList::write(&file_io, &manifest_list_path, &[manifest_meta])
             .await
             .unwrap();
@@ -1040,6 +1076,7 @@ mod tests {
         let total = result.iter().find(|r| r.source == "total").unwrap();
         assert_eq!(total.data_file_count, 2);
         assert_eq!(total.data_file_size, 100 + sidecar_content.len() as i64);
+        assert_eq!(total.manifest_file_count, 4);
     }
 
     #[tokio::test]