1use crate::dictionary::NodeId;
9use crate::error::{Result, TdbError};
10use crate::index::Triple;
11use parking_lot::RwLock;
12use scirs2_core::ndarray_ext::Array1;
13use scirs2_core::parallel_ops::{IntoParallelIterator, ParallelIterator};
14use scirs2_core::profiling::{Profiler, Timer};
15use std::sync::Arc;
16
17pub struct SimdPatternMatcher {
21 profiler: Arc<RwLock<Profiler>>,
23}
24
25impl SimdPatternMatcher {
26 pub fn new() -> Self {
28 Self {
29 profiler: Arc::new(RwLock::new(Profiler::new())),
30 }
31 }
32
33 pub fn match_pattern(
37 &self,
38 triples: &[Triple],
39 subject: Option<NodeId>,
40 predicate: Option<NodeId>,
41 object: Option<NodeId>,
42 ) -> Result<Vec<Triple>> {
43 let timer = Timer::start("simd_pattern_match");
44
45 if triples.is_empty() {
47 timer.stop();
48 return Ok(Vec::new());
49 }
50
51 let results: Vec<Triple> = triples
53 .iter()
54 .filter(|triple| {
55 if let Some(s) = subject {
57 if triple.subject != s {
58 return false;
59 }
60 }
61
62 if let Some(p) = predicate {
64 if triple.predicate != p {
65 return false;
66 }
67 }
68
69 if let Some(o) = object {
71 if triple.object != o {
72 return false;
73 }
74 }
75
76 true
77 })
78 .copied()
79 .collect();
80
81 timer.stop();
82
83 Ok(results)
84 }
85
86 pub fn get_stats(&self) -> String {
88 "SIMD Pattern Matcher - optimized for vectorized operations".to_string()
89 }
90}
91
92impl Default for SimdPatternMatcher {
93 fn default() -> Self {
94 Self::new()
95 }
96}
97
98pub struct ParallelQueryExecutor {
103 num_threads: usize,
105 profiler: Arc<RwLock<Profiler>>,
107}
108
109impl ParallelQueryExecutor {
110 pub fn new(num_threads: usize) -> Self {
112 Self {
113 num_threads,
114 profiler: Arc::new(RwLock::new(Profiler::new())),
115 }
116 }
117
118 pub fn parallel_match(
127 &self,
128 triples: &[Triple],
129 subject: Option<NodeId>,
130 predicate: Option<NodeId>,
131 object: Option<NodeId>,
132 ) -> Result<Vec<Triple>> {
133 let timer = Timer::start("parallel_match");
134
135 let results: Vec<Triple> = triples
137 .into_par_iter()
138 .filter(|triple| {
139 if let Some(s) = subject {
141 if triple.subject != s {
142 return false;
143 }
144 }
145
146 if let Some(p) = predicate {
148 if triple.predicate != p {
149 return false;
150 }
151 }
152
153 if let Some(o) = object {
155 if triple.object != o {
156 return false;
157 }
158 }
159
160 true
161 })
162 .copied()
163 .collect();
164
165 timer.stop();
166
167 Ok(results)
168 }
169
170 pub fn get_stats(&self) -> String {
172 format!(
173 "Parallel Query Executor - {} threads configured",
174 self.num_threads
175 )
176 }
177}
178
179pub struct HighPerfBloomFilter {
183 bits: Array1<f64>,
185 num_hashes: usize,
187 size: usize,
189 profiler: Arc<RwLock<Profiler>>,
191}
192
193impl HighPerfBloomFilter {
194 pub fn new(expected_items: usize, false_positive_rate: f64) -> Self {
200 let size = Self::optimal_size(expected_items, false_positive_rate);
202 let num_hashes = Self::optimal_hashes(expected_items, size);
203
204 let bits = Array1::zeros(size);
205
206 Self {
207 bits,
208 num_hashes,
209 size,
210 profiler: Arc::new(RwLock::new(Profiler::new())),
211 }
212 }
213
214 fn optimal_size(n: usize, p: f64) -> usize {
216 let ln2 = std::f64::consts::LN_2;
217 let size = -(n as f64 * p.ln()) / (ln2 * ln2);
218 size.ceil() as usize
219 }
220
221 fn optimal_hashes(n: usize, m: usize) -> usize {
223 let ln2 = std::f64::consts::LN_2;
224 let k = (m as f64 / n as f64) * ln2;
225 k.ceil() as usize
226 }
227
228 pub fn insert(&mut self, triple: &Triple) -> Result<()> {
230 let timer = Timer::start("bloom_insert");
231
232 let hash_base = self.hash_triple(triple);
233
234 for i in 0..self.num_hashes {
235 let hash = self.nth_hash(hash_base, i);
236 let index = (hash % self.size as u64) as usize;
237 self.bits[index] = 1.0;
238 }
239
240 timer.stop();
241 Ok(())
242 }
243
244 pub fn contains(&self, triple: &Triple) -> bool {
250 let timer = Timer::start("bloom_contains");
251
252 let hash_base = self.hash_triple(triple);
253
254 for i in 0..self.num_hashes {
255 let hash = self.nth_hash(hash_base, i);
256 let index = (hash % self.size as u64) as usize;
257 if self.bits[index] < 0.5 {
258 timer.stop();
259 return false;
260 }
261 }
262
263 timer.stop();
264 true
265 }
266
267 fn hash_triple(&self, triple: &Triple) -> u64 {
269 use std::collections::hash_map::DefaultHasher;
270 use std::hash::{Hash, Hasher};
271
272 let mut hasher = DefaultHasher::new();
273 triple.subject.as_u64().hash(&mut hasher);
274 triple.predicate.as_u64().hash(&mut hasher);
275 triple.object.as_u64().hash(&mut hasher);
276 hasher.finish()
277 }
278
279 fn nth_hash(&self, base: u64, n: usize) -> u64 {
281 let hash1 = base;
283 let hash2 = base.wrapping_mul(0x9e3779b97f4a7c15); hash1.wrapping_add((n as u64).wrapping_mul(hash2))
285 }
286
287 pub fn stats(&self) -> BloomFilterStats {
289 let bits_set = self.bits.iter().filter(|&&b| b > 0.5).count();
290 let load_factor = bits_set as f64 / self.size as f64;
291
292 BloomFilterStats {
293 size: self.size,
294 bits_set,
295 load_factor,
296 num_hashes: self.num_hashes,
297 }
298 }
299
300 pub fn get_profiling_stats(&self) -> String {
302 format!(
303 "Bloom Filter - size: {}, hashes: {}",
304 self.size, self.num_hashes
305 )
306 }
307}
308
309#[derive(Debug, Clone)]
311pub struct BloomFilterStats {
312 pub size: usize,
314 pub bits_set: usize,
316 pub load_factor: f64,
318 pub num_hashes: usize,
320}
321
322pub struct MemoryEfficientScanner {
327 chunk_size: usize,
329 profiler: Arc<RwLock<Profiler>>,
331}
332
333impl MemoryEfficientScanner {
334 pub fn new(chunk_size: usize) -> Self {
336 Self {
337 chunk_size,
338 profiler: Arc::new(RwLock::new(Profiler::new())),
339 }
340 }
341
342 pub fn scan<F>(&self, triples: &[Triple], mut callback: F) -> Result<()>
347 where
348 F: FnMut(&[Triple]) -> Result<()>,
349 {
350 let timer = Timer::start("memory_efficient_scan");
351
352 for chunk in triples.chunks(self.chunk_size) {
353 callback(chunk)?;
354 }
355
356 timer.stop();
357 Ok(())
358 }
359
360 pub fn get_stats(&self) -> String {
362 format!("Memory Efficient Scanner - chunk size: {}", self.chunk_size)
363 }
364}
365
366#[cfg(test)]
367mod tests {
368 use super::*;
369 use crate::dictionary::NodeId;
370
371 #[test]
372 fn test_simd_pattern_matcher_creation() {
373 let _matcher = SimdPatternMatcher::new();
374 }
375
376 #[test]
377 fn test_simd_pattern_matcher_empty() {
378 let matcher = SimdPatternMatcher::new();
379 let results = matcher.match_pattern(&[], None, None, None).unwrap();
380 assert_eq!(results.len(), 0);
381 }
382
383 #[test]
384 fn test_simd_pattern_matcher_wildcard() {
385 let matcher = SimdPatternMatcher::new();
386 let triples = vec![
387 Triple::new(NodeId::from(1), NodeId::from(2), NodeId::from(3)),
388 Triple::new(NodeId::from(4), NodeId::from(5), NodeId::from(6)),
389 ];
390
391 let results = matcher.match_pattern(&triples, None, None, None).unwrap();
392 assert_eq!(results.len(), 2);
393 }
394
395 #[test]
396 fn test_simd_pattern_matcher_subject() {
397 let matcher = SimdPatternMatcher::new();
398 let triples = vec![
399 Triple::new(NodeId::from(1), NodeId::from(2), NodeId::from(3)),
400 Triple::new(NodeId::from(1), NodeId::from(5), NodeId::from(6)),
401 Triple::new(NodeId::from(4), NodeId::from(2), NodeId::from(3)),
402 ];
403
404 let results = matcher
405 .match_pattern(&triples, Some(NodeId::from(1)), None, None)
406 .unwrap();
407 assert_eq!(results.len(), 2);
408 }
409
410 #[test]
411 fn test_simd_pattern_matcher_full_pattern() {
412 let matcher = SimdPatternMatcher::new();
413 let triples = vec![
414 Triple::new(NodeId::from(1), NodeId::from(2), NodeId::from(3)),
415 Triple::new(NodeId::from(1), NodeId::from(5), NodeId::from(6)),
416 Triple::new(NodeId::from(4), NodeId::from(2), NodeId::from(3)),
417 ];
418
419 let results = matcher
420 .match_pattern(
421 &triples,
422 Some(NodeId::from(1)),
423 Some(NodeId::from(2)),
424 Some(NodeId::from(3)),
425 )
426 .unwrap();
427 assert_eq!(results.len(), 1);
428 assert_eq!(results[0].subject, NodeId::from(1));
429 }
430
431 #[test]
432 fn test_parallel_query_executor_creation() {
433 let _executor = ParallelQueryExecutor::new(4);
434 }
435
436 #[test]
437 fn test_parallel_query_executor_match() {
438 let executor = ParallelQueryExecutor::new(4);
439 let triples = vec![
440 Triple::new(NodeId::from(1), NodeId::from(2), NodeId::from(3)),
441 Triple::new(NodeId::from(1), NodeId::from(5), NodeId::from(6)),
442 Triple::new(NodeId::from(4), NodeId::from(2), NodeId::from(3)),
443 ];
444
445 let results = executor
446 .parallel_match(&triples, Some(NodeId::from(1)), None, None)
447 .unwrap();
448 assert_eq!(results.len(), 2);
449 }
450
451 #[test]
452 fn test_parallel_query_executor_large_dataset() {
453 let executor = ParallelQueryExecutor::new(4);
454
455 let mut triples = Vec::new();
457 for i in 0..10000 {
458 triples.push(Triple::new(
459 NodeId::from(i % 100),
460 NodeId::from(i % 50),
461 NodeId::from(i % 200),
462 ));
463 }
464
465 let results = executor
466 .parallel_match(&triples, Some(NodeId::from(1)), None, None)
467 .unwrap();
468
469 assert!(!results.is_empty());
471 for triple in &results {
472 assert_eq!(triple.subject, NodeId::from(1));
473 }
474 }
475
476 #[test]
477 fn test_bloom_filter_creation() {
478 let _filter = HighPerfBloomFilter::new(1000, 0.01);
479 }
480
481 #[test]
482 fn test_bloom_filter_insert_contains() {
483 let mut filter = HighPerfBloomFilter::new(1000, 0.01);
484 let triple = Triple::new(NodeId::from(1), NodeId::from(2), NodeId::from(3));
485
486 filter.insert(&triple).unwrap();
487 assert!(filter.contains(&triple));
488 }
489
490 #[test]
491 fn test_bloom_filter_negative() {
492 let mut filter = HighPerfBloomFilter::new(1000, 0.01);
493 let triple1 = Triple::new(NodeId::from(1), NodeId::from(2), NodeId::from(3));
494 let triple2 = Triple::new(NodeId::from(4), NodeId::from(5), NodeId::from(6));
495
496 filter.insert(&triple1).unwrap();
497
498 }
502
503 #[test]
504 fn test_bloom_filter_stats() {
505 let mut filter = HighPerfBloomFilter::new(100, 0.01);
506
507 for i in 0..50 {
508 let triple = Triple::new(NodeId::from(i), NodeId::from(i + 1), NodeId::from(i + 2));
509 filter.insert(&triple).unwrap();
510 }
511
512 let stats = filter.stats();
513 assert_eq!(stats.size, filter.size);
514 assert!(stats.bits_set > 0);
515 assert!(stats.load_factor > 0.0 && stats.load_factor <= 1.0);
516 }
517
518 #[test]
519 fn test_memory_efficient_scanner() {
520 let scanner = MemoryEfficientScanner::new(100);
521 let triples = vec![
522 Triple::new(NodeId::from(1), NodeId::from(2), NodeId::from(3)),
523 Triple::new(NodeId::from(4), NodeId::from(5), NodeId::from(6)),
524 ];
525
526 let mut count = 0;
527 scanner
528 .scan(&triples, |chunk| {
529 count += chunk.len();
530 Ok(())
531 })
532 .unwrap();
533
534 assert_eq!(count, 2);
535 }
536
537 #[test]
538 fn test_memory_efficient_scanner_large() {
539 let scanner = MemoryEfficientScanner::new(100);
540
541 let mut triples = Vec::new();
543 for i in 0..1000 {
544 triples.push(Triple::new(
545 NodeId::from(i),
546 NodeId::from(i + 1),
547 NodeId::from(i + 2),
548 ));
549 }
550
551 let mut chunk_count = 0;
552 let mut total_triples = 0;
553
554 scanner
555 .scan(&triples, |chunk| {
556 chunk_count += 1;
557 total_triples += chunk.len();
558 assert!(chunk.len() <= 100);
559 Ok(())
560 })
561 .unwrap();
562
563 assert_eq!(total_triples, 1000);
564 assert_eq!(chunk_count, 10); }
566}