1use std::sync::{Arc, Mutex};
21
22use rustc_hash::{FxHashMap, FxHashSet};
23
24use crate::aggregation::{AggregationExecutorExt, AggregationHost};
25use crate::context::ExecutionContext;
26use crate::expression::RowFilter;
27use crate::mutation::host::ActiveTransaction;
28use crate::operators::reference_unique_lookup::{
29 execute_unique_lookup_join_batch, materialize_unique_lookup_candidates, LookupEdgeCardinality,
30 LookupEdgeFallback, UniqueLookupIntegrity, UniqueLookupRows,
31};
32use crate::result::ExecutorResult;
33use crate::utils::{combine_predicates_with_and, flatten_and_predicates, RetainedRowsBudget};
34use radixdb_core::{
35 DataType, Error, NavigationErrorCode, ReferenceTargetKey, Result, Row, RowVec, SchemaColumnId,
36 SchemaTableId, Value,
37};
38use radixdb_sql::ast::{
39 Expression, GroupByClause, GroupByModifier, Identifier, InfixExpression, InfixOperator,
40 JoinTableSource, NullLiteral, QualifiedIdentifier, SelectStatement, SimpleTableSource,
41 Statement, WindowFrame, WindowFrameBound,
42};
43use radixdb_storage::mvcc::engine::MVCCEngine;
44use radixdb_storage::traits::{Engine, QueryResult, ScanPlan, Table};
45
46mod binding;
47mod execution;
48mod rewrite;
49
50#[cfg(any(test, feature = "test-hooks"))]
51#[doc(hidden)]
52pub type SourceMaterializedTestHook =
53 Arc<dyn Fn(&ReferenceExpandPlan, &ExecutionContext) + Send + Sync>;
54
55#[cfg(any(test, feature = "test-hooks"))]
56static SOURCE_MATERIALIZED_TEST_HOOK: std::sync::LazyLock<
57 Mutex<Option<SourceMaterializedTestHook>>,
58> = std::sync::LazyLock::new(|| Mutex::new(None));
59
60#[cfg(any(test, feature = "test-hooks"))]
61static SOURCE_MATERIALIZED_TEST_HOOK_OWNER: Mutex<()> = Mutex::new(());
62
63#[cfg(any(test, feature = "test-hooks"))]
64#[doc(hidden)]
65pub fn run_source_materialized_test_hook(plan: &ReferenceExpandPlan, context: &ExecutionContext) {
66 let hook = SOURCE_MATERIALIZED_TEST_HOOK
67 .lock()
68 .unwrap_or_else(|poisoned| poisoned.into_inner())
69 .clone();
70 if let Some(hook) = hook {
71 hook(plan, context);
72 }
73}
74
75#[cfg(any(test, feature = "test-hooks"))]
76#[doc(hidden)]
77pub struct SourceMaterializedTestHookGuard {
78 _owner: std::sync::MutexGuard<'static, ()>,
79}
80
81#[cfg(any(test, feature = "test-hooks"))]
82impl SourceMaterializedTestHookGuard {
83 #[doc(hidden)]
84 pub fn install(hook: SourceMaterializedTestHook) -> Self {
85 let owner = SOURCE_MATERIALIZED_TEST_HOOK_OWNER
86 .lock()
87 .unwrap_or_else(|poisoned| poisoned.into_inner());
88 *SOURCE_MATERIALIZED_TEST_HOOK
89 .lock()
90 .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(hook);
91 Self { _owner: owner }
92 }
93}
94
95#[cfg(any(test, feature = "test-hooks"))]
96impl Drop for SourceMaterializedTestHookGuard {
97 fn drop(&mut self) {
98 *SOURCE_MATERIALIZED_TEST_HOOK
99 .lock()
100 .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
101 }
102}
103
104pub use binding::{
105 bind_navigation_paths, bind_reference_expand_plan, bind_reference_expand_plan_for_execution,
106 reject_navigation_in_write_statement,
107};
108#[doc(hidden)]
109pub use execution::verify_unique_lookup_integrity;
110
111pub trait NavigationHost: AggregationHost {
115 fn navigation_engine(&self) -> &Arc<MVCCEngine>;
116 fn navigation_active_transaction(&self) -> &Mutex<Option<ActiveTransaction>>;
117 fn navigation_execute_select(
118 &self,
119 statement: &SelectStatement,
120 context: &ExecutionContext,
121 ) -> Result<Box<dyn QueryResult>>;
122 fn navigation_project_rows_with_alias(
123 &self,
124 select_expressions: &[Expression],
125 rows: RowVec,
126 columns: &[String],
127 columns_lower: Option<&[String]>,
128 context: &ExecutionContext,
129 table_alias: Option<&str>,
130 ) -> Result<RowVec>;
131 fn navigation_source_materialized(
132 &self,
133 _plan: &ReferenceExpandPlan,
134 _context: &ExecutionContext,
135 ) {
136 }
137}
138
139pub struct NavigationExecutor<'a, H: NavigationHost + ?Sized> {
141 host: &'a H,
142}
143
144impl<'a, H: NavigationHost + ?Sized> NavigationExecutor<'a, H> {
145 fn new(host: &'a H) -> Self {
146 Self { host }
147 }
148}
149
150pub trait NavigationExecutorExt: NavigationHost {
152 fn execute_reference_projection(
153 &self,
154 select: &SelectStatement,
155 plan: &ReferenceExpandPlan,
156 context: &ExecutionContext,
157 ) -> Result<Box<dyn QueryResult>> {
158 NavigationExecutor::new(self).execute_reference_projection(select, plan, context)
159 }
160
161 fn execute_reference_projection_with_metrics(
162 &self,
163 select: &SelectStatement,
164 plan: &ReferenceExpandPlan,
165 context: &ExecutionContext,
166 ) -> Result<(Box<dyn QueryResult>, ReferenceExpandMetrics)> {
167 NavigationExecutor::new(self)
168 .execute_reference_projection_with_metrics(select, plan, context)
169 }
170}
171
172impl<T: NavigationHost + ?Sized> NavigationExecutorExt for T {}
173
174use execution::*;
175use rewrite::*;
176
177#[derive(Debug, Clone, PartialEq, Eq, Hash)]
179pub struct RootRelationInstance {
180 ordinal: u32,
181 table: SchemaTableId,
182}
183
184#[allow(dead_code)] impl RootRelationInstance {
186 pub fn ordinal(&self) -> u32 {
187 self.ordinal
188 }
189
190 pub fn table(&self) -> &SchemaTableId {
191 &self.table
192 }
193}
194
195#[derive(Debug, Clone, PartialEq, Eq, Hash)]
196pub struct ReferenceStepIdentity {
197 source_column: SchemaColumnId,
198 target_key_column: SchemaColumnId,
199}
200
201impl ReferenceExpandEdgeIdentity {
202 pub fn depth(&self) -> usize {
204 self.steps.len()
205 }
206}
207
208#[derive(Debug, Clone, PartialEq, Eq)]
209pub struct ReferenceStep {
210 identity: ReferenceStepIdentity,
211 target_key: ReferenceTargetKey,
212 source_nullable: bool,
213}
214
215#[allow(dead_code)] impl ReferenceStep {
217 pub fn source_column(&self) -> &SchemaColumnId {
218 &self.identity.source_column
219 }
220
221 pub fn target_table(&self) -> &SchemaTableId {
222 self.identity.target_key_column.table()
223 }
224
225 pub fn target_key_column(&self) -> &SchemaColumnId {
226 &self.identity.target_key_column
227 }
228
229 pub fn target_key(&self) -> ReferenceTargetKey {
230 self.target_key
231 }
232
233 pub fn source_nullable(&self) -> bool {
234 self.source_nullable
235 }
236}
237
238#[derive(Debug, Clone, PartialEq, Eq, Hash)]
240pub struct NavigationPathIdentity {
241 root: RootRelationInstance,
242 steps: Vec<ReferenceStepIdentity>,
243 terminal_column: SchemaColumnId,
244}
245
246#[derive(Debug, Clone, PartialEq, Eq)]
248pub struct NavigationExpr {
249 identity: NavigationPathIdentity,
250 steps: Vec<ReferenceStep>,
251 terminal_type: DataType,
252 nullable: bool,
253 display_path: String,
254}
255
256#[derive(Debug, Clone, Copy, PartialEq, Eq)]
257pub enum ReferenceSemantics {
258 Left,
259}
260
261#[derive(Debug, Clone, Copy, PartialEq, Eq)]
262pub enum ReferenceIntegrityCheck {
263 Required,
264}
265
266#[derive(Debug, Clone, PartialEq, Eq, Hash)]
267pub struct ReferenceExpandEdgeIdentity {
268 root: RootRelationInstance,
269 steps: Vec<ReferenceStepIdentity>,
270}
271
272#[derive(Debug, Clone, PartialEq, Eq)]
274pub struct ReferenceExpandEdge {
275 identity: ReferenceExpandEdgeIdentity,
276 source_column: SchemaColumnId,
277 target_key_column: SchemaColumnId,
278 target_key: ReferenceTargetKey,
279 required_columns: Vec<SchemaColumnId>,
280 semantics: ReferenceSemantics,
281 integrity_check: ReferenceIntegrityCheck,
282}
283
284#[allow(dead_code)] impl ReferenceExpandEdge {
286 pub fn identity(&self) -> &ReferenceExpandEdgeIdentity {
287 &self.identity
288 }
289
290 pub fn source_column(&self) -> &SchemaColumnId {
291 &self.source_column
292 }
293
294 pub fn target_key_column(&self) -> &SchemaColumnId {
295 &self.target_key_column
296 }
297
298 pub fn target_key(&self) -> ReferenceTargetKey {
299 self.target_key
300 }
301
302 pub fn required_columns(&self) -> &[SchemaColumnId] {
303 &self.required_columns
304 }
305
306 pub fn semantics(&self) -> ReferenceSemantics {
307 self.semantics
308 }
309
310 pub fn integrity_check(&self) -> ReferenceIntegrityCheck {
311 self.integrity_check
312 }
313}
314
315#[derive(Debug, Clone, PartialEq, Eq)]
317pub struct ReferenceExpandPath {
318 identity: NavigationPathIdentity,
319 edge_indices: Vec<usize>,
320 terminal_type: DataType,
321 nullable: bool,
322 display_paths: Vec<String>,
323}
324
325#[allow(dead_code)] impl ReferenceExpandPath {
327 pub fn identity(&self) -> &NavigationPathIdentity {
328 &self.identity
329 }
330
331 pub fn edge_indices(&self) -> &[usize] {
332 &self.edge_indices
333 }
334
335 pub fn terminal_type(&self) -> DataType {
336 self.terminal_type
337 }
338
339 pub fn nullable(&self) -> bool {
340 self.nullable
341 }
342
343 pub fn display_paths(&self) -> &[String] {
344 &self.display_paths
345 }
346}
347
348#[doc(hidden)]
353#[derive(Debug, Clone, PartialEq, Eq)]
354pub struct ReferenceExpandPlan {
355 schema_scope_id: u64,
356 schema_generation: u64,
357 paths: Vec<ReferenceExpandPath>,
358 edges: Vec<ReferenceExpandEdge>,
359}
360
361#[doc(hidden)]
362#[derive(Debug, Clone, Default, PartialEq, Eq)]
363pub enum CachedReferenceExpand {
364 #[default]
365 Unknown,
366 NoPaths {
367 schema_scope_id: u64,
368 schema_generation: u64,
369 },
370 Plan(ReferenceExpandPlan),
371}
372
373#[derive(Debug, Clone, Copy, PartialEq, Eq)]
374pub enum ReferenceLookupStrategy {
375 DirectUnique,
376 IndexNestedLoop,
377 UniqueBatch,
378 TargetHashScan,
379 MergeJoin,
380 SnapshotScanFallback,
381}
382
383impl ReferenceLookupStrategy {
384 fn explain_name(self) -> &'static str {
385 match self {
386 Self::DirectUnique => "direct_unique_lookup",
387 Self::IndexNestedLoop => "index_nested_loop",
388 Self::UniqueBatch => "unique_batch_lookup",
389 Self::TargetHashScan => "target_hash_scan",
390 Self::MergeJoin => "merge_join",
391 Self::SnapshotScanFallback => "snapshot_scan_fallback",
392 }
393 }
394}
395
396#[derive(Debug, Clone, Copy, PartialEq, Eq)]
397pub enum ReferenceStorageMode {
398 HotMvcc,
399 ColdArtifact,
400 HybridArtifactHot,
401}
402
403impl ReferenceStorageMode {
404 fn explain_name(self) -> &'static str {
405 match self {
406 Self::HotMvcc => "hot_mvcc",
407 Self::ColdArtifact => "cold_artifact",
408 Self::HybridArtifactHot => "mixed_cold_artifact_hot",
409 }
410 }
411}
412
413#[derive(Debug, Clone, Copy, PartialEq, Eq)]
414pub enum ReferenceExecutionDirection {
415 SourceFirst,
416 TargetFirst,
417}
418
419impl ReferenceExecutionDirection {
420 fn explain_name(self) -> &'static str {
421 match self {
422 Self::SourceFirst => "source_first",
423 Self::TargetFirst => "target_first",
424 }
425 }
426}
427
428#[derive(Debug, Clone, PartialEq, Eq)]
429struct ReferenceEdgeExecution {
430 edge_index: usize,
431 strategy: ReferenceLookupStrategy,
432 storage_mode: ReferenceStorageMode,
433 direction: ReferenceExecutionDirection,
434 distinct_keys: usize,
435 projected_columns: usize,
436 reverse_source_index_eligible: bool,
437 target_predicate_pushdown: bool,
438 left_to_inner: bool,
439 rejected_distinct_keys: usize,
440}
441
442#[derive(Debug, Clone, Default, PartialEq, Eq)]
443pub struct ReferenceExpandMetrics {
444 paths_planned: usize,
445 paths_executed: usize,
446 source_rows: usize,
447 null_source_keys: usize,
448 distinct_keys: usize,
449 repeated_keys_eliminated: usize,
450 lookup_batches: usize,
451 lookup_hits: usize,
452 lookup_misses: usize,
453 direct_edges: usize,
454 index_nested_loop_edges: usize,
455 batch_edges: usize,
456 hash_edges: usize,
457 merge_edges: usize,
458 fallback_edges: usize,
459 target_first_edges: usize,
460 hot_edges: usize,
461 cold_edges: usize,
462 hybrid_edges: usize,
463 target_predicate_edges: usize,
464 left_to_inner_edges: usize,
465 target_predicate_keys_rejected: usize,
466 planner_left_join_edges: usize,
467 edge_executions: Vec<ReferenceEdgeExecution>,
468}
469
470#[allow(dead_code)] impl ReferenceExpandMetrics {
472 pub fn paths_planned(&self) -> usize {
473 self.paths_planned
474 }
475
476 pub fn paths_executed(&self) -> usize {
477 self.paths_executed
478 }
479
480 pub fn source_rows(&self) -> usize {
481 self.source_rows
482 }
483
484 pub fn null_source_keys(&self) -> usize {
485 self.null_source_keys
486 }
487
488 pub fn distinct_keys(&self) -> usize {
489 self.distinct_keys
490 }
491
492 pub fn repeated_keys_eliminated(&self) -> usize {
493 self.repeated_keys_eliminated
494 }
495
496 pub fn lookup_batches(&self) -> usize {
497 self.lookup_batches
498 }
499
500 pub fn lookup_hits(&self) -> usize {
501 self.lookup_hits
502 }
503
504 pub fn lookup_misses(&self) -> usize {
505 self.lookup_misses
506 }
507
508 pub fn direct_edges(&self) -> usize {
509 self.direct_edges
510 }
511
512 pub fn batch_edges(&self) -> usize {
513 self.batch_edges
514 }
515
516 pub fn index_nested_loop_edges(&self) -> usize {
517 self.index_nested_loop_edges
518 }
519
520 pub fn hash_edges(&self) -> usize {
521 self.hash_edges
522 }
523
524 pub fn merge_edges(&self) -> usize {
525 self.merge_edges
526 }
527
528 pub fn fallback_edges(&self) -> usize {
529 self.fallback_edges
530 }
531
532 pub fn target_first_edges(&self) -> usize {
533 self.target_first_edges
534 }
535
536 pub fn target_predicate_edges(&self) -> usize {
537 self.target_predicate_edges
538 }
539
540 pub fn left_to_inner_edges(&self) -> usize {
541 self.left_to_inner_edges
542 }
543
544 pub fn target_predicate_keys_rejected(&self) -> usize {
545 self.target_predicate_keys_rejected
546 }
547
548 pub fn planner_left_join_edges(&self) -> usize {
549 self.planner_left_join_edges
550 }
551
552 pub fn hot_edges(&self) -> usize {
553 self.hot_edges
554 }
555
556 pub fn edge_projected_columns(&self, index: usize) -> Option<usize> {
557 self.edge_executions
558 .get(index)
559 .map(|edge| edge.projected_columns)
560 }
561
562 pub fn edge_reverse_source_index_eligible(&self, index: usize) -> Option<bool> {
563 self.edge_executions
564 .get(index)
565 .map(|edge| edge.reverse_source_index_eligible)
566 }
567
568 fn instrumentation(&self) -> radixdb_storage::instrumentation::NavigationQueryCounters {
569 radixdb_storage::instrumentation::NavigationQueryCounters {
570 paths_planned: self.paths_planned as u64,
571 paths_executed: self.paths_executed as u64,
572 source_rows: self.source_rows as u64,
573 distinct_source_keys: self.distinct_keys as u64,
574 repeated_keys_eliminated: self.repeated_keys_eliminated as u64,
575 lookup_batches: self.lookup_batches as u64,
576 lookup_hits: self.lookup_hits as u64,
577 lookup_misses: self.lookup_misses as u64,
578 direct_edges: self.direct_edges as u64,
579 index_nested_loop_edges: self.index_nested_loop_edges as u64,
580 batch_edges: self.batch_edges as u64,
581 hash_edges: self.hash_edges as u64,
582 merge_edges: self.merge_edges as u64,
583 fallback_edges: self.fallback_edges as u64,
584 planner_left_join_edges: self.planner_left_join_edges as u64,
585 }
586 }
587}
588
589#[derive(Debug, Clone, PartialEq)]
590struct ReferenceGraphExecutionPlan {
591 source_select: SelectStatement,
592 augmented_columns: Vec<String>,
593 source_column_positions: Vec<Option<usize>>,
594 rewritten_where: Option<Expression>,
595 rewritten_projection: Vec<Expression>,
596 output_columns: Vec<String>,
597 edge_key_positions: Vec<usize>,
598 edge_key_types: Vec<DataType>,
599 hidden_path_positions: Vec<usize>,
600 target_predicates: Vec<Vec<Expression>>,
601 table_alias: Option<String>,
602 limit: Option<Box<Expression>>,
603 offset: Option<Box<Expression>>,
604 aggregation_select: Option<SelectStatement>,
605}
606
607const MAX_NAVIGATION_STEPS: usize = 8;
608pub const MAX_NAVIGATION_PATHS: usize = 256;
609const MAX_NAVIGATION_EDGES: usize = 512;
610
611impl ReferenceExpandPlan {
612 pub fn build(paths: Vec<NavigationExpr>) -> Result<Self> {
613 if paths.len() > MAX_NAVIGATION_PATHS {
614 return Err(Error::navigation(
615 NavigationErrorCode::UnsupportedReferenceShape,
616 format!(
617 "statement contains {} navigable paths; maximum is {MAX_NAVIGATION_PATHS}",
618 paths.len()
619 ),
620 ));
621 }
622 let schema_generation = paths
623 .first()
624 .map_or(0, |path| path.root_relation().table().schema_generation());
625 let schema_scope_id = paths
626 .first()
627 .map_or(0, |path| path.root_relation().table().scope_id());
628 let mut plan = Self {
629 schema_scope_id,
630 schema_generation,
631 paths: Vec::new(),
632 edges: Vec::new(),
633 };
634 let mut path_index = FxHashMap::default();
635 let mut edge_index = FxHashMap::default();
636
637 for path in paths {
638 if path.root_relation().table().schema_generation() != schema_generation {
639 return Err(Error::navigation(
640 NavigationErrorCode::SchemaChanged,
641 "bound navigation paths span more than one schema generation",
642 ));
643 }
644
645 if let Some(&index) = path_index.get(path.identity()) {
646 let existing: &mut ReferenceExpandPath = &mut plan.paths[index];
647 if !existing.display_paths.contains(&path.display_path) {
648 existing.display_paths.push(path.display_path);
649 }
650 continue;
651 }
652
653 let mut edge_indices = Vec::with_capacity(path.steps.len());
654 for step_index in 0..path.steps.len() {
655 let identity = ReferenceExpandEdgeIdentity {
656 root: path.identity.root.clone(),
657 steps: path.identity.steps[..=step_index].to_vec(),
658 };
659 let required = if step_index + 1 < path.steps.len() {
660 path.steps[step_index + 1].source_column().clone()
661 } else {
662 path.terminal_column().clone()
663 };
664
665 let index = if let Some(&index) = edge_index.get(&identity) {
666 let edge: &mut ReferenceExpandEdge = &mut plan.edges[index];
667 if !edge.required_columns.contains(&required) {
668 edge.required_columns.push(required);
669 }
670 index
671 } else {
672 let step = &path.steps[step_index];
673 if plan.edges.len() == MAX_NAVIGATION_EDGES {
674 return Err(Error::navigation(
675 NavigationErrorCode::UnsupportedReferenceShape,
676 format!(
677 "navigation graph exceeds the {MAX_NAVIGATION_EDGES}-edge compile limit"
678 ),
679 ));
680 }
681 let index = plan.edges.len();
682 plan.edges.push(ReferenceExpandEdge {
683 identity: identity.clone(),
684 source_column: step.source_column().clone(),
685 target_key_column: step.target_key_column().clone(),
686 target_key: step.target_key(),
687 required_columns: vec![required],
688 semantics: ReferenceSemantics::Left,
689 integrity_check: ReferenceIntegrityCheck::Required,
690 });
691 edge_index.insert(identity, index);
692 index
693 };
694 edge_indices.push(index);
695 }
696
697 let identity = path.identity.clone();
698 let index = plan.paths.len();
699 plan.paths.push(ReferenceExpandPath {
700 identity: identity.clone(),
701 edge_indices,
702 terminal_type: path.terminal_type,
703 nullable: path.nullable,
704 display_paths: vec![path.display_path],
705 });
706 path_index.insert(identity, index);
707 }
708
709 Ok(plan)
710 }
711
712 pub fn schema_generation(&self) -> u64 {
713 self.schema_generation
714 }
715
716 pub fn schema_scope_id(&self) -> u64 {
717 self.schema_scope_id
718 }
719
720 #[allow(dead_code)] pub fn paths(&self) -> &[ReferenceExpandPath] {
722 &self.paths
723 }
724
725 #[allow(dead_code)] pub fn edges(&self) -> &[ReferenceExpandEdge] {
727 &self.edges
728 }
729
730 pub fn validate(&self, engine: &dyn Engine) -> Result<()> {
731 for edge in &self.edges {
732 engine.validate_schema_table_id(edge.identity.root.table())?;
733 engine.validate_schema_table_id(edge.source_column.table())?;
734 engine.validate_schema_table_id(edge.target_key_column.table())?;
735 for required in &edge.required_columns {
736 engine.validate_schema_table_id(required.table())?;
737 }
738 }
739 Ok(())
740 }
741
742 pub fn explain_lines(&self, engine: &dyn Engine) -> Result<Vec<String>> {
743 self.explain_lines_with_metrics(engine, None)
744 }
745
746 pub fn explain_lines_with_metrics(
747 &self,
748 engine: &dyn Engine,
749 metrics: Option<&ReferenceExpandMetrics>,
750 ) -> Result<Vec<String>> {
751 self.validate(engine)?;
752 let mut lines = vec!["Reference Navigation".to_string()];
753 lines.push(format!(" Schema Generation: {}", self.schema_generation));
754 lines.push(" Semantics: LEFT".to_string());
755 lines.push(" Snapshot: statement".to_string());
756 lines.push(" Authorization: same_as_explicit_left_join".to_string());
757 if let Some(metrics) = metrics {
758 if metrics.planner_left_join_edges > 0 {
759 lines.push(format!(
760 " Counters: paths_planned={}, paths_executed={}, delegated_reference_edges={}",
761 metrics.paths_planned, metrics.paths_executed, metrics.planner_left_join_edges
762 ));
763 lines.push(format!(
764 " Actual Work: delegated_to_join_executor, reference_edges={}",
765 metrics.planner_left_join_edges
766 ));
767 } else {
768 lines.push(format!(
769 " Counters: paths_planned={}, paths_executed={}, source_rows={}, distinct_source_keys={}, repeated_keys_eliminated={}, lookup_batches={}, target_lookup_hits={}, target_lookup_misses={}",
770 metrics.paths_planned,
771 metrics.paths_executed,
772 metrics.source_rows,
773 metrics.distinct_keys,
774 metrics.repeated_keys_eliminated,
775 metrics.lookup_batches,
776 metrics.lookup_hits,
777 metrics.lookup_misses
778 ));
779 lines.push(format!(
780 " Strategies: direct={}, index_nested_loop={}, batch={}, hash={}, merge={}, fallback={}",
781 metrics.direct_edges,
782 metrics.index_nested_loop_edges,
783 metrics.batch_edges,
784 metrics.hash_edges,
785 metrics.merge_edges,
786 metrics.fallback_edges
787 ));
788 lines.push(format!(
789 " Actual Work: source_rows={}, distinct_keys={}, lookup_batches={}, hits={}",
790 metrics.source_rows,
791 metrics.distinct_keys,
792 metrics.lookup_batches,
793 metrics.lookup_hits
794 ));
795 lines.push(format!(
796 " Predicate Rewrite: target_edges={}, left_to_inner_edges={}, rejected_distinct_keys={}",
797 metrics.target_predicate_edges,
798 metrics.left_to_inner_edges,
799 metrics.target_predicate_keys_rejected
800 ));
801 }
802 } else {
803 lines.push(
804 " Physical Strategy: adaptive_unique_lookup (direct_unique_lookup | index_nested_loop | unique_batch_lookup | target_hash_scan)"
805 .to_string(),
806 );
807 lines.push(" Merge Join: disabled_without_ordering_certificate".to_string());
808 lines.push(" Counters: available_with_explain_analyze".to_string());
809 }
810 for (index, edge) in self.edges.iter().enumerate() {
811 let source_name = column_name(engine, edge.source_column())?;
812 let target_name = column_name(engine, edge.target_key_column())?;
813 let required = edge
814 .required_columns()
815 .iter()
816 .map(|column| column_name(engine, column))
817 .collect::<Result<Vec<_>>>()?;
818 lines.push(format!(
819 " Edge {}: {}.{} -> {}.{}",
820 index + 1,
821 edge.source_column().table().table_name(),
822 source_name,
823 edge.target_key_column().table().table_name(),
824 target_name
825 ));
826 lines.push(format!(" Required Columns: {}", required.join(", ")));
827 lines.push(" Integrity Check: enabled".to_string());
828 if let Some(execution) = metrics.and_then(|metrics| {
829 metrics
830 .edge_executions
831 .iter()
832 .find(|execution| execution.edge_index == index)
833 }) {
834 lines.push(format!(
835 " Actual Strategy: {}",
836 execution.strategy.explain_name()
837 ));
838 lines.push(format!(
839 " Storage Mode: {}",
840 execution.storage_mode.explain_name()
841 ));
842 lines.push(format!(
843 " Physical Direction: {}",
844 execution.direction.explain_name()
845 ));
846 lines.push(format!(" Distinct Keys: {}", execution.distinct_keys));
847 lines.push(format!(
848 " Target Projection Columns: {}",
849 execution.projected_columns
850 ));
851 lines.push(format!(
852 " Reverse Source Index: {}",
853 if execution.reverse_source_index_eligible {
854 "eligible_but_not_used_without_target_predicate"
855 } else {
856 "not_eligible"
857 }
858 ));
859 lines.push(format!(
860 " Target Predicate Pushdown: {}",
861 if execution.target_predicate_pushdown {
862 "enabled"
863 } else {
864 "disabled"
865 }
866 ));
867 lines.push(format!(
868 " LEFT-to-INNER: {}",
869 if execution.left_to_inner {
870 "proven_null_rejecting"
871 } else {
872 "disabled"
873 }
874 ));
875 if execution.target_predicate_pushdown {
876 lines.push(format!(
877 " Rejected Distinct Keys: {}",
878 execution.rejected_distinct_keys
879 ));
880 }
881 } else if metrics.is_some_and(|metrics| metrics.planner_left_join_edges > 0) {
882 lines.push(" Actual Strategy: planner_left_join".to_string());
883 lines.push(" Storage Mode: delegated_to_join_executor".to_string());
884 }
885 }
886 for (index, path) in self.paths.iter().enumerate() {
887 lines.push(format!(" Path {}: {}", index + 1, path.display_paths()[0]));
888 lines.push(format!(
889 " Steps: {}",
890 path.edge_indices
891 .iter()
892 .map(|edge| (edge + 1).to_string())
893 .collect::<Vec<_>>()
894 .join(" -> ")
895 ));
896 }
897 Ok(lines)
898 }
899
900 fn path_index_for_display(&self, display: &str) -> Option<usize> {
901 self.paths
902 .iter()
903 .position(|path| path.display_paths.iter().any(|item| item == display))
904 }
905
906 fn prepare_graph_execution(
907 &self,
908 select: &SelectStatement,
909 engine: &dyn Engine,
910 ) -> Result<ReferenceGraphExecutionPlan> {
911 let navigable_where = select
912 .where_clause
913 .as_deref()
914 .filter(|expression| expression_contains_bound_navigation(expression, self));
915 let graph_aggregation = self.can_execute_graph_aggregation(select);
916 if !graph_aggregation
917 && (select.distinct
918 || !select.distinct_on.is_empty()
919 || !select.group_by.columns.is_empty()
920 || !matches!(select.group_by.modifier, GroupByModifier::None)
921 || select.having.is_some()
922 || !select.window_defs.is_empty()
923 || !select.set_operations.is_empty()
924 || select
925 .columns
926 .iter()
927 .any(crate::utils::expression_contains_aggregate))
928 {
929 return Err(Error::NotSupported(
930 "navigable predicates in aggregate, DISTINCT, window, or set contexts require NR-10"
931 .to_string(),
932 ));
933 }
934 if navigable_where.is_some_and(expression_contains_subquery) {
935 return Err(Error::NotSupported(
936 "subqueries combined with navigable predicates require NR-10".to_string(),
937 ));
938 }
939
940 let table = single_root_table_source(select)?;
941 let root = self
942 .paths
943 .first()
944 .ok_or_else(|| Error::internal("ReferenceExpand plan has no paths"))?
945 .identity
946 .root
947 .table();
948 if !table.name.value().eq_ignore_ascii_case(root.table_name()) {
949 return Err(Error::navigation(
950 NavigationErrorCode::SchemaChanged,
951 "navigation root no longer matches the physical source table",
952 ));
953 }
954 let schema = engine.get_table_schema(root.table_name())?;
955 let source_columns = schema.column_names_owned().to_vec();
956 let visible_root = table
957 .alias
958 .as_ref()
959 .unwrap_or(&table.name)
960 .value()
961 .to_string();
962 let table_alias = table.alias.as_ref().map(|alias| alias.value().to_string());
963 let projection_aliases: FxHashSet<String> = select
964 .columns
965 .iter()
966 .filter_map(|expression| match expression {
967 Expression::Aliased(aliased) => Some(aliased.alias.value_lower().to_string()),
968 _ => None,
969 })
970 .collect();
971 for order in &select.order_by {
972 if matches!(order.expression, Expression::IntegerLiteral(_))
973 || matches!(
974 &order.expression,
975 Expression::Identifier(identifier)
976 if projection_aliases.contains(identifier.value_lower())
977 )
978 {
979 return Err(Error::NotSupported(
980 "ORDER BY projection alias/position with a navigable predicate requires NR-10"
981 .to_string(),
982 ));
983 }
984 }
985
986 let mut outside_where_and_projection = select.clone();
989 outside_where_and_projection.columns.clear();
990 outside_where_and_projection.where_clause = None;
991 if graph_aggregation {
992 outside_where_and_projection.group_by = GroupByClause::default();
993 outside_where_and_projection.having = None;
994 }
995 let mut unsupported_path = None;
996 radixdb_sql::ast::walk_select_tree(&outside_where_and_projection, &mut |expression| {
997 if unsupported_path.is_none() {
998 if let Expression::QualifiedIdentifier(path) = expression {
999 let display = path.to_string();
1000 if self.path_index_for_display(&display).is_some() {
1001 unsupported_path = Some(display);
1002 }
1003 }
1004 }
1005 });
1006 if let Some(path) = unsupported_path {
1007 return Err(Error::NotSupported(format!(
1008 "navigable reference path '{path}' outside projection/WHERE requires NR-10"
1009 )));
1010 }
1011
1012 let mut edge_key_names = Vec::with_capacity(self.edges.len());
1013 let mut occupied: FxHashSet<String> = source_columns
1014 .iter()
1015 .map(|column| column.to_lowercase())
1016 .collect();
1017 for edge_index in 0..self.edges.len() {
1018 let mut suffix = 0usize;
1019 let hidden = loop {
1020 let candidate = if suffix == 0 {
1021 format!(
1022 "__radix_reference_edge_{}_{}",
1023 self.schema_scope_id, edge_index
1024 )
1025 } else {
1026 format!(
1027 "__radix_reference_edge_{}_{}_{}",
1028 self.schema_scope_id, edge_index, suffix
1029 )
1030 };
1031 if occupied.insert(candidate.to_lowercase()) {
1032 break candidate;
1033 }
1034 suffix = suffix.saturating_add(1);
1035 };
1036 edge_key_names.push(hidden);
1037 }
1038 let mut hidden_names = Vec::with_capacity(self.paths.len());
1039 for path_index in 0..self.paths.len() {
1040 let mut suffix = 0usize;
1041 let hidden = loop {
1042 let candidate = if suffix == 0 {
1043 format!("__radix_reference_{}_{}", self.schema_scope_id, path_index)
1044 } else {
1045 format!(
1046 "__radix_reference_{}_{}_{}",
1047 self.schema_scope_id, path_index, suffix
1048 )
1049 };
1050 if occupied.insert(candidate.to_lowercase()) {
1051 break candidate;
1052 }
1053 suffix = suffix.saturating_add(1);
1054 };
1055 hidden_names.push(hidden);
1056 }
1057 let rewritten_where = navigable_where
1058 .map(|where_clause| {
1059 rewrite_navigation_expression(where_clause, self, &hidden_names, &visible_root)
1060 })
1061 .transpose()?;
1062
1063 let mut rewritten_projection = Vec::new();
1064 let mut output_columns = Vec::new();
1065 for (index, expression) in select.columns.iter().enumerate() {
1066 match expression {
1067 Expression::Star(_) => {
1068 append_root_projection(
1069 &mut rewritten_projection,
1070 &mut output_columns,
1071 &source_columns,
1072 &select.token,
1073 );
1074 }
1075 Expression::QualifiedStar(star)
1076 if star.qualifier.eq_ignore_ascii_case(&visible_root) =>
1077 {
1078 append_root_projection(
1079 &mut rewritten_projection,
1080 &mut output_columns,
1081 &source_columns,
1082 &select.token,
1083 );
1084 }
1085 Expression::QualifiedStar(_) => {
1086 return Err(Error::NotSupported(
1087 "qualified star outside the navigation root requires NR-10".to_string(),
1088 ));
1089 }
1090 _ => {
1091 if !graph_aggregation
1092 && expression_contains_bound_navigation(expression, self)
1093 && !is_direct_navigation_projection(expression, self)
1094 {
1095 return Err(Error::NotSupported(
1096 "navigable references inside projection expressions require NR-10"
1097 .to_string(),
1098 ));
1099 }
1100 rewritten_projection.push(rewrite_navigation_expression(
1101 expression,
1102 self,
1103 &hidden_names,
1104 &visible_root,
1105 )?);
1106 let output_name = match expression {
1107 Expression::QualifiedIdentifier(path)
1108 if self.path_index_for_display(&path.to_string()).is_some() =>
1109 {
1110 path.to_string()
1111 }
1112 _ => reference_output_name(expression, index),
1113 };
1114 output_columns.push(output_name);
1115 }
1116 }
1117 }
1118
1119 let conjuncts = navigable_where
1120 .map(flatten_and_predicates)
1121 .unwrap_or_default();
1122 let source_predicates = conjuncts
1123 .iter()
1124 .filter(|predicate| !expression_contains_bound_navigation(predicate, self))
1125 .cloned()
1126 .collect();
1127 let mut target_predicates = vec![Vec::new(); self.edges.len()];
1128 for predicate in &conjuncts {
1129 if let Some(edge_index) = null_rejecting_target_edge(predicate, self) {
1130 target_predicates[edge_index].push(rewrite_navigation_expression(
1131 predicate,
1132 self,
1133 &hidden_names,
1134 &visible_root,
1135 )?);
1136 }
1137 }
1138
1139 let edge_key_types = self
1140 .edges
1141 .iter()
1142 .map(|edge| {
1143 let schema = engine.get_table_schema(edge.source_column.table().table_name())?;
1144 schema
1145 .get_column(edge.source_column.ordinal())
1146 .map(|column| column.data_type)
1147 .ok_or_else(|| {
1148 Error::navigation(
1149 NavigationErrorCode::SchemaChanged,
1150 "navigation source column moved while preparing execution",
1151 )
1152 })
1153 })
1154 .collect::<Result<Vec<_>>>()?;
1155
1156 let aggregation_select = if graph_aggregation {
1157 let mut rewritten = select.clone();
1158 rewritten.columns = rewritten_projection.clone();
1159 rewritten.where_clause = None;
1160 for expression in &mut rewritten.group_by.columns {
1161 *expression =
1162 rewrite_navigation_expression(expression, self, &hidden_names, &visible_root)?;
1163 }
1164 if let GroupByModifier::GroupingSets(sets) = &mut rewritten.group_by.modifier {
1165 for set in sets {
1166 for expression in set {
1167 *expression = rewrite_navigation_expression(
1168 expression,
1169 self,
1170 &hidden_names,
1171 &visible_root,
1172 )?;
1173 }
1174 }
1175 }
1176 if let Some(having) = &mut rewritten.having {
1177 **having =
1178 rewrite_navigation_expression(having, self, &hidden_names, &visible_root)?;
1179 }
1180 Some(rewritten)
1181 } else {
1182 None
1183 };
1184
1185 let source_projection_ordinals = if graph_aggregation {
1186 graph_aggregation_source_ordinals(
1187 &source_columns,
1188 &visible_root,
1189 root.table_name(),
1190 &self.edges,
1191 aggregation_select
1192 .as_ref()
1193 .expect("graph aggregation owns its rewritten SELECT"),
1194 rewritten_where.as_ref(),
1195 )
1196 } else {
1197 (0..source_columns.len()).collect()
1198 };
1199 let projected_source_columns = source_projection_ordinals
1200 .iter()
1201 .map(|&ordinal| source_columns[ordinal].clone())
1202 .collect::<Vec<_>>();
1203 let mut source_column_positions = vec![None; source_columns.len()];
1204 for (position, &ordinal) in source_projection_ordinals.iter().enumerate() {
1205 source_column_positions[ordinal] = Some(position);
1206 }
1207 let mut augmented_columns = projected_source_columns.clone();
1208 let edge_key_positions = edge_key_names
1209 .iter()
1210 .map(|name| {
1211 let position = augmented_columns.len();
1212 augmented_columns.push(name.clone());
1213 position
1214 })
1215 .collect::<Vec<_>>();
1216 let hidden_path_positions = hidden_names
1217 .iter()
1218 .map(|name| {
1219 let position = augmented_columns.len();
1220 augmented_columns.push(name.clone());
1221 position
1222 })
1223 .collect::<Vec<_>>();
1224
1225 let mut source_select = select.clone();
1226 source_select.columns = projected_source_columns
1227 .iter()
1228 .map(|column| {
1229 Expression::Identifier(radixdb_sql::ast::Identifier::new(
1230 select.token.clone(),
1231 column.clone(),
1232 ))
1233 })
1234 .collect();
1235 if navigable_where.is_some() {
1236 source_select.where_clause =
1237 combine_predicates_with_and(source_predicates).map(Box::new);
1238 }
1239 source_select.limit = None;
1240 source_select.offset = None;
1241 if graph_aggregation {
1242 source_select.group_by = GroupByClause::default();
1243 source_select.having = None;
1244 source_select.order_by.clear();
1245 }
1246
1247 Ok(ReferenceGraphExecutionPlan {
1248 source_select,
1249 augmented_columns,
1250 source_column_positions,
1251 rewritten_where,
1252 rewritten_projection,
1253 output_columns,
1254 edge_key_positions,
1255 edge_key_types,
1256 hidden_path_positions,
1257 target_predicates,
1258 table_alias,
1259 limit: select.limit.clone(),
1260 offset: select.offset.clone(),
1261 aggregation_select,
1262 })
1263 }
1264
1265 #[doc(hidden)]
1268 pub fn graph_source_projection_len(
1269 &self,
1270 select: &SelectStatement,
1271 engine: &dyn Engine,
1272 ) -> Result<usize> {
1273 self.prepare_graph_execution(select, engine)
1274 .map(|graph| graph.source_select.columns.len())
1275 }
1276
1277 fn can_execute_graph_aggregation(&self, select: &SelectStatement) -> bool {
1278 let has_aggregation = !select.group_by.columns.is_empty()
1279 || !matches!(select.group_by.modifier, GroupByModifier::None)
1280 || select.having.is_some()
1281 || select
1282 .columns
1283 .iter()
1284 .any(crate::utils::expression_contains_aggregate);
1285 if !has_aggregation
1286 || select.with.is_some()
1287 || !select.set_operations.is_empty()
1288 || !matches!(
1289 select.table_expr.as_deref(),
1290 Some(Expression::TableSource(_))
1291 )
1292 || select.distinct
1293 || !select.distinct_on.is_empty()
1294 || !select.window_defs.is_empty()
1295 || !select.order_by.is_empty()
1296 || select.limit.is_some()
1297 || select.offset.is_some()
1298 {
1299 return false;
1300 }
1301
1302 !select.columns.iter().any(expression_contains_subquery)
1303 && !select
1304 .group_by
1305 .columns
1306 .iter()
1307 .any(expression_contains_subquery)
1308 && !matches!(
1309 &select.group_by.modifier,
1310 GroupByModifier::GroupingSets(sets)
1311 if sets.iter().flatten().any(expression_contains_subquery)
1312 )
1313 && !select
1314 .having
1315 .as_deref()
1316 .is_some_and(expression_contains_subquery)
1317 && !select
1318 .where_clause
1319 .as_deref()
1320 .is_some_and(expression_contains_subquery)
1321 }
1322
1323 fn requires_planner_left_join(&self, select: &SelectStatement) -> bool {
1324 if select.with.is_some()
1325 || !select.set_operations.is_empty()
1326 || !matches!(
1327 select.table_expr.as_deref(),
1328 Some(Expression::TableSource(_))
1329 )
1330 || select.distinct
1331 || !select.distinct_on.is_empty()
1332 || !select.group_by.columns.is_empty()
1333 || !matches!(select.group_by.modifier, GroupByModifier::None)
1334 || select.having.is_some()
1335 || !select.window_defs.is_empty()
1336 || select
1337 .columns
1338 .iter()
1339 .any(crate::utils::expression_contains_aggregate)
1340 || select.columns.iter().any(|expression| {
1341 expression_contains_bound_navigation(expression, self)
1342 && !is_direct_navigation_projection(expression, self)
1343 })
1344 || select.where_clause.as_deref().is_some_and(|expression| {
1345 expression_contains_subquery(expression)
1346 && expression_contains_bound_navigation(expression, self)
1347 })
1348 {
1349 return true;
1350 }
1351
1352 let navigation_aliases: FxHashSet<String> = select
1353 .columns
1354 .iter()
1355 .enumerate()
1356 .filter(|(_, expression)| expression_contains_bound_navigation(expression, self))
1357 .map(|(index, expression)| match expression {
1358 Expression::Aliased(aliased) => aliased.alias.value_lower().to_string(),
1359 _ => (index + 1).to_string(),
1360 })
1361 .collect();
1362 select.order_by.iter().any(|order| {
1363 expression_contains_bound_navigation(&order.expression, self)
1364 || matches!(
1365 &order.expression,
1366 Expression::Identifier(identifier)
1367 if navigation_aliases.contains(identifier.value_lower())
1368 )
1369 || matches!(
1370 &order.expression,
1371 Expression::IntegerLiteral(position)
1372 if position.value > 0
1373 && navigation_aliases.contains(&position.value.to_string())
1374 )
1375 })
1376 }
1377
1378 fn lower_to_planner_left_joins(
1379 &self,
1380 select: &SelectStatement,
1381 engine: &dyn Engine,
1382 ) -> Result<SelectStatement> {
1383 self.validate(engine)?;
1384 let mut lowered = select.clone();
1385 let needs_correlated_visibility = select_has_correlated_navigation(select, self);
1386 let source = lowered.table_expr.take().ok_or_else(|| {
1387 Error::navigation(
1388 NavigationErrorCode::UnsupportedReferenceShape,
1389 "navigable references require a physical FROM root",
1390 )
1391 })?;
1392 let aliases = reference_edge_aliases(source.as_ref(), self, &select.token);
1393 let parent_edges = reference_parent_edges(self)?;
1394 let mut relation_ordinal = 0u32;
1395 lowered.table_expr = Some(Box::new(attach_reference_edges_to_sources(
1396 *source,
1397 self,
1398 engine,
1399 &aliases,
1400 &parent_edges,
1401 &mut relation_ordinal,
1402 )?));
1403
1404 add_canonical_navigation_result_aliases(&mut lowered.columns, self)?;
1405 rewrite_select_navigation_current_scope(&mut lowered, self, engine, &aliases)?;
1406 if needs_correlated_visibility {
1407 retain_correlated_navigation_columns(&mut lowered, self, engine, &aliases)?;
1408 }
1409 Ok(lowered)
1410 }
1411}