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