Skip to content

feat(parquet): wire in additional parquet writer settings #2887

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xanderbailey wants to merge 3 commits into
base: main
Choose a base branch
from
+416 −13
Open

feat(parquet): wire in additional parquet writer settings #2887

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ca7d5ba
feat(paquet): wire in additional parquet writer settings
xanderbailey Jul 24, 2026
67c28d8
fix compression level
xanderbailey Jul 30, 2026
da6cb46
fmt
xanderbailey Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion crates/iceberg/public-api.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2750,6 +2750,12 @@ pub iceberg::spec::TableProperties::max_ref_age_ms: i64
pub iceberg::spec::TableProperties::max_snapshot_age_ms: i64
pub iceberg::spec::TableProperties::metadata_compression_codec: iceberg::compression::CompressionCodec
pub iceberg::spec::TableProperties::min_snapshots_to_keep: usize
pub iceberg::spec::TableProperties::parquet_compression_codec: alloc::string::String
pub iceberg::spec::TableProperties::parquet_compression_level: core::option::Option<i32>
pub iceberg::spec::TableProperties::parquet_dict_size_bytes: usize
pub iceberg::spec::TableProperties::parquet_page_row_limit: usize
pub iceberg::spec::TableProperties::parquet_page_size_bytes: usize
pub iceberg::spec::TableProperties::parquet_row_group_size_bytes: usize
pub iceberg::spec::TableProperties::write_datafusion_fanout_enabled: bool
pub iceberg::spec::TableProperties::write_format_default: alloc::string::String
pub iceberg::spec::TableProperties::write_metadata_path: core::option::Option<alloc::string::String>
Expand Down Expand Up @@ -2798,6 +2804,17 @@ pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE: &
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT: usize
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT: usize
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT: usize
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT: usize
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES: &str
pub const iceberg::spec::TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT: usize
pub const iceberg::spec::TableProperties::PROPERTY_SNAPSHOT_COUNT: &str
pub const iceberg::spec::TableProperties::PROPERTY_UUID: &str
pub const iceberg::spec::TableProperties::PROPERTY_WRITE_METADATA_PATH: &str
Expand Down Expand Up @@ -3270,7 +3287,7 @@ pub async fn iceberg::writer::file_writer::ParquetWriter::close(self) -> iceberg
pub async fn iceberg::writer::file_writer::ParquetWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()>
pub struct iceberg::writer::file_writer::ParquetWriterBuilder
impl iceberg::writer::file_writer::ParquetWriterBuilder
pub fn iceberg::writer::file_writer::ParquetWriterBuilder::from_table_properties(table_props: &iceberg::spec::TableProperties, schema: iceberg::spec::SchemaRef) -> Self
pub fn iceberg::writer::file_writer::ParquetWriterBuilder::from_table_properties(table_props: &iceberg::spec::TableProperties, schema: iceberg::spec::SchemaRef) -> iceberg::Result<Self>
pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef) -> Self
pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new_with_match_mode(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef, match_mode: iceberg::arrow::FieldMatchMode) -> Self
pub fn iceberg::writer::file_writer::ParquetWriterBuilder::with_match_mode(self, match_mode: iceberg::arrow::FieldMatchMode) -> Self
Expand Down
172 changes: 172 additions & 0 deletions crates/iceberg/src/spec/table_properties.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,28 @@ where
})
}

/// Parse an optional property, returning `None` when the key is absent and an
/// error when the value is present but fails to parse.
fn parse_optional_property<T: FromStr>(
properties: &HashMap<String, String>,
key: &str,
) -> Result<Option<T>>
where
<T as FromStr>::Err: Display,
{
properties
.get(key)
.map(|value| {
value.parse::<T>().map_err(|e| {
Error::new(
ErrorKind::DataInvalid,
format!("Invalid value for {key}: {e}"),
)
})
})
.transpose()
}

