Merge pull request #8684 from youngsofun/fix2

fix(copy): fix wrong asserts.

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

@@ -352,7 +352,10 @@ impl<T: InputFormatTextBase> AligningStateTrait for AligningState<T> {
             T::align(self, &buf)?
         } else {
             if let Some(decoder) = &self.decoder {
-                assert_eq!(decoder.state(), DecompressState::Done)
+                let state = decoder.state();
+                if !matches!(state, DecompressState::Done | DecompressState::Reading) {
+                    tracing::warn!("decompressor end with state {:?}", state)
+                }
             }
             self.flush()
         };

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

@@ -21,6 +21,7 @@ use common_base::base::tokio::sync::mpsc::Sender;
 use common_base::base::GlobalIORuntime;
 use common_base::base::TrySpawn;
 use common_datablocks::DataBlock;
+use common_exception::ErrorCode;
 use common_exception::Result;
 use common_pipeline_core::Pipeline;
 use futures::AsyncRead;
@@ -288,6 +289,12 @@ pub trait InputFormatPipe: Sized + Send + 'static {
             let mut batch = vec![0u8; batch_size];
             let n = read_full(&mut reader, &mut batch[0..]).await?;
             if n == 0 {
+                if total_read != size {
+                    return Err(ErrorCode::BadBytes(format!(
+                        "split {} expect {} bytes, read only {} bytes",
+                        split_info, size, total_read
+                    )));
+                }
                 break;
             } else {
                 total_read += n;
@@ -299,7 +306,6 @@ pub trait InputFormatPipe: Sized + Send + 'static {
                 }
             }
         }
-        assert_eq!(total_read, size);
         tracing::debug!("finished");
         Ok(())
     }