Add no_hardlinks option to LocalConfig and fix error handling

object_store_factory/src/local.rs

@@ -6,20 +6,33 @@ use std::sync::Arc;
 #[derive(Deserialize, Debug, PartialEq, Eq, Clone)]
 pub struct LocalConfig {
     pub data_dir: String,
+    #[serde(default = "default_false")]
+    pub no_hardlinks: bool,
+}
+
+fn default_false() -> bool {
+    false
 }
 
 impl LocalConfig {
     pub fn from_hashmap(
         map: &HashMap<String, String>,
     ) -> Result<Self, object_store::Error> {
         Ok(Self {
-            data_dir: map.get("data_dir").unwrap().clone(),
+          data_dir: map.get("data_dir")
+          .ok_or_else(|| object_store::Error::Generic {
+              store: "local",
+              source: "Missing data_dir".into(),
+          })?
+          .clone(),
+            no_hardlinks: map.get("no_hardlinks").map(|s| s == "true").unwrap_or(false),
         })
     }
 
     pub fn to_hashmap(&self) -> HashMap<String, String> {
         let mut map = HashMap::new();
         map.insert("data_dir".to_string(), self.data_dir.clone());
+        map.insert("no_hardlinks".to_string(), self.no_hardlinks.to_string());
         map
     }
 
@@ -44,6 +57,31 @@ mod tests {
         let config = LocalConfig::from_hashmap(&map)
             .expect("Failed to create config from hashmap");
         assert_eq!(config.data_dir, "/tmp/data".to_string());
+        assert_eq!(config.no_hardlinks, false); // Default value
+    }
+
+    #[test]
+    fn test_config_from_hashmap_with_no_hardlinks() {
+        let mut map = HashMap::new();
+        map.insert("data_dir".to_string(), "/tmp/data".to_string());
+        map.insert("no_hardlinks".to_string(), "true".to_string());
+
+        let config = LocalConfig::from_hashmap(&map)
+            .expect("Failed to create config from hashmap");
+        assert_eq!(config.data_dir, "/tmp/data".to_string());
+        assert_eq!(config.no_hardlinks, true);
+    }
+
+    #[test]
+    fn test_config_from_hashmap_with_no_hardlinks_false() {
+        let mut map = HashMap::new();
+        map.insert("data_dir".to_string(), "/tmp/data".to_string());
+        map.insert("no_hardlinks".to_string(), "false".to_string());
+
+        let config = LocalConfig::from_hashmap(&map)
+            .expect("Failed to create config from hashmap");
+        assert_eq!(config.data_dir, "/tmp/data".to_string());
+        assert_eq!(config.no_hardlinks, false);
     }
 
     #[test]
@@ -64,6 +102,7 @@ mod tests {
 
         let result = LocalConfig {
             data_dir: data_dir.to_string(),
+            no_hardlinks: false,
         }
         .build_local_storage();
         assert!(result.is_ok(), "Expected Ok, got Err: {:?}", result);
@@ -73,6 +112,7 @@ mod tests {
     fn test_build_local_storage_with_invalid_path() {
         let result = LocalConfig {
             data_dir: "".to_string(),
+            no_hardlinks: false,
         }
         .build_local_storage();
         assert!(result.is_err(), "Expected Err due to invalid path, got Ok");
@@ -82,10 +122,44 @@ mod tests {
     fn test_to_hashmap() {
         let local_config = LocalConfig {
             data_dir: "path/to/data".to_string(),
+            no_hardlinks: true,
         };
 
         let hashmap = local_config.to_hashmap();
 
         assert_eq!(hashmap.get("data_dir"), Some(&"path/to/data".to_string()));
+        assert_eq!(hashmap.get("no_hardlinks"), Some(&"true".to_string()));
     }
-}
+
+    #[test]
+    fn test_default_false() {
+        assert_eq!(default_false(), false);
+    }
+
+    #[test]
+    fn test_deserialize_with_default() {
+        let json = r#"
+        {
+            "data_dir": "/tmp/data"
+        }
+        "#;
+
+        let config: LocalConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(config.data_dir, "/tmp/data");
+        assert_eq!(config.no_hardlinks, false);
+    }
+
+    #[test]
+    fn test_deserialize_with_no_hardlinks() {
+        let json = r#"
+        {
+            "data_dir": "/tmp/data",
+            "no_hardlinks": true
+        }
+        "#;
+
+        let config: LocalConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(config.data_dir, "/tmp/data");
+        assert_eq!(config.no_hardlinks, true);
+    }
+}

src/config/schema.rs

@@ -637,6 +637,7 @@ cache_control = "private, max-age=86400"
             SeafowlConfig {
                 object_store: Some(ObjectStoreConfig::Local(LocalConfig {
                     data_dir: "./seafowl-data".to_string(),
+                    no_hardlinks: false,
                 })),
                 catalog: Some(Catalog::Postgres(Postgres {
                     dsn: "postgresql://user:pass@localhost:5432/somedb".to_string(),
@@ -732,6 +733,7 @@ cache_control = "private, max-age=86400"
             SeafowlConfig {
                 object_store: Some(ObjectStoreConfig::Local(LocalConfig {
                     data_dir: "some_other_path".to_string(),
+                    no_hardlinks: false,
                 })),
                 catalog: Some(Catalog::Sqlite(Sqlite {
                     dsn: "sqlite://file.sqlite".to_string(),

src/context/delta.rs

@@ -528,6 +528,7 @@ mod tests {
                     Arc::new(LocalFileSystem::new_with_prefix(tmp_dir.path()).unwrap()),
                     ObjectStoreConfig::Local(LocalConfig {
                         data_dir: tmp_dir.path().to_string_lossy().to_string(),
+                        no_hardlinks: false,
                     }),
                 ),
                 Some(tmp_dir),

src/object_store/wrapped.rs

@@ -17,6 +17,7 @@ use url::Url;
 use object_store_factory::aws::S3Config;
 use object_store_factory::google::GCSConfig;
 use object_store_factory::ObjectStoreConfig;
+use object_store_factory::local::LocalConfig;
 
 // Wrapper around the object_store crate that holds on to the original config
 // in order to provide a more efficient "upload" for the local object store
@@ -239,6 +240,9 @@ impl ObjectStore for InternalObjectStore {
     ///
     /// Will return an error if the destination already has an object.
     async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
+      if let  ObjectStoreConfig::Local(LocalConfig { no_hardlinks: true, .. }) = self.config {
+        return self.inner.copy(from, to).await;
+      }
         self.inner.copy_if_not_exists(from, to).await
     }
 
@@ -254,6 +258,9 @@ impl ObjectStore for InternalObjectStore {
             // this with a lock too, so look into using that down the line instead.
             return self.inner.rename(from, to).await;
         }
+        if let ObjectStoreConfig::Local(LocalConfig { no_hardlinks: true, .. }) = self.config {
+            return self.inner.rename(from, to).await;
+        }
         self.inner.rename_if_not_exists(from, to).await
     }
 }
@@ -264,7 +271,7 @@ mod tests {
     use crate::object_store::wrapped::InternalObjectStore;
     use datafusion::common::Result;
     use rstest::rstest;
-
+    
     use object_store_factory::aws::S3Config;
     use object_store_factory::ObjectStoreConfig;