refactor: default to num_cpu

Cargo.lock

@@ -824,6 +824,7 @@ version = "0.1.0"
 dependencies = [
  "chrono",
  "influxdb_line_protocol",
+ "num_cpus",
  "observability_deps",
  "percent-encoding",
  "regex",
@@ -1278,6 +1279,7 @@ dependencies = [
  "data_types",
  "futures",
  "google_types",
+ "num_cpus",
  "observability_deps",
  "proc-macro2",
  "prost",

data_types/Cargo.toml

@@ -9,6 +9,7 @@ readme = "README.md"
 [dependencies] # In alphabetical order
 chrono = { version = "0.4", features = ["serde"] }
 influxdb_line_protocol = { path = "../influxdb_line_protocol" }
+num_cpus = "1.13.0"
 percent-encoding = "2.1.0"
 regex = "1.4"
 serde = { version = "1.0", features = ["rc", "derive"] }

data_types/src/database_rules.rs

@@ -118,7 +118,6 @@ pub const DEFAULT_MUB_ROW_THRESHOLD: usize = 100_000;
 pub const DEFAULT_PERSIST_ROW_THRESHOLD: usize = 1_000_000;
 pub const DEFAULT_PERSIST_AGE_THRESHOLD_SECONDS: u32 = 30 * 60;
 pub const DEFAULT_LATE_ARRIVE_WINDOW_SECONDS: u32 = 5 * 60;
-pub const DEFAULT_MAX_ACTIVE_COMPACTIONS: u32 = 14;
 
 /// Configures how data automatically flows through the system
 #[derive(Debug, Eq, PartialEq, Clone)]
@@ -185,7 +184,7 @@ impl Default for LifecycleRules {
             persist: false,
             immutable: false,
             worker_backoff_millis: NonZeroU64::new(DEFAULT_WORKER_BACKOFF_MILLIS).unwrap(),
-            max_active_compactions: NonZeroU32::new(DEFAULT_MAX_ACTIVE_COMPACTIONS).unwrap(),
+            max_active_compactions: NonZeroU32::new(num_cpus::get() as u32).unwrap(), // defaults to number of CPU threads
             catalog_transactions_until_checkpoint: NonZeroU64::new(
                 DEFAULT_CATALOG_TRANSACTIONS_UNTIL_CHECKPOINT,
             )

generated_types/Cargo.toml

@@ -13,6 +13,7 @@ data_types = { path = "../data_types" }
 futures = "0.3"
 google_types = { path = "../google_types" }
 observability_deps = { path = "../observability_deps" }
+num_cpus = "1.13.0"
 prost = "0.7"
 prost-types = "0.7"
 regex = "1.4"

generated_types/src/database_rules/lifecycle.rs

@@ -3,7 +3,7 @@ use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
 
 use data_types::database_rules::{
     LifecycleRules, DEFAULT_CATALOG_TRANSACTIONS_UNTIL_CHECKPOINT,
-    DEFAULT_LATE_ARRIVE_WINDOW_SECONDS, DEFAULT_MAX_ACTIVE_COMPACTIONS, DEFAULT_MUB_ROW_THRESHOLD,
+    DEFAULT_LATE_ARRIVE_WINDOW_SECONDS, DEFAULT_MUB_ROW_THRESHOLD,
     DEFAULT_PERSIST_AGE_THRESHOLD_SECONDS, DEFAULT_PERSIST_ROW_THRESHOLD,
     DEFAULT_WORKER_BACKOFF_MILLIS,
 };
@@ -52,7 +52,7 @@ impl TryFrom<management::LifecycleRules> for LifecycleRules {
             worker_backoff_millis: NonZeroU64::new(proto.worker_backoff_millis)
                 .unwrap_or_else(|| NonZeroU64::new(DEFAULT_WORKER_BACKOFF_MILLIS).unwrap()),
             max_active_compactions: NonZeroU32::new(proto.max_active_compactions)
-                .unwrap_or_else(|| NonZeroU32::new(DEFAULT_MAX_ACTIVE_COMPACTIONS).unwrap()),
+                .unwrap_or_else(|| NonZeroU32::new(num_cpus::get() as u32).unwrap()), // default to num CPU threads
             catalog_transactions_until_checkpoint: NonZeroU64::new(
                 proto.catalog_transactions_until_checkpoint,
             )