// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// 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

#include <gtest/gtest.h>
#include <stdint.h>
#include <algorithm>
#include <initializer_list>
#include <iostream>
#include <memory>
#include <optional>
#include <set>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include "NamedType/underlying_functionalities.hpp"
#include "common/Common.h"
#include "common/Consts.h"
#include "common/Schema.h"
#include "common/Types.h"
#include "common/protobuf_utils.h"
#include "filemanager/InputStream.h"
#include "gtest/gtest.h"
#include "index/IndexFactory.h"
#include "index/Meta.h"
#include "knowhere/comp/index_param.h"
#include "pb/common.pb.h"
#include "pb/index_cgo_msg.pb.h"
#include "pb/index_coord.pb.h"
#include "pb/segcore.pb.h"
#include "segcore/SegmentLoadInfo.h"
#include "segcore/Types.h"

using namespace milvus;
using namespace milvus::segcore;

class SegmentLoadInfoTest : public ::testing::Test {
 protected:
    void
    SetUp() override {
        // Setup schema with test fields (sequential field IDs starting from 100)
        schema_ = std::make_shared<Schema>();
        // Field IDs: 100=pk, 101=vec, 102=json, 103=varchar, 104-110=additional fields
        schema_->AddDebugField("pk", DataType::INT64);
        schema_->AddDebugField(
            "vec", DataType::VECTOR_FLOAT, 128, knowhere::metric::L2);
        schema_->AddDebugField("json_field", DataType::JSON);
        schema_->AddDebugField("bm25_field", DataType::JSON);
        schema_->AddDebugField("group_field", DataType::INT64);
        schema_->AddDebugField("child_field1", DataType::FLOAT);
        schema_->AddDebugField("child_field2", DataType::FLOAT);
        schema_->AddDebugField("child_field3", DataType::FLOAT);
        schema_->AddDebugField("extra_field1", DataType::INT64);
        schema_->AddDebugField("extra_field2", DataType::FLOAT);
        schema_->AddDebugField("extra_field3", DataType::FLOAT);
        schema_->set_primary_field_id(FieldId(100));

        // Setup a basic SegmentLoadInfo proto
        proto_.set_segmentid(12345);
        proto_.set_partitionid(100);
        proto_.set_collectionid(200);
        proto_.set_dbid(1);
        proto_.set_num_of_rows(10000);
        proto_.set_flush_time(1234567890);
        proto_.set_readableversion(5);
        proto_.set_storageversion(1);
        proto_.set_is_sorted(true);
        proto_.set_insert_channel("test_channel");
        proto_.set_manifest_path("/path/to/manifest");
        proto_.set_priority(proto::common::LoadPriority::LOW);

        // Add compaction from
        proto_.add_compactionfrom(111);
        proto_.add_compactionfrom(222);

        // Add index info (field 101=vec, field 102=json_field)
        auto* index_info = proto_.add_index_infos();
        index_info->set_fieldid(101);
        index_info->set_indexid(1001);
        index_info->set_buildid(2001);
        index_info->set_index_version(1);
        index_info->add_index_file_paths("/path/to/index1");
        index_info->add_index_file_paths("/path/to/index2");
        // Add required index_type parameter for vector field
        auto* index_param1 = index_info->add_index_params();
        index_param1->set_key("index_type");
        index_param1->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

        auto* index_info2 = proto_.add_index_infos();
        index_info2->set_fieldid(102);
        index_info2->set_indexid(1002);
        index_info2->add_index_file_paths("/path/to/index3");
        // Add required index_type parameter for scalar field
        auto* index_param2 = index_info2->add_index_params();
        index_param2->set_key("index_type");
        index_param2->set_value(milvus::index::INVERTED_INDEX_TYPE);

        // Add binlog paths (field 101)
        auto* binlog = proto_.add_binlog_paths();
        binlog->set_fieldid(101);
        auto* log1 = binlog->add_binlogs();
        log1->set_log_path("/path/to/binlog1");
        log1->set_entries_num(500);
        auto* log2 = binlog->add_binlogs();
        log2->set_log_path("/path/to/binlog2");
        log2->set_entries_num(500);

        // Add column group binlog (field 104 with child fields 105, 106)
        auto* group_binlog = proto_.add_binlog_paths();
        group_binlog->set_fieldid(104);
        group_binlog->add_child_fields(105);
        group_binlog->add_child_fields(106);
        auto* group_log = group_binlog->add_binlogs();
        group_log->set_log_path("/path/to/group_binlog");
        group_log->set_entries_num(1000);

        // Add statslogs
        auto* statslog = proto_.add_statslogs();
        statslog->set_fieldid(101);
        auto* stat_log = statslog->add_binlogs();
        stat_log->set_log_path("/path/to/statslog");

        // Add deltalogs
        auto* deltalog = proto_.add_deltalogs();
        deltalog->set_fieldid(0);
        auto* delta_log = deltalog->add_binlogs();
        delta_log->set_log_path("/path/to/deltalog");

        // Add text stats
        auto& text_stats = (*proto_.mutable_textstatslogs())[101];
        text_stats.set_fieldid(101);
        text_stats.set_version(1);
        text_stats.add_files("/path/to/text_stats");

        // Add json key stats
        auto& json_stats = (*proto_.mutable_jsonkeystatslogs())[102];
        json_stats.set_fieldid(102);
        json_stats.set_version(1);

        // Add bm25 logs
        auto* bm25log = proto_.add_bm25logs();
        bm25log->set_fieldid(103);
        auto* bm25_binlog = bm25log->add_binlogs();
        bm25_binlog->set_log_path("/path/to/bm25log");
    }

    SchemaPtr schema_;
    proto::segcore::SegmentLoadInfo proto_;
};

TEST_F(SegmentLoadInfoTest, EmptyProtoConstructor) {
    proto::segcore::SegmentLoadInfo empty_proto;
    SegmentLoadInfo info(empty_proto, schema_);
    EXPECT_TRUE(info.IsEmpty());
    EXPECT_EQ(info.GetSegmentID(), 0);
    EXPECT_EQ(info.GetNumOfRows(), 0);
}

TEST_F(SegmentLoadInfoTest, ConstructFromProto) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_FALSE(info.IsEmpty());
    EXPECT_EQ(info.GetSegmentID(), 12345);
    EXPECT_EQ(info.GetPartitionID(), 100);
    EXPECT_EQ(info.GetCollectionID(), 200);
    EXPECT_EQ(info.GetDbID(), 1);
    EXPECT_EQ(info.GetNumOfRows(), 10000);
    EXPECT_EQ(info.GetFlushTime(), 1234567890);
    EXPECT_EQ(info.GetReadableVersion(), 5);
    EXPECT_EQ(info.GetStorageVersion(), 1);
    EXPECT_TRUE(info.IsSorted());
    EXPECT_EQ(info.GetInsertChannel(), "test_channel");
    EXPECT_EQ(info.GetManifestPath(), "/path/to/manifest");
    EXPECT_TRUE(info.HasManifestPath());
    EXPECT_EQ(info.GetPriority(), proto::common::LoadPriority::LOW);
}

TEST_F(SegmentLoadInfoTest, MoveConstructor) {
    SegmentLoadInfo info1(proto_, schema_);
    SegmentLoadInfo info2(std::move(info1));

    EXPECT_EQ(info2.GetSegmentID(), 12345);
    EXPECT_EQ(info2.GetNumOfRows(), 10000);
}

TEST_F(SegmentLoadInfoTest, CopyAssignment) {
    SegmentLoadInfo info1(proto_, schema_);
    proto::segcore::SegmentLoadInfo empty_proto;
    SegmentLoadInfo info2(empty_proto, schema_);
    info2 = info1;

    EXPECT_EQ(info2.GetSegmentID(), 12345);
    EXPECT_EQ(info2.GetNumOfRows(), 10000);
}

TEST_F(SegmentLoadInfoTest, SetMethod) {
    proto::segcore::SegmentLoadInfo empty_proto;
    SegmentLoadInfo info(empty_proto, schema_);
    info.Set(proto_, schema_);

    EXPECT_EQ(info.GetSegmentID(), 12345);
    EXPECT_EQ(info.GetNumOfRows(), 10000);
}

TEST_F(SegmentLoadInfoTest, CompactionInfo) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_TRUE(info.IsCompacted());
    EXPECT_EQ(info.GetCompactionFromCount(), 2);
    EXPECT_EQ(info.GetCompactionFrom()[0], 111);
    EXPECT_EQ(info.GetCompactionFrom()[1], 222);
}

TEST_F(SegmentLoadInfoTest, IndexInfo) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_EQ(info.GetIndexInfoCount(), 2);
    EXPECT_TRUE(info.HasIndexInfo(FieldId(101)));
    EXPECT_TRUE(info.HasIndexInfo(FieldId(102)));
    EXPECT_FALSE(info.HasIndexInfo(FieldId(999)));

    auto index_infos = info.GetFieldIndexInfos(FieldId(101));
    EXPECT_EQ(index_infos.size(), 1);
    EXPECT_EQ(index_infos[0].index_id, 1001);

    auto indexed_fields = info.GetIndexedFieldIds();
    EXPECT_EQ(indexed_fields.size(), 2);
    EXPECT_TRUE(indexed_fields.count(FieldId(101)) > 0);
    EXPECT_TRUE(indexed_fields.count(FieldId(102)) > 0);

    // Test non-existent field
    auto empty_infos = info.GetFieldIndexInfos(FieldId(999));
    EXPECT_TRUE(empty_infos.empty());
}

TEST_F(SegmentLoadInfoTest, LoadIndexInfoCopyPreservesIndexStorePathVersion) {
    LoadIndexInfo original;
    original.index_store_path_version = proto::index::IndexStorePathVersion::
        INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED;

    LoadIndexInfo copied(original);
    EXPECT_EQ(copied.index_store_path_version,
              proto::index::IndexStorePathVersion::
                  INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED);

    LoadIndexInfo assigned;
    assigned = original;
    EXPECT_EQ(assigned.index_store_path_version,
              proto::index::IndexStorePathVersion::
                  INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED);
}

TEST_F(SegmentLoadInfoTest,
       ConvertFieldIndexInfoToLoadIndexInfoPreservesIndexStorePathVersion) {
    auto* index_info = proto_.add_index_infos();
    index_info->set_fieldid(101);
    index_info->set_indexid(201);
    index_info->set_buildid(301);
    index_info->set_index_version(1);
    index_info->set_current_index_version(1);
    index_info->set_index_size(1024);
    index_info->set_num_rows(1000);
    index_info->add_index_file_paths("index_v1/1/2/3/301/1/index_data");
    auto* index_param = index_info->add_index_params();
    index_param->set_key("index_type");
    index_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);
    index_info->set_index_store_path_version(
        proto::index::IndexStorePathVersion::
            INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED);

    SegmentLoadInfo info(proto_, schema_);
    auto load_index_info = info.ConvertFieldIndexInfoToLoadIndexInfo(
        index_info, proto_.segmentid());

    EXPECT_EQ(load_index_info.index_store_path_version,
              proto::index::IndexStorePathVersion::
                  INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED);
}

TEST_F(SegmentLoadInfoTest, BuildCacheSkipsHybridScalarResourceEstimate) {
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    auto* hybrid_index = proto.add_index_infos();
    hybrid_index->set_fieldid(108);
    hybrid_index->set_indexid(5001);
    hybrid_index->add_index_file_paths("/path/to/hybrid_index");
    auto* hybrid_param = hybrid_index->add_index_params();
    hybrid_param->set_key(milvus::index::INDEX_TYPE);
    hybrid_param->set_value(milvus::index::HYBRID_INDEX_TYPE);

    auto* inverted_index = proto.add_index_infos();
    inverted_index->set_fieldid(109);
    inverted_index->set_indexid(5002);
    inverted_index->add_index_file_paths("/path/to/inverted_index");
    auto* inverted_param = inverted_index->add_index_params();
    inverted_param->set_key(milvus::index::INDEX_TYPE);
    inverted_param->set_value(milvus::index::INVERTED_INDEX_TYPE);

    SegmentLoadInfo segment_info(proto, schema_);

    auto hybrid_infos = segment_info.GetFieldIndexInfos(FieldId(108));
    ASSERT_EQ(hybrid_infos.size(), 1);
    EXPECT_FALSE(hybrid_infos[0].load_resource_request.has_value());

    auto inverted_infos = segment_info.GetFieldIndexInfos(FieldId(109));
    ASSERT_EQ(inverted_infos.size(), 1);
    EXPECT_TRUE(inverted_infos[0].load_resource_request.has_value());
}

TEST_F(SegmentLoadInfoTest, CompactRuntimeInfoForManifest) {
    auto proto = proto_;
    for (int i = 0; i < 16; ++i) {
        proto.add_index_infos()->CopyFrom(proto.index_infos(0));
        proto.add_binlog_paths()->CopyFrom(proto.binlog_paths(0));
        proto.add_statslogs()->CopyFrom(proto.statslogs(0));
        proto.add_deltalogs()->CopyFrom(proto.deltalogs(0));
        proto.add_bm25logs()->CopyFrom(proto.bm25logs(0));
    }
    SegmentLoadInfo info(proto, schema_);
    auto index_capacity = info.GetProto().index_infos().Capacity();
    auto binlog_capacity = info.GetProto().binlog_paths().Capacity();
    auto stats_capacity = info.GetProto().statslogs().Capacity();
    auto delta_capacity = info.GetProto().deltalogs().Capacity();
    auto bm25_capacity = info.GetProto().bm25logs().Capacity();

    info.CompactRuntimeInfoForManifest();

    EXPECT_EQ(info.GetIndexInfoCount(), 0);
    EXPECT_FALSE(info.HasIndexInfo(FieldId(101)));
    EXPECT_FALSE(info.HasIndexInfo(FieldId(102)));
    auto index_infos = info.GetFieldIndexInfos(FieldId(101));
    EXPECT_TRUE(index_infos.empty());
    EXPECT_EQ(info.GetBinlogPathCount(), 0);
    EXPECT_FALSE(info.HasBinlogPath(FieldId(101)));
    EXPECT_EQ(info.GetStatslogCount(), 0);
    EXPECT_EQ(info.GetDeltalogCount(), 0);
    EXPECT_EQ(info.GetBm25logCount(), 0);
    EXPECT_LT(info.GetProto().index_infos().Capacity(), index_capacity);
    EXPECT_LT(info.GetProto().binlog_paths().Capacity(), binlog_capacity);
    EXPECT_LT(info.GetProto().statslogs().Capacity(), stats_capacity);
    EXPECT_LT(info.GetProto().deltalogs().Capacity(), delta_capacity);
    EXPECT_LT(info.GetProto().bm25logs().Capacity(), bm25_capacity);
    EXPECT_LE(info.GetProto().index_infos().Capacity(), 1);
    EXPECT_LE(info.GetProto().binlog_paths().Capacity(), 1);
    EXPECT_LE(info.GetProto().statslogs().Capacity(), 1);
    EXPECT_LE(info.GetProto().deltalogs().Capacity(), 1);
    EXPECT_LE(info.GetProto().bm25logs().Capacity(), 1);
    EXPECT_TRUE(info.HasTextStatsLog(101));
    EXPECT_TRUE(info.HasJsonKeyStatsLog(102));
    EXPECT_EQ(info.GetManifestPath(), "/path/to/manifest");
}

TEST_F(SegmentLoadInfoTest,
       CompactRuntimeInfoForManifestKeepsIndexDiffIdentity) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    current_proto.set_manifest_path("/path/to/current_manifest");
    auto* cur_index = current_proto.add_index_infos();
    cur_index->set_fieldid(101);
    cur_index->set_indexid(1001);
    cur_index->add_index_file_paths("/path/to/old_index");
    auto* cur_param = cur_index->add_index_params();
    cur_param->set_key("index_type");
    cur_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    new_proto.set_manifest_path("/path/to/current_manifest");
    auto* new_index = new_proto.add_index_infos();
    new_index->set_fieldid(101);
    new_index->set_indexid(2001);
    new_index->add_index_file_paths("/path/to/new_index");
    auto* new_param = new_index->add_index_params();
    new_param->set_key("index_type");
    new_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    current_info.SetColumnGroupsForTesting(
        std::make_shared<milvus_storage::api::ColumnGroups>());
    new_info.SetColumnGroupsForTesting(
        std::make_shared<milvus_storage::api::ColumnGroups>());

    current_info.CompactRuntimeInfoForManifest();
    EXPECT_EQ(current_info.GetIndexInfoCount(), 0);
    EXPECT_FALSE(current_info.HasIndexInfo(FieldId(101)));
    EXPECT_TRUE(current_info.GetFieldIndexInfos(FieldId(101)).empty());

    auto diff = current_info.ComputeDiff(new_info);
    EXPECT_EQ(diff.indexes_to_replace.size(), 1);
    ASSERT_TRUE(diff.indexes_to_replace.count(FieldId(101)) > 0);
    ASSERT_EQ(diff.indexes_to_replace[FieldId(101)].size(), 1);
    EXPECT_EQ(diff.indexes_to_replace[FieldId(101)][0].index_id, 2001);
    EXPECT_EQ(diff.indexes_to_drop.size(), 1);
    EXPECT_TRUE(diff.indexes_to_drop.count(FieldId(101)) > 0);
}

