feat: schema mismatch alignment

This PR addresses the OOM issue (or reduces the chances of running into OOM when snapshotting) by doing following main changes
- defaults gen 1 duration to 1m (instead of 10m)
- snapshot chunks are built lazily and
- sort/dedupe step itself is done serially (i.e 1 at a time)

As an optimisation when _not_ forcing a snapshot it aggregates up to 10m worth of chunks and writes them in parallel assumption is given it's a normal snapshot, there is enough memory to run it.

closes: #25991

Author: praveen-influx

Cargo.lock

@@ -181,7 +181,7 @@ pub struct Config {
     #[clap(
         long = "gen1-duration",
         env = "INFLUXDB3_GEN1_DURATION",
-        default_value = "10m",
+        default_value = "1m",
         action
     )]
     pub gen1_duration: Gen1Duration,
@@ -577,9 +577,7 @@ pub async fn command(config: Config) -> Result<()> {
 
     info!("setting up background mem check for query buffer");
     background_buffer_checker(
-        // config.force_snapshot_mem_threshold.bytes(),
-        734003200,
-        // 536870912,
+        config.force_snapshot_mem_threshold.as_num_bytes(),
         &write_buffer_impl,
     )
     .await;

influxdb3/src/commands/serve.rs

@@ -207,6 +207,16 @@ impl Gen1Duration {
         self.0.as_nanos() as i64
     }
 
+    pub fn as_10m(&self) -> u64 {
+        let duration_secs = self.0.as_secs();
+        let ten_min_secs = 600;
+        if duration_secs >= ten_min_secs {
+            1
+        } else {
+            ten_min_secs / duration_secs
+        }
+    }
+
     pub fn new_1m() -> Self {
         Self(Duration::from_secs(60))
     }
@@ -237,7 +247,7 @@ impl Default for Gen1Duration {
 
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
 pub struct NoopDetails {
-    timestamp_ns: i64,
+    pub timestamp_ns: i64,
 }
 
 #[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]

influxdb3_wal/src/lib.rs

@@ -63,6 +63,7 @@ serde_json.workspace = true
 serde_with.workspace = true
 sha2.workspace = true
 snap.workspace = true
+sysinfo.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 url.workspace = true

influxdb3_write/Cargo.toml

@@ -55,7 +55,6 @@ 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,
@@ -64,26 +63,14 @@ 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 = 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
-            ))
-        };
+        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
+        ));
         Self(path)
     }
 }
@@ -156,7 +143,6 @@ 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")
     );
@@ -176,7 +162,6 @@ fn parquet_file_percent_encoded() {
                 .timestamp_nanos_opt()
                 .unwrap(),
             WalFileSequenceNumber::new(100),
-            None,
         )
         .as_ref()
         .as_ref(),

influxdb3_write/src/paths.rs

@@ -968,7 +968,6 @@ 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())

influxdb3_write/src/persister.rs

@@ -221,6 +221,8 @@ impl WriteBufferImpl {
             distinct_cache_provider: Arc::clone(&distinct_cache),
             persisted_files: Arc::clone(&persisted_files),
             parquet_cache: parquet_cache.clone(),
+            gen1_duration: wal_config.gen1_duration,
+            max_size_per_parquet_file_bytes: 50_000_000,
         }));
 
         // create the wal instance, which will replay into the queryable buffer and start
@@ -2721,10 +2723,7 @@ mod tests {
     #[test_log::test(tokio::test)]
     async fn test_out_of_order_data() {
         let tmp_dir = test_helpers::tmp_dir().unwrap();
-        debug!(
-            ?tmp_dir,
-            ">>> using tmp dir for test_check_mem_and_force_snapshot"
-        );
+        debug!(?tmp_dir, ">>> using tmp dir");
         let obj_store: Arc<dyn ObjectStore> =
             Arc::new(LocalFileSystem::new_with_prefix(tmp_dir).unwrap());
         let (write_buffer, _, _) = setup(
@@ -2795,6 +2794,9 @@ mod tests {
                 "| a    | us     | 1970-01-01T00:00:28Z | 10.0  |",
                 "| a    | us     | 1970-01-01T00:00:29Z | 10.0  |",
                 "| a    | us     | 1970-01-01T00:00:30Z | 10.0  |",
+                "| a    | us     | 1970-01-01T00:00:20Z | 10.0  |",
+                "| a    | us     | 1970-01-01T00:00:21Z | 10.0  |",
+                "| a    | us     | 1970-01-01T00:00:22Z | 10.0  |",
                 "| a    | us     | 1970-01-01T00:01:40Z | 10.0  |",
                 "| a    | us     | 1970-01-01T00:01:41Z | 10.0  |",
                 "| a    | us     | 1970-01-01T00:01:42Z | 10.0  |",
@@ -2807,9 +2809,6 @@ mod tests {
                 "| a    | us     | 1970-01-01T00:01:49Z | 10.0  |",
                 "| a    | us     | 1970-01-01T00:01:50Z | 10.0  |",
                 "| a    | us     | 1970-01-01T00:01:51Z | 10.0  |",
-                "| a    | us     | 1970-01-01T00:00:20Z | 10.0  |",
-                "| a    | us     | 1970-01-01T00:00:21Z | 10.0  |",
-                "| a    | us     | 1970-01-01T00:00:22Z | 10.0  |",
                 "+------+--------+----------------------+-------+",
             ],
             &actual