Skip to content

Commit 381f978

Browse files
terjekvterjekv
authored
Enforce pipeline stage shape contracts (#66)
1 parent b7d4ab6 commit 381f978

9 files changed

Lines changed: 453 additions & 18 deletions

File tree

‎crates/hubuum-filter/src/eval.rs‎

Lines changed: 8 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -13,17 +13,8 @@ use crate::verbs::search::{
1313
};
1414

1515
impl PipeStage {
16-
pub fn apply_all(
17-
stages: &[Self],
18-
mut lines: Vec<String>,
19-
) -> Result<Vec<String>, PipelineError> {
20-
for stage in stages {
21-
lines = stage.apply(lines)?;
22-
}
23-
Ok(lines)
24-
}
25-
2616
fn apply(&self, lines: Vec<String>) -> Result<Vec<String>, PipelineError> {
17+
self.validate_input_shape(OutputShape::Lines)?;
2718
match self {
2819
Self::Grep(pattern) | Self::ValueSearch(pattern) => {
2920
let regex = Regex::new(pattern)?;
@@ -64,9 +55,7 @@ impl PipeStage {
6455
| Self::CollapseGroups
6556
| Self::Unroll(_)
6657
| Self::Jq(_)
67-
| Self::Value(_) => Err(PipelineError::Pipe(
68-
"Pipe stage requires structured table output".to_string(),
69-
)),
58+
| Self::Value(_) => unreachable!("line input shape was validated"),
7059
}
7160
}
7261
}
@@ -86,6 +75,7 @@ fn apply_semantic_stage(
8675
envelope: OutputEnvelope,
8776
stage: &PipeStage,
8877
) -> Result<OutputEnvelope, PipelineError> {
78+
stage.validate_input_shape(envelope.shape)?;
8979
if envelope.shape == OutputShape::Lines {
9080
let lines = envelope
9181
.value
@@ -94,6 +84,11 @@ fn apply_semantic_stage(
9484
.flatten()
9585
.filter_map(|value| value.as_str().map(str::to_string))
9686
.collect::<Vec<_>>();
87+
if matches!(stage, PipeStage::Count) {
88+
return Ok(OutputEnvelope::values(vec![serde_json::Value::Number(
89+
lines.len().into(),
90+
)]));
91+
}
9792
return Ok(OutputEnvelope::lines(stage.apply(lines)?));
9893
}
9994

‎crates/hubuum-filter/src/model.rs‎

Lines changed: 200 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,192 @@ pub enum PipeStage {
3737
Value(Selector),
3838
}
3939

40+
const ALL_SHAPES: &[OutputShape] = &[
41+
OutputShape::Empty,
42+
OutputShape::Lines,
43+
OutputShape::Rows,
44+
OutputShape::Detail,
45+
OutputShape::Message,
46+
OutputShape::Values,
47+
OutputShape::Groups,
48+
];
49+
const STRUCTURED_SHAPES: &[OutputShape] = &[
50+
OutputShape::Empty,
51+
OutputShape::Rows,
52+
OutputShape::Detail,
53+
OutputShape::Message,
54+
OutputShape::Values,
55+
OutputShape::Groups,
56+
];
57+
const COLLECTION_SHAPES: &[OutputShape] = &[
58+
OutputShape::Empty,
59+
OutputShape::Lines,
60+
OutputShape::Rows,
61+
OutputShape::Values,
62+
OutputShape::Groups,
63+
];
64+
const STRUCTURED_COLLECTION_SHAPES: &[OutputShape] = &[
65+
OutputShape::Empty,
66+
OutputShape::Rows,
67+
OutputShape::Values,
68+
OutputShape::Groups,
69+
];
70+
const PROJECT_SHAPES: &[OutputShape] = &[
71+
OutputShape::Empty,
72+
OutputShape::Rows,
73+
OutputShape::Detail,
74+
OutputShape::Message,
75+
OutputShape::Groups,
76+
];
77+
const GROUP_INPUT_SHAPES: &[OutputShape] = &[
78+
OutputShape::Empty,
79+
OutputShape::Rows,
80+
OutputShape::Detail,
81+
OutputShape::Message,
82+
OutputShape::Values,
83+
];
84+
const GROUPS_ONLY: &[OutputShape] = &[OutputShape::Groups];
85+
const EMPTY_ONLY: &[OutputShape] = &[OutputShape::Empty];
86+
const LINES_ONLY: &[OutputShape] = &[OutputShape::Lines];
87+
const ROWS_ONLY: &[OutputShape] = &[OutputShape::Rows];
88+
const DETAIL_ONLY: &[OutputShape] = &[OutputShape::Detail];
89+
const VALUES_ONLY: &[OutputShape] = &[OutputShape::Values];
90+
const DETAIL_OR_EMPTY: &[OutputShape] = &[OutputShape::Detail, OutputShape::Empty];
91+
const MESSAGE_OR_EMPTY: &[OutputShape] = &[OutputShape::Message, OutputShape::Empty];
92+
const JQ_OUTPUT_SHAPES: &[OutputShape] = &[
93+
OutputShape::Empty,
94+
OutputShape::Rows,
95+
OutputShape::Detail,
96+
OutputShape::Message,
97+
OutputShape::Values,
98+
];
99+
100+
impl PipeStage {
101+
pub fn name(&self) -> &'static str {
102+
match self {
103+
Self::Grep(_) => "F",
104+
Self::ValueSearch(_) => "V",
105+
Self::KeySearch(_) => "K",
106+
Self::Truthy(_) => "?",
107+
Self::Reject(_) => "reject",
108+
Self::Head { .. } => "L",
109+
Self::Tail(_) => "tail",
110+
Self::Count => "C",
111+
Self::SortLines { .. } | Self::SortColumn { .. } => "S",
112+
Self::Columns(_) => "P",
113+
Self::Group(_) => "G",
114+
Self::Aggregate(_) => "A",
115+
Self::CollapseGroups => "Z",
116+
Self::Unroll(_) => "U",
117+
Self::Jq(_) => "JQ",
118+
Self::Value(_) => "VALUE",
119+
}
120+
}
121+
122+
pub fn accepted_input_shapes(&self) -> &'static [OutputShape] {
123+
match self {
124+
Self::Grep(_) | Self::ValueSearch(_) | Self::Reject(_) | Self::Count => ALL_SHAPES,
125+
Self::KeySearch(_) | Self::Truthy(_) | Self::Jq(_) | Self::Value(_) => {
126+
STRUCTURED_SHAPES
127+
}
128+
Self::Head { .. } | Self::Tail(_) | Self::SortLines { .. } => COLLECTION_SHAPES,
129+
Self::Columns(_) => PROJECT_SHAPES,
130+
Self::SortColumn { .. } | Self::Unroll(_) => STRUCTURED_COLLECTION_SHAPES,
131+
Self::Group(_) => GROUP_INPUT_SHAPES,
132+
Self::Aggregate(_) | Self::CollapseGroups => GROUPS_ONLY,
133+
}
134+
}
135+
136+
pub fn resulting_shapes(
137+
&self,
138+
input: OutputShape,
139+
) -> Result<&'static [OutputShape], PipelineError> {
140+
self.validate_input_shape(input)?;
141+
let shapes = match self {
142+
Self::Grep(_) | Self::ValueSearch(_) | Self::Reject(_) => match input {
143+
OutputShape::Empty => EMPTY_ONLY,
144+
OutputShape::Lines => LINES_ONLY,
145+
OutputShape::Rows => ROWS_ONLY,
146+
OutputShape::Detail => DETAIL_OR_EMPTY,
147+
OutputShape::Message => MESSAGE_OR_EMPTY,
148+
OutputShape::Values => VALUES_ONLY,
149+
OutputShape::Groups => GROUPS_ONLY,
150+
},
151+
Self::KeySearch(_) => match input {
152+
OutputShape::Empty => EMPTY_ONLY,
153+
OutputShape::Rows | OutputShape::Values => ROWS_ONLY,
154+
OutputShape::Detail | OutputShape::Message => DETAIL_OR_EMPTY,
155+
OutputShape::Groups => GROUPS_ONLY,
156+
OutputShape::Lines => unreachable!("validated input shape"),
157+
},
158+
Self::Truthy(_) => match input {
159+
OutputShape::Empty => EMPTY_ONLY,
160+
OutputShape::Rows => ROWS_ONLY,
161+
OutputShape::Detail => DETAIL_OR_EMPTY,
162+
OutputShape::Message => MESSAGE_OR_EMPTY,
163+
OutputShape::Values => VALUES_ONLY,
164+
OutputShape::Groups => GROUPS_ONLY,
165+
OutputShape::Lines => unreachable!("validated input shape"),
166+
},
167+
Self::Head { .. } | Self::Tail(_) | Self::SortLines { .. } => match input {
168+
OutputShape::Empty => EMPTY_ONLY,
169+
OutputShape::Lines => LINES_ONLY,
170+
OutputShape::Rows => ROWS_ONLY,
171+
OutputShape::Values => VALUES_ONLY,
172+
OutputShape::Groups => GROUPS_ONLY,
173+
OutputShape::Detail | OutputShape::Message => {
174+
unreachable!("validated input shape")
175+
}
176+
},
177+
Self::Count => match input {
178+
OutputShape::Groups => ROWS_ONLY,
179+
_ => VALUES_ONLY,
180+
},
181+
Self::Columns(_) => match input {
182+
OutputShape::Empty => EMPTY_ONLY,
183+
OutputShape::Rows => ROWS_ONLY,
184+
OutputShape::Detail | OutputShape::Message => DETAIL_ONLY,
185+
OutputShape::Groups => GROUPS_ONLY,
186+
OutputShape::Lines | OutputShape::Values => {
187+
unreachable!("validated input shape")
188+
}
189+
},
190+
Self::SortColumn { .. } | Self::Unroll(_) => match input {
191+
OutputShape::Empty => EMPTY_ONLY,
192+
OutputShape::Rows => ROWS_ONLY,
193+
OutputShape::Values => VALUES_ONLY,
194+
OutputShape::Groups => GROUPS_ONLY,
195+
OutputShape::Lines | OutputShape::Detail | OutputShape::Message => {
196+
unreachable!("validated input shape")
197+
}
198+
},
199+
Self::Group(_) => GROUPS_ONLY,
200+
Self::Aggregate(_) => GROUPS_ONLY,
201+
Self::CollapseGroups => ROWS_ONLY,
202+
Self::Jq(_) => JQ_OUTPUT_SHAPES,
203+
Self::Value(_) => VALUES_ONLY,
204+
};
205+
Ok(shapes)
206+
}
207+
208+
pub(crate) fn validate_input_shape(&self, input: OutputShape) -> Result<(), PipelineError> {
209+
let accepted = self.accepted_input_shapes();
210+
if accepted.contains(&input) {
211+
return Ok(());
212+
}
213+
214+
let expected = accepted
215+
.iter()
216+
.map(ToString::to_string)
217+
.collect::<Vec<_>>()
218+
.join(", ");
219+
Err(PipelineError::Pipe(format!(
220+
"Pipe stage '{}' does not accept {input} output; expected one of: {expected}",
221+
self.name()
222+
)))
223+
}
224+
}
225+
40226
#[derive(Debug, Clone, PartialEq, Eq)]
41227
pub struct ProjectTerm {
42228
selector: Selector,
@@ -204,6 +390,20 @@ pub enum OutputShape {
204390
Groups,
205391
}
206392

393+
impl Display for OutputShape {
394+
fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
395+
formatter.write_str(match self {
396+
Self::Empty => "Empty",
397+
Self::Lines => "Lines",
398+
Self::Rows => "Rows",
399+
Self::Detail => "Detail",
400+
Self::Message => "Message",
401+
Self::Values => "Values",
402+
Self::Groups => "Groups",
403+
})
404+
}
405+
}
406+
207407
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
208408
pub struct OutputEnvelope {
209409
pub shape: OutputShape,

0 commit comments

Comments
 (0)