feat: introduce num wal files to keep

influxdb3/src/commands/serve.rs

@@ -208,6 +208,17 @@ pub struct Config {
     )]
     pub wal_max_write_buffer_size: usize,
 
+    /// Number of snapshotted wal files to retain in object store, wal flush does not clear
+    /// the wal files immediately instead they are only deleted when snapshotted and num wal files
+    /// count exceeds this size
+    #[clap(
+        long = "snapshotted-wal-files-to-keep",
+        env = "INFLUXDB3_NUM_WAL_FILES_TO_KEEP",
+        default_value = "300",
+        action
+    )]
+    pub snapshotted_wal_files_to_keep: u64,
+
     // TODO - tune this default:
     /// The size of the query log. Up to this many queries will remain in the log before
     /// old queries are evicted to make room for new ones.
@@ -502,6 +513,7 @@ pub async fn command(config: Config) -> Result<()> {
         wal_config,
         parquet_cache,
         metric_registry: Arc::clone(&metrics),
+        snapshotted_wal_files_to_keep: config.snapshotted_wal_files_to_keep,
     })
     .await
     .map_err(|e| Error::WriteBufferInit(e.into()))?;

influxdb3_processing_engine/src/lib.rs

@@ -865,6 +865,7 @@ mod tests {
             wal_config,
             parquet_cache: None,
             metric_registry: Arc::clone(&metric_registry),
+            snapshotted_wal_files_to_keep: 10,
         })
         .await
         .unwrap();

influxdb3_server/src/lib.rs

@@ -778,6 +778,7 @@ mod tests {
                 wal_config: WalConfig::test_config(),
                 parquet_cache: Some(parquet_cache),
                 metric_registry: Arc::clone(&metrics),
+                snapshotted_wal_files_to_keep: 100,
             },
         )
         .await

influxdb3_server/src/query_executor/mod.rs

@@ -713,6 +713,7 @@ mod tests {
             },
             parquet_cache: Some(parquet_cache),
             metric_registry: Default::default(),
+            snapshotted_wal_files_to_keep: 1,
         })
         .await
         .unwrap();