1use crate::ast::{JoinQuery, JoinType, QueryExpr};
14use crate::sql_lowering::{effective_table_filter, effective_vector_filter};
15
16pub trait OptimizationPass: Send + Sync {
18 fn name(&self) -> &str;
20
21 fn apply(&self, query: QueryExpr) -> QueryExpr;
23
24 fn benefit(&self) -> u32;
26}
27
28pub struct QueryOptimizer {
30 passes: Vec<Box<dyn OptimizationPass>>,
32 cost_based: bool,
34}
35
36impl QueryOptimizer {
37 pub fn new() -> Self {
39 let passes: Vec<Box<dyn OptimizationPass>> = vec![
40 Box::new(PredicatePushdownPass),
41 Box::new(ProjectionPushdownPass),
42 Box::new(JoinReorderingPass),
43 Box::new(IndexSelectionPass),
44 Box::new(LimitPushdownPass),
45 ];
46
47 Self {
48 passes,
49 cost_based: true,
50 }
51 }
52
53 pub fn add_pass(&mut self, pass: Box<dyn OptimizationPass>) {
55 self.passes.push(pass);
56 self.passes.sort_by_key(|b| std::cmp::Reverse(b.benefit()));
58 }
59
60 pub fn optimize(&self, query: QueryExpr) -> (QueryExpr, Vec<String>) {
62 let mut optimized = query;
63 let mut applied_passes = Vec::new();
64
65 for pass in &self.passes {
66 let before = format!("{:?}", optimized);
67 optimized = pass.apply(optimized);
68 let after = format!("{:?}", optimized);
69
70 if before != after {
71 applied_passes.push(pass.name().to_string());
72 }
73 }
74
75 (optimized, applied_passes)
76 }
77
78 pub fn optimize_with_hints(&self, query: QueryExpr, hints: &OptimizationHints) -> QueryExpr {
80 let mut optimized = query;
81
82 for pass in &self.passes {
83 if hints.disabled_passes.contains(&pass.name().to_string()) {
85 continue;
86 }
87
88 optimized = pass.apply(optimized);
89 }
90
91 optimized
92 }
93}
94
95impl Default for QueryOptimizer {
96 fn default() -> Self {
97 Self::new()
98 }
99}
100
101#[derive(Debug, Clone, Default)]
103pub struct OptimizationHints {
104 pub disabled_passes: Vec<String>,
106 pub join_order: Option<Vec<String>>,
108 pub force_index: Option<String>,
110 pub no_parallel: bool,
112}
113
114struct PredicatePushdownPass;
120
121impl OptimizationPass for PredicatePushdownPass {
122 fn name(&self) -> &str {
123 "PredicatePushdown"
124 }
125
126 fn apply(&self, query: QueryExpr) -> QueryExpr {
127 match query {
128 QueryExpr::Join(jq) => self.optimize_join(jq),
129 other => other,
130 }
131 }
132
133 fn benefit(&self) -> u32 {
134 100 }
136}
137
138impl PredicatePushdownPass {
139 fn optimize_join(&self, query: JoinQuery) -> QueryExpr {
140 let left = self.apply(*query.left);
145 let right = self.apply(*query.right);
146
147 QueryExpr::Join(JoinQuery {
148 left: Box::new(left),
149 right: Box::new(right),
150 ..query
151 })
152 }
153}
154
155struct ProjectionPushdownPass;
157
158impl OptimizationPass for ProjectionPushdownPass {
159 fn name(&self) -> &str {
160 "ProjectionPushdown"
161 }
162
163 fn apply(&self, query: QueryExpr) -> QueryExpr {
164 match query {
165 QueryExpr::Join(jq) => {
166 let left = self.apply(*jq.left);
168 let right = self.apply(*jq.right);
169
170 QueryExpr::Join(JoinQuery {
171 left: Box::new(left),
172 right: Box::new(right),
173 ..jq
174 })
175 }
176 QueryExpr::Table(tq) => {
177 QueryExpr::Table(tq)
180 }
181 other => other,
182 }
183 }
184
185 fn benefit(&self) -> u32 {
186 80 }
188}
189
190struct JoinReorderingPass;
192
193impl OptimizationPass for JoinReorderingPass {
194 fn name(&self) -> &str {
195 "JoinReordering"
196 }
197
198 fn apply(&self, query: QueryExpr) -> QueryExpr {
199 match query {
200 QueryExpr::Join(jq) => {
201 self.optimize_join_order(jq)
204 }
205 other => other,
206 }
207 }
208
209 fn benefit(&self) -> u32 {
210 90 }
212}
213
214impl JoinReorderingPass {
215 fn optimize_join_order(&self, query: JoinQuery) -> QueryExpr {
216 let left_size = Self::estimate_size(&query.left);
218 let right_size = Self::estimate_size(&query.right);
219
220 if left_size > right_size && query.join_type == JoinType::Inner {
222 let JoinQuery {
224 left,
225 right,
226 join_type,
227 on,
228 filter,
229 order_by,
230 limit,
231 offset,
232 return_items,
233 return_,
234 } = query;
235 QueryExpr::Join(JoinQuery {
236 left: right,
237 right: left,
238 join_type,
239 on: swap_condition(on),
240 filter,
241 order_by,
242 limit,
243 offset,
244 return_items,
245 return_,
246 })
247 } else {
248 QueryExpr::Join(query)
249 }
250 }
251
252 fn estimate_size(query: &QueryExpr) -> f64 {
253 match query {
254 QueryExpr::Table(tq) => {
255 let base = 1000.0;
256 if effective_table_filter(tq).is_some() {
257 base * 0.1
258 } else if tq.limit.is_some() {
259 tq.limit.unwrap() as f64
260 } else {
261 base
262 }
263 }
264 QueryExpr::Graph(_) => 100.0,
265 QueryExpr::Join(jq) => {
266 Self::estimate_size(&jq.left) * Self::estimate_size(&jq.right) * 0.1
267 }
268 QueryExpr::Path(_) => 10.0,
269 QueryExpr::Vector(vq) => {
270 if effective_vector_filter(vq).is_some() {
272 (vq.k as f64).min(100.0)
273 } else {
274 vq.k as f64
275 }
276 }
277 QueryExpr::Hybrid(hq) => {
278 let structured_size = Self::estimate_size(&hq.structured);
280 let vector_size = hq.vector.k as f64;
281 let base = structured_size.min(vector_size);
283 hq.limit.map(|l| base.min(l as f64)).unwrap_or(base)
284 }
285 QueryExpr::Insert(_)
287 | QueryExpr::Update(_)
288 | QueryExpr::Delete(_)
289 | QueryExpr::CreateTable(_)
290 | QueryExpr::CreateCollection(_)
291 | QueryExpr::CreateVector(_)
292 | QueryExpr::DropTable(_)
293 | QueryExpr::DropGraph(_)
294 | QueryExpr::DropVector(_)
295 | QueryExpr::DropDocument(_)
296 | QueryExpr::DropKv(_)
297 | QueryExpr::DropCollection(_)
298 | QueryExpr::Truncate(_)
299 | QueryExpr::AlterTable(_)
300 | QueryExpr::CreateVcsRef(_)
301 | QueryExpr::DropVcsRef(_)
302 | QueryExpr::GraphCommand(_)
303 | QueryExpr::SearchCommand(_)
304 | QueryExpr::CreateIndex(_)
305 | QueryExpr::DropIndex(_)
306 | QueryExpr::ProbabilisticCommand(_)
307 | QueryExpr::Ask(_)
308 | QueryExpr::SetConfig { .. }
309 | QueryExpr::ShowConfig { .. }
310 | QueryExpr::SetSecret { .. }
311 | QueryExpr::DeleteSecret { .. }
312 | QueryExpr::ShowSecrets { .. }
313 | QueryExpr::SetTenant(_)
314 | QueryExpr::ShowTenant
315 | QueryExpr::CreateTimeSeries(_)
316 | QueryExpr::CreateMetric(_)
317 | QueryExpr::AlterMetric(_)
318 | QueryExpr::CreateSlo(_)
319 | QueryExpr::DropTimeSeries(_)
320 | QueryExpr::CreateQueue(_)
321 | QueryExpr::AlterQueue(_)
322 | QueryExpr::DropQueue(_)
323 | QueryExpr::QueueSelect(_)
324 | QueryExpr::QueueCommand(_)
325 | QueryExpr::KvCommand(_)
326 | QueryExpr::ConfigCommand(_)
327 | QueryExpr::CreateTree(_)
328 | QueryExpr::DropTree(_)
329 | QueryExpr::TreeCommand(_)
330 | QueryExpr::ExplainAlter(_)
331 | QueryExpr::TransactionControl(_)
332 | QueryExpr::MaintenanceCommand(_)
333 | QueryExpr::VcsCommand(_)
334 | QueryExpr::CreateSchema(_)
335 | QueryExpr::DropSchema(_)
336 | QueryExpr::CreateSequence(_)
337 | QueryExpr::DropSequence(_)
338 | QueryExpr::CopyFrom(_)
339 | QueryExpr::CreateView(_)
340 | QueryExpr::DropView(_)
341 | QueryExpr::RefreshMaterializedView(_)
342 | QueryExpr::CreatePolicy(_)
343 | QueryExpr::DropPolicy(_)
344 | QueryExpr::CreateServer(_)
345 | QueryExpr::DropServer(_)
346 | QueryExpr::CreateForeignTable(_)
347 | QueryExpr::DropForeignTable(_)
348 | QueryExpr::Grant(_)
349 | QueryExpr::Revoke(_)
350 | QueryExpr::AlterUser(_)
351 | QueryExpr::CreateUser(_)
352 | QueryExpr::CreateIamPolicy { .. }
353 | QueryExpr::DropIamPolicy { .. }
354 | QueryExpr::AttachPolicy { .. }
355 | QueryExpr::DetachPolicy { .. }
356 | QueryExpr::ShowPolicies { .. }
357 | QueryExpr::ShowEffectivePermissions { .. }
358 | QueryExpr::RankOf(_)
359 | QueryExpr::ApproxRankOf(_)
360 | QueryExpr::RankRange(_)
361 | QueryExpr::SimulatePolicy { .. }
362 | QueryExpr::LintPolicy { .. }
363 | QueryExpr::MigratePolicyMode { .. }
364 | QueryExpr::CreateMigration(_)
365 | QueryExpr::ApplyMigration(_)
366 | QueryExpr::RollbackMigration(_)
367 | QueryExpr::ExplainMigration(_)
368 | QueryExpr::EventsBackfill(_)
369 | QueryExpr::EventsBackfillStatus { .. } => 1.0,
370 }
371 }
372}
373
374struct IndexSelectionPass;
387
388impl OptimizationPass for IndexSelectionPass {
389 fn name(&self) -> &str {
390 "IndexSelection"
391 }
392
393 fn apply(&self, query: QueryExpr) -> QueryExpr {
394 match query {
395 QueryExpr::Table(mut tq) => {
396 if let Some(filter) = effective_table_filter(&tq).as_ref() {
397 if let Some(hint) = Self::analyze_filter(filter) {
398 let expand = tq.expand.get_or_insert_with(Default::default);
400 expand.index_hint = Some(hint);
401 }
402 }
403 QueryExpr::Table(tq)
404 }
405 other => other,
406 }
407 }
408
409 fn benefit(&self) -> u32 {
410 70
411 }
412}
413
414impl IndexSelectionPass {
415 fn analyze_filter(filter: &crate::ast::Filter) -> Option<IndexHint> {
417 match filter {
418 crate::ast::Filter::Compare { field, op, .. } if *op == crate::ast::CompareOp::Eq => {
420 let col = Self::field_name(field);
421 Some(IndexHint {
422 method: IndexHintMethod::Hash,
423 column: col,
424 })
425 }
426 crate::ast::Filter::Compare {
428 field,
429 op:
430 crate::ast::CompareOp::Lt
431 | crate::ast::CompareOp::Le
432 | crate::ast::CompareOp::Gt
433 | crate::ast::CompareOp::Ge,
434 ..
435 } => {
436 let col = Self::field_name(field);
437 Some(IndexHint {
438 method: IndexHintMethod::BTree,
439 column: col,
440 })
441 }
442 crate::ast::Filter::Between { field, .. } => {
444 let col = Self::field_name(field);
445 Some(IndexHint {
446 method: IndexHintMethod::BTree,
447 column: col,
448 })
449 }
450 crate::ast::Filter::In { field, values } if values.len() <= 10 => {
452 let col = Self::field_name(field);
453 Some(IndexHint {
454 method: IndexHintMethod::Bitmap,
455 column: col,
456 })
457 }
458 crate::ast::Filter::And(left, right) => {
460 Self::analyze_filter(left).or_else(|| Self::analyze_filter(right))
461 }
462 _ => None,
463 }
464 }
465
466 fn field_name(field: &crate::ast::FieldRef) -> String {
467 match field {
468 crate::ast::FieldRef::TableColumn { column, .. } => column.clone(),
469 crate::ast::FieldRef::NodeProperty { property, .. } => property.clone(),
470 crate::ast::FieldRef::EdgeProperty { property, .. } => property.clone(),
471 crate::ast::FieldRef::NodeId { alias } => {
472 format!("{}.id", alias)
473 }
474 }
475 }
476}
477
478pub use reddb_types::index_hint::{IndexHint, IndexHintMethod};
483
484struct LimitPushdownPass;
486
487impl OptimizationPass for LimitPushdownPass {
488 fn name(&self) -> &str {
489 "LimitPushdown"
490 }
491
492 fn apply(&self, query: QueryExpr) -> QueryExpr {
493 match query {
494 QueryExpr::Join(jq) => {
495 let left = self.apply(*jq.left);
497 let right = self.apply(*jq.right);
498
499 QueryExpr::Join(JoinQuery {
500 left: Box::new(left),
501 right: Box::new(right),
502 ..jq
503 })
504 }
505 other => other,
506 }
507 }
508
509 fn benefit(&self) -> u32 {
510 60
511 }
512}
513
514fn swap_condition(condition: crate::ast::JoinCondition) -> crate::ast::JoinCondition {
519 crate::ast::JoinCondition {
520 left_field: condition.right_field,
521 right_field: condition.left_field,
522 }
523}
524
525#[cfg(test)]
526mod tests {
527 use super::*;
528 use crate::ast::{
529 CompareOp, DistanceMetric, FieldRef, Filter, FusionStrategy, JoinCondition, Projection,
530 TableQuery,
531 };
532 use reddb_types::Value;
533
534 fn make_table_query(name: &str) -> QueryExpr {
535 QueryExpr::Table(TableQuery {
536 table: name.to_string(),
537 source: None,
538 alias: Some(name.to_string()),
539 select_items: Vec::new(),
540 columns: vec![Projection::All],
541 where_expr: None,
542 filter: None,
543 group_by_exprs: Vec::new(),
544 group_by: Vec::new(),
545 having_expr: None,
546 having: None,
547 order_by: vec![],
548 limit: None,
549 limit_param: None,
550 offset: None,
551 offset_param: None,
552 expand: None,
553 as_of: None,
554 sessionize: None,
555 distinct: false,
556 })
557 }
558
559 #[test]
560 fn test_optimizer_applies_passes() {
561 let optimizer = QueryOptimizer::new();
562 let query = make_table_query("hosts");
563
564 let (optimized, passes) = optimizer.optimize(query);
565 assert!(matches!(optimized, QueryExpr::Table(_)));
567 }
568
569 #[test]
570 fn test_join_reordering() {
571 let optimizer = QueryOptimizer::new();
572
573 let small = QueryExpr::Table(TableQuery {
574 table: "small".to_string(),
575 source: None,
576 alias: None,
577 select_items: Vec::new(),
578 columns: vec![Projection::All],
579 where_expr: None,
580 filter: None,
581 group_by_exprs: Vec::new(),
582 group_by: Vec::new(),
583 having_expr: None,
584 having: None,
585 order_by: vec![],
586 limit: Some(10), limit_param: None,
588 offset: None,
589 offset_param: None,
590 expand: None,
591 as_of: None,
592 sessionize: None,
593 distinct: false,
594 });
595
596 let large = QueryExpr::Table(TableQuery {
597 table: "large".to_string(),
598 source: None,
599 alias: None,
600 select_items: Vec::new(),
601 columns: vec![Projection::All],
602 where_expr: None,
603 filter: None,
604 group_by_exprs: Vec::new(),
605 group_by: Vec::new(),
606 having_expr: None,
607 having: None,
608 order_by: vec![],
609 limit: None, limit_param: None,
611 offset: None,
612 offset_param: None,
613 expand: None,
614 as_of: None,
615 sessionize: None,
616 distinct: false,
617 });
618
619 let join = QueryExpr::Join(JoinQuery {
620 left: Box::new(large.clone()),
621 right: Box::new(small.clone()),
622 join_type: JoinType::Inner,
623 on: JoinCondition {
624 left_field: FieldRef::TableColumn {
625 table: "large".to_string(),
626 column: "id".to_string(),
627 },
628 right_field: FieldRef::TableColumn {
629 table: "small".to_string(),
630 column: "id".to_string(),
631 },
632 },
633 filter: None,
634 order_by: Vec::new(),
635 limit: None,
636 offset: None,
637 return_items: Vec::new(),
638 return_: Vec::new(),
639 });
640
641 let (optimized, passes) = optimizer.optimize(join);
642
643 assert!(passes.iter().any(|pass| pass == "JoinReordering"));
645 if let QueryExpr::Join(jq) = optimized {
646 if let QueryExpr::Table(left) = jq.left.as_ref() {
648 assert_eq!(left.table, "small");
649 }
650 assert!(matches!(
651 &jq.on.left_field,
652 FieldRef::TableColumn { table, column } if table == "small" && column == "id"
653 ));
654 }
655 }
656
657 #[test]
658 fn optimize_with_hints_can_disable_join_reordering() {
659 let optimizer = QueryOptimizer::new();
660
661 let large = make_table_query("large");
662 let mut small_table = TableQuery::new("small");
663 small_table.limit = Some(1);
664 let small = QueryExpr::Table(small_table);
665
666 let join = QueryExpr::Join(JoinQuery {
667 left: Box::new(large),
668 right: Box::new(small),
669 join_type: JoinType::Inner,
670 on: JoinCondition {
671 left_field: FieldRef::TableColumn {
672 table: "large".to_string(),
673 column: "id".to_string(),
674 },
675 right_field: FieldRef::TableColumn {
676 table: "small".to_string(),
677 column: "id".to_string(),
678 },
679 },
680 filter: None,
681 order_by: Vec::new(),
682 limit: None,
683 offset: None,
684 return_items: Vec::new(),
685 return_: Vec::new(),
686 });
687 let hints = OptimizationHints {
688 disabled_passes: vec!["JoinReordering".to_string()],
689 ..OptimizationHints::default()
690 };
691
692 let optimized = optimizer.optimize_with_hints(join, &hints);
693 let QueryExpr::Join(join) = optimized else {
694 panic!("expected join query");
695 };
696 let QueryExpr::Table(left) = join.left.as_ref() else {
697 panic!("expected table on left side");
698 };
699
700 assert_eq!(left.table, "large");
701 assert!(matches!(
702 &join.on.left_field,
703 FieldRef::TableColumn { table, column } if table == "large" && column == "id"
704 ));
705 }
706
707 #[test]
708 fn optimizer_sets_index_hint_on_table_filters() {
709 let optimizer = QueryOptimizer::new();
710 let mut table = TableQuery::new("hosts");
711 table.filter = Some(Filter::Compare {
712 field: FieldRef::column("", "host_id"),
713 op: CompareOp::Eq,
714 value: Value::Integer(7),
715 });
716
717 let (optimized, passes) = optimizer.optimize(QueryExpr::Table(table));
718 let QueryExpr::Table(table) = optimized else {
719 panic!("expected table query");
720 };
721 let hint = table
722 .expand
723 .and_then(|expand| expand.index_hint)
724 .expect("expected optimizer index hint");
725
726 assert!(passes.iter().any(|pass| pass == "IndexSelection"));
727 assert_eq!(hint.method, IndexHintMethod::Hash);
728 assert_eq!(hint.column, "host_id");
729 }
730
731 #[test]
732 fn index_selection_analyzes_supported_filter_shapes() {
733 let range = IndexSelectionPass::analyze_filter(&Filter::Between {
734 field: FieldRef::node_prop("n", "score"),
735 low: Value::Integer(1),
736 high: Value::Integer(9),
737 })
738 .expect("expected range hint");
739 assert_eq!(range.method, IndexHintMethod::BTree);
740 assert_eq!(range.column, "score");
741
742 let bitmap = IndexSelectionPass::analyze_filter(&Filter::In {
743 field: FieldRef::edge_prop("e", "kind"),
744 values: vec![Value::text("http"), Value::text("ssh")],
745 })
746 .expect("expected bitmap hint");
747 assert_eq!(bitmap.method, IndexHintMethod::Bitmap);
748 assert_eq!(bitmap.column, "kind");
749
750 assert!(IndexSelectionPass::analyze_filter(&Filter::In {
751 field: FieldRef::column("", "status"),
752 values: (0..11).map(Value::Integer).collect(),
753 })
754 .is_none());
755
756 let fallback_right = IndexSelectionPass::analyze_filter(&Filter::And(
757 Box::new(Filter::IsNull(FieldRef::column("", "deleted_at"))),
758 Box::new(Filter::Compare {
759 field: FieldRef::node_id("n"),
760 op: CompareOp::Eq,
761 value: Value::Integer(1),
762 }),
763 ))
764 .expect("expected right-side AND hint");
765 assert_eq!(fallback_right.method, IndexHintMethod::Hash);
766 assert_eq!(fallback_right.column, "n.id");
767 }
768}