Skip to main content

hawdb_system_sql/
lib.rs

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