From 30c3535f2d34bc204b0cdfd4a546040729a2a507 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Wed, 12 Aug 2026 10:52:20 +0800 Subject: [PATCH] refactor: Migrate TableProperties consumers to getters Replace direct TableProperties field access throughout Iceberg and DataFusion with the generated read-only accessors, then make the backing fields private and update the public API snapshot.\n\nCloses #2969. --- crates/iceberg/public-api.txt | 28 ---------- crates/iceberg/src/catalog/utils.rs | 2 +- crates/iceberg/src/encryption/manager.rs | 4 +- crates/iceberg/src/spec/table_metadata.rs | 11 ++-- crates/iceberg/src/spec/table_properties.rs | 56 +++++++++---------- .../src/transaction/expire_snapshots.rs | 10 ++-- crates/iceberg/src/transaction/mod.rs | 10 ++-- .../writer/file_writer/location_generator.rs | 14 +++-- .../src/writer/file_writer/parquet_writer.rs | 18 +++--- .../datafusion/src/physical_plan/write.rs | 6 +- 10 files changed, 68 insertions(+), 91 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index a4040b4ec6..ad71f12774 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -2757,34 +2757,6 @@ pub fn iceberg::spec::TableMetadataBuilder::clone(&self) -> iceberg::spec::Table impl core::fmt::Debug for iceberg::spec::TableMetadataBuilder pub fn iceberg::spec::TableMetadataBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result pub struct iceberg::spec::TableProperties -pub iceberg::spec::TableProperties::cdc_enabled: bool -pub iceberg::spec::TableProperties::cdc_max_chunk_size: usize -pub iceberg::spec::TableProperties::cdc_min_chunk_size: usize -pub iceberg::spec::TableProperties::cdc_norm_level: i32 -pub iceberg::spec::TableProperties::commit_max_retry_wait_ms: u64 -pub iceberg::spec::TableProperties::commit_min_retry_wait_ms: u64 -pub iceberg::spec::TableProperties::commit_num_retries: usize -pub iceberg::spec::TableProperties::commit_total_retry_timeout_ms: u64 -pub iceberg::spec::TableProperties::encryption_data_key_length: usize -pub iceberg::spec::TableProperties::encryption_key_id: core::option::Option -pub iceberg::spec::TableProperties::gc_enabled: bool -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: iceberg::compression::CompressionCodec -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_data_location: core::option::Option -pub iceberg::spec::TableProperties::write_datafusion_fanout_enabled: bool -pub iceberg::spec::TableProperties::write_folder_storage_location: core::option::Option -pub iceberg::spec::TableProperties::write_format_default: alloc::string::String -pub iceberg::spec::TableProperties::write_metadata_path: core::option::Option -pub iceberg::spec::TableProperties::write_object_storage_location: core::option::Option -pub iceberg::spec::TableProperties::write_object_storage_partitioned_paths: bool -pub iceberg::spec::TableProperties::write_target_file_size_bytes: usize impl iceberg::spec::TableProperties pub const iceberg::spec::TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS: &str pub const iceberg::spec::TableProperties::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT: u64 diff --git a/crates/iceberg/src/catalog/utils.rs b/crates/iceberg/src/catalog/utils.rs index 8e743e7d7d..c25904f682 100644 --- a/crates/iceberg/src/catalog/utils.rs +++ b/crates/iceberg/src/catalog/utils.rs @@ -61,7 +61,7 @@ pub async fn drop_table_data(table_info: &Table) -> Result<()> { } // Delete data files only if gc.enabled is true, to avoid corrupting shared tables - if metadata.table_properties()?.gc_enabled { + if metadata.table_properties()?.gc_enabled() { delete_data_files(io, &manifests_to_delete).await?; } diff --git a/crates/iceberg/src/encryption/manager.rs b/crates/iceberg/src/encryption/manager.rs index e2294c2f2c..3bceb2a8ec 100644 --- a/crates/iceberg/src/encryption/manager.rs +++ b/crates/iceberg/src/encryption/manager.rs @@ -119,7 +119,7 @@ impl EncryptionManager { } let table_properties = metadata.table_properties()?; - let Some(table_key_id) = table_properties.encryption_key_id else { + let Some(table_key_id) = table_properties.encryption_key_id().clone() else { if kms_client.is_some() { tracing::warn!( "KeyManagementClient provided but table does not have encryption.key-id set" @@ -140,7 +140,7 @@ impl EncryptionManager { .table_key_id(table_key_id) .encryption_keys(metadata.encryption_keys.clone()) .key_size(AesKeySize::from_key_length( - table_properties.encryption_data_key_length, + table_properties.encryption_data_key_length(), )?) .build(); Ok(Some(Arc::new(em))) diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs index ecc0586680..7620be2e70 100644 --- a/crates/iceberg/src/spec/table_metadata.rs +++ b/crates/iceberg/src/spec/table_metadata.rs @@ -371,7 +371,8 @@ impl TableMetadata { pub fn metadata_location(&self) -> Result { Ok(self .table_properties()? - .write_metadata_path + .write_metadata_path() + .clone() .unwrap_or_else(|| format!("{}/{}", self.location(), METADATA_FOLDER_NAME))) } @@ -4043,11 +4044,11 @@ mod tests { let props = metadata.table_properties().unwrap(); assert_eq!( - props.commit_num_retries, + props.commit_num_retries(), TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT ); assert_eq!( - props.write_target_file_size_bytes, + props.write_target_file_size_bytes(), TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT ); } @@ -4089,8 +4090,8 @@ mod tests { let props = metadata.table_properties().unwrap(); - assert_eq!(props.commit_num_retries, 10); - assert_eq!(props.write_target_file_size_bytes, 1024); + assert_eq!(props.commit_num_retries(), 10); + assert_eq!(props.write_target_file_size_bytes(), 1024); } #[test] diff --git a/crates/iceberg/src/spec/table_properties.rs b/crates/iceberg/src/spec/table_properties.rs index 9a784f59a0..98f7c2c043 100644 --- a/crates/iceberg/src/spec/table_properties.rs +++ b/crates/iceberg/src/spec/table_properties.rs @@ -153,42 +153,42 @@ pub struct TableProperties { default = Self::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT, getter )] - pub commit_num_retries: usize, + commit_num_retries: usize, /// The minimum wait time between retries. #[property( key = Self::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS, default = Self::PROPERTY_COMMIT_MIN_RETRY_WAIT_MS_DEFAULT, getter )] - pub commit_min_retry_wait_ms: u64, + commit_min_retry_wait_ms: u64, /// The maximum wait time between retries. #[property( key = Self::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS, default = Self::PROPERTY_COMMIT_MAX_RETRY_WAIT_MS_DEFAULT, getter )] - pub commit_max_retry_wait_ms: u64, + commit_max_retry_wait_ms: u64, /// The total timeout for commit retries. #[property( key = Self::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS, default = Self::PROPERTY_COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT, getter )] - pub commit_total_retry_timeout_ms: u64, + commit_total_retry_timeout_ms: u64, /// The default format for files. #[property( key = Self::PROPERTY_DEFAULT_FILE_FORMAT, default = Self::PROPERTY_DEFAULT_FILE_FORMAT_DEFAULT, getter )] - pub write_format_default: String, + write_format_default: String, /// The target file size for files. #[property( key = Self::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES, default = Self::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT, getter )] - pub write_target_file_size_bytes: usize, + write_target_file_size_bytes: usize, /// Base directory for metadata files (manifests, manifest lists), with any /// trailing slash trimmed. `None` if `write.metadata.path` is not set. #[property( @@ -197,7 +197,7 @@ pub struct TableProperties { parse_with = parse_location_property, getter )] - pub write_metadata_path: Option, + write_metadata_path: Option, /// Compression codec for metadata files (JSON) #[property( key = Self::PROPERTY_METADATA_COMPRESSION_CODEC, @@ -205,14 +205,14 @@ pub struct TableProperties { parse_with = parse_metadata_compression, getter )] - pub metadata_compression_codec: CompressionCodec, + metadata_compression_codec: CompressionCodec, /// Whether to use `FanoutWriter` for partitioned tables. #[property( key = Self::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED, default = Self::PROPERTY_DATAFUSION_WRITE_FANOUT_ENABLED_DEFAULT, getter )] - pub write_datafusion_fanout_enabled: bool, + write_datafusion_fanout_enabled: bool, /// Whether garbage collection is enabled on drop. /// When `false`, data files will not be deleted when a table is dropped. #[property( @@ -220,28 +220,28 @@ pub struct TableProperties { default = Self::PROPERTY_GC_ENABLED_DEFAULT, getter )] - pub gc_enabled: bool, + gc_enabled: bool, /// Default maximum age of a snapshot to keep when expiring snapshots. #[property( key = Self::PROPERTY_MAX_SNAPSHOT_AGE_MS, default = Self::PROPERTY_MAX_SNAPSHOT_AGE_MS_DEFAULT, getter )] - pub max_snapshot_age_ms: i64, + max_snapshot_age_ms: i64, /// Default minimum number of snapshots to keep per branch when expiring snapshots. #[property( key = Self::PROPERTY_MIN_SNAPSHOTS_TO_KEEP, default = Self::PROPERTY_MIN_SNAPSHOTS_TO_KEEP_DEFAULT, getter )] - pub min_snapshots_to_keep: usize, + min_snapshots_to_keep: usize, /// Default maximum age of a snapshot reference to keep when expiring snapshots. #[property( key = Self::PROPERTY_MAX_REF_AGE_MS, default = Self::PROPERTY_MAX_REF_AGE_MS_DEFAULT, getter )] - pub max_ref_age_ms: i64, + max_ref_age_ms: i64, /// Whether content-defined chunking is enabled. /// `true` only when `write.parquet.content-defined-chunking.enabled = "true"`. #[property( @@ -249,28 +249,28 @@ pub struct TableProperties { default = Self::PROPERTY_PARQUET_CDC_ENABLED_DEFAULT, getter )] - pub cdc_enabled: bool, + cdc_enabled: bool, /// Content-defined chunking minimum chunk size in bytes. #[property( key = Self::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE, default = Self::PROPERTY_PARQUET_CDC_MIN_CHUNK_SIZE_DEFAULT, getter )] - pub cdc_min_chunk_size: usize, + cdc_min_chunk_size: usize, /// Content-defined chunking maximum chunk size in bytes. #[property( key = Self::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE, default = Self::PROPERTY_PARQUET_CDC_MAX_CHUNK_SIZE_DEFAULT, getter )] - pub cdc_max_chunk_size: usize, + cdc_max_chunk_size: usize, /// Content-defined chunking normalization level (gearhash bit adjustment). #[property( key = Self::PROPERTY_PARQUET_CDC_NORM_LEVEL, default = Self::PROPERTY_PARQUET_CDC_NORM_LEVEL_DEFAULT, getter )] - pub cdc_norm_level: i32, + cdc_norm_level: i32, /// Parquet compression codec for data files, with the resolved compression /// level folded in (from `write.parquet.compression-level`, or the codec's /// default when unset). @@ -281,35 +281,35 @@ pub struct TableProperties { parse_properties_with = parse_parquet_compression, getter )] - pub parquet_compression_codec: CompressionCodec, + parquet_compression_codec: CompressionCodec, /// Approximate maximum Parquet row group size in bytes. #[property( key = Self::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES, default = Self::PROPERTY_PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT, getter )] - pub parquet_row_group_size_bytes: usize, + parquet_row_group_size_bytes: usize, /// Approximate maximum Parquet data page size in bytes. #[property( key = Self::PROPERTY_PARQUET_PAGE_SIZE_BYTES, default = Self::PROPERTY_PARQUET_PAGE_SIZE_BYTES_DEFAULT, getter )] - pub parquet_page_size_bytes: usize, + parquet_page_size_bytes: usize, /// Maximum number of rows per Parquet data page. #[property( key = Self::PROPERTY_PARQUET_PAGE_ROW_LIMIT, default = Self::PROPERTY_PARQUET_PAGE_ROW_LIMIT_DEFAULT, getter )] - pub parquet_page_row_limit: usize, + parquet_page_row_limit: usize, /// Approximate maximum Parquet dictionary page size in bytes. #[property( key = Self::PROPERTY_PARQUET_DICT_SIZE_BYTES, default = Self::PROPERTY_PARQUET_DICT_SIZE_BYTES_DEFAULT, getter )] - pub parquet_dict_size_bytes: usize, + 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. #[property( @@ -317,21 +317,21 @@ pub struct TableProperties { default = None, getter )] - pub encryption_key_id: Option, + encryption_key_id: Option, /// The encryption data encryption key length in bytes. #[property( key = Self::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH, default = Self::PROPERTY_ENCRYPTION_DATA_KEY_LENGTH_DEFAULT, getter )] - pub encryption_data_key_length: usize, + encryption_data_key_length: usize, /// Base directory for data files #[property( key = Self::PROPERTY_WRITE_DATA_LOCATION, default = None, getter )] - pub write_data_location: Option, + write_data_location: Option, /// Deprecated table property for data file write location. /// /// Property will be removed at a later date. @@ -341,7 +341,7 @@ pub struct TableProperties { default = None, getter )] - pub write_folder_storage_location: Option, + write_folder_storage_location: Option, /// Deprecated table property for data file write location for object storage location generator. /// /// Property will be removed at a later date. @@ -351,14 +351,14 @@ pub struct TableProperties { default = None, getter )] - pub write_object_storage_location: Option, + write_object_storage_location: Option, /// Whether partition values are included in object storage paths. #[property( key = Self::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS, default = Self::PROPERTY_WRITE_OBJECT_STORAGE_PARTITIONED_PATHS_DEFAULT, getter )] - pub write_object_storage_partitioned_paths: bool, + write_object_storage_partitioned_paths: bool, } impl TableProperties { diff --git a/crates/iceberg/src/transaction/expire_snapshots.rs b/crates/iceberg/src/transaction/expire_snapshots.rs index b2420a1dff..66b0291965 100644 --- a/crates/iceberg/src/transaction/expire_snapshots.rs +++ b/crates/iceberg/src/transaction/expire_snapshots.rs @@ -115,8 +115,10 @@ impl ExpireSnapshotsAction { // days) the age path always runs, so even an explicit-id-only call applies the default cutoff. let default_cutoff = self .older_than_ms - .unwrap_or_else(|| now.saturating_sub(properties.max_snapshot_age_ms)); - let default_min_to_keep = self.retain_last.unwrap_or(properties.min_snapshots_to_keep); + .unwrap_or_else(|| now.saturating_sub(properties.max_snapshot_age_ms())); + let default_min_to_keep = self + .retain_last + .unwrap_or(properties.min_snapshots_to_keep()); // Ref aging: `main` is always kept; any other ref whose head is older than its // `max_ref_age_ms` (defaulting to `history.expire.max-ref-age-ms`) is dropped, like Java's @@ -125,7 +127,7 @@ impl ExpireSnapshotsAction { let mut retained_refs: Vec<&SnapshotReference> = vec![]; for (ref_name, snapshot_ref) in &metadata.refs { if ref_name == MAIN_BRANCH - || !Self::ref_aged_out(metadata, snapshot_ref, now, properties.max_ref_age_ms) + || !Self::ref_aged_out(metadata, snapshot_ref, now, properties.max_ref_age_ms()) { retained_refs.push(snapshot_ref); } else { @@ -302,7 +304,7 @@ impl TransactionAction for ExpireSnapshotsAction { let properties = metadata.table_properties()?; // Expiring metadata defeats a user's explicit decision to disable GC (Java refuses too). - if !properties.gc_enabled { + if !properties.gc_enabled() { return Err(Error::new( ErrorKind::DataInvalid, "Cannot expire snapshots: gc.enabled is false", diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index 3e0a4e9391..9897011e4f 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -181,7 +181,7 @@ impl Transaction { let table_props = self.table.metadata().table_properties()?; // TODO(https://github.com/apache/iceberg-rust/issues/2034): remove once encrypted writes are supported - if table_props.encryption_key_id.is_some() { + if table_props.encryption_key_id().is_some() { return Err(Error::new( ErrorKind::FeatureUnsupported, "Cannot commit to an encrypted table: encrypted writes are not yet supported", @@ -205,12 +205,12 @@ impl Transaction { fn build_backoff(props: TableProperties) -> Result { Ok(ExponentialBuilder::new() - .with_min_delay(Duration::from_millis(props.commit_min_retry_wait_ms)) - .with_max_delay(Duration::from_millis(props.commit_max_retry_wait_ms)) + .with_min_delay(Duration::from_millis(props.commit_min_retry_wait_ms())) + .with_max_delay(Duration::from_millis(props.commit_max_retry_wait_ms())) .with_total_delay(Some(Duration::from_millis( - props.commit_total_retry_timeout_ms, + props.commit_total_retry_timeout_ms(), ))) - .with_max_times(props.commit_num_retries) + .with_max_times(props.commit_num_retries()) .with_factor(2.0) .build()) } diff --git a/crates/iceberg/src/writer/file_writer/location_generator.rs b/crates/iceberg/src/writer/file_writer/location_generator.rs index caaeb7b683..b8a612dc1a 100644 --- a/crates/iceberg/src/writer/file_writer/location_generator.rs +++ b/crates/iceberg/src/writer/file_writer/location_generator.rs @@ -69,8 +69,9 @@ impl DefaultLocationGenerator { let table_location = strip_trailing_slash(table_metadata.location()); let prop = TableProperties::try_from(table_metadata.properties())?; let data_location = strip_trailing_slash( - prop.write_data_location - .or(prop.write_folder_storage_location) + prop.write_data_location() + .clone() + .or_else(|| prop.write_folder_storage_location().clone()) .unwrap_or(format!("{table_location}{DEFAULT_DATA_DIR}")) .as_ref(), ) @@ -136,9 +137,10 @@ impl ObjectStorageLocationGenerator { let table_location = strip_trailing_slash(table_metadata.location()); let prop = TableProperties::try_from(table_metadata.properties())?; let storage_location = strip_trailing_slash( - prop.write_data_location - .or(prop.write_object_storage_location) - .or(prop.write_folder_storage_location) + prop.write_data_location() + .clone() + .or_else(|| prop.write_object_storage_location().clone()) + .or_else(|| prop.write_folder_storage_location().clone()) .unwrap_or(format!("{table_location}{DEFAULT_DATA_DIR}")) .as_ref(), ) @@ -152,7 +154,7 @@ impl ObjectStorageLocationGenerator { Some(path_context(table_location)) }; - let include_partition_paths = prop.write_object_storage_partitioned_paths; + let include_partition_paths = prop.write_object_storage_partitioned_paths(); Ok(Self { storage_location, diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index 4e70520abe..4b92f238b9 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -84,19 +84,19 @@ impl ParquetWriterBuilder { /// schema, translating `write.parquet.*` settings into `WriterProperties` /// instead of using parquet-rs defaults. pub fn from_table_properties(table_props: &TableProperties, schema: SchemaRef) -> Result { - let cdc = table_props.cdc_enabled.then_some(CdcOptions { - min_chunk_size: table_props.cdc_min_chunk_size, - max_chunk_size: table_props.cdc_max_chunk_size, - norm_level: table_props.cdc_norm_level, + let cdc = table_props.cdc_enabled().then_some(CdcOptions { + min_chunk_size: table_props.cdc_min_chunk_size(), + max_chunk_size: table_props.cdc_max_chunk_size(), + norm_level: table_props.cdc_norm_level(), }); - let compression = parquet_compression(table_props.parquet_compression_codec)?; + let compression = parquet_compression(*table_props.parquet_compression_codec())?; let props = WriterProperties::builder() .set_content_defined_chunking(cdc) .set_compression(compression) - .set_max_row_group_bytes(Some(table_props.parquet_row_group_size_bytes)) - .set_data_page_size_limit(table_props.parquet_page_size_bytes) - .set_data_page_row_count_limit(table_props.parquet_page_row_limit) - .set_dictionary_page_size_limit(table_props.parquet_dict_size_bytes) + .set_max_row_group_bytes(Some(table_props.parquet_row_group_size_bytes())) + .set_data_page_size_limit(table_props.parquet_page_size_bytes()) + .set_data_page_row_count_limit(table_props.parquet_page_row_limit()) + .set_dictionary_page_size_limit(table_props.parquet_dict_size_bytes()) .build(); Ok(Self::new_with_match_mode(props, schema, FieldMatchMode::Id)) } diff --git a/crates/integrations/datafusion/src/physical_plan/write.rs b/crates/integrations/datafusion/src/physical_plan/write.rs index 1afbab124b..fd0d5d2c4c 100644 --- a/crates/integrations/datafusion/src/physical_plan/write.rs +++ b/crates/integrations/datafusion/src/physical_plan/write.rs @@ -209,7 +209,7 @@ impl ExecutionPlan for IcebergWriteExec { .map_err(to_datafusion_error)?; // Check data file format - let file_format = DataFileFormat::from_str(&table_props.write_format_default) + let file_format = DataFileFormat::from_str(table_props.write_format_default()) .map_err(to_datafusion_error)?; if file_format != DataFileFormat::Parquet { return Err(to_datafusion_error(Error::new( @@ -227,7 +227,7 @@ impl ExecutionPlan for IcebergWriteExec { ) .map_err(to_datafusion_error)? .with_match_mode(FieldMatchMode::Name); - let target_file_size = table_props.write_target_file_size_bytes; + let target_file_size = table_props.write_target_file_size_bytes(); let file_io = self.table.file_io().clone(); // todo location_gen and file_name_gen should be configurable @@ -246,7 +246,7 @@ impl ExecutionPlan for IcebergWriteExec { let data_file_writer_builder = DataFileWriterBuilder::new(rolling_writer_builder); // Create TaskWriter - let fanout_enabled = table_props.write_datafusion_fanout_enabled; + let fanout_enabled = table_props.write_datafusion_fanout_enabled(); let schema = self.table.metadata().current_schema().clone(); let partition_spec = self.table.metadata().default_partition_spec().clone(); let task_writer = TaskWriter::try_new(