add RowBatch.batch_id to debug aligner.

Author: youngsofun

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

@@ -177,6 +177,7 @@ impl InputFormatTextBase for InputFormatCSV {
             row_ends: vec![],
             field_ends: vec![],
             path: state.path.to_string(),
+            batch_id: state.batch_id,
             offset: 0,
             start_row: None,
         };
@@ -256,6 +257,7 @@ impl InputFormatTextBase for InputFormatCSV {
             } else {
                 vec![last_remain, out_tmp].concat()
             };
+            state.batch_id += 1;
             Ok(vec![row_batch])
         }
     }

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

@@ -34,6 +34,7 @@ impl InputFormatTSV {
         deserializers: &mut Vec<common_datavalues::TypeDeserializerImpl>,
         format_settings: &FormatSettings,
         path: &str,
+        batch_id: usize,
         offset: usize,
         row_index: Option<usize>,
     ) -> Result<()> {
@@ -86,8 +87,9 @@ impl InputFormatTSV {
                 String::new()
             };
             let mut msg = format!(
-                "fail to parse tsv {} at offset {}, {} reason={}, row data: ",
+                "fail to parse tsv {} batch {} at offset {}, {} reason={}, row data: ",
                 path,
+                batch_id,
                 offset + pos,
                 row_info,
                 m
@@ -110,6 +112,13 @@ impl InputFormatTextBase for InputFormatTSV {
     }
 
     fn deserialize(builder: &mut BlockBuilder<Self>, batch: RowBatch) -> Result<()> {
+        tracing::debug!(
+            "tsv deserializing row batch {}, id={}, start_row={:?}, offset={}",
+            batch.path,
+            batch.id,
+            batch.start_row,
+            batch.offset
+        );
         let columns = &mut builder.mutable_columns;
         let mut start = 0usize;
         let start_row = batch.start_row;
@@ -120,6 +129,7 @@ impl InputFormatTextBase for InputFormatTSV {
                 columns,
                 &builder.ctx.format_settings,
                 &batch.path,
+                batch.batch_id,
                 batch.offset + start,
                 start_row.map(|n| n + i),
             )?;

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

@@ -143,6 +143,7 @@ pub struct RowBatch {
 
     // for error info
     pub path: String,
+    pub batch_id: usize,
     pub offset: usize,
     pub start_row: Option<usize>,
 }
@@ -151,6 +152,7 @@ pub struct AligningState<T> {
     pub path: String,
     pub record_delimiter_end: u8,
     pub field_delimiter: u8,
+    pub batch_id: usize,
     pub rows: usize,
     pub offset: usize,
     pub rows_to_skip: usize,
@@ -203,7 +205,12 @@ impl<T: InputFormatTextBase> AligningState<T> {
             output.data.extend_from_slice(&buf[0..last + 1]);
             let size = output.data.len();
             output.path = self.path.to_string();
+            output.start_row = Some(self.rows);
+            output.offset = self.offset;
+            output.batch_id = self.batch_id;
             self.offset += size;
+            self.rows += rows.len();
+            self.batch_id += 1;
             tracing::debug!(
                 "align {} bytes to {} rows: {} .. {}",
                 size,
@@ -227,8 +234,9 @@ impl<T: InputFormatTextBase> AligningState<T> {
                 row_ends: vec![end],
                 field_ends: vec![],
                 path: self.path.to_string(),
+                batch_id: self.batch_id,
                 offset: self.offset,
-                start_row: None,
+                start_row: Some(self.rows),
             };
             vec![row_batch]
         }
@@ -260,6 +268,7 @@ impl<T: InputFormatTextBase> AligningStateTrait for AligningState<T> {
             csv_reader,
             tail_of_last_batch: vec![],
             rows: 0,
+            batch_id: 0,
             num_fields: ctx.schema.num_fields(),
             offset: split_info.offset,
             record_delimiter_end: ctx.record_delimiter.end(),