feat(typed-builder): migrate builder to use TypedBuilder in aws crate

fvaleye · rtyler · commit 7064361a78c8 · 2025-10-30T05:32:09.000-07:00
Signed-off-by: Florian Valeye &lt;florian.valeye@gmail.com&gt;
diff --git a/Cargo.toml b/Cargo.toml
@@ -69,6 +69,7 @@ uuid = { version = "1" }
 async-trait = { version = "0.1" }
 futures = { version = "0.3" }
 tokio = { version = "1" }
+typed-builder = { version = "0.23.0" }
 
 # opentelemetry
 tracing-opentelemetry = { version = "0.32" }
diff --git a/crates/aws/Cargo.toml b/crates/aws/Cargo.toml
@@ -25,6 +25,7 @@ object_store = { workspace = true, features = ["aws"] }
 regex = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+typed-builder = { workspace = true }
 uuid = { workspace = true, features = ["serde", "v4"] }
 url = { workspace = true }
 
diff --git a/crates/aws/src/lib.rs b/crates/aws/src/lib.rs
@@ -44,6 +44,7 @@ use storage::S3StorageOptionsConversion;
 use storage::{S3ObjectStoreFactory, S3StorageOptions};
 use tracing::debug;
 use tracing::warn;
+use typed_builder::TypedBuilder;
 use url::Url;
 
 #[derive(Clone, Debug, Default)]
@@ -118,36 +119,29 @@ pub fn register_handlers(_additional_prefixes: Option<Url>) {
 /// - temp_path: String - name of temporary file containing commit info
 /// - complete: bool - operation completed, i.e. atomic rename from `tempPath` to `fileName` succeeded
 /// - expire_time: `Option<SystemTime>` - epoch seconds at which this external commit entry is safe to be deleted
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, TypedBuilder)]
+#[builder(doc)]
 pub struct CommitEntry {
-    /// Commit version, stored as file name (e.g., 00000N.json) in dynamodb (relative to `_delta_log/`
+    /// Commit version, stored as file name (e.g., 00000N.json) in dynamodb (relative to `_delta_log/`)
     pub version: i64,
-    /// Path to temp file for this commit, relative to the `_delta_log
+    /// Path to temp file for this commit, relative to the `_delta_log` directory
+    #[builder(setter(into))]
     pub temp_path: Path,
     /// true if delta json file is successfully copied to its destination location, else false
+    #[builder(default = false)]
     pub complete: bool,
     /// If complete = true, epoch seconds at which this external commit entry is safe to be deleted
+    #[builder(default, setter(strip_option))]
     pub expire_time: Option<SystemTime>,
 }
 
-impl CommitEntry {
-    /// Create a new log entry for the given version.
-    /// Initial log entry state is incomplete.
-    pub fn new(version: i64, temp_path: Path) -> CommitEntry {
-        Self {
-            version,
-            temp_path,
-            complete: false,
-            expire_time: None,
-        }
-    }
-}
-
 /// Lock client backed by DynamoDb.
+#[derive(TypedBuilder)]
+#[builder(doc)]
 pub struct DynamoDbLockClient {
     /// DynamoDb client
     dynamodb_client: Client,
-    /// configuration of the
+    /// Configuration of the lock client
     config: DynamoDbConfig,
 }
 
@@ -204,16 +198,16 @@ impl DynamoDbLockClient {
             )
             .map_err(|err| DynamoDbConfigError::ParseMaxElapsedRequestTime { source: err })?;
 
-        let config = DynamoDbConfig {
-            billing_mode,
-            lock_table_name,
-            max_elapsed_request_time,
-            sdk_config: sdk_config.clone(),
-        };
-        Ok(Self {
-            dynamodb_client,
-            config,
-        })
+        let config = DynamoDbConfig::builder()
+            .billing_mode(billing_mode)
+            .lock_table_name(lock_table_name)
+            .max_elapsed_request_time(max_elapsed_request_time)
+            .sdk_config(sdk_config.clone())
+            .build();
+        Ok(Self::builder()
+            .dynamodb_client(dynamodb_client)
+            .config(config)
+            .build())
     }
     fn create_dynamodb_sdk_config(
         sdk_config: &SdkConfig,
@@ -612,10 +606,13 @@ impl TryFrom<&HashMap<String, AttributeValue>> for CommitEntry {
                 })
                 .transpose()?
                 .map(epoch_to_system_time);
