fix: skip dropped-field column groups when loading StorageV3 segments (#51275) (#51277)

issue: #51275

### Problem

Under StorageV3 growing-source flush, a manifest legally keeps column
groups whose fields were later dropped: drop does not rewrite flushed
data, and compaction removes the leftover lazily. Loaders however
projected such groups against the **current** schema and treated the
mismatch as an error:

- **Growing recovery** (`SegmentGrowingImpl::LoadColumnsGroups`)
submitted every manifest group with a full-schema projection; a group
whose fields were all dropped has an empty projection, which
milvus-storage rejects by contract (`ChunkReaderImpl::open`: "No needed
columns found in column group"). The channel-recovery level retried the
deterministic failure every second — observed locally: **4140 identical
failures over ~3 minutes**, the channel unavailable (`no available shard
leaders`) the whole time, until a compaction rewrite happened to remove
the group.
- **Sealed load** (`ChunkedSegmentSealedImpl::LoadColumnGroups`)
resolved every manifest column to a field id without a membership check
and tripped the schema-lookup assert on any group still carrying a
dropped column — mixed groups included.

### Fix

Treat a dropped-field column as a legal leftover at the loaders:

- Growing: skip a column group before submitting its load task when none
of its columns exist in the segment schema (logged).
- Sealed: filter resolved field ids absent from the schema snapshot
while building `cg_field_ids` (logged); a group filtered to empty
produces no load task. Mixed groups keep loading their live columns via
projection, unchanged.

### Verification

- Local A/B with a schema-evolution chaos workload (concurrent add/drop
ordinary + BM25 function field, DML/query/search, `useLoonFFI` +
growing-source flush enabled, StreamingNode SIGKILL/restart injections):
before the fix, recovery wedged in the retry loop (4140 errors / ~3 min
/ final query 503); after the fix the same scenario logs 2 group skips,
the channel recovers in seconds, and the full serial consistency
validation passes.
- New UT `LoadGrowingSegmentSkipsDroppedFieldColumnGroup`: flushes a
manifest with a field-exclusive column group (writer pattern
`"0|1|100,101"`), reloads the segment under a schema without that field,
and asserts the load succeeds with the group skipped (plus a guard
assertion that the manifest really contains the exclusive group).
- The sealed-path filter is exercised by code inspection and mirrors the
growing-side semantics; on current master the external-table load path
derives its projection from schema-owned column names and is not exposed
to this class.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

internal/core/src/segcore/ChunkedSegmentSealedImpl.cpp

@@ -1885,15 +1885,31 @@ ChunkedSegmentSealedImpl::LoadColumnGroups(
         reader_create_ms,
         column_groups->size());
 
+    // A manifest column whose field was dropped from the schema is a legal
+    // leftover of drop semantics: filter it out (a group filtered to empty
+    // produces no load task) instead of tripping the schema lookup on it.
     std::vector<std::pair<int, std::vector<FieldId>>> cg_field_ids;
     cg_field_ids.reserve(column_groups->size());
     for (size_t i = 0; i < column_groups->size(); ++i) {
         auto cg = column_groups->at(i);
         std::vector<FieldId> field_ids;
+        std::vector<int64_t> dropped_fields;
         field_ids.reserve(cg->columns.size());
         for (auto& column : cg->columns) {
-            field_ids.emplace_back(
-                schema_snapshot->ResolveColumnFieldId(column));
+            auto field_id = schema_snapshot->ResolveColumnFieldId(column);
+            if (!schema_snapshot->has_field(field_id)) {
+                dropped_fields.push_back(field_id.get());
+                continue;
+            }
+            field_ids.emplace_back(field_id);
+        }
+        if (!dropped_fields.empty()) {
+            LOG_INFO(
+                "segment {} skips dropped fields {} of column group {} on "
+                "load",
+                id_,
+                fmt::format("{}", dropped_fields),
+                i);
         }
         cg_field_ids.emplace_back(static_cast<int>(i), std::move(field_ids));
     }

internal/core/src/segcore/SegmentGrowingImpl.cpp

@@ -2752,11 +2752,37 @@ SegmentGrowingImpl::LoadColumnsGroups(std::string manifest_path) {
     reader_ = milvus_storage::api::Reader::create(
         column_groups, arrow_schema, nullptr, *properties);
 
+    // A column group whose fields were all dropped from the segment schema
+    // has an empty read projection; opening it violates the reader contract
+    // (ChunkReaderImpl::open rejects a group containing none of the needed
+    // columns). Such a group is a legal leftover of drop semantics — skip it
+    // and let compaction remove it from the manifest.
+    auto schema_snapshot = schema_;
+    auto group_has_live_field = [&](int64_t group_index) {
+        for (const auto& column : column_groups->at(group_index)->columns) {
+            // An untranslatable column name is a broken manifest and throws
+            // (same contract as the sealed loader).
+            auto field_id = schema_snapshot->ResolveColumnFieldId(column);
+            if (schema_snapshot->has_field(field_id)) {
+                return true;
+            }
+        }
+        return false;
+    };
+
     auto& pool = ThreadPools::GetThreadPool(milvus::ThreadPoolPriority::MIDDLE);
     std::vector<
         std::future<std::unordered_map<FieldId, std::vector<FieldDataPtr>>>>
         load_group_futures;
     for (int64_t i = 0; i < column_groups->size(); ++i) {
+        if (!group_has_live_field(i)) {
+            LOG_INFO(
+                "skip loading column group {} of segment {}: all of its "
+                "fields are dropped from the schema",
+                i,
+                id_);
+            continue;
+        }
         auto future =
             pool.Submit([this, column_groups, properties, i, num_rows] {
                 return LoadColumnGroup(column_groups, properties, i, num_rows);

internal/core/unittest/test_storage.cpp

@@ -42,6 +42,7 @@
 #include "storage/Types.h"
 #include "storage/Util.h"
 #include "storage/loon_ffi/property_singleton.h"
+#include "storage/loon_ffi/util.h"
 #include "storage/minio/MinioChunkManager.h"
 #include "storage/storage_c.h"
 #include "test_utils/Constants.h"
@@ -488,6 +489,130 @@ TEST_F(StorageTest, FlushGrowingSegmentSkipsEmptyFunctionOutputColumn) {
     cleanup();
 }
 
+// A manifest column group whose only field was dropped from the segment
+// schema before recovery is a legal leftover of drop semantics: reloading
+// the growing segment must skip that group instead of failing the whole
+// load with milvus-storage "No needed columns found in column group".
+TEST_F(StorageTest, LoadGrowingSegmentSkipsDroppedFieldColumnGroup) {
+    std::string test_dir =
+        "/tmp/load_skip_dropped_group_" +
+        std::to_string(
+            std::chrono::system_clock::now().time_since_epoch().count());
+    std::filesystem::create_directories(test_dir);
+    auto cleanup = [&]() {
+        if (std::filesystem::exists(test_dir)) {
+            std::filesystem::remove_all(test_dir);
+        }
+    };
+
+    // Reduced schema: what remains after field 101 is dropped.
+    milvus::proto::schema::CollectionSchema reduced_proto;
+    reduced_proto.set_name("load_skip_dropped_group");
+    auto* row_id_field = reduced_proto.add_fields();
+    row_id_field->set_fieldid(0);
+    row_id_field->set_name("RowID");
+    row_id_field->set_data_type(milvus::proto::schema::DataType::Int64);
+    auto* ts_field = reduced_proto.add_fields();
+    ts_field->set_fieldid(1);
+    ts_field->set_name("Timestamp");
+    ts_field->set_data_type(milvus::proto::schema::DataType::Int64);
+    auto* pk_field = reduced_proto.add_fields();
+    pk_field->set_fieldid(100);
+    pk_field->set_name("pk");
+    pk_field->set_data_type(milvus::proto::schema::DataType::Int64);
+    pk_field->set_is_primary_key(true);
+
+    // Full schema at flush time still carries field 101.
+    auto full_proto = reduced_proto;
+    auto* extra_field = full_proto.add_fields();
+    extra_field->set_fieldid(101);
+    extra_field->set_name("extra");
+    extra_field->set_data_type(milvus::proto::schema::DataType::Int64);
+
+    auto full_schema = Schema::ParseFrom(full_proto);
+    auto segment = CreateGrowingSegment(full_schema, empty_index_meta);
+    ASSERT_NE(segment, nullptr);
+
+    constexpr int N = 3;
+    std::vector<int64_t> row_ids = {0, 1, 2};
+    std::vector<Timestamp> timestamps = {10, 11, 12};
+    std::vector<int64_t> pks = {100, 101, 102};
+    std::vector<int64_t> extras = {200, 201, 202};
+
+    auto insert_data = std::make_unique<InsertRecordProto>();
+    insert_data->set_num_rows(N);
+    insert_data->mutable_fields_data()->AddAllocated(
+        CreateDataArrayFrom(
+            pks.data(), nullptr, N, (*full_schema)[FieldId(100)])
+            .release());
+    insert_data->mutable_fields_data()->AddAllocated(
+        CreateDataArrayFrom(
+            extras.data(), nullptr, N, (*full_schema)[FieldId(101)])
+            .release());
+
+    segment->PreInsert(N);
+    segment->Insert(0, N, row_ids.data(), timestamps.data(), insert_data.get());
+
+    std::string schema_blob = full_proto.SerializeAsString();
+
+    CFlushConfig config{};
+    std::string segment_path = test_dir + "/collection/partition/segment_lg";
+    config.segment_path = segment_path.c_str();
+    config.read_version = -1;
+    config.retry_limit = 3;
+    config.schema_blob = schema_blob.data();
+    config.schema_length = static_cast<int64_t>(schema_blob.size());
+    // Force field 101 into its own column group so that dropping the field
+    // leaves a group with no live field.
+    config.schema_based_pattern = "0|1|100,101";
+
+    CFlushResult result{};
+    auto status =
+        FlushGrowingSegmentData(segment.get(), 0, N, &config, &result);
+
+    ASSERT_EQ(status.error_code, Success) << status.error_msg;
+    ASSERT_EQ(result.num_rows, N);
+    std::string manifest_json =
+        "{\"base_path\":\"" + segment_path +
+        "\",\"ver\":" + std::to_string(result.committed_version) + "}";
+    FreeFlushResult(&result);
+
+    // Guard against a vacuous pass: the split pattern must have produced a
+    // column group holding field 101 exclusively, or the load below would
+    // succeed via projection without exercising the skip branch.
+    auto properties = LoonFFIPropertiesSingleton::GetInstance().GetProperties();
+    ASSERT_NE(properties, nullptr);
+    auto manifest = GetLoonManifest(manifest_json, properties);
+    ASSERT_EQ(manifest->columnGroups().size(), 2u);
+    bool has_exclusive_101_group = false;
+    for (const auto& cg : manifest->columnGroups()) {
+        if (cg->columns.size() == 1 && cg->columns[0] == "101") {
+            has_exclusive_101_group = true;
+        }
+    }
+    ASSERT_TRUE(has_exclusive_101_group)
+        << "expected a column group holding field 101 exclusively";
+
+    // Reload the manifest under the reduced schema (field 101 dropped).
+    auto reduced_schema = Schema::ParseFrom(reduced_proto);
+    auto reloaded = CreateGrowingSegment(reduced_schema, empty_index_meta);
+    ASSERT_NE(reloaded, nullptr);
+
+    milvus::proto::segcore::SegmentLoadInfo load_info;
+    load_info.set_segmentid(1);
+    load_info.set_num_of_rows(N);
+    load_info.set_manifest_path(manifest_json);
+    reloaded->SetLoadInfo(load_info);
+
+    milvus::tracer::TraceContext trace_ctx;
+    ASSERT_NO_THROW(reloaded->Load(trace_ctx, nullptr));
+    EXPECT_EQ(reloaded->get_row_count(), N);
+    EXPECT_TRUE(reloaded->HasFieldData(FieldId(100)));
+    EXPECT_FALSE(reloaded->HasFieldData(FieldId(101)));
+
+    cleanup();
+}
+
 TEST_F(StorageTest, GetLocalUsedSize) {
     int64_t size = 0;
     auto lcm = LocalChunkManagerSingleton::GetInstance().GetChunkManager();