feat: return per-entry region write failures (#9728)

* feat: return per-entry region write failures

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

* fix: preserve retry hints for single-entry region writes

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

* fix: preserve region write error compatibility

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

* refactor: simplify region partial result handling

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

* fix: align partial failure summaries with input order

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

* refactor: separate region write response handling

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

* test: fix partial result expectations and region map guard

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

* chore: limit lockfile changes to greptime proto

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

* fix: make partial region results an explicit client opt-in

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

* chore: pin greptime-proto to merged partial-result changes

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

---------

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

Cargo.lock

@@ -6083,7 +6083,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=9533078d6fd4bdaddbca0750cb9bf7ad64b2e568#9533078d6fd4bdaddbca0750cb9bf7ad64b2e568"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=acb247415b8f4c2f0750783dab4e1f38c71f5860#acb247415b8f4c2f0750783dab4e1f38c71f5860"
 dependencies = [
  "prost 0.14.1",
  "prost-types 0.14.1",

Cargo.toml

@@ -162,7 +162,7 @@ fs2 = "0.4"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "9533078d6fd4bdaddbca0750cb9bf7ad64b2e568" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "acb247415b8f4c2f0750783dab4e1f38c71f5860" }
 hdfs-native = "0.14.6"
 hex = "0.4"
 hostname = "0.4.0"

src/api/src/region.rs

@@ -15,7 +15,19 @@
 use std::collections::HashMap;
 
 use common_base::AffectedRows;
-use greptime_proto::v1::region::RegionResponse as RegionResponseV1;
+use greptime_proto::v1::region::{RegionRequestHeader, RegionResponse as RegionResponseV1};
+
+/// Request extension advertising support for per-entry write failures.
+pub const ACCEPT_PARTIAL_RESULT: &str = "accept_partial_result";
+
+/// Advertises support for per-entry write failures in the request header.
+pub fn set_accept_partial_result(header: &mut RegionRequestHeader) {
+    header
+        .query_context
+        .get_or_insert_with(Default::default)
+        .extensions
+        .insert(ACCEPT_PARTIAL_RESULT.to_string(), "true".to_string());
+}
 
 /// This result struct is derived from [RegionResponseV1]
 #[derive(Debug)]
@@ -25,6 +37,21 @@ pub struct RegionResponse {
     pub metadata: Vec<u8>,
 }
 
+/// Successful entries and failed-entry errors indexed by input order.
+#[derive(Debug)]
+pub struct BatchRegionResponse<E> {
+    pub affected_rows: AffectedRows,
+    pub extensions: HashMap<String, Vec<u8>>,
+    pub failures: Vec<(usize, E)>,
+}
+
+impl<E> BatchRegionResponse<E> {
+    /// Returns the failure with the smallest original request index.
+    pub fn first_failure(&self) -> Option<&(usize, E)> {
+        self.failures.iter().min_by_key(|(index, _)| *index)
+    }
+}
+
 impl RegionResponse {
     pub fn from_region_response(region_response: RegionResponseV1) -> Self {
         Self {

src/client/src/error.rs

@@ -18,7 +18,7 @@ use common_error::define_from_tonic_status;
 use common_error::ext::{BoxedError, ErrorExt, RetryHint};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use snafu::{Location, Snafu, location};
+use snafu::{Location, OptionExt, Snafu, location};
 use tonic::Code;
 use tonic::metadata::errors::InvalidMetadataValue;
 
@@ -26,6 +26,15 @@ use tonic::metadata::errors::InvalidMetadataValue;
 #[snafu(visibility(pub))]
 #[stack_trace_debug]
 pub enum Error {
+    #[snafu(display(
+        "Region request has {} failed entries; {}",
+        response.failures.len(),
+        response.first_failure().map(|(index, error)| format!("entry {index}: {}", error.output_msg())).unwrap_or_default()
+    ))]
+    PartialRegion {
+        response: Box<api::region::BatchRegionResponse<Error>>,
+    },
+
     #[snafu(display("Illegal Flight messages, reason: {}", reason))]
     IllegalFlightMessages {
         reason: String,
@@ -109,11 +118,12 @@ pub enum Error {
         location: Location,
     },
 
-    // Server error carried in Tonic Status's metadata.
+    // Server error carried in a protobuf response status.
     #[snafu(display("{}", msg))]
     Server {
         code: StatusCode,
         msg: String,
+        retry_hint: RetryHint,
         #[snafu(implicit)]
         location: Location,
     },
@@ -163,6 +173,9 @@ pub type Result<T> = std::result::Result<T, Error>;
 impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         match self {
+            Error::PartialRegion { response } => response
+                .first_failure()
+                .map_or(StatusCode::Internal, |(_, error)| error.status_code()),
             Error::IllegalFlightMessages { .. }
             | Error::MissingField { .. }
             | Error::IllegalDatabaseResponse { .. } => StatusCode::Internal,
@@ -188,7 +201,8 @@ impl ErrorExt for Error {
 
     fn retry_hint(&self) -> RetryHint {
         match self {
-            Error::Tonic { retry_hint, .. } => *retry_hint,
+            Error::PartialRegion { .. } => RetryHint::Retryable,
+            Error::Tonic { retry_hint, .. } | Error::Server { retry_hint, .. } => *retry_hint,
             Error::FlightGet { source, .. }
             | Error::FlightStream { source, .. }
             | Error::FlowServer { source, .. }
@@ -220,9 +234,22 @@ define_from_tonic_status!(Error, Tonic);
 
 impl From<Error> for common_meta::error::Error {
     fn from(error: Error) -> Self {
-        Self::External {
-            source: BoxedError::new(error),
-            location: location!(),
+        match error {
+            Error::PartialRegion { response } => Self::PartialRegion {
+                response: Box::new(api::region::BatchRegionResponse {
+                    affected_rows: response.affected_rows,
+                    extensions: response.extensions,
+                    failures: response
+                        .failures
+                        .into_iter()
+                        .map(|(index, error)| (index, error.into()))
+                        .collect(),
+                }),
+            },
+            error => Self::External {
+                source: BoxedError::new(error),
+                location: location!(),
+            },
         }
     }
 }
@@ -258,3 +285,20 @@ impl Error {
             )
     }
 }
+
+impl Error {
+    /// Decodes an application error carried in a protobuf response.
+    pub(crate) fn try_from_response_status(status: &api::v1::Status) -> Result<Self> {
+        let code = StatusCode::from_u32(status.status_code).with_context(|| {
+            IllegalDatabaseResponseSnafu {
+                err_msg: format!("Unknown server status: {}", status.status_code),
+            }
+        })?;
+        Ok(ServerSnafu {
+            code,
+            msg: status.err_msg.clone(),
+            retry_hint: status.retry_hint.parse().unwrap_or(RetryHint::NonRetryable),
+        }
+        .build())
+    }
+}

src/client/src/lib.rs

@@ -35,7 +35,7 @@ use snafu::OptionExt;
 pub use self::client::{Client, ClientOptions};
 pub use self::database::{Database, OutputMetrics, OutputWithMetrics};
 pub use self::error::{Error, Result};
-use crate::error::{IllegalDatabaseResponseSnafu, ServerSnafu};
+use crate::error::IllegalDatabaseResponseSnafu;
 
 pub fn from_grpc_response(response: GreptimeResponse) -> Result<u32> {
     let header = response.header.context(IllegalDatabaseResponseSnafu {
@@ -53,14 +53,6 @@ pub fn from_grpc_response(response: GreptimeResponse) -> Result<u32> {
             Response::AffectedRows(AffectedRows { value }) => Ok(value),
         }
     } else {
-        let status_code =
-            StatusCode::from_u32(status.status_code).context(IllegalDatabaseResponseSnafu {
-                err_msg: format!("invalid status: {:?}", status),
-            })?;
-        ServerSnafu {
-            code: status_code,
-            msg: status.err_msg,
-        }
-        .fail()
+        Err(Error::try_from_response_status(&status)?)
     }
 }

src/client/src/region.rs

@@ -18,8 +18,9 @@ use std::borrow::Borrow;
 use std::sync::Arc;
 use std::time::Duration;
 
-use api::region::RegionResponse;
+use api::region::{BatchRegionResponse, RegionResponse};
 use api::v1::ResponseHeader;
+use api::v1::region::RegionResponse as PbRegionResponse;
 use api::v1::region::{
     RegionRequest, RegionRequestHeader, RemoteDynFilterRequest, RemoteDynFilterUnregister,
     RemoteDynFilterUpdate, region_request, remote_dyn_filter_request,
@@ -49,7 +50,7 @@ use tonic::codec::CompressionEncoding;
 
 use crate::error::{
     self, FlightGetSnafu, IllegalDatabaseResponseSnafu, IllegalFlightMessagesSnafu,
-    MissingFieldSnafu, Result, ServerSnafu,
+    MissingFieldSnafu, Result,
 };
 use crate::flight::{FlightMessageReader, decode_flight_data};
 use crate::region::codec::RegionCodec;
@@ -141,7 +142,8 @@ impl RegionRequester {
     ///
     /// Accepts both [`RegionRequest`] and [`Arc<RegionRequest>`]. Callers can retain a
     /// shared request for another attempt; this method does not retry or determine
-    /// whether replaying the request is safe.
+    /// whether replaying the request is safe. Callers accepting partial results in the
+    /// request header must inspect failed entries before retrying.
     pub async fn handle<R>(&self, request: R) -> Result<RegionResponse>
     where
         R: Borrow<RegionRequest> + Send + Sync + 'static,
@@ -180,6 +182,11 @@ impl RegionRequester {
             .body
             .as_ref()
             .context(MissingFieldSnafu { field: "body" })?;
+        let request_count = match request_body {
+            region_request::Body::Inserts(requests) => Some(requests.requests.len()),
+            region_request::Body::Deletes(requests) => Some(requests.requests.len()),
+            _ => None,
+        };
         let is_insert = matches!(
             request_body,
             region_request::Body::Inserts(_) | region_request::Body::BulkInsert(_)
@@ -199,6 +206,13 @@ impl RegionRequester {
             }
         }
 
+        let accept_partial_result = request
+            .borrow()
+            .header
+            .as_ref()
+            .and_then(|header| header.query_context.as_ref())
+            .and_then(|context| context.extensions.get(api::region::ACCEPT_PARTIAL_RESULT))
+            .is_some_and(|value| value == "true");
         let response = async {
             client
                 .ready()
@@ -223,9 +237,10 @@ impl RegionRequester {
         .map_err(|status| map_region_error(status, addr))?
         .into_inner();
 
-        check_response_header(&response.header)?;
-
-        Ok(RegionResponse::from_region_response(response))
+        match request_count {
+            Some(count) if accept_partial_result => decode_write_response(response, count),
+            _ => decode_region_response(response),
+        }
     }
 }
 
@@ -385,6 +400,7 @@ fn build_remote_dyn_filter_request(
     }
 }
 
+/// Checks that the response contains a successful application status.
 pub fn check_response_header(header: &Option<ResponseHeader>) -> Result<()> {
     let status = header
         .as_ref()
@@ -396,35 +412,107 @@ pub fn check_response_header(header: &Option<ResponseHeader>) -> Result<()> {
     if StatusCode::is_success(status.status_code) {
         Ok(())
     } else {
-        let code =
-            StatusCode::from_u32(status.status_code).context(IllegalDatabaseResponseSnafu {
-                err_msg: format!("unknown server status: {:?}", status),
-            })?;
-        ServerSnafu {
-            code,
-            msg: status.err_msg.clone(),
-        }
-        .fail()
+        Err(error::Error::try_from_response_status(status)?)
     }
 }
 
+/// Decodes an ordinary region response without per-entry write failures.
+pub fn decode_region_response(response: PbRegionResponse) -> Result<RegionResponse> {
+    if !response.failures.is_empty() {
+        return IllegalDatabaseResponseSnafu {
+            err_msg: "Failure details are only valid for inserts/deletes",
+        }
+        .fail();
+    }
+    check_response_header(&response.header)?;
+    Ok(RegionResponse::from_region_response(response))
+}
+
+/// Decodes an inserts/deletes response, validating failures against the original entry count.
+/// A failed single-entry request returns its underlying error.
+fn decode_write_response(response: PbRegionResponse, entry_count: usize) -> Result<RegionResponse> {
+    if response.failures.is_empty() {
+        if response
+            .header
+            .as_ref()
+            .and_then(|header| header.status.as_ref())
+            .is_some_and(|status| status.status_code == StatusCode::PartialFailure as u32)
+        {
+            return IllegalDatabaseResponseSnafu {
+                err_msg: "Partial failure has no failed entries",
+            }
+            .fail();
+        }
+        return decode_region_response(response);
+    }
+    let status = response
+        .header
+        .as_ref()
+        .and_then(|header| header.status.as_ref())
+        .context(IllegalDatabaseResponseSnafu {
+            err_msg: "Missing region response status",
+        })?;
+    if status.status_code != StatusCode::PartialFailure as u32 {
+        return IllegalDatabaseResponseSnafu {
+            err_msg: "Failure details require a partial-failure status",
+        }
+        .fail();
+    }
+    let mut seen = std::collections::HashSet::with_capacity(response.failures.len());
+    let mut failures = Vec::with_capacity(response.failures.len());
+    for error in response.failures {
+        let index = error.request_index as usize;
+        if index >= entry_count || !seen.insert(index) {
+            return IllegalDatabaseResponseSnafu {
+                err_msg: "Invalid or duplicate batch error index",
+            }
+            .fail();
+        }
+        let status = error.status.context(IllegalDatabaseResponseSnafu {
+            err_msg: "Missing batch error status",
+        })?;
+        if StatusCode::is_success(status.status_code) {
+            return IllegalDatabaseResponseSnafu {
+                err_msg: "Batch error contains a success status",
+            }
+            .fail();
+        }
+        failures.push((index, error::Error::try_from_response_status(&status)?));
+    }
+    // A single failed entry has no successful peers to preserve.
+    if entry_count == 1
+        && let Some((_, error)) = failures.pop()
+    {
+        return Err(error);
+    }
+    Err(error::Error::PartialRegion {
+        response: Box::new(BatchRegionResponse {
+            affected_rows: response.affected_rows as _,
+            extensions: response.extensions,
+            failures,
+        }),
+    })
+}
+
 #[cfg(test)]
 #[allow(deprecated)]
 mod tests {
     use std::collections::VecDeque;
     use std::sync::Mutex;
 
+    use api::v1::ResponseHeader;
     use api::v1::region::region_server::{Region, RegionServer};
     use api::v1::region::{
-        BulkInsertRequest, InsertRequest, InsertRequests, RegionResponse as PbRegionResponse,
-        RemoteDynFilterUnregister, RemoteDynFilterUpdate, region_request,
-        remote_dyn_filter_request,
+        BulkInsertRequest, InsertRequest, InsertRequests, RegionRequestFailure,
+        RegionResponse as PbRegionResponse, RemoteDynFilterUnregister, RemoteDynFilterUpdate,
+        region_request, remote_dyn_filter_request,
     };
     use api::v1::value::ValueData;
     use api::v1::{
         ColumnDataType, ColumnSchema as PbColumnSchema, Row, Rows, Status as PbStatus, Value,
     };
     use common_error::ext::RetryHint;
+    use common_error::status_code::StatusCode;
     use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
     use common_recordbatch::adapter::RecordBatchMetrics;
     use datatypes::arrow::array::Int32Array;
@@ -579,6 +667,20 @@ mod tests {
         }
     }
 
+    fn with_partial_result(mut request: RegionRequest) -> RegionRequest {
+        request
+            .header
+            .get_or_insert_with(Default::default)
+            .query_context
+            .get_or_insert_with(Default::default)
+            .extensions
+            .insert(
+                api::region::ACCEPT_PARTIAL_RESULT.to_string(),
+                "true".to_string(),
+            );
+        request
+    }
+
     #[tokio::test]
     async fn test_owned_and_shared_requests_preserve_payload_and_compression() {
         let server = RegionTestServer::start().await;
@@ -636,6 +738,54 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_partial_result_requires_explicit_acceptance() {
+        let server = RegionTestServer::start().await;
+        let requester = server.create_requester(ChannelConfig::default(), false, false);
+        for accepted in [false, true] {
+            let request = if accepted {
+                with_partial_result(build_insert_request())
+            } else {
+                build_insert_request()
+            };
+            let shared = Arc::new(request);
+            server
+                .service
+                .responses
+                .lock()
+                .unwrap()
+                .push_back(Ok(PbRegionResponse {
+                    header: Some(ResponseHeader {
+                        status: Some(PbStatus {
+                            status_code: StatusCode::PartialFailure as u32,
+                            ..Default::default()
+                        }),
+                    }),
+                    failures: vec![RegionRequestFailure {
+                        request_index: 0,
+                        status: Some(PbStatus {
+                            status_code: StatusCode::RegionNotFound as u32,
+                            retry_hint: RetryHint::RetryAfterRouteRefresh.as_str().to_string(),
+                            ..Default::default()
+                        }),
+                    }],
+                    ..Default::default()
+                }));
+            let error = requester.handle(Arc::clone(&shared)).await.unwrap_err();
+            if accepted {
+                assert_eq!(error.status_code(), StatusCode::RegionNotFound);
+                assert_eq!(error.retry_hint(), RetryHint::RetryAfterRouteRefresh);
+            } else {
+                assert!(matches!(error, Error::IllegalDatabaseResponse { .. }));
+            }
+            assert_eq!(
+                &server.service.requests.lock().unwrap().last().unwrap().0,
+                shared.as_ref()
+            );
+            assert_eq!(Arc::strong_count(&shared), 1);
+        }
+    }
+
     #[tokio::test]
     async fn test_shared_request_can_be_resent_after_transport_error() {
         let server = RegionTestServer::start().await;
@@ -762,7 +912,10 @@ mod tests {
                 assert_eq!(error.tonic_code(), baseline.tonic_code());
             }
             let datanode: &dyn Datanode = &requester;
-            let error = datanode.handle(build_insert_request()).await.unwrap_err();
+            let error = datanode
+                .handle_request(Arc::new(build_insert_request()))
+                .await
+                .unwrap_err();
             assert_eq!(error.status_code(), StatusCode::RegionNotFound);
             assert_eq!(error.retry_hint(), hint);
             assert!(matches!(error, meta_error::Error::External { .. }));
@@ -832,10 +985,16 @@ mod tests {
                     body: Some(body),
                 };
                 let shared = Arc::new(request);
-                let shared_error = datanode.handle(Arc::clone(&shared)).await.unwrap_err();
+                let shared_error = datanode
+                    .handle_request(Arc::clone(&shared))
+                    .await
+                    .unwrap_err();
                 assert_eq!(Arc::strong_count(&shared), 1);
                 let request = Arc::try_unwrap(shared).unwrap();
-                let owned_error = datanode.handle(request).await.unwrap_err();
+                let owned_error = datanode
+                    .handle_request(Arc::new(request))
+                    .await
+                    .unwrap_err();
                 for error in [owned_error, shared_error] {
                     assert_eq!(
                         error.retry_hint(),
@@ -1053,8 +1212,281 @@ mod tests {
         .unwrap()
     }
 
+    #[tokio::test]
+    async fn test_partial_response_validation_and_compression() {
+        let server = RegionTestServer::start().await;
+        let requester = server.create_requester(ChannelConfig::default(), true, true);
+        let mut request = build_insert_request();
+        if let Some(region_request::Body::Inserts(inserts)) = &mut request.body {
+            inserts.requests.push(inserts.requests[0].clone());
+        }
+        let request = Arc::new(with_partial_result(request));
+        let failure = RegionRequestFailure {
+            request_index: 0,
+            status: Some(PbStatus {
+                status_code: StatusCode::RegionNotFound as u32,
+                err_msg: "moved".to_string(),
+                retry_hint: "retry_after_route_refresh".to_string(),
+            }),
+        };
+        for (failures, valid) in [
+            (vec![], true),
+            (vec![failure.clone()], true),
+            (vec![failure.clone(), failure.clone()], false),
+            (
+                vec![RegionRequestFailure {
+                    request_index: u32::MAX,
+                    ..failure.clone()
+                }],
+                false,
+            ),
+            (
+                vec![RegionRequestFailure {
+                    request_index: 0,
+                    status: None,
+                }],
+                false,
+            ),
+            (
+                vec![RegionRequestFailure {
+                    request_index: 0,
+                    status: Some(PbStatus::default()),
+                }],
+                false,
+            ),
+            (
+                vec![RegionRequestFailure {
+                    request_index: 0,
+                    status: Some(PbStatus {
+                        status_code: u32::MAX,
+                        ..Default::default()
+                    }),
+                }],
+                false,
+            ),
+        ] {
+            let has_error = !failures.is_empty();
+            server
+                .service
+                .responses
+                .lock()
+                .unwrap()
+                .push_back(Ok(PbRegionResponse {
+                    header: Some(ResponseHeader {
+                        status: Some(PbStatus {
+                            status_code: if has_error {
+                                StatusCode::PartialFailure
+                            } else {
+                                StatusCode::Success
+                            } as u32,
+                            ..Default::default()
+                        }),
+                    }),
+                    affected_rows: 3,
+                    extensions: std::collections::HashMap::from([("test".to_string(), vec![1, 2])]),
+                    failures,
+                    ..Default::default()
+                }));
+            let response = requester.handle_inner(Arc::clone(&request)).await;
+            if valid {
+                if has_error {
+                    let error = response.unwrap_err();
+                    assert!(error.output_msg().contains("entry 0"));
+                    assert!(error.output_msg().contains("moved"));
+                    let error: common_meta::error::Error = error.into();
+                    assert!(error.output_msg().contains("moved"));
+                    let common_meta::error::Error::PartialRegion { response } = error else {
+                        panic!("Expected partial region failure");
+                    };
+                    assert_eq!(response.affected_rows, 3);
+                    assert_eq!(response.extensions["test"], vec![1, 2]);
+                    assert_eq!(
+                        response.failures[0].1.retry_hint(),
+                        RetryHint::RetryAfterRouteRefresh
+                    );
+                } else {
+                    let response = response.unwrap();
+                    assert_eq!(response.affected_rows, 3);
+                    assert_eq!(response.extensions["test"], vec![1, 2]);
+                }
+            } else {
+                assert!(matches!(
+                    response.unwrap_err(),
+                    Error::IllegalDatabaseResponse { .. }
+                ));
+            }
+        }
+        for (received, send, accept) in server.service.requests.lock().unwrap().iter() {
+            assert_eq!(received, request.as_ref());
+            assert!(*send && *accept);
+        }
+        assert_eq!(Arc::strong_count(&request), 1);
+    }
+
+    #[test]
+    fn test_partial_response_envelope() {
+        let failure = RegionRequestFailure {
+            request_index: 0,
+            status: Some(PbStatus {
+                status_code: StatusCode::RegionNotFound as u32,
+                retry_hint: RetryHint::RetryAfterRouteRefresh.as_str().to_string(),
+                ..Default::default()
+            }),
+        };
+        for (status, failures, count) in [
+            (Some(StatusCode::Success), vec![failure.clone()], Some(2)),
+            (Some(StatusCode::PartialFailure), vec![], Some(2)),
+            (
+                Some(StatusCode::PartialFailure),
+                vec![failure.clone()],
+                None,
+            ),
+            (None, vec![failure.clone()], Some(2)),
+        ] {
+            let response = PbRegionResponse {
+                header: status.map(|code| ResponseHeader {
+                    status: Some(PbStatus {
+                        status_code: code as u32,
+                        ..Default::default()
+                    }),
+                }),
+                failures,
+                ..Default::default()
+            };
+            assert!(matches!(
+                match count {
+                    Some(count) => decode_write_response(response, count),
+                    None => decode_region_response(response),
+                },
+                Err(Error::IllegalDatabaseResponse { .. })
+            ));
+        }
+        let response = PbRegionResponse {
+            header: Some(ResponseHeader {
+                status: Some(PbStatus {
+                    status_code: StatusCode::PartialFailure as u32,
+                    ..Default::default()
+                }),
+            }),
+            failures: vec![failure],
+            ..Default::default()
+        };
+        // A partial envelope must not be interpreted as success by the header-only decoder.
+        assert!(check_response_header(&response.header).is_err());
+        let error = decode_write_response(response, 1).unwrap_err();
+        assert_eq!(error.status_code(), StatusCode::RegionNotFound);
+        assert_eq!(error.retry_hint(), RetryHint::RetryAfterRouteRefresh);
+    }
+
+    #[test]
+    fn test_partial_error_preserves_status_and_retry_hint() {
+        let expected_message = "Region request has 2 failed entries; entry 0: first input failed";
+        for code in [StatusCode::StorageUnavailable, StatusCode::InvalidArguments] {
+            let response = PbRegionResponse {
+                header: Some(ResponseHeader {
+                    status: Some(PbStatus {
+                        status_code: StatusCode::PartialFailure as u32,
+                        ..Default::default()
+                    }),
+                }),
+                failures: vec![
+                    RegionRequestFailure {
+                        request_index: 1,
+                        status: Some(PbStatus {
+                            status_code: StatusCode::Internal as u32,
+                            err_msg: "later input failed".to_string(),
+                            ..Default::default()
+                        }),
+                    },
+                    RegionRequestFailure {
+                        request_index: 0,
+                        status: Some(PbStatus {
+                            status_code: code as u32,
+                            err_msg: "first input failed".to_string(),
+                            retry_hint: RetryHint::NonRetryable.as_str().to_string(),
+                        }),
+                    },
+                ],
+                ..Default::default()
+            };
+            let error = decode_write_response(response, 2).unwrap_err();
+            assert_eq!(code, error.status_code());
+            assert_eq!(expected_message, error.to_string());
+            assert_eq!(RetryHint::Retryable, error.retry_hint());
+            let error: common_meta::error::Error = error.into();
+            assert_eq!(code, error.status_code());
+            assert_eq!(expected_message, error.to_string());
+            assert_eq!(RetryHint::Retryable, error.retry_hint());
+            let common_meta::error::Error::PartialRegion { response } = error else {
+                panic!("expected partial result")
+            };
+            assert!(
+                response
+                    .failures
+                    .iter()
+                    .all(|(_, error)| error.retry_hint() == RetryHint::NonRetryable)
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_region_rpc_errors() {
+        let server = RegionTestServer::start().await;
+        let requester = server.create_requester(ChannelConfig::default(), false, false);
+        let request = Arc::new(build_insert_request());
+        let mut business_error = Status::unimplemented("unsupported write");
+        business_error.metadata_mut().insert(
+            common_error::GREPTIME_DB_HEADER_ERROR_CODE,
+            (StatusCode::Unsupported as u32)
+                .to_string()
+                .parse()
+                .unwrap(),
+        );
+        for status in [
+            Status::unimplemented("unknown method"),
+            business_error,
+            Status::unavailable("lost response"),
+            Status::deadline_exceeded("timeout"),
+        ] {
+            let code = status.code();
+            server
+                .service
+                .responses
+                .lock()
+                .unwrap()
+                .push_back(Err(status));
+            let error = requester
+                .handle_inner(Arc::clone(&request))
+                .await
+                .unwrap_err();
+            assert!(matches!(error, Error::RegionServer { code: actual, .. } if actual == code));
+        }
+        assert_eq!(server.service.requests.lock().unwrap().len(), 4);
+    }
+
     #[test]
     fn test_check_response_header() {
+        for (hint, expected) in [
+            ("retryable", RetryHint::Retryable),
+            (
+                "retry_after_route_refresh",
+                RetryHint::RetryAfterRouteRefresh,
+            ),
+            ("non_retryable", RetryHint::NonRetryable),
+            ("", RetryHint::NonRetryable),
+            ("future_hint", RetryHint::NonRetryable),
+        ] {
+            let error = check_response_header(&Some(ResponseHeader {
+                status: Some(PbStatus {
+                    status_code: StatusCode::RegionNotFound as u32,
+                    err_msg: "region moved".to_string(),
+                    retry_hint: hint.to_string(),
+                }),
+            }))
+            .unwrap_err();
+            assert_eq!(error.retry_hint(), expected, "{hint}");
+        }
+
         let result = check_response_header(&None);
         assert!(matches!(
             result.unwrap_err(),
@@ -1071,6 +1503,7 @@ mod tests {
             status: Some(PbStatus {
                 status_code: StatusCode::Success as u32,
                 err_msg: String::default(),
+                ..Default::default()
             }),
         }));
         assert!(result.is_ok());
@@ -1079,6 +1512,7 @@ mod tests {
             status: Some(PbStatus {
                 status_code: u32::MAX,
                 err_msg: String::default(),
+                ..Default::default()
             }),
         }));
         assert!(matches!(
@@ -1090,6 +1524,7 @@ mod tests {
             status: Some(PbStatus {
                 status_code: StatusCode::Internal as u32,
                 err_msg: "blabla".to_string(),
+                ..Default::default()
             }),
         }));
         let Error::Server { code, msg, .. } = result.unwrap_err() else {

src/common/error/src/status_code.rs

@@ -45,6 +45,8 @@ pub enum StatusCode {
     DeadlineExceeded = 1008,
     /// Service got suspended for various reason. For example, resources exceed limit.
     Suspended = 1009,
+    /// One or more entries failed; inspect per-entry results before retrying.
+    PartialFailure = 1010,
     // ====== End of common status code ================
 
     // ====== Begin of SQL related status code =========
@@ -150,7 +152,8 @@ impl StatusCode {
             | StatusCode::RuntimeResourcesExhausted
             | StatusCode::External => true,
 
-            StatusCode::Success
+            StatusCode::PartialFailure
+            | StatusCode::Success
             | StatusCode::Unsupported
             | StatusCode::InvalidArguments
             | StatusCode::InvalidSyntax
@@ -300,7 +303,7 @@ macro_rules! define_into_tonic_status {
 pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
     match status_code {
         StatusCode::Success => Code::Ok,
-        StatusCode::Unknown | StatusCode::External => Code::Unknown,
+        StatusCode::Unknown | StatusCode::External | StatusCode::PartialFailure => Code::Unknown,
         StatusCode::Unsupported => Code::Unimplemented,
         StatusCode::Unexpected
         | StatusCode::IllegalState

src/common/meta/src/error.rs

@@ -41,6 +41,15 @@ pub use retry_hint::{retry_hint_from_postgres_error, retry_hint_from_postgres_po
 #[snafu(visibility(pub))]
 #[stack_trace_debug]
 pub enum Error {
+    #[snafu(display(
+        "Region request has {} failed entries; {}",
+        response.failures.len(),
+        response.first_failure().map(|(index, error)| format!("entry {index}: {}", error.output_msg())).unwrap_or_default()
+    ))]
+    PartialRegion {
+        response: Box<api::region::BatchRegionResponse<Error>>,
+    },
+
     #[snafu(display("Empty key is not allowed"))]
     EmptyKey {
         #[snafu(implicit)]
@@ -1195,6 +1204,9 @@ impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         use Error::*;
         match self {
+            Error::PartialRegion { response } => response
+                .first_failure()
+                .map_or(StatusCode::Internal, |(_, error)| error.status_code()),
             IllegalServerState { .. }
             | EtcdTxnOpResponse { .. }
             | EtcdFailed { .. }
@@ -1361,6 +1373,7 @@ impl ErrorExt for Error {
         use Error::*;
 
         match self {
+            Error::PartialRegion { .. } => RetryHint::Retryable,
             RetryLater { source, .. } => match source.retry_hint() {
                 RetryHint::RetryAfterRouteRefresh => RetryHint::RetryAfterRouteRefresh,
                 RetryHint::Retryable | RetryHint::NonRetryable => RetryHint::Retryable,

src/datanode/src/error.rs

@@ -30,6 +30,9 @@ use tokio::time::error::Elapsed;
 #[snafu(visibility(pub))]
 #[stack_trace_debug]
 pub enum Error {
+    #[snafu(display("Failed to write a batch group"))]
+    BatchWrite { source: std::sync::Arc<Error> },
+
     #[snafu(display("Failed to execute async task"))]
     AsyncTaskExecute {
         #[snafu(implicit)]
@@ -479,6 +482,7 @@ impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         use Error::*;
         match self {
+            Self::BatchWrite { source } => source.status_code(),
             NewPlanDecoder { source, .. } | ExecuteLogicalPlan { source, .. } => {
                 source.status_code()
             }
@@ -561,6 +565,7 @@ impl ErrorExt for Error {
         use Error::*;
 
         match self {
+            Self::BatchWrite { source } => source.retry_hint(),
             RegionNotReady { .. } | RegionNotFound { .. } => RetryHint::RetryAfterRouteRefresh,
             RegionBusy { .. }
             | ConcurrentQueryLimiterClosed { .. }

src/datanode/src/region_server/flow_write.rs

@@ -129,6 +129,7 @@ fn response(reply: Reply) -> ServerResult<api::v1::region::RegionResponse> {
         affected_rows: 0,
         extensions,
         metadata: Vec::new(),
+        failures: Vec::new(),
     })
 }
 

src/meta-srv/src/procedure/utils.rs

@@ -356,11 +356,13 @@ pub mod mock {
                     status: Some(PbStatus {
                         status_code: 0,
                         err_msg: String::default(),
+                        ..Default::default()
                     }),
                 }),
                 affected_rows: 0,
                 extensions: Default::default(),
                 metadata: Vec::new(),
+                failures: Vec::new(),
             })
         }
     }

src/metric-engine/AGENTS.md

@@ -51,6 +51,7 @@ Mito2 data region. Queries against a physical region pass straight through.
 ## Public surface
 
 - Entry: `MetricEngine` in `src/metric-engine/src/engine.rs`, built via `try_new(mito, config)`.
+- `MetricEngine::put_regions_batch` returns each physical group's outcome with its input indices; a failed group does not stop other groups.
 - Trait: `impl RegionEngine for MetricEngine` (name = `"metric"`).
 - Depends on `mito2`, `store-api`, and `mito-codec` (sparse primary key codec).
 

src/metric-engine/src/engine.rs

@@ -488,6 +488,17 @@ impl RegionEngine for MetricEngine {
     }
 }
 
+/// Outcome of a group of input requests.
+///
+/// Requests rejected before physical-region grouping produce singleton failures.
+#[derive(Debug)]
+pub struct PutGroupResult {
+    /// Positions in the input batch, including separate entries for the same region.
+    pub request_indices: Vec<usize>,
+    /// Aggregate rows on success; no other physical group's outcome is implied.
+    pub result: Result<AffectedRows>,
+}
+
 impl MetricEngine {
     pub fn try_new(mito: MitoEngine, mut config: EngineConfig) -> Result<Self> {
         let metadata_region = MetadataRegion::new(mito.clone());
@@ -520,13 +531,11 @@ impl MetricEngine {
         self.inner.mito.clone()
     }
 
-    /// Batch put operation for multiple logical regions.
-    /// Requests are grouped by physical region; a failure can leave earlier
-    /// physical-region groups committed.
+    /// Writes all physical-region groups and retains each group's outcome.
     pub async fn put_regions_batch(
         &self,
         requests: impl ExactSizeIterator<Item = (RegionId, RegionPutRequest)>,
-    ) -> Result<AffectedRows> {
+    ) -> Vec<PutGroupResult> {
         self.inner.put_regions_batch(requests).await
     }
 

src/metric-engine/src/engine/put.rs

@@ -29,7 +29,7 @@ use store_api::region_request::{
 };
 use store_api::storage::{RegionId, TableId};
 
-use crate::engine::MetricEngineInner;
+use crate::engine::{MetricEngineInner, PutGroupResult};
 use crate::error::{
     ColumnNotFoundSnafu, CreateDefaultSnafu, ForbiddenPhysicalWriteSnafu, InvalidRequestSnafu,
     LogicalRegionNotFoundSnafu, PhysicalRegionNotFoundSnafu, Result, UnexpectedRequestSnafu,
@@ -39,6 +39,12 @@ use crate::metrics::{FORBIDDEN_OPERATION_COUNT, MITO_OPERATION_ELAPSED};
 use crate::row_modifier::{RowsIter, TableIdInput};
 use crate::utils::to_data_region_id;
 
+#[derive(Default)]
+struct PutGroup {
+    request_indices: Vec<usize>,
+    requests: Vec<(RegionId, RegionPutRequest)>,
+}
+
 impl MetricEngineInner {
     /// Dispatch region put request
     pub async fn put_region(
@@ -61,67 +67,63 @@ impl MetricEngineInner {
         }
     }
 
-    /// Batch write multiple logical regions to the same physical region.
-    ///
-    /// Dispatch region put requests in batch.
-    ///
-    /// Requests may span multiple physical regions. We group them by physical
-    /// region and write sequentially. This method fails fast on validation or
-    /// preparation errors within a group and stops at the first failure.
-    /// Writes in earlier physical-region groups are not rolled back if a later
-    /// group fails.
+    /// Retains every group's outcome instead of returning on the first failure.
     pub async fn put_regions_batch(
         &self,
-        requests: impl ExactSizeIterator<Item = (RegionId, RegionPutRequest)>,
-    ) -> Result<AffectedRows> {
+        mut requests: impl ExactSizeIterator<Item = (RegionId, RegionPutRequest)>,
+    ) -> Vec<PutGroupResult> {
         let len = requests.len();
-
         if len == 0 {
-            return Ok(0);
+            return Vec::new();
         }
-
         let _timer = MITO_OPERATION_ELAPSED
             .with_label_values(&["put_batch"])
             .start_timer();
-
-        // Fast path: single request, no batching overhead
         if len == 1 {
-            let (region_id, req) = requests.into_iter().next().unwrap();
-            let is_putting_physical_region =
-                self.state.read().unwrap().exist_physical_region(region_id);
-            if is_putting_physical_region {
-                FORBIDDEN_OPERATION_COUNT.inc();
-                return ForbiddenPhysicalWriteSnafu.fail();
+            let (region_id, request) = requests.next().unwrap();
+            return vec![PutGroupResult {
+                request_indices: vec![0],
+                result: self.put_region(region_id, request).await,
+            }];
+        }
+        let mut groups = HashMap::new();
+        let mut results = Vec::new();
+        for (index, (region_id, request)) in requests.enumerate() {
+            if let Err(error) = self.group_put_request(&mut groups, index, region_id, request) {
+                results.push(PutGroupResult {
+                    request_indices: vec![index],
+                    result: Err(error),
+                });
             }
-
-            return self.put_logical_region(region_id, req).await;
         }
-
-        let mut requests_per_physical: HashMap<RegionId, Vec<(RegionId, RegionPutRequest)>> =
-            HashMap::new();
-        for (region_id, request) in requests {
-            let is_putting_physical_region =
-                self.state.read().unwrap().exist_physical_region(region_id);
-            if is_putting_physical_region {
-                FORBIDDEN_OPERATION_COUNT.inc();
-                return ForbiddenPhysicalWriteSnafu.fail();
-            }
-            let physical_region_id = self.find_physical_region_id(region_id)?;
-            requests_per_physical
-                .entry(physical_region_id)
-                .or_default()
-                .push((region_id, request));
+        for (physical_region_id, group) in groups {
+            let result = self
+                .put_regions_batch_single_physical(physical_region_id, group.requests)
+                .await;
+            results.push(PutGroupResult {
+                request_indices: group.request_indices,
+                result,
+            });
         }
+        results
+    }
 
-        let mut total_affected_rows: AffectedRows = 0;
-        for (physical_region_id, requests) in requests_per_physical {
-            let affected_rows = self
-                .put_regions_batch_single_physical(physical_region_id, requests)
-                .await?;
-            total_affected_rows += affected_rows;
+    fn group_put_request(
+        &self,
+        groups: &mut HashMap<RegionId, PutGroup>,
+        index: usize,
+        region_id: RegionId,
+        request: RegionPutRequest,
+    ) -> Result<()> {
+        if self.state.read().unwrap().exist_physical_region(region_id) {
+            FORBIDDEN_OPERATION_COUNT.inc();
+            return ForbiddenPhysicalWriteSnafu.fail();
         }
-
-        Ok(total_affected_rows)
+        let physical_region_id = self.find_physical_region_id(region_id)?;
+        let group = groups.entry(physical_region_id).or_default();
+        group.request_indices.push(index);
+        group.requests.push((region_id, request));
+        Ok(())
     }
 
     /// Write a batch of requests that all belong to the same physical region.
@@ -1001,7 +1003,9 @@ mod tests {
                     },
                 )
             });
-        let affected_rows = engine.inner.put_regions_batch(requests).await.unwrap();
+        let results = engine.put_regions_batch(requests).await;
+        assert_eq!(results.len(), 1);
+        let affected_rows = results.into_iter().next().unwrap().result.unwrap();
         assert_eq!(affected_rows, 6);
         assert_eq!(
             scan_timestamp_values(&engine, logical_region_id).await,
@@ -1376,12 +1380,13 @@ mod tests {
             for ((_, request), skip_wal) in mixed_requests.iter_mut().zip(policies) {
                 request.skip_wal = skip_wal;
             }
-            let err = env
+            let results = env
                 .metric()
-                .inner
                 .put_regions_batch(mixed_requests.into_iter())
-                .await
-                .unwrap_err();
+                .await;
+            assert_eq!(results.len(), 1);
+            assert_eq!(results[0].request_indices, vec![0, 1]);
+            let err = results.into_iter().next().unwrap().result.unwrap_err();
             assert!(err.to_string().contains("inconsistent WAL policy in batch"));
             for logical_region_id in [logical_region_1, logical_region_2] {
                 assert!(
@@ -1392,12 +1397,12 @@ mod tests {
             }
         }
 
-        let affected_rows = env
+        let results = env
             .metric()
-            .inner
             .put_regions_batch(build_requests().into_iter())
-            .await
-            .unwrap();
+            .await;
+        assert_eq!(results.len(), 1);
+        let affected_rows = results.into_iter().next().unwrap().result.unwrap();
         assert_eq!(affected_rows, 5);
 
         let request = ScanRequest::default();
@@ -1736,11 +1741,9 @@ mod tests {
         ];
 
         // Batch write
-        let affected_rows = engine
-            .inner
-            .put_regions_batch(requests.into_iter())
-            .await
-            .unwrap();
+        let results = engine.put_regions_batch(requests.into_iter()).await;
+        assert_eq!(results.len(), 1);
+        let affected_rows = results.into_iter().next().unwrap().result.unwrap();
         assert_eq!(affected_rows, 10);
 
         // Verify physical region contains data from all logical regions
@@ -1760,72 +1763,70 @@ mod tests {
     async fn test_batch_write_with_partial_failure() {
         let env = TestEnv::new().await;
         env.init_metric_region().await;
-        let engine = env.metric();
 
-        let physical_region_id = env.default_physical_region_id();
-        let logical_region_1 = env.default_logical_region_id();
-        let logical_region_2 = RegionId::new(1024, 1);
-        let nonexistent_region = RegionId::new(9999, 9999);
-
-        env.create_logical_region(physical_region_id, logical_region_2)
+        let physical = env.default_physical_region_id();
+        let logical = env.default_logical_region_id();
+        let other_physical = RegionId::new(4096, 0);
+        let other_logical = RegionId::new(4097, 0);
+        env.create_physical_region(other_physical, "batch_results", vec![])
             .await;
-
-        // Prepare batch with one invalid region
-        let schema = test_util::row_schema_with_tags(&["job"]);
-        let requests = vec![
-            (
-                logical_region_1,
-                RegionPutRequest {
-                    skip_wal: false,
-                    rows: Rows {
-                        schema: schema.clone(),
-                        rows: test_util::build_rows(1, 3),
-                    },
-                    hint: None,
-                    partition_expr_version: None,
+        env.create_logical_region(other_physical, other_logical)
+            .await;
+        let request = |index: i64, version| {
+            let mut rows = test_util::build_rows(1, 1);
+            rows[0].values[0].value_data =
+                Some(api::v1::value::ValueData::TimestampMillisecondValue(index));
+            RegionPutRequest {
+                rows: Rows {
+                    schema: test_util::row_schema_with_tags(&["job"]),
+                    rows,
                 },
-            ),
-            (
-                nonexistent_region,
-                RegionPutRequest {
-                    skip_wal: false,
-                    rows: Rows {
-                        schema: schema.clone(),
-                        rows: test_util::build_rows(1, 2),
-                    },
-                    hint: None,
-                    partition_expr_version: None,
-                },
-            ),
-            (
-                logical_region_2,
-                RegionPutRequest {
-                    skip_wal: false,
-                    rows: Rows {
-                        schema: schema.clone(),
-                        rows: test_util::build_rows(1, 5),
-                    },
-                    hint: None,
-                    partition_expr_version: None,
-                },
-            ),
-        ];
-
-        // Batch write
-        let result = engine.inner.put_regions_batch(requests.into_iter()).await;
-        assert!(result.is_err());
-
-        // Invalid region is detected before any write, so the physical region remains empty.
-        // Fail-fast is per physical-region group; cross-group partial success is possible.
-        let request = ScanRequest::default();
-        let stream = env
+                skip_wal: false,
+                hint: None,
+                partition_expr_version: version,
+            }
+        };
+        let mut results = env
             .metric()
-            .scan_to_stream(physical_region_id, request)
-            .await
-            .unwrap();
-        let batches = RecordBatches::try_collect(stream).await.unwrap();
-
-        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 0);
+            .put_regions_batch(
+                vec![
+                    (logical, request(0, None)),
+                    (RegionId::new(9999, 0), request(1, None)),
+                    (other_logical, request(2, Some(1))),
+                    (other_logical, request(3, None)),
+                    (logical, request(4, None)),
+                    (physical, request(5, None)),
+                ]
+                .into_iter(),
+            )
+            .await;
+        results.sort_by_key(|result| result.request_indices[0]);
+        assert_eq!(
+            results
+                .iter()
+                .map(|result| result.request_indices.clone())
+                .collect::<Vec<_>>(),
+            vec![vec![0, 4], vec![1], vec![2, 3], vec![5]]
+        );
+        assert_eq!(results[0].result.as_ref().unwrap(), &2);
+        assert!(results[1].result.is_err());
+        assert!(results[2].result.is_err());
+        assert!(matches!(
+            results[3].result,
+            Err(crate::error::Error::ForbiddenPhysicalWrite { .. })
+        ));
+        for (region, expected) in [(physical, 2), (other_physical, 0)] {
+            let stream = env
+                .metric()
+                .scan_to_stream(region, ScanRequest::default())
+                .await
+                .unwrap();
+            let batches = RecordBatches::try_collect(stream).await.unwrap();
+            assert_eq!(
+                batches.iter().map(|batch| batch.num_rows()).sum::<usize>(),
+                expected
+            );
+        }
     }
 
     #[tokio::test]
@@ -1849,11 +1850,10 @@ mod tests {
             },
         )];
 
-        let err = engine
-            .inner
-            .put_regions_batch(requests.into_iter())
-            .await
-            .unwrap_err();
+        let results = engine.put_regions_batch(requests.into_iter()).await;
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].request_indices, vec![0]);
+        let err = results.into_iter().next().unwrap().result.unwrap_err();
 
         assert!(matches!(
             err,
@@ -1897,15 +1897,15 @@ mod tests {
             ),
         ];
 
-        let err = engine
-            .inner
-            .put_regions_batch(requests.into_iter())
-            .await
-            .unwrap_err();
-
+        let mut results = engine.put_regions_batch(requests.into_iter()).await;
+        results.sort_by_key(|group| group.request_indices[0]);
+        assert_eq!(results.len(), 2);
+        assert_eq!(results[0].request_indices, vec![0]);
+        assert_eq!(results[0].result.as_ref().unwrap(), &1);
+        assert_eq!(results[1].request_indices, vec![1]);
         assert!(matches!(
-            err,
-            crate::error::Error::ForbiddenPhysicalWrite { .. }
+            results[1].result,
+            Err(crate::error::Error::ForbiddenPhysicalWrite { .. })
         ));
     }
 
@@ -1932,11 +1932,10 @@ mod tests {
             },
         )];
 
-        let affected_rows = engine
-            .inner
-            .put_regions_batch(requests.into_iter())
-            .await
-            .unwrap();
+        let results = engine.put_regions_batch(requests.into_iter()).await;
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].request_indices, vec![0]);
+        let affected_rows = results.into_iter().next().unwrap().result.unwrap();
         assert_eq!(affected_rows, 5);
     }
 
@@ -1946,15 +1945,12 @@ mod tests {
         env.init_metric_region().await;
         let engine = env.metric();
 
-        // Empty batch should return zero affected rows
-        let requests = vec![];
-        let affected_rows = engine
-            .inner
-            .put_regions_batch(requests.into_iter())
-            .await
-            .unwrap();
-
-        assert_eq!(affected_rows, 0);
+        assert!(
+            engine
+                .put_regions_batch(vec![].into_iter())
+                .await
+                .is_empty()
+        );
     }
 
     #[tokio::test]

src/operator/src/write_dispatcher.rs

@@ -203,10 +203,13 @@ impl WriteDispatcher {
     pub(crate) async fn send(
         &self,
         peer: PeerRef,
-        request: RegionRequest,
+        mut request: RegionRequest,
         histogram: Option<&Histogram>,
     ) -> WriteAttempt {
         if let Some(retry) = &self.retry {
+            if matches!(&request.body, Some(Body::Inserts(_) | Body::Deletes(_))) {
+                api::region::set_accept_partial_result(request.header.get_or_insert_default());
+            }
             return retry
                 .send(
                     RoutedWrite {

src/operator/src/write_dispatcher/tests.rs

@@ -356,6 +356,15 @@ async fn test_write_retry_disabled_and_enabled_ablation() {
             );
             assert_eq!(result.is_ok(), max_retries > 0);
             assert_eq!(attempts, max_retries + 1);
+            for request in server.service.requests.lock().unwrap().iter() {
+                let accepts_partial = request
+                    .header
+                    .as_ref()
+                    .and_then(|header| header.query_context.as_ref())
+                    .and_then(|context| context.extensions.get(api::region::ACCEPT_PARTIAL_RESULT))
+                    .is_some_and(|value| value == "true");
+                assert_eq!(accepts_partial, max_retries > 0);
+            }
             assert!(server.invalidator.calls.lock().unwrap().is_empty());
         }
     }
@@ -1709,11 +1718,11 @@ async fn check_write_retry_rebuilds_logical_rows_in_mixed_table_envelope(kind: W
     );
     let mut requests = server.service.requests.lock().unwrap().clone();
     requests.sort_by_key(|request| write_region_id(request).unwrap().as_u64());
-    assert_eq!(
-        requests,
-        vec![physical, rebuilt_upper, logical, rebuilt_lower],
-        "{kind:?}"
-    );
+    let mut expected = vec![physical, rebuilt_upper, logical, rebuilt_lower];
+    for request in &mut expected {
+        api::region::set_accept_partial_result(request.header.get_or_insert_default());
+    }
+    assert_eq!(requests, expected, "{kind:?}");
     let calls = invalidator.calls.lock().unwrap();
     assert_eq!(calls.len(), 2, "{kind:?}");
     assert_eq!(calls[0].len(), 2, "{kind:?}");

src/servers/src/error.rs

@@ -1014,6 +1014,7 @@ pub fn status_code_to_http_status(status_code: &StatusCode) -> HttpStatusCode {
         StatusCode::Internal
         | StatusCode::Unexpected
         | StatusCode::IllegalState
+        | StatusCode::PartialFailure
         | StatusCode::Unknown
         | StatusCode::RuntimeResourcesExhausted
         | StatusCode::EngineExecuteQuery => HttpStatusCode::INTERNAL_SERVER_ERROR,

src/servers/src/mysql/writer.rs

@@ -512,7 +512,9 @@ pub fn create_mysql_column_def(schema: &SchemaRef) -> Result<Vec<Column>> {
 fn mysql_error_kind(status_code: &StatusCode) -> ErrorKind {
     match status_code {
         StatusCode::Success => ErrorKind::ER_YES,
-        StatusCode::Unknown | StatusCode::External => ErrorKind::ER_UNKNOWN_ERROR,
+        StatusCode::PartialFailure | StatusCode::Unknown | StatusCode::External => {
+            ErrorKind::ER_UNKNOWN_ERROR
+        }
         StatusCode::Unsupported => ErrorKind::ER_NOT_SUPPORTED_YET,
         StatusCode::Cancelled | StatusCode::DeadlineExceeded => ErrorKind::ER_QUERY_INTERRUPTED,
         StatusCode::RuntimeResourcesExhausted => ErrorKind::ER_OUT_OF_RESOURCES,

src/servers/src/postgres/types/error.rs

@@ -382,7 +382,8 @@ impl From<StatusCode> for PgErrorCode {
             StatusCode::DeadlineExceeded => PgErrorCode::Ec57000,
             StatusCode::External => PgErrorCode::Ec58000,
 
-            StatusCode::Unknown
+            StatusCode::PartialFailure
+            | StatusCode::Unknown
             | StatusCode::Unexpected
             | StatusCode::IllegalState
             | StatusCode::Internal => PgErrorCode::EcXX000,

src/standalone/Cargo.toml

@@ -46,5 +46,7 @@ tokio.workspace = true
 url.workspace = true
 
 [dev-dependencies]
+datanode = { workspace = true, features = ["testing"] }
+mito2 = { workspace = true, features = ["test"] }
 temp-env.workspace = true
 toml.workspace = true

src/standalone/src/datanode_manager.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 use api::region::RegionResponse;
 use api::v1::region::{RegionRequest, RegionResponse as RegionResponseV1};
 use async_trait::async_trait;
-use client::region::check_response_header;
+use client::region;
 use common_error::ext::BoxedError;
 use common_meta::error::{self as meta_error, Result as MetaResult};
 use common_meta::node_manager::{
@@ -96,10 +96,7 @@ impl Datanode for RegionInvoker {
             .await
             .map_err(BoxedError::new)
             .context(meta_error::ExternalSnafu)?;
-        check_response_header(&response.header)
-            .map_err(BoxedError::new)
-            .context(meta_error::ExternalSnafu)?;
-        Ok(RegionResponse::from_region_response(response))
+        region::decode_region_response(response).map_err(Into::into)
     }
 
     async fn handle_query(&self, request: QueryRequest) -> MetaResult<SendableRecordBatchStream> {
@@ -121,3 +118,65 @@ impl Datanode for RegionInvoker {
             .context(meta_error::ExternalSnafu)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use api::v1::Rows;
+    use api::v1::region::{InsertRequest, InsertRequests, region_request};
+    use common_error::ext::ErrorExt;
+    use common_error::mock::MockError;
+    use common_error::status_code::StatusCode;
+    use datanode::tests::{MockRegionEngine, mock_region_server};
+    use mito2::engine::MITO_ENGINE_NAME;
+    use mito2::test_util::CreateRequestBuilder;
+    use store_api::region_request::RegionRequest as StoreRegionRequest;
+    use store_api::storage::RegionId;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn test_write_preserves_original_error() {
+        for code in [StatusCode::StorageUnavailable, StatusCode::InvalidArguments] {
+            let mut server = mock_region_server();
+            let (engine, _receiver) = MockRegionEngine::with_mock_fn(
+                MITO_ENGINE_NAME,
+                Box::new(move |region_id, request| {
+                    if matches!(request, StoreRegionRequest::Put(_)) {
+                        Err(BoxedError::new(MockError::new(code)))
+                            .context(datanode::error::HandleRegionRequestSnafu { region_id })
+                    } else {
+                        Ok(0)
+                    }
+                }),
+            );
+            server.register_engine(engine);
+            let mut requests = Vec::new();
+            for number in 0..2 {
+                let region_id = RegionId::new(1, number);
+                server
+                    .handle_request(
+                        region_id,
+                        StoreRegionRequest::Create(CreateRequestBuilder::new().build()),
+                    )
+                    .await
+                    .unwrap();
+                requests.push(InsertRequest {
+                    region_id: region_id.as_u64(),
+                    rows: Some(Rows::default()),
+                    ..Default::default()
+                });
+            }
+            let datanode: DatanodeRef = RegionInvoker::arc(server);
+            let error = datanode
+                .handle(RegionRequest {
+                    header: Some(Default::default()),
+                    body: Some(region_request::Body::Inserts(InsertRequests { requests })),
+                })
+                .await
+                .unwrap_err();
+            assert_eq!(code, error.status_code());
+            assert_eq!(MockError::new(code).retry_hint(), error.retry_hint());
+            assert!(!matches!(error, meta_error::Error::PartialRegion { .. }));
+        }
+    }
+}

src/store-api/src/region_request.rs

@@ -213,47 +213,55 @@ impl RegionRequest {
     }
 }
 
+/// Converts one insert, treating absent rows as a no-op like vector inserts.
+pub fn make_region_put(
+    request: api::v1::region::InsertRequest,
+) -> Option<(RegionId, RegionPutRequest)> {
+    request.rows.map(|rows| {
+        (
+            request.region_id.into(),
+            RegionPutRequest {
+                rows,
+                hint: None,
+                skip_wal: request.skip_wal,
+                partition_expr_version: request.partition_expr_version.map(|version| version.value),
+            },
+        )
+    })
+}
+
 fn make_region_puts(inserts: InsertRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
-    let requests = inserts
+    Ok(inserts
         .requests
         .into_iter()
-        .filter_map(|r| {
-            let region_id = r.region_id.into();
-            r.rows.map(|rows| {
-                (
-                    region_id,
-                    RegionRequest::Put(RegionPutRequest {
-                        rows,
-                        hint: None,
-                        skip_wal: r.skip_wal,
-                        partition_expr_version: r.partition_expr_version.map(|v| v.value),
-                    }),
-                )
-            })
-        })
-        .collect();
-    Ok(requests)
+        .filter_map(make_region_put)
+        .map(|(region_id, request)| (region_id, RegionRequest::Put(request)))
+        .collect())
+}
+
+/// Converts one delete, treating absent rows as a no-op like vector deletes.
+pub fn make_region_delete(
+    request: api::v1::region::DeleteRequest,
+) -> Option<(RegionId, RegionDeleteRequest)> {
+    request.rows.map(|rows| {
+        (
+            request.region_id.into(),
+            RegionDeleteRequest {
+                rows,
+                hint: None,
+                partition_expr_version: request.partition_expr_version.map(|version| version.value),
+            },
+        )
+    })
 }
 
 fn make_region_deletes(deletes: DeleteRequests) -> Result<Vec<(RegionId, RegionRequest)>> {
-    let requests = deletes
+    Ok(deletes
         .requests
         .into_iter()
-        .filter_map(|r| {
-            let region_id = r.region_id.into();
-            r.rows.map(|rows| {
-                (
-                    region_id,
-                    RegionRequest::Delete(RegionDeleteRequest {
-                        rows,
-                        hint: None,
-                        partition_expr_version: r.partition_expr_version.map(|v| v.value),
-                    }),
-                )
-            })
-        })
-        .collect();
-    Ok(requests)
+        .filter_map(make_region_delete)
+        .map(|(region_id, request)| (region_id, RegionRequest::Delete(request)))
+        .collect())
 }
 
 fn parse_region_create(create: CreateRequest) -> Result<(RegionId, RegionCreateRequest)> {

tests/runner/src/client.rs

@@ -169,6 +169,7 @@ impl MultiProtocolClient {
                         result = ServerSnafu {
                             code: status_code,
                             msg,
+                            retry_hint: e.retry_hint(),
                         }
                         .fail();
                     }