+        let complete = extract_required_string_field(item, constants::ATTR_COMPLETE)? == "true";
+
+        // Construct directly to avoid TypedBuilder's type state issues with conditional fields
         Ok(Self {
             version,
             temp_path,
-            complete: extract_required_string_field(item, constants::ATTR_COMPLETE)? == "true",
+            complete,
             expire_time,
         })
     }
@@ -663,11 +660,18 @@ fn create_value_map(
     value_map
 }
 
-#[derive(Debug)]
+/// Configuration for DynamoDb lock client
+#[derive(Debug, TypedBuilder)]
+#[builder(doc)]
 pub struct DynamoDbConfig {
+    /// Billing mode for the DynamoDb table
     pub billing_mode: BillingMode,
+    /// Name of the lock table
+    #[builder(setter(into))]
     pub lock_table_name: String,
+    /// Maximum time to wait for DynamoDB requests
     pub max_elapsed_request_time: Duration,
+    /// AWS SDK configuration
     pub sdk_config: SdkConfig,
 }
 
@@ -773,18 +777,20 @@ mod tests {
                     .unwrap()
                     .as_secs(),
             );
-        commit_entry_roundtrip(&CommitEntry {
-            version: 0,
-            temp_path: Path::from("_delta_log/tmp/0_abc.json"),
-            complete: true,
-            expire_time: Some(system_time),
-        })?;
-        commit_entry_roundtrip(&CommitEntry {
-            version: 139,
-            temp_path: Path::from("_delta_log/tmp/0_abc.json"),
-            complete: false,
-            expire_time: None,
-        })?;
+        commit_entry_roundtrip(
+            &CommitEntry::builder()
+                .version(0)
+                .temp_path(Path::from("_delta_log/tmp/0_abc.json"))
+                .complete(true)
+                .expire_time(system_time)
+                .build(),
+        )?;
+        commit_entry_roundtrip(
+            &CommitEntry::builder()
+                .version(139)
+                .temp_path(Path::from("_delta_log/tmp/0_abc.json"))
+                .build(),
+        )?;
         Ok(())
     }
 
diff --git a/crates/aws/src/logstore/dynamodb_logstore.rs b/crates/aws/src/logstore/dynamodb_logstore.rs
@@ -10,6 +10,7 @@ use crate::{constants, CommitEntry, DynamoDbLockClient, UpdateLogEntryResult};
 use bytes::Bytes;
 use deltalake_core::{ObjectStoreError, Path};
 use tracing::{debug, error, warn};
+use typed_builder::TypedBuilder;
 use url::Url;
 
 use deltalake_core::logstore::*;
@@ -22,11 +23,19 @@ const STORE_NAME: &str = "DeltaS3ObjectStore";
 const MAX_REPAIR_RETRIES: i64 = 3;
 
 /// [`LogStore`] implementation backed by DynamoDb
+#[derive(TypedBuilder)]
+#[builder(doc)]
 pub struct S3DynamoDbLogStore {
+    /// Object store for delta log operations
     prefixed_store: ObjectStoreRef,
+    /// Root object store
     root_store: ObjectStoreRef,
+    /// DynamoDB lock client for transaction coordination
     lock_client: DynamoDbLockClient,
+    /// Log store configuration
     config: LogStoreConfig,
+    /// Table path URI
+    #[builder(setter(into))]
     table_path: String,
 }
 
@@ -72,16 +81,16 @@ impl S3DynamoDbLogStore {
             },
         })?;
         let table_path = to_uri(&location, &Path::from(""));
-        Ok(Self {
-            prefixed_store,
-            root_store,
-            lock_client,
-            config: LogStoreConfig {
+        Ok(Self::builder()
+            .prefixed_store(prefixed_store)
+            .root_store(root_store)
+            .lock_client(lock_client)
+            .config(LogStoreConfig {
                 location,
                 options: options.clone(),
-            },
-            table_path,
-        })
+            })
+            .table_path(table_path)
+            .build())
     }
 
     /// Attempt to repair an incomplete log entry by moving the temporary commit file
