enhance: classify iceberg planning errors

.github/workflows/cpp-ci.yml

@@ -29,8 +29,12 @@ jobs:
 
       # Conan 2.x CMakeDeps doesn't propagate system_libs through shared library targets,
       # so libaio (required by folly) must be explicitly installed and linked.
+      # protobuf-compiler is needed by the Rust bridge test step below: `cargo test`
+      # uses the dev profile, which rebuilds the dependency tree and runs
+      # lance-encoding's build script (prost-build wants protoc). The Build step
+      # above gets protoc from conan, but that environment is not inherited here.
       - name: Install system libraries
-        run: sudo apt-get update && sudo apt-get install -y libaio-dev
+        run: sudo apt-get update && sudo apt-get install -y libaio-dev protobuf-compiler
 
       - name: Setup Rust
         id: rust-toolchain
@@ -67,6 +71,14 @@ jobs:
         run: |
           make build USE_ASAN=True BUILD_TYPE=Release
 
+      - name: Test Rust bridge error classifier
+        working-directory: ./cpp
+        env:
+          CARGO_TARGET_DIR: ${{ github.workspace }}/cpp/build/Release/cargo/build
+          RUSTFLAGS: -C force-frame-pointers=yes
+        run: |
+          cargo test --locked --manifest-path src/format/bridge/rust/Cargo.toml --lib bridge_error::tests
+
       - name: Save rust build cache
         if: github.ref == 'refs/heads/main' && steps.rust-cache.outputs.cache-hit != 'true'
         uses: actions/cache/save@v4

cpp/benchmark/benchmark_format_read.cpp

