enhance: classify packed extend status codes into segcore error codes

cpp/conanfile.py

@@ -167,7 +167,7 @@ def requirements(self):
         self.requires("libavrocpp/1.12.1.1@milvus/dev#cde7bb587a29f6f233bae7e18b71815d")
         self.requires("google-cloud-cpp/2.28.0@milvus/dev#468918b43cec43624531a0340398cf43")
         self.requires("opentelemetry-cpp/1.23.0@milvus/dev#11bc565ec6e82910ae8f7471da756720")
-        self.requires("milvus-common/1.0.0-9ca5ea6@milvus/dev#274d428d85f1d3d996e1092f0c9c7144")
+        self.requires("milvus-common/1.0.0-60a563c@milvus/dev#a7448f82ed17d10934eacb6d1b152fd8")
         # azure-sdk-for-cpp is a transitive dep of Arrow, but must be declared
         # as a direct dep so CMakeDeps generates standalone cmake config files.
         # Without this, find_package(Azure) can't find include directories.

cpp/include/milvus-storage/common/extend_status.h

@@ -22,12 +22,31 @@
 #include <arrow/status.h>
 #include <arrow/result.h>
 
+// from milvus-common repo
+#include "common/EasyAssert.h"
+
 namespace milvus_storage {
 enum class ExtendStatusCode : char {
   // arrow::StatusCode biggest is 45
+  // Packed-specific error codes.
+  PackedInvalidArgs = 50,
+  PackedStorageIO = 51,
+  PackedMetadataCorrupted = 52,
+  PackedFileCorrupted = 53,
+  PackedArrowError = 54,
+  PackedUnexpected = 55,
+
   AwsErrorNoSuchUpload = 101,
   AwsErrorConflict = 102,
   AwsErrorPreConditionFailed = 103,
+  // Permanently-failing object-storage errors that must NOT be classified as
+  // transient/retriable by consumers: the object/bucket is gone (retrying or
+  // rerouting to another replica hits the same shared object store and fails
+  // identically), the credentials/permissions are wrong, or the AWS SDK itself
+  // judged the error non-retryable (AWSError::ShouldRetry() == false).
+  AwsErrorNotFound = 104,      // NoSuchKey / NoSuchBucket / ResourceNotFound
+  AwsErrorAccessDenied = 105,  // AccessDenied / InvalidAccessKeyId / SignatureDoesNotMatch
+  AwsErrorNonRetryable = 106,  // any other error with ShouldRetry() == false
 
   // Transaction-specific error codes
   TxnExhaustedRetry = 111,
@@ -67,6 +86,12 @@ class ExtendStatusDetail : public arrow::StatusDetail {
   std::string extra_info_;
 };
 
-arrow::Status MakeExtendError(ExtendStatusCode code, std::string message, std::string extra_info);
+arrow::Status MakeExtendError(ExtendStatusCode code, std::string message, std::string extra_info = "");
+
+arrow::Status WrapExtendError(ExtendStatusCode code, std::string message, const arrow::Status& cause);
+
+milvus::ErrorCode ToSegcoreErrorCode(ExtendStatusCode code);
+
+milvus::SegcoreError ToSegcoreError(const arrow::Status& status);
 
-}  // namespace milvus_storage
+}  // namespace milvus_storage

cpp/include/milvus-storage/filesystem/s3/s3_internal.h

