feat: produce snapshot chunks lazily

Author: praveen-influx

influxdb3/src/commands/serve.rs

@@ -579,6 +579,7 @@ pub async fn command(config: Config) -> Result<()> {
     background_buffer_checker(
         // config.force_snapshot_mem_threshold.bytes(),
         734003200,
+        // 536870912,
         &write_buffer_impl,
     )
     .await;

influxdb3_write/src/paths.rs

@@ -55,6 +55,7 @@ pub struct ParquetFilePath(ObjPath);
 impl ParquetFilePath {
     /// Generate a parquet file path using the given arguments. This will convert the provided
     /// `chunk_time` into a date time string with format `'YYYY-MM-DD/HH-MM'`
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         host_prefix: &str,
         db_name: &str,
@@ -63,14 +64,26 @@ impl ParquetFilePath {
         table_id: u32,
         chunk_time: i64,
         wal_file_sequence_number: WalFileSequenceNumber,
+        sub_chunk_index: Option<u64>,
     ) -> Self {
         let date_time = DateTime::<Utc>::from_timestamp_nanos(chunk_time);
-        let path = ObjPath::from(format!(
-            "{host_prefix}/dbs/{db_name}-{db_id}/{table_name}-{table_id}/{date_string}/{wal_seq:010}.{ext}",
-            date_string = date_time.format("%Y-%m-%d/%H-%M"),
-            wal_seq = wal_file_sequence_number.as_u64(),
-            ext = PARQUET_FILE_EXTENSION
-        ));
+        let path = if sub_chunk_index.is_some() {
+            ObjPath::from(format!(
+                "{host_prefix}/dbs/{db_name}-{db_id}/{table_name}-{table_id}/{date_string}/{wal_seq:010}-{chunk_idx}.{ext}",
+                date_string = date_time.format("%Y-%m-%d/%H-%M"),
+                wal_seq = wal_file_sequence_number.as_u64(),
+                chunk_idx = sub_chunk_index.unwrap(),
+                ext = PARQUET_FILE_EXTENSION
+            ))
+
+        } else {
+            ObjPath::from(format!(
+                "{host_prefix}/dbs/{db_name}-{db_id}/{table_name}-{table_id}/{date_string}/{wal_seq:010}.{ext}",
+                date_string = date_time.format("%Y-%m-%d/%H-%M"),
+                wal_seq = wal_file_sequence_number.as_u64(),
+                ext = PARQUET_FILE_EXTENSION
+            ))
+        };
         Self(path)
     }
 }
@@ -143,6 +156,7 @@ fn parquet_file_path_new() {
                 .timestamp_nanos_opt()
                 .unwrap(),
             WalFileSequenceNumber::new(1337),
+            None,
         ),
         ObjPath::from("my_host/dbs/my_db-0/my_table-0/2038-01-19/03-14/0000001337.parquet")
     );
@@ -162,6 +176,7 @@ fn parquet_file_percent_encoded() {
                 .timestamp_nanos_opt()
                 .unwrap(),
             WalFileSequenceNumber::new(100),
+            None,
         )
         .as_ref()
         .as_ref(),

influxdb3_write/src/persister.rs

@@ -968,6 +968,7 @@ mod tests {
             0,
             Utc::now().timestamp_nanos_opt().unwrap(),
             WalFileSequenceNumber::new(1),
+            None,
         );
         let (bytes_written, meta, _) = persister
             .persist_parquet_file(path.clone(), stream_builder.build())