pub mod constantconstraint;
pub mod equalityconstraint;
pub mod frontier;
#[doc(hidden)]
pub mod hashmapconstraint;
pub mod hashsetconstraint;
pub mod intersectionconstraint;
mod liveness;
mod ordinal;
pub mod patchconstraint;
#[doc(hidden)]
pub mod rangeconstraint;
pub mod register;
pub mod sortedsliceconstraint;
pub mod unionconstraint;
mod variableset;
use std::cmp::Reverse;
use std::fmt;
use std::marker::PhantomData;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use arrayvec::ArrayVec;
use constantconstraint::*;
use ordinal::RowSelection;
use crate::inline::encodings::genid::GenId;
use crate::inline::Inline;
use crate::inline::InlineEncoding;
use crate::inline::RawInline;
pub use variableset::VariableSet;
pub use liveness::{and_words, or_words, Candidates, ProposalBuffer, LIVENESS_WORD_BITS};
impl<T: InlineEncoding> Term<T> {
pub fn erase(self) -> RawTerm {
match self {
Term::Var(v) => RawTerm::Var(v.index),
Term::Const(c) => RawTerm::Const(c.raw),
}
}
}
pub trait TriblePattern {
type PatternConstraint<'a>: Constraint<'a> + Send + Sync
where
Self: 'a;
fn pattern<'a, V: InlineEncoding>(
&'a self,
e: impl Into<Term<GenId>>,
a: impl Into<Term<GenId>>,
v: impl Into<Term<V>>,
) -> Self::PatternConstraint<'a>;
}
pub type VariableId = usize;
#[derive(Debug)]
pub struct VariableContext {
pub next_index: VariableId,
}
impl Default for VariableContext {
fn default() -> Self {
Self::new()
}
}
impl VariableContext {
pub fn new() -> Self {
VariableContext { next_index: 0 }
}
pub fn next_variable<T: InlineEncoding>(&mut self) -> Variable<T> {
assert!(
self.next_index < 128,
"currently queries support at most 128 variables"
);
let v = Variable::new(self.next_index);
self.next_index += 1;
v
}
}
#[derive(Debug)]
pub struct Variable<T: InlineEncoding> {
pub index: VariableId,
typed: PhantomData<T>,
}
impl<T: InlineEncoding> Copy for Variable<T> {}
impl<T: InlineEncoding> Clone for Variable<T> {
fn clone(&self) -> Self {
*self
}
}
impl<T: InlineEncoding> Variable<T> {
pub fn new(index: VariableId) -> Self {
Variable {
index,
typed: PhantomData,
}
}
pub fn extract<'b>(self, binding: &Binding<'b>) -> &'b Inline<T> {
let raw = binding.get(self.index).unwrap_or_else(|| {
panic!(
"query variable (idx {}) was never bound before projection. This usually means the variable was projected in `find!` but never appeared in any constraint. If you intended a pure existence query, use `find!((), ...)` or `exists!(constraint)`.",
self.index
)
});
Inline::as_transmute_raw(raw)
}
}
#[derive(Debug)]
pub enum Term<T: InlineEncoding> {
Var(Variable<T>),
Const(Inline<T>),
}
impl<T: InlineEncoding> Copy for Term<T> {}
impl<T: InlineEncoding> Clone for Term<T> {
fn clone(&self) -> Self {
*self
}
}
impl<T: InlineEncoding> From<Variable<T>> for Term<T> {
fn from(v: Variable<T>) -> Self {
Term::Var(v)
}
}
impl<T: InlineEncoding> From<Inline<T>> for Term<T> {
fn from(value: Inline<T>) -> Self {
Term::Const(value)
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RawTerm {
Var(VariableId),
Const(RawInline),
}
impl RawTerm {
#[inline]
pub fn is_var(&self, variable: VariableId) -> bool {
matches!(self, RawTerm::Var(v) if *v == variable)
}
#[inline]
pub fn position_value<'b>(&'b self, binding: &'b Binding) -> Option<&'b RawInline> {
match self {
RawTerm::Var(v) => binding.get(*v),
RawTerm::Const(c) => Some(c),
}
}
#[inline]
pub fn add_to(&self, set: &mut VariableSet) {
if let RawTerm::Var(v) = self {
set.set(*v);
}
}
}
pub trait ContainsConstraint<'a, T: InlineEncoding> {
type Constraint: Constraint<'a>;
fn has(self, v: Variable<T>) -> Self::Constraint;
}
impl<T: InlineEncoding> Variable<T> {
pub fn is(self, constant: Inline<T>) -> ConstantConstraint {
ConstantConstraint::new(self, constant)
}
}
#[derive(Clone, Copy)]
pub struct Binding<'a> {
pub bound: VariableSet,
indexes: &'a [u32],
levels: &'a [LevelValues],
}
impl fmt::Debug for Binding<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_map()
.entries(self.bound.into_iter().map(|v| (v, self.get(v))))
.finish()
}
}
impl<'a> Binding<'a> {
pub fn get(&self, variable: VariableId) -> Option<&'a RawInline> {
if self.bound.is_set(variable) {
Some(&self.levels[variable].buffer()[self.indexes[variable] as usize])
} else {
None
}
}
}
static NO_LEVELS: [LevelValues; 128] = [const { LevelValues::empty() }; 128];
static NO_INDEXES: [u32; 128] = [0; 128];
impl Default for Binding<'_> {
fn default() -> Self {
Binding {
bound: VariableSet::new_empty(),
indexes: &NO_INDEXES,
levels: &NO_LEVELS,
}
}
}
#[derive(Clone, Copy)]
pub struct Frontier<'a> {
bound: VariableSet,
block: &'a [u32],
stride: usize,
select: RowSelection<'a>,
levels: &'a [LevelValues],
parallel: bool,
}
impl fmt::Debug for Frontier<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_list().entries(self.rows()).finish()
}
}
impl Default for Frontier<'_> {
fn default() -> Self {
Frontier {
bound: VariableSet::new_empty(),
block: &NO_INDEXES,
stride: NO_INDEXES.len(),
select: RowSelection::identity(1),
levels: &NO_LEVELS,
parallel: false,
}
}
}
impl<'a> Frontier<'a> {
pub fn len(&self) -> usize {
self.select.len()
}
pub fn is_empty(&self) -> bool {
self.select.len() == 0
}
pub fn bound(&self) -> VariableSet {
self.bound
}
#[cfg(feature = "parallel")]
pub(crate) fn parallel(&self) -> bool {
self.parallel
}
pub fn row(&self, i: usize) -> Binding<'a> {
Binding {
bound: self.bound,
indexes: self.row_indexes(i),
levels: self.levels,
}
}
pub fn rows(&self) -> impl Iterator<Item = Binding<'a>> + '_ {
(0..self.len()).map(|i| self.row(i))
}
fn row_indexes(&self, i: usize) -> &'a [u32] {
let row = self.select.get(i).expect("frontier row index is in bounds") as usize;
&self.block[row * self.stride..(row + 1) * self.stride]
}
pub fn compose(&self, positions: impl IntoIterator<Item = u32>, out: &mut Vec<u32>) {
out.extend(positions.into_iter().map(|position| {
self.select
.get(position as usize)
.expect("frontier composition index is in bounds")
}));
}
pub fn with_select<'b>(&self, select: &'b [u32]) -> Frontier<'b>
where
'a: 'b,
{
Frontier {
bound: self.bound,
block: self.block,
stride: self.stride,
select: RowSelection::explicit(select),
levels: self.levels,
parallel: self.parallel,
}
}
}
#[derive(Clone)]
pub struct BindingStore {
bound: VariableSet,
indexes: [u32; 128],
levels: [LevelValues; 128],
}
impl fmt::Debug for BindingStore {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt::Debug::fmt(&self.view(), f)
}
}
impl Default for BindingStore {
fn default() -> Self {
BindingStore {
bound: VariableSet::new_empty(),
indexes: [0; 128],
levels: std::array::from_fn(|_| LevelValues::default()),
}
}
}
impl BindingStore {
pub fn new() -> Self {
Self::default()
}
pub fn view(&self) -> Binding<'_> {
Binding {
bound: self.bound,
indexes: &self.indexes,
levels: &self.levels,
}
}
pub fn frontier(&self) -> Frontier<'_> {
Frontier {
bound: self.bound,
block: &self.indexes,
stride: self.indexes.len(),
select: RowSelection::identity(1),
levels: &self.levels,
parallel: false,
}
}
pub fn bind(&mut self, variable: VariableId, value: &RawInline) {
let level = &mut self.levels[variable];
let buffer = Arc::make_mut(
level
.buffer
.get_or_insert_with(|| Arc::new(ProposalBuffer::new())),
);
let index = buffer.len();
buffer.push(*value);
self.indexes[variable] = index as u32;
self.bound.set(variable);
}
pub fn unset(&mut self, variable: VariableId) {
self.bound.unset(variable);
}
pub fn bound(&self) -> VariableSet {
self.bound
}
pub(crate) fn refill(
&mut self,
variable: VariableId,
block: &[u32],
select: RowSelection<'_>,
stride: usize,
parallel: bool,
propose: impl FnOnce(&Frontier<'_>, &mut ProposalBuffer),
) -> usize {
debug_assert!(
!self.bound.is_set(variable),
"refilling a bound variable's level would strand its binding"
);
let mut level = std::mem::take(&mut self.levels[variable]);
let mut buffer = level
.buffer
.take()
.unwrap_or_else(|| Arc::new(ProposalBuffer::new()));
if Arc::get_mut(&mut buffer).is_none() {
buffer = Arc::new(ProposalBuffer::new());
}
level.pos = 0;
let proposed = {
let buffer = Arc::get_mut(&mut buffer).expect("sole owner after replacement");
buffer.clear();
propose(
&Frontier {
bound: self.bound,
block,
stride,
select,
levels: &self.levels,
parallel,
},
buffer,
);
buffer.len()
};
level.buffer = Some(buffer);
self.levels[variable] = level;
proposed
}
pub(crate) fn batch<'s>(
&'s self,
block: &'s [u32],
select: RowSelection<'s>,
stride: usize,
parallel: bool,
) -> Frontier<'s> {
Frontier {
bound: self.bound,
block,
stride,
select,
levels: &self.levels,
parallel,
}
}
pub(crate) fn consumed(&self, variable: VariableId) -> usize {
self.levels[variable].pos
}
pub(crate) fn take_chunk(
&mut self,
variable: VariableId,
width: usize,
parent_block: &[u32],
parent_select: RowSelection<'_>,
stride: usize,
out: &mut Vec<u32>,
parents_out: &mut Vec<u32>,
) -> usize {
out.clear();
parents_out.clear();
let level = &mut self.levels[variable];
let buffer = level
.buffer
.as_deref()
.expect("an active level must have a proposal buffer");
let pos = &mut level.pos;
let mut rows = 0;
while rows < width {
let Some(i) = buffer.next_live(*pos) else {
break;
};
*pos = i + 1;
let parent = parent_select
.get(buffer.parent_of(i) as usize)
.expect("candidate parent tag is in bounds") as usize;
out.extend_from_slice(&parent_block[parent * stride..(parent + 1) * stride]);
let base = out.len() - stride;
out[base + variable] = i as u32;
parents_out.push(parent as u32);
rows += 1;
}
self.bound.set_value(variable, rows != 0);
rows
}
pub(crate) fn draw(
&mut self,
variable: VariableId,
width: usize,
parent_select: RowSelection<'_>,
drawn_out: &mut Vec<u32>,
parents_out: &mut Vec<u32>,
) {
drawn_out.clear();
parents_out.clear();
let level = &mut self.levels[variable];
let buffer = level
.buffer
.as_deref()
.expect("an active level must have a proposal buffer");
let pos = &mut level.pos;
while drawn_out.len() < width {
let Some(i) = buffer.next_live(*pos) else {
break;
};
*pos = i + 1;
drawn_out.push(i as u32);
parents_out.push(
parent_select
.get(buffer.parent_of(i) as usize)
.expect("candidate parent tag is in bounds"),
);
}
self.bound.set_value(variable, !drawn_out.is_empty());
}
pub(crate) fn spent(&self, variable: VariableId) -> bool {
let level = &self.levels[variable];
level.buffer().next_live(level.pos).is_none()
}
}
pub trait Constraint<'a> {
fn variables(&self) -> VariableSet;
fn estimate(&self, variable: VariableId, binding: &Binding) -> Option<usize>;
fn propose(
&self,
variable: VariableId,
frontier: &Frontier<'_>,
proposals: &mut ProposalBuffer,
);
fn confirm(&self, variable: VariableId, frontier: &Frontier<'_>, cands: &mut Candidates<'_>);
fn satisfied(&self, _binding: &Binding) -> bool {
true
}
fn influence(&self, variable: VariableId) -> VariableSet {
let mut vars = self.variables();
if vars.is_set(variable) {
vars.unset(variable);
vars
} else {
VariableSet::new_empty()
}
}
}
impl<'a, T: Constraint<'a> + ?Sized> Constraint<'a> for Box<T> {
fn variables(&self) -> VariableSet {
let inner: &T = self;
inner.variables()
}
fn estimate(&self, variable: VariableId, binding: &Binding) -> Option<usize> {
let inner: &T = self;
inner.estimate(variable, binding)
}
fn propose(
&self,
variable: VariableId,
frontier: &Frontier<'_>,
proposals: &mut ProposalBuffer,
) {
let inner: &T = self;
inner.propose(variable, frontier, proposals)
}
fn confirm(&self, variable: VariableId, frontier: &Frontier<'_>, cands: &mut Candidates<'_>) {
let inner: &T = self;
inner.confirm(variable, frontier, cands)
}
fn satisfied(&self, binding: &Binding) -> bool {
let inner: &T = self;
inner.satisfied(binding)
}
fn influence(&self, variable: VariableId) -> VariableSet {
let inner: &T = self;
inner.influence(variable)
}
}
impl<'a, T: Constraint<'a> + ?Sized> Constraint<'a> for std::sync::Arc<T> {
fn variables(&self) -> VariableSet {
let inner: &T = self;
inner.variables()
}
fn estimate(&self, variable: VariableId, binding: &Binding) -> Option<usize> {
let inner: &T = self;
inner.estimate(variable, binding)
}
fn propose(
&self,
variable: VariableId,
frontier: &Frontier<'_>,
proposals: &mut ProposalBuffer,
) {
let inner: &T = self;
inner.propose(variable, frontier, proposals)
}
fn confirm(&self, variable: VariableId, frontier: &Frontier<'_>, cands: &mut Candidates<'_>) {
let inner: &T = self;
inner.confirm(variable, frontier, cands)
}
fn satisfied(&self, binding: &Binding) -> bool {
let inner: &T = self;
inner.satisfied(binding)
}
fn influence(&self, variable: VariableId) -> VariableSet {
let inner: &T = self;
inner.influence(variable)
}
}
#[derive(Clone, Debug)]
struct LevelValues {
buffer: Option<Arc<ProposalBuffer>>,
pos: usize,
}
impl LevelValues {
const fn empty() -> Self {
LevelValues {
buffer: None,
pos: 0,
}
}
fn buffer(&self) -> &ProposalBuffer {
self.buffer
.as_deref()
.expect("a bound or active level must have a proposal buffer")
}
}
impl Default for LevelValues {
fn default() -> Self {
Self::empty()
}
}
pub const DEFAULT_FRONTIER_WIDTH: usize = 16384;
pub const INITIAL_FRONTIER_WIDTH: usize = 1;
pub const FRONTIER_RAMP_BASE: usize = 8;
pub struct Query<C, P: Fn(&Binding<'_>) -> Option<R>, R> {
constraint: C,
postprocessing: P,
mode: Search,
bindings: BindingStore,
influences: [VariableSet; 128],
unbound: VariableSet,
slots: usize,
width: usize,
parallel: bool,
stack: ArrayVec<Level, 128>,
depths: Vec<Depth>,
depth: usize,
choice: Vec<u32>,
drawn: Vec<u32>,
parents: Vec<u32>,
stats: Arc<FrontierStats>,
}
#[derive(Clone, Copy, Debug)]
struct Level {
variable: VariableId,
width: usize,
proposed: usize,
}
fn reset_shared<T: Clone>(slot: &mut Arc<Vec<T>>, capacity: usize) -> &mut Vec<T> {
if Arc::get_mut(slot).is_none() {
*slot = Arc::new(Vec::with_capacity(capacity));
}
let buffer = Arc::get_mut(slot).expect("sole owner after replacement");
buffer.clear();
buffer.reserve(capacity);
buffer
}
#[derive(Clone, Debug, Default)]
enum DepthPlan {
#[default]
Unplanned,
Universal(VariableId),
Split(Arc<SplitPlan>),
}
#[derive(Debug, Default)]
struct SplitPlan {
order_override: Vec<u32>,
groups: Vec<(VariableId, usize)>,
}
impl DepthPlan {
fn selection(&self, rows: usize) -> RowSelection<'_> {
match self {
Self::Split(plan) if !plan.order_override.is_empty() => {
debug_assert_eq!(plan.order_override.len(), rows);
RowSelection::explicit(&plan.order_override)
}
Self::Unplanned | Self::Universal(_) | Self::Split(_) => RowSelection::identity(rows),
}
}
fn group_count(&self) -> usize {
match self {
Self::Unplanned => 0,
Self::Universal(_) => 1,
Self::Split(plan) => plan.groups.len(),
}
}
fn group(&self, group: usize, rows: usize) -> (VariableId, std::ops::Range<usize>) {
match self {
Self::Unplanned => panic!("an unplanned frontier has no variable groups"),
Self::Universal(variable) => {
assert_eq!(group, 0, "universal plan has exactly one group");
(*variable, 0..rows)
}
Self::Split(plan) => {
let start = if group == 0 {
0
} else {
plan.groups[group - 1].1
};
let (variable, end) = plan.groups[group];
(variable, start..end)
}
}
}
}
#[derive(Clone, Debug, Default)]
struct Depth {
block: Arc<Vec<u32>>,
rows: usize,
estimates: Arc<Vec<usize>>,
plan: DepthPlan,
group: usize,
group_limit: usize,
emit: usize,
}
impl Depth {
fn selection(&self) -> RowSelection<'_> {
self.plan.selection(self.rows)
}
fn group(&self, group: usize) -> (VariableId, std::ops::Range<usize>) {
self.plan.group(group, self.rows)
}
}
#[derive(Debug, Default)]
pub struct FrontierStats {
expansions: AtomicU64,
rows: AtomicU64,
variable_groups: AtomicU64,
proposals: AtomicU64,
peak_region: AtomicU64,
widest: AtomicU64,
inplace_descents: AtomicU64,
copied_descents: AtomicU64,
}
impl FrontierStats {
pub fn expansions(&self) -> u64 {
self.expansions.load(Ordering::Relaxed)
}
pub fn rows(&self) -> u64 {
self.rows.load(Ordering::Relaxed)
}
pub fn variable_groups(&self) -> u64 {
self.variable_groups.load(Ordering::Relaxed)
}
pub fn proposals(&self) -> u64 {
self.proposals.load(Ordering::Relaxed)
}
pub fn peak_region(&self) -> u64 {
self.peak_region.load(Ordering::Relaxed)
}
pub fn widest(&self) -> u64 {
self.widest.load(Ordering::Relaxed)
}
pub fn inplace_descents(&self) -> u64 {
self.inplace_descents.load(Ordering::Relaxed)
}
pub fn copied_descents(&self) -> u64 {
self.copied_descents.load(Ordering::Relaxed)
}
pub fn mean_width(&self) -> f64 {
let expansions = self.expansions();
if expansions == 0 {
0.0
} else {
self.rows() as f64 / expansions as f64
}
}
pub fn mean_variable_groups(&self) -> f64 {
let expansions = self.expansions();
if expansions == 0 {
0.0
} else {
self.variable_groups() as f64 / expansions as f64
}
}
}
#[cfg(feature = "parallel")]
impl<C, P, R> Clone for Query<C, P, R>
where
C: Clone,
P: Fn(&Binding<'_>) -> Option<R> + Clone,
{
fn clone(&self) -> Self {
Self {
constraint: self.constraint.clone(),
postprocessing: self.postprocessing.clone(),
mode: self.mode,
bindings: self.bindings.clone(),
influences: self.influences,
unbound: self.unbound,
slots: self.slots,
width: self.width,
parallel: self.parallel,
stack: self.stack.clone(),
depths: self.depths[..=self.depth].to_vec(),
depth: self.depth,
choice: Vec::new(),
drawn: Vec::new(),
parents: Vec::new(),
stats: Arc::clone(&self.stats),
}
}
}
#[cfg(feature = "parallel")]
impl<C, P, R> Query<C, P, R>
where
C: Clone,
P: Fn(&Binding<'_>) -> Option<R> + Clone,
{
fn has_ancestor_continuation(&self) -> bool {
(0..self.depth).any(|depth| {
let frontier = &self.depths[depth];
frontier.group < frontier.group_limit
|| self
.stack
.get(depth)
.is_some_and(|level| level.proposed > self.bindings.consumed(level.variable))
})
}
fn split_pending_group(&mut self) -> Option<Self> {
debug_assert_eq!(self.mode, Search::NextGroup);
let depth = self.depth;
let current = self.depths[depth].group;
let group_limit = self.depths[depth].group_limit;
let later_group = current + 1 < group_limit;
if current >= group_limit || !(later_group || self.has_ancestor_continuation()) {
return None;
}
let mut right = self.clone();
self.depths[depth].group += 1;
drop(right.stack.drain(..depth));
drop(right.depths.drain(..depth));
right.depth = 0;
let root = &mut right.depths[0];
root.group_limit = root.group + 1;
Some(right)
}
fn split_terminal_frontier(&mut self) -> Option<Self> {
debug_assert_eq!(self.mode, Search::Emit);
let depth = self.depth;
if depth == 0
|| self.depths[depth].emit >= self.depths[depth].rows
|| !self.has_ancestor_continuation()
{
return None;
}
let mut right = self.clone();
self.depth -= 1;
self.mode = Search::NextChunk;
drop(right.stack.drain(..depth));
drop(right.depths.drain(..depth));
right.depth = 0;
Some(right)
}
}
impl<'a, C: Constraint<'a>, P: Fn(&Binding<'_>) -> Option<R>, R> Query<C, P, R> {
pub fn new(constraint: C, postprocessing: P) -> Self {
let variables = constraint.variables();
let influences = std::array::from_fn(|v| {
if variables.is_set(v) {
constraint.influence(v)
} else {
VariableSet::new_empty()
}
});
let bindings = BindingStore::new();
let slots = variables.find_last_set().map(|v| v + 1).unwrap_or(0);
let estimates: Vec<usize> = (0..slots)
.map(|v| {
if variables.is_set(v) {
constraint
.estimate(v, &bindings.view())
.expect("unconstrained variable in query")
} else {
usize::MAX
}
})
.collect();
let mode = if constraint.satisfied(&bindings.view()) {
Search::Plan
} else {
Search::Done
};
let root = Depth {
block: Arc::new(vec![0u32; slots]),
rows: 1,
estimates: Arc::new(estimates),
plan: DepthPlan::Unplanned,
group: 0,
group_limit: 0,
emit: 0,
};
Query {
constraint,
postprocessing,
mode,
bindings,
influences,
unbound: variables,
slots,
width: DEFAULT_FRONTIER_WIDTH,
parallel: false,
stack: ArrayVec::new(),
depths: vec![root],
depth: 0,
choice: Vec::new(),
drawn: Vec::new(),
parents: Vec::new(),
stats: Arc::new(FrontierStats::default()),
}
}
pub fn with_frontier_width(mut self, width: usize) -> Self {
assert!(width > 0, "frontier width must be at least one binding");
self.width = width;
self
}
pub fn stats(&self) -> Arc<FrontierStats> {
Arc::clone(&self.stats)
}
fn frontier(&self) -> Frontier<'_> {
let depth = &self.depths[self.depth];
self.bindings
.batch(&depth.block, depth.selection(), self.slots, self.parallel)
}
fn plan(&mut self) {
if self.unbound.is_empty() {
self.depths[self.depth].emit = 0;
self.mode = Search::Emit;
return;
}
let slots = self.slots;
let influences = &self.influences;
let unbound = self.unbound;
let depth = &mut self.depths[self.depth];
let rows = depth.rows;
assert!(rows != 0, "the engine never plans an empty frontier");
u32::try_from(rows - 1).expect("frontier row ordinal exceeds u32");
let preferred = |row: usize| {
let row_estimates = &depth.estimates[row * slots..(row + 1) * slots];
unbound
.into_iter()
.max_by_key(|&v| {
(
Reverse(
row_estimates[v]
.checked_ilog2()
.map(|magnitude| magnitude + 1)
.unwrap_or(0),
),
influences[v].count(),
)
})
.expect("non-empty unbound")
};
self.choice.clear();
let first = preferred(0);
let disagreement = (1..rows).find_map(|row| {
let variable = preferred(row);
(variable != first).then_some((row, variable))
});
if let Some((row, variable)) = disagreement {
self.choice.resize(
row,
u32::try_from(first).expect("query variables fit in u32"),
);
self.choice
.push(u32::try_from(variable).expect("query variables fit in u32"));
for row in row + 1..rows {
self.choice
.push(u32::try_from(preferred(row)).expect("query variables fit in u32"));
}
let mut counts = [0usize; 128];
for &variable in &self.choice {
counts[variable as usize] += 1;
}
let mut starts = [0usize; 128];
let mut split = SplitPlan::default();
let mut offset = 0;
for variable in 0..slots {
starts[variable] = offset;
if counts[variable] != 0 {
offset += counts[variable];
split.groups.push((variable, offset));
}
}
debug_assert_eq!(offset, rows);
debug_assert!(split.groups.len() > 1);
let identity = self.choice.windows(2).all(|pair| pair[0] <= pair[1]);
if !identity {
split.order_override.resize(rows, 0);
for (row, &variable) in self.choice.iter().enumerate() {
split.order_override[starts[variable as usize]] =
u32::try_from(row).expect("frontier row ordinal exceeds u32");
starts[variable as usize] += 1;
}
}
depth.plan = DepthPlan::Split(Arc::new(split));
} else {
depth.plan = DepthPlan::Universal(first);
}
depth.group = 0;
depth.group_limit = depth.plan.group_count();
self.stats.expansions.fetch_add(1, Ordering::Relaxed);
self.stats.rows.fetch_add(rows as u64, Ordering::Relaxed);
self.stats.widest.fetch_max(rows as u64, Ordering::Relaxed);
self.stats.variable_groups.fetch_add(
self.depths[self.depth].group_limit as u64,
Ordering::Relaxed,
);
self.mode = Search::NextGroup;
}
fn next_group(&mut self) {
let depth = &self.depths[self.depth];
if depth.group >= depth.group_limit {
if self.depth == 0 {
self.mode = Search::Done;
} else {
self.depth -= 1;
self.mode = Search::NextChunk;
}
return;
}
let group = depth.group;
let (variable, range) = depth.group(group);
self.depths[self.depth].group += 1;
self.unbound.unset(variable);
let constraint = &self.constraint;
let depth = &self.depths[self.depth];
let block = Arc::clone(&depth.block);
let selection = depth.selection().slice(range);
let proposed = self.bindings.refill(
variable,
&block,
selection,
self.slots,
self.parallel,
|frontier, proposals| constraint.propose(variable, frontier, proposals),
);
self.stack.push(Level {
variable,
width: INITIAL_FRONTIER_WIDTH.min(self.width),
proposed,
});
self.stats
.proposals
.fetch_add(proposed as u64, Ordering::Relaxed);
self.stats
.peak_region
.fetch_max(proposed as u64, Ordering::Relaxed);
self.mode = Search::NextChunk;
}
fn next_chunk(&mut self) {
let level = *self.stack.last().expect("a level to chunk");
let variable = level.variable;
let parent = self.depth;
let group = self.depths[parent].group - 1;
let (_, range) = self.depths[parent].group(group);
let slots = self.slots;
if self.depths.len() <= parent + 1 {
self.depths.push(Depth::default());
}
let source_rows = range.len();
let speculate = level.proposed == source_rows
&& level.width >= source_rows
&& self.depths[parent].group >= self.depths[parent].group_limit;
let rows = if speculate {
let selection = self.depths[parent].selection().slice(range.clone());
self.bindings.draw(
variable,
level.width,
selection,
&mut self.drawn,
&mut self.parents,
);
self.drawn.len()
} else {
let (head, tail) = self.depths.split_at_mut(parent + 1);
let selection = head[parent].selection().slice(range);
self.bindings.take_chunk(
variable,
level.width,
&head[parent].block,
selection,
slots,
Arc::make_mut(&mut tail[0].block),
&mut self.parents,
)
};
let consumed = self.bindings.consumed(variable);
if let Some(top) = self.stack.last_mut() {
let next = top.width.saturating_mul(FRONTIER_RAMP_BASE).min(self.width);
let remaining = top.proposed.saturating_sub(consumed);
top.width = if remaining < next.saturating_mul(2) {
remaining.max(next).min(self.width)
} else {
next
};
}
if rows == 0 {
self.stack.pop();
self.bindings.unset(variable);
self.unbound.set(variable);
self.mode = Search::NextGroup;
return;
}
let reusable = speculate
&& rows == source_rows
&& self
.parents
.iter()
.enumerate()
.all(|(row, &parent_row)| parent_row as usize == row)
&& self.bindings.spent(variable);
let (head, tail) = self.depths.split_at_mut(parent + 1);
let source = &mut head[parent];
let child = &mut tail[0];
let inplace = reusable
&& Arc::get_mut(&mut source.block).is_some()
&& Arc::get_mut(&mut source.estimates).is_some();
if inplace {
std::mem::swap(&mut source.block, &mut child.block);
std::mem::swap(&mut source.estimates, &mut child.estimates);
let block = Arc::get_mut(&mut child.block).expect("unique across the swap");
for (row, &entry) in self.drawn.iter().enumerate() {
block[row * slots + variable] = entry;
}
self.stats.inplace_descents.fetch_add(1, Ordering::Relaxed);
} else {
if speculate {
let block = reset_shared(&mut child.block, rows * slots);
for (&entry, &parent_row) in self.drawn.iter().zip(self.parents.iter()) {
let parent_row = parent_row as usize;
block.extend_from_slice(
&source.block[parent_row * slots..(parent_row + 1) * slots],
);
let base = block.len() - slots;
block[base + variable] = entry;
}
}
let estimates = reset_shared(&mut child.estimates, rows * slots);
for &parent_row in self.parents.iter() {
let parent_row = parent_row as usize;
estimates.extend_from_slice(
&source.estimates[parent_row * slots..(parent_row + 1) * slots],
);
}
self.stats.copied_descents.fetch_add(1, Ordering::Relaxed);
}
child.rows = rows;
child.plan = DepthPlan::Unplanned;
child.group = 0;
child.group_limit = 0;
child.emit = 0;
let stale = self.influences[variable].intersect(self.unbound);
if !stale.is_empty() {
let child = &mut self.depths[parent + 1];
let block = Arc::clone(&child.block);
let estimates = Arc::make_mut(&mut child.estimates);
let batch =
self.bindings
.batch(&block, RowSelection::identity(rows), slots, self.parallel);
for row in 0..rows {
let binding = batch.row(row);
for v in stale {
estimates[row * slots + v] = self
.constraint
.estimate(v, &binding)
.expect("unconstrained variable in query");
}
}
}
self.depth = parent + 1;
self.mode = Search::Plan;
}
}
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
enum Search {
Plan,
NextGroup,
NextChunk,
Emit,
Done,
}
impl<'a, C: Constraint<'a>, P: Fn(&Binding<'_>) -> Option<R>, R> Iterator for Query<C, P, R> {
type Item = R;
fn next(&mut self) -> Option<Self::Item> {
loop {
match self.mode {
Search::Plan => self.plan(),
Search::NextGroup => self.next_group(),
Search::NextChunk => self.next_chunk(),
Search::Emit => {
let frontier = self.frontier();
let rows = frontier.len();
let mut emit = self.depths[self.depth].emit;
let mut result = None;
while emit < rows {
let binding = frontier.row(emit);
emit += 1;
if let Some(row) = (self.postprocessing)(&binding) {
result = Some(row);
break;
}
}
self.depths[self.depth].emit = emit;
if result.is_some() {
return result;
}
if self.depth == 0 {
self.mode = Search::Done;
} else {
self.depth -= 1;
self.mode = Search::NextChunk;
}
}
Search::Done => return None,
}
}
}
}
impl<'a, C: Constraint<'a>, P: Fn(&Binding<'_>) -> Option<R>, R> fmt::Debug for Query<C, P, R> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Query")
.field("constraint", &std::any::type_name::<C>())
.field("mode", &self.mode)
.field("width", &self.width)
.field("frontier", &self.frontier())
.field("stack", &self.stack)
.field("unbound", &self.unbound)
.finish()
}
}
#[cfg(feature = "parallel")]
pub use parallel::QueryParIter;
#[cfg(feature = "parallel")]
mod parallel {
use super::*;
use rayon::iter::plumbing::{bridge_unindexed, Folder, UnindexedConsumer, UnindexedProducer};
use rayon::iter::{IntoParallelIterator, ParallelIterator};
pub struct QueryParIter<C, P: Fn(&Binding<'_>) -> Option<R>, R> {
inner: Box<Query<C, P, R>>,
}
impl<'a, C, P, R> IntoParallelIterator for Query<C, P, R>
where
C: Constraint<'a> + Clone + Send + 'a,
P: Fn(&Binding<'_>) -> Option<R> + Clone + Send,
R: Send,
{
type Item = R;
type Iter = QueryParIter<C, P, R>;
fn into_par_iter(mut self) -> Self::Iter {
self.parallel = true;
QueryParIter {
inner: Box::new(self),
}
}
}
impl<'a, C, P, R> UnindexedProducer for QueryParIter<C, P, R>
where
C: Constraint<'a> + Clone + Send + 'a,
P: Fn(&Binding<'_>) -> Option<R> + Clone + Send,
R: Send,
{
type Item = R;
fn split(mut self) -> (Self, Option<Self>) {
let q = &mut *self.inner;
loop {
match q.mode {
Search::Plan => q.plan(),
Search::NextGroup => {
if let Some(right) = q.split_pending_group() {
return (
self,
Some(QueryParIter {
inner: Box::new(right),
}),
);
}
q.next_group();
}
Search::Emit => {
if let Some(right) = q.split_terminal_frontier() {
return (
self,
Some(QueryParIter {
inner: Box::new(right),
}),
);
}
return (self, None);
}
Search::Done => return (self, None),
Search::NextChunk => {
q.next_chunk();
}
}
}
}
fn fold_with<F: Folder<R>>(self, mut folder: F) -> F {
let QueryParIter { inner: mut q } = self;
while !folder.full() {
match q.next() {
Some(item) => folder = folder.consume(item),
None => break,
}
}
folder
}
}
impl<'a, C, P, R> ParallelIterator for QueryParIter<C, P, R>
where
C: Constraint<'a> + Clone + Send + 'a,
P: Fn(&Binding<'_>) -> Option<R> + Clone + Send,
R: Send,
{
type Item = R;
fn drive_unindexed<Con>(self, consumer: Con) -> Con::Result
where
Con: UnindexedConsumer<Self::Item>,
{
bridge_unindexed(self, consumer)
}
}
}
#[macro_export]
macro_rules! find {
($($tokens:tt)*) => {
{
#[allow(unused_mut, unused_variables)]
let mut ctx = $crate::query::VariableContext::new();
macro_rules! __local_find_context {
() => { &mut ctx }
}
$crate::macros::__find_impl!($crate, ctx, $($tokens)*)
}
};
}
pub use find;
#[macro_export]
macro_rules! exists {
(($($vars:tt)*), $Constraint:expr) => {
$crate::query::find!(($($vars)*), $Constraint).next().is_some()
};
($Constraint:expr) => {
$crate::query::find!((), $Constraint).next().is_some()
};
}
pub use exists;
#[macro_export]
macro_rules! temp {
(($Var:ident), $body:expr) => {{
let $Var = __local_find_context!().next_variable();
$body
}};
(($Var:ident,), $body:expr) => {
$crate::temp!(($Var), $body)
};
(($Var:ident, $($rest:ident),+ $(,)?), $body:expr) => {{
$crate::temp!(
($Var),
$crate::temp!(($($rest),+), $body)
)
}};
}
pub use temp;
#[cfg(test)]
mod tests {
use inlineencodings::ShortString;
use crate::prelude::inlineencodings::*;
use crate::prelude::*;
use crate::examples::literature;
use fake::faker::lorem::en::Sentence;
use fake::faker::lorem::en::Words;
use fake::faker::name::raw::*;
use fake::locales::*;
use fake::Fake;
use std::collections::HashSet;
use super::*;
fn variable_set(indices: impl IntoIterator<Item = VariableId>) -> VariableSet {
let mut variables = VariableSet::new_empty();
for variable in indices {
variables.set(variable);
}
variables
}
pub mod knights {
use crate::prelude::*;
attributes! {
"8143F46E812E88C4544E7094080EC523" unsafe as loves: inlineencodings::GenId;
"D6E0F2A6E5214E1330565B4D4138E55C" unsafe as name: inlineencodings::ShortString;
}
}
mod shared_level_buffers {
use super::*;
fn value(byte: u8) -> RawInline {
let mut value = [0; 32];
value[31] = byte;
value
}
fn refill(store: &mut BindingStore, values: &[RawInline]) {
store.refill(
0,
&[0],
RowSelection::identity(1),
1,
false,
|frontier, proposals| {
assert_eq!(frontier.len(), 1);
proposals.open(0);
proposals.extend(values.iter().copied());
},
);
}
#[test]
fn clone_shares_published_buffer_but_not_consumption_cursor() {
let mut left = BindingStore::new();
refill(&mut left, &[value(1), value(2), value(3)]);
let right = left.clone();
assert!(Arc::ptr_eq(
left.levels[0].buffer.as_ref().unwrap(),
right.levels[0].buffer.as_ref().unwrap(),
));
let mut drawn = Vec::new();
let mut parents = Vec::new();
left.draw(0, 1, RowSelection::identity(1), &mut drawn, &mut parents);
assert_eq!(left.levels[0].pos, 1);
assert_eq!(right.levels[0].pos, 0);
assert_eq!(right.levels[0].buffer()[0], value(1));
}
#[test]
fn refill_replaces_a_shared_buffer_without_touching_its_sibling() {
let mut left = BindingStore::new();
refill(&mut left, &[value(1), value(2)]);
let right = left.clone();
refill(&mut left, &[value(9)]);
assert!(!Arc::ptr_eq(
left.levels[0].buffer.as_ref().unwrap(),
right.levels[0].buffer.as_ref().unwrap(),
));
assert_eq!(&right.levels[0].buffer()[..], &[value(1), value(2)]);
assert_eq!(&left.levels[0].buffer()[..], &[value(9)]);
}
#[test]
fn refill_reuses_a_uniquely_owned_buffer() {
let mut store = BindingStore::new();
refill(&mut store, &[value(1), value(2), value(3)]);
let allocation = Arc::as_ptr(store.levels[0].buffer.as_ref().unwrap());
refill(&mut store, &[value(4)]);
assert_eq!(
Arc::as_ptr(store.levels[0].buffer.as_ref().unwrap()),
allocation,
"unique refill should retain the ProposalBuffer allocation",
);
assert_eq!(&store.levels[0].buffer()[..], &[value(4)]);
}
}
mod social {
use crate::prelude::*;
attributes! {
"A19EC1D9DD534BA9896223A457A6B9C9" unsafe as name: inlineencodings::ShortString;
"C21DE0AA5BA3446AB886C9640BA60244" unsafe as friend: inlineencodings::GenId;
}
}
#[test]
fn and_set() {
let mut books = HashSet::<String>::new();
let mut movies = HashSet::<Inline<ShortString>>::new();
books.insert("LOTR".to_string());
books.insert("Dragonrider".to_string());
books.insert("Highlander".to_string());
movies.insert("LOTR".to_inline());
movies.insert("Highlander".to_inline());
let inter: Vec<_> =
find!((a: Inline<ShortString>), and!(books.has(a), movies.has(a))).collect();
assert_eq!(inter.len(), 2);
let cross: Vec<_> =
find!((a: Inline<ShortString>, b: Inline<ShortString>), and!(books.has(a), movies.has(b))).collect();
assert_eq!(cross.len(), 6);
let one: Vec<_> = find!((a: Inline<ShortString>),
and!(books.has(a), a.is(ShortString::inline_from("LOTR")))
)
.collect();
assert_eq!(one.len(), 1);
}
#[test]
fn pattern() {
let mut kb = TribleSet::new();
(0..1000).for_each(|_| {
let author = fucid();
let book = fucid();
kb += entity! { &author @
literature::firstname: FirstName(EN).fake::<String>(),
literature::lastname: LastName(EN).fake::<String>(),
};
kb += entity! { &book @
literature::author: &author,
literature::title: Words(1..3).fake::<Vec<String>>().join(" "),
literature::quote: Sentence(5..25).fake::<String>().to_blob().get_handle()
};
});
let author = fucid();
let book = fucid();
kb += entity! { &author @
literature::firstname: "Frank",
literature::lastname: "Herbert",
};
kb += entity! { &book @
literature::author: &author,
literature::title: "Dune",
literature::quote: "I must not fear. Fear is the \
mind-killer. Fear is the little-death that brings total \
obliteration. I will face my fear. I will permit it to \
pass over me and through me. And when it has gone past I \
will turn the inner eye to see its path. Where the fear \
has gone there will be nothing. Only I will remain.".to_blob().get_handle()
};
(0..100).for_each(|_| {
let author = fucid();
let book = fucid();
kb += entity! { &author @
literature::firstname: "Fake",
literature::lastname: "Herbert",
};
kb += entity! { &book @
literature::author: &author,
literature::title: Words(1..3).fake::<Vec<String>>().join(" "),
literature::quote: Sentence(5..25).fake::<String>().to_blob().get_handle()
};
});
let r: Vec<_> = find!(
(author: Inline<_>, book: Inline<_>, title: Inline<_>, quote: Inline<_>),
pattern!(&kb, [
{?author @
literature::firstname: "Frank",
literature::lastname: "Herbert"},
{?book @
literature::author: ?author,
literature::title: ?title,
literature::quote: ?quote
}]))
.collect();
assert_eq!(1, r.len())
}
#[test]
fn constant() {
let r: Vec<_> = find! {
(string: Inline<_>, number: Inline<_>),
and!(
string.is(ShortString::inline_from("Hello World!")),
number.is(I256BE::inline_from(42))
)
}
.collect();
assert_eq!(1, r.len())
}
#[test]
fn exists_true() {
assert!(exists!((a: Inline<_>), a.is(I256BE::inline_from(42))));
}
#[test]
fn exists_false() {
assert!(!exists!(
(a: Inline<_>),
and!(a.is(I256BE::inline_from(1)), a.is(I256BE::inline_from(2)))
));
}
#[test]
fn exists_no_variables_true() {
let mut ctx = VariableContext::new();
let a = ctx.next_variable::<I256BE>();
assert!(exists!(a.is(I256BE::inline_from(42))));
}
#[test]
fn find_no_variables_yields_unit() {
let mut ctx = VariableContext::new();
let a = ctx.next_variable::<I256BE>();
let rows: Vec<()> = find!((), a.is(I256BE::inline_from(42))).collect();
assert_eq!(rows, vec![()]);
}
#[test]
fn temp_variables_span_patterns() {
use social::*;
let mut kb = TribleSet::new();
let alice = fucid();
let bob = fucid();
kb += entity! { &alice @ name: "Alice", friend: &bob };
kb += entity! { &bob @ name: "Bob" };
let matches: Vec<_> = find!(
(person_name: Inline<_>),
temp!((mutual_friend),
and!(
pattern!(&kb, [{ _?person @ name: ?person_name, friend: ?mutual_friend }]),
pattern!(&kb, [{ ?mutual_friend @ name: "Bob" }])
)
)
)
.collect();
assert_eq!(matches.len(), 1);
assert_eq!(matches[0].0.try_from_inline::<&str>().unwrap(), "Alice");
}
#[test]
fn estimate_override_debug_order() {
use std::cell::RefCell;
use std::rc::Rc;
let mut ctx = VariableContext::new();
let a = ctx.next_variable::<ShortString>();
let b = ctx.next_variable::<ShortString>();
let base = and!(
a.is(ShortString::inline_from("A")),
b.is(ShortString::inline_from("B"))
);
let mut wrapper = crate::debug::query::EstimateOverrideConstraint::new(base);
wrapper.set_estimate(a.index, 10);
wrapper.set_estimate(b.index, 1);
let record = Rc::new(RefCell::new(Vec::new()));
let debug = crate::debug::query::DebugConstraint::new(wrapper, Rc::clone(&record));
let q: Query<_, _, _> = Query::new(debug, |_| Some(()));
let r: Vec<_> = q.collect();
assert_eq!(1, r.len());
assert_eq!(&*record.borrow(), &[b.index, a.index]);
}
#[cfg(feature = "parallel")]
mod split_fence {
use super::*;
use rayon::iter::plumbing::UnindexedProducer;
use rayon::iter::{IntoParallelIterator, ParallelIterator};
use std::sync::atomic::{AtomicU8, Ordering};
#[derive(Clone)]
struct Fixture {
fanout: u32,
anchors: [u8; 4],
}
#[derive(Clone)]
struct OneToOne(usize);
#[derive(Clone)]
struct IntentRecorder(Arc<AtomicU8>);
impl<'a> Constraint<'a> for IntentRecorder {
fn variables(&self) -> VariableSet {
variable_set([0])
}
fn estimate(&self, variable: VariableId, _binding: &Binding) -> Option<usize> {
(variable == 0).then_some(1)
}
fn propose(
&self,
_variable: VariableId,
frontier: &Frontier<'_>,
proposals: &mut ProposalBuffer,
) {
self.0.fetch_or(
if frontier.parallel() { 0b10 } else { 0b01 },
Ordering::Relaxed,
);
for row in 0..frontier.len() {
proposals.open(row as u32);
proposals.push(Fixture::value(0x55, row as u32));
}
}
fn confirm(
&self,
_variable: VariableId,
_frontier: &Frontier<'_>,
_candidates: &mut Candidates<'_>,
) {
}
}
impl<'a> Constraint<'a> for OneToOne {
fn variables(&self) -> VariableSet {
variable_set(0..self.0)
}
fn estimate(&self, variable: VariableId, binding: &Binding) -> Option<usize> {
(variable < self.0).then_some(if variable == binding.bound.count() {
1
} else {
1 << 20
})
}
fn propose(
&self,
variable: VariableId,
frontier: &Frontier<'_>,
proposals: &mut ProposalBuffer,
) {
for row in 0..frontier.len() {
proposals.open(row as u32);
proposals.push(Fixture::value(0x7f, variable as u32));
}
}
fn confirm(
&self,
_variable: VariableId,
_frontier: &Frontier<'_>,
_candidates: &mut Candidates<'_>,
) {
}
}
impl Fixture {
fn identity_split(fanout: u32) -> Self {
Self {
fanout,
anchors: [3, 0, 1, 2],
}
}
fn nonidentity_split(fanout: u32) -> Self {
Self {
fanout,
anchors: [3, 1, 0, 2],
}
}
fn value(tag: u8, index: u32) -> RawInline {
let mut value = [0; 32];
value[0] = tag;
value[28..].copy_from_slice(&index.to_be_bytes());
value
}
fn count(&self, group: u8, variable: VariableId) -> u32 {
match (group, variable) {
(3 | 0, 1) => 0,
(1, 2) => self.fanout,
_ => 1,
}
}
}
impl<'a> Constraint<'a> for Fixture {
fn variables(&self) -> VariableSet {
variable_set(0..4)
}
fn estimate(&self, variable: VariableId, binding: &Binding) -> Option<usize> {
let preferred = binding
.get(0)
.map(|anchor| match anchor[31] {
0 => 1,
1 => 2,
2 => 3,
_ => 1,
})
.unwrap_or(0);
(variable < 4).then_some(if variable == 0 {
4
} else if variable == preferred {
1
} else {
8
})
}
fn propose(
&self,
variable: VariableId,
frontier: &Frontier<'_>,
proposals: &mut ProposalBuffer,
) {
for row in 0..frontier.len() {
proposals.open(row as u32);
if variable == 0 {
proposals.extend(
self.anchors
.into_iter()
.map(|group| Self::value(0, group as u32)),
);
} else {
let group = frontier.row(row).get(0).expect("anchor bound")[31];
proposals.extend(
(0..self.count(group, variable))
.map(|index| Self::value(variable as u8, index)),
);
}
}
}
fn confirm(
&self,
_variable: VariableId,
_frontier: &Frontier<'_>,
_candidates: &mut Candidates<'_>,
) {
}
}
type TestQuery = Query<Fixture, fn(&Binding<'_>) -> Option<RawInline>, RawInline>;
fn project(binding: &Binding<'_>) -> Option<RawInline> {
binding.get(0).copied()
}
fn query(fanout: u32) -> TestQuery {
fixture_query(Fixture::identity_split(fanout))
}
fn nonidentity_query(fanout: u32) -> TestQuery {
fixture_query(Fixture::nonidentity_split(fanout))
}
fn fixture_query(fixture: Fixture) -> TestQuery {
Query::new(fixture, project as fn(&Binding<'_>) -> Option<RawInline>)
.with_frontier_width(4)
}
#[test]
fn only_into_par_iter_sets_parallel_frontier_intent() {
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(4)
.build()
.unwrap();
let sequential = Arc::new(AtomicU8::new(0));
let sequential_rows = pool.install(|| {
Query::new(IntentRecorder(Arc::clone(&sequential)), |_| Some(())).count()
});
assert_eq!(sequential_rows, 1);
assert_eq!(
sequential.load(Ordering::Relaxed),
0b01,
"an ordinary iterator stays serial even inside a Rayon pool"
);
let parallel = Arc::new(AtomicU8::new(0));
let parallel_rows = pool.install(|| {
Query::new(IntentRecorder(Arc::clone(¶llel)), |_| Some(()))
.into_par_iter()
.count()
});
assert_eq!(parallel_rows, 1);
assert_eq!(parallel.load(Ordering::Relaxed), 0b10);
}
fn advance_to_three_groups(query: &mut TestQuery) {
query.plan();
query.next_group();
query.next_chunk();
query.plan();
query.next_group();
query.next_chunk();
query.next_group();
query.next_chunk();
query.plan();
assert_eq!(query.depths[query.depth].plan.group_count(), 3);
assert_eq!(query.depths[query.depth].group, 0);
assert_eq!(query.depths[query.depth].group_limit, 3);
}
fn split_plan(query: &TestQuery) -> &SplitPlan {
match &query.depths[query.depth].plan {
DepthPlan::Split(plan) => plan,
plan => panic!("expected split plan, got {plan:?}"),
}
}
#[test]
fn universal_plan_stays_generated_without_choice_storage() {
let mut representation = query(16);
assert_eq!(representation.choice.capacity(), 0);
representation.plan();
assert!(matches!(
representation.depths[0].plan,
DepthPlan::Universal(0)
));
assert_eq!(
representation.depths[0].selection(),
RowSelection::identity(1)
);
assert_eq!(
representation.choice.capacity(),
0,
"unanimity must not allocate planner choice storage"
);
}
#[test]
fn identity_partition_omits_the_row_order_override() {
let mut representation = query(16);
advance_to_three_groups(&mut representation);
let plan = split_plan(&representation);
assert_eq!(plan.groups, [(1, 1), (2, 2), (3, 3)]);
assert!(plan.order_override.is_empty());
assert_eq!(
representation.depths[representation.depth].selection(),
RowSelection::identity(3)
);
}
#[test]
fn nonidentity_partition_stores_only_the_row_order_override() {
let mut representation = nonidentity_query(16);
advance_to_three_groups(&mut representation);
let plan = split_plan(&representation);
assert_eq!(plan.groups, [(1, 1), (2, 2), (3, 3)]);
assert_eq!(plan.order_override, [1, 0, 2]);
assert_eq!(
representation.depths[representation.depth].selection(),
RowSelection::explicit(&[1, 0, 2])
);
let mut expected: Vec<_> = query(16).collect();
let mut actual: Vec<_> = nonidentity_query(16).collect();
expected.sort_unstable();
actual.sort_unstable();
assert_eq!(
actual, expected,
"row permutation must preserve the exact bag"
);
}
#[test]
fn split_transfers_one_whole_group_and_fences_it() {
const FANOUT: u32 = 64;
let mut expected: Vec<_> = query(FANOUT).collect();
expected.sort_unstable();
assert_eq!(expected.len(), FANOUT as usize + 1);
let mut left = query(FANOUT);
advance_to_three_groups(&mut left);
let source_len = left.bindings.levels[0].buffer().len();
let source_pos = left.bindings.levels[0].pos;
let right = left.split_pending_group().expect("transfer first group");
assert_eq!(
(right.depth, right.depths.len(), right.stack.len()),
(0, 1, 0)
);
assert_eq!(
(
right.depths[0].group,
right.depths[0].group_limit,
right.depths[0].plan.group_count(),
),
(0, 1, 3)
);
assert!(Arc::ptr_eq(&left.depths[1].block, &right.depths[0].block));
assert_eq!(left.depths[1].selection(), right.depths[0].selection());
match (&left.depths[1].plan, &right.depths[0].plan) {
(DepthPlan::Split(left), DepthPlan::Split(right)) => {
assert!(Arc::ptr_eq(left, right), "fencing must not clone the plan");
}
plans => panic!("expected shared split plans, got {plans:?}"),
}
assert_eq!(left.depths[1].group, 1, "left skips the owned group");
assert_eq!(left.bindings.levels[0].buffer().len(), source_len);
assert_eq!(right.bindings.levels[0].buffer().len(), source_len);
assert_eq!(left.bindings.levels[0].pos, source_pos);
assert_eq!(right.bindings.levels[0].pos, source_pos);
let mut fenced = right.clone();
assert!(fenced.split_pending_group().is_none());
let second = left.split_pending_group().expect("transfer second group");
assert!(
left.split_pending_group().is_none(),
"the last group stays with the only producer that can execute it"
);
assert_eq!(left.depths[1].group + 1, left.depths[1].group_limit);
let mut actual: Vec<_> = left.chain(right).chain(second).collect();
actual.sort_unstable();
assert_eq!(actual, expected, "splitting must preserve the exact bag");
assert_eq!(actual.iter().filter(|row| row[31] == 2).count(), 1);
}
fn advance_to_emit(query: &mut TestQuery) {
loop {
match query.mode {
Search::Plan => query.plan(),
Search::NextGroup => query.next_group(),
Search::NextChunk => query.next_chunk(),
Search::Emit => return,
Search::Done => panic!("fixture ended before a terminal frontier"),
}
}
}
#[test]
fn terminal_page_is_transferred_without_slicing_rows() {
let mut left = query(32);
advance_to_emit(&mut left);
assert!(left.depth > 0);
let terminal_rows = left.depths[left.depth].rows;
let terminal_block = Arc::clone(&left.depths[left.depth].block);
let terminal_plan = left.depths[left.depth].plan.clone();
let mut expected = left.clone().collect::<Vec<_>>();
let right = left
.split_terminal_frontier()
.expect("terminal page has a parent continuation");
assert_eq!(right.depth, 0);
assert_eq!(right.stack.len(), 0);
assert_eq!(right.mode, Search::Emit);
assert_eq!(right.depths[0].rows, terminal_rows);
assert!(Arc::ptr_eq(&terminal_block, &right.depths[0].block));
assert_eq!(
terminal_plan.selection(terminal_rows),
right.depths[0].selection()
);
for variable in left.bindings.bound {
assert!(Arc::ptr_eq(
left.bindings.levels[variable]
.buffer
.as_ref()
.expect("bound level has a published buffer"),
right.bindings.levels[variable]
.buffer
.as_ref()
.expect("bound level has a published buffer"),
));
}
let mut actual: Vec<_> = left.chain(right).collect();
expected.sort_unstable();
actual.sort_unstable();
assert_eq!(actual, expected);
}
#[test]
fn indivisible_one_to_one_chain_never_splits_or_forces_copies() {
const HOPS: usize = 60;
let query = Query::new(OneToOne(HOPS), |binding: &Binding| {
binding.get(HOPS - 1).copied()
});
let stats = query.stats();
let (producer, right) = query.into_par_iter().split();
assert!(right.is_none());
let rows = producer.collect::<Vec<_>>();
assert_eq!(rows, vec![Fixture::value(0x7f, HOPS as u32 - 1)]);
assert_eq!(stats.inplace_descents(), HOPS as u64);
assert_eq!(
stats.copied_descents(),
0,
"a fruitless split search must not destroy Arc uniqueness"
);
}
#[test]
fn repeated_manual_producer_splits_preserve_explicit_order_and_bag() {
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(4)
.build()
.unwrap();
for fixture in [query as fn(u32) -> TestQuery, nonidentity_query] {
let mut expected: Vec<_> = fixture(96).collect();
expected.sort_unstable();
let mut queue = vec![fixture(96).into_par_iter()];
let mut leaves = Vec::new();
for _ in 0..64 {
let Some(producer) = queue.pop() else {
break;
};
let (left, right) = producer.split();
if let Some(right) = right {
queue.push(left);
queue.push(right);
} else {
leaves.push(left);
}
}
let mut actual = Vec::new();
for producer in queue.into_iter().chain(leaves) {
actual.extend(producer.collect::<Vec<_>>());
}
actual.sort_unstable();
assert_eq!(actual, expected);
for _ in 0..8 {
let mut scheduled =
pool.install(|| fixture(96).into_par_iter().collect::<Vec<_>>());
scheduled.sort_unstable();
assert_eq!(scheduled, expected);
}
}
}
}
}