Remove periodic data transfer

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<u64>,
-    /// 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<u64>,
     /// 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<u64> {
-        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<u64>,
-        storage_engine: Arc<RwLock<StorageEngine>>,
-    ) -> 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();

crates/modelardb_server/src/main.rs

@@ -74,15 +74,10 @@ pub(crate) struct ServerArgs {
     compressed_reserved_memory_in_bytes: Option<u64>,
 
     /// 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<u64>,
 
-    /// 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<u64>,
-
     /// 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.

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)?;
 

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<String, u64>,
     /// 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<u64>,
-    /// 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<TaskJoinHandle<()>>,
     /// Tables that have been dropped and should not be transferred to the remote data folder.
     dropped_tables: HashSet<String>,
 }
@@ -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<u64>,
-        data_transfer: Arc<RwLock<Option<Self>>>,
-    ) {
-        // 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

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<u64>,
-    ) -> 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(),
-            ))
-        }
-    }
 }

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(

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.

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