[adapters] Fix record loss when splitting a nested input buffer - #6787
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
Nice catch. The old remaining >= len guard was almost tautological — take_some(n) returns at most n, so it fired on nearly every partial drain and dropped the surviving records with the buffer. Fix reads cleanly: only advance drained once the source buffer is actually empty, break on the first one that kept records.
Tests are exactly what I'd want here — take_some_loses_no_records_across_batches sweeping batch sizes 1..=9 against three nested buffers is a strong regression net, and the S3 test wiring a PassthroughPreprocessorFactory + capped max_batch_size reproduces the #6709 hang scenario end-to-end. Good comment work explaining why NDJSON matters (LineSplitter granularity) versus CSV.
One tiny wart: if let Some(buffer) = v.take_some(remaining) then unconditionally v.len().records > 0 — if take_some returned None (empty source) we still fall through to the same len().records > 0 check, which correctly reads 0 and lets us advance drained. Works, but it means "buffer that couldn't produce anything" and "buffer that was fully drained" take the same path. Fine as written; only noting because I had to re-read it once.
mihaibudiu
left a comment
There was a problem hiding this comment.
wasn't this also losing some data?
`InputBuffer::take_some` on `Vec<Box<dyn InputBuffer>>` decided a nested
element was fully consumed from the size of the chunk that element handed
back:
if remaining >= len { index += 1; } // "completely used"
...
self.drain(0..index);
`take_some(n)` returns at most `n`, so that test holds on essentially every
iteration. A partially drained element was therefore counted as spent and
deleted along with its leftover records. Draining `[1,2,3][4,5][6,7,8,9]` one
record at a time delivered only `[1, 4, 6]`.
The lost records stay counted in `total_input_records`, because the connector
reported them through `InputConsumer::buffered` before queueing them, but
never reach `processed_data`. `total_processed_records` then trails
`total_input_records` forever, so `ControllerStatus::pipeline_complete` can
never return true and a pipeline waiting for completion hangs.
Drop an element only once it holds no more records, and stop at the first
element that kept some.
Signed-off-by: Leonid Ryzhyk <leonid@feldera.com>
2712197 to
034539a
Compare
blp
left a comment
There was a problem hiding this comment.
This seems like it was a serious bug. I am surprised that it did not cause serious problems. Maybe we did not use small n much.
Apparently only the S3 connector was affected |
Fix #6709
InputBuffer::take_someonVec<Box<dyn InputBuffer>>decided a nested element was fully consumed from the size of the chunk that element handed back:take_some(n)returns at mostn, so that test holds on essentially every iteration. A partially drained element was therefore counted as spent and deleted along with its leftover records. Draining[1,2,3][4,5][6,7,8,9]one record at a time delivered only[1, 4, 6].The lost records stay counted in
total_input_records, because the connector reported them throughInputConsumer::bufferedbefore queueing them, but never reachprocessed_data.total_processed_recordsthen trailstotal_input_recordsforever, soControllerStatus::pipeline_completecan never return true and a pipeline waiting for completion hangs.Drop an element only once it holds no more records, and stop at the first element that kept some.
Describe Manual Test Plan
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes