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