feat(data-plane): columnar engine on the Planner's Arrow operators; compile once; per-sample panes from columns

Cargo.toml

@@ -10,8 +10,8 @@ version = "0.1.0"
 # The backend depends on two Planner crates, at one immutable revision:
 # `asap-planner` to plan and `asap-executor` to execute. Planner IR types
 # are reached through `asap_executor::planner`.
-asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" }
-asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" }
+asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "92bc1296fd33802e8ce3fcab6b75f8e3205c3960" }
+asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "92bc1296fd33802e8ce3fcab6b75f8e3205c3960" }
 
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

README.md

@@ -46,11 +46,12 @@ across the ingestion workers by a hash of their label set; a pane closes
 passes its end, and later samples for it are dropped. The first pane after
 startup is skipped because it may be partial. Each query, including every
 step of a range query, runs on one query thread; queries read the stored
-panes concurrently, by shared reference. Ingestion updates each open pane's summary
-state per sample, with the updater the Planner executor's batch build uses;
-a definition whose build has another shape (a row filter, a keyed family,
-operators between its range and its summary) is built from the pane's
-buffered samples instead. The data plane uses jemalloc as its allocator.
+panes concurrently, by shared reference. The plan is compiled once and shared by every thread. Ingestion updates
+each open pane's summary state per sample with the executor's
+`StateColumnUpdater`, and builds the pane from columns; a definition whose
+build has another shape (a row filter, a keyed family, operators between
+its range and its summary, groups that are not series) is built once over
+the pane's samples instead. Batches are Arrow columns throughout. The data plane uses jemalloc as its allocator.
 
 An instant query whose text is one of the plan's queries is answered by the
 plan (`x-asap-execution: plan`). Its stored inputs read the panes ending at

data_plane/src/engine.rs

@@ -1,10 +1,10 @@
 //! The data plane's threads: ingestion workers that build stored panes, and
 //! a pool of query threads that answer planned queries.
 //!
-//! Planner IR and the executor's compiled plans hold `Rc`s, so they cannot
-//! be shared across threads: every worker and query thread reads the plan
-//! and compiles its own. What threads share is plain data: the stored panes,
-//! the kept query-time results, and the watermarks.
+//! The plan is compiled once, when it is installed: its ingestion side
+//! ([`Ingestion`]) and its query roots ([`QueryEngine`]). Every thread
+//! shares them, with the stored panes, the kept query-time results and the
+//! watermarks.
 //!
 //! - **Ingestion.** Series are sharded by a hash of their label set; each
 //!   worker owns the open panes of its shard. A worker's watermark is its
@@ -22,7 +22,7 @@ use std::sync::{Arc, Mutex};
 use asap_planner::InstallablePlan;
 use tokio::sync::{mpsc, oneshot};
 
-use crate::precompute::{PrecomputeStats, RawSample, ShardWorker, Shared};
+use crate::precompute::{Ingestion, PrecomputeStats, RawSample, ShardWorker, Shared};
 use crate::query::{KeptResults, QueryEngine, RawRange, ResultSample, Unanswered};
 use crate::raw_buffer::RawBuffer;
 use crate::sds::Sds;
@@ -152,14 +152,16 @@ impl EngineHandle {
             horizons.values_mut().for_each(|ms| *ms = override_ms);
         }
         tracing::info!(?horizons, "local raw buffer");
+        let ingestion = Arc::new(Ingestion::new(&plan, &sds, true)?);
+        let kept = Arc::new(KeptResults::default());
+        let queries_engine = Arc::new(QueryEngine::with_kept(&plan, &sds, kept));
         drop(plan);
         let raw_buffer = (!horizons.is_empty()).then(|| {
             Arc::new(RawBuffer::with_horizons(
                 horizons,
                 threads.ingestion_workers,
             ))
         });
-        let plan_json = Arc::new(plan_json);
         let precompute = Arc::new(Shared::new(threads.ingestion_workers, allowed_lateness_ms));
         let (senders, receivers): (Vec<_>, Vec<_>) = (0..precompute.workers())
             .map(|_| mpsc::channel(QUEUE))
@@ -168,27 +170,11 @@ impl EngineHandle {
         // senders let the channels close when the handle is dropped.
         let peers: Vec<_> = senders.iter().map(mpsc::Sender::downgrade).collect();
         for (index, mut commands) in receivers.into_iter().enumerate() {
-            let (plan_json, shared, peers, raw_buffer) = (
-                plan_json.clone(),
-                precompute.clone(),
-                peers.clone(),
-                raw_buffer.clone(),
-            );
+            let mut worker = ShardWorker::new(index, ingestion.clone(), precompute.clone())
+                .with_raw_buffer(raw_buffer.clone());
+            let peers = peers.clone();
             spawn(format!("ingest-{index}"), move |ready| {
-                let started = install(&plan_json).and_then(|(plan, sds)| {
-                    Ok(ShardWorker::new(&plan, &sds, index, shared, true)?
-                        .with_raw_buffer(raw_buffer))
-                });
-                let mut worker = match started {
-                    Ok(worker) => {
-                        let _ = ready.send(Ok(()));
-                        worker
-                    }
-                    Err(error) => {
-                        let _ = ready.send(Err(error));
-                        return;
-                    }
-                };
+                let _ = ready.send(Ok(()));
                 while let Some(command) = commands.blocking_recv() {
                     match command {
                         WorkerCommand::Ingest(samples) => {
@@ -214,22 +200,9 @@ impl EngineHandle {
         }
         let (queries, jobs) = sync_channel(QUEUE);
         let jobs = Arc::new(Mutex::new(jobs));
-        let kept = Arc::new(KeptResults::default());
         for index in 0..threads.query_threads.max(1) {
-            let (plan_json, shared, kept, jobs) = (
-                plan_json.clone(),
-                precompute.clone(),
-                kept.clone(),
-                jobs.clone(),
-            );
+            let (engine, shared, jobs) = (queries_engine.clone(), precompute.clone(), jobs.clone());
             spawn(format!("query-{index}"), move |ready| {
-                let engine = match install(&plan_json) {
-                    Ok((plan, sds)) => QueryEngine::with_kept(&plan, &sds, kept),
-                    Err(error) => {
-                        let _ = ready.send(Err(error));
-                        return;
-                    }
-                };
                 let _ = ready.send(Ok(()));
                 serve_queries(&engine, &shared, &jobs);
             })?;

data_plane/src/lib.rs

@@ -17,6 +17,7 @@ pub mod raw_buffer;
 pub mod remote_read;
 pub mod remote_write;
 pub mod sds;
+mod series;
 
 use std::collections::BTreeMap;
 use std::sync::atomic::{AtomicU64, Ordering};