diff --git a/crates/modelardb_embedded/bindings/python/modelardb/operations.py b/crates/modelardb_embedded/bindings/python/modelardb/operations.py index 75a246591..80e13f669 100644 --- a/crates/modelardb_embedded/bindings/python/modelardb/operations.py +++ b/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) diff --git a/crates/modelardb_embedded/bindings/python/tests/test_operations.py b/crates/modelardb_embedded/bindings/python/tests/test_operations.py index 1c96982b7..4e62c8cd4 100644 --- a/crates/modelardb_embedded/bindings/python/tests/test_operations.py +++ b/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)) diff --git a/crates/modelardb_embedded/src/capi.rs b/crates/modelardb_embedded/src/capi.rs index 604a24a8a..64f865c1a 100644 --- a/crates/modelardb_embedded/src/capi.rs +++ b/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::() + .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 diff --git a/crates/modelardb_embedded/src/operations/client.rs b/crates/modelardb_embedded/src/operations/client.rs index 90d2f3808..b12be79f0 100644 --- a/crates/modelardb_embedded/src/operations/client.rs +++ b/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, + ) -> 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(()) diff --git a/crates/modelardb_embedded/src/operations/data_folder.rs b/crates/modelardb_embedded/src/operations/data_folder.rs index c9fb7dca1..f01c0c12f 100644 --- a/crates/modelardb_embedded/src/operations/data_folder.rs +++ b/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, + ) -> 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> = 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(), diff --git a/crates/modelardb_embedded/src/operations/mod.rs b/crates/modelardb_embedded/src/operations/mod.rs index 650fd8545..e3185c86b 100644 --- a/crates/modelardb_embedded/src/operations/mod.rs +++ b/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, + ) -> Result<()>; } /// Use the time series table metadata in `table_name`, `schema`, `error_bounds`, and `generated_columns` diff --git a/crates/modelardb_manager/src/remote.rs b/crates/modelardb_manager/src/remote.rs index c647b2d8d..e487b1375 100644 --- a/crates/modelardb_manager/src/remote.rs +++ b/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, + ) -> 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. diff --git a/crates/modelardb_server/src/configuration.rs b/crates/modelardb_server/src/configuration.rs index 4663152bc..b85643f9d 100644 --- a/crates/modelardb_server/src/configuration.rs +++ b/crates/modelardb_server/src/configuration.rs @@ -45,8 +45,6 @@ pub struct ConfigurationManager { /// The number of seconds between each transfer of data to the remote object store. If [`None`], /// data is not transferred based on time. transfer_time_in_seconds: Option, - /// The number of seconds to retain deleted data in storage before it can be removed by vacuum. - retention_period_in_seconds: usize, /// Number of threads to allocate for converting multivariate time series to univariate /// time series. pub(crate) ingestion_threads: usize, @@ -76,9 +74,6 @@ impl ConfigurationManager { let transfer_time_in_seconds = env::var("MODELARDBD_TRANSFER_TIME_IN_SECONDS") .map_or(None, |value| Some(value.parse().unwrap())); - let retention_period_in_seconds = env::var("MODELARDBD_RETENTION_PERIOD_IN_SECONDS") - .map_or(60 * 60 * 24 * 7, |value| value.parse().unwrap()); - Self { cluster_mode, multivariate_reserved_memory_in_bytes, @@ -86,7 +81,6 @@ impl ConfigurationManager { compressed_reserved_memory_in_bytes, transfer_batch_size_in_bytes, transfer_time_in_seconds, - retention_period_in_seconds, // TODO: Add support for running multiple threads per component. The individual // components in the storage engine have not been validated with multiple threads, e.g., // UncompressedDataManager may have race conditions finishing buffers if multiple @@ -223,18 +217,6 @@ impl ConfigurationManager { Ok(()) } - pub(crate) fn retention_period_in_seconds(&self) -> usize { - self.retention_period_in_seconds - } - - /// Set the new value for the retention period in seconds. - pub(crate) fn set_retention_period_in_seconds( - &mut self, - new_retention_period_in_seconds: usize, - ) { - self.retention_period_in_seconds = new_retention_period_in_seconds; - } - /// Encode the current configuration into a [`Configuration`](protocol::Configuration) /// protobuf message and serialize it. pub(crate) fn encode_and_serialize(&self) -> Vec { @@ -246,7 +228,6 @@ impl ConfigurationManager { compressed_reserved_memory_in_bytes: self.compressed_reserved_memory_in_bytes as u64, transfer_batch_size_in_bytes: self.transfer_batch_size_in_bytes.map(|v| v as u64), transfer_time_in_seconds: self.transfer_time_in_seconds.map(|v| v as u64), - retention_period_in_seconds: self.retention_period_in_seconds as u64, ingestion_threads: self.ingestion_threads as u32, compression_threads: self.compression_threads as u32, writer_threads: self.writer_threads as u32, @@ -421,34 +402,6 @@ mod tests { ); } - #[tokio::test] - async fn test_set_retention_period_in_seconds() { - let temp_dir = tempfile::tempdir().unwrap(); - let (_, configuration_manager) = create_components(&temp_dir).await; - - assert_eq!( - configuration_manager - .read() - .await - .retention_period_in_seconds(), - 60 * 60 * 24 * 7 - ); - - let new_value = 60; - configuration_manager - .write() - .await - .set_retention_period_in_seconds(new_value); - - assert_eq!( - configuration_manager - .read() - .await - .retention_period_in_seconds(), - new_value - ); - } - /// Create a [`StorageEngine`] and a [`ConfigurationManager`]. async fn create_components( temp_dir: &TempDir, diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index 97bea4ca1..a116298aa 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -331,19 +331,20 @@ impl Context { Ok(()) } - /// Vacuum the table with `table_name` if it exists. If the table does not exist or if it could - /// not be vacuumed, [`ModelarDbServerError`] is returned. - pub async fn vacuum_table(&self, table_name: &str) -> Result<()> { - let retention_period_in_seconds = self - .configuration_manager - .read() - .await - .retention_period_in_seconds(); - + /// Vacuum the table with `table_name` if it exists. If a retention period is not given, the + /// default retention period of 7 days is used. If the retention period is larger than + /// [`MAX_RETENTION_PERIOD_IN_SECONDS`](modelardb_types::types::MAX_RETENTION_PERIOD_IN_SECONDS) + /// seconds, the table does not exist, or if it could not be vacuumed, + /// [`ModelarDbServerError`] is returned. + pub async fn vacuum_table( + &self, + table_name: &str, + maybe_retention_period_in_seconds: Option, + ) -> Result<()> { self.data_folders .local_data_folder .delta_lake - .vacuum_table(table_name, retention_period_in_seconds) + .vacuum_table(table_name, maybe_retention_period_in_seconds) .await?; Ok(()) @@ -426,6 +427,7 @@ mod tests { use super::*; use modelardb_test::table::{self, NORMAL_TABLE_NAME, TIME_SERIES_TABLE_NAME}; + use modelardb_types::types::MAX_RETENTION_PERIOD_IN_SECONDS; use tempfile::TempDir; use crate::data_folders::DataFolder; @@ -756,12 +758,6 @@ mod tests { let temp_dir = tempfile::tempdir().unwrap(); let context = create_context_with_normal_table(&temp_dir).await; - context - .configuration_manager - .write() - .await - .set_retention_period_in_seconds(0); - context.truncate_table(NORMAL_TABLE_NAME).await.unwrap(); // The files should still exist on disk even though they are no longer active. @@ -773,7 +769,10 @@ mod tests { let files = std::fs::read_dir(&table_path).unwrap(); assert_eq!(files.count(), 2); - context.vacuum_table(NORMAL_TABLE_NAME).await.unwrap(); + context + .vacuum_table(NORMAL_TABLE_NAME, Some(0)) + .await + .unwrap(); // Only the _delta_log folder should remain. let files = std::fs::read_dir(&table_path).unwrap(); @@ -808,12 +807,6 @@ mod tests { let temp_dir = tempfile::tempdir().unwrap(); let context = create_context_with_time_series_table(&temp_dir).await; - context - .configuration_manager - .write() - .await - .set_retention_period_in_seconds(0); - context .truncate_table(TIME_SERIES_TABLE_NAME) .await @@ -828,13 +821,32 @@ mod tests { let files = std::fs::read_dir(&column_path).unwrap(); assert_eq!(files.count(), 1); - context.vacuum_table(TIME_SERIES_TABLE_NAME).await.unwrap(); + context + .vacuum_table(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(); assert_eq!(files.count(), 0); } + #[tokio::test] + async fn test_vacuum_table_with_out_of_bounds_retention_period() { + let temp_dir = tempfile::tempdir().unwrap(); + let context = create_context_with_time_series_table(&temp_dir).await; + + assert!( + context + .vacuum_table( + TIME_SERIES_TABLE_NAME, + Some(MAX_RETENTION_PERIOD_IN_SECONDS + 1) + ) + .await + .is_err() + ); + } + /// Create a [`Context`] with a time series table named `TIME_SERIES_TABLE_NAME` and write data /// to it. async fn create_context_with_time_series_table(temp_dir: &TempDir) -> Arc { @@ -864,7 +876,12 @@ mod tests { let temp_dir = tempfile::tempdir().unwrap(); let context = create_context(&temp_dir).await; - assert!(context.vacuum_table(TIME_SERIES_TABLE_NAME).await.is_err()); + assert!( + context + .vacuum_table(TIME_SERIES_TABLE_NAME, None) + .await + .is_err() + ); } #[tokio::test] diff --git a/crates/modelardb_server/src/remote.rs b/crates/modelardb_server/src/remote.rs index 5579a7580..66d509e4b 100644 --- a/crates/modelardb_server/src/remote.rs +++ b/crates/modelardb_server/src/remote.rs @@ -516,7 +516,7 @@ impl FlightService for FlightServiceHandler { Ok(empty_record_batch_stream()) } - 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 @@ -528,7 +528,7 @@ impl FlightService for FlightServiceHandler { for table_name in table_names { self.context - .vacuum_table(&table_name) + .vacuum_table(&table_name, maybe_retention_period_in_seconds) .await .map_err(error_to_status_invalid_argument)?; } @@ -770,13 +770,6 @@ impl FlightService for FlightServiceHandler { .await .map_err(error_to_status_internal) } - Ok(protocol::update_configuration::Setting::RetentionPeriodInSeconds) => { - let new_value = new_value.ok_or(invalid_null_error)?; - - configuration_manager.set_retention_period_in_seconds(new_value); - - Ok(()) - } _ => Err(Status::unimplemented(format!( "{setting} is not an updatable setting in the server configuration." ))), diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index 623385100..d4812921c 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -258,8 +258,15 @@ impl TestContext { async fn vacuum_table( &mut self, table_name: &str, + maybe_retention_period_in_seconds: Option, ) -> Result>, Status> { - let ticket = Ticket::new(format!("VACUUM {table_name}")); + 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.client.do_get(ticket).await } @@ -695,13 +702,6 @@ async fn test_cannot_truncate_missing_table() { #[tokio::test] async fn test_can_vacuum_normal_table() { let mut test_context = TestContext::new().await; - test_context - .update_configuration( - protocol::update_configuration::Setting::RetentionPeriodInSeconds as i32, - Some(0), - ) - .await - .unwrap(); let time_series = TestContext::generate_time_series_with_tag(false, None, Some("location")); ingest_time_series_and_flush_data( @@ -726,7 +726,10 @@ async fn test_can_vacuum_normal_table() { let files = std::fs::read_dir(&table_path).unwrap(); assert_eq!(files.count(), 2); - test_context.vacuum_table(NORMAL_TABLE_NAME).await.unwrap(); + test_context + .vacuum_table(NORMAL_TABLE_NAME, Some(0)) + .await + .unwrap(); // Only the _delta_log folder should remain. let files = std::fs::read_dir(&table_path).unwrap(); @@ -736,13 +739,6 @@ async fn test_can_vacuum_normal_table() { #[tokio::test] async fn test_can_vacuum_time_series_table() { let mut test_context = TestContext::new().await; - test_context - .update_configuration( - protocol::update_configuration::Setting::RetentionPeriodInSeconds as i32, - Some(0), - ) - .await - .unwrap(); let time_series = TestContext::generate_time_series_with_tag(false, None, Some("location")); ingest_time_series_and_flush_data( @@ -768,7 +764,7 @@ async fn test_can_vacuum_time_series_table() { assert_eq!(files.count(), 1); test_context - .vacuum_table(TIME_SERIES_TABLE_NAME) + .vacuum_table(TIME_SERIES_TABLE_NAME, Some(0)) .await .unwrap(); @@ -781,7 +777,7 @@ async fn test_can_vacuum_time_series_table() { async fn test_cannot_vacuum_missing_table() { let mut test_context = TestContext::new().await; - let result = test_context.vacuum_table(NORMAL_TABLE_NAME).await; + let result = test_context.vacuum_table(NORMAL_TABLE_NAME, None).await; assert!(result.is_err()); } @@ -1273,7 +1269,6 @@ async fn test_can_get_configuration() { Some(64 * 1024 * 1024) ); assert_eq!(configuration.transfer_time_in_seconds, None); - assert_eq!(configuration.retention_period_in_seconds, 60 * 60 * 24 * 7); assert_eq!(configuration.ingestion_threads, 1); assert_eq!(configuration.compression_threads, 1); assert_eq!(configuration.writer_threads, 1); @@ -1315,16 +1310,6 @@ async fn test_can_update_compressed_reserved_memory_in_bytes() { assert_eq!(updated_configuration.compressed_reserved_memory_in_bytes, 1); } -#[tokio::test] -async fn test_can_update_retention_period_in_seconds() { - let updated_configuration = update_and_get_configuration( - protocol::update_configuration::Setting::RetentionPeriodInSeconds as i32, - ) - .await; - - assert_eq!(updated_configuration.retention_period_in_seconds, 1); -} - async fn update_and_get_configuration(setting: i32) -> protocol::Configuration { let mut test_context = TestContext::new().await; test_context @@ -1376,7 +1361,6 @@ async fn test_cannot_update_non_nullable_setting_with_null_value() { protocol::update_configuration::Setting::MultivariateReservedMemoryInBytes as i32, protocol::update_configuration::Setting::UncompressedReservedMemoryInBytes as i32, protocol::update_configuration::Setting::CompressedReservedMemoryInBytes as i32, - protocol::update_configuration::Setting::RetentionPeriodInSeconds as i32, ] { update_configuration_and_assert_error( setting, diff --git a/crates/modelardb_storage/src/delta_lake.rs b/crates/modelardb_storage/src/delta_lake.rs index 8d8662903..c80893e11 100644 --- a/crates/modelardb_storage/src/delta_lake.rs +++ b/crates/modelardb_storage/src/delta_lake.rs @@ -37,7 +37,7 @@ use deltalake::{DeltaOps, DeltaTable, DeltaTableError}; use futures::{StreamExt, TryStreamExt}; use modelardb_types::flight::protocol; use modelardb_types::schemas::{COMPRESSED_SCHEMA, FIELD_COLUMN}; -use modelardb_types::types::TimeSeriesTableMetadata; +use modelardb_types::types::{MAX_RETENTION_PERIOD_IN_SECONDS, TimeSeriesTableMetadata}; use object_store::ObjectStore; use object_store::aws::AmazonS3Builder; use object_store::local::LocalFileSystem; @@ -466,19 +466,24 @@ impl DeltaLake { Ok(()) } - /// Vacuum the Delta Lake table with `table_name` by deleting all files that are older than - /// `retention_period_in_seconds` seconds. If the retention period is out of bounds or the - /// files could not be deleted, a [`ModelarDbStorageError`] is returned. + /// Vacuum the Delta Lake table with `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 retention period is larger than + /// [`MAX_RETENTION_PERIOD_IN_SECONDS`] seconds or the files could not be deleted, a + /// [`ModelarDbStorageError`] is returned. pub async fn vacuum_table( &self, table_name: &str, - retention_period_in_seconds: usize, + maybe_retention_period_in_seconds: Option, ) -> Result<()> { let delta_table_ops = self.delta_ops(table_name).await?; + let retention_period_in_seconds = + maybe_retention_period_in_seconds.unwrap_or(60 * 60 * 24 * 7); + let retention_period = TimeDelta::new(retention_period_in_seconds as i64, 0).ok_or( ModelarDbStorageError::InvalidArgument(format!( - "Retention period of {retention_period_in_seconds} seconds is out of bounds." + "Retention period cannot be more than {MAX_RETENTION_PERIOD_IN_SECONDS} seconds." )), )?; diff --git a/crates/modelardb_storage/src/parser.rs b/crates/modelardb_storage/src/parser.rs index c5da7a146..b35f43894 100644 --- a/crates/modelardb_storage/src/parser.rs +++ b/crates/modelardb_storage/src/parser.rs @@ -35,7 +35,8 @@ use datafusion::sql::TableReference; use datafusion::sql::planner::{ContextProvider, PlannerContext, SqlToRel}; use modelardb_types::functions::normalize_name; // Fully imported to not conflict. use modelardb_types::types::{ - ArrowTimestamp, ArrowValue, ErrorBound, GeneratedColumn, TimeSeriesTableMetadata, + ArrowTimestamp, ArrowValue, ErrorBound, GeneratedColumn, MAX_RETENTION_PERIOD_IN_SECONDS, + TimeSeriesTableMetadata, }; use sqlparser::ast::{ CascadeOption, ColumnDef, ColumnOption, ColumnOptionDef, CreateTable, DataType as SQLDataType, @@ -68,7 +69,7 @@ pub enum ModelarDbStatement { /// TRUNCATE TABLE. TruncateTable(Vec), /// VACUUM. - Vacuum(Vec), + Vacuum(Vec, Option), } /// Tokenizes and parses the SQL statement in `sql` and returns its parsed representation in the form @@ -131,10 +132,11 @@ pub fn tokenize_and_parse_sql_statement(sql_statement: &str) -> Result Ok(ModelarDbStatement::Vacuum( - variable.into_iter().map(|ident| ident.value).collect(), + Statement::NOTIFY { channel, payload } => Ok(ModelarDbStatement::Vacuum( + channel.value.split_terminator(';').map(|s| s.to_owned()).collect(), + payload.and_then(|p| p.parse::().ok()), )), Statement::Explain { .. } => Ok(ModelarDbStatement::Statement(statement)), Statement::Query(ref boxed_query) => { @@ -173,7 +175,7 @@ pub fn tokenize_and_parse_sql_expression( /// SQL dialect that extends `sqlparsers's` [`GenericDialect`] with support for parsing CREATE TIME /// SERIES TABLE table_name DDL statements, INCLUDE 'address'\[, 'address'\]+ DQL statements, and -/// VACUUM \[table_name\[, table_name\]+\] statements. +/// VACUUM \[table_name\[, table_name\]+\] \[RETAIN num_seconds\] statements. #[derive(Debug)] struct ModelarDbDialect { /// Dialect to use for identifying identifiers. @@ -497,21 +499,26 @@ impl ModelarDbDialect { } } - /// Parse VACUUM \[table_name\[, table_name\]+\] to a [`Statement::ShowVariable`] with the - /// table names in the `variable` field. Note that [`Statement::ShowVariable`] is used since - /// [`Statement`] does not have a `Vacuum` variant. A [`ParserError`] is returned if VACUUM is - /// not the first word or the table names cannot be extracted. + /// Parse VACUUM \[table_name\[, table_name\]+\] \[RETAIN num_seconds\] to a [`Statement::NOTIFY`] + /// with the table names in the `channel` field and the optional retention period in the `payload` + /// field. Note that [`Statement::NOTIFY`] is used since [`Statement`] does not have a `Vacuum` + /// variant. A [`ParserError`] is returned if VACUUM is not the first word, the table names + /// cannot be extracted, or the retention period is not a valid positive integer that is at + /// most [`MAX_RETENTION_PERIOD_IN_SECONDS`] seconds. fn parse_vacuum(&self, parser: &mut Parser) -> StdResult { // VACUUM. parser.expect_keyword(Keyword::VACUUM)?; let mut table_names = vec![]; - if Token::EOF != parser.peek_nth_token(0).token { + // If the next token is a word that is not RETAIN, attempt to parse table names. + if let Token::Word(word) = parser.peek_nth_token(0).token + && word.keyword != Keyword::RETAIN + { loop { match self.parse_word_value(parser) { Ok(table_name) => { - table_names.push(Ident::new(table_name)); + table_names.push(table_name); if Token::Comma == parser.peek_nth_token(0).token { parser.next_token(); } else { @@ -523,11 +530,45 @@ impl ModelarDbDialect { } } - // Return Statement::ShowVariable as a substitute for Vacuum. - Ok(Statement::ShowVariable { - variable: table_names, + // If the next token is RETAIN, attempt to parse the retention period in seconds. + let maybe_retention_period_in_seconds = if let Token::Word(word) = + parser.peek_nth_token(0).token + && word.keyword == Keyword::RETAIN + { + parser.expect_keyword(Keyword::RETAIN)?; + let retention_period_in_seconds = self.parse_unsigned_literal_u64(parser)?; + + if retention_period_in_seconds > MAX_RETENTION_PERIOD_IN_SECONDS { + return Err(ParserError::ParserError(format!( + "Retention period cannot be more than {MAX_RETENTION_PERIOD_IN_SECONDS} seconds." + ))); + } + + Some(retention_period_in_seconds) + } else { + None + }; + + // Return Statement::NOTIFY as a substitute for Vacuum. + Ok(Statement::NOTIFY { + channel: Ident::new(table_names.join(";")), + payload: maybe_retention_period_in_seconds.map(|period| period.to_string()), }) } + + /// Return its value as a [`u64`] if the next [`Token`] is a [`Token::Number`], otherwise a + /// [`ParserError`] is returned. + fn parse_unsigned_literal_u64(&self, parser: &mut Parser) -> StdResult { + let token_with_location = parser.next_token(); + match token_with_location.token { + Token::Number(maybe_u64, _) => maybe_u64.parse::().map_err(|error| { + ParserError::ParserError(format!( + "Failed to parse '{maybe_u64}' into a u64 due to: {error}" + )) + }), + _ => parser.expected("literal integer", token_with_location), + } + } } /// Create a [`Setting`] with `key`, `quote_style`, and `value`. @@ -558,9 +599,9 @@ impl Dialect for ModelarDbDialect { /// as a CREATE TIME SERIES TABLE DDL statement. If not, check if the next token is INCLUDE, if so, /// attempt to parse the token stream as an INCLUDE 'address'\[, 'address'\]+ DQL statement. /// If not, check if the next token is VACUUM, if so, attempt to parse the token stream as a - /// VACUUM \[table_name\[, table_name\]+\] statement. If all checks fail, [`None`] is returned - /// so [`sqlparser`] uses its parsing methods for all other statements. If parsing succeeds, a - /// [`Statement`] is returned, and if not, a [`ParserError`] is returned. + /// VACUUM \[table_name\[, table_name\]+\] \[RETAIN num_seconds\] statement. If all checks fail, + /// [`None`] is returned so [`sqlparser`] uses its parsing methods for all other statements. + /// If parsing succeeds, a [`Statement`] is returned, and if not, a [`ParserError`] is returned. fn parse_statement(&self, parser: &mut Parser) -> Option> { if self.next_tokens_are_create_time_series_table(parser) { Some(self.parse_create_time_series_table(parser)) @@ -1661,33 +1702,62 @@ mod tests { #[test] fn test_tokenize_and_parse_vacuum_all_tables() { - let table_names = parse_vacuum_and_extract_table_names("VACUUM"); + let (table_names, maybe_retention_period_in_seconds) = + parse_vacuum_and_extract_table_names("VACUUM"); assert!(table_names.is_empty()); + assert!(maybe_retention_period_in_seconds.is_none()); } #[test] fn test_tokenize_and_parse_vacuum_single_table() { - let table_names = parse_vacuum_and_extract_table_names("VACUUM table_name"); + let (table_names, maybe_retention_period_in_seconds) = + parse_vacuum_and_extract_table_names("VACUUM table_name"); assert_eq!(table_names, vec!["table_name".to_owned()]); + assert!(maybe_retention_period_in_seconds.is_none()); } #[test] fn test_tokenize_and_parse_vacuum_multiple_tables() { - let table_names = parse_vacuum_and_extract_table_names("VACUUM table_name_1, table_name_2"); + let (table_names, maybe_retention_period_in_seconds) = + parse_vacuum_and_extract_table_names("VACUUM table_name_1, table_name_2"); assert_eq!( table_names, vec!["table_name_1".to_owned(), "table_name_2".to_owned()] ); + assert!(maybe_retention_period_in_seconds.is_none()); } - fn parse_vacuum_and_extract_table_names(sql_statement: &str) -> Vec { + #[test] + fn test_tokenize_and_parse_vacuum_with_retention_period() { + let (table_names, maybe_retention_period_in_seconds) = + parse_vacuum_and_extract_table_names("VACUUM RETAIN 30"); + + assert!(table_names.is_empty()); + assert_eq!(maybe_retention_period_in_seconds, Some(30)); + } + + #[test] + fn test_tokenize_and_parse_vacuum_multiple_tables_with_retention_period() { + let (table_names, maybe_retention_period_in_seconds) = + parse_vacuum_and_extract_table_names("VACUUM table_name_1, table_name_2 RETAIN 30"); + + assert_eq!( + table_names, + vec!["table_name_1".to_owned(), "table_name_2".to_owned()] + ); + assert_eq!(maybe_retention_period_in_seconds, Some(30)); + } + + fn parse_vacuum_and_extract_table_names(sql_statement: &str) -> (Vec, Option) { let modelardb_statement = tokenize_and_parse_sql_statement(sql_statement).unwrap(); match modelardb_statement { - ModelarDbStatement::Vacuum(table_names) => table_names, + ModelarDbStatement::Vacuum(table_names, maybe_retention_period_in_seconds) => { + (table_names, maybe_retention_period_in_seconds) + } _ => panic!("Expected ModelarDbStatement::Vacuum."), } } @@ -1702,8 +1772,69 @@ mod tests { assert!(tokenize_and_parse_sql_statement("VACUUM ,table_name").is_err()); } + #[test] + fn test_tokenize_and_parse_vacuum_only_comma() { + assert!(tokenize_and_parse_sql_statement("VACUUM,").is_err()); + } + #[test] fn test_tokenize_and_parse_vacuum_quoted_table_name() { assert!(tokenize_and_parse_sql_statement("VACUUM 'table_name'").is_err()); } + + #[test] + fn test_tokenize_and_parse_vacuum_retain_without_number() { + assert!(tokenize_and_parse_sql_statement("VACUUM RETAIN").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_number_without_retain() { + assert!(tokenize_and_parse_sql_statement("VACUUM 30").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_retain_with_float() { + assert!(tokenize_and_parse_sql_statement("VACUUM RETAIN 30.5").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_retain_with_non_numeric() { + assert!(tokenize_and_parse_sql_statement("VACUUM RETAIN thirty").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_retain_with_negative() { + assert!(tokenize_and_parse_sql_statement("VACUUM RETAIN -30").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_retain_with_max_plus_one() { + assert!( + tokenize_and_parse_sql_statement(&format!( + "VACUUM RETAIN {}", + MAX_RETENTION_PERIOD_IN_SECONDS + 1 + )) + .is_err() + ); + } + + #[test] + fn test_tokenize_and_parse_vacuum_retain_twice() { + assert!(tokenize_and_parse_sql_statement("VACUUM RETAIN 30 RETAIN 30").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_multiple_tables_retain_without_number() { + assert!(tokenize_and_parse_sql_statement("VACUUM table_1, table_2 RETAIN").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_tables_and_retain_mixed() { + assert!(tokenize_and_parse_sql_statement("VACUUM table_1, RETAIN 30, table_2").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_retain_first() { + assert!(tokenize_and_parse_sql_statement("VACUUM RETAIN 30 table_1, table_2").is_err()); + } } diff --git a/crates/modelardb_types/src/flight/protocol.proto b/crates/modelardb_types/src/flight/protocol.proto index 5ddec3592..8d2babd82 100644 --- a/crates/modelardb_types/src/flight/protocol.proto +++ b/crates/modelardb_types/src/flight/protocol.proto @@ -107,17 +107,14 @@ message Configuration { // The number of seconds between each transfer of data to the remote object store. optional uint64 transfer_time_in_seconds = 5; - // The number of seconds to retain deleted data in storage before it can be removed by vacuum. - uint64 retention_period_in_seconds = 6; - // Number of threads to allocate for converting multivariate time series to univariate time series. - uint32 ingestion_threads = 7; + uint32 ingestion_threads = 6; // Number of threads to allocate for compressing univariate time series to segments. - uint32 compression_threads = 8; + uint32 compression_threads = 7; // Number of threads to allocate for writing segments to a local and/or remote data folder. - uint32 writer_threads = 9; + uint32 writer_threads = 8; } // Request to update the configuration of a ModelarDB node. @@ -128,7 +125,6 @@ message UpdateConfiguration { COMPRESSED_RESERVED_MEMORY_IN_BYTES = 2; TRANSFER_BATCH_SIZE_IN_BYTES = 3; TRANSFER_TIME_IN_SECONDS = 4; - RETENTION_PERIOD_IN_SECONDS = 5; } // Setting to update in the configuration. diff --git a/crates/modelardb_types/src/types.rs b/crates/modelardb_types/src/types.rs index 7d27a2c4f..4bcfca1d2 100644 --- a/crates/modelardb_types/src/types.rs +++ b/crates/modelardb_types/src/types.rs @@ -58,6 +58,12 @@ pub struct QueryCompressedSchema(pub Arc); #[derive(Clone)] pub struct GridSchema(pub Arc); +/// Maximum period in seconds that data can be retained in ModelarDB before it is deleted by a +/// VACUUM operation. The period is equal to the maximum value of an i64 in milliseconds. +/// The limitation is imposed by the use of `chrono::TimeDelta` when vacuuming data, which uses +/// an i64 internally to represent time in milliseconds. +pub const MAX_RETENTION_PERIOD_IN_SECONDS: u64 = (i64::MAX / 1000) as u64; + /// Types of tables supported by ModelarDB. pub enum Table { NormalTable(String, Schema),