@@ -196,22 +196,52 @@ arrow::Status ErrorToStatus(const std::string& prefix,
   switch (error_type) {
     case Aws::S3::S3Errors::NO_SUCH_UPLOAD:
       return MakeExtendError(ExtendStatusCode::AwsErrorNoSuchUpload, message, message /* extra_info */);
+    // Permanent errors: mark them so consumers do not classify them as
+    // transient/retriable (a retry or replica-reroute hits the same shared
+    // object store and fails identically).
+    case Aws::S3::S3Errors::NO_SUCH_BUCKET:
+    case Aws::S3::S3Errors::NO_SUCH_KEY:
+    case Aws::S3::S3Errors::RESOURCE_NOT_FOUND:
+      return MakeExtendError(ExtendStatusCode::AwsErrorNotFound, message, message /* extra_info */);
+    case Aws::S3::S3Errors::ACCESS_DENIED:
+    case Aws::S3::S3Errors::INVALID_ACCESS_KEY_ID:
+    case Aws::S3::S3Errors::SIGNATURE_DOES_NOT_MATCH:
+      return MakeExtendError(ExtendStatusCode::AwsErrorAccessDenied, message, message /* extra_info */);
     case Aws::S3::S3Errors::UNKNOWN: {
       switch (error.GetResponseCode()) {
         case Aws::Http::HttpResponseCode::PRECONDITION_FAILED:
           return MakeExtendError(ExtendStatusCode::AwsErrorPreConditionFailed, message, message /* extra_info */);
         case Aws::Http::HttpResponseCode::CONFLICT:
           return MakeExtendError(ExtendStatusCode::AwsErrorConflict, message, message /* extra_info */);
         default:
-          [[fallthrough]];
+          break;
       };
-
-      // fallthrough
+      break;
     }
     default:
-      [[fallthrough]];
+      break;
+  }
+
+  // The AWS SDK carries its own retryability verdict (the same one its internal
+  // retry loop used). An escaped error the SDK itself would not retry is
+  // permanent -- tag it so it does not fall into the plain-IOError bucket that
+  // consumers treat as transient.
+  //
+  // BUT only trust that verdict for error types the SDK actually recognized:
+  // S3-compatible backends (e.g. MinIO) return genuine transients as UNKNOWN
+  // with the non-retryable flag set -- "SlowDown" rate limiting arrives exactly
+  // this way (that is why IsConnectError() special-cases it by exception name).
+  // Tagging those permanent would invert a transient into a hard failure, so
+  // UNKNOWN and connect-style errors fall through to the plain-IOError bucket
+  // (fail-open to a bounded upper-layer retry, the safe direction).
+  if (error_type != Aws::S3::S3Errors::UNKNOWN && !IsConnectError(error) && !error.ShouldRetry()) {
+    return MakeExtendError(ExtendStatusCode::AwsErrorNonRetryable, message, message /* extra_info */);
   }
 
+  // Transient escapee (throttle / 5xx / timeout / connection error): the SDK
+  // retry budget is spent, but a distinct upper-layer retry (e.g. querynode
+  // rerouting to another replica) can still succeed. Plain IOError -> consumers
+  // classify it as retriable.
   return arrow::Status::IOError(prefix, "AWS Error ", ss.str(), " during ", operation,
                                 " operation: ", error.GetMessage(), wrong_region_msg.value_or(""));
 }

cpp/include/milvus-storage/packed/splitter/indices_based_splitter.h

