1use crate::error::{Result, TdbError};
47use crate::statistics::StatisticsSnapshot;
48use anyhow::Context;
49use parking_lot::Mutex;
50use serde::{Deserialize, Serialize};
51use std::collections::{HashMap, HashSet};
52use std::sync::Arc;
53
54#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
56pub struct TriplePattern {
57 pub subject: String,
59 pub predicate: String,
61 pub object: String,
63 pub estimated_cardinality: Option<u64>,
65}
66
67impl TriplePattern {
68 pub fn new(subject: &str, predicate: &str, object: &str) -> Self {
70 Self {
71 subject: subject.to_string(),
72 predicate: predicate.to_string(),
73 object: object.to_string(),
74 estimated_cardinality: None,
75 }
76 }
77
78 pub fn is_variable(term: &str) -> bool {
80 term.starts_with('?')
81 }
82
83 pub fn variables(&self) -> HashSet<String> {
85 let mut vars = HashSet::new();
86 if Self::is_variable(&self.subject) {
87 vars.insert(self.subject.clone());
88 }
89 if Self::is_variable(&self.predicate) {
90 vars.insert(self.predicate.clone());
91 }
92 if Self::is_variable(&self.object) {
93 vars.insert(self.object.clone());
94 }
95 vars
96 }
97
98 pub fn join_variables(&self, other: &TriplePattern) -> HashSet<String> {
100 self.variables()
101 .intersection(&other.variables())
102 .cloned()
103 .collect()
104 }
105
106 pub fn bound_count(&self) -> usize {
108 let mut count = 0;
109 if !Self::is_variable(&self.subject) {
110 count += 1;
111 }
112 if !Self::is_variable(&self.predicate) {
113 count += 1;
114 }
115 if !Self::is_variable(&self.object) {
116 count += 1;
117 }
118 count
119 }
120}
121
122#[derive(Debug, Clone, Serialize, Deserialize)]
124pub struct JoinNode {
125 pub left: Box<JoinPlan>,
127 pub right: Box<JoinPlan>,
129 pub join_vars: HashSet<String>,
131 pub cost: f64,
133 pub cardinality: u64,
135}
136
137#[derive(Debug, Clone, Serialize, Deserialize)]
139pub enum JoinPlan {
140 Pattern(TriplePattern),
142 Join(JoinNode),
144}
145
146impl JoinPlan {
147 pub fn variables(&self) -> HashSet<String> {
149 match self {
150 JoinPlan::Pattern(pattern) => pattern.variables(),
151 JoinPlan::Join(node) => {
152 let mut vars = node.left.variables();
153 vars.extend(node.right.variables());
154 vars
155 }
156 }
157 }
158
159 pub fn cardinality(&self) -> u64 {
161 match self {
162 JoinPlan::Pattern(pattern) => pattern.estimated_cardinality.unwrap_or(1000),
163 JoinPlan::Join(node) => node.cardinality,
164 }
165 }
166
167 pub fn cost(&self) -> f64 {
169 match self {
170 JoinPlan::Pattern(_) => 0.0,
171 JoinPlan::Join(node) => node.cost,
172 }
173 }
174}
175
176#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
178pub enum JoinAlgorithm {
179 Greedy,
181 DynamicProgramming,
183 Genetic,
185}
186
187#[derive(Debug, Clone, Serialize, Deserialize)]
189pub struct JoinOptimizerConfig {
190 pub algorithm: JoinAlgorithm,
192 pub dp_max_patterns: usize,
194 pub default_join_selectivity: f64,
196 pub cost_based: bool,
198}
199
200impl Default for JoinOptimizerConfig {
201 fn default() -> Self {
202 Self {
203 algorithm: JoinAlgorithm::Greedy,
204 dp_max_patterns: 12,
205 default_join_selectivity: 0.1,
206 cost_based: true,
207 }
208 }
209}
210
211pub struct JoinOptimizer {
215 config: JoinOptimizerConfig,
217 stats: Option<Arc<StatisticsSnapshot>>,
219 opt_stats: Arc<Mutex<OptimizationStats>>,
221}
222
223#[derive(Debug, Clone, Default, Serialize, Deserialize)]
225pub struct OptimizationStats {
226 pub total_optimizations: u64,
228 pub greedy_optimizations: u64,
230 pub dp_optimizations: u64,
232 pub avg_optimization_time_ms: f64,
234 total_time_ms: f64,
236}
237
238impl JoinOptimizer {
239 pub fn new(config: JoinOptimizerConfig) -> Self {
241 Self {
242 config,
243 stats: None,
244 opt_stats: Arc::new(Mutex::new(OptimizationStats::default())),
245 }
246 }
247
248 pub fn set_statistics(&mut self, stats: Arc<StatisticsSnapshot>) {
250 self.stats = Some(stats);
251 }
252
253 pub fn optimize(&mut self, mut patterns: Vec<TriplePattern>) -> Result<JoinPlan> {
255 if patterns.is_empty() {
256 return Err(TdbError::Other("No patterns to optimize".to_string()));
257 }
258
259 if patterns.len() == 1 {
260 return Ok(JoinPlan::Pattern(
261 patterns
262 .pop()
263 .expect("collection validated to be non-empty"),
264 ));
265 }
266
267 self.estimate_cardinalities(&mut patterns)?;
269
270 let algorithm = if patterns.len() > self.config.dp_max_patterns {
272 JoinAlgorithm::Greedy
273 } else {
274 self.config.algorithm
275 };
276
277 let start = std::time::Instant::now();
278
279 let plan = match algorithm {
280 JoinAlgorithm::Greedy => self.greedy_optimize(patterns)?,
281 JoinAlgorithm::DynamicProgramming => self.dp_optimize(patterns)?,
282 JoinAlgorithm::Genetic => {
283 self.greedy_optimize(patterns)?
286 }
287 };
288
289 let duration = start.elapsed().as_millis() as f64;
290
291 let mut opt_stats = self.opt_stats.lock();
293 opt_stats.total_optimizations += 1;
294 opt_stats.total_time_ms += duration;
295 opt_stats.avg_optimization_time_ms =
296 opt_stats.total_time_ms / opt_stats.total_optimizations as f64;
297
298 match algorithm {
299 JoinAlgorithm::Greedy => opt_stats.greedy_optimizations += 1,
300 JoinAlgorithm::DynamicProgramming => opt_stats.dp_optimizations += 1,
301 JoinAlgorithm::Genetic => opt_stats.greedy_optimizations += 1,
302 }
303
304 Ok(plan)
305 }
306
307 fn estimate_cardinalities(&self, patterns: &mut [TriplePattern]) -> Result<()> {
309 for pattern in patterns.iter_mut() {
310 let cardinality = self.estimate_pattern_cardinality(pattern);
311 pattern.estimated_cardinality = Some(cardinality);
312 }
313 Ok(())
314 }
315
316 fn estimate_pattern_cardinality(&self, pattern: &TriplePattern) -> u64 {
318 if let Some(ref stats) = self.stats {
319 let bound_count = pattern.bound_count();
321
322 match bound_count {
323 3 => 1, 2 => stats.total_triples / 100, 1 => stats.total_triples / 10, 0 => stats.total_triples, _ => stats.total_triples / 10,
328 }
329 } else {
330 match pattern.bound_count() {
332 3 => 1,
333 2 => 1000,
334 1 => 10000,
335 0 => 100000,
336 _ => 10000,
337 }
338 }
339 }
340
341 fn greedy_optimize(&self, patterns: Vec<TriplePattern>) -> Result<JoinPlan> {
348 if patterns.len() == 1 {
349 return Ok(JoinPlan::Pattern(patterns[0].clone()));
350 }
351
352 let mut remaining: Vec<_> = patterns.clone();
354 let start_idx = remaining
355 .iter()
356 .enumerate()
357 .min_by_key(|(_, p)| p.estimated_cardinality.unwrap_or(u64::MAX))
358 .map(|(i, _)| i)
359 .expect("collection validated to be non-empty");
360
361 let mut current_plan = JoinPlan::Pattern(remaining.remove(start_idx));
362 let mut current_vars = current_plan.variables();
363
364 while !remaining.is_empty() {
366 let (best_idx, _best_cost) = remaining
368 .iter()
369 .enumerate()
370 .map(|(i, pattern)| {
371 let join_vars = current_vars
372 .intersection(&pattern.variables())
373 .cloned()
374 .collect();
375 let cost = self.estimate_join_cost(¤t_plan, pattern, &join_vars);
376 (i, cost)
377 })
378 .min_by(|(_, cost1), (_, cost2)| {
379 cost1
380 .partial_cmp(cost2)
381 .unwrap_or(std::cmp::Ordering::Equal)
382 })
383 .expect("collection validated to be non-empty");
384
385 let next_pattern = remaining.remove(best_idx);
386 let join_vars = current_vars
387 .intersection(&next_pattern.variables())
388 .cloned()
389 .collect();
390
391 current_plan = self.create_join(current_plan, next_pattern, join_vars);
393 current_vars = current_plan.variables();
394 }
395
396 Ok(current_plan)
397 }
398
399 fn dp_optimize(&self, patterns: Vec<TriplePattern>) -> Result<JoinPlan> {
404 let n = patterns.len();
405 if n == 1 {
406 return Ok(JoinPlan::Pattern(patterns[0].clone()));
407 }
408
409 let mut dp: HashMap<u64, JoinPlan> = HashMap::new();
411
412 for (i, pattern) in patterns.iter().enumerate() {
414 let mask = 1u64 << i;
415 dp.insert(mask, JoinPlan::Pattern(pattern.clone()));
416 }
417
418 for size in 2..=n {
420 let subsets = self.generate_subsets(n, size);
422
423 for subset in subsets {
424 let mut best_plan: Option<JoinPlan> = None;
426 let mut best_cost = f64::MAX;
427
428 for left_mask in 1..subset {
430 if (left_mask & subset) != left_mask {
431 continue;
432 }
433
434 let right_mask = subset & !left_mask;
435 if right_mask == 0 {
436 continue;
437 }
438
439 let left_plan = dp.get(&left_mask).cloned();
441 let right_plan = dp.get(&right_mask).cloned();
442
443 if let (Some(left), Some(right)) = (left_plan, right_plan) {
444 let join_vars = left
445 .variables()
446 .intersection(&right.variables())
447 .cloned()
448 .collect();
449
450 let cost = self.estimate_join_cost_plans(&left, &right, &join_vars);
451
452 if cost < best_cost {
453 best_cost = cost;
454 best_plan = Some(self.create_join_plans(left, right, join_vars));
455 }
456 }
457 }
458
459 if let Some(plan) = best_plan {
460 dp.insert(subset, plan);
461 }
462 }
463 }
464
465 let full_mask = (1u64 << n) - 1;
467 dp.remove(&full_mask)
468 .ok_or_else(|| TdbError::Other("DP optimization failed".to_string()))
469 }
470
471 fn generate_subsets(&self, n: usize, k: usize) -> Vec<u64> {
473 let mut subsets = Vec::new();
474 self.generate_subsets_recursive(n, k, 0, 0, &mut subsets);
475 subsets
476 }
477
478 #[allow(clippy::only_used_in_recursion)]
480 fn generate_subsets_recursive(
481 &self,
482 n: usize,
483 k: usize,
484 start: usize,
485 current: u64,
486 subsets: &mut Vec<u64>,
487 ) {
488 if k == 0 {
489 subsets.push(current);
490 return;
491 }
492
493 for i in start..n {
494 let next = current | (1u64 << i);
495 self.generate_subsets_recursive(n, k - 1, i + 1, next, subsets);
496 }
497 }
498
499 fn estimate_join_cost(
501 &self,
502 plan: &JoinPlan,
503 pattern: &TriplePattern,
504 join_vars: &HashSet<String>,
505 ) -> f64 {
506 let left_card = plan.cardinality() as f64;
507 let right_card = pattern.estimated_cardinality.unwrap_or(1000) as f64;
508
509 let selectivity = if join_vars.is_empty() {
510 1.0
512 } else {
513 self.config.default_join_selectivity
515 };
516
517 left_card * right_card * selectivity
519 }
520
521 fn estimate_join_cost_plans(
523 &self,
524 left: &JoinPlan,
525 right: &JoinPlan,
526 join_vars: &HashSet<String>,
527 ) -> f64 {
528 let left_card = left.cardinality() as f64;
529 let right_card = right.cardinality() as f64;
530
531 let selectivity = if join_vars.is_empty() {
532 1.0
533 } else {
534 self.config.default_join_selectivity
535 };
536
537 left_card * right_card * selectivity + left.cost() + right.cost()
538 }
539
540 fn create_join(
542 &self,
543 left: JoinPlan,
544 right: TriplePattern,
545 join_vars: HashSet<String>,
546 ) -> JoinPlan {
547 let right_plan = JoinPlan::Pattern(right);
548 self.create_join_plans(left, right_plan, join_vars)
549 }
550
551 fn create_join_plans(
553 &self,
554 left: JoinPlan,
555 right: JoinPlan,
556 join_vars: HashSet<String>,
557 ) -> JoinPlan {
558 let cost = self.estimate_join_cost_plans(&left, &right, &join_vars);
559 let left_card = left.cardinality();
560 let right_card = right.cardinality();
561
562 let cardinality = if join_vars.is_empty() {
563 left_card.saturating_mul(right_card)
565 } else {
566 ((left_card as f64 * right_card as f64 * self.config.default_join_selectivity) as u64)
567 .max(1)
568 };
569
570 JoinPlan::Join(JoinNode {
571 left: Box::new(left),
572 right: Box::new(right),
573 join_vars,
574 cost,
575 cardinality,
576 })
577 }
578
579 pub fn stats(&self) -> OptimizationStats {
581 self.opt_stats.lock().clone()
582 }
583}
584
585#[cfg(test)]
586mod tests {
587 use super::*;
588
589 #[test]
590 fn test_triple_pattern_variables() {
591 let pattern = TriplePattern::new("?s", "type", "Person");
592 let vars = pattern.variables();
593
594 assert_eq!(vars.len(), 1);
595 assert!(vars.contains("?s"));
596 }
597
598 #[test]
599 fn test_triple_pattern_bound_count() {
600 let pattern1 = TriplePattern::new("?s", "type", "Person");
601 assert_eq!(pattern1.bound_count(), 2);
602
603 let pattern2 = TriplePattern::new("?s", "?p", "?o");
604 assert_eq!(pattern2.bound_count(), 0);
605
606 let pattern3 = TriplePattern::new("Alice", "name", "?name");
607 assert_eq!(pattern3.bound_count(), 2);
608 }
609
610 #[test]
611 fn test_join_variables() {
612 let pattern1 = TriplePattern::new("?s", "type", "Person");
613 let pattern2 = TriplePattern::new("?s", "name", "?name");
614
615 let join_vars = pattern1.join_variables(&pattern2);
616 assert_eq!(join_vars.len(), 1);
617 assert!(join_vars.contains("?s"));
618 }
619
620 #[test]
621 fn test_optimizer_creation() {
622 let config = JoinOptimizerConfig::default();
623 let optimizer = JoinOptimizer::new(config);
624
625 assert_eq!(optimizer.config.algorithm, JoinAlgorithm::Greedy);
626 }
627
628 #[test]
629 fn test_single_pattern_optimization() {
630 let config = JoinOptimizerConfig::default();
631 let mut optimizer = JoinOptimizer::new(config);
632
633 let patterns = vec![TriplePattern::new("?s", "type", "Person")];
634
635 let plan = optimizer.optimize(patterns).unwrap();
636
637 match plan {
638 JoinPlan::Pattern(pattern) => {
639 assert_eq!(pattern.subject, "?s");
640 assert_eq!(pattern.predicate, "type");
641 assert_eq!(pattern.object, "Person");
642 }
643 _ => panic!("Expected pattern plan"),
644 }
645 }
646
647 #[test]
648 fn test_greedy_optimization() {
649 let config = JoinOptimizerConfig {
650 algorithm: JoinAlgorithm::Greedy,
651 ..Default::default()
652 };
653 let mut optimizer = JoinOptimizer::new(config);
654
655 let patterns = vec![
656 TriplePattern::new("?s", "type", "Person"),
657 TriplePattern::new("?s", "name", "?name"),
658 TriplePattern::new("?s", "age", "?age"),
659 ];
660
661 let plan = optimizer.optimize(patterns).unwrap();
662
663 match plan {
665 JoinPlan::Join(_) => {
666 }
668 _ => panic!("Expected join plan"),
669 }
670
671 let stats = optimizer.stats();
672 assert_eq!(stats.total_optimizations, 1);
673 assert_eq!(stats.greedy_optimizations, 1);
674 }
675
676 #[test]
677 fn test_dp_optimization() {
678 let config = JoinOptimizerConfig {
679 algorithm: JoinAlgorithm::DynamicProgramming,
680 ..Default::default()
681 };
682 let mut optimizer = JoinOptimizer::new(config);
683
684 let patterns = vec![
685 TriplePattern::new("?s", "type", "Person"),
686 TriplePattern::new("?s", "name", "?name"),
687 TriplePattern::new("?s", "age", "?age"),
688 ];
689
690 let plan = optimizer.optimize(patterns).unwrap();
691
692 match plan {
693 JoinPlan::Join(_) => {
694 }
696 _ => panic!("Expected join plan"),
697 }
698
699 let stats = optimizer.stats();
700 assert_eq!(stats.dp_optimizations, 1);
701 }
702
703 #[test]
704 fn test_cardinality_estimation() {
705 let config = JoinOptimizerConfig::default();
706 let optimizer = JoinOptimizer::new(config);
707
708 let pattern1 = TriplePattern::new("Alice", "name", "\"Alice\"");
710 let card1 = optimizer.estimate_pattern_cardinality(&pattern1);
711 assert_eq!(card1, 1);
712
713 let pattern2 = TriplePattern::new("?s", "name", "\"Alice\"");
715 let card2 = optimizer.estimate_pattern_cardinality(&pattern2);
716 assert!(card2 > 1 && card2 < 10000);
717
718 let pattern3 = TriplePattern::new("?s", "?p", "?o");
720 let card3 = optimizer.estimate_pattern_cardinality(&pattern3);
721 assert!(card3 > 10000);
722 }
723
724 #[test]
725 fn test_join_plan_variables() {
726 let pattern1 = TriplePattern::new("?s", "type", "Person");
727 let pattern2 = TriplePattern::new("?s", "name", "?name");
728
729 let plan1 = JoinPlan::Pattern(pattern1.clone());
730 let plan2 = JoinPlan::Pattern(pattern2.clone());
731
732 let join_vars: HashSet<String> = vec!["?s".to_string()].into_iter().collect();
733
734 let join_plan = JoinPlan::Join(JoinNode {
735 left: Box::new(plan1),
736 right: Box::new(plan2),
737 join_vars,
738 cost: 100.0,
739 cardinality: 1000,
740 });
741
742 let vars = join_plan.variables();
743 assert_eq!(vars.len(), 2);
744 assert!(vars.contains("?s"));
745 assert!(vars.contains("?name"));
746 }
747
748 #[test]
749 fn test_multiple_optimizations() {
750 let config = JoinOptimizerConfig::default();
751 let mut optimizer = JoinOptimizer::new(config);
752
753 let patterns1 = vec![
755 TriplePattern::new("?s", "type", "Person"),
756 TriplePattern::new("?s", "name", "?name"),
757 ];
758 optimizer.optimize(patterns1).unwrap();
759
760 let patterns2 = vec![
762 TriplePattern::new("?x", "knows", "?y"),
763 TriplePattern::new("?y", "age", "?age"),
764 ];
765 optimizer.optimize(patterns2).unwrap();
766
767 let stats = optimizer.stats();
768 assert_eq!(stats.total_optimizations, 2);
769 assert!(stats.avg_optimization_time_ms >= 0.0);
771 }
772
773 #[test]
774 fn test_empty_patterns() {
775 let config = JoinOptimizerConfig::default();
776 let mut optimizer = JoinOptimizer::new(config);
777
778 let patterns = vec![];
779 let result = optimizer.optimize(patterns);
780
781 assert!(result.is_err());
782 }
783
784 #[test]
785 fn test_large_pattern_set() {
786 let config = JoinOptimizerConfig {
787 algorithm: JoinAlgorithm::DynamicProgramming,
788 dp_max_patterns: 5,
789 ..Default::default()
790 };
791 let mut optimizer = JoinOptimizer::new(config);
792
793 let patterns: Vec<_> = (0..10)
795 .map(|i| TriplePattern::new(&format!("?s{}", i), "type", "Person"))
796 .collect();
797
798 let plan = optimizer.optimize(patterns).unwrap();
799
800 let stats = optimizer.stats();
802 assert_eq!(stats.greedy_optimizations, 1);
803 assert_eq!(stats.dp_optimizations, 0);
804 }
805}