@@ -218,7 +227,10 @@ impl LogStore for S3DynamoDbLogStore {
             CommitOrBytes::TmpCommit(tmp_commit) => tmp_commit,
             _ => unreachable!(), // S3DynamoDBLogstore should never get Bytes
         };
-        let entry = CommitEntry::new(version, tmp_commit.clone());
+        let entry = CommitEntry::builder()
+            .version(version)
+            .temp_path(tmp_commit.clone())
+            .build();
         debug!("Writing commit entry for {self:?}: {entry:?}");
         // create log entry in dynamo db: complete = false, no expireTime
         self.lock_client
diff --git a/crates/aws/src/storage.rs b/crates/aws/src/storage.rs
@@ -22,6 +22,7 @@ use futures::Future;
 use object_store::aws::AmazonS3;
 use object_store::client::SpawnedReqwestConnector;
 use tracing::log::*;
+use typed_builder::TypedBuilder;
 use url::Url;
 
 use crate::constants;
@@ -30,6 +31,11 @@ use crate::errors::DynamoDbConfigError;
 
 const STORE_NAME: &str = "DeltaS3ObjectStore";
 
+// Default values for S3StorageOptions
+const DEFAULT_S3_POOL_IDLE_TIMEOUT_SECONDS: u64 = 15;
+const DEFAULT_STS_POOL_IDLE_TIMEOUT_SECONDS: u64 = 10;
+const DEFAULT_S3_GET_INTERNAL_SERVER_ERROR_RETRIES: usize = 10;
+
 #[derive(Clone, Default, Debug)]
 pub struct S3ObjectStoreFactory {}
 
