fix(spec): support deflate for Avro manifest files

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 }

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<Cow<'a, [u8]>> {
         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<u8> {
+        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<u8> {
+        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<u8> {
+        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}");
+    }
 }

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<Codec> {
         "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::<ManifestFileMeta>(&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)] {

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::<usize>() > 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();