feat: Significantly decrease startup times for WAL

This commit does a few important things to speedup startup times:
1. We avoid changing an Arc<str> to a String with the series key as the
   From<String> impl will call with_column which will then turn it into
   an Arc<str> again. Instead we can just call `with_column` directly
   and pass in the iterator without also collecting into a Vec<String>
2. We switch to using bitcode as the serialization format for the WAL.
   This significantly reduces startup time as this format is faster to
   use instead of JSON, which was eating up massive amounts of time.
   Part of this change involves not using the tag feature of serde as
   it's currently not supported by bincode
3. We also parallelize reading and deserializing the WAL files before
   we then apply them in order. This reduces time waiting on IO and we
   eagerly evaluate each spawned task in order as much as possible.

This gives us about a 189% speedup over what we were doing before.

Closes #25534

Cargo.toml

@@ -50,6 +50,7 @@ async-trait = "0.1"
 backtrace = "0.3"
 base64 = "0.22.0"
 bimap = "0.6.3"
+bitcode = { version = "0.6.3", features = ["serde"] }
 byteorder = "1.3.4"
 bytes = "1.9"
 chrono = "0.4"

influxdb3_cache/src/last_cache/snapshots/influxdb3_cache__last_cache__tests__catalog_initialization.snap

@@ -1,6 +1,7 @@
 ---
 source: influxdb3_cache/src/last_cache/mod.rs
 expression: caches
+snapshot_kind: text
 ---
 [
   {
@@ -11,9 +12,7 @@ expression: caches
       0,
       1
     ],
-    "value_columns": {
-      "type": "all_non_key_columns"
-    },
+    "value_columns": "all_non_key_columns",
     "count": 1,
     "ttl": 600
   },
@@ -25,11 +24,12 @@ expression: caches
       6
     ],
     "value_columns": {
-      "type": "explicit",
-      "columns": [
-        8,
-        7
-      ]
+      "explicit": {
+        "columns": [
+          8,
+          7
+        ]
+      }
     },
     "count": 5,
     "ttl": 60
@@ -40,11 +40,12 @@ expression: caches
     "name": "test_cache_3",
     "key_columns": [],
     "value_columns": {
-      "type": "explicit",
-      "columns": [
-        9,
-        7
-      ]
+      "explicit": {
+        "columns": [
+          9,
+          7
+        ]
+      }
     },
     "count": 10,
     "ttl": 500

influxdb3_client/src/lib.rs

