From 2d19085987991b61606b5d62b8b29db301424f66 Mon Sep 17 00:00:00 2001
From: shyjsarah <44659226+shyjsarah@users.noreply.github.com>
Date: Mon, 24 Aug 2026 01:15:08 -0700
Subject: [PATCH] fix: write manifest file metadata as Avro record

---
 crates/paimon/src/spec/manifest_entry.rs | 22 +++++++++++++++++++---
 crates/paimon/src/spec/objects_file.rs   | 12 +++++++++++-
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/crates/paimon/src/spec/manifest_entry.rs b/crates/paimon/src/spec/manifest_entry.rs
index 0e66891a..63473247 100644
--- a/crates/paimon/src/spec/manifest_entry.rs
+++ b/crates/paimon/src/spec/manifest_entry.rs
@@ -192,7 +192,7 @@ pub const MANIFEST_ENTRY_SCHEMA: &str = r#"["null", {
         {"name": "_PARTITION", "type": "bytes"},
         {"name": "_BUCKET", "type": "int"},
         {"name": "_TOTAL_BUCKETS", "type": "int"},
-        {"name": "_FILE", "type": ["null", {
+        {"name": "_FILE", "type": {
             "type": "record",
             "name": "record__FILE",
             "fields": [
@@ -233,14 +233,14 @@ pub const MANIFEST_ENTRY_SCHEMA: &str = r#"["null", {
                 {"name": "_FIRST_ROW_ID", "type": ["null", "long"], "default": null},
                 {"name": "_WRITE_COLS", "type": ["null", {"type": "array", "items": "string"}], "default": null}
             ]
-        }], "default": null}
+        }}
     ]
 }]"#;
 
 #[cfg(test)]
 mod tests {
     use super::{Identifier, MANIFEST_ENTRY_SCHEMA};
-    use crate::spec::avro::schema::WriterSchema;
+    use crate::spec::avro::schema::{FieldSchema, WriterSchema};
     use std::collections::HashSet;
 
     #[test]
@@ -265,6 +265,22 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_manifest_entry_file_schema_matches_java_record_type() {
+        let schema = WriterSchema::parse(MANIFEST_ENTRY_SCHEMA).unwrap();
+        let file = schema
+            .fields
+            .iter()
+            .find(|field| field.name == "_FILE")
+            .unwrap();
+
+        assert!(
+            !file.nullable,
+            "Java ManifestAvroReader requires _FILE to be a RECORD, not a nullable UNION"
+        );
+        assert!(matches!(file.schema, FieldSchema::Record(_)));
+    }
+
     fn ident(file_name: &str, level: i32) -> Identifier {
         Identifier {
             partition: vec![1, 2, 3],
diff --git a/crates/paimon/src/spec/objects_file.rs b/crates/paimon/src/spec/objects_file.rs
index 7f2c1436..7db066fa 100644
--- a/crates/paimon/src/spec/objects_file.rs
+++ b/crates/paimon/src/spec/objects_file.rs
@@ -190,7 +190,7 @@ mod tests {
     }
 
     #[test]
-    fn test_read_manifest_entry_with_legacy_rust_field_order() {
+    fn test_read_manifest_entry_with_legacy_rust_schema() {
         let mut schema: serde_json::Value = serde_json::from_str(MANIFEST_ENTRY_SCHEMA).unwrap();
         let fields = schema.as_array_mut().unwrap()[1]
             .as_object_mut()
@@ -199,6 +199,16 @@ mod tests {
             .unwrap()
             .as_array_mut()
             .unwrap();
+        let file = fields
+            .iter_mut()
+            .find(|field| field.get("name").and_then(|name| name.as_str()) == Some("_FILE"))
+            .unwrap()
+            .as_object_mut()
+            .unwrap();
+        let file_type = file.remove("type").unwrap();
+        file.insert("type".to_string(), serde_json::json!(["null", file_type]));
+        file.insert("default".to_string(), serde_json::Value::Null);
+
         let version = fields.remove(0);
         fields.push(version);
         let legacy_schema = serde_json::to_string(&schema).unwrap();