Skip to content

Commit dee387b

Browse files
committed
fix: support reading compressed metadata
The spec mentions that metadata files "may be compressed with GZIP", here: https://iceberg.apache.org/spec/#table-metadata-and-snapshots
1 parent 833739e commit dee387b

File tree

5 files changed

+53
-4
lines changed

5 files changed

+53
-4
lines changed

Cargo.lock

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

Cargo.toml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ enum-ordinalize = "4.3.0"
7171
env_logger = "0.11.8"
7272
expect-test = "1"
7373
faststr = "0.2.31"
74+
flate2 = "1.1.5"
7475
fnv = "1.0.7"
7576
fs-err = "3.1.0"
7677
futures = "0.3"

crates/iceberg/Cargo.toml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@ bytes = { workspace = true }
6363
chrono = { workspace = true }
6464
derive_builder = { workspace = true }
6565
expect-test = { workspace = true }
66+
flate2 = { workspace = true }
6667
fnv = { workspace = true }
6768
futures = { workspace = true }
6869
itertools = { workspace = true }

crates/iceberg/src/spec/table_metadata.rs

Lines changed: 49 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,12 @@ use std::cmp::Ordering;
2222
use std::collections::HashMap;
2323
use std::fmt::{Display, Formatter};
2424
use std::hash::Hash;
25+
use std::io::Read as _;
2526
use std::sync::Arc;
2627

2728
use _serde::TableMetadataEnum;
2829
use chrono::{DateTime, Utc};
30+
use flate2::read::GzDecoder;
2931
use serde::{Deserialize, Serialize};
3032
use serde_repr::{Deserialize_repr, Serialize_repr};
3133
use uuid::Uuid;
@@ -413,9 +415,30 @@ impl TableMetadata {
413415
file_io: &FileIO,
414416
metadata_location: impl AsRef<str>,
415417
) -> Result<TableMetadata> {
418+
let metadata_location = metadata_location.as_ref();
416419
let input_file = file_io.new_input(metadata_location)?;
417420
let metadata_content = input_file.read().await?;
418-
let metadata = serde_json::from_slice::<TableMetadata>(&metadata_content)?;
421+
422+
// Check if the file is compressed by looking for the gzip "magic number".
423+
let metadata = if metadata_content.len() > 2
424+
&& metadata_content[0] == 0x1F
425+
&& metadata_content[1] == 0x8B
426+
{
427+
let mut decoder = GzDecoder::new(metadata_content.as_ref());
428+
let mut decompressed_data = Vec::new();
429+
decoder.read_to_end(&mut decompressed_data).map_err(|e| {
430+
Error::new(
431+
ErrorKind::DataInvalid,
432+
"Trying to read compressed metadata file",
433+
)
434+
.with_context("file_path", metadata_location)
435+
.with_source(e)
436+
})?;
437+
serde_json::from_slice(&decompressed_data)?
438+
} else {
439+
serde_json::from_slice(&metadata_content)?
440+
};
441+
419442
Ok(metadata)
420443
}
421444

@@ -1314,6 +1337,7 @@ impl SnapshotLog {
13141337
mod tests {
13151338
use std::collections::HashMap;
13161339
use std::fs;
1340+
use std::io::Write as _;
13171341
use std::sync::Arc;
13181342

13191343
use anyhow::Result;
@@ -3047,6 +3071,30 @@ mod tests {
30473071
assert_eq!(read_metadata, original_metadata);
30483072
}
30493073

3074+
#[tokio::test]
3075+
async fn test_table_metadata_read_compressed() {
3076+
let temp_dir = TempDir::new().unwrap();
3077+
let metadata_location = temp_dir.path().join("v1.gz.metadata.json");
3078+
3079+
let original_metadata: TableMetadata = get_test_table_metadata("TableMetadataV2Valid.json");
3080+
let json = serde_json::to_string(&original_metadata).unwrap();
3081+
3082+
let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
3083+
encoder.write_all(json.as_bytes()).unwrap();
3084+
std::fs::write(&metadata_location, encoder.finish().unwrap())
3085+
.expect("failed to write metadata");
3086+
3087+
// Read the metadata back
3088+
let file_io = FileIOBuilder::new_fs_io().build().unwrap();
3089+
let metadata_location = metadata_location.to_str().unwrap();
3090+
let read_metadata = TableMetadata::read_from(&file_io, metadata_location)
3091+
.await
3092+
.unwrap();
3093+
3094+
// Verify the metadata matches
3095+
assert_eq!(read_metadata, original_metadata);
3096+
}
3097+
30503098
#[tokio::test]
30513099
async fn test_table_metadata_read_nonexistent_file() {
30523100
// Create a FileIO instance

crates/iceberg/src/table.rs

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -292,9 +292,7 @@ impl StaticTable {
292292
table_ident: TableIdent,
293293
file_io: FileIO,
294294
) -> Result<Self> {
295-
let metadata_file = file_io.new_input(metadata_location)?;
296-
let metadata_file_content = metadata_file.read().await?;
297-
let metadata = serde_json::from_slice::<TableMetadata>(&metadata_file_content)?;
295+
let metadata = TableMetadata::read_from(&file_io, metadata_location).await?;
298296

299297
let table = Table::builder()
300298
.metadata(metadata)

0 commit comments

Comments
 (0)