/// Strips trailing slashes from a location, preserving a bare URI scheme root
fn strip_trailing_slash(path: &str) -> &str {
let mut path = path;
Expand Down Expand Up @@ -168,6 +190,20 @@ pub struct TableProperties {
pub cdc_max_chunk_size: usize,
/// Content-defined chunking normalization level (gearhash bit adjustment).
pub cdc_norm_level: i32,
/// Parquet compression codec name (e.g. `zstd`, `gzip`). Validated when the
/// writer is built, not when properties are parsed.
pub parquet_compression_codec: String,
Comment on lines +193 to +195

@dannycjones dannycjones Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Validated when the writer is built, not when properties are parsed.

This is a good tenet for the whole module - i.e. we push validation to access, since we may not care if we're writing ORC files (as an example).

If you end up pushing another commit, I suggest we add it to the module comment.

/// Parquet compression level for codecs that accept one. `None` uses the
/// codec's default level.
pub parquet_compression_level: Option<i32>,
/// Approximate maximum Parquet row group size in bytes.
pub parquet_row_group_size_bytes: usize,
/// Approximate maximum Parquet data page size in bytes.
pub parquet_page_size_bytes: usize,
/// Maximum number of rows per Parquet data page.
pub parquet_page_row_limit: usize,
/// Approximate maximum Parquet dictionary page size in bytes.
pub parquet_dict_size_bytes: usize,
/// The master key id used to encrypt this table's manifest list and data
/// files. `None` if `encryption.key-id` is not set.
pub encryption_key_id: Option<String>,
Expand Down Expand Up @@ -317,6 +353,36 @@ impl TableProperties {
/// Default matches `parquet::file::properties::DEFAULT_CDC_NORM_LEVEL`.
pub const PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32 = 0;

/// Compression codec for Parquet data files (e.g. `zstd`, `gzip`, `snappy`,
/// `lz4`, `brotli`, `uncompressed`). The codec name is validated when the
/// writer is built, not when properties are parsed.
pub const PROPERTY_PARQUET_COMPRESSION_CODEC: &str = "write.parquet.compression-codec";
/// Default Parquet compression codec.
pub const PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT: &str = "zstd";

@xanderbailey xanderbailey Jul 24, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Have called this out in the PR description but current default is actually uncompressed from what I can see.

This opens up a question generally as to if we should copy the Java defaults for these settings which might be more intuitive for user or keep the defaults as the arrow-rs defaults... I don't have a strong opinion here so interested to hear people's thoughts.

@zakariya-s zakariya-s Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

On the matter of inconsistency, there are other places where arrow-rs has different defaults to parquet-java, like the compression level for zstd too:

@dannycjones dannycjones Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm aligned with Matt's "Principle of Least Astonishment" and with the outcome from the Rust sync - let's adopt Java defaults unless we have a reason to diverge.

/// Compression level for Parquet data files, for codecs that take one
/// (`gzip`, `zstd`, `brotli`). When unset, the codec's default level is used.
pub const PROPERTY_PARQUET_COMPRESSION_LEVEL: &str = "write.parquet.compression-level";

/// Approximate maximum size of a Parquet row group in bytes.
pub const PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES: &str = "write.parquet.row-group-size-bytes";
/// Default Parquet row group size in bytes.
pub const PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT: usize = 128 * 1024 * 1024;

/// Approximate maximum size of a Parquet data page in bytes.
pub const PROPERTY_PARQUET_PAGE_SIZE_BYTES: &str = "write.parquet.page-size-bytes";
/// Default Parquet page size in bytes.
pub const PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT: usize = 1024 * 1024;

/// Maximum number of rows per Parquet data page.
pub const PROPERTY_PARQUET_PAGE_ROW_LIMIT: &str = "write.parquet.page-row-limit";
/// Default Parquet page row limit.
pub const PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT: usize = 20000;

/// Approximate maximum size of the Parquet dictionary page in bytes.
pub const PROPERTY_PARQUET_DICT_SIZE_BYTES: &str = "write.parquet.dict-size-bytes";
/// Default Parquet dictionary page size in bytes.
pub const PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT: usize = 2 * 1024 * 1024;

/// Property key for the master key id used to encrypt the table's manifest
/// list and data files as defined in https://iceberg.apache.org/docs/nightly/encryption/.
pub const PROPERTY_ENCRYPTION_KEY_ID: &str = "encryption.key-id";
Expand Down Expand Up @@ -413,6 +479,35 @@ impl TryFrom<&HashMap<String, String>> for TableProperties {
TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL,
TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT,
)?,
parquet_compression_codec: parse_property(
props,
TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC,
TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT.to_string(),
)?,
parquet_compression_level: parse_optional_property(
props,
TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL,
)?,
parquet_row_group_size_bytes: parse_property(
props,
TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES,
TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT,
)?,
parquet_page_size_bytes: parse_property(
props,
TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES,
TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT,
)?,
parquet_page_row_limit: parse_property(
props,
TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT,
TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT,
)?,
parquet_dict_size_bytes: parse_property(
props,
TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES,
TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT,
)?,
encryption_key_id: props
.get(TableProperties::PROPERTY_ENCRYPTION_KEY_ID)
.cloned(),
Expand Down Expand Up @@ -947,4 +1042,81 @@ mod tests {
let tp = TableProperties::try_from(&props).unwrap();
assert!(!tp.cdc_enabled);
}

#[test]
fn test_parquet_sizing_defaults() {
let tp = TableProperties::try_from(&HashMap::new()).unwrap();
assert_eq!(
tp.parquet_compression_codec,
TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT
);
assert_eq!(tp.parquet_compression_level, None);
assert_eq!(
tp.parquet_row_group_size_bytes,
TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT
);
assert_eq!(
tp.parquet_page_size_bytes,
TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT
);
assert_eq!(
tp.parquet_page_row_limit,
TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT
);
assert_eq!(
tp.parquet_dict_size_bytes,
TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT
);
}

#[test]
fn test_parquet_sizing_overrides() {
let props = HashMap::from([
(
TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC.to_string(),
"gzip".to_string(),
),
(
TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL.to_string(),
"4".to_string(),
),
(
TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES.to_string(),
"1048576".to_string(),
),
(
TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES.to_string(),
"65536".to_string(),
),
(
TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT.to_string(),
"5000".to_string(),
),
(
TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES.to_string(),
"131072".to_string(),
),
]);
let tp = TableProperties::try_from(&props).unwrap();
assert_eq!(tp.parquet_compression_codec, "gzip");
assert_eq!(tp.parquet_compression_level, Some(4));
assert_eq!(tp.parquet_row_group_size_bytes, 1048576);
assert_eq!(tp.parquet_page_size_bytes, 65536);
assert_eq!(tp.parquet_page_row_limit, 5000);
assert_eq!(tp.parquet_dict_size_bytes, 131072);
}

#[test]
fn test_parquet_invalid_sizing_rejected() {
let props = HashMap::from([(
TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES.to_string(),
"not_a_number".to_string(),
)]);
let err = TableProperties::try_from(&props).unwrap_err();
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(
err.to_string()
.contains(TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES)
);
}
}
Loading
Loading