diff --git a/Cargo.lock b/Cargo.lock index c979ec908..f636a6bad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -104,6 +104,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "aliasable" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd" + [[package]] name = "aligned-vec" version = "0.6.4" @@ -266,24 +272,24 @@ checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" [[package]] name = "apache-avro" -version = "0.21.0" +version = "0.22.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf" +checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c" dependencies = [ "bigdecimal", "bon", - "digest 0.10.7", + "digest 0.11.3", "log", - "miniz_oxide", + "miniz_oxide 0.9.1", "num-bigint 0.4.8", + "ouroboros", "quad-rand", - "rand 0.9.5", + "rand 0.10.2", "regex-lite", "serde", "serde_bytes", "serde_json", - "strum 0.27.2", - "strum_macros 0.27.2", + "strum", "thiserror 2.0.20", "uuid", ] @@ -1458,7 +1464,7 @@ dependencies = [ "addr2line", "cfg-if", "libc", - "miniz_oxide", + "miniz_oxide 0.8.9", "object 0.37.3", "rustc-demangle", "windows-link", @@ -2003,7 +2009,7 @@ version = "4.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d012d2b9d65aca7f18f4d9878a045bc17899bba951561ba5ec3c2ba1eed9a061" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn 3.0.3", @@ -4184,7 +4190,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", - "miniz_oxide", + "miniz_oxide 0.8.9", "zlib-rs", ] @@ -4854,6 +4860,12 @@ dependencies = [ "stable_deref_trait", ] +[[package]] +name = "heck" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8" + [[package]] name = "heck" version = "0.5.0" @@ -6391,6 +6403,15 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", +] + [[package]] name = "minlz" version = "1.2.3" @@ -6485,7 +6506,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4db8a44120571277accfaa3f3d91e7d3989d601d817c2fc01a9391b86135666" dependencies = [ "darling 0.23.0", - "heck", + "heck 0.5.0", "manyhow", "num-bigint 0.4.8", "proc-macro-crate", @@ -7190,6 +7211,30 @@ dependencies = [ "num-traits", ] +[[package]] +name = "ouroboros" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0f050db9c44b97a94723127e6be766ac5c340c48f2c4bb3ffa11713744be59" +dependencies = [ + "aliasable", + "ouroboros_macro", + "static_assertions", +] + +[[package]] +name = "ouroboros_macro" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c7028bdd3d43083f6d8d4d5187680d0d3560d54df4cc9d752005268b41e64d0" +dependencies = [ + "heck 0.4.1", + "proc-macro2", + "proc-macro2-diagnostics", + "quote", + "syn 2.0.119", +] + [[package]] name = "outref" version = "0.5.2" @@ -7925,6 +7970,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proc-macro2-diagnostics" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af066a9c399a26e020ada66a034357a868728e72cd426f3adcd35f80d88d88c8" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "version_check", + "yansi", +] + [[package]] name = "prometheus" version = "0.14.0" @@ -7984,7 +8042,7 @@ version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ - "heck", + "heck 0.5.0", "itertools 0.14.0", "log", "multimap", @@ -8004,7 +8062,7 @@ version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ - "heck", + "heck 0.5.0", "itertools 0.14.0", "log", "multimap", @@ -9924,7 +9982,7 @@ dependencies = [ "rustfs-crypto", "serde", "serde_json", - "strum 0.28.0", + "strum", "temp-env", "test-case", "thiserror 2.0.20", @@ -11513,31 +11571,13 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" -[[package]] -name = "strum" -version = "0.27.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" - [[package]] name = "strum" version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" dependencies = [ - "strum_macros 0.28.0", -] - -[[package]] -name = "strum_macros" -version = "0.27.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" -dependencies = [ - "heck", - "proc-macro2", - "quote", - "syn 2.0.119", + "strum_macros", ] [[package]] @@ -11546,7 +11586,7 @@ version = "0.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.119", diff --git a/Cargo.toml b/Cargo.toml index 5cca4c433..b22bf20d0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -171,7 +171,7 @@ tower = { version = "0.5.3" } tower-http = { version = "0.7.0" } # Serialization and Data Formats -apache-avro = "0.21.0" +apache-avro = "0.22.0" bytes = { version = "1.12.1" } bytesize = "2.7.0" byteorder = "1.5.0" diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index da3a685c2..bfcf4f321 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -7313,10 +7313,10 @@ fn test_manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, i32, "#, ) .expect("manifest list avro schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize"); for (manifest_path, partition_spec_id, sequence_number, snapshot_id) in manifests { writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ( "manifest_path".to_string(), apache_avro::types::Value::String((*manifest_path).to_string()), @@ -7368,10 +7368,10 @@ fn test_manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec { "#, ) .expect("manifest avro schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize"); for (file_path, content, status, snapshot_id, sequence_number) in files { writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("status".to_string(), apache_avro::types::Value::Int(*status)), ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)), ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)), @@ -7427,10 +7427,10 @@ fn test_manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i32, i6 "#, ) .expect("manifest avro schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize"); for (file_path, content, status, snapshot_id, sequence_number) in files { writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("status".to_string(), apache_avro::types::Value::Int(*status)), ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)), ("sequence_number".to_string(), test_nullable_long(*sequence_number)), diff --git a/rustfs/src/table_catalog/maintenance/planner.rs b/rustfs/src/table_catalog/maintenance/planner.rs index 646f2ee7b..e6b39354e 100644 --- a/rustfs/src/table_catalog/maintenance/planner.rs +++ b/rustfs/src/table_catalog/maintenance/planner.rs @@ -921,9 +921,11 @@ pub(crate) fn compacted_manifest_list_avro_bytes(summary: CompactionManifestList "#, ) .map_err(|err| TableCatalogStoreError::Internal(format!("failed to build compaction manifest list schema: {err}")))?; - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).map_err(|err| { + TableCatalogStoreError::Internal(format!("failed to initialize compaction manifest list writer: {err}")) + })?; writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ( "manifest_path".to_string(), apache_avro::types::Value::String(summary.manifest_path.to_string()), @@ -1095,14 +1097,15 @@ fn compaction_partition_field_schema(value: &apache_avro::types::Value) -> Optio pub(crate) fn compacted_manifest_avro_bytes(data_files: &[CompactedDataFile]) -> TableCatalogStoreResult> { let schema = compacted_manifest_avro_schema(data_files)?; - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()) + .map_err(|err| TableCatalogStoreError::Internal(format!("failed to initialize compaction manifest writer: {err}")))?; for data_file in data_files { let sort_order_id = match data_file.sort_order_id { Some(sort_order_id) => apache_avro::types::Value::Union(1, Box::new(apache_avro::types::Value::Int(sort_order_id))), None => apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)), }; writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("status".to_string(), apache_avro::types::Value::Int(data_file.status)), ("snapshot_id".to_string(), apache_avro::types::Value::Long(data_file.snapshot_id)), ("sequence_number".to_string(), apache_avro::types::Value::Long(data_file.sequence_number)), diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index 9533ae862..997fa49ab 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -985,10 +985,10 @@ fn manifest_list_avro_bytes_with_spec(manifest_paths: &[&str], partition_spec_id "#, ) .expect("manifest list avro schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize"); for manifest_path in manifest_paths { writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ( "manifest_path".to_string(), apache_avro::types::Value::String((*manifest_path).to_string()), @@ -1027,9 +1027,9 @@ fn v1_manifest_list_avro_bytes(manifest_path: &str) -> Vec { "#, ) .expect("v1 manifest list schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("v1 manifest list writer should initialize"); writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("manifest_path".to_string(), apache_avro::types::Value::String(manifest_path.to_string())), ("manifest_length".to_string(), apache_avro::types::Value::Long(1)), ("partition_spec_id".to_string(), apache_avro::types::Value::Int(0)), @@ -1067,9 +1067,9 @@ fn v1_manifest_avro_bytes(data_file_path: &str) -> Vec { "#, ) .expect("v1 manifest schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("v1 manifest writer should initialize"); writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("status".to_string(), apache_avro::types::Value::Int(1)), ("snapshot_id".to_string(), apache_avro::types::Value::Long(10)), ( @@ -1346,9 +1346,10 @@ fn iceberg_manifest_validation_accepts_deflate_and_rejects_unknown_content() { "#, ) .expect("manifest list schema should parse"); - let mut writer = apache_avro::Writer::with_codec(&schema, Vec::new(), apache_avro::Codec::Deflate(Default::default())); + let mut writer = apache_avro::Writer::with_codec(&schema, Vec::new(), apache_avro::Codec::Deflate(Default::default())) + .expect("compressed manifest list writer should initialize"); writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ( "manifest_path".to_string(), apache_avro::types::Value::String("s3://warehouse/tables/table-id/metadata/manifest.avro".to_string()), @@ -1368,6 +1369,45 @@ fn iceberg_manifest_validation_accepts_deflate_and_rejects_unknown_content() { assert!(matches!(error, TableCatalogStoreError::Invalid(_))); } +#[test] +fn apache_avro_021_iceberg_manifest_fixtures_remain_readable() { + // Fixed 0.21 output prevents this compatibility check from becoming a current-version round trip. + let manifest_list = include_bytes!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/table_catalog/apache-avro-0.21-manifest-list.avro" + )); + let manifest_list = decode_manifest_list_avro(manifest_list).expect("apache-avro 0.21 manifest list should remain readable"); + assert_eq!(manifest_list.references.len(), 1); + let manifest_list_reference = &manifest_list.references[0]; + assert_eq!(manifest_list_reference.format_version, 2); + assert_eq!( + manifest_list_reference.manifest_path, + "s3://warehouse/tables/table-id/metadata/manifest-021.avro" + ); + assert_eq!(manifest_list_reference.manifest_length, Some(4096)); + assert_eq!(manifest_list_reference.partition_spec_id, Some(3)); + assert_eq!(manifest_list_reference.sequence_number, Some(9)); + assert_eq!(manifest_list_reference.min_sequence_number, Some(8)); + assert_eq!(manifest_list_reference.added_snapshot_id, Some(101)); + + let manifest = include_bytes!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/table_catalog/apache-avro-0.21-manifest.avro" + )); + let manifest = decode_manifest_avro(manifest).expect("apache-avro 0.21 manifest should remain readable"); + assert_eq!(manifest.references.len(), 1); + let data_file = &manifest.references[0]; + assert_eq!(data_file.format_version, 2); + assert_eq!(data_file.location, "s3://warehouse/tables/table-id/data/part-021.parquet"); + assert_eq!(data_file.snapshot_id, Some(101)); + assert_eq!(data_file.sequence_number, Some(9)); + assert_eq!(data_file.file_sequence_number, Some(9)); + assert_eq!(data_file.record_count, Some(10)); + assert_eq!(data_file.file_size_bytes, Some(1024)); + assert_eq!(data_file.sort_order_id, Some(7)); + assert_eq!(data_file.partition, vec![("dt".to_string(), apache_avro::types::Value::Date(20_312))]); +} + #[tokio::test] async fn iceberg_snapshot_graph_rejects_unknown_partition_specs() { let backend = TestCatalogObjectBackend::default(); @@ -1856,10 +1896,10 @@ fn manifest_avro_bytes_with_status(files: &[(&str, i32, i32)]) -> Vec { "#, ) .expect("manifest avro schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize"); for (file_path, content, status) in files { writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("status".to_string(), apache_avro::types::Value::Int(*status)), ("snapshot_id".to_string(), apache_avro::types::Value::Long(20)), ("sequence_number".to_string(), apache_avro::types::Value::Long(7)), @@ -1911,10 +1951,10 @@ fn manifest_avro_bytes_with_dt_partition(files: &[(&str, i32, &str)]) -> Vec "#, ) .expect("partitioned manifest avro schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("partitioned manifest writer should initialize"); for (file_path, content, partition_value) in files { writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("status".to_string(), apache_avro::types::Value::Int(1)), ("snapshot_id".to_string(), apache_avro::types::Value::Long(20)), ("sequence_number".to_string(), apache_avro::types::Value::Long(7)), @@ -1971,10 +2011,10 @@ fn manifest_avro_bytes_with_sort_order(files: &[(&str, i32, i32)]) -> Vec { "#, ) .expect("sort-order manifest avro schema should parse"); - let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("sorted manifest writer should initialize"); for (file_path, content, sort_order_id) in files { writer - .append(apache_avro::types::Value::Record(vec![ + .append_value(apache_avro::types::Value::Record(vec![ ("status".to_string(), apache_avro::types::Value::Int(1)), ("snapshot_id".to_string(), apache_avro::types::Value::Long(20)), ("sequence_number".to_string(), apache_avro::types::Value::Long(7)), diff --git a/rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest-list.avro b/rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest-list.avro new file mode 100644 index 000000000..d824c3495 Binary files /dev/null and b/rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest-list.avro differ diff --git a/rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest.avro b/rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest.avro new file mode 100644 index 000000000..fdac5e493 Binary files /dev/null and b/rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest.avro differ