1use crate::model::{Object, Predicate, Subject, Triple, TriplePattern};
41use crate::model::{ObjectPattern, PredicatePattern, SubjectPattern};
42use crate::OxirsError;
43
44use scirs2_core::ndarray_ext::{Array1, Array2};
46
47#[cfg(feature = "parallel")]
49use rayon::prelude::*;
50
51use scirs2_core::metrics::{Counter, Timer};
53
54use std::sync::atomic::{AtomicU64, Ordering};
56use std::sync::Arc;
57
58pub type Result<T> = std::result::Result<T, OxirsError>;
60
61#[derive(Debug, Clone)]
63pub struct MatcherStats {
64 pub total_matches: u64,
66 pub total_triples_processed: u64,
68 pub simd_time_ns: u64,
70 pub scalar_time_ns: u64,
72 pub simd_calls: u64,
74 pub scalar_calls: u64,
76 pub avg_speedup: f64,
78}
79
80pub struct SimdTripleMatcher {
92 chunk_size: usize,
94 match_counter: Arc<Counter>,
96 simd_timer: Arc<Timer>,
98 scalar_timer: Arc<Timer>,
100 triples_processed: Arc<AtomicU64>,
102 simd_calls: Arc<AtomicU64>,
104 scalar_calls: Arc<AtomicU64>,
106}
107
108impl SimdTripleMatcher {
109 pub fn new() -> Self {
111 let match_counter = Arc::new(Counter::new("simd_triple_matches".to_string()));
112 let simd_timer = Arc::new(Timer::new("simd_matching".to_string()));
113 let scalar_timer = Arc::new(Timer::new("scalar_matching".to_string()));
114
115 Self {
116 chunk_size: Self::optimal_chunk_size(),
117 match_counter,
118 simd_timer,
119 scalar_timer,
120 triples_processed: Arc::new(AtomicU64::new(0)),
121 simd_calls: Arc::new(AtomicU64::new(0)),
122 scalar_calls: Arc::new(AtomicU64::new(0)),
123 }
124 }
125
126 pub fn with_chunk_size(chunk_size: usize) -> Self {
128 let mut matcher = Self::new();
129 matcher.chunk_size = chunk_size;
130 matcher
131 }
132
133 pub fn stats(&self) -> MatcherStats {
135 let simd_stats = self.simd_timer.get_stats();
136 let scalar_stats = self.scalar_timer.get_stats();
137
138 let simd_time_ns = (simd_stats.sum * 1_000_000_000.0) as u64;
139 let scalar_time_ns = (scalar_stats.sum * 1_000_000_000.0) as u64;
140 let simd_calls = self.simd_calls.load(Ordering::Relaxed);
141 let scalar_calls = self.scalar_calls.load(Ordering::Relaxed);
142
143 let avg_speedup = if simd_stats.mean > 0.0 && scalar_stats.mean > 0.0 {
145 scalar_stats.mean / simd_stats.mean
146 } else {
147 1.0
148 };
149
150 MatcherStats {
151 total_matches: self.match_counter.get(),
152 total_triples_processed: self.triples_processed.load(Ordering::Relaxed),
153 simd_time_ns,
154 scalar_time_ns,
155 simd_calls,
156 scalar_calls,
157 avg_speedup,
158 }
159 }
160
161 pub fn reset_stats(&self) {
166 self.triples_processed.store(0, Ordering::Relaxed);
167 self.simd_calls.store(0, Ordering::Relaxed);
168 self.scalar_calls.store(0, Ordering::Relaxed);
169 }
170
171 fn optimal_chunk_size() -> usize {
173 #[cfg(any(target_arch = "x86", target_arch = "x86_64"))]
175 {
176 if is_x86_feature_detected!("avx512f") {
177 16 } else {
179 8 }
181 }
182
183 #[cfg(target_arch = "aarch64")]
184 {
185 4 }
187
188 #[cfg(not(any(target_arch = "x86", target_arch = "x86_64", target_arch = "aarch64")))]
189 {
190 8 }
192 }
193
194 pub fn match_batch(&self, pattern: &TriplePattern, triples: &[Triple]) -> Result<Vec<usize>> {
208 if triples.is_empty() {
209 return Ok(Vec::new());
210 }
211
212 if triples.len() < self.chunk_size * 2 {
214 return Ok(self.match_scalar(pattern, triples));
215 }
216
217 self.match_simd(pattern, triples)
219 }
220
221 fn match_scalar(&self, pattern: &TriplePattern, triples: &[Triple]) -> Vec<usize> {
223 let _timer_guard = self.scalar_timer.start();
225 self.scalar_calls.fetch_add(1, Ordering::Relaxed);
226 self.triples_processed
227 .fetch_add(triples.len() as u64, Ordering::Relaxed);
228
229 let matches: Vec<usize> = triples
230 .iter()
231 .enumerate()
232 .filter_map(|(idx, triple)| {
233 if pattern.matches(triple) {
234 Some(idx)
235 } else {
236 None
237 }
238 })
239 .collect();
240
241 self.match_counter.add(matches.len() as u64);
242 matches
243 }
244
245 fn match_simd(&self, pattern: &TriplePattern, triples: &[Triple]) -> Result<Vec<usize>> {
247 let _timer_guard = self.simd_timer.start();
249 self.simd_calls.fetch_add(1, Ordering::Relaxed);
250 self.triples_processed
251 .fetch_add(triples.len() as u64, Ordering::Relaxed);
252
253 let mut matches = Vec::with_capacity(triples.len() / 4); let pattern_mask = self.pattern_to_mask(pattern);
257
258 #[cfg(feature = "parallel")]
259 {
260 if triples.len() > self.chunk_size * 8 {
262 return self.match_simd_parallel(pattern, triples, &pattern_mask);
263 }
264 }
265
266 for (chunk_idx, chunk) in triples.chunks(self.chunk_size).enumerate() {
268 let base_idx = chunk_idx * self.chunk_size;
269
270 let triple_masks = self.triples_to_masks(chunk);
272
273 let match_results = self.simd_compare_masks(&pattern_mask, &triple_masks)?;
275
276 for (i, &matched) in match_results.iter().enumerate() {
278 if matched != 0.0 {
279 matches.push(base_idx + i);
280 }
281 }
282 }
283
284 self.match_counter.add(matches.len() as u64);
285 Ok(matches)
286 }
287
288 #[cfg(feature = "parallel")]
290 fn match_simd_parallel(
291 &self,
292 _pattern: &TriplePattern,
293 triples: &[Triple],
294 pattern_mask: &[f32; 3],
295 ) -> Result<Vec<usize>> {
296 use std::sync::Mutex;
297
298 let matches = Arc::new(Mutex::new(Vec::new()));
300 let chunk_size = self.chunk_size;
301
302 let chunks: Vec<&[Triple]> = triples.chunks(chunk_size * 4).collect();
304
305 chunks.par_iter().for_each(|chunk_group| {
306 let mut local_matches = Vec::new();
307 for (chunk_idx, chunk) in chunk_group.chunks(chunk_size).enumerate() {
308 let base_idx = chunk_idx * chunk_size;
309 let triple_masks = self.triples_to_masks(chunk);
310
311 if let Ok(match_results) = self.simd_compare_masks(pattern_mask, &triple_masks) {
313 for (i, &matched) in match_results.iter().enumerate() {
314 if matched != 0.0 {
315 local_matches.push(base_idx + i);
316 }
317 }
318 }
319 }
320
321 if let Ok(mut global) = matches.lock() {
323 global.extend(local_matches);
324 }
325 });
326
327 let final_matches = match Arc::try_unwrap(matches) {
328 Ok(mutex) => mutex.into_inner().unwrap_or_default(),
329 Err(arc) => arc.lock().expect("lock should not be poisoned").clone(),
330 };
331
332 self.match_counter.add(final_matches.len() as u64);
333 Ok(final_matches)
334 }
335
336 fn pattern_to_mask(&self, pattern: &TriplePattern) -> [f32; 3] {
342 let subject_mask = match &pattern.subject {
343 None => 0.0, Some(SubjectPattern::Variable(_)) => 0.0, Some(SubjectPattern::NamedNode(nn)) => self.hash_term(nn.as_str()),
346 Some(SubjectPattern::BlankNode(bn)) => self.hash_term(bn.as_str()),
347 Some(SubjectPattern::QuotedTriple(_)) => 0.0,
350 };
351
352 let predicate_mask = match &pattern.predicate {
353 None => 0.0,
354 Some(PredicatePattern::Variable(_)) => 0.0,
355 Some(PredicatePattern::NamedNode(nn)) => self.hash_term(nn.as_str()),
356 };
357
358 let object_mask = match &pattern.object {
359 None => 0.0,
360 Some(ObjectPattern::Variable(_)) => 0.0,
361 Some(ObjectPattern::NamedNode(nn)) => self.hash_term(nn.as_str()),
362 Some(ObjectPattern::BlankNode(bn)) => self.hash_term(bn.as_str()),
363 Some(ObjectPattern::Literal(lit)) => self.hash_term(lit.value()),
364 Some(ObjectPattern::QuotedTriple(_)) => 0.0,
366 };
367
368 [subject_mask, predicate_mask, object_mask]
369 }
370
371 fn triples_to_masks(&self, triples: &[Triple]) -> Vec<[f32; 3]> {
373 triples
374 .iter()
375 .map(|triple| {
376 [
377 self.hash_subject(triple.subject()),
378 self.hash_predicate(triple.predicate()),
379 self.hash_object(triple.object()),
380 ]
381 })
382 .collect()
383 }
384
385 fn simd_compare_masks(
390 &self,
391 pattern: &[f32; 3],
392 triple_masks: &[[f32; 3]],
393 ) -> Result<Vec<f32>> {
394 if triple_masks.is_empty() {
395 return Ok(Vec::new());
396 }
397
398 if triple_masks.len() < 4 {
400 return Ok(self.scalar_compare_masks(pattern, triple_masks));
401 }
402
403 let num_triples = triple_masks.len();
405 let mut triple_matrix = Vec::with_capacity(num_triples * 3);
406 for mask in triple_masks {
407 triple_matrix.extend_from_slice(mask);
408 }
409
410 let triple_array = Array2::from_shape_vec((num_triples, 3), triple_matrix)
412 .map_err(|e| OxirsError::Query(format!("Failed to create triple array: {}", e)))?;
413
414 let pattern_array = Array1::from_vec(pattern.to_vec());
416
417 let mut results = vec![1.0; num_triples];
419
420 for (i, triple_view) in triple_array.outer_iter().enumerate() {
422 let mut matches = true;
423
424 for j in 0..3 {
426 let pattern_val = pattern_array[j];
427 let triple_val = triple_view[j];
428
429 if pattern_val == 0.0 {
431 continue;
432 }
433
434 if (pattern_val - triple_val).abs() >= 0.0001 {
436 matches = false;
437 break;
438 }
439 }
440
441 results[i] = if matches { 1.0 } else { 0.0 };
442 }
443
444 Ok(results)
445 }
446
447 fn scalar_compare_masks(&self, pattern: &[f32; 3], triple_masks: &[[f32; 3]]) -> Vec<f32> {
449 triple_masks
450 .iter()
451 .map(|triple_mask| {
452 let matches_all = (0..3).all(|j| {
453 let pattern_val = pattern[j];
454 let triple_val = triple_mask[j];
455
456 if pattern_val == 0.0 {
458 return true;
459 }
460
461 (pattern_val - triple_val).abs() < 0.0001
463 });
464
465 if matches_all {
466 1.0
467 } else {
468 0.0
469 }
470 })
471 .collect()
472 }
473
474 #[allow(dead_code)]
476 fn matches_mask(&self, pattern: &[f32; 3], triple: &Triple) -> bool {
477 let triple_mask = [
478 self.hash_subject(triple.subject()),
479 self.hash_predicate(triple.predicate()),
480 self.hash_object(triple.object()),
481 ];
482
483 (0..3).all(|i| {
484 let pattern_val = pattern[i];
485 let triple_val = triple_mask[i];
486
487 pattern_val == 0.0 || (pattern_val - triple_val).abs() < 0.0001
489 })
490 }
491
492 fn hash_term(&self, term: &str) -> f32 {
497 use std::collections::hash_map::DefaultHasher;
498 use std::hash::{Hash, Hasher};
499
500 let mut hasher = DefaultHasher::new();
501 term.hash(&mut hasher);
502 let hash = hasher.finish();
503
504 ((hash % (i32::MAX as u64)) as f32) + 1.0
507 }
508
509 fn hash_subject(&self, subject: &Subject) -> f32 {
511 match subject {
512 Subject::NamedNode(nn) => self.hash_term(nn.as_str()),
513 Subject::BlankNode(bn) => self.hash_term(bn.as_str()),
514 Subject::Variable(v) => self.hash_term(v.as_str()),
515 Subject::QuotedTriple(qt) => {
516 let repr = format!("<<{:?}>>", qt);
518 self.hash_term(&repr)
519 }
520 }
521 }
522
523 fn hash_predicate(&self, predicate: &Predicate) -> f32 {
525 match predicate {
526 Predicate::NamedNode(nn) => self.hash_term(nn.as_str()),
527 Predicate::Variable(v) => self.hash_term(v.as_str()),
528 }
529 }
530
531 fn hash_object(&self, object: &Object) -> f32 {
533 match object {
534 Object::NamedNode(nn) => self.hash_term(nn.as_str()),
535 Object::BlankNode(bn) => self.hash_term(bn.as_str()),
536 Object::Literal(lit) => self.hash_term(lit.value()),
537 Object::Variable(v) => self.hash_term(v.as_str()),
538 Object::QuotedTriple(qt) => {
539 let repr = format!("<<{:?}>>", qt);
541 self.hash_term(&repr)
542 }
543 }
544 }
545
546 pub fn estimate_selectivity(&self, pattern: &TriplePattern, _total_triples: usize) -> f32 {
550 let num_wildcards = pattern.subject.is_none() as i32
551 + pattern.predicate.is_none() as i32
552 + pattern.object.is_none() as i32;
553
554 match num_wildcards {
556 3 => 1.0, 2 => 0.5, 1 => 0.1, 0 => 0.001, _ => 0.5,
561 }
562 }
563}
564
565impl Default for SimdTripleMatcher {
566 fn default() -> Self {
567 Self::new()
568 }
569}
570
571#[cfg(test)]
572mod tests {
573 use super::*;
574 use crate::model::{Literal, NamedNode};
575
576 #[test]
577 fn test_simd_matcher_creation() {
578 let matcher = SimdTripleMatcher::new();
579 assert!(matcher.chunk_size >= 4);
580 assert!(matcher.chunk_size <= 16);
581 }
582
583 #[test]
584 fn test_match_empty_batch() {
585 let matcher = SimdTripleMatcher::new();
586 let pattern = TriplePattern::new(None, None, None);
587 let triples = vec![];
588
589 let matches = matcher
590 .match_batch(&pattern, &triples)
591 .expect("operation should succeed");
592 assert_eq!(matches.len(), 0);
593 }
594
595 #[test]
596 fn test_match_all_pattern() -> Result<()> {
597 let matcher = SimdTripleMatcher::new();
598 let pattern = TriplePattern::new(None, None, None); let s = Subject::NamedNode(NamedNode::new("http://example.org/s")?);
602 let p = Predicate::NamedNode(NamedNode::new("http://example.org/p")?);
603 let o = Object::Literal(Literal::new("test"));
604
605 let triples = vec![
606 Triple::new(s.clone(), p.clone(), o.clone()),
607 Triple::new(s.clone(), p.clone(), o.clone()),
608 Triple::new(s, p, o),
609 ];
610
611 let matches = matcher.match_batch(&pattern, &triples)?;
612 assert_eq!(matches.len(), 3); Ok(())
615 }
616
617 #[test]
618 fn test_hash_term_non_zero() {
619 let matcher = SimdTripleMatcher::new();
620 let hash1 = matcher.hash_term("http://example.org/test");
621 let hash2 = matcher.hash_term("http://example.org/other");
622
623 assert!(hash1 > 0.0);
625 assert!(hash2 > 0.0);
626
627 assert_ne!(hash1, hash2);
630 }
631
632 #[test]
633 fn test_optimal_chunk_size() {
634 let size = SimdTripleMatcher::optimal_chunk_size();
635 assert!((4..=16).contains(&size));
637 }
638
639 #[test]
640 fn test_estimate_selectivity() {
641 let matcher = SimdTripleMatcher::new();
642
643 let pattern_all = TriplePattern::new(None, None, None);
645 assert_eq!(matcher.estimate_selectivity(&pattern_all, 1000), 1.0);
646
647 let s =
649 SubjectPattern::NamedNode(NamedNode::new("http://example.org/s").expect("valid IRI"));
650 let p =
651 PredicatePattern::NamedNode(NamedNode::new("http://example.org/p").expect("valid IRI"));
652 let o = ObjectPattern::Literal(Literal::new("test"));
653 let pattern_none = TriplePattern::new(Some(s), Some(p), Some(o));
654 assert_eq!(matcher.estimate_selectivity(&pattern_none, 1000), 0.001);
655 }
656}