refactor: tidy ups

Author: praveen-influx

influxdb3/src/commands/serve.rs

@@ -1,7 +1,6 @@
 //! Entrypoint for InfluxDB 3 Core Server
 
 use anyhow::{Context, bail};
-use datafusion_util::config::register_iox_object_store;
 use futures::{FutureExt, future::FusedFuture, pin_mut};
 use influxdb3_authz::TokenAuthenticator;
 use influxdb3_cache::{
@@ -546,8 +545,9 @@ 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
+    // When you have extra executor, you need separate metrics registry! It is not clear what
+    // the impact would be
+    // TODO: confirm this is not going to mess up downstream metrics consumers
     let write_path_metrics = setup_metric_registry();
 
     // Install custom panic handler and forget about it.
@@ -646,6 +646,13 @@ pub async fn command(config: Config) -> Result<()> {
             metric_registry: Arc::clone(&write_path_metrics),
             // use as much memory for persistence, can this be UnboundedMemoryPool?
             mem_pool_size: usize::MAX,
+            // These are new additions, just skimming through the code it does not look like we can
+            // achieve the same effect as having a separate executor. It looks like it's for "all"
+            // queries, it'd be nice to have a filter to say when the query matches this pattern
+            // apply these limits. If that's possible maybe we could avoid creating a separate
+            // executor.
+            per_query_mem_pool_config: PerQueryMemoryPoolConfig::Disabled,
+            heap_memory_limit: None,
         },
         DedicatedExecutor::new(
             "datafusion_write_path",
@@ -656,9 +663,6 @@ pub async fn command(config: Config) -> Result<()> {
         ),
     ));
 
-    let runtime_env = exec.new_context().inner().runtime_env();
-    register_iox_object_store(runtime_env, parquet_store.id(), Arc::clone(&object_store));
-
     let trace_header_parser = TraceHeaderParser::new()
         .with_jaeger_trace_context_header_name(
             config

influxdb3_wal/src/lib.rs

@@ -21,7 +21,6 @@ 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;
@@ -558,26 +557,15 @@ pub fn background_wal_flush<W: Wal>(
                     {
                         let snapshot_wal = Arc::clone(&wal);
                         tokio::spawn(async move {
-                            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);
-                                }
-                            }
+                            // since we're using separate executor with unlimited memory,
+                            // the errors here will never be due to lack of resources. Only OS
+                            // (operating system) can OOM kill the whole process
+                            let snapshot_details = snapshot_complete.await.expect("snapshot failed");
+                            assert_eq!(snapshot_info, snapshot_details);
+
+                            snapshot_wal
+                                .cleanup_snapshot(snapshot_info, snapshot_permit)
+                                .await;
                         });
                     }
                 }