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
121
122
123
124
125
126
127
128
129
use crate::filter::Filter;
use crate::CanonicalValue;
/// Row-level transformations applied between reading and writing.
///
/// Transforms are applied in order: filter → rename → select → limit.
/// All transforms are optional; an empty `Transforms` is a no-op.
#[derive(Debug, Clone, Default)]
pub struct Transforms {
pub filter: Option<Filter>,
pub select: Option<Vec<String>>,
pub rename: Vec<(String, String)>,
pub limit: Option<usize>,
pub dedup: Option<Vec<String>>,
}
impl Transforms {
pub fn is_empty(&self) -> bool {
self.filter.is_none()
&& self.select.is_none()
&& self.rename.is_empty()
&& self.limit.is_none()
&& self.dedup.is_none()
}
/// Applies transforms to a single row. Returns `None` if the row was
/// filtered out.
pub fn apply_to_row(
&self,
mut row: Vec<(String, CanonicalValue)>,
) -> Option<Vec<(String, CanonicalValue)>> {
if let Some(f) = &self.filter {
let wrapped = CanonicalValue::Object(row.clone());
if !f.matches(&wrapped) {
return None;
}
}
for (from, to) in &self.rename {
for (k, _) in row.iter_mut() {
if k == from {
*k = to.clone();
break;
}
}
}
if let Some(sel) = &self.select {
let mut new_row = Vec::with_capacity(sel.len());
for name in sel {
match row.iter().find(|(k, _)| k == name) {
Some((k, v)) => new_row.push((k.clone(), v.clone())),
None => new_row.push((name.clone(), CanonicalValue::Null)),
}
}
row = new_row;
}
Some(row)
}
/// Applies transforms to a `CanonicalValue::Array` of objects.
pub fn apply_to_array(&self, value: &CanonicalValue) -> CanonicalValue {
if let CanonicalValue::Object(o) = value {
if o.len() == 1
&& matches!(
&o[0].1,
CanonicalValue::Array(_) | CanonicalValue::Object(_)
)
{
let inner = self.apply_to_array(&o[0].1);
return CanonicalValue::Object(vec![(o[0].0.clone(), inner)]);
}
}
// Single row: object whose fields are all scalars. XML with one item
// and JSON with one root object both produce this shape.
if let CanonicalValue::Object(o) = value {
let all_scalar = o.iter().all(|(_, v)| {
matches!(
v,
CanonicalValue::Null
| CanonicalValue::Bool(_)
| CanonicalValue::Number(_)
| CanonicalValue::String(_)
)
});
if all_scalar && !o.is_empty() {
if let Some(new_row) = self.apply_to_row(o.clone()) {
return CanonicalValue::Object(new_row);
}
return CanonicalValue::Null;
}
}
let rows = match value {
CanonicalValue::Array(a) => a,
_ => return value.clone(),
};
let limit = self.limit.unwrap_or(usize::MAX);
let mut out = Vec::new();
let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
for row in rows {
if out.len() >= limit {
break;
}
match row {
CanonicalValue::Object(o) => {
if let Some(new_row) = self.apply_to_row(o.clone()) {
if let Some(cols) = &self.dedup {
let key: String = cols
.iter()
.map(|c| {
new_row
.iter()
.find(|(k, _)| k == c)
.map(|(_, v)| format!("{v:?}"))
.unwrap_or_default()
})
.collect::<Vec<_>>()
.join("\x1f");
if !seen.insert(key) {
continue;
}
}
out.push(CanonicalValue::Object(new_row));
}
}
other => out.push(other.clone()),
}
}
CanonicalValue::Array(out)
}
}