src/vendor/datafrog/src/join.rs

   1 //! Join functionality.
   2
   3 use super::{Variable, Relation};
   4
   5 pub fn join_into<Key: Ord, Val1: Ord, Val2: Ord, Result: Ord>(
   6     input1: &Variable<(Key, Val1)>,
   7     input2: &Variable<(Key, Val2)>,
   8     output: &Variable<Result>,
   9     mut logic: impl FnMut(&Key, &Val1, &Val2)->Result) {
  10
  11     let mut results = Vec::new();
  12
  13     let recent1 = input1.recent.borrow();
  14     let recent2 = input2.recent.borrow();
  15
  16     {   // scoped to let `closure` drop borrow of `results`.
  17
  18         let mut closure = |k: &Key, v1: &Val1, v2: &Val2| results.push(logic(k,v1,v2));
  19
  20         for batch2 in input2.stable.borrow().iter() {
  21             join_helper(&recent1, &batch2, &mut closure);
  22         }
  23
  24         for batch1 in input1.stable.borrow().iter() {
  25             join_helper(&batch1, &recent2, &mut closure);
  26         }
  27
  28         join_helper(&recent1, &recent2, &mut closure);
  29     }
  30
  31     output.insert(Relation::from_vec(results));
  32 }
  33
  34 /// Moves all recent tuples from `input1` that are not present in `input2` into `output`.
  35 pub fn antijoin_into<Key: Ord, Val: Ord, Result: Ord>(
  36     input1: &Variable<(Key, Val)>,
  37     input2: &Relation<Key>,
  38     output: &Variable<Result>,
  39     mut logic: impl FnMut(&Key, &Val)->Result) {
  40
  41     let mut results = Vec::new();
  42     let mut tuples2 = &input2[..];
  43
  44     for &(ref key, ref val) in input1.recent.borrow().iter() {
  45         tuples2 = gallop(tuples2, |k| k < key);
  46         if tuples2.first() != Some(key) {
  47             results.push(logic(key, val));
  48         }
  49     }
  50
  51     output.insert(Relation::from_vec(results));
  52 }
  53
  54 fn join_helper<K: Ord, V1, V2>(
  55     mut slice1: &[(K,V1)],
  56     mut slice2: &[(K,V2)],
  57     mut result: impl FnMut(&K,&V1,&V2)) {
  58
  59     while !slice1.is_empty() && !slice2.is_empty() {
  60
  61         use std::cmp::Ordering;
  62
  63         // If the keys match produce tuples, else advance the smaller key until they might.
  64         match slice1[0].0.cmp(&slice2[0].0) {
  65             Ordering::Less => {
  66                 slice1 = gallop(slice1, |x| x.0 < slice2[0].0);
  67             },
  68             Ordering::Equal => {
  69
  70                 // Determine the number of matching keys in each slice.
  71                 let count1 = slice1.iter().take_while(|x| x.0 == slice1[0].0).count();
  72                 let count2 = slice2.iter().take_while(|x| x.0 == slice2[0].0).count();
  73
  74                 // Produce results from the cross-product of matches.
  75                 for index1 in 0 .. count1 {
  76                     for index2 in 0 .. count2 {
  77                         result(&slice1[0].0, &slice1[index1].1, &slice2[index2].1);
  78                     }
  79                 }
  80
  81                 // Advance slices past this key.
  82                 slice1 = &slice1[count1..];
  83                 slice2 = &slice2[count2..];
  84             }
  85             Ordering::Greater => {
  86                 slice2 = gallop(slice2, |x| x.0 < slice1[0].0);
  87             }
  88         }
  89     }
  90 }
  91
  92 pub fn gallop<T>(mut slice: &[T], mut cmp: impl FnMut(&T)->bool) -> &[T] {
  93     // if empty slice, or already >= element, return
  94     if slice.len() > 0 && cmp(&slice[0]) {
  95         let mut step = 1;
  96         while step < slice.len() && cmp(&slice[step]) {
  97             slice = &slice[step..];
  98             step = step << 1;
  99         }
 100
 101         step = step >> 1;
 102         while step > 0 {
 103             if step < slice.len() && cmp(&slice[step]) {
 104                 slice = &slice[step..];
 105             }
 106             step = step >> 1;
 107         }
 108
 109         slice = &slice[1..]; // advance one, as we always stayed < value
 110     }
 111
 112     return slice;
 113 }