@@ -23,7 +23,7 @@ class IndicesBasedSplitter : public SplitterPlugin {
   public:
   explicit IndicesBasedSplitter(const std::vector<std::vector<int>>& column_indices);
 
-  std::vector<ColumnGroup> Split(const std::shared_ptr<arrow::RecordBatch>& record) override;
+  arrow::Result<std::vector<ColumnGroup>> Split(const std::shared_ptr<arrow::RecordBatch>& record) override;
 
   private:
   std::vector<std::vector<int>> column_indices_;

cpp/include/milvus-storage/packed/splitter/size_based_splitter.h

@@ -25,10 +25,11 @@ class SizeBasedSplitter : public SplitterPlugin {
    */
   explicit SizeBasedSplitter(size_t max_group_size);
 
-  std::vector<ColumnGroup> Split(const std::shared_ptr<arrow::RecordBatch>& record) override;
+  arrow::Result<std::vector<ColumnGroup>> Split(const std::shared_ptr<arrow::RecordBatch>& record) override;
 
   private:
-  std::vector<ColumnGroup> SplitRecordBatches(const std::vector<std::shared_ptr<arrow::RecordBatch>>& batches);
+  arrow::Result<std::vector<ColumnGroup>> SplitRecordBatches(
+      const std::vector<std::shared_ptr<arrow::RecordBatch>>& batches);
 
   private:
   size_t max_group_size_;

cpp/include/milvus-storage/packed/splitter/splitter_plugin.h

@@ -17,6 +17,7 @@
 #include <vector>
 #include <memory>
 #include <arrow/record_batch.h>
+#include <arrow/result.h>
 #include <milvus-storage/packed/column_group.h>
 
 namespace milvus_storage {
@@ -25,8 +26,11 @@ class SplitterPlugin {
   public:
   virtual ~SplitterPlugin() = default;
 
-  // Split the input record batch into multiple groups of columns
-  virtual std::vector<ColumnGroup> Split(const std::shared_ptr<arrow::RecordBatch>& record) = 0;
+  // Split the input record batch into multiple groups of columns.
+  // Returns an error status instead of aborting when a column selection or
+  // batch accumulation fails (the previous bare-vector signature forced
+  // ValueOrDie, which crashed the whole process on failure).
+  virtual arrow::Result<std::vector<ColumnGroup>> Split(const std::shared_ptr<arrow::RecordBatch>& record) = 0;
 };
 
 }  // namespace milvus_storage

cpp/src/common/extend_status.cpp

@@ -20,6 +20,7 @@
 
 #include <arrow/status.h>
 #include <arrow/result.h>
+#include <fmt/format.h>
 
 namespace milvus_storage {
 
@@ -39,12 +40,30 @@ std::string ExtendStatusDetail::extra_info() const { return extra_info_; }
 
 std::string ExtendStatusDetail::CodeAsString() const {
   switch (code()) {
+    case ExtendStatusCode::PackedInvalidArgs:
+      return "PackedInvalidArgs";
+    case ExtendStatusCode::PackedStorageIO:
+      return "PackedStorageIO";
+    case ExtendStatusCode::PackedMetadataCorrupted:
+      return "PackedMetadataCorrupted";
+    case ExtendStatusCode::PackedFileCorrupted:
+      return "PackedFileCorrupted";
+    case ExtendStatusCode::PackedArrowError:
+      return "PackedArrowError";
+    case ExtendStatusCode::PackedUnexpected:
+      return "PackedUnexpected";
     case ExtendStatusCode::AwsErrorNoSuchUpload:
       return "AwsErrorNoSuchUpload";
     case ExtendStatusCode::AwsErrorConflict:
       return "AwsErrorConflict";
     case ExtendStatusCode::AwsErrorPreConditionFailed:
       return "AwsErrorPreConditionFailed";
+    case ExtendStatusCode::AwsErrorNotFound:
+      return "AwsErrorNotFound";
+    case ExtendStatusCode::AwsErrorAccessDenied:
+      return "AwsErrorAccessDenied";
+    case ExtendStatusCode::AwsErrorNonRetryable:
+      return "AwsErrorNonRetryable";
     case ExtendStatusCode::TxnExhaustedRetry:
       return "TxnExhaustedRetry";
     case ExtendStatusCode::TxnResolutionFailed:
@@ -64,8 +83,131 @@ std::shared_ptr<ExtendStatusDetail> ExtendStatusDetail::UnwrapStatus(const arrow
 }
 
 arrow::Status MakeExtendError(ExtendStatusCode code, std::string message, std::string extra_info) {
-  arrow::StatusCode arrow_code = arrow::StatusCode::IOError;
+  auto arrow_code =
+      code == ExtendStatusCode::PackedInvalidArgs ? arrow::StatusCode::Invalid : arrow::StatusCode::IOError;
   return {arrow_code, std::move(message), std::make_shared<ExtendStatusDetail>(code, std::move(extra_info))};
 }
 
+arrow::Status WrapExtendError(ExtendStatusCode code, std::string message, const arrow::Status& cause) {
+  auto detail = ExtendStatusDetail::UnwrapStatus(cause);
+  auto wrapped_code = detail ? detail->code() : code;
+  auto cause_message = cause.ToString();
+  return MakeExtendError(wrapped_code, fmt::format("{}: {}", message, cause_message), cause_message);
+}
+
+// Map a producer-side ExtendStatusCode to the shared milvus ErrorCode that the
+// segcore boundary (and ultimately the Go retry policy) consumes. This is the
+// single place milvus-storage classifies its own codes ("producer owns
+// classification").
+//
+// It is deliberately a switch with NO `default:` plus a post-switch fallback:
+//   * a `default:` inside the switch would suppress -Wswitch, so a newly added
+//     ExtendStatusCode could silently fall through to the wrong bucket;
+//   * the post-switch `return` satisfies -Wreturn-type and guards out-of-range
+//     values, without suppressing the exhaustiveness warning.
+// The surrounding pragma turns -Wswitch into an error so adding an
+// ExtendStatusCode without classifying it here breaks the build (the
+// extend_status_test.cpp coverage is the runtime backstop).
+//
+// Retriability model (do not repeat the "v2 retries, v3 doesn't" myth):
+// object-storage IO retry does NOT live in the packed / format / api::Reader
+// layers. It lives once in the shared S3 ArrowFileSystem (AWS SDK
+// DefaultRetryStrategy), which every read path -- v1 binlog, v2
+// FileRowGroupReader, v3 api::Reader -- runs on top of. So an IO error that
+// propagates up here already spent the S3 SDK retry budget, equally for v2 and
+// v3; there is no per-generation retry asymmetry.
+//
+// Retriability is therefore decided by whether a DISTINCT upper-layer retry can
+// still help: querynode can reroute a failed read to another replica/node (a
+// different network path / endpoint), or the failure was a node-local transient.
+// Plain IO does not assume that path and is classified conservatively as
+// non-retriable StorageError/2044.
+//
+// Two callers reach segcore ErrorCode differently:
+//   1. A status carrying an ExtendStatusDetail (Packed*/Aws*/Txn) is classified
+//      by this switch. NOTE: as of this writing NO live milvus consumer routes a
+//      Packed* status through here -- packed_reader_c/packed_writer_c hardcode
+//      FileReadFailed/FileWriteFailed and drop the ExtendStatusCode -- so this
+//      switch is a reserved, forward-looking classification, not a hot path.
+//   2. A status with NO detail (plain arrow) is the LIVE segcore/storage read
+//      path; its plain IO is classified as non-retriable StorageError/2044 via
+//      the no-detail fallback of ToSegcoreError below, NOT this switch.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic error "-Wswitch"
+milvus::ErrorCode ToSegcoreErrorCode(ExtendStatusCode code) {
+  switch (code) {
+    case ExtendStatusCode::PackedInvalidArgs:
+      return milvus::InvalidParameter;  // 2042, caller's fault (non-retriable input)
+    case ExtendStatusCode::PackedStorageIO:
+      // Conservatively non-retriable, but this is a DORMANT branch: no live
+      // consumer routes a Packed* status here (the packed C-APIs hardcode
+      // FileReadFailed/FileWriteFailed and drop the code). Do NOT justify this
+      // with "v2 retries internally" -- the S3 SDK retry is shared by v2 and v3
+      // alike. If a real direct-link consumer ever appears, revisit: validate
+      // its retry semantics before changing this non-retriable classification.
+      return milvus::StorageError;  // 2044 (dormant; conservative)
+    case ExtendStatusCode::PackedMetadataCorrupted:
+    case ExtendStatusCode::PackedFileCorrupted:
+      return milvus::DataFormatBroken;  // 2024, permanent data corruption
+    case ExtendStatusCode::PackedArrowError:
+    case ExtendStatusCode::PackedUnexpected:
+      return milvus::StorageError;  // 2044, permanent internal storage error
+    case ExtendStatusCode::AwsErrorNoSuchUpload:
+    case ExtendStatusCode::AwsErrorConflict:
+    case ExtendStatusCode::AwsErrorPreConditionFailed:
+    case ExtendStatusCode::TxnExhaustedRetry:
+    case ExtendStatusCode::TxnResolutionFailed:
+      // S3 multipart / precondition / transaction failures: conservatively
+      // permanent here (the retry budget is already spent or the precondition
+      // genuinely failed). Promote to a more specific code if a real retriable
+      // case is identified.
+      return milvus::StorageError;  // 2044
+    case ExtendStatusCode::AwsErrorNotFound:
+      // The object/bucket is gone: permanent, and fine-grained -- consumers can
+      // distinguish "data missing" (stale loadinfo, GC'd file) from a generic
+      // storage failure. Never transient/2045: a retry/reroute hits the same
+      // shared object store and fails identically.
+      return milvus::ObjectNotExist;  // 2017, permanent
+    case ExtendStatusCode::AwsErrorAccessDenied:
+    case ExtendStatusCode::AwsErrorNonRetryable:
+      // Bad credentials/permissions, or the AWS SDK itself judged the error
+      // non-retryable. Same rule: never transient/2045, or querynode would
+      // retry-storm a request that can never succeed.
+      return milvus::StorageError;  // 2044, permanent
+  }
+  return milvus::StorageError;  // out-of-range value: safe non-retriable fallback
+}
+#pragma GCC diagnostic pop
+
+milvus::SegcoreError ToSegcoreError(const arrow::Status& status) {
+  if (status.ok()) {
+    return milvus::SegcoreError::success();
+  }
+
+  auto detail = ExtendStatusDetail::UnwrapStatus(status);
+  if (detail) {
+    return {ToSegcoreErrorCode(detail->code()), status.ToString()};
+  }
+
+  // No structured ExtendStatusDetail attached: this is the LIVE read path (plain
+  // arrow from FileRowGroupReader / v3 api::Reader / ArrowFileSystem). A
+  // propagated IO error here already spent the shared S3 SDK retry budget, AND
+  // permanently-failing S3 errors (NotFound / AccessDenied / SDK-judged
+  // non-retryable) were already tagged with an ExtendStatusDetail upstream in
+  // ErrorToStatus. A *plain* IOError that reaches this branch is classified
+  // conservatively as non-retriable StorageError/2044. OOM is retriable;
+  // malformed data is permanent corruption; anything else internal.
+  milvus::ErrorCode code;
+  if (status.IsOutOfMemory()) {
+    code = milvus::MemAllocateFailed;  // 2034, retriable
+  } else if (status.IsIOError()) {
+    code = milvus::StorageError;  // 2044, non-retriable
+  } else if (status.IsInvalid() || status.IsTypeError() || status.IsKeyError()) {
+    code = milvus::DataFormatBroken;  // 2024, permanent corruption
+  } else {
+    code = milvus::StorageError;  // 2044, permanent internal error
+  }
+  return {code, status.ToString()};
+}
+
 }  // namespace milvus_storage

cpp/src/filesystem/s3/s3_client.cpp

@@ -179,7 +179,9 @@ arrow::Result<std::string> S3Client::GetBucketRegionFromError(const std::string&
   if (!region.empty()) {
     return region;
   } else if (error.GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
-    return arrow::Status::IOError("Bucket '", bucket, "' not found");
+    // Permanent: the bucket does not exist; a retry/reroute fails identically.
+    return MakeExtendError(ExtendStatusCode::AwsErrorNotFound, "Bucket '" + bucket + "' not found",
+                           "" /* extra_info */);
   } else {
     return arrow::Status::IOError("When resolving region for bucket: ", bucket);
   }
@@ -199,7 +201,9 @@ arrow::Result<std::string> S3Client::GetBucketRegion(const std::string& bucket,
   if (!region.empty()) {
     return region;
   } else if (outcome.GetResult().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) {
-    return arrow::Status::IOError("Bucket '", request.GetBucket(), "' not found");
+    // Permanent: the bucket does not exist; a retry/reroute fails identically.
+    return MakeExtendError(ExtendStatusCode::AwsErrorNotFound,
+                           "Bucket '" + std::string(request.GetBucket().c_str()) + "' not found", "" /* extra_info */);
   } else {
     return arrow::Status::IOError("When resolving region for bucket '", request.GetBucket(),
                                   "': missing 'x-amz-bucket-region' header in response");