return error when copy
@@ -211,13 +211,9 @@ pub trait InputFormatPipe: Sized + Send + 'static {
while let Some(row_batch) = futs.next().await {
match row_batch {
Ok(row_batch) => {
- if row_batch.is_ok() {
if data_tx2.send(row_batch).await.is_err() {
break;
}
- } else {
- break;
- }
}
Err(cause) => {
data_tx2.send(Err(cause)).await.ok();
