differential-dataflow 0.9.0

An incremental data-parallel dataflow platform
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

//! Implementation using ordered keys and exponential search.

use std::rc::Rc;

use owning_ref::{OwningRef, Erased};

use super::{Trie, Cursor, Builder, MergeBuilder, TupleBuilder};

/// A layer of unordered values. 
#[derive(Debug)]
pub struct UnorderedLayer<K> {
	/// Unordered values.
	pub vals: Vec<K>,
}

impl<K: Clone> Trie for UnorderedLayer<K> {
	type Item = K;
	type Cursor = UnorderedCursor;
	type MergeBuilder = UnorderedBuilder<K>;
	type TupleBuilder = UnorderedBuilder<K>;
	fn keys(&self) -> usize { self.vals.len() }
	fn tuples(&self) -> usize { <UnorderedLayer<K> as Trie>::keys(&self) }
	fn cursor_from(&self, lower: usize, upper: usize) -> Self::Cursor {	
		// println!("unordered: {} .. {}", lower, upper);
		UnorderedCursor {
			// vals: owned_self.map(|x| &x.vals[..]),
			bounds: (lower, upper),
			pos: lower,
		}
	}
}

/// A builder for unordered values.
pub struct UnorderedBuilder<K> {
	/// Unordered values.
	pub vals: Vec<K>,
}

impl<K: Clone> Builder for UnorderedBuilder<K> {
	type Trie = UnorderedLayer<K>; 
	fn boundary(&mut self) -> usize { self.vals.len() } 
	fn done(self) -> Self::Trie { UnorderedLayer { vals: self.vals } }
}

impl<K: Clone> MergeBuilder for UnorderedBuilder<K> {
	fn with_capacity(other1: &Self::Trie, other2: &Self::Trie) -> Self {
		UnorderedBuilder {
			vals: Vec::with_capacity(<UnorderedLayer<K> as Trie>::keys(other1) + <UnorderedLayer<K> as Trie>::keys(other2)),
		}
	}
	fn copy_range(&mut self, other: &Self::Trie, lower: usize, upper: usize) {
		self.vals.extend_from_slice(&other.vals[lower .. upper]);
	}
	fn push_merge(&mut self, other1: (&Self::Trie, usize, usize), other2: (&Self::Trie, usize, usize)) -> usize {
		<UnorderedBuilder<K> as MergeBuilder>::copy_range(self, &other1.0, other1.1, other1.2);
		<UnorderedBuilder<K> as MergeBuilder>::copy_range(self, &other2.0, other2.1, other2.2);
		self.vals.len()
	}
}

impl<K: Clone> TupleBuilder for UnorderedBuilder<K> {
	type Item = K;
	fn new() -> Self { UnorderedBuilder { vals: Vec::new() } }
	fn with_capacity(cap: usize) -> Self { UnorderedBuilder { vals: Vec::with_capacity(cap) } }
	#[inline(always)] fn push_tuple(&mut self, tuple: K) { self.vals.push(tuple) }
}

/// A cursor for walking through an unordered sequence of values.
///
/// This cursor does not support `seek`, though I'm not certain how to expose this.
#[derive(Debug)]
pub struct UnorderedCursor {
	// vals: OwningRef<Rc<Erased>, [K]>,
	pos: usize,
	bounds: (usize, usize),
}

impl<K: Clone> Cursor<UnorderedLayer<K>> for UnorderedCursor {
	type Key = K;
	fn key<'a>(&self, storage: &'a UnorderedLayer<K>) -> &'a Self::Key { &storage.vals[self.pos] }
	fn step(&mut self, _storage: &UnorderedLayer<K>) {
		self.pos += 1; 
		if !self.valid() {
			self.pos = self.bounds.1;
		}
	}
	fn seek(&mut self, _storage: &UnorderedLayer<K>, _key: &Self::Key) {
		panic!("seeking in an UnorderedCursor");
	}
	fn valid(&self, _storage: &UnorderedLayer<K>) -> bool { self.pos < self.bounds.1 }
	fn rewind(&mut self, _storage: &UnorderedLayer<K>) {
		self.pos = self.bounds.0;
	}
	fn reposition(&mut self, _storage: &UnorderedLayer<K>, lower: usize, upper: usize) {
		self.pos = lower;
		self.bounds = (lower, upper);
	}
}