hawdb_system_sql/
lib.rs

1// Copyright 2026 Nowledge
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Virtual `system`, `information_schema`, and `pg_catalog` query execution.
16//!
17//! This crate owns the bounded, storage-backed system-catalog query engine.
18//! The embedded facade supplies a snapshot of its live state and retains all
19//! database lifecycle and transaction coordination.
20
21use hawdb_core::time::{SystemTime, UNIX_EPOCH};
22use hawdb_core::{
23    Catalog, ConstraintKind, ConstraintSubject, GraphStatistics, HawDBError, IndexKind,
24    IndexStatisticsSample, LabelId, PropertyType, RelTypeId, Result, RuntimeCapabilities,
25    SchemaObjectState, TableKind, Value,
26};
27use hawdb_executor::{binding::map_payload_bytes, QueryOutput, Row, VectorExecutionReport};
28use hawdb_plan_cache::PlanCacheStats;
29use hawdb_query::QueryIdentity;
30use hawdb_sql::{Expr, ExprKind};
31use hawdb_sql::{
32    SelectProjection, SelectStatement, SqlBound, SqlColumnRef, SqlComparisonOp, SqlOrderDirection,
33    SqlOrderItem, SqlPredicate, SqlStatement, SqlValue,
34};
35use hawdb_storage::{
36    append_table::{AppendOrderMode, AppendState, AppendStorageResidencyReport},
37    config::{RelationalIndexMode, StorageResidencyMode},
38    projection::{ProjectedGraphStatus, SearchProjectionChangefeedStatus},
39    relational::{
40        RelationalColumnDefault, RelationalScalarType, RelationalState, RelationalTableSchema,
41        RelationalValue,
42    },
43};
44use std::cmp::Ordering;
45use std::collections::{BTreeMap, VecDeque};
46
47#[doc(hidden)]
48pub const DEFAULT_SLOW_QUERY_LOG_CAPACITY: usize = 256;
49#[doc(hidden)]
50pub const DEFAULT_SLOW_QUERY_LOG_THRESHOLD_MICROS: u128 = 300_000;
51#[doc(hidden)]
52pub const DEFAULT_STATEMENT_SUMMARY_CAPACITY: usize = 256;
53pub const SLOW_QUERY_LOG_EVENT_PROTOCOL: &str = "hawdb-slow-query-log-event-v1";
54const MAX_SLOW_QUERY_TEXT_BYTES: usize = 4096;
55const MAX_STATEMENT_TEXT_BYTES: usize = 4096;
56const MAX_STATEMENT_ERROR_BYTES: usize = 1024;
57
58#[derive(Debug, Clone, Default, PartialEq, Eq)]
59pub struct SlowQueryLogExportOptions {
60    pub include_query_text: bool,
61}
62
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub struct SlowQueryLogRecordSummary {
65    pub sequence: u64,
66    pub query_language: String,
67    pub statement_kind: String,
68    pub query_digest: String,
69    pub started_unix_micros: i64,
70    pub elapsed_micros: i64,
71    pub row_count: i64,
72    pub success: bool,
73    pub slow_log_candidate: bool,
74    pub access_control_policy_epoch: Option<u64>,
75}
76
77#[derive(Debug, Clone, PartialEq, Eq)]
78#[doc(hidden)]
79pub struct SlowQueryRecord {
80    pub sequence: u64,
81    pub query_language: String,
82    pub statement_kind: String,
83    pub query_digest: String,
84    pub query_text_hash: String,
85    pub query_text: String,
86    pub started_unix_micros: i64,
87    pub elapsed_micros: i64,
88    pub row_count: i64,
89    pub success: bool,
90    pub error: Option<String>,
91    pub slow_log_candidate: bool,
92    pub access_control_policy_epoch: Option<u64>,
93    pub vector_execution_reports: Vec<VectorExecutionReport>,
94}
95
96#[doc(hidden)]
97pub struct SlowQueryCompletion<'a> {
98    pub query_language: &'a str,
99    pub statement_kind: &'a str,
100    pub query_text: &'a str,
101    pub query_identity: &'a QueryIdentity,
102    pub elapsed_micros: u128,
103    pub row_count: usize,
104    pub success: bool,
105    pub error: Option<String>,
106    pub slow_log_candidate: bool,
107    pub access_control_policy_epoch: Option<u64>,
108    pub vector_execution_reports: Vec<VectorExecutionReport>,
109}
110
111#[derive(Debug, Clone, PartialEq, Eq)]
112#[doc(hidden)]
113pub struct SlowQueryLog {
114    capacity: usize,
115    next_sequence: u64,
116    records: VecDeque<SlowQueryRecord>,
117}
118
119#[derive(Debug, Clone, PartialEq, Eq)]
120#[doc(hidden)]
121pub struct StatementExecution {
122    pub query_language: String,
123    pub query_text: String,
124    pub statement_kind: String,
125    pub query_digest: String,
126    pub query_text_hash: String,
127    pub elapsed_micros: i64,
128    pub row_count: i64,
129    pub success: bool,
130    pub error: Option<String>,
131}
132
133#[derive(Debug, Clone, PartialEq, Eq)]
134#[doc(hidden)]
135pub struct StatementSummaryRecord {
136    pub digest: String,
137    pub query_language: String,
138    pub query_text: String,
139    pub sample_query_text_hash: String,
140    pub statement_kind: String,
141    pub execution_count: i64,
142    pub success_count: i64,
143    pub error_count: i64,
144    pub total_elapsed_micros: i64,
145    pub max_elapsed_micros: i64,
146    pub total_row_count: i64,
147    pub last_seen_unix_micros: i64,
148    pub last_elapsed_micros: i64,
149    pub last_row_count: i64,
150    pub last_success: bool,
151    pub last_error: Option<String>,
152}
153
154#[derive(Debug, Clone, PartialEq, Eq)]
155#[doc(hidden)]
156pub struct StatementSummary {
157    capacity: usize,
158    records: BTreeMap<String, StatementSummaryRecord>,
159    insertion_order: VecDeque<String>,
160}
161
162/// Supplies the storage state needed by bounded virtual catalog queries.
163///
164/// The embedded facade owns the concrete store and adapts it at this boundary.
165#[doc(hidden)]
166pub trait SystemSqlStore {
167    fn commit_epoch(&self) -> u64;
168    fn append_storage_residency_report(&self) -> AppendStorageResidencyReport;
169    fn statistics(&self, catalog: &Catalog) -> GraphStatistics;
170    fn projected_graph_statuses(&self) -> Vec<ProjectedGraphStatus>;
171    fn search_projection_changefeed_status(&self) -> SearchProjectionChangefeedStatus;
172}
173
174#[doc(hidden)]
175pub struct SystemSqlContext<'a, Store: SystemSqlStore> {
176    pub catalog: &'a Catalog,
177    pub store: &'a Store,
178    pub relational_state: &'a RelationalState,
179    pub append_state: &'a AppendState,
180    pub runtime: SystemRuntimeSnapshot,
181    pub plan_cache_stats: &'a PlanCacheStats,
182    pub slow_queries: &'a [SlowQueryRecord],
183    pub statement_summaries: &'a [StatementSummaryRecord],
184}
185
186#[derive(Debug, Clone, Copy, PartialEq, Eq)]
187#[doc(hidden)]
188pub struct SystemRuntimeSnapshot {
189    read_only: bool,
190    max_read_result_rows: Option<usize>,
191    max_read_result_payload_bytes: Option<usize>,
192    storage_residency_mode: StorageResidencyMode,
193    relational_index_mode: RelationalIndexMode,
194    runtime_capabilities: RuntimeCapabilities,
195}
196
197impl SystemRuntimeSnapshot {
198    pub fn new(
199        read_only: bool,
200        max_read_result_rows: Option<usize>,
201        max_read_result_payload_bytes: Option<usize>,
202        storage_residency_mode: StorageResidencyMode,
203        relational_index_mode: RelationalIndexMode,
204        runtime_capabilities: RuntimeCapabilities,
205    ) -> Self {
206        Self {
207            read_only,
208            max_read_result_rows,
209            max_read_result_payload_bytes,
210            storage_residency_mode,
211            relational_index_mode,
212            runtime_capabilities,
213        }
214    }
215}
216
217#[derive(Debug, Clone, PartialEq, Eq)]
218pub(crate) enum SqlLogicalPlan {
219    SystemTableScan(SystemTableScan),
220}
221
222#[derive(Debug, Clone, PartialEq, Eq)]
223pub(crate) enum SqlPhysicalPlan {
224    SystemTableScanExec(SystemTableScan),
225}
226
227#[derive(Debug, Clone, PartialEq, Eq)]
228pub(crate) struct SystemTableScan {
229    table: SystemTable,
230    projection: Vec<SelectProjection>,
231    predicate: Option<SqlPredicate>,
232    order_by: Vec<SqlOrderItem>,
233    offset: Option<u64>,
234    limit: Option<u64>,
235}
236
237#[derive(Debug, Clone, Copy, PartialEq, Eq)]
238enum SystemTable {
239    Tables,
240    Properties,
241    Indexes,
242    Constraints,
243    AppendTables,
244    AppendStorage,
245    RuntimeStatus,
246    RuntimeCapabilities,
247    GraphStatistics,
248    ProjectedGraphs,
249    SearchProjectionChangefeed,
250    PlanCache,
251    SlowQueries,
252    StatementSummary,
253    InformationSchemaTables,
254    InformationSchemaColumns,
255    PgTables,
256    PgIndexes,
257}
258
259#[doc(hidden)]
260pub fn is_virtual_catalog_select(select: &SelectStatement) -> bool {
261    let Some(from) = &select.from else {
262        return false;
263    };
264    matches!(
265        (from.schema.as_deref(), from.name.as_str()),
266        (Some("system" | "information_schema" | "pg_catalog"), _)
267            | (None, "pg_tables" | "pg_indexes")
268    )
269}
270
271impl SlowQueryRecord {
272    pub fn completed(completion: SlowQueryCompletion<'_>) -> Self {
273        Self {
274            sequence: 0,
275            query_language: completion.query_language.to_string(),
276            statement_kind: completion.statement_kind.to_string(),
277            query_digest: completion.query_identity.query_digest().to_string(),
278            query_text_hash: completion.query_identity.query_text_hash().to_string(),
279            query_text: truncate_utf8(completion.query_text, MAX_SLOW_QUERY_TEXT_BYTES),
280            started_unix_micros: unix_now_micros(),
281            elapsed_micros: saturating_i64_from_u128(completion.elapsed_micros),
282            row_count: i64::try_from(completion.row_count).unwrap_or(i64::MAX),
283            success: completion.success,
284            error: completion.error,
285            slow_log_candidate: completion.slow_log_candidate,
286            access_control_policy_epoch: completion.access_control_policy_epoch,
287            vector_execution_reports: completion.vector_execution_reports,
288        }
289    }
290}
291
292impl SlowQueryLog {
293    pub fn new(capacity: usize) -> Self {
294        Self {
295            capacity,
296            next_sequence: 1,
297            records: VecDeque::with_capacity(capacity),
298        }
299    }
300
301    pub fn push(&mut self, mut record: SlowQueryRecord) {
302        if self.capacity == 0 {
303            return;
304        }
305        record.sequence = self.next_sequence;
306        self.next_sequence = self.next_sequence.saturating_add(1);
307        while self.records.len() >= self.capacity {
308            self.records.pop_front();
309        }
310        self.records.push_back(record);
311    }
312
313    pub fn snapshot(&self) -> Vec<SlowQueryRecord> {
314        self.records.iter().cloned().collect()
315    }
316}
317
318pub fn slow_query_log_jsonl(
319    records: &[SlowQueryRecord],
320    include_query_text: bool,
321) -> Result<String> {
322    let mut jsonl = String::new();
323    for record in records {
324        let line = serde_json::to_string(&slow_query_record_json(record, include_query_text))
325            .map_err(|error| {
326                HawDBError::Execution(format!("slow query log JSON error: {error}"))
327            })?;
328        jsonl.push_str(&line);
329        jsonl.push('\n');
330    }
331    Ok(jsonl)
332}
333
334pub fn slow_query_record_summary(record: &SlowQueryRecord) -> SlowQueryLogRecordSummary {
335    SlowQueryLogRecordSummary {
336        sequence: record.sequence,
337        query_language: record.query_language.clone(),
338        statement_kind: record.statement_kind.clone(),
339        query_digest: record.query_digest.clone(),
340        started_unix_micros: record.started_unix_micros,
341        elapsed_micros: record.elapsed_micros,
342        row_count: record.row_count,
343        success: record.success,
344        slow_log_candidate: record.slow_log_candidate,
345        access_control_policy_epoch: record.access_control_policy_epoch,
346    }
347}
348
349fn slow_query_record_json(record: &SlowQueryRecord, include_query_text: bool) -> serde_json::Value {
350    let mut object = serde_json::json!({
351        "protocol": SLOW_QUERY_LOG_EVENT_PROTOCOL,
352        "protocol_version": 1,
353        "sequence": record.sequence,
354        "query_language": record.query_language,
355        "statement_kind": record.statement_kind,
356        "query_digest": record.query_digest,
357        "started_unix_micros": record.started_unix_micros,
358        "elapsed_micros": record.elapsed_micros,
359        "row_count": record.row_count,
360        "success": record.success,
361        "slow_log_candidate": record.slow_log_candidate,
362        "access_control_policy_epoch": record.access_control_policy_epoch,
363        "redaction": {
364            "query_text_copied": include_query_text,
365            "parameters_copied": false,
366            "access_control_policy_inputs_copied": false
367        }
368    });
369    if include_query_text {
370        object["query_text"] = serde_json::Value::String(record.query_text.clone());
371    }
372    if let Some(error) = &record.error {
373        object["error"] =
374            serde_json::Value::String(truncate_utf8(error, MAX_STATEMENT_ERROR_BYTES));
375    }
376    object["vector_execution_report_count"] =
377        serde_json::json!(record.vector_execution_reports.len());
378    object["vector_execution_reports"] = serde_json::Value::Array(
379        record
380            .vector_execution_reports
381            .iter()
382            .map(vector_execution_report_json)
383            .collect(),
384    );
385    object
386}
387
388fn vector_execution_report_json(report: &VectorExecutionReport) -> serde_json::Value {
389    serde_json::json!({
390        "backend": report.backend.as_str(),
391        "compression_mode": report.compression_mode.as_str(),
392        "candidate_source": report.candidate_source.as_str(),
393        "backend_selection_reason": report.backend_selection_reason.map(|reason| reason.as_str()),
394        "estimated_raw_vector_bytes": report.estimated_raw_vector_bytes,
395        "filter_selectivity_per_million": report.filter_selectivity_per_million,
396        "candidate_score_source": report.candidate_score_source.as_str(),
397        "final_score_source": report.final_score_source.as_str(),
398        "generated_candidate_count": report.generated_candidate_count,
399        "descriptor_pruned_count": report.descriptor_pruned_count,
400        "scalar_filtered_count": report.scalar_filtered_count,
401        "residual_filtered_count": report.residual_filtered_count,
402        "candidate_scan_rounds": report.candidate_scan_rounds,
403        "reranked_candidate_count": report.reranked_candidate_count,
404        "returned_count": report.returned_count,
405        "raw_vector_bytes_read": report.raw_vector_bytes_read,
406        "candidate_scan": report.candidate_scan_metrics.as_ref().map(|metrics| serde_json::json!({
407            "kernel": metrics.kernel,
408            "worker_count": metrics.worker_count,
409            "segment_count": metrics.segment_count,
410            "scanned_segment_count": metrics.scanned_segment_count,
411            "scored_document_count": metrics.scored_document_count,
412            "filtered_document_count": metrics.filtered_document_count,
413            "scanned_block_count": metrics.scanned_block_count,
414            "skipped_block_count": metrics.skipped_block_count,
415            "payload_bytes_read": metrics.payload_bytes_read,
416            "admitted_working_bytes": metrics.admitted_working_bytes,
417        })),
418        "index_covered_document_count": report.index_covered_document_count,
419        "index_candidate_document_count": report.index_candidate_document_count,
420        "index_coverage_complete": report.index_coverage_complete,
421        "fallback_reason_codes": report.fallback_reason_codes.iter().map(|code| code.as_str()).collect::<Vec<_>>(),
422    })
423}
424
425impl StatementExecution {
426    pub fn completed(
427        query_language: &str,
428        query_text: &str,
429        statement_kind: &str,
430        query_identity: &QueryIdentity,
431        elapsed_micros: u128,
432        row_count: usize,
433    ) -> Self {
434        Self {
435            query_language: query_language.to_string(),
436            query_text: truncate_utf8(query_text, MAX_STATEMENT_TEXT_BYTES),
437            statement_kind: statement_kind.to_string(),
438            query_digest: query_identity.query_digest().to_string(),
439            query_text_hash: query_identity.query_text_hash().to_string(),
440            elapsed_micros: saturating_i64_from_u128(elapsed_micros),
441            row_count: i64::try_from(row_count).unwrap_or(i64::MAX),
442            success: true,
443            error: None,
444        }
445    }
446
447    pub fn failed(
448        query_language: &str,
449        query_text: &str,
450        statement_kind: &str,
451        query_identity: &QueryIdentity,
452        elapsed_micros: u128,
453        error: String,
454    ) -> Self {
455        Self {
456            query_language: query_language.to_string(),
457            query_text: truncate_utf8(query_text, MAX_STATEMENT_TEXT_BYTES),
458            statement_kind: statement_kind.to_string(),
459            query_digest: query_identity.query_digest().to_string(),
460            query_text_hash: query_identity.query_text_hash().to_string(),
461            elapsed_micros: saturating_i64_from_u128(elapsed_micros),
462            row_count: 0,
463            success: false,
464            error: Some(truncate_utf8(&error, MAX_STATEMENT_ERROR_BYTES)),
465        }
466    }
467}
468
469impl StatementSummary {
470    pub fn new(capacity: usize) -> Self {
471        Self {
472            capacity,
473            records: BTreeMap::new(),
474            insertion_order: VecDeque::with_capacity(capacity),
475        }
476    }
477
478    pub fn record(&mut self, execution: StatementExecution) {
479        if self.capacity == 0 {
480            return;
481        }
482
483        let digest = execution.query_digest.clone();
484        if let Some(record) = self.records.get_mut(&digest) {
485            record.apply(execution);
486            return;
487        }
488
489        while self.records.len() >= self.capacity {
490            let Some(evicted) = self.insertion_order.pop_front() else {
491                break;
492            };
493            self.records.remove(&evicted);
494        }
495
496        self.insertion_order.push_back(digest.clone());
497        self.records.insert(
498            digest.clone(),
499            StatementSummaryRecord::from_execution(digest, execution),
500        );
501    }
502
503    pub fn snapshot(&self) -> Vec<StatementSummaryRecord> {
504        self.insertion_order
505            .iter()
506            .filter_map(|digest| self.records.get(digest).cloned())
507            .collect()
508    }
509}
510
511impl StatementSummaryRecord {
512    fn from_execution(digest: String, execution: StatementExecution) -> Self {
513        let now = unix_now_micros();
514        let success_count = i64::from(execution.success);
515        let error_count = i64::from(!execution.success);
516        Self {
517            digest,
518            query_language: execution.query_language,
519            query_text: execution.query_text,
520            sample_query_text_hash: execution.query_text_hash,
521            statement_kind: execution.statement_kind,
522            execution_count: 1,
523            success_count,
524            error_count,
525            total_elapsed_micros: execution.elapsed_micros,
526            max_elapsed_micros: execution.elapsed_micros,
527            total_row_count: execution.row_count,
528            last_seen_unix_micros: now,
529            last_elapsed_micros: execution.elapsed_micros,
530            last_row_count: execution.row_count,
531            last_success: execution.success,
532            last_error: execution.error,
533        }
534    }
535
536    fn apply(&mut self, execution: StatementExecution) {
537        self.execution_count = self.execution_count.saturating_add(1);
538        if execution.success {
539            self.success_count = self.success_count.saturating_add(1);
540        } else {
541            self.error_count = self.error_count.saturating_add(1);
542        }
543        self.total_elapsed_micros = self
544            .total_elapsed_micros
545            .saturating_add(execution.elapsed_micros);
546        self.max_elapsed_micros = self.max_elapsed_micros.max(execution.elapsed_micros);
547        self.total_row_count = self.total_row_count.saturating_add(execution.row_count);
548        self.last_seen_unix_micros = unix_now_micros();
549        self.last_elapsed_micros = execution.elapsed_micros;
550        self.last_row_count = execution.row_count;
551        self.last_success = execution.success;
552        self.last_error = execution.error;
553    }
554}
555
556#[cfg(test)]
557pub fn query_sql<Store: SystemSqlStore>(
558    sql_text: &str,
559    max_rows: Option<usize>,
560    max_payload_bytes: Option<usize>,
561    context: &SystemSqlContext<'_, Store>,
562) -> Result<QueryOutput> {
563    query_sql_with_params(sql_text, &[], max_rows, max_payload_bytes, context)
564}
565
566pub fn query_sql_with_params<Store: SystemSqlStore>(
567    sql_text: &str,
568    parameters: &[Value],
569    max_rows: Option<usize>,
570    max_payload_bytes: Option<usize>,
571    context: &SystemSqlContext<'_, Store>,
572) -> Result<QueryOutput> {
573    let logical = plan_sql(sql_text, parameters)?;
574    let physical = optimize_sql(logical);
575    let rows = execute_sql(physical, context, max_rows)?;
576    let payload_bytes = rows.iter().fold(0usize, |total, row| {
577        total.saturating_add(map_payload_bytes(row))
578    });
579    if max_payload_bytes.is_some_and(|limit| payload_bytes > limit) {
580        return Err(HawDBError::Execution(format!(
581            "SQL query payload uses {payload_bytes} bytes, exceeding max_read_result_payload_bytes {}",
582            max_payload_bytes.unwrap_or_default()
583        )));
584    }
585    Ok(QueryOutput { rows: rows.into() })
586}
587
588fn plan_sql(sql_text: &str, parameters: &[Value]) -> Result<SqlLogicalPlan> {
589    let prepared = hawdb_sql::prepare_postgres_sql(sql_text)?;
590    if prepared.parameters.len() != parameters.len() {
591        return Err(HawDBError::Semantic(format!(
592            "PostgreSQL statement requires {} parameters, but {} parameters were supplied",
593            prepared.parameters.len(),
594            parameters.len()
595        )));
596    }
597    let SqlStatement::Select(select) = prepared.statement else {
598        return Err(HawDBError::Semantic(
599            "system SQL only supports SELECT statements".to_string(),
600        ));
601    };
602    validate_system_select_shape(&select)?;
603    let table = system_table(&select)?;
604    validate_projection(table, &select.projection)?;
605    validate_predicate_columns(table, select.selection.as_ref())?;
606    validate_order_columns(table, &select.order_by)?;
607    Ok(SqlLogicalPlan::SystemTableScan(SystemTableScan {
608        table,
609        projection: select.projection,
610        predicate: select
611            .selection
612            .map(|predicate| bind_predicate(predicate, parameters))
613            .transpose()?,
614        order_by: select.order_by,
615        offset: select
616            .offset
617            .map(|bound| bind_bound(bound, parameters, "OFFSET"))
618            .transpose()?,
619        limit: select
620            .limit
621            .map(|bound| bind_bound(bound, parameters, "LIMIT"))
622            .transpose()?,
623    }))
624}
625
626fn validate_system_select_shape(select: &SelectStatement) -> Result<()> {
627    if select.having.is_some() {
628        return Err(HawDBError::Semantic(
629            "system SQL does not support HAVING".into(),
630        ));
631    }
632    if select.distinct
633        || select.from_alias.is_some()
634        || !select.joins.is_empty()
635        || !select.group_by.is_empty()
636        || select.lock_strength.is_some()
637    {
638        return Err(HawDBError::Semantic(
639            "system SQL does not support DISTINCT, table aliases, joins, GROUP BY, or locking clauses"
640                .to_string(),
641        ));
642    }
643    if select
644        .order_by
645        .iter()
646        .any(|item| item.nulls != hawdb_sql::SqlNullOrder::DialectDefault)
647    {
648        return Err(HawDBError::Semantic(
649            "system SQL does not support explicit NULLS FIRST/LAST".to_string(),
650        ));
651    }
652    Ok(())
653}
654
655fn bind_predicate(mut predicate: SqlPredicate, parameters: &[Value]) -> Result<SqlPredicate> {
656    predicate.try_visit_mut(&mut |expression| {
657        if let ExprKind::Value(value) = &mut expression.kind {
658            let unbound = std::mem::replace(value, SqlValue::Literal(Value::Null));
659            *value = SqlValue::Literal(bind_value(unbound, parameters)?);
660        }
661        if let ExprKind::Like {
662            pattern,
663            escape,
664            case_insensitive,
665            ..
666        } = &expression.kind
667            && let SqlValue::Literal(Value::String(pattern)) = pattern.require_value()?
668        {
669            hawdb_sql::sql_like_matches("", pattern, *escape, *case_insensitive)?;
670        }
671        Ok::<_, HawDBError>(())
672    })?;
673    Ok(predicate)
674}
675
676fn bind_value(value: SqlValue, parameters: &[Value]) -> Result<Value> {
677    match value {
678        SqlValue::Literal(value) => Ok(value),
679        SqlValue::Parameter(position) => parameters.get(position - 1).cloned().ok_or_else(|| {
680            HawDBError::Semantic(format!("missing PostgreSQL parameter ${position}"))
681        }),
682    }
683}
684
685fn bind_bound(bound: SqlBound, parameters: &[Value], name: &str) -> Result<u64> {
686    match bound {
687        SqlBound::Literal(value) => Ok(value),
688        SqlBound::Parameter(position) => match parameters.get(position - 1) {
689            Some(Value::Int(value)) if *value >= 0 => Ok(*value as u64),
690            Some(_) => Err(HawDBError::Semantic(format!(
691                "PostgreSQL {name} parameter ${position} must be a non-negative integer"
692            ))),
693            None => Err(HawDBError::Semantic(format!(
694                "missing PostgreSQL parameter ${position}"
695            ))),
696        },
697    }
698}
699
700fn optimize_sql(logical: SqlLogicalPlan) -> SqlPhysicalPlan {
701    match logical {
702        SqlLogicalPlan::SystemTableScan(scan) => SqlPhysicalPlan::SystemTableScanExec(scan),
703    }
704}
705
706fn execute_sql<Store: SystemSqlStore>(
707    physical: SqlPhysicalPlan,
708    context: &SystemSqlContext<'_, Store>,
709    max_rows: Option<usize>,
710) -> Result<Vec<Row>> {
711    match physical {
712        SqlPhysicalPlan::SystemTableScanExec(scan) => {
713            execute_system_table_scan(scan, context, max_rows)
714        }
715    }
716}
717
718fn execute_system_table_scan<Store: SystemSqlStore>(
719    scan: SystemTableScan,
720    context: &SystemSqlContext<'_, Store>,
721    max_rows: Option<usize>,
722) -> Result<Vec<Row>> {
723    let mut rows = match scan.table {
724        SystemTable::Tables => table_rows(context.catalog),
725        SystemTable::Properties => property_rows(context.catalog),
726        SystemTable::Indexes => index_rows(context.catalog),
727        SystemTable::Constraints => constraint_rows(context.catalog),
728        SystemTable::AppendTables => append_table_rows(context.append_state),
729        SystemTable::AppendStorage => append_storage_rows(context.store, context.append_state),
730        SystemTable::RuntimeStatus => runtime_status_rows(context),
731        SystemTable::RuntimeCapabilities => runtime_capability_rows(context.runtime),
732        SystemTable::GraphStatistics => {
733            graph_statistics_rows(context.catalog, &context.store.statistics(context.catalog))
734        }
735        SystemTable::ProjectedGraphs => {
736            projected_graph_rows(context.store.projected_graph_statuses())
737        }
738        SystemTable::SearchProjectionChangefeed => {
739            search_projection_changefeed_rows(context.store.search_projection_changefeed_status())
740        }
741        SystemTable::PlanCache => plan_cache_rows(context.plan_cache_stats),
742        SystemTable::SlowQueries => slow_query_rows(context.slow_queries),
743        SystemTable::StatementSummary => statement_summary_rows(context.statement_summaries),
744        SystemTable::InformationSchemaTables => {
745            information_schema_table_rows(context.relational_state)
746        }
747        SystemTable::InformationSchemaColumns => {
748            information_schema_column_rows(context.relational_state)
749        }
750        SystemTable::PgTables => pg_table_rows(context.relational_state),
751        SystemTable::PgIndexes => pg_index_rows(context.relational_state),
752    };
753
754    if let Some(predicate) = &scan.predicate {
755        rows.retain(|row| predicate_matches(predicate, row));
756    }
757
758    if !scan.order_by.is_empty() {
759        rows.sort_by(|left, right| compare_ordered_rows(left, right, &scan.order_by));
760    }
761
762    let offset = scan.offset.unwrap_or(0);
763    let limit = effective_limit(scan.limit, max_rows)?;
764    rows = rows
765        .into_iter()
766        .skip(usize::try_from(offset).unwrap_or(usize::MAX))
767        .take(limit.unwrap_or(usize::MAX))
768        .collect();
769
770    if let Some(max_rows) = max_rows
771        && rows.len() > max_rows
772    {
773        return Err(HawDBError::Execution(format!(
774                "SQL query returned more than {max_rows} rows, exceeding max_read_result_rows {max_rows}"
775            )));
776    }
777
778    project_rows(rows, &scan.projection)
779}
780
781fn effective_limit(query_limit: Option<u64>, max_rows: Option<usize>) -> Result<Option<usize>> {
782    let query_limit = query_limit
783        .map(|limit| {
784            usize::try_from(limit)
785                .map_err(|_| HawDBError::Semantic("SQL LIMIT is too large".to_string()))
786        })
787        .transpose()?;
788    Ok(match (query_limit, max_rows) {
789        (Some(query_limit), Some(max_rows)) => Some(query_limit.min(max_rows.saturating_add(1))),
790        (Some(query_limit), None) => Some(query_limit),
791        (None, Some(max_rows)) => Some(max_rows.saturating_add(1)),
792        (None, None) => None,
793    })
794}
795
796fn project_rows(rows: Vec<Row>, projection: &[SelectProjection]) -> Result<Vec<Row>> {
797    if projection
798        .iter()
799        .any(|projection| matches!(projection, SelectProjection::Wildcard))
800    {
801        return Ok(rows);
802    }
803    rows.into_iter()
804        .map(|row| {
805            projection
806                .iter()
807                .map(|projection| {
808                    let SelectProjection::Expression {
809                        expression:
810                            Expr {
811                                kind: ExprKind::Column(name),
812                                ..
813                            },
814                        alias,
815                        ..
816                    } = projection
817                    else {
818                        unreachable!("wildcard handled above");
819                    };
820                    let value = row.get(&name.name).cloned().unwrap_or(Value::Null);
821                    Ok((alias.clone().unwrap_or_else(|| name.name.clone()), value))
822                })
823                .collect()
824        })
825        .collect()
826}
827
828fn table_rows(catalog: &Catalog) -> Vec<Row> {
829    catalog
830        .table_descriptors()
831        .map(|table| {
832            BTreeMap::from([
833                ("table_id".to_string(), u32_value(table.id.0)),
834                ("table_name".to_string(), Value::String(table.name.clone())),
835                (
836                    "table_kind".to_string(),
837                    Value::String(table_kind_name(table.kind).to_string()),
838                ),
839                (
840                    "state".to_string(),
841                    Value::String(schema_object_state_name(table.state).to_string()),
842                ),
843            ])
844        })
845        .collect()
846}
847
848fn information_schema_table_rows(state: &RelationalState) -> Vec<Row> {
849    state
850        .table_schemas()
851        .map(|schema| {
852            BTreeMap::from([
853                (
854                    "table_catalog".to_string(),
855                    Value::String("hawdb".to_string()),
856                ),
857                (
858                    "table_schema".to_string(),
859                    Value::String("public".to_string()),
860                ),
861                ("table_name".to_string(), Value::String(schema.name.clone())),
862                (
863                    "table_type".to_string(),
864                    Value::String("BASE TABLE".to_string()),
865                ),
866                ("self_referencing_column_name".to_string(), Value::Null),
867                ("reference_generation".to_string(), Value::Null),
868                ("user_defined_type_catalog".to_string(), Value::Null),
869                ("user_defined_type_schema".to_string(), Value::Null),
870                ("user_defined_type_name".to_string(), Value::Null),
871                (
872                    "is_insertable_into".to_string(),
873                    Value::String("YES".to_string()),
874                ),
875                ("is_typed".to_string(), Value::String("NO".to_string())),
876                ("commit_action".to_string(), Value::Null),
877            ])
878        })
879        .collect()
880}
881
882fn information_schema_column_rows(state: &RelationalState) -> Vec<Row> {
883    state
884        .table_schemas()
885        .flat_map(|schema| {
886            schema
887                .columns
888                .iter()
889                .enumerate()
890                .map(move |(position, column)| {
891                    let type_info = information_schema_type(column.scalar_type);
892                    let ordinal_position = position.saturating_add(1);
893                    BTreeMap::from([
894                        (
895                            "table_catalog".to_string(),
896                            Value::String("hawdb".to_string()),
897                        ),
898                        (
899                            "table_schema".to_string(),
900                            Value::String("public".to_string()),
901                        ),
902                        ("table_name".to_string(), Value::String(schema.name.clone())),
903                        (
904                            "column_name".to_string(),
905                            Value::String(column.name.clone()),
906                        ),
907                        (
908                            "ordinal_position".to_string(),
909                            usize_value(ordinal_position),
910                        ),
911                        (
912                            "column_default".to_string(),
913                            relational_default_value(column.default.as_ref()),
914                        ),
915                        (
916                            "is_nullable".to_string(),
917                            Value::String(if column.nullable { "YES" } else { "NO" }.to_string()),
918                        ),
919                        (
920                            "data_type".to_string(),
921                            Value::String(type_info.data_type.to_string()),
922                        ),
923                        (
924                            "character_maximum_length".to_string(),
925                            option_i64_value(type_info.character_maximum_length),
926                        ),
927                        (
928                            "character_octet_length".to_string(),
929                            option_i64_value(type_info.character_octet_length),
930                        ),
931                        (
932                            "numeric_precision".to_string(),
933                            option_i64_value(type_info.numeric_precision),
934                        ),
935                        (
936                            "numeric_precision_radix".to_string(),
937                            option_i64_value(type_info.numeric_precision_radix),
938                        ),
939                        (
940                            "numeric_scale".to_string(),
941                            option_i64_value(type_info.numeric_scale),
942                        ),
943                        ("datetime_precision".to_string(), Value::Null),
944                        ("interval_type".to_string(), Value::Null),
945                        ("interval_precision".to_string(), Value::Null),
946                        ("character_set_catalog".to_string(), Value::Null),
947                        ("character_set_schema".to_string(), Value::Null),
948                        ("character_set_name".to_string(), Value::Null),
949                        ("collation_catalog".to_string(), Value::Null),
950                        ("collation_schema".to_string(), Value::Null),
951                        ("collation_name".to_string(), Value::Null),
952                        ("domain_catalog".to_string(), Value::Null),
953                        ("domain_schema".to_string(), Value::Null),
954                        ("domain_name".to_string(), Value::Null),
955                        (
956                            "udt_catalog".to_string(),
957                            Value::String("hawdb".to_string()),
958                        ),
959                        (
960                            "udt_schema".to_string(),
961                            Value::String("pg_catalog".to_string()),
962                        ),
963                        (
964                            "udt_name".to_string(),
965                            Value::String(type_info.udt_name.to_string()),
966                        ),
967                        ("scope_catalog".to_string(), Value::Null),
968                        ("scope_schema".to_string(), Value::Null),
969                        ("scope_name".to_string(), Value::Null),
970                        ("maximum_cardinality".to_string(), Value::Null),
971                        (
972                            "dtd_identifier".to_string(),
973                            Value::String(ordinal_position.to_string()),
974                        ),
975                        (
976                            "is_self_referencing".to_string(),
977                            Value::String("NO".to_string()),
978                        ),
979                        ("is_identity".to_string(), Value::String("NO".to_string())),
980                        ("identity_generation".to_string(), Value::Null),
981                        ("identity_start".to_string(), Value::Null),
982                        ("identity_increment".to_string(), Value::Null),
983                        ("identity_maximum".to_string(), Value::Null),
984                        ("identity_minimum".to_string(), Value::Null),
985                        ("identity_cycle".to_string(), Value::Null),
986                        (
987                            "is_generated".to_string(),
988                            Value::String("NEVER".to_string()),
989                        ),
990                        ("generation_expression".to_string(), Value::Null),
991                        ("is_updatable".to_string(), Value::String("YES".to_string())),
992                    ])
993                })
994        })
995        .collect()
996}
997
998#[derive(Debug, Clone, Copy)]
999struct InformationSchemaType {
1000    data_type: &'static str,
1001    udt_name: &'static str,
1002    character_maximum_length: Option<i64>,
1003    character_octet_length: Option<i64>,
1004    numeric_precision: Option<i64>,
1005    numeric_precision_radix: Option<i64>,
1006    numeric_scale: Option<i64>,
1007}
1008
1009fn information_schema_type(scalar_type: RelationalScalarType) -> InformationSchemaType {
1010    match scalar_type {
1011        RelationalScalarType::Boolean => InformationSchemaType {
1012            data_type: "boolean",
1013            udt_name: "bool",
1014            character_maximum_length: None,
1015            character_octet_length: None,
1016            numeric_precision: None,
1017            numeric_precision_radix: None,
1018            numeric_scale: None,
1019        },
1020        RelationalScalarType::BigInt => InformationSchemaType {
1021            data_type: "bigint",
1022            udt_name: "int8",
1023            character_maximum_length: None,
1024            character_octet_length: None,
1025            numeric_precision: Some(64),
1026            numeric_precision_radix: Some(2),
1027            numeric_scale: Some(0),
1028        },
1029        RelationalScalarType::DoublePrecision => InformationSchemaType {
1030            data_type: "double precision",
1031            udt_name: "float8",
1032            character_maximum_length: None,
1033            character_octet_length: None,
1034            numeric_precision: Some(53),
1035            numeric_precision_radix: Some(2),
1036            numeric_scale: None,
1037        },
1038        RelationalScalarType::Text => InformationSchemaType {
1039            data_type: "text",
1040            udt_name: "text",
1041            character_maximum_length: None,
1042            character_octet_length: None,
1043            numeric_precision: None,
1044            numeric_precision_radix: None,
1045            numeric_scale: None,
1046        },
1047        RelationalScalarType::Bytea => InformationSchemaType {
1048            data_type: "bytea",
1049            udt_name: "bytea",
1050            character_maximum_length: None,
1051            character_octet_length: None,
1052            numeric_precision: None,
1053            numeric_precision_radix: None,
1054            numeric_scale: None,
1055        },
1056        RelationalScalarType::Uuid => InformationSchemaType {
1057            data_type: "uuid",
1058            udt_name: "uuid",
1059            character_maximum_length: None,
1060            character_octet_length: None,
1061            numeric_precision: None,
1062            numeric_precision_radix: None,
1063            numeric_scale: None,
1064        },
1065    }
1066}
1067
1068fn relational_default_value(default: Option<&RelationalColumnDefault>) -> Value {
1069    match default {
1070        None | Some(RelationalColumnDefault::Literal(RelationalValue::Overflow(_))) => Value::Null,
1071        Some(RelationalColumnDefault::Literal(RelationalValue::Null)) => {
1072            Value::String("NULL".to_string())
1073        }
1074        Some(RelationalColumnDefault::Literal(RelationalValue::Boolean(value))) => {
1075            Value::String(value.to_string())
1076        }
1077        Some(RelationalColumnDefault::Literal(RelationalValue::BigInt(value))) => {
1078            Value::String(value.to_string())
1079        }
1080        Some(RelationalColumnDefault::Literal(RelationalValue::DoublePrecision(value))) => {
1081            Value::String(value.to_string())
1082        }
1083        Some(RelationalColumnDefault::Literal(RelationalValue::Text(value))) => {
1084            Value::String(format!("'{}'::text", value.replace('\'', "''")))
1085        }
1086        Some(RelationalColumnDefault::Literal(RelationalValue::Bytea(value))) => {
1087            Value::String(format!("'\\x{}'::bytea", encode_hex(value)))
1088        }
1089        Some(RelationalColumnDefault::Literal(RelationalValue::Uuid(value))) => {
1090            Value::String(format!("'{value}'::uuid"))
1091        }
1092        Some(RelationalColumnDefault::UuidV7) => Value::String("uuidv7()".to_string()),
1093    }
1094}
1095
1096fn encode_hex(bytes: &[u8]) -> String {
1097    const HEX: &[u8; 16] = b"0123456789abcdef";
1098    let mut encoded = String::with_capacity(bytes.len().saturating_mul(2));
1099    for byte in bytes {
1100        encoded.push(HEX[usize::from(byte >> 4)] as char);
1101        encoded.push(HEX[usize::from(byte & 0x0f)] as char);
1102    }
1103    encoded
1104}
1105
1106fn pg_table_rows(state: &RelationalState) -> Vec<Row> {
1107    state
1108        .table_schemas()
1109        .map(|schema| {
1110            BTreeMap::from([
1111                (
1112                    "schemaname".to_string(),
1113                    Value::String("public".to_string()),
1114                ),
1115                ("tablename".to_string(), Value::String(schema.name.clone())),
1116                ("tableowner".to_string(), Value::String("hawdb".to_string())),
1117                ("tablespace".to_string(), Value::Null),
1118                (
1119                    "hasindexes".to_string(),
1120                    Value::Bool(
1121                        !schema.primary_key.is_empty()
1122                            || !schema.unique_constraints.is_empty()
1123                            || !schema.indexes.is_empty(),
1124                    ),
1125                ),
1126                ("hasrules".to_string(), Value::Bool(false)),
1127                ("hastriggers".to_string(), Value::Bool(false)),
1128                ("rowsecurity".to_string(), Value::Bool(false)),
1129            ])
1130        })
1131        .collect()
1132}
1133
1134fn pg_index_rows(state: &RelationalState) -> Vec<Row> {
1135    state
1136        .table_schemas()
1137        .flat_map(relational_schema_index_rows)
1138        .collect()
1139}
1140
1141fn relational_schema_index_rows(schema: &RelationalTableSchema) -> Vec<Row> {
1142    let mut rows = Vec::with_capacity(
1143        1usize
1144            .saturating_add(schema.unique_constraints.len())
1145            .saturating_add(schema.indexes.len()),
1146    );
1147    if !schema.primary_key.is_empty() {
1148        let name = format!("{}_pkey", schema.name);
1149        rows.push(pg_index_row(schema, &name, &schema.primary_key, true));
1150    }
1151    rows.extend(schema.unique_constraints.iter().map(|columns| {
1152        let name = format!("{}_{}_key", schema.name, columns.join("_"));
1153        pg_index_row(schema, &name, columns, true)
1154    }));
1155    rows.extend(
1156        schema
1157            .indexes
1158            .iter()
1159            .map(|index| pg_index_row(schema, &index.name, &index.columns, index.unique)),
1160    );
1161    rows
1162}
1163
1164fn pg_index_row(
1165    schema: &RelationalTableSchema,
1166    index_name: &str,
1167    columns: &[String],
1168    unique: bool,
1169) -> Row {
1170    let unique = if unique { "UNIQUE " } else { "" };
1171    let columns = columns
1172        .iter()
1173        .map(|column| quote_postgres_identifier(column))
1174        .collect::<Vec<_>>()
1175        .join(", ");
1176    let index_definition = format!(
1177        "CREATE {unique}INDEX {} ON public.{} USING btree ({columns})",
1178        quote_postgres_identifier(index_name),
1179        quote_postgres_identifier(&schema.name),
1180    );
1181    BTreeMap::from([
1182        (
1183            "schemaname".to_string(),
1184            Value::String("public".to_string()),
1185        ),
1186        ("tablename".to_string(), Value::String(schema.name.clone())),
1187        (
1188            "indexname".to_string(),
1189            Value::String(index_name.to_string()),
1190        ),
1191        ("tablespace".to_string(), Value::Null),
1192        ("indexdef".to_string(), Value::String(index_definition)),
1193    ])
1194}
1195
1196fn quote_postgres_identifier(identifier: &str) -> String {
1197    format!("\"{}\"", identifier.replace('"', "\"\""))
1198}
1199
1200fn property_rows(catalog: &Catalog) -> Vec<Row> {
1201    catalog
1202        .property_descriptors()
1203        .map(|property| {
1204            let table = catalog.table_descriptor(property.table_id);
1205            BTreeMap::from([
1206                ("property_id".to_string(), u32_value(property.id.0)),
1207                ("table_id".to_string(), u32_value(property.table_id.0)),
1208                (
1209                    "table_name".to_string(),
1210                    table
1211                        .map(|table| Value::String(table.name.clone()))
1212                        .unwrap_or(Value::Null),
1213                ),
1214                (
1215                    "table_kind".to_string(),
1216                    table
1217                        .map(|table| Value::String(table_kind_name(table.kind).to_string()))
1218                        .unwrap_or(Value::Null),
1219                ),
1220                (
1221                    "property_name".to_string(),
1222                    Value::String(property.name.clone()),
1223                ),
1224                (
1225                    "value_type".to_string(),
1226                    Value::String(property_type_name(property.value_type).to_string()),
1227                ),
1228                ("nullable".to_string(), Value::Bool(property.nullable)),
1229                (
1230                    "state".to_string(),
1231                    Value::String(schema_object_state_name(property.state).to_string()),
1232                ),
1233            ])
1234        })
1235        .collect()
1236}
1237
1238fn index_rows(catalog: &Catalog) -> Vec<Row> {
1239    let mut rows = catalog
1240        .property_indexes()
1241        .map(|index| {
1242            BTreeMap::from([
1243                ("index_id".to_string(), u32_value(index.id.0)),
1244                (
1245                    "subject_kind".to_string(),
1246                    Value::String("node".to_string()),
1247                ),
1248                ("subject_id".to_string(), u32_value(index.label_id.0)),
1249                (
1250                    "subject_name".to_string(),
1251                    optional_string_value(catalog.label_name(index.label_id)),
1252                ),
1253                (
1254                    "index_kind".to_string(),
1255                    Value::String(index_kind_name(index.kind).to_string()),
1256                ),
1257                (
1258                    "property_names".to_string(),
1259                    Value::List(vec![Value::String(index.property.clone())]),
1260                ),
1261            ])
1262        })
1263        .chain(catalog.composite_property_indexes().map(|index| {
1264            BTreeMap::from([
1265                ("index_id".to_string(), u32_value(index.id.0)),
1266                (
1267                    "subject_kind".to_string(),
1268                    Value::String("node".to_string()),
1269                ),
1270                ("subject_id".to_string(), u32_value(index.label_id.0)),
1271                (
1272                    "subject_name".to_string(),
1273                    optional_string_value(catalog.label_name(index.label_id)),
1274                ),
1275                (
1276                    "index_kind".to_string(),
1277                    Value::String("composite_equality".to_string()),
1278                ),
1279                (
1280                    "property_names".to_string(),
1281                    Value::List(
1282                        index
1283                            .properties
1284                            .iter()
1285                            .cloned()
1286                            .map(Value::String)
1287                            .collect(),
1288                    ),
1289                ),
1290            ])
1291        }))
1292        .collect::<Vec<_>>();
1293    rows.sort_by_key(|row| row.get("index_id").cloned());
1294    rows
1295}
1296
1297fn constraint_rows(catalog: &Catalog) -> Vec<Row> {
1298    let constraints = catalog
1299        .unique_constraints()
1300        .chain(catalog.node_property_exists_constraints())
1301        .chain(catalog.relationship_property_exists_constraints())
1302        .chain(catalog.relationship_unique_constraints());
1303    let mut rows = constraints
1304        .map(|constraint| {
1305            let (subject_kind, subject_id, subject_name) = match constraint.subject {
1306                ConstraintSubject::Node(label_id) => (
1307                    "node",
1308                    label_id.0,
1309                    optional_string_value(catalog.label_name(label_id)),
1310                ),
1311                ConstraintSubject::Relationship(rel_type_id) => (
1312                    "relationship",
1313                    rel_type_id.0,
1314                    optional_string_value(catalog.rel_type_name(rel_type_id)),
1315                ),
1316            };
1317            BTreeMap::from([
1318                ("constraint_id".to_string(), u32_value(constraint.id.0)),
1319                (
1320                    "subject_kind".to_string(),
1321                    Value::String(subject_kind.to_string()),
1322                ),
1323                ("subject_id".to_string(), u32_value(subject_id)),
1324                ("subject_name".to_string(), subject_name),
1325                (
1326                    "property_name".to_string(),
1327                    Value::String(constraint.property.clone()),
1328                ),
1329                (
1330                    "constraint_kind".to_string(),
1331                    Value::String(constraint_kind_name(constraint.kind).to_string()),
1332                ),
1333            ])
1334        })
1335        .collect::<Vec<_>>();
1336    rows.sort_by_key(|row| row.get("constraint_id").cloned());
1337    rows
1338}
1339
1340fn runtime_status_rows<Store: SystemSqlStore>(context: &SystemSqlContext<'_, Store>) -> Vec<Row> {
1341    vec![BTreeMap::from([
1342        (
1343            "commit_epoch".to_string(),
1344            u64_value(context.store.commit_epoch()),
1345        ),
1346        (
1347            "read_only".to_string(),
1348            Value::Bool(context.runtime.read_only),
1349        ),
1350        (
1351            "max_read_result_rows".to_string(),
1352            option_usize_value(context.runtime.max_read_result_rows),
1353        ),
1354        (
1355            "max_read_result_payload_bytes".to_string(),
1356            option_usize_value(context.runtime.max_read_result_payload_bytes),
1357        ),
1358        (
1359            "storage_residency_mode".to_string(),
1360            Value::String(
1361                storage_residency_mode_name(context.runtime.storage_residency_mode).to_string(),
1362            ),
1363        ),
1364        (
1365            "relational_index_mode".to_string(),
1366            Value::String(
1367                relational_index_mode_name(context.runtime.relational_index_mode).to_string(),
1368            ),
1369        ),
1370    ])]
1371}
1372
1373fn append_table_rows(state: &AppendState) -> Vec<Row> {
1374    state
1375        .schemas()
1376        .values()
1377        .map(|schema| {
1378            BTreeMap::from([
1379                ("table_name".to_string(), Value::String(schema.name.clone())),
1380                (
1381                    "storage_mode".to_string(),
1382                    Value::String("strict_append".to_string()),
1383                ),
1384                (
1385                    "partition_key".to_string(),
1386                    Value::List(
1387                        schema
1388                            .partition_key
1389                            .iter()
1390                            .cloned()
1391                            .map(Value::String)
1392                            .collect(),
1393                    ),
1394                ),
1395                (
1396                    "order_key".to_string(),
1397                    Value::List(
1398                        schema
1399                            .order_key
1400                            .iter()
1401                            .cloned()
1402                            .map(Value::String)
1403                            .collect(),
1404                    ),
1405                ),
1406                (
1407                    "order_mode".to_string(),
1408                    Value::String(
1409                        match schema.order_mode {
1410                            AppendOrderMode::CallerProvided => "caller_provided",
1411                            AppendOrderMode::CommitSequence => "commit_sequence",
1412                        }
1413                        .to_string(),
1414                    ),
1415                ),
1416                (
1417                    "generated_order_watermark".to_string(),
1418                    state
1419                        .generated_order_watermark(&schema.name)
1420                        .map_or(Value::Null, Value::Int),
1421                ),
1422                (
1423                    "column_count".to_string(),
1424                    Value::Int(i64::try_from(schema.columns.len()).unwrap_or(i64::MAX)),
1425                ),
1426            ])
1427        })
1428        .collect()
1429}
1430
1431fn append_storage_rows(store: &impl SystemSqlStore, state: &AppendState) -> Vec<Row> {
1432    let mut report = store.append_storage_residency_report();
1433    report.live_rows = state.live_rows();
1434    report.live_payload_bytes = state.live_payload_bytes();
1435    vec![BTreeMap::from([
1436        (
1437            "canonical_segment_count".to_string(),
1438            Value::Int(i64::try_from(report.canonical_segment_count).unwrap_or(i64::MAX)),
1439        ),
1440        (
1441            "canonical_segment_bytes".to_string(),
1442            Value::Int(i64::try_from(report.canonical_segment_bytes).unwrap_or(i64::MAX)),
1443        ),
1444        (
1445            "resident_segment_payload_bytes".to_string(),
1446            Value::Int(i64::try_from(report.resident_segment_payload_bytes).unwrap_or(i64::MAX)),
1447        ),
1448        (
1449            "resident_descriptor_count".to_string(),
1450            Value::Int(i64::try_from(report.resident_descriptor_count).unwrap_or(i64::MAX)),
1451        ),
1452        (
1453            "live_rows".to_string(),
1454            Value::Int(i64::try_from(report.live_rows).unwrap_or(i64::MAX)),
1455        ),
1456        (
1457            "live_payload_bytes".to_string(),
1458            Value::Int(i64::try_from(report.live_payload_bytes).unwrap_or(i64::MAX)),
1459        ),
1460    ])]
1461}
1462
1463fn runtime_capability_rows(runtime: SystemRuntimeSnapshot) -> Vec<Row> {
1464    [
1465        (
1466            "access_control",
1467            runtime.runtime_capabilities.access_control,
1468        ),
1469        (
1470            "full_text_search",
1471            runtime.runtime_capabilities.full_text_search,
1472        ),
1473        ("vector_search", runtime.runtime_capabilities.vector_search),
1474        (
1475            "graph_analytics",
1476            runtime.runtime_capabilities.graph_analytics,
1477        ),
1478        (
1479            "background_maintenance",
1480            runtime.runtime_capabilities.background_maintenance,
1481        ),
1482    ]
1483    .into_iter()
1484    .map(|(capability, enabled)| {
1485        BTreeMap::from([
1486            (
1487                "capability".to_string(),
1488                Value::String(capability.to_string()),
1489            ),
1490            ("enabled".to_string(), Value::Bool(enabled)),
1491        ])
1492    })
1493    .collect()
1494}
1495
1496fn graph_statistics_rows(catalog: &Catalog, statistics: &GraphStatistics) -> Vec<Row> {
1497    let mut rows = Vec::with_capacity(
1498        2usize
1499            .saturating_add(statistics.label_counts.len())
1500            .saturating_add(statistics.rel_type_counts.len())
1501            .saturating_add(statistics.rel_type_source_counts.len())
1502            .saturating_add(statistics.rel_type_target_counts.len())
1503            .saturating_add(statistics.path_counts.len())
1504            .saturating_add(statistics.path_source_distinct_counts.len())
1505            .saturating_add(statistics.path_target_distinct_counts.len())
1506            .saturating_add(statistics.bounded_path_counts.len())
1507            .saturating_add(statistics.bounded_path_source_distinct_counts.len())
1508            .saturating_add(statistics.bounded_path_target_distinct_counts.len())
1509            .saturating_add(statistics.index_samples.len())
1510            .saturating_add(statistics.property_distinct_counts.len())
1511            .saturating_add(statistics.rel_property_distinct_counts.len())
1512            .saturating_add(statistics.property_histograms.len())
1513            .saturating_add(statistics.rel_property_histograms.len()),
1514    );
1515    rows.push(graph_statistic_count_row(
1516        statistics,
1517        "node_count",
1518        statistics.node_count,
1519    ));
1520    rows.push(graph_statistic_count_row(
1521        statistics,
1522        "relationship_count",
1523        statistics.relationship_count,
1524    ));
1525
1526    for (label_id, count) in &statistics.label_counts {
1527        let mut row = graph_statistic_count_row(statistics, "label_count", *count);
1528        row.insert(
1529            "label_name".to_string(),
1530            optional_string_value(catalog.label_name(*label_id)),
1531        );
1532        rows.push(row);
1533    }
1534    for (rel_type_id, count) in &statistics.rel_type_counts {
1535        let mut row = graph_statistic_count_row(statistics, "relationship_type_count", *count);
1536        row.insert(
1537            "relationship_type_name".to_string(),
1538            optional_string_value(catalog.rel_type_name(*rel_type_id)),
1539        );
1540        rows.push(row);
1541    }
1542    for (kind, counts) in [
1543        (
1544            "relationship_source_count",
1545            &statistics.rel_type_source_counts,
1546        ),
1547        (
1548            "relationship_target_count",
1549            &statistics.rel_type_target_counts,
1550        ),
1551    ] {
1552        for (rel_type_id, count) in counts {
1553            let mut row = graph_statistic_count_row(statistics, kind, *count);
1554            row.insert(
1555                "relationship_type_name".to_string(),
1556                optional_string_value(catalog.rel_type_name(*rel_type_id)),
1557            );
1558            rows.push(row);
1559        }
1560    }
1561    for index in catalog.property_indexes() {
1562        let Some(sample) = statistics.index_samples.get(&index.id) else {
1563            continue;
1564        };
1565        let mut row = graph_statistic_base_row(statistics, "index_sample");
1566        row.insert("index_id".to_string(), u32_value(index.id.0));
1567        row.insert(
1568            "index_kind".to_string(),
1569            Value::String(index_kind_name(index.kind).to_string()),
1570        );
1571        row.insert(
1572            "label_name".to_string(),
1573            optional_string_value(catalog.label_name(index.label_id)),
1574        );
1575        row.insert(
1576            "property_name".to_string(),
1577            Value::String(index.property.clone()),
1578        );
1579        populate_index_sample(&mut row, *sample);
1580        rows.push(row);
1581    }
1582    for index in catalog.composite_property_indexes() {
1583        let Some(sample) = statistics.index_samples.get(&index.id) else {
1584            continue;
1585        };
1586        let mut row = graph_statistic_base_row(statistics, "index_sample");
1587        row.insert("index_id".to_string(), u32_value(index.id.0));
1588        row.insert(
1589            "index_kind".to_string(),
1590            Value::String("composite".to_string()),
1591        );
1592        row.insert(
1593            "label_name".to_string(),
1594            optional_string_value(catalog.label_name(index.label_id)),
1595        );
1596        row.insert(
1597            "index_properties".to_string(),
1598            Value::List(
1599                index
1600                    .properties
1601                    .iter()
1602                    .cloned()
1603                    .map(Value::String)
1604                    .collect(),
1605            ),
1606        );
1607        populate_index_sample(&mut row, *sample);
1608        rows.push(row);
1609    }
1610    for (kind, counts) in [
1611        ("path_count", &statistics.path_counts),
1612        (
1613            "path_source_distinct_count",
1614            &statistics.path_source_distinct_counts,
1615        ),
1616        (
1617            "path_target_distinct_count",
1618            &statistics.path_target_distinct_counts,
1619        ),
1620    ] {
1621        for ((source_label_id, rel_type_id, target_label_id), count) in counts {
1622            let mut row = graph_statistic_count_row(statistics, kind, *count);
1623            populate_path_statistic_names(
1624                &mut row,
1625                catalog,
1626                *source_label_id,
1627                *rel_type_id,
1628                *target_label_id,
1629            );
1630            rows.push(row);
1631        }
1632    }
1633    for (kind, counts) in [
1634        ("bounded_path_count", &statistics.bounded_path_counts),
1635        (
1636            "bounded_path_source_distinct_count",
1637            &statistics.bounded_path_source_distinct_counts,
1638        ),
1639        (
1640            "bounded_path_target_distinct_count",
1641            &statistics.bounded_path_target_distinct_counts,
1642        ),
1643    ] {
1644        for ((source_label_id, rel_type_id, target_label_id, depth), count) in counts {
1645            let mut row = graph_statistic_count_row(statistics, kind, *count);
1646            populate_path_statistic_names(
1647                &mut row,
1648                catalog,
1649                *source_label_id,
1650                *rel_type_id,
1651                *target_label_id,
1652            );
1653            row.insert("depth".to_string(), usize_value(*depth));
1654            rows.push(row);
1655        }
1656    }
1657    for ((label_id, property_name), count) in &statistics.property_distinct_counts {
1658        let mut row = graph_statistic_count_row(statistics, "node_property_distinct_count", *count);
1659        row.insert(
1660            "label_name".to_string(),
1661            optional_string_value(catalog.label_name(*label_id)),
1662        );
1663        row.insert(
1664            "property_name".to_string(),
1665            Value::String(property_name.clone()),
1666        );
1667        rows.push(row);
1668    }
1669    for ((rel_type_id, property_name), count) in &statistics.rel_property_distinct_counts {
1670        let mut row =
1671            graph_statistic_count_row(statistics, "relationship_property_distinct_count", *count);
1672        row.insert(
1673            "relationship_type_name".to_string(),
1674            optional_string_value(catalog.rel_type_name(*rel_type_id)),
1675        );
1676        row.insert(
1677            "property_name".to_string(),
1678            Value::String(property_name.clone()),
1679        );
1680        rows.push(row);
1681    }
1682    for ((label_id, property_name), histogram) in &statistics.property_histograms {
1683        let mut row = graph_statistic_base_row(statistics, "node_property_histogram");
1684        row.insert(
1685            "label_name".to_string(),
1686            optional_string_value(catalog.label_name(*label_id)),
1687        );
1688        row.insert(
1689            "property_name".to_string(),
1690            Value::String(property_name.clone()),
1691        );
1692        row.insert("histogram".to_string(), Value::List(histogram.clone()));
1693        row.insert(
1694            "sampled".to_string(),
1695            Value::Bool(
1696                statistics
1697                    .sampled_property_histograms
1698                    .get(&(*label_id, property_name.clone()))
1699                    .copied()
1700                    .unwrap_or(false),
1701            ),
1702        );
1703        rows.push(row);
1704    }
1705    for ((rel_type_id, property_name), histogram) in &statistics.rel_property_histograms {
1706        let mut row = graph_statistic_base_row(statistics, "relationship_property_histogram");
1707        row.insert(
1708            "relationship_type_name".to_string(),
1709            optional_string_value(catalog.rel_type_name(*rel_type_id)),
1710        );
1711        row.insert(
1712            "property_name".to_string(),
1713            Value::String(property_name.clone()),
1714        );
1715        row.insert("histogram".to_string(), Value::List(histogram.clone()));
1716        row.insert(
1717            "sampled".to_string(),
1718            Value::Bool(
1719                statistics
1720                    .sampled_rel_property_histograms
1721                    .get(&(*rel_type_id, property_name.clone()))
1722                    .copied()
1723                    .unwrap_or(false),
1724            ),
1725        );
1726        rows.push(row);
1727    }
1728    rows
1729}
1730
1731fn graph_statistic_count_row(statistics: &GraphStatistics, kind: &str, count: u64) -> Row {
1732    let mut row = graph_statistic_base_row(statistics, kind);
1733    row.insert("count".to_string(), u64_value(count));
1734    row
1735}
1736
1737fn populate_index_sample(row: &mut Row, sample: IndexStatisticsSample) {
1738    row.insert("index_size".to_string(), u64_value(sample.index_size));
1739    row.insert("unique_values".to_string(), u64_value(sample.unique_values));
1740    row.insert("sample_size".to_string(), u64_value(sample.sample_size));
1741    row.insert(
1742        "updates_since_sample".to_string(),
1743        u64_value(sample.updates_since_sample),
1744    );
1745    row.insert("stale".to_string(), Value::Bool(sample.is_stale()));
1746}
1747
1748fn graph_statistic_base_row(statistics: &GraphStatistics, kind: &str) -> Row {
1749    BTreeMap::from([
1750        (
1751            "statistic_kind".to_string(),
1752            Value::String(kind.to_string()),
1753        ),
1754        (
1755            "computed_at_commit_epoch".to_string(),
1756            u64_value(statistics.computed_at_commit_epoch),
1757        ),
1758        (
1759            "advanced_statistics_complete".to_string(),
1760            Value::Bool(statistics.advanced_statistics_complete),
1761        ),
1762        (
1763            "histogram_sample_limit".to_string(),
1764            usize_value(statistics.histogram_sample_limit),
1765        ),
1766        ("label_name".to_string(), Value::Null),
1767        ("relationship_type_name".to_string(), Value::Null),
1768        ("source_label_name".to_string(), Value::Null),
1769        ("target_label_name".to_string(), Value::Null),
1770        ("depth".to_string(), Value::Null),
1771        ("property_name".to_string(), Value::Null),
1772        ("index_id".to_string(), Value::Null),
1773        ("index_kind".to_string(), Value::Null),
1774        ("index_properties".to_string(), Value::Null),
1775        ("index_size".to_string(), Value::Null),
1776        ("unique_values".to_string(), Value::Null),
1777        ("sample_size".to_string(), Value::Null),
1778        ("updates_since_sample".to_string(), Value::Null),
1779        ("stale".to_string(), Value::Null),
1780        ("count".to_string(), Value::Null),
1781        ("histogram".to_string(), Value::Null),
1782        ("sampled".to_string(), Value::Null),
1783    ])
1784}
1785
1786fn populate_path_statistic_names(
1787    row: &mut Row,
1788    catalog: &Catalog,
1789    source_label_id: LabelId,
1790    rel_type_id: RelTypeId,
1791    target_label_id: LabelId,
1792) {
1793    row.insert(
1794        "source_label_name".to_string(),
1795        optional_string_value(catalog.label_name(source_label_id)),
1796    );
1797    row.insert(
1798        "relationship_type_name".to_string(),
1799        optional_string_value(catalog.rel_type_name(rel_type_id)),
1800    );
1801    row.insert(
1802        "target_label_name".to_string(),
1803        optional_string_value(catalog.label_name(target_label_id)),
1804    );
1805}
1806
1807fn projected_graph_rows(statuses: Vec<ProjectedGraphStatus>) -> Vec<Row> {
1808    statuses
1809        .into_iter()
1810        .map(|status| {
1811            BTreeMap::from([
1812                ("name".to_string(), Value::String(status.name)),
1813                (
1814                    "node_labels".to_string(),
1815                    Value::List(status.node_labels.into_iter().map(Value::String).collect()),
1816                ),
1817                (
1818                    "relationship_types".to_string(),
1819                    Value::List(status.rel_types.into_iter().map(Value::String).collect()),
1820                ),
1821                (
1822                    "projection_epoch".to_string(),
1823                    option_u64_value(status.projection_epoch),
1824                ),
1825                (
1826                    "commit_epoch".to_string(),
1827                    option_u64_value(status.commit_epoch),
1828                ),
1829                (
1830                    "node_count".to_string(),
1831                    option_usize_value(status.node_count),
1832                ),
1833                (
1834                    "edge_count".to_string(),
1835                    option_usize_value(status.edge_count),
1836                ),
1837                ("reusable".to_string(), Value::Bool(status.reusable)),
1838            ])
1839        })
1840        .collect()
1841}
1842
1843fn search_projection_changefeed_rows(status: SearchProjectionChangefeedStatus) -> Vec<Row> {
1844    vec![BTreeMap::from([
1845        (
1846            "graph_commit_epoch".to_string(),
1847            u64_value(status.graph_commit_epoch),
1848        ),
1849        (
1850            "resume_floor_commit_epoch".to_string(),
1851            u64_value(status.resume_floor_commit_epoch),
1852        ),
1853        (
1854            "oldest_retained_mutation_id".to_string(),
1855            option_u64_value(
1856                status
1857                    .oldest_retained_mutation_id
1858                    .map(|id| id.commit_epoch()),
1859            ),
1860        ),
1861        (
1862            "newest_retained_mutation_id".to_string(),
1863            option_u64_value(
1864                status
1865                    .newest_retained_mutation_id
1866                    .map(|id| id.commit_epoch()),
1867            ),
1868        ),
1869        (
1870            "retained_mutation_count".to_string(),
1871            usize_value(status.retained_mutation_count),
1872        ),
1873        (
1874            "restart_recoverable".to_string(),
1875            Value::Bool(status.restart_recoverable),
1876        ),
1877    ])]
1878}
1879
1880fn optional_string_value(value: Option<&str>) -> Value {
1881    value
1882        .map(|value| Value::String(value.to_string()))
1883        .unwrap_or(Value::Null)
1884}
1885
1886const fn table_kind_name(kind: TableKind) -> &'static str {
1887    match kind {
1888        TableKind::Node => "node",
1889        TableKind::Relationship => "relationship",
1890    }
1891}
1892
1893const fn schema_object_state_name(state: SchemaObjectState) -> &'static str {
1894    match state {
1895        SchemaObjectState::DeleteOnly => "delete_only",
1896        SchemaObjectState::WriteOnly => "write_only",
1897        SchemaObjectState::Backfill => "backfill",
1898        SchemaObjectState::Validating => "validating",
1899        SchemaObjectState::Public => "public",
1900        SchemaObjectState::Gc => "gc",
1901    }
1902}
1903
1904const fn property_type_name(value_type: PropertyType) -> &'static str {
1905    match value_type {
1906        PropertyType::Any => "any",
1907        PropertyType::Bool => "bool",
1908        PropertyType::Int => "int",
1909        PropertyType::Float => "float",
1910        PropertyType::String => "string",
1911        PropertyType::Text => "text",
1912        PropertyType::List => "list",
1913    }
1914}
1915
1916const fn index_kind_name(kind: IndexKind) -> &'static str {
1917    match kind {
1918        IndexKind::Equality => "equality",
1919        IndexKind::Range => "range",
1920        IndexKind::FullText => "full_text",
1921    }
1922}
1923
1924const fn constraint_kind_name(kind: ConstraintKind) -> &'static str {
1925    match kind {
1926        ConstraintKind::NodePropertyUnique => "node_property_unique",
1927        ConstraintKind::NodePropertyExists => "node_property_exists",
1928        ConstraintKind::RelationshipPropertyUnique => "relationship_property_unique",
1929        ConstraintKind::RelationshipPropertyExists => "relationship_property_exists",
1930    }
1931}
1932
1933fn plan_cache_rows(stats: &PlanCacheStats) -> Vec<Row> {
1934    [
1935        ("max_entries", option_usize_value(stats.max_entries)),
1936        ("entries", usize_value(stats.entries)),
1937        ("hits", u64_value(stats.hits)),
1938        ("misses", u64_value(stats.misses)),
1939        ("admissions", u64_value(stats.admissions)),
1940        ("disabled_misses", u64_value(stats.disabled_misses)),
1941        ("bypasses", u64_value(stats.bypasses)),
1942        ("evictions", u64_value(stats.evictions)),
1943        (
1944            "memory_pressure_events",
1945            u64_value(stats.memory_pressure_events),
1946        ),
1947    ]
1948    .into_iter()
1949    .map(|(metric, value)| {
1950        BTreeMap::from([
1951            ("metric".to_string(), Value::String(metric.to_string())),
1952            ("value".to_string(), value),
1953        ])
1954    })
1955    .collect()
1956}
1957
1958fn slow_query_rows(records: &[SlowQueryRecord]) -> Vec<Row> {
1959    records
1960        .iter()
1961        .map(|record| {
1962            BTreeMap::from([
1963                ("sequence".to_string(), u64_value(record.sequence)),
1964                (
1965                    "query_language".to_string(),
1966                    Value::String(record.query_language.clone()),
1967                ),
1968                (
1969                    "statement_kind".to_string(),
1970                    Value::String(record.statement_kind.clone()),
1971                ),
1972                (
1973                    "query_digest".to_string(),
1974                    Value::String(record.query_digest.clone()),
1975                ),
1976                (
1977                    "query_text_hash".to_string(),
1978                    Value::String(record.query_text_hash.clone()),
1979                ),
1980                (
1981                    "query_text".to_string(),
1982                    Value::String(record.query_text.clone()),
1983                ),
1984                (
1985                    "started_unix_micros".to_string(),
1986                    Value::Int(record.started_unix_micros),
1987                ),
1988                (
1989                    "elapsed_micros".to_string(),
1990                    Value::Int(record.elapsed_micros),
1991                ),
1992                ("row_count".to_string(), Value::Int(record.row_count)),
1993                ("success".to_string(), Value::Bool(record.success)),
1994                (
1995                    "error".to_string(),
1996                    record
1997                        .error
1998                        .as_ref()
1999                        .map(|error| Value::String(error.clone()))
2000                        .unwrap_or(Value::Null),
2001                ),
2002                (
2003                    "slow_log_candidate".to_string(),
2004                    Value::Bool(record.slow_log_candidate),
2005                ),
2006                (
2007                    "access_control_policy_epoch".to_string(),
2008                    record
2009                        .access_control_policy_epoch
2010                        .map(u64_value)
2011                        .unwrap_or(Value::Null),
2012                ),
2013            ])
2014        })
2015        .collect()
2016}
2017
2018fn statement_summary_rows(records: &[StatementSummaryRecord]) -> Vec<Row> {
2019    records
2020        .iter()
2021        .map(|record| {
2022            BTreeMap::from([
2023                ("digest".to_string(), Value::String(record.digest.clone())),
2024                (
2025                    "query_language".to_string(),
2026                    Value::String(record.query_language.clone()),
2027                ),
2028                (
2029                    "query_text".to_string(),
2030                    Value::String(record.query_text.clone()),
2031                ),
2032                (
2033                    "sample_query_text_hash".to_string(),
2034                    Value::String(record.sample_query_text_hash.clone()),
2035                ),
2036                (
2037                    "statement_kind".to_string(),
2038                    Value::String(record.statement_kind.clone()),
2039                ),
2040                (
2041                    "execution_count".to_string(),
2042                    Value::Int(record.execution_count),
2043                ),
2044                (
2045                    "success_count".to_string(),
2046                    Value::Int(record.success_count),
2047                ),
2048                ("error_count".to_string(), Value::Int(record.error_count)),
2049                (
2050                    "total_elapsed_micros".to_string(),
2051                    Value::Int(record.total_elapsed_micros),
2052                ),
2053                (
2054                    "max_elapsed_micros".to_string(),
2055                    Value::Int(record.max_elapsed_micros),
2056                ),
2057                (
2058                    "avg_elapsed_micros".to_string(),
2059                    Value::Int(avg_i64(record.total_elapsed_micros, record.execution_count)),
2060                ),
2061                (
2062                    "total_row_count".to_string(),
2063                    Value::Int(record.total_row_count),
2064                ),
2065                (
2066                    "last_seen_unix_micros".to_string(),
2067                    Value::Int(record.last_seen_unix_micros),
2068                ),
2069                (
2070                    "last_elapsed_micros".to_string(),
2071                    Value::Int(record.last_elapsed_micros),
2072                ),
2073                (
2074                    "last_row_count".to_string(),
2075                    Value::Int(record.last_row_count),
2076                ),
2077                ("last_success".to_string(), Value::Bool(record.last_success)),
2078                (
2079                    "last_error".to_string(),
2080                    record
2081                        .last_error
2082                        .as_ref()
2083                        .map(|error| Value::String(error.clone()))
2084                        .unwrap_or(Value::Null),
2085                ),
2086            ])
2087        })
2088        .collect()
2089}
2090
2091fn predicate_matches(predicate: &SqlPredicate, row: &Row) -> bool {
2092    match &predicate.kind {
2093        ExprKind::And(left, right) => predicate_matches(left, row) && predicate_matches(right, row),
2094        ExprKind::Or(left, right) => predicate_matches(left, row) || predicate_matches(right, row),
2095        ExprKind::Not(inner) => !predicate_matches(inner, row),
2096        ExprKind::Compare { left, op, right } => system_expression_value(left, row)
2097            .zip(system_expression_value(right, row))
2098            .is_some_and(|(left, right)| compare_values(left, *op, right)),
2099        ExprKind::InList {
2100            left,
2101            values,
2102            negated,
2103        } => {
2104            let matched = system_expression_value(left, row).is_some_and(|left| {
2105                values.iter().any(|value| {
2106                    system_expression_value(value, row).is_some_and(|value| left == value)
2107                })
2108            });
2109            matched != *negated
2110        }
2111        ExprKind::Like {
2112            left,
2113            pattern,
2114            case_insensitive,
2115            negated,
2116            escape,
2117        } => {
2118            let Some((Value::String(value), Value::String(pattern))) =
2119                system_expression_value(left, row).zip(system_expression_value(pattern, row))
2120            else {
2121                return false;
2122            };
2123            hawdb_sql::sql_like_matches(value, pattern, *escape, *case_insensitive)
2124                .is_ok_and(|matched| matched != *negated)
2125        }
2126        ExprKind::IsNull {
2127            expression,
2128            negated,
2129        } => {
2130            let matched = system_expression_value(expression, row)
2131                .is_none_or(|value| matches!(value, Value::Null));
2132            matched != *negated
2133        }
2134        _ => false,
2135    }
2136}
2137
2138fn system_expression_value<'a>(expression: &'a Expr, row: &'a Row) -> Option<&'a Value> {
2139    match &expression.kind {
2140        ExprKind::Column(column) => row.get(&column.name),
2141        ExprKind::Value(SqlValue::Literal(value)) => Some(value),
2142        _ => None,
2143    }
2144}
2145
2146fn compare_values(left: &Value, op: SqlComparisonOp, right: &Value) -> bool {
2147    if matches!(left, Value::Null) || matches!(right, Value::Null) {
2148        return false;
2149    }
2150    match op {
2151        SqlComparisonOp::Eq => left == right,
2152        SqlComparisonOp::NotEq => left != right,
2153        SqlComparisonOp::Lt => left < right,
2154        SqlComparisonOp::Lte => left <= right,
2155        SqlComparisonOp::Gt => left > right,
2156        SqlComparisonOp::Gte => left >= right,
2157    }
2158}
2159
2160fn compare_ordered_rows(left: &Row, right: &Row, order_by: &[SqlOrderItem]) -> Ordering {
2161    for item in order_by {
2162        let column = item
2163            .expression
2164            .as_column()
2165            .expect("system ORDER BY columns were validated");
2166        let ordering = left.get(&column.name).cmp(&right.get(&column.name));
2167        let ordering = match item.direction {
2168            SqlOrderDirection::Asc => ordering,
2169            SqlOrderDirection::Desc => ordering.reverse(),
2170        };
2171        if ordering != Ordering::Equal {
2172            return ordering;
2173        }
2174    }
2175    Ordering::Equal
2176}
2177
2178fn system_table(select: &SelectStatement) -> Result<SystemTable> {
2179    let from = select.from_table();
2180    match (from.schema.as_deref(), from.name.as_str()) {
2181        (Some("system"), "tables") => Ok(SystemTable::Tables),
2182        (Some("system"), "properties") => Ok(SystemTable::Properties),
2183        (Some("system"), "indexes") => Ok(SystemTable::Indexes),
2184        (Some("system"), "constraints") => Ok(SystemTable::Constraints),
2185        (Some("system"), "append_tables") => Ok(SystemTable::AppendTables),
2186        (Some("system"), "append_storage") => Ok(SystemTable::AppendStorage),
2187        (Some("system"), "runtime_status") => Ok(SystemTable::RuntimeStatus),
2188        (Some("system"), "runtime_capabilities") => Ok(SystemTable::RuntimeCapabilities),
2189        (Some("system"), "graph_statistics") => Ok(SystemTable::GraphStatistics),
2190        (Some("system"), "projected_graphs") => Ok(SystemTable::ProjectedGraphs),
2191        (Some("system"), "search_projection_changefeed") => {
2192            Ok(SystemTable::SearchProjectionChangefeed)
2193        }
2194        (Some("system"), "plan_cache") => Ok(SystemTable::PlanCache),
2195        (Some("system"), "slow_queries") => Ok(SystemTable::SlowQueries),
2196        (Some("system"), "statement_summary") => Ok(SystemTable::StatementSummary),
2197        (Some("information_schema"), "tables") => Ok(SystemTable::InformationSchemaTables),
2198        (Some("information_schema"), "columns") => Ok(SystemTable::InformationSchemaColumns),
2199        (Some("pg_catalog"), "pg_tables") => Ok(SystemTable::PgTables),
2200        (Some("pg_catalog"), "pg_indexes") => Ok(SystemTable::PgIndexes),
2201        (None, "pg_tables") => Ok(SystemTable::PgTables),
2202        (None, "pg_indexes") => Ok(SystemTable::PgIndexes),
2203        _ => Err(HawDBError::Semantic(format!(
2204            "unknown SQL virtual catalog table {}",
2205            format_table_name(select)
2206        ))),
2207    }
2208}
2209
2210fn validate_projection(table: SystemTable, projection: &[SelectProjection]) -> Result<()> {
2211    for projection in projection {
2212        match projection {
2213            SelectProjection::Wildcard => {}
2214            SelectProjection::Expression {
2215                expression:
2216                    Expr {
2217                        kind: ExprKind::Column(name),
2218                        ..
2219                    },
2220                ..
2221            } => validate_column(table, name)?,
2222            SelectProjection::Expression { .. } => {
2223                return Err(HawDBError::Semantic(
2224                    "system SQL aggregate expressions are not supported".to_string(),
2225                ));
2226            }
2227        }
2228    }
2229    Ok(())
2230}
2231
2232fn validate_predicate_columns(table: SystemTable, predicate: Option<&SqlPredicate>) -> Result<()> {
2233    let mut result = Ok(());
2234    if let Some(predicate) = predicate {
2235        predicate.visit(&mut |expression| {
2236            if result.is_ok()
2237                && let Some(column) = expression.as_column()
2238            {
2239                result = validate_column(table, column);
2240            }
2241        });
2242    }
2243    result
2244}
2245
2246fn validate_order_columns(table: SystemTable, order_by: &[SqlOrderItem]) -> Result<()> {
2247    for item in order_by {
2248        validate_column(table, item.expression.require_column()?)?;
2249    }
2250    Ok(())
2251}
2252
2253fn validate_column(table: SystemTable, column: &SqlColumnRef) -> Result<()> {
2254    if let Some(qualifier) = &column.qualifier {
2255        let table_name = match table {
2256            SystemTable::Tables => "tables",
2257            SystemTable::Properties => "properties",
2258            SystemTable::Indexes => "indexes",
2259            SystemTable::Constraints => "constraints",
2260            SystemTable::AppendTables => "append_tables",
2261            SystemTable::AppendStorage => "append_storage",
2262            SystemTable::RuntimeStatus => "runtime_status",
2263            SystemTable::RuntimeCapabilities => "runtime_capabilities",
2264            SystemTable::GraphStatistics => "graph_statistics",
2265            SystemTable::ProjectedGraphs => "projected_graphs",
2266            SystemTable::SearchProjectionChangefeed => "search_projection_changefeed",
2267            SystemTable::PlanCache => "plan_cache",
2268            SystemTable::SlowQueries => "slow_queries",
2269            SystemTable::StatementSummary => "statement_summary",
2270            SystemTable::InformationSchemaTables => "tables",
2271            SystemTable::InformationSchemaColumns => "columns",
2272            SystemTable::PgTables => "pg_tables",
2273            SystemTable::PgIndexes => "pg_indexes",
2274        };
2275        if qualifier != table_name {
2276            return Err(HawDBError::Semantic(format!(
2277                "unknown SQL column qualifier {qualifier}"
2278            )));
2279        }
2280    }
2281    if table_columns(table).contains(&column.name.as_str()) {
2282        Ok(())
2283    } else {
2284        Err(HawDBError::Semantic(format!(
2285            "unknown SQL column {}",
2286            column.name
2287        )))
2288    }
2289}
2290
2291fn table_columns(table: SystemTable) -> &'static [&'static str] {
2292    match table {
2293        SystemTable::Tables => &["table_id", "table_name", "table_kind", "state"],
2294        SystemTable::Properties => &[
2295            "property_id",
2296            "table_id",
2297            "table_name",
2298            "table_kind",
2299            "property_name",
2300            "value_type",
2301            "nullable",
2302            "state",
2303        ],
2304        SystemTable::Indexes => &[
2305            "index_id",
2306            "subject_kind",
2307            "subject_id",
2308            "subject_name",
2309            "index_kind",
2310            "property_names",
2311        ],
2312        SystemTable::Constraints => &[
2313            "constraint_id",
2314            "subject_kind",
2315            "subject_id",
2316            "subject_name",
2317            "property_name",
2318            "constraint_kind",
2319        ],
2320        SystemTable::AppendTables => &[
2321            "table_name",
2322            "storage_mode",
2323            "partition_key",
2324            "order_key",
2325            "order_mode",
2326            "generated_order_watermark",
2327            "column_count",
2328        ],
2329        SystemTable::AppendStorage => &[
2330            "canonical_segment_count",
2331            "canonical_segment_bytes",
2332            "resident_segment_payload_bytes",
2333            "resident_descriptor_count",
2334            "live_rows",
2335            "live_payload_bytes",
2336        ],
2337        SystemTable::RuntimeStatus => &[
2338            "commit_epoch",
2339            "read_only",
2340            "max_read_result_rows",
2341            "max_read_result_payload_bytes",
2342            "storage_residency_mode",
2343            "relational_index_mode",
2344        ],
2345        SystemTable::RuntimeCapabilities => &["capability", "enabled"],
2346        SystemTable::GraphStatistics => &[
2347            "statistic_kind",
2348            "computed_at_commit_epoch",
2349            "advanced_statistics_complete",
2350            "histogram_sample_limit",
2351            "label_name",
2352            "relationship_type_name",
2353            "source_label_name",
2354            "target_label_name",
2355            "depth",
2356            "property_name",
2357            "index_id",
2358            "index_kind",
2359            "index_properties",
2360            "index_size",
2361            "unique_values",
2362            "sample_size",
2363            "updates_since_sample",
2364            "stale",
2365            "count",
2366            "histogram",
2367            "sampled",
2368        ],
2369        SystemTable::ProjectedGraphs => &[
2370            "name",
2371            "node_labels",
2372            "relationship_types",
2373            "projection_epoch",
2374            "commit_epoch",
2375            "node_count",
2376            "edge_count",
2377            "reusable",
2378        ],
2379        SystemTable::SearchProjectionChangefeed => &[
2380            "graph_commit_epoch",
2381            "resume_floor_commit_epoch",
2382            "oldest_retained_mutation_id",
2383            "newest_retained_mutation_id",
2384            "retained_mutation_count",
2385            "restart_recoverable",
2386        ],
2387        SystemTable::PlanCache => &["metric", "value"],
2388        SystemTable::SlowQueries => &[
2389            "sequence",
2390            "query_language",
2391            "statement_kind",
2392            "query_digest",
2393            "query_text_hash",
2394            "query_text",
2395            "started_unix_micros",
2396            "elapsed_micros",
2397            "row_count",
2398            "success",
2399            "error",
2400            "slow_log_candidate",
2401            "access_control_policy_epoch",
2402        ],
2403        SystemTable::StatementSummary => &[
2404            "digest",
2405            "query_language",
2406            "query_text",
2407            "sample_query_text_hash",
2408            "statement_kind",
2409            "execution_count",
2410            "success_count",
2411            "error_count",
2412            "total_elapsed_micros",
2413            "max_elapsed_micros",
2414            "avg_elapsed_micros",
2415            "total_row_count",
2416            "last_seen_unix_micros",
2417            "last_elapsed_micros",
2418            "last_row_count",
2419            "last_success",
2420            "last_error",
2421        ],
2422        SystemTable::InformationSchemaTables => &[
2423            "table_catalog",
2424            "table_schema",
2425            "table_name",
2426            "table_type",
2427            "self_referencing_column_name",
2428            "reference_generation",
2429            "user_defined_type_catalog",
2430            "user_defined_type_schema",
2431            "user_defined_type_name",
2432            "is_insertable_into",
2433            "is_typed",
2434            "commit_action",
2435        ],
2436        SystemTable::InformationSchemaColumns => &[
2437            "table_catalog",
2438            "table_schema",
2439            "table_name",
2440            "column_name",
2441            "ordinal_position",
2442            "column_default",
2443            "is_nullable",
2444            "data_type",
2445            "character_maximum_length",
2446            "character_octet_length",
2447            "numeric_precision",
2448            "numeric_precision_radix",
2449            "numeric_scale",
2450            "datetime_precision",
2451            "interval_type",
2452            "interval_precision",
2453            "character_set_catalog",
2454            "character_set_schema",
2455            "character_set_name",
2456            "collation_catalog",
2457            "collation_schema",
2458            "collation_name",
2459            "domain_catalog",
2460            "domain_schema",
2461            "domain_name",
2462            "udt_catalog",
2463            "udt_schema",
2464            "udt_name",
2465            "scope_catalog",
2466            "scope_schema",
2467            "scope_name",
2468            "maximum_cardinality",
2469            "dtd_identifier",
2470            "is_self_referencing",
2471            "is_identity",
2472            "identity_generation",
2473            "identity_start",
2474            "identity_increment",
2475            "identity_maximum",
2476            "identity_minimum",
2477            "identity_cycle",
2478            "is_generated",
2479            "generation_expression",
2480            "is_updatable",
2481        ],
2482        SystemTable::PgTables => &[
2483            "schemaname",
2484            "tablename",
2485            "tableowner",
2486            "tablespace",
2487            "hasindexes",
2488            "hasrules",
2489            "hastriggers",
2490            "rowsecurity",
2491        ],
2492        SystemTable::PgIndexes => &[
2493            "schemaname",
2494            "tablename",
2495            "indexname",
2496            "tablespace",
2497            "indexdef",
2498        ],
2499    }
2500}
2501
2502fn format_table_name(select: &SelectStatement) -> String {
2503    let from = select.from_table();
2504    match &from.schema {
2505        Some(schema) => format!("{schema}.{}", from.name),
2506        None => from.name.clone(),
2507    }
2508}
2509
2510fn option_usize_value(value: Option<usize>) -> Value {
2511    value.map(usize_value).unwrap_or(Value::Null)
2512}
2513
2514fn option_i64_value(value: Option<i64>) -> Value {
2515    value.map(Value::Int).unwrap_or(Value::Null)
2516}
2517
2518fn option_u64_value(value: Option<u64>) -> Value {
2519    value.map(u64_value).unwrap_or(Value::Null)
2520}
2521
2522fn usize_value(value: usize) -> Value {
2523    Value::Int(i64::try_from(value).unwrap_or(i64::MAX))
2524}
2525
2526fn u32_value(value: u32) -> Value {
2527    Value::Int(i64::from(value))
2528}
2529
2530fn u64_value(value: u64) -> Value {
2531    Value::Int(i64::try_from(value).unwrap_or(i64::MAX))
2532}
2533
2534const fn storage_residency_mode_name(mode: StorageResidencyMode) -> &'static str {
2535    match mode {
2536        StorageResidencyMode::Auto => "auto",
2537        StorageResidencyMode::Materialized => "materialized",
2538        StorageResidencyMode::OutOfCore => "out_of_core",
2539    }
2540}
2541
2542const fn relational_index_mode_name(mode: RelationalIndexMode) -> &'static str {
2543    match mode {
2544        RelationalIndexMode::Materialized => "materialized",
2545        RelationalIndexMode::Shadow => "shadow",
2546        RelationalIndexMode::DemandPaged => "demand_paged",
2547        RelationalIndexMode::Authoritative => "authoritative",
2548    }
2549}
2550
2551fn avg_i64(total: i64, count: i64) -> i64 {
2552    if count <= 0 {
2553        0
2554    } else {
2555        total / count
2556    }
2557}
2558
2559fn saturating_i64_from_u128(value: u128) -> i64 {
2560    i64::try_from(value).unwrap_or(i64::MAX)
2561}
2562
2563fn unix_now_micros() -> i64 {
2564    let micros = SystemTime::now()
2565        .duration_since(UNIX_EPOCH)
2566        .map(|duration| duration.as_micros())
2567        .unwrap_or(0);
2568    saturating_i64_from_u128(micros)
2569}
2570
2571fn truncate_utf8(input: &str, max_bytes: usize) -> String {
2572    if input.len() <= max_bytes {
2573        return input.to_string();
2574    }
2575    let mut end = max_bytes;
2576    while !input.is_char_boundary(end) {
2577        end -= 1;
2578    }
2579    input[..end].to_string()
2580}
2581
2582mod graph_store_adapter;
2583
2584#[cfg(test)]
2585mod tests {
2586    use super::*;
2587
2588    #[derive(Debug, Default)]
2589    struct TestStore;
2590
2591    impl SystemSqlStore for TestStore {
2592        fn commit_epoch(&self) -> u64 {
2593            0
2594        }
2595
2596        fn append_storage_residency_report(&self) -> AppendStorageResidencyReport {
2597            AppendStorageResidencyReport::default()
2598        }
2599
2600        fn statistics(&self, _catalog: &Catalog) -> GraphStatistics {
2601            GraphStatistics::default()
2602        }
2603
2604        fn projected_graph_statuses(&self) -> Vec<ProjectedGraphStatus> {
2605            Vec::new()
2606        }
2607
2608        fn search_projection_changefeed_status(&self) -> SearchProjectionChangefeedStatus {
2609            SearchProjectionChangefeedStatus {
2610                graph_commit_epoch: 0,
2611                resume_floor_commit_epoch: 0,
2612                oldest_retained_mutation_id: None,
2613                newest_retained_mutation_id: None,
2614                first_rebuild_required_mutation_id: None,
2615                retained_mutation_count: 0,
2616                retained_bytes: 0,
2617                max_retained_bytes: None,
2618                restart_recoverable: true,
2619            }
2620        }
2621    }
2622
2623    fn default_runtime() -> SystemRuntimeSnapshot {
2624        SystemRuntimeSnapshot::new(
2625            false,
2626            Some(512),
2627            Some(4 * 1024 * 1024),
2628            StorageResidencyMode::Auto,
2629            RelationalIndexMode::default(),
2630            RuntimeCapabilities::default(),
2631        )
2632    }
2633
2634    #[test]
2635    fn predicate_binding_preserves_source_spans_and_system_null_behavior() {
2636        let SqlStatement::Select(select) = hawdb_sql::prepare_postgres_sql(
2637            "SELECT value FROM system.plan_cache WHERE \
2638             (value IN ($1, $2) OR NOT value = $3) AND \
2639             (metric LIKE $4 OR value IS NULL)",
2640        )
2641        .unwrap()
2642        .statement
2643        else {
2644            panic!("expected SELECT");
2645        };
2646        let predicate = select.selection.unwrap();
2647        let mut before = Vec::new();
2648        predicate.visit(&mut |node| before.push(node.span));
2649        let bound = bind_predicate(
2650            predicate.clone(),
2651            &[
2652                Value::Null,
2653                Value::Int(5),
2654                Value::Null,
2655                Value::String("hit%".to_owned()),
2656            ],
2657        )
2658        .unwrap();
2659        let mut after = Vec::new();
2660        let mut parameters = Vec::new();
2661        bound.visit(&mut |node| {
2662            after.push(node.span);
2663            if let ExprKind::Value(SqlValue::Parameter(position)) = &node.kind {
2664                parameters.push(*position);
2665            }
2666        });
2667        assert_eq!(after, before);
2668        assert!(parameters.is_empty());
2669        assert!(predicate_matches(
2670            &bound,
2671            &BTreeMap::from([
2672                ("value".to_owned(), Value::Null),
2673                ("metric".to_owned(), Value::String("misses".to_owned())),
2674            ])
2675        ));
2676        assert!(bind_predicate(
2677            predicate,
2678            &[
2679                Value::Null,
2680                Value::Int(5),
2681                Value::Null,
2682                Value::String("dangling\\".to_owned()),
2683            ]
2684        )
2685        .unwrap_err()
2686        .to_string()
2687        .contains("ends with its escape"));
2688    }
2689
2690    #[test]
2691    fn query_plan_cache_virtual_table_with_predicate_and_projection() {
2692        let catalog = Catalog::default();
2693        let store = TestStore;
2694        let relational_state = RelationalState::default();
2695        let append_state = AppendState::default();
2696        let stats = PlanCacheStats {
2697            max_entries: Some(128),
2698            entries: 3,
2699            hits: 5,
2700            misses: 7,
2701            admissions: 4,
2702            disabled_misses: 0,
2703            bypasses: 2,
2704            evictions: 1,
2705            memory_pressure_events: 1,
2706        };
2707
2708        let output = query_sql(
2709            "SELECT value FROM system.plan_cache WHERE metric = 'hits'",
2710            None,
2711            None,
2712            &SystemSqlContext {
2713                catalog: &catalog,
2714                store: &store,
2715                relational_state: &relational_state,
2716                append_state: &append_state,
2717                runtime: default_runtime(),
2718                plan_cache_stats: &stats,
2719                slow_queries: &[],
2720                statement_summaries: &[],
2721            },
2722        )
2723        .expect("system plan cache query");
2724
2725        assert_eq!(
2726            output.rows,
2727            vec![BTreeMap::from([("value".to_string(), Value::Int(5))])]
2728        );
2729    }
2730
2731    #[test]
2732    fn query_system_table_with_postgres_parameters() {
2733        let catalog = Catalog::default();
2734        let store = TestStore;
2735        let relational_state = RelationalState::default();
2736        let append_state = AppendState::default();
2737        let stats = PlanCacheStats {
2738            max_entries: Some(128),
2739            entries: 3,
2740            hits: 5,
2741            misses: 7,
2742            admissions: 4,
2743            disabled_misses: 0,
2744            bypasses: 2,
2745            evictions: 1,
2746            memory_pressure_events: 1,
2747        };
2748        let context = SystemSqlContext {
2749            catalog: &catalog,
2750            store: &store,
2751            relational_state: &relational_state,
2752            append_state: &append_state,
2753            runtime: default_runtime(),
2754            plan_cache_stats: &stats,
2755            slow_queries: &[],
2756            statement_summaries: &[],
2757        };
2758
2759        let output = query_sql_with_params(
2760            "SELECT metric, value FROM system.plan_cache \
2761             WHERE metric IN ($1, $2) ORDER BY metric ASC LIMIT $3 OFFSET $4",
2762            &[
2763                Value::String("hits".to_string()),
2764                Value::String("misses".to_string()),
2765                Value::Int(1),
2766                Value::Int(1),
2767            ],
2768            None,
2769            None,
2770            &context,
2771        )
2772        .expect("parameterized system table query");
2773
2774        assert_eq!(
2775            output.rows,
2776            vec![BTreeMap::from([
2777                ("metric".to_string(), Value::String("misses".to_string())),
2778                ("value".to_string(), Value::Int(7)),
2779            ])]
2780        );
2781    }
2782
2783    #[test]
2784    fn query_system_table_rejects_parameter_contract_mismatch() {
2785        let catalog = Catalog::default();
2786        let store = TestStore;
2787        let relational_state = RelationalState::default();
2788        let append_state = AppendState::default();
2789        let stats = PlanCacheStats {
2790            max_entries: None,
2791            entries: 0,
2792            hits: 0,
2793            misses: 0,
2794            admissions: 0,
2795            disabled_misses: 0,
2796            bypasses: 0,
2797            evictions: 0,
2798            memory_pressure_events: 0,
2799        };
2800        let context = SystemSqlContext {
2801            catalog: &catalog,
2802            store: &store,
2803            relational_state: &relational_state,
2804            append_state: &append_state,
2805            runtime: default_runtime(),
2806            plan_cache_stats: &stats,
2807            slow_queries: &[],
2808            statement_summaries: &[],
2809        };
2810
2811        let missing = query_sql_with_params(
2812            "SELECT * FROM system.plan_cache WHERE metric = $1",
2813            &[],
2814            None,
2815            None,
2816            &context,
2817        )
2818        .expect_err("missing parameter must fail");
2819        assert!(missing.to_string().contains("requires 1 parameters"));
2820
2821        let invalid_bound = query_sql_with_params(
2822            "SELECT * FROM system.plan_cache LIMIT $1",
2823            &[Value::String("one".to_string())],
2824            None,
2825            None,
2826            &context,
2827        )
2828        .expect_err("non-integer LIMIT must fail");
2829        assert!(invalid_bound
2830            .to_string()
2831            .contains("must be a non-negative integer"));
2832    }
2833
2834    #[test]
2835    fn query_slow_queries_pushes_filter_order_and_limit_into_scan() {
2836        let catalog = Catalog::default();
2837        let store = TestStore;
2838        let relational_state = RelationalState::default();
2839        let append_state = AppendState::default();
2840        let stats = PlanCacheStats {
2841            max_entries: None,
2842            entries: 0,
2843            hits: 0,
2844            misses: 0,
2845            admissions: 0,
2846            disabled_misses: 0,
2847            bypasses: 0,
2848            evictions: 0,
2849            memory_pressure_events: 0,
2850        };
2851        let records = vec![
2852            SlowQueryRecord {
2853                sequence: 1,
2854                query_language: "cypher".to_string(),
2855                statement_kind: "match_return".to_string(),
2856                query_digest: "q1:first".to_string(),
2857                query_text_hash: "t1:first".to_string(),
2858                query_text: "MATCH (m:Memory) RETURN m".to_string(),
2859                started_unix_micros: 10,
2860                elapsed_micros: 200,
2861                row_count: 1,
2862                success: true,
2863                error: None,
2864                slow_log_candidate: false,
2865                access_control_policy_epoch: None,
2866                vector_execution_reports: Vec::new(),
2867            },
2868            SlowQueryRecord {
2869                sequence: 2,
2870                query_language: "cypher".to_string(),
2871                statement_kind: "match_return".to_string(),
2872                query_digest: "q1:second".to_string(),
2873                query_text_hash: "t1:second".to_string(),
2874                query_text: "MATCH (m:Memory) RETURN m ORDER BY m.id".to_string(),
2875                started_unix_micros: 20,
2876                elapsed_micros: 500,
2877                row_count: 2,
2878                success: true,
2879                error: None,
2880                slow_log_candidate: true,
2881                access_control_policy_epoch: None,
2882                vector_execution_reports: Vec::new(),
2883            },
2884            SlowQueryRecord {
2885                sequence: 3,
2886                query_language: "cypher".to_string(),
2887                statement_kind: "match_return".to_string(),
2888                query_digest: "q1:third".to_string(),
2889                query_text_hash: "t1:third".to_string(),
2890                query_text: "MATCH (m:Memory {id: 'x'}) RETURN m".to_string(),
2891                started_unix_micros: 30,
2892                elapsed_micros: 300,
2893                row_count: 1,
2894                success: true,
2895                error: None,
2896                slow_log_candidate: true,
2897                access_control_policy_epoch: None,
2898                vector_execution_reports: Vec::new(),
2899            },
2900        ];
2901
2902        let output = query_sql(
2903            "SELECT sequence, elapsed_micros FROM system.slow_queries \
2904             WHERE slow_log_candidate = true \
2905             ORDER BY elapsed_micros DESC LIMIT 1",
2906            None,
2907            None,
2908            &SystemSqlContext {
2909                catalog: &catalog,
2910                store: &store,
2911                relational_state: &relational_state,
2912                append_state: &append_state,
2913                runtime: default_runtime(),
2914                plan_cache_stats: &stats,
2915                slow_queries: &records,
2916                statement_summaries: &[],
2917            },
2918        )
2919        .expect("system slow query scan");
2920
2921        assert_eq!(
2922            output.rows,
2923            vec![BTreeMap::from([
2924                ("elapsed_micros".to_string(), Value::Int(500)),
2925                ("sequence".to_string(), Value::Int(2)),
2926            ])]
2927        );
2928    }
2929
2930    #[test]
2931    fn query_catalog_virtual_tables_expose_schema_without_typed_getters() {
2932        let mut catalog = Catalog::default();
2933        let store = TestStore;
2934        let relational_state = RelationalState::default();
2935        let append_state = AppendState::default();
2936        let memory_label = catalog.get_or_create_label("Memory");
2937        let memory_table = catalog.get_or_create_table(TableKind::Node, "Memory");
2938        catalog.get_or_create_property(memory_table, "id", PropertyType::String, false);
2939        catalog.get_or_create_property_index_with_kind(memory_label, "id", IndexKind::Equality);
2940        catalog.get_or_create_unique_constraint(memory_label, "id");
2941        let stats = PlanCacheStats {
2942            max_entries: None,
2943            entries: 0,
2944            hits: 0,
2945            misses: 0,
2946            admissions: 0,
2947            disabled_misses: 0,
2948            bypasses: 0,
2949            evictions: 0,
2950            memory_pressure_events: 0,
2951        };
2952        let context = SystemSqlContext {
2953            catalog: &catalog,
2954            store: &store,
2955            relational_state: &relational_state,
2956            append_state: &append_state,
2957            runtime: default_runtime(),
2958            plan_cache_stats: &stats,
2959            slow_queries: &[],
2960            statement_summaries: &[],
2961        };
2962
2963        let tables = query_sql(
2964            "SELECT table_id, table_name, table_kind, state FROM system.tables",
2965            None,
2966            None,
2967            &context,
2968        )
2969        .expect("system tables query");
2970        assert_eq!(
2971            tables.rows,
2972            vec![BTreeMap::from([
2973                ("state".to_string(), Value::String("public".to_string())),
2974                ("table_id".to_string(), Value::Int(0)),
2975                ("table_kind".to_string(), Value::String("node".to_string()),),
2976                (
2977                    "table_name".to_string(),
2978                    Value::String("Memory".to_string()),
2979                ),
2980            ])]
2981        );
2982
2983        let properties = query_sql(
2984            "SELECT table_name, property_name, value_type, nullable \
2985             FROM system.properties WHERE table_name = 'Memory'",
2986            None,
2987            None,
2988            &context,
2989        )
2990        .expect("system properties query");
2991        assert_eq!(
2992            properties.rows,
2993            vec![BTreeMap::from([
2994                ("nullable".to_string(), Value::Bool(false)),
2995                ("property_name".to_string(), Value::String("id".to_string()),),
2996                (
2997                    "table_name".to_string(),
2998                    Value::String("Memory".to_string()),
2999                ),
3000                (
3001                    "value_type".to_string(),
3002                    Value::String("string".to_string()),
3003                ),
3004            ])]
3005        );
3006
3007        let indexes = query_sql(
3008            "SELECT subject_name, index_kind, property_names FROM system.indexes",
3009            None,
3010            None,
3011            &context,
3012        )
3013        .expect("system indexes query");
3014        assert_eq!(
3015            indexes.rows,
3016            vec![BTreeMap::from([
3017                (
3018                    "index_kind".to_string(),
3019                    Value::String("equality".to_string()),
3020                ),
3021                (
3022                    "property_names".to_string(),
3023                    Value::List(vec![Value::String("id".to_string())]),
3024                ),
3025                (
3026                    "subject_name".to_string(),
3027                    Value::String("Memory".to_string()),
3028                ),
3029            ])]
3030        );
3031
3032        let constraints = query_sql(
3033            "SELECT subject_name, property_name, constraint_kind FROM system.constraints",
3034            None,
3035            None,
3036            &context,
3037        )
3038        .expect("system constraints query");
3039        assert_eq!(
3040            constraints.rows,
3041            vec![BTreeMap::from([
3042                (
3043                    "constraint_kind".to_string(),
3044                    Value::String("node_property_unique".to_string()),
3045                ),
3046                ("property_name".to_string(), Value::String("id".to_string()),),
3047                (
3048                    "subject_name".to_string(),
3049                    Value::String("Memory".to_string()),
3050                ),
3051            ])]
3052        );
3053    }
3054}
3055
3056#[cfg(all(test, feature = "loom-tests"))]
3057pub(crate) mod loom_tests {
3058    use super::{SlowQueryCompletion, SlowQueryLog, SlowQueryRecord};
3059    use hawdb_query::QueryIdentity;
3060    use loom::sync::{Arc, Mutex};
3061    use loom::thread;
3062
3063    #[test]
3064    fn slow_query_ring_preserves_bounds_under_modeled_concurrent_access() {
3065        loom::model(|| {
3066            let log = Arc::new(Mutex::new(SlowQueryLog::new(2)));
3067
3068            let first_writer = spawn_slow_query_writer(Arc::clone(&log), "first");
3069            let second_writer = spawn_slow_query_writer(Arc::clone(&log), "second");
3070            let snapshotter = {
3071                let log = Arc::clone(&log);
3072                thread::spawn(move || {
3073                    let snapshot = log.lock().unwrap().snapshot();
3074                    assert_snapshot_invariants(&snapshot, 2);
3075                })
3076            };
3077
3078            first_writer.join().unwrap();
3079            second_writer.join().unwrap();
3080            snapshotter.join().unwrap();
3081
3082            let snapshot = log.lock().unwrap().snapshot();
3083            assert_snapshot_invariants(&snapshot, 2);
3084            assert_eq!(snapshot.len(), 2);
3085            assert_eq!(snapshot.last().map(|record| record.sequence), Some(2));
3086        });
3087    }
3088
3089    fn spawn_slow_query_writer(
3090        log: Arc<Mutex<SlowQueryLog>>,
3091        query_text: &'static str,
3092    ) -> thread::JoinHandle<()> {
3093        thread::spawn(move || {
3094            let query_identity = QueryIdentity::new("cypher", query_text);
3095            log.lock()
3096                .unwrap()
3097                .push(SlowQueryRecord::completed(SlowQueryCompletion {
3098                    query_language: "cypher",
3099                    statement_kind: "match_return",
3100                    query_text,
3101                    query_identity: &query_identity,
3102                    elapsed_micros: 1,
3103                    row_count: 1,
3104                    success: true,
3105                    error: None,
3106                    slow_log_candidate: true,
3107                    access_control_policy_epoch: None,
3108                    vector_execution_reports: Vec::new(),
3109                }));
3110        })
3111    }
3112
3113    fn assert_snapshot_invariants(snapshot: &[SlowQueryRecord], capacity: usize) {
3114        assert!(snapshot.len() <= capacity);
3115        for pair in snapshot.windows(2) {
3116            assert!(pair[0].sequence < pair[1].sequence);
3117        }
3118    }
3119}
hawdb_system_sql/lib.rs

hawdb_system_sql/
lib.rs
