Add support for vacuum to delete files no longer referenced by a table

crates/modelardb_embedded/bindings/python/modelardb/operations.py

@@ -188,6 +188,10 @@ def __find_library(build: str) -> str:
             int modelardb_embedded_drop(void* maybe_operations_ptr,
                                         bool is_data_folder,
                                         char* table_name_ptr);
+
+            int modelardb_embedded_vacuum(void* maybe_operations_ptr,
+                                          bool is_data_folder,
+                                          char* table_name_ptr);
 
             char* modelardb_embedded_error();
             """
@@ -724,6 +728,19 @@ def drop(self, table_name: str):
         )
         self.__check_return_code_and_raise_error(return_code)
 
+    def vacuum(self, table_name: str):
+        """Vacuum the table with `table_name`.
+
+        :param table_name: The name of the table to vacuum.
+        :type table_name: str
+        :raises ValueError: If incorrect arguments are provided.
+        """
+        table_name_ptr = ffi.new("char[]", bytes(table_name, "UTF-8"))
+        return_code = self.__library.modelardb_embedded_vacuum(
+            self.__operations_ptr, self.__is_data_folder, table_name_ptr
+        )
+        self.__check_return_code_and_raise_error(return_code)
+
     def __check_return_code_and_raise_error(self, return_code: int):
         """Raises an appropriate exception based on the return code.
 

crates/modelardb_embedded/bindings/python/tests/test_operations.py

@@ -437,6 +437,37 @@ def test_data_folder_drop_error(self):
             )
             self.assertEqual(error_message, str(context.exception))
 
+    def test_data_folder_vacuum(self):
+        with TemporaryDirectory() as temp_dir:
+            os.environ["MODELARDBD_RETENTION_PERIOD_IN_SECONDS"] = "0"
+
+            data_folder = Operations.open_local(temp_dir)
+            create_tables_in_data_folder(data_folder)
+
+            data_folder.write(TIME_SERIES_TABLE_NAME, time_series_table_data())
+            data_folder.truncate(TIME_SERIES_TABLE_NAME)
+
+            # The files should still exist on disk even though they are no longer active.
+            folder_path = os.path.join(temp_dir, "tables", TIME_SERIES_TABLE_NAME, "field_column=2")
+            file_count = len(os.listdir(folder_path))
+            self.assertEqual(file_count, 1)
+
+            data_folder.vacuum(TIME_SERIES_TABLE_NAME)
+
+            # No files should remain in the column folder.
+            file_count = len(os.listdir(folder_path))
+            self.assertEqual(file_count, 0)
+
+    def test_data_folder_vacuum_error(self):
+        with TemporaryDirectory() as temp_dir:
+            data_folder = Operations.open_local(temp_dir)
+
+            with self.assertRaises(RuntimeError) as context:
+                data_folder.vacuum(MISSING_TABLE_NAME)
+
+            error_message = f"Invalid Argument Error: Table with name '{MISSING_TABLE_NAME}' does not exist."
+            self.assertEqual(error_message, str(context.exception))
+
 
 def create_tables_in_data_folder(data_folder: Operations):
     table_type = NormalTable(normal_table_schema())

crates/modelardb_embedded/src/capi.rs

@@ -932,6 +932,31 @@ unsafe fn drop(
     TOKIO_RUNTIME.block_on(modelardb.drop(table_name))
 }
 
+/// Vacuums the table with the name in `table_name_ptr` in the [`DataFolder`] or [`Client`] in
+/// `maybe_operations_ptr`. Assumes `maybe_operations_ptr` points to a [`DataFolder`] or [`Client`];
+/// and `table_name_ptr` points to a valid C string.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn modelardb_embedded_vacuum(
+    maybe_operations_ptr: *mut c_void,
+    is_data_folder: bool,
+    table_name_ptr: *const c_char,
+) -> c_int {
+    let maybe_unit = unsafe { vacuum(maybe_operations_ptr, is_data_folder, table_name_ptr) };
+    set_error_and_return_code(maybe_unit)
+}
+
+/// See documentation for [`modelardb_embedded_vacuum`].
+unsafe fn vacuum(
+    maybe_operations_ptr: *mut c_void,
+    is_data_folder: bool,
+    table_name_ptr: *const c_char,
+) -> Result<()> {
+    let modelardb = unsafe { c_void_to_operations(maybe_operations_ptr, is_data_folder)? };
+    let table_name = unsafe { c_char_ptr_to_str(table_name_ptr)? };
+
+    TOKIO_RUNTIME.block_on(modelardb.vacuum(table_name))
+}
+
 /// Return a read-only [`*const c_char`] with a human-readable representation of the last error the
 /// current thread encountered. The lifetime of the returned [`*const c_char`] ends when
 /// [`modelardb_embedded_error()`] is called again. If no errors have occurred, a zero-initialized

