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