feat: separate out write path executor with unbounded memory limit

closes: #26422

Author: praveen-influx

influxdb3/src/commands/serve.rs

@@ -543,6 +543,19 @@ pub async fn command(config: Config) -> Result<()> {
     let f = SendPanicsToTracing::new_with_metrics(&metrics);
     std::mem::forget(f);
 
+    // hmmm when you have extra executor, you need extra metrics! This is expected to be a
+    // singleton
+    let write_path_metrics = setup_metric_registry();
+
+    // Install custom panic handler and forget about it.
+    //
+    // This leaks the handler and prevents it from ever being dropped during the
+    // lifetime of the program - this is actually a good thing, as it prevents
+    // the panic handler from being removed while unwinding a panic (which in
+    // turn, causes a panic - see #548)
+    let write_path_panic_handler_fn = SendPanicsToTracing::new_with_metrics(&write_path_metrics);
+    std::mem::forget(write_path_panic_handler_fn);
+
     // Construct a token to trigger clean shutdown
     let frontend_shutdown = CancellationToken::new();
     let shutdown_manager = ShutdownManager::new(frontend_shutdown.clone());
@@ -612,6 +625,30 @@ pub async fn command(config: Config) -> Result<()> {
             Arc::clone(&metrics),
         ),
     ));
+
+    // Note: using same metrics registry causes runtime panic.
+    let write_path_executor = Arc::new(Executor::new_with_config_and_executor(
+        ExecutorConfig {
+            // should this be divided? or should this contend for threads with executor that's
+            // setup for querying only
+            target_query_partitions: tokio_datafusion_config.num_threads.unwrap(),
+            object_stores: [&parquet_store]
+                .into_iter()
+                .map(|store| (store.id(), Arc::clone(store.object_store())))
+                .collect(),
+            metric_registry: Arc::clone(&write_path_metrics),
+            // use as much memory for persistence, can this be UnboundedMemoryPool?
+            mem_pool_size: usize::MAX,
+        },
+        DedicatedExecutor::new(
+            "datafusion_write_path",
+            tokio_datafusion_config
+                .builder()
+                .map_err(Error::TokioRuntime)?,
+            Arc::clone(&write_path_metrics),
+        ),
+    ));
+
     let runtime_env = exec.new_context().inner().runtime_env();
     register_iox_object_store(runtime_env, parquet_store.id(), Arc::clone(&object_store));
 
@@ -678,7 +715,7 @@ pub async fn command(config: Config) -> Result<()> {
         last_cache,
         distinct_cache,
         time_provider: Arc::<SystemProvider>::clone(&time_provider),
-        executor: Arc::clone(&exec),
+        executor: Arc::clone(&write_path_executor),
         wal_config,
         parquet_cache,
         metric_registry: Arc::clone(&metrics),

influxdb3_wal/src/lib.rs

@@ -17,10 +17,12 @@ use influxdb_line_protocol::v3::SeriesValue;
 use influxdb3_id::{ColumnId, DbId, SerdeVecMap, TableId};
 use influxdb3_shutdown::ShutdownToken;
 use iox_time::Time;
+use observability_deps::tracing::error;
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
 use std::cmp::Ordering;
 use std::fmt::Debug;
+use std::process;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
@@ -566,12 +568,26 @@ pub fn background_wal_flush<W: Wal>(
                     {
                         let snapshot_wal = Arc::clone(&wal);
                         tokio::spawn(async move {
-                            let snapshot_details = snapshot_complete.await.expect("snapshot failed");
-                            assert_eq!(snapshot_info, snapshot_details);
-
-                            snapshot_wal
-                                .cleanup_snapshot(snapshot_info, snapshot_permit)
-                                .await;
+                            let snapshot_details = snapshot_complete.await;
+                            match snapshot_details {
+                                Ok(snapshot_details) => {
+                                    if snapshot_info != snapshot_details {
+                                        drop(snapshot_permit);
+                                        panic!("snapshot details are different");
+                                    }
+                                    snapshot_wal
+                                        .cleanup_snapshot(snapshot_info, snapshot_permit)
+                                        .await;
+
+                                },
+                                Err(e) => {
+                                    drop(snapshot_permit);
+                                    error!(error = ?e, "snapshotting failed with error");
+                                    // shutdown the whole process if snapshot fails (defaulting to
+                                    // 137 as OOMs are the most common reason to fail at this point)
+                                    process::exit(137);
+                                }
+                            }
                         });
                     }
                 }