From 42f87f763b308f1088b478909c4d9c5fecd5f686 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Sun, 17 Aug 2025 21:49:22 +0200 Subject: [PATCH 01/31] Add methods to parse VACUUM statements --- crates/modelardb_storage/src/parser.rs | 46 ++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/crates/modelardb_storage/src/parser.rs b/crates/modelardb_storage/src/parser.rs index 2f2a5cc56..2c70ab2d5 100644 --- a/crates/modelardb_storage/src/parser.rs +++ b/crates/modelardb_storage/src/parser.rs @@ -476,6 +476,52 @@ impl ModelarDbDialect { Ok(statement) } + + /// Return [`true`] if the token stream starts with VACUUM, otherwise [`false`] is returned. + /// The method does not consume tokens. + fn next_token_is_vacuum(&self, parser: &Parser) -> bool { + // VACUUM. + if let Token::Word(word) = parser.peek_nth_token(0).token { + word.keyword == Keyword::VACUUM + } else { + false + } + } + + /// 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 + /// typed incorrectly or the table names cannot be extracted. + 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) { + loop { + match self.parse_word_value(parser) { + Ok(table_name) => { + table_names.push(Ident::new(table_name)); + if Token::Comma == parser.peek_nth_token(0).token { + parser.next_token(); + } else { + break; + }; + } + Err(error) => return Err(error), + } + } + } + + // Return Statement::ShowVariable as a substitute for Vacuum. + Ok(Statement::ShowVariable { + variable: table_names + }) + } } /// Create a [`Setting`] with `key`, `quote_style`, and `value`. From 393d0d95e47e6deb4bfe53b89425c7f63a8de284 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Sun, 17 Aug 2025 23:15:45 +0200 Subject: [PATCH 02/31] Add support for VACUUM in SQL parser --- crates/modelardb_storage/src/parser.rs | 72 ++++++++++++++++++++------ 1 file changed, 57 insertions(+), 15 deletions(-) diff --git a/crates/modelardb_storage/src/parser.rs b/crates/modelardb_storage/src/parser.rs index 2c70ab2d5..5a754cad8 100644 --- a/crates/modelardb_storage/src/parser.rs +++ b/crates/modelardb_storage/src/parser.rs @@ -50,8 +50,9 @@ use sqlparser::tokenizer::{Span, Token}; use crate::error::{ModelarDbStorageError, Result}; -/// A top-level statement (CREATE, INSERT, SELECT, TRUNCATE, DROP, etc.) that have been tokenized, -/// parsed, and for which semantic checks have verified that it is compatible with ModelarDB. +/// A top-level statement (CREATE, INSERT, SELECT, TRUNCATE, DROP, VACUUM etc.) that have been +/// tokenized, parsed, and for which semantic checks have verified that it is compatible with +/// ModelarDB. #[derive(Debug)] pub enum ModelarDbStatement { /// CREATE TABLE. @@ -66,12 +67,14 @@ pub enum ModelarDbStatement { DropTable(Vec), /// TRUNCATE TABLE. TruncateTable(Vec), + /// VACUUM. + Vacuum(Vec), } /// Tokenizes and parses the SQL statement in `sql` and return its parsed representation in the form /// of a [`ModelarDbStatement`]. Returns a [`ModelarDbStorageError`] if `sql` is empty, contain /// multiple statements, or the statement is unsupported. Currently, CREATE TABLE, CREATE TIME SERIES -/// TABLE, INSERT, EXPLAIN, INCLUDE, SELECT, TRUNCATE TABLE, and DROP TABLE are supported. +/// TABLE, INSERT, EXPLAIN, INCLUDE, SELECT, TRUNCATE TABLE, DROP TABLE, and VACUUM are supported. pub fn tokenize_and_parse_sql_statement(sql_statement: &str) -> Result { let mut statements = Parser::parse_sql(&ModelarDbDialect::new(), sql_statement)?; @@ -128,6 +131,11 @@ pub fn tokenize_and_parse_sql_statement(sql_statement: &str) -> Result Ok(ModelarDbStatement::Vacuum( + variable.into_iter().map(|ident| ident.value).collect(), + )), Statement::Explain { .. } => Ok(ModelarDbStatement::Statement(statement)), Statement::Query(ref boxed_query) => { if let Some(addresses) = extract_include_addresses(boxed_query) { @@ -138,7 +146,7 @@ pub fn tokenize_and_parse_sql_statement(sql_statement: &str) -> Result Ok(ModelarDbStatement::Statement(statement)), _ => Err(ModelarDbStorageError::InvalidArgument( - "Only CREATE, DROP, TRUNCATE, EXPLAIN, INCLUDE, SELECT, and INSERT are supported." + "Only CREATE, DROP, TRUNCATE, EXPLAIN, INCLUDE, SELECT, INSERT, and VACUUM are supported." .to_owned(), )), } @@ -164,7 +172,8 @@ 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 and INCLUDE 'address'[, 'address']+ DQL statements. +/// SERIES TABLE table_name DDL statements, INCLUDE 'address'[, 'address']+ DQL statements, and +/// VACUUM \[table_name\[, table_name\]+\] statements. #[derive(Debug)] struct ModelarDbDialect { /// Dialect to use for identifying identifiers. @@ -488,14 +497,11 @@ impl ModelarDbDialect { } } - /// Parse VACUUM \[table_name \[, table_name\]+\] to a [`Statement::ShowVariable`] with the + /// 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 /// typed incorrectly or the table names cannot be extracted. - fn parse_vacuum( - &self, - parser: &mut Parser, - ) -> StdResult { + fn parse_vacuum(&self, parser: &mut Parser) -> StdResult { // VACUUM. parser.expect_keyword(Keyword::VACUUM)?; @@ -519,7 +525,7 @@ impl ModelarDbDialect { // Return Statement::ShowVariable as a substitute for Vacuum. Ok(Statement::ShowVariable { - variable: table_names + variable: table_names, }) } } @@ -550,15 +556,18 @@ impl Dialect for ModelarDbDialect { /// Check if the next tokens are CREATE TIME SERIES TABLE, if so, attempt to parse the token stream /// 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 - /// both 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. + /// 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. 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)) } else if self.next_token_is_include(parser) { Some(self.parse_include_query(parser)) + } else if self.next_token_is_vacuum(parser) { + Some(self.parse_vacuum(parser)) } else { None } @@ -1649,4 +1658,37 @@ mod tests { fn test_tokenize_and_parse_include_zero_addresses_select() { assert!(tokenize_and_parse_sql_statement("INCLUDE SELECT * FROM table_name",).is_err()); } + + #[test] + fn test_tokenize_and_parse_vacuum_all_tables() { + let table_names = parse_vacuum_and_extract_table_names("VACUUM"); + + assert!(table_names.is_empty()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_single_table() { + let table_names = parse_vacuum_and_extract_table_names("VACUUM table_name"); + + assert_eq!(table_names, vec!["table_name".to_owned()]); + } + + #[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"); + + assert_eq!( + table_names, + vec!["table_name_1".to_owned(), "table_name_2".to_owned()] + ); + } + + fn parse_vacuum_and_extract_table_names(sql_statement: &str) -> Vec { + let modelardb_statement = tokenize_and_parse_sql_statement(sql_statement).unwrap(); + + match modelardb_statement { + ModelarDbStatement::Vacuum(table_names) => table_names, + _ => panic!("Expected ModelarDbStatement::Vacuum."), + } + } } From 885df791680ac3f9b1ce82b85a0b8da2712201e2 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 13:49:09 +0200 Subject: [PATCH 03/31] Add support for setting retention period in seconds in configuration --- crates/modelardb_server/src/configuration.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/crates/modelardb_server/src/configuration.rs b/crates/modelardb_server/src/configuration.rs index 7826be7e7..bff2723d0 100644 --- a/crates/modelardb_server/src/configuration.rs +++ b/crates/modelardb_server/src/configuration.rs @@ -45,6 +45,8 @@ 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, @@ -74,6 +76,9 @@ 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, @@ -81,6 +86,7 @@ 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 @@ -217,6 +223,18 @@ 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 { From 5e711cb5ae8f149c0acbe72a0b7e290baa004ee7 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 14:09:48 +0200 Subject: [PATCH 04/31] Add retention period to configuration protocol buffer --- crates/modelardb_server/src/configuration.rs | 1 + crates/modelardb_server/src/remote.rs | 7 +++++++ crates/modelardb_types/src/flight/protocol.proto | 10 +++++++--- 3 files changed, 15 insertions(+), 3 deletions(-) diff --git a/crates/modelardb_server/src/configuration.rs b/crates/modelardb_server/src/configuration.rs index bff2723d0..5095337ca 100644 --- a/crates/modelardb_server/src/configuration.rs +++ b/crates/modelardb_server/src/configuration.rs @@ -246,6 +246,7 @@ 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, diff --git a/crates/modelardb_server/src/remote.rs b/crates/modelardb_server/src/remote.rs index 57c1cf6fa..e69b842a0 100644 --- a/crates/modelardb_server/src/remote.rs +++ b/crates/modelardb_server/src/remote.rs @@ -737,6 +737,13 @@ 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_types/src/flight/protocol.proto b/crates/modelardb_types/src/flight/protocol.proto index d54ba68e6..77e1524a5 100644 --- a/crates/modelardb_types/src/flight/protocol.proto +++ b/crates/modelardb_types/src/flight/protocol.proto @@ -107,14 +107,17 @@ 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 = 6; + uint32 ingestion_threads = 7; // Number of threads to allocate for compressing univariate time series to segments. - uint32 compression_threads = 7; + uint32 compression_threads = 8; // Number of threads to allocate for writing segments to a local and/or remote data folder. - uint32 writer_threads = 8; + uint32 writer_threads = 9; } // Request to update the configuration of a ModelarDB node. @@ -125,6 +128,7 @@ 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. From c983fa9975569210eb0be837e5e2469da585e190 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 14:16:26 +0200 Subject: [PATCH 05/31] Add test for setting retention period in configuration --- crates/modelardb_server/src/configuration.rs | 27 ++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/crates/modelardb_server/src/configuration.rs b/crates/modelardb_server/src/configuration.rs index 5095337ca..3ef100635 100644 --- a/crates/modelardb_server/src/configuration.rs +++ b/crates/modelardb_server/src/configuration.rs @@ -418,6 +418,33 @@ 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 + ); + + configuration_manager + .write() + .await + .set_retention_period_in_seconds(60); + + assert_eq!( + configuration_manager + .read() + .await + .retention_period_in_seconds(), + 60 + ); + } + /// Create a [`StorageEngine`] and a [`ConfigurationManager`]. async fn create_components( temp_dir: &TempDir, From bd039a8d3c4e00b5ac400028307192d37bb91e25 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 14:22:35 +0200 Subject: [PATCH 06/31] Add integration test for updating retention period --- crates/modelardb_server/tests/integration_test.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index b23c0a254..cf10d6588 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -1129,6 +1129,7 @@ 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); @@ -1170,6 +1171,16 @@ 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 @@ -1221,6 +1232,7 @@ 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, From b803c0795404ca33db47a7743dc49a565f141bc6 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 16:19:50 +0200 Subject: [PATCH 07/31] Add method to DeltaLake to vacuum delta table --- Cargo.lock | 1 + crates/modelardb_storage/Cargo.toml | 1 + crates/modelardb_storage/src/delta_lake.rs | 25 ++++++++++++++++++++++ 3 files changed, 27 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 1db9f7d62..da6719404 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3150,6 +3150,7 @@ dependencies = [ "arrow", "async-trait", "bytes", + "chrono", "dashmap", "datafusion", "datafusion-proto", diff --git a/crates/modelardb_storage/Cargo.toml b/crates/modelardb_storage/Cargo.toml index a868e5e03..55c00c69f 100644 --- a/crates/modelardb_storage/Cargo.toml +++ b/crates/modelardb_storage/Cargo.toml @@ -26,6 +26,7 @@ workspace = true arrow.workspace = true async-trait.workspace = true bytes.workspace = true +chrono = "0.4.41" dashmap.workspace = true datafusion.workspace = true datafusion-proto.workspace = true diff --git a/crates/modelardb_storage/src/delta_lake.rs b/crates/modelardb_storage/src/delta_lake.rs index 585763739..9c85864eb 100644 --- a/crates/modelardb_storage/src/delta_lake.rs +++ b/crates/modelardb_storage/src/delta_lake.rs @@ -22,6 +22,7 @@ use std::sync::Arc; use arrow::array::{ArrayRef, RecordBatch}; use arrow::datatypes::{DataType, Field, Schema}; +use chrono::TimeDelta; use dashmap::DashMap; use datafusion::catalog::TableProvider; use datafusion::parquet::file::properties::WriterProperties; @@ -465,6 +466,30 @@ 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. + pub async fn vacuum_table( + &self, + table_name: &str, + retention_period_in_seconds: usize, + ) -> Result<()> { + let delta_table_ops = self.delta_ops(table_name).await?; + + 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." + )), + )?; + + delta_table_ops + .vacuum() + .with_retention_period(retention_period) + .await?; + + Ok(()) + } + /// Write `columns` to a metadata Delta Lake table with `table_name`. Returns an updated /// [`DeltaTable`] version if the file was written successfully, otherwise returns /// [`ModelarDbStorageError`]. From a49afe53ce5e9eb33c24e50b7649fbdd50b97d4b Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 16:27:57 +0200 Subject: [PATCH 08/31] Add context method to vacuum a table --- crates/modelardb_server/src/context.rs | 18 ++++++++++++++++++ crates/modelardb_server/src/remote.rs | 10 ++++++++++ 2 files changed, 28 insertions(+) diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index ac3f36d04..c43437e10 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -350,6 +350,24 @@ 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(); + + self.data_folders + .local_data_folder + .delta_lake + .vacuum_table(table_name, retention_period_in_seconds) + .await?; + + Ok(()) + } + /// Lookup the [`TimeSeriesTableMetadata`] of the time series table with name `table_name` if it /// exists. Specifically, the method returns: /// * [`TimeSeriesTableMetadata`] if a time series table with the name `table_name` exists. diff --git a/crates/modelardb_server/src/remote.rs b/crates/modelardb_server/src/remote.rs index e69b842a0..2fa91f135 100644 --- a/crates/modelardb_server/src/remote.rs +++ b/crates/modelardb_server/src/remote.rs @@ -514,6 +514,16 @@ impl FlightService for FlightServiceHandler { .map_err(error_to_status_invalid_argument)?; } + Ok(empty_record_batch_stream()) + } + ModelarDbStatement::Vacuum(table_names) => { + for table_name in table_names { + self.context + .vacuum_table(&table_name) + .await + .map_err(error_to_status_invalid_argument)?; + } + Ok(empty_record_batch_stream()) } } From f73eadb1b2bf9a1ef0eb6c560a3c51207ab0dcad Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 16:52:30 +0200 Subject: [PATCH 09/31] Add utility test functions to create table and write to it --- crates/modelardb_server/src/context.rs | 89 +++++++++++++++----------- 1 file changed, 53 insertions(+), 36 deletions(-) diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index c43437e10..6aa814636 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -444,10 +444,11 @@ impl Context { mod tests { use super::*; - use crate::data_folders::DataFolder; use modelardb_test::table::{self, NORMAL_TABLE_NAME, TIME_SERIES_TABLE_NAME}; use tempfile::TempDir; + use crate::data_folders::DataFolder; + // Tests for Context. #[tokio::test] async fn test_create_tables_from_bytes() { @@ -724,12 +725,7 @@ mod tests { #[tokio::test] async fn test_truncate_normal_table() { let temp_dir = tempfile::tempdir().unwrap(); - let context = create_context(&temp_dir).await; - - context - .create_normal_table(NORMAL_TABLE_NAME, &table::normal_table_schema()) - .await - .unwrap(); + let context = create_context_with_normal_table(&temp_dir).await; let local_data_folder = &context.data_folders.local_data_folder; let mut delta_table = local_data_folder @@ -738,17 +734,6 @@ mod tests { .await .unwrap(); - // Write data to the normal table that should be deleted when the table is truncated. - local_data_folder - .delta_lake - .write_record_batches_to_normal_table( - NORMAL_TABLE_NAME, - vec![table::normal_table_record_batch()], - ) - .await - .unwrap(); - - delta_table.load().await.unwrap(); assert_eq!(delta_table.get_files_count(), 1); context.truncate_table(NORMAL_TABLE_NAME).await.unwrap(); @@ -770,12 +755,7 @@ mod tests { #[tokio::test] async fn test_truncate_time_series_table() { let temp_dir = tempfile::tempdir().unwrap(); - let context = create_context(&temp_dir).await; - - context - .create_time_series_table(&table::time_series_table_metadata()) - .await - .unwrap(); + let context = create_context_with_time_series_table(&temp_dir).await; let local_data_folder = &context.data_folders.local_data_folder; let mut delta_table = local_data_folder @@ -784,18 +764,6 @@ mod tests { .await .unwrap(); - // Write data to the time series table that should be deleted when the table is truncated. - let record_batch = table::compressed_segments_record_batch(); - local_data_folder - .delta_lake - .write_compressed_segments_to_time_series_table( - TIME_SERIES_TABLE_NAME, - vec![record_batch], - ) - .await - .unwrap(); - - delta_table.load().await.unwrap(); assert_eq!(delta_table.get_files_count(), 1); context @@ -830,6 +798,55 @@ mod tests { ); } + + /// Create a [`Context`] with a normal table named `NORMAL_TABLE_NAME` and write data to it. + async fn create_context_with_normal_table(temp_dir: &TempDir) -> Arc { + let context = create_context(&temp_dir).await; + + context + .create_normal_table(NORMAL_TABLE_NAME, &table::normal_table_schema()) + .await + .unwrap(); + + // Write data to the normal table. + let local_data_folder = &context.data_folders.local_data_folder; + local_data_folder + .delta_lake + .write_record_batches_to_normal_table( + NORMAL_TABLE_NAME, + vec![table::normal_table_record_batch()], + ) + .await + .unwrap(); + + context + } + + + /// 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 { + let context = create_context(&temp_dir).await; + + context + .create_time_series_table(&table::time_series_table_metadata()) + .await + .unwrap(); + + // Write data to the time series table. + let local_data_folder = &context.data_folders.local_data_folder; + local_data_folder + .delta_lake + .write_compressed_segments_to_time_series_table( + TIME_SERIES_TABLE_NAME, + vec![table::compressed_segments_record_batch()], + ) + .await + .unwrap(); + + context + } + #[tokio::test] async fn test_time_series_table_metadata_from_default_database_schema() { let temp_dir = tempfile::tempdir().unwrap(); From f7e469e5d9fc4d843a9dbde3d9bc25fe68e49996 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 17:44:39 +0200 Subject: [PATCH 10/31] Add test for vacuuming normal tables --- crates/modelardb_server/src/context.rs | 37 ++++++++++++++++++++++ crates/modelardb_storage/src/delta_lake.rs | 1 + 2 files changed, 38 insertions(+) diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index 6aa814636..ed3e130c7 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -798,6 +798,43 @@ mod tests { ); } + #[tokio::test] + async fn test_vacuum_normal_table() { + 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); + + let local_data_folder = &context.data_folders.local_data_folder; + let mut delta_table = local_data_folder + .delta_lake + .delta_table(NORMAL_TABLE_NAME) + .await + .unwrap(); + + context.truncate_table(NORMAL_TABLE_NAME).await.unwrap(); + delta_table.load().await.unwrap(); + assert_eq!(delta_table.get_files_count(), 0); + + // 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); + + context.vacuum_table(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); + } /// Create a [`Context`] with a normal table named `NORMAL_TABLE_NAME` and write data to it. async fn create_context_with_normal_table(temp_dir: &TempDir) -> Arc { diff --git a/crates/modelardb_storage/src/delta_lake.rs b/crates/modelardb_storage/src/delta_lake.rs index 9c85864eb..8d8662903 100644 --- a/crates/modelardb_storage/src/delta_lake.rs +++ b/crates/modelardb_storage/src/delta_lake.rs @@ -485,6 +485,7 @@ impl DeltaLake { delta_table_ops .vacuum() .with_retention_period(retention_period) + .with_enforce_retention_duration(false) .await?; Ok(()) From 32dfef6c47733ae69c74572306c64f33f3093554 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 17:44:52 +0200 Subject: [PATCH 11/31] Add test for vacuuming missing tables --- crates/modelardb_server/src/context.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index ed3e130c7..1fd3565d7 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -884,6 +884,14 @@ mod tests { context } + #[tokio::test] + async fn test_vacuum_missing_table() { + let temp_dir = tempfile::tempdir().unwrap(); + let context = create_context(&temp_dir).await; + + assert!(context.vacuum_table(TIME_SERIES_TABLE_NAME).await.is_err()); + } + #[tokio::test] async fn test_time_series_table_metadata_from_default_database_schema() { let temp_dir = tempfile::tempdir().unwrap(); From a1ad4092e95e35c748e1e9af656853f1294bba34 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 17:58:25 +0200 Subject: [PATCH 12/31] Add test for vacuuming time series tables --- crates/modelardb_server/src/context.rs | 40 ++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index 1fd3565d7..cfc9e4d82 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -859,6 +859,46 @@ mod tests { context } + #[tokio::test] + async fn test_vacuum_time_series_table() { + 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); + + let local_data_folder = &context.data_folders.local_data_folder; + let mut delta_table = local_data_folder + .delta_lake + .delta_table(TIME_SERIES_TABLE_NAME) + .await + .unwrap(); + + context + .truncate_table(TIME_SERIES_TABLE_NAME) + .await + .unwrap(); + delta_table.load().await.unwrap(); + assert_eq!(delta_table.get_files_count(), 0); + + // The files should still exist on disk even though they are no longer active. + let column_path = format!( + "{}/tables/{}/field_column=0", + temp_dir.path().to_str().unwrap(), + TIME_SERIES_TABLE_NAME + ); + let files = std::fs::read_dir(&column_path).unwrap(); + assert_eq!(files.count(), 1); + + context.vacuum_table(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); + } /// Create a [`Context`] with a time series table named `TIME_SERIES_TABLE_NAME` and write data /// to it. From eb090f05fcbe924c07829329a55bcd965e8f5b5b Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 18:10:23 +0200 Subject: [PATCH 13/31] Add integration test for vacuuming missing table --- crates/modelardb_server/tests/integration_test.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index cf10d6588..39b794002 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -673,6 +673,15 @@ async fn test_cannot_truncate_missing_table() { assert!(result.is_err()); } +#[tokio::test] +async fn test_cannot_vacuum_missing_table() { + let mut test_context = TestContext::new().await; + + let ticket = Ticket::new(format!("VACUUM {TABLE_NAME}")); + let result = test_context.client.do_get(ticket).await; + assert!(result.is_err()); +} + #[tokio::test] async fn test_can_get_schema() { let mut test_context = TestContext::new().await; From 2291cb093f9b90ce68fa347ad409b4fe4a4d6813 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 18:17:05 +0200 Subject: [PATCH 14/31] Add integration test for vacuuming normal tables --- .../tests/integration_test.rs | 49 ++++++++++++++++++- 1 file changed, 47 insertions(+), 2 deletions(-) diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index 39b794002..149efb97f 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -255,6 +255,15 @@ impl TestContext { self.client.do_get(ticket).await } + /// Vacuum a table in the server through the `do_get()` method. + async fn vacuum_table( + &mut self, + table_name: &str, + ) -> Result>, Status> { + let ticket = Ticket::new(format!("VACUUM {table_name}")); + self.client.do_get(ticket).await + } + /// Return a [`RecordBatch`] containing a time series with regular or irregular time stamps /// depending on `generate_irregular_timestamps`, generated values with noise depending on /// `multiply_noise_range`, and an optional tag. @@ -673,12 +682,48 @@ async fn test_cannot_truncate_missing_table() { assert!(result.is_err()); } +#[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( + &mut test_context, + slice::from_ref(&time_series), + TableType::NormalTable, + ) + .await; + + test_context.truncate_table(TABLE_NAME).await.unwrap(); + + // The files should still exist on disk even though they are no longer active. + let table_path = format!( + "{}/tables/{}", + test_context.temp_dir.path().to_str().unwrap(), + TABLE_NAME + ); + let files = std::fs::read_dir(&table_path).unwrap(); + assert_eq!(files.count(), 2); + + test_context.vacuum_table(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_cannot_vacuum_missing_table() { let mut test_context = TestContext::new().await; - let ticket = Ticket::new(format!("VACUUM {TABLE_NAME}")); - let result = test_context.client.do_get(ticket).await; + let result = test_context.vacuum_table(TABLE_NAME).await; assert!(result.is_err()); } From a82469a29e9feab11543e6b9d000f15d3d0be333 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 18:22:27 +0200 Subject: [PATCH 15/31] Add integration test for vacuuming time series tables --- .../tests/integration_test.rs | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index 149efb97f..d9cbef36f 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -719,6 +719,43 @@ async fn test_can_vacuum_normal_table() { assert_eq!(files.count(), 1); } +#[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( + &mut test_context, + slice::from_ref(&time_series), + TableType::TimeSeriesTable, + ) + .await; + + test_context.truncate_table(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=1", + test_context.temp_dir.path().to_str().unwrap(), + TABLE_NAME + ); + let files = std::fs::read_dir(&column_path).unwrap(); + assert_eq!(files.count(), 1); + + test_context.vacuum_table(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_cannot_vacuum_missing_table() { let mut test_context = TestContext::new().await; From 11e035dca50804eaa45faf589e5ef888c6e0fef6 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 18:39:11 +0200 Subject: [PATCH 16/31] Add handler for Vacuum statements in manager --- crates/modelardb_manager/src/remote.rs | 38 ++++++++++++++++++++++++-- crates/modelardb_server/src/remote.rs | 2 +- 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/crates/modelardb_manager/src/remote.rs b/crates/modelardb_manager/src/remote.rs index 11d3f08ad..f6c76f1d8 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; @@ -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 as usize) + .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, @@ -497,6 +524,11 @@ impl FlightService for FlightServiceHandler { self.drop_cluster_table(&table_name).await?; } } + ModelarDbStatement::Vacuum(table_names) => { + 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( diff --git a/crates/modelardb_server/src/remote.rs b/crates/modelardb_server/src/remote.rs index 2fa91f135..3a78bad25 100644 --- a/crates/modelardb_server/src/remote.rs +++ b/crates/modelardb_server/src/remote.rs @@ -434,7 +434,7 @@ 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, EXPLAIN, INCLUDE, - /// SELECT, INSERT, TRUNCATE TABLE, and DROP TABLE are supported. + /// SELECT, INSERT, TRUNCATE TABLE, DROP TABLE, and VACUUM are supported. async fn do_get( &self, request: Request, From 6a843b194cb6b3dffb984769100087b7a9e521e7 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 19:46:23 +0200 Subject: [PATCH 17/31] Vacuum all tables if no table names are provided --- crates/modelardb_manager/src/remote.rs | 14 +++++++++++++- crates/modelardb_server/src/remote.rs | 11 ++++++++++- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/crates/modelardb_manager/src/remote.rs b/crates/modelardb_manager/src/remote.rs index f6c76f1d8..40e02bfbc 100644 --- a/crates/modelardb_manager/src/remote.rs +++ b/crates/modelardb_manager/src/remote.rs @@ -524,7 +524,19 @@ impl FlightService for FlightServiceHandler { self.drop_cluster_table(&table_name).await?; } } - ModelarDbStatement::Vacuum(table_names) => { + 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?; } diff --git a/crates/modelardb_server/src/remote.rs b/crates/modelardb_server/src/remote.rs index 3a78bad25..fb9074033 100644 --- a/crates/modelardb_server/src/remote.rs +++ b/crates/modelardb_server/src/remote.rs @@ -516,7 +516,16 @@ impl FlightService for FlightServiceHandler { Ok(empty_record_batch_stream()) } - ModelarDbStatement::Vacuum(table_names) => { + ModelarDbStatement::Vacuum(mut table_names) => { + // Vacuum all tables if no table names are provided. + if table_names.is_empty() { + table_names = self + .context + .default_database_schema() + .map_err(error_to_status_internal)? + .table_names(); + }; + for table_name in table_names { self.context .vacuum_table(&table_name) From c98bcb4ebd98d3e5f5f77f49d04f035579e90b91 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 20:17:02 +0200 Subject: [PATCH 18/31] Add support for Vacuum to Operations --- crates/modelardb_embedded/src/operations/client.rs | 9 +++++++++ .../src/operations/data_folder.rs | 13 +++++++++++++ crates/modelardb_embedded/src/operations/mod.rs | 3 +++ crates/modelardb_manager/src/remote.rs | 2 +- 4 files changed, 26 insertions(+), 1 deletion(-) diff --git a/crates/modelardb_embedded/src/operations/client.rs b/crates/modelardb_embedded/src/operations/client.rs index 81021a85c..711d004bb 100644 --- a/crates/modelardb_embedded/src/operations/client.rs +++ b/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(()) + } } diff --git a/crates/modelardb_embedded/src/operations/data_folder.rs b/crates/modelardb_embedded/src/operations/data_folder.rs index 22a6a3199..cadbde292 100644 --- a/crates/modelardb_embedded/src/operations/data_folder.rs +++ b/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,18 @@ 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<()> { + 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()) + } } /// Sort the `uncompressed_data` from the time series table with `time_series_table_metadata` diff --git a/crates/modelardb_embedded/src/operations/mod.rs b/crates/modelardb_embedded/src/operations/mod.rs index 1e0f81e0a..650fd8545 100644 --- a/crates/modelardb_embedded/src/operations/mod.rs +++ b/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` diff --git a/crates/modelardb_manager/src/remote.rs b/crates/modelardb_manager/src/remote.rs index 40e02bfbc..60ae293ce 100644 --- a/crates/modelardb_manager/src/remote.rs +++ b/crates/modelardb_manager/src/remote.rs @@ -345,7 +345,7 @@ impl FlightServiceHandler { self.context .remote_data_folder .delta_lake - .vacuum_table(table_name, retention_period_in_seconds as usize) + .vacuum_table(table_name, retention_period_in_seconds) .await .map_err(error_to_status_internal)?; From 7ba8d12724fd845918afd74e3e989fedac2cc9ca Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 20:21:57 +0200 Subject: [PATCH 19/31] Add test for vacuuming missing table --- .../src/operations/data_folder.rs | 34 +++++++++++++++---- 1 file changed, 28 insertions(+), 6 deletions(-) diff --git a/crates/modelardb_embedded/src/operations/data_folder.rs b/crates/modelardb_embedded/src/operations/data_folder.rs index cadbde292..9f1d93341 100644 --- a/crates/modelardb_embedded/src/operations/data_folder.rs +++ b/crates/modelardb_embedded/src/operations/data_folder.rs @@ -856,13 +856,19 @@ impl Operations for DataFolder { /// 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<()> { - let retention_period_in_seconds = env::var("MODELARDBD_RETENTION_PERIOD_IN_SECONDS") - .map_or(60 * 60 * 24 * 7, |value| value.parse().unwrap()); + 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()) + 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." + ))) + } } } @@ -2553,6 +2559,22 @@ mod tests { ); } + + #[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; From ed56dbc5c7c29bdc50793ac32e887a0ceff6a2ce Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 20:29:48 +0200 Subject: [PATCH 20/31] Add test for vacuuming normal tables --- .../src/operations/data_folder.rs | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/crates/modelardb_embedded/src/operations/data_folder.rs b/crates/modelardb_embedded/src/operations/data_folder.rs index 9f1d93341..6a974fc9f 100644 --- a/crates/modelardb_embedded/src/operations/data_folder.rs +++ b/crates/modelardb_embedded/src/operations/data_folder.rs @@ -941,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; @@ -961,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> = LazyLock::new(|| Mutex::new(())); + #[tokio::test] async fn test_create_normal_table() { let (_temp_dir, data_folder) = create_data_folder_with_normal_table().await; @@ -2559,6 +2564,39 @@ 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_missing_table() { From 6b6893f680750a0faeb0fbd7d3dc9e506c65773a Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 20:32:47 +0200 Subject: [PATCH 21/31] Add test for vacuuming time series tables --- .../src/operations/data_folder.rs | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/crates/modelardb_embedded/src/operations/data_folder.rs b/crates/modelardb_embedded/src/operations/data_folder.rs index 6a974fc9f..52d5c4cc0 100644 --- a/crates/modelardb_embedded/src/operations/data_folder.rs +++ b/crates/modelardb_embedded/src/operations/data_folder.rs @@ -2597,6 +2597,38 @@ mod tests { 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() { From 55382ca040a157a51c72fddd5210399c1202042b Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 20:39:48 +0200 Subject: [PATCH 22/31] Add functions to C-API to support Vacuum --- crates/modelardb_embedded/src/capi.rs | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/crates/modelardb_embedded/src/capi.rs b/crates/modelardb_embedded/src/capi.rs index a894cc480..604a24a8a 100644 --- a/crates/modelardb_embedded/src/capi.rs +++ b/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 From 163d97761ba209f24c5370d4bdf39880bc40b5f2 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 20:59:34 +0200 Subject: [PATCH 23/31] Add method to Python bindings to support vacuum --- .../bindings/python/modelardb/operations.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/crates/modelardb_embedded/bindings/python/modelardb/operations.py b/crates/modelardb_embedded/bindings/python/modelardb/operations.py index 7430f7b6e..75a246591 100644 --- a/crates/modelardb_embedded/bindings/python/modelardb/operations.py +++ b/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. From 8f616fd7768120c48817de383da49d6e4d20bb9d Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 21:18:42 +0200 Subject: [PATCH 24/31] Add Python unit tests for data folder vacuum --- .../bindings/python/tests/test_operations.py | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/crates/modelardb_embedded/bindings/python/tests/test_operations.py b/crates/modelardb_embedded/bindings/python/tests/test_operations.py index 95e3a8681..1c96982b7 100644 --- a/crates/modelardb_embedded/bindings/python/tests/test_operations.py +++ b/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()) From 45a5f39f6a35390e9e2b1f8609bbfdb41001137b Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Mon, 18 Aug 2025 21:46:00 +0200 Subject: [PATCH 25/31] Fix clippy issues and run Rustfmt --- crates/modelardb_embedded/src/operations/data_folder.rs | 2 +- crates/modelardb_server/src/context.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/modelardb_embedded/src/operations/data_folder.rs b/crates/modelardb_embedded/src/operations/data_folder.rs index 52d5c4cc0..c9fb7dca1 100644 --- a/crates/modelardb_embedded/src/operations/data_folder.rs +++ b/crates/modelardb_embedded/src/operations/data_folder.rs @@ -942,7 +942,7 @@ 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; diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index cfc9e4d82..65f02d20c 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -838,7 +838,7 @@ mod tests { /// Create a [`Context`] with a normal table named `NORMAL_TABLE_NAME` and write data to it. async fn create_context_with_normal_table(temp_dir: &TempDir) -> Arc { - let context = create_context(&temp_dir).await; + let context = create_context(temp_dir).await; context .create_normal_table(NORMAL_TABLE_NAME, &table::normal_table_schema()) @@ -903,7 +903,7 @@ mod tests { /// 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 { - let context = create_context(&temp_dir).await; + let context = create_context(temp_dir).await; context .create_time_series_table(&table::time_series_table_metadata()) From d67dfc0dec759468cb64e36007b9a7eb6be1026e Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Tue, 19 Aug 2025 20:41:46 +0200 Subject: [PATCH 26/31] Remove extra check from context vacuum tests --- crates/modelardb_server/src/context.rs | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index 65f02d20c..e30675a44 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -809,16 +809,7 @@ mod tests { .await .set_retention_period_in_seconds(0); - let local_data_folder = &context.data_folders.local_data_folder; - let mut delta_table = local_data_folder - .delta_lake - .delta_table(NORMAL_TABLE_NAME) - .await - .unwrap(); - context.truncate_table(NORMAL_TABLE_NAME).await.unwrap(); - delta_table.load().await.unwrap(); - assert_eq!(delta_table.get_files_count(), 0); // The files should still exist on disk even though they are no longer active. let table_path = format!( @@ -870,19 +861,10 @@ mod tests { .await .set_retention_period_in_seconds(0); - let local_data_folder = &context.data_folders.local_data_folder; - let mut delta_table = local_data_folder - .delta_lake - .delta_table(TIME_SERIES_TABLE_NAME) - .await - .unwrap(); - context .truncate_table(TIME_SERIES_TABLE_NAME) .await .unwrap(); - delta_table.load().await.unwrap(); - assert_eq!(delta_table.get_files_count(), 0); // The files should still exist on disk even though they are no longer active. let column_path = format!( From 4d97689677c2a8d07cc195d22d03c3a650d4fe4d Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Tue, 19 Aug 2025 20:51:07 +0200 Subject: [PATCH 27/31] Use token field when checking for EOF --- crates/modelardb_storage/src/parser.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/modelardb_storage/src/parser.rs b/crates/modelardb_storage/src/parser.rs index 5a754cad8..db8352f95 100644 --- a/crates/modelardb_storage/src/parser.rs +++ b/crates/modelardb_storage/src/parser.rs @@ -507,7 +507,7 @@ impl ModelarDbDialect { let mut table_names = vec![]; - if Token::EOF != parser.peek_nth_token(0) { + if Token::EOF != parser.peek_nth_token(0).token { loop { match self.parse_word_value(parser) { Ok(table_name) => { From a9dfdd8c630341f11884ef004ff342ce91686652 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Wed, 20 Aug 2025 19:13:50 +0200 Subject: [PATCH 28/31] Use variable in configuration tests --- crates/modelardb_server/src/configuration.rs | 30 ++++++++++++-------- 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/crates/modelardb_server/src/configuration.rs b/crates/modelardb_server/src/configuration.rs index 3ef100635..f29f937d6 100644 --- a/crates/modelardb_server/src/configuration.rs +++ b/crates/modelardb_server/src/configuration.rs @@ -287,10 +287,11 @@ mod tests { 512 * 1024 * 1024 ); + let new_value = 1024; configuration_manager .write() .await - .set_multivariate_reserved_memory_in_bytes(1024, storage_engine) + .set_multivariate_reserved_memory_in_bytes(new_value, storage_engine) .await; assert_eq!( @@ -298,7 +299,7 @@ mod tests { .read() .await .multivariate_reserved_memory_in_bytes(), - 1024 + new_value ); } @@ -315,10 +316,11 @@ mod tests { 512 * 1024 * 1024 ); + let new_value = 1024; configuration_manager .write() .await - .set_uncompressed_reserved_memory_in_bytes(1024, storage_engine) + .set_uncompressed_reserved_memory_in_bytes(new_value, storage_engine) .await .unwrap(); @@ -327,7 +329,7 @@ mod tests { .read() .await .uncompressed_reserved_memory_in_bytes(), - 1024 + new_value ); } @@ -344,10 +346,11 @@ mod tests { 512 * 1024 * 1024 ); + let new_value = 1024; configuration_manager .write() .await - .set_compressed_reserved_memory_in_bytes(1024, storage_engine) + .set_compressed_reserved_memory_in_bytes(new_value, storage_engine) .await .unwrap(); @@ -356,7 +359,7 @@ mod tests { .read() .await .compressed_reserved_memory_in_bytes(), - 1024 + new_value ); } @@ -373,10 +376,11 @@ mod tests { Some(64 * 1024 * 1024) ); + let new_value = Some(1024); configuration_manager .write() .await - .set_transfer_batch_size_in_bytes(Some(1024), storage_engine) + .set_transfer_batch_size_in_bytes(new_value, storage_engine) .await .unwrap(); @@ -385,7 +389,7 @@ mod tests { .read() .await .transfer_batch_size_in_bytes(), - Some(1024) + new_value ); } @@ -402,10 +406,11 @@ mod tests { None ); + let new_value = Some(60); configuration_manager .write() .await - .set_transfer_time_in_seconds(Some(60), storage_engine) + .set_transfer_time_in_seconds(new_value, storage_engine) .await .unwrap(); @@ -414,7 +419,7 @@ mod tests { .read() .await .transfer_time_in_seconds(), - Some(60) + new_value ); } @@ -431,17 +436,18 @@ mod tests { 60 * 60 * 24 * 7 ); + let new_value = 60; configuration_manager .write() .await - .set_retention_period_in_seconds(60); + .set_retention_period_in_seconds(new_value); assert_eq!( configuration_manager .read() .await .retention_period_in_seconds(), - 60 + new_value ); } From c7e2a2557b496b8cb9287bcdd86da57fee273f24 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Wed, 20 Aug 2025 19:23:42 +0200 Subject: [PATCH 29/31] Remove TABLE_NAME constant from integration tests --- .../tests/integration_test.rs | 181 +++++++++++------- 1 file changed, 117 insertions(+), 64 deletions(-) diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index d9cbef36f..11746b361 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -48,7 +48,6 @@ use tokio::time; use tonic::transport::Channel; use tonic::{Request, Response, Status, Streaming}; -const TABLE_NAME: &str = "table_name"; const HOST: &str = "127.0.0.1"; /// The next port to be used for the server in an integration test. Each test uses a unique port and @@ -473,13 +472,13 @@ async fn test_can_create_normal_table() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::NormalTable) + .create_table(NORMAL_TABLE_NAME, TableType::NormalTable) .await; let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); assert_eq!(retrieved_table_names.len(), 1); - assert_eq!(retrieved_table_names[0], TABLE_NAME); + assert_eq!(retrieved_table_names[0], NORMAL_TABLE_NAME); } #[tokio::test] @@ -487,7 +486,7 @@ async fn test_can_register_normal_table_after_restart() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::NormalTable) + .create_table(NORMAL_TABLE_NAME, TableType::NormalTable) .await; test_context.restart_server().await; @@ -495,7 +494,7 @@ async fn test_can_register_normal_table_after_restart() { let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); assert_eq!(retrieved_table_names.len(), 1); - assert_eq!(retrieved_table_names[0], TABLE_NAME); + assert_eq!(retrieved_table_names[0], NORMAL_TABLE_NAME); } #[tokio::test] @@ -503,13 +502,13 @@ async fn test_can_create_time_series_table() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTable) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable) .await; let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); assert_eq!(retrieved_table_names.len(), 1); - assert_eq!(retrieved_table_names[0], TABLE_NAME); + assert_eq!(retrieved_table_names[0], TIME_SERIES_TABLE_NAME); } #[tokio::test] @@ -517,7 +516,7 @@ async fn test_can_register_time_series_table_after_restart() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTable) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable) .await; test_context.restart_server().await; @@ -525,7 +524,7 @@ async fn test_can_register_time_series_table_after_restart() { let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); assert_eq!(retrieved_table_names.len(), 1); - assert_eq!(retrieved_table_names[0], TABLE_NAME); + assert_eq!(retrieved_table_names[0], TIME_SERIES_TABLE_NAME); } #[tokio::test] @@ -580,13 +579,13 @@ async fn test_can_create_register_and_list_multiple_normal_tables_and_time_serie async fn test_can_drop_normal_table() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::NormalTable) + .create_table(NORMAL_TABLE_NAME, TableType::NormalTable) .await; let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); - assert_eq!(retrieved_table_names[0], TABLE_NAME); + assert_eq!(retrieved_table_names[0], NORMAL_TABLE_NAME); - test_context.drop_table(TABLE_NAME).await.unwrap(); + test_context.drop_table(NORMAL_TABLE_NAME).await.unwrap(); let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); assert_eq!(retrieved_table_names.len(), 0); @@ -594,7 +593,7 @@ async fn test_can_drop_normal_table() { // It should be possible to create a normal table, drop it, and then create a new normal table // with the same name. test_context - .create_table(TABLE_NAME, TableType::NormalTable) + .create_table(NORMAL_TABLE_NAME, TableType::NormalTable) .await; } @@ -602,13 +601,16 @@ async fn test_can_drop_normal_table() { async fn test_can_drop_time_series_table() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTable) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable) .await; let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); - assert_eq!(retrieved_table_names[0], TABLE_NAME); + assert_eq!(retrieved_table_names[0], TIME_SERIES_TABLE_NAME); - test_context.drop_table(TABLE_NAME).await.unwrap(); + test_context + .drop_table(TIME_SERIES_TABLE_NAME) + .await + .unwrap(); let retrieved_table_names = test_context.retrieve_all_table_names().await.unwrap(); assert_eq!(retrieved_table_names.len(), 0); @@ -616,7 +618,7 @@ async fn test_can_drop_time_series_table() { // It should be possible to create a time series table, drop it, and then create a new time // series table with the same name. test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTable) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable) .await; } @@ -624,7 +626,7 @@ async fn test_can_drop_time_series_table() { async fn test_cannot_drop_missing_table() { let mut test_context = TestContext::new().await; - let result = test_context.drop_table(TABLE_NAME).await; + let result = test_context.drop_table(NORMAL_TABLE_NAME).await; assert!(result.is_err()); } @@ -636,14 +638,18 @@ async fn test_can_truncate_normal_table() { ingest_time_series_and_flush_data( &mut test_context, slice::from_ref(&time_series), + NORMAL_TABLE_NAME, TableType::NormalTable, ) .await; - test_context.truncate_table(TABLE_NAME).await.unwrap(); + test_context + .truncate_table(NORMAL_TABLE_NAME) + .await + .unwrap(); let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {NORMAL_TABLE_NAME}")) .await .unwrap(); @@ -659,14 +665,18 @@ async fn test_can_truncate_time_series_table() { ingest_time_series_and_flush_data( &mut test_context, slice::from_ref(&time_series), + TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable, ) .await; - test_context.truncate_table(TABLE_NAME).await.unwrap(); + test_context + .truncate_table(TIME_SERIES_TABLE_NAME) + .await + .unwrap(); let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {TIME_SERIES_TABLE_NAME}")) .await .unwrap(); @@ -678,7 +688,7 @@ async fn test_can_truncate_time_series_table() { async fn test_cannot_truncate_missing_table() { let mut test_context = TestContext::new().await; - let result = test_context.truncate_table(TABLE_NAME).await; + let result = test_context.truncate_table(NORMAL_TABLE_NAME).await; assert!(result.is_err()); } @@ -697,22 +707,26 @@ async fn test_can_vacuum_normal_table() { ingest_time_series_and_flush_data( &mut test_context, slice::from_ref(&time_series), + NORMAL_TABLE_NAME, TableType::NormalTable, ) .await; - test_context.truncate_table(TABLE_NAME).await.unwrap(); + test_context + .truncate_table(NORMAL_TABLE_NAME) + .await + .unwrap(); // The files should still exist on disk even though they are no longer active. let table_path = format!( "{}/tables/{}", test_context.temp_dir.path().to_str().unwrap(), - TABLE_NAME + NORMAL_TABLE_NAME ); let files = std::fs::read_dir(&table_path).unwrap(); assert_eq!(files.count(), 2); - test_context.vacuum_table(TABLE_NAME).await.unwrap(); + test_context.vacuum_table(NORMAL_TABLE_NAME).await.unwrap(); // Only the _delta_log folder should remain. let files = std::fs::read_dir(&table_path).unwrap(); @@ -734,22 +748,29 @@ async fn test_can_vacuum_time_series_table() { ingest_time_series_and_flush_data( &mut test_context, slice::from_ref(&time_series), + TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable, ) .await; - test_context.truncate_table(TABLE_NAME).await.unwrap(); + test_context + .truncate_table(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=1", test_context.temp_dir.path().to_str().unwrap(), - TABLE_NAME + TIME_SERIES_TABLE_NAME ); let files = std::fs::read_dir(&column_path).unwrap(); assert_eq!(files.count(), 1); - test_context.vacuum_table(TABLE_NAME).await.unwrap(); + test_context + .vacuum_table(TIME_SERIES_TABLE_NAME) + .await + .unwrap(); // No files should remain in the column folder. let files = std::fs::read_dir(&column_path).unwrap(); @@ -760,7 +781,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(TABLE_NAME).await; + let result = test_context.vacuum_table(NORMAL_TABLE_NAME).await; assert!(result.is_err()); } @@ -769,10 +790,10 @@ async fn test_can_get_schema() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTable) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable) .await; - let schema = test_context.retrieve_schema(TABLE_NAME).await; + let schema = test_context.retrieve_schema(TIME_SERIES_TABLE_NAME).await; assert_eq!( schema, @@ -828,12 +849,13 @@ async fn test_do_put_can_ingest_time_series_with_tags() { ingest_time_series_and_flush_data( &mut test_context, slice::from_ref(&time_series), + TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable, ) .await; let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {TIME_SERIES_TABLE_NAME}")) .await .unwrap(); @@ -844,12 +866,12 @@ async fn test_do_put_can_ingest_time_series_with_tags() { async fn test_insert_can_ingest_time_series_with_tags() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTable) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable) .await; let insert_result = test_context .execute_query(format!( - "INSERT INTO {TABLE_NAME} VALUES\ + "INSERT INTO {TIME_SERIES_TABLE_NAME} VALUES\ ('2020-01-01 13:00:00', 1, 2, 3, 4, 5, 'Aalborg'),\ ('2020-01-01 13:00:01', 1, 2, 3, 4, 5, 'Aalborg'),\ ('2020-01-01 13:00:02', 1, 2, 3, 4, 5, 'Aalborg'),\ @@ -862,7 +884,7 @@ async fn test_insert_can_ingest_time_series_with_tags() { test_context.flush_data_to_disk().await; let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {TIME_SERIES_TABLE_NAME}")) .await .unwrap(); @@ -879,12 +901,13 @@ async fn test_do_put_can_ingest_time_series_without_tags() { ingest_time_series_and_flush_data( &mut test_context, slice::from_ref(&time_series), + TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTableNoTag, ) .await; let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {TIME_SERIES_TABLE_NAME}")) .await .unwrap(); @@ -895,12 +918,12 @@ async fn test_do_put_can_ingest_time_series_without_tags() { async fn test_insert_can_ingest_time_series_without_tags() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTableNoTag) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTableNoTag) .await; let insert_result = test_context .execute_query(format!( - "INSERT INTO {TABLE_NAME} VALUES\ + "INSERT INTO {TIME_SERIES_TABLE_NAME} VALUES\ ('2020-01-01 13:00:00', 1, 2, 3, 4, 5),\ ('2020-01-01 13:00:01', 1, 2, 3, 4, 5),\ ('2020-01-01 13:00:02', 1, 2, 3, 4, 5),\ @@ -913,7 +936,7 @@ async fn test_insert_can_ingest_time_series_without_tags() { test_context.flush_data_to_disk().await; let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {TIME_SERIES_TABLE_NAME}")) .await .unwrap(); @@ -930,13 +953,16 @@ async fn test_do_put_can_ingest_time_series_with_generated_field() { ingest_time_series_and_flush_data( &mut test_context, slice::from_ref(&time_series), + TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTableAsField, ) .await; // The optimizer is allowed to add SortedJoinExec between SortedJoinExec and GeneratedAsExec. let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME} ORDER BY timestamp")) + .execute_query(format!( + "SELECT * FROM {TIME_SERIES_TABLE_NAME} ORDER BY timestamp" + )) .await .unwrap(); @@ -951,12 +977,12 @@ async fn test_do_put_can_ingest_time_series_with_generated_field() { async fn test_insert_can_ingest_time_series_with_generated_field() { let mut test_context = TestContext::new().await; test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTableAsField) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTableAsField) .await; let insert_result = test_context .execute_query(format!( - "INSERT INTO {TABLE_NAME} VALUES\ + "INSERT INTO {TIME_SERIES_TABLE_NAME} VALUES\ ('2020-01-01 13:00:00', 1, 2, 3, 4),\ ('2020-01-01 13:00:01', 1, 2, 3, 4),\ ('2020-01-01 13:00:02', 1, 2, 3, 4),\ @@ -969,7 +995,7 @@ async fn test_insert_can_ingest_time_series_with_generated_field() { test_context.flush_data_to_disk().await; let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {TIME_SERIES_TABLE_NAME}")) .await .unwrap(); @@ -988,12 +1014,17 @@ async fn test_do_put_can_ingest_multiple_time_series_with_different_tags() { TestContext::generate_time_series_with_tag(false, None, Some("tag_two")); let time_series = &[time_series_with_tag_one, time_series_with_tag_two]; - ingest_time_series_and_flush_data(&mut test_context, time_series, TableType::TimeSeriesTable) - .await; + ingest_time_series_and_flush_data( + &mut test_context, + time_series, + TIME_SERIES_TABLE_NAME, + TableType::TimeSeriesTable, + ) + .await; let query_result = test_context .execute_query(format!( - "SELECT * FROM {TABLE_NAME} ORDER BY tag, timestamp" + "SELECT * FROM {TIME_SERIES_TABLE_NAME} ORDER BY tag, timestamp" )) .await .unwrap(); @@ -1012,11 +1043,13 @@ async fn test_do_put_can_ingest_multiple_time_series_with_different_tags() { async fn test_cannot_ingest_invalid_time_series() { let mut test_context = TestContext::new().await; let time_series = TestContext::generate_time_series_with_tag(false, None, None); - let flight_data = - TestContext::create_flight_data_from_time_series(TABLE_NAME.to_owned(), &[time_series]); + let flight_data = TestContext::create_flight_data_from_time_series( + TIME_SERIES_TABLE_NAME.to_owned(), + &[time_series], + ); test_context - .create_table(TABLE_NAME, TableType::TimeSeriesTable) + .create_table(TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable) .await; assert!( @@ -1029,7 +1062,7 @@ async fn test_cannot_ingest_invalid_time_series() { test_context.flush_data_to_disk().await; let query_result = test_context - .execute_query(format!("SELECT * FROM {TABLE_NAME}")) + .execute_query(format!("SELECT * FROM {TIME_SERIES_TABLE_NAME}")) .await .unwrap(); assert_eq!(query_result.num_rows(), 0); @@ -1056,6 +1089,7 @@ async fn execute_and_assert_include_select(address_count: usize) { ingest_time_series_and_flush_data( &mut test_context, &[time_series], + TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable, ) .await; @@ -1066,7 +1100,9 @@ async fn execute_and_assert_include_select(address_count: usize) { let address = addresses_separate.join(", "); let query_result = test_context - .execute_query(format!("INCLUDE {address} SELECT * FROM {TABLE_NAME}")) + .execute_query(format!( + "INCLUDE {address} SELECT * FROM {TIME_SERIES_TABLE_NAME}" + )) .await .unwrap(); @@ -1075,32 +1111,47 @@ async fn execute_and_assert_include_select(address_count: usize) { #[tokio::test] async fn test_count_from_segments_equals_count_from_data_points() { - assert_ne_query_plans_and_eq_result(format!("SELECT COUNT(field_one) FROM {TABLE_NAME}"), 0.0) - .await; + assert_ne_query_plans_and_eq_result( + format!("SELECT COUNT(field_one) FROM {TIME_SERIES_TABLE_NAME}"), + 0.0, + ) + .await; } #[tokio::test] async fn test_min_from_segments_equals_min_from_data_points() { - assert_ne_query_plans_and_eq_result(format!("SELECT MIN(field_one) FROM {TABLE_NAME}"), 0.0) - .await; + assert_ne_query_plans_and_eq_result( + format!("SELECT MIN(field_one) FROM {TIME_SERIES_TABLE_NAME}"), + 0.0, + ) + .await; } #[tokio::test] async fn test_max_from_segments_equals_max_from_data_points() { - assert_ne_query_plans_and_eq_result(format!("SELECT MAX(field_one) FROM {TABLE_NAME}"), 0.0) - .await; + assert_ne_query_plans_and_eq_result( + format!("SELECT MAX(field_one) FROM {TIME_SERIES_TABLE_NAME}"), + 0.0, + ) + .await; } #[tokio::test] async fn test_sum_from_segments_equals_sum_from_data_points() { - assert_ne_query_plans_and_eq_result(format!("SELECT SUM(field_one) FROM {TABLE_NAME}"), 0.001) - .await; + assert_ne_query_plans_and_eq_result( + format!("SELECT SUM(field_one) FROM {TIME_SERIES_TABLE_NAME}"), + 0.001, + ) + .await; } #[tokio::test] async fn test_avg_from_segments_equals_avg_from_data_points() { - assert_ne_query_plans_and_eq_result(format!("SELECT AVG(field_one) FROM {TABLE_NAME}"), 0.001) - .await; + assert_ne_query_plans_and_eq_result( + format!("SELECT AVG(field_one) FROM {TIME_SERIES_TABLE_NAME}"), + 0.001, + ) + .await; } /// Asserts that the query executed on segments in `segment_query` returns a result within @@ -1121,6 +1172,7 @@ async fn assert_ne_query_plans_and_eq_result(segment_query: String, error_bound: ingest_time_series_and_flush_data( &mut test_context, &[time_series], + TIME_SERIES_TABLE_NAME, TableType::TimeSeriesTable, ) .await; @@ -1181,12 +1233,13 @@ async fn assert_ne_query_plans_and_eq_result(segment_query: String, error_bound: async fn ingest_time_series_and_flush_data( test_context: &mut TestContext, time_series: &[RecordBatch], + table_name: &str, table_type: TableType, ) { let flight_data = - TestContext::create_flight_data_from_time_series(TABLE_NAME.to_owned(), time_series); + TestContext::create_flight_data_from_time_series(table_name.to_owned(), time_series); - test_context.create_table(TABLE_NAME, table_type).await; + test_context.create_table(table_name, table_type).await; test_context .send_time_series_to_server(flight_data) From 7be5938f3f9535af4564f07d3cf7b70b054ec1c0 Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Wed, 20 Aug 2025 19:36:47 +0200 Subject: [PATCH 30/31] Update based on comments from @skejserjensen --- crates/modelardb_storage/src/parser.rs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/crates/modelardb_storage/src/parser.rs b/crates/modelardb_storage/src/parser.rs index db8352f95..b0727c790 100644 --- a/crates/modelardb_storage/src/parser.rs +++ b/crates/modelardb_storage/src/parser.rs @@ -172,7 +172,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 +/// SERIES TABLE table_name DDL statements, INCLUDE 'address'\[, 'address'\]+ DQL statements, and /// VACUUM \[table_name\[, table_name\]+\] statements. #[derive(Debug)] struct ModelarDbDialect { @@ -556,7 +556,7 @@ impl Dialect for ModelarDbDialect { /// Check if the next tokens are CREATE TIME SERIES TABLE, if so, attempt to parse the token stream /// 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. + /// 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 @@ -1691,4 +1691,19 @@ mod tests { _ => panic!("Expected ModelarDbStatement::Vacuum."), } } + + #[test] + fn test_tokenize_and_parse_vacuum_trailing_comma() { + assert!(tokenize_and_parse_sql_statement("VACUUM table_name,").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_leading_comma() { + assert!(tokenize_and_parse_sql_statement("VACUUM ,table_name").is_err()); + } + + #[test] + fn test_tokenize_and_parse_vacuum_quoted_table_name() { + assert!(tokenize_and_parse_sql_statement("VACUUM 'table_name'").is_err()); + } } From e95eedabcf085a1c573baf1275e7092a61861c0c Mon Sep 17 00:00:00 2001 From: CGodiksen <36046286+CGodiksen@users.noreply.github.com> Date: Thu, 21 Aug 2025 10:35:29 +0200 Subject: [PATCH 31/31] Update based on comments from @chrthomsen --- crates/modelardb_storage/src/parser.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/modelardb_storage/src/parser.rs b/crates/modelardb_storage/src/parser.rs index b0727c790..c5da7a146 100644 --- a/crates/modelardb_storage/src/parser.rs +++ b/crates/modelardb_storage/src/parser.rs @@ -50,7 +50,7 @@ use sqlparser::tokenizer::{Span, Token}; use crate::error::{ModelarDbStorageError, Result}; -/// A top-level statement (CREATE, INSERT, SELECT, TRUNCATE, DROP, VACUUM etc.) that have been +/// A top-level statement (CREATE, INSERT, SELECT, TRUNCATE, DROP, VACUUM etc.) that has been /// tokenized, parsed, and for which semantic checks have verified that it is compatible with /// ModelarDB. #[derive(Debug)] @@ -71,8 +71,8 @@ pub enum ModelarDbStatement { Vacuum(Vec), } -/// Tokenizes and parses the SQL statement in `sql` and return its parsed representation in the form -/// of a [`ModelarDbStatement`]. Returns a [`ModelarDbStorageError`] if `sql` is empty, contain +/// Tokenizes and parses the SQL statement in `sql` and returns its parsed representation in the form +/// of a [`ModelarDbStatement`]. Returns a [`ModelarDbStorageError`] if `sql` is empty, contains /// multiple statements, or the statement is unsupported. Currently, CREATE TABLE, CREATE TIME SERIES /// TABLE, INSERT, EXPLAIN, INCLUDE, SELECT, TRUNCATE TABLE, DROP TABLE, and VACUUM are supported. pub fn tokenize_and_parse_sql_statement(sql_statement: &str) -> Result { @@ -500,7 +500,7 @@ 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 - /// typed incorrectly or the table names cannot be extracted. + /// not the first word or the table names cannot be extracted. fn parse_vacuum(&self, parser: &mut Parser) -> StdResult { // VACUUM. parser.expect_keyword(Keyword::VACUUM)?;