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 ae806e47 fix(spec): preserve manifest extra files (#848)
ae806e47 is described below
commit ae806e47582130aefd0e0cdf88c03130808eb4f3
Author: jianguotian <[email protected]>
AuthorDate: Thu Sep 17 08:08:23 2026 +0800
fix(spec): preserve manifest extra files (#848)
---
.../paimon/src/spec/avro/manifest_entry_decode.rs | 2 +-
.../src/spec/avro/manifest_file_meta_decode.rs | 6 ++++
crates/paimon/src/spec/manifest_file_meta.rs | 31 +++++++++++++++-
crates/paimon/src/spec/manifest_list.rs | 4 ++-
crates/paimon/src/spec/objects_file.rs | 6 ++++
crates/paimon/src/table/referenced_files.rs | 41 ++++++++++++++++++++--
6 files changed, 85 insertions(+), 5 deletions(-)
diff --git a/crates/paimon/src/spec/avro/manifest_entry_decode.rs
b/crates/paimon/src/spec/avro/manifest_entry_decode.rs
index 0de57aae..b4af8a36 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<Option<Vec<String>>> {
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 fa5cd09e..6beff616 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<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,
))
}
}
diff --git a/crates/paimon/src/spec/manifest_file_meta.rs
b/crates/paimon/src/spec/manifest_file_meta.rs
index 540f4409..adcc95c6 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<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}
]
}]"#;
diff --git a/crates/paimon/src/spec/manifest_list.rs
b/crates/paimon/src/spec/manifest_list.rs
index 6c59c684..638e8f8b 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 56f2fe4d..78421b0a 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::<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]
diff --git a/crates/paimon/src/table/referenced_files.rs
b/crates/paimon/src/table/referenced_files.rs
index d4bb617b..812cf566 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::<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]
