fix align_by_record_delimiter

Author: youngsofun

src/query/pipeline/sources/src/processors/sources/input_formats/impls/input_format_ndjson.rs

@@ -105,7 +105,7 @@ impl InputFormatTextBase for InputFormatNDJson {
                     return Err(ErrorCode::BadBytes(msg));
                 }
             }
-            start = *end + 1;
+            start = *end;
         }
         Ok(())
     }

src/query/pipeline/sources/src/processors/sources/input_formats/impls/input_format_tsv.rs

@@ -115,15 +115,15 @@ impl InputFormatTextBase for InputFormatTSV {
         tracing::debug!(
             "tsv deserializing row batch {}, id={}, start_row={:?}, offset={}",
             batch.path,
-            batch.id,
+            batch.batch_id,
             batch.start_row,
             batch.offset
         );
         let columns = &mut builder.mutable_columns;
         let mut start = 0usize;
         let start_row = batch.start_row;
         for (i, end) in batch.row_ends.iter().enumerate() {
-            let buf = &batch.data[start..*end];
+            let buf = &batch.data[start..*end]; // include \n
             Self::read_row(
                 buf,
                 columns,
@@ -133,7 +133,7 @@ impl InputFormatTextBase for InputFormatTSV {
                 batch.offset + start,
                 start_row.map(|n| n + i),
             )?;
-            start = *end + 1;
+            start = *end;
         }
         Ok(())
     }

src/query/pipeline/sources/src/processors/sources/input_formats/input_format_text.rs

@@ -166,6 +166,7 @@ pub struct AligningState<T> {
 impl<T: InputFormatTextBase> AligningState<T> {
     pub fn align_by_record_delimiter(&mut self, buf_in: &[u8]) -> Vec<RowBatch> {
         let record_delimiter_end = self.record_delimiter_end;
+        let size_last_remain = self.tail_of_last_batch.len();
         let mut buf = buf_in;
         if self.rows_to_skip > 0 {
             let mut i = 0;
@@ -193,16 +194,17 @@ impl<T: InputFormatTextBase> AligningState<T> {
         let rows = &mut output.row_ends;
         for (i, b) in buf.iter().enumerate() {
             if *b == b'\n' {
-                rows.push(i)
+                rows.push(i + 1 + size_last_remain)
             }
         }
-        let last = rows[rows.len() - 1];
+        let batch_end = rows[rows.len() - 1] - size_last_remain;
         if rows.is_empty() {
             self.tail_of_last_batch.extend_from_slice(buf);
             vec![]
         } else {
             output.data = mem::take(&mut self.tail_of_last_batch);
-            output.data.extend_from_slice(&buf[0..last + 1]);
+            output.data.extend_from_slice(&buf[..batch_end]);
+            self.tail_of_last_batch.extend_from_slice(&buf[batch_end..]);
             let size = output.data.len();
             output.path = self.path.to_string();
             output.start_row = Some(self.rows);
@@ -212,11 +214,12 @@ impl<T: InputFormatTextBase> AligningState<T> {
             self.rows += rows.len();
             self.batch_id += 1;
             tracing::debug!(
-                "align {} bytes to {} rows: {} .. {}",
-                size,
+                "align batch {}, {} + {} + {} bytes to {} rows",
+                output.batch_id,
+                size_last_remain,
+                batch_end,
+                self.tail_of_last_batch.len(),
                 rows.len(),
-                rows[0],
-                last
             );
             vec![output]
         }
@@ -238,6 +241,12 @@ impl<T: InputFormatTextBase> AligningState<T> {
                 offset: self.offset,
                 start_row: Some(self.rows),
             };
+            tracing::debug!(
+                "align flush batch {}, bytes = {}, start_row = {}",
+                row_batch.batch_id,
+                self.tail_of_last_batch.len(),
+                self.rows
+            );
             vec![row_batch]
         }
     }