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