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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
// #![allow(dead_code)]
use crate::sparse_io_vector::*;
use log::info;
/// `sparse_io_stack` is a stack of `sparse_io_vector`
pub struct SparseIoStack {
pub stack: Vec<SparseIoVec>,
column_names: Vec<Box<str>>,
}
impl Default for SparseIoStack {
/// an empty sparse io vector for horizontal data integration
fn default() -> Self {
Self::new()
}
}
impl SparseIoStack {
pub fn new() -> Self {
Self {
stack: vec![],
column_names: vec![],
}
}
pub fn push(&mut self, data: SparseIoVec) -> anyhow::Result<()> {
if self.stack.is_empty() {
self.column_names.extend(data.column_names()?);
self.stack.push(data);
return Ok(());
}
info!("Checking column names...");
if self.column_names != data.column_names()? {
return Err(anyhow::anyhow!("column names don't match"));
}
self.stack.push(data);
Ok(())
}
/// number of data types
pub fn num_types(&self) -> usize {
self.stack.len()
}
/// number of shared columns
pub fn num_columns(&self) -> anyhow::Result<usize> {
self.stack
.iter()
.map(|x| x.num_columns())
.max()
.ok_or(anyhow::anyhow!("can't figure out the max"))
}
/// Get the shared column names.
pub fn column_names(&self) -> anyhow::Result<Vec<Box<str>>> {
if self.column_names.len() != self.num_columns()? {
return Err(anyhow::anyhow!("inconsistent columns"));
}
Ok(self.column_names.clone())
}
/// Exclude columns (cells) across every layer in the stack in
/// lockstep. `keep[global_col]` is `true` to keep, `false` to drop.
/// Mirror of [`SparseIoVec::mask_columns`]; the members share one
/// synchronized cell axis, so the same mask applies to all. MUST be
/// called before batch/group registration.
pub fn mask_columns_all(&mut self, keep: &[bool]) -> anyhow::Result<()> {
let n = self.num_columns()?;
if keep.len() != n {
return Err(anyhow::anyhow!(
"mask_columns_all: keep.len()={} != num_columns={}",
keep.len(),
n
));
}
for layer in self.stack.iter_mut() {
layer.mask_columns(keep)?;
}
// Refresh the cached shared column names from the (now filtered)
// first member; all members were masked identically.
self.column_names = match self.stack.first() {
Some(first) => first.column_names()?,
None => vec![],
};
Ok(())
}
/// Register batch membership for all layers in the stack.
pub fn register_batch_membership<T>(&mut self, batch_membership: &[T])
where
T: Sync + Send + std::hash::Hash + Eq + Clone + ToString,
{
for layer in self.stack.iter_mut() {
layer.register_batch_membership(batch_membership);
}
}
/// Get the row names combined across all the types. We will
/// append additional data type index: `format!("{}/{}", x, d)`
///
pub fn row_names(&self) -> anyhow::Result<Vec<Box<str>>> {
Ok(self
.stack
.iter()
.enumerate()
.map(|(d, x)| {
x.row_names().map(|x| {
x.into_iter()
.map(|y| format!("{}/{}", y, d).into_boxed_str())
.collect::<Vec<_>>()
})
})
.collect::<anyhow::Result<Vec<_>>>()?
.into_iter()
.flatten()
.collect())
}
}