use std::collections::HashMap;
use std::sync::Arc;
use crate::ast::Value;
use crate::dsl::traversal::Traversal;
use crate::iteration::comprehension::runtime::{RuntimeTuple, evaluate_for_iteration};
use crate::iteration::cursor_partition::{cursor_extent_on, cursor_over_partitions_on};
use crate::kernel::Kernel;
use super::{PolydatKernel, PolydatProgram};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CursorSlice {
pub cursor: String,
pub start: u64,
pub end: u64,
}
impl CursorSlice {
pub fn len(&self) -> u64 {
self.end.saturating_sub(self.start)
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
pub struct Activation<K = PolydatKernel> {
pub index: u64,
pub coords: Vec<(String, Value)>,
pub kernel: K,
pub cursor: Option<CursorSlice>,
}
impl std::fmt::Debug for Activation {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Activation")
.field("index", &self.index)
.field("coords", &self.coords)
.field("cursor", &self.cursor)
.field("program_nodes", &self.kernel.program().node_count())
.finish()
}
}
impl std::fmt::Debug for Activation<Box<dyn Kernel>> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Activation")
.field("index", &self.index)
.field("coords", &self.coords)
.field("cursor", &self.cursor)
.field("engine", &self.kernel.engine())
.finish()
}
}
impl<K> Activation<K> {
pub fn cycle_count(&self) -> u64 {
match &self.cursor {
Some(slice) => slice.len(),
None => 1,
}
}
pub fn coord(&self, name: &str) -> Option<&Value> {
self.coords.iter().find(|(n, _)| n == name).map(|(_, v)| v)
}
}
impl Activation<Box<dyn Kernel>> {
pub fn cycle(&mut self, i: u64) -> &mut dyn Kernel {
self.kernel.set_inputs(&[i]);
if let Some(slice) = &self.cursor {
let ordinal = slice.start.saturating_add(i);
let slot = format!("{}__ordinal", slice.cursor);
let _ = self.kernel.set_input(&slot, Value::U64(ordinal));
}
self.kernel.as_mut()
}
pub fn for_each_cycle(&mut self, mut f: impl FnMut(u64, &mut dyn Kernel)) {
for i in 0..self.cycle_count() {
let kernel = self.cycle(i);
f(i, kernel);
}
}
}
impl Activation {
pub fn cycle(&mut self, i: u64) -> &mut PolydatKernel {
self.kernel.set_inputs(&[i]);
if let Some(slice) = &self.cursor {
let ordinal = slice.start.saturating_add(i);
let slot = format!("{}__ordinal", slice.cursor);
if let Some(idx) = self.kernel.program().find_input(&slot) {
self.kernel.state().set_input(idx, Value::U64(ordinal));
}
}
&mut self.kernel
}
pub fn for_each_cycle(&mut self, mut f: impl FnMut(u64, &mut PolydatKernel)) {
for i in 0..self.cycle_count() {
let kernel = self.cycle(i);
f(i, kernel);
}
}
}
pub struct TraversalStream {
traversal: Traversal,
tuples: Vec<RuntimeTuple>,
cascade: Vec<(String, Value)>,
next: usize,
}
impl TraversalStream {
pub fn len(&self) -> usize {
self.tuples.len()
}
pub fn is_empty(&self) -> bool {
self.tuples.is_empty()
}
pub fn traversal(&self) -> &Traversal {
&self.traversal
}
pub fn seek(&mut self, index: usize) {
self.next = index.min(self.tuples.len());
}
pub fn position(&self) -> usize {
self.next
}
pub fn advance(&mut self) -> Result<Option<Activation>, String> {
if self.next >= self.tuples.len() {
return Ok(None);
}
let i = self.next;
self.next += 1;
self.activation(i).map(Some)
}
pub fn activation(&self, index: usize) -> Result<Activation, String> {
let tuple = self.tuples.get(index).ok_or_else(|| {
format!(
"activation index {index} is out of range; traversal has {} tuples",
self.tuples.len()
)
})?;
let program = self.traversal.program.clone();
let mut kernel = PolydatKernel::from_program(program);
bind_by_name(&mut kernel, tuple);
bind_by_name(&mut kernel, &self.cascade);
let cursor = narrow_cursors(&mut kernel)?;
Ok(Activation {
index: index as u64,
coords: tuple.clone(),
kernel,
cursor,
})
}
pub fn activate(&self, index: usize) -> Result<Activation<Box<dyn Kernel>>, String> {
self.activation_on(index, crate::Engine::default())
}
pub fn activation_on(
&self,
index: usize,
engine: crate::Engine,
) -> Result<Activation<Box<dyn Kernel>>, String> {
let tuple = self.tuples.get(index).ok_or_else(|| {
format!(
"activation index {index} is out of range; traversal has {} tuples",
self.tuples.len()
)
})?;
let program = self
.traversal
.program_on(engine)
.map_err(|e| e.to_string())?;
let mut kernel = program.create_kernel();
bind_by_name_on(kernel.as_mut(), tuple)?;
bind_by_name_on(kernel.as_mut(), &self.cascade)?;
let cursor = narrow_cursors_on(kernel.as_mut())?;
Ok(Activation {
index: index as u64,
coords: tuple.clone(),
kernel,
cursor,
})
}
}
fn bind_by_name_on(kernel: &mut dyn Kernel, values: &[(String, Value)]) -> Result<(), String> {
let declared: std::collections::HashSet<String> = kernel.input_names().into_iter().collect();
for (name, value) in values {
if declared.contains(name) {
kernel.set_input(name, value.clone())?;
}
}
Ok(())
}
fn bind_by_name(kernel: &mut PolydatKernel, values: &[(String, Value)]) {
for (name, value) in values {
if let Some(idx) = kernel.program().find_input(name) {
kernel.state().set_input(idx, value.clone());
}
}
}
fn narrow_cursors(kernel: &mut PolydatKernel) -> Result<Option<CursorSlice>, String> {
narrow_cursors_on(kernel)
}
fn narrow_cursors_on(kernel: &mut dyn Kernel) -> Result<Option<CursorSlice>, String> {
let schemas: Vec<crate::iteration::source::SourceSchema> = kernel.cursor_schemas().to_vec();
let mut narrowest: Option<CursorSlice> = None;
for schema in &schemas {
let slice = if schema.partition_output.is_some() {
let parts = cursor_over_partitions_on(kernel, schema)?;
let partition = match parts.len() {
1 => parts[0],
0 => {
return Err(format!(
"cursor '{}': its `over` value resolved to no partitions",
schema.name
));
}
n => {
return Err(format!(
"cursor '{}': its `over` value resolved to {n} partitions; inside a traversal, bind the list \
with an enclosing `for p in ...` and declare the cursor `over p`",
schema.name
));
}
};
kernel.set_cursor(&schema.name, &partition)?;
CursorSlice {
cursor: schema.name.clone(),
start: partition.start_ord,
end: partition.end_ord,
}
} else {
let extent = cursor_extent_on(kernel, schema);
CursorSlice {
cursor: schema.name.clone(),
start: 0,
end: extent,
}
};
narrowest = Some(match narrowest {
Some(prev) if prev.len() <= slice.len() => prev,
_ => slice,
});
}
Ok(narrowest)
}
impl PolydatKernel {
pub fn over(program: Arc<PolydatProgram>) -> Self {
PolydatKernel::from_program(program)
}
pub fn traverse(&mut self, index: usize) -> Result<TraversalStream, String> {
let program = self.program().clone();
let traversal = program.traversals().get(index).cloned().ok_or_else(|| {
format!(
"no traversal at index {index}; the program declares {}",
program.traversals().len()
)
})?;
open_traversal(self, traversal)
}
}
pub fn program_identity(kernel: &PolydatKernel) -> *const PolydatProgram {
Arc::as_ptr(kernel.program())
}
pub fn open_traversal(
parent: &mut dyn Kernel,
traversal: Traversal,
) -> Result<TraversalStream, String> {
let mut cascade = Vec::with_capacity(traversal.cascade.len());
for (name, _) in &traversal.cascade {
let value = if parent.output_type(name).is_some() {
parent.pull(name)
} else {
parent.input_value(name).unwrap_or(Value::None)
};
cascade.push((name.clone(), value));
}
let mut canonical = PolydatKernel::from_program(traversal.program.clone());
bind_by_name(&mut canonical, &cascade);
let params: HashMap<String, String> = HashMap::new();
let tuples = evaluate_for_iteration(&traversal.comprehension, &canonical, ¶ms, |_| Ok(()))
.map_err(|e| {
format!(
"`for {}` at line {}, col {}: {e}",
traversal.source_text, traversal.span.line, traversal.span.col
)
})?;
Ok(TraversalStream {
traversal,
tuples,
cascade,
next: 0,
})
}