fluidB
/
fluidb
1
0
Fork 0
Code Pull Requests Activity

restore some cases of prefix join in neg join

Browse Source
main
Ziyang Hu 2 years ago
parent 115873e61a
commit c5ac7d36e1
1 changed files with 181 additions and 84 deletions
Show Stats Download Patch File Download Diff File

265
src/query/relation.rs
Unescape Escape View File

@ -756,48 +756,96 @@ impl RelationRA {
left_to_prefix_indices.push(left_join_indices[*idx]); left_to_prefix_indices.push(left_join_indices[*idx]);
} }
// TODO don't build the whole thing is prefix scan suffices if join_is_prefix(&right_join_indices) {
let mut right_join_vals = BTreeSet::new(); Ok(Box::new(
for tuple in self.storage.scan_all(tx) { left_iter
let tuple = tuple?; .map_ok(move |tuple| -> Result<Option<Tuple>> {
let to_join: Box<[DataValue]> = right_join_indices.iter().map(|i| { let prefix = Tuple(
tuple.0[*i].clone() left_to_prefix_indices
}).collect(); .iter()
right_join_vals.insert(to_join); .map(|i| tuple.0[*i].clone())
} .collect_vec(),
);
Ok(Box::new( 'outer: for found in self.storage.scan_prefix(tx, &prefix) {
left_iter let found = found?;
.map_ok(move |tuple| -> Result<Option<Tuple>> { for (left_idx, right_idx) in
let left_join_vals: Box<[DataValue]> = left_join_indices.iter().map(|i| { left_join_indices.iter().zip(right_join_indices.iter())
tuple.0[*i].clone() {
}).collect(); if tuple.0[*left_idx] != found.0[*right_idx] {
if right_join_vals.contains(&left_join_vals) { continue 'outer;
return Ok(None) }
} }
return Ok(None);
}
Ok(Some(if !eliminate_indices.is_empty() { Ok(Some(if !eliminate_indices.is_empty() {
Tuple( Tuple(
tuple
.0
.into_iter()
.enumerate()
.filter_map(|(i, v)| {
if eliminate_indices.contains(&i) {
None
} else {
Some(v)
}
})
.collect_vec(),
)
} else {
tuple tuple
.0 }))
.into_iter() })
.enumerate() .map(flatten_err)
.filter_map(|(i, v)| { .filter_map(invert_option_err),
if eliminate_indices.contains(&i) { ))
None } else {
} else { let mut right_join_vals = BTreeSet::new();
Some(v)
} for tuple in self.storage.scan_all(tx) {
}) let tuple = tuple?;
.collect_vec(), let to_join: Box<[DataValue]> = right_join_indices
) .iter()
} else { .map(|i| tuple.0[*i].clone())
tuple .collect();
})) right_join_vals.insert(to_join);
}) }
.map(flatten_err) Ok(Box::new(
.filter_map(invert_option_err), left_iter
)) .map_ok(move |tuple| -> Result<Option<Tuple>> {
let left_join_vals: Box<[DataValue]> = left_join_indices
.iter()
.map(|i| tuple.0[*i].clone())
.collect();
if right_join_vals.contains(&left_join_vals) {
return Ok(None);
}
Ok(Some(if !eliminate_indices.is_empty() {
Tuple(
tuple
.0
.into_iter()
.enumerate()
.filter_map(|(i, v)| {
if eliminate_indices.contains(&i) {
None
} else {
Some(v)
}
})
.collect_vec(),
)
} else {
tuple
}))
})
.map(flatten_err)
.filter_map(invert_option_err),
))
}
} }
fn iter(&self, tx: &SessionTx) -> Result<TupleIter<'_>> { fn iter(&self, tx: &SessionTx) -> Result<TupleIter<'_>> {
@ -808,12 +856,13 @@ impl RelationRA {
Box::new(filter_iter(self.filters.clone(), it)) Box::new(filter_iter(self.filters.clone(), it))
}) })
} }
fn join_is_prefix(&self, right_join_indices: &[usize]) -> bool { }
let mut indices = right_join_indices.to_vec();
indices.sort(); fn join_is_prefix(right_join_indices: &[usize]) -> bool {
let l = indices.len(); let mut indices = right_join_indices.to_vec();
indices.into_iter().eq(0..l) indices.sort();
} let l = indices.len();
indices.into_iter().eq(0..l)
} }
#[derive(Debug)] #[derive(Debug)]
@ -900,47 +949,95 @@ impl StoredRelationRA {
} }
left_to_prefix_indices.push(left_join_indices[*idx]); left_to_prefix_indices.push(left_join_indices[*idx]);
} }
if join_is_prefix(&right_join_indices) {
Ok(Box::new(
left_iter
.map_ok(move |tuple| -> Result<Option<Tuple>> {
let prefix = Tuple(
left_to_prefix_indices
.iter()
.map(|i| tuple.0[*i].clone())
.collect_vec(),
);
let mut right_join_vals = BTreeSet::new(); 'outer: for found in self.storage.scan_prefix(&prefix) {
for tuple in self.storage.scan_all() { let found = found?;
let tuple = tuple?; for (left_idx, right_idx) in
let to_join: Box<[DataValue]> = right_join_indices.iter().map(|i| { left_join_indices.iter().zip(right_join_indices.iter())
tuple.0[*i].clone() {
}).collect(); if tuple.0[*left_idx] != found.0[*right_idx] {
right_join_vals.insert(to_join); continue 'outer;
} }
}
return Ok(None);
}
Ok(Box::new( Ok(Some(if !eliminate_indices.is_empty() {
left_iter Tuple(
.map_ok(move |tuple| -> Result<Option<Tuple>> { tuple
let left_join_vals: Box<[DataValue]> = left_join_indices.iter().map(|i| { .0
tuple.0[*i].clone() .into_iter()
}).collect(); .enumerate()
if right_join_vals.contains(&left_join_vals) { .filter_map(|(i, v)| {
return Ok(None) if eliminate_indices.contains(&i) {
} None
Ok(Some(if !eliminate_indices.is_empty() { } else {
Tuple( Some(v)
}
})
.collect_vec(),
)
} else {
tuple tuple
.0 }))
.into_iter() })
.enumerate() .map(flatten_err)
.filter_map(|(i, v)| { .filter_map(invert_option_err),
if eliminate_indices.contains(&i) { ))
None } else {
} else { let mut right_join_vals = BTreeSet::new();
Some(v) for tuple in self.storage.scan_all() {
} let tuple = tuple?;
}) let to_join: Box<[DataValue]> = right_join_indices
.collect_vec(), .iter()
) .map(|i| tuple.0[*i].clone())
} else { .collect();
tuple right_join_vals.insert(to_join);
})) }
})
.map(flatten_err) Ok(Box::new(
.filter_map(invert_option_err), left_iter
)) .map_ok(move |tuple| -> Result<Option<Tuple>> {
let left_join_vals: Box<[DataValue]> = left_join_indices
.iter()
.map(|i| tuple.0[*i].clone())
.collect();
if right_join_vals.contains(&left_join_vals) {
return Ok(None);
}
Ok(Some(if !eliminate_indices.is_empty() {
Tuple(
tuple
.0
.into_iter()
.enumerate()
.filter_map(|(i, v)| {
if eliminate_indices.contains(&i) {
None
} else {
Some(v)
}
})
.collect_vec(),
)
} else {
tuple
}))
})
.map(flatten_err)
.filter_map(invert_option_err),
))
}
} }
fn prefix_join<'a>( fn prefix_join<'a>(
&'a self, &'a self,
@ -1305,7 +1402,7 @@ impl InnerJoin {
&self.right.bindings_after_eliminate(), &self.right.bindings_after_eliminate(),
) )
.unwrap(); .unwrap();
if r.join_is_prefix(&join_indices.1) { if join_is_prefix(&join_indices.1) {
r.prefix_join( r.prefix_join(
tx, tx,
self.left.iter(tx, epoch, use_delta)?, self.left.iter(tx, epoch, use_delta)?,

Write Preview
Loading…
Cancel
Save