TEST_F(SegmentLoadInfoTest,
       CompactRuntimeInfoForManifestSchemaCopyKeepsIndexIdentity) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    current_proto.set_manifest_path("/path/to/current_manifest");
    auto* cur_index = current_proto.add_index_infos();
    cur_index->set_fieldid(101);
    cur_index->set_indexid(1001);
    cur_index->add_index_file_paths("/path/to/index");
    auto* cur_param = cur_index->add_index_params();
    cur_param->set_key("index_type");
    cur_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    SegmentLoadInfo current_info(current_proto, schema_);
    current_info.SetColumnGroupsForTesting(
        std::make_shared<milvus_storage::api::ColumnGroups>());
    current_info.CompactRuntimeInfoForManifest();

    auto new_schema = std::make_shared<Schema>();
    new_schema->AddDebugField("pk", DataType::INT64);
    new_schema->AddDebugField(
        "vec", DataType::VECTOR_FLOAT, 128, knowhere::metric::L2);
    new_schema->AddDebugField("new_field", DataType::INT64);
    new_schema->set_primary_field_id(FieldId(100));
    new_schema->set_schema_version(schema_->get_schema_version() + 1);

    SegmentLoadInfo new_info(current_info);
    new_info.ReplaceSchemaForReopen(new_schema);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.indexes_to_load.empty());
    EXPECT_TRUE(diff.indexes_to_replace.empty());
    EXPECT_TRUE(diff.indexes_to_drop.empty());
    EXPECT_NE(std::find(diff.fields_to_fill_default.begin(),
                        diff.fields_to_fill_default.end(),
                        FieldId(102)),
              diff.fields_to_fill_default.end());
}

TEST_F(SegmentLoadInfoTest,
       ReplaceSchemaForReopenPrunesDroppedFieldRuntimeState) {
    SegmentLoadInfo current_info(proto_, schema_);
    current_info.SetColumnGroupsForTesting(
        std::make_shared<milvus_storage::api::ColumnGroups>());
    current_info.SetFieldFilledWithDefault(FieldId(101));
    current_info.SetFieldFilledWithDefault(FieldId(102));
    current_info.SetTextIndexCreated(FieldId(101));
    current_info.SetTextIndexCreated(FieldId(102));
    current_info.CompactRuntimeInfoForManifest();

    auto new_schema = std::make_shared<Schema>();
    new_schema->AddField(
        FieldName("pk"), FieldId(100), DataType::INT64, false, std::nullopt);
    new_schema->AddField(FieldName("json_field"),
                         FieldId(102),
                         DataType::JSON,
                         false,
                         std::nullopt);
    new_schema->set_primary_field_id(FieldId(100));
    new_schema->set_schema_version(schema_->get_schema_version() + 1);

    SegmentLoadInfo new_info(current_info);
    new_info.ReplaceSchemaForReopen(new_schema);

    EXPECT_FALSE(new_info.HasIndexInfo(FieldId(101)));
    EXPECT_FALSE(new_info.HasIndexInfo(FieldId(102)));
    EXPECT_FALSE(new_info.IsFieldFilledWithDefault(FieldId(101)));
    EXPECT_TRUE(new_info.IsFieldFilledWithDefault(FieldId(102)));
    EXPECT_FALSE(new_info.HasTextIndexCreated(FieldId(101)));
    EXPECT_TRUE(new_info.HasTextIndexCreated(FieldId(102)));

    auto diff = current_info.ComputeDiff(new_info);
    EXPECT_TRUE(diff.indexes_to_load.empty());
    EXPECT_TRUE(diff.indexes_to_replace.empty());
    EXPECT_EQ(diff.indexes_to_drop, std::set<FieldId>({FieldId(101)}));

    SegmentLoadInfo next_info(new_info);
    next_info.ReplaceSchemaForReopen(new_schema);
    auto second_diff = new_info.ComputeDiff(next_info);
    EXPECT_TRUE(second_diff.indexes_to_load.empty());
    EXPECT_TRUE(second_diff.indexes_to_replace.empty());
    EXPECT_TRUE(second_diff.indexes_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, BinlogInfo) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_EQ(info.GetBinlogPathCount(), 2);
    EXPECT_TRUE(info.HasBinlogPath(FieldId(101)));
    EXPECT_TRUE(info.HasBinlogPath(FieldId(104)));
    EXPECT_FALSE(info.HasBinlogPath(FieldId(999)));

    auto binlog = info.GetFieldBinlog(FieldId(101));
    EXPECT_NE(binlog, nullptr);
    EXPECT_EQ(binlog->binlogs_size(), 2);

    auto paths = info.GetFieldBinlogPaths(FieldId(101));
    EXPECT_EQ(paths.size(), 2);
    EXPECT_EQ(paths[0], "/path/to/binlog1");
    EXPECT_EQ(paths[1], "/path/to/binlog2");

    auto row_count = info.GetFieldBinlogRowCount(FieldId(101));
    EXPECT_EQ(row_count, 1000);

    // Test non-existent field
    auto empty_paths = info.GetFieldBinlogPaths(FieldId(999));
    EXPECT_TRUE(empty_paths.empty());

    auto zero_count = info.GetFieldBinlogRowCount(FieldId(999));
    EXPECT_EQ(zero_count, 0);
}

TEST_F(SegmentLoadInfoTest, ColumnGroup) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_TRUE(info.IsColumnGroup(FieldId(104)));
    EXPECT_FALSE(info.IsColumnGroup(FieldId(101)));
    EXPECT_FALSE(info.IsColumnGroup(FieldId(999)));

    auto child_fields = info.GetChildFieldIds(FieldId(104));
    EXPECT_EQ(child_fields.size(), 2);
    EXPECT_EQ(child_fields[0], 105);
    EXPECT_EQ(child_fields[1], 106);

    auto empty_children = info.GetChildFieldIds(FieldId(101));
    EXPECT_TRUE(empty_children.empty());
}

TEST_F(SegmentLoadInfoTest, StatsAndDeltaLogs) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_EQ(info.GetStatslogCount(), 1);
    EXPECT_EQ(info.GetDeltalogCount(), 1);

    const auto& statslog = info.GetStatslog(0);
    EXPECT_EQ(statslog.fieldid(), 101);

    const auto& deltalog = info.GetDeltalog(0);
    EXPECT_EQ(deltalog.fieldid(), 0);
}

TEST_F(SegmentLoadInfoTest, TextStats) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_TRUE(info.HasTextStatsLog(101));
    EXPECT_FALSE(info.HasTextStatsLog(999));

    auto text_stats = info.GetTextStatsLog(101);
    EXPECT_NE(text_stats, nullptr);
    EXPECT_EQ(text_stats->fieldid(), 101);
    EXPECT_EQ(text_stats->version(), 1);

    auto null_stats = info.GetTextStatsLog(999);
    EXPECT_EQ(null_stats, nullptr);
}

TEST_F(SegmentLoadInfoTest, JsonKeyStats) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_TRUE(info.HasJsonKeyStatsLog(102));
    EXPECT_FALSE(info.HasJsonKeyStatsLog(999));

    auto json_stats = info.GetJsonKeyStatsLog(102);
    EXPECT_NE(json_stats, nullptr);
    EXPECT_EQ(json_stats->fieldid(), 102);

    auto null_stats = info.GetJsonKeyStatsLog(999);
    EXPECT_EQ(null_stats, nullptr);
}

TEST_F(SegmentLoadInfoTest, Bm25Logs) {
    SegmentLoadInfo info(proto_, schema_);

    EXPECT_EQ(info.GetBm25logCount(), 1);
    const auto& bm25log = info.GetBm25log(0);
    EXPECT_EQ(bm25log.fieldid(), 103);
}

TEST_F(SegmentLoadInfoTest, UnderlyingProtoAccess) {
    SegmentLoadInfo info(proto_, schema_);

    const auto& proto = info.GetProto();
    EXPECT_EQ(proto.segmentid(), 12345);

    auto* mutable_proto = info.MutableProto();
    mutable_proto->set_segmentid(99999);
    info.RebuildCache();

    EXPECT_EQ(info.GetSegmentID(), 99999);
}

TEST_F(SegmentLoadInfoTest, EmptyManifestPath) {
    proto::segcore::SegmentLoadInfo empty_proto;
    empty_proto.set_segmentid(1);
    empty_proto.set_num_of_rows(100);

    SegmentLoadInfo info(empty_proto, schema_);
    EXPECT_FALSE(info.HasManifestPath());
    EXPECT_TRUE(info.GetManifestPath().empty());
}

TEST_F(SegmentLoadInfoTest, IndexWithoutFiles) {
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(1);

    // Add index info without files - should be ignored in cache
    auto* index_info = test_proto.add_index_infos();
    index_info->set_fieldid(101);
    index_info->set_indexid(1001);
    // No index_file_paths added

    SegmentLoadInfo info(test_proto, schema_);

    // Index without files should not be in cache
    EXPECT_FALSE(info.HasIndexInfo(FieldId(101)));
    EXPECT_EQ(info.GetIndexInfoCount(), 1);  // Proto still has it
}

// ==================== GetLoadDiff Tests ====================

TEST_F(SegmentLoadInfoTest, GetLoadDiffWithIndexesOnly) {
    // Create info with only indexes (no binlogs, no manifest)
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add two indexes
    auto* index1 = test_proto.add_index_infos();
    index1->set_fieldid(101);
    index1->set_indexid(1001);
    index1->add_index_file_paths("/path/to/index1");
    auto* index_param1 = index1->add_index_params();
    index_param1->set_key("index_type");
    index_param1->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    auto* index2 = test_proto.add_index_infos();
    index2->set_fieldid(102);
    index2->set_indexid(1002);
    index2->add_index_file_paths("/path/to/index2");
    auto* index_param2 = index2->add_index_params();
    index_param2->set_key("index_type");
    index_param2->set_value(milvus::index::INVERTED_INDEX_TYPE);

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();
    std::cout << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Both indexes should be in indexes_to_load
    EXPECT_EQ(diff.indexes_to_load.size(), 2);
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(101)) > 0);
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(102)) > 0);
    EXPECT_EQ(diff.indexes_to_load[FieldId(101)].size(), 1);
    EXPECT_EQ(diff.indexes_to_load[FieldId(102)].size(), 1);

    // No drops or other changes
    EXPECT_TRUE(diff.indexes_to_drop.empty());
    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffWithBinlogsOnly) {
    // Create info with only binlogs (no indexes, no manifest)
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add binlog with child fields (field 104 with children 105, 106)
    auto* binlog = test_proto.add_binlog_paths();
    binlog->set_fieldid(104);
    binlog->add_child_fields(105);
    binlog->add_child_fields(106);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/binlog");
    log->set_entries_num(500);

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();
    std::cout << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Binlogs should be in binlogs_to_load
    EXPECT_EQ(diff.binlogs_to_load.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first.size(), 2);  // 2 child fields
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(), 105);
    EXPECT_EQ(diff.binlogs_to_load[0].first[1].get(), 106);

    // No index changes
    EXPECT_TRUE(diff.indexes_to_load.empty());
    EXPECT_TRUE(diff.indexes_to_drop.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffWithIndexesAndBinlogs) {
    // Create info with both indexes and binlogs
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add index
    auto* index = test_proto.add_index_infos();
    index->set_fieldid(101);
    index->set_indexid(1001);
    index->add_index_file_paths("/path/to/index");
    auto* index_param = index->add_index_params();
    index_param->set_key("index_type");
    index_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    // Add binlog with child fields (field 104 with child 105)
    auto* binlog = test_proto.add_binlog_paths();
    binlog->set_fieldid(104);
    binlog->add_child_fields(105);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/binlog");
    log->set_entries_num(500);

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();
    std::cout << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Index should be in indexes_to_load
    EXPECT_EQ(diff.indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(101)) > 0);

    // Binlog should be in binlogs_to_load
    EXPECT_EQ(diff.binlogs_to_load.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(), 105);

    // No drops
    EXPECT_TRUE(diff.indexes_to_drop.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffIgnoresIndexesWithoutFiles) {
    // Indexes without files should be ignored in GetLoadDiff
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add index with files
    auto* index1 = test_proto.add_index_infos();
    index1->set_fieldid(101);
    index1->set_indexid(1001);
    index1->add_index_file_paths("/path/to/index");
    auto* index_param1 = index1->add_index_params();
    index_param1->set_key("index_type");
    index_param1->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    // Add index without files - should be ignored
    auto* index2 = test_proto.add_index_infos();
    index2->set_fieldid(102);
    index2->set_indexid(1002);
    // No index_file_paths added

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();
    std::cout << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Only index with files should be in indexes_to_load
    EXPECT_EQ(diff.indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(101)) > 0);
    EXPECT_FALSE(diff.indexes_to_load.count(FieldId(102)) > 0);
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffWithMultipleIndexesPerField) {
    // A field can have multiple indexes (e.g., JSON field with multiple paths)
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add two indexes for the same field
    auto* index1 = test_proto.add_index_infos();
    index1->set_fieldid(101);
    index1->set_indexid(1001);
    index1->add_index_file_paths("/path/to/index1");
    auto* index_param1 = index1->add_index_params();
    index_param1->set_key("index_type");
    index_param1->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    auto* index2 = test_proto.add_index_infos();
    index2->set_fieldid(101);
    index2->set_indexid(1002);
    index2->add_index_file_paths("/path/to/index2");
    auto* index_param2 = index2->add_index_params();
    index_param2->set_key("index_type");
    index_param2->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();

    EXPECT_TRUE(diff.HasChanges());
    // Both indexes should be in indexes_to_load for the same field
    EXPECT_EQ(diff.indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(101)) > 0);
    EXPECT_EQ(diff.indexes_to_load[FieldId(101)].size(), 2);
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffWithMultipleBinlogGroups) {
    // Test with multiple binlog groups
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add first binlog group (field 104 with children 105, 106)
    auto* binlog1 = test_proto.add_binlog_paths();
    binlog1->set_fieldid(104);
    binlog1->add_child_fields(105);
    binlog1->add_child_fields(106);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/binlog1");
    log1->set_entries_num(500);

    // Add second binlog group (field 108 with child 109)
    auto* binlog2 = test_proto.add_binlog_paths();
    binlog2->set_fieldid(108);
    binlog2->add_child_fields(109);
    auto* log2 = binlog2->add_binlogs();
    log2->set_log_path("/path/to/binlog2");
    log2->set_entries_num(500);

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();

    EXPECT_TRUE(diff.HasChanges());
    // Both binlog groups should be in binlogs_to_load
    EXPECT_EQ(diff.binlogs_to_load.size(), 2);

    // Check first group
    EXPECT_EQ(diff.binlogs_to_load[0].first.size(), 2);
    // Check second group
    EXPECT_EQ(diff.binlogs_to_load[1].first.size(), 1);
}

// ==================== Legacy Format (v1) Tests ====================
// Test binlogs without child_fields (v1/legacy format where group_id == field_id)

