fix: introduce batch flush error with its table batcher consumer

Signed-off-by: WenyXu <wenymedia@gmail.com>

src/operator/src/error.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::any::Any;
+use std::sync::Arc;
 
 use common_datasource::file_format::Format;
 use common_error::define_into_tonic_status;
@@ -101,6 +102,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to flush pending batch: {source}"))]
+    BatchFlush {
+        source: Arc<Error>,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("external error"))]
     External {
         #[snafu(implicit)]
@@ -1051,6 +1059,7 @@ impl ErrorExt for Error {
             | Error::DescribeStatement { source, .. } => source.status_code(),
             Error::AlterExprToRequest { source, .. } => source.status_code(),
             Error::External { source, .. } => source.status_code(),
+            Error::BatchFlush { source, .. } => source.status_code(),
             Error::FindTablePartitionRule { source, .. }
             | Error::SplitInsert { source, .. }
             | Error::SplitDelete { source, .. }
@@ -1156,6 +1165,7 @@ impl ErrorExt for Error {
             Error::Catalog { source, .. } => source.retry_hint(),
             Error::SubstraitCodec { source, .. } => source.retry_hint(),
             Error::External { source, .. } => source.retry_hint(),
+            Error::BatchFlush { source, .. } => source.retry_hint(),
             Error::BuildRecordBatch { source, .. } => source.retry_hint(),
             Error::DecodeFlightData { source, .. } => source.retry_hint(),
             Error::SqlCommon { source, .. } => source.retry_hint(),