From bcf1f5cc1cf2f2409f149d327afdf6a1a7d64a59 Mon Sep 17 00:00:00 2001 From: jackylee-ch Date: Tue, 15 Sep 2026 19:05:35 +0800 Subject: [PATCH] fix(spec): support deflate for Avro manifest files --- Cargo.lock | 1 + crates/paimon/Cargo.toml | 1 + crates/paimon/src/spec/avro/ocf.rs | 135 +++++++++++++++++++++++- crates/paimon/src/spec/objects_file.rs | 62 ++++++++++- crates/paimon/src/table/table_commit.rs | 75 +++++++++++++ 5 files changed, 272 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6b3bc6796..cb2addac8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4676,6 +4676,7 @@ dependencies = [ "lz4_flex 0.13.1", "lzokay-native", "md-5 0.10.6", + "miniz_oxide", "opendal-core", "opendal-http-transport-reqwest", "opendal-layer-retry", diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index a3d116fbd..230aea3f4 100644 --- a/crates/paimon/Cargo.toml +++ b/crates/paimon/Cargo.toml @@ -103,6 +103,7 @@ roaring = "0.11" crc32fast = "1" zstd = "0.13" snap = "1" +miniz_oxide = "0.8" lz4_flex = "0.13" lzokay-native = { version = "0.1", default-features = false, features = ["decompress"] } arrow-array = { workspace = true } diff --git a/crates/paimon/src/spec/avro/ocf.rs b/crates/paimon/src/spec/avro/ocf.rs index aed7ee259..9d6103148 100644 --- a/crates/paimon/src/spec/avro/ocf.rs +++ b/crates/paimon/src/spec/avro/ocf.rs @@ -34,6 +34,7 @@ pub struct OcfHeader { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum OcfCodec { Null, + Deflate, Snappy, Zstandard, } @@ -101,6 +102,24 @@ impl<'a> OcfBlockIter<'a> { fn decompress(&mut self, data: &'a [u8]) -> crate::Result> { match self.codec { OcfCodec::Null => Ok(Cow::Borrowed(data)), + OcfCodec::Deflate => { + // Avro's `deflate` block is *raw* RFC 1951 — no zlib header, no + // adler32, because Avro Java deflates with `nowrap=true`. This is the + // same entry point apache-avro uses for `Codec::Deflate`, so a file + // this reader accepts is one `apache_avro::Reader` accepts too. It is + // also the strict choice: `flate2`'s `read_to_end` returns the partial + // output for a truncated stream, while this errors. + if data.is_empty() { + return Ok(Cow::Borrowed(data)); + } + let decompressed = miniz_oxide::inflate::decompress_to_vec(data).map_err(|e| { + Error::UnexpectedError { + message: format!("avro ocf: deflate decompression failed: {:?}", e.status), + source: None, + } + })?; + Ok(Cow::Owned(decompressed)) + } OcfCodec::Snappy => { if data.len() < 4 { return Err(Error::UnexpectedError { @@ -121,7 +140,7 @@ impl<'a> OcfBlockIter<'a> { if actual_crc != expected_crc { return Err(Error::UnexpectedError { message: format!( - "avro ocf: snappy CRC32C mismatch: expected {expected_crc:#010x}, got {actual_crc:#010x}" + "avro ocf: snappy CRC32 mismatch: expected {expected_crc:#010x}, got {actual_crc:#010x}" ), source: None, }); @@ -164,6 +183,7 @@ pub fn parse_ocf_streaming(bytes: &[u8]) -> crate::Result<(OcfHeader, OcfBlockIt let codec = match meta.get("avro.codec").map(|s| s.as_str()) { None | Some("null") => OcfCodec::Null, + Some("deflate") => OcfCodec::Deflate, Some("snappy") => OcfCodec::Snappy, Some("zstandard") => OcfCodec::Zstandard, Some(other) => { @@ -296,4 +316,117 @@ mod tests { assert_eq!(blocks.len(), 1); assert_eq!(blocks[0].object_count, 1); } + + #[test] + fn test_parse_ocf_deflate() { + use apache_avro::{Codec, DeflateSettings, Schema, Writer}; + + let schema = Schema::parse_str( + r#"{"type": "record", "name": "test", "fields": [{"name": "x", "type": "long"}]}"#, + ) + .unwrap(); + let mut writer = Writer::with_codec( + &schema, + Vec::new(), + Codec::Deflate(DeflateSettings::default()), + ); + let mut record = apache_avro::types::Record::new(&schema).unwrap(); + record.put("x", 24680i64); + writer.append(record).unwrap(); + let bytes = writer.into_inner().unwrap(); + + let (header, blocks) = parse_ocf(&bytes).unwrap(); + assert_eq!(header.codec, OcfCodec::Deflate); + assert_eq!(blocks.len(), 1); + assert_eq!(blocks[0].object_count, 1); + } + + /// Avro zigzag varint, so the fixture below owes nothing to any encoder we ship. + fn avro_long(value: i64) -> Vec { + let mut zigzag = ((value << 1) ^ (value >> 63)) as u64; + let mut out = Vec::new(); + loop { + if zigzag & !0x7f == 0 { + out.push(zigzag as u8); + return out; + } + out.push((zigzag as u8 & 0x7f) | 0x80); + zigzag >>= 7; + } + } + + fn avro_bytes(value: &[u8]) -> Vec { + let mut out = avro_long(value.len() as i64); + out.extend_from_slice(value); + out + } + + /// A *stored* (BTYPE=00) final deflate block: raw RFC 1951, no zlib wrapper. + fn raw_deflate_stored(payload: &[u8]) -> Vec { + let len = u16::try_from(payload.len()).expect("fixture block fits in a stored block"); + let mut out = vec![0x01]; + out.extend_from_slice(&len.to_le_bytes()); + out.extend_from_slice(&(!len).to_le_bytes()); + out.extend_from_slice(payload); + out + } + + #[test] + fn test_parse_ocf_deflate_accepts_a_hand_built_raw_stream() { + // Built byte by byte, without any Rust deflate implementation, so it pins the + // on-disk contract rather than a round trip through one crate: Avro's deflate + // is raw RFC 1951, and a zlib reader would reject exactly this input. + const SCHEMA: &str = + r#"{"type":"record","name":"test","fields":[{"name":"x","type":"long"}]}"#; + let sync = [7u8; SYNC_MARKER_LEN]; + + let mut bytes = Vec::new(); + bytes.extend_from_slice(AVRO_MAGIC); + bytes.extend_from_slice(&avro_long(2)); // header map entry count + bytes.extend_from_slice(&avro_bytes(b"avro.schema")); + bytes.extend_from_slice(&avro_bytes(SCHEMA.as_bytes())); + bytes.extend_from_slice(&avro_bytes(b"avro.codec")); + bytes.extend_from_slice(&avro_bytes(b"deflate")); + bytes.extend_from_slice(&avro_long(0)); // end of map + bytes.extend_from_slice(&sync); + + // One block holding two records: x = 1, x = -2. + let mut body = avro_long(1); + body.extend_from_slice(&avro_long(-2)); + let block = raw_deflate_stored(&body); + bytes.extend_from_slice(&avro_long(2)); // object count + bytes.extend_from_slice(&avro_long(block.len() as i64)); + bytes.extend_from_slice(&block); + bytes.extend_from_slice(&sync); + + let (header, blocks) = parse_ocf(&bytes).unwrap(); + assert_eq!(header.codec, OcfCodec::Deflate); + assert_eq!(header.schema_json, SCHEMA); + assert_eq!(blocks.len(), 1); + assert_eq!(blocks[0].object_count, 2); + // The decompressed body must be the exact bytes we wrapped. + assert_eq!(blocks[0].data.as_ref(), body.as_slice()); + } + + #[test] + fn test_parse_ocf_still_rejects_codecs_we_do_not_implement() { + // Java's `CodecFactory` also accepts bzip2 and xz; we do not, and the + // whitelist must keep saying so rather than becoming permissive. + let sync = [0u8; SYNC_MARKER_LEN]; + let mut bytes = Vec::new(); + bytes.extend_from_slice(AVRO_MAGIC); + bytes.extend_from_slice(&avro_long(2)); + bytes.extend_from_slice(&avro_bytes(b"avro.schema")); + bytes.extend_from_slice(&avro_bytes(br#"{"type":"null"}"#)); + bytes.extend_from_slice(&avro_bytes(b"avro.codec")); + bytes.extend_from_slice(&avro_bytes(b"bzip2")); + bytes.extend_from_slice(&avro_long(0)); + bytes.extend_from_slice(&sync); + + let error = match parse_ocf(&bytes) { + Ok(_) => panic!("bzip2 is not implemented here and must not be accepted"), + Err(error) => error.to_string(), + }; + assert!(error.contains("unsupported codec: bzip2"), "{error}"); + } } diff --git a/crates/paimon/src/spec/objects_file.rs b/crates/paimon/src/spec/objects_file.rs index 56f2fe4d2..28b376d9f 100644 --- a/crates/paimon/src/spec/objects_file.rs +++ b/crates/paimon/src/spec/objects_file.rs @@ -15,7 +15,9 @@ // specific language governing permissions and limitations // under the License. -use apache_avro::{from_value, to_value, Codec, Reader, Schema, Writer, ZstandardSettings}; +use apache_avro::{ + from_value, to_value, Codec, DeflateSettings, Reader, Schema, Writer, ZstandardSettings, +}; use serde::de::DeserializeOwned; use serde::Serialize; @@ -71,6 +73,16 @@ pub(crate) fn avro_codec(compression: &str) -> crate::Result { "zstd" | "zstandard" => Ok(Codec::Zstandard(ZstandardSettings::default())), "null" | "none" | "uncompressed" => Ok(Codec::Null), "snappy" => Ok(Codec::Snappy), + // Ask for level 6, matching Avro Java's `Deflater.DEFAULT_COMPRESSION`. + // `DeflateSettings::default()` looks like the obvious choice and is not: + // apache-avro's `DefaultCompression` is `-1`, `compression_level()` casts it + // `as u8` to 255, and miniz_oxide clamps that to 10 search probes — 6.7x the + // CPU of level 6 at our 16 KB block size, for byte-identical output. + // miniz_oxide's probe count is not zlib's level algorithm, so this is + // comparable effort rather than byte-identical output to Java. + "deflate" => Ok(Codec::Deflate(DeflateSettings::new( + miniz_oxide::deflate::CompressionLevel::DefaultLevel, + ))), other => Err(crate::Error::Unsupported { message: format!("Unsupported Avro compression: {other}"), }), @@ -90,6 +102,54 @@ mod tests { use apache_avro::types::Value; use chrono::{DateTime, Utc}; + #[test] + fn test_avro_codec_accepts_deflate_and_still_rejects_the_rest() { + // Java's `CodecFactory.fromString` also takes bzip2 and xz; each would be a new + // dependency with no evidence of use, so they stay rejected here. We are also + // laxer than Java on case and on the `zstd`/`none`/`uncompressed` aliases — + // pre-existing and deliberate, since these are Paimon option names rather than + // Avro header names. + assert!(matches!(avro_codec("deflate").unwrap(), Codec::Deflate(_))); + assert!(matches!(avro_codec("DEFLATE").unwrap(), Codec::Deflate(_))); + let error = avro_codec("bzip2").unwrap_err().to_string(); + assert!( + error.contains("Unsupported Avro compression: bzip2"), + "{error}" + ); + } + + #[test] + fn test_every_writable_compression_round_trips_through_the_ocf_reader() { + // The invariant this whole change exists to restore: whatever `avro_codec` + // accepts must produce a header `parse_ocf` accepts. The two whitelists live in + // different string domains (Paimon option names vs Avro header names), so + // nothing but a loop keeps them composable. + let metas = vec![ManifestFileMeta::new( + "manifest-test-0".to_string(), + 1024, + 5, + 2, + BinaryTableStats::new(vec![0, 0, 0, 2], vec![0, 0, 0, 3], vec![Some(1)]), + 0, + )]; + for compression in [ + "zstd", + "zstandard", + "null", + "none", + "uncompressed", + "snappy", + "deflate", + ] { + let bytes = + to_avro_bytes_with_compression(MANIFEST_FILE_META_SCHEMA, &metas, compression) + .unwrap(); + let decoded = from_avro_bytes_fast::(&bytes) + .unwrap_or_else(|error| panic!("{compression}: {error}")); + assert_eq!(decoded, metas, "{compression}"); + } + } + // Check the record decoded from the OCF writer schema, including fields whose // values were omitted by serde and filled from Avro defaults. fn assert_record_field_order<'a>(value: &'a Value, expected: &[&str]) -> &'a [(String, Value)] { diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 676d53587..a1c22f45b 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -5839,6 +5839,81 @@ mod tests { assert_eq!(stats.null_counts(), &vec![Some(1)]); } + /// `manifest.compression` comes from the persisted table schema, so a table + /// created by Java with `deflate` makes every paimon-rust process both write and + /// read that codec. The second commit is the load-bearing half: it makes the + /// commit path *read back* the first snapshot's manifest list and manifest files, + /// which is exactly what fails on such a table today. + #[tokio::test] + async fn test_deflate_manifest_compression_round_trips() { + use crate::spec::avro::ocf::{parse_ocf, OcfCodec}; + + let file_io = test_file_io(); + let table_path = "memory:/test_manifest_deflate"; + setup_dirs(&file_io, table_path).await; + + let table = test_table_with_options( + &file_io, + table_path, + HashMap::from([("manifest.compression".to_string(), "deflate".to_string())]), + ); + let commit = TableCommit::new(table, "test-user".to_string()); + commit + .commit(vec![CommitMessage::new( + vec![], + 0, + vec![test_data_file("data-0.parquet", 3)], + )]) + .await + .unwrap(); + // Reads the previous snapshot's manifests through the production commit path. + commit + .commit(vec![CommitMessage::new( + vec![], + 0, + vec![test_data_file("data-1.parquet", 4)], + )]) + .await + .unwrap(); + + let snapshot = latest_snapshot(&file_io, table_path).await.unwrap(); + assert_eq!(snapshot.id(), 2); + let manifest_dir = format!("{table_path}/manifest"); + let mut names = Vec::new(); + // The base list carries what snapshot 1 wrote and the commit path just read + // back; the delta list is what this commit wrote. + for list_name in [ + snapshot.base_manifest_list(), + snapshot.delta_manifest_list(), + ] { + let list_path = format!("{manifest_dir}/{list_name}"); + let list_bytes = file_io.new_input(&list_path).unwrap().read().await.unwrap(); + let (header, blocks) = parse_ocf(&list_bytes).unwrap(); + assert_eq!(header.codec, OcfCodec::Deflate, "{list_path}"); + assert!(blocks.iter().map(|block| block.object_count).sum::() > 0); + + for meta in ManifestList::read(&file_io, &list_path).await.unwrap() { + let manifest_path = format!("{manifest_dir}/{}", meta.file_name()); + let manifest_bytes = file_io + .new_input(&manifest_path) + .unwrap() + .read() + .await + .unwrap(); + assert_eq!( + parse_ocf(&manifest_bytes).unwrap().0.codec, + OcfCodec::Deflate, + "{manifest_path}" + ); + for entry in Manifest::read(&file_io, &manifest_path).await.unwrap() { + names.push(entry.file().file_name.clone()); + } + } + } + names.sort(); + assert_eq!(names, ["data-0.parquet", "data-1.parquet"]); + } + #[tokio::test] async fn test_manifest_files_roll_by_target_size_and_preserve_entries() { let file_io = test_file_io();