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::SetKv { .. }
314 | QueryExpr::DeleteKv { .. }
315 | QueryExpr::SetTenant(_)
316 | QueryExpr::ShowTenant
317 | QueryExpr::CreateTimeSeries(_)
318 | QueryExpr::CreateMetric(_)
319 | QueryExpr::AlterMetric(_)
320 | QueryExpr::CreateSlo(_)
321 | QueryExpr::DropTimeSeries(_)
322 | QueryExpr::CreateQueue(_)
323 | QueryExpr::AlterQueue(_)
324 | QueryExpr::DropQueue(_)
325 | QueryExpr::QueueSelect(_)
326 | QueryExpr::QueueCommand(_)
327 | QueryExpr::KvCommand(_)
328 | QueryExpr::ConfigCommand(_)
329 | QueryExpr::CreateTree(_)
330 | QueryExpr::DropTree(_)
331 | QueryExpr::TreeCommand(_)
332 | QueryExpr::ExplainAlter(_)
333 | QueryExpr::TransactionControl(_)
334 | QueryExpr::MaintenanceCommand(_)
335 | QueryExpr::VcsCommand(_)
336 | QueryExpr::CreateSchema(_)
337 | QueryExpr::DropSchema(_)
338 | QueryExpr::CreateSequence(_)
339 | QueryExpr::DropSequence(_)
340 | QueryExpr::CopyFrom(_)
341 | QueryExpr::CreateView(_)
342 | QueryExpr::DropView(_)
343 | QueryExpr::RefreshMaterializedView(_)
344 | QueryExpr::CreatePolicy(_)
345 | QueryExpr::DropPolicy(_)
346 | QueryExpr::CreateServer(_)
347 | QueryExpr::DropServer(_)
348 | QueryExpr::CreateForeignTable(_)
349 | QueryExpr::DropForeignTable(_)
350 | QueryExpr::Grant(_)
351 | QueryExpr::Revoke(_)
352 | QueryExpr::AlterUser(_)
353 | QueryExpr::CreateUser(_)
354 | QueryExpr::CreateIamPolicy { .. }
355 | QueryExpr::DropIamPolicy { .. }
356 | QueryExpr::AttachPolicy { .. }
357 | QueryExpr::DetachPolicy { .. }
358 | QueryExpr::ShowPolicies { .. }
359 | QueryExpr::ShowEffectivePermissions { .. }
360 | QueryExpr::RankOf(_)
361 | QueryExpr::ApproxRankOf(_)
362 | QueryExpr::RankRange(_)
363 | QueryExpr::SimulatePolicy { .. }
364 | QueryExpr::LintPolicy { .. }
365 | QueryExpr::MigratePolicyMode { .. }
366 | QueryExpr::CreateMigration(_)
367 | QueryExpr::ApplyMigration(_)
368 | QueryExpr::RollbackMigration(_)
369 | QueryExpr::ExplainMigration(_)
370 | QueryExpr::EventsBackfill(_)
371 | QueryExpr::EventsBackfillStatus { .. } => 1.0,
372 }
373 }
374}
375
376struct IndexSelectionPass;
389
390impl OptimizationPass for IndexSelectionPass {
391 fn name(&self) -> &str {
392 "IndexSelection"
393 }
394
395 fn apply(&self, query: QueryExpr) -> QueryExpr {
396 match query {
397 QueryExpr::Table(mut tq) => {
398 if let Some(filter) = effective_table_filter(&tq).as_ref() {
399 if let Some(hint) = Self::analyze_filter(filter) {
400 let expand = tq.expand.get_or_insert_with(Default::default);
402 expand.index_hint = Some(hint);
403 }
404 }
405 QueryExpr::Table(tq)
406 }
407 other => other,
408 }
409 }
410
411 fn benefit(&self) -> u32 {
412 70
413 }
414}
415
416impl IndexSelectionPass {
417 fn analyze_filter(filter: &crate::ast::Filter) -> Option<IndexHint> {
419 match filter {
420 crate::ast::Filter::Compare { field, op, .. } if *op == crate::ast::CompareOp::Eq => {
422 let col = Self::field_name(field);
423 Some(IndexHint {
424 method: IndexHintMethod::Hash,
425 column: col,
426 })
427 }
428 crate::ast::Filter::Compare {
430 field,
431 op:
432 crate::ast::CompareOp::Lt
433 | crate::ast::CompareOp::Le
434 | crate::ast::CompareOp::Gt
435 | crate::ast::CompareOp::Ge,
436 ..
437 } => {
438 let col = Self::field_name(field);
439 Some(IndexHint {
440 method: IndexHintMethod::BTree,
441 column: col,
442 })
443 }
444 crate::ast::Filter::Between { field, .. } => {
446 let col = Self::field_name(field);
447 Some(IndexHint {
448 method: IndexHintMethod::BTree,
449 column: col,
450 })
451 }
452 crate::ast::Filter::In { field, values } if values.len() <= 10 => {
454 let col = Self::field_name(field);
455 Some(IndexHint {
456 method: IndexHintMethod::Bitmap,
457 column: col,
458 })
459 }
460 crate::ast::Filter::And(left, right) => {
462 Self::analyze_filter(left).or_else(|| Self::analyze_filter(right))
463 }
464 _ => None,
465 }
466 }
467
468 fn field_name(field: &crate::ast::FieldRef) -> String {
469 match field {
470 crate::ast::FieldRef::TableColumn { column, .. } => column.clone(),
471 crate::ast::FieldRef::NodeProperty { property, .. } => property.clone(),
472 crate::ast::FieldRef::EdgeProperty { property, .. } => property.clone(),
473 crate::ast::FieldRef::NodeId { alias } => {
474 format!("{}.id", alias)
475 }
476 }
477 }
478}
479
480pub use reddb_types::index_hint::{IndexHint, IndexHintMethod};
485
486struct LimitPushdownPass;
488
489impl OptimizationPass for LimitPushdownPass {
490 fn name(&self) -> &str {
491 "LimitPushdown"
492 }
493
494 fn apply(&self, query: QueryExpr) -> QueryExpr {
495 match query {
496 QueryExpr::Join(jq) => {
497 let left = self.apply(*jq.left);
499 let right = self.apply(*jq.right);
500
501 QueryExpr::Join(JoinQuery {
502 left: Box::new(left),
503 right: Box::new(right),
504 ..jq
505 })
506 }
507 other => other,
508 }
509 }
510
511 fn benefit(&self) -> u32 {
512 60
513 }
514}
515
516fn swap_condition(condition: crate::ast::JoinCondition) -> crate::ast::JoinCondition {
521 crate::ast::JoinCondition {
522 left_field: condition.right_field,
523 right_field: condition.left_field,
524 }
525}
526
527#[cfg(test)]
528mod tests {
529 use super::*;
530 use crate::ast::{
531 CompareOp, DistanceMetric, FieldRef, Filter, FusionStrategy, JoinCondition, Projection,
532 TableQuery,
533 };
534 use reddb_types::Value;
535
536 fn make_table_query(name: &str) -> QueryExpr {
537 QueryExpr::Table(TableQuery {
538 table: name.to_string(),
539 source: None,
540 alias: Some(name.to_string()),
541 select_items: Vec::new(),
542 columns: vec![Projection::All],
543 where_expr: None,
544 filter: None,
545 group_by_exprs: Vec::new(),
546 group_by: Vec::new(),
547 having_expr: None,
548 having: None,
549 order_by: vec![],
550 limit: None,
551 limit_param: None,
552 offset: None,
553 offset_param: None,
554 expand: None,
555 as_of: None,
556 sessionize: None,
557 distinct: false,
558 })
559 }
560
561 #[test]
562 fn test_optimizer_applies_passes() {
563 let optimizer = QueryOptimizer::new();
564 let query = make_table_query("hosts");
565
566 let (optimized, passes) = optimizer.optimize(query);
567 assert!(matches!(optimized, QueryExpr::Table(_)));
569 }
570
571 #[test]
572 fn test_join_reordering() {
573 let optimizer = QueryOptimizer::new();
574
575 let small = QueryExpr::Table(TableQuery {
576 table: "small".to_string(),
577 source: None,
578 alias: None,
579 select_items: Vec::new(),
580 columns: vec![Projection::All],
581 where_expr: None,
582 filter: None,
583 group_by_exprs: Vec::new(),
584 group_by: Vec::new(),
585 having_expr: None,
586 having: None,
587 order_by: vec![],
588 limit: Some(10), limit_param: None,
590 offset: None,
591 offset_param: None,
592 expand: None,
593 as_of: None,
594 sessionize: None,
595 distinct: false,
596 });
597
598 let large = QueryExpr::Table(TableQuery {
599 table: "large".to_string(),
600 source: None,
601 alias: None,
602 select_items: Vec::new(),
603 columns: vec![Projection::All],
604 where_expr: None,
605 filter: None,
606 group_by_exprs: Vec::new(),
607 group_by: Vec::new(),
608 having_expr: None,
609 having: None,
610 order_by: vec![],
611 limit: None, limit_param: None,
613 offset: None,
614 offset_param: None,
615 expand: None,
616 as_of: None,
617 sessionize: None,
618 distinct: false,
619 });
620
621 let join = QueryExpr::Join(JoinQuery {
622 left: Box::new(large.clone()),
623 right: Box::new(small.clone()),
624 join_type: JoinType::Inner,
625 on: JoinCondition {
626 left_field: FieldRef::TableColumn {
627 table: "large".to_string(),
628 column: "id".to_string(),
629 },
630 right_field: FieldRef::TableColumn {
631 table: "small".to_string(),
632 column: "id".to_string(),
633 },
634 },
635 filter: None,
636 order_by: Vec::new(),
637 limit: None,
638 offset: None,
639 return_items: Vec::new(),
640 return_: Vec::new(),
641 });
642
643 let (optimized, passes) = optimizer.optimize(join);
644
645 assert!(passes.iter().any(|pass| pass == "JoinReordering"));
647 if let QueryExpr::Join(jq) = optimized {
648 if let QueryExpr::Table(left) = jq.left.as_ref() {
650 assert_eq!(left.table, "small");
651 }
652 assert!(matches!(
653 &jq.on.left_field,
654 FieldRef::TableColumn { table, column } if table == "small" && column == "id"
655 ));
656 }
657 }
658
659 #[test]
660 fn optimize_with_hints_can_disable_join_reordering() {
661 let optimizer = QueryOptimizer::new();
662
663 let large = make_table_query("large");
664 let mut small_table = TableQuery::new("small");
665 small_table.limit = Some(1);
666 let small = QueryExpr::Table(small_table);
667
668 let join = QueryExpr::Join(JoinQuery {
669 left: Box::new(large),
670 right: Box::new(small),
671 join_type: JoinType::Inner,
672 on: JoinCondition {
673 left_field: FieldRef::TableColumn {
674 table: "large".to_string(),
675 column: "id".to_string(),
676 },
677 right_field: FieldRef::TableColumn {
678 table: "small".to_string(),
679 column: "id".to_string(),
680 },
681 },
682 filter: None,
683 order_by: Vec::new(),
684 limit: None,
685 offset: None,
686 return_items: Vec::new(),
687 return_: Vec::new(),
688 });
689 let hints = OptimizationHints {
690 disabled_passes: vec!["JoinReordering".to_string()],
691 ..OptimizationHints::default()
692 };
693
694 let optimized = optimizer.optimize_with_hints(join, &hints);
695 let QueryExpr::Join(join) = optimized else {
696 panic!("expected join query");
697 };
698 let QueryExpr::Table(left) = join.left.as_ref() else {
699 panic!("expected table on left side");
700 };
701
702 assert_eq!(left.table, "large");
703 assert!(matches!(
704 &join.on.left_field,
705 FieldRef::TableColumn { table, column } if table == "large" && column == "id"
706 ));
707 }
708
709 #[test]
710 fn optimizer_sets_index_hint_on_table_filters() {
711 let optimizer = QueryOptimizer::new();
712 let mut table = TableQuery::new("hosts");
713 table.filter = Some(Filter::Compare {
714 field: FieldRef::column("", "host_id"),
715 op: CompareOp::Eq,
716 value: Value::Integer(7),
717 });
718
719 let (optimized, passes) = optimizer.optimize(QueryExpr::Table(table));
720 let QueryExpr::Table(table) = optimized else {
721 panic!("expected table query");
722 };
723 let hint = table
724 .expand
725 .and_then(|expand| expand.index_hint)
726 .expect("expected optimizer index hint");
727
728 assert!(passes.iter().any(|pass| pass == "IndexSelection"));
729 assert_eq!(hint.method, IndexHintMethod::Hash);
730 assert_eq!(hint.column, "host_id");
731 }
732
733 #[test]
734 fn index_selection_analyzes_supported_filter_shapes() {
735 let range = IndexSelectionPass::analyze_filter(&Filter::Between {
736 field: FieldRef::node_prop("n", "score"),
737 low: Value::Integer(1),
738 high: Value::Integer(9),
739 })
740 .expect("expected range hint");
741 assert_eq!(range.method, IndexHintMethod::BTree);
742 assert_eq!(range.column, "score");
743
744 let bitmap = IndexSelectionPass::analyze_filter(&Filter::In {
745 field: FieldRef::edge_prop("e", "kind"),
746 values: vec![Value::text("http"), Value::text("ssh")],
747 })
748 .expect("expected bitmap hint");
749 assert_eq!(bitmap.method, IndexHintMethod::Bitmap);
750 assert_eq!(bitmap.column, "kind");
751
752 assert!(IndexSelectionPass::analyze_filter(&Filter::In {
753 field: FieldRef::column("", "status"),
754 values: (0..11).map(Value::Integer).collect(),
755 })
756 .is_none());
757
758 let fallback_right = IndexSelectionPass::analyze_filter(&Filter::And(
759 Box::new(Filter::IsNull(FieldRef::column("", "deleted_at"))),
760 Box::new(Filter::Compare {
761 field: FieldRef::node_id("n"),
762 op: CompareOp::Eq,
763 value: Value::Integer(1),
764 }),
765 ))
766 .expect("expected right-side AND hint");
767 assert_eq!(fallback_right.method, IndexHintMethod::Hash);
768 assert_eq!(fallback_right.column, "n.id");
769 }
770}