crates/modelardb_embedded/src/operations/client.rs

@@ -357,4 +357,13 @@ impl Operations for Client {
 
         Ok(())
     }
+
+    /// Vacuum the table with the name in `table_name`. If the table could not be vacuumed,
+    /// [`ModelarDbEmbeddedError`] is returned.
+    async fn vacuum(&mut self, table_name: &str) -> Result<()> {
+        let ticket = Ticket::new(format!("VACUUM {table_name}"));
+        self.flight_client.do_get(ticket).await?;
+
+        Ok(())
+    }
 }

crates/modelardb_embedded/src/operations/data_folder.rs

@@ -17,6 +17,7 @@
 
 use std::any::Any;
 use std::collections::HashMap;
+use std::env;
 use std::fmt::{Debug, Formatter, Result as FmtResult};
 use std::path::Path as StdPath;
 use std::pin::Pin;
@@ -851,6 +852,24 @@ impl Operations for DataFolder {
 
         Ok(())
     }
+
+    /// Vacuum the table with the name in `table_name`. If the table does not exist or the
+    /// table could not be vacuumed, [`ModelarDbEmbeddedError`] is returned.
+    async fn vacuum(&mut self, table_name: &str) -> Result<()> {
+        if self.tables().await?.contains(&table_name.to_owned()) {
+            let retention_period_in_seconds = env::var("MODELARDBD_RETENTION_PERIOD_IN_SECONDS")
+                .map_or(60 * 60 * 24 * 7, |value| value.parse().unwrap());
+
+            self.delta_lake
+                .vacuum_table(table_name, retention_period_in_seconds)
+                .await
+                .map_err(|error| error.into())
+        } else {
+            Err(ModelarDbEmbeddedError::InvalidArgument(format!(
+                "Table with name '{table_name}' does not exist."
+            )))
+        }
+    }
 }
 
 /// Sort the `uncompressed_data` from the time series table with `time_series_table_metadata`