@@ -122,21 +128,47 @@ fn is_aws(options: &HashMap<String, String>) -> bool {
 /// Options used to configure the [S3StorageBackend].
 ///
 /// Available options are described in [constants].
-#[derive(Clone, Debug)]
-#[allow(missing_docs)]
+#[derive(Clone, Debug, TypedBuilder)]
+#[builder(doc)]
 pub struct S3StorageOptions {
+    /// Whether to use virtual hosted-style requests
+    #[builder(default = false)]
     pub virtual_hosted_style_request: bool,
+    /// Locking provider to use (e.g., "dynamodb")
+    #[builder(default, setter(strip_option, into))]
     pub locking_provider: Option<String>,
+    /// Override endpoint for DynamoDB
+    #[builder(default, setter(strip_option, into))]
     pub dynamodb_endpoint: Option<String>,
+    /// Override region for DynamoDB
+    #[builder(default, setter(strip_option, into))]
     pub dynamodb_region: Option<String>,
+    /// Override access key ID for DynamoDB
+    #[builder(default, setter(strip_option, into))]
     pub dynamodb_access_key_id: Option<String>,
+    /// Override secret access key for DynamoDB
+    #[builder(default, setter(strip_option, into))]
     pub dynamodb_secret_access_key: Option<String>,
+    /// Override session token for DynamoDB
+    #[builder(default, setter(strip_option, into))]
     pub dynamodb_session_token: Option<String>,
+    /// Idle timeout for S3 connection pool
+    #[builder(default = Duration::from_secs(DEFAULT_S3_POOL_IDLE_TIMEOUT_SECONDS))]
     pub s3_pool_idle_timeout: Duration,
+    /// Idle timeout for STS connection pool
+    #[builder(default = Duration::from_secs(DEFAULT_STS_POOL_IDLE_TIMEOUT_SECONDS))]
     pub sts_pool_idle_timeout: Duration,
+    /// Number of retries for S3 internal server errors
+    #[builder(default = DEFAULT_S3_GET_INTERNAL_SERVER_ERROR_RETRIES)]
     pub s3_get_internal_server_error_retries: usize,
+    /// Allow unsafe rename operations
+    #[builder(default = false)]
     pub allow_unsafe_rename: bool,
+    /// Extra storage options not handled by other fields
+    #[builder(default)]
     pub extra_opts: HashMap<String, String>,
+    /// AWS SDK configuration
+    #[builder(default, setter(strip_option))]
     pub sdk_config: Option<SdkConfig>,
 }
 
@@ -177,15 +209,21 @@ impl S3StorageOptions {
         Self::ensure_env_var(options, constants::AWS_WEB_IDENTITY_TOKEN_FILE);
         Self::ensure_env_var(options, constants::AWS_ROLE_ARN);
         Self::ensure_env_var(options, constants::AWS_ROLE_SESSION_NAME);
-        let s3_pool_idle_timeout =
-            Self::u64_or_default(options, constants::AWS_S3_POOL_IDLE_TIMEOUT_SECONDS, 15);
-        let sts_pool_idle_timeout =
-            Self::u64_or_default(options, constants::AWS_STS_POOL_IDLE_TIMEOUT_SECONDS, 10);
+        let s3_pool_idle_timeout = Self::u64_or_default(
+            options,
+            constants::AWS_S3_POOL_IDLE_TIMEOUT_SECONDS,
+            DEFAULT_S3_POOL_IDLE_TIMEOUT_SECONDS,
+        );
+        let sts_pool_idle_timeout = Self::u64_or_default(
+            options,
+            constants::AWS_STS_POOL_IDLE_TIMEOUT_SECONDS,
+            DEFAULT_STS_POOL_IDLE_TIMEOUT_SECONDS,
+        );
 
         let s3_get_internal_server_error_retries = Self::u64_or_default(
             options,
             constants::AWS_S3_GET_INTERNAL_SERVER_ERROR_RETRIES,
-            10,
+            DEFAULT_S3_GET_INTERNAL_SERVER_ERROR_RETRIES as u64,
         ) as usize;
 
         let virtual_hosted_style_request: bool =
@@ -568,26 +606,15 @@ mod tests {
 
             let options = S3StorageOptions::try_default().unwrap();
             assert_eq!(
-                S3StorageOptions {
-                    sdk_config: Some(
+                S3StorageOptions::builder()
+                    .sdk_config(
                         SdkConfig::builder()
                             .endpoint_url("http://localhost".to_string())
                             .region(Region::from_static("us-west-1"))
                             .build()
-                    ),
-                    virtual_hosted_style_request: false,
-                    locking_provider: Some("dynamodb".to_string()),
-                    dynamodb_endpoint: None,
-                    dynamodb_region: None,
-                    dynamodb_access_key_id: None,
-                    dynamodb_secret_access_key: None,
-                    dynamodb_session_token: None,
-                    s3_pool_idle_timeout: Duration::from_secs(15),
-                    sts_pool_idle_timeout: Duration::from_secs(10),
-                    s3_get_internal_server_error_retries: 10,
-                    extra_opts: HashMap::new(),
-                    allow_unsafe_rename: false,
-                },
+                    )
+                    .locking_provider("dynamodb")
+                    .build(),
                 options
             );
         });
@@ -803,26 +830,19 @@ mod tests {
             .unwrap();
 
             assert_eq!(
-                S3StorageOptions {
-                    sdk_config: Some(
+                S3StorageOptions::builder()
+                    .sdk_config(
                         SdkConfig::builder()
                             .endpoint_url("http://localhost".to_string())
                             .region(Region::from_static("us-west-2"))
                             .build()
-                    ),
-                    virtual_hosted_style_request: false,
-                    locking_provider: Some("dynamodb".to_string()),
-                    dynamodb_endpoint: Some("http://localhost:dynamodb".to_string()),
-                    dynamodb_region: None,
-                    dynamodb_access_key_id: None,
-                    dynamodb_secret_access_key: None,
-                    dynamodb_session_token: None,
-                    s3_pool_idle_timeout: Duration::from_secs(1),
-                    sts_pool_idle_timeout: Duration::from_secs(2),
-                    s3_get_internal_server_error_retries: 3,
-                    extra_opts: HashMap::new(),
-                    allow_unsafe_rename: false,
-                },
+                    )
+                    .locking_provider("dynamodb")
+                    .dynamodb_endpoint("http://localhost:dynamodb")
+                    .s3_pool_idle_timeout(Duration::from_secs(1))
+                    .sts_pool_idle_timeout(Duration::from_secs(2))
+                    .s3_get_internal_server_error_retries(3)
+                    .build(),
                 options
             );
         });
diff --git a/crates/aws/tests/integration_s3_dynamodb.rs b/crates/aws/tests/integration_s3_dynamodb.rs
