chore(format): aligner not pass empty row batch to deserializer.
@@ -123,9 +123,11 @@ impl<I: InputFormatPipe> Processor for Aligner<I> {
let eof = read_batch.is_none();
let row_batches = state.align(read_batch)?;
for b in row_batches.into_iter() {
+ if b.size() > 0 {
process_values.rows += b.rows();
self.row_batches.push_back(b);
}
+ }
if eof {
assert!(self.is_flushing_split);
}
