laminar_db/
show_commands.rs

1//! SHOW and DESCRIBE command builders; reopens `impl LaminarDB` to keep `db.rs` focused.
2
3use std::sync::Arc;
4
5use arrow::array::{BooleanArray, RecordBatch, StringArray, UInt64Array};
6use arrow::datatypes::{DataType, Field, Schema};
7
8use crate::db::LaminarDB;
9use crate::error::DbError;
10
11impl LaminarDB {
12    /// Build a SHOW CHECKPOINT STATUS metadata result. Public so the server's
13    /// `GET /api/v1/cluster/checkpoints` endpoint can surface it.
14    ///
15    /// # Errors
16    ///
17    /// Returns [`DbError::Checkpoint`] if the metadata batch cannot be assembled.
18    pub async fn build_show_checkpoint_status(&self) -> Result<RecordBatch, DbError> {
19        let coordinator = self.coordinator.lock().await;
20        let (cp_id, epoch, ts_ms, sources, sinks, completed_this_runtime) =
21            if let Some(coordinator) = coordinator.as_ref() {
22                let completed = coordinator.stats().completed;
23                coordinator.last_committed_manifest().map_or_else(
24                    || (0, 0, 0, String::new(), String::new(), completed),
25                    |manifest| {
26                        (
27                            manifest.checkpoint_id,
28                            manifest.epoch,
29                            manifest.timestamp_ms,
30                            manifest.source_names.join(", "),
31                            manifest.sink_names.join(", "),
32                            completed,
33                        )
34                    },
35                )
36            } else {
37                (0, 0, 0, String::new(), String::new(), 0)
38            };
39
40        let schema = Arc::new(Schema::new(vec![
41            Field::new("checkpoint_id", DataType::UInt64, false),
42            Field::new("epoch", DataType::UInt64, false),
43            Field::new("timestamp_ms", DataType::UInt64, false),
44            Field::new("sources", DataType::Utf8, false),
45            Field::new("sinks", DataType::Utf8, false),
46            Field::new("completed_this_runtime", DataType::UInt64, false),
47        ]));
48
49        RecordBatch::try_new(
50            schema,
51            vec![
52                Arc::new(UInt64Array::from(vec![cp_id])),
53                Arc::new(UInt64Array::from(vec![epoch])),
54                Arc::new(UInt64Array::from(vec![ts_ms])),
55                Arc::new(StringArray::from(vec![sources.as_str()])),
56                Arc::new(StringArray::from(vec![sinks.as_str()])),
57                Arc::new(UInt64Array::from(vec![completed_this_runtime])),
58            ],
59        )
60        .map_err(|e| DbError::Checkpoint(format!("failed to build checkpoint status batch: {e}")))
61    }
62
63    /// Build a SHOW MATERIALIZED VIEWS metadata result.
64    pub(crate) fn build_show_materialized_views(&self) -> RecordBatch {
65        let registry = self.mv_registry.lock();
66        let mut names = Vec::new();
67        let mut sqls = Vec::new();
68        let mut states = Vec::new();
69        for view in registry.views() {
70            let info = crate::handle::MaterializedViewInfo::from(view);
71            names.push(info.name);
72            sqls.push(info.sql);
73            states.push(info.state);
74        }
75        let names_ref: Vec<&str> = names.iter().map(String::as_str).collect();
76        let sqls_ref: Vec<&str> = sqls.iter().map(String::as_str).collect();
77        let states_ref: Vec<&str> = states.iter().map(String::as_str).collect();
78        let schema = Arc::new(Schema::new(vec![
79            Field::new("view_name", DataType::Utf8, false),
80            Field::new("sql", DataType::Utf8, false),
81            Field::new("state", DataType::Utf8, false),
82        ]));
83        RecordBatch::try_new(
84            schema,
85            vec![
86                Arc::new(StringArray::from(names_ref)),
87                Arc::new(StringArray::from(sqls_ref)),
88                Arc::new(StringArray::from(states_ref)),
89            ],
90        )
91        .expect("show materialized views: schema matches columns")
92    }
93
94    /// Build a SHOW SOURCES metadata result with connector metadata.
95    pub(crate) fn build_show_sources(&self) -> RecordBatch {
96        let sources = self.sources();
97        let mgr = self.connector_manager.lock();
98        let regs = mgr.sources();
99
100        let mut names = Vec::with_capacity(sources.len());
101        let mut connectors: Vec<Option<&str>> = Vec::with_capacity(sources.len());
102        let mut formats: Vec<Option<&str>> = Vec::with_capacity(sources.len());
103        let mut watermarks: Vec<Option<&str>> = Vec::with_capacity(sources.len());
104
105        for s in &sources {
106            names.push(s.name.as_str());
107            if let Some(reg) = regs.get(&s.name) {
108                connectors.push(reg.connector_type.as_deref());
109                formats.push(reg.format.as_deref());
110            } else {
111                connectors.push(None);
112                formats.push(None);
113            }
114            watermarks.push(s.watermark_column.as_deref());
115        }
116
117        let schema = Arc::new(Schema::new(vec![
118            Field::new("source_name", DataType::Utf8, false),
119            Field::new("connector", DataType::Utf8, true),
120            Field::new("format", DataType::Utf8, true),
121            Field::new("watermark_column", DataType::Utf8, true),
122        ]));
123        RecordBatch::try_new(
124            schema,
125            vec![
126                Arc::new(StringArray::from(names)),
127                Arc::new(StringArray::from(connectors)),
128                Arc::new(StringArray::from(formats)),
129                Arc::new(StringArray::from(watermarks)),
130            ],
131        )
132        .expect("show sources: schema matches columns")
133    }
134
135    /// Build a SHOW SINKS metadata result with connector metadata.
136    pub(crate) fn build_show_sinks(&self) -> RecordBatch {
137        let sinks = self.sinks();
138        let mgr = self.connector_manager.lock();
139        let regs = mgr.sinks();
140
141        let mut names = Vec::with_capacity(sinks.len());
142        let mut inputs: Vec<Option<String>> = Vec::with_capacity(sinks.len());
143        let mut connectors: Vec<Option<&str>> = Vec::with_capacity(sinks.len());
144        let mut formats: Vec<Option<&str>> = Vec::with_capacity(sinks.len());
145
146        for s in &sinks {
147            names.push(s.name.as_str());
148            let catalog_input = self.catalog.get_sink_input(&s.name);
149            if let Some(reg) = regs.get(&s.name) {
150                inputs.push(Some(reg.input.clone()));
151                connectors.push(reg.connector_type.as_deref());
152                formats.push(reg.format.as_deref());
153            } else {
154                inputs.push(catalog_input);
155                connectors.push(None);
156                formats.push(None);
157            }
158        }
159
160        let schema = Arc::new(Schema::new(vec![
161            Field::new("sink_name", DataType::Utf8, false),
162            Field::new("input", DataType::Utf8, true),
163            Field::new("connector", DataType::Utf8, true),
164            Field::new("format", DataType::Utf8, true),
165        ]));
166        RecordBatch::try_new(
167            schema,
168            vec![
169                Arc::new(StringArray::from(names)),
170                Arc::new(StringArray::from(inputs)),
171                Arc::new(StringArray::from(connectors)),
172                Arc::new(StringArray::from(formats)),
173            ],
174        )
175        .expect("show sinks: schema matches columns")
176    }
177
178    /// Build a SHOW QUERIES metadata result.
179    pub(crate) fn build_show_queries(&self) -> RecordBatch {
180        let queries = self.queries();
181        let ids: Vec<u64> = queries.iter().map(|q| q.id).collect();
182        let sqls: Vec<&str> = queries.iter().map(|q| q.sql.as_str()).collect();
183        let actives: Vec<bool> = queries.iter().map(|q| q.active).collect();
184        let schema = Arc::new(Schema::new(vec![
185            Field::new("query_id", DataType::UInt64, false),
186            Field::new("sql", DataType::Utf8, false),
187            Field::new("active", DataType::Boolean, false),
188        ]));
189        RecordBatch::try_new(
190            schema,
191            vec![
192                Arc::new(UInt64Array::from(ids)),
193                Arc::new(StringArray::from(sqls)),
194                Arc::new(BooleanArray::from(actives)),
195            ],
196        )
197        .expect("show queries: schema matches columns")
198    }
199
200    /// Build a SHOW STREAMS metadata result with SQL definitions.
201    pub(crate) fn build_show_streams(&self) -> RecordBatch {
202        let streams: Vec<String> = self
203            .catalog
204            .list_streams()
205            .into_iter()
206            .filter(|name| crate::sql_analysis::stage_owner(name).is_none())
207            .collect();
208        let mgr = self.connector_manager.lock();
209        let regs = mgr.streams();
210
211        let mut names = Vec::with_capacity(streams.len());
212        let mut sqls: Vec<Option<&str>> = Vec::with_capacity(streams.len());
213
214        for name in &streams {
215            names.push(name.as_str());
216            sqls.push(regs.get(name.as_str()).map(|r| r.query_sql.as_str()));
217        }
218
219        let schema = Arc::new(Schema::new(vec![
220            Field::new("stream_name", DataType::Utf8, false),
221            Field::new("sql", DataType::Utf8, true),
222        ]));
223        RecordBatch::try_new(
224            schema,
225            vec![
226                Arc::new(StringArray::from(names)),
227                Arc::new(StringArray::from(sqls)),
228            ],
229        )
230        .expect("show streams: schema matches columns")
231    }
232
233    /// Build a SHOW CREATE SOURCE result.
234    pub(crate) fn build_show_create_source(&self, name: &str) -> Result<RecordBatch, DbError> {
235        if self.catalog.get_source(name).is_none() {
236            return Err(DbError::SourceNotFound(name.to_string()));
237        }
238        let mgr = self.connector_manager.lock();
239        let ddl = mgr
240            .get_ddl(name)
241            .ok_or_else(|| DbError::InvalidOperation(format!("No stored DDL for source '{name}'")))?
242            .to_string();
243        drop(mgr);
244
245        let schema = Arc::new(Schema::new(vec![Field::new(
246            "create_statement",
247            DataType::Utf8,
248            false,
249        )]));
250        Ok(RecordBatch::try_new(
251            schema,
252            vec![Arc::new(StringArray::from(vec![ddl.as_str()]))],
253        )
254        .expect("show create source: schema matches columns"))
255    }
256
257    /// Build a SHOW CREATE SINK result.
258    pub(crate) fn build_show_create_sink(&self, name: &str) -> Result<RecordBatch, DbError> {
259        if self.catalog.get_sink_input(name).is_none() {
260            return Err(DbError::SinkNotFound(name.to_string()));
261        }
262        let mgr = self.connector_manager.lock();
263        let ddl = mgr
264            .get_ddl(name)
265            .ok_or_else(|| DbError::InvalidOperation(format!("No stored DDL for sink '{name}'")))?
266            .to_string();
267        drop(mgr);
268
269        let schema = Arc::new(Schema::new(vec![Field::new(
270            "create_statement",
271            DataType::Utf8,
272            false,
273        )]));
274        Ok(RecordBatch::try_new(
275            schema,
276            vec![Arc::new(StringArray::from(vec![ddl.as_str()]))],
277        )
278        .expect("show create sink: schema matches columns"))
279    }
280
281    /// Build a SHOW TABLES metadata result.
282    pub(crate) fn build_show_tables(&self) -> RecordBatch {
283        let ts = self.table_store.read();
284        let mut names = Vec::new();
285        let mut pks = Vec::new();
286        let mut row_counts = Vec::new();
287        let mut connectors = Vec::new();
288
289        for name in ts.table_names() {
290            let pk = ts.primary_key(&name).unwrap_or("").to_string();
291            let count = ts.table_row_count(&name) as u64;
292            let conn = ts.connector(&name).unwrap_or("").to_string();
293
294            names.push(name);
295            pks.push(pk);
296            row_counts.push(count);
297            connectors.push(conn);
298        }
299
300        let names_ref: Vec<&str> = names.iter().map(String::as_str).collect();
301        let pks_ref: Vec<&str> = pks.iter().map(String::as_str).collect();
302        let connectors_ref: Vec<&str> = connectors.iter().map(String::as_str).collect();
303
304        let schema = Arc::new(Schema::new(vec![
305            Field::new("name", DataType::Utf8, false),
306            Field::new("primary_key", DataType::Utf8, false),
307            Field::new("row_count", DataType::UInt64, false),
308            Field::new("connector", DataType::Utf8, false),
309        ]));
310
311        RecordBatch::try_new(
312            schema,
313            vec![
314                Arc::new(StringArray::from(names_ref)),
315                Arc::new(StringArray::from(pks_ref)),
316                Arc::new(UInt64Array::from(row_counts)),
317                Arc::new(StringArray::from(connectors_ref)),
318            ],
319        )
320        .expect("show tables: schema matches columns")
321    }
322
323    /// Build a DESCRIBE result.
324    pub(crate) fn build_describe(&self, name: &str) -> Result<RecordBatch, DbError> {
325        let committed = {
326            let manager = self.connector_manager.lock();
327            manager.schema_binding(name).cloned().map(|binding| {
328                let generation = manager
329                    .ordered_ddl()
330                    .into_iter()
331                    .find(|(object, _, _)| object == name)
332                    .map_or(1, |(_, _, generation)| generation);
333                (binding, generation)
334            })
335        };
336        if let Some((binding, generation)) = committed {
337            return schema_binding_describe_batch(&binding, generation);
338        }
339        let schema = if let Some(s) = self.catalog.describe_source(name) {
340            s
341        } else if let Some(s) = self.table_store.read().table_schema(name) {
342            s
343        } else if let Some(s) = self
344            .mv_registry
345            .lock()
346            .get(name)
347            .map(|mv| mv.schema.clone())
348        {
349            s
350        } else if self.catalog.list_sinks().contains(&name.to_string()) {
351            return Err(DbError::InvalidOperation(
352                "DESCRIBE is not supported for sinks. Use SHOW SINKS for details.".to_string(),
353            ));
354        } else if self.catalog.list_streams().contains(&name.to_string()) {
355            return Err(DbError::InvalidOperation(format!(
356                "Stream '{name}' exists but schema is only available after pipeline start"
357            )));
358        } else {
359            return Err(DbError::TableNotFound(name.to_string()));
360        };
361
362        schema_to_describe_batch(&schema)
363    }
364}
365
366/// Convert an Arrow schema to a DESCRIBE result `RecordBatch`.
367pub(crate) fn schema_to_describe_batch(schema: &Schema) -> Result<RecordBatch, DbError> {
368    let col_names: Vec<String> = schema.fields().iter().map(|f| f.name().clone()).collect();
369    let col_types: Vec<String> = schema
370        .fields()
371        .iter()
372        .map(|f| format!("{}", f.data_type()))
373        .collect();
374    let col_nullable: Vec<bool> = schema.fields().iter().map(|f| f.is_nullable()).collect();
375
376    let names_ref: Vec<&str> = col_names.iter().map(String::as_str).collect();
377    let types_ref: Vec<&str> = col_types.iter().map(String::as_str).collect();
378
379    let result_schema = Arc::new(Schema::new(vec![
380        Field::new("column_name", DataType::Utf8, false),
381        Field::new("data_type", DataType::Utf8, false),
382        Field::new("nullable", DataType::Boolean, false),
383    ]));
384
385    RecordBatch::try_new(
386        result_schema,
387        vec![
388            Arc::new(StringArray::from(names_ref)),
389            Arc::new(StringArray::from(types_ref)),
390            Arc::new(BooleanArray::from(col_nullable)),
391        ],
392    )
393    .map_err(|e| DbError::InvalidOperation(format!("describe metadata: {e}")))
394}
395
396fn schema_binding_describe_batch(
397    binding: &laminar_core::schema_binding::SchemaBinding,
398    generation: u64,
399) -> Result<RecordBatch, DbError> {
400    let base = schema_to_describe_batch(&binding.logical)?;
401    let origin = match binding.origin {
402        laminar_core::schema_binding::SchemaOrigin::Explicit => "explicit",
403        laminar_core::schema_binding::SchemaOrigin::BuiltIn => "built_in",
404        laminar_core::schema_binding::SchemaOrigin::Metadata => "metadata",
405        laminar_core::schema_binding::SchemaOrigin::Query => "query",
406        laminar_core::schema_binding::SchemaOrigin::Sample => "sample",
407    };
408    let identity = serde_json::to_string(&serde_json::json!({
409        "value": binding.value.as_ref().map(|native| (&native.format, &native.identity)),
410        "key": binding.key.as_ref().map(|native| (&native.format, &native.identity)),
411        "representation_version": binding.version,
412        "control_fields": binding.control_fields,
413        "policy": "frozen logical contract; exact field names; external preparation requires explicit configuration"
414    })).map_err(|error| DbError::Config(error.to_string()))?;
415    let fingerprint = binding
416        .fingerprint()
417        .map_err(|error| DbError::Config(error.to_string()))?;
418    let mappings = binding.logical.fields().iter().map(|field| {
419        binding
420            .mapping
421            .iter()
422            .find(|mapping| mapping.logical == *field.name())
423            .map(|mapping| mapping.external.as_str())
424    });
425    let mut fields = base.schema().fields().to_vec();
426    fields.extend([
427        Field::new("schema_origin", DataType::Utf8, false).into(),
428        Field::new("contract_generation", DataType::UInt64, false).into(),
429        Field::new("external_field", DataType::Utf8, true).into(),
430        Field::new("native_contract", DataType::Utf8, true).into(),
431        Field::new("schema_fingerprint", DataType::Utf8, true).into(),
432    ]);
433    let count = base.num_rows();
434    let mut columns = base.columns().to_vec();
435    columns.extend([
436        Arc::new(StringArray::from(vec![origin; count])) as arrow::array::ArrayRef,
437        Arc::new(UInt64Array::from(vec![generation; count])),
438        Arc::new(StringArray::from_iter(mappings)),
439        Arc::new(StringArray::from_iter(
440            (0..count).map(|row| (row == 0).then_some(identity.as_str())),
441        )),
442        Arc::new(StringArray::from_iter(
443            (0..count).map(|row| (row == 0).then_some(fingerprint.as_str())),
444        )),
445    ]);
446    RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)
447        .map_err(|error| DbError::Config(error.to_string()))
448}
laminar_db/show_commands.rs

laminar_db/
show_commands.rs
