gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-16 01:48:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(table-catalog): support apache-avro 0.22 (#5975)

Browse Source 
fix(avro): support apache-avro 0.22

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
2026-08-12 15:08:52 +08:00
committed by GitHub
parent 3fdf2964c8
commit 537d34b8cd
7 changed files with 142 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/table_catalog/tests.rs
+6 -6
View File
@@ -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<u8> {
"#,
)
.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)),
rustfs/src/table_catalog/maintenance/planner.rs
+7 -4
View File
@@ -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<Vec<u8>> {
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)),
rustfs/src/table_catalog/tests.rs
+54 -14
View File
@@ -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<u8> {
"#,
)
.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<u8> {
"#,
)
.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<u8> {
"#,
)
.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<u8>
"#,
)
.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<u8> {
"#,
)
.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)),
rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest-list.avro
Vendored
BIN
View File
Binary file not shown.
rustfs/tests/fixtures/table_catalog/apache-avro-0.21-manifest.avro
Vendored
BIN
View File
Binary file not shown.