Always remove dropped tables from remote data folder cache

crates/modelardb_server/src/context.rs

@@ -335,6 +335,13 @@ impl Context {
         // Drop the table from the Delta Lake.
         local_data_folder.drop_table(table_name).await?;
 
+        // In a cluster, the table is dropped from the remote data folder by the node that received
+        // the statement. The other nodes still have the table in their cache, so it is removed here
+        // to avoid using the cached table if a table with the same name is created later.
+        if let Some(remote_data_folder) = &self.data_folders.maybe_remote_data_folder {
+            remote_data_folder.remove_delta_table_from_cache(table_name);
+        }
+
         Ok(())
     }
 

crates/modelardb_storage/src/data_folder/mod.rs

@@ -588,7 +588,11 @@ impl DataFolder {
         self.delete_table_metadata(table_name).await?;
 
         let table_path = format!("{TABLE_FOLDER}/{table_name}");
-        self.delete_table_files(&table_path).await
+        let deleted_paths = self.delete_table_files(&table_path).await?;
+
+        self.remove_delta_table_from_cache(table_name);
+
+        Ok(deleted_paths)
     }
 
     /// Depending on the type of the table with `table_name`, delete either the normal table metadata
@@ -634,17 +638,18 @@ impl DataFolder {
             .map_ok(|object_meta| object_meta.location)
             .boxed();
 
-        let deleted_paths = self
-            .object_store
+        self.object_store
             .delete_stream(file_locations)
             .try_collect::<Vec<Path>>()
-            .await?;
-
-        // Remove the table from the cache.
-        let delta_table_path = format!("{}/{}", self.location, table_path);
-        self.delta_table_cache.remove(&delta_table_path);
+            .await
+            .map_err(|error| error.into())
+    }
 
-        Ok(deleted_paths)
+    /// Remove the [`DeltaTable`] for the table with `table_name` from the cache so the table is
+    /// opened from the Delta Lake again the next time it is used.
+    pub fn remove_delta_table_from_cache(&self, table_name: &str) {
+        self.delta_table_cache
+            .remove(&self.location_of_table(table_name));
     }
 
     /// Truncate the Delta Lake table with `table_name` by deleting all rows in the table. If the