enhance:[2.6] classify packed extend status codes into segcore error codes (#575)

cpp/CMakeLists.txt

@@ -164,4 +164,4 @@ install(DIRECTORY "${PROJECT_SOURCE_DIR}/include/milvus-storage"
 
 set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib)
 
-set(CMAKE_INCLUDE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/include)
+set(CMAKE_INCLUDE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/include)

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

@@ -25,7 +25,29 @@
 namespace milvus_storage {
 enum class ExtendStatusCode : char {
   // arrow::StatusCode biggest is 45
-  NoSuchUpload = 101,
+  // 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,
+  TxnResolutionFailed = 112,
 };
 
 class ExtendStatusDetail : public arrow::StatusDetail {
@@ -61,6 +83,8 @@ 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);
 
-}  // namespace milvus_storage
+}  // namespace milvus_storage

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

@@ -189,12 +189,58 @@ arrow::Status ErrorToStatus(const std::string& prefix,
     }
   }
 
-  if (error_type == Aws::S3::S3Errors::NO_SUCH_UPLOAD) {
-    std::string message = "AWS Error " + ss.str() + " during " + operation + " operation: " + error.GetMessage() +
-                          wrong_region_msg.value_or("");
-    return MakeExtendError(ExtendStatusCode::NoSuchUpload, message, message /* extra_info */);
+  std::string message = "AWS Error " + ss.str() + " during " + operation + " operation: " + error.GetMessage() +
+                        wrong_region_msg.value_or("");
+
+  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:
+          break;
+      };
+      break;
+    }
+    default:
+      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

@@ -39,8 +39,34 @@ std::string ExtendStatusDetail::extra_info() const { return extra_info_; }
 
 std::string ExtendStatusDetail::CodeAsString() const {
   switch (code()) {
-    case ExtendStatusCode::NoSuchUpload:
-      return "NoSuchUpload";
+    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:
+      return "TxnResolutionFailed";
     default:
       return "Unknown";
   }
@@ -56,9 +82,16 @@ 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;
-  return arrow::Status(arrow_code, std::move(message),
-                       std::make_shared<ExtendStatusDetail>(code, std::move(extra_info)));
+  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, std::move(message) + ": " + cause_message, cause_message);
 }
 
 }  // namespace milvus_storage

cpp/src/filesystem/s3/s3_client.cpp

@@ -177,7 +177,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);
   }
@@ -197,7 +199,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");

cpp/src/format/parquet/file_reader.cpp

@@ -147,11 +147,11 @@ arrow::Status FileRowGroupReader::SetRowGroupOffsetAndCount(int row_group_offset
 }
 
 // Helper function to match schema and fill null columns
-void MatchSchemaAndFillNullColumns(const std::shared_ptr<arrow::Table>& table,
-                                   const std::shared_ptr<arrow::Schema>& schema,
-                                   const FieldIDList& field_id_list,
-                                   const std::map<FieldID, ColumnOffset>& field_id_mapping,
-                                   std::shared_ptr<arrow::Table>* out) {
+arrow::Status MatchSchemaAndFillNullColumns(const std::shared_ptr<arrow::Table>& table,
+                                            const std::shared_ptr<arrow::Schema>& schema,
+                                            const FieldIDList& field_id_list,
+                                            const std::map<FieldID, ColumnOffset>& field_id_mapping,
+                                            std::shared_ptr<arrow::Table>* out) {
   std::vector<std::shared_ptr<arrow::ChunkedArray>> columns;
 
   for (size_t i = 0; i < field_id_list.size(); ++i) {
@@ -160,12 +160,15 @@ void MatchSchemaAndFillNullColumns(const std::shared_ptr<arrow::Table>& table,
       int col = field_id_mapping.at(field_id).col_index;
       columns.emplace_back(table->column(col));
     } else {
-      auto null_array = arrow::MakeArrayOfNull(schema->field(i)->type(), table->num_rows()).ValueOrDie();
+      // MakeArrayOfNull allocates; ValueOrDie here aborted the whole process
+      // on failure instead of reporting a status.
+      ARROW_ASSIGN_OR_RAISE(auto null_array, arrow::MakeArrayOfNull(schema->field(i)->type(), table->num_rows()));
       columns.emplace_back(std::make_shared<arrow::ChunkedArray>(null_array));
     }
   }
 
   *out = arrow::Table::Make(schema, columns);
+  return arrow::Status::OK();
 }
 
 arrow::Status FileRowGroupReader::SliceRowGroupFromTable(std::shared_ptr<arrow::Table>* out) {
@@ -243,8 +246,8 @@ arrow::Status FileRowGroupReader::ReadNextRowGroup(std::shared_ptr<arrow::Table>
 
   // Match schema and fill null columns
   std::shared_ptr<arrow::Table> matched_table;
-  MatchSchemaAndFillNullColumns(new_table, schema_, field_id_list_, file_metadata_->GetFieldIDMapping(),
-                                &matched_table);
+  ARROW_RETURN_NOT_OK(MatchSchemaAndFillNullColumns(new_table, schema_, field_id_list_,
+                                                    file_metadata_->GetFieldIDMapping(), &matched_table));
 
   // Merge with existing buffer table if needed
   if (buffer_table_ != nullptr) {

cpp/src/packed/column_group.cpp

@@ -14,6 +14,7 @@
 
 #include "milvus-storage/packed/column_group.h"
 #include "milvus-storage/common/arrow_util.h"
+#include "milvus-storage/common/extend_status.h"
 #include <arrow/table.h>
 #include <arrow/status.h>
 
@@ -32,7 +33,7 @@ ColumnGroup::ColumnGroup(GroupId group_id,
 
 arrow::Status ColumnGroup::AddRecordBatch(const std::shared_ptr<arrow::RecordBatch>& batch) {
   if (!batch) {
-    return arrow::Status::IOError("ColumnGroup::AddRecordBatch: batch is null");
+    return MakeExtendError(ExtendStatusCode::PackedInvalidArgs, "ColumnGroup::AddRecordBatch: batch is null");
   }
   batches_.emplace_back(batch);
 
@@ -46,9 +47,11 @@ arrow::Status ColumnGroup::AddRecordBatch(const std::shared_ptr<arrow::RecordBat
 
 arrow::Status ColumnGroup::Merge(const ColumnGroup& other) {
   for (auto& batch : other.batches_) {
-    if (!AddRecordBatch(batch).ok()) {
-      return arrow::Status::IOError("ColumnGroup::Merge: failed to merge record batch");
-    };
+    auto status = AddRecordBatch(batch);
+    if (!status.ok()) {
+      return WrapExtendError(ExtendStatusCode::PackedUnexpected, "ColumnGroup::Merge: failed to merge record batch",
+                             status);
+    }
   }
   return arrow::Status::OK();
 }