This is an automated email from the ASF dual-hosted git repository.
jerry-024 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 b36d533f fix: align manifest entry schema with Java Paimon (#742)
b36d533f is described below
commit b36d533f744d01b8988e2240a022faf78325d0c4
Author: shyjsarah <[email protected]>
AuthorDate: Mon Aug 24 14:16:02 2026 +0800
fix: align manifest entry schema with Java Paimon (#742)
---
crates/paimon/src/spec/manifest_entry.rs | 38 ++++++++--
crates/paimon/src/spec/objects_file.rs | 115 +++++++++++++++++++------------
crates/paimon/src/table/table_commit.rs | 42 ++++++++++-
3 files changed, 142 insertions(+), 53 deletions(-)
diff --git a/crates/paimon/src/spec/manifest_entry.rs
b/crates/paimon/src/spec/manifest_entry.rs
index 194fc314..0e66891a 100644
--- a/crates/paimon/src/spec/manifest_entry.rs
+++ b/crates/paimon/src/spec/manifest_entry.rs
@@ -55,6 +55,10 @@ impl Hash for Identifier {
/// Impl Reference:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java>
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ManifestEntry {
+ // Java's VersionedObjectSerializer requires the version field at position
0.
+ #[serde(rename = "_VERSION")]
+ version: i32,
+
#[serde(rename = "_KIND")]
kind: FileKind,
@@ -69,9 +73,6 @@ pub struct ManifestEntry {
#[serde(rename = "_FILE")]
pub(crate) file: DataFileMeta,
-
- #[serde(rename = "_VERSION")]
- version: i32,
}
#[allow(dead_code)]
@@ -151,12 +152,12 @@ impl ManifestEntry {
version: i32,
) -> Self {
ManifestEntry {
+ version,
kind,
partition,
bucket,
total_buckets,
file,
- version,
}
}
@@ -186,6 +187,7 @@ pub const MANIFEST_ENTRY_SCHEMA: &str = r#"["null", {
"name": "record",
"namespace": "org.apache.paimon.avro.generated",
"fields": [
+ {"name": "_VERSION", "type": "int"},
{"name": "_KIND", "type": "int"},
{"name": "_PARTITION", "type": "bytes"},
{"name": "_BUCKET", "type": "int"},
@@ -231,16 +233,38 @@ 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},
- {"name": "_VERSION", "type": "int"}
+ }], "default": null}
]
}]"#;
#[cfg(test)]
mod tests {
- use super::Identifier;
+ use super::{Identifier, MANIFEST_ENTRY_SCHEMA};
+ use crate::spec::avro::schema::WriterSchema;
use std::collections::HashSet;
+ #[test]
+ fn test_manifest_entry_schema_matches_java_field_order() {
+ let schema = WriterSchema::parse(MANIFEST_ENTRY_SCHEMA).unwrap();
+ let field_names = schema
+ .fields
+ .iter()
+ .map(|field| field.name.as_str())
+ .collect::<Vec<_>>();
+
+ assert_eq!(
+ field_names,
+ vec![
+ "_VERSION",
+ "_KIND",
+ "_PARTITION",
+ "_BUCKET",
+ "_TOTAL_BUCKETS",
+ "_FILE"
+ ]
+ );
+ }
+
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 2b16f55c..7f2c1436 100644
--- a/crates/paimon/src/spec/objects_file.rs
+++ b/crates/paimon/src/spec/objects_file.rs
@@ -80,6 +80,7 @@ pub(crate) fn avro_codec(compression: &str) ->
crate::Result<Codec> {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::spec::avro::from_avro_bytes_fast;
use crate::spec::manifest_common::FileKind;
use crate::spec::manifest_entry::{ManifestEntry, MANIFEST_ENTRY_SCHEMA};
use crate::spec::manifest_file_meta::MANIFEST_FILE_META_SCHEMA;
@@ -87,6 +88,54 @@ mod tests {
use crate::spec::{DataFileMeta, ManifestFileMeta};
use chrono::{DateTime, Utc};
+ fn manifest_entry() -> ManifestEntry {
+ let value_bytes = vec![
+ 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 49, 0, 0, 0, 0, 0, 0, 129, 1,
0, 0, 0, 0, 0, 0, 0,
+ ];
+ let single_value = vec![0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0,
0, 0, 0, 0, 0];
+ ManifestEntry::new(
+ FileKind::Add,
+ single_value.clone(),
+ 1,
+ 10,
+ DataFileMeta {
+ file_name: "test.parquet".to_string(),
+ file_size: 100,
+ row_count: 50,
+ min_key: single_value.clone(),
+ max_key: single_value,
+ key_stats: BinaryTableStats::new(
+ value_bytes.clone(),
+ value_bytes.clone(),
+ vec![Some(1), Some(2)],
+ ),
+ value_stats: BinaryTableStats::new(
+ value_bytes.clone(),
+ value_bytes,
+ vec![Some(1), Some(2)],
+ ),
+ min_sequence_number: 1,
+ max_sequence_number: 50,
+ schema_id: 0,
+ level: 0,
+ extra_files: vec![],
+ creation_time: Some(
+ "2024-09-06T07:45:55.039+00:00"
+ .parse::<DateTime<Utc>>()
+ .unwrap(),
+ ),
+ delete_row_count: Some(0),
+ embedded_index: None,
+ first_row_id: None,
+ write_cols: None,
+ external_path: None,
+ file_source: None,
+ value_stats_cols: None,
+ },
+ 2,
+ )
+ }
+
#[test]
fn test_roundtrip_manifest_file_meta() {
let value_bytes = vec![
@@ -134,56 +183,32 @@ mod tests {
#[test]
fn test_roundtrip_manifest_entry() {
- let value_bytes = vec![
- 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 49, 0, 0, 0, 0, 0, 0, 129, 1,
0, 0, 0, 0, 0, 0, 0,
- ];
- let single_value = vec![0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0,
0, 0, 0, 0, 0];
- let original = vec![ManifestEntry::new(
- FileKind::Add,
- single_value.clone(),
- 1,
- 10,
- DataFileMeta {
- file_name: "test.parquet".to_string(),
- file_size: 100,
- row_count: 50,
- min_key: single_value.clone(),
- max_key: single_value.clone(),
- key_stats: BinaryTableStats::new(
- value_bytes.clone(),
- value_bytes.clone(),
- vec![Some(1), Some(2)],
- ),
- value_stats: BinaryTableStats::new(
- value_bytes.clone(),
- value_bytes.clone(),
- vec![Some(1), Some(2)],
- ),
- min_sequence_number: 1,
- max_sequence_number: 50,
- schema_id: 0,
- level: 0,
- extra_files: vec![],
- creation_time: Some(
- "2024-09-06T07:45:55.039+00:00"
- .parse::<DateTime<Utc>>()
- .unwrap(),
- ),
- delete_row_count: Some(0),
- embedded_index: None,
- first_row_id: None,
- write_cols: None,
- external_path: None,
- file_source: None,
- value_stats_cols: None,
- },
- 2,
- )];
+ let original = vec![manifest_entry()];
let bytes = to_avro_bytes(MANIFEST_ENTRY_SCHEMA, &original).unwrap();
let decoded = from_avro_bytes::<ManifestEntry>(&bytes).unwrap();
assert_eq!(original, decoded);
}
+ #[test]
+ fn test_read_manifest_entry_with_legacy_rust_field_order() {
+ 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()
+ .unwrap()
+ .get_mut("fields")
+ .unwrap()
+ .as_array_mut()
+ .unwrap();
+ let version = fields.remove(0);
+ fields.push(version);
+ let legacy_schema = serde_json::to_string(&schema).unwrap();
+
+ let original = vec![manifest_entry()];
+ let bytes = to_avro_bytes(&legacy_schema, &original).unwrap();
+ let decoded = from_avro_bytes_fast::<ManifestEntry>(&bytes).unwrap();
+ assert_eq!(original, decoded);
+ }
+
#[tokio::test]
async fn test_read_manifest_list() {
let workdir =
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 39757ced..e5f09f9e 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -2870,7 +2870,7 @@ impl TableCommit {
msg.bucket,
msg.total_buckets.unwrap_or(self.total_buckets),
file.clone(),
- 0,
+ 2,
)
})
})
@@ -3142,6 +3142,7 @@ mod tests {
BinaryRowBuilder, DataFileMeta, DeletionVectorMeta, GlobalIndexMeta,
IndexFileMeta,
ManifestList, TableSchema, POSTPONE_BUCKET,
};
+ use apache_avro::types::Value;
use chrono::{DateTime, Utc};
#[tokio::test]
@@ -5366,6 +5367,45 @@ mod tests {
assert_eq!(snapshot.changelog_record_count(), Some(3));
assert!(snapshot.changelog_manifest_list().is_some());
assert!(snapshot.changelog_manifest_list_size().unwrap() > 0);
+
+ let manifest_dir = format!("{table_path}/manifest");
+ let changelog_manifest_list =
snapshot.changelog_manifest_list().unwrap();
+ let changelog_metas = ManifestList::read(
+ &file_io,
+ &format!("{manifest_dir}/{changelog_manifest_list}"),
+ )
+ .await
+ .unwrap();
+ assert_eq!(changelog_metas.len(), 1);
+
+ let manifest_bytes = file_io
+ .new_input(&format!(
+ "{manifest_dir}/{}",
+ changelog_metas[0].file_name()
+ ))
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let values = apache_avro::Reader::new(manifest_bytes.as_ref())
+ .unwrap()
+ .collect::<std::result::Result<Vec<_>, _>>()
+ .unwrap();
+ assert_eq!(values.len(), 1);
+
+ let value = match &values[0] {
+ Value::Union(_, value) => value.as_ref(),
+ value => value,
+ };
+ let Value::Record(fields) = value else {
+ panic!("manifest entry must be an Avro record");
+ };
+ let version = fields
+ .iter()
+ .find(|(name, _)| name == "_VERSION")
+ .map(|(_, value)| value)
+ .expect("manifest entry format identifier");
+ assert_eq!(version, &Value::Int(2));
}
#[tokio::test]
