This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 d81476a0 feat(spec): support the Avro deflate codec (#982)
d81476a0 is described below
commit d81476a0a8006f4659c1e164ba148c46a4c07791
Author: jackylee <[email protected]>
AuthorDate: Fri Oct 2 10:08:01 2026 +0800
feat(spec): support the Avro deflate codec (#982)
---
crates/paimon/src/spec/avro/ocf.rs | 49 ++++++++++++++++++++++++++++++++++
crates/paimon/src/spec/objects_file.rs | 17 +++++++++++-
2 files changed, 65 insertions(+), 1 deletion(-)
diff --git a/crates/paimon/src/spec/avro/ocf.rs
b/crates/paimon/src/spec/avro/ocf.rs
index 58b89ca7..1004c23d 100644
--- a/crates/paimon/src/spec/avro/ocf.rs
+++ b/crates/paimon/src/spec/avro/ocf.rs
@@ -36,6 +36,7 @@ pub struct OcfHeader {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OcfCodec {
Null,
+ Deflate,
Snappy,
Zstandard,
}
@@ -113,6 +114,20 @@ 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 => {
+ // The Avro `deflate` codec is raw DEFLATE (RFC 1951), with no
+ // zlib or gzip wrapper.
+ use std::io::Read;
+ let mut decoder = flate2::read::DeflateDecoder::new(data);
+ let mut decompressed = Vec::new();
+ decoder
+ .read_to_end(&mut decompressed)
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("avro ocf: deflate decompression
failed: {e}"),
+ source: None,
+ })?;
+ Ok(Cow::Owned(decompressed))
+ }
OcfCodec::Snappy => {
if data.len() < 4 {
return Err(Error::UnexpectedError {
@@ -176,6 +191,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) => {
@@ -309,4 +325,37 @@ mod tests {
assert_eq!(blocks.len(), 1);
assert_eq!(blocks[0].object_count, 1);
}
+
+ #[test]
+ fn test_parse_ocf_deflate() {
+ use apache_avro::{from_avro_datum, types::Value, 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", 424242i64);
+ 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);
+
+ // The decompressed block must decode back to the original record,
which
+ // proves the deflate output is correct rather than merely
non-erroring.
+ let mut cursor = blocks[0].data.as_ref();
+ let value = from_avro_datum(&schema, &mut cursor, None).unwrap();
+ assert_eq!(
+ value,
+ Value::Record(vec![("x".to_string(), Value::Long(424242))])
+ );
+ }
}
diff --git a/crates/paimon/src/spec/objects_file.rs
b/crates/paimon/src/spec/objects_file.rs
index c30ecc8f..a4e60c4f 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;
@@ -70,6 +72,7 @@ pub(crate) fn avro_codec(compression: &str) ->
crate::Result<Codec> {
match compression.to_ascii_lowercase().as_str() {
"zstd" | "zstandard" =>
Ok(Codec::Zstandard(ZstandardSettings::default())),
"null" | "none" | "uncompressed" => Ok(Codec::Null),
+ "deflate" => Ok(Codec::Deflate(DeflateSettings::default())),
"snappy" => Ok(Codec::Snappy),
other => Err(crate::Error::Unsupported {
message: format!("Unsupported Avro compression: {other}"),
@@ -333,6 +336,18 @@ mod tests {
assert_eq!(original, decoded);
}
+ #[test]
+ fn test_roundtrip_manifest_entry_deflate() {
+ // Writing with the deflate codec and reading back through the fast
+ // decoder must round-trip, exercising both the write-side codec
mapping
+ // and the read-side deflate decompression.
+ let original = vec![manifest_entry()];
+ let bytes =
+ to_avro_bytes_with_compression(MANIFEST_ENTRY_SCHEMA, &original,
"deflate").unwrap();
+ let decoded = from_avro_bytes_fast::<ManifestEntry>(&bytes).unwrap();
+ assert_eq!(original, decoded);
+ }
+
#[test]
fn test_roundtrip_index_manifest_field_order() {
let entry: IndexManifestEntry =
serde_json::from_value(serde_json::json!({