@@ -922,6 +941,8 @@ fn schemas_are_compatible(source_schema: &Schema, target_schema: &Schema) -> boo
 mod tests {
     use super::*;
 
+    use std::sync::{LazyLock, Mutex};
+
     use arrow::array::{Array, Float64Array, Int8Array, Int16Array, Int32Array, Int64Array};
     use arrow::datatypes::{ArrowPrimitiveType, DataType, Field};
     use arrow_flight::flight_service_client::FlightServiceClient;
@@ -942,6 +963,9 @@ mod tests {
     const TIME_SERIES_TABLE_WITH_GENERATED_COLUMN_NAME: &str = "time_series_table_with_generated";
     const INVALID_COLUMN_NAME: &str = "invalid_column";
 
+    /// Lock used for env::set_var() as it is not guaranteed to be thread-safe.
+    static SET_VAR_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
+
     #[tokio::test]
     async fn test_create_normal_table() {
         let (_temp_dir, data_folder) = create_data_folder_with_normal_table().await;
@@ -2540,6 +2564,87 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_vacuum_normal_table() {
+        // env::set_var is safe to call in a single-threaded program.
+        unsafe {
+            let _mutex_guard = SET_VAR_LOCK.lock();
+            env::set_var("MODELARDBD_RETENTION_PERIOD_IN_SECONDS", "0");
+        }
+
+        let (temp_dir, mut data_folder) = create_data_folder_with_normal_table().await;
+
+        data_folder
+            .write(NORMAL_TABLE_NAME, normal_table_data())
+            .await
+            .unwrap();
+
+        data_folder.truncate(NORMAL_TABLE_NAME).await.unwrap();
+
+        // The files should still exist on disk even though they are no longer active.
+        let table_path = format!(
+            "{}/tables/{}",
+            temp_dir.path().to_str().unwrap(),
+            NORMAL_TABLE_NAME
+        );
+        let files = std::fs::read_dir(&table_path).unwrap();
+        assert_eq!(files.count(), 2);
+
+        data_folder.vacuum(NORMAL_TABLE_NAME).await.unwrap();
+
+        // Only the _delta_log folder should remain.
+        let files = std::fs::read_dir(&table_path).unwrap();
+        assert_eq!(files.count(), 1);
+    }
+
+    #[tokio::test]
+    async fn test_vacuum_time_series_table() {
+        // env::set_var is safe to call in a single-threaded program.
+        unsafe {
+            let _mutex_guard = SET_VAR_LOCK.lock();
+            env::set_var("MODELARDBD_RETENTION_PERIOD_IN_SECONDS", "0");
+        }
+
+        let (temp_dir, mut data_folder) = create_data_folder_with_time_series_table().await;
+
+        data_folder
+            .write(TIME_SERIES_TABLE_NAME, time_series_table_data())
+            .await
+            .unwrap();
+
+        data_folder.truncate(TIME_SERIES_TABLE_NAME).await.unwrap();
+
+        // The files should still exist on disk even though they are no longer active.
+        let column_path = format!(
+            "{}/tables/{}/field_column=3",
+            temp_dir.path().to_str().unwrap(),
+            TIME_SERIES_TABLE_NAME
+        );
+        let files = std::fs::read_dir(&column_path).unwrap();
+        assert_eq!(files.count(), 1);
+
+        data_folder.vacuum(TIME_SERIES_TABLE_NAME).await.unwrap();
+
+        // No files should remain in the column folder.
+        let files = std::fs::read_dir(&column_path).unwrap();
+        assert_eq!(files.count(), 0);
+    }
+
+    #[tokio::test]
+    async fn test_vacuum_missing_table() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let mut data_folder = DataFolder::open_local(temp_dir.path()).await.unwrap();
+
+        let result = data_folder.vacuum(MISSING_TABLE_NAME).await;
+
+        assert_eq!(
+            result.unwrap_err().to_string(),
+            format!(
+                "Invalid Argument Error: Table with name '{MISSING_TABLE_NAME}' does not exist."
+            )
+        );
+    }
+
     #[tokio::test]
     async fn test_move_normal_table_to_normal_table() {
         let (_temp_dir, mut source) = create_data_folder_with_normal_table().await;

crates/modelardb_embedded/src/operations/mod.rs

@@ -107,6 +107,9 @@ pub trait Operations: Sync + Send {
 
     /// Drop the table with the name in `table_name`.
     async fn drop(&mut self, table_name: &str) -> Result<()>;
+
+    /// Vacuum the table with the name in `table_name`.
+    async fn vacuum(&mut self, table_name: &str) -> Result<()>;
 }
 
 /// Use the time series table metadata in `table_name`, `schema`, `error_bounds`, and `generated_columns`

crates/modelardb_manager/src/remote.rs

@@ -21,8 +21,8 @@ use std::error::Error;
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::result::Result as StdResult;
-use std::str;
 use std::sync::Arc;
+use std::{env, str};
 
 use arrow::datatypes::Schema;
 use arrow::ipc::writer::IpcWriteOptions;
@@ -333,6 +333,33 @@ impl FlightServiceHandler {
 
         Ok(())
     }
+
+    /// Vacuum the table in the remote data folder and at each node controlled by the manager. If
+    /// the table does not exist or the table cannot be vacuumed in the remote data folder
+    /// and at each node, return [`Status`].
+    async fn vacuum_cluster_table(&self, table_name: &str) -> StdResult<(), Status> {
+        let retention_period_in_seconds = env::var("MODELARDBD_RETENTION_PERIOD_IN_SECONDS")
+            .map_or(60 * 60 * 24 * 7, |value| value.parse().unwrap());
+
+        // Vacuum the table in the remote data folder Delta lake.
+        self.context
+            .remote_data_folder
+            .delta_lake
+            .vacuum_table(table_name, retention_period_in_seconds)
+            .await
+            .map_err(error_to_status_internal)?;
+
+        // Vacuum the table in the nodes controlled by the manager.
+        self.context
+            .cluster
+            .read()
+            .await
+            .cluster_do_get(&format!("VACUUM {table_name}"), &self.context.key)
+            .await
+            .map_err(error_to_status_internal)?;
+
+        Ok(())
+    }
 }
 
 #[tonic::async_trait]
@@ -453,8 +480,8 @@ impl FlightService for FlightServiceHandler {
     }
 
     /// Execute a SQL statement provided in UTF-8 and return the schema of the result followed by
-    /// the result itself. Currently, CREATE TABLE, CREATE TIME SERIES TABLE, TRUNCATE TABLE, and
-    /// DROP TABLE are supported.
+    /// the result itself. Currently, CREATE TABLE, CREATE TIME SERIES TABLE, TRUNCATE TABLE,
+    /// DROP TABLE, and VACUUM are supported.
     async fn do_get(
         &self,
         request: Request<Ticket>,
@@ -497,6 +524,23 @@ impl FlightService for FlightServiceHandler {
                     self.drop_cluster_table(&table_name).await?;
                 }
             }
+            ModelarDbStatement::Vacuum(mut table_names) => {
+                // Vacuum all tables if no table names are provided.
+                if table_names.is_empty() {
+                    table_names = self
+                        .context
+                        .remote_data_folder
+                        .metadata_manager
+                        .table_metadata_manager
+                        .table_names()
+                        .await
+                        .map_err(error_to_status_internal)?;
+                }
+
+                for table_name in table_names {
+                    self.vacuum_cluster_table(&table_name).await?;
+                }
+            }
             // .. is not used so a compile error is raised if a new ModelarDbStatement is added.
             ModelarDbStatement::Statement(_) | ModelarDbStatement::IncludeSelect(..) => {
                 return Err(Status::invalid_argument(