feat: add a PropagateError plan

Signed-off-by: ekexium <ekexium@gmail.com>

src/request/mod.rs

@@ -11,7 +11,7 @@ use tikv_client_store::{HasError, Request};
 pub use self::{
     plan::{
         Collect, CollectError, DefaultProcessor, Dispatch, Merge, MergeResponse, MultiRegion, Plan,
-        Process, ProcessResponse, ResolveLock, RetryRegion,
+        Process, ProcessResponse, PropagateError, ResolveLock, RetryRegion,
     },
     plan_builder::{PlanBuilder, SingleKey},
     shard::Shardable,

src/request/plan.rs

@@ -249,6 +249,35 @@ where
     }
 }
 
+pub struct PropagateError<P: Plan> {
+    pub inner: P,
+}
+
+impl<P: Plan> Clone for PropagateError<P> {
+    fn clone(&self) -> Self {
+        PropagateError {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+#[async_trait]
+impl<P: Plan> Plan for PropagateError<P>
+where
+    P::Result: HasError,
+{
+    type Result = P::Result;
+
+    async fn execute(&self) -> Result<Self::Result> {
+        let mut result = self.inner.execute().await?;
+        if let Some(error) = result.error() {
+            Err(error)
+        } else {
+            Ok(result)
+        }
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

src/request/plan_builder.rs

@@ -5,7 +5,7 @@ use crate::{
     pd::PdClient,
     request::{
         DefaultProcessor, Dispatch, KvRequest, Merge, MergeResponse, MultiRegion, Plan, Process,
-        ProcessResponse, ResolveLock, RetryRegion, Shardable,
+        ProcessResponse, PropagateError, ResolveLock, RetryRegion, Shardable,
     },
     store::Store,
     transaction::HasLocks,
@@ -161,6 +161,19 @@ impl<PdC: PdClient, R: KvRequest> PlanBuilder<PdC, Dispatch<R>, NoTarget> {
     }
 }
 
+impl<PdC: PdClient, P: Plan> PlanBuilder<PdC, P, Targetted>
+where
+    P::Result: HasError,
+{
+    pub fn propagate_error(self) -> PlanBuilder<PdC, PropagateError<P>, Targetted> {
+        PlanBuilder {
+            pd_client: self.pd_client,
+            plan: PropagateError { inner: self.plan },
+            phantom: self.phantom,
+        }
+    }
+}
+
 fn set_single_region_store<PdC: PdClient, R: KvRequest>(
     mut plan: Dispatch<R>,
     store: Store,