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

hawdb_system_sql/
lib.rs
