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
use super::*;
pub(super) fn handle(
op: &mut UnaryOperator,
input: &mut StreamingExecutor,
) -> Result<Option<DataChunk>, QueryError> {
let UnaryOperatorKind::Flatten {
// Plan level group identifier, carried end to end
// (plan -> spec -> operator) and visible in EXPLAIN /
// cbo_notes. The streaming engine stores flat row batches
// only, so flatten replays the child selection; the
// `group_columns`/`expected_groups` carried alongside are
// EXPLAIN traceability; the stale-position check runs in
// `open()` so a wrong group never replays silently.
group_pos,
group_columns: _,
expected_groups: _,
current_idx,
size_to_flatten,
saved_sel_vector,
buffered_chunk,
batch_size,
} = &mut op.kind
else {
return Err(QueryError::execution(
"flatten::handle called for a non-flatten unary operator".to_string(),
));
};
log::debug!("flatten: replaying selection for group {group_pos}");
if *batch_size <= 1 {
crate::executor::streaming::operators::flatten::flatten_next_inner(
current_idx,
size_to_flatten,
saved_sel_vector,
buffered_chunk,
input,
)
} else {
crate::executor::streaming::operators::flatten::flatten_next_batch(
current_idx,
size_to_flatten,
saved_sel_vector,
buffered_chunk,
input,
*batch_size,
)
}
}