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 782a6d80 Complete manifest codec support and typed partition specs
(#1170)
782a6d80 is described below
commit 782a6d8075a57e9f93508c20f5f19ab7c9f39818
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 18:05:35 2026 +0800
Complete manifest codec support and typed partition specs (#1170)
---
Cargo.lock | 3 +
bindings/c/DEPENDENCIES.rust.tsv | 2 +
bindings/go/DEPENDENCIES.rust.tsv | 2 +
crates/integration_tests/DEPENDENCIES.rust.tsv | 2 +
crates/paimon-rest-server/DEPENDENCIES.rust.tsv | 2 +
crates/paimon/Cargo.toml | 3 +-
crates/paimon/DEPENDENCIES.rust.tsv | 2 +
crates/paimon/src/spec/avro/codec_tests.rs | 312 +++++++++++++++++++++
crates/paimon/src/spec/avro/mod.rs | 3 +
crates/paimon/src/spec/avro/ocf.rs | 36 ++-
crates/paimon/src/spec/objects_file.rs | 40 ++-
.../src/table/key_normalization/encoded_tests.rs | 76 +++++
crates/paimon/src/table/partition_spec.rs | 67 ++++-
.../testdata/avro_codec/GenerateCodecFixtures.java | 73 +++++
crates/paimon/testdata/avro_codec/README.md | 38 +++
.../paimon/testdata/avro_codec/fastavro_bzip2.avro | Bin 0 -> 478 bytes
crates/paimon/testdata/avro_codec/fastavro_xz.avro | Bin 0 -> 520 bytes
crates/paimon/testdata/avro_codec/generate.py | 35 +++
crates/paimon/testdata/avro_codec/java_bzip2.avro | Bin 0 -> 462 bytes
crates/paimon/testdata/avro_codec/java_xz.avro | Bin 0 -> 504 bytes
crates/paimon/testdata/avro_codec/records.avsc | 9 +
scripts/release_licenses.py | 2 +-
scripts/tests/test_release_licenses.py | 39 +++
23 files changed, 723 insertions(+), 23 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 61d3be37..d73fbb31 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -153,8 +153,10 @@ checksum =
"36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf"
dependencies = [
"bigdecimal",
"bon",
+ "bzip2",
"crc32fast",
"digest 0.10.7",
+ "liblzma",
"log",
"miniz_oxide",
"num-bigint",
@@ -4680,6 +4682,7 @@ dependencies = [
"lexical-write-float",
"libc",
"libloading 0.9.0",
+ "liblzma",
"log",
"lru 0.18.2",
"lz4_flex 0.13.1",
diff --git a/bindings/c/DEPENDENCIES.rust.tsv b/bindings/c/DEPENDENCIES.rust.tsv
index d3a70ef8..9dbbc6e9 100644
--- a/bindings/c/DEPENDENCIES.rust.tsv
+++ b/bindings/c/DEPENDENCIES.rust.tsv
@@ -181,6 +181,8 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X X
X
[email protected]
X
diff --git a/bindings/go/DEPENDENCIES.rust.tsv
b/bindings/go/DEPENDENCIES.rust.tsv
index d3a70ef8..9dbbc6e9 100644
--- a/bindings/go/DEPENDENCIES.rust.tsv
+++ b/bindings/go/DEPENDENCIES.rust.tsv
@@ -181,6 +181,8 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X X
X
[email protected]
X
diff --git a/crates/integration_tests/DEPENDENCIES.rust.tsv
b/crates/integration_tests/DEPENDENCIES.rust.tsv
index 3c350db3..4d16dab4 100644
--- a/crates/integration_tests/DEPENDENCIES.rust.tsv
+++ b/crates/integration_tests/DEPENDENCIES.rust.tsv
@@ -171,6 +171,8 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X X
X
[email protected]
X
diff --git a/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
b/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
index 55d7caf6..1ef833b0 100644
--- a/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
+++ b/crates/paimon-rest-server/DEPENDENCIES.rust.tsv
@@ -174,6 +174,8 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X X
X
[email protected]
X
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 31302b4a..34644e6c 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -103,7 +103,7 @@ opendal-service-obs = { version = "0.58.0", optional = true
}
opendal-service-oss = { version = "0.58.0", optional = true }
opendal-service-s3 = { version = "0.58.0", optional = true }
pretty_assertions = "1"
-apache-avro = { version = "0.21", features = ["snappy", "zstandard"] }
+apache-avro = { version = "0.21", features = ["snappy", "zstandard", "bzip",
"xz"] }
indexmap = "2.5.0"
lru = "0.18.2"
roaring = "0.11"
@@ -118,6 +118,7 @@ snap = "1"
lz4_flex = "0.13"
flate2 = "1"
bzip2 = "0.6"
+liblzma = { version = "0.4", default-features = false, features = ["static"] }
lzokay-native = { version = "0.1", default-features = false, features =
["decompress"] }
arrow-array = { workspace = true }
arrow-arith = { workspace = true }
diff --git a/crates/paimon/DEPENDENCIES.rust.tsv
b/crates/paimon/DEPENDENCIES.rust.tsv
index 3a0678c2..cceabf93 100644
--- a/crates/paimon/DEPENDENCIES.rust.tsv
+++ b/crates/paimon/DEPENDENCIES.rust.tsv
@@ -263,6 +263,8 @@ [email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X X
X
diff --git a/crates/paimon/src/spec/avro/codec_tests.rs
b/crates/paimon/src/spec/avro/codec_tests.rs
new file mode 100644
index 00000000..694055b1
--- /dev/null
+++ b/crates/paimon/src/spec/avro/codec_tests.rs
@@ -0,0 +1,312 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::cursor::AvroCursor;
+use super::ocf::{parse_ocf_streaming, OcfCodec};
+use crate::spec::objects_file::new_avro_writer;
+use apache_avro::types::Value;
+use apache_avro::{Reader, Schema};
+
+const FIXTURES: &[(&str, &[u8], OcfCodec)] = &[
+ (
+ "java/bzip2",
+ include_bytes!("../../../testdata/avro_codec/java_bzip2.avro"),
+ OcfCodec::Bzip2,
+ ),
+ (
+ "java/xz",
+ include_bytes!("../../../testdata/avro_codec/java_xz.avro"),
+ OcfCodec::Xz,
+ ),
+ (
+ "fastavro/bzip2",
+ include_bytes!("../../../testdata/avro_codec/fastavro_bzip2.avro"),
+ OcfCodec::Bzip2,
+ ),
+ (
+ "fastavro/xz",
+ include_bytes!("../../../testdata/avro_codec/fastavro_xz.avro"),
+ OcfCodec::Xz,
+ ),
+];
+
+fn expected_records() -> Vec<Value> {
+ [
+ (-7, "before-epoch".to_string(), vec![0, 1, 255]),
+ (0, "\u{4e8c}".to_string(), vec![]),
+ (
+ i64::MAX,
+ format!("end{}", "x".repeat(96)),
+ (0..32).collect(),
+ ),
+ ]
+ .into_iter()
+ .map(|(id, label, payload)| {
+ Value::Record(vec![
+ ("id".to_string(), Value::Long(id)),
+ ("label".to_string(), Value::String(label)),
+ ("payload".to_string(), Value::Bytes(payload)),
+ ])
+ })
+ .collect()
+}
+
+fn assert_blocks(bytes: &[u8], expected_codec: OcfCodec) {
+ let (header, mut blocks) = parse_ocf_streaming(bytes).unwrap();
+ assert_eq!(header.codec, expected_codec);
+ let schema = Schema::parse_str(&header.schema_json).unwrap();
+ let mut encoded_end = header.encoded_len;
+ for expected in expected_records() {
+ let block = blocks.next_block().unwrap().expect("missing OCF block");
+ assert_eq!(block.object_count, 1);
+ assert_eq!(block.encoded_offset, encoded_end);
+ encoded_end += block.encoded_len;
+ assert_eq!(&bytes[encoded_end - 16..encoded_end], &header.sync_marker);
+ let mut data = block.data.as_ref();
+ assert_eq!(
+ apache_avro::from_avro_datum(&schema, &mut data, None).unwrap(),
+ expected
+ );
+ assert!(data.is_empty(), "unconsumed bytes in decoded block");
+ }
+ assert!(blocks.next_block().unwrap().is_none());
+ assert_eq!(encoded_end, bytes.len());
+
+ // Both the streaming OCF reader and the reference reader must observe the
+ // same records, rather than merely accepting the codec metadata.
+ let reference = Reader::new(bytes)
+ .unwrap()
+ .collect::<apache_avro::AvroResult<Vec<_>>>()
+ .unwrap();
+ assert_eq!(reference, expected_records());
+
+ #[derive(Debug, PartialEq, serde::Deserialize)]
+ struct Record {
+ id: i64,
+ label: String,
+ #[serde(with = "serde_bytes")]
+ payload: Vec<u8>,
+ }
+ let generic = crate::spec::from_avro_bytes::<Record>(bytes).unwrap();
+ assert_eq!(
+ generic,
+ vec![
+ Record {
+ id: -7,
+ label: "before-epoch".to_string(),
+ payload: vec![0, 1, 255],
+ },
+ Record {
+ id: 0,
+ label: "\u{4e8c}".to_string(),
+ payload: vec![],
+ },
+ Record {
+ id: i64::MAX,
+ label: format!("end{}", "x".repeat(96)),
+ payload: (0..32).collect(),
+ },
+ ]
+ );
+}
+
+#[test]
+fn test_bzip2_xz_java_and_fastavro_blocks() {
+ for (name, bytes, codec) in FIXTURES {
+ assert_eq!(
+ parse_ocf_streaming(bytes).unwrap().0.sync_marker,
+ std::array::from_fn(|i| i as u8),
+ "{name}"
+ );
+ assert_blocks(bytes, *codec);
+ }
+}
+
+#[test]
+fn test_bzip2_xz_writer_multiple_blocks() {
+ let schema =
+
Schema::parse_str(include_str!("../../../testdata/avro_codec/records.avsc")).unwrap();
+ for (compression, codec) in [
+ ("bzip2", OcfCodec::Bzip2),
+ ("xz", OcfCodec::Xz),
+ ("BZIP2", OcfCodec::Bzip2),
+ ("XZ", OcfCodec::Xz),
+ ] {
+ let mut writer = new_avro_writer(&schema, compression, 1).unwrap();
+ for record in expected_records() {
+ writer.append(record).unwrap();
+ }
+ assert_blocks(&writer.into_inner().unwrap(), codec);
+ }
+}
+
+#[test]
+fn test_bzip2_xz_public_reader_resolves_named_records() {
+ #[derive(Debug, PartialEq, serde::Deserialize, serde::Serialize)]
+ struct Child {
+ id: i64,
+ }
+ #[derive(Debug, PartialEq, serde::Deserialize, serde::Serialize)]
+ struct Parent {
+ first: Child,
+ second: Option<Child>,
+ }
+ let schema = Schema::parse_str(
+ r#"{
+ "type": "record", "name": "Parent", "fields": [
+ {"name": "first", "type": {
+ "type": "record", "name": "Child",
+ "fields": [{"name": "id", "type": "long"}]
+ }},
+ {"name": "second", "type": ["null", "Child"]}
+ ]
+ }"#,
+ )
+ .unwrap();
+ let expected = vec![
+ Parent {
+ first: Child { id: -7 },
+ second: None,
+ },
+ Parent {
+ first: Child { id: i64::MIN },
+ second: Some(Child { id: i64::MAX }),
+ },
+ ];
+ for compression in ["null", "deflate", "snappy", "zstd", "bzip2", "xz"] {
+ let mut writer = new_avro_writer(&schema, compression, 1).unwrap();
+ for record in &expected {
+ writer
+ .append(
+ apache_avro::to_value(record)
+ .unwrap()
+ .resolve(&schema)
+ .unwrap(),
+ )
+ .unwrap();
+ }
+ let bytes = writer.into_inner().unwrap();
+ assert_eq!(
+ crate::spec::from_avro_bytes::<Parent>(&bytes).unwrap(),
+ expected,
+ "{compression}"
+ );
+ }
+}
+
+#[test]
+fn test_bzip2_xz_public_reader_empty_container_and_oversized_count() {
+ for (name, bytes, _) in FIXTURES {
+ let (header, _) = parse_ocf_streaming(bytes).unwrap();
+ assert!(
+
crate::spec::from_avro_bytes::<serde_json::Value>(&bytes[..header.encoded_len])
+ .unwrap()
+ .is_empty()
+ );
+
+ let mut cursor = AvroCursor::new(&bytes[header.encoded_len..]);
+ cursor.read_long().unwrap();
+ let mut changed = bytes[..header.encoded_len].to_vec();
+ write_long(i64::MAX as usize, &mut changed);
+ changed.extend_from_slice(&bytes[header.encoded_len +
cursor.position()..]);
+ // The untrusted count must not cause a capacity-overflow allocation.
+ // Decode the actual payload and fail when the block runs out of data.
+ assert!(
+
crate::spec::from_avro_bytes::<serde::de::IgnoredAny>(&changed).is_err(),
+ "{name}"
+ );
+ }
+}
+
+fn write_long(value: usize, output: &mut Vec<u8>) {
+ let mut encoded = (value as u64) << 1;
+ while encoded >= 0x80 {
+ output.push((encoded as u8) | 0x80);
+ encoded >>= 7;
+ }
+ output.push(encoded as u8);
+}
+
+fn replace_first_payload(bytes: &[u8], update: impl FnOnce(&mut Vec<u8>)) ->
Vec<u8> {
+ let (header, _) = parse_ocf_streaming(bytes).unwrap();
+ let mut cursor = AvroCursor::new(&bytes[header.encoded_len..]);
+ let count = cursor.read_long().unwrap() as usize;
+ let size = cursor.read_long().unwrap() as usize;
+ let mut payload = cursor.read_fixed(size).unwrap().to_vec();
+ update(&mut payload);
+ let mut changed = bytes[..header.encoded_len].to_vec();
+ write_long(count, &mut changed);
+ write_long(payload.len(), &mut changed);
+ changed.extend_from_slice(&payload);
+ // Preserve the rest of the container, including the original sync markers.
+ changed.extend_from_slice(&bytes[header.encoded_len +
cursor.position()..]);
+ changed
+}
+
+#[test]
+fn test_bzip2_xz_corrupt_and_truncated_compressed_blocks_return_errors() {
+ for (name, bytes, _) in FIXTURES {
+ for damage in [0, 1, 2] {
+ let corrupted = replace_first_payload(bytes, |payload| match
damage {
+ 0 => payload.clear(),
+ 1 => payload.truncate(payload.len() / 2),
+ 2 => payload[0] ^= 0xff,
+ _ => unreachable!(),
+ });
+ let (_, mut blocks) = parse_ocf_streaming(&corrupted).unwrap();
+ let error = blocks
+ .next_block()
+ .err()
+ .expect("invalid codec block accepted");
+ assert!(
+ error.to_string().contains("decompression failed"),
+ "{name}, damage {damage}: {error}"
+ );
+ let generic_error =
crate::spec::from_avro_bytes::<serde::de::IgnoredAny>(&corrupted)
+ .expect_err("public OCF reader accepted invalid codec block");
+ assert!(
+ generic_error.to_string().contains("decompression failed"),
+ "public OCF reader, {name}, damage {damage}: {generic_error}"
+ );
+ }
+ }
+}
+
+#[test]
+fn test_bzip2_xz_container_truncation_and_sync_mismatch_return_errors() {
+ for (name, bytes, _) in FIXTURES {
+ let (_, mut blocks) = parse_ocf_streaming(&bytes[..bytes.len() -
1]).unwrap();
+ assert!(blocks.next_block().unwrap().is_some());
+ assert!(blocks.next_block().unwrap().is_some());
+ assert!(blocks.next_block().is_err(), "{name}");
+
+ let mut corrupted = bytes.to_vec();
+ *corrupted.last_mut().unwrap() ^= 1;
+ let (_, mut blocks) = parse_ocf_streaming(&corrupted).unwrap();
+ assert!(blocks.next_block().unwrap().is_some());
+ assert!(blocks.next_block().unwrap().is_some());
+ assert!(
+ blocks
+ .next_block()
+ .err()
+ .unwrap()
+ .to_string()
+ .contains("sync marker mismatch"),
+ "{name}"
+ );
+ }
+}
diff --git a/crates/paimon/src/spec/avro/mod.rs
b/crates/paimon/src/spec/avro/mod.rs
index 490c234b..b89db19f 100644
--- a/crates/paimon/src/spec/avro/mod.rs
+++ b/crates/paimon/src/spec/avro/mod.rs
@@ -24,6 +24,9 @@ mod manifest_file_meta_decode;
pub mod ocf;
pub mod schema;
+#[cfg(test)]
+mod codec_tests;
+
use cursor::AvroCursor;
use decode::AvroRecordDecode;
use ocf::parse_ocf_streaming;
diff --git a/crates/paimon/src/spec/avro/ocf.rs
b/crates/paimon/src/spec/avro/ocf.rs
index 1004c23d..a6e9d1ed 100644
--- a/crates/paimon/src/spec/avro/ocf.rs
+++ b/crates/paimon/src/spec/avro/ocf.rs
@@ -39,6 +39,8 @@ pub enum OcfCodec {
Deflate,
Snappy,
Zstandard,
+ Bzip2,
+ Xz,
}
/// A single data block from the OCF.
@@ -117,16 +119,8 @@ impl<'a> OcfBlockIter<'a> {
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))
+ read_decompressed("deflate",
flate2::read::DeflateDecoder::new(data))
+ .map(Cow::Owned)
}
OcfCodec::Snappy => {
if data.len() < 4 {
@@ -163,10 +157,30 @@ impl<'a> OcfBlockIter<'a> {
})?;
Ok(Cow::Owned(decompressed))
}
+ // The apache-avro decoders unwrap I/O errors for these two codecs.
+ // Read them directly so corrupt manifests return an error instead
+ // of panicking during a scan or commit.
+ OcfCodec::Bzip2 => {
+ read_decompressed("bzip2",
bzip2::read::BzDecoder::new(data)).map(Cow::Owned)
+ }
+ OcfCodec::Xz => {
+ read_decompressed("xz",
liblzma::read::XzDecoder::new(data)).map(Cow::Owned)
+ }
}
}
}
+fn read_decompressed(codec: &str, mut decoder: impl std::io::Read) ->
crate::Result<Vec<u8>> {
+ let mut decompressed = Vec::new();
+ decoder
+ .read_to_end(&mut decompressed)
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("avro ocf: {codec} decompression failed: {e}"),
+ source: None,
+ })?;
+ Ok(decompressed)
+}
+
/// Parse an Avro OCF header and return a streaming block iterator.
pub fn parse_ocf_streaming(bytes: &[u8]) -> crate::Result<(OcfHeader,
OcfBlockIter<'_>)> {
let mut cursor = AvroCursor::new(bytes);
@@ -194,6 +208,8 @@ pub fn parse_ocf_streaming(bytes: &[u8]) ->
crate::Result<(OcfHeader, OcfBlockIt
Some("deflate") => OcfCodec::Deflate,
Some("snappy") => OcfCodec::Snappy,
Some("zstandard") => OcfCodec::Zstandard,
+ Some("bzip2") => OcfCodec::Bzip2,
+ Some("xz") => OcfCodec::Xz,
Some(other) => {
return Err(Error::UnexpectedError {
message: format!("avro ocf: unsupported codec: {other}"),
diff --git a/crates/paimon/src/spec/objects_file.rs
b/crates/paimon/src/spec/objects_file.rs
index a4e60c4f..8116ae56 100644
--- a/crates/paimon/src/spec/objects_file.rs
+++ b/crates/paimon/src/spec/objects_file.rs
@@ -16,7 +16,8 @@
// under the License.
use apache_avro::{
- from_value, to_value, Codec, DeflateSettings, Reader, Schema, Writer,
ZstandardSettings,
+ from_avro_datum, from_value, to_value, Bzip2Settings, Codec,
DeflateSettings, Schema, Writer,
+ XzSettings, ZstandardSettings,
};
use serde::de::DeserializeOwned;
use serde::Serialize;
@@ -25,12 +26,17 @@ pub(crate) const DEFAULT_AVRO_BLOCK_SIZE: usize = 16_000;
pub(crate) const DEFAULT_AVRO_COMPRESSION: &str = "zstd";
pub fn from_avro_bytes<T: DeserializeOwned>(bytes: &[u8]) ->
crate::Result<Vec<T>> {
- Reader::new(bytes)?
- .map(|r| {
- let value = r?;
- from_value::<T>(&value).map_err(crate::Error::from)
- })
- .collect()
+ let (header, mut blocks) = super::avro::ocf::parse_ocf_streaming(bytes)?;
+ let schema = Schema::parse_str(&header.schema_json)?;
+ let mut records = Vec::new();
+ while let Some(block) = blocks.next_block()? {
+ let mut data = block.data.as_ref();
+ for _ in 0..block.object_count {
+ let value = from_avro_datum(&schema, &mut data, None)?;
+ records.push(from_value::<T>(&value)?);
+ }
+ }
+ Ok(records)
}
/// Serialize records into Avro Object Container File bytes.
@@ -74,6 +80,10 @@ pub(crate) fn avro_codec(compression: &str) ->
crate::Result<Codec> {
"null" | "none" | "uncompressed" => Ok(Codec::Null),
"deflate" => Ok(Codec::Deflate(DeflateSettings::default())),
"snappy" => Ok(Codec::Snappy),
+ "bzip2" => Ok(Codec::Bzip2(Bzip2Settings::default())),
+ // Java's CodecFactory.fromString("xz") uses level 6. The Rust Avro
+ // default is 9, which also needs a much larger compression dictionary.
+ "xz" => Ok(Codec::Xz(XzSettings::new(6))),
other => Err(crate::Error::Unsupported {
message: format!("Unsupported Avro compression: {other}"),
}),
@@ -91,6 +101,7 @@ mod tests {
use crate::spec::stats::BinaryTableStats;
use crate::spec::{DataFileMeta, IndexManifestEntry, ManifestFileMeta};
use apache_avro::types::Value;
+ use apache_avro::Reader;
use chrono::{DateTime, Utc};
// Check the record decoded from the OCF writer schema, including fields
whose
@@ -348,6 +359,21 @@ mod tests {
assert_eq!(original, decoded);
}
+ #[test]
+ fn test_roundtrip_manifest_entry_bzip2_and_xz() {
+ let original = vec![manifest_entry()];
+ for compression in ["bzip2", "xz"] {
+ let bytes =
+ to_avro_bytes_with_compression(MANIFEST_ENTRY_SCHEMA,
&original, compression)
+ .unwrap();
+ assert_eq!(from_avro_bytes::<ManifestEntry>(&bytes).unwrap(),
original);
+ assert_eq!(
+ from_avro_bytes_fast::<ManifestEntry>(&bytes).unwrap(),
+ original
+ );
+ }
+ }
+
#[test]
fn test_roundtrip_index_manifest_field_order() {
let entry: IndexManifestEntry =
serde_json::from_value(serde_json::json!({
diff --git a/crates/paimon/src/table/key_normalization/encoded_tests.rs
b/crates/paimon/src/table/key_normalization/encoded_tests.rs
index 0740e3b4..efaab155 100644
--- a/crates/paimon/src/table/key_normalization/encoded_tests.rs
+++ b/crates/paimon/src/table/key_normalization/encoded_tests.rs
@@ -209,6 +209,82 @@ fn
union_key_conversion_never_uses_lossy_assignment_or_text_parsing() {
assert!(normalize_match_key(&input.slice(2, 1),
&DataType::Int32).is_err());
}
+#[test]
+fn typed_null_string_keys_match_without_casting_unselected_dictionary_values()
{
+ let plain: ArrayRef = Arc::new(StringArray::from(vec![None::<&str>; 3]));
+ let dictionary: ArrayRef = Arc::new(
+ DictionaryArray::<Int8Type>::try_new(
+ Int8Array::from(vec![Some(0), Some(1), None, Some(1), Some(0)]),
+ Arc::new(StringArray::from(vec![Some("not an integer"), None])),
+ )
+ .unwrap(),
+ );
+ for input in [plain, dictionary.slice(1, 3)] {
+ let normalized = normalize_match_key(&input,
&DataType::Int32).unwrap();
+ assert_eq!(normalized.data_type(), &DataType::Int32);
+ assert_eq!(normalized.len(), 3);
+ assert_eq!(normalized.null_count(), 3);
+ }
+ assert!(normalize_match_key(&dictionary, &DataType::Int32).is_err());
+ let non_null: ArrayRef = Arc::new(StringArray::from(vec![None,
Some("1")]));
+ assert!(normalize_match_key(&non_null, &DataType::Int32).is_err());
+}
+
+#[test]
+fn all_null_union_keys_need_no_type_compatible_child() {
+ for dense in [true, false] {
+ let fields = UnionFields::try_new(
+ [7, 42],
+ [
+ Field::new("text", DataType::Utf8, true),
+ Field::new("bytes", DataType::Binary, true),
+ ],
+ )
+ .unwrap();
+ let input: ArrayRef = Arc::new(
+ UnionArray::try_new(
+ fields,
+ ScalarBuffer::from(vec![7, 42, 7]),
+ dense.then(|| ScalarBuffer::from(vec![0, 1, 1])),
+ if dense {
+ vec![
+ Arc::new(StringArray::from(vec![Some("not an
integer"), None])) as ArrayRef,
+
Arc::new(BinaryArray::from(vec![Some(b"inactive".as_slice()), None])),
+ ]
+ } else {
+ vec![
+ Arc::new(StringArray::from(vec![
+ Some("not an integer"),
+ Some("inactive"),
+ None,
+ ])) as ArrayRef,
+ Arc::new(BinaryArray::from(vec![
+ Some(b"inactive".as_slice()),
+ None,
+ Some(b"inactive".as_slice()),
+ ])),
+ ]
+ },
+ )
+ .unwrap(),
+ );
+ assert!(normalize_match_key(&input, &DataType::Int32).is_err());
+ let selected = input.slice(1, 2);
+ let dictionary: ArrayRef = Arc::new(
+ DictionaryArray::<Int8Type>::try_new(
+ Int8Array::from(vec![Some(1), None, Some(0)]),
+ selected.clone(),
+ )
+ .unwrap(),
+ );
+ for keys in [selected, dictionary] {
+ let normalized = normalize_match_key(&keys,
&DataType::Int32).unwrap();
+ assert_eq!(normalized.data_type(), &DataType::Int32);
+ assert_eq!(normalized.null_count(), keys.len());
+ }
+ }
+}
+
#[test]
fn typed_null_union_children_match_null_keys_without_casting_their_type() {
for dense in [true, false] {
diff --git a/crates/paimon/src/table/partition_spec.rs
b/crates/paimon/src/table/partition_spec.rs
index 1d1c8615..bcba1dae 100644
--- a/crates/paimon/src/table/partition_spec.rs
+++ b/crates/paimon/src/table/partition_spec.rs
@@ -96,6 +96,23 @@ pub fn parse_partition_spec_value(
| DataType::Decimal(_) => {}
_ => return Err(invalid()),
}
+ if matches!(data_type, DataType::Timestamp(ty) if ty.precision() == 0) {
+ let digits = text.strip_prefix('-').unwrap_or(text);
+ if !digits.is_empty() && digits.bytes().all(|byte|
byte.is_ascii_digit()) {
+ // Java converts numeric seconds with long multiplication. Overflow
+ // can leave subsecond milliseconds, which a partition Datum can
+ // retain even though an Arrow second-precision array cannot.
+ return text
+ .parse::<i64>()
+ .map(|seconds| {
+ Some(Datum::Timestamp {
+ millis: seconds.wrapping_mul(1_000),
+ nanos: 0,
+ })
+ })
+ .map_err(|_| invalid());
+ }
+ }
let input: ArrayRef = Arc::new(StringArray::from(vec![text]));
let array = crate::arrow::cast_array_for_schema_evolution(
&input,
@@ -163,14 +180,19 @@ fn parse_time_spec(text: &str) -> crate::Result<i32> {
}
None => (0, ""),
};
- // Java's fraction parser rounds using the fourth character.
+ // Java's fraction parser counts UTF-16 code units, including the fourth
+ // unit used for rounding. Supplementary characters occupy two positions.
let mut millis = 0;
- let mut chars = fraction.chars();
+ let mut chars = fraction.encode_utf16();
for multiplier in [100, 10, 1] {
let Some(ch) = chars.next() else { break };
- millis += ch.to_digit(10).filter(|_| ch.is_ascii()).unwrap_or(0) as
i32 * multiplier;
+ let digit = ch
+ .checked_sub(u16::from(b'0'))
+ .filter(|digit| *digit < 10)
+ .unwrap_or(0);
+ millis += i32::from(digit) * multiplier;
}
- if chars.next().is_some_and(|ch| ch >= '5') {
+ if chars.next().is_some_and(|ch| ch >= u16::from(b'5')) {
millis += 1;
}
Ok(hours
@@ -207,6 +229,43 @@ mod tests {
}
}
+ #[test]
+ fn time_partition_fractions_use_java_utf16_character_positions() {
+ let time = DataType::Time(TimeType::new(3).unwrap());
+ for (text, millis) in [
+ ("12:34:56.😀123", 45_296_001),
+ ("12:34:56.1😀23", 45_296_100),
+ ("12:34:56.12😀3", 45_296_121),
+ ("12:34:56.123😀", 45_296_124),
+ ] {
+ assert_eq!(
+ parse_partition_spec_value(text, &time).unwrap(),
+ Some(Datum::Time(millis)),
+ "{text}"
+ );
+ }
+ }
+
+ #[test]
+ fn
numeric_second_precision_partition_specs_keep_java_wrapped_milliseconds() {
+ let data_type = DataType::Timestamp(TimestampType::new(0).unwrap());
+ for (text, millis) in [
+ ("18446744073709552", 384),
+ ("-18446744073709552", -384),
+ ("9223372036854775807", -1000),
+ ("1", 1000),
+ ] {
+ assert_eq!(
+ parse_partition_spec_value(text, &data_type).unwrap(),
+ Some(Datum::Timestamp { millis, nanos: 0 }),
+ "{text}"
+ );
+ }
+ for text in ["9223372036854775808", "+1", "1.5"] {
+ assert!(parse_partition_spec_value(text, &data_type).is_err());
+ }
+ }
+
#[test]
fn partition_specs_use_java_typed_text_conversion() {
assert_eq!(
diff --git a/crates/paimon/testdata/avro_codec/GenerateCodecFixtures.java
b/crates/paimon/testdata/avro_codec/GenerateCodecFixtures.java
new file mode 100644
index 00000000..89305c2e
--- /dev/null
+++ b/crates/paimon/testdata/avro_codec/GenerateCodecFixtures.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.avro.Schema;
+import org.apache.avro.file.CodecFactory;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.nio.ByteBuffer;
+
+/** Produces the same records and block boundaries as generate.py. */
+public final class GenerateCodecFixtures {
+ public static void main(String[] args) throws Exception {
+ File directory = new File(args[0]);
+ Schema schema = new Schema.Parser().parse(new File(directory,
"records.avsc"));
+ byte[] sync = new byte[16];
+ for (int i = 0; i < sync.length; i++) {
+ sync[i] = (byte) i;
+ }
+ byte[] payload = new byte[32];
+ for (int i = 0; i < payload.length; i++) {
+ payload[i] = (byte) i;
+ }
+ for (String codec : new String[] {"bzip2", "xz"}) {
+ try (DataFileWriter<GenericRecord> writer =
+ new DataFileWriter<>(new GenericDatumWriter<>(schema))) {
+ writer.setCodec(CodecFactory.fromString(codec));
+ writer.create(
+ schema,
+ new FileOutputStream(new File(directory, "java_" +
codec + ".avro")),
+ sync);
+ append(writer, schema, -7L, "before-epoch", new byte[] {0, 1,
(byte) 255});
+ append(writer, schema, 0L, "\u4e8c", new byte[0]);
+ append(writer, schema, Long.MAX_VALUE,
+ "end" + new String(new char[96]).replace('\0', 'x'),
payload);
+ }
+ }
+ }
+
+ private static void append(
+ DataFileWriter<GenericRecord> writer,
+ Schema schema,
+ long id,
+ String label,
+ byte[] payload) throws Exception {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put("id", id);
+ record.put("label", label);
+ record.put("payload", ByteBuffer.wrap(payload));
+ writer.append(record);
+ writer.sync();
+ }
+}
diff --git a/crates/paimon/testdata/avro_codec/README.md
b/crates/paimon/testdata/avro_codec/README.md
new file mode 100644
index 00000000..374683cb
--- /dev/null
+++ b/crates/paimon/testdata/avro_codec/README.md
@@ -0,0 +1,38 @@
+# Avro manifest codec fixtures
+
+These four Avro Object Container Files use the same three records and schema
+(`records.avsc`), with one record per block and sync marker bytes `00` through
+`0f`. They exercise the `bzip2` and `xz` codec names used by Java Paimon and
+PyPaimon manifests. The records cover a negative long, zero, `Long.MAX_VALUE`,
+UTF-8 text, empty bytes, and non-text bytes.
+
+Producers:
+
+- `java_*.avro`: Apache Avro Java **1.11.4**, the version in the Paimon Java
+ build, using `CodecFactory.fromString`. Generated with
+ `GenerateCodecFixtures.java`, JDK 8, Commons Compress 1.26.2, and XZ 1.9.
+- `fastavro_*.avro`: fastavro **1.12.2**, generated with `generate.py` using
+ the ordinary `fastavro.writer` API used by PyPaimon.
+
+The Rust tests decode every record, check block boundaries and encoded offsets,
+and check damaged compressed blocks and container sync markers. The fixture
+files are static, so running the Rust tests does not require Java or Python.
+
+To regenerate the Python fixtures:
+
+```sh
+python3 crates/paimon/testdata/avro_codec/generate.py
+```
+
+To regenerate the Java fixtures, set `AVRO_CP` to a classpath containing the
+Avro version and its dependencies above, plus Jackson and SLF4J, then run from
+the repository root:
+
+```sh
+mkdir -p /tmp/paimon-avro-codec-fixtures
+javac -source 8 -target 8 -encoding UTF-8 -cp "$AVRO_CP" \
+ -d /tmp/paimon-avro-codec-fixtures \
+ crates/paimon/testdata/avro_codec/GenerateCodecFixtures.java
+java -cp "/tmp/paimon-avro-codec-fixtures:$AVRO_CP" \
+ GenerateCodecFixtures crates/paimon/testdata/avro_codec
+```
diff --git a/crates/paimon/testdata/avro_codec/fastavro_bzip2.avro
b/crates/paimon/testdata/avro_codec/fastavro_bzip2.avro
new file mode 100644
index 00000000..f4425840
Binary files /dev/null and
b/crates/paimon/testdata/avro_codec/fastavro_bzip2.avro differ
diff --git a/crates/paimon/testdata/avro_codec/fastavro_xz.avro
b/crates/paimon/testdata/avro_codec/fastavro_xz.avro
new file mode 100644
index 00000000..2222a53c
Binary files /dev/null and b/crates/paimon/testdata/avro_codec/fastavro_xz.avro
differ
diff --git a/crates/paimon/testdata/avro_codec/generate.py
b/crates/paimon/testdata/avro_codec/generate.py
new file mode 100644
index 00000000..dddaf0c3
--- /dev/null
+++ b/crates/paimon/testdata/avro_codec/generate.py
@@ -0,0 +1,35 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Regenerate the fastavro fixtures, with one record per OCF block."""
+
+import json
+from pathlib import Path
+
+import fastavro
+
+fixture_dir = Path(__file__).resolve().parent
+schema = json.loads((fixture_dir / 'records.avsc').read_text())
+records = [
+ {'id': -7, 'label': 'before-epoch', 'payload': bytes([0, 1, 255])},
+ {'id': 0, 'label': '\u4e8c', 'payload': b''},
+ {'id': (1 << 63) - 1, 'label': 'end' + 'x' * 96, 'payload':
bytes(range(32))},
+]
+for codec in ['bzip2', 'xz']:
+ with (fixture_dir / ('fastavro_' + codec + '.avro')).open('wb') as output:
+ fastavro.writer(output, schema, records, codec=codec, sync_interval=1,
+ sync_marker=bytes(range(16)))
diff --git a/crates/paimon/testdata/avro_codec/java_bzip2.avro
b/crates/paimon/testdata/avro_codec/java_bzip2.avro
new file mode 100644
index 00000000..b3152eb5
Binary files /dev/null and b/crates/paimon/testdata/avro_codec/java_bzip2.avro
differ
diff --git a/crates/paimon/testdata/avro_codec/java_xz.avro
b/crates/paimon/testdata/avro_codec/java_xz.avro
new file mode 100644
index 00000000..21dddb8c
Binary files /dev/null and b/crates/paimon/testdata/avro_codec/java_xz.avro
differ
diff --git a/crates/paimon/testdata/avro_codec/records.avsc
b/crates/paimon/testdata/avro_codec/records.avsc
new file mode 100644
index 00000000..42e6fa04
--- /dev/null
+++ b/crates/paimon/testdata/avro_codec/records.avsc
@@ -0,0 +1,9 @@
+{
+ "type": "record",
+ "name": "CodecFixture",
+ "fields": [
+ {"name": "id", "type": "long"},
+ {"name": "label", "type": "string"},
+ {"name": "payload", "type": "bytes"}
+ ]
+}
diff --git a/scripts/release_licenses.py b/scripts/release_licenses.py
index 5785cf80..7b2fda97 100644
--- a/scripts/release_licenses.py
+++ b/scripts/release_licenses.py
@@ -186,7 +186,7 @@ BUNDLED_COMPONENTS = (
component_url="https://github.com/tukaani-project/xz/tree/v5.8.3";,
license_name="BSD Zero Clause License",
anchor="bundled-xz-utils-0bsd",
- components=("python",),
+ components=("python", "go"),
required=True,
relationship="statically linked through",
crate_version="0.4.7",
diff --git a/scripts/tests/test_release_licenses.py
b/scripts/tests/test_release_licenses.py
index 2cfe1b8a..ca882fb6 100644
--- a/scripts/tests/test_release_licenses.py
+++ b/scripts/tests/test_release_licenses.py
@@ -22,7 +22,9 @@ from pathlib import Path
from scripts.release_licenses import (
LicenseCorrection,
PYTHON_MIT_CORRECTIONS,
+ Report,
binary_notice,
+ bundled_component_html,
correction_html,
resolved_packages,
)
@@ -217,5 +219,42 @@ class ResolvedPackagesTest(unittest.TestCase):
self.assertNotIn("Optional dependency notice", notice)
+class ManifestCodecLicenseTest(unittest.TestCase):
+ def test_python_and_go_reports_include_statically_linked_xz_license(self)
-> None:
+ with tempfile.TemporaryDirectory() as temp:
+ root = Path(temp)
+ crate_dir = root / "liblzma-sys"
+ license_path = crate_dir / "xz" / "COPYING.0BSD"
+ license_path.parent.mkdir(parents=True)
+ license_path.write_text("XZ Utils bundled license\n",
encoding="utf-8")
+ package = {
+ "name": "liblzma-sys",
+ "id": "liblzma-sys-package",
+ "version": "0.4.7",
+ "manifest_path": str(crate_dir / "Cargo.toml"),
+ }
+ metadata = {
+ "normal_packages": {("liblzma-sys", "0.4.7")},
+ "packages": [package],
+ "resolve": {
+ "nodes": [{"id": package["id"], "features": ["static"]}],
+ },
+ }
+ report_text = (
+ '<ul><li><a href="https://crates.io/crates/liblzma-sys";>'
+ 'liblzma-sys 0.4.7</a></li></ul>'
+ )
+ for component in ("python", "go"):
+ with self.subTest(component=component):
+ report = Report(component, "", "aarch64-apple-darwin", "")
+ result = bundled_component_html(root, report, report_text,
metadata)
+ self.assertIn("XZ Utils bundled license", result)
+ self.assertIn("statically linked through", result)
+ metadata["resolve"]["nodes"][0]["features"] = []
+ with self.assertRaisesRegex(RuntimeError, "missing
required features"):
+ bundled_component_html(root, report, report_text,
metadata)
+ metadata["resolve"]["nodes"][0]["features"] = ["static"]
+
+
if __name__ == "__main__":
unittest.main()
