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

hawdb_system_sql/
lib.rs
