fix: PR feedback

Signed-off-by: Robert Pack <[email protected]>

Author: roeap

crates/core/src/logstore/config.rs

@@ -1,3 +1,11 @@
+//! Configuration for the Delta Log Store.
+//!
+//! This module manages the various pieces of configuration for the Delta Log Store.
+//! It provides methods for parsing and updating configuration settings. All configuration
+//! is parsed from String -> String mappings.
+//!
+//! Specific pieces of configuration must implement the `TryUpdateKey` trait which
+//! defines how to update internal fields based on key-value pairs.
 use std::collections::HashMap;
 
 use ::object_store::RetryConfig;
@@ -42,10 +50,15 @@ where
     }
 }
 
+/// Generic container for parsing configuration
 pub(super) struct ParseResult<T> {
+    /// Parsed configuration
     pub config: T,
+    /// Unrecognized key value pairs.
     pub unparsed: HashMap<String, String>,
+    /// Errors encountered during parsing
     pub errors: Vec<(String, String)>,
+    /// Whether the configuration is defaults only - i.e. no custom values were provided
     pub is_default: bool,
 }
 
@@ -112,8 +125,15 @@ pub struct StorageConfig {
     /// Configuration to limit the number of concurrent requests to the object store.
     pub limit: Option<LimitConfig>,
 
+    /// Properties that are not recognized by the storage configuration.
+    ///
+    /// These properties are ignored by the storage configuration and can be used for custom purposes.
     pub unknown_properties: HashMap<String, String>,
 
+    /// Original unprocessed properties.
+    ///
+    /// Since we remove properties during processing, but downstream integrations may
+    /// use them for their own purposes, we keep a copy of the original properties.
     pub raw: HashMap<String, String>,
 }
 
@@ -202,6 +222,15 @@ impl StorageConfig {
         self.raw.iter()
     }
 
+    /// Parse options into a StorageConfig.
+    ///
+    /// This method will raise if it cannot parse a value. StorageConfig can also
+    /// be constructed from an iterator of key-value pairs which will ignore any
+    /// parsing errors.
+    ///
+    /// # Raises
+    ///
+    /// Raises a `DeltaError` if any of the options are invalid - i.e. cannot be parsed into target type.
     pub fn parse_options<K, V, I>(options: I) -> DeltaResult<Self>
     where
         I: IntoIterator<Item = (K, V)>,

crates/core/src/logstore/storage/cache.rs

@@ -99,23 +99,19 @@ impl config::TryUpdateKey for LogCacheConfig {
 
 async fn get_default_cache(
     config: LogCacheConfig,
-) -> DeltaResult<(DeltaLogCache, Option<TempDir>), CachedStoreError> {
-    let dir = tempfile::tempdir()?;
-
-    Ok((
-        Arc::new(
-            HybridCacheBuilder::new()
-                .memory(config.max_size_memory_mb * 1024 * 1024)
-                .storage(Engine::Large)
-                .with_device_options(
-                    DirectFsDeviceOptions::new(dir.path())
-                        .with_capacity(config.max_size_disk_mb * 1024 * 1024),
-                )
-                .build()
-                .await
-                .map_err(|_| CachedStoreError::CacheInitialization)?,
-        ),
-        Some(dir),
+    cache_dir: std::path::PathBuf,
+) -> DeltaResult<DeltaLogCache, CachedStoreError> {
+    Ok(Arc::new(
+        HybridCacheBuilder::new()
+            .memory(config.max_size_memory_mb * 1024 * 1024)
+            .storage(Engine::Large)
+            .with_device_options(
+                DirectFsDeviceOptions::new(cache_dir.as_path())
+                    .with_capacity(config.max_size_disk_mb * 1024 * 1024),
+            )
+            .build()
+            .await
+            .map_err(|_| CachedStoreError::CacheInitialization)?,
     ))
 }
 
@@ -181,7 +177,7 @@ impl std::fmt::Display for CommitCacheObjectStore {
     }
 }
 
-fn cache_delta(path: &Url) -> bool {
+fn is_delta_commit(path: &Url) -> bool {
     let Ok(Some(log_path)) = ParsedLogPath::try_from(path.clone()) else {
         return false;
     };
@@ -212,7 +208,14 @@ impl CommitCacheObjectStore {
         root_url: url::Url,
         config: LogCacheConfig,
     ) -> DeltaResult<Self> {
-        let (cache, cache_dir) = get_default_cache(config).await?;
+        let (path, cache_dir) = if let Some(dir) = &config.cache_directory {
+            let path = std::fs::canonicalize(dir)?;
+            (path, None)
+        } else {
+            let tmp_dir = tempfile::tempdir()?;
+            (tmp_dir.path().to_path_buf(), Some(tmp_dir))
+        };
+        let cache = get_default_cache(config, path).await?;
         Ok(Self::new(inner, root_url, cache, cache_dir))
     }
 
@@ -231,7 +234,7 @@ impl CommitCacheObjectStore {
     ) -> ObjectStoreResult<GetResult> {
         let cache_key = self.cache_key(location)?;
 
-        if options.range.is_some() || !cache_delta(&cache_key) || options.head {
+        if options.range.is_some() || !is_delta_commit(&cache_key) || options.head {
             return self.inner.get_opts(location, options).await;
         }
 

crates/core/src/logstore/storage/runtime.rs

@@ -68,10 +68,15 @@ fn io_rt(config: Option<&RuntimeConfig>) -> &Runtime {
 /// Configuration for Tokio runtime
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct RuntimeConfig {
+    /// Whether to use a multi-threaded runtime
     pub(crate) multi_threaded: Option<bool>,
+    /// Number of worker threads to use
     pub(crate) worker_threads: Option<usize>,
+    /// Name of the thread
     pub(crate) thread_name: Option<String>,
+    /// Whether to enable IO
     pub(crate) enable_io: Option<bool>,
+    /// Whether to enable time
     pub(crate) enable_time: Option<bool>,
 }