diff --git a/crates/paimon/src/spec/avro/manifest_entry_decode.rs b/crates/paimon/src/spec/avro/manifest_entry_decode.rs index 0de57aaec..b4af8a36f 100644 --- a/crates/paimon/src/spec/avro/manifest_entry_decode.rs +++ b/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>> { 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..6beff6166 100644 --- a/crates/paimon/src/spec/avro/manifest_file_meta_decode.rs +++ b/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 = None; let mut min_row_id: Option = None; let mut max_row_id: Option = None; + let mut extra_files: Option> = 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, )) } } diff --git a/crates/paimon/src/spec/manifest_file_meta.rs b/crates/paimon/src/spec/manifest_file_meta.rs index 540f44090..adcc95c6f 100644 --- a/crates/paimon/src/spec/manifest_file_meta.rs +++ b/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, + + /// 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>, } 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>) -> 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, min_row_id: Option, max_row_id: Option, + extra_files: Option>, ) -> 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} ] }]"#; diff --git a/crates/paimon/src/spec/manifest_list.rs b/crates/paimon/src/spec/manifest_list.rs index 6c59c684b..638e8f8b0 100644 --- a/crates/paimon/src/spec/manifest_list.rs +++ b/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); } } diff --git a/crates/paimon/src/spec/objects_file.rs b/crates/paimon/src/spec/objects_file.rs index 56f2fe4d2..78421b0a0 100644 --- a/crates/paimon/src/spec/objects_file.rs +++ b/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::(&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] diff --git a/crates/paimon/src/table/referenced_files.rs b/crates/paimon/src/table/referenced_files.rs index d4bb617bb..812cf566a 100644 --- a/crates/paimon/src/table/referenced_files.rs +++ b/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::>(); + if !manifest_extra_names.is_empty() { + let manifest_extra_names = manifest_extra_names.into_iter().collect::>(); + let manifest_extra_paths = manifest_extra_names + .iter() + .map(|name| manifest_sm.manifest_path(name)) + .collect::>(); + 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 = 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]