TEST_F(SegmentLoadInfoTest, GetLoadDiffWithBinlogsLegacyFormat) {
    // Create info with binlogs without child_fields (legacy/v1 format)
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add binlog WITHOUT child_fields - this is the legacy format
    auto* binlog = test_proto.add_binlog_paths();
    binlog->set_fieldid(101);
    // Note: no child_fields added - this triggers the legacy handling
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/binlog");
    log->set_entries_num(500);

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();
    std::cout << "Legacy format diff: " << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // In legacy format, field_id itself is used as the child_id
    EXPECT_EQ(diff.binlogs_to_load.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(),
              101);  // field_id as child_id

    // No index changes
    EXPECT_TRUE(diff.indexes_to_load.empty());
    EXPECT_TRUE(diff.indexes_to_drop.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffWithMultipleBinlogsLegacyFormat) {
    // Test multiple binlogs in legacy format
    proto::segcore::SegmentLoadInfo test_proto;
    test_proto.set_segmentid(100);
    test_proto.set_num_of_rows(1000);

    // Add first binlog without child_fields
    auto* binlog1 = test_proto.add_binlog_paths();
    binlog1->set_fieldid(101);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/binlog1");
    log1->set_entries_num(500);

    // Add second binlog without child_fields
    auto* binlog2 = test_proto.add_binlog_paths();
    binlog2->set_fieldid(102);
    auto* log2 = binlog2->add_binlogs();
    log2->set_log_path("/path/to/binlog2");
    log2->set_entries_num(500);

    SegmentLoadInfo info(test_proto, schema_);
    auto diff = info.GetLoadDiff();

    EXPECT_TRUE(diff.HasChanges());
    EXPECT_EQ(diff.binlogs_to_load.size(), 2);
    EXPECT_EQ(diff.binlogs_to_load[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(), 101);
    EXPECT_EQ(diff.binlogs_to_load[1].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[1].first[0].get(), 102);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffWithBinlogsLegacyFormat) {
    // Test ComputeDiff between two SegmentLoadInfos using legacy format

    // Current: has field 101
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* binlog1 = current_proto.add_binlog_paths();
    binlog1->set_fieldid(101);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/binlog1");
    log1->set_entries_num(500);

    // New: has field 101 and 102
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog1 = new_proto.add_binlog_paths();
    new_binlog1->set_fieldid(101);
    auto* new_log1 = new_binlog1->add_binlogs();
    new_log1->set_log_path("/path/to/binlog1");
    new_log1->set_entries_num(500);

    auto* new_binlog2 = new_proto.add_binlog_paths();
    new_binlog2->set_fieldid(102);
    auto* new_log2 = new_binlog2->add_binlogs();
    new_log2->set_log_path("/path/to/binlog2");
    new_log2->set_entries_num(500);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "ComputeDiff legacy format: " << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Field 102 should be added
    EXPECT_EQ(diff.binlogs_to_load.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(), 102);

    // No fields should be dropped or replaced
    EXPECT_TRUE(diff.field_data_to_drop.empty());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDropFieldLegacyFormat) {
    // Test ComputeDiff when dropping a field in legacy format

    // Current: has fields 101 and 102
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    auto* binlog1 = current_proto.add_binlog_paths();
    binlog1->set_fieldid(101);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/binlog1");
    log1->set_entries_num(500);

    auto* binlog2 = current_proto.add_binlog_paths();
    binlog2->set_fieldid(102);
    auto* log2 = binlog2->add_binlogs();
    log2->set_log_path("/path/to/binlog2");
    log2->set_entries_num(500);

    // New: only has field 101 (102 is dropped)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);

    auto* new_binlog1 = new_proto.add_binlog_paths();
    new_binlog1->set_fieldid(101);
    auto* new_log1 = new_binlog1->add_binlogs();
    new_log1->set_log_path("/path/to/binlog1");
    new_log1->set_entries_num(500);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.HasChanges());
    // No new binlogs to load or replace
    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
    // Field 102 should be dropped
    EXPECT_EQ(diff.field_data_to_drop.size(), 1);
    EXPECT_TRUE(diff.field_data_to_drop.count(FieldId(102)) > 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffMixedFormats) {
    // Test ComputeDiff with mixed formats: legacy and column group

    // Current: has field 101 (legacy) and column group 104 with child fields 105, 106
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    // Legacy format binlog
    auto* binlog1 = current_proto.add_binlog_paths();
    binlog1->set_fieldid(101);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/binlog1");
    log1->set_entries_num(500);

    // Column group format binlog
    auto* binlog2 = current_proto.add_binlog_paths();
    binlog2->set_fieldid(104);
    binlog2->add_child_fields(105);
    binlog2->add_child_fields(106);
    auto* log2 = binlog2->add_binlogs();
    log2->set_log_path("/path/to/group_binlog");
    log2->set_entries_num(1000);

    // New: has all existing + new field 103 (legacy) + new field 107 in column group
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);

    auto* new_binlog1 = new_proto.add_binlog_paths();
    new_binlog1->set_fieldid(101);
    auto* new_log1 = new_binlog1->add_binlogs();
    new_log1->set_log_path("/path/to/binlog1");
    new_log1->set_entries_num(500);

    // New legacy field
    auto* new_binlog3 = new_proto.add_binlog_paths();
    new_binlog3->set_fieldid(103);
    auto* new_log3 = new_binlog3->add_binlogs();
    new_log3->set_log_path("/path/to/binlog3");
    new_log3->set_entries_num(500);

    // Updated column group with additional field
    auto* new_binlog2 = new_proto.add_binlog_paths();
    new_binlog2->set_fieldid(104);
    new_binlog2->add_child_fields(105);
    new_binlog2->add_child_fields(106);
    new_binlog2->add_child_fields(107);  // New field added to group
    auto* new_log2 = new_binlog2->add_binlogs();
    new_log2->set_log_path("/path/to/group_binlog_new");
    new_log2->set_entries_num(1500);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Mixed format diff: " << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Field 103 (legacy) and field 107 (in column group) should be loaded
    EXPECT_EQ(diff.binlogs_to_load.size(), 2);

    // Fields 105, 106 were in current (group 104) and group 104 is not found
    // as a child_id key in current_fields, so they go to binlogs_to_replace
    EXPECT_EQ(diff.binlogs_to_replace.size(), 1);
    EXPECT_EQ(diff.binlogs_to_replace[0].first.size(), 2);
    std::set<int64_t> replace_field_ids;
    for (const auto& fid : diff.binlogs_to_replace[0].first) {
        replace_field_ids.insert(fid.get());
    }
    EXPECT_TRUE(replace_field_ids.count(105) > 0);
    EXPECT_TRUE(replace_field_ids.count(106) > 0);

    // No fields should be dropped
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffNoChangesLegacyFormat) {
    // Test ComputeDiff when there are no changes (same binlogs)
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(101);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/binlog");
    log->set_entries_num(500);

    SegmentLoadInfo current_info(proto, schema_);
    auto diff = current_info.GetLoadDiff();
    for (const auto& field_id : diff.fields_to_fill_default) {
        current_info.SetFieldFilledWithDefault(field_id);
    }
    SegmentLoadInfo new_info(proto, schema_);
    diff = current_info.ComputeDiff(new_info);

    EXPECT_FALSE(diff.HasChanges());
    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
    EXPECT_TRUE(diff.indexes_to_replace.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

// ==================== Default Value Filling Tests ====================

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFieldsBasic) {
    // Test that fields without data sources are added to fields_to_fill_default
    // Schema has fields 100-110, but we only provide binlog for field 100 (pk)
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    // Only add binlog for pk field (100)
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    SegmentLoadInfo empty_info(milvus::proto::segcore::SegmentLoadInfo(),
                               schema_);
    SegmentLoadInfo new_info(proto, schema_);

    auto diff = empty_info.ComputeDiff(new_info);

    // Fields 101-110 should be in fields_to_fill_default (no data source)
    // Field 100 has binlog, so it should NOT be in fields_to_fill_default
    EXPECT_TRUE(diff.HasChanges());
    EXPECT_FALSE(diff.fields_to_fill_default.empty());

    // Check that field 100 is NOT in fields_to_fill_default
    for (const auto& field_id : diff.fields_to_fill_default) {
        EXPECT_NE(field_id.get(), 100);
    }

    // Fields 101-110 should be in fields_to_fill_default
    std::set<int64_t> expected_fields = {
        101, 102, 103, 104, 105, 106, 107, 108, 109, 110};
    std::set<int64_t> actual_fields;
    for (const auto& field_id : diff.fields_to_fill_default) {
        actual_fields.insert(field_id.get());
    }
    EXPECT_EQ(actual_fields, expected_fields);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFieldsSkipsFunctionOutput) {
    auto schema = std::make_shared<Schema>();
    schema->AddDebugField("pk", DataType::INT64);
    schema->AddDebugField(
        "vec", DataType::VECTOR_FLOAT, 128, knowhere::metric::L2);
    schema->AddDebugField(
        "sparse_out", DataType::VECTOR_SPARSE_U32_F32, 0, std::nullopt);
    schema->set_primary_field_id(FieldId(100));
    const FieldId sparse_field_id = FieldId(102);
    schema->add_function_output_field_id(sparse_field_id);

    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    SegmentLoadInfo empty_info(milvus::proto::segcore::SegmentLoadInfo(),
                               schema);
    SegmentLoadInfo new_info(proto, schema);
    auto diff = empty_info.ComputeDiff(new_info);

    for (const auto& field_id : diff.fields_to_fill_default) {
        EXPECT_NE(field_id, sparse_field_id);
    }
    EXPECT_NE(std::find(diff.fields_to_fill_default.begin(),
                        diff.fields_to_fill_default.end(),
                        FieldId(101)),
              diff.fields_to_fill_default.end());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFieldsSkipAlreadyFilled) {
    // Test that fields already filled with default values are skipped
    // on subsequent ComputeDiff calls
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    // Only add binlog for pk field (100)
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    // First diff: empty -> proto (simulates initial load)
    SegmentLoadInfo empty_info(milvus::proto::segcore::SegmentLoadInfo(),
                               schema_);
    SegmentLoadInfo first_new_info(proto, schema_);
    auto first_diff = empty_info.ComputeDiff(first_new_info);

    // Verify fields were added to fields_to_fill_default
    EXPECT_FALSE(first_diff.fields_to_fill_default.empty());
    for (const auto& field_id : first_diff.fields_to_fill_default) {
        first_new_info.SetFieldFilledWithDefault(field_id);
    }

    SegmentLoadInfo second_new_info(proto, schema_);
    auto second_diff = first_new_info.ComputeDiff(second_new_info);

    // All previously filled fields should be skipped
    EXPECT_TRUE(second_diff.fields_to_fill_default.empty());

    second_new_info.SetFieldsFilledWithDefault(
        first_new_info.GetDefaultFilledFieldsForNewInfo(second_new_info));
    SegmentLoadInfo third_new_info(proto, schema_);
    auto third_diff = second_new_info.ComputeDiff(third_new_info);
    EXPECT_TRUE(third_diff.fields_to_fill_default.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFieldsWithIndex) {
    // Test that fields with index (that has raw data) are NOT added
    // to fields_to_fill_default
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    // Add binlog for pk field (100)
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    // Add index for field 101 (vec field) - FLAT index has raw data
    auto* index_info = proto.add_index_infos();
    index_info->set_fieldid(101);
    index_info->set_indexid(1001);
    index_info->add_index_file_paths("/path/to/index");
    auto* index_param = index_info->add_index_params();
    index_param->set_key("index_type");
    index_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IDMAP);

    SegmentLoadInfo empty_info(milvus::proto::segcore::SegmentLoadInfo(),
                               schema_);
    SegmentLoadInfo new_info(proto, schema_);
    auto diff = empty_info.ComputeDiff(new_info);

    // Field 101 should NOT be in fields_to_fill_default (has index with raw
    // data)
    for (const auto& field_id : diff.fields_to_fill_default) {
        EXPECT_NE(field_id.get(), 101);
    }

    // Field 101 should be in indexes_to_load
    EXPECT_TRUE(diff.indexes_to_load.find(FieldId(101)) !=
                diff.indexes_to_load.end());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFieldsNotInBinlogsToLoad) {
    // Test that fields being loaded via binlogs_to_load are NOT added
    // to fields_to_fill_default
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    // Current has only pk field
    auto* binlog1 = current_proto.add_binlog_paths();
    binlog1->set_fieldid(100);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/pk_binlog");
    log1->set_entries_num(1000);

    // New has pk + field 101
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);

    auto* new_binlog1 = new_proto.add_binlog_paths();
    new_binlog1->set_fieldid(100);
    auto* new_log1 = new_binlog1->add_binlogs();
    new_log1->set_log_path("/path/to/pk_binlog");
    new_log1->set_entries_num(1000);

    auto* new_binlog2 = new_proto.add_binlog_paths();
    new_binlog2->set_fieldid(101);
    auto* new_log2 = new_binlog2->add_binlogs();
    new_log2->set_log_path("/path/to/field101_binlog");
    new_log2->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    // Field 101 should be in binlogs_to_load, NOT in fields_to_fill_default
    bool found_in_binlogs = false;
    for (const auto& [field_ids, binlog_ptr] : diff.binlogs_to_load) {
        for (const auto& fid : field_ids) {
            if (fid.get() == 101) {
                found_in_binlogs = true;
                break;
            }
        }
    }
    EXPECT_TRUE(found_in_binlogs);

    // Field 101 should NOT be in fields_to_fill_default
    for (const auto& field_id : diff.fields_to_fill_default) {
        EXPECT_NE(field_id.get(), 101);
    }
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffDefaultFields) {
    // Test GetLoadDiff includes fields_to_fill_default for initial load
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    // Only add binlog for pk field (100) and vec field (101)
    auto* binlog1 = proto.add_binlog_paths();
    binlog1->set_fieldid(100);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/pk_binlog");
    log1->set_entries_num(1000);

    auto* binlog2 = proto.add_binlog_paths();
    binlog2->set_fieldid(101);
    auto* log2 = binlog2->add_binlogs();
    log2->set_log_path("/path/to/vec_binlog");
    log2->set_entries_num(1000);

    SegmentLoadInfo info(proto, schema_);
    auto diff = info.GetLoadDiff();

    // Fields 102-110 should be in fields_to_fill_default
    std::set<int64_t> expected_fields = {
        102, 103, 104, 105, 106, 107, 108, 109, 110};
    std::set<int64_t> actual_fields;
    for (const auto& field_id : diff.fields_to_fill_default) {
        actual_fields.insert(field_id.get());
    }
    EXPECT_EQ(actual_fields, expected_fields);
}

TEST_F(SegmentLoadInfoTest, LoadDiffToStringIncludesDefaultFields) {
    // Test that LoadDiff::ToString includes fields_to_fill_default
    LoadDiff diff;
    diff.fields_to_fill_default.push_back(FieldId(102));
    diff.fields_to_fill_default.push_back(FieldId(103));

    std::string str = diff.ToString();
    EXPECT_TRUE(str.find("fields_to_fill_default") != std::string::npos);
    EXPECT_TRUE(str.find("102") != std::string::npos);
    EXPECT_TRUE(str.find("103") != std::string::npos);
}

TEST_F(SegmentLoadInfoTest, LoadDiffHasChangesWithDefaultFields) {
    // Test that HasChanges returns true when only fields_to_fill_default
    // is non-empty
    LoadDiff diff;
    EXPECT_FALSE(diff.HasChanges());

    diff.fields_to_fill_default.push_back(FieldId(102));
    EXPECT_TRUE(diff.HasChanges());
}

TEST_F(SegmentLoadInfoTest, CopyPreservesDefaultFilledFields) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* current_binlog = current_proto.add_binlog_paths();
    current_binlog->set_fieldid(100);
    auto* current_log = current_binlog->add_binlogs();
    current_log->set_log_path("/path/to/pk");
    current_log->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);
    current_info.SetFieldFilledWithDefault(FieldId(101));

    SegmentLoadInfo copied(current_info);
    EXPECT_TRUE(copied.IsFieldFilledWithDefault(FieldId(101)));

    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(101);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/field101");
    new_log->set_entries_num(1000);

    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = copied.ComputeDiff(new_info);

    bool found_replace = false;
    for (const auto& [field_ids, binlog] : diff.binlogs_to_replace) {
        (void)binlog;
        for (auto field_id : field_ids) {
            found_replace = found_replace || field_id == FieldId(101);
        }
    }
    EXPECT_TRUE(found_replace);
}

TEST_F(SegmentLoadInfoTest, MovePreservesRawDataIndexFieldsForDefaultDiff) {
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk");
    log->set_entries_num(1000);

    auto* index_info = proto.add_index_infos();
    index_info->set_fieldid(101);
    index_info->set_indexid(1001);
    index_info->add_index_file_paths("/path/to/index");
    auto* index_param = index_info->add_index_params();
    index_param->set_key("index_type");
    index_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IDMAP);

    auto assert_field_101_has_data = [](SegmentLoadInfo& info) {
        auto diff = info.GetLoadDiff();
        EXPECT_TRUE(diff.indexes_to_load.count(FieldId(101)) > 0);
        for (const auto& field_id : diff.fields_to_fill_default) {
            EXPECT_NE(field_id, FieldId(101));
        }
    };

    SegmentLoadInfo original(proto, schema_);
    SegmentLoadInfo moved(std::move(original));
    assert_field_101_has_data(moved);

    SegmentLoadInfo original_for_assignment(proto, schema_);
    proto::segcore::SegmentLoadInfo empty_proto;
    SegmentLoadInfo assigned(empty_proto, schema_);
    assigned = std::move(original_for_assignment);
    assert_field_101_has_data(assigned);
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffClearsDefaultFilledWhenDataSourceAppears) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* current_binlog = current_proto.add_binlog_paths();
    current_binlog->set_fieldid(100);
    auto* current_log = current_binlog->add_binlogs();
    current_log->set_log_path("/path/to/pk");
    current_log->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);
    current_info.SetFieldFilledWithDefault(FieldId(101));

    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    auto* field_binlog = new_proto.add_binlog_paths();
    field_binlog->set_fieldid(101);
    auto* field_log = field_binlog->add_binlogs();
    field_log->set_log_path("/path/to/field101");
    field_log->set_entries_num(1000);

    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    for (const auto& field_id : diff.fields_to_fill_default) {
        EXPECT_NE(field_id, FieldId(101));
    }
    EXPECT_FALSE(new_info.IsFieldFilledWithDefault(FieldId(101)));
}

// ==================== Text Index Tests ====================

// Helper: create a schema with varchar field that has enable_match=true
static SchemaPtr
CreateSchemaWithTextMatchField() {
    auto schema = std::make_shared<Schema>();
    // 100=pk, 101=vec, 102=varchar_match, 103=varchar_no_match
    schema->AddDebugField("pk", DataType::INT64);
    schema->AddDebugField(
        "vec", DataType::VECTOR_FLOAT, 128, knowhere::metric::L2);

    // Varchar field with enable_match=true (field 102)
    std::map<std::string, std::string> params;
    schema->AddDebugVarcharField(FieldName("text_field"),
                                 DataType::VARCHAR,
                                 /*max_length=*/65535,
                                 /*nullable=*/false,
                                 /*enable_match=*/true,
                                 /*enable_analyzer=*/true,
                                 params,
                                 std::nullopt);

    // Varchar field without enable_match (field 103)
    schema->AddDebugField("plain_varchar", DataType::VARCHAR);

    schema->set_primary_field_id(FieldId(100));
    return schema;
}

TEST_F(SegmentLoadInfoTest, SetTextIndexCreatedAndHas) {
    // Test SetTextIndexCreated/HasTextIndexCreated tracking
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    SegmentLoadInfo info(proto, schema_);

    EXPECT_FALSE(info.HasTextIndexCreated(FieldId(101)));

    info.SetTextIndexCreated(FieldId(101));
    EXPECT_TRUE(info.HasTextIndexCreated(FieldId(101)));
    EXPECT_FALSE(info.HasTextIndexCreated(FieldId(102)));

    info.SetTextIndexCreated(FieldId(102));
    EXPECT_TRUE(info.HasTextIndexCreated(FieldId(102)));
}

