Some more optimizations

Author: shefeek-jinnah

src/catalog.rs

@@ -20,8 +20,6 @@ use crate::metadata_writer::MetadataWriter;
 struct WriteConfig {
     /// Metadata writer for catalog operations
     writer: Arc<dyn MetadataWriter>,
-    /// Base data path for writing files
-    data_path: String,
 }
 
 /// DuckLake catalog provider
@@ -123,7 +121,6 @@ impl DuckLakeCatalog {
             catalog_path,
             write_config: Some(WriteConfig {
                 writer,
-                data_path: data_path_str,
             }),
         })
     }
@@ -197,7 +194,7 @@ impl CatalogProvider for DuckLakeCatalog {
                 // Configure writer if this catalog is writable
                 #[cfg(feature = "write")]
                 let schema = if let Some(ref config) = self.write_config {
-                    schema.with_writer(Arc::clone(&config.writer), config.data_path.clone())
+                    schema.with_writer(Arc::clone(&config.writer))
                 } else {
                     schema
                 };

src/insert_exec.rs

@@ -39,7 +39,6 @@ pub struct DuckLakeInsertExec {
     arrow_schema: SchemaRef,
     write_mode: WriteMode,
     object_store_url: Arc<ObjectStoreUrl>,
-    data_path: String,
     cache: PlanProperties,
 }
 
@@ -53,7 +52,6 @@ impl DuckLakeInsertExec {
         arrow_schema: SchemaRef,
         write_mode: WriteMode,
         object_store_url: Arc<ObjectStoreUrl>,
-        data_path: String,
     ) -> Self {
         let cache = Self::compute_properties();
         Self {
@@ -64,7 +62,6 @@ impl DuckLakeInsertExec {
             arrow_schema,
             write_mode,
             object_store_url,
-            data_path,
             cache,
         }
     }
@@ -85,7 +82,6 @@ impl Debug for DuckLakeInsertExec {
             .field("schema_name", &self.schema_name)
             .field("table_name", &self.table_name)
             .field("write_mode", &self.write_mode)
-            .field("data_path", &self.data_path)
             .finish_non_exhaustive()
     }
 }
@@ -140,7 +136,6 @@ impl ExecutionPlan for DuckLakeInsertExec {
             Arc::clone(&self.arrow_schema),
             self.write_mode,
             self.object_store_url.clone(),
-            self.data_path.clone(),
         )))
     }
 

src/table.rs

@@ -95,9 +95,6 @@ pub struct DuckLakeTable {
     /// Metadata writer for write operations (when write feature is enabled)
     #[cfg(feature = "write")]
     writer: Option<Arc<dyn MetadataWriter>>,
-    /// Data path for write operations (when write feature is enabled)
-    #[cfg(feature = "write")]
-    data_path: Option<String>,
 }
 
 impl std::fmt::Debug for DuckLakeTable {
@@ -172,8 +169,6 @@ impl DuckLakeTable {
             schema_name: None,
             #[cfg(feature = "write")]
             writer: None,
-            #[cfg(feature = "write")]
-            data_path: None,
         })
     }
 
@@ -395,17 +390,10 @@ impl DuckLakeTable {
     /// # Arguments
     /// * `schema_name` - Name of the schema this table belongs to
     /// * `writer` - Metadata writer for catalog operations
-    /// * `data_path` - Base path for data files
     #[cfg(feature = "write")]
-    pub fn with_writer(
-        mut self,
-        schema_name: String,
-        writer: Arc<dyn MetadataWriter>,
-        data_path: String,
-    ) -> Self {
+    pub fn with_writer(mut self, schema_name: String, writer: Arc<dyn MetadataWriter>) -> Self {
         self.schema_name = Some(schema_name);
         self.writer = Some(writer);
-        self.data_path = Some(data_path);
         self
     }
 
@@ -588,10 +576,6 @@ impl TableProvider for DuckLakeTable {
             DataFusionError::Internal("Schema name not set for writable table".to_string())
         })?;
 
-        let data_path = self.data_path.as_ref().ok_or_else(|| {
-            DataFusionError::Internal("Data path not set for writable table".to_string())
-        })?;
-
         let write_mode = match insert_op {
             InsertOp::Append => WriteMode::Append,
             InsertOp::Overwrite | InsertOp::Replace => WriteMode::Replace,
@@ -605,7 +589,6 @@ impl TableProvider for DuckLakeTable {
             self.schema(),
             write_mode,
             self.object_store_url.clone(),
-            data_path.clone(),
         )))
     }
 }

tests/concurrent_write_tests.rs

@@ -159,11 +159,12 @@ async fn test_write_session_cleanup_on_drop() {
     let (writer, _): (SqliteMetadataWriter, _) = create_test_writer(&temp_dir).await;
     let writer: Arc<dyn MetadataWriter> = Arc::new(writer);
     let schema = create_user_schema();
+    let object_store = create_object_store();
 
     // Dropped session should NOT upload data (buffer is just dropped)
     let file_path_str = {
         let table_writer =
-            DuckLakeTableWriter::new(Arc::clone(&writer), create_object_store()).unwrap();
+            DuckLakeTableWriter::new(Arc::clone(&writer), Arc::clone(&object_store)).unwrap();
         let mut session = table_writer
             .begin_write("main", "dropped_table", &schema, WriteMode::Replace)
             .unwrap();
@@ -172,16 +173,16 @@ async fn test_write_session_cleanup_on_drop() {
         session.file_path().to_string()
     };
     // With buffer approach, no file is created until finish() is called
-    let path = std::path::Path::new(&file_path_str);
+    let dropped_path = object_store::path::Path::from(file_path_str);
     assert!(
-        !path.exists(),
-        "No file should exist since session was dropped without finish()"
+        object_store.get(&dropped_path).await.is_err(),
+        "No object should exist since session was dropped without finish()"
     );
 
-    // Finished session should upload and keep file
+    // Finished session should upload the file
     let finished_path_str = {
         let table_writer =
-            DuckLakeTableWriter::new(Arc::clone(&writer), create_object_store()).unwrap();
+            DuckLakeTableWriter::new(Arc::clone(&writer), Arc::clone(&object_store)).unwrap();
         let mut session = table_writer
             .begin_write("main", "finished_table", &schema, WriteMode::Replace)
             .unwrap();
@@ -191,10 +192,11 @@ async fn test_write_session_cleanup_on_drop() {
         session.finish().await.unwrap();
         p
     };
-    // LocalFileSystem stores files at /<object_path>, so prepend /
-    let finished_fs_path = format!("/{}", finished_path_str);
-    let finished_path = std::path::Path::new(&finished_fs_path);
-    assert!(finished_path.exists(), "Finished file should exist");
+    let finished_path = object_store::path::Path::from(finished_path_str);
+    assert!(
+        object_store.get(&finished_path).await.is_ok(),
+        "Finished file should exist in object store"
+    );
 }
 
 #[tokio::test(flavor = "multi_thread")]