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