TEST_F(SegmentLoadInfoTest, CreatedTextIndexesCopyConstructor) {
    // Test that created_text_indexes_ is preserved via copy constructor
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    SegmentLoadInfo info1(proto, schema_);
    info1.SetTextIndexCreated(FieldId(101));
    info1.SetTextIndexCreated(FieldId(102));

    // Copy constructor
    SegmentLoadInfo info2(info1);
    EXPECT_TRUE(info2.HasTextIndexCreated(FieldId(101)));
    EXPECT_TRUE(info2.HasTextIndexCreated(FieldId(102)));
    EXPECT_FALSE(info2.HasTextIndexCreated(FieldId(103)));
}

TEST_F(SegmentLoadInfoTest, CreatedTextIndexesMoveConstructor) {
    // Test that created_text_indexes_ is preserved via move constructor
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    SegmentLoadInfo info1(proto, schema_);
    info1.SetTextIndexCreated(FieldId(101));

    SegmentLoadInfo info2(std::move(info1));
    EXPECT_TRUE(info2.HasTextIndexCreated(FieldId(101)));
}

TEST_F(SegmentLoadInfoTest, CreatedTextIndexesCopyAssignment) {
    // Test that created_text_indexes_ is preserved via copy assignment
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    SegmentLoadInfo info1(proto, schema_);
    info1.SetTextIndexCreated(FieldId(101));

    proto::segcore::SegmentLoadInfo empty_proto;
    SegmentLoadInfo info2(empty_proto, schema_);
    info2 = info1;

    EXPECT_TRUE(info2.HasTextIndexCreated(FieldId(101)));
}

TEST_F(SegmentLoadInfoTest, CreatedTextIndexesMoveAssignment) {
    // Test that created_text_indexes_ is preserved via move assignment
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    SegmentLoadInfo info1(proto, schema_);
    info1.SetTextIndexCreated(FieldId(101));

    proto::segcore::SegmentLoadInfo empty_proto;
    SegmentLoadInfo info2(empty_proto, schema_);
    info2 = std::move(info1);

    EXPECT_TRUE(info2.HasTextIndexCreated(FieldId(101)));
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffTextIndexesFromStats) {
    // Test GetLoadDiff loads pre-built text indexes from textstatslogs
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_collectionid(200);
    proto.set_partitionid(300);
    proto.set_num_of_rows(1000);

    // Add a binlog to satisfy GetLoadDiff precondition
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    // Add text stats for field 101 (simulates pre-built text index)
    auto& text_stats = (*proto.mutable_textstatslogs())[101];
    text_stats.set_fieldid(101);
    text_stats.set_version(1);
    text_stats.set_buildid(5001);
    text_stats.set_memory_size(1024);
    text_stats.set_current_scalar_index_version(3);
    text_stats.add_files("/path/to/text_index_file1");
    text_stats.add_files("/path/to/text_index_file2");

    SegmentLoadInfo info(proto, schema_);
    auto diff = info.GetLoadDiff();

    // The text index should be in text_indexes_to_load
    EXPECT_EQ(diff.text_indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.text_indexes_to_load.count(FieldId(101)) > 0);

    // Verify the converted LoadTextIndexInfo
    auto& load_info = diff.text_indexes_to_load[FieldId(101)];
    EXPECT_NE(load_info, nullptr);
    EXPECT_EQ(load_info->fieldid(), 101);
    EXPECT_EQ(load_info->version(), 1);
    EXPECT_EQ(load_info->buildid(), 5001);
    EXPECT_EQ(load_info->files_size(), 2);
    EXPECT_EQ(load_info->files(0), "/path/to/text_index_file1");
    EXPECT_EQ(load_info->files(1), "/path/to/text_index_file2");
    EXPECT_EQ(load_info->collectionid(), 200);
    EXPECT_EQ(load_info->partitionid(), 300);
    EXPECT_EQ(load_info->index_size(), 1024);
    EXPECT_EQ(load_info->current_scalar_index_version(), 3);
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffTextIndexesToCreate) {
    // Test GetLoadDiff creates text indexes for enable_match fields
    // without pre-built text indexes
    auto text_schema = CreateSchemaWithTextMatchField();

    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    // Add a binlog to satisfy GetLoadDiff precondition
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    // No text stats provided - the text_field (102) has enable_match
    // so it should be in text_indexes_to_create
    SegmentLoadInfo info(proto, text_schema);
    auto diff = info.GetLoadDiff();

    // text_field (102) should be in text_indexes_to_create
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(102)) > 0);
    // plain_varchar (103) should NOT be (no enable_match)
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(103)) == 0);
    // No text indexes to load (no stats)
    EXPECT_TRUE(diff.text_indexes_to_load.empty());
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffTextIndexesPreBuiltSkipsCreate) {
    // When a pre-built text index exists, the field should NOT be
    // in text_indexes_to_create
    auto text_schema = CreateSchemaWithTextMatchField();

    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_collectionid(200);
    proto.set_partitionid(300);
    proto.set_num_of_rows(1000);

    // Add a binlog to satisfy GetLoadDiff precondition
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    // Add pre-built text index for field 102 (text_field)
    auto& text_stats = (*proto.mutable_textstatslogs())[102];
    text_stats.set_fieldid(102);
    text_stats.set_version(1);
    text_stats.set_buildid(5001);
    text_stats.add_files("/path/to/text_index");

    SegmentLoadInfo info(proto, text_schema);
    auto diff = info.GetLoadDiff();

    // Field 102 should be in text_indexes_to_load (pre-built)
    EXPECT_EQ(diff.text_indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.text_indexes_to_load.count(FieldId(102)) > 0);

    // Field 102 should NOT be in text_indexes_to_create (has pre-built)
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(102)) == 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffTextIndexNewTextStats) {
    // ComputeDiff: new_info has text stats that current doesn't -> load
    auto text_schema = CreateSchemaWithTextMatchField();

    // Current: no text stats
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    // New: has text stats for field 102
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_collectionid(200);
    new_proto.set_partitionid(300);
    new_proto.set_num_of_rows(1000);
    auto& text_stats = (*new_proto.mutable_textstatslogs())[102];
    text_stats.set_fieldid(102);
    text_stats.set_version(1);
    text_stats.set_buildid(5001);
    text_stats.add_files("/path/to/text_index");

    SegmentLoadInfo current_info(current_proto, text_schema);
    SegmentLoadInfo new_info(new_proto, text_schema);
    auto diff = current_info.ComputeDiff(new_info);

    // Field 102 should be in text_indexes_to_load
    EXPECT_EQ(diff.text_indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.text_indexes_to_load.count(FieldId(102)) > 0);

    // Field 102 should NOT be in text_indexes_to_create (has pre-built in new)
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(102)) == 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffTextIndexAlreadyLoaded) {
    // ComputeDiff: both current and new have same text stats -> no diff
    auto text_schema = CreateSchemaWithTextMatchField();

    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_collectionid(200);
    proto.set_partitionid(300);
    proto.set_num_of_rows(1000);

    // Add a binlog to satisfy GetLoadDiff precondition
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    auto& text_stats = (*proto.mutable_textstatslogs())[102];
    text_stats.set_fieldid(102);
    text_stats.set_version(1);
    text_stats.set_buildid(5001);
    text_stats.add_files("/path/to/text_index");

    SegmentLoadInfo current_info(proto, text_schema);
    // Calculate first diff to populate default fields etc.
    auto first_diff = current_info.GetLoadDiff();
    SegmentLoadInfo new_info(proto, text_schema);
    auto diff = current_info.ComputeDiff(new_info);

    // No text index changes - both have the same text stats
    EXPECT_TRUE(diff.text_indexes_to_load.empty());
    EXPECT_TRUE(diff.text_indexes_to_create.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffTextIndexCreatedFromRawData) {
    // ComputeDiff: current already created text index from raw data ->
    // should not be re-created or re-loaded
    auto text_schema = CreateSchemaWithTextMatchField();

    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);

    SegmentLoadInfo current_info(proto, text_schema);
    // Mark field 102 as having been created from raw data
    current_info.SetTextIndexCreated(FieldId(102));

    SegmentLoadInfo new_info(proto, text_schema);
    auto diff = current_info.ComputeDiff(new_info);

    // Field 102 should NOT be in text_indexes_to_create (already created)
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(102)) == 0);
    // Field 102 should NOT be in text_indexes_to_load (no stats in new)
    EXPECT_TRUE(diff.text_indexes_to_load.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffTextIndexCreateForUnindexed) {
    // ComputeDiff: enable_match field without pre-built index and not yet
    // created -> should be in text_indexes_to_create
    auto text_schema = CreateSchemaWithTextMatchField();

    // Current: no text stats, no created indexes
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    // New: also no text stats for field 102
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);

    SegmentLoadInfo current_info(current_proto, text_schema);
    SegmentLoadInfo new_info(new_proto, text_schema);
    auto diff = current_info.ComputeDiff(new_info);

    // Field 102 (enable_match=true) should be in text_indexes_to_create
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(102)) > 0);
    // Field 103 (no enable_match) should NOT be in text_indexes_to_create
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(103)) == 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffExternalTextIndexCreateForUnindexed) {
    auto text_schema = CreateSchemaWithTextMatchField();
    text_schema->set_external_source("s3://external-bucket/table");

    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);

    SegmentLoadInfo current_info(current_proto, text_schema);
    SegmentLoadInfo new_info(new_proto, text_schema);
    auto diff = current_info.ComputeDiff(new_info);

    // Match internal collection behavior: without persisted text index stats,
    // load can create a runtime text index from raw field data.
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(102)) > 0);
    EXPECT_TRUE(diff.text_indexes_to_load.empty());
}

TEST_F(SegmentLoadInfoTest, ConvertTextIndexStatsToLoadTextIndexInfo) {
    // Test the conversion of TextIndexStats to LoadTextIndexInfo
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_collectionid(200);
    proto.set_partitionid(300);
    proto.set_num_of_rows(1000);
    proto.set_priority(proto::common::LoadPriority::HIGH);

    proto::segcore::TextIndexStats text_stats;
    text_stats.set_fieldid(101);
    text_stats.set_version(3);
    text_stats.set_buildid(5001);
    text_stats.set_memory_size(2048);
    text_stats.set_current_scalar_index_version(3);
    text_stats.add_files("/path/to/file1");
    text_stats.add_files("/path/to/file2");
    text_stats.add_files("/path/to/file3");

    SegmentLoadInfo info(proto, schema_);
    auto load_info =
        info.ConvertTextIndexStatsToLoadTextIndexInfo(text_stats, FieldId(101));

    EXPECT_NE(load_info, nullptr);
    EXPECT_EQ(load_info->fieldid(), 101);
    EXPECT_EQ(load_info->version(), 3);
    EXPECT_EQ(load_info->buildid(), 5001);
    EXPECT_EQ(load_info->index_size(), 2048);
    EXPECT_EQ(load_info->current_scalar_index_version(), 3);
    EXPECT_EQ(load_info->files_size(), 3);
    EXPECT_EQ(load_info->files(0), "/path/to/file1");
    EXPECT_EQ(load_info->files(1), "/path/to/file2");
    EXPECT_EQ(load_info->files(2), "/path/to/file3");
    EXPECT_EQ(load_info->collectionid(), 200);
    EXPECT_EQ(load_info->partitionid(), 300);
    EXPECT_EQ(load_info->load_priority(), proto::common::LoadPriority::HIGH);
    // Schema should be populated
    EXPECT_TRUE(load_info->has_schema());
}

TEST_F(SegmentLoadInfoTest, LoadDiffHasChangesWithTextIndexesToLoad) {
    // Test that HasChanges returns true when text_indexes_to_load is non-empty
    LoadDiff diff;
    EXPECT_FALSE(diff.HasChanges());

    diff.text_indexes_to_load[FieldId(101)] =
        std::make_shared<proto::indexcgo::LoadTextIndexInfo>();
    EXPECT_TRUE(diff.HasChanges());
}

TEST_F(SegmentLoadInfoTest, LoadDiffHasChangesWithTextIndexesToCreate) {
    // Test that HasChanges returns true when text_indexes_to_create is non-empty
    LoadDiff diff;
    EXPECT_FALSE(diff.HasChanges());

    diff.text_indexes_to_create.insert(FieldId(102));
    EXPECT_TRUE(diff.HasChanges());
}

TEST_F(SegmentLoadInfoTest, LoadDiffToStringIncludesTextIndexes) {
    // Test that ToString includes text index fields
    LoadDiff diff;
    diff.text_indexes_to_load[FieldId(101)] =
        std::make_shared<proto::indexcgo::LoadTextIndexInfo>();
    diff.text_indexes_to_create.insert(FieldId(102));

    std::string str = diff.ToString();
    EXPECT_TRUE(str.find("text_indexes_to_load") != std::string::npos);
    EXPECT_TRUE(str.find("101") != std::string::npos);
    EXPECT_TRUE(str.find("text_indexes_to_create") != std::string::npos);
    EXPECT_TRUE(str.find("102") != std::string::npos);
}

TEST_F(SegmentLoadInfoTest, ConvertJsonKeyStatsToLoadJsonKeyIndexInfo) {
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_collectionid(200);
    proto.set_partitionid(300);
    proto.set_num_of_rows(1000);
    proto.set_priority(proto::common::LoadPriority::HIGH);

    proto::segcore::JsonKeyStats json_stats;
    json_stats.set_fieldid(102);
    json_stats.set_version(3);
    json_stats.set_buildid(5001);
    json_stats.set_log_size(4096);
    json_stats.set_memory_size(8192);
    json_stats.set_json_key_stats_data_format(
        std::stoll(JSON_STATS_DATA_FORMAT_VERSION));
    json_stats.set_base_path("/path/to/json_stats");
    json_stats.add_files("meta.json");
    json_stats.add_files("shared_key_index/file1");

    SegmentLoadInfo info(proto, schema_);
    auto load_info = info.ConvertJsonKeyStatsToLoadJsonKeyIndexInfo(
        json_stats, FieldId(102));

    EXPECT_NE(load_info, nullptr);
    EXPECT_EQ(load_info->fieldid(), 102);
    EXPECT_EQ(load_info->version(), 3);
    EXPECT_EQ(load_info->buildid(), 5001);
    EXPECT_EQ(load_info->stats_size(), 4096);
    EXPECT_EQ(load_info->base_path(), "/path/to/json_stats");
    EXPECT_EQ(load_info->files_size(), 2);
    EXPECT_EQ(load_info->files(0), "meta.json");
    EXPECT_EQ(load_info->files(1), "shared_key_index/file1");
    EXPECT_EQ(load_info->collectionid(), 200);
    EXPECT_EQ(load_info->partitionid(), 300);
    EXPECT_EQ(load_info->load_priority(), proto::common::LoadPriority::HIGH);
    EXPECT_TRUE(load_info->has_schema());
}

TEST_F(SegmentLoadInfoTest, LoadDiffHasChangesWithJsonStats) {
    LoadDiff diff;
    EXPECT_FALSE(diff.HasChanges());

    diff.json_stats_to_load[FieldId(102)] =
        std::make_shared<proto::indexcgo::LoadJsonKeyIndexInfo>();
    EXPECT_TRUE(diff.HasChanges());

    diff.json_stats_to_load.clear();
    diff.json_stats_to_replace[FieldId(102)] =
        std::make_shared<proto::indexcgo::LoadJsonKeyIndexInfo>();
    EXPECT_TRUE(diff.HasChanges());

    diff.json_stats_to_replace.clear();
    diff.json_stats_to_drop.insert(FieldId(102));
    EXPECT_TRUE(diff.HasChanges());
}

TEST_F(SegmentLoadInfoTest, LoadDiffToStringIncludesJsonStats) {
    LoadDiff diff;
    diff.json_stats_to_load[FieldId(102)] =
        std::make_shared<proto::indexcgo::LoadJsonKeyIndexInfo>();
    diff.json_stats_to_replace[FieldId(103)] =
        std::make_shared<proto::indexcgo::LoadJsonKeyIndexInfo>();
    diff.json_stats_to_drop.insert(FieldId(104));

    auto str = diff.ToString();
    EXPECT_TRUE(str.find("json_stats_to_load") != std::string::npos);
    EXPECT_TRUE(str.find("102") != std::string::npos);
    EXPECT_TRUE(str.find("json_stats_to_replace") != std::string::npos);
    EXPECT_TRUE(str.find("103") != std::string::npos);
    EXPECT_TRUE(str.find("json_stats_to_drop") != std::string::npos);
    EXPECT_TRUE(str.find("104") != std::string::npos);
}

