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 hawdb_sql::bind_sql_value(&value, parameters)
739}
740
741fn bind_bound(bound: SqlBound, parameters: &[Value], name: &str) -> Result<u64> {
742 match bound {
743 SqlBound::Literal(value) => Ok(value),
744 SqlBound::Parameter(position) => match parameters.get(position - 1) {
745 Some(Value::Int(value)) if *value >= 0 => Ok(*value as u64),
746 Some(_) => Err(HawDBError::Semantic(format!(
747 "PostgreSQL {name} parameter ${position} must be a non-negative integer"
748 ))),
749 None => Err(HawDBError::Semantic(format!(
750 "missing PostgreSQL parameter ${position}"
751 ))),
752 },
753 }
754}
755
756fn optimize_sql(logical: SqlLogicalPlan) -> SqlPhysicalPlan {
757 match logical {
758 SqlLogicalPlan::SystemTableScan(scan) => SqlPhysicalPlan::SystemTableScanExec(scan),
759 }
760}
761
762fn execute_sql<Store: SystemSqlStore>(
763 physical: SqlPhysicalPlan,
764 context: &SystemSqlContext<'_, Store>,
765 max_rows: Option<usize>,
766) -> Result<Vec<Row>> {
767 match physical {
768 SqlPhysicalPlan::SystemTableScanExec(scan) => {
769 execute_system_table_scan(scan, context, max_rows)
770 }
771 }
772}
773
774fn execute_system_table_scan<Store: SystemSqlStore>(
775 scan: SystemTableScan,
776 context: &SystemSqlContext<'_, Store>,
777 max_rows: Option<usize>,
778) -> Result<Vec<Row>> {
779 let mut rows = match scan.table {
780 SystemTable::Tables => table_rows(context.catalog),
781 SystemTable::VectorColumns => vector_column_rows(context.relational_state),
782 SystemTable::Properties => property_rows(context.catalog),
783 SystemTable::Indexes => index_rows(context.catalog),
784 SystemTable::Constraints => constraint_rows(context.catalog),
785 SystemTable::AppendTables => append_table_rows(context.append_state),
786 SystemTable::AppendStorage => append_storage_rows(context.store, context.append_state),
787 SystemTable::RuntimeStatus => runtime_status_rows(context),
788 SystemTable::RuntimeCapabilities => runtime_capability_rows(context.runtime),
789 SystemTable::GraphStatistics => {
790 graph_statistics_rows(context.catalog, &context.store.statistics(context.catalog))
791 }
792 SystemTable::ProjectedGraphs => {
793 projected_graph_rows(context.store.projected_graph_statuses())
794 }
795 SystemTable::SearchProjectionChangefeed => {
796 search_projection_changefeed_rows(context.store.search_projection_changefeed_status())
797 }
798 SystemTable::PlanCache => plan_cache_rows(context.plan_cache_stats),
799 SystemTable::SlowQueries => slow_query_rows(context.slow_queries),
800 SystemTable::StatementSummary => statement_summary_rows(context.statement_summaries),
801 SystemTable::InformationSchemaTables => {
802 information_schema_table_rows(context.relational_state)
803 }
804 SystemTable::InformationSchemaColumns => {
805 information_schema_column_rows(context.relational_state)
806 }
807 SystemTable::PgTables => pg_table_rows(context.relational_state),
808 SystemTable::PgIndexes => pg_index_rows(context.relational_state),
809 };
810
811 if let Some(predicate) = &scan.predicate {
812 let mut retained = 0;
813 for index in 0..rows.len() {
814 if predicate_matches(predicate, &rows[index])? {
815 rows.swap(retained, index);
816 retained += 1;
817 }
818 }
819 rows.truncate(retained);
820 }
821
822 if !scan.order_by.is_empty() {
823 rows.sort_by(|left, right| compare_ordered_rows(left, right, &scan.order_by));
824 }
825
826 let offset = scan.offset.unwrap_or(0);
827 let limit = effective_limit(scan.limit, max_rows)?;
828 rows = rows
829 .into_iter()
830 .skip(usize::try_from(offset).unwrap_or(usize::MAX))
831 .take(limit.unwrap_or(usize::MAX))
832 .collect();
833
834 if let Some(max_rows) = max_rows
835 && rows.len() > max_rows
836 {
837 return Err(HawDBError::read_budget_exceeded(
838 hawdb_core::ReadBudgetResource::Rows,
839 max_rows,
840 format!(
841 "SQL query returned more than {max_rows} rows, exceeding max_read_result_rows {max_rows}"
842 ),
843 ));
844 }
845
846 Ok(project_rows(rows, &scan.projection))
847}
848
849fn effective_limit(query_limit: Option<u64>, max_rows: Option<usize>) -> Result<Option<usize>> {
850 let query_limit = query_limit
851 .map(|limit| {
852 usize::try_from(limit)
853 .map_err(|_| HawDBError::Semantic("SQL LIMIT is too large".to_string()))
854 })
855 .transpose()?;
856 Ok(match (query_limit, max_rows) {
857 (Some(query_limit), Some(max_rows)) => Some(query_limit.min(max_rows.saturating_add(1))),
858 (Some(query_limit), None) => Some(query_limit),
859 (None, Some(max_rows)) => Some(max_rows.saturating_add(1)),
860 (None, None) => None,
861 })
862}
863
864fn projection_columns(
865 table: SystemTable,
866 projection: &[SelectProjection],
867) -> Vec<(String, String)> {
868 let mut columns = Vec::new();
869 for item in projection {
870 match item {
871 SelectProjection::Wildcard => columns.extend(
872 table_columns(table)
873 .iter()
874 .map(|name| (name.to_string(), name.to_string())),
875 ),
876 SelectProjection::Expression {
877 expression:
878 Expr {
879 kind: ExprKind::Column(column),
880 ..
881 },
882 alias,
883 ..
884 } => columns.push((
885 column.name.clone(),
886 alias.clone().unwrap_or_else(|| column.name.clone()),
887 )),
888 _ => unreachable!("catalog projections were validated during planning"),
889 }
890 }
891 columns
892}
893
894fn project_rows(rows: Vec<Row>, projection: &[(String, String)]) -> Vec<Row> {
895 rows.into_iter()
896 .map(|row| {
897 projection
898 .iter()
899 .map(|(source, name)| {
900 let value = row.get(source).cloned().unwrap_or(Value::Null);
901 (name.clone(), value)
902 })
903 .collect()
904 })
905 .collect()
906}
907
908fn table_rows(catalog: &Catalog) -> Vec<Row> {
909 catalog
910 .table_descriptors()
911 .map(|table| {
912 BTreeMap::from([
913 ("table_id".to_string(), u32_value(table.id.0)),
914 ("table_name".to_string(), Value::String(table.name.clone())),
915 (
916 "table_kind".to_string(),
917 Value::String(table_kind_name(table.kind).to_string()),
918 ),
919 (
920 "state".to_string(),
921 Value::String(schema_object_state_name(table.state).to_string()),
922 ),
923 ])
924 })
925 .collect()
926}
927
928fn information_schema_table_rows(state: &RelationalState) -> Vec<Row> {
929 state
930 .table_schemas()
931 .map(|schema| {
932 BTreeMap::from([
933 (
934 "table_catalog".to_string(),
935 Value::String("hawdb".to_string()),
936 ),
937 (
938 "table_schema".to_string(),
939 Value::String("public".to_string()),
940 ),
941 ("table_name".to_string(), Value::String(schema.name.clone())),
942 (
943 "table_type".to_string(),
944 Value::String(
945 if schema.materialized_view.is_some() {
946 "MATERIALIZED VIEW"
947 } else {
948 "BASE TABLE"
949 }
950 .to_string(),
951 ),
952 ),
953 ("self_referencing_column_name".to_string(), Value::Null),
954 ("reference_generation".to_string(), Value::Null),
955 ("user_defined_type_catalog".to_string(), Value::Null),
956 ("user_defined_type_schema".to_string(), Value::Null),
957 ("user_defined_type_name".to_string(), Value::Null),
958 (
959 "is_insertable_into".to_string(),
960 Value::String(
961 if schema.materialized_view.is_some() {
962 "NO"
963 } else {
964 "YES"
965 }
966 .to_string(),
967 ),
968 ),
969 ("is_typed".to_string(), Value::String("NO".to_string())),
970 ("commit_action".to_string(), Value::Null),
971 ])
972 })
973 .collect()
974}
975
976fn vector_column_rows(state: &RelationalState) -> Vec<Row> {
977 state
978 .table_schemas()
979 .flat_map(|schema| {
980 schema.columns.iter().filter_map(move |column| {
981 let RelationalScalarType::Vector(dimension) = column.scalar_type else {
982 return None;
983 };
984 Some(BTreeMap::from([
985 ("table_schema".into(), Value::String("public".into())),
986 ("table_name".into(), Value::String(schema.name.clone())),
987 ("column_name".into(), Value::String(column.name.clone())),
988 ("dimensions".into(), Value::Int(i64::from(dimension.get()))),
989 ("element_type".into(), Value::String("float32".into())),
990 ("nullable".into(), Value::Bool(column.nullable)),
991 ]))
992 })
993 })
994 .collect()
995}
996
997fn information_schema_column_rows(state: &RelationalState) -> Vec<Row> {
998 state
999 .table_schemas()
1000 .flat_map(|schema| {
1001 schema
1002 .visible_columns()
1003 .iter()
1004 .enumerate()
1005 .map(move |(position, column)| {
1006 let type_info = information_schema_type(column.scalar_type);
1007 let ordinal_position = position.saturating_add(1);
1008 BTreeMap::from([
1009 (
1010 "table_catalog".to_string(),
1011 Value::String("hawdb".to_string()),
1012 ),
1013 (
1014 "table_schema".to_string(),
1015 Value::String("public".to_string()),
1016 ),
1017 ("table_name".to_string(), Value::String(schema.name.clone())),
1018 (
1019 "column_name".to_string(),
1020 Value::String(column.name.clone()),
1021 ),
1022 (
1023 "ordinal_position".to_string(),
1024 usize_value(ordinal_position),
1025 ),
1026 (
1027 "column_default".to_string(),
1028 relational_default_value(column.default.as_ref()),
1029 ),
1030 (
1031 "is_nullable".to_string(),
1032 Value::String(if column.nullable { "YES" } else { "NO" }.to_string()),
1033 ),
1034 (
1035 "data_type".to_string(),
1036 Value::String(type_info.data_type.to_string()),
1037 ),
1038 (
1039 "character_maximum_length".to_string(),
1040 option_i64_value(type_info.character_maximum_length),
1041 ),
1042 (
1043 "character_octet_length".to_string(),
1044 option_i64_value(type_info.character_octet_length),
1045 ),
1046 (
1047 "numeric_precision".to_string(),
1048 option_i64_value(type_info.numeric_precision),
1049 ),
1050 (
1051 "numeric_precision_radix".to_string(),
1052 option_i64_value(type_info.numeric_precision_radix),
1053 ),
1054 (
1055 "numeric_scale".to_string(),
1056 option_i64_value(type_info.numeric_scale),
1057 ),
1058 ("datetime_precision".to_string(), Value::Null),
1059 ("interval_type".to_string(), Value::Null),
1060 ("interval_precision".to_string(), Value::Null),
1061 ("character_set_catalog".to_string(), Value::Null),
1062 ("character_set_schema".to_string(), Value::Null),
1063 ("character_set_name".to_string(), Value::Null),
1064 ("collation_catalog".to_string(), Value::Null),
1065 ("collation_schema".to_string(), Value::Null),
1066 ("collation_name".to_string(), Value::Null),
1067 ("domain_catalog".to_string(), Value::Null),
1068 ("domain_schema".to_string(), Value::Null),
1069 ("domain_name".to_string(), Value::Null),
1070 (
1071 "udt_catalog".to_string(),
1072 Value::String("hawdb".to_string()),
1073 ),
1074 (
1075 "udt_schema".to_string(),
1076 Value::String(
1077 if matches!(column.scalar_type, RelationalScalarType::Vector(_)) {
1078 "public"
1079 } else {
1080 "pg_catalog"
1081 }
1082 .to_string(),
1083 ),
1084 ),
1085 (
1086 "udt_name".to_string(),
1087 Value::String(type_info.udt_name.to_string()),
1088 ),
1089 ("scope_catalog".to_string(), Value::Null),
1090 ("scope_schema".to_string(), Value::Null),
1091 ("scope_name".to_string(), Value::Null),
1092 ("maximum_cardinality".to_string(), Value::Null),
1093 (
1094 "dtd_identifier".to_string(),
1095 Value::String(ordinal_position.to_string()),
1096 ),
1097 (
1098 "is_self_referencing".to_string(),
1099 Value::String("NO".to_string()),
1100 ),
1101 ("is_identity".to_string(), Value::String("NO".to_string())),
1102 ("identity_generation".to_string(), Value::Null),
1103 ("identity_start".to_string(), Value::Null),
1104 ("identity_increment".to_string(), Value::Null),
1105 ("identity_maximum".to_string(), Value::Null),
1106 ("identity_minimum".to_string(), Value::Null),
1107 ("identity_cycle".to_string(), Value::Null),
1108 (
1109 "is_generated".to_string(),
1110 Value::String("NEVER".to_string()),
1111 ),
1112 ("generation_expression".to_string(), Value::Null),
1113 ("is_updatable".to_string(), Value::String("YES".to_string())),
1114 ])
1115 })
1116 })
1117 .collect()
1118}
1119
1120#[derive(Debug, Clone, Copy)]
1121struct InformationSchemaType {
1122 data_type: &'static str,
1123 udt_name: &'static str,
1124 character_maximum_length: Option<i64>,
1125 character_octet_length: Option<i64>,
1126 numeric_precision: Option<i64>,
1127 numeric_precision_radix: Option<i64>,
1128 numeric_scale: Option<i64>,
1129}
1130
1131fn information_schema_type(scalar_type: RelationalScalarType) -> InformationSchemaType {
1132 match scalar_type {
1133 RelationalScalarType::Decimal(kind) => InformationSchemaType {
1134 data_type: "numeric",
1135 udt_name: "numeric",
1136 character_maximum_length: None,
1137 character_octet_length: None,
1138 numeric_precision: Some(i64::from(kind.precision())),
1139 numeric_precision_radix: Some(10),
1140 numeric_scale: Some(i64::from(kind.scale())),
1141 },
1142 RelationalScalarType::Boolean => InformationSchemaType {
1143 data_type: "boolean",
1144 udt_name: "bool",
1145 character_maximum_length: None,
1146 character_octet_length: None,
1147 numeric_precision: None,
1148 numeric_precision_radix: None,
1149 numeric_scale: None,
1150 },
1151 RelationalScalarType::BigInt => InformationSchemaType {
1152 data_type: "bigint",
1153 udt_name: "int8",
1154 character_maximum_length: None,
1155 character_octet_length: None,
1156 numeric_precision: Some(64),
1157 numeric_precision_radix: Some(2),
1158 numeric_scale: Some(0),
1159 },
1160 RelationalScalarType::DoublePrecision => InformationSchemaType {
1161 data_type: "double precision",
1162 udt_name: "float8",
1163 character_maximum_length: None,
1164 character_octet_length: None,
1165 numeric_precision: Some(53),
1166 numeric_precision_radix: Some(2),
1167 numeric_scale: None,
1168 },
1169 RelationalScalarType::Text => InformationSchemaType {
1170 data_type: "text",
1171 udt_name: "text",
1172 character_maximum_length: None,
1173 character_octet_length: None,
1174 numeric_precision: None,
1175 numeric_precision_radix: None,
1176 numeric_scale: None,
1177 },
1178 RelationalScalarType::Bytea => InformationSchemaType {
1179 data_type: "bytea",
1180 udt_name: "bytea",
1181 character_maximum_length: None,
1182 character_octet_length: None,
1183 numeric_precision: None,
1184 numeric_precision_radix: None,
1185 numeric_scale: None,
1186 },
1187 RelationalScalarType::Uuid => InformationSchemaType {
1188 data_type: "uuid",
1189 udt_name: "uuid",
1190 character_maximum_length: None,
1191 character_octet_length: None,
1192 numeric_precision: None,
1193 numeric_precision_radix: None,
1194 numeric_scale: None,
1195 },
1196 RelationalScalarType::Vector(_) => InformationSchemaType {
1197 data_type: "USER-DEFINED",
1198 udt_name: "vector",
1199 character_maximum_length: None,
1200 character_octet_length: None,
1201 numeric_precision: None,
1202 numeric_precision_radix: None,
1203 numeric_scale: None,
1204 },
1205 }
1206}
1207
1208fn relational_default_value(default: Option<&RelationalColumnDefault>) -> Value {
1209 match default {
1210 None | Some(RelationalColumnDefault::Literal(RelationalValue::Overflow(_))) => Value::Null,
1211 Some(RelationalColumnDefault::Literal(RelationalValue::Null)) => {
1212 Value::String("NULL".to_string())
1213 }
1214 Some(RelationalColumnDefault::Literal(RelationalValue::Boolean(value))) => {
1215 Value::String(value.to_string())
1216 }
1217 Some(RelationalColumnDefault::Literal(RelationalValue::BigInt(value))) => {
1218 Value::String(value.to_string())
1219 }
1220 Some(RelationalColumnDefault::Literal(RelationalValue::Decimal(value))) => {
1221 Value::String(value.to_string())
1222 }
1223 Some(RelationalColumnDefault::Literal(RelationalValue::DoublePrecision(value))) => {
1224 Value::String(value.to_string())
1225 }
1226 Some(RelationalColumnDefault::Literal(RelationalValue::Text(value))) => {
1227 Value::String(format!("'{}'::text", value.replace('\'', "''")))
1228 }
1229 Some(RelationalColumnDefault::Literal(RelationalValue::Bytea(value))) => {
1230 Value::String(format!("'\\x{}'::bytea", encode_hex(value)))
1231 }
1232 Some(RelationalColumnDefault::Literal(RelationalValue::Uuid(value))) => {
1233 Value::String(format!("'{value}'::uuid"))
1234 }
1235 Some(RelationalColumnDefault::Literal(RelationalValue::Vector(value))) => {
1236 Value::String(Value::Vector(value.clone()).to_string())
1237 }
1238 Some(RelationalColumnDefault::UuidV7) => Value::String("uuidv7()".to_string()),
1239 }
1240}
1241
1242fn encode_hex(bytes: &[u8]) -> String {
1243 const HEX: &[u8; 16] = b"0123456789abcdef";
1244 let mut encoded = String::with_capacity(bytes.len().saturating_mul(2));
1245 for byte in bytes {
1246 encoded.push(HEX[usize::from(byte >> 4)] as char);
1247 encoded.push(HEX[usize::from(byte & 0x0f)] as char);
1248 }
1249 encoded
1250}
1251
1252fn pg_table_rows(state: &RelationalState) -> Vec<Row> {
1253 state
1254 .table_schemas()
1255 .map(|schema| {
1256 BTreeMap::from([
1257 (
1258 "schemaname".to_string(),
1259 Value::String("public".to_string()),
1260 ),
1261 ("tablename".to_string(), Value::String(schema.name.clone())),
1262 ("tableowner".to_string(), Value::String("hawdb".to_string())),
1263 ("tablespace".to_string(), Value::Null),
1264 (
1265 "hasindexes".to_string(),
1266 Value::Bool(
1267 !schema.primary_key.is_empty()
1268 || !schema.unique_constraints.is_empty()
1269 || !schema.indexes.is_empty(),
1270 ),
1271 ),
1272 ("hasrules".to_string(), Value::Bool(false)),
1273 ("hastriggers".to_string(), Value::Bool(false)),
1274 ("rowsecurity".to_string(), Value::Bool(false)),
1275 ])
1276 })
1277 .collect()
1278}
1279
1280fn pg_index_rows(state: &RelationalState) -> Vec<Row> {
1281 state
1282 .table_schemas()
1283 .flat_map(relational_schema_index_rows)
1284 .collect()
1285}
1286
1287fn relational_schema_index_rows(schema: &RelationalTableSchema) -> Vec<Row> {
1288 let mut rows = Vec::with_capacity(
1289 1usize
1290 .saturating_add(schema.unique_constraints.len())
1291 .saturating_add(schema.indexes.len()),
1292 );
1293 if !schema.primary_key.is_empty() {
1294 let name = format!("{}_pkey", schema.name);
1295 rows.push(pg_index_row(schema, &name, &schema.primary_key, true));
1296 }
1297 rows.extend(schema.unique_constraints.iter().map(|columns| {
1298 let name = format!("{}_{}_key", schema.name, columns.join("_"));
1299 pg_index_row(schema, &name, columns, true)
1300 }));
1301 rows.extend(
1302 schema
1303 .indexes
1304 .iter()
1305 .map(|index| pg_index_row(schema, &index.name, &index.columns, index.unique)),
1306 );
1307 rows
1308}
1309
1310fn pg_index_row(
1311 schema: &RelationalTableSchema,
1312 index_name: &str,
1313 columns: &[String],
1314 unique: bool,
1315) -> Row {
1316 let unique = if unique { "UNIQUE " } else { "" };
1317 let columns = columns
1318 .iter()
1319 .map(|column| quote_postgres_identifier(column))
1320 .collect::<Vec<_>>()
1321 .join(", ");
1322 let index_definition = format!(
1323 "CREATE {unique}INDEX {} ON public.{} USING btree ({columns})",
1324 quote_postgres_identifier(index_name),
1325 quote_postgres_identifier(&schema.name),
1326 );
1327 BTreeMap::from([
1328 (
1329 "schemaname".to_string(),
1330 Value::String("public".to_string()),
1331 ),
1332 ("tablename".to_string(), Value::String(schema.name.clone())),
1333 (
1334 "indexname".to_string(),
1335 Value::String(index_name.to_string()),
1336 ),
1337 ("tablespace".to_string(), Value::Null),
1338 ("indexdef".to_string(), Value::String(index_definition)),
1339 ])
1340}
1341
1342fn quote_postgres_identifier(identifier: &str) -> String {
1343 format!("\"{}\"", identifier.replace('"', "\"\""))
1344}
1345
1346fn property_rows(catalog: &Catalog) -> Vec<Row> {
1347 catalog
1348 .property_descriptors()
1349 .map(|property| {
1350 let table = catalog.table_descriptor(property.table_id);
1351 BTreeMap::from([
1352 ("property_id".to_string(), u32_value(property.id.0)),
1353 ("table_id".to_string(), u32_value(property.table_id.0)),
1354 (
1355 "table_name".to_string(),
1356 table
1357 .map(|table| Value::String(table.name.clone()))
1358 .unwrap_or(Value::Null),
1359 ),
1360 (
1361 "table_kind".to_string(),
1362 table
1363 .map(|table| Value::String(table_kind_name(table.kind).to_string()))
1364 .unwrap_or(Value::Null),
1365 ),
1366 (
1367 "property_name".to_string(),
1368 Value::String(property.name.clone()),
1369 ),
1370 (
1371 "value_type".to_string(),
1372 Value::String(property_type_name(property.value_type).to_string()),
1373 ),
1374 ("nullable".to_string(), Value::Bool(property.nullable)),
1375 (
1376 "state".to_string(),
1377 Value::String(schema_object_state_name(property.state).to_string()),
1378 ),
1379 ])
1380 })
1381 .collect()
1382}
1383
1384fn index_rows(catalog: &Catalog) -> Vec<Row> {
1385 let mut rows = catalog
1386 .property_indexes()
1387 .map(|index| {
1388 BTreeMap::from([
1389 ("index_id".to_string(), u32_value(index.id.0)),
1390 (
1391 "subject_kind".to_string(),
1392 Value::String("node".to_string()),
1393 ),
1394 ("subject_id".to_string(), u32_value(index.label_id.0)),
1395 (
1396 "subject_name".to_string(),
1397 optional_string_value(catalog.label_name(index.label_id)),
1398 ),
1399 (
1400 "index_kind".to_string(),
1401 Value::String(index_kind_name(index.kind).to_string()),
1402 ),
1403 (
1404 "property_names".to_string(),
1405 Value::List(vec![Value::String(index.property.clone())]),
1406 ),
1407 ])
1408 })
1409 .chain(catalog.composite_property_indexes().map(|index| {
1410 BTreeMap::from([
1411 ("index_id".to_string(), u32_value(index.id.0)),
1412 (
1413 "subject_kind".to_string(),
1414 Value::String("node".to_string()),
1415 ),
1416 ("subject_id".to_string(), u32_value(index.label_id.0)),
1417 (
1418 "subject_name".to_string(),
1419 optional_string_value(catalog.label_name(index.label_id)),
1420 ),
1421 (
1422 "index_kind".to_string(),
1423 Value::String("composite_equality".to_string()),
1424 ),
1425 (
1426 "property_names".to_string(),
1427 Value::List(
1428 index
1429 .properties
1430 .iter()
1431 .cloned()
1432 .map(Value::String)
1433 .collect(),
1434 ),
1435 ),
1436 ])
1437 }))
1438 .collect::<Vec<_>>();
1439 rows.sort_by_key(|row| row.get("index_id").cloned());
1440 rows
1441}
1442
1443fn constraint_rows(catalog: &Catalog) -> Vec<Row> {
1444 let constraints = catalog
1445 .unique_constraints()
1446 .chain(catalog.node_property_exists_constraints())
1447 .chain(catalog.relationship_property_exists_constraints())
1448 .chain(catalog.relationship_unique_constraints());
1449 let mut rows = constraints
1450 .map(|constraint| {
1451 let (subject_kind, subject_id, subject_name) = match constraint.subject {
1452 ConstraintSubject::Node(label_id) => (
1453 "node",
1454 label_id.0,
1455 optional_string_value(catalog.label_name(label_id)),
1456 ),
1457 ConstraintSubject::Relationship(rel_type_id) => (
1458 "relationship",
1459 rel_type_id.0,
1460 optional_string_value(catalog.rel_type_name(rel_type_id)),
1461 ),
1462 };
1463 BTreeMap::from([
1464 ("constraint_id".to_string(), u32_value(constraint.id.0)),
1465 (
1466 "subject_kind".to_string(),
1467 Value::String(subject_kind.to_string()),
1468 ),
1469 ("subject_id".to_string(), u32_value(subject_id)),
1470 ("subject_name".to_string(), subject_name),
1471 (
1472 "property_name".to_string(),
1473 Value::String(constraint.property.clone()),
1474 ),
1475 (
1476 "constraint_kind".to_string(),
1477 Value::String(constraint_kind_name(constraint.kind).to_string()),
1478 ),
1479 ])
1480 })
1481 .collect::<Vec<_>>();
1482 rows.sort_by_key(|row| row.get("constraint_id").cloned());
1483 rows
1484}
1485
1486fn runtime_status_rows<Store: SystemSqlStore>(context: &SystemSqlContext<'_, Store>) -> Vec<Row> {
1487 vec![BTreeMap::from([
1488 (
1489 "commit_epoch".to_string(),
1490 u64_value(context.store.commit_epoch()),
1491 ),
1492 (
1493 "read_only".to_string(),
1494 Value::Bool(context.runtime.read_only),
1495 ),
1496 (
1497 "max_read_result_rows".to_string(),
1498 option_usize_value(context.runtime.max_read_result_rows),
1499 ),
1500 (
1501 "max_read_result_payload_bytes".to_string(),
1502 option_usize_value(context.runtime.max_read_result_payload_bytes),
1503 ),
1504 (
1505 "storage_residency_mode".to_string(),
1506 Value::String(
1507 storage_residency_mode_name(context.runtime.storage_residency_mode).to_string(),
1508 ),
1509 ),
1510 (
1511 "relational_index_mode".to_string(),
1512 Value::String(
1513 relational_index_mode_name(context.runtime.relational_index_mode).to_string(),
1514 ),
1515 ),
1516 ])]
1517}
1518
1519fn append_table_rows(state: &AppendState) -> Vec<Row> {
1520 state
1521 .schemas()
1522 .values()
1523 .map(|schema| {
1524 BTreeMap::from([
1525 ("table_name".to_string(), Value::String(schema.name.clone())),
1526 (
1527 "storage_mode".to_string(),
1528 Value::String("strict_append".to_string()),
1529 ),
1530 (
1531 "partition_key".to_string(),
1532 Value::List(
1533 schema
1534 .partition_key
1535 .iter()
1536 .cloned()
1537 .map(Value::String)
1538 .collect(),
1539 ),
1540 ),
1541 (
1542 "order_key".to_string(),
1543 Value::List(
1544 schema
1545 .order_key
1546 .iter()
1547 .cloned()
1548 .map(Value::String)
1549 .collect(),
1550 ),
1551 ),
1552 (
1553 "order_mode".to_string(),
1554 Value::String(
1555 match schema.order_mode {
1556 AppendOrderMode::CallerProvided => "caller_provided",
1557 AppendOrderMode::CommitSequence => "commit_sequence",
1558 }
1559 .to_string(),
1560 ),
1561 ),
1562 (
1563 "generated_order_watermark".to_string(),
1564 state
1565 .generated_order_watermark(&schema.name)
1566 .map_or(Value::Null, Value::Int),
1567 ),
1568 (
1569 "column_count".to_string(),
1570 Value::Int(i64::try_from(schema.columns.len()).unwrap_or(i64::MAX)),
1571 ),
1572 ])
1573 })
1574 .collect()
1575}
1576
1577fn append_storage_rows(store: &impl SystemSqlStore, state: &AppendState) -> Vec<Row> {
1578 let mut report = store.append_storage_residency_report();
1579 report.live_rows = state.live_rows();
1580 report.live_payload_bytes = state.live_payload_bytes();
1581 vec![BTreeMap::from([
1582 (
1583 "canonical_segment_count".to_string(),
1584 Value::Int(i64::try_from(report.canonical_segment_count).unwrap_or(i64::MAX)),
1585 ),
1586 (
1587 "canonical_segment_bytes".to_string(),
1588 Value::Int(i64::try_from(report.canonical_segment_bytes).unwrap_or(i64::MAX)),
1589 ),
1590 (
1591 "resident_segment_payload_bytes".to_string(),
1592 Value::Int(i64::try_from(report.resident_segment_payload_bytes).unwrap_or(i64::MAX)),
1593 ),
1594 (
1595 "resident_descriptor_count".to_string(),
1596 Value::Int(i64::try_from(report.resident_descriptor_count).unwrap_or(i64::MAX)),
1597 ),
1598 (
1599 "live_rows".to_string(),
1600 Value::Int(i64::try_from(report.live_rows).unwrap_or(i64::MAX)),
1601 ),
1602 (
1603 "live_payload_bytes".to_string(),
1604 Value::Int(i64::try_from(report.live_payload_bytes).unwrap_or(i64::MAX)),
1605 ),
1606 ])]
1607}
1608
1609fn runtime_capability_rows(runtime: SystemRuntimeSnapshot) -> Vec<Row> {
1610 [
1611 (
1612 "access_control",
1613 runtime.runtime_capabilities.access_control,
1614 ),
1615 (
1616 "full_text_search",
1617 runtime.runtime_capabilities.full_text_search,
1618 ),
1619 ("vector_search", runtime.runtime_capabilities.vector_search),
1620 (
1621 "graph_analytics",
1622 runtime.runtime_capabilities.graph_analytics,
1623 ),
1624 (
1625 "background_maintenance",
1626 runtime.runtime_capabilities.background_maintenance,
1627 ),
1628 ]
1629 .into_iter()
1630 .map(|(capability, enabled)| {
1631 BTreeMap::from([
1632 (
1633 "capability".to_string(),
1634 Value::String(capability.to_string()),
1635 ),
1636 ("enabled".to_string(), Value::Bool(enabled)),
1637 ])
1638 })
1639 .collect()
1640}
1641
1642fn graph_statistics_rows(catalog: &Catalog, statistics: &GraphStatistics) -> Vec<Row> {
1643 let mut rows = Vec::with_capacity(
1644 2usize
1645 .saturating_add(statistics.label_counts.len())
1646 .saturating_add(statistics.rel_type_counts.len())
1647 .saturating_add(statistics.rel_type_source_counts.len())
1648 .saturating_add(statistics.rel_type_target_counts.len())
1649 .saturating_add(statistics.path_counts.len())
1650 .saturating_add(statistics.path_source_distinct_counts.len())
1651 .saturating_add(statistics.path_target_distinct_counts.len())
1652 .saturating_add(statistics.bounded_path_counts.len())
1653 .saturating_add(statistics.bounded_path_source_distinct_counts.len())
1654 .saturating_add(statistics.bounded_path_target_distinct_counts.len())
1655 .saturating_add(statistics.index_samples.len())
1656 .saturating_add(statistics.property_distinct_counts.len())
1657 .saturating_add(statistics.rel_property_distinct_counts.len())
1658 .saturating_add(statistics.property_histograms.len())
1659 .saturating_add(statistics.rel_property_histograms.len()),
1660 );
1661 rows.push(graph_statistic_count_row(
1662 statistics,
1663 "node_count",
1664 statistics.node_count,
1665 ));
1666 rows.push(graph_statistic_count_row(
1667 statistics,
1668 "relationship_count",
1669 statistics.relationship_count,
1670 ));
1671
1672 for (label_id, count) in &statistics.label_counts {
1673 let mut row = graph_statistic_count_row(statistics, "label_count", *count);
1674 row.insert(
1675 "label_name".to_string(),
1676 optional_string_value(catalog.label_name(*label_id)),
1677 );
1678 rows.push(row);
1679 }
1680 for (rel_type_id, count) in &statistics.rel_type_counts {
1681 let mut row = graph_statistic_count_row(statistics, "relationship_type_count", *count);
1682 row.insert(
1683 "relationship_type_name".to_string(),
1684 optional_string_value(catalog.rel_type_name(*rel_type_id)),
1685 );
1686 rows.push(row);
1687 }
1688 for (kind, counts) in [
1689 (
1690 "relationship_source_count",
1691 &statistics.rel_type_source_counts,
1692 ),
1693 (
1694 "relationship_target_count",
1695 &statistics.rel_type_target_counts,
1696 ),
1697 ] {
1698 for (rel_type_id, count) in counts {
1699 let mut row = graph_statistic_count_row(statistics, kind, *count);
1700 row.insert(
1701 "relationship_type_name".to_string(),
1702 optional_string_value(catalog.rel_type_name(*rel_type_id)),
1703 );
1704 rows.push(row);
1705 }
1706 }
1707 for index in catalog.property_indexes() {
1708 let Some(sample) = statistics.index_samples.get(&index.id) else {
1709 continue;
1710 };
1711 let mut row = graph_statistic_base_row(statistics, "index_sample");
1712 row.insert("index_id".to_string(), u32_value(index.id.0));
1713 row.insert(
1714 "index_kind".to_string(),
1715 Value::String(index_kind_name(index.kind).to_string()),
1716 );
1717 row.insert(
1718 "label_name".to_string(),
1719 optional_string_value(catalog.label_name(index.label_id)),
1720 );
1721 row.insert(
1722 "property_name".to_string(),
1723 Value::String(index.property.clone()),
1724 );
1725 populate_index_sample(&mut row, *sample);
1726 rows.push(row);
1727 }
1728 for index in catalog.composite_property_indexes() {
1729 let Some(sample) = statistics.index_samples.get(&index.id) else {
1730 continue;
1731 };
1732 let mut row = graph_statistic_base_row(statistics, "index_sample");
1733 row.insert("index_id".to_string(), u32_value(index.id.0));
1734 row.insert(
1735 "index_kind".to_string(),
1736 Value::String("composite".to_string()),
1737 );
1738 row.insert(
1739 "label_name".to_string(),
1740 optional_string_value(catalog.label_name(index.label_id)),
1741 );
1742 row.insert(
1743 "index_properties".to_string(),
1744 Value::List(
1745 index
1746 .properties
1747 .iter()
1748 .cloned()
1749 .map(Value::String)
1750 .collect(),
1751 ),
1752 );
1753 populate_index_sample(&mut row, *sample);
1754 rows.push(row);
1755 }
1756 for (kind, counts) in [
1757 ("path_count", &statistics.path_counts),
1758 (
1759 "path_source_distinct_count",
1760 &statistics.path_source_distinct_counts,
1761 ),
1762 (
1763 "path_target_distinct_count",
1764 &statistics.path_target_distinct_counts,
1765 ),
1766 ] {
1767 for ((source_label_id, rel_type_id, target_label_id), count) in counts {
1768 let mut row = graph_statistic_count_row(statistics, kind, *count);
1769 populate_path_statistic_names(
1770 &mut row,
1771 catalog,
1772 *source_label_id,
1773 *rel_type_id,
1774 *target_label_id,
1775 );
1776 rows.push(row);
1777 }
1778 }
1779 for (kind, counts) in [
1780 ("bounded_path_count", &statistics.bounded_path_counts),
1781 (
1782 "bounded_path_source_distinct_count",
1783 &statistics.bounded_path_source_distinct_counts,
1784 ),
1785 (
1786 "bounded_path_target_distinct_count",
1787 &statistics.bounded_path_target_distinct_counts,
1788 ),
1789 ] {
1790 for ((source_label_id, rel_type_id, target_label_id, depth), count) in counts {
1791 let mut row = graph_statistic_count_row(statistics, kind, *count);
1792 populate_path_statistic_names(
1793 &mut row,
1794 catalog,
1795 *source_label_id,
1796 *rel_type_id,
1797 *target_label_id,
1798 );
1799 row.insert("depth".to_string(), usize_value(*depth));
1800 rows.push(row);
1801 }
1802 }
1803 for ((label_id, property_name), count) in &statistics.property_distinct_counts {
1804 let mut row = graph_statistic_count_row(statistics, "node_property_distinct_count", *count);
1805 row.insert(
1806 "label_name".to_string(),
1807 optional_string_value(catalog.label_name(*label_id)),
1808 );
1809 row.insert(
1810 "property_name".to_string(),
1811 Value::String(property_name.clone()),
1812 );
1813 rows.push(row);
1814 }
1815 for ((rel_type_id, property_name), count) in &statistics.rel_property_distinct_counts {
1816 let mut row =
1817 graph_statistic_count_row(statistics, "relationship_property_distinct_count", *count);
1818 row.insert(
1819 "relationship_type_name".to_string(),
1820 optional_string_value(catalog.rel_type_name(*rel_type_id)),
1821 );
1822 row.insert(
1823 "property_name".to_string(),
1824 Value::String(property_name.clone()),
1825 );
1826 rows.push(row);
1827 }
1828 for ((label_id, property_name), histogram) in &statistics.property_histograms {
1829 let mut row = graph_statistic_base_row(statistics, "node_property_histogram");
1830 row.insert(
1831 "label_name".to_string(),
1832 optional_string_value(catalog.label_name(*label_id)),
1833 );
1834 row.insert(
1835 "property_name".to_string(),
1836 Value::String(property_name.clone()),
1837 );
1838 row.insert("histogram".to_string(), Value::List(histogram.clone()));
1839 row.insert(
1840 "sampled".to_string(),
1841 Value::Bool(
1842 statistics
1843 .sampled_property_histograms
1844 .get(&(*label_id, property_name.clone()))
1845 .copied()
1846 .unwrap_or(false),
1847 ),
1848 );
1849 rows.push(row);
1850 }
1851 for ((rel_type_id, property_name), histogram) in &statistics.rel_property_histograms {
1852 let mut row = graph_statistic_base_row(statistics, "relationship_property_histogram");
1853 row.insert(
1854 "relationship_type_name".to_string(),
1855 optional_string_value(catalog.rel_type_name(*rel_type_id)),
1856 );
1857 row.insert(
1858 "property_name".to_string(),
1859 Value::String(property_name.clone()),
1860 );
1861 row.insert("histogram".to_string(), Value::List(histogram.clone()));
1862 row.insert(
1863 "sampled".to_string(),
1864 Value::Bool(
1865 statistics
1866 .sampled_rel_property_histograms
1867 .get(&(*rel_type_id, property_name.clone()))
1868 .copied()
1869 .unwrap_or(false),
1870 ),
1871 );
1872 rows.push(row);
1873 }
1874 rows
1875}
1876
1877fn graph_statistic_count_row(statistics: &GraphStatistics, kind: &str, count: u64) -> Row {
1878 let mut row = graph_statistic_base_row(statistics, kind);
1879 row.insert("count".to_string(), u64_value(count));
1880 row
1881}
1882
1883fn populate_index_sample(row: &mut Row, sample: IndexStatisticsSample) {
1884 row.insert("index_size".to_string(), u64_value(sample.index_size));
1885 row.insert("unique_values".to_string(), u64_value(sample.unique_values));
1886 row.insert("sample_size".to_string(), u64_value(sample.sample_size));
1887 row.insert(
1888 "updates_since_sample".to_string(),
1889 u64_value(sample.updates_since_sample),
1890 );
1891 row.insert("stale".to_string(), Value::Bool(sample.is_stale()));
1892}
1893
1894fn graph_statistic_base_row(statistics: &GraphStatistics, kind: &str) -> Row {
1895 BTreeMap::from([
1896 (
1897 "statistic_kind".to_string(),
1898 Value::String(kind.to_string()),
1899 ),
1900 (
1901 "computed_at_commit_epoch".to_string(),
1902 u64_value(statistics.computed_at_commit_epoch),
1903 ),
1904 (
1905 "advanced_statistics_complete".to_string(),
1906 Value::Bool(statistics.advanced_statistics_complete),
1907 ),
1908 (
1909 "histogram_sample_limit".to_string(),
1910 usize_value(statistics.histogram_sample_limit),
1911 ),
1912 ("label_name".to_string(), Value::Null),
1913 ("relationship_type_name".to_string(), Value::Null),
1914 ("source_label_name".to_string(), Value::Null),
1915 ("target_label_name".to_string(), Value::Null),
1916 ("depth".to_string(), Value::Null),
1917 ("property_name".to_string(), Value::Null),
1918 ("index_id".to_string(), Value::Null),
1919 ("index_kind".to_string(), Value::Null),
1920 ("index_properties".to_string(), Value::Null),
1921 ("index_size".to_string(), Value::Null),
1922 ("unique_values".to_string(), Value::Null),
1923 ("sample_size".to_string(), Value::Null),
1924 ("updates_since_sample".to_string(), Value::Null),
1925 ("stale".to_string(), Value::Null),
1926 ("count".to_string(), Value::Null),
1927 ("histogram".to_string(), Value::Null),
1928 ("sampled".to_string(), Value::Null),
1929 ])
1930}
1931
1932fn populate_path_statistic_names(
1933 row: &mut Row,
1934 catalog: &Catalog,
1935 source_label_id: LabelId,
1936 rel_type_id: RelTypeId,
1937 target_label_id: LabelId,
1938) {
1939 row.insert(
1940 "source_label_name".to_string(),
1941 optional_string_value(catalog.label_name(source_label_id)),
1942 );
1943 row.insert(
1944 "relationship_type_name".to_string(),
1945 optional_string_value(catalog.rel_type_name(rel_type_id)),
1946 );
1947 row.insert(
1948 "target_label_name".to_string(),
1949 optional_string_value(catalog.label_name(target_label_id)),
1950 );
1951}
1952
1953fn projected_graph_rows(statuses: Vec<ProjectedGraphStatus>) -> Vec<Row> {
1954 statuses
1955 .into_iter()
1956 .map(|status| {
1957 BTreeMap::from([
1958 ("name".to_string(), Value::String(status.name)),
1959 (
1960 "node_labels".to_string(),
1961 Value::List(status.node_labels.into_iter().map(Value::String).collect()),
1962 ),
1963 (
1964 "relationship_types".to_string(),
1965 Value::List(status.rel_types.into_iter().map(Value::String).collect()),
1966 ),
1967 (
1968 "projection_epoch".to_string(),
1969 option_u64_value(status.projection_epoch),
1970 ),
1971 (
1972 "commit_epoch".to_string(),
1973 option_u64_value(status.commit_epoch),
1974 ),
1975 (
1976 "node_count".to_string(),
1977 option_usize_value(status.node_count),
1978 ),
1979 (
1980 "edge_count".to_string(),
1981 option_usize_value(status.edge_count),
1982 ),
1983 ("reusable".to_string(), Value::Bool(status.reusable)),
1984 ])
1985 })
1986 .collect()
1987}
1988
1989fn search_projection_changefeed_rows(status: SearchProjectionChangefeedStatus) -> Vec<Row> {
1990 vec![BTreeMap::from([
1991 (
1992 "graph_commit_epoch".to_string(),
1993 u64_value(status.graph_commit_epoch),
1994 ),
1995 (
1996 "resume_floor_commit_epoch".to_string(),
1997 u64_value(status.resume_floor_commit_epoch),
1998 ),
1999 (
2000 "oldest_retained_mutation_id".to_string(),
2001 option_u64_value(
2002 status
2003 .oldest_retained_mutation_id
2004 .map(|id| id.commit_epoch()),
2005 ),
2006 ),
2007 (
2008 "newest_retained_mutation_id".to_string(),
2009 option_u64_value(
2010 status
2011 .newest_retained_mutation_id
2012 .map(|id| id.commit_epoch()),
2013 ),
2014 ),
2015 (
2016 "retained_mutation_count".to_string(),
2017 usize_value(status.retained_mutation_count),
2018 ),
2019 (
2020 "restart_recoverable".to_string(),
2021 Value::Bool(status.restart_recoverable),
2022 ),
2023 ])]
2024}
2025
2026fn optional_string_value(value: Option<&str>) -> Value {
2027 value
2028 .map(|value| Value::String(value.to_string()))
2029 .unwrap_or(Value::Null)
2030}
2031
2032const fn table_kind_name(kind: TableKind) -> &'static str {
2033 match kind {
2034 TableKind::Node => "node",
2035 TableKind::Relationship => "relationship",
2036 }
2037}
2038
2039const fn schema_object_state_name(state: SchemaObjectState) -> &'static str {
2040 match state {
2041 SchemaObjectState::DeleteOnly => "delete_only",
2042 SchemaObjectState::WriteOnly => "write_only",
2043 SchemaObjectState::Backfill => "backfill",
2044 SchemaObjectState::Validating => "validating",
2045 SchemaObjectState::Public => "public",
2046 SchemaObjectState::Gc => "gc",
2047 }
2048}
2049
2050fn property_type_name(value_type: PropertyType) -> std::borrow::Cow<'static, str> {
2051 match value_type {
2052 PropertyType::Any => "any",
2053 PropertyType::Bool => "bool",
2054 PropertyType::Int => "int",
2055 PropertyType::Float => "float",
2056 PropertyType::String => "string",
2057 PropertyType::Text => "text",
2058 PropertyType::List => "list",
2059 PropertyType::Vector(dimension) => return format!("vector({})", dimension.get()).into(),
2060 }
2061 .into()
2062}
2063
2064const fn index_kind_name(kind: IndexKind) -> &'static str {
2065 match kind {
2066 IndexKind::Equality => "equality",
2067 IndexKind::Range => "range",
2068 IndexKind::FullText => "full_text",
2069 }
2070}
2071
2072const fn constraint_kind_name(kind: ConstraintKind) -> &'static str {
2073 match kind {
2074 ConstraintKind::NodePropertyUnique => "node_property_unique",
2075 ConstraintKind::NodePropertyExists => "node_property_exists",
2076 ConstraintKind::RelationshipPropertyUnique => "relationship_property_unique",
2077 ConstraintKind::RelationshipPropertyExists => "relationship_property_exists",
2078 }
2079}
2080
2081fn plan_cache_rows(stats: &PlanCacheStats) -> Vec<Row> {
2082 [
2083 ("max_entries", option_usize_value(stats.max_entries)),
2084 ("max_bytes", option_usize_value(stats.max_bytes)),
2085 ("max_entry_bytes", option_usize_value(stats.max_entry_bytes)),
2086 ("charged_bytes", option_usize_value(stats.charged_bytes)),
2087 ("rejected_admissions", u64_value(stats.rejected_admissions)),
2088 ("entries", usize_value(stats.entries)),
2089 ("hits", u64_value(stats.hits)),
2090 ("misses", u64_value(stats.misses)),
2091 ("admissions", u64_value(stats.admissions)),
2092 ("disabled_misses", u64_value(stats.disabled_misses)),
2093 ("bypasses", u64_value(stats.bypasses)),
2094 ("evictions", u64_value(stats.evictions)),
2095 (
2096 "memory_pressure_events",
2097 u64_value(stats.memory_pressure_events),
2098 ),
2099 ]
2100 .into_iter()
2101 .map(|(metric, value)| {
2102 BTreeMap::from([
2103 ("metric".to_string(), Value::String(metric.to_string())),
2104 ("value".to_string(), value),
2105 ])
2106 })
2107 .collect()
2108}
2109
2110fn slow_query_rows(records: &[SlowQueryRecord]) -> Vec<Row> {
2111 records
2112 .iter()
2113 .map(|record| {
2114 BTreeMap::from([
2115 ("sequence".to_string(), u64_value(record.sequence)),
2116 (
2117 "query_language".to_string(),
2118 Value::String(record.query_language.clone()),
2119 ),
2120 (
2121 "statement_kind".to_string(),
2122 Value::String(record.statement_kind.clone()),
2123 ),
2124 (
2125 "query_digest".to_string(),
2126 Value::String(record.query_digest.clone()),
2127 ),
2128 (
2129 "query_text_hash".to_string(),
2130 Value::String(record.query_text_hash.clone()),
2131 ),
2132 (
2133 "query_text".to_string(),
2134 Value::String(record.query_text.clone()),
2135 ),
2136 (
2137 "started_unix_micros".to_string(),
2138 Value::Int(record.started_unix_micros),
2139 ),
2140 (
2141 "elapsed_micros".to_string(),
2142 Value::Int(record.elapsed_micros),
2143 ),
2144 ("row_count".to_string(), Value::Int(record.row_count)),
2145 ("success".to_string(), Value::Bool(record.success)),
2146 (
2147 "error".to_string(),
2148 record
2149 .error
2150 .as_ref()
2151 .map(|error| Value::String(error.clone()))
2152 .unwrap_or(Value::Null),
2153 ),
2154 (
2155 "slow_log_candidate".to_string(),
2156 Value::Bool(record.slow_log_candidate),
2157 ),
2158 (
2159 "access_control_policy_epoch".to_string(),
2160 record
2161 .access_control_policy_epoch
2162 .map(u64_value)
2163 .unwrap_or(Value::Null),
2164 ),
2165 ])
2166 })
2167 .collect()
2168}
2169
2170fn statement_summary_rows(records: &[StatementSummaryRecord]) -> Vec<Row> {
2171 records
2172 .iter()
2173 .map(|record| {
2174 BTreeMap::from([
2175 ("digest".to_string(), Value::String(record.digest.clone())),
2176 (
2177 "query_language".to_string(),
2178 Value::String(record.query_language.clone()),
2179 ),
2180 (
2181 "query_text".to_string(),
2182 Value::String(record.query_text.clone()),
2183 ),
2184 (
2185 "sample_query_text_hash".to_string(),
2186 Value::String(record.sample_query_text_hash.clone()),
2187 ),
2188 (
2189 "statement_kind".to_string(),
2190 Value::String(record.statement_kind.clone()),
2191 ),
2192 (
2193 "execution_count".to_string(),
2194 Value::Int(record.execution_count),
2195 ),
2196 (
2197 "success_count".to_string(),
2198 Value::Int(record.success_count),
2199 ),
2200 ("error_count".to_string(), Value::Int(record.error_count)),
2201 (
2202 "total_elapsed_micros".to_string(),
2203 Value::Int(record.total_elapsed_micros),
2204 ),
2205 (
2206 "max_elapsed_micros".to_string(),
2207 Value::Int(record.max_elapsed_micros),
2208 ),
2209 (
2210 "avg_elapsed_micros".to_string(),
2211 Value::Int(avg_i64(record.total_elapsed_micros, record.execution_count)),
2212 ),
2213 (
2214 "total_row_count".to_string(),
2215 Value::Int(record.total_row_count),
2216 ),
2217 (
2218 "last_seen_unix_micros".to_string(),
2219 Value::Int(record.last_seen_unix_micros),
2220 ),
2221 (
2222 "last_elapsed_micros".to_string(),
2223 Value::Int(record.last_elapsed_micros),
2224 ),
2225 (
2226 "last_row_count".to_string(),
2227 Value::Int(record.last_row_count),
2228 ),
2229 ("last_success".to_string(), Value::Bool(record.last_success)),
2230 (
2231 "last_error".to_string(),
2232 record
2233 .last_error
2234 .as_ref()
2235 .map(|error| Value::String(error.clone()))
2236 .unwrap_or(Value::Null),
2237 ),
2238 ])
2239 })
2240 .collect()
2241}
2242
2243fn predicate_matches(predicate: &SqlPredicate, row: &Row) -> Result<bool> {
2244 Ok(match &predicate.kind {
2245 ExprKind::And(left, right) => {
2246 predicate_matches(left, row)? && predicate_matches(right, row)?
2247 }
2248 ExprKind::Or(left, right) => {
2249 predicate_matches(left, row)? || predicate_matches(right, row)?
2250 }
2251 ExprKind::Not(inner) => !predicate_matches(inner, row)?,
2252 ExprKind::Compare { left, op, right } => {
2253 let left = system_expression_value(left, row)?;
2254 let right = system_expression_value(right, row)?;
2255 compare_values(&left, *op, &right)?
2256 }
2257 ExprKind::InList {
2258 left,
2259 values,
2260 negated,
2261 } => {
2262 let left = system_expression_value(left, row)?;
2263 let mut matched = false;
2264 for value in values {
2265 let value = system_expression_value(value, row)?;
2266 if compare_values(&left, SqlComparisonOp::Eq, &value)? {
2267 matched = true;
2268 break;
2269 }
2270 }
2271 matched != *negated
2272 }
2273 ExprKind::Like {
2274 left,
2275 pattern,
2276 case_insensitive,
2277 negated,
2278 escape,
2279 } => {
2280 let left = system_expression_value(left, row)?;
2281 let pattern = system_expression_value(pattern, row)?;
2282 let (Value::String(value), Value::String(pattern)) = (left.as_ref(), pattern.as_ref())
2283 else {
2284 return Ok(false);
2285 };
2286 hawdb_sql::sql_like_matches(value, pattern, *escape, *case_insensitive)? != *negated
2287 }
2288 ExprKind::IsNull {
2289 expression,
2290 negated,
2291 } => {
2292 matches!(
2293 system_expression_value(expression, row)?.as_ref(),
2294 Value::Null
2295 ) != *negated
2296 }
2297 _ => {
2298 return Err(HawDBError::Semantic(
2299 "unsupported system SQL predicate".into(),
2300 ))
2301 }
2302 })
2303}
2304
2305fn system_expression_value<'a>(
2306 expression: &'a Expr,
2307 row: &'a Row,
2308) -> Result<std::borrow::Cow<'a, Value>> {
2309 use std::borrow::Cow;
2310 let value = match &expression.kind {
2311 ExprKind::Column(column) => Cow::Borrowed(row.get(&column.name).unwrap_or(&Value::Null)),
2312 ExprKind::Value(SqlValue::Literal(value)) => Cow::Borrowed(value),
2313 ExprKind::Arithmetic { left, op, right } => {
2314 let left = system_expression_value(left, row)?;
2315 let right = system_expression_value(right, row)?;
2316 Cow::Owned(hawdb_core::numeric::evaluate_numeric(
2317 left.as_ref().as_ref(),
2318 *op,
2319 right.as_ref().as_ref(),
2320 )?)
2321 }
2322 ExprKind::Negate(expression) => {
2323 let value = system_expression_value(expression, row)?;
2324 Cow::Owned(hawdb_core::numeric::negate_numeric(
2325 value.as_ref().as_ref(),
2326 )?)
2327 }
2328 ExprKind::Cast { expression, target } => {
2329 let value = system_expression_value(expression, row)?;
2330 Cow::Owned(hawdb_core::numeric::cast_numeric(
2331 value.as_ref().as_ref(),
2332 *target,
2333 )?)
2334 }
2335 _ => {
2336 return Err(HawDBError::Semantic(
2337 "unsupported system SQL scalar expression".into(),
2338 ))
2339 }
2340 };
2341 Ok(value)
2342}
2343
2344fn compare_values(left: &Value, op: SqlComparisonOp, right: &Value) -> Result<bool> {
2345 if matches!(left, Value::Null) || matches!(right, Value::Null) {
2346 return Ok(false);
2347 }
2348 let ordering = if left.logical_type().is_some_and(|kind| kind.is_numeric())
2349 && right.logical_type().is_some_and(|kind| kind.is_numeric())
2350 {
2351 hawdb_core::numeric::compare_numeric(left.as_ref(), right.as_ref())?
2352 } else {
2353 left.cmp(right)
2354 };
2355 Ok(match op {
2356 SqlComparisonOp::Eq => ordering.is_eq(),
2357 SqlComparisonOp::NotEq => !ordering.is_eq(),
2358 SqlComparisonOp::Lt => ordering.is_lt(),
2359 SqlComparisonOp::Lte => !ordering.is_gt(),
2360 SqlComparisonOp::Gt => ordering.is_gt(),
2361 SqlComparisonOp::Gte => !ordering.is_lt(),
2362 })
2363}
2364
2365fn compare_ordered_rows(left: &Row, right: &Row, order_by: &[SqlOrderItem]) -> Ordering {
2366 for item in order_by {
2367 let column = item
2368 .expression
2369 .as_column()
2370 .expect("system ORDER BY columns were validated");
2371 let ordering = left.get(&column.name).cmp(&right.get(&column.name));
2372 let ordering = match item.direction {
2373 SqlOrderDirection::Asc => ordering,
2374 SqlOrderDirection::Desc => ordering.reverse(),
2375 };
2376 if ordering != Ordering::Equal {
2377 return ordering;
2378 }
2379 }
2380 Ordering::Equal
2381}
2382
2383fn system_table(select: &SelectStatement) -> Result<SystemTable> {
2384 let from = select.from_table();
2385 match (from.schema.as_deref(), from.name.as_str()) {
2386 (Some("system"), "tables") => Ok(SystemTable::Tables),
2387 (Some("system"), "vector_columns") => Ok(SystemTable::VectorColumns),
2388 (Some("system"), "properties") => Ok(SystemTable::Properties),
2389 (Some("system"), "indexes") => Ok(SystemTable::Indexes),
2390 (Some("system"), "constraints") => Ok(SystemTable::Constraints),
2391 (Some("system"), "append_tables") => Ok(SystemTable::AppendTables),
2392 (Some("system"), "append_storage") => Ok(SystemTable::AppendStorage),
2393 (Some("system"), "runtime_status") => Ok(SystemTable::RuntimeStatus),
2394 (Some("system"), "runtime_capabilities") => Ok(SystemTable::RuntimeCapabilities),
2395 (Some("system"), "graph_statistics") => Ok(SystemTable::GraphStatistics),
2396 (Some("system"), "projected_graphs") => Ok(SystemTable::ProjectedGraphs),
2397 (Some("system"), "search_projection_changefeed") => {
2398 Ok(SystemTable::SearchProjectionChangefeed)
2399 }
2400 (Some("system"), "plan_cache") => Ok(SystemTable::PlanCache),
2401 (Some("system"), "slow_queries") => Ok(SystemTable::SlowQueries),
2402 (Some("system"), "statement_summary") => Ok(SystemTable::StatementSummary),
2403 (Some("information_schema"), "tables") => Ok(SystemTable::InformationSchemaTables),
2404 (Some("information_schema"), "columns") => Ok(SystemTable::InformationSchemaColumns),
2405 (Some("pg_catalog"), "pg_tables") => Ok(SystemTable::PgTables),
2406 (Some("pg_catalog"), "pg_indexes") => Ok(SystemTable::PgIndexes),
2407 (None, "pg_tables") => Ok(SystemTable::PgTables),
2408 (None, "pg_indexes") => Ok(SystemTable::PgIndexes),
2409 _ => Err(HawDBError::Semantic(format!(
2410 "unknown SQL virtual catalog table {}",
2411 format_table_name(select)
2412 ))),
2413 }
2414}
2415
2416fn validate_projection(table: SystemTable, projection: &[SelectProjection]) -> Result<()> {
2417 for projection in projection {
2418 match projection {
2419 SelectProjection::Wildcard => {}
2420 SelectProjection::Expression {
2421 expression:
2422 Expr {
2423 kind: ExprKind::Column(name),
2424 ..
2425 },
2426 ..
2427 } => validate_column(table, name)?,
2428 SelectProjection::Expression { .. } => {
2429 return Err(HawDBError::Semantic(
2430 "system SQL aggregate expressions are not supported".to_string(),
2431 ));
2432 }
2433 }
2434 }
2435 Ok(())
2436}
2437
2438fn validate_predicate_columns(table: SystemTable, predicate: Option<&SqlPredicate>) -> Result<()> {
2439 let mut result = Ok(());
2440 if let Some(predicate) = predicate {
2441 predicate.visit(&mut |expression| {
2442 if result.is_ok()
2443 && let Some(column) = expression.as_column()
2444 {
2445 result = validate_column(table, column);
2446 }
2447 });
2448 }
2449 result
2450}
2451
2452fn validate_order_columns(table: SystemTable, order_by: &[SqlOrderItem]) -> Result<()> {
2453 for item in order_by {
2454 validate_column(table, item.expression.require_column()?)?;
2455 }
2456 Ok(())
2457}
2458
2459fn validate_column(table: SystemTable, column: &SqlColumnRef) -> Result<()> {
2460 if let Some(qualifier) = &column.qualifier {
2461 let table_name = match table {
2462 SystemTable::Tables => "tables",
2463 SystemTable::VectorColumns => "vector_columns",
2464 SystemTable::Properties => "properties",
2465 SystemTable::Indexes => "indexes",
2466 SystemTable::Constraints => "constraints",
2467 SystemTable::AppendTables => "append_tables",
2468 SystemTable::AppendStorage => "append_storage",
2469 SystemTable::RuntimeStatus => "runtime_status",
2470 SystemTable::RuntimeCapabilities => "runtime_capabilities",
2471 SystemTable::GraphStatistics => "graph_statistics",
2472 SystemTable::ProjectedGraphs => "projected_graphs",
2473 SystemTable::SearchProjectionChangefeed => "search_projection_changefeed",
2474 SystemTable::PlanCache => "plan_cache",
2475 SystemTable::SlowQueries => "slow_queries",
2476 SystemTable::StatementSummary => "statement_summary",
2477 SystemTable::InformationSchemaTables => "tables",
2478 SystemTable::InformationSchemaColumns => "columns",
2479 SystemTable::PgTables => "pg_tables",
2480 SystemTable::PgIndexes => "pg_indexes",
2481 };
2482 if qualifier != table_name {
2483 return Err(HawDBError::Semantic(format!(
2484 "unknown SQL column qualifier {qualifier}"
2485 )));
2486 }
2487 }
2488 if table_columns(table).contains(&column.name.as_str()) {
2489 Ok(())
2490 } else {
2491 Err(HawDBError::Semantic(format!(
2492 "unknown SQL column {}",
2493 column.name
2494 )))
2495 }
2496}
2497
2498fn table_columns(table: SystemTable) -> &'static [&'static str] {
2499 match table {
2500 SystemTable::Tables => &["table_id", "table_name", "table_kind", "state"],
2501 SystemTable::VectorColumns => &[
2502 "table_schema",
2503 "table_name",
2504 "column_name",
2505 "dimensions",
2506 "element_type",
2507 "nullable",
2508 ],
2509 SystemTable::Properties => &[
2510 "property_id",
2511 "table_id",
2512 "table_name",
2513 "table_kind",
2514 "property_name",
2515 "value_type",
2516 "nullable",
2517 "state",
2518 ],
2519 SystemTable::Indexes => &[
2520 "index_id",
2521 "subject_kind",
2522 "subject_id",
2523 "subject_name",
2524 "index_kind",
2525 "property_names",
2526 ],
2527 SystemTable::Constraints => &[
2528 "constraint_id",
2529 "subject_kind",
2530 "subject_id",
2531 "subject_name",
2532 "property_name",
2533 "constraint_kind",
2534 ],
2535 SystemTable::AppendTables => &[
2536 "table_name",
2537 "storage_mode",
2538 "partition_key",
2539 "order_key",
2540 "order_mode",
2541 "generated_order_watermark",
2542 "column_count",
2543 ],
2544 SystemTable::AppendStorage => &[
2545 "canonical_segment_count",
2546 "canonical_segment_bytes",
2547 "resident_segment_payload_bytes",
2548 "resident_descriptor_count",
2549 "live_rows",
2550 "live_payload_bytes",
2551 ],
2552 SystemTable::RuntimeStatus => &[
2553 "commit_epoch",
2554 "read_only",
2555 "max_read_result_rows",
2556 "max_read_result_payload_bytes",
2557 "storage_residency_mode",
2558 "relational_index_mode",
2559 ],
2560 SystemTable::RuntimeCapabilities => &["capability", "enabled"],
2561 SystemTable::GraphStatistics => &[
2562 "statistic_kind",
2563 "computed_at_commit_epoch",
2564 "advanced_statistics_complete",
2565 "histogram_sample_limit",
2566 "label_name",
2567 "relationship_type_name",
2568 "source_label_name",
2569 "target_label_name",
2570 "depth",
2571 "property_name",
2572 "index_id",
2573 "index_kind",
2574 "index_properties",
2575 "index_size",
2576 "unique_values",
2577 "sample_size",
2578 "updates_since_sample",
2579 "stale",
2580 "count",
2581 "histogram",
2582 "sampled",
2583 ],
2584 SystemTable::ProjectedGraphs => &[
2585 "name",
2586 "node_labels",
2587 "relationship_types",
2588 "projection_epoch",
2589 "commit_epoch",
2590 "node_count",
2591 "edge_count",
2592 "reusable",
2593 ],
2594 SystemTable::SearchProjectionChangefeed => &[
2595 "graph_commit_epoch",
2596 "resume_floor_commit_epoch",
2597 "oldest_retained_mutation_id",
2598 "newest_retained_mutation_id",
2599 "retained_mutation_count",
2600 "restart_recoverable",
2601 ],
2602 SystemTable::PlanCache => &["metric", "value"],
2603 SystemTable::SlowQueries => &[
2604 "sequence",
2605 "query_language",
2606 "statement_kind",
2607 "query_digest",
2608 "query_text_hash",
2609 "query_text",
2610 "started_unix_micros",
2611 "elapsed_micros",
2612 "row_count",
2613 "success",
2614 "error",
2615 "slow_log_candidate",
2616 "access_control_policy_epoch",
2617 ],
2618 SystemTable::StatementSummary => &[
2619 "digest",
2620 "query_language",
2621 "query_text",
2622 "sample_query_text_hash",
2623 "statement_kind",
2624 "execution_count",
2625 "success_count",
2626 "error_count",
2627 "total_elapsed_micros",
2628 "max_elapsed_micros",
2629 "avg_elapsed_micros",
2630 "total_row_count",
2631 "last_seen_unix_micros",
2632 "last_elapsed_micros",
2633 "last_row_count",
2634 "last_success",
2635 "last_error",
2636 ],
2637 SystemTable::InformationSchemaTables => &[
2638 "table_catalog",
2639 "table_schema",
2640 "table_name",
2641 "table_type",
2642 "self_referencing_column_name",
2643 "reference_generation",
2644 "user_defined_type_catalog",
2645 "user_defined_type_schema",
2646 "user_defined_type_name",
2647 "is_insertable_into",
2648 "is_typed",
2649 "commit_action",
2650 ],
2651 SystemTable::InformationSchemaColumns => &[
2652 "table_catalog",
2653 "table_schema",
2654 "table_name",
2655 "column_name",
2656 "ordinal_position",
2657 "column_default",
2658 "is_nullable",
2659 "data_type",
2660 "character_maximum_length",
2661 "character_octet_length",
2662 "numeric_precision",
2663 "numeric_precision_radix",
2664 "numeric_scale",
2665 "datetime_precision",
2666 "interval_type",
2667 "interval_precision",
2668 "character_set_catalog",
2669 "character_set_schema",
2670 "character_set_name",
2671 "collation_catalog",
2672 "collation_schema",
2673 "collation_name",
2674 "domain_catalog",
2675 "domain_schema",
2676 "domain_name",
2677 "udt_catalog",
2678 "udt_schema",
2679 "udt_name",
2680 "scope_catalog",
2681 "scope_schema",
2682 "scope_name",
2683 "maximum_cardinality",
2684 "dtd_identifier",
2685 "is_self_referencing",
2686 "is_identity",
2687 "identity_generation",
2688 "identity_start",
2689 "identity_increment",
2690 "identity_maximum",
2691 "identity_minimum",
2692 "identity_cycle",
2693 "is_generated",
2694 "generation_expression",
2695 "is_updatable",
2696 ],
2697 SystemTable::PgTables => &[
2698 "schemaname",
2699 "tablename",
2700 "tableowner",
2701 "tablespace",
2702 "hasindexes",
2703 "hasrules",
2704 "hastriggers",
2705 "rowsecurity",
2706 ],
2707 SystemTable::PgIndexes => &[
2708 "schemaname",
2709 "tablename",
2710 "indexname",
2711 "tablespace",
2712 "indexdef",
2713 ],
2714 }
2715}
2716
2717fn format_table_name(select: &SelectStatement) -> String {
2718 let from = select.from_table();
2719 match &from.schema {
2720 Some(schema) => format!("{schema}.{}", from.name),
2721 None => from.name.clone(),
2722 }
2723}
2724
2725fn option_usize_value(value: Option<usize>) -> Value {
2726 value.map(usize_value).unwrap_or(Value::Null)
2727}
2728
2729fn option_i64_value(value: Option<i64>) -> Value {
2730 value.map(Value::Int).unwrap_or(Value::Null)
2731}
2732
2733fn option_u64_value(value: Option<u64>) -> Value {
2734 value.map(u64_value).unwrap_or(Value::Null)
2735}
2736
2737fn usize_value(value: usize) -> Value {
2738 Value::Int(i64::try_from(value).unwrap_or(i64::MAX))
2739}
2740
2741fn u32_value(value: u32) -> Value {
2742 Value::Int(i64::from(value))
2743}
2744
2745fn u64_value(value: u64) -> Value {
2746 Value::Int(i64::try_from(value).unwrap_or(i64::MAX))
2747}
2748
2749const fn storage_residency_mode_name(mode: StorageResidencyMode) -> &'static str {
2750 match mode {
2751 StorageResidencyMode::Auto => "auto",
2752 StorageResidencyMode::Materialized => "materialized",
2753 StorageResidencyMode::OutOfCore => "out_of_core",
2754 }
2755}
2756
2757const fn relational_index_mode_name(mode: RelationalIndexMode) -> &'static str {
2758 match mode {
2759 RelationalIndexMode::Materialized => "materialized",
2760 RelationalIndexMode::Shadow => "shadow",
2761 RelationalIndexMode::DemandPaged => "demand_paged",
2762 RelationalIndexMode::Authoritative => "authoritative",
2763 }
2764}
2765
2766fn avg_i64(total: i64, count: i64) -> i64 {
2767 if count <= 0 {
2768 0
2769 } else {
2770 total / count
2771 }
2772}
2773
2774fn saturating_i64_from_u128(value: u128) -> i64 {
2775 i64::try_from(value).unwrap_or(i64::MAX)
2776}
2777
2778fn unix_now_micros() -> i64 {
2779 let micros = SystemTime::now()
2780 .duration_since(UNIX_EPOCH)
2781 .map(|duration| duration.as_micros())
2782 .unwrap_or(0);
2783 saturating_i64_from_u128(micros)
2784}
2785
2786fn truncate_utf8(input: &str, max_bytes: usize) -> String {
2787 if input.len() <= max_bytes {
2788 return input.to_string();
2789 }
2790 let mut end = max_bytes;
2791 while !input.is_char_boundary(end) {
2792 end -= 1;
2793 }
2794 input[..end].to_string()
2795}
2796
2797mod graph_store_adapter;
2798
2799#[cfg(test)]
2800mod statement_summary_tests;
2801
2802#[cfg(test)]
2803mod tests {
2804 use super::*;
2805
2806 #[test]
2807 fn catalog_numeric_predicates_share_exact_arithmetic_and_comparison() {
2808 let row = BTreeMap::from([("value".into(), Value::Int(9_007_199_254_740_993))]);
2809 for sql in [
2810 "SELECT value FROM system.plan_cache WHERE value = 9007199254740993.00",
2811 "SELECT value FROM system.plan_cache WHERE CAST(value AS DECIMAL(38,2)) + 0.01 = $1",
2812 "SELECT value FROM system.plan_cache WHERE value IN (9007199254740993.00)",
2813 ] {
2814 let SqlStatement::Select(select) = hawdb_sql::parse_postgres_sql(sql).unwrap() else {
2815 panic!("SELECT");
2816 };
2817 let predicate = bind_predicate(
2818 select.selection.unwrap(),
2819 &[Value::Decimal("9007199254740993.01".parse().unwrap())],
2820 )
2821 .unwrap();
2822 assert!(predicate_matches(&predicate, &row).unwrap(), "{sql}");
2823 }
2824 let SqlStatement::Select(select) = hawdb_sql::parse_postgres_sql(
2825 "SELECT value FROM system.plan_cache WHERE value / 0 > 1",
2826 )
2827 .unwrap() else {
2828 panic!("SELECT");
2829 };
2830 assert!(predicate_matches(&select.selection.unwrap(), &row).is_err());
2831 }
2832
2833 #[derive(Debug, Default)]
2834 struct TestStore;
2835
2836 impl SystemSqlStore for TestStore {
2837 fn commit_epoch(&self) -> u64 {
2838 0
2839 }
2840
2841 fn append_storage_residency_report(&self) -> AppendStorageResidencyReport {
2842 AppendStorageResidencyReport::default()
2843 }
2844
2845 fn statistics(&self, _catalog: &Catalog) -> GraphStatistics {
2846 GraphStatistics::default()
2847 }
2848
2849 fn projected_graph_statuses(&self) -> Vec<ProjectedGraphStatus> {
2850 Vec::new()
2851 }
2852
2853 fn search_projection_changefeed_status(&self) -> SearchProjectionChangefeedStatus {
2854 SearchProjectionChangefeedStatus {
2855 graph_commit_epoch: 0,
2856 resume_floor_commit_epoch: 0,
2857 oldest_retained_mutation_id: None,
2858 newest_retained_mutation_id: None,
2859 first_rebuild_required_mutation_id: None,
2860 retained_mutation_count: 0,
2861 retained_bytes: 0,
2862 max_retained_bytes: None,
2863 restart_recoverable: true,
2864 }
2865 }
2866 }
2867
2868 fn default_runtime() -> SystemRuntimeSnapshot {
2869 SystemRuntimeSnapshot::new(
2870 false,
2871 Some(512),
2872 Some(4 * 1024 * 1024),
2873 StorageResidencyMode::Auto,
2874 RelationalIndexMode::default(),
2875 RuntimeCapabilities::default(),
2876 )
2877 }
2878
2879 #[test]
2880 fn predicate_binding_preserves_source_spans_and_system_null_behavior() {
2881 let SqlStatement::Select(select) = hawdb_sql::prepare_postgres_sql(
2882 "SELECT value FROM system.plan_cache WHERE \
2883 (value IN ($1, $2) OR NOT value = $3) AND \
2884 (metric LIKE $4 OR value IS NULL)",
2885 )
2886 .unwrap()
2887 .statement
2888 else {
2889 panic!("expected SELECT");
2890 };
2891 let predicate = select.selection.unwrap();
2892 let mut before = Vec::new();
2893 predicate.visit(&mut |node| before.push(node.span));
2894 let bound = bind_predicate(
2895 predicate.clone(),
2896 &[
2897 Value::Null,
2898 Value::Int(5),
2899 Value::Null,
2900 Value::String("hit%".to_owned()),
2901 ],
2902 )
2903 .unwrap();
2904 let mut after = Vec::new();
2905 let mut parameters = Vec::new();
2906 bound.visit(&mut |node| {
2907 after.push(node.span);
2908 if let ExprKind::Value(SqlValue::Parameter(position)) = &node.kind {
2909 parameters.push(*position);
2910 }
2911 });
2912 assert_eq!(after, before);
2913 assert!(parameters.is_empty());
2914 assert!(predicate_matches(
2915 &bound,
2916 &BTreeMap::from([
2917 ("value".to_owned(), Value::Null),
2918 ("metric".to_owned(), Value::String("misses".to_owned())),
2919 ])
2920 )
2921 .unwrap());
2922 assert!(bind_predicate(
2923 predicate,
2924 &[
2925 Value::Null,
2926 Value::Int(5),
2927 Value::Null,
2928 Value::String("dangling\\".to_owned()),
2929 ]
2930 )
2931 .unwrap_err()
2932 .to_string()
2933 .contains("ends with its escape"));
2934 }
2935
2936 #[test]
2937 fn query_plan_cache_virtual_table_with_predicate_and_projection() {
2938 let catalog = Catalog::default();
2939 let store = TestStore;
2940 let relational_state = RelationalState::default();
2941 let append_state = AppendState::default();
2942 let stats = PlanCacheStats {
2943 max_bytes: Some(16384),
2944 max_entry_bytes: Some(8192),
2945 charged_bytes: Some(4096),
2946 rejected_admissions: 2,
2947 max_entries: Some(128),
2948 entries: 3,
2949 hits: 5,
2950 misses: 7,
2951 admissions: 4,
2952 disabled_misses: 0,
2953 bypasses: 2,
2954 evictions: 1,
2955 memory_pressure_events: 1,
2956 };
2957
2958 let context = SystemSqlContext {
2959 catalog: &catalog,
2960 store: &store,
2961 relational_state: &relational_state,
2962 append_state: &append_state,
2963 runtime: default_runtime(),
2964 plan_cache_stats: &stats,
2965 slow_queries: &[],
2966 statement_summaries: &[],
2967 };
2968 let sql = "SELECT value FROM system.plan_cache WHERE metric = 'hits'";
2969 let output = query_sql(sql, None, None, &context).expect("system plan cache query");
2970
2971 assert_eq!(
2972 output.rows,
2973 vec![BTreeMap::from([("value".to_string(), Value::Int(5))])]
2974 );
2975 let payload_bytes = output.payload_bytes();
2976 for (metric, expected) in [
2977 ("max_bytes", 16384),
2978 ("max_entry_bytes", 8192),
2979 ("charged_bytes", 4096),
2980 ("rejected_admissions", 2),
2981 ] {
2982 let output = query_sql(
2983 &format!("SELECT value FROM system.plan_cache WHERE metric = '{metric}'"),
2984 Some(1),
2985 Some(1024),
2986 &context,
2987 )
2988 .unwrap();
2989 assert_eq!(
2990 output.rows,
2991 vec![BTreeMap::from([("value".into(), Value::Int(expected))])]
2992 );
2993 }
2994 let exact = query_sql(sql, Some(1), Some(payload_bytes), &context).unwrap();
2995 assert_eq!(exact.rows, output.rows);
2996 for (rows, payload, resource, limit) in [
2997 (Some(0), Some(0), hawdb_core::ReadBudgetResource::Rows, 0),
2998 (
2999 Some(1),
3000 Some(payload_bytes - 1),
3001 hawdb_core::ReadBudgetResource::PayloadBytes,
3002 payload_bytes - 1,
3003 ),
3004 ] {
3005 let error = query_sql(sql, rows, payload, &context).unwrap_err();
3006 assert!(
3007 matches!(&error, HawDBError::ReadBudgetExceeded(cause)
3008 if cause.resource == resource && cause.limit == limit),
3009 "system SQL lost its result budget cause: {error}"
3010 );
3011 }
3012 }
3013
3014 #[test]
3015 fn query_system_table_with_postgres_parameters() {
3016 let catalog = Catalog::default();
3017 let store = TestStore;
3018 let relational_state = RelationalState::default();
3019 let append_state = AppendState::default();
3020 let stats = PlanCacheStats {
3021 max_bytes: None,
3022 max_entry_bytes: None,
3023 charged_bytes: None,
3024 rejected_admissions: 0,
3025 max_entries: Some(128),
3026 entries: 3,
3027 hits: 5,
3028 misses: 7,
3029 admissions: 4,
3030 disabled_misses: 0,
3031 bypasses: 2,
3032 evictions: 1,
3033 memory_pressure_events: 1,
3034 };
3035 let context = SystemSqlContext {
3036 catalog: &catalog,
3037 store: &store,
3038 relational_state: &relational_state,
3039 append_state: &append_state,
3040 runtime: default_runtime(),
3041 plan_cache_stats: &stats,
3042 slow_queries: &[],
3043 statement_summaries: &[],
3044 };
3045
3046 let output = query_sql_with_params(
3047 "SELECT metric, value FROM system.plan_cache \
3048 WHERE metric IN ($1, $2) ORDER BY metric ASC LIMIT $3 OFFSET $4",
3049 &[
3050 Value::String("hits".to_string()),
3051 Value::String("misses".to_string()),
3052 Value::Int(1),
3053 Value::Int(1),
3054 ],
3055 None,
3056 None,
3057 &context,
3058 )
3059 .expect("parameterized system table query");
3060
3061 assert_eq!(
3062 output.rows,
3063 vec![BTreeMap::from([
3064 ("metric".to_string(), Value::String("misses".to_string())),
3065 ("value".to_string(), Value::Int(7)),
3066 ])]
3067 );
3068 }
3069
3070 #[test]
3071 fn query_system_table_rejects_parameter_contract_mismatch() {
3072 let catalog = Catalog::default();
3073 let store = TestStore;
3074 let relational_state = RelationalState::default();
3075 let append_state = AppendState::default();
3076 let stats = PlanCacheStats {
3077 max_bytes: None,
3078 max_entry_bytes: None,
3079 charged_bytes: None,
3080 rejected_admissions: 0,
3081 max_entries: None,
3082 entries: 0,
3083 hits: 0,
3084 misses: 0,
3085 admissions: 0,
3086 disabled_misses: 0,
3087 bypasses: 0,
3088 evictions: 0,
3089 memory_pressure_events: 0,
3090 };
3091 let context = SystemSqlContext {
3092 catalog: &catalog,
3093 store: &store,
3094 relational_state: &relational_state,
3095 append_state: &append_state,
3096 runtime: default_runtime(),
3097 plan_cache_stats: &stats,
3098 slow_queries: &[],
3099 statement_summaries: &[],
3100 };
3101
3102 let missing = query_sql_with_params(
3103 "SELECT * FROM system.plan_cache WHERE metric = $1",
3104 &[],
3105 None,
3106 None,
3107 &context,
3108 )
3109 .expect_err("missing parameter must fail");
3110 assert!(missing.to_string().contains("requires 1 parameters"));
3111
3112 let invalid_bound = query_sql_with_params(
3113 "SELECT * FROM system.plan_cache LIMIT $1",
3114 &[Value::String("one".to_string())],
3115 None,
3116 None,
3117 &context,
3118 )
3119 .expect_err("non-integer LIMIT must fail");
3120 assert!(invalid_bound
3121 .to_string()
3122 .contains("must be a non-negative integer"));
3123 }
3124
3125 #[test]
3126 fn query_slow_queries_pushes_filter_order_and_limit_into_scan() {
3127 let catalog = Catalog::default();
3128 let store = TestStore;
3129 let relational_state = RelationalState::default();
3130 let append_state = AppendState::default();
3131 let stats = PlanCacheStats {
3132 max_bytes: None,
3133 max_entry_bytes: None,
3134 charged_bytes: None,
3135 rejected_admissions: 0,
3136 max_entries: None,
3137 entries: 0,
3138 hits: 0,
3139 misses: 0,
3140 admissions: 0,
3141 disabled_misses: 0,
3142 bypasses: 0,
3143 evictions: 0,
3144 memory_pressure_events: 0,
3145 };
3146 let records = vec![
3147 SlowQueryRecord {
3148 sequence: 1,
3149 query_language: "cypher".to_string(),
3150 statement_kind: "match_return".to_string(),
3151 query_digest: "q1:first".to_string(),
3152 query_text_hash: "t1:first".to_string(),
3153 query_text: "MATCH (m:Memory) RETURN m".to_string(),
3154 started_unix_micros: 10,
3155 elapsed_micros: 200,
3156 row_count: 1,
3157 success: true,
3158 error: None,
3159 slow_log_candidate: false,
3160 access_control_policy_epoch: None,
3161 vector_execution_reports: Vec::new(),
3162 },
3163 SlowQueryRecord {
3164 sequence: 2,
3165 query_language: "cypher".to_string(),
3166 statement_kind: "match_return".to_string(),
3167 query_digest: "q1:second".to_string(),
3168 query_text_hash: "t1:second".to_string(),
3169 query_text: "MATCH (m:Memory) RETURN m ORDER BY m.id".to_string(),
3170 started_unix_micros: 20,
3171 elapsed_micros: 500,
3172 row_count: 2,
3173 success: true,
3174 error: None,
3175 slow_log_candidate: true,
3176 access_control_policy_epoch: None,
3177 vector_execution_reports: Vec::new(),
3178 },
3179 SlowQueryRecord {
3180 sequence: 3,
3181 query_language: "cypher".to_string(),
3182 statement_kind: "match_return".to_string(),
3183 query_digest: "q1:third".to_string(),
3184 query_text_hash: "t1:third".to_string(),
3185 query_text: "MATCH (m:Memory {id: 'x'}) RETURN m".to_string(),
3186 started_unix_micros: 30,
3187 elapsed_micros: 300,
3188 row_count: 1,
3189 success: true,
3190 error: None,
3191 slow_log_candidate: true,
3192 access_control_policy_epoch: None,
3193 vector_execution_reports: Vec::new(),
3194 },
3195 ];
3196
3197 let output = query_sql(
3198 "SELECT sequence, elapsed_micros FROM system.slow_queries \
3199 WHERE slow_log_candidate = true \
3200 ORDER BY elapsed_micros DESC LIMIT 1",
3201 None,
3202 None,
3203 &SystemSqlContext {
3204 catalog: &catalog,
3205 store: &store,
3206 relational_state: &relational_state,
3207 append_state: &append_state,
3208 runtime: default_runtime(),
3209 plan_cache_stats: &stats,
3210 slow_queries: &records,
3211 statement_summaries: &[],
3212 },
3213 )
3214 .expect("system slow query scan");
3215
3216 assert_eq!(
3217 output.rows,
3218 vec![BTreeMap::from([
3219 ("elapsed_micros".to_string(), Value::Int(500)),
3220 ("sequence".to_string(), Value::Int(2)),
3221 ])]
3222 );
3223 }
3224
3225 #[test]
3226 fn query_catalog_virtual_tables_expose_schema_without_typed_getters() {
3227 let mut catalog = Catalog::default();
3228 let store = TestStore;
3229 let relational_state = RelationalState::default();
3230 let append_state = AppendState::default();
3231 let memory_label = catalog.get_or_create_label("Memory");
3232 let memory_table = catalog.get_or_create_table(TableKind::Node, "Memory");
3233 catalog.get_or_create_property(memory_table, "id", PropertyType::String, false);
3234 catalog.get_or_create_property_index_with_kind(memory_label, "id", IndexKind::Equality);
3235 catalog.get_or_create_unique_constraint(memory_label, "id");
3236 let stats = PlanCacheStats {
3237 max_bytes: None,
3238 max_entry_bytes: None,
3239 charged_bytes: None,
3240 rejected_admissions: 0,
3241 max_entries: None,
3242 entries: 0,
3243 hits: 0,
3244 misses: 0,
3245 admissions: 0,
3246 disabled_misses: 0,
3247 bypasses: 0,
3248 evictions: 0,
3249 memory_pressure_events: 0,
3250 };
3251 let context = SystemSqlContext {
3252 catalog: &catalog,
3253 store: &store,
3254 relational_state: &relational_state,
3255 append_state: &append_state,
3256 runtime: default_runtime(),
3257 plan_cache_stats: &stats,
3258 slow_queries: &[],
3259 statement_summaries: &[],
3260 };
3261
3262 let tables = query_sql(
3263 "SELECT table_id, table_name, table_kind, state FROM system.tables",
3264 None,
3265 None,
3266 &context,
3267 )
3268 .expect("system tables query");
3269 assert_eq!(
3270 tables.rows,
3271 vec![BTreeMap::from([
3272 ("state".to_string(), Value::String("public".to_string())),
3273 ("table_id".to_string(), Value::Int(0)),
3274 ("table_kind".to_string(), Value::String("node".to_string()),),
3275 (
3276 "table_name".to_string(),
3277 Value::String("Memory".to_string()),
3278 ),
3279 ])]
3280 );
3281
3282 let properties = query_sql(
3283 "SELECT table_name, property_name, value_type, nullable \
3284 FROM system.properties WHERE table_name = 'Memory'",
3285 None,
3286 None,
3287 &context,
3288 )
3289 .expect("system properties query");
3290 assert_eq!(
3291 properties.rows,
3292 vec![BTreeMap::from([
3293 ("nullable".to_string(), Value::Bool(false)),
3294 ("property_name".to_string(), Value::String("id".to_string()),),
3295 (
3296 "table_name".to_string(),
3297 Value::String("Memory".to_string()),
3298 ),
3299 (
3300 "value_type".to_string(),
3301 Value::String("string".to_string()),
3302 ),
3303 ])]
3304 );
3305
3306 let indexes = query_sql(
3307 "SELECT subject_name, index_kind, property_names FROM system.indexes",
3308 None,
3309 None,
3310 &context,
3311 )
3312 .expect("system indexes query");
3313 assert_eq!(
3314 indexes.rows,
3315 vec![BTreeMap::from([
3316 (
3317 "index_kind".to_string(),
3318 Value::String("equality".to_string()),
3319 ),
3320 (
3321 "property_names".to_string(),
3322 Value::List(vec![Value::String("id".to_string())]),
3323 ),
3324 (
3325 "subject_name".to_string(),
3326 Value::String("Memory".to_string()),
3327 ),
3328 ])]
3329 );
3330
3331 let constraints = query_sql(
3332 "SELECT subject_name, property_name, constraint_kind FROM system.constraints",
3333 None,
3334 None,
3335 &context,
3336 )
3337 .expect("system constraints query");
3338 assert_eq!(
3339 constraints.rows,
3340 vec![BTreeMap::from([
3341 (
3342 "constraint_kind".to_string(),
3343 Value::String("node_property_unique".to_string()),
3344 ),
3345 ("property_name".to_string(), Value::String("id".to_string()),),
3346 (
3347 "subject_name".to_string(),
3348 Value::String("Memory".to_string()),
3349 ),
3350 ])]
3351 );
3352 }
3353}
3354
3355#[cfg(all(test, feature = "loom-tests"))]
3356pub(crate) mod loom_tests {
3357 use super::{SlowQueryCompletion, SlowQueryLog, SlowQueryRecord};
3358 use hawdb_query::QueryIdentity;
3359 use loom::sync::{Arc, Mutex};
3360 use loom::thread;
3361
3362 #[test]
3363 fn slow_query_ring_preserves_bounds_under_modeled_concurrent_access() {
3364 loom::model(|| {
3365 let log = Arc::new(Mutex::new(SlowQueryLog::new(2)));
3366
3367 let first_writer = spawn_slow_query_writer(Arc::clone(&log), "first");
3368 let second_writer = spawn_slow_query_writer(Arc::clone(&log), "second");
3369 let snapshotter = {
3370 let log = Arc::clone(&log);
3371 thread::spawn(move || {
3372 let snapshot = log.lock().unwrap().snapshot();
3373 assert_snapshot_invariants(&snapshot, 2);
3374 })
3375 };
3376
3377 first_writer.join().unwrap();
3378 second_writer.join().unwrap();
3379 snapshotter.join().unwrap();
3380
3381 let snapshot = log.lock().unwrap().snapshot();
3382 assert_snapshot_invariants(&snapshot, 2);
3383 assert_eq!(snapshot.len(), 2);
3384 assert_eq!(snapshot.last().map(|record| record.sequence), Some(2));
3385 });
3386 }
3387
3388 fn spawn_slow_query_writer(
3389 log: Arc<Mutex<SlowQueryLog>>,
3390 query_text: &'static str,
3391 ) -> thread::JoinHandle<()> {
3392 thread::spawn(move || {
3393 let query_identity = QueryIdentity::new("cypher", query_text);
3394 log.lock()
3395 .unwrap()
3396 .push(SlowQueryRecord::completed(SlowQueryCompletion {
3397 query_language: "cypher",
3398 statement_kind: "match_return",
3399 query_text,
3400 query_identity: &query_identity,
3401 elapsed_micros: 1,
3402 row_count: 1,
3403 success: true,
3404 error: None,
3405 slow_log_candidate: true,
3406 access_control_policy_epoch: None,
3407 vector_execution_reports: Vec::new(),
3408 }));
3409 })
3410 }
3411
3412 fn assert_snapshot_invariants(snapshot: &[SlowQueryRecord], capacity: usize) {
3413 assert!(snapshot.len() <= capacity);
3414 for pair in snapshot.windows(2) {
3415 assert!(pair[0].sequence < pair[1].sequence);
3416 }
3417 }
3418}