laminar_db/db/
committed_read.rs

1//! Queries that read every materialized view at one committed checkpoint.
2
3use std::time::Instant;
4
5use arrow::array::RecordBatch;
6use arrow::datatypes::SchemaRef;
7
8use super::LaminarDB;
9use crate::error::DbError;
10use crate::mv_store::{CommittedRead, CommittedReadError, VIEW_READ_DEADLINE};
11
12/// Rows of a committed read, and the committed cut every view in it was read at.
13#[derive(Debug, Clone)]
14pub struct CommittedQuery {
15    /// Epoch of the cut.
16    pub epoch: u64,
17    /// Checkpoint that committed the cut.
18    pub checkpoint_id: u64,
19    /// When the cut was taken, in Unix milliseconds; the result is as old as now minus this.
20    pub taken_at_ms: u64,
21    /// Schema of the result.
22    pub schema: SchemaRef,
23    /// Result rows.
24    pub batches: Vec<RecordBatch>,
25}
26
27impl LaminarDB {
28    /// Run the query `sql` with every materialized view it scans read at the one committed
29    /// checkpoint `at` resolves to.
30    ///
31    /// The result never mixes checkpoints and is never rolled back, and a later
32    /// [`CommittedRead::Latest`] read on this database never sees an earlier epoch. It is as fresh
33    /// as the checkpoint cadence. Pass a returned epoch back as [`CommittedRead::AtLeast`] to read
34    /// no earlier than a previous result. [`CommittedRead::Recent`] waits until the newest cut
35    /// holds every record the views' sources had upstream when the read arrived, so a client
36    /// reads what it wrote to a source; every source behind the views must report its upstream
37    /// position (reference tables are looked up and are not waited for). An append view with a
38    /// row-count retention is not exact: which rows a cap keeps depends on arrival order.
39    ///
40    /// The statement may scan only materialized views. The checkpoint is held against retention
41    /// for at most 30 seconds, which bounds the read, including any wait for
42    /// [`CommittedRead::AtLeast`] or [`CommittedRead::Recent`]. A cluster node reads every participant's committed state, so
43    /// any node serves any view; it reads only cuts at or after the durable pin it holds. A node
44    /// outside the checkpoint participants (draining, or not yet assigned) learns of each cut of
45    /// its own topology from the cluster authority within about a second.
46    ///
47    /// # Errors
48    /// [`DbError::CommittedRead`] when checkpointing is disabled, the requested epoch has not
49    /// committed in time or is no longer retained, the statement scans live state, a recent
50    /// read's source cannot report or failed to report its upstream position, or the read passes
51    /// its deadline; any planning or execution error of `sql`.
52    pub async fn query_committed(
53        &self,
54        sql: &str,
55        at: CommittedRead,
56    ) -> Result<CommittedQuery, DbError> {
57        self.query_committed_until(sql, at, Instant::now() + VIEW_READ_DEADLINE)
58            .await
59    }
60
61    pub(crate) async fn query_committed_until(
62        &self,
63        sql: &str,
64        at: CommittedRead,
65        deadline: Instant,
66    ) -> Result<CommittedQuery, DbError> {
67        if self.shutdown.load(std::sync::atomic::Ordering::Relaxed) {
68            return Err(DbError::Shutdown);
69        }
70        let (cut, stream) = self.committed_stream(sql, at, deadline).await?;
71        let schema = stream.schema();
72        let batches = tokio::time::timeout_at(
73            tokio::time::Instant::from_std(deadline),
74            futures::TryStreamExt::try_collect::<Vec<_>>(stream),
75        )
76        .await
77        .map_err(|_| CommittedReadError::DeadlineExceeded { epoch: cut.epoch })?
78        .map_err(crate::query_rows::execution_error)?;
79        Ok(CommittedQuery {
80            epoch: cut.epoch,
81            checkpoint_id: cut.checkpoint_id,
82            taken_at_ms: cut.taken_at_ms,
83            schema,
84            batches,
85        })
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use std::sync::Arc;
92    use std::time::Duration;
93
94    use arrow::array::{Array, Int64Array};
95    use laminar_core::streaming::StreamCheckpointConfig;
96
97    use super::*;
98    use crate::config::LaminarConfig;
99
100    async fn pipeline(dir: &std::path::Path) -> Arc<LaminarDB> {
101        let db = LaminarDB::open_with_config(LaminarConfig {
102            storage_dir: Some(dir.to_path_buf()),
103            checkpoint: Some(StreamCheckpointConfig {
104                interval_ms: None,
105                ..StreamCheckpointConfig::default()
106            }),
107            ..LaminarConfig::default()
108        })
109        .unwrap();
110        for sql in [
111            "CREATE SOURCE events (k BIGINT)",
112            "CREATE MATERIALIZED VIEW tail AS SELECT k FROM events WITH (retention = '10 rows')",
113            "CREATE MATERIALIZED VIEW counts AS SELECT k, COUNT(*) AS n FROM events GROUP BY k",
114        ] {
115            db.execute(sql).await.unwrap();
116        }
117        db.start().await.unwrap();
118        db
119    }
120
121    async fn push_and_checkpoint(db: &LaminarDB, k: i64) -> u64 {
122        let batch = RecordBatch::try_from_iter_with_nullable(vec![(
123            "k",
124            Arc::new(Int64Array::from(vec![k])) as arrow::array::ArrayRef,
125            true,
126        )])
127        .unwrap();
128        db.source_untyped("events")
129            .unwrap()
130            .push_arrow(batch)
131            .unwrap();
132        tokio::time::sleep(Duration::from_millis(200)).await;
133        let result = db.checkpoint().await.unwrap();
134        assert!(result.success, "{:?}", result.error);
135        result.epoch
136    }
137
138    fn manifest_exists(dir: &std::path::Path, checkpoint_id: u64) -> bool {
139        let suffix = std::path::Path::new(&format!("{checkpoint_id:020}")).join("manifest.json");
140        let mut pending = vec![dir.to_path_buf()];
141        while let Some(path) = pending.pop() {
142            let Ok(entries) = std::fs::read_dir(&path) else {
143                continue;
144            };
145            for entry in entries.flatten() {
146                let path = entry.path();
147                if path.ends_with(&suffix) {
148                    return true;
149                }
150                if path.is_dir() {
151                    pending.push(path);
152                }
153            }
154        }
155        false
156    }
157
158    fn keys(batches: &[RecordBatch]) -> Vec<i64> {
159        let mut keys: Vec<i64> = batches
160            .iter()
161            .flat_map(|batch| {
162                let column = batch
163                    .column(0)
164                    .as_any()
165                    .downcast_ref::<Int64Array>()
166                    .unwrap();
167                (0..column.len())
168                    .map(|row| column.value(row))
169                    .collect::<Vec<_>>()
170            })
171            .collect();
172        keys.sort_unstable();
173        keys
174    }
175
176    #[tokio::test]
177    async fn a_pinned_cut_outlives_later_checkpoints_until_it_is_released() {
178        let dir = tempfile::tempdir().unwrap();
179        let db = pipeline(dir.path()).await;
180        let first = push_and_checkpoint(&db, 1).await;
181        let views = db.mv_store.read().committed_views().unwrap();
182        let held = views
183            .pin(
184                CommittedRead::Exact(first),
185                Instant::now() + VIEW_READ_DEADLINE,
186            )
187            .await
188            .unwrap();
189        // Each checkpoint rewrites key 1's frame, so nothing later references the first one.
190        push_and_checkpoint(&db, 1).await;
191        push_and_checkpoint(&db, 1).await;
192        tokio::time::sleep(Duration::from_millis(300)).await;
193
194        let pinned = held.checkpoint_id;
195        assert!(
196            manifest_exists(dir.path(), pinned),
197            "retention keeps the pinned cut"
198        );
199        let shape = db.mv_store.read().view_shape("counts").unwrap();
200        let rows = held.read_view(&views, &shape, &[]).await.unwrap();
201        let counts = rows[0]
202            .column(1)
203            .as_any()
204            .downcast_ref::<Int64Array>()
205            .unwrap();
206        assert_eq!(counts.values().to_vec(), [1]);
207
208        drop(held);
209        push_and_checkpoint(&db, 1).await;
210        for _ in 0..50 {
211            if !manifest_exists(dir.path(), pinned) {
212                break;
213            }
214            tokio::time::sleep(Duration::from_millis(100)).await;
215        }
216        assert!(
217            !manifest_exists(dir.path(), pinned),
218            "a released cut is retired"
219        );
220        let mut error = None;
221        for _ in 0..50 {
222            match db
223                .query_committed("SELECT n FROM counts", CommittedRead::Exact(first))
224                .await
225            {
226                Ok(_) => tokio::time::sleep(Duration::from_millis(100)).await,
227                Err(other) => {
228                    error = Some(other);
229                    break;
230                }
231            }
232        }
233        assert!(
234            matches!(
235                error,
236                Some(DbError::CommittedRead(CommittedReadError::Retired { requested }))
237                    if requested == first
238            ),
239            "{error:?}"
240        );
241        db.shutdown().await.unwrap();
242    }
243
244    #[tokio::test]
245    async fn a_declared_freshness_survives_a_pipeline_restart() {
246        let dir = tempfile::tempdir().unwrap();
247        let db = LaminarDB::open_with_config(LaminarConfig {
248            storage_dir: Some(dir.path().to_path_buf()),
249            checkpoint: Some(StreamCheckpointConfig {
250                interval_ms: Some(1_000),
251                ..StreamCheckpointConfig::default()
252            }),
253            ..LaminarConfig::default()
254        })
255        .unwrap();
256        db.execute("CREATE SOURCE events (k BIGINT)").await.unwrap();
257        db.execute(
258            "CREATE MATERIALIZED VIEW counts AS SELECT k, COUNT(*) AS n FROM events GROUP BY k \
259             WITH (freshness = '5 seconds')",
260        )
261        .await
262        .unwrap();
263        db.start().await.unwrap();
264        db.stop_pipeline().await.unwrap();
265        db.start().await.unwrap();
266        let contracts = db.mv_store.read().freshness_contracts();
267        assert_eq!(
268            contracts.declared_freshness("counts"),
269            Some(Duration::from_secs(5))
270        );
271        db.stop_pipeline().await.unwrap();
272        db.execute("DROP MATERIALIZED VIEW counts").await.unwrap();
273        assert_eq!(contracts.declared_freshness("counts"), None);
274        db.shutdown().await.unwrap();
275    }
276
277    #[tokio::test]
278    async fn a_read_waits_for_its_epoch_until_the_deadline() {
279        let dir = tempfile::tempdir().unwrap();
280        let db = pipeline(dir.path()).await;
281        let first = push_and_checkpoint(&db, 1).await;
282        let error = db
283            .query_committed_until(
284                "SELECT k FROM tail",
285                CommittedRead::AtLeast(first + 100),
286                Instant::now() + Duration::from_millis(200),
287            )
288            .await
289            .unwrap_err();
290        assert!(
291            matches!(
292                error,
293                DbError::CommittedRead(CommittedReadError::NotCommitted {
294                    requested: Some(requested),
295                    latest: Some(latest),
296                }) if requested == first + 100 && latest == first
297            ),
298            "{error:?}"
299        );
300
301        let writer = Arc::clone(&db);
302        let next = tokio::spawn(async move {
303            tokio::time::sleep(Duration::from_millis(100)).await;
304            push_and_checkpoint(&writer, 2).await
305        });
306        let read = db
307            .query_committed("SELECT k FROM tail", CommittedRead::AtLeast(first + 1))
308            .await
309            .unwrap();
310        let second = next.await.unwrap();
311        assert_eq!(read.epoch, second);
312        assert_eq!(keys(&read.batches), [1, 2]);
313        db.shutdown().await.unwrap();
314    }
315}
laminar_db/db/committed_read.rs

laminar_db/db/
committed_read.rs