@@ -845,7 +845,7 @@ pub struct LastCacheCreatedResponse {
 /// A last cache will either store values for an explicit set of columns, or will accept all
 /// non-key columns
 #[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]
-#[serde(tag = "type", rename_all = "snake_case")]
+#[serde(rename_all = "snake_case")]
 pub enum LastCacheValueColumnsDef {
     /// Explicit list of column names
     Explicit { columns: Vec<u32> },
@@ -1220,8 +1220,9 @@ mod tests {
                     "name": "cache_name",
                     "key_columns": [0, 1],
                     "value_columns": {
-                        "type": "explicit",
-                        "columns": [2, 3]
+                        "explicit": {
+                            "columns": [2, 3]
+                        }
                     },
                     "ttl": 120,
                     "count": 5

influxdb3_server/src/lib.rs

@@ -773,6 +773,7 @@ mod tests {
     }
 
     async fn setup_server(start_time: i64) -> (String, CancellationToken, Arc<dyn WriteBuffer>) {
+        let server_start_time = tokio::time::Instant::now();
         let trace_header_parser = trace_http::ctx::TraceHeaderParser::new();
         let metrics = Arc::new(metric::Registry::new());
         let object_store: Arc<DynObjectStore> = Arc::new(object_store::memory::InMemory::new());
@@ -865,7 +866,7 @@ mod tests {
         let frontend_shutdown = CancellationToken::new();
         let shutdown = frontend_shutdown.clone();
 
-        tokio::spawn(async move { serve(server, frontend_shutdown).await });
+        tokio::spawn(async move { serve(server, frontend_shutdown, server_start_time).await });
 
         (format!("http://{addr}"), shutdown, write_buffer)
     }

influxdb3_wal/Cargo.toml

@@ -18,6 +18,7 @@ influxdb3_id = { path = "../influxdb3_id" }
 
 # crates.io dependencies
 async-trait.workspace = true
+bitcode.workspace = true
 bytes.workspace = true
 byteorder.workspace = true
 crc32fast.workspace  = true
@@ -27,7 +28,6 @@ indexmap.workspace = true
 object_store.workspace = true
 parking_lot.workspace = true
 serde.workspace = true
-serde_json.workspace = true
 serde_with.workspace = true
 thiserror.workspace = true
 tokio.workspace = true

influxdb3_wal/src/lib.rs

@@ -40,6 +40,9 @@ pub enum Error {
     #[error("deserialize error: {0}")]
     Serialize(#[from] crate::serialize::Error),
 
+    #[error("join error: {0}")]
+    Join(#[from] tokio::task::JoinError),
+
     #[error("object store error: {0}")]
     ObjectStoreError(#[from] ::object_store::Error),
 
@@ -426,7 +429,7 @@ impl LastCacheDefinition {
 /// A last cache will either store values for an explicit set of columns, or will accept all
 /// non-key columns
 #[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]
-#[serde(tag = "type", rename_all = "snake_case")]
+#[serde(rename_all = "snake_case")]
 pub enum LastCacheValueColumnsDef {
     /// Explicit list of column names
     Explicit { columns: Vec<ColumnId> },

influxdb3_wal/src/object_store.rs

@@ -104,9 +104,22 @@ impl WalObjectStore {
                 .last_snapshot_sequence_number()
         };
 
+        async fn get_contents(
+            object_store: Arc<dyn ObjectStore>,
+            path: Path,
+        ) -> Result<WalContents, crate::Error> {
+            let file_bytes = object_store.get(&path).await?.bytes().await?;
+            Ok(verify_file_type_and_deserialize(file_bytes)?)
+        }
+
+        let mut replay_tasks = Vec::new();
         for path in paths {
-            let file_bytes = self.object_store.get(&path).await?.bytes().await?;
-            let wal_contents = verify_file_type_and_deserialize(file_bytes)?;
+            let object_store = Arc::clone(&self.object_store);
+            replay_tasks.push(tokio::spawn(get_contents(object_store, path)));
+        }
+
+        for wal_contents in replay_tasks {
+            let wal_contents = wal_contents.await??;
 
             // add this to the snapshot tracker, so we know what to clear out later if the replay
             // was a wal file that had a snapshot
@@ -120,6 +133,7 @@ impl WalObjectStore {
                 ));
 
             match wal_contents.snapshot {
+                // This branch uses so much time
                 None => self.file_notifier.notify(wal_contents),
                 Some(snapshot_details) => {
                     let snapshot_info = {

influxdb3_wal/src/serialize.rs

@@ -17,8 +17,8 @@ pub enum Error {
     #[error("crc32 checksum mismatch")]
     Crc32Mismatch,
 
-    #[error("Serde error: {0}")]
-    Serde(#[from] serde_json::Error),
+    #[error("bitcode error: {0}")]
+    Bitcode(#[from] bitcode::Error),
 
     #[error("IO error: {0}")]
     Io(#[from] std::io::Error),
@@ -32,6 +32,7 @@ pub(crate) type Result<T, E = Error> = std::result::Result<T, E>;
 /// The first bytes written into a wal file to identify it and its version.
 const FILE_TYPE_IDENTIFIER: &[u8] = b"idb3.001";
 
+#[inline(always)]
 pub fn verify_file_type_and_deserialize(b: Bytes) -> Result<WalContents> {
     let contents = b.to_vec();
 
@@ -61,7 +62,7 @@ pub fn verify_file_type_and_deserialize(b: Bytes) -> Result<WalContents> {
     }
 
     // Deserialize the data into a WalContents
-    let contents: WalContents = serde_json::from_slice(data)?;
+    let contents: WalContents = bitcode::deserialize(data)?;
 
     Ok(contents)
 }
@@ -70,8 +71,8 @@ pub(crate) fn serialize_to_file_bytes(contents: &WalContents) -> Result<Vec<u8>>
     let mut buf = Vec::new();
     buf.extend_from_slice(FILE_TYPE_IDENTIFIER);
 
-    // serialize the contents into json bytes
-    let data = serde_json::to_vec(contents)?;
+    // serialize the contents into bitcode bytes
+    let data = bitcode::serialize(contents)?;
 
     // calculate the crc32 checksum
     let mut hasher = crc32fast::Hasher::new();

influxdb3_write/src/write_buffer/queryable_buffer.rs

@@ -513,11 +513,10 @@ impl BufferState {
                 let sort_key = table_def
                     .series_key
                     .iter()
-                    .map(|c| table_def.column_id_to_name_unchecked(c).to_string())
-                    .collect::<Vec<_>>();
+                    .map(|c| Arc::clone(&table_def.column_id_to_name_unchecked(c)));
                 let index_columns = table_def.index_column_ids();
 
-                TableBuffer::new(index_columns, SortKey::from(sort_key))
+                TableBuffer::new(index_columns, SortKey::from_columns(sort_key))
             });
             for (chunk_time, chunk) in table_chunks.chunk_time_to_chunk {
                 table_buffer.buffer_chunk(chunk_time, chunk.rows);