TEST_F(SegmentLoadInfoTest, GetLoadDiffJsonStatsFromStats) {
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_collectionid(200);
    proto.set_partitionid(300);
    proto.set_num_of_rows(1000);
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    auto& json_stats = (*proto.mutable_jsonkeystatslogs())[102];
    json_stats.set_fieldid(102);
    json_stats.set_version(1);
    json_stats.set_buildid(5001);
    json_stats.set_log_size(4096);
    json_stats.set_memory_size(8192);
    json_stats.set_json_key_stats_data_format(
        std::stoll(JSON_STATS_DATA_FORMAT_VERSION));
    json_stats.set_base_path("/path/to/json_stats");
    json_stats.add_files("meta.json");

    SegmentLoadInfo info(proto, schema_);
    auto diff = info.GetLoadDiff();

    EXPECT_EQ(diff.json_stats_to_load.size(), 1);
    EXPECT_TRUE(diff.json_stats_to_load.count(FieldId(102)) > 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffJsonStatsNewSameReplaceDropAndSkip) {
    const auto current_format = std::stoll(JSON_STATS_DATA_FORMAT_VERSION);

    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_collectionid(200);
    current_proto.set_partitionid(300);
    current_proto.set_num_of_rows(1000);
    auto& current_stats = (*current_proto.mutable_jsonkeystatslogs())[102];
    current_stats.set_fieldid(102);
    current_stats.set_version(1);
    current_stats.set_buildid(5001);
    current_stats.set_log_size(4096);
    current_stats.set_memory_size(8192);
    current_stats.set_json_key_stats_data_format(current_format);
    current_stats.set_base_path("/path/to/json_stats");
    current_stats.add_files("meta.json");

    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_same_info(new_proto, schema_);
    auto same_diff = current_info.ComputeDiff(new_same_info);
    EXPECT_TRUE(same_diff.json_stats_to_load.empty());
    EXPECT_TRUE(same_diff.json_stats_to_replace.empty());
    EXPECT_TRUE(same_diff.json_stats_to_drop.empty());

    auto& new_stats = (*new_proto.mutable_jsonkeystatslogs())[103];
    new_stats.set_fieldid(103);
    new_stats.set_version(1);
    new_stats.set_buildid(5002);
    new_stats.set_log_size(1024);
    new_stats.set_memory_size(2048);
    new_stats.set_json_key_stats_data_format(current_format);
    new_stats.set_base_path("/path/to/new_json_stats");
    new_stats.add_files("meta.json");
    SegmentLoadInfo new_info(new_proto, schema_);
    auto new_diff = current_info.ComputeDiff(new_info);
    EXPECT_EQ(new_diff.json_stats_to_load.size(), 1);
    EXPECT_TRUE(new_diff.json_stats_to_load.count(FieldId(103)) > 0);

    (*new_proto.mutable_jsonkeystatslogs())[102].set_version(2);
    SegmentLoadInfo replace_info(new_proto, schema_);
    auto replace_diff = current_info.ComputeDiff(replace_info);
    EXPECT_EQ(replace_diff.json_stats_to_replace.size(), 1);
    EXPECT_TRUE(replace_diff.json_stats_to_replace.count(FieldId(102)) > 0);

    proto::segcore::SegmentLoadInfo drop_proto;
    drop_proto.set_segmentid(100);
    drop_proto.set_collectionid(200);
    drop_proto.set_partitionid(300);
    drop_proto.set_num_of_rows(1000);
    SegmentLoadInfo drop_info(drop_proto, schema_);
    auto drop_diff = current_info.ComputeDiff(drop_info);
    EXPECT_EQ(drop_diff.json_stats_to_drop.size(), 1);
    EXPECT_TRUE(drop_diff.json_stats_to_drop.count(FieldId(102)) > 0);

    proto::segcore::SegmentLoadInfo invalid_proto;
    invalid_proto.set_segmentid(100);
    invalid_proto.set_collectionid(200);
    invalid_proto.set_partitionid(300);
    invalid_proto.set_num_of_rows(1000);
    auto& invalid_stats = (*invalid_proto.mutable_jsonkeystatslogs())[102];
    invalid_stats.CopyFrom(current_stats);
    invalid_stats.set_json_key_stats_data_format(current_format - 1);
    SegmentLoadInfo invalid_info(invalid_proto, schema_);
    auto invalid_diff = current_info.ComputeDiff(invalid_info);
    EXPECT_TRUE(invalid_diff.json_stats_to_load.empty());
    EXPECT_TRUE(invalid_diff.json_stats_to_replace.empty());
    EXPECT_EQ(invalid_diff.json_stats_to_drop.size(), 1);
    EXPECT_TRUE(invalid_diff.json_stats_to_drop.count(FieldId(102)) > 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffJsonStatsDisabled) {
    SetDefaultJSONKeyStatsEnable(false);
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_collectionid(200);
    current_proto.set_partitionid(300);
    current_proto.set_num_of_rows(1000);

    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    auto& json_stats = (*new_proto.mutable_jsonkeystatslogs())[102];
    json_stats.set_fieldid(102);
    json_stats.set_version(1);
    json_stats.set_buildid(5001);
    json_stats.set_json_key_stats_data_format(
        std::stoll(JSON_STATS_DATA_FORMAT_VERSION));
    json_stats.add_files("meta.json");

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.json_stats_to_load.empty());
    EXPECT_TRUE(diff.json_stats_to_replace.empty());
    EXPECT_TRUE(diff.json_stats_to_drop.empty());
    SetDefaultJSONKeyStatsEnable(true);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffTextIndexMultipleFields) {
    // Test with multiple text stats fields in new_info
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    // Current has text stats for field 101
    auto& current_stats = (*current_proto.mutable_textstatslogs())[101];
    current_stats.set_fieldid(101);
    current_stats.set_version(1);
    current_stats.add_files("/path/to/old_text_index");

    // New has text stats for fields 101 and 102
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_collectionid(200);
    new_proto.set_partitionid(300);
    new_proto.set_num_of_rows(1000);

    auto& new_stats1 = (*new_proto.mutable_textstatslogs())[101];
    new_stats1.set_fieldid(101);
    new_stats1.set_version(1);
    new_stats1.add_files("/path/to/old_text_index");

    auto& new_stats2 = (*new_proto.mutable_textstatslogs())[102];
    new_stats2.set_fieldid(102);
    new_stats2.set_version(1);
    new_stats2.set_buildid(5002);
    new_stats2.add_files("/path/to/new_text_index");

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    // Only field 102 should be in text_indexes_to_load (101 already exists)
    EXPECT_EQ(diff.text_indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.text_indexes_to_load.count(FieldId(102)) > 0);
    EXPECT_TRUE(diff.text_indexes_to_load.count(FieldId(101)) == 0);
}

// ==================== Replace Tests ====================
// Tests for indexes_to_replace, binlogs_to_replace, column_groups_to_replace

TEST_F(SegmentLoadInfoTest, ComputeDiffIndexReplace) {
    // When current has index for field 101 (index_id=1001) and new has
    // index for field 101 with different index_id (2001), it should go
    // to indexes_to_replace (not indexes_to_load)

    // Current: field 101 has index 1001
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_index = current_proto.add_index_infos();
    cur_index->set_fieldid(101);
    cur_index->set_indexid(1001);
    cur_index->add_index_file_paths("/path/to/old_index");
    auto* cur_param = cur_index->add_index_params();
    cur_param->set_key("index_type");
    cur_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    // New: field 101 has different index 2001
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_index = new_proto.add_index_infos();
    new_index->set_fieldid(101);
    new_index->set_indexid(2001);
    new_index->add_index_file_paths("/path/to/new_index");
    auto* new_param = new_index->add_index_params();
    new_param->set_key("index_type");
    new_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Index replace diff: " << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Field 101 should be in indexes_to_replace (already had index)
    EXPECT_EQ(diff.indexes_to_replace.size(), 1);
    EXPECT_TRUE(diff.indexes_to_replace.count(FieldId(101)) > 0);
    EXPECT_EQ(diff.indexes_to_replace[FieldId(101)].size(), 1);
    EXPECT_EQ(diff.indexes_to_replace[FieldId(101)][0].index_id, 2001);

    // Not in indexes_to_load
    EXPECT_TRUE(diff.indexes_to_load.empty());
    // Old index should be dropped
    EXPECT_EQ(diff.indexes_to_drop.size(), 1);
    EXPECT_TRUE(diff.indexes_to_drop.count(FieldId(101)) > 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffIndexReplaceMixed) {
    // Mixed scenario: some indexes replaced, some new, some unchanged

    // Current: field 101 index_id=1001, field 102 index_id=1002
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    auto* cur_idx1 = current_proto.add_index_infos();
    cur_idx1->set_fieldid(101);
    cur_idx1->set_indexid(1001);
    cur_idx1->add_index_file_paths("/path/to/idx1");
    auto* cur_param1 = cur_idx1->add_index_params();
    cur_param1->set_key("index_type");
    cur_param1->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    auto* cur_idx2 = current_proto.add_index_infos();
    cur_idx2->set_fieldid(102);
    cur_idx2->set_indexid(1002);
    cur_idx2->add_index_file_paths("/path/to/idx2");
    auto* cur_param2 = cur_idx2->add_index_params();
    cur_param2->set_key("index_type");
    cur_param2->set_value(milvus::index::INVERTED_INDEX_TYPE);

    // New: field 101 index_id=2001 (replace), field 102 index_id=1002 (same),
    //      field 103 index_id=3001 (new)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);

    auto* new_idx1 = new_proto.add_index_infos();
    new_idx1->set_fieldid(101);
    new_idx1->set_indexid(2001);
    new_idx1->add_index_file_paths("/path/to/new_idx1");
    auto* new_param1 = new_idx1->add_index_params();
    new_param1->set_key("index_type");
    new_param1->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    auto* new_idx2 = new_proto.add_index_infos();
    new_idx2->set_fieldid(102);
    new_idx2->set_indexid(1002);
    new_idx2->add_index_file_paths("/path/to/idx2");
    auto* new_param2 = new_idx2->add_index_params();
    new_param2->set_key("index_type");
    new_param2->set_value(milvus::index::INVERTED_INDEX_TYPE);

    auto* new_idx3 = new_proto.add_index_infos();
    new_idx3->set_fieldid(103);
    new_idx3->set_indexid(3001);
    new_idx3->add_index_file_paths("/path/to/idx3");
    auto* new_param3 = new_idx3->add_index_params();
    new_param3->set_key("index_type");
    new_param3->set_value(milvus::index::INVERTED_INDEX_TYPE);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Index replace mixed diff: " << diff.ToString() << "\n";

    // Field 101: old index_id=1001 dropped, new index_id=2001 in replace
    EXPECT_EQ(diff.indexes_to_replace.size(), 1);
    EXPECT_TRUE(diff.indexes_to_replace.count(FieldId(101)) > 0);
    EXPECT_EQ(diff.indexes_to_replace[FieldId(101)][0].index_id, 2001);

    // Field 103: new index, should be in indexes_to_load
    EXPECT_EQ(diff.indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(103)) > 0);

    // Field 102: unchanged (same index_id), should not appear
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(102)) == 0);
    EXPECT_TRUE(diff.indexes_to_replace.count(FieldId(102)) == 0);

    // Old index_id=1001 for field 101 should be dropped
    EXPECT_EQ(diff.indexes_to_drop.size(), 1);
    EXPECT_TRUE(diff.indexes_to_drop.count(FieldId(101)) > 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffIndexReplaceDropConflict) {
    // Verify that when a field has both indexes_to_replace and indexes_to_drop,
    // the drop should be skipped during ApplyLoadDiff to avoid dropping the
    // just-loaded replacement index. ComputeDiff correctly produces both entries;
    // ApplyLoadDiff filters the drop.

    // Current: field 101 has index 1001
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_index = current_proto.add_index_infos();
    cur_index->set_fieldid(101);
    cur_index->set_indexid(1001);
    cur_index->add_index_file_paths("/path/to/old_index");
    auto* cur_param = cur_index->add_index_params();
    cur_param->set_key("index_type");
    cur_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    // New: field 101 has different index 2001 (replacement)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_index = new_proto.add_index_infos();
    new_index->set_fieldid(101);
    new_index->set_indexid(2001);
    new_index->add_index_file_paths("/path/to/new_index");
    auto* new_param = new_index->add_index_params();
    new_param->set_key("index_type");
    new_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    // ComputeDiff should produce BOTH replace and drop for field 101
    EXPECT_TRUE(diff.indexes_to_replace.count(FieldId(101)) > 0);
    EXPECT_TRUE(diff.indexes_to_drop.count(FieldId(101)) > 0);

    // Verify the replacement has the correct new index_id
    EXPECT_EQ(diff.indexes_to_replace[FieldId(101)][0].index_id, 2001);

    // Simulate the ApplyLoadDiff filter: fields with replace should be skipped
    // in the drop loop
    std::vector<FieldId> actually_dropped;
    for (auto field_id : diff.indexes_to_drop) {
        if (diff.indexes_to_replace.count(field_id) > 0 ||
            diff.indexes_to_load.count(field_id) > 0) {
            continue;
        }
        actually_dropped.push_back(field_id);
    }
    // Field 101 has a replacement, so it should NOT be dropped
    EXPECT_TRUE(actually_dropped.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDropsJsonIndexByNestedPath) {
    auto add_json_index = [](proto::segcore::SegmentLoadInfo& proto,
                             int64_t index_id,
                             const std::string& nested_path) {
        auto* index = proto.add_index_infos();
        index->set_fieldid(102);
        index->set_indexid(index_id);
        index->add_index_file_paths("/path/to/json_index_" +
                                    std::to_string(index_id));
        auto* type = index->add_index_params();
        type->set_key(milvus::index::INDEX_TYPE);
        type->set_value(milvus::index::INVERTED_INDEX_TYPE);
        auto* path = index->add_index_params();
        path->set_key(JSON_PATH);
        path->set_value(nested_path);
        auto* cast = index->add_index_params();
        cast->set_key(JSON_CAST_TYPE);
        cast->set_value("DOUBLE");
    };

    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    current_proto.set_manifest_path("/path/to/manifest");
    add_json_index(current_proto, 5001, "a");
    add_json_index(current_proto, 5002, "b");

    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    new_proto.set_manifest_path("/path/to/manifest");
    add_json_index(new_proto, 5002, "b");

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    current_info.SetColumnGroupsForTesting(
        std::make_shared<milvus_storage::api::ColumnGroups>());
    new_info.SetColumnGroupsForTesting(
        std::make_shared<milvus_storage::api::ColumnGroups>());
    current_info.CompactRuntimeInfoForManifest();
    EXPECT_EQ(current_info.GetIndexInfoCount(), 0);
    auto diff = current_info.ComputeDiff(new_info);

    ASSERT_EQ(diff.json_indexes_to_drop.count(FieldId(102)), 1);
    EXPECT_EQ(diff.json_indexes_to_drop.at(FieldId(102)).count("a"), 1);
    EXPECT_EQ(diff.json_indexes_to_drop.at(FieldId(102)).count("b"), 0);
    EXPECT_EQ(diff.indexes_to_drop.count(FieldId(102)), 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffBinlogReplace) {
    // When a field's binlog group changes, it goes to binlogs_to_replace

    // Current: field 101 in group 101 (legacy format)
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(101);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_old");
    cur_log->set_entries_num(500);

    // New: field 101 in a column group 200 with children 101, 102
    // Field 101 already exists -> replace; field 102 is new -> load
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(200);
    new_binlog->add_child_fields(101);
    new_binlog->add_child_fields(102);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/group_binlog");
    new_log->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Binlog replace diff: " << diff.ToString() << "\n";

    EXPECT_TRUE(diff.HasChanges());
    // Field 102 is new -> binlogs_to_load
    EXPECT_EQ(diff.binlogs_to_load.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(), 102);

    // Field 101 already existed -> binlogs_to_replace
    EXPECT_EQ(diff.binlogs_to_replace.size(), 1);
    EXPECT_EQ(diff.binlogs_to_replace[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_replace[0].first[0].get(), 101);

    // Nothing dropped (field 101 still present, just in different group)
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffBinlogNoReplace) {
    // When binlogs don't change, no replacement should occur

    // Current and new: same field 101 in same group (legacy)
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(101);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/binlog");
    log->set_entries_num(500);

    SegmentLoadInfo current_info(proto, schema_);
    auto first_diff = current_info.GetLoadDiff();
    for (const auto& field_id : first_diff.fields_to_fill_default) {
        current_info.SetFieldFilledWithDefault(field_id);
    }

    SegmentLoadInfo new_info(proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_FALSE(diff.HasChanges());
    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
    EXPECT_TRUE(diff.indexes_to_load.empty());
    EXPECT_TRUE(diff.indexes_to_replace.empty());
}

TEST_F(SegmentLoadInfoTest, LoadDiffHasChangesWithReplace) {
    // Test that HasChanges returns true for each replace field
    {
        LoadDiff diff;
        EXPECT_FALSE(diff.HasChanges());
        diff.indexes_to_replace[FieldId(101)].push_back(LoadIndexInfo{});
        EXPECT_TRUE(diff.HasChanges());
    }
    {
        LoadDiff diff;
        diff.binlogs_to_replace.emplace_back(std::vector<FieldId>{FieldId(101)},
                                             proto::segcore::FieldBinlog{});
        EXPECT_TRUE(diff.HasChanges());
    }
    {
        LoadDiff diff;
        diff.column_groups_to_replace.emplace_back(
            0, std::vector<FieldId>{FieldId(101)});
        EXPECT_TRUE(diff.HasChanges());
    }
    {
        LoadDiff diff;
        diff.column_groups_to_lazyreplace.emplace_back(
            0, std::vector<FieldId>{FieldId(101)});
        EXPECT_TRUE(diff.HasChanges());
    }
}

TEST_F(SegmentLoadInfoTest, LoadDiffToStringIncludesReplace) {
    // Test that ToString includes all replace fields
    LoadDiff diff;
    diff.indexes_to_replace[FieldId(101)].push_back(LoadIndexInfo{});
    diff.binlogs_to_replace.emplace_back(std::vector<FieldId>{FieldId(102)},
                                         proto::segcore::FieldBinlog{});
    diff.column_groups_to_replace.emplace_back(
        0, std::vector<FieldId>{FieldId(103)});
    diff.column_groups_to_lazyreplace.emplace_back(
        1, std::vector<FieldId>{FieldId(104)});

    std::string str = diff.ToString();
    EXPECT_TRUE(str.find("indexes_to_replace") != std::string::npos);
    EXPECT_TRUE(str.find("binlogs_to_replace") != std::string::npos);
    EXPECT_TRUE(str.find("column_groups_to_replace") != std::string::npos);
    EXPECT_TRUE(str.find("column_groups_to_lazyreplace") != std::string::npos);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFilledFieldBecomesReplace) {
    // When a field was previously filled with default values (no binlog)
    // and the new LoadInfo provides actual binlog data for it,
    // it should go to binlogs_to_replace (not binlogs_to_load)
    // because the segment already has data loaded for it.

    // Step 1: Initial load with only pk field (100).
    // Fields 101-110 will be filled with default values.
    proto::segcore::SegmentLoadInfo initial_proto;
    initial_proto.set_segmentid(100);
    initial_proto.set_num_of_rows(1000);
    auto* binlog1 = initial_proto.add_binlog_paths();
    binlog1->set_fieldid(100);
    auto* log1 = binlog1->add_binlogs();
    log1->set_log_path("/path/to/pk_binlog");
    log1->set_entries_num(1000);

    SegmentLoadInfo current_info(initial_proto, schema_);
    auto initial_diff = current_info.GetLoadDiff();
    EXPECT_FALSE(initial_diff.fields_to_fill_default.empty());
    for (const auto& field_id : initial_diff.fields_to_fill_default) {
        current_info.SetFieldFilledWithDefault(field_id);
    }

    // Step 2: New LoadInfo adds binlog for field 101 (was default-filled)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog1 = new_proto.add_binlog_paths();
    new_binlog1->set_fieldid(100);
    auto* new_log1 = new_binlog1->add_binlogs();
    new_log1->set_log_path("/path/to/pk_binlog");
    new_log1->set_entries_num(1000);

    auto* new_binlog2 = new_proto.add_binlog_paths();
    new_binlog2->set_fieldid(101);
    auto* new_log2 = new_binlog2->add_binlogs();
    new_log2->set_log_path("/path/to/vec_binlog");
    new_log2->set_entries_num(1000);

    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Default-filled replace diff: " << diff.ToString() << "\n";

    // Field 101 was default-filled → should be in binlogs_to_replace
    EXPECT_FALSE(diff.binlogs_to_replace.empty());
    bool found_101_in_replace = false;
    for (const auto& [field_ids, binlog] : diff.binlogs_to_replace) {
        for (const auto& fid : field_ids) {
            if (fid.get() == 101) {
                found_101_in_replace = true;
            }
        }
    }
    EXPECT_TRUE(found_101_in_replace);

    // Field 101 should NOT be in binlogs_to_load
    bool found_101_in_load = false;
    for (const auto& [field_ids, binlog] : diff.binlogs_to_load) {
        for (const auto& fid : field_ids) {
            if (fid.get() == 101) {
                found_101_in_load = true;
            }
        }
    }
    EXPECT_FALSE(found_101_in_load);

    // Field 101 should NOT be in fields_to_fill_default (has data source now)
    for (const auto& fid : diff.fields_to_fill_default) {
        EXPECT_NE(fid.get(), 101);
    }
}

// ==================== Raw Data Lifecycle Tests ====================
// Tests for field_data_to_drop generation and ComputeDiffReloadFields behavior

TEST_F(SegmentLoadInfoTest, ComputeDiffNewIndexWithRawDataNoFieldDrop) {
    // Test 1.1: When a new index has raw data (ASCENDING_SORT for scalar),
    // field_data_to_drop should NOT include that field.
    // Raw-data indexes no longer trigger field drop; field data is kept
    // (binlog mode) or lazified (column groups mode).
    // Scenario: current has no index -> new has ASCENDING_SORT index (raw=true)

    // Current: only binlog for field 108 (INT64, extra_field1)
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(108);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_108");
    cur_log->set_entries_num(1000);

    // New: same binlog + ASCENDING_SORT index for field 108 (has raw data)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(108);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/binlog_108");
    new_log->set_entries_num(1000);

    auto* index_info = new_proto.add_index_infos();
    index_info->set_fieldid(108);
    index_info->set_indexid(5001);
    index_info->add_index_file_paths("/path/to/sort_index");
    auto* param = index_info->add_index_params();
    param->set_key("index_type");
    param->set_value(milvus::index::ASCENDING_SORT);

    SegmentLoadInfo current_info(current_proto, schema_);
    // Initialize default fields
    (void)current_info.GetLoadDiff();
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "New index with raw data diff: " << diff.ToString() << "\n";

    // Index should be in indexes_to_load
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(108)) > 0);
    // field_data_to_drop should NOT include field 108
    // (raw-data index no longer triggers drop)
    EXPECT_TRUE(diff.field_data_to_drop.count(FieldId(108)) == 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffReplaceIndexWithRawDataNoFieldDrop) {
    // Test 1.2: When replacing an index and the new one has raw data,
    // field_data_to_drop should NOT include that field.
    // Raw-data indexes no longer trigger field drop.
    // Scenario: current has INVERTED index -> new has ASCENDING_SORT index

    // Current: field 108 has INVERTED index (no raw data)
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(108);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_108");
    cur_log->set_entries_num(1000);
    auto* cur_index = current_proto.add_index_infos();
    cur_index->set_fieldid(108);
    cur_index->set_indexid(5001);
    cur_index->add_index_file_paths("/path/to/inverted_index");
    auto* cur_param = cur_index->add_index_params();
    cur_param->set_key("index_type");
    cur_param->set_value(milvus::index::INVERTED_INDEX_TYPE);

    // New: field 108 has ASCENDING_SORT index (has raw data), different index_id
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(108);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/binlog_108");
    new_log->set_entries_num(1000);
    auto* new_index = new_proto.add_index_infos();
    new_index->set_fieldid(108);
    new_index->set_indexid(5002);
    new_index->add_index_file_paths("/path/to/sort_index");
    auto* new_param = new_index->add_index_params();
    new_param->set_key("index_type");
    new_param->set_value(milvus::index::ASCENDING_SORT);

    SegmentLoadInfo current_info(current_proto, schema_);
    (void)current_info.GetLoadDiff();
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Replace index with raw data diff: " << diff.ToString()
              << "\n";

    // Index should be in indexes_to_replace
    EXPECT_TRUE(diff.indexes_to_replace.count(FieldId(108)) > 0);
    // field_data_to_drop should NOT include field 108
    // (raw-data index no longer triggers drop)
    EXPECT_TRUE(diff.field_data_to_drop.count(FieldId(108)) == 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffNewIndexNoRawDataNoFieldDrop) {
    // Test 1.3: When a new index has no raw data (INVERTED_INDEX_TYPE),
    // field_data_to_drop should NOT include that field.

    // Current: only binlog for field 108
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(108);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_108");
    cur_log->set_entries_num(1000);

    // New: same binlog + INVERTED index for field 108 (no raw data)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(108);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/binlog_108");
    new_log->set_entries_num(1000);
    auto* index_info = new_proto.add_index_infos();
    index_info->set_fieldid(108);
    index_info->set_indexid(5001);
    index_info->add_index_file_paths("/path/to/inverted_index");
    auto* param = index_info->add_index_params();
    param->set_key("index_type");
    param->set_value(milvus::index::INVERTED_INDEX_TYPE);

    SegmentLoadInfo current_info(current_proto, schema_);
    (void)current_info.GetLoadDiff();
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "New index no raw data diff: " << diff.ToString() << "\n";

    // Index should be in indexes_to_load
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(108)) > 0);
    // field_data_to_drop should NOT include field 108 (no raw data)
    EXPECT_TRUE(diff.field_data_to_drop.count(FieldId(108)) == 0);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffUnchangedIndexNoFieldDrop) {
    // Test 1.4: When the index is unchanged (same index_id),
    // field_data_to_drop should be empty.

    // Both current and new: field 108 has ASCENDING_SORT index with same id
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(108);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/binlog_108");
    log->set_entries_num(1000);
    auto* index_info = proto.add_index_infos();
    index_info->set_fieldid(108);
    index_info->set_indexid(5001);
    index_info->add_index_file_paths("/path/to/sort_index");
    auto* param = index_info->add_index_params();
    param->set_key("index_type");
    param->set_value(milvus::index::ASCENDING_SORT);

    SegmentLoadInfo current_info(proto, schema_);
    (void)current_info.GetLoadDiff();
    SegmentLoadInfo new_info(proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    // No index changes
    EXPECT_TRUE(diff.indexes_to_load.empty());
    EXPECT_TRUE(diff.indexes_to_replace.empty());
    EXPECT_TRUE(diff.indexes_to_drop.empty());
    // field_data_to_drop should be empty (no change)
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffReloadFieldRawToNoRaw) {
    // Test 1.5: When index changes from raw=true to raw=false,
    // binlogs_to_replace should include the field's binlog.
    // Scenario: current has ASCENDING_SORT (raw=true) -> new has INVERTED (raw=false)

    // Current: field 108 with ASCENDING_SORT index + binlog
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(108);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_108");
    cur_log->set_entries_num(1000);
    auto* cur_index = current_proto.add_index_infos();
    cur_index->set_fieldid(108);
    cur_index->set_indexid(5001);
    cur_index->add_index_file_paths("/path/to/sort_index");
    auto* cur_param = cur_index->add_index_params();
    cur_param->set_key("index_type");
    cur_param->set_value(milvus::index::ASCENDING_SORT);

    // New: field 108 with INVERTED index (no raw) + same binlog
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(108);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/binlog_108");
    new_log->set_entries_num(1000);
    auto* new_index = new_proto.add_index_infos();
    new_index->set_fieldid(108);
    new_index->set_indexid(5002);
    new_index->add_index_file_paths("/path/to/inverted_index");
    auto* new_param = new_index->add_index_params();
    new_param->set_key("index_type");
    new_param->set_value(milvus::index::INVERTED_INDEX_TYPE);

    SegmentLoadInfo current_info(current_proto, schema_);
    (void)current_info.GetLoadDiff();
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Raw->NoRaw reload diff: " << diff.ToString() << "\n";

    // Index should be replaced
    EXPECT_TRUE(diff.indexes_to_replace.count(FieldId(108)) > 0);

    // binlogs_to_replace should include field 108 (need to reload raw data)
    bool found_108_in_replace = false;
    for (const auto& [field_ids, binlog] : diff.binlogs_to_replace) {
        for (const auto& fid : field_ids) {
            if (fid.get() == 108) {
                found_108_in_replace = true;
            }
        }
    }
    EXPECT_TRUE(found_108_in_replace);

    // fields_to_reload should be empty (new path uses binlogs_to_replace)
    EXPECT_TRUE(diff.fields_to_reload.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffReloadFieldIndexDropped) {
    // Test 1.6: When an index is dropped and it had raw data,
    // binlogs_to_replace should include the field's binlog.
    // Scenario: current has ASCENDING_SORT (raw=true) -> new has no index

    // Current: field 108 with ASCENDING_SORT index + binlog
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(108);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_108");
    cur_log->set_entries_num(1000);
    auto* cur_index = current_proto.add_index_infos();
    cur_index->set_fieldid(108);
    cur_index->set_indexid(5001);
    cur_index->add_index_file_paths("/path/to/sort_index");
    auto* cur_param = cur_index->add_index_params();
    cur_param->set_key("index_type");
    cur_param->set_value(milvus::index::ASCENDING_SORT);

    // New: field 108 with binlog only (no index)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(108);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/binlog_108");
    new_log->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);
    (void)current_info.GetLoadDiff();
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);
    std::cout << "Index dropped reload diff: " << diff.ToString() << "\n";

    // Index should be dropped
    EXPECT_TRUE(diff.indexes_to_drop.count(FieldId(108)) > 0);

    // binlogs_to_replace should include field 108
    bool found_108_in_replace = false;
    for (const auto& [field_ids, binlog] : diff.binlogs_to_replace) {
        for (const auto& fid : field_ids) {
            if (fid.get() == 108) {
                found_108_in_replace = true;
            }
        }
    }
    EXPECT_TRUE(found_108_in_replace);
}

TEST_F(SegmentLoadInfoTest, ComputeDiffNoReloadWhenIndexStaysRaw) {
    // Test 1.7: When index stays raw=true (same index_id),
    // no reload should happen.

    // Both: field 108 with same ASCENDING_SORT index
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(108);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/binlog_108");
    log->set_entries_num(1000);
    auto* index_info = proto.add_index_infos();
    index_info->set_fieldid(108);
    index_info->set_indexid(5001);
    index_info->add_index_file_paths("/path/to/sort_index");
    auto* param = index_info->add_index_params();
    param->set_key("index_type");
    param->set_value(milvus::index::ASCENDING_SORT);

    SegmentLoadInfo current_info(proto, schema_);
    auto initial_diff = current_info.GetLoadDiff();
    for (const auto& field_id : initial_diff.fields_to_fill_default) {
        current_info.SetFieldFilledWithDefault(field_id);
    }
    SegmentLoadInfo new_info(proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    // No changes at all
    EXPECT_FALSE(diff.HasChanges());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
    EXPECT_TRUE(diff.fields_to_reload.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

// -----------------------------------------------------------------------------
// Tests for same-group file-change detection in ComputeDiffBinlogs and
// ComputeDiffColumnGroups.
// When compaction rewrites the data under the same group id / column group
// index, the files list changes but the field layout does not. These tests
// verify that the diff emits replace entries so the loader can evict stale
// cached columns.
// -----------------------------------------------------------------------------

TEST_F(SegmentLoadInfoTest,
       ComputeDiffBinlogSameGroupDifferentFilesTriggersReplace) {
    // Current: legacy field 105 pointing at binlog A
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(105);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_A");
    cur_log->set_entries_num(1000);

    // New: same group id, same child field, but a different log_path
    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    new_proto.mutable_binlog_paths(0)->mutable_binlogs(0)->set_log_path(
        "/path/to/binlog_B");

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.HasChanges());
    ASSERT_EQ(diff.binlogs_to_replace.size(), 1);
    ASSERT_EQ(diff.binlogs_to_replace[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_replace[0].first[0].get(), 105);
    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffBinlogSameGroupDifferentFileCountTriggersReplace) {
    // Current: legacy field 105 with one binlog file
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(105);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_A");
    cur_log->set_entries_num(500);

    // New: same group id, same child, but an additional binlog file (post
    // compaction merging more files into the same group).
    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    auto* extra_log = new_proto.mutable_binlog_paths(0)->add_binlogs();
    extra_log->set_log_path("/path/to/binlog_B");
    extra_log->set_entries_num(500);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    ASSERT_EQ(diff.binlogs_to_replace.size(), 1);
    ASSERT_EQ(diff.binlogs_to_replace[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_replace[0].first[0].get(), 105);
    EXPECT_TRUE(diff.binlogs_to_load.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffBinlogSameGroupIdenticalFilesNoDiff) {
    // Current and new have identical binlog contents under group 105.
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(105);
    auto* log1 = binlog->add_binlogs();
    log1->set_log_path("/path/to/binlog_A");
    log1->set_entries_num(500);
    auto* log2 = binlog->add_binlogs();
    log2->set_log_path("/path/to/binlog_B");
    log2->set_entries_num(500);

    SegmentLoadInfo current_info(proto, schema_);
    (void)current_info.GetLoadDiff();
    SegmentLoadInfo new_info(proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffBinlogMultiFieldGroupIdenticalFilesNoDiff) {
    // Multi-field (storage v2) binlog group: group id 200 carries children
    // 105 and 106 under the same log file. Cur and new are identical, so
    // the diff must be empty. Regression for a bug where the mapping
    // lookup used the group id (which is not a map key) and spuriously
    // routed both children into binlogs_to_replace.
    proto::segcore::SegmentLoadInfo proto;
    proto.set_segmentid(100);
    proto.set_num_of_rows(1000);
    auto* binlog = proto.add_binlog_paths();
    binlog->set_fieldid(200);
    binlog->add_child_fields(105);
    binlog->add_child_fields(106);
    auto* log1 = binlog->add_binlogs();
    log1->set_log_path("/path/to/group_binlog_A");
    log1->set_entries_num(1000);

    SegmentLoadInfo current_info(proto, schema_);
    SegmentLoadInfo new_info(proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffBinlogMultiFieldGroupDifferentFilesReplacesAllChildren) {
    // Same multi-field group shape on both sides but the underlying
    // binlog file path changes — every child must land in
    // binlogs_to_replace so the loader evicts the stale cached columns.
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(200);
    cur_binlog->add_child_fields(105);
    cur_binlog->add_child_fields(106);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/group_binlog_A");
    cur_log->set_entries_num(1000);

    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    new_proto.mutable_binlog_paths(0)->mutable_binlogs(0)->set_log_path(
        "/path/to/group_binlog_B");

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    ASSERT_EQ(diff.binlogs_to_replace.size(), 1);
    std::set<int64_t> replaced;
    for (const auto& fid : diff.binlogs_to_replace[0].first) {
        replaced.insert(fid.get());
    }
    EXPECT_EQ(replaced, (std::set<int64_t>{105, 106}));
    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffBinlogSameGroupFilesChangedWithNewChildSplitsReplaceAndLoad) {
    // Current: multi-field group 200 contains only child 105
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(200);
    cur_binlog->add_child_fields(105);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/group_binlog_A");
    cur_log->set_entries_num(1000);

    // New: same group id 200 but files changed AND child 106 is newly added.
    // Expect: 105 -> replace (already loaded, cache stale), 106 -> load (new).
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(200);
    new_binlog->add_child_fields(105);
    new_binlog->add_child_fields(106);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/group_binlog_B");
    new_log->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, schema_);
    auto diff = current_info.ComputeDiff(new_info);

    ASSERT_EQ(diff.binlogs_to_replace.size(), 1);
    ASSERT_EQ(diff.binlogs_to_replace[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_replace[0].first[0].get(), 105);

    ASSERT_EQ(diff.binlogs_to_load.size(), 1);
    ASSERT_EQ(diff.binlogs_to_load[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(), 106);

    EXPECT_TRUE(diff.field_data_to_drop.empty());
}

// ---- Column group same-index file-change detection ----

namespace {

// Build a ColumnGroups payload from a list of (fields, files) pairs so
// tests can exercise ComputeDiffColumnGroups without a real manifest.
std::shared_ptr<milvus_storage::api::ColumnGroups>
MakeColumnGroups(
    std::initializer_list<
        std::pair<std::vector<int64_t>, std::vector<std::string>>> groups) {
    auto cgs = std::make_shared<milvus_storage::api::ColumnGroups>();
    for (const auto& [field_ids, paths] : groups) {
        auto cg = std::make_shared<milvus_storage::api::ColumnGroup>();
        cg->format = "parquet";
        for (auto fid : field_ids) {
            cg->columns.push_back(std::to_string(fid));
        }
        for (const auto& p : paths) {
            milvus_storage::api::ColumnGroupFile f;
            f.path = p;
            f.start_index = 0;
            f.end_index = 0;
            cg->files.push_back(std::move(f));
        }
        cgs->push_back(std::move(cg));
    }
    return cgs;
}

std::shared_ptr<milvus_storage::api::ColumnGroups>
MakeExternalColumnGroups(
    std::initializer_list<
        std::pair<std::vector<std::string>, std::vector<std::string>>> groups) {
    auto cgs = std::make_shared<milvus_storage::api::ColumnGroups>();
    for (const auto& [columns, paths] : groups) {
        auto cg = std::make_shared<milvus_storage::api::ColumnGroup>();
        cg->format = "parquet";
        for (const auto& column : columns) {
            cg->columns.push_back(column);
        }
        for (const auto& p : paths) {
            milvus_storage::api::ColumnGroupFile f;
            f.path = p;
            f.start_index = 0;
            f.end_index = 0;
            cg->files.push_back(std::move(f));
        }
        cgs->push_back(std::move(cg));
    }
    return cgs;
}

proto::segcore::SegmentLoadInfo
MakeManifestProto(const std::string& manifest_path) {
    proto::segcore::SegmentLoadInfo p;
    p.set_segmentid(100);
    p.set_num_of_rows(1000);
    p.set_manifest_path(manifest_path);
    return p;
}

SchemaPtr
MakeSchemaWithFieldIds(std::initializer_list<int64_t> field_ids) {
    auto schema = std::make_shared<Schema>();
    for (auto field_id : field_ids) {
        schema->AddField(FieldName("field_" + std::to_string(field_id)),
                         FieldId(field_id),
                         DataType::INT64,
                         false,
                         std::nullopt);
    }
    if (std::find(field_ids.begin(), field_ids.end(), 100) != field_ids.end()) {
        schema->set_primary_field_id(FieldId(100));
    }
    return schema;
}

}  // namespace

TEST_F(SegmentLoadInfoTest, ConstructSkipsIndexInfoForDroppedField) {
    proto::segcore::SegmentLoadInfo p;
    p.set_segmentid(100);
    p.set_num_of_rows(1000);
    auto* index_info = p.add_index_infos();
    index_info->set_fieldid(102);
    index_info->set_indexid(1002);
    index_info->add_index_file_paths("/path/to/stale_index");
    auto* index_param = index_info->add_index_params();
    index_param->set_key("index_type");
    index_param->set_value(milvus::index::INVERTED_INDEX_TYPE);

    SegmentLoadInfo info(p, MakeSchemaWithFieldIds({100, 101}));

    EXPECT_FALSE(info.HasIndexInfo(FieldId(102)));
    EXPECT_TRUE(info.GetIndexedFieldIds().empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffSkipsNewBinlogForDroppedField) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* stale_binlog = new_proto.add_binlog_paths();
    stale_binlog->set_fieldid(102);
    auto* log = stale_binlog->add_binlogs();
    log->set_log_path("/path/to/stale_binlog");
    log->set_entries_num(1000);

    auto latest_schema = MakeSchemaWithFieldIds({100, 101});
    SegmentLoadInfo current_info(current_proto, latest_schema);
    SegmentLoadInfo new_info(new_proto, latest_schema);
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.binlogs_to_load.empty());
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffFiltersDroppedChildFromMixedBinlogGroup) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* mixed_binlog = new_proto.add_binlog_paths();
    mixed_binlog->set_fieldid(104);
    mixed_binlog->add_child_fields(105);
    mixed_binlog->add_child_fields(106);
    auto* log = mixed_binlog->add_binlogs();
    log->set_log_path("/path/to/mixed_binlog");
    log->set_entries_num(1000);

    auto latest_schema = MakeSchemaWithFieldIds({100, 105});
    SegmentLoadInfo current_info(current_proto, latest_schema);
    SegmentLoadInfo new_info(new_proto, latest_schema);
    auto diff = current_info.ComputeDiff(new_info);

    ASSERT_EQ(diff.binlogs_to_load.size(), 1);
    ASSERT_EQ(diff.binlogs_to_load[0].first.size(), 1);
    EXPECT_EQ(diff.binlogs_to_load[0].first[0].get(), 105);
    EXPECT_TRUE(diff.binlogs_to_replace.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffSkipsNewManifestColumnForDroppedField) {
    auto latest_schema = MakeSchemaWithFieldIds({100, 101});
    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"),
                                 latest_schema);
    current_info.SetColumnGroupsForTesting(MakeColumnGroups({}));
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/new"), latest_schema);
    new_info.SetColumnGroupsForTesting(
        MakeColumnGroups({{{102}, {"/path/to/stale.parquet"}}}));

    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.column_groups_to_load.empty());
    EXPECT_TRUE(diff.column_groups_to_replace.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyload.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyreplace.empty());
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDropsManifestFieldRemovedFromSchema) {
    auto current_schema = MakeSchemaWithFieldIds({100, 102});
    auto latest_schema = MakeSchemaWithFieldIds({100});
    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"),
                                 current_schema);
    current_info.SetColumnGroupsForTesting(
        MakeColumnGroups({{{102}, {"/path/to/old.parquet"}}}));
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/new"), latest_schema);
    new_info.SetColumnGroupsForTesting(
        MakeColumnGroups({{{102}, {"/path/to/stale.parquet"}}}));

    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.field_data_to_drop.count(FieldId(102)) > 0);
    EXPECT_TRUE(diff.column_groups_to_load.empty());
    EXPECT_TRUE(diff.column_groups_to_replace.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyload.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyreplace.empty());
}

TEST_F(SegmentLoadInfoTest, SchemaReopenLoadsNewFieldWithBinlog) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* current_binlog = current_proto.add_binlog_paths();
    current_binlog->set_fieldid(100);
    auto* current_log = current_binlog->add_binlogs();
    current_log->set_log_path("/path/to/pk");
    current_log->set_entries_num(1000);

    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    auto* field_binlog = new_proto.add_binlog_paths();
    field_binlog->set_fieldid(101);
    auto* field_log = field_binlog->add_binlogs();
    field_log->set_log_path("/path/to/field101");
    field_log->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, MakeSchemaWithFieldIds({100}));
    SegmentLoadInfo new_info(new_proto, MakeSchemaWithFieldIds({100, 101}));
    auto diff = current_info.ComputeDiff(new_info);

    bool found_load = false;
    for (const auto& [field_ids, binlog] : diff.binlogs_to_load) {
        (void)binlog;
        for (auto field_id : field_ids) {
            found_load = found_load || field_id == FieldId(101);
        }
    }
    EXPECT_TRUE(found_load);
    for (auto field_id : diff.fields_to_fill_default) {
        EXPECT_NE(field_id, FieldId(101));
    }
}

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFieldsUsesNewSchema) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* current_binlog = current_proto.add_binlog_paths();
    current_binlog->set_fieldid(100);
    auto* current_log = current_binlog->add_binlogs();
    current_log->set_log_path("/path/to/pk");
    current_log->set_entries_num(1000);

    proto::segcore::SegmentLoadInfo new_proto = current_proto;

    SegmentLoadInfo current_info(current_proto, MakeSchemaWithFieldIds({100}));
    SegmentLoadInfo new_info(new_proto, MakeSchemaWithFieldIds({100, 101}));
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_EQ(diff.fields_to_fill_default.size(), 1);
    EXPECT_EQ(diff.fields_to_fill_default[0].get(), 101);
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffDropsResourcesForFieldRemovedFromSchema) {
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* index_info = current_proto.add_index_infos();
    index_info->set_fieldid(102);
    index_info->set_indexid(1002);
    index_info->add_index_file_paths("/path/to/stale_index");
    auto* index_param = index_info->add_index_params();
    index_param->set_key("index_type");
    index_param->set_value(milvus::index::INVERTED_INDEX_TYPE);
    auto* binlog = current_proto.add_binlog_paths();
    binlog->set_fieldid(102);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/stale_binlog");
    log->set_entries_num(1000);

    proto::segcore::SegmentLoadInfo new_proto = current_proto;
    SegmentLoadInfo current_info(current_proto,
                                 MakeSchemaWithFieldIds({100, 102}));
    SegmentLoadInfo new_info(new_proto, MakeSchemaWithFieldIds({100}));
    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.indexes_to_drop.count(FieldId(102)) > 0);
    EXPECT_TRUE(diff.field_data_to_drop.count(FieldId(102)) > 0);
    EXPECT_TRUE(diff.indexes_to_load.empty());
    EXPECT_TRUE(diff.binlogs_to_load.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffExternalManifestWithNewIndexDoesNotParseColumnNames) {
    schema_->set_external_source("s3://external-bucket/table");

    auto cgs = MakeExternalColumnGroups(
        {{{"id"}, {"/external/id.parquet"}},
         {{"vector", "tenant"}, {"/external/data.parquet"}}});

    auto current_proto = MakeManifestProto("/manifest/v1");
    SegmentLoadInfo current_info(current_proto, schema_);
    current_info.SetColumnGroupsForTesting(cgs);

    auto new_proto = MakeManifestProto("/manifest/v1");
    auto* index_info = new_proto.add_index_infos();
    index_info->set_fieldid(101);
    index_info->set_indexid(2001);
    index_info->set_buildid(3001);
    index_info->set_index_version(1);
    index_info->add_index_file_paths("/path/to/index");
    auto* index_param = index_info->add_index_params();
    index_param->set_key("index_type");
    index_param->set_value(knowhere::IndexEnum::INDEX_FAISS_IVFSQ8);

    SegmentLoadInfo new_info(new_proto, schema_);
    new_info.SetColumnGroupsForTesting(cgs);

    LoadDiff diff;
    EXPECT_NO_THROW({ diff = current_info.ComputeDiff(new_info); });
    ASSERT_EQ(diff.indexes_to_load.size(), 1);
    EXPECT_TRUE(diff.indexes_to_load.count(FieldId(101)) > 0);
    EXPECT_FALSE(diff.load_external_manifest);
    EXPECT_TRUE(diff.column_groups_to_load.empty());
    EXPECT_TRUE(diff.column_groups_to_replace.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyload.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyreplace.empty());
}

TEST_F(SegmentLoadInfoTest, HasManifestColumnUsesManifestColumnNames) {
    schema_->set_external_source("s3://external-bucket/table");

    SegmentLoadInfo info(MakeManifestProto("/manifest/v1"), schema_);
    info.SetColumnGroupsForTesting(MakeExternalColumnGroups(
        {{{"id"}, {"/external/id.parquet"}},
         {{"vector", "tenant"}, {"/external/data.parquet"}}}));

    EXPECT_TRUE(info.HasManifestColumn("id"));
    EXPECT_TRUE(info.HasManifestColumn("vector"));
    EXPECT_TRUE(info.HasManifestColumn("tenant"));
    EXPECT_FALSE(info.HasManifestColumn("missing"));
    EXPECT_FALSE(info.HasManifestColumn("101"));
}

TEST_F(SegmentLoadInfoTest, CachedManifestColumnsCanBeInherited) {
    schema_->set_external_source("s3://external-bucket/table");

    SegmentLoadInfo current(MakeManifestProto("/manifest/v1"), schema_);
    current.SetColumnGroupsForTesting(MakeExternalColumnGroups(
        {{{"id"}, {"/external/id.parquet"}},
         {{"vector", "tenant"}, {"/external/data.parquet"}}}));

    SegmentLoadInfo next(current.GetProto(), schema_);
    next.InheritCachedColumnGroupsFrom(current);

    EXPECT_TRUE(next.HasManifestColumn("id"));
    EXPECT_TRUE(next.HasManifestColumn("vector"));
    EXPECT_TRUE(next.HasManifestColumn("tenant"));
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffExternalManifestUpdateReloadsExternalManifest) {
    schema_->set_external_source("s3://external-bucket/table");

    auto current_cgs = MakeExternalColumnGroups(
        {{{"id"}, {"/external/id_v1.parquet"}},
         {{"vector", "tenant"}, {"/external/data_v1.parquet"}}});
    auto new_cgs = MakeExternalColumnGroups(
        {{{"id"}, {"/external/id_v2.parquet"}},
         {{"vector", "tenant"}, {"/external/data_v2.parquet"}}});

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/v1"), schema_);
    current_info.SetColumnGroupsForTesting(current_cgs);
    current_info.SetFieldFilledWithDefault(FieldId(101));
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/v2"), schema_);
    new_info.SetColumnGroupsForTesting(new_cgs);

    LoadDiff diff;
    EXPECT_NO_THROW({ diff = current_info.ComputeDiff(new_info); });
    EXPECT_TRUE(diff.manifest_updated);
    EXPECT_EQ(diff.new_manifest_path, "/manifest/v2");
    EXPECT_TRUE(diff.load_external_manifest);
    EXPECT_TRUE(diff.column_groups_to_load.empty());
    EXPECT_TRUE(diff.column_groups_to_replace.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyload.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyreplace.empty());

    std::set<FieldId> default_fields;
    EXPECT_NO_THROW({
        default_fields =
            current_info.GetDefaultFilledFieldsForNewInfo(new_info);
    });
    EXPECT_TRUE(default_fields.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffColumnGroupSameIndexDifferentFilesTriggersReplace) {
    // Two column groups with identical shape {105,106} at index 0, but the
    // new manifest points at different parquet files. The pk (100) is in
    // its own group at index 1 so it stays untouched.
    auto current_cgs =
        MakeColumnGroups({{{100}, {"/pk/file_A.parquet"}},
                          {{105, 106}, {"/user/file_A.parquet"}}});
    auto new_cgs = MakeColumnGroups({{{100}, {"/pk/file_A.parquet"}},
                                     {{105, 106}, {"/user/file_B.parquet"}}});

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"), schema_);
    current_info.SetColumnGroupsForTesting(current_cgs);
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/new"), schema_);
    new_info.SetColumnGroupsForTesting(new_cgs);

    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_TRUE(diff.manifest_updated);
    // User fields (105, 106) at the unchanged index 1 must be replaced
    // since the underlying parquet file changed.
    bool saw_user_group = false;
    for (const auto& [idx, fields] : diff.column_groups_to_replace) {
        if (idx == 1) {
            saw_user_group = true;
            std::set<int64_t> ids;
            for (auto f : fields) {
                ids.insert(f.get());
            }
            EXPECT_EQ(ids, (std::set<int64_t>{105, 106}));
        }
    }
    EXPECT_TRUE(saw_user_group);
    // PK group (index 0) had identical files; no replace entry for it.
    for (const auto& [idx, fields] : diff.column_groups_to_replace) {
        EXPECT_NE(idx, 0);
    }
    EXPECT_TRUE(diff.column_groups_to_load.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffColumnGroupSameIndexIdenticalFilesNoReplace) {
    auto cgs = MakeColumnGroups(
        {{{100}, {"/pk/file.parquet"}}, {{105, 106}, {"/user/file.parquet"}}});

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/v1"), schema_);
    current_info.SetColumnGroupsForTesting(cgs);
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/v1"), schema_);
    new_info.SetColumnGroupsForTesting(cgs);

    auto diff = current_info.ComputeDiff(new_info);

    EXPECT_FALSE(diff.manifest_updated);
    EXPECT_TRUE(diff.column_groups_to_load.empty());
    EXPECT_TRUE(diff.column_groups_to_replace.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyload.empty());
    EXPECT_TRUE(diff.column_groups_to_lazyreplace.empty());
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffColumnGroupSameIndexDifferentFileCountTriggersReplace) {
    auto current_cgs = MakeColumnGroups(
        {{{100}, {"/pk/file.parquet"}}, {{105, 106}, {"/user/a.parquet"}}});
    auto new_cgs = MakeColumnGroups(
        {{{100}, {"/pk/file.parquet"}},
         {{105, 106}, {"/user/a.parquet", "/user/b.parquet"}}});

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"), schema_);
    current_info.SetColumnGroupsForTesting(current_cgs);
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/new"), schema_);
    new_info.SetColumnGroupsForTesting(new_cgs);

    auto diff = current_info.ComputeDiff(new_info);

    bool saw_user_group = false;
    for (const auto& [idx, fields] : diff.column_groups_to_replace) {
        if (idx == 1) {
            saw_user_group = true;
            std::set<int64_t> ids;
            for (auto f : fields) {
                ids.insert(f.get());
            }
            EXPECT_EQ(ids, (std::set<int64_t>{105, 106}));
        }
    }
    EXPECT_TRUE(saw_user_group);
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffColumnGroupMovedFieldNotDoubleEmittedWithFilesChanged) {
    // Current: field 106 lives at column-group index 0 (alongside 105).
    // New: field 106 moves to a new column-group index 2, and the files
    //      at index 0 also differ. 106 must appear exactly once in the
    //      diff — as a "moved" entry under group 2, not also under 0.
    auto current_cgs =
        MakeColumnGroups({{{100}, {"/pk/file.parquet"}},
                          {{105, 106}, {"/user/file_A.parquet"}}});
    auto new_cgs = MakeColumnGroups({{{100}, {"/pk/file.parquet"}},
                                     {{105}, {"/user/file_B.parquet"}},
                                     {{106}, {"/moved/file.parquet"}}});

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"), schema_);
    current_info.SetColumnGroupsForTesting(current_cgs);
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/new"), schema_);
    new_info.SetColumnGroupsForTesting(new_cgs);

    auto diff = current_info.ComputeDiff(new_info);

    int emissions_for_106 = 0;
    for (const auto& [idx, fields] : diff.column_groups_to_replace) {
        for (auto f : fields) {
            if (f.get() == 106) {
                emissions_for_106++;
            }
        }
    }
    for (const auto& [idx, fields] : diff.column_groups_to_lazyreplace) {
        for (auto f : fields) {
            if (f.get() == 106) {
                emissions_for_106++;
            }
        }
    }
    for (const auto& [idx, fields] : diff.column_groups_to_load) {
        for (auto f : fields) {
            if (f.get() == 106) {
                emissions_for_106++;
            }
        }
    }
    EXPECT_EQ(emissions_for_106, 1);
}

TEST_F(SegmentLoadInfoTest,
       ComputeDiffColumnGroupSamePathDifferentRowRangeTriggersReplace) {
    // Same file path on both sides, but the row range (start_index /
    // end_index) differs — e.g. compaction re-slices a packed file.
    // path-only comparison would miss this and leave stale cache.
    auto make_cgs = [](int64_t start, int64_t end) {
        auto cgs = std::make_shared<milvus_storage::api::ColumnGroups>();
        auto pk_cg = std::make_shared<milvus_storage::api::ColumnGroup>();
        pk_cg->format = "parquet";
        pk_cg->columns.push_back("100");
        milvus_storage::api::ColumnGroupFile pk_file;
        pk_file.path = "/pk/file.parquet";
        pk_file.start_index = 0;
        pk_file.end_index = 0;
        pk_cg->files.push_back(std::move(pk_file));
        cgs->push_back(std::move(pk_cg));

        auto user_cg = std::make_shared<milvus_storage::api::ColumnGroup>();
        user_cg->format = "parquet";
        user_cg->columns.push_back("105");
        user_cg->columns.push_back("106");
        milvus_storage::api::ColumnGroupFile user_file;
        user_file.path = "/user/packed.parquet";
        user_file.start_index = start;
        user_file.end_index = end;
        user_cg->files.push_back(std::move(user_file));
        cgs->push_back(std::move(user_cg));
        return cgs;
    };

    auto current_cgs = make_cgs(0, 500);
    auto new_cgs = make_cgs(500, 1000);

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"), schema_);
    current_info.SetColumnGroupsForTesting(current_cgs);
    SegmentLoadInfo new_info(MakeManifestProto("/manifest/new"), schema_);
    new_info.SetColumnGroupsForTesting(new_cgs);

    auto diff = current_info.ComputeDiff(new_info);

    bool saw_user_group = false;
    for (const auto& [idx, fields] : diff.column_groups_to_replace) {
        std::set<int64_t> ids;
        for (auto f : fields) {
            ids.insert(f.get());
        }
        if (ids == std::set<int64_t>{105, 106}) {
            saw_user_group = true;
        }
    }
    EXPECT_TRUE(saw_user_group);
    EXPECT_TRUE(diff.column_groups_to_load.empty());
}

// ==================== Drop Field Tests ====================
// These tests verify that ComputeDiff correctly handles fields that exist
// in the old schema but have been dropped from the new schema.

TEST_F(SegmentLoadInfoTest, ComputeDiffDefaultFieldsDroppedField) {
    // Bug: ComputeDiffDefaultFields iterates schema_->get_fields() (old schema)
    // which still includes the dropped field. If the field has no data source
    // in either old or new, it gets added to fields_to_fill_default.
    //
    // Setup: old schema has field 110 (extra_field3). New schema drops it.
    // Old_info has no data for field 110 (never loaded).
    // Expected: field 110 should NOT appear in fields_to_fill_default.

    auto new_schema = std::make_shared<Schema>();
    new_schema->AddDebugField("pk", DataType::INT64);
    new_schema->AddDebugField(
        "vec", DataType::VECTOR_FLOAT, 128, knowhere::metric::L2);
    new_schema->AddDebugField("json_field", DataType::JSON);
    new_schema->AddDebugField("bm25_field", DataType::JSON);
    new_schema->AddDebugField("group_field", DataType::INT64);
    new_schema->AddDebugField("child_field1", DataType::FLOAT);
    new_schema->AddDebugField("child_field2", DataType::FLOAT);
    new_schema->AddDebugField("child_field3", DataType::FLOAT);
    new_schema->AddDebugField("extra_field1", DataType::INT64);
    new_schema->AddDebugField("extra_field2", DataType::FLOAT);
    // field 110 (extra_field3) intentionally NOT added — it is dropped
    new_schema->set_primary_field_id(FieldId(100));

    // Current: only has pk binlog, all other fields were default-filled
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* binlog = current_proto.add_binlog_paths();
    binlog->set_fieldid(100);
    auto* log = binlog->add_binlogs();
    log->set_log_path("/path/to/pk_binlog");
    log->set_entries_num(1000);

    // New: same, no data for field 110
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog = new_proto.add_binlog_paths();
    new_binlog->set_fieldid(100);
    auto* new_log = new_binlog->add_binlogs();
    new_log->set_log_path("/path/to/pk_binlog");
    new_log->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);  // old schema has 110
    SegmentLoadInfo new_info(new_proto, new_schema);       // new schema no 110
    auto diff = current_info.ComputeDiff(new_info);

    // Field 110 is dropped — it should NOT be in fields_to_fill_default
    for (const auto& fid : diff.fields_to_fill_default) {
        EXPECT_NE(fid.get(), 110)
            << "Dropped field 110 should not be in fields_to_fill_default";
    }
}

TEST_F(SegmentLoadInfoTest, ComputeDiffTextIndexesDroppedField) {
    // Bug: ComputeDiffTextIndexes iterates schema_->get_fields() (old schema)
    // and checks enable_match(). If the dropped field had enable_match, it
    // gets added to text_indexes_to_create.
    //
    // Setup: old schema has text_field (102) with enable_match=true.
    // New schema drops field 102.
    // Expected: field 102 should NOT appear in text_indexes_to_create.

    auto old_schema = CreateSchemaWithTextMatchField();
    // 100=pk, 101=vec, 102=text_field(enable_match), 103=plain_varchar

    // New schema: drops field 102 (text_field)
    auto new_schema = std::make_shared<Schema>();
    new_schema->AddDebugField("pk", DataType::INT64);
    new_schema->AddDebugField(
        "vec", DataType::VECTOR_FLOAT, 128, knowhere::metric::L2);
    // Skip field 102 (dropped)
    new_schema->AddDebugField("plain_varchar", DataType::VARCHAR);
    new_schema->set_primary_field_id(FieldId(100));

    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);

    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);

    SegmentLoadInfo current_info(current_proto, old_schema);
    SegmentLoadInfo new_info(new_proto, new_schema);
    auto diff = current_info.ComputeDiff(new_info);

    // Field 102 is dropped — it should NOT be in text_indexes_to_create
    EXPECT_TRUE(diff.text_indexes_to_create.count(FieldId(102)) == 0)
        << "Dropped field 102 should not be in text_indexes_to_create";
}

