Add support for setting retention period using RETAIN syntax

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

@@ -191,7 +191,8 @@ def __find_library(build: str) -> str:
 
             int modelardb_embedded_vacuum(void* maybe_operations_ptr,
                                           bool is_data_folder,
-                                          char* table_name_ptr);
+                                          char* table_name_ptr,
+                                          char* retention_period_in_seconds_ptr);
 
             char* modelardb_embedded_error();
             """
@@ -728,16 +729,28 @@ def drop(self, table_name: str):
         )
         self.__check_return_code_and_raise_error(return_code)
 
-    def vacuum(self, table_name: str):
+    def vacuum(self, table_name: str, retention_period_in_seconds: None | int = None):
         """Vacuum the table with `table_name`.
 
         :param table_name: The name of the table to vacuum.
         :type table_name: str
+        :param retention_period_in_seconds: The retention period in seconds. Data older than the retention
+        period is deleted. If `None`, the default retention period of 7 days is used.
+        :type retention_period_in_seconds: int, optional
         :raises ValueError: If incorrect arguments are provided.
         """
         table_name_ptr = ffi.new("char[]", bytes(table_name, "UTF-8"))
+
+        if retention_period_in_seconds is not None:
+            # Convert the retention period to a string to avoid issues with converting an int to a C type that uses
+            # an inconsistent amount of bits across platforms and then converting that to a 64-bit integer in Rust.
+            # The string is converted directly to an unsigned 64-bit integer in Rust.
+            retention_period_in_seconds_ptr = ffi.new("char[]", bytes(str(retention_period_in_seconds), "UTF-8"))
+        else:
+            retention_period_in_seconds_ptr = ffi.NULL
+
         return_code = self.__library.modelardb_embedded_vacuum(
-            self.__operations_ptr, self.__is_data_folder, table_name_ptr
+            self.__operations_ptr, self.__is_data_folder, table_name_ptr, retention_period_in_seconds_ptr
         )
         self.__check_return_code_and_raise_error(return_code)
 

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

@@ -439,8 +439,6 @@ def test_data_folder_drop_error(self):
 
     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)
 
@@ -452,7 +450,7 @@ def test_data_folder_vacuum(self):
             file_count = len(os.listdir(folder_path))
             self.assertEqual(file_count, 1)
 
-            data_folder.vacuum(TIME_SERIES_TABLE_NAME)
+            data_folder.vacuum(TIME_SERIES_TABLE_NAME, retention_period_in_seconds=0)
 
             # No files should remain in the column folder.
             file_count = len(os.listdir(folder_path))

crates/modelardb_embedded/src/capi.rs

@@ -933,15 +933,27 @@ unsafe fn drop(
 }
 
 /// 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.
+/// `maybe_operations_ptr` by deleting stale files that are older than `retention_period_in_seconds_ptr`
+/// seconds. Assumes `maybe_operations_ptr` points to a [`DataFolder`] or [`Client`];
+/// `table_name_ptr` points to a valid C string; and `retention_period_in_seconds_ptr` points to a
+/// valid C string. A C string is used for the retention period to avoid issues with different
+/// platforms using an inconsistent amount of bits for integer types. The string is converted
+/// directly to an unsigned 64-bit integer in Rust.
 #[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,
+    retention_period_in_seconds_ptr: *const c_char,
 ) -> c_int {
-    let maybe_unit = unsafe { vacuum(maybe_operations_ptr, is_data_folder, table_name_ptr) };
+    let maybe_unit = unsafe {
+        vacuum(
+            maybe_operations_ptr,
+            is_data_folder,
+            table_name_ptr,
+            retention_period_in_seconds_ptr,
+        )
+    };
     set_error_and_return_code(maybe_unit)
 }
 
@@ -950,11 +962,26 @@ unsafe fn vacuum(
     maybe_operations_ptr: *mut c_void,
     is_data_folder: bool,
     table_name_ptr: *const c_char,
+    retention_period_in_seconds_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))
+    let maybe_retention_period_in_seconds_str =
+        unsafe { c_char_ptr_to_maybe_str(retention_period_in_seconds_ptr)? };
+
+    let maybe_retention_period_in_seconds = maybe_retention_period_in_seconds_str
+        .map(|retention_period_in_seconds_str| {
+            retention_period_in_seconds_str
+                .parse::<u64>()
+                .map_err(|error| {
+                    ModelarDbEmbeddedError::InvalidArgument(format!(
+                        "Retention period is not a valid u64: {error}"
+                    ))
+                })
+        })
+        .transpose()?;
+
+    TOKIO_RUNTIME.block_on(modelardb.vacuum(table_name, maybe_retention_period_in_seconds))
 }
 
 /// Return a read-only [`*const c_char`] with a human-readable representation of the last error the

crates/modelardb_embedded/src/operations/client.rs

@@ -358,10 +358,24 @@ impl Operations for Client {
         Ok(())
     }
 
-    /// Vacuum the table with the name in `table_name`. If the table could not be vacuumed,
+    /// Vacuum the table with the name in `table_name` by deleting stale files that are older than
+    /// `maybe_retention_period_in_seconds` seconds. If a retention period is not given, the
+    /// default retention period of 7 days is used. If the table does not exist, the table could
+    /// not be vacuumed, or the retention period is larger than
+    /// [`MAX_RETENTION_PERIOD_IN_SECONDS`](modelardb_types::types::MAX_RETENTION_PERIOD_IN_SECONDS),
     /// [`ModelarDbEmbeddedError`] is returned.
-    async fn vacuum(&mut self, table_name: &str) -> Result<()> {
-        let ticket = Ticket::new(format!("VACUUM {table_name}"));
+    async fn vacuum(
+        &mut self,
+        table_name: &str,
+        maybe_retention_period_in_seconds: Option<u64>,
+    ) -> Result<()> {
+        let sql = if let Some(retention_period_in_seconds) = maybe_retention_period_in_seconds {
+            format!("VACUUM {table_name} RETAIN {retention_period_in_seconds}")
+        } else {
+            format!("VACUUM {table_name}")
+        };
+
+        let ticket = Ticket::new(sql);
         self.flight_client.do_get(ticket).await?;
 
         Ok(())

crates/modelardb_embedded/src/operations/data_folder.rs

@@ -17,7 +17,6 @@
 
 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;
@@ -853,15 +852,20 @@ 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<()> {
+    /// Vacuum the table with the name in `table_name` by deleting stale files that are older than
+    /// `maybe_retention_period_in_seconds` seconds. If a retention period is not given, the
+    /// default retention period of 7 days is used. If the table does not exist, the table could
+    /// not be vacuumed, or the retention period is larger than
+    /// [`MAX_RETENTION_PERIOD_IN_SECONDS`](modelardb_types::types::MAX_RETENTION_PERIOD_IN_SECONDS),
+    /// [`ModelarDbEmbeddedError`] is returned.
+    async fn vacuum(
+        &mut self,
+        table_name: &str,
+        maybe_retention_period_in_seconds: Option<u64>,
+    ) -> 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)
+                .vacuum_table(table_name, maybe_retention_period_in_seconds)
                 .await
                 .map_err(|error| error.into())
         } else {
@@ -941,8 +945,6 @@ 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;
@@ -963,9 +965,6 @@ 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;
@@ -2566,12 +2565,6 @@ 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
@@ -2590,7 +2583,10 @@ mod tests {
         let files = std::fs::read_dir(&table_path).unwrap();
         assert_eq!(files.count(), 2);
 
-        data_folder.vacuum(NORMAL_TABLE_NAME).await.unwrap();
+        data_folder
+            .vacuum(NORMAL_TABLE_NAME, Some(0))
+            .await
+            .unwrap();
 
         // Only the _delta_log folder should remain.
         let files = std::fs::read_dir(&table_path).unwrap();
@@ -2599,12 +2595,6 @@ mod tests {
 
     #[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
@@ -2623,7 +2613,10 @@ mod tests {
         let files = std::fs::read_dir(&column_path).unwrap();
         assert_eq!(files.count(), 1);
 
-        data_folder.vacuum(TIME_SERIES_TABLE_NAME).await.unwrap();
+        data_folder
+            .vacuum(TIME_SERIES_TABLE_NAME, Some(0))
+            .await
+            .unwrap();
 
         // No files should remain in the column folder.
         let files = std::fs::read_dir(&column_path).unwrap();
@@ -2635,7 +2628,7 @@ mod tests {
         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;
+        let result = data_folder.vacuum(MISSING_TABLE_NAME, None).await;
 
         assert_eq!(
             result.unwrap_err().to_string(),

crates/modelardb_embedded/src/operations/mod.rs

@@ -108,8 +108,14 @@ 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<()>;
+    /// Vacuum the table with the name in `table_name` by deleting stale files that are older than
+    /// `maybe_retention_period_in_seconds` seconds. If a retention period is not given, the
+    /// default retention period of 7 days is used.
+    async fn vacuum(
+        &mut self,
+        table_name: &str,
+        maybe_retention_period_in_seconds: Option<u64>,
+    ) -> 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;
@@ -337,24 +337,31 @@ impl FlightServiceHandler {
     /// 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());
-
+    async fn vacuum_cluster_table(
+        &self,
+        table_name: &str,
+        maybe_retention_period_in_seconds: Option<u64>,
+    ) -> StdResult<(), Status> {
         // 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)
+            .vacuum_table(table_name, maybe_retention_period_in_seconds)
             .await
             .map_err(error_to_status_internal)?;
 
         // Vacuum the table in the nodes controlled by the manager.
+        let vacuum_sql = if let Some(retention_period) = maybe_retention_period_in_seconds {
+            format!("VACUUM {table_name} RETAIN {retention_period}")
+        } else {
+            format!("VACUUM {table_name}")
+        };
+
         self.context
             .cluster
             .read()
             .await
-            .cluster_do_get(&format!("VACUUM {table_name}"), &self.context.key)
+            .cluster_do_get(&vacuum_sql, &self.context.key)
             .await
             .map_err(error_to_status_internal)?;
 
@@ -524,7 +531,7 @@ impl FlightService for FlightServiceHandler {
                     self.drop_cluster_table(&table_name).await?;
                 }
             }
-            ModelarDbStatement::Vacuum(mut table_names) => {
+            ModelarDbStatement::Vacuum(mut table_names, maybe_retention_period_in_seconds) => {
                 // Vacuum all tables if no table names are provided.
                 if table_names.is_empty() {
                     table_names = self
@@ -538,7 +545,8 @@ impl FlightService for FlightServiceHandler {
                 }
 
                 for table_name in table_names {
-                    self.vacuum_cluster_table(&table_name).await?;
+                    self.vacuum_cluster_table(&table_name, maybe_retention_period_in_seconds)
+                        .await?;
                 }
             }
             // .. is not used so a compile error is raised if a new ModelarDbStatement is added.