|
|
|
|
@ -6,7 +6,6 @@
|
|
|
|
|
* You can obtain one at https://mozilla.org/MPL/2.0/.
|
|
|
|
|
*/
|
|
|
|
|
|
|
|
|
|
use std::cmp::Reverse;
|
|
|
|
|
use std::collections::{BTreeMap, BTreeSet};
|
|
|
|
|
use std::fmt::{Debug, Formatter};
|
|
|
|
|
use std::iter;
|
|
|
|
|
@ -14,14 +13,15 @@ use std::iter;
|
|
|
|
|
use either::{Left, Right};
|
|
|
|
|
use itertools::Itertools;
|
|
|
|
|
use log::{debug, error};
|
|
|
|
|
use miette::{Diagnostic, Result};
|
|
|
|
|
use miette::{bail, Diagnostic, Result};
|
|
|
|
|
use thiserror::Error;
|
|
|
|
|
|
|
|
|
|
use crate::data::expr::{compute_bounds, Expr};
|
|
|
|
|
use crate::data::program::MagicSymbol;
|
|
|
|
|
use crate::data::relation::{ColType, NullableColType};
|
|
|
|
|
use crate::data::symb::Symbol;
|
|
|
|
|
use crate::data::tuple::{Tuple, TupleIter};
|
|
|
|
|
use crate::data::value::DataValue;
|
|
|
|
|
use crate::data::value::{DataValue, ValidityTs};
|
|
|
|
|
use crate::parse::SourceSpan;
|
|
|
|
|
use crate::runtime::relation::RelationHandle;
|
|
|
|
|
use crate::runtime::temp_store::EpochStore;
|
|
|
|
|
@ -317,6 +317,14 @@ impl Debug for RelAlgebra {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#[derive(Debug, Error, Diagnostic)]
|
|
|
|
|
#[error("Invalid time travel on relation {0}")]
|
|
|
|
|
#[diagnostic(code(eval::invalid_time_travel))]
|
|
|
|
|
#[diagnostic(help(
|
|
|
|
|
"Time travel scanning requires the last key column of the relation to be of type 'Validity'"
|
|
|
|
|
))]
|
|
|
|
|
pub(crate) struct InvalidTimeTravelScanning(pub(crate) String, #[label] pub(crate) SourceSpan);
|
|
|
|
|
|
|
|
|
|
impl RelAlgebra {
|
|
|
|
|
pub(crate) fn fill_binding_indices(&mut self) -> Result<()> {
|
|
|
|
|
match self {
|
|
|
|
|
@ -380,25 +388,31 @@ impl RelAlgebra {
|
|
|
|
|
bindings: Vec<Symbol>,
|
|
|
|
|
storage: RelationHandle,
|
|
|
|
|
span: SourceSpan,
|
|
|
|
|
validity: Option<Reverse<i64>>,
|
|
|
|
|
) -> Self {
|
|
|
|
|
validity: Option<ValidityTs>,
|
|
|
|
|
) -> Result<Self> {
|
|
|
|
|
match validity {
|
|
|
|
|
None => {
|
|
|
|
|
Self::Stored(StoredRA {
|
|
|
|
|
None => Ok(Self::Stored(StoredRA {
|
|
|
|
|
bindings,
|
|
|
|
|
storage,
|
|
|
|
|
filters: vec![],
|
|
|
|
|
span,
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
})),
|
|
|
|
|
Some(vld) => {
|
|
|
|
|
Self::StoredWithValidity(StoredWithValidityRA {
|
|
|
|
|
if storage.metadata.keys.last().unwrap().typing
|
|
|
|
|
!= (NullableColType {
|
|
|
|
|
coltype: ColType::Validity,
|
|
|
|
|
nullable: false,
|
|
|
|
|
})
|
|
|
|
|
{
|
|
|
|
|
bail!(InvalidTimeTravelScanning(storage.name.to_string(), span));
|
|
|
|
|
};
|
|
|
|
|
Ok(Self::StoredWithValidity(StoredWithValidityRA {
|
|
|
|
|
bindings,
|
|
|
|
|
storage,
|
|
|
|
|
filters: vec![],
|
|
|
|
|
valid_at: vld,
|
|
|
|
|
span,
|
|
|
|
|
})
|
|
|
|
|
}))
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
@ -789,7 +803,7 @@ pub(crate) struct StoredWithValidityRA {
|
|
|
|
|
pub(crate) bindings: Vec<Symbol>,
|
|
|
|
|
pub(crate) storage: RelationHandle,
|
|
|
|
|
pub(crate) filters: Vec<Expr>,
|
|
|
|
|
pub(crate) valid_at: Reverse<i64>,
|
|
|
|
|
pub(crate) valid_at: ValidityTs,
|
|
|
|
|
pub(crate) span: SourceSpan,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -808,13 +822,12 @@ impl StoredWithValidityRA {
|
|
|
|
|
Ok(())
|
|
|
|
|
}
|
|
|
|
|
fn iter<'a>(&'a self, tx: &'a SessionTx<'_>) -> Result<TupleIter<'a>> {
|
|
|
|
|
todo!()
|
|
|
|
|
// let it = self.storage.scan_all(tx);
|
|
|
|
|
// Ok(if self.filters.is_empty() {
|
|
|
|
|
// Box::new(it)
|
|
|
|
|
// } else {
|
|
|
|
|
// Box::new(filter_iter(self.filters.clone(), it))
|
|
|
|
|
// })
|
|
|
|
|
let it = self.storage.skip_scan_all(tx, self.valid_at);
|
|
|
|
|
Ok(if self.filters.is_empty() {
|
|
|
|
|
Box::new(it)
|
|
|
|
|
} else {
|
|
|
|
|
Box::new(filter_iter(self.filters.clone(), it))
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
fn prefix_join<'a>(
|
|
|
|
|
&'a self,
|
|
|
|
|
@ -822,89 +835,81 @@ impl StoredWithValidityRA {
|
|
|
|
|
left_iter: TupleIter<'a>,
|
|
|
|
|
(left_join_indices, right_join_indices): (Vec<usize>, Vec<usize>),
|
|
|
|
|
eliminate_indices: BTreeSet<usize>,
|
|
|
|
|
left_tuple_len: usize,
|
|
|
|
|
) -> Result<TupleIter<'a>> {
|
|
|
|
|
todo!()
|
|
|
|
|
// let mut right_invert_indices = right_join_indices.iter().enumerate().collect_vec();
|
|
|
|
|
// right_invert_indices.sort_by_key(|(_, b)| **b);
|
|
|
|
|
// let left_to_prefix_indices = right_invert_indices
|
|
|
|
|
// .into_iter()
|
|
|
|
|
// .map(|(a, _)| left_join_indices[a])
|
|
|
|
|
// .collect_vec();
|
|
|
|
|
//
|
|
|
|
|
// let key_len = self.storage.metadata.keys.len();
|
|
|
|
|
// if left_to_prefix_indices.len() >= key_len {
|
|
|
|
|
// return self.point_lookup_join(
|
|
|
|
|
// tx,
|
|
|
|
|
// left_iter,
|
|
|
|
|
// key_len,
|
|
|
|
|
// left_to_prefix_indices,
|
|
|
|
|
// eliminate_indices,
|
|
|
|
|
// left_tuple_len,
|
|
|
|
|
// );
|
|
|
|
|
// }
|
|
|
|
|
//
|
|
|
|
|
// let mut skip_range_check = false;
|
|
|
|
|
// // In some cases, maybe we can stop as soon as we get one result?
|
|
|
|
|
// let it = left_iter
|
|
|
|
|
// .map_ok(move |tuple| {
|
|
|
|
|
// let prefix = left_to_prefix_indices
|
|
|
|
|
// .iter()
|
|
|
|
|
// .map(|i| tuple[*i].clone())
|
|
|
|
|
// .collect_vec();
|
|
|
|
|
//
|
|
|
|
|
// if !skip_range_check && !self.filters.is_empty() {
|
|
|
|
|
// let other_bindings = &self.bindings[right_join_indices.len()..];
|
|
|
|
|
// let (l_bound, u_bound) = match compute_bounds(&self.filters, other_bindings) {
|
|
|
|
|
// Ok(b) => b,
|
|
|
|
|
// _ => (vec![], vec![]),
|
|
|
|
|
// };
|
|
|
|
|
// if !l_bound.iter().all(|v| *v == DataValue::Null)
|
|
|
|
|
// || !u_bound.iter().all(|v| *v == DataValue::Bot)
|
|
|
|
|
// {
|
|
|
|
|
// return Left(
|
|
|
|
|
// self.storage
|
|
|
|
|
// .scan_bounded_prefix(tx, &prefix, &l_bound, &u_bound)
|
|
|
|
|
// .map(move |res_found| -> Result<Option<Tuple>> {
|
|
|
|
|
// let found = res_found?;
|
|
|
|
|
// for p in self.filters.iter() {
|
|
|
|
|
// if !p.eval_pred(&found)? {
|
|
|
|
|
// return Ok(None);
|
|
|
|
|
// }
|
|
|
|
|
// }
|
|
|
|
|
// let mut ret = tuple.clone();
|
|
|
|
|
// ret.extend(found);
|
|
|
|
|
// Ok(Some(ret))
|
|
|
|
|
// })
|
|
|
|
|
// .filter_map(swap_option_result),
|
|
|
|
|
// );
|
|
|
|
|
// }
|
|
|
|
|
// }
|
|
|
|
|
// skip_range_check = true;
|
|
|
|
|
// Right(
|
|
|
|
|
// self.storage
|
|
|
|
|
// .scan_prefix(tx, &prefix)
|
|
|
|
|
// .map(move |res_found| -> Result<Option<Tuple>> {
|
|
|
|
|
// let found = res_found?;
|
|
|
|
|
// for p in self.filters.iter() {
|
|
|
|
|
// if !p.eval_pred(&found)? {
|
|
|
|
|
// return Ok(None);
|
|
|
|
|
// }
|
|
|
|
|
// }
|
|
|
|
|
// let mut ret = tuple.clone();
|
|
|
|
|
// ret.extend(found);
|
|
|
|
|
// Ok(Some(ret))
|
|
|
|
|
// })
|
|
|
|
|
// .filter_map(swap_option_result),
|
|
|
|
|
// )
|
|
|
|
|
// })
|
|
|
|
|
// .flatten_ok()
|
|
|
|
|
// .map(flatten_err);
|
|
|
|
|
// Ok(if eliminate_indices.is_empty() {
|
|
|
|
|
// Box::new(it)
|
|
|
|
|
// } else {
|
|
|
|
|
// Box::new(it.map_ok(move |t| eliminate_from_tuple(t, &eliminate_indices)))
|
|
|
|
|
// })
|
|
|
|
|
let mut right_invert_indices = right_join_indices.iter().enumerate().collect_vec();
|
|
|
|
|
right_invert_indices.sort_by_key(|(_, b)| **b);
|
|
|
|
|
let left_to_prefix_indices = right_invert_indices
|
|
|
|
|
.into_iter()
|
|
|
|
|
.map(|(a, _)| left_join_indices[a])
|
|
|
|
|
.collect_vec();
|
|
|
|
|
|
|
|
|
|
let mut skip_range_check = false;
|
|
|
|
|
|
|
|
|
|
let it = left_iter
|
|
|
|
|
.map_ok(move |tuple| {
|
|
|
|
|
let prefix = left_to_prefix_indices
|
|
|
|
|
.iter()
|
|
|
|
|
.map(|i| tuple[*i].clone())
|
|
|
|
|
.collect_vec();
|
|
|
|
|
|
|
|
|
|
if !skip_range_check && !self.filters.is_empty() {
|
|
|
|
|
let other_bindings = &self.bindings[right_join_indices.len()..];
|
|
|
|
|
let (l_bound, u_bound) = match compute_bounds(&self.filters, other_bindings) {
|
|
|
|
|
Ok(b) => b,
|
|
|
|
|
_ => (vec![], vec![]),
|
|
|
|
|
};
|
|
|
|
|
if !l_bound.iter().all(|v| *v == DataValue::Null)
|
|
|
|
|
|| !u_bound.iter().all(|v| *v == DataValue::Bot)
|
|
|
|
|
{
|
|
|
|
|
return Left(
|
|
|
|
|
self.storage
|
|
|
|
|
.skip_scan_bounded_prefix(
|
|
|
|
|
tx,
|
|
|
|
|
&prefix,
|
|
|
|
|
&l_bound,
|
|
|
|
|
&u_bound,
|
|
|
|
|
self.valid_at,
|
|
|
|
|
)
|
|
|
|
|
.map(move |res_found| -> Result<Option<Tuple>> {
|
|
|
|
|
let found = res_found?;
|
|
|
|
|
for p in self.filters.iter() {
|
|
|
|
|
if !p.eval_pred(&found)? {
|
|
|
|
|
return Ok(None);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
let mut ret = tuple.clone();
|
|
|
|
|
ret.extend(found);
|
|
|
|
|
Ok(Some(ret))
|
|
|
|
|
})
|
|
|
|
|
.filter_map(swap_option_result),
|
|
|
|
|
);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
skip_range_check = true;
|
|
|
|
|
Right(
|
|
|
|
|
self.storage
|
|
|
|
|
.skip_scan_prefix(tx, &prefix, self.valid_at)
|
|
|
|
|
.map(move |res_found| -> Result<Option<Tuple>> {
|
|
|
|
|
let found = res_found?;
|
|
|
|
|
for p in self.filters.iter() {
|
|
|
|
|
if !p.eval_pred(&found)? {
|
|
|
|
|
return Ok(None);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
let mut ret = tuple.clone();
|
|
|
|
|
ret.extend(found);
|
|
|
|
|
Ok(Some(ret))
|
|
|
|
|
})
|
|
|
|
|
.filter_map(swap_option_result),
|
|
|
|
|
)
|
|
|
|
|
})
|
|
|
|
|
.flatten_ok()
|
|
|
|
|
.map(flatten_err);
|
|
|
|
|
Ok(if eliminate_indices.is_empty() {
|
|
|
|
|
Box::new(it)
|
|
|
|
|
} else {
|
|
|
|
|
Box::new(it.map_ok(move |t| eliminate_from_tuple(t, &eliminate_indices)))
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -1856,13 +1861,11 @@ impl InnerJoin {
|
|
|
|
|
)
|
|
|
|
|
.unwrap();
|
|
|
|
|
if join_is_prefix(&join_indices.1) {
|
|
|
|
|
let left_len = self.left.bindings_after_eliminate().len();
|
|
|
|
|
r.prefix_join(
|
|
|
|
|
tx,
|
|
|
|
|
self.left.iter(tx, delta_rule, stores)?,
|
|
|
|
|
join_indices,
|
|
|
|
|
eliminate_indices,
|
|
|
|
|
left_len,
|
|
|
|
|
)
|
|
|
|
|
} else {
|
|
|
|
|
self.materialized_join(tx, eliminate_indices, delta_rule, stores)
|
|
|
|
|
|
Ziyang Hu
2 years ago