hawdb_system_sql/
lib.rs

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

hawdb_system_sql/
lib.rs
