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: #[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::>(); + + 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 { #[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::>() + .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::>() - .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::(&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::(&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::, _>>() + .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]