// NOTE: ComputeDiffIndexes also has a schema filter for dropped fields, but
// it cannot be unit-tested because BuildCache() requires schema to contain all
// fields referenced by index_infos in the proto (it accesses schema[field_id]).
// The filter serves as defense-in-depth; Go layer filters dropped-field indexes
// before constructing the proto.

TEST_F(SegmentLoadInfoTest, ComputeDiffBinlogsDroppedField) {
    // Verify that ComputeDiffBinlogs skips binlogs for dropped fields.
    //
    // Setup: new_info has binlog for field 102 (legacy format).
    // New schema drops field 102.
    // The binlog should NOT appear in binlogs_to_load.

    auto new_schema = std::make_shared<Schema>();
    new_schema->AddDebugField("pk", DataType::INT64);
    new_schema->AddDebugField(
        "vec", DataType::VECTOR_FLOAT, 128, knowhere::metric::L2);
    // field 102 (json_field) intentionally NOT added — dropped
    new_schema->set_primary_field_id(FieldId(100));

    // Current: binlog for field 101 only
    proto::segcore::SegmentLoadInfo current_proto;
    current_proto.set_segmentid(100);
    current_proto.set_num_of_rows(1000);
    auto* cur_binlog = current_proto.add_binlog_paths();
    cur_binlog->set_fieldid(101);
    auto* cur_log = cur_binlog->add_binlogs();
    cur_log->set_log_path("/path/to/binlog_101");
    cur_log->set_entries_num(1000);

    // New: binlogs for fields 101 and 102 (legacy format)
    proto::segcore::SegmentLoadInfo new_proto;
    new_proto.set_segmentid(100);
    new_proto.set_num_of_rows(1000);
    auto* new_binlog1 = new_proto.add_binlog_paths();
    new_binlog1->set_fieldid(101);
    auto* new_log1 = new_binlog1->add_binlogs();
    new_log1->set_log_path("/path/to/binlog_101");
    new_log1->set_entries_num(1000);
    auto* new_binlog2 = new_proto.add_binlog_paths();
    new_binlog2->set_fieldid(102);
    auto* new_log2 = new_binlog2->add_binlogs();
    new_log2->set_log_path("/path/to/binlog_102");
    new_log2->set_entries_num(1000);

    SegmentLoadInfo current_info(current_proto, schema_);
    SegmentLoadInfo new_info(new_proto, new_schema);
    auto diff = current_info.ComputeDiff(new_info);

    // Field 102 is dropped — should NOT appear in binlogs_to_load
    for (const auto& [field_ids, binlog] : diff.binlogs_to_load) {
        for (const auto& fid : field_ids) {
            EXPECT_NE(fid.get(), 102)
                << "Dropped field 102 should not be in binlogs_to_load";
        }
    }
    // Field 102 should be in field_data_to_drop (was filtered from new)
    // but since it wasn't in current either, it won't be there.
    // Just verify no crash and dropped field is excluded from load.
}

