diff --git a/crates/modelardb_embedded/bindings/c/modelardb_embedded.h b/crates/modelardb_embedded/bindings/c/modelardb_embedded.h index 0fe32dcd..34a2e728 100644 --- a/crates/modelardb_embedded/bindings/c/modelardb_embedded.h +++ b/crates/modelardb_embedded/bindings/c/modelardb_embedded.h @@ -199,7 +199,7 @@ int modelardb_embedded_vacuum(void* maybe_operations_ptr, const char* table_name_ptr, const uint64_t* retention_period_in_seconds_ptr); -// Optimize the table by compacting its many small files into fewer larger files. +// Optimize the table by merging its many small files into fewer larger files. int modelardb_embedded_optimize(void* maybe_operations_ptr, bool is_data_folder, const char* table_name_ptr, diff --git a/crates/modelardb_embedded/bindings/python/modelardb/operations.py b/crates/modelardb_embedded/bindings/python/modelardb/operations.py index c975767a..e172f5df 100644 --- a/crates/modelardb_embedded/bindings/python/modelardb/operations.py +++ b/crates/modelardb_embedded/bindings/python/modelardb/operations.py @@ -685,7 +685,7 @@ def optimize(self, table_name: str, target_size_in_bytes: None | int = None): :param table_name: The name of the table to optimize. :type table_name: str - :param target_size_in_bytes: The target file size in bytes. Many small files are compacted + :param target_size_in_bytes: The target file size in bytes. Many small files are merged into fewer larger files of approximately this size. If `None`, the default target size of 64 MiB is used. :type target_size_in_bytes: int, optional diff --git a/crates/modelardb_embedded/bindings/python/tests/test_operations.py b/crates/modelardb_embedded/bindings/python/tests/test_operations.py index 24fea831..0d8fe3ca 100644 --- a/crates/modelardb_embedded/bindings/python/tests/test_operations.py +++ b/crates/modelardb_embedded/bindings/python/tests/test_operations.py @@ -512,10 +512,10 @@ def test_data_folder_optimize(self): data_folder.optimize(TIME_SERIES_TABLE_NAME) - # Vacuum to remove the compacted files. + # Vacuum to remove the stale files left by the merge. data_folder.vacuum(TIME_SERIES_TABLE_NAME, retention_period_in_seconds=0) - # The small files should be compacted into a single active file. + # The small files should be merged into a single active file. file_count = len(os.listdir(folder_path)) self.assertEqual(file_count, 1) diff --git a/crates/modelardb_embedded/src/capi.rs b/crates/modelardb_embedded/src/capi.rs index 5447abf6..2f7e1e7f 100644 --- a/crates/modelardb_embedded/src/capi.rs +++ b/crates/modelardb_embedded/src/capi.rs @@ -1007,7 +1007,7 @@ unsafe fn vacuum( } /// Optimizes the table with the name in `table_name_ptr` in the [`DataFolder`] or [`Client`] in -/// `maybe_operations_ptr` by compacting its many small files into fewer larger files of +/// `maybe_operations_ptr` by merging its many small files into fewer larger files of /// approximately `target_size_in_bytes_ptr` bytes. Assumes `maybe_operations_ptr` points to a /// [`DataFolder`] or [`Client`]; `table_name_ptr` points to a valid C string; and /// `target_size_in_bytes_ptr` points to a valid `u64`, or is null to use the default target size. diff --git a/crates/modelardb_embedded/src/operations/client.rs b/crates/modelardb_embedded/src/operations/client.rs index b71d93b4..4456d067 100644 --- a/crates/modelardb_embedded/src/operations/client.rs +++ b/crates/modelardb_embedded/src/operations/client.rs @@ -316,7 +316,7 @@ impl Operations for Client { Ok(()) } - /// Optimize the table with the name in `table_name` by compacting its many small files into + /// Optimize the table with the name in `table_name` by merging its many small files into /// fewer larger files of approximately `maybe_target_size_in_bytes` bytes. If a target size is /// not given, the default target size of 64 MiB is used. If the table does not exist, the table /// could not be optimized, or the target size is zero, [`ModelarDbEmbeddedError`] is returned. diff --git a/crates/modelardb_embedded/src/operations/data_folder.rs b/crates/modelardb_embedded/src/operations/data_folder.rs index bb293dbf..57fe7938 100644 --- a/crates/modelardb_embedded/src/operations/data_folder.rs +++ b/crates/modelardb_embedded/src/operations/data_folder.rs @@ -540,7 +540,7 @@ impl Operations for DataFolder { } } - /// Optimize the table with the name in `table_name` by compacting its many small files into + /// Optimize the table with the name in `table_name` by merging its many small files into /// fewer larger files of approximately `maybe_target_size_in_bytes` bytes. If a target size is /// not given, the default target size of 64 MiB is used. If the table does not exist, the table /// could not be optimized, or the target size is zero, [`ModelarDbEmbeddedError`] is returned. @@ -2314,7 +2314,7 @@ mod tests { data_folder.optimize(NORMAL_TABLE_NAME, None).await.unwrap(); - // The small files should be compacted into a single active file. + // The small files should be merged into a single active file. let delta_table = data_folder.delta_table(NORMAL_TABLE_NAME).await.unwrap(); assert_eq!(delta_table.get_file_uris().unwrap().count(), 1); } @@ -2343,7 +2343,7 @@ mod tests { .await .unwrap(); - // The files in each of the two partitions should be compacted into a single active file. + // The files in each of the two partitions should be merged into a single active file. let delta_table = data_folder .delta_table(TIME_SERIES_TABLE_NAME) .await diff --git a/crates/modelardb_embedded/src/operations/mod.rs b/crates/modelardb_embedded/src/operations/mod.rs index 2ff33b49..24a5793b 100644 --- a/crates/modelardb_embedded/src/operations/mod.rs +++ b/crates/modelardb_embedded/src/operations/mod.rs @@ -146,7 +146,7 @@ pub trait Operations: Sync + Send { maybe_retention_period_in_seconds: Option, ) -> Result<()>; - /// Optimize the table with the name in `table_name` by compacting its many small files into + /// Optimize the table with the name in `table_name` by merging its many small files into /// fewer larger files of approximately `maybe_target_size_in_bytes` bytes. If a target size is /// not given, the default target size of 64 MiB is used. async fn optimize( diff --git a/crates/modelardb_server/src/configuration.rs b/crates/modelardb_server/src/configuration.rs index 3bdf4a14..39b1ae59 100644 --- a/crates/modelardb_server/src/configuration.rs +++ b/crates/modelardb_server/src/configuration.rs @@ -27,6 +27,7 @@ use std::sync::Arc; use modelardb_storage::data_folder::DataFolder; use modelardb_storage::write_ahead_log::WriteAheadLog; use modelardb_types::flight::protocol; +use modelardb_types::types::MAX_RETENTION_PERIOD_IN_SECONDS; use object_store::path::Path; use object_store::{Error, ObjectStoreExt, PutPayload}; use prost::Message; @@ -63,6 +64,14 @@ struct Configuration { /// The approximate maximum size, in bytes, of a single WAL segment file before it is closed and /// a new one is started. segment_size_threshold_in_bytes: u64, + /// Target size, in bytes, of the files produced when automatically compacting a table's + /// storage. This is also the default value used when an OPTIMIZE query is executed without an + /// explicit target size. + optimize_target_file_size_in_bytes: u64, + /// Retention period, in seconds, used when automatically vacuuming a table during compaction. + /// This is also the default value used when a VACUUM query is executed without an explicit + /// retention period. + vacuum_retention_period_in_seconds: u64, /// Number of threads to allocate for converting multivariate time series to univariate /// time series. ingestion_threads: u8, @@ -99,6 +108,14 @@ impl Configuration { self.segment_size_threshold_in_bytes = value; } + if let Some(value) = args.optimize_target_file_size_in_bytes { + self.optimize_target_file_size_in_bytes = value; + } + + if let Some(value) = args.vacuum_retention_period_in_seconds { + self.vacuum_retention_period_in_seconds = value; + } + if let Some(value) = args.wal_enabled { self.wal_enabled = value; } @@ -114,10 +131,22 @@ impl Configuration { if self.ingestion_threads != 1 || self.compression_threads != 1 || self.writer_threads != 1 { return Err(ModelarDbServerError::InvalidState( - "Only one thread per component is currently supported.".to_string(), + "Only one thread per component is currently supported.".to_owned(), )); }; + if self.optimize_target_file_size_in_bytes == 0 { + return Err(ModelarDbServerError::InvalidState( + "Optimize target file size must be greater than zero.".to_owned(), + )); + } + + if self.vacuum_retention_period_in_seconds > MAX_RETENTION_PERIOD_IN_SECONDS { + return Err(ModelarDbServerError::InvalidState(format!( + "Vacuum retention period cannot be more than {MAX_RETENTION_PERIOD_IN_SECONDS} seconds." + ))); + } + Ok(()) } @@ -147,6 +176,8 @@ impl Default for Configuration { compressed_reserved_memory_in_bytes: 512 * 1024 * 1024, transfer_batch_size_in_bytes: Some(64 * 1024 * 1024), segment_size_threshold_in_bytes: 64 * 1024 * 1024, + optimize_target_file_size_in_bytes: 64 * 1024 * 1024, + vacuum_retention_period_in_seconds: 60 * 60 * 24 * 7, ingestion_threads: 1, compression_threads: 1, writer_threads: 1, @@ -389,6 +420,71 @@ impl ConfigurationManager { .await } + pub(crate) fn optimize_target_file_size_in_bytes(&self) -> u64 { + self.configuration.optimize_target_file_size_in_bytes + } + + /// Set the target file size used by the data storage compactor and as the default for OPTIMIZE + /// queries without an explicit target size. If the new value is zero or the new configuration + /// could not be saved to the configuration file, return [`ModelarDbServerError`]. + pub(crate) async fn set_optimize_target_file_size_in_bytes( + &mut self, + new_optimize_target_file_size_in_bytes: u64, + storage_engine: Arc>, + ) -> Result<()> { + if new_optimize_target_file_size_in_bytes == 0 { + return Err(ModelarDbServerError::InvalidArgument( + "Optimize target file size must be greater than zero.".to_owned(), + )); + } + + storage_engine + .write() + .await + .set_optimize_target_file_size_in_bytes(new_optimize_target_file_size_in_bytes) + .await; + + self.configuration.optimize_target_file_size_in_bytes = + new_optimize_target_file_size_in_bytes; + + self.configuration + .save_to_toml(&self.local_data_folder) + .await + } + + pub(crate) fn vacuum_retention_period_in_seconds(&self) -> u64 { + self.configuration.vacuum_retention_period_in_seconds + } + + /// Set the retention period used by the data storage compactor and as the default for VACUUM + /// queries without an explicit retention period. If the new value is larger than + /// [`MAX_RETENTION_PERIOD_IN_SECONDS`] or the new configuration could not be saved to the + /// configuration file, return [`ModelarDbServerError`]. + pub(crate) async fn set_vacuum_retention_period_in_seconds( + &mut self, + new_vacuum_retention_period_in_seconds: u64, + storage_engine: Arc>, + ) -> Result<()> { + if new_vacuum_retention_period_in_seconds > MAX_RETENTION_PERIOD_IN_SECONDS { + return Err(ModelarDbServerError::InvalidArgument(format!( + "Vacuum retention period cannot be more than {MAX_RETENTION_PERIOD_IN_SECONDS} seconds." + ))); + } + + storage_engine + .write() + .await + .set_vacuum_retention_period_in_seconds(new_vacuum_retention_period_in_seconds) + .await; + + self.configuration.vacuum_retention_period_in_seconds = + new_vacuum_retention_period_in_seconds; + + self.configuration + .save_to_toml(&self.local_data_folder) + .await + } + pub(crate) fn ingestion_threads(&self) -> u8 { self.configuration.ingestion_threads } @@ -418,6 +514,12 @@ impl ConfigurationManager { compression_threads: self.configuration.compression_threads as u32, writer_threads: self.configuration.writer_threads as u32, wal_enabled: self.configuration.wal_enabled, + optimize_target_file_size_in_bytes: self + .configuration + .optimize_target_file_size_in_bytes, + vacuum_retention_period_in_seconds: self + .configuration + .vacuum_retention_period_in_seconds, }; configuration.encode_to_vec() @@ -464,6 +566,8 @@ mod tests { compressed_reserved_memory_in_bytes: 1, transfer_batch_size_in_bytes: Some(1), segment_size_threshold_in_bytes: 1, + optimize_target_file_size_in_bytes: 1, + vacuum_retention_period_in_seconds: 1, ..Configuration::default() }; @@ -537,6 +641,39 @@ mod tests { ); } + #[test] + fn test_validate_rejects_zero_optimize_target_file_size() { + let configuration = Configuration { + optimize_target_file_size_in_bytes: 0, + ..Configuration::default() + }; + + let result = configuration.validate(); + + assert_eq!( + result.unwrap_err().to_string(), + "Invalid State Error: Optimize target file size must be greater than zero." + ); + } + + #[test] + fn test_validate_rejects_too_large_vacuum_retention_period() { + let configuration = Configuration { + vacuum_retention_period_in_seconds: MAX_RETENTION_PERIOD_IN_SECONDS + 1, + ..Configuration::default() + }; + + let result = configuration.validate(); + + assert_eq!( + result.unwrap_err().to_string(), + format!( + "Invalid State Error: Vacuum retention period cannot be more than {} seconds.", + MAX_RETENTION_PERIOD_IN_SECONDS + ) + ); + } + #[tokio::test] async fn test_set_ingested_reserved_memory_in_bytes() { let temp_dir = tempfile::tempdir().unwrap(); @@ -735,6 +872,118 @@ mod tests { ); } + #[tokio::test] + async fn test_set_optimize_target_file_size_in_bytes() { + let temp_dir = tempfile::tempdir().unwrap(); + let (storage_engine, configuration_manager) = create_components(&temp_dir).await; + + assert_eq!( + configuration_manager + .read() + .await + .optimize_target_file_size_in_bytes(), + 64 * 1024 * 1024 + ); + + let new_value = 1024; + configuration_manager + .write() + .await + .set_optimize_target_file_size_in_bytes(new_value, storage_engine) + .await + .unwrap(); + + assert_eq!( + configuration_manager + .read() + .await + .optimize_target_file_size_in_bytes(), + new_value + ); + + let configuration_from_file = configuration_from_file(&temp_dir).await; + assert_eq!( + configuration_from_file.optimize_target_file_size_in_bytes, + new_value + ); + } + + #[tokio::test] + async fn test_set_optimize_target_file_size_in_bytes_rejects_zero() { + let temp_dir = tempfile::tempdir().unwrap(); + let (storage_engine, configuration_manager) = create_components(&temp_dir).await; + + let result = configuration_manager + .write() + .await + .set_optimize_target_file_size_in_bytes(0, storage_engine) + .await; + + assert_eq!( + result.unwrap_err().to_string(), + "Invalid Argument Error: Optimize target file size must be greater than zero." + ); + } + + #[tokio::test] + async fn test_set_vacuum_retention_period_in_seconds() { + let temp_dir = tempfile::tempdir().unwrap(); + let (storage_engine, configuration_manager) = create_components(&temp_dir).await; + + assert_eq!( + configuration_manager + .read() + .await + .vacuum_retention_period_in_seconds(), + 60 * 60 * 24 * 7 + ); + + let new_value = 60; + configuration_manager + .write() + .await + .set_vacuum_retention_period_in_seconds(new_value, storage_engine) + .await + .unwrap(); + + assert_eq!( + configuration_manager + .read() + .await + .vacuum_retention_period_in_seconds(), + new_value + ); + + let configuration_from_file = configuration_from_file(&temp_dir).await; + assert_eq!( + configuration_from_file.vacuum_retention_period_in_seconds, + new_value + ); + } + + #[tokio::test] + async fn test_set_vacuum_retention_period_in_seconds_rejects_too_large_value() { + let temp_dir = tempfile::tempdir().unwrap(); + let (storage_engine, configuration_manager) = create_components(&temp_dir).await; + + let result = configuration_manager + .write() + .await + .set_vacuum_retention_period_in_seconds( + MAX_RETENTION_PERIOD_IN_SECONDS + 1, + storage_engine, + ) + .await; + + assert_eq!( + result.unwrap_err().to_string(), + format!( + "Invalid Argument Error: Vacuum retention period cannot be more than {} seconds.", + MAX_RETENTION_PERIOD_IN_SECONDS + ) + ); + } + /// Return the configuration from the configuration file at the root of `temp_dir`. async fn configuration_from_file(temp_dir: &TempDir) -> Configuration { let configuration_file_path = temp_dir.path().join(CONFIGURATION_FILE_NAME); diff --git a/crates/modelardb_server/src/context.rs b/crates/modelardb_server/src/context.rs index c94985b5..9d600959 100644 --- a/crates/modelardb_server/src/context.rs +++ b/crates/modelardb_server/src/context.rs @@ -401,7 +401,7 @@ impl Context { Ok(()) } - /// Optimize the table with `table_name` if it exists by compacting its small files into larger + /// Optimize the table with `table_name` if it exists by merging its small files into larger /// files of approximately `maybe_target_size_in_bytes` bytes. If a target size is not given, a /// default target size of 64 MiB is used. If the target size is zero, the table does not exist, /// or it could not be optimized, [`ModelarDbServerError`] is returned. @@ -922,7 +922,7 @@ mod tests { .await .unwrap(); - // The small files should be compacted into a single active file. + // The small files should be merged into a single active file. assert_eq!(active_file_count(&context, NORMAL_TABLE_NAME).await, 1); } @@ -970,7 +970,7 @@ mod tests { .await .unwrap(); - // The small files should be compacted into a single active file. + // The small files should be merged into a single active file. assert_eq!(active_file_count(&context, TIME_SERIES_TABLE_NAME).await, 1); } diff --git a/crates/modelardb_server/src/main.rs b/crates/modelardb_server/src/main.rs index c873db51..673e2847 100644 --- a/crates/modelardb_server/src/main.rs +++ b/crates/modelardb_server/src/main.rs @@ -84,6 +84,19 @@ pub(crate) struct ServerArgs { #[arg(long, env = "MODELARDBD_SEGMENT_SIZE_THRESHOLD_IN_BYTES")] segment_size_threshold_in_bytes: Option, + /// Target size, in bytes, of the files produced when automatically compacting a table's + /// storage. This is also the default value used when an OPTIMIZE query is executed without an + /// explicit target size. + #[arg(long, env = "MODELARDBD_OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES")] + optimize_target_file_size_in_bytes: Option, + + /// Retention period, in seconds, used when automatically vacuuming a table during compaction. + /// This is also the default value used when a VACUUM query is executed without an explicit + /// retention period. Note that a very low value can delete files an in-progress query is still + /// scanning. + #[arg(long, env = "MODELARDBD_VACUUM_RETENTION_PERIOD_IN_SECONDS")] + vacuum_retention_period_in_seconds: Option, + /// Whether the write-ahead log is enabled. #[arg(long, env = "MODELARDBD_WAL_ENABLED")] wal_enabled: Option, diff --git a/crates/modelardb_server/src/remote/mod.rs b/crates/modelardb_server/src/remote/mod.rs index 720f8521..8f245963 100644 --- a/crates/modelardb_server/src/remote/mod.rs +++ b/crates/modelardb_server/src/remote/mod.rs @@ -471,9 +471,15 @@ impl FlightServiceHandler { } } + // If the retention period is not specified, use the local configuration retention period. + // The local retention period is not passed to the peer nodes above, so they can use their + // own local retention period. + let retention_period_in_seconds = maybe_retention_period_in_seconds + .unwrap_or(configuration_manager.vacuum_retention_period_in_seconds()); + for table_name in table_names { self.context - .vacuum_table(table_name, maybe_retention_period_in_seconds) + .vacuum_table(table_name, Some(retention_period_in_seconds)) .await .map_err(error_to_status_invalid_argument)?; } @@ -503,9 +509,15 @@ impl FlightServiceHandler { } } + // If the target size is not specified, use the local configuration target size. The local + // target size is not passed to the peer nodes above, so they can use their own local target + // size. + let target_size_in_bytes = maybe_target_size_in_bytes + .unwrap_or(configuration_manager.optimize_target_file_size_in_bytes()); + for table_name in table_names { self.context - .optimize_table(table_name, maybe_target_size_in_bytes) + .optimize_table(table_name, Some(target_size_in_bytes)) .await .map_err(error_to_status_invalid_argument)?; } @@ -1021,6 +1033,22 @@ impl FlightService for FlightServiceHandler { .await .map_err(error_to_status_internal) } + Ok(protocol::update_configuration::Setting::OptimizeTargetFileSizeInBytes) => { + let new_value = maybe_new_value.ok_or(invalid_null_error)?; + + configuration_manager + .set_optimize_target_file_size_in_bytes(new_value, storage_engine) + .await + .map_err(error_to_status_internal) + } + Ok(protocol::update_configuration::Setting::VacuumRetentionPeriodInSeconds) => { + let new_value = maybe_new_value.ok_or(invalid_null_error)?; + + configuration_manager + .set_vacuum_retention_period_in_seconds(new_value, storage_engine) + .await + .map_err(error_to_status_internal) + } _ => Err(Status::unimplemented(format!( "{setting} is not an updatable setting in the server configuration." ))), diff --git a/crates/modelardb_server/src/storage/compressed_data_manager.rs b/crates/modelardb_server/src/storage/compressed_data_manager.rs index 0c3e88c0..94cb11e4 100644 --- a/crates/modelardb_server/src/storage/compressed_data_manager.rs +++ b/crates/modelardb_server/src/storage/compressed_data_manager.rs @@ -29,6 +29,7 @@ use tracing::{debug, error, info}; use crate::configuration::WalMode; use crate::error::Result; use crate::storage::compressed_data_buffer::{CompressedDataBuffer, CompressedSegmentBatch}; +use crate::storage::data_storage_compactor::DataStorageCompactor; use crate::storage::data_transfer::DataTransfer; use crate::storage::types::Message; use crate::storage::types::{Channels, MemoryPool}; @@ -36,6 +37,9 @@ use crate::storage::types::{Channels, MemoryPool}; /// Stores data points compressed as segments containing metadata and models in memory to batch the /// compressed segments before saving them to Apache Parquet files. pub(super) struct CompressedDataManager { + /// Component that compacts a table by merging the small compressed files that accumulate for it + /// into fewer larger files and vacuuming the files left behind. + pub(super) data_storage_compactor: Arc>, /// Component that transfers saved compressed data to the remote data folder when it is necessary. pub(super) data_transfer: Arc>>, /// Folder containing all compressed data managed by the [`StorageEngine`](crate::storage::StorageEngine). @@ -57,6 +61,7 @@ pub(super) struct CompressedDataManager { impl CompressedDataManager { pub(super) fn new( + data_storage_compactor: Arc>, data_transfer: Arc>>, local_data_folder: DataFolder, channels: Arc, @@ -64,6 +69,7 @@ impl CompressedDataManager { wal_mode: WalMode, ) -> Self { Self { + data_storage_compactor, data_transfer, local_data_folder, compressed_data_buffers: DashMap::new(), @@ -287,6 +293,14 @@ impl CompressedDataManager { self.memory_pool.remaining_compressed_memory_in_bytes() ); + // Compact the compressed data for table_name on disk once enough new data has been written + // since the last compaction. + self.data_storage_compactor + .read() + .await + .increase_estimated_compactable_size(table_name, compressed_data_buffer_size_in_bytes) + .await?; + Ok(()) } @@ -580,9 +594,18 @@ mod tests { .unwrap(), )); + let compactor = DataStorageCompactor::try_new( + local_data_folder.clone(), + 64 * 1024 * 1024, + 60 * 60 * 24 * 7, + ) + .await + .unwrap(); + ( temp_dir, CompressedDataManager::new( + Arc::new(RwLock::new(compactor)), Arc::new(RwLock::new(None)), local_data_folder, channels, diff --git a/crates/modelardb_server/src/storage/data_storage_compactor.rs b/crates/modelardb_server/src/storage/data_storage_compactor.rs new file mode 100644 index 00000000..4b502351 --- /dev/null +++ b/crates/modelardb_server/src/storage/data_storage_compactor.rs @@ -0,0 +1,345 @@ +/* Copyright 2026 The ModelarDB Contributors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +//! Support for automatically compacting the compressed data stored on disk. As compressed data is +//! saved, many small Apache Parquet files accumulate for each table. This component compacts a +//! table by merging those small files into fewer larger ones and vacuuming the files left behind, +//! reducing storage use and query time. + +use dashmap::DashMap; +use modelardb_storage::data_folder::DataFolder; +use tracing::debug; + +use crate::error::Result; + +/// Compacts each table by merging the many small Apache Parquet files that accumulate for it into +/// fewer larger files and vacuuming the files left behind. The component accumulates an estimate of +/// how much compactable data each table has and, once the estimate reaches the target file size, +/// compacts that table. The component only operates on the local data folder. +pub(super) struct DataStorageCompactor { + /// The data folder containing all compressed data managed by the + /// [`StorageEngine`](crate::storage::StorageEngine). + local_data_folder: DataFolder, + /// The target size, in bytes, of the files produced when a table is optimized. A table is + /// compacted once its `estimated_compactable_size_in_bytes` reaches this size, so the same + /// value decides both when to compact and how large the optimized files are. + optimize_target_file_size_in_bytes: u64, + /// The retention period, in seconds, used when a table is vacuumed as part of compaction. + /// Note that a very low value can let the vacuum physically delete files that an in-progress + /// query is still scanning, causing that query to fail. + vacuum_retention_period_in_seconds: u64, + /// Map from table names to an estimate of how many bytes each table has in files smaller than + /// the target size. The estimate over-approximates the on-disk size since it is increased by + /// the in-memory size of the compressed data, so it ignores the compression applied by Apache + /// Parquet. Note that it is not reduced when a table is truncated by the data transfer + /// component for simplicity. + estimated_compactable_size_in_bytes: DashMap, +} + +impl DataStorageCompactor { + /// Create a new [`DataStorageCompactor`] that compacts the tables in `local_data_folder`, + /// producing files of approximately `optimize_target_file_size_in_bytes` bytes and vacuuming + /// with a retention period of `vacuum_retention_period_in_seconds` seconds. The estimate for + /// each table is initialized with the combined size of its files smaller than the target size, + /// so small files written before a restart are not forgotten. If the files in `local_data_folder` + /// could not be read, return [`ModelarDbServerError`](crate::error::ModelarDbServerError). + pub(super) async fn try_new( + local_data_folder: DataFolder, + optimize_target_file_size_in_bytes: u64, + vacuum_retention_period_in_seconds: u64, + ) -> Result { + let table_names = local_data_folder.table_names().await?; + + let estimated_compactable_size_in_bytes = DashMap::with_capacity(table_names.len()); + for table_name in table_names { + let compactable_size_in_bytes: u64 = local_data_folder + .table_file_sizes(&table_name) + .await? + .into_iter() + .filter(|size_in_bytes| *size_in_bytes < optimize_target_file_size_in_bytes) + .sum(); + + estimated_compactable_size_in_bytes.insert(table_name, compactable_size_in_bytes); + } + + Ok(Self { + local_data_folder, + optimize_target_file_size_in_bytes, + vacuum_retention_period_in_seconds, + estimated_compactable_size_in_bytes, + }) + } + + /// Increase the estimated compactable size of the table with `table_name` by `size_in_bytes`. + /// If the estimate has reached `optimize_target_file_size_in_bytes`, the table is compacted. + /// The trigger assumes each newly written file is smaller than the target size. If the target + /// is set below the size of a typical file, compaction is attempted on nearly every write, but + /// is a harmless no-op. Returns [`Ok`] if the table did not need compacting or was compacted + /// successfully, otherwise [`ModelarDbServerError`](crate::error::ModelarDbServerError). + pub(super) async fn increase_estimated_compactable_size( + &self, + table_name: &str, + size_in_bytes: u64, + ) -> Result<()> { + // entry() is not used as it would require the allocation of a new String for each lookup as + // it must be given as a K, while get_mut() accepts the key as a &K so one K can be used. + if !self + .estimated_compactable_size_in_bytes + .contains_key(table_name) + { + self.estimated_compactable_size_in_bytes + .insert(table_name.to_owned(), 0); + } + *self + .estimated_compactable_size_in_bytes + .get_mut(table_name) + .unwrap() += size_in_bytes; + + let estimate_reached_target = *self + .estimated_compactable_size_in_bytes + .get(table_name) + .expect("table_name should have been added to estimated_compactable_size_in_bytes.") + .value() + >= self.optimize_target_file_size_in_bytes; + + if estimate_reached_target { + self.compact_table(table_name).await?; + } + + Ok(()) + } + + /// Compact the table with `table_name` by merging its small files into files of approximately + /// `optimize_target_file_size_in_bytes` bytes, vacuuming the files left behind, and resetting + /// the table's estimated compactable size. Note that the vacuum can physically delete files + /// that an in-progress query is still scanning if `vacuum_retention_period_in_seconds` is very + /// low. Returns [`Ok`] if the table was compacted successfully, otherwise + /// [`ModelarDbServerError`](crate::error::ModelarDbServerError). + async fn compact_table(&self, table_name: &str) -> Result<()> { + debug!("Compacting the storage of the table '{table_name}'."); + + self.local_data_folder + .optimize_table(table_name, Some(self.optimize_target_file_size_in_bytes)) + .await?; + + self.local_data_folder + .vacuum_table(table_name, Some(self.vacuum_retention_period_in_seconds)) + .await?; + + // Reset the estimate so the next compaction only counts data written from now on. + *self + .estimated_compactable_size_in_bytes + .get_mut(table_name) + .expect("table_name should be in estimated_compactable_size_in_bytes.") = 0; + + Ok(()) + } + + /// Set the target size, in bytes, of the files produced when a table is optimized to + /// `new_optimize_target_file_size_in_bytes`. The new target takes effect the next time each + /// table is written to. Tables are not re-compacted here to keep configuration updates cheap + /// and to avoid having to re-check all files on disk to see if they are compactable. + pub(super) fn set_optimize_target_file_size_in_bytes( + &mut self, + new_optimize_target_file_size_in_bytes: u64, + ) { + self.optimize_target_file_size_in_bytes = new_optimize_target_file_size_in_bytes; + } + + /// Set the retention period, in seconds, used when a table is vacuumed as part of compaction to + /// `new_vacuum_retention_period_in_seconds`. + pub(super) fn set_vacuum_retention_period_in_seconds( + &mut self, + new_vacuum_retention_period_in_seconds: u64, + ) { + self.vacuum_retention_period_in_seconds = new_vacuum_retention_period_in_seconds; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use modelardb_test::table::{self, TIME_SERIES_TABLE_NAME}; + use tempfile::{self, TempDir}; + + const OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES: u64 = 1024 * 1024; + const VACUUM_RETENTION_PERIOD_IN_SECONDS: u64 = 0; + const BATCH_COUNT: u8 = 3; + + // Tests for try_new(). + #[tokio::test] + async fn test_initialize_estimate_from_existing_small_files() { + let (_temp_dir, local_data_folder) = create_local_data_folder_with_table().await; + write_batches_to_table(&local_data_folder, BATCH_COUNT).await; + + // The compactor is created after the data is written, so its estimate includes the small + // files already on disk. + let compactor = create_data_storage_compactor(local_data_folder.clone()).await; + + let expected_estimate: u64 = local_data_folder + .table_file_sizes(TIME_SERIES_TABLE_NAME) + .await + .unwrap() + .into_iter() + .sum(); + assert!(expected_estimate > 0); + + assert_eq!( + *compactor + .estimated_compactable_size_in_bytes + .get(TIME_SERIES_TABLE_NAME) + .unwrap(), + expected_estimate + ); + } + + #[tokio::test] + async fn test_initialize_estimate_excludes_files_at_or_above_target() { + let (_temp_dir, local_data_folder) = create_local_data_folder_with_table().await; + write_batches_to_table(&local_data_folder, BATCH_COUNT).await; + + // With a one-byte target, every existing file is already at or above the target, so none of + // them count towards the compactable backlog. + let compactor = DataStorageCompactor::try_new(local_data_folder.clone(), 1, 0) + .await + .unwrap(); + + assert_eq!(table_file_count(&local_data_folder), BATCH_COUNT); + + assert_eq!( + *compactor + .estimated_compactable_size_in_bytes + .get(TIME_SERIES_TABLE_NAME) + .unwrap(), + 0 + ); + } + + // Tests for increase_estimated_compactable_size(). + #[tokio::test] + async fn test_compact_table_when_estimate_reaches_target() { + let (_temp_dir, local_data_folder) = create_local_data_folder_with_table().await; + let compactor = create_data_storage_compactor(local_data_folder.clone()).await; + + write_batches_to_table(&local_data_folder, BATCH_COUNT).await; + + let initial_file_count = table_file_count(&local_data_folder); + assert_eq!(initial_file_count, BATCH_COUNT); + + compactor + .increase_estimated_compactable_size( + TIME_SERIES_TABLE_NAME, + OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES, + ) + .await + .unwrap(); + + // The small files should have been compacted into a single file. + assert_eq!(table_file_count(&local_data_folder), 1); + + // The estimate should have been reset after compacting. + assert_eq!( + *compactor + .estimated_compactable_size_in_bytes + .get(TIME_SERIES_TABLE_NAME) + .unwrap(), + 0 + ); + } + + #[tokio::test] + async fn test_do_not_compact_table_when_estimate_below_target() { + let (_temp_dir, local_data_folder) = create_local_data_folder_with_table().await; + let compactor = create_data_storage_compactor(local_data_folder.clone()).await; + + write_batches_to_table(&local_data_folder, BATCH_COUNT).await; + + let initial_file_count = table_file_count(&local_data_folder); + assert_eq!(initial_file_count, BATCH_COUNT); + + compactor + .increase_estimated_compactable_size( + TIME_SERIES_TABLE_NAME, + OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES - 1, + ) + .await + .unwrap(); + + // No files should have been compacted since the estimate did not reach the target. + assert_eq!(table_file_count(&local_data_folder), initial_file_count); + + // The estimate should have accumulated without being reset. + assert_eq!( + *compactor + .estimated_compactable_size_in_bytes + .get(TIME_SERIES_TABLE_NAME) + .unwrap(), + OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES - 1 + ); + } + + /// Create a [`DataFolder`] in a local [`TempDir`] containing a single time series table. + async fn create_local_data_folder_with_table() -> (TempDir, DataFolder) { + let temp_dir = tempfile::tempdir().unwrap(); + let temp_dir_url = temp_dir.path().to_str().unwrap(); + let local_data_folder = DataFolder::open_local_url(temp_dir_url).await.unwrap(); + + let time_series_table_metadata = table::time_series_table_metadata(); + local_data_folder + .create_time_series_table(&time_series_table_metadata) + .await + .unwrap(); + + (temp_dir, local_data_folder) + } + + /// Write `batch_count` batches of compressed segments to the time series table in + /// `local_data_folder`, each as a separate file. + async fn write_batches_to_table(local_data_folder: &DataFolder, batch_count: u8) { + for _ in 0..batch_count { + local_data_folder + .write_record_batches( + TIME_SERIES_TABLE_NAME, + vec![table::compressed_segments_record_batch()], + ) + .await + .unwrap(); + } + } + + /// Return the number of physical Apache Parquet files in the time series table in + /// `local_data_folder`. + fn table_file_count(local_data_folder: &DataFolder) -> u8 { + let column_path = format!( + "{}/tables/{}/field_column=0", + local_data_folder.location(), + TIME_SERIES_TABLE_NAME + ); + + std::fs::read_dir(column_path).unwrap().count() as u8 + } + + /// Create a [`DataStorageCompactor`] that compacts the tables in `local_data_folder`. + async fn create_data_storage_compactor(local_data_folder: DataFolder) -> DataStorageCompactor { + DataStorageCompactor::try_new( + local_data_folder, + OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES, + VACUUM_RETENTION_PERIOD_IN_SECONDS, + ) + .await + .unwrap() + } +} diff --git a/crates/modelardb_server/src/storage/data_transfer.rs b/crates/modelardb_server/src/storage/data_transfer.rs index 182e896e..885d530c 100644 --- a/crates/modelardb_server/src/storage/data_transfer.rs +++ b/crates/modelardb_server/src/storage/data_transfer.rs @@ -22,7 +22,6 @@ use dashmap::DashMap; use deltalake::arrow::array::RecordBatch; use futures::TryStreamExt; use modelardb_storage::data_folder::DataFolder; -use object_store::ObjectStoreExt; use tracing::debug; use crate::error::Result; @@ -61,15 +60,13 @@ impl DataTransfer { // The size of tables is computed manually as datafusion_table_statistics() is not exact. let table_size_in_bytes = DashMap::with_capacity(table_names.len()); for table_name in table_names { - let delta_table = local_data_folder.delta_table(&table_name).await?; + let size_in_bytes: u64 = local_data_folder + .table_file_sizes(&table_name) + .await? + .into_iter() + .sum(); - let mut table_size_in_bytes = table_size_in_bytes.entry(table_name).or_insert(0); - - let object_store = delta_table.object_store(); - for file_path in delta_table.get_files_by_partitions(&[]).await? { - let object_meta = object_store.head(&file_path).await?; - *table_size_in_bytes += object_meta.size; - } + table_size_in_bytes.insert(table_name, size_in_bytes); } let data_transfer = Self { @@ -460,15 +457,12 @@ mod tests { /// Return the total size of the files in the table with `table_name` in `local_data_folder`. async fn table_files_size(local_data_folder: &DataFolder, table_name: &str) -> u64 { - let delta_table = local_data_folder.delta_table(table_name).await.unwrap(); - - let mut files_size = 0; - for file_path in delta_table.get_files_by_partitions(&[]).await.unwrap() { - let object_meta = delta_table.object_store().head(&file_path).await; - files_size += object_meta.unwrap().size; - } - - files_size + local_data_folder + .table_file_sizes(table_name) + .await + .unwrap() + .into_iter() + .sum() } /// Create a data transfer component with a target object store that is deleted once the test is finished. diff --git a/crates/modelardb_server/src/storage/mod.rs b/crates/modelardb_server/src/storage/mod.rs index 0d323396..19eb2718 100644 --- a/crates/modelardb_server/src/storage/mod.rs +++ b/crates/modelardb_server/src/storage/mod.rs @@ -24,6 +24,7 @@ mod compressed_data_buffer; mod compressed_data_manager; pub(super) mod data_sinks; // pub(super) so it can be used in context.rs. +mod data_storage_compactor; mod data_transfer; mod types; mod uncompressed_data_buffer; @@ -42,6 +43,7 @@ use crate::configuration::{ConfigurationManager, WalMode}; use crate::data_folders::DataFolders; use crate::error::{ModelarDbServerError, Result}; use crate::storage::compressed_data_manager::CompressedDataManager; +use crate::storage::data_storage_compactor::DataStorageCompactor; use crate::storage::data_transfer::DataTransfer; use crate::storage::types::{Channels, MemoryPool, Message}; use crate::storage::uncompressed_data_buffer::IngestedDataBuffer; @@ -146,6 +148,13 @@ impl StorageEngine { } // Create the compressed data manager. + let data_storage_compactor = DataStorageCompactor::try_new( + data_folders.local_data_folder.clone(), + configuration_manager.optimize_target_file_size_in_bytes(), + configuration_manager.vacuum_retention_period_in_seconds(), + ) + .await?; + let data_transfer = if let Some(remote_data_folder) = data_folders.maybe_remote_data_folder { let data_transfer = DataTransfer::try_new( @@ -161,6 +170,7 @@ impl StorageEngine { }; let compressed_data_manager = Arc::new(CompressedDataManager::new( + Arc::new(RwLock::new(data_storage_compactor)), Arc::new(RwLock::new(data_transfer)), data_folders.local_data_folder, channels.clone(), @@ -388,4 +398,23 @@ impl StorageEngine { )) } } + + /// Set the target file size used when automatically compacting a table's storage to `new_value`. + pub(super) async fn set_optimize_target_file_size_in_bytes(&self, new_value: u64) { + self.compressed_data_manager + .data_storage_compactor + .write() + .await + .set_optimize_target_file_size_in_bytes(new_value); + } + + /// Set the retention period used when automatically vacuuming a table during compaction to + /// `new_value`. + pub(super) async fn set_vacuum_retention_period_in_seconds(&self, new_value: u64) { + self.compressed_data_manager + .data_storage_compactor + .write() + .await + .set_vacuum_retention_period_in_seconds(new_value); + } } diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index 3adf3c7e..ea5dad15 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -831,7 +831,7 @@ async fn test_can_optimize_normal_table() { .await; // ingest_time_series_and_flush_data() writes one file. Ingest and flush three more times so - // there are four small files to compact. + // there are four small files to merge. for _ in 0..3 { let flight_data = TestContext::create_flight_data_from_time_series( NORMAL_TABLE_NAME.to_owned(), @@ -886,7 +886,7 @@ async fn test_can_optimize_time_series_table() { .await; // ingest_time_series_and_flush_data() writes one file per field column partition. ingest and - // flush three more times so each partition has four small files to compact. + // flush three more times so each partition has four small files to merge. for _ in 0..3 { let flight_data = TestContext::create_flight_data_from_time_series( TIME_SERIES_TABLE_NAME.to_owned(), @@ -921,7 +921,7 @@ async fn test_can_optimize_time_series_table() { .await .unwrap(); - // The four files in the partition should be compacted into a single active file. + // The four files in the partition should be merged into a single active file. let files = std::fs::read_dir(&column_path).unwrap(); assert_eq!(files.count(), 1); } @@ -1429,6 +1429,14 @@ async fn test_can_get_configuration() { configuration.segment_size_threshold_in_bytes, 64 * 1024 * 1024 ); + assert_eq!( + configuration.optimize_target_file_size_in_bytes, + 64 * 1024 * 1024 + ); + assert_eq!( + configuration.vacuum_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); @@ -1478,6 +1486,26 @@ async fn test_can_update_segment_size_threshold_in_bytes() { assert_eq!(updated_configuration.segment_size_threshold_in_bytes, 1); } +#[tokio::test] +async fn test_can_update_optimize_target_file_size_in_bytes() { + let updated_configuration = update_and_get_configuration( + protocol::update_configuration::Setting::OptimizeTargetFileSizeInBytes as i32, + ) + .await; + + assert_eq!(updated_configuration.optimize_target_file_size_in_bytes, 1); +} + +#[tokio::test] +async fn test_can_update_vacuum_retention_period_in_seconds() { + let updated_configuration = update_and_get_configuration( + protocol::update_configuration::Setting::VacuumRetentionPeriodInSeconds as i32, + ) + .await; + + assert_eq!(updated_configuration.vacuum_retention_period_in_seconds, 1); +} + async fn update_and_get_configuration(setting: i32) -> protocol::Configuration { let mut test_context = TestContext::new().await; test_context @@ -1518,6 +1546,8 @@ async fn test_cannot_update_non_nullable_setting_with_null_value() { protocol::update_configuration::Setting::UncompressedReservedMemoryInBytes as i32, protocol::update_configuration::Setting::CompressedReservedMemoryInBytes as i32, protocol::update_configuration::Setting::SegmentSizeThresholdInBytes as i32, + protocol::update_configuration::Setting::OptimizeTargetFileSizeInBytes as i32, + protocol::update_configuration::Setting::VacuumRetentionPeriodInSeconds as i32, ] { update_configuration_and_assert_error( setting, diff --git a/crates/modelardb_storage/src/data_folder/mod.rs b/crates/modelardb_storage/src/data_folder/mod.rs index 4fe76801..4d9b4f2b 100644 --- a/crates/modelardb_storage/src/data_folder/mod.rs +++ b/crates/modelardb_storage/src/data_folder/mod.rs @@ -48,11 +48,11 @@ use modelardb_types::types::{ ArrowValue, CloudCredentials, ErrorBound, GeneratedColumn, MAX_RETENTION_PERIOD_IN_SECONDS, TimeSeriesTableMetadata, }; -use object_store::ObjectStore; use object_store::aws::AmazonS3Builder; use object_store::local::LocalFileSystem; use object_store::memory::InMemory; use object_store::path::Path; +use object_store::{ObjectStore, ObjectStoreExt}; use url::Url; use crate::data_folder::delta_table_writer::DeltaTableWriter; @@ -685,12 +685,12 @@ impl DataFolder { Ok(()) } - /// Optimize the Delta Lake table with `table_name` by compacting its many small files into + /// Optimize the Delta Lake table with `table_name` by merging its many small files into /// fewer larger files of approximately `maybe_target_size_in_bytes` bytes. If a target size is - /// not given, a default target size of 64 MiB is used. Compaction only rewrites files smaller + /// not given, a default target size of 64 MiB is used. Optimize only rewrites files smaller /// than the target, so it is safe to call repeatedly. Note that the small files are only marked /// as removed and are not deleted from disk until the table is vacuumed. If the target size is - /// zero, the table does not exist, or the files could not be compacted, a + /// zero, the table does not exist, or the files could not be merged, a /// [`ModelarDbStorageError`] is returned. pub async fn optimize_table( &self, @@ -711,6 +711,21 @@ impl DataFolder { Ok(()) } + /// Return the size in bytes of each Apache Parquet file that makes up the Delta Lake table with + /// `table_name`. If the table does not exist or the size of a file could not be read, a + /// [`ModelarDbStorageError`] is returned. + pub async fn table_file_sizes(&self, table_name: &str) -> Result> { + let delta_table = self.delta_table(table_name).await?; + let object_store = delta_table.object_store(); + + let mut file_sizes_in_bytes = Vec::new(); + for file_path in delta_table.get_files_by_partitions(&[]).await? { + file_sizes_in_bytes.push(object_store.head(&file_path).await?.size); + } + + Ok(file_sizes_in_bytes) + } + /// Return a [`DeltaTableWriter`] for writing to the table with `table_name` in the Delta Lake, /// or a [`ModelarDbStorageError`] if a connection to the Delta Lake cannot be established or /// the table does not exist. @@ -1604,7 +1619,7 @@ mod tests { .await .unwrap(); - // The small files should be compacted into a single active file with no rows lost or + // The small files should be merged into a single active file with no rows lost or // duplicated. assert_eq!(active_file_count(&data_folder, "normal_table_1").await, 1); assert_eq!(row_count(&data_folder, "normal_table_1").await, rows_before); @@ -1636,7 +1651,7 @@ mod tests { .await .unwrap(); - // The small files should be compacted into a single active file with no rows lost or + // The small files should be merged into a single active file with no rows lost or // duplicated. assert_eq!( active_file_count(&data_folder, TIME_SERIES_TABLE_NAME).await, @@ -1675,7 +1690,7 @@ mod tests { assert_eq!(files_before, 4); // A one-byte target is smaller than every existing file, so none of them are candidates for - // compaction, and the files should be left untouched. + // merging, and the files should be left untouched. data_folder .optimize_table("normal_table_1", Some(1)) .await diff --git a/crates/modelardb_storage/src/write_ahead_log.rs b/crates/modelardb_storage/src/write_ahead_log.rs index 34748cd0..786fda82 100644 --- a/crates/modelardb_storage/src/write_ahead_log.rs +++ b/crates/modelardb_storage/src/write_ahead_log.rs @@ -779,6 +779,51 @@ mod tests { assert_eq!(*persisted, BTreeSet::from([0, 1, 2])); } + #[tokio::test] + async fn test_wal_recovery_survives_optimize_and_vacuum() { + let (temp_dir, data_folder) = create_data_folder_with_time_series_table().await; + + // Write three separate commits, each carrying its own batch ids. + write_compressed_segments_with_batch_ids(&data_folder, HashSet::from([0, 1, 2])).await; + write_compressed_segments_with_batch_ids(&data_folder, HashSet::from([3, 4, 5])).await; + write_compressed_segments_with_batch_ids(&data_folder, HashSet::from([6, 7, 8])).await; + + let column_path = format!( + "{}/tables/{}/field_column=0", + temp_dir.path().to_str().unwrap(), + TIME_SERIES_TABLE_NAME + ); + assert_eq!(std::fs::read_dir(&column_path).unwrap().count(), 3); + + // Optimize merges the small files into one, and vacuum physically deletes the stale files + // left behind. Vacuum should only remove Parquet data files, never the _delta_log commits. + data_folder + .optimize_table(TIME_SERIES_TABLE_NAME, None) + .await + .unwrap(); + data_folder + .vacuum_table(TIME_SERIES_TABLE_NAME, Some(0)) + .await + .unwrap(); + + // Only the single merged Parquet file should remain on disk. + assert_eq!(std::fs::read_dir(&column_path).unwrap().count(), 1); + + // Rebuilding the WAL from the same folder must still recover every persisted batch id from + // the Delta commit history. This proves optimize and vacuum did not discard the commits + // that crash recovery relies on to exclude already persisted data from replay. + let wal = WriteAheadLog::try_new(&data_folder, SEGMENT_SIZE_THRESHOLD_IN_BYTES) + .await + .unwrap(); + + let persisted = wal.table_logs[TIME_SERIES_TABLE_NAME] + .persisted_batch_ids + .lock() + .unwrap(); + + assert_eq!(*persisted, BTreeSet::from([0, 1, 2, 3, 4, 5, 6, 7, 8])); + } + #[tokio::test] async fn test_try_new_fails_for_non_local_data_folder() { let data_folder = DataFolder::open_memory().await.unwrap(); diff --git a/crates/modelardb_types/src/flight/protocol.proto b/crates/modelardb_types/src/flight/protocol.proto index 082f91b0..141816e3 100644 --- a/crates/modelardb_types/src/flight/protocol.proto +++ b/crates/modelardb_types/src/flight/protocol.proto @@ -79,6 +79,15 @@ message Configuration { // Whether the write-ahead log is enabled. bool wal_enabled = 9; + + // Target size, in bytes, of the files produced when automatically compacting a table's storage. This is also the + // default value used when an OPTIMIZE query is executed without an explicit target size. + uint64 optimize_target_file_size_in_bytes = 10; + + // Retention period, in seconds, used when automatically vacuuming a table during compaction. This is also the + // default value used when a VACUUM query is executed without an explicit retention period. Note that a very low + // value can delete files an in-progress query is still scanning. + uint64 vacuum_retention_period_in_seconds = 11; } // Request to update the configuration of a ModelarDB node. @@ -89,6 +98,8 @@ message UpdateConfiguration { COMPRESSED_RESERVED_MEMORY_IN_BYTES = 2; TRANSFER_BATCH_SIZE_IN_BYTES = 3; SEGMENT_SIZE_THRESHOLD_IN_BYTES = 4; + OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES = 5; + VACUUM_RETENTION_PERIOD_IN_SECONDS = 6; } // Setting to update in the configuration. diff --git a/docs/user/README.md b/docs/user/README.md index 6e7ca86f..8a64f20f 100644 --- a/docs/user/README.md +++ b/docs/user/README.md @@ -1,9 +1,11 @@ # ModelarDB Installation and Usage + This document describes how to set up and use ModelarDB. Installation instructions are provided for Linux, macOS, FreeBSD, and Windows. To support running ModelarDB in a containerized environment, instructions for setting up a Docker environment are also provided. Once installed, using ModelarDB is consistent across all platforms. ## Installation from Builds + Builds for `aarch64 macOS`, `x86_64 Windows`, and `x86_64 Linux` are created for each commit to the `main` branch using GitHub Actions. As these builds are created for each commit, they are not considered stable release builds. Also, since they are built using GitHub Actions, they are only available for 90 days, as this is GitHub's maximum artifact retention @@ -12,23 +14,29 @@ completed successfully](https://github.com/ModelarData/ModelarDB-RS/actions/workflows/build-lint-test-and-upload.yml?query=branch%3Amain). ## Installation from Source + ### Linux + The following commands are for Ubuntu Server. However, equivalent commands should work for other Linux distributions. 1. Install [build-essential](https://packages.ubuntu.com/jammy/build-essential): `sudo apt install build-essential` ### macOS + 1. Install the Xcode Command Line Developer Tools: `xcode-select --install` ### FreeBSD -1. Install [cURL](https://curl.se/) as the *root* user: `pkg install curl` + +1. Install [cURL](https://curl.se/) as the _root_ user: `pkg install curl` ### Windows + 1. Install the latest versions of the Microsoft Visual C++ Prerequisites for Rust: - Microsoft Visual C++ Prerequisites for Rust: see [The rustup book](https://rust-lang.github.io/rustup/installation/windows-msvc.html). ### All + 2. Install the latest stable [Rust Toolchain](https://rustup.rs/). 3. Clone the repository: `git clone https://github.com/ModelarData/ModelarDB-RS` 4. Build, test, and run the system using Cargo: @@ -43,6 +51,7 @@ The following commands are for Ubuntu Server. However, equivalent commands shoul - Run Tests: `python3 -m unittest` ## Usage + ModelarDB consists of three binaries and a library with bindings: `modelardbd` is a DBMS server that manages data and executes SQL queries, `modelardb` is a command-line client for connecting to a `modelardbd` instance and executing commands and SQL queries, `modelardbb` is a command-line bulk loader that operates without `modelardbd` as it reads @@ -66,6 +75,7 @@ the cloud to use for executing each query, thus providing a workload-balanced in in the object store using the `modelardbd` instances in the cloud. ### Start Server + There are three options available when starting `modelardbd` depending on the desired deployment use case. Each option has different requirements and supports different features. @@ -87,7 +97,7 @@ modelardbd edge path_to_local_data_folder is required. This configuration supports ingesting data to a local folder in the cloud, transferring data in the local data folder to the object store, and querying the data in the object store in the cloud. -The following flags (or the corresponding environment variables) must be provided if an Amazon S3-compatible object +The following flags (or the corresponding environment variables) must be provided if an Amazon S3-compatible object store is used: ```shell @@ -115,7 +125,7 @@ modelardbd edge path_to_local_data_folder s3://wind-turbine ``` `modelardbd` also supports using [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/) -for the remote object store. To use Azure Blob Storage, provide the following flags or the corresponding environment +for the remote object store. To use Azure Blob Storage, provide the following flags or the corresponding environment variables: ```shell @@ -137,7 +147,7 @@ mode the `modelardbd` instance will execute queries against the object store and modelardbd cloud path_to_local_data_folder s3://wind-turbine ``` -Note that `modelardbd` uses `127.0.0.1` and `9999` as the default host and port. Both can be changed using flags or the +Note that `modelardbd` uses `127.0.0.1` and `9999` as the default host and port. Both can be changed using flags or the corresponding environment variables: ```shell @@ -149,6 +159,7 @@ MODELARDBD_HOST=0.0.0.0 modelardbd edge path_to_local_data_folder ``` ### Ingest Data + Before data can be ingested into `modelardbd`, tables must be created. `modelardbd` supports two types of tables, standard relational tables created with `CREATE TABLE` statements and time series tables created with `CREATE TIME SERIES TABLE` statements. From a user's perspective, a time series table functions like a standard relational table and can be queried @@ -237,6 +248,7 @@ in edge mode in a cluster or in cloud mode in a cluster. When `modelardbd` is de the ingested data is only stored in local storage. ### Execute Queries + ModelarDB includes a command-line client in the form of `modelardb`. To interactively execute SQL statements against a local instance of `modelardbd` through a REPL, simply run `modelardb`: @@ -347,6 +359,7 @@ for flight_stream_chunk in flight_stream_reader: ``` ### Embed Library + ModelarDB includes an embeddable library in the form of `modelardb_embedded`. It allows programming languages to execute queries against or write to `modelardbd` or a data folder directly. A C-API allows other programming languages than Rust to also use `modelardb_embedded`. The location where queries and writes are executed is specified @@ -371,27 +384,31 @@ modelardb_node = modelardb.connect(url) ``` ## ModelarDB configuration + When the server is started for the first time, a configuration file is created in the root of the data folder named `modelardbd.toml`. If the file is changed manually, the changes are only applied when the server is restarted. -`modelardbd` can be configured before the server is started using command line flags or environment variables. Flags take -precedence over environment variables, which take precedence over the configuration file, which takes precedence over -the built-in defaults. Note that the connection settings `--host` and `--port` are not persisted in the configuration -file. Variables marked with ✓ in the **Updatable** column can also be updated while the server is running using the +`modelardbd` can be configured before the server is started using command line flags or environment variables. Flags take +precedence over environment variables, which take precedence over the configuration file, which takes precedence over +the built-in defaults. Note that the connection settings `--host` and `--port` are not persisted in the configuration +file. Variables marked with ✓ in the **Updatable** column can also be updated while the server is running using the `UpdateConfiguration` action without requiring a restart. The update is persisted in the configuration file. -| **CLI Flag** | **Environment Variable** | **Default** | **Updatable** | **Description** | -|-------------------------------------------|----------------------------------------------------|-------------|---------------|--------------------------------------------------------------------------------------------------------------------------------| -| `--host` | `MODELARDBD_HOST` | 127.0.0.1 | | The host address of the `modelardbd` server. | -| `--port` | `MODELARDBD_PORT` | 9999 | | The port of the `modelardbd` server. | -| `--wal-enabled` | `MODELARDBD_WAL_ENABLED` | true | | Whether the write-ahead log is enabled. | -| `--ingested-reserved-memory-in-bytes` | `MODELARDBD_INGESTED_RESERVED_MEMORY_IN_BYTES` | 512 MB | ✓ | The amount of memory to reserve for storing ingested time series. | -| `--uncompressed-reserved-memory-in-bytes` | `MODELARDBD_UNCOMPRESSED_RESERVED_MEMORY_IN_BYTES` | 512 MB | ✓ | The amount of memory to reserve for storing uncompressed data buffers. | -| `--compressed-reserved-memory-in-bytes` | `MODELARDBD_COMPRESSED_RESERVED_MEMORY_IN_BYTES` | 512 MB | ✓ | The amount of memory to reserve for storing compressed data buffers. | -| `--transfer-batch-size-in-bytes` | `MODELARDBD_TRANSFER_BATCH_SIZE_IN_BYTES` | 64 MB | ✓ | The amount of data that must be collected before transferring a batch to the remote object store. | -| `--segment-size-threshold-in-bytes` | `MODELARDBD_SEGMENT_SIZE_THRESHOLD_IN_BYTES` | 64 MB | ✓ | The approximate maximum size of a single WAL segment file before a new one is started. Only updatable when the WAL is enabled. | +| **CLI Flag** | **Environment Variable** | **Default** | **Updatable** | **Description** | +|-------------------------------------------|----------------------------------------------------|-------------|---------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `--host` | `MODELARDBD_HOST` | 127.0.0.1 | | The host address of the `modelardbd` server. | +| `--port` | `MODELARDBD_PORT` | 9999 | | The port of the `modelardbd` server. | +| `--wal-enabled` | `MODELARDBD_WAL_ENABLED` | true | | Whether the write-ahead log is enabled. | +| `--ingested-reserved-memory-in-bytes` | `MODELARDBD_INGESTED_RESERVED_MEMORY_IN_BYTES` | 512 MB | ✓ | The amount of memory to reserve for storing ingested time series. | +| `--uncompressed-reserved-memory-in-bytes` | `MODELARDBD_UNCOMPRESSED_RESERVED_MEMORY_IN_BYTES` | 512 MB | ✓ | The amount of memory to reserve for storing uncompressed data buffers. | +| `--compressed-reserved-memory-in-bytes` | `MODELARDBD_COMPRESSED_RESERVED_MEMORY_IN_BYTES` | 512 MB | ✓ | The amount of memory to reserve for storing compressed data buffers. | +| `--transfer-batch-size-in-bytes` | `MODELARDBD_TRANSFER_BATCH_SIZE_IN_BYTES` | 64 MB | ✓ | The amount of data that must be collected before transferring a batch to the remote object store. | +| `--segment-size-threshold-in-bytes` | `MODELARDBD_SEGMENT_SIZE_THRESHOLD_IN_BYTES` | 64 MB | ✓ | The approximate maximum size of a single WAL segment file before a new one is started. Only updatable when the WAL is enabled. | +| `--optimize-target-file-size-in-bytes` | `MODELARDBD_OPTIMIZE_TARGET_FILE_SIZE_IN_BYTES` | 64 MB | ✓ | The target size of the files produced when automatically compacting a table's storage. Also used as the default when `OPTIMIZE` is run without an explicit `TARGET`. | +| `--vacuum-retention-period-in-seconds` | `MODELARDBD_VACUUM_RETENTION_PERIOD_IN_SECONDS` | 7 days | ✓ | The retention period used when automatically vacuuming a table during compaction. Also used as the default when `VACUUM` is run without an explicit `RETAIN`. Note that a very low value can delete files an in-progress query is still scanning. | ## Docker + Two different [Docker](https://docs.docker.com/) environments are included to make it easy to experiment with the different use cases of ModelarDB. The first environment sets up a single instance of `modelardbd` that only uses local storage. Data can be ingested into this instance, compressed, and saved to local storage. The compressed data in local @@ -407,6 +424,7 @@ software that utilizes ModelarDB. Downloading [Docker Desktop](https://docs.dock make maintenance of the created containers easier. ### Single edge deployment + Once [Docker](https://docs.docker.com/) is set up, the single edge deployment can be started by running the following command from the root of the ModelarDB repository: @@ -428,6 +446,7 @@ Arrow Flight as described above. Tables can be created and data can be ingested, The compressed data on local disk can then be queried. ### Cluster deployment + Once Docker is set up, the cluster deployment can be started by running the following command from the root of the ModelarDB repository: