diff --git a/crates/modelardb_server/src/configuration.rs b/crates/modelardb_server/src/configuration.rs index afc6b3ff..c510f566 100644 --- a/crates/modelardb_server/src/configuration.rs +++ b/crates/modelardb_server/src/configuration.rs @@ -58,11 +58,8 @@ struct Configuration { /// Amount of memory to reserve for storing compressed data buffers. compressed_reserved_memory_in_bytes: u64, /// The number of bytes that are required before transferring a batch of data to the remote - /// object store. If [`None`], data is not transferred based on batch size. + /// object store. If [`None`], data is only transferred on an explicit flush. transfer_batch_size_in_bytes: Option, - /// 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 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, @@ -98,10 +95,6 @@ impl Configuration { self.transfer_batch_size_in_bytes = Some(value); } - if let Some(value) = args.transfer_time_in_seconds { - self.transfer_time_in_seconds = Some(value); - } - if let Some(value) = args.segment_size_threshold_in_bytes { self.segment_size_threshold_in_bytes = value; } @@ -153,7 +146,6 @@ impl Default for Configuration { uncompressed_reserved_memory_in_bytes: 512 * 1024 * 1024, compressed_reserved_memory_in_bytes: 512 * 1024 * 1024, transfer_batch_size_in_bytes: Some(64 * 1024 * 1024), - transfer_time_in_seconds: None, segment_size_threshold_in_bytes: 64 * 1024 * 1024, ingestion_threads: 1, compression_threads: 1, @@ -364,31 +356,6 @@ impl ConfigurationManager { .await } - pub(crate) fn transfer_time_in_seconds(&self) -> Option { - self.configuration.transfer_time_in_seconds - } - - /// Set the new value and update the transfer time in the storage engine. If the transfer time - /// could not be updated or if the new configuration could not be saved to the configuration - /// file, return [`ModelarDbServerError`]. - pub(crate) async fn set_transfer_time_in_seconds( - &mut self, - new_transfer_time_in_seconds: Option, - storage_engine: Arc>, - ) -> Result<()> { - storage_engine - .write() - .await - .set_transfer_time_in_seconds(new_transfer_time_in_seconds) - .await?; - - self.configuration.transfer_time_in_seconds = new_transfer_time_in_seconds; - - self.configuration - .save_to_toml(&self.local_data_folder) - .await - } - #[allow(dead_code)] pub(crate) fn segment_size_threshold_in_bytes(&self) -> u64 { self.configuration.segment_size_threshold_in_bytes @@ -448,7 +415,6 @@ impl ConfigurationManager { .configuration .compressed_reserved_memory_in_bytes, transfer_batch_size_in_bytes: self.configuration.transfer_batch_size_in_bytes, - transfer_time_in_seconds: self.configuration.transfer_time_in_seconds, segment_size_threshold_in_bytes: self.configuration.segment_size_threshold_in_bytes, ingestion_threads: self.configuration.ingestion_threads as u32, compression_threads: self.configuration.compression_threads as u32, @@ -499,7 +465,6 @@ mod tests { uncompressed_reserved_memory_in_bytes: 1, compressed_reserved_memory_in_bytes: 1, transfer_batch_size_in_bytes: Some(1), - transfer_time_in_seconds: Some(1), segment_size_threshold_in_bytes: 1, ..Configuration::default() }; @@ -718,39 +683,6 @@ mod tests { ); } - #[tokio::test] - async fn test_set_transfer_time_in_seconds() { - let temp_dir = tempfile::tempdir().unwrap(); - let (storage_engine, configuration_manager) = create_components(&temp_dir).await; - - assert_eq!( - configuration_manager - .read() - .await - .transfer_time_in_seconds(), - None - ); - - let new_value = Some(60); - configuration_manager - .write() - .await - .set_transfer_time_in_seconds(new_value, storage_engine) - .await - .unwrap(); - - assert_eq!( - configuration_manager - .read() - .await - .transfer_time_in_seconds(), - new_value - ); - - let configuration_from_file = configuration_from_file(&temp_dir).await; - assert_eq!(configuration_from_file.transfer_time_in_seconds, new_value) - } - #[tokio::test] async fn test_set_segment_size_threshold_in_bytes() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/crates/modelardb_server/src/main.rs b/crates/modelardb_server/src/main.rs index 9a015b3e..e117d821 100644 --- a/crates/modelardb_server/src/main.rs +++ b/crates/modelardb_server/src/main.rs @@ -74,15 +74,10 @@ pub(crate) struct ServerArgs { compressed_reserved_memory_in_bytes: Option, /// Number of bytes required before transferring a batch of data to the remote object store. - /// If not set, data is not transferred based on batch size. + /// If not set, data is only transferred on an explicit flush. #[arg(long, env = "MODELARDBD_TRANSFER_BATCH_SIZE_IN_BYTES")] transfer_batch_size_in_bytes: Option, - /// Number of seconds between each transfer of data to the remote object store. If not set, - /// data is not transferred based on time. - #[arg(long, env = "MODELARDBD_TRANSFER_TIME_IN_SECONDS")] - transfer_time_in_seconds: Option, - /// Approximate maximum size in bytes of a single WAL segment file before a new one is started. /// The size is approximate since the in-memory size of each batch is used instead of its /// on-disk size to avoid the overhead of reading the file size after each write. diff --git a/crates/modelardb_server/src/remote/mod.rs b/crates/modelardb_server/src/remote/mod.rs index 7558da13..cf63bc3c 100644 --- a/crates/modelardb_server/src/remote/mod.rs +++ b/crates/modelardb_server/src/remote/mod.rs @@ -1013,12 +1013,6 @@ impl FlightService for FlightServiceHandler { .await .map_err(error_to_status_internal) } - Ok(protocol::update_configuration::Setting::TransferTimeInSeconds) => { - configuration_manager - .set_transfer_time_in_seconds(maybe_new_value, storage_engine) - .await - .map_err(error_to_status_internal) - } Ok(protocol::update_configuration::Setting::SegmentSizeThresholdInBytes) => { let new_value = maybe_new_value.ok_or(invalid_null_error)?; diff --git a/crates/modelardb_server/src/storage/data_transfer.rs b/crates/modelardb_server/src/storage/data_transfer.rs index 246e8c16..182e896e 100644 --- a/crates/modelardb_server/src/storage/data_transfer.rs +++ b/crates/modelardb_server/src/storage/data_transfer.rs @@ -17,16 +17,12 @@ //! is managed here until it is of a sufficient size to be transferred efficiently. use std::collections::HashSet; -use std::sync::Arc; -use std::time::Duration; use dashmap::DashMap; use deltalake::arrow::array::RecordBatch; use futures::TryStreamExt; use modelardb_storage::data_folder::DataFolder; use object_store::ObjectStoreExt; -use tokio::sync::RwLock; -use tokio::task::JoinHandle as TaskJoinHandle; use tracing::debug; use crate::error::Result; @@ -44,11 +40,8 @@ pub struct DataTransfer { /// Map from table names to the current size of the table in bytes. table_size_in_bytes: DashMap, /// The number of bytes that are required before transferring a batch of data to the remote - /// Delta Lake. If [`None`], data is not transferred based on batch size. + /// Delta Lake. If [`None`], data is only transferred on an explicit flush. transfer_batch_size_in_bytes: Option, - /// Handle to the task that transfers data periodically to the remote object store. If [`None`], - /// data is not transferred based on time. - transfer_scheduler_handle: Option>, /// Tables that have been dropped and should not be transferred to the remote data folder. dropped_tables: HashSet, } @@ -84,7 +77,6 @@ impl DataTransfer { remote_data_folder, table_size_in_bytes: table_size_in_bytes.clone(), transfer_batch_size_in_bytes, - transfer_scheduler_handle: None, dropped_tables: HashSet::new(), }; @@ -160,43 +152,6 @@ impl DataTransfer { Ok(()) } - /// If a new transfer time is given, stop the existing task transferring data periodically - /// if there is one, and start a new task. If `new_value` is [`None`], the task is just stopped. - /// `data_transfer` is needed as an argument instead of using `self` so it can be moved into the - /// periodic task. - pub(super) fn set_transfer_time_in_seconds( - &mut self, - new_value: Option, - data_transfer: Arc>>, - ) { - // Stop the current task periodically transferring data if there is one. - if let Some(task) = &self.transfer_scheduler_handle { - task.abort(); - } - - // If a transfer time was given, create the task that periodically transfers data. - self.transfer_scheduler_handle = if let Some(transfer_time) = new_value { - let join_handle = tokio::spawn(async move { - loop { - tokio::time::sleep(Duration::from_secs(transfer_time)).await; - - data_transfer - .write() - .await - .as_ref() - .unwrap() - .transfer_larger_than_threshold(0) - .await - .expect("Periodic data transfer failed."); - } - }); - - Some(join_handle) - } else { - None - }; - } - /// Transfer all compressed files from tables currently using more than `threshold` bytes in the /// data folder to the remote object store. Return [`Ok`] if all files were transferred /// successfully, otherwise [`ModelarDbServerError`](crate::error::ModelarDbServerError). Note diff --git a/crates/modelardb_server/src/storage/mod.rs b/crates/modelardb_server/src/storage/mod.rs index 6f9a3c1a..8460b51b 100644 --- a/crates/modelardb_server/src/storage/mod.rs +++ b/crates/modelardb_server/src/storage/mod.rs @@ -160,7 +160,6 @@ impl StorageEngine { None }; - let data_transfer_is_some = data_transfer.is_some(); let compressed_data_manager = Arc::new(CompressedDataManager::new( Arc::new(RwLock::new(data_transfer)), data_folders.local_data_folder, @@ -187,24 +186,14 @@ impl StorageEngine { )?; } - let mut storage_engine = Self { + Ok(Self { uncompressed_data_manager, compressed_data_manager, memory_pool, join_handles, channels, wal_mode, - }; - - // Start the task that transfers data periodically if a remote data folder is given and - // time-based data transfer is enabled. - if data_transfer_is_some { - storage_engine - .set_transfer_time_in_seconds(configuration_manager.transfer_time_in_seconds()) - .await?; - } - - Ok(storage_engine) + }) } /// Start `num_threads` threads with `name` that executes `function` and whose [`JoinHandle`] is @@ -399,26 +388,4 @@ impl StorageEngine { )) } } - - /// Set the transfer time in the data transfer component to `new_value` if it exists. If a data - /// transfer component does not exist, return [`ModelarDbServerError`]. - pub(super) async fn set_transfer_time_in_seconds( - &mut self, - new_value: Option, - ) -> Result<()> { - if let Some(ref mut data_transfer) = - *self.compressed_data_manager.data_transfer.write().await - { - data_transfer.set_transfer_time_in_seconds( - new_value, - self.compressed_data_manager.data_transfer.clone(), - ); - - Ok(()) - } else { - Err(ModelarDbServerError::InvalidState( - "Storage engine is not configured to transfer data.".to_owned(), - )) - } - } } diff --git a/crates/modelardb_server/tests/integration_test.rs b/crates/modelardb_server/tests/integration_test.rs index e8b99825..164bbb84 100644 --- a/crates/modelardb_server/tests/integration_test.rs +++ b/crates/modelardb_server/tests/integration_test.rs @@ -1425,7 +1425,6 @@ async fn test_can_get_configuration() { configuration.transfer_batch_size_in_bytes, Some(64 * 1024 * 1024) ); - assert_eq!(configuration.transfer_time_in_seconds, None); assert_eq!( configuration.segment_size_threshold_in_bytes, 64 * 1024 * 1024 @@ -1505,18 +1504,6 @@ async fn test_cannot_update_transfer_batch_size_in_bytes() { .await; } -#[tokio::test] -async fn test_cannot_update_transfer_time_in_seconds() { - // It is only possible to test that this fails since we cannot start the server with a - // remote data folder. - update_configuration_and_assert_error( - protocol::update_configuration::Setting::TransferTimeInSeconds as i32, - Some(1), - "Invalid State Error: Storage engine is not configured to transfer data.", - ) - .await; -} - #[tokio::test] async fn test_cannot_update_non_updatable_setting() { update_configuration_and_assert_error( diff --git a/crates/modelardb_types/src/flight/protocol.proto b/crates/modelardb_types/src/flight/protocol.proto index 58462cee..89dbbb9b 100644 --- a/crates/modelardb_types/src/flight/protocol.proto +++ b/crates/modelardb_types/src/flight/protocol.proto @@ -61,26 +61,24 @@ message Configuration { // Amount of memory to reserve for storing compressed data buffers. uint64 compressed_reserved_memory_in_bytes = 3; - // The number of bytes that are required before transferring a batch of data to the remote object store. + // The number of bytes that are required before transferring a batch of data to the remote object store. If not set, + // data is only transferred on an explicit flush. optional uint64 transfer_batch_size_in_bytes = 4; - // The number of seconds between each transfer of data to the remote object store. - optional uint64 transfer_time_in_seconds = 5; - // The approximate maximum size, in bytes, of a single WAL segment file before it is closed and a new one is started. - uint64 segment_size_threshold_in_bytes = 6; + uint64 segment_size_threshold_in_bytes = 5; // Number of threads to allocate for converting multivariate time series to univariate time series. - uint32 ingestion_threads = 7; + uint32 ingestion_threads = 6; // Number of threads to allocate for compressing univariate time series to segments. - uint32 compression_threads = 8; + uint32 compression_threads = 7; // Number of threads to allocate for writing segments to a local and/or remote data folder. - uint32 writer_threads = 9; + uint32 writer_threads = 8; // Whether the write-ahead log is enabled. - bool wal_enabled = 10; + bool wal_enabled = 9; } // Request to update the configuration of a ModelarDB node. @@ -90,8 +88,7 @@ message UpdateConfiguration { UNCOMPRESSED_RESERVED_MEMORY_IN_BYTES = 1; COMPRESSED_RESERVED_MEMORY_IN_BYTES = 2; TRANSFER_BATCH_SIZE_IN_BYTES = 3; - TRANSFER_TIME_IN_SECONDS = 4; - SEGMENT_SIZE_THRESHOLD_IN_BYTES = 5; + SEGMENT_SIZE_THRESHOLD_IN_BYTES = 4; } // Setting to update in the configuration. diff --git a/docs/user/README.md b/docs/user/README.md index ab7821ec..0dfd82f9 100644 --- a/docs/user/README.md +++ b/docs/user/README.md @@ -389,7 +389,6 @@ file. Variables marked with ✓ in the **Updatable** column can also be updated | `--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. | -| `--transfer-time-in-seconds` | `MODELARDBD_TRANSFER_TIME_IN_SECONDS` | Disabled | ✓ | The number of seconds between each transfer of data 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. | ## Docker