// NOTE: ComputeDiffColumnGroups also has a schema filter for dropped fields,
// but it cannot be unit-tested because GetColumnGroups() requires Loon FFI
// (real manifest file access). The filter uses the same pattern
// (new_info.schema_->has_field()) as ComputeDiffBinlogs tested above.

// ==================== storage-v3 raw-index skip (vector-only) ================
// Regression tests for the DiskANN double-footprint fix (PR #51541): on
// storage-v3 the skip is restricted to VECTOR fields — a vector index
// (IVF_FLAT / DiskANN / HNSW-with-raw) reconstructs the raw vector at retrieve
// (get_vector), so its raw column is dropped. Scalar / JSON / ARRAY raw columns
// always stay resident (their index-read paths do not cover every consumer).
// The vector skip + retrieve-from-index correctness is exercised end-to-end by
// the storage-v3 e2e / go-sdk suites; these unit tests pin the LoadDiff
// classification, in particular that scalars are NOT skipped.
// CanUseIndexRawDataForField additionally excludes JSON/ARRAY/VECTOR_ARRAY at
// the source.

namespace {

// True if `field_id` appears in any (group_index, fields) entry.
bool
ColumnGroupFieldPresent(
    const std::vector<std::pair<int, std::vector<FieldId>>>& groups,
    int64_t field_id) {
    for (const auto& [group_index, fields] : groups) {
        for (const auto& f : fields) {
            if (f.get() == field_id) {
                return true;
            }
        }
    }
    return false;
}

// Attach a scalar index (index_type) to a field so BuildCache populates
// field_index_has_raw_data_ (ASCENDING_SORT/STL_SORT => has raw data,
// INVERTED => no raw data).
void
AddScalarIndex(proto::segcore::SegmentLoadInfo& proto,
               int64_t field_id,
               int64_t index_id,
               const std::string& index_type) {
    auto* idx = proto.add_index_infos();
    idx->set_fieldid(field_id);
    idx->set_indexid(index_id);
    idx->add_index_file_paths("/idx/" + std::to_string(field_id));
    auto* param = idx->add_index_params();
    param->set_key("index_type");
    param->set_value(index_type);
}

}  // namespace