@@ -335,16 +335,12 @@ class FormatReadBenchmark : public FormatBenchFixtureBase<> {
   arrow::Result<PreparedReaderFile> PrepareIcebergReaderFile() const {
     ARROW_ASSIGN_OR_RAISE(auto table_uri, MakeIcebergTableUri(GetUniquePath("iceberg_read_test")));
 
-    iceberg::IcebergTestTableInfo table_info;
-    std::vector<iceberg::IcebergFileInfo> file_infos;
-    try {
-      auto storage_options = iceberg::ToStorageOptions(fs_config_);
-      table_info =
-          iceberg::CreateTestTable(table_uri, static_cast<uint64_t>(GetLoaderNumRows()), false, {}, storage_options);
-      file_infos = iceberg::PlanFiles(table_info.metadata_location, table_info.snapshot_id, storage_options);
-    } catch (const std::exception& e) {
-      return arrow::Status::IOError("Failed to create Iceberg benchmark table: ", e.what());
-    }
+    auto storage_options = iceberg::ToStorageOptions(fs_config_);
+    ARROW_ASSIGN_OR_RAISE(
+        auto table_info,
+        iceberg::CreateTestTable(table_uri, static_cast<uint64_t>(GetLoaderNumRows()), false, {}, storage_options));
+    ARROW_ASSIGN_OR_RAISE(auto file_infos,
+                          iceberg::PlanFiles(table_info.metadata_location, table_info.snapshot_id, storage_options));
 
     if (file_infos.empty()) {
       return arrow::Status::Invalid("Iceberg PlanFiles returned no files");

cpp/benchmark/benchmark_storage_layer.cpp

@@ -310,11 +310,7 @@ class StorageLayerFixture : public FormatBenchFixtureBase<> {
     ArrowArrayStream stream;
     ARROW_RETURN_NOT_OK(arrow::ExportRecordBatchReader(batch_reader, &stream));
 
-    try {
-      auto dataset = lance::BlockingDataset::WriteDataset(lance_uri, &stream, storage_options);
-    } catch (const lance::LanceException& e) {
-      return arrow::Status::IOError("Lance write failed: ", e.what());
-    }
+    ARROW_RETURN_NOT_OK(lance::BlockingDataset::WriteDataset(lance_uri, &stream, storage_options).status());
     return arrow::Status::OK();
   }
 
@@ -539,13 +535,13 @@ BENCHMARK_DEFINE_F(StorageLayerFixture, LanceNative_OpenRead)(::benchmark::State
 
   // Lambda to read lance dataset
   auto read_lance = [&](bool collect_stats, int64_t& out_rows, int64_t& out_bytes) -> arrow::Status {
-    auto dataset = lance::BlockingDataset::Open(lance_uri, storage_options);
+    ARROW_ASSIGN_OR_RAISE(auto dataset, lance::BlockingDataset::Open(lance_uri, storage_options));
 
     ArrowSchema c_schema;
     ARROW_RETURN_NOT_OK(arrow::ExportSchema(*schema_, &c_schema));
 
-    auto scanner = dataset->Scan(c_schema, 8192);
-    auto stream = scanner->OpenStream();
+    ARROW_ASSIGN_OR_RAISE(auto scanner, dataset->Scan(c_schema, 8192));
+    ARROW_ASSIGN_OR_RAISE(auto stream, scanner->OpenStream());
 
     ARROW_ASSIGN_OR_RAISE(auto reader, arrow::ImportRecordBatchReader(&stream));
 
@@ -612,12 +608,12 @@ BENCHMARK_DEFINE_F(StorageLayerFixture, LanceNative_Take)(::benchmark::State& st
 
   // Lambda to take from lance dataset
   auto take_lance = [&](bool collect_stats, int64_t& out_rows, int64_t& out_bytes) -> arrow::Status {
-    auto dataset = lance::BlockingDataset::Open(lance_uri, storage_options);
+    ARROW_ASSIGN_OR_RAISE(auto dataset, lance::BlockingDataset::Open(lance_uri, storage_options));
 
     ArrowSchema c_schema;
     ARROW_RETURN_NOT_OK(arrow::ExportSchema(*schema_, &c_schema));
 
-    auto stream = dataset->Take(indices, c_schema);
+    ARROW_ASSIGN_OR_RAISE(auto stream, dataset->Take(indices, c_schema));
 
     ARROW_ASSIGN_OR_RAISE(auto reader, arrow::ImportRecordBatchReader(&stream));
 
@@ -718,13 +714,13 @@ BENCHMARK_DEFINE_F(StorageLayerFixture, LanceNative_MultiReader)(::benchmark::St
     for (int i = 0; i < num_readers; ++i) {
       reader_threads.emplace_back([&, i]() {
         auto read_all = [&]() -> arrow::Status {
-          auto dataset = lance::BlockingDataset::Open(lance_uri, storage_options);
+          ARROW_ASSIGN_OR_RAISE(auto dataset, lance::BlockingDataset::Open(lance_uri, storage_options));
 
           ArrowSchema c_schema;
           ARROW_RETURN_NOT_OK(arrow::ExportSchema(*schema_, &c_schema));
 
-          auto scanner = dataset->Scan(c_schema, 8192);
-          auto stream = scanner->OpenStream();
+          ARROW_ASSIGN_OR_RAISE(auto scanner, dataset->Scan(c_schema, 8192));
+          ARROW_ASSIGN_OR_RAISE(auto stream, scanner->OpenStream());
 
           ARROW_ASSIGN_OR_RAISE(auto reader, arrow::ImportRecordBatchReader(&stream));
 

cpp/ffi_exports.map

@@ -31,6 +31,8 @@
     loon_errcode_transient_service;
     loon_errcode_txn_exhausted_retry;
     loon_errcode_txn_resolution_failed;
+    loon_errcode_lance_write_contention;
+    loon_errcode_lance_resource_not_found;
 
     # Properties interface
     loon_properties_create;

cpp/ffi_exports_mac.map

@@ -29,6 +29,8 @@ _loon_errcode_transient_throttling
 _loon_errcode_transient_service
 _loon_errcode_txn_exhausted_retry
 _loon_errcode_txn_resolution_failed
+_loon_errcode_lance_write_contention
+_loon_errcode_lance_resource_not_found
 
 # Properties interface
 _loon_properties_create

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

@@ -58,6 +58,11 @@ enum class ExtendStatusCode : char {
   // Transaction-specific error codes
   TxnExhaustedRetry = LOON_TXN_EXHAUSTED_RETRY,
   TxnResolutionFailed = LOON_TXN_RESOLUTION_FAILED,
+
+  // Lance-specific error codes. These preserve Lance semantics without
+  // overloading object-storage throttling or the dataset-absence ENOENT signal.
+  LanceWriteContention = LOON_LANCE_WRITE_CONTENTION,
+  LanceResourceNotFound = LOON_LANCE_RESOURCE_NOT_FOUND,
 };
 
 class ExtendStatusDetail : public arrow::StatusDetail {

cpp/include/milvus-storage/ffi_c.h

@@ -58,6 +58,8 @@ FFI_EXPORT extern const int loon_errcode_transient_throttling;
 FFI_EXPORT extern const int loon_errcode_transient_service;
 FFI_EXPORT extern const int loon_errcode_txn_exhausted_retry;
 FFI_EXPORT extern const int loon_errcode_txn_resolution_failed;
+FFI_EXPORT extern const int loon_errcode_lance_write_contention;
+FFI_EXPORT extern const int loon_errcode_lance_resource_not_found;
 
 // usage example(caller must free the message string):
 //

cpp/include/milvus-storage/ffi_internal/ffi_error_code.h

@@ -40,3 +40,5 @@
 #define LOON_TRANSIENT_SERVICE 110
 #define LOON_TXN_EXHAUSTED_RETRY 111
 #define LOON_TXN_RESOLUTION_FAILED 112
+#define LOON_LANCE_WRITE_CONTENTION 113
+#define LOON_LANCE_RESOURCE_NOT_FOUND 114

cpp/scripts/error_handling_baseline.tsv

@@ -1,8 +1,6 @@
 throw	cpp/src/common/metadata.cpp	4
 throw	cpp/src/ffi/ffi_fiu_c.cpp	1
 throw	cpp/src/ffi/v2_column_groups_builder.cpp	5
-throw	cpp/src/format/bridge/rust/src/iceberg_bridge.cpp	2
-throw	cpp/src/format/bridge/rust/src/lance_bridge.cpp	21
 throw	cpp/src/format/iceberg/iceberg_common.cpp	2
 throw	cpp/src/format/lance/lance_common.cpp	2
 throw	cpp/src/format/vortex/vortex_translater.cpp	6

cpp/src/common/extend_status.cpp

@@ -54,6 +54,8 @@ constexpr ExtendStatusCodeMetadata kExtendStatusCodeMetadata[] = {
     {ExtendStatusCode::StorageTransientService, "StorageTransientService", true},
     {ExtendStatusCode::TxnExhaustedRetry, "TxnExhaustedRetry", false},
     {ExtendStatusCode::TxnResolutionFailed, "TxnResolutionFailed", false},
+    {ExtendStatusCode::LanceWriteContention, "LanceWriteContention", true},
+    {ExtendStatusCode::LanceResourceNotFound, "LanceResourceNotFound", false},
 };
 
 const ExtendStatusCodeMetadata* FindExtendStatusCodeMetadata(ExtendStatusCode code) {
@@ -203,6 +205,7 @@ milvus::ErrorCode ToSegcoreErrorCode(ExtendStatusCode code) {
     case ExtendStatusCode::StorageTransientTimeout:
     case ExtendStatusCode::StorageTransientThrottling:
     case ExtendStatusCode::StorageTransientService:
+    case ExtendStatusCode::LanceWriteContention:
       return milvus::StorageTransientError;  // 2045
     case ExtendStatusCode::AwsErrorConflict:
     case ExtendStatusCode::AwsErrorPreConditionFailed:
@@ -213,6 +216,7 @@ milvus::ErrorCode ToSegcoreErrorCode(ExtendStatusCode code) {
       // budget is already spent).
       return milvus::StorageError;  // 2044
     case ExtendStatusCode::AwsErrorNotFound:
+    case ExtendStatusCode::LanceResourceNotFound:
       // 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

cpp/src/ffi/result_c.cpp

@@ -44,6 +44,8 @@ extern FFI_EXPORT const int loon_errcode_transient_throttling = LOON_TRANSIENT_T
 extern FFI_EXPORT const int loon_errcode_transient_service = LOON_TRANSIENT_SERVICE;
 extern FFI_EXPORT const int loon_errcode_txn_exhausted_retry = LOON_TXN_EXHAUSTED_RETRY;
 extern FFI_EXPORT const int loon_errcode_txn_resolution_failed = LOON_TXN_RESOLUTION_FAILED;
+extern FFI_EXPORT const int loon_errcode_lance_write_contention = LOON_LANCE_WRITE_CONTENTION;
+extern FFI_EXPORT const int loon_errcode_lance_resource_not_found = LOON_LANCE_RESOURCE_NOT_FOUND;
 
 }  // extern "C"
 
@@ -72,6 +74,8 @@ std::string error_to_string(int code) {
       {LOON_TRANSIENT_SERVICE, "StorageTransientService"},
       {LOON_TXN_EXHAUSTED_RETRY, "TxnExhaustedRetry"},
       {LOON_TXN_RESOLUTION_FAILED, "TxnResolutionFailed"},
+      {LOON_LANCE_WRITE_CONTENTION, "LanceWriteContention"},
+      {LOON_LANCE_RESOURCE_NOT_FOUND, "LanceResourceNotFound"},
   };
 
   if (auto it = error_strings.find(code); it != error_strings.end()) {

cpp/src/format/bridge/rust/.gitignore

@@ -10,3 +10,6 @@ build/*
 
 #ignore others
 .DS_Store
+
+# vendored vortex sources generated by patch_vortex.sh
+_vortex_patched/

cpp/src/format/bridge/rust/include/bridge_error.h

@@ -0,0 +1,66 @@
+// Copyright 2025 Zilliz
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <memory>
+#include <string_view>
+
+#include <arrow/record_batch.h>
+#include <arrow/status.h>
+
+namespace milvus_storage::bridge {
+
+// Shared decoding of classified errors coming out of the Rust cxx bridges
+// (vortex / lance / iceberg).
+//
+// The cxx boundary only carries an error as a display string; the Rust side
+// (rust/src/bridge_error.rs, vortex's filesystem_c.rs) embeds the
+// classification as "__LOON_RUST_BRIDGE_ERRCODE__=<code>; message". The
+// helpers here parse and strip that marker and rebuild a structured
+// arrow::Status:
+//   * code 12 (LOON_FILE_NOT_FOUND)      -> IOError + ENOENT detail
+//   * ExtendStatusCode values (101-114)  -> IOError + ExtendStatusDetail
+//   * bridge-private codes (>= 1000, never cross the C ABI):
+//       1000 unclassified   -> plain IOError          (conservative fallback)
+//       1001 data-corrupt   -> Status::Invalid       (permanent data error)
+//       1002 not-supported  -> Status::NotImplemented
+//   * no / unknown marker                -> plain IOError (conservative
+//     non-retriable fallback; never invent retriability)
+
+// Bridge-private marker codes; keep in sync with rust/src/bridge_error.rs.
+inline constexpr int kBridgeErrCodeUnclassified = 1000;
+inline constexpr int kBridgeErrCodeDataCorrupt = 1001;
+inline constexpr int kBridgeErrCodeNotSupported = 1002;
+
+/// Decode a raw bridge error message (marker stripped) into a structured
+/// arrow::Status per the table above.
+arrow::Status MakeBridgeErrorStatus(std::string_view message);
+
+/// Prefix `context` onto `status`'s message, preserving its StatusCode and
+/// detail (ExtendStatusDetail / errno). OK statuses pass through.
+arrow::Status WithBridgeContext(std::string_view context, const arrow::Status& status);
+
+/// Translate any status that may carry an (encoded or already-structured)
+/// bridge error: already-classified statuses just gain context; otherwise the
+/// message is scanned for the marker and rebuilt.
+arrow::Status TranslateBridgeStatus(std::string_view context, const arrow::Status& status);
+
+/// Wrap a RecordBatchReader whose ReadNext/Close surface raw bridge error
+/// strings (arrow FFI stringification of Rust stream errors) so mid-scan
+/// errors are decoded too. `context` is prefixed onto translated errors.
+std::shared_ptr<arrow::RecordBatchReader> WrapBridgeRecordBatchReader(std::shared_ptr<arrow::RecordBatchReader> inner,
+                                                                      std::string context);
+
+}  // namespace milvus_storage::bridge

cpp/src/format/bridge/rust/include/iceberg_bridge.h

@@ -18,14 +18,15 @@
 #include <string>
 #include <unordered_map>
 #include <vector>
-#include <stdexcept>
 
-namespace milvus_storage::iceberg {
+#include <arrow/result.h>
 
-class IcebergException : public std::runtime_error {
-  public:
-  explicit IcebergException(const std::string& message) : std::runtime_error(message) {}
-};
+// Error model: fallible APIs return arrow::Result. Bridge errors surface as
+// plain (non-retriable) IOError for now — iceberg-rust error kinds are not
+// yet classified on the Rust side (follow-up); no exception leaves the
+// library.
+
+namespace milvus_storage::iceberg {
 
 /// Per-file info returned from PlanFiles
 struct IcebergFileInfo {
@@ -42,9 +43,10 @@ struct IcebergFileInfo {
 /// @param snapshot_id Which snapshot to scan
 /// @param storage_options S3/cloud config as key-value pairs
 /// @return Vector of file info, one per data file in the snapshot
-std::vector<IcebergFileInfo> PlanFiles(const std::string& metadata_location,
-                                       int64_t snapshot_id,
-                                       const std::unordered_map<std::string, std::string>& storage_options);
+arrow::Result<std::vector<IcebergFileInfo>> PlanFiles(
+    const std::string& metadata_location,
+    int64_t snapshot_id,
+    const std::unordered_map<std::string, std::string>& storage_options);
 
 /// Info returned after creating a test Iceberg table.
 struct IcebergTestTableInfo {
@@ -66,11 +68,12 @@ struct IcebergTestTableInfo {
 /// intending to read via native `gs://` with SA impersonation; the Rust side
 /// will byte-rewrite the embedded scheme across every level of the metadata
 /// tree so iceberg-rust's `plan_files` can traverse it under a `gs://` FileIO.
-IcebergTestTableInfo CreateTestTable(const std::string& table_dir,
-                                     uint64_t num_rows,
-                                     bool with_positional_deletes,
-                                     const std::vector<int64_t>& deleted_positions,
-                                     const std::unordered_map<std::string, std::string>& storage_options = {},
-                                     const std::string& record_scheme_override = "");
+arrow::Result<IcebergTestTableInfo> CreateTestTable(
+    const std::string& table_dir,
+    uint64_t num_rows,
+    bool with_positional_deletes,
+    const std::vector<int64_t>& deleted_positions,
+    const std::unordered_map<std::string, std::string>& storage_options = {},
+    const std::string& record_scheme_override = "");
 
 }  // namespace milvus_storage::iceberg