iceberg/spec/
table_properties.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::HashMap;
19use std::fmt::Display;
20use std::str::FromStr;
21
22use crate::compression::CompressionCodec;
23use crate::error::{Error, ErrorKind, Result};
24use crate::util::location::strip_trailing_slash;
25
26fn parse_property<T: FromStr>(
27    properties: &HashMap<String, String>,
28    key: &str,
29    default: T,
30) -> Result<T>
31where
32    <T as FromStr>::Err: Display,
33{
34    properties.get(key).map_or(Ok(default), |value| {
35        value.parse::<T>().map_err(|e| {
36            Error::new(
37                ErrorKind::DataInvalid,
38                format!("Invalid value for {key}: {e}"),
39            )
40        })
41    })
42}
43
44/// Parse an optional property, returning `None` when the key is absent and an
45/// error when the value is present but fails to parse.
46fn parse_optional_property<T: FromStr>(
47    properties: &HashMap<String, String>,
48    key: &str,
49) -> Result<Option<T>>
50where
51    <T as FromStr>::Err: Display,
52{
53    properties
54        .get(key)
55        .map(|value| {
56            value.parse::<T>().map_err(|e| {
57                Error::new(
58                    ErrorKind::DataInvalid,
59                    format!("Invalid value for {key}: {e}"),
60                )
61            })
62        })
63        .transpose()
64}
65
66fn parse_location_property(
67    properties: &HashMap<String, String>,
68    key: &str,
69) -> Result<Option<String>> {
70    properties
71        .get(key)
72        .map(|path| {
73            if path.is_empty() {
74                return Err(Error::new(
75                    ErrorKind::DataInvalid,
76                    format!("Invalid value for {key}: path must not be empty"),
77                ));
78            }
79
80            Ok(strip_trailing_slash(path).to_string())
81        })
82        .transpose()
83}
84
85/// Parse compression codec for metadata files from table properties.
86/// Retrieves the compression codec property, applies defaults, and parses the value.
87/// Only "none" (or empty string) and "gzip" are supported for metadata compression.
88///
89/// # Arguments
90///
91/// * `properties` - HashMap containing table properties
92///
93/// # Errors
94///
95/// Returns an error if the codec is not "none", "", or "gzip" (case-insensitive).
96/// Lz4 and Zstd are not supported for metadata file compression.
97pub(crate) fn parse_metadata_file_compression(
98    properties: &HashMap<String, String>,
99) -> Result<CompressionCodec> {
100    let value = properties
101        .get(TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC)
102        .map(|s| s.as_str())
103        .unwrap_or(TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC_DEFAULT);
104
105    // Handle empty string as None
106    if value.is_empty() {
107        return Ok(CompressionCodec::None);
108    }
109
110    // Lowercase the value for case-insensitive parsing
111    let lowercase_value = value.to_lowercase();
112
113    // Use serde to parse the codec (which has rename_all = "lowercase")
114    let codec: CompressionCodec = serde_json::from_value(serde_json::Value::String(
115        lowercase_value,
116    ))
117    .map_err(|_| {
118        Error::new(
119            ErrorKind::DataInvalid,
120            format!(
121                "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported.",
122                CompressionCodec::None.name(),
123                CompressionCodec::gzip_default().name()
124            ),
125        )
126    })?;
127
128    // Validate that only None and Gzip are used for metadata
129    match codec {
130        CompressionCodec::None | CompressionCodec::Gzip(_) => Ok(codec),
131        _ => Err(Error::new(
132            ErrorKind::DataInvalid,
133            format!(
134                "Invalid metadata compression codec: {value}. Only '{}' and '{}' are supported for metadata files.",
135                CompressionCodec::None.name(),
136                CompressionCodec::gzip_default().name()
137            ),
138        )),
139    }
140}
141
142/// Parse the Parquet data-file compression codec (`write.parquet.compression-codec`)
143/// and fold in the compression level (`write.parquet.compression-level`) for the
144/// codecs that accept one (`zstd`, `gzip`, `brotli`).
145fn parse_parquet_compression(properties: &HashMap<String, String>) -> Result<CompressionCodec> {
146    let value = properties
147        .get(TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC)
148        .map(|s| s.as_str())
149        .unwrap_or(TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT);
150
151    let codec: CompressionCodec =
152        serde_json::from_value(serde_json::Value::String(value.to_lowercase())).map_err(|_| {
153            Error::new(
154                ErrorKind::DataInvalid,
155                format!(
156                    "Invalid Parquet compression codec: {value}. Supported codecs: \
157                     uncompressed, snappy, gzip, lzo, brotli, lz4, lz4_raw, zstd"
158                ),
159            )
160        })?;
161
162    let level: Option<u8> = parse_optional_property(
163        properties,
164        TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL,
165    )?;
166
167    Ok(match (codec, level) {
168        (CompressionCodec::Zstd(_), Some(level)) => CompressionCodec::Zstd(level),
169        (CompressionCodec::Gzip(_), Some(level)) => CompressionCodec::Gzip(level),
170        (CompressionCodec::Brotli(_), Some(level)) => CompressionCodec::Brotli(level),
171        (codec, _) => codec,
172    })
173}
174
175/// Parse boolean property case insensitively
176/// Rust standard library only accepts "true" and "false", see https://doc.rust-lang.org/std/primitive.bool.html#method.from_str
177/// Users might accidentally trigger fallback with valid configuration values such as "False" or "True"
178fn parse_property_bool(
179    properties: &HashMap<String, String>,
180    key: &str,
181    default: bool,
182) -> Result<bool> {
183    properties.get(key).map_or(Ok(default), |value| {
184        value.to_lowercase().parse::<bool>().map_err(|e| {
185            Error::new(
186                ErrorKind::DataInvalid,
187                format!("Invalid value for {key}: {e}"),
188            )
189        })
190    })
191}
192
193/// TableProperties that contains the properties of a table.
194#[derive(Debug)]
195pub struct TableProperties {
196    /// The number of times to retry a commit.
197    pub commit_num_retries: usize,
198    /// The minimum wait time between retries.
199    pub commit_min_retry_wait_ms: u64,
200    /// The maximum wait time between retries.
201    pub commit_max_retry_wait_ms: u64,
202    /// The total timeout for commit retries.
203    pub commit_total_retry_timeout_ms: u64,
204    /// The default format for files.
205    pub write_format_default: String,
206    /// The target file size for files.
207    pub write_target_file_size_bytes: usize,
208    /// Base directory for metadata files (manifests, manifest lists), with any
209    /// trailing slash trimmed. `None` if `write.metadata.path` is not set.
210    pub write_metadata_path: Option<String>,
211    /// Compression codec for metadata files (JSON)
212    pub metadata_compression_codec: CompressionCodec,
213    /// Whether to use `FanoutWriter` for partitioned tables.
214    pub write_datafusion_fanout_enabled: bool,
215    /// Whether garbage collection is enabled on drop.
216    /// When `false`, data files will not be deleted when a table is dropped.
217    pub gc_enabled: bool,
218    /// Default maximum age of a snapshot to keep when expiring snapshots.
219    pub max_snapshot_age_ms: i64,
220    /// Default minimum number of snapshots to keep per branch when expiring snapshots.
221    pub min_snapshots_to_keep: usize,
222    /// Default maximum age of a snapshot reference to keep when expiring snapshots.
223    pub max_ref_age_ms: i64,
224    /// Whether content-defined chunking is enabled.
225    /// `true` only when `write.parquet.content-defined-chunking.enabled = "true"`.
226    pub cdc_enabled: bool,
227    /// Content-defined chunking minimum chunk size in bytes.
228    pub cdc_min_chunk_size: usize,
229    /// Content-defined chunking maximum chunk size in bytes.
230    pub cdc_max_chunk_size: usize,
231    /// Content-defined chunking normalization level (gearhash bit adjustment).
232    pub cdc_norm_level: i32,
233    /// Parquet compression codec for data files, with the resolved compression
234    /// level folded in (from `write.parquet.compression-level`, or the codec's
235    /// default when unset).
236    pub parquet_compression_codec: CompressionCodec,
237    /// Approximate maximum Parquet row group size in bytes.
238    pub parquet_row_group_size_bytes: usize,
239    /// Approximate maximum Parquet data page size in bytes.
240    pub parquet_page_size_bytes: usize,
241    /// Maximum number of rows per Parquet data page.
242    pub parquet_page_row_limit: usize,
243    /// Approximate maximum Parquet dictionary page size in bytes.
244    pub parquet_dict_size_bytes: usize,
245    /// The master key id used to encrypt this table's manifest list and data
246    /// files. `None` if `encryption.key-id` is not set.
247    pub encryption_key_id: Option<String>,
248    /// The encryption data encryption key length in bytes.
249    pub encryption_data_key_length: usize,
250    /// Base directory for data files
251    pub write_data_location: Option<String>,
252    /// Deprecated table property for data file write location.
253    ///
254    /// Property will be removed at a later date.
255    /// Superseded by [write_data_location].
256    pub write_folder_storage_location: Option<String>,
257    /// Deprecated table property for data file write location for object storage location generator.
258    ///
259    /// Property will be removed at a later date.
260    /// Superseded by [write_data_location].
261    pub write_object_storage_location: Option<String>,
262    /// Whether partition values are included in object storage paths.
263    pub write_object_storage_partitioned_paths: bool,
264}
265
266impl TableProperties {
267    /// Reserved table property for table format version.
268    ///
269    /// Iceberg will default a new table's format version to the latest stable and recommended
270    /// version. This reserved property keyword allows users to override the Iceberg format version of
271    /// the table metadata.
272    ///
273    /// If this table property exists when creating a table, the table will use the specified format
274    /// version. If a table updates this property, it will try to upgrade to the specified format
275    /// version.
276    pub const PROPERTY_FORMAT_VERSION: &str = "format-version";
277    /// Reserved table property for table UUID.
278    pub const PROPERTY_UUID: &str = "uuid";
279    /// Reserved table property for the total number of snapshots.
280    pub const PROPERTY_SNAPSHOT_COUNT: &str = "snapshot-count";
281    /// Reserved table property for current snapshot summary.
282    pub const PROPERTY_CURRENT_SNAPSHOT_SUMMARY: &str = "current-snapshot-summary";
283    /// Reserved table property for current snapshot id.
284    pub const PROPERTY_CURRENT_SNAPSHOT_ID: &str = "current-snapshot-id";
285    /// Reserved table property for current snapshot timestamp.
286    pub const PROPERTY_CURRENT_SNAPSHOT_TIMESTAMP: &str = "current-snapshot-timestamp-ms";
287    /// Reserved table property for the JSON representation of current schema.
288    pub const PROPERTY_CURRENT_SCHEMA: &str = "current-schema";
289    /// Reserved table property for the JSON representation of current(default) partition spec.
290    pub const PROPERTY_DEFAULT_PARTITION_SPEC: &str = "default-partition-spec";
291    /// Reserved table property for the JSON representation of current(default) sort order.
292    pub const PROPERTY_DEFAULT_SORT_ORDER: &str = "default-sort-order";
293
294    /// Property key for max number of previous versions to keep.
295    pub const PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX: &str =
296        "write.metadata.previous-versions-max";
297    /// Default value for max number of previous versions to keep.
298    pub const PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT: usize = 100;
299
300    /// Property key for max number of partitions to keep summary stats for.
301    pub const PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT: &str = "write.summary.partition-limit";
302    /// Default value for the max number of partitions to keep summary stats for.
303    pub const PROPERTY_WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT: u64 = 0;
304
305    /// Reserved Iceberg table properties list.
306    ///
307    /// Reserved table properties are only used to control behaviors when creating or updating a
308    /// table. The value of these properties are not persisted as a part of the table metadata.
309    pub const RESERVED_PROPERTIES: [&str; 9] = [
310        Self::PROPERTY_FORMAT_VERSION,
311        Self::PROPERTY_UUID,
312        Self::PROPERTY_SNAPSHOT_COUNT,
313        Self::PROPERTY_CURRENT_SNAPSHOT_ID,
314        Self::PROPERTY_CURRENT_SNAPSHOT_SUMMARY,
315        Self::PROPERTY_CURRENT_SNAPSHOT_TIMESTAMP,
316        Self::PROPERTY_CURRENT_SCHEMA,
317        Self::PROPERTY_DEFAULT_PARTITION_SPEC,
318        Self::PROPERTY_DEFAULT_SORT_ORDER,
319    ];
320
321    /// Property key for number of commit retries.
322    pub const PROPERTY_COMMIT_NUM_RETRIES: &str = "commit.retry.num-retries";
323    /// Default value for number of commit retries.
324    pub const PROPERTY_COMMIT_NUM_RETRIES_DEFAULT: usize = 4;
325
326    /// Property key for minimum wait time (ms) between retries.
327    pub const PROPERTY_COMMIT_MIN_RETRY_WAIT_MS: &str = "commit.retry.min-wait-ms";
328    /// Default value for minimum wait time (ms) between retries.
329    pub const PROPERTY_COMMIT_MIN_RETRY_WAIT_MS_DEFAULT: u64 = 100;
330
331    /// Property key for maximum wait time (ms) between retries.
332    pub const PROPERTY_COMMIT_MAX_RETRY_WAIT_MS: &str = "commit.retry.max-wait-ms";
333    /// Default value for maximum wait time (ms) between retries.
334    pub const PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT: u64 = 60 * 1000; // 1 minute
335
336    /// Property key for total maximum retry time (ms).
337    pub const PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS: &str = "commit.retry.total-timeout-ms";
338    /// Default value for total maximum retry time (ms).
339    pub const PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT: u64 = 30 * 60 * 1000; // 30 minutes
340
341    /// Default file format for data files
342    pub const PROPERTY_DEFAULT_FILE_FORMAT: &str = "write.format.default";
343    /// Default file format for delete files
344    pub const PROPERTY_DELETE_DEFAULT_FILE_FORMAT: &str = "write.delete.format.default";
345    /// Default value for data file format
346    pub const PROPERTY_DEFAULT_FILE_FORMAT_DEFAULT: &str = "parquet";
347
348    /// Target file size for newly written files.
349    pub const PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES: &str = "write.target-file-size-bytes";
350    /// Default target file size
351    pub const PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT: usize = 512 * 1024 * 1024; // 512 MB
352
353    /// Base location for metadata files (manifests, manifest lists, table metadata).
354    /// When unset, metadata files default to the `metadata` directory under the table
355    /// location.
356    pub const PROPERTY_WRITE_METADATA_PATH: &str = "write.metadata.path";
357
358    /// Compression codec for metadata files (JSON)
359    pub const PROPERTY_METADATA_COMPRESSION_CODEC: &str = "write.metadata.compression-codec";
360    /// Default metadata compression codec - uncompressed
361    pub const PROPERTY_METADATA_COMPRESSION_CODEC_DEFAULT: &str = "none";
362    /// Whether to use `FanoutWriter` for partitioned tables (handles unsorted data).
363    /// If false, uses `ClusteredWriter` (requires sorted data, more memory efficient).
364    pub const PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED: &str = "write.datafusion.fanout.enabled";
365    /// Default value for fanout writer enabled
366    pub const PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED_DEFAULT: bool = true;
367
368    /// Property key for enabling garbage collection on drop.
369    /// When set to `false`, data files will not be deleted when a table is dropped.
370    /// Defaults to `true`.
371    pub const PROPERTY_GC_ENABLED: &str = "gc.enabled";
372    /// Default value for gc.enabled
373    pub const PROPERTY_GC_ENABLED_DEFAULT: bool = true;
374
375    /// Property key for the default maximum age of a snapshot to keep when expiring snapshots.
376    pub const PROPERTY_MAX_SNAPSHOT_AGE_MS: &str = "history.expire.max-snapshot-age-ms";
377    /// Default value for history.expire.max-snapshot-age-ms (5 days).
378    pub const PROPERTY_MAX_SNAPSHOT_AGE_MS_DEFAULT: i64 = 5 * 24 * 60 * 60 * 1000;
379    /// Property key for the default minimum number of snapshots to keep when expiring snapshots.
380    pub const PROPERTY_MIN_SNAPSHOTS_TO_KEEP: &str = "history.expire.min-snapshots-to-keep";
381    /// Default value for history.expire.min-snapshots-to-keep.
382    pub const PROPERTY_MIN_SNAPSHOTS_TO_KEEP_DEFAULT: usize = 1;
383    /// Property key for the default maximum age of a snapshot reference to keep when expiring.
384    pub const PROPERTY_MAX_REF_AGE_MS: &str = "history.expire.max-ref-age-ms";
385    /// Default value for history.expire.max-ref-age-ms (effectively never expire refs).
386    pub const PROPERTY_MAX_REF_AGE_MS_DEFAULT: i64 = i64::MAX;
387
388    /// Enable content-defined chunking with parquet defaults (or per-property overrides).
389    pub const PROPERTY_PARQUET_CDC_ENABLED: &str = "write.parquet.content-defined-chunking.enabled";
390    /// Default value for content-defined chunking enabled.
391    pub const PROPERTY_PARQUET_CDC_ENABLED_DEFAULT: bool = false;
392    /// Minimum chunk size in bytes for content-defined chunking.
393    pub const PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE: &str =
394        "write.parquet.content-defined-chunking.min-chunk-size";
395    /// Default matches `parquet::file::properties::DEFAULT_CDC_MIN_CHUNK_SIZE`.
396    pub const PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT: usize = 256 * 1024;
397    /// Maximum chunk size in bytes for content-defined chunking.
398    pub const PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE: &str =
399        "write.parquet.content-defined-chunking.max-chunk-size";
400    /// Default matches `parquet::file::properties::DEFAULT_CDC_MAX_CHUNK_SIZE`.
401    pub const PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT: usize = 1024 * 1024;
402    /// Normalization level (gearhash bit adjustment) for content-defined chunking.
403    pub const PROPERTY_PARQUET_CDC_NORM_LEVEL: &str =
404        "write.parquet.content-defined-chunking.norm-level";
405    /// Default matches `parquet::file::properties::DEFAULT_CDC_NORM_LEVEL`.
406    pub const PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT: i32 = 0;
407
408    /// Compression codec for Parquet data files (e.g. `zstd`, `gzip`, `snappy`,
409    /// `lz4`, `lz4_raw`, `brotli`, `lzo`, `uncompressed`). The codec name is
410    /// parsed into a [`CompressionCodec`] when properties are parsed; the level's
411    /// range is validated when the writer is built.
412    pub const PROPERTY_PARQUET_COMPRESSION_CODEC: &str = "write.parquet.compression-codec";
413    /// Default Parquet compression codec.
414    pub const PROPERTY_PARQUET_COMPRESSION_CODEC_DEFAULT: &str = "zstd";
415    /// Compression level for Parquet data files, for codecs that take one
416    /// (`gzip`, `zstd`, `brotli`). When unset, the codec's default level is used.
417    pub const PROPERTY_PARQUET_COMPRESSION_LEVEL: &str = "write.parquet.compression-level";
418
419    /// Approximate maximum size of a Parquet row group in bytes.
420    pub const PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES: &str = "write.parquet.row-group-size-bytes";
421    /// Default Parquet row group size in bytes.
422    pub const PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT: usize = 128 * 1024 * 1024;
423
424    /// Approximate maximum size of a Parquet data page in bytes.
425    pub const PROPERTY_PARQUET_PAGE_SIZE_BYTES: &str = "write.parquet.page-size-bytes";
426    /// Default Parquet page size in bytes.
427    pub const PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT: usize = 1024 * 1024;
428
429    /// Maximum number of rows per Parquet data page.
430    pub const PROPERTY_PARQUET_PAGE_ROW_LIMIT: &str = "write.parquet.page-row-limit";
431    /// Default Parquet page row limit.
432    pub const PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT: usize = 20000;
433
434    /// Approximate maximum size of the Parquet dictionary page in bytes.
435    pub const PROPERTY_PARQUET_DICT_SIZE_BYTES: &str = "write.parquet.dict-size-bytes";
436    /// Default Parquet dictionary page size in bytes.
437    pub const PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT: usize = 2 * 1024 * 1024;
438
439    /// Property key for the master key id used to encrypt the table's manifest
440    /// list and data files as defined in https://iceberg.apache.org/docs/nightly/encryption/.
441    pub const PROPERTY_ENCRYPTION_KEY_ID: &str = "encryption.key-id";
442
443    /// Property key for the encryption data encryption key (DEK) length in bytes.
444    pub const PROPERTY_ENCRYPTION_DATA_KEY_LENGTH: &str = "encryption.data-key-length";
445    /// Default value for the encryption DEK length (16 bytes = AES-128).
446    pub const PROPERTY_ENCRYPTION_DATA_KEY_LENGTH_DEFAULT: usize = 16;
447    /// Property key for the base directory for data files
448    pub const PROPERTY_WRITE_DATA_LOCATION: &str = "write.data.path";
449    /// Property key for deprecated [write_folder_storage_location]
450    pub const PROPERTY_WRITE_FOLDER_STORAGE_LOCATION: &str = "write.folder-storage.path";
451    /// Property key for deprecated object storage path, kept as a fallback for compatibility.
452    pub const PROPERTY_WRITE_OBJECT_STORAGE_LOCATION: &str = "write.object-storage.path";
453    /// Property key for controlling whether partition values are included in object storage paths.
454    pub const PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS: &str =
455        "write.object-storage.partitioned-paths";
456    /// Default value for [PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS]
457    pub const PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT: bool = true;
458}
459
460impl TryFrom<&HashMap<String, String>> for TableProperties {
461    // parse by entry key or use default value
462    type Error = Error;
463
464    fn try_from(props: &HashMap<String, String>) -> Result<Self> {
465        Ok(TableProperties {
466            commit_num_retries: parse_property(
467                props,
468                TableProperties::PROPERTY_COMMIT_NUM_RETRIES,
469                TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT,
470            )?,
471            commit_min_retry_wait_ms: parse_property(
472                props,
473                TableProperties::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS,
474                TableProperties::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS_DEFAULT,
475            )?,
476            commit_max_retry_wait_ms: parse_property(
477                props,
478                TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS,
479                TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT,
480            )?,
481            commit_total_retry_timeout_ms: parse_property(
482                props,
483                TableProperties::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS,
484                TableProperties::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT,
485            )?,
486            write_format_default: parse_property(
487                props,
488                TableProperties::PROPERTY_DEFAULT_FILE_FORMAT,
489                TableProperties::PROPERTY_DEFAULT_FILE_FORMAT_DEFAULT.to_string(),
490            )?,
491            write_target_file_size_bytes: parse_property(
492                props,
493                TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES,
494                TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT,
495            )?,
496            write_metadata_path: parse_location_property(
497                props,
498                TableProperties::PROPERTY_WRITE_METADATA_PATH,
499            )?,
500            metadata_compression_codec: parse_metadata_file_compression(props)?,
501            write_datafusion_fanout_enabled: parse_property_bool(
502                props,
503                TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED,
504                TableProperties::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED_DEFAULT,
505            )?,
506            gc_enabled: parse_property_bool(
507                props,
508                TableProperties::PROPERTY_GC_ENABLED,
509                TableProperties::PROPERTY_GC_ENABLED_DEFAULT,
510            )?,
511            max_snapshot_age_ms: parse_property(
512                props,
513                TableProperties::PROPERTY_MAX_SNAPSHOT_AGE_MS,
514                TableProperties::PROPERTY_MAX_SNAPSHOT_AGE_MS_DEFAULT,
515            )?,
516            min_snapshots_to_keep: parse_property(
517                props,
518                TableProperties::PROPERTY_MIN_SNAPSHOTS_TO_KEEP,
519                TableProperties::PROPERTY_MIN_SNAPSHOTS_TO_KEEP_DEFAULT,
520            )?,
521            max_ref_age_ms: parse_property(
522                props,
523                TableProperties::PROPERTY_MAX_REF_AGE_MS,
524                TableProperties::PROPERTY_MAX_REF_AGE_MS_DEFAULT,
525            )?,
526            cdc_enabled: parse_property_bool(
527                props,
528                TableProperties::PROPERTY_PARQUET_CDC_ENABLED,
529                TableProperties::PROPERTY_PARQUET_CDC_ENABLED_DEFAULT,
530            )?,
531            cdc_min_chunk_size: parse_property(
532                props,
533                TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE,
534                TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT,
535            )?,
536            cdc_max_chunk_size: parse_property(
537                props,
538                TableProperties::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE,
539                TableProperties::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT,
540            )?,
541            cdc_norm_level: parse_property(
542                props,
543                TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL,
544                TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT,
545            )?,
546            parquet_compression_codec: parse_parquet_compression(props)?,
547            parquet_row_group_size_bytes: parse_property(
548                props,
549                TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES,
550                TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT,
551            )?,
552            parquet_page_size_bytes: parse_property(
553                props,
554                TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES,
555                TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT,
556            )?,
557            parquet_page_row_limit: parse_property(
558                props,
559                TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT,
560                TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT,
561            )?,
562            parquet_dict_size_bytes: parse_property(
563                props,
564                TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES,
565                TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT,
566            )?,
567            encryption_key_id: props
568                .get(TableProperties::PROPERTY_ENCRYPTION_KEY_ID)
569                .cloned(),
570            encryption_data_key_length: parse_property(
571                props,
572                TableProperties::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH,
573                TableProperties::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH_DEFAULT,
574            )?,
575            write_data_location: props
576                .get(TableProperties::PROPERTY_WRITE_DATA_LOCATION)
577                .cloned(),
578            write_folder_storage_location: props
579                .get(TableProperties::PROPERTY_WRITE_FOLDER_STORAGE_LOCATION)
580                .cloned(),
581            write_object_storage_location: props
582                .get(TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_LOCATION)
583                .cloned(),
584            write_object_storage_partitioned_paths: parse_property_bool(
585                props,
586                TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS,
587                TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT,
588            )?,
589        })
590    }
591}
592
593#[cfg(test)]
594mod tests {
595    use super::*;
596    use crate::compression::CompressionCodec;
597
598    #[test]
599    fn test_table_properties_default() {
600        let props = HashMap::new();
601        let table_properties = TableProperties::try_from(&props).unwrap();
602        assert_eq!(
603            table_properties.commit_num_retries,
604            TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT
605        );
606        assert_eq!(
607            table_properties.commit_min_retry_wait_ms,
608            TableProperties::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
609        );
610        assert_eq!(
611            table_properties.commit_max_retry_wait_ms,
612            TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
613        );
614        assert_eq!(
615            table_properties.write_format_default,
616            TableProperties::PROPERTY_DEFAULT_FILE_FORMAT_DEFAULT.to_string()
617        );
618        assert_eq!(
619            table_properties.write_target_file_size_bytes,
620            TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT
621        );
622        // Test compression defaults (none means CompressionCodec::None)
623        assert_eq!(
624            table_properties.metadata_compression_codec,
625            CompressionCodec::None
626        );
627        assert_eq!(
628            table_properties.gc_enabled,
629            TableProperties::PROPERTY_GC_ENABLED_DEFAULT
630        );
631        assert_eq!(
632            table_properties.max_snapshot_age_ms,
633            TableProperties::PROPERTY_MAX_SNAPSHOT_AGE_MS_DEFAULT
634        );
635        assert_eq!(
636            table_properties.min_snapshots_to_keep,
637            TableProperties::PROPERTY_MIN_SNAPSHOTS_TO_KEEP_DEFAULT
638        );
639        assert_eq!(
640            table_properties.max_ref_age_ms,
641            TableProperties::PROPERTY_MAX_REF_AGE_MS_DEFAULT
642        );
643    }
644
645    #[test]
646    fn test_table_properties_history_expire_overrides() {
647        let props = HashMap::from([
648            (
649                TableProperties::PROPERTY_MAX_SNAPSHOT_AGE_MS.to_string(),
650                "1234".to_string(),
651            ),
652            (
653                TableProperties::PROPERTY_MIN_SNAPSHOTS_TO_KEEP.to_string(),
654                "7".to_string(),
655            ),
656            (
657                TableProperties::PROPERTY_MAX_REF_AGE_MS.to_string(),
658                "5678".to_string(),
659            ),
660        ]);
661        let table_properties = TableProperties::try_from(&props).unwrap();
662        assert_eq!(table_properties.max_snapshot_age_ms, 1234);
663        assert_eq!(table_properties.min_snapshots_to_keep, 7);
664        assert_eq!(table_properties.max_ref_age_ms, 5678);
665    }
666
667    #[test]
668    fn test_table_properties_write_metadata_path() {
669        // Test unset
670        let table_properties = TableProperties::try_from(&HashMap::new()).unwrap();
671        assert_eq!(table_properties.write_metadata_path, None);
672
673        // Test empty path is invalid
674        let props = HashMap::from([(
675            TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(),
676            String::new(),
677        )]);
678        let error = TableProperties::try_from(&props).unwrap_err();
679        assert_eq!(error.kind(), ErrorKind::DataInvalid);
680        assert!(
681            error
682                .message()
683                .contains(TableProperties::PROPERTY_WRITE_METADATA_PATH)
684        );
685
686        let props = HashMap::from([(
687            TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(),
688            "s3://other-bucket/custom-meta/".to_string(),
689        )]);
690        let table_properties = TableProperties::try_from(&props).unwrap();
691        assert_eq!(
692            table_properties.write_metadata_path.as_deref(),
693            Some("s3://other-bucket/custom-meta")
694        );
695    }
696
697    #[test]
698    fn test_table_properties_compression() {
699        let props = HashMap::from([(
700            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
701            "gzip".to_string(),
702        )]);
703        let table_properties = TableProperties::try_from(&props).unwrap();
704        assert_eq!(
705            table_properties.metadata_compression_codec,
706            CompressionCodec::gzip_default()
707        );
708    }
709
710    #[test]
711    fn test_table_properties_compression_none() {
712        let props = HashMap::from([(
713            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
714            "none".to_string(),
715        )]);
716        let table_properties = TableProperties::try_from(&props).unwrap();
717        assert_eq!(
718            table_properties.metadata_compression_codec,
719            CompressionCodec::None
720        );
721    }
722
723    #[test]
724    fn test_table_properties_compression_case_insensitive() {
725        // Test uppercase
726        let props_upper = HashMap::from([(
727            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
728            "GZIP".to_string(),
729        )]);
730        let table_properties = TableProperties::try_from(&props_upper).unwrap();
731        assert_eq!(
732            table_properties.metadata_compression_codec,
733            CompressionCodec::gzip_default()
734        );
735
736        // Test mixed case
737        let props_mixed = HashMap::from([(
738            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
739            "GzIp".to_string(),
740        )]);
741        let table_properties = TableProperties::try_from(&props_mixed).unwrap();
742        assert_eq!(
743            table_properties.metadata_compression_codec,
744            CompressionCodec::gzip_default()
745        );
746
747        // Test "NONE" should also be case-insensitive
748        let props_none_upper = HashMap::from([(
749            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
750            "NONE".to_string(),
751        )]);
752        let table_properties = TableProperties::try_from(&props_none_upper).unwrap();
753        assert_eq!(
754            table_properties.metadata_compression_codec,
755            CompressionCodec::None
756        );
757    }
758
759    #[test]
760    fn test_table_properties_valid() {
761        let props = HashMap::from([
762            (
763                TableProperties::PROPERTY_COMMIT_NUM_RETRIES.to_string(),
764                "10".to_string(),
765            ),
766            (
767                TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS.to_string(),
768                "20".to_string(),
769            ),
770            (
771                TableProperties::PROPERTY_DEFAULT_FILE_FORMAT.to_string(),
772                "avro".to_string(),
773            ),
774            (
775                TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES.to_string(),
776                "512".to_string(),
777            ),
778            (
779                TableProperties::PROPERTY_GC_ENABLED.to_string(),
780                "false".to_string(),
781            ),
782        ]);
783        let table_properties = TableProperties::try_from(&props).unwrap();
784        assert_eq!(table_properties.commit_num_retries, 10);
785        assert_eq!(table_properties.commit_max_retry_wait_ms, 20);
786        assert_eq!(table_properties.write_format_default, "avro".to_string());
787        assert_eq!(table_properties.write_target_file_size_bytes, 512);
788        assert!(!table_properties.gc_enabled);
789    }
790
791    #[test]
792    fn test_table_properties_invalid() {
793        let invalid_retries = HashMap::from([(
794            TableProperties::PROPERTY_COMMIT_NUM_RETRIES.to_string(),
795            "abc".to_string(),
796        )]);
797
798        let table_properties = TableProperties::try_from(&invalid_retries).unwrap_err();
799        assert!(
800            table_properties.to_string().contains(
801                "Invalid value for commit.retry.num-retries: invalid digit found in string"
802            )
803        );
804
805        let invalid_min_wait = HashMap::from([(
806            TableProperties::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS.to_string(),
807            "abc".to_string(),
808        )]);
809        let table_properties = TableProperties::try_from(&invalid_min_wait).unwrap_err();
810        assert!(
811            table_properties.to_string().contains(
812                "Invalid value for commit.retry.min-wait-ms: invalid digit found in string"
813            )
814        );
815
816        let invalid_max_wait = HashMap::from([(
817            TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS.to_string(),
818            "abc".to_string(),
819        )]);
820        let table_properties = TableProperties::try_from(&invalid_max_wait).unwrap_err();
821        assert!(
822            table_properties.to_string().contains(
823                "Invalid value for commit.retry.max-wait-ms: invalid digit found in string"
824            )
825        );
826
827        let invalid_target_size = HashMap::from([(
828            TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES.to_string(),
829            "abc".to_string(),
830        )]);
831        let table_properties = TableProperties::try_from(&invalid_target_size).unwrap_err();
832        assert!(table_properties.to_string().contains(
833            "Invalid value for write.target-file-size-bytes: invalid digit found in string"
834        ));
835
836        let invalid_gc_enabled = HashMap::from([(
837            TableProperties::PROPERTY_GC_ENABLED.to_string(),
838            "notabool".to_string(),
839        )]);
840        let table_properties = TableProperties::try_from(&invalid_gc_enabled).unwrap_err();
841        assert!(
842            table_properties
843                .to_string()
844                .contains("Invalid value for gc.enabled")
845        );
846    }
847
848    #[test]
849    fn test_table_properties_compression_invalid_rejected() {
850        let invalid_codecs = ["lz4", "zstd", "snappy"];
851
852        for codec in invalid_codecs {
853            let props = HashMap::from([(
854                TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
855                codec.to_string(),
856            )]);
857            let err = TableProperties::try_from(&props).unwrap_err();
858            let err_msg = err.to_string();
859            assert!(
860                err_msg.contains(&format!("Invalid metadata compression codec: {codec}")),
861                "Expected error message to contain codec '{codec}', got: {err_msg}"
862            );
863            assert!(
864                err_msg.contains("Only 'none' and 'gzip' are supported"),
865                "Expected error message to contain supported codecs, got: {err_msg}"
866            );
867        }
868    }
869
870    #[test]
871    fn test_parse_metadata_file_compression_valid() {
872        // Test with "none"
873        let props = HashMap::from([(
874            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
875            "none".to_string(),
876        )]);
877        assert_eq!(
878            parse_metadata_file_compression(&props).unwrap(),
879            CompressionCodec::None
880        );
881
882        // Test with empty string
883        let props = HashMap::from([(
884            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
885            "".to_string(),
886        )]);
887        assert_eq!(
888            parse_metadata_file_compression(&props).unwrap(),
889            CompressionCodec::None
890        );
891
892        // Test with "gzip"
893        let props = HashMap::from([(
894            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
895            "gzip".to_string(),
896        )]);
897        assert_eq!(
898            parse_metadata_file_compression(&props).unwrap(),
899            CompressionCodec::gzip_default()
900        );
901
902        // Test case insensitivity - "NONE"
903        let props = HashMap::from([(
904            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
905            "NONE".to_string(),
906        )]);
907        assert_eq!(
908            parse_metadata_file_compression(&props).unwrap(),
909            CompressionCodec::None
910        );
911
912        // Test case insensitivity - "GZIP"
913        let props = HashMap::from([(
914            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
915            "GZIP".to_string(),
916        )]);
917        assert_eq!(
918            parse_metadata_file_compression(&props).unwrap(),
919            CompressionCodec::gzip_default()
920        );
921
922        // Test case insensitivity - "GzIp"
923        let props = HashMap::from([(
924            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
925            "GzIp".to_string(),
926        )]);
927        assert_eq!(
928            parse_metadata_file_compression(&props).unwrap(),
929            CompressionCodec::gzip_default()
930        );
931
932        // Test default when property is missing
933        let props = HashMap::new();
934        assert_eq!(
935            parse_metadata_file_compression(&props).unwrap(),
936            CompressionCodec::None
937        );
938    }
939
940    #[test]
941    fn test_parse_metadata_file_compression_invalid() {
942        let invalid_codecs = ["lz4", "zstd", "snappy"];
943
944        for codec in invalid_codecs {
945            let props = HashMap::from([(
946                TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
947                codec.to_string(),
948            )]);
949            let err = parse_metadata_file_compression(&props).unwrap_err();
950            let err_msg = err.to_string();
951            assert!(
952                err_msg.contains("Invalid metadata compression codec"),
953                "Expected error message to contain 'Invalid metadata compression codec', got: {err_msg}"
954            );
955            assert!(
956                err_msg.contains("Only 'none' and 'gzip' are supported"),
957                "Expected error message to contain supported codecs, got: {err_msg}"
958            );
959        }
960    }
961
962    #[test]
963    fn test_cdc_disabled_by_default() {
964        let props = HashMap::new();
965        let tp = TableProperties::try_from(&props).unwrap();
966        assert!(!tp.cdc_enabled);
967    }
968
969    #[test]
970    fn test_cdc_enabled_via_flag() {
971        let props = HashMap::from([(
972            TableProperties::PROPERTY_PARQUET_CDC_ENABLED.to_string(),
973            "true".to_string(),
974        )]);
975        let tp = TableProperties::try_from(&props).unwrap();
976        assert!(tp.cdc_enabled);
977        assert_eq!(tp.cdc_min_chunk_size, 256 * 1024);
978        assert_eq!(tp.cdc_max_chunk_size, 1024 * 1024);
979        assert_eq!(tp.cdc_norm_level, 0);
980    }
981
982    #[test]
983    fn test_cdc_size_props_alone_do_not_enable() {
984        let props = HashMap::from([(
985            TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE.to_string(),
986            "262144".to_string(),
987        )]);
988        let tp = TableProperties::try_from(&props).unwrap();
989        assert!(!tp.cdc_enabled);
990    }
991
992    #[test]
993    fn test_cdc_custom_values() {
994        let props = HashMap::from([
995            (
996                TableProperties::PROPERTY_PARQUET_CDC_ENABLED.to_string(),
997                "true".to_string(),
998            ),
999            (
1000                TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE.to_string(),
1001                "200000".to_string(),
1002            ),
1003            (
1004                TableProperties::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE.to_string(),
1005                "900000".to_string(),
1006            ),
1007            (
1008                TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL.to_string(),
1009                "1".to_string(),
1010            ),
1011        ]);
1012        let tp = TableProperties::try_from(&props).unwrap();
1013        assert!(tp.cdc_enabled);
1014        assert_eq!(tp.cdc_min_chunk_size, 200000);
1015        assert_eq!(tp.cdc_max_chunk_size, 900000);
1016        assert_eq!(tp.cdc_norm_level, 1);
1017    }
1018
1019    #[test]
1020    fn test_cdc_partial_override() {
1021        let props = HashMap::from([
1022            (
1023                TableProperties::PROPERTY_PARQUET_CDC_ENABLED.to_string(),
1024                "true".to_string(),
1025            ),
1026            (
1027                TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL.to_string(),
1028                "2".to_string(),
1029            ),
1030        ]);
1031        let tp = TableProperties::try_from(&props).unwrap();
1032        assert!(tp.cdc_enabled);
1033        assert_eq!(tp.cdc_min_chunk_size, 256 * 1024);
1034        assert_eq!(tp.cdc_max_chunk_size, 1024 * 1024);
1035        assert_eq!(tp.cdc_norm_level, 2);
1036    }
1037
1038    #[test]
1039    fn test_cdc_negative_norm_level() {
1040        let props = HashMap::from([
1041            (
1042                TableProperties::PROPERTY_PARQUET_CDC_ENABLED.to_string(),
1043                "true".to_string(),
1044            ),
1045            (
1046                TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL.to_string(),
1047                "-2".to_string(),
1048            ),
1049        ]);
1050        let tp = TableProperties::try_from(&props).unwrap();
1051        assert_eq!(tp.cdc_norm_level, -2);
1052    }
1053
1054    #[test]
1055    fn test_cdc_invalid_min_chunk_size() {
1056        let props = HashMap::from([
1057            (
1058                TableProperties::PROPERTY_PARQUET_CDC_ENABLED.to_string(),
1059                "true".to_string(),
1060            ),
1061            (
1062                TableProperties::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE.to_string(),
1063                "not_a_number".to_string(),
1064            ),
1065        ]);
1066        let err = TableProperties::try_from(&props).unwrap_err();
1067        assert!(
1068            err.to_string().contains(
1069                "Invalid value for write.parquet.content-defined-chunking.min-chunk-size"
1070            )
1071        );
1072    }
1073
1074    #[test]
1075    fn test_cdc_invalid_norm_level() {
1076        let props = HashMap::from([
1077            (
1078                TableProperties::PROPERTY_PARQUET_CDC_ENABLED.to_string(),
1079                "true".to_string(),
1080            ),
1081            (
1082                TableProperties::PROPERTY_PARQUET_CDC_NORM_LEVEL.to_string(),
1083                "not_a_number".to_string(),
1084            ),
1085        ]);
1086        let err = TableProperties::try_from(&props).unwrap_err();
1087        assert!(
1088            err.to_string()
1089                .contains("Invalid value for write.parquet.content-defined-chunking.norm-level")
1090        );
1091    }
1092
1093    #[test]
1094    fn test_cdc_no_properties() {
1095        let props = HashMap::from([("some.other.property".to_string(), "value".to_string())]);
1096        let tp = TableProperties::try_from(&props).unwrap();
1097        assert!(!tp.cdc_enabled);
1098    }
1099
1100    #[test]
1101    fn test_parquet_sizing_defaults() {
1102        let tp = TableProperties::try_from(&HashMap::new()).unwrap();
1103        // Default codec is zstd at its default level.
1104        assert_eq!(
1105            tp.parquet_compression_codec,
1106            CompressionCodec::zstd_default()
1107        );
1108        assert_eq!(
1109            tp.parquet_row_group_size_bytes,
1110            TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT
1111        );
1112        assert_eq!(
1113            tp.parquet_page_size_bytes,
1114            TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT
1115        );
1116        assert_eq!(
1117            tp.parquet_page_row_limit,
1118            TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT
1119        );
1120        assert_eq!(
1121            tp.parquet_dict_size_bytes,
1122            TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT
1123        );
1124    }
1125
1126    #[test]
1127    fn test_parquet_sizing_overrides() {
1128        let props = HashMap::from([
1129            (
1130                TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC.to_string(),
1131                "gzip".to_string(),
1132            ),
1133            (
1134                TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL.to_string(),
1135                "4".to_string(),
1136            ),
1137            (
1138                TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES.to_string(),
1139                "1048576".to_string(),
1140            ),
1141            (
1142                TableProperties::PROPERTY_PARQUET_PAGE_SIZE_BYTES.to_string(),
1143                "65536".to_string(),
1144            ),
1145            (
1146                TableProperties::PROPERTY_PARQUET_PAGE_ROW_LIMIT.to_string(),
1147                "5000".to_string(),
1148            ),
1149            (
1150                TableProperties::PROPERTY_PARQUET_DICT_SIZE_BYTES.to_string(),
1151                "131072".to_string(),
1152            ),
1153        ]);
1154        let tp = TableProperties::try_from(&props).unwrap();
1155        // Codec name and level are folded into a single CompressionCodec.
1156        assert_eq!(tp.parquet_compression_codec, CompressionCodec::Gzip(4));
1157        assert_eq!(tp.parquet_row_group_size_bytes, 1048576);
1158        assert_eq!(tp.parquet_page_size_bytes, 65536);
1159        assert_eq!(tp.parquet_page_row_limit, 5000);
1160        assert_eq!(tp.parquet_dict_size_bytes, 131072);
1161    }
1162
1163    #[test]
1164    fn test_parquet_invalid_sizing_rejected() {
1165        let props = HashMap::from([(
1166            TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES.to_string(),
1167            "not_a_number".to_string(),
1168        )]);
1169        let err = TableProperties::try_from(&props).unwrap_err();
1170        assert_eq!(err.kind(), ErrorKind::DataInvalid);
1171        assert!(
1172            err.to_string()
1173                .contains(TableProperties::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES)
1174        );
1175    }
1176
1177    #[test]
1178    fn test_parquet_all_codecs_parse() {
1179        // Every codec name parquet-java supports must parse (parity with Java's
1180        // `CompressionCodecName.valueOf`).
1181        for (name, expected) in [
1182            ("uncompressed", CompressionCodec::None),
1183            ("snappy", CompressionCodec::Snappy),
1184            ("gzip", CompressionCodec::gzip_default()),
1185            ("lzo", CompressionCodec::Lzo),
1186            ("brotli", CompressionCodec::brotli_default()),
1187            ("lz4", CompressionCodec::Lz4),
1188            ("lz4_raw", CompressionCodec::Lz4Raw),
1189            ("zstd", CompressionCodec::zstd_default()),
1190        ] {
1191            let props = HashMap::from([(
1192                TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC.to_string(),
1193                name.to_string(),
1194            )]);
1195            let tp = TableProperties::try_from(&props).unwrap();
1196            assert_eq!(tp.parquet_compression_codec, expected, "codec {name}");
1197        }
1198    }
1199
1200    #[test]
1201    fn test_parquet_compression_level_ignored_for_levelless_codec() {
1202        // A level set alongside a codec that carries none (e.g. snappy) is
1203        // ignored rather than rejected, matching parquet-java.
1204        let props = HashMap::from([
1205            (
1206                TableProperties::PROPERTY_PARQUET_COMPRESSION_CODEC.to_string(),
1207                "snappy".to_string(),
1208            ),
1209            (
1210                TableProperties::PROPERTY_PARQUET_COMPRESSION_LEVEL.to_string(),
1211                "5".to_string(),
1212            ),
1213        ]);
1214        let tp = TableProperties::try_from(&props).unwrap();
1215        assert_eq!(tp.parquet_compression_codec, CompressionCodec::Snappy);
1216    }
1217
1218    #[test]
1219    fn test_parse_boolean_property_case_insensitive() {
1220        let false_variants = ["False", "FALSE"];
1221        let true_variants = ["True", "TRUE"];
1222
1223        for f in false_variants {
1224            let props = HashMap::from([(
1225                TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS.to_string(),
1226                f.to_string(),
1227            )]);
1228            let tp = TableProperties::try_from(&props).unwrap();
1229            assert!(!tp.write_object_storage_partitioned_paths);
1230        }
1231
1232        for t in true_variants {
1233            let props = HashMap::from([(
1234                TableProperties::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS.to_string(),
1235                t.to_string(),
1236            )]);
1237            let tp = TableProperties::try_from(&props).unwrap();
1238            assert!(tp.write_object_storage_partitioned_paths);
1239        }
1240    }
1241}
iceberg/spec/table_properties.rs

iceberg/spec/
table_properties.rs
