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