TEST_F(SegmentLoadInfoTest,
       ComputeDiffColumnGroupDoesNotSkipScalarRawDataIndex) {
    // The skip is vector-only. A scalar field (108, INT64) with a raw-data
    // index (STL_SORT) must NOT be skipped: its raw column stays resident
    // (loaded) and is never dropped, because scalar index-read paths do not
    // cover every consumer (nullable Reverse_Lookup, column-scan-only exprs).
    auto new_proto = MakeManifestProto("/manifest/new");
    AddScalarIndex(new_proto, 108, 6001, milvus::index::ASCENDING_SORT);

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"), schema_);
    current_info.SetColumnGroupsForTesting(MakeColumnGroups({}));
    SegmentLoadInfo new_info(new_proto, schema_);
    new_info.SetColumnGroupsForTesting(
        MakeColumnGroups({{{108}, {"/cg/108.parquet"}}}));

    auto diff = current_info.ComputeDiff(new_info);

    bool loaded = ColumnGroupFieldPresent(diff.column_groups_to_load, 108) ||
                  ColumnGroupFieldPresent(diff.column_groups_to_lazyload, 108);
    EXPECT_TRUE(loaded)
        << "scalar raw column must stay resident (skip is vector-only)";
    EXPECT_EQ(diff.field_data_to_drop.count(FieldId(108)), 0u)
        << "scalar raw column must not be dropped";
}

TEST_F(SegmentLoadInfoTest, ComputeDiffColumnGroupNeverSkipsPrimaryKey) {
    // The pk (field 100) column must stay resident even when its index carries
    // raw data: sorted-segment row navigation (num_rows_until_chunk /
    // get_chunk_by_offset) reads the pk column directly, and a pk scalar index
    // is not registered for expression index-reads.
    auto new_proto = MakeManifestProto("/manifest/new");
    AddScalarIndex(new_proto, 100, 6002, milvus::index::ASCENDING_SORT);

    SegmentLoadInfo current_info(MakeManifestProto("/manifest/old"), schema_);
    current_info.SetColumnGroupsForTesting(MakeColumnGroups({}));
    SegmentLoadInfo new_info(new_proto, schema_);
    new_info.SetColumnGroupsForTesting(
        MakeColumnGroups({{{100}, {"/cg/100.parquet"}}}));

    auto diff = current_info.ComputeDiff(new_info);

    // pk must NOT be skipped: it lands in a load list (lazy, since its index
    // reports raw data, but still resident).
    bool pk_loaded =
        ColumnGroupFieldPresent(diff.column_groups_to_load, 100) ||
        ColumnGroupFieldPresent(diff.column_groups_to_lazyload, 100);
    EXPECT_TRUE(pk_loaded)
        << "primary key column must load even when its index has raw data";
}

TEST(IndexFactoryRawDataTest,
     CanUseIndexRawDataForFieldExcludesNonSubstitutableTypes) {
    // JSON and ARRAY indexes only index a projection (a JSON path / array
    // elements). VECTOR_ARRAY currently keeps its field column for struct
    // offsets and parent-row validity. None of them may substitute for the raw
    // column, regardless of the index's own has_raw_data flag.
    EXPECT_FALSE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::JSON, true));
    EXPECT_FALSE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::ARRAY, true));
    EXPECT_FALSE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::VECTOR_ARRAY, true));
    // Other field types pass the index's has_raw_data through unchanged.
    EXPECT_TRUE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::INT64, true));
    EXPECT_TRUE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::VECTOR_FLOAT, true));
    EXPECT_TRUE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::VARCHAR, true));
    // has_raw_data == false is always false.
    EXPECT_FALSE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::INT64, false));
    EXPECT_FALSE(milvus::index::IndexFactory::CanUseIndexRawDataForField(
        DataType::JSON, false));
}
