// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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 <filesystem>
#include <memory>
#include "common/FastMem.h"

#include "arrow/array/builder_binary.h"
#include "arrow/array/builder_nested.h"
#include "arrow/array/builder_primitive.h"
#include "arrow/array/concatenate.h"
#include "arrow/buffer_builder.h"
#include <arrow/c/bridge.h>
#include "arrow/scalar.h"
#include "arrow/type_fwd.h"
#include "common/type_c.h"
#include "fmt/format.h"
#include "index/Utils.h"
#include "log/Log.h"

#include "common/Consts.h"
#include "common/EasyAssert.h"
#include "common/FieldData.h"
#include "common/Geometry.h"
#include "common/Schema.h"
#include "common/FieldDataInterface.h"
#include "pb/common.pb.h"
#include "storage/StorageV2FSCache.h"
#ifdef AZURE_BUILD_DIR
#include "storage/azure/AzureChunkManager.h"
#endif
#ifdef ENABLE_GCP_NATIVE
#include "storage/gcp-native-storage/GcpNativeChunkManager.h"
#endif
#include "storage/ChunkManager.h"
#include "storage/DiskFileManagerImpl.h"
#include "storage/InsertData.h"
#include "storage/LocalChunkManager.h"
#include "storage/MemFileManagerImpl.h"
#include "storage/minio/MinioChunkManager.h"
#include "storage/Types.h"
#include "storage/Util.h"
#include "common/Common.h"
#include "common/Types.h"
#include "common/VectorArray.h"
#include "storage/ThreadPools.h"
#include "storage/MemFileManagerImpl.h"
#include "storage/DiskFileManagerImpl.h"
#include "storage/KeyRetriever.h"
#include "segcore/memory_planner.h"
#include "mmap/Types.h"
#include "storage/loon_ffi/ffi_reader_c.h"
#include "storage/loon_ffi/util.h"
#include "milvus-storage/common/constants.h"
#include "milvus-storage/ffi_c.h"
#include "milvus-storage/format/parquet/file_reader.h"
#include "milvus-storage/filesystem/fs.h"
#include "milvus-storage/lob_column/lob_column_manager.h"
#include "milvus-storage/reader.h"
#include "milvus-storage/segment/segment_reader.h"
#include "nlohmann/json.hpp"

namespace milvus::storage {

constexpr const char* TEMP = "tmp";

std::map<std::string, ChunkManagerType> ChunkManagerType_Map = {
    {"local", ChunkManagerType::Local},
    {"minio", ChunkManagerType::Minio},
    {"remote", ChunkManagerType::Remote}};

static std::shared_ptr<arrow::Array>
NormalizeExternalArrowByType(const std::shared_ptr<arrow::Array>& array,
                             DataType data_type,
                             int64_t dim,
                             bool nullable,
                             DataType element_type,
                             const FieldMeta& field_meta);

enum class CloudProviderType : int8_t {
    UNKNOWN = 0,
    AWS = 1,
    GCP = 2,
    ALIYUN = 3,
    AZURE = 4,
    TENCENTCLOUD = 5,
    GCPNATIVE = 6,
    HUAWEICLOUD = 7,
};

std::map<std::string, CloudProviderType> CloudProviderType_Map = {
    {"aws", CloudProviderType::AWS},
    {"gcp", CloudProviderType::GCP},
    {"aliyun", CloudProviderType::ALIYUN},
    {"azure", CloudProviderType::AZURE},
    {"tencent", CloudProviderType::TENCENTCLOUD},
    {"gcpnative", CloudProviderType::GCPNATIVE},
    {"huawei", CloudProviderType::HUAWEICLOUD}};

std::map<std::string, int> ReadAheadPolicy_Map = {
    {"normal", MADV_NORMAL},
    {"random", MADV_RANDOM},
    {"sequential", MADV_SEQUENTIAL},
    {"willneed", MADV_WILLNEED},
    {"dontneed", MADV_DONTNEED}};

// in arrow, null_bitmap read from the least significant bit
std::vector<uint8_t>
genValidIter(const uint8_t* valid_data, int length) {
    std::vector<uint8_t> valid_data_;
    valid_data_.reserve(length);
    for (size_t i = 0; i < length; ++i) {
        auto bit = (valid_data[i >> 3] >> (i & 0x07)) & 1;
        valid_data_.push_back(bit);
    }
    return valid_data_;
}

void
ReadMediumType(BinlogReaderPtr reader) {
    AssertInfo(reader->Tell() == 0,
               "medium type must be parsed from stream header");
    int32_t magic_num;
    auto ret = reader->Read(sizeof(magic_num), &magic_num);
    AssertInfo(ret.ok(), "read binlog failed: {}", ret.what());
    AssertInfo(magic_num == MAGIC_NUM, "invalid magic num: {}", magic_num);
}

void
add_vector_payload(std::shared_ptr<arrow::ArrayBuilder> builder,
                   uint8_t* values,
                   const uint8_t* valid_data,
                   bool nullable,
                   int length,
                   int byte_width) {
    AssertInfo(builder != nullptr, "empty arrow builder");
    AssertInfo((nullable && valid_data) || !nullable,
               "valid_data is required for nullable vectors");
    arrow::Status ast;

    if (nullable) {
        auto binary_builder =
            std::dynamic_pointer_cast<arrow::BinaryBuilder>(builder);
        int valid_index = 0;
        for (int i = 0; i < length; ++i) {
            auto bit = (valid_data[i >> 3] >> (i & 0x07)) & 1;
            if (bit) {
                ast = binary_builder->Append(values + valid_index * byte_width,
                                             byte_width);
                valid_index++;
            } else {
                ast = binary_builder->AppendNull();
            }
            AssertInfo(ast.ok(),
                       "append value to arrow builder failed: {}",
                       ast.ToString());
        }
    } else {
        auto binary_builder =
            std::dynamic_pointer_cast<arrow::FixedSizeBinaryBuilder>(builder);
        ast = binary_builder->AppendValues(values, length);
        AssertInfo(ast.ok(),
                   "append value to arrow builder failed: {}",
                   ast.ToString());
    }
}

// append values for numeric data
template <typename DT, typename BT>
void
add_numeric_payload(std::shared_ptr<arrow::ArrayBuilder> builder,
                    DT* start,
                    const uint8_t* valid_data,
                    bool nullable,
                    int length) {
    AssertInfo(builder != nullptr, "empty arrow builder");
    auto numeric_builder = std::dynamic_pointer_cast<BT>(builder);
    arrow::Status ast;
    if (nullable) {
        // need iter to read valid_data when write
        auto iter = genValidIter(valid_data, length);
        ast =
            numeric_builder->AppendValues(start, start + length, iter.begin());
        AssertInfo(ast.ok(), "append value to arrow builder failed");
    } else {
        ast = numeric_builder->AppendValues(start, start + length);
        AssertInfo(ast.ok(), "append value to arrow builder failed");
    }
}

void
AddPayloadToArrowBuilder(std::shared_ptr<arrow::ArrayBuilder> builder,
                         const Payload& payload) {
    AssertInfo(builder != nullptr, "empty arrow builder");
    auto raw_data = const_cast<uint8_t*>(payload.raw_data);
    auto length = payload.rows;
    auto data_type = payload.data_type;
    auto nullable = payload.nullable;

    switch (data_type) {
        case DataType::BOOL: {
            auto bool_data = reinterpret_cast<bool*>(raw_data);
            add_numeric_payload<bool, arrow::BooleanBuilder>(
                builder, bool_data, payload.valid_data, nullable, length);
            break;
        }
        case DataType::INT8: {
            auto int8_data = reinterpret_cast<int8_t*>(raw_data);
            add_numeric_payload<int8_t, arrow::Int8Builder>(
                builder, int8_data, payload.valid_data, nullable, length);
            break;
        }
        case DataType::INT16: {
            auto int16_data = reinterpret_cast<int16_t*>(raw_data);
            add_numeric_payload<int16_t, arrow::Int16Builder>(
                builder, int16_data, payload.valid_data, nullable, length);
            break;
        }
        case DataType::INT32: {
            auto int32_data = reinterpret_cast<int32_t*>(raw_data);
            add_numeric_payload<int32_t, arrow::Int32Builder>(
                builder, int32_data, payload.valid_data, nullable, length);
            break;
        }
        case DataType::INT64: {
            auto int64_data = reinterpret_cast<int64_t*>(raw_data);
            add_numeric_payload<int64_t, arrow::Int64Builder>(
                builder, int64_data, payload.valid_data, nullable, length);
            break;
        }
        case DataType::FLOAT: {
            auto float_data = reinterpret_cast<float*>(raw_data);
            add_numeric_payload<float, arrow::FloatBuilder>(
                builder, float_data, payload.valid_data, nullable, length);
            break;
        }
        case DataType::DOUBLE: {
            auto double_data = reinterpret_cast<double_t*>(raw_data);
            add_numeric_payload<double, arrow::DoubleBuilder>(
                builder, double_data, payload.valid_data, nullable, length);
            break;
        }
        case DataType::TIMESTAMPTZ: {
            auto timestamptz_data = reinterpret_cast<int64_t*>(raw_data);
            add_numeric_payload<int64_t, arrow::Int64Builder>(
                builder,
                timestamptz_data,
                payload.valid_data,
                nullable,
                length);
            break;
        }

        case DataType::VECTOR_FLOAT: {
            AssertInfo(payload.dimension.has_value(),
                       "dimension is required for VECTOR_FLOAT");
            int byte_width = payload.dimension.value() * sizeof(float);
            add_vector_payload(builder,
                               const_cast<uint8_t*>(raw_data),
                               payload.valid_data,
                               nullable,
                               length,
                               byte_width);
            break;
        }
        case DataType::VECTOR_BINARY: {
            AssertInfo(payload.dimension.has_value(),
                       "dimension is required for VECTOR_BINARY");
            int byte_width = (payload.dimension.value() + 7) / 8;
            add_vector_payload(builder,
                               const_cast<uint8_t*>(raw_data),
                               payload.valid_data,
                               nullable,
                               length,
                               byte_width);
            break;
        }
        case DataType::VECTOR_FLOAT16: {
            AssertInfo(payload.dimension.has_value(),
                       "dimension is required for VECTOR_FLOAT16");
            int byte_width = payload.dimension.value() * 2;
            add_vector_payload(builder,
                               const_cast<uint8_t*>(raw_data),
                               payload.valid_data,
                               nullable,
                               length,
                               byte_width);
            break;
        }
        case DataType::VECTOR_BFLOAT16: {
            AssertInfo(payload.dimension.has_value(),
                       "dimension is required for VECTOR_BFLOAT16");
            int byte_width = payload.dimension.value() * 2;
            add_vector_payload(builder,
                               const_cast<uint8_t*>(raw_data),
                               payload.valid_data,
                               nullable,
                               length,
                               byte_width);
            break;
        }
        case DataType::VECTOR_INT8: {
            AssertInfo(payload.dimension.has_value(),
                       "dimension is required for VECTOR_INT8");
            int byte_width = payload.dimension.value() * sizeof(int8_t);
            add_vector_payload(builder,
                               const_cast<uint8_t*>(raw_data),
                               payload.valid_data,
                               nullable,
                               length,
                               byte_width);
            break;
        }
        case DataType::VECTOR_SPARSE_U32_F32: {
            ThrowInfo(DataTypeInvalid,
                      "Sparse Float Vector payload should be added by calling "
                      "add_one_binary_payload",
                      data_type);
        }
        case DataType::VECTOR_ARRAY: {
            auto list_builder =
                std::dynamic_pointer_cast<arrow::ListBuilder>(builder);
            AssertInfo(list_builder != nullptr,
                       "builder must be ListBuilder for VECTOR_ARRAY");

            auto vector_arrays = reinterpret_cast<VectorArray*>(raw_data);
            auto valid_data = payload.valid_data;
            AssertInfo((nullable && valid_data) || !nullable,
                       "valid_data is required for nullable VectorArray");

            if (length > 0) {
                auto value_builder =
                    static_cast<arrow::FixedSizeBinaryBuilder*>(
                        list_builder->value_builder());
                AssertInfo(value_builder != nullptr,
                           "value_builder must be FixedSizeBinaryBuilder for "
                           "VectorArray");

                DataType element_type = DataType::NONE;
                auto append_vector_array = [&](const VectorArray& array) {
                    if (element_type == DataType::NONE) {
                        element_type = array.get_element_type();
                        switch (element_type) {
                            case DataType::VECTOR_FLOAT:
                            case DataType::VECTOR_BINARY:
                            case DataType::VECTOR_FLOAT16:
                            case DataType::VECTOR_BFLOAT16:
                            case DataType::VECTOR_INT8:
                                break;
                            default:
                                ThrowInfo(
                                    DataTypeInvalid,
                                    "Unsupported element type in VectorArray: "
                                    "{}",
                                    element_type);
                        }
                    } else {
                        AssertInfo(array.get_element_type() == element_type,
                                   "Inconsistent element types in "
                                   "VectorArray");
                    }
                    auto status = list_builder->Append();
                    AssertInfo(status.ok(),
                               "Failed to append list: {}",
                               status.ToString());

                    int num_vectors = array.length();
                    if (num_vectors > 0) {
                        auto ast = value_builder->AppendValues(
                            reinterpret_cast<const uint8_t*>(array.data()),
                            num_vectors);
                        AssertInfo(ast.ok(),
                                   "Failed to batch append vectors: {}",
                                   ast.ToString());
                    }
                };

                int valid_index = 0;
                for (int i = 0; i < length; ++i) {
                    if (nullable) {
                        auto bit = (valid_data[i >> 3] >> (i & 0x07)) & 1;
                        if (!bit) {
                            auto status = list_builder->AppendNull();
                            AssertInfo(status.ok(),
                                       "Failed to append null list: {}",
                                       status.ToString());
                            continue;
                        }
                        append_vector_array(vector_arrays[valid_index++]);
                    } else {
                        append_vector_array(vector_arrays[i]);
                    }
                }
            }
            break;
        }
        default: {
            ThrowInfo(DataTypeInvalid, "unsupported data type {}", data_type);
        }
    }
}

void
AddOneStringToArrowBuilder(std::shared_ptr<arrow::ArrayBuilder> builder,
                           const char* str,
                           int str_size) {
    AssertInfo(builder != nullptr, "empty arrow builder");
    auto string_builder =
        std::dynamic_pointer_cast<arrow::StringBuilder>(builder);
    arrow::Status ast;
    if (str == nullptr || str_size < 0) {
        ast = string_builder->AppendNull();
    } else {
        ast = string_builder->Append(str, str_size);
    }
    AssertInfo(
        ast.ok(), "append value to arrow builder failed: {}", ast.ToString());
}

void
AddOneBinaryToArrowBuilder(std::shared_ptr<arrow::ArrayBuilder> builder,
                           const uint8_t* data,
                           int length) {
    AssertInfo(builder != nullptr, "empty arrow builder");
    auto binary_builder =
        std::dynamic_pointer_cast<arrow::BinaryBuilder>(builder);
    arrow::Status ast;
    if (data == nullptr || length < 0) {
        ast = binary_builder->AppendNull();
    } else {
        ast = binary_builder->Append(data, length);
    }
    AssertInfo(
        ast.ok(), "append value to arrow builder failed: {}", ast.ToString());
}

std::shared_ptr<arrow::ArrayBuilder>
CreateArrowBuilder(DataType data_type) {
    switch (static_cast<DataType>(data_type)) {
        case DataType::BOOL: {
            return std::make_shared<arrow::BooleanBuilder>();
        }
        case DataType::INT8: {
            return std::make_shared<arrow::Int8Builder>();
        }
        case DataType::INT16: {
            return std::make_shared<arrow::Int16Builder>();
        }
        case DataType::INT32: {
            return std::make_shared<arrow::Int32Builder>();
        }
        case DataType::INT64: {
            return std::make_shared<arrow::Int64Builder>();
        }
        case DataType::FLOAT: {
            return std::make_shared<arrow::FloatBuilder>();
        }
        case DataType::DOUBLE: {
            return std::make_shared<arrow::DoubleBuilder>();
        }
        case DataType::TIMESTAMPTZ: {
            return std::make_shared<arrow::Int64Builder>();
        }
        case DataType::VARCHAR:
        case DataType::STRING:
        case DataType::TEXT: {
            return std::make_shared<arrow::StringBuilder>();
        }
        case DataType::ARRAY:
        case DataType::JSON:
        case DataType::GEOMETRY: {
            return std::make_shared<arrow::BinaryBuilder>();
        }
        // sparse float vector doesn't require a dim
        case DataType::VECTOR_SPARSE_U32_F32: {
            return std::make_shared<arrow::BinaryBuilder>();
        }
        default: {
            ThrowInfo(
                DataTypeInvalid, "unsupported numeric data type {}", data_type);
        }
    }
}

std::shared_ptr<arrow::ArrayBuilder>
CreateArrowBuilder(DataType data_type,
                   DataType element_type,
                   int dim,
                   bool nullable) {
    switch (static_cast<DataType>(data_type)) {
        case DataType::VECTOR_FLOAT: {
            AssertInfo(dim > 0, "invalid dim value: {}", dim);
            if (nullable) {
                return std::make_shared<arrow::BinaryBuilder>();
            }
            return std::make_shared<arrow::FixedSizeBinaryBuilder>(
                arrow::fixed_size_binary(dim * sizeof(float)));
        }
        case DataType::VECTOR_BINARY: {
            AssertInfo(dim % 8 == 0 && dim > 0, "invalid dim value: {}", dim);
            if (nullable) {
                return std::make_shared<arrow::BinaryBuilder>();
            }
            return std::make_shared<arrow::FixedSizeBinaryBuilder>(
                arrow::fixed_size_binary(dim / 8));
        }
        case DataType::VECTOR_FLOAT16: {
            AssertInfo(dim > 0, "invalid dim value: {}", dim);
            if (nullable) {
                return std::make_shared<arrow::BinaryBuilder>();
            }
            return std::make_shared<arrow::FixedSizeBinaryBuilder>(
                arrow::fixed_size_binary(dim * sizeof(float16)));
        }
        case DataType::VECTOR_BFLOAT16: {
            AssertInfo(dim > 0, "invalid dim value");
            if (nullable) {
                return std::make_shared<arrow::BinaryBuilder>();
            }
            return std::make_shared<arrow::FixedSizeBinaryBuilder>(
                arrow::fixed_size_binary(dim * sizeof(bfloat16)));
        }
        case DataType::VECTOR_INT8: {
            AssertInfo(dim > 0, "invalid dim value");
            if (nullable) {
                return std::make_shared<arrow::BinaryBuilder>();
            }
            return std::make_shared<arrow::FixedSizeBinaryBuilder>(
                arrow::fixed_size_binary(dim * sizeof(int8)));
        }
        case DataType::VECTOR_SPARSE_U32_F32: {
            return std::make_shared<arrow::BinaryBuilder>();
        }
        case DataType::VECTOR_ARRAY: {
            AssertInfo(dim > 0, "invalid dim value");
            AssertInfo(element_type != DataType::NONE,
                       "element_type must be specified for VECTOR_ARRAY");

            std::shared_ptr<arrow::ArrayBuilder> value_builder;
            switch (element_type) {
                case DataType::VECTOR_FLOAT: {
                    int byte_width = dim * sizeof(float);
                    value_builder =
                        std::make_shared<arrow::FixedSizeBinaryBuilder>(
                            arrow::fixed_size_binary(byte_width));
                    break;
                }
                case DataType::VECTOR_BINARY: {
                    int byte_width = (dim + 7) / 8;
                    value_builder =
                        std::make_shared<arrow::FixedSizeBinaryBuilder>(
                            arrow::fixed_size_binary(byte_width));
                    break;
                }
                case DataType::VECTOR_FLOAT16: {
                    int byte_width = dim * 2;
                    value_builder =
                        std::make_shared<arrow::FixedSizeBinaryBuilder>(
                            arrow::fixed_size_binary(byte_width));
                    break;
                }
                case DataType::VECTOR_BFLOAT16: {
                    int byte_width = dim * 2;
                    value_builder =
                        std::make_shared<arrow::FixedSizeBinaryBuilder>(
                            arrow::fixed_size_binary(byte_width));
                    break;
                }
                case DataType::VECTOR_INT8: {
                    int byte_width = dim;
                    value_builder =
                        std::make_shared<arrow::FixedSizeBinaryBuilder>(
                            arrow::fixed_size_binary(byte_width));
                    break;
                }
                default: {
                    ThrowInfo(DataTypeInvalid,
                              "unsupported element type {} for VECTOR_ARRAY",
                              GetDataTypeName(element_type));
                }
            }

            return std::make_shared<arrow::ListBuilder>(
                arrow::default_memory_pool(), value_builder);
        }
        default: {
            ThrowInfo(
                DataTypeInvalid, "unsupported vector data type {}", data_type);
        }
    }
}

std::shared_ptr<arrow::Scalar>
CreateArrowScalarFromDefaultValue(const FieldMeta& field_meta) {
    auto default_var = field_meta.default_value();
    AssertInfo(default_var.has_value(),
               "cannot create Arrow Scalar from empty default value");
    auto default_value = default_var.value();
    switch (field_meta.get_data_type()) {
        case DataType::BOOL:
            return std::make_shared<arrow::BooleanScalar>(
                default_value.bool_data());
        case DataType::INT8:
            return std::make_shared<arrow::Int8Scalar>(
                default_value.int_data());
        case DataType::INT16:
            return std::make_shared<arrow::Int16Scalar>(
                default_value.int_data());
        case DataType::INT32:
            return std::make_shared<arrow::Int32Scalar>(
                default_value.int_data());
        case DataType::INT64:
            return std::make_shared<arrow::Int64Scalar>(
                default_value.long_data());
        case DataType::FLOAT:
            return std::make_shared<arrow::FloatScalar>(
                default_value.float_data());
        case DataType::DOUBLE:
            return std::make_shared<arrow::DoubleScalar>(
                default_value.double_data());
        case DataType::TIMESTAMPTZ:
            return std::make_shared<arrow::Int64Scalar>(
                default_value.timestamptz_data());
        case DataType::VARCHAR:
        case DataType::STRING:
        case DataType::TEXT:
            return std::make_shared<arrow::StringScalar>(
                default_value.string_data());
        case DataType::JSON:
            return std::make_shared<arrow::BinaryScalar>(
                default_value.bytes_data());
        default:
            ThrowInfo(DataTypeInvalid,
                      "unsupported default value data type {}",
                      field_meta.get_data_type());
    }
}

std::shared_ptr<arrow::Schema>
CreateArrowSchema(DataType data_type, bool nullable) {
    switch (static_cast<DataType>(data_type)) {
        case DataType::BOOL: {
            return arrow::schema(
                {arrow::field("val", arrow::boolean(), nullable)});
        }
        case DataType::INT8: {
            return arrow::schema(
                {arrow::field("val", arrow::int8(), nullable)});
        }
        case DataType::INT16: {
            return arrow::schema(
                {arrow::field("val", arrow::int16(), nullable)});
        }
        case DataType::INT32: {
            return arrow::schema(
                {arrow::field("val", arrow::int32(), nullable)});
        }
        case DataType::INT64: {
            return arrow::schema(
                {arrow::field("val", arrow::int64(), nullable)});
        }
        case DataType::FLOAT: {
            return arrow::schema(
                {arrow::field("val", arrow::float32(), nullable)});
        }
        case DataType::DOUBLE: {
            return arrow::schema(
                {arrow::field("val", arrow::float64(), nullable)});
        }
        case DataType::TIMESTAMPTZ: {
            return arrow::schema(
                {arrow::field("val", arrow::int64(), nullable)});
        }
        case DataType::VARCHAR:
        case DataType::STRING:
        case DataType::TEXT: {
            return arrow::schema(
                {arrow::field("val", arrow::utf8(), nullable)});
        }
        case DataType::ARRAY:
        case DataType::JSON:
        case DataType::GEOMETRY: {
            return arrow::schema(
                {arrow::field("val", arrow::binary(), nullable)});
        }
        // sparse float vector doesn't require a dim
        case DataType::VECTOR_SPARSE_U32_F32: {
            return arrow::schema(
                {arrow::field("val", arrow::binary(), nullable)});
        }
        default: {
            ThrowInfo(
                DataTypeInvalid, "unsupported numeric data type {}", data_type);
        }
    }
}

std::shared_ptr<arrow::Schema>
CreateArrowSchema(DataType data_type, int dim, bool nullable) {
    switch (static_cast<DataType>(data_type)) {
        case DataType::VECTOR_FLOAT: {
            AssertInfo(dim > 0, "invalid dim value: {}", dim);
            if (nullable) {
                auto metadata = std::shared_ptr<arrow::KeyValueMetadata>(
                    new arrow::KeyValueMetadata());
                metadata->Append(DIM_KEY, std::to_string(dim));
                return arrow::schema(
                    {arrow::field("val", arrow::binary(), nullable, metadata)});
            }
            return arrow::schema(
                {arrow::field("val",
                              arrow::fixed_size_binary(dim * sizeof(float)),
                              nullable)});
        }
        case DataType::VECTOR_BINARY: {
            AssertInfo(dim % 8 == 0 && dim > 0, "invalid dim value: {}", dim);
            if (nullable) {
                auto metadata = std::shared_ptr<arrow::KeyValueMetadata>(
                    new arrow::KeyValueMetadata());
                metadata->Append(DIM_KEY, std::to_string(dim));
                return arrow::schema(
                    {arrow::field("val", arrow::binary(), nullable, metadata)});
            }
            return arrow::schema({arrow::field(
                "val", arrow::fixed_size_binary(dim / 8), nullable)});
        }
        case DataType::VECTOR_FLOAT16: {
            AssertInfo(dim > 0, "invalid dim value: {}", dim);
            if (nullable) {
                auto metadata = std::shared_ptr<arrow::KeyValueMetadata>(
                    new arrow::KeyValueMetadata());
                metadata->Append(DIM_KEY, std::to_string(dim));
                return arrow::schema(
                    {arrow::field("val", arrow::binary(), nullable, metadata)});
            }
            return arrow::schema(
                {arrow::field("val",
                              arrow::fixed_size_binary(dim * sizeof(float16)),
                              nullable)});
        }
        case DataType::VECTOR_BFLOAT16: {
            AssertInfo(dim > 0, "invalid dim value");
            if (nullable) {
                auto metadata = std::shared_ptr<arrow::KeyValueMetadata>(
                    new arrow::KeyValueMetadata());
                metadata->Append(DIM_KEY, std::to_string(dim));
                return arrow::schema(
                    {arrow::field("val", arrow::binary(), nullable, metadata)});
            }
            return arrow::schema(
                {arrow::field("val",
                              arrow::fixed_size_binary(dim * sizeof(bfloat16)),
                              nullable)});
        }
        case DataType::VECTOR_SPARSE_U32_F32: {
            return arrow::schema(
                {arrow::field("val", arrow::binary(), nullable)});
        }
        case DataType::VECTOR_INT8: {
            AssertInfo(dim > 0, "invalid dim value");
            if (nullable) {
                auto metadata = std::shared_ptr<arrow::KeyValueMetadata>(
                    new arrow::KeyValueMetadata());
                metadata->Append(DIM_KEY, std::to_string(dim));
                return arrow::schema(
                    {arrow::field("val", arrow::binary(), nullable, metadata)});
            }
            return arrow::schema(
                {arrow::field("val",
                              arrow::fixed_size_binary(dim * sizeof(int8)),
                              nullable)});
        }
        case DataType::VECTOR_ARRAY: {
            // VectorArray should not use this overload - should call the one with element_type
            ThrowInfo(
                NotImplemented,
                "VectorArray requires element_type parameter. Use "
                "CreateArrowSchema(data_type, dim, element_type, nullable)");
        }
        default: {
            ThrowInfo(
                DataTypeInvalid, "unsupported vector data type {}", data_type);
        }
    }
}

std::shared_ptr<arrow::Schema>
CreateArrowSchema(DataType data_type,
                  int dim,
                  DataType element_type,
                  bool nullable) {
    AssertInfo(data_type == DataType::VECTOR_ARRAY,
               "This overload is only for VECTOR_ARRAY type");
    AssertInfo(dim > 0, "invalid dim value");

    auto value_type = GetArrowDataTypeForVectorArray(element_type, dim);
    auto metadata = arrow::KeyValueMetadata::Make(
        {ELEMENT_TYPE_KEY_FOR_ARROW, DIM_KEY},
        {std::to_string(static_cast<int>(element_type)), std::to_string(dim)});

    auto field =
        arrow::field("val", value_type, nullable)->WithMetadata(metadata);
    return arrow::schema({field});
}

int
GetDimensionFromFileMetaData(const parquet::ColumnDescriptor* schema,
                             DataType data_type) {
    switch (data_type) {
        case DataType::VECTOR_FLOAT: {
            return schema->type_length() / sizeof(float);
        }
        case DataType::VECTOR_BINARY: {
            return schema->type_length() * 8;
        }
        case DataType::VECTOR_FLOAT16: {
            return schema->type_length() / sizeof(float16);
        }
        case DataType::VECTOR_BFLOAT16: {
            return schema->type_length() / sizeof(bfloat16);
        }
        case DataType::VECTOR_SPARSE_U32_F32: {
            ThrowInfo(DataTypeInvalid,
                      fmt::format("GetDimensionFromFileMetaData should not be "
                                  "called for sparse vector"));
        }
        case DataType::VECTOR_INT8: {
            return schema->type_length() / sizeof(int8);
        }
        default:
            ThrowInfo(DataTypeInvalid, "unsupported data type {}", data_type);
    }
}

std::string
GenIndexPathIdentifier(int64_t build_id,
                       int64_t index_version,
                       int64_t segment_id,
                       int64_t field_id) {
    return fmt::format(
        "{}_{}_{}_{}/", build_id, index_version, segment_id, field_id);
}

std::string
GenIndexPathPrefix(ChunkManagerPtr cm,
                   int64_t build_id,
                   int64_t index_version,
                   int64_t segment_id,
                   int64_t field_id,
                   bool is_temp) {
    return GenIndexPathPrefixByType(cm,
                                    build_id,
                                    index_version,
                                    segment_id,
                                    field_id,
                                    INDEX_ROOT_PATH,
                                    is_temp);
}

std::string
GenIndexPathPrefixByType(ChunkManagerPtr cm,
                         int64_t build_id,
                         int64_t index_version,
                         int64_t segment_id,
                         int64_t field_id,
                         const std::string& index_type,
                         bool is_temp) {
    boost::filesystem::path prefix = cm->GetRootPath();

    if (is_temp) {
        prefix = prefix / TEMP;
    }

    boost::filesystem::path path = std::string(index_type);
    boost::filesystem::path path1 =
        GenIndexPathIdentifier(build_id, index_version, segment_id, field_id);
    return NormalizePath(prefix / path / path1);
}

std::string
GenTextIndexPathPrefix(ChunkManagerPtr cm,
                       int64_t build_id,
                       int64_t index_version,
                       int64_t segment_id,
                       int64_t field_id,
                       bool is_temp) {
    return GenIndexPathPrefixByType(cm,
                                    build_id,
                                    index_version,
                                    segment_id,
                                    field_id,
                                    TEXT_LOG_ROOT_PATH,
                                    is_temp);
}

std::string
GenJsonStatsPathPrefix(ChunkManagerPtr cm,
                       int64_t build_id,
                       int64_t index_version,
                       int64_t segment_id,
                       int64_t field_id,
                       bool is_temp) {
    boost::filesystem::path prefix = cm->GetRootPath();

    if (is_temp) {
        prefix = prefix / TEMP;
    }

    boost::filesystem::path path = std::string(JSON_STATS_ROOT_PATH);
    boost::filesystem::path path1 =
        GenIndexPathIdentifier(build_id, index_version, segment_id, field_id);

    return NormalizePath(prefix / path / path1);
}

std::string
GenJsonStatsPathIdentifier(int64_t build_id,
                           int64_t index_version,
                           int64_t collection_id,
                           int64_t partition_id,
                           int64_t segment_id,
                           int64_t field_id) {
    boost::filesystem::path p =
        boost::filesystem::path(std::to_string(build_id)) /
        std::to_string(index_version) / std::to_string(collection_id) /
        std::to_string(partition_id) / std::to_string(segment_id) /
        std::to_string(field_id);
    return NormalizePath(p);
}

std::string
GenRemoteJsonStatsPathPrefix(ChunkManagerPtr cm,
                             int64_t build_id,
                             int64_t index_version,
                             int64_t collection_id,
                             int64_t partition_id,
                             int64_t segment_id,
                             int64_t field_id) {
    boost::filesystem::path p = cm->GetRootPath();
    p /= std::string(JSON_STATS_ROOT_PATH);
    p /= std::string(JSON_STATS_DATA_FORMAT_VERSION);
    p /= GenJsonStatsPathIdentifier(build_id,
                                    index_version,
                                    collection_id,
                                    partition_id,
                                    segment_id,
                                    field_id);
    return NormalizePath(p);
}

std::string
GenNgramIndexPrefix(ChunkManagerPtr cm,
                    int64_t build_id,
                    int64_t index_version,
                    int64_t segment_id,
                    int64_t field_id,
                    bool is_temp) {
    return GenIndexPathPrefixByType(cm,
                                    build_id,
                                    index_version,
                                    segment_id,
                                    field_id,
                                    NGRAM_LOG_ROOT_PATH,
                                    is_temp);
}

std::string
GenFieldRawDataPathPrefix(ChunkManagerPtr cm,
                          int64_t segment_id,
                          int64_t field_id) {
    boost::filesystem::path prefix = cm->GetRootPath();
    boost::filesystem::path path = std::string(RAWDATA_ROOT_PATH);
    boost::filesystem::path path1 = fmt::format("{}_{}/", segment_id, field_id);
    return NormalizePath(prefix / path / path1);
}

std::pair<std::string, size_t>
EncodeAndUploadIndexSlice(ChunkManager* chunk_manager,
                          uint8_t* buf,
                          int64_t batch_size,
                          IndexMeta index_meta,
                          FieldDataMeta field_meta,
                          std::string object_key,
                          std::shared_ptr<CPluginContext> plugin_context) {
    std::shared_ptr<IndexData> index_data = nullptr;
    if (index_meta.index_non_encoding) {
        index_data = std::make_shared<IndexData>(buf, batch_size);
        // index-build tasks assigned from new milvus-coord nodes to none-encoding
    } else {
        auto field_data =
            CreateFieldData(DataType::INT8, DataType::NONE, false);
        field_data->FillFieldData(buf, batch_size);
        auto payload_reader = std::make_shared<PayloadReader>(field_data);
        index_data = std::make_shared<IndexData>(payload_reader);
        // index-build tasks assigned from old milvus-coord nodes, fallback to int8 encoding
    }
    // index not use valid_data, so no need to set nullable==true
    index_data->set_index_meta(index_meta);
    index_data->SetFieldDataMeta(field_meta);
    auto serialized_index_data =
        index_data->serialize_to_remote_file(plugin_context);
    auto serialized_index_size = serialized_index_data.size();
    chunk_manager->Write(
        object_key, serialized_index_data.data(), serialized_index_size);
    return std::make_pair(std::move(object_key), serialized_index_size);
}

std::vector<std::future<std::unique_ptr<DataCodec>>>
GetObjectData(ChunkManager* remote_chunk_manager,
              const std::vector<std::string>& remote_files,
              milvus::ThreadPoolPriority priority,
              bool is_field_data) {
    auto& pool = ThreadPools::GetThreadPool(priority);
    std::vector<std::future<std::unique_ptr<DataCodec>>> futures;
    futures.reserve(remote_files.size());

    auto DownloadAndDeserialize = [](ChunkManager* chunk_manager,
                                     bool is_field_data,
                                     const std::string file) {
        // TODO remove this Size() cost
        auto fileSize = chunk_manager->Size(file);
        auto buf = std::shared_ptr<uint8_t[]>(new uint8_t[fileSize]);
        chunk_manager->Read(file, buf.get(), fileSize);
        auto res = DeserializeFileData(buf, fileSize, is_field_data);
        return res;
    };

    for (auto& file : remote_files) {
        futures.emplace_back(pool.Submit(
            DownloadAndDeserialize, remote_chunk_manager, is_field_data, file));
    }
    return futures;
}

std::map<std::string, int64_t>
PutIndexData(ChunkManager* remote_chunk_manager,
             const std::vector<const uint8_t*>& data_slices,
             const std::vector<int64_t>& slice_sizes,
             const std::vector<std::string>& slice_names,
             FieldDataMeta& field_meta,
             IndexMeta& index_meta,
             std::shared_ptr<CPluginContext> plugin_context) {
    auto& pool = ThreadPools::GetThreadPool(milvus::ThreadPoolPriority::MIDDLE);
    std::vector<std::future<std::pair<std::string, size_t>>> futures;
    AssertInfo(data_slices.size() == slice_sizes.size(),
               "inconsistent data slices size {} with slice sizes {}",
               data_slices.size(),
               slice_sizes.size());
    AssertInfo(data_slices.size() == slice_names.size(),
               "inconsistent data slices size {} with slice names size {}",
               data_slices.size(),
               slice_names.size());

    for (int64_t i = 0; i < data_slices.size(); ++i) {
        futures.push_back(pool.Submit(EncodeAndUploadIndexSlice,
                                      remote_chunk_manager,
                                      const_cast<uint8_t*>(data_slices[i]),
                                      slice_sizes[i],
                                      index_meta,
                                      field_meta,
                                      slice_names[i],
                                      plugin_context));
    }

    std::map<std::string, int64_t> remote_paths_to_size;
    std::exception_ptr first_exception = nullptr;
    for (auto& future : futures) {
        try {
            auto res = future.get();
            remote_paths_to_size[res.first] = res.second;
        } catch (...) {
            if (!first_exception) {
                first_exception = std::current_exception();
            }
        }
    }
    ReleaseArrowUnused();
    if (first_exception) {
        std::rethrow_exception(first_exception);
    }

    return remote_paths_to_size;
}

int64_t
GetTotalNumRowsForFieldDatas(const std::vector<FieldDataPtr>& field_datas) {
    int64_t count = 0;
    for (auto& field_data : field_datas) {
        count += field_data->get_num_rows();
    }

    return count;
}

size_t
GetNumRowsForLoadInfo(const LoadFieldDataInfo& load_info) {
    if (load_info.field_infos.empty()) {
        return 0;
    }

    auto& field = load_info.field_infos.begin()->second;
    return field.row_count;
}

void
ReleaseArrowUnused() {
    static std::mutex release_mutex;

    // While multiple threads are releasing memory,
    // we don't need everyone do releasing,
    // just let some of them do this also works well
    if (release_mutex.try_lock()) {
        arrow::default_memory_pool()->ReleaseUnused();
        release_mutex.unlock();
    }
}

ChunkManagerPtr
CreateChunkManager(const StorageConfig& storage_config) {
    auto storage_type = ChunkManagerType_Map[storage_config.storage_type];

    switch (storage_type) {
        case ChunkManagerType::Local: {
            return std::make_shared<LocalChunkManager>(
                storage_config.root_path);
        }
        case ChunkManagerType::Minio: {
            return std::make_shared<MinioChunkManager>(storage_config);
        }
        case ChunkManagerType::Remote: {
            auto cloud_provider_type =
                CloudProviderType_Map[storage_config.cloud_provider];
            switch (cloud_provider_type) {
                case CloudProviderType::AWS: {
                    return std::make_shared<AwsChunkManager>(storage_config);
                }
                case CloudProviderType::GCP: {
                    return std::make_shared<GcpChunkManager>(storage_config);
                }
                case CloudProviderType::ALIYUN: {
                    return std::make_shared<AliyunChunkManager>(storage_config);
                }
                case CloudProviderType::TENCENTCLOUD: {
                    return std::make_shared<TencentCloudChunkManager>(
                        storage_config);
                }
                case CloudProviderType::HUAWEICLOUD: {
                    return std::make_shared<HuaweiCloudChunkManager>(
                        storage_config);
                }
#ifdef AZURE_BUILD_DIR
                case CloudProviderType::AZURE: {
                    return std::make_shared<AzureChunkManager>(storage_config);
                }
#endif
#ifdef ENABLE_GCP_NATIVE
                case CloudProviderType::GCPNATIVE: {
                    return std::make_shared<GcpNativeChunkManager>(
                        storage_config);
                }
#endif
                default: {
                    return std::make_shared<MinioChunkManager>(storage_config);
                }
            }
        }
        default: {
            ThrowInfo(ConfigInvalid,
                      "unsupported storage_config.storage_type {}",
                      fmt::underlying(storage_type));
        }
    }
}

milvus_storage::ArrowFileSystemPtr
InitArrowFileSystem(milvus::storage::StorageConfig storage_config) {
    StorageV2FSCache::Key conf;
    if (storage_config.storage_type == "local") {
        std::string path(storage_config.root_path);
        conf.root_path = path;
        conf.storage_type = "local";
    } else {
        conf.address = std::string(storage_config.address);
        conf.bucket_name = std::string(storage_config.bucket_name);
        conf.access_key_id = std::string(storage_config.access_key_id);
        conf.access_key_value = std::string(storage_config.access_key_value);
        conf.root_path = std::string(storage_config.root_path);
        conf.storage_type = std::string(storage_config.storage_type);
        conf.cloud_provider = std::string(storage_config.cloud_provider);
        conf.iam_endpoint = std::string(storage_config.iam_endpoint);
        conf.log_level = std::string(storage_config.log_level);
        conf.region = std::string(storage_config.region);
        conf.useSSL = storage_config.useSSL;
        conf.sslCACert = std::string(storage_config.sslCACert);
        conf.useIAM = storage_config.useIAM;
        conf.useVirtualHost = storage_config.useVirtualHost;
        conf.requestTimeoutMs = storage_config.requestTimeoutMs;
        conf.gcp_credential_json =
            std::string(storage_config.gcp_credential_json);
        conf.use_custom_part_upload = true;
        conf.max_connections = storage_config.max_connections;
        conf.tls_min_version = storage_config.tls_min_version;
        conf.use_crc32c_checksum = storage_config.use_crc32c_checksum;
    }
    return StorageV2FSCache::Instance().Get(conf);
}

FieldDataPtr
CreateFieldData(const DataType& type,
                const DataType& element_type,
                bool nullable,
                int64_t dim,
                int64_t total_num_rows) {
    switch (type) {
        case DataType::BOOL:
            return std::make_shared<FieldData<bool>>(
                type, nullable, total_num_rows);
        case DataType::INT8:
            return std::make_shared<FieldData<int8_t>>(
                type, nullable, total_num_rows);
        case DataType::INT16:
            return std::make_shared<FieldData<int16_t>>(
                type, nullable, total_num_rows);
        case DataType::INT32:
            return std::make_shared<FieldData<int32_t>>(
                type, nullable, total_num_rows);
        case DataType::INT64:
            return std::make_shared<FieldData<int64_t>>(
                type, nullable, total_num_rows);
        case DataType::FLOAT:
            return std::make_shared<FieldData<float>>(
                type, nullable, total_num_rows);
        case DataType::DOUBLE:
            return std::make_shared<FieldData<double>>(
                type, nullable, total_num_rows);
        case DataType::TIMESTAMPTZ:
            return std::make_shared<FieldData<int64_t>>(
                type, nullable, total_num_rows);
        case DataType::STRING:
        case DataType::VARCHAR:
        case DataType::TEXT:
            return std::make_shared<FieldData<std::string>>(
                type, nullable, total_num_rows);
        case DataType::JSON:
            return std::make_shared<FieldData<Json>>(
                type, nullable, total_num_rows);
        case DataType::GEOMETRY:
            return std::make_shared<FieldData<Geometry>>(
                type, nullable, total_num_rows);
        case DataType::ARRAY:
            return std::make_shared<FieldData<Array>>(
                type, nullable, total_num_rows);
        case DataType::VECTOR_FLOAT:
            return std::make_shared<FieldData<FloatVector>>(
                dim, type, nullable, total_num_rows);
        case DataType::VECTOR_BINARY:
            return std::make_shared<FieldData<BinaryVector>>(
                dim, type, nullable, total_num_rows);
        case DataType::VECTOR_FLOAT16:
            return std::make_shared<FieldData<Float16Vector>>(
                dim, type, nullable, total_num_rows);
        case DataType::VECTOR_BFLOAT16:
            return std::make_shared<FieldData<BFloat16Vector>>(
                dim, type, nullable, total_num_rows);
        case DataType::VECTOR_SPARSE_U32_F32:
            return std::make_shared<FieldData<SparseFloatVector>>(
                type, nullable, total_num_rows);
        case DataType::VECTOR_INT8:
            return std::make_shared<FieldData<Int8Vector>>(
                dim, type, nullable, total_num_rows);
        case DataType::VECTOR_ARRAY:
            return std::make_shared<FieldData<VectorArray>>(
                dim, element_type, nullable, total_num_rows);
        default:
            ThrowInfo(DataTypeInvalid,
                      "CreateFieldData not support data type " +
                          GetDataTypeName(type));
    }
}

int64_t
GetByteSizeOfFieldDatas(const std::vector<FieldDataPtr>& field_datas) {
    int64_t result = 0;
    for (auto& data : field_datas) {
        result += data->Size();
    }

    return result;
}

std::vector<FieldDataPtr>
CollectFieldDataChannel(FieldDataChannelPtr& channel) {
    std::vector<FieldDataPtr> result;
    FieldDataPtr field_data;
    while (channel->pop(field_data)) {
        result.push_back(field_data);
    }
    return result;
}

FieldDataPtr
MergeFieldData(std::vector<FieldDataPtr>& data_array) {
    if (data_array.size() == 0) {
        return nullptr;
    }

    if (data_array.size() == 1) {
        return data_array[0];
    }

    size_t total_length = 0;
    for (const auto& data : data_array) {
        total_length += data->Length();
    }

    auto element_type = DataType::NONE;
    auto vector_array_data =
        dynamic_cast<FieldData<VectorArray>*>(data_array[0].get());
    if (vector_array_data) {
        element_type = vector_array_data->get_element_type();
    }

    auto merged_data = storage::CreateFieldData(data_array[0]->get_data_type(),
                                                element_type,
                                                data_array[0]->IsNullable());
    merged_data->Reserve(total_length);
    for (const auto& data : data_array) {
        if (merged_data->IsNullable()) {
            merged_data->FillFieldData(
                data->Data(), data->ValidData(), data->Length(), 0);
        } else {
            merged_data->FillFieldData(data->Data(), data->Length());
        }
    }
    return merged_data;
}

std::vector<FieldDataPtr>
FetchFieldData(ChunkManager* cm, const std::vector<std::string>& remote_files) {
    std::vector<FieldDataPtr> field_datas;
    std::vector<std::string> batch_files;
    auto FetchRawData = [&]() {
        auto fds = GetObjectData(cm, batch_files);
        ProcessFuturesInOrder(fds, [&](std::unique_ptr<DataCodec> codec) {
            field_datas.emplace_back(codec->GetFieldData());
        });
    };

    auto parallel_degree =
        uint64_t(DEFAULT_FIELD_MAX_MEMORY_LIMIT / FILE_SLICE_SIZE);

    for (auto& file : remote_files) {
        if (batch_files.size() >= parallel_degree) {
            FetchRawData();
            batch_files.clear();
        }
        batch_files.emplace_back(file);
    }
    if (batch_files.size() > 0) {
        FetchRawData();
    }
    return field_datas;
}

std::vector<FieldDataPtr>
GetFieldDatasFromStorageV2(std::vector<std::vector<std::string>>& remote_files,
                           int64_t field_id,
                           DataType data_type,
                           DataType element_type,
                           int64_t dim,
                           milvus_storage::ArrowFileSystemPtr fs) {
    AssertInfo(remote_files.size() > 0, "[StorageV2] remote files size is 0");
    std::vector<FieldDataPtr> field_data_list;

    // remote files might not followed the sequence of column group id,
    // so we need to put into map<column_group_id, remote_chunk_files>
    std::unordered_map<int64_t, std::vector<std::string>> column_group_files;
    for (auto& remote_chunk_files : remote_files) {
        AssertInfo(remote_chunk_files.size() > 0,
                   "[StorageV2] remote files size is 0");
        int64_t group_id = ExtractGroupIdFromPath(remote_chunk_files[0]);
        column_group_files[group_id] = remote_chunk_files;
    }

    std::vector<std::string> remote_chunk_files;
    int64_t column_group_id = -1;
    size_t col_offset = -1;
    if (column_group_files.find(field_id) == column_group_files.end()) {
        for (auto& [group_id, files] : column_group_files) {
            if (group_id >= START_USER_FIELDID) {
                continue;
            }
            milvus_storage::FieldIDList field_id_list = storage::GetFieldIDList(
                FieldId(group_id), files[0], nullptr, fs);
            for (size_t i = 0; i < field_id_list.size(); ++i) {
                if (field_id_list.Get(i) == field_id) {
                    remote_chunk_files = files;
                    column_group_id = group_id;
                    col_offset = i;
                    break;
                }
            }
            if (column_group_id != -1) {
                break;
            }
        }
    } else {
        remote_chunk_files = column_group_files[field_id];
        column_group_id = field_id;
    }

    if (column_group_id == -1) {
        LOG_INFO(
            "[StorageV2] field {} not found in any column group, return "
            "empty result set",
            field_id);
        return field_data_list;
    }
    AssertInfo(remote_chunk_files.size() > 0,
               "[StorageV2] remote files size is 0");

    // find column offset
    if (col_offset == -1) {
        milvus_storage::FieldIDList field_id_list = storage::GetFieldIDList(
            FieldId(column_group_id), remote_chunk_files[0], nullptr, fs);

        for (size_t i = 0; i < field_id_list.size(); ++i) {
            if (field_id_list.Get(i) == field_id) {
                col_offset = i;
                break;
            }
        }
    }
    // field not found, must be newly added field, return empty resultset
    if (col_offset == -1) {
        LOG_INFO(
            "[StorageV2] field {} not found in column group {}, return empty "
            "result set",
            field_id,
            column_group_id);
        return field_data_list;
    }

    AssertInfo(fs != nullptr,
               "[StorageV2] storage v2 arrow file system is not initialized");

    // set up channel for arrow reader
    auto field_data_info = FieldDataInfo();
    auto parallel_degree =
        static_cast<uint64_t>(DEFAULT_FIELD_MAX_MEMORY_LIMIT / FILE_SLICE_SIZE);
    field_data_info.arrow_reader_channel->set_capacity(parallel_degree);

    auto& pool = ThreadPools::GetThreadPool(milvus::ThreadPoolPriority::MIDDLE);

    for (auto& column_group_file : remote_chunk_files) {
        // get all row groups for each file
        std::vector<std::vector<int64_t>> row_group_lists;
        auto result = milvus_storage::FileRowGroupReader::Make(
            fs,
            column_group_file,
            milvus_storage::DEFAULT_READ_BUFFER_SIZE,
            GetReaderProperties(),
            GetArrowReaderProperties());
        AssertInfo(result.ok(),
                   "[StorageV2] Failed to create file row group reader: " +
                       result.status().ToString());
        auto reader = result.ValueOrDie();

        auto row_group_num =
            reader->file_metadata()->GetRowGroupMetadataVector().size();
        std::vector<int64_t> all_row_groups(row_group_num);
        std::iota(all_row_groups.begin(), all_row_groups.end(), 0);
        row_group_lists.push_back(all_row_groups);

        // create a schema with only the field id
        auto field_schema = reader->schema()->field(col_offset)->Copy();
        auto arrow_schema = arrow::schema({field_schema});
        auto status = reader->Close();
        AssertInfo(status.ok(),
                   "[StorageV2] failed to close file reader when get arrow "
                   "schema from file: " +
                       column_group_file + " with error: " + status.ToString());

        // split row groups for parallel reading
        auto strategy = std::make_unique<segcore::ParallelDegreeSplitStrategy>(
            parallel_degree);
        auto load_future = pool.Submit([&]() {
            return LoadWithStrategy(std::vector<std::string>{column_group_file},
                                    field_data_info.arrow_reader_channel,
                                    DEFAULT_FIELD_MAX_MEMORY_LIMIT,
                                    std::move(strategy),
                                    row_group_lists,
                                    fs,
                                    nullptr,
                                    milvus::proto::common::LoadPriority::HIGH);
        });
        // read field data from channel
        try {
            std::shared_ptr<milvus::ArrowDataWrapper> r;
            while (field_data_info.arrow_reader_channel->pop(r)) {
                size_t num_rows = 0;
                std::vector<std::shared_ptr<arrow::ChunkedArray>>
                    chunked_arrays;
                for (const auto& table_info : r->arrow_tables) {
                    num_rows += table_info.table->num_rows();
                    chunked_arrays.push_back(
                        table_info.table->column(col_offset));
                }
                auto field_data =
                    storage::CreateFieldData(data_type,
                                             element_type,
                                             field_schema->nullable(),
                                             dim,
                                             num_rows);
                for (const auto& chunked_array : chunked_arrays) {
                    field_data->FillFieldData(chunked_array);
                }
                field_data_list.push_back(field_data);
            }
        } catch (...) {
            // The load task captures this frame by reference and may be
            // blocked pushing into the bounded channel. Unblock it, then
            // wait for it to finish before unwinding (see #46958).
            try {
                std::shared_ptr<milvus::ArrowDataWrapper> discard;
                while (field_data_info.arrow_reader_channel->pop(discard)) {
                }
            } catch (...) {
            }
            DrainFuture(load_future);
            throw;
        }
        // access underlying feature to get exception if any
        load_future.get();
    }
    return field_data_list;
}

std::vector<FieldDataPtr>
GetFieldDatasFromManifest(
    const std::string& manifest_path,
    const std::shared_ptr<milvus_storage::api::Properties>& loon_ffi_properties,
    const FieldDataMeta& field_meta,
    std::optional<DataType> data_type,
    int64_t dim,
    std::optional<DataType> element_type,
    std::optional<StorageColumnMapping> storage_column_mapping) {
    auto loon_manifest = GetLoonManifest(manifest_path, loon_ffi_properties);
    auto column_groups = std::make_shared<milvus_storage::api::ColumnGroups>(
        loon_manifest->columnGroups());

    // Determine the column name to use. New index-build callers pass the
    // storage column mapping through FileManagerContext. The fallback keeps
    // direct test callers and older internal paths on the existing behavior.
    std::string column_name;
    const auto& ext_field = field_meta.field_schema.external_field();
    bool is_external = false;
    if (storage_column_mapping.has_value()) {
        column_name = storage_column_mapping->storage_column_name;
        is_external = storage_column_mapping->is_external_column;
    } else if (!ext_field.empty()) {
        column_name = ext_field;
        is_external = true;
    } else {
        column_name = std::to_string(field_meta.field_id);
    }
    if (column_name.empty()) {
        column_name = std::to_string(field_meta.field_id);
    }

    // TODO remove manual check after loon support read null for non-exists field
    auto column_exists = [&](const std::string& name) {
        for (size_t i = 0; i < column_groups->size(); i++) {
            for (const auto& column : column_groups->at(i)->columns) {
                if (column == name) {
                    return true;
                }
            }
        }
        return false;
    };
    bool field_exists = column_exists(column_name);
    if (!field_exists) {
        return {};
    }

    std::vector<std::string> needed_columns = {column_name};

    bool nullable = field_meta.field_schema.nullable();
    std::optional<FieldMeta> normalize_field_meta;
    if (is_external) {
        auto schema = field_meta.field_schema;
        if (schema.fieldid() == 0) {
            schema.set_fieldid(field_meta.field_id);
        }
        normalize_field_meta.emplace(FieldMeta::ParseFrom(schema));
    }

    // External tables: schemaless reader - let the reader derive types from
    // file metadata, then normalize Arrow types to Milvus internal format.
    // Internal tables: explicit schema reader - the reader returns data in
    // the exact Milvus-native Arrow types, no normalization needed.
    std::shared_ptr<arrow::Schema> reader_schema = nullptr;
    if (!is_external) {
        std::shared_ptr<arrow::Schema> arrow_schema;
        if (IsVectorDataType(data_type.value())) {
            if (data_type.value() == DataType::VECTOR_ARRAY) {
                arrow_schema = CreateArrowSchema(data_type.value(),
                                                 static_cast<int>(dim),
                                                 element_type.value(),
                                                 nullable);
            } else if (IsSparseFloatVectorDataType(data_type.value())) {
                arrow_schema = CreateArrowSchema(data_type.value(), nullable);
            } else {
                arrow_schema = CreateArrowSchema(
                    data_type.value(), static_cast<int>(dim), nullable);
            }
        } else if (data_type.value() == DataType::ARRAY) {
            arrow_schema = CreateArrowSchema(data_type.value(), nullable);
        } else {
            arrow_schema = CreateArrowSchema(data_type.value(), nullable);
        }
        reader_schema = std::make_shared<arrow::Schema>(
            arrow::Schema({arrow_schema->field(0)->WithName(column_name)}));
    }

    auto needed_cols_ptr =
        std::make_shared<std::vector<std::string>>(needed_columns);
    auto reader = milvus_storage::api::Reader::create(
        column_groups, reader_schema, needed_cols_ptr, *loon_ffi_properties);

    AssertInfo(reader != nullptr, "Failed to create reader");

    auto reader_result = reader->get_record_batch_reader("");
    AssertInfo(reader_result.ok(),
               "Failed to get record batch reader: " +
                   reader_result.status().ToString());

    auto record_batch_reader = reader_result.ValueOrDie();

    std::vector<FieldDataPtr> field_datas;
    while (true) {
        std::shared_ptr<arrow::RecordBatch> batch;
        auto status = record_batch_reader->ReadNext(&batch);
        AssertInfo(status.ok(),
                   "Failed to read record batch: " + status.ToString());
        if (batch == nullptr) {
            break;
        }

        auto num_rows = batch->num_rows();
        if (num_rows == 0) {
            continue;
        }

        auto raw_column = batch->GetColumnByName(column_name);
        if (is_external) {
            raw_column =
                NormalizeExternalArrowByType(raw_column,
                                             data_type.value(),
                                             dim,
                                             nullable,
                                             element_type.value(),
                                             normalize_field_meta.value());
        }
        auto chunked_array = std::make_shared<arrow::ChunkedArray>(raw_column);
        auto field_data = CreateFieldData(
            data_type.value(), element_type.value(), nullable, dim, num_rows);
        field_data->FillFieldData(chunked_array);
        field_datas.push_back(field_data);
    }

    return field_datas;
}

static bool
ManifestContainsColumn(
    const std::shared_ptr<milvus_storage::api::Manifest>& manifest,
    const std::string& column_name) {
    const auto& column_groups = manifest->columnGroups();
    for (size_t i = 0; i < column_groups.size(); i++) {
        for (const auto& column : column_groups.at(i)->columns) {
            if (column == column_name) {
                return true;
            }
        }
    }
    return false;
}

static std::string
GetSegmentBasePathFromManifestPath(const std::string& manifest_path,
                                   int64_t field_id) {
    try {
        auto j = nlohmann::json::parse(manifest_path);
        return j.at("base_path").get<std::string>();
    } catch (const std::exception& e) {
        ThrowInfo(ErrorCode::UnexpectedError,
                  "Failed to parse manifest path for TEXT field {}: {}",
                  field_id,
                  e.what());
    }
    return {};
}

static std::string
BuildTextLobBasePath(const std::string& segment_base_path, int64_t field_id) {
    // segment_base_path format: {root}/{collectionID}/{partitionID}/{segmentID}
    // lob_base_path format: {root}/{collectionID}/{partitionID}/lobs/{field_id}
    std::filesystem::path segment_fs_path(segment_base_path);
    auto partition_path = segment_fs_path.parent_path();
    return (partition_path / "lobs" / std::to_string(field_id)).string();
}

static std::shared_ptr<arrow::Schema>
BuildTextFieldReaderSchema(const FieldDataMeta& field_meta,
                           const std::string& column_name) {
    auto metadata =
        arrow::key_value_metadata({milvus_storage::ARROW_FIELD_ID_KEY},
                                  {std::to_string(field_meta.field_id)});
    auto field = arrow::field(column_name,
                              arrow::utf8(),
                              field_meta.field_schema.nullable(),
                              metadata);
    return arrow::schema({field});
}

static milvus_storage::segment::SegmentReaderConfig
BuildTextSegmentReaderConfig(const milvus_storage::api::Properties& properties,
                             int64_t field_id,
                             const std::string& lob_base_path) {
    milvus_storage::segment::SegmentReaderConfig config;
    config.properties = properties;

    milvus_storage::lob_column::LobColumnConfig lob_config;
    lob_config.data_type = milvus_storage::lob_column::LobDataType::kText;
    lob_config.field_id = field_id;
    lob_config.lob_base_path = lob_base_path;
    lob_config.properties = properties;
    config.lob_columns[field_id] = std::move(lob_config);
    return config;
}

static std::unique_ptr<milvus_storage::segment::SegmentReader>
OpenTextFieldSegmentReader(
    const std::string& manifest_path,
    const std::shared_ptr<milvus_storage::api::Properties>& loon_ffi_properties,
    const std::shared_ptr<milvus_storage::api::Manifest>& loon_manifest,
    const FieldDataMeta& field_meta,
    const std::string& column_name) {
    auto segment_base_path =
        GetSegmentBasePathFromManifestPath(manifest_path, field_meta.field_id);
    auto lob_base_path =
        BuildTextLobBasePath(segment_base_path, field_meta.field_id);

    auto fs_result = milvus_storage::FilesystemCache::getInstance().get(
        *loon_ffi_properties, segment_base_path);
    AssertInfo(fs_result.ok(),
               "Failed to get filesystem for TEXT field {}: {}",
               field_meta.field_id,
               fs_result.status().ToString());
    auto fs = std::move(fs_result.ValueOrDie());

    auto schema = BuildTextFieldReaderSchema(field_meta, column_name);
    auto config = BuildTextSegmentReaderConfig(
        *loon_ffi_properties, field_meta.field_id, lob_base_path);

    auto reader_result = milvus_storage::segment::SegmentReader::Open(
        fs, loon_manifest, schema, {column_name}, config);
    AssertInfo(reader_result.ok(),
               "Failed to open SegmentReader for TEXT field {}: {}",
               field_meta.field_id,
               reader_result.status().ToString());
    return std::move(reader_result.ValueOrDie());
}

std::vector<FieldDataPtr>
GetTextFieldDatasFromManifest(
    const std::string& manifest_path,
    const std::shared_ptr<milvus_storage::api::Properties>& loon_ffi_properties,
    const FieldDataMeta& field_meta) {
    AssertInfo(loon_ffi_properties != nullptr,
               "loon ffi properties is null when read text field data from "
               "manifest");

    auto loon_manifest = GetLoonManifest(manifest_path, loon_ffi_properties);

    std::string column_name = std::to_string(field_meta.field_id);

    if (!ManifestContainsColumn(loon_manifest, column_name)) {
        LOG_INFO("TEXT field {} not found in manifest", field_meta.field_id);
        return {};
    }

    auto reader = OpenTextFieldSegmentReader(manifest_path,
                                             loon_ffi_properties,
                                             loon_manifest,
                                             field_meta,
                                             column_name);

    std::vector<FieldDataPtr> field_datas;
    while (true) {
        std::shared_ptr<arrow::RecordBatch> batch;
        auto status = reader->ReadNext(&batch);
        AssertInfo(status.ok(),
                   "Failed to read TEXT field {} from manifest: {}",
                   field_meta.field_id,
                   status.ToString());
        if (batch == nullptr) {
            break;
        }

        auto num_rows = batch->num_rows();
        if (num_rows == 0) {
            continue;
        }

        auto raw_column = batch->GetColumnByName(column_name);
        AssertInfo(raw_column != nullptr,
                   "TEXT field {} column {} not found in SegmentReader batch",
                   field_meta.field_id,
                   column_name);
        auto chunked_array = std::make_shared<arrow::ChunkedArray>(raw_column);
        auto field_data = CreateFieldData(DataType::TEXT,
                                          DataType::NONE,
                                          field_meta.field_schema.nullable(),
                                          1,
                                          num_rows);
        field_data->FillFieldData(chunked_array);
        field_datas.push_back(field_data);
    }

    auto status = reader->Close();
    AssertInfo(status.ok(),
               "Failed to close SegmentReader for TEXT field {}: {}",
               field_meta.field_id,
               status.ToString());

    return field_datas;
}

std::vector<FieldDataPtr>
CacheRawDataAndFillMissing(const MemFileManagerImplPtr& file_manager,
                           const Config& config) {
    // download field data
    auto field_datas = file_manager->CacheRawDataToMemory(config);

    int64_t lack_binlog_rows =
        index::GetValueFromConfig<int64_t>(config, INDEX_NUM_ROWS_KEY)
            .value_or(0);
    for (auto& field_data : field_datas) {
        lack_binlog_rows -= field_data->get_num_rows();
    }

    if (lack_binlog_rows > 0) {
        LOG_INFO("create index lack binlog detected, lack row num: {}",
                 lack_binlog_rows);
        auto field_schema = file_manager->GetFieldDataMeta().field_schema;
        auto default_value = [&]() -> std::optional<DefaultValueType> {
            if (!field_schema.has_default_value()) {
                return std::nullopt;
            }
            return field_schema.default_value();
        }();
        auto field_data = storage::CreateFieldData(
            static_cast<DataType>(field_schema.data_type()),
            static_cast<DataType>(field_schema.element_type()),
            true,
            1,
            lack_binlog_rows);
        field_data->FillFieldData(default_value, lack_binlog_rows);
        field_datas.insert(field_datas.begin(), field_data);
    }

    return field_datas;
}

int64_t
ExtractGroupIdFromPath(const std::string& path) {
    // find second last of / to get group_id
    size_t last_slash = path.find_last_of("/");
    size_t second_last_slash = path.find_last_of("/", last_slash - 1);
    return std::stol(
        path.substr(second_last_slash + 1, last_slash - second_last_slash - 1));
}

// if it is multi-field column group, read field id list from file metadata
// if it is single-field column group, return the column group id
milvus_storage::FieldIDList
GetFieldIDList(FieldId column_group_id,
               const std::string& filepath,
               const std::shared_ptr<arrow::Schema>& arrow_schema,
               milvus_storage::ArrowFileSystemPtr fs) {
    milvus_storage::FieldIDList field_id_list;
    if (column_group_id >= FieldId(START_USER_FIELDID)) {
        field_id_list.Add(column_group_id.get());
        return field_id_list;
    }
    auto result = milvus_storage::FileRowGroupReader::Make(
        fs,
        filepath,
        arrow_schema,
        milvus_storage::DEFAULT_READ_BUFFER_SIZE,
        GetReaderProperties(),
        GetArrowReaderProperties());
    AssertInfo(result.ok(),
               "[StorageV2] Failed to create file row group reader: " +
                   result.status().ToString());
    auto file_reader = result.ValueOrDie();
    field_id_list =
        file_reader->file_metadata()->GetGroupFieldIDList().GetFieldIDList(
            column_group_id.get());
    auto status = file_reader->Close();
    AssertInfo(status.ok(),
               "failed to close file reader when get field id list from {}",
               filepath);
    return field_id_list;
}

std::string
FieldErrorSuffix(const FieldMeta& field_meta) {
    return fmt::format(" for field '{}' (external_field '{}')",
                       field_meta.get_name().get(),
                       field_meta.get_external_field());
}

void
ValidateFixedSizeBinaryVectorWidth(const std::shared_ptr<arrow::Array>& array,
                                   DataType data_type,
                                   int dim,
                                   const FieldMeta& field_meta) {
    auto fsb_array =
        std::static_pointer_cast<arrow::FixedSizeBinaryArray>(array);
    int byte_width = GetDataTypeSize(data_type, dim);
    AssertInfo(fsb_array->byte_width() == byte_width,
               "vector byte width mismatch{}, expected {} bytes for "
               "dim {}, actual {} bytes",
               FieldErrorSuffix(field_meta),
               byte_width,
               dim,
               fsb_array->byte_width());
}

void
ValidateBinaryVectorWidth(const std::shared_ptr<arrow::Array>& array,
                          DataType data_type,
                          int dim,
                          const FieldMeta& field_meta) {
    auto binary_array = std::static_pointer_cast<arrow::BinaryArray>(array);
    int byte_width = GetDataTypeSize(data_type, dim);
    for (int64_t i = 0; i < binary_array->length(); ++i) {
        if (binary_array->IsNull(i)) {
            continue;
        }
        auto actual_width = binary_array->value_length(i);
        AssertInfo(actual_width == byte_width,
                   "vector byte width mismatch{}, expected {} bytes for "
                   "dim {}, actual {} bytes at row {}",
                   FieldErrorSuffix(field_meta),
                   byte_width,
                   dim,
                   actual_width,
                   i);
    }
}

void
ValidateNoNullValuesInRange(const std::shared_ptr<arrow::Array>& values,
                            int64_t begin,
                            int64_t end,
                            const char* context) {
    if (values->null_count() == 0) {
        return;
    }
    for (int64_t i = begin; i < end; ++i) {
        AssertInfo(values->IsValid(i),
                   "{} contains null element at child offset {}",
                   context,
                   i);
    }
}

bool
IsByteVectorListInput(DataType data_type) {
    return data_type == DataType::VECTOR_BINARY ||
           data_type == DataType::VECTOR_BFLOAT16;
}

// External function output vectors can round-trip through schemaless Arrow
// paths that erase the semantic element type and surface the column as a raw
// list<uint8> (the physical Milvus vector bytes). In that case the uint8 bytes
// already match our on-disk storage layout, so we accept them as raw bytes
// instead of demanding the semantic element type (e.g. FLOAT for FloatVector).
//
// Restrictions:
//   - Only dense vector columns qualify (IsVectorDataType).
//   - Sparse float vectors are excluded: they are not fixed-width and do not
//     use the raw fixed-size-binary layout.
//   - VECTOR_ARRAY is excluded: its inner vectors are normalized recursively by
//     the per-element semantic type, not as a flat byte list at this level.
bool
CanTreatVectorListAsRawBytes(
    DataType data_type, const std::shared_ptr<arrow::DataType>& actual_type) {
    return actual_type->id() == arrow::Type::UINT8 &&
           IsVectorDataType(data_type) &&
           !IsSparseFloatVectorDataType(data_type) &&
           !IsVectorArrayDataType(data_type);
}

arrow::Type::type
ExpectedVectorListElementArrowType(DataType data_type,
                                   const FieldMeta& field_meta) {
    // This only validates vector columns encoded as List/FixedSizeList input.
    // FixedSizeBinary and nullable Binary inputs bypass this helper and are
    // validated by their byte-width checks instead.
    switch (data_type) {
        case DataType::VECTOR_FLOAT:
            return arrow::Type::FLOAT;
        case DataType::VECTOR_INT8:
            return arrow::Type::INT8;
        case DataType::VECTOR_FLOAT16:
            return arrow::Type::HALF_FLOAT;
        case DataType::VECTOR_BINARY:
        case DataType::VECTOR_BFLOAT16:
            return arrow::Type::UINT8;
        default:
            ThrowInfo(ErrorCode::Unsupported,
                      "unsupported vector list input{} for {}",
                      FieldErrorSuffix(field_meta),
                      data_type);
    }
}

const char*
ArrowTypeName(arrow::Type::type type) {
    switch (type) {
        case arrow::Type::FLOAT:
            return "float";
        case arrow::Type::INT8:
            return "int8";
        case arrow::Type::HALF_FLOAT:
            return "halffloat";
        case arrow::Type::UINT8:
            return "uint8";
        default:
            return "unsupported";
    }
}

int
ExpectedVectorListLength(DataType data_type,
                         int dim,
                         const std::shared_ptr<arrow::DataType>& actual_type) {
    // Float-like vector lists are element-counted by dim. Byte vector lists are
    // raw-byte encoded, so their list length must match the physical byte width.
    // The same byte-width rule applies when a semantic vector column arrives as
    // a raw uint8 byte list (CanTreatVectorListAsRawBytes), hence actual_type is
    // needed here to disambiguate float-element lists from raw-byte lists.
    if (IsByteVectorListInput(data_type) ||
        CanTreatVectorListAsRawBytes(data_type, actual_type)) {
        return GetDataTypeSize(data_type, dim);
    }
    return dim;
}

void
ValidateVectorListElementType(
    const std::shared_ptr<arrow::DataType>& actual_type,
    DataType data_type,
    const FieldMeta& field_meta) {
    auto expected_type =
        ExpectedVectorListElementArrowType(data_type, field_meta);
    // Accept either the semantic element type (e.g. FLOAT for FloatVector) or a
    // raw uint8 byte list carrying the physical vector bytes. The list-length
    // check downstream still enforces the correct byte width, so mismatched
    // non-uint8 element types remain rejected below.
    if (actual_type->id() == expected_type ||
        CanTreatVectorListAsRawBytes(data_type, actual_type)) {
        return;
    }
    AssertInfo(false,
               "vector element type mismatch{}, expected {} or raw uint8 "
               "bytes, actual {}",
               FieldErrorSuffix(field_meta),
               ArrowTypeName(expected_type),
               actual_type->ToString());
}

arrow::ArrayVector
NormalizeVectorArraysToFixedSizeBinary(const arrow::ArrayVector& arrays,
                                       DataType data_type,
                                       int dim,
                                       const FieldMeta& field_meta) {
    int byte_width = GetDataTypeSize(data_type, dim);
    auto fsb_type = arrow::fixed_size_binary(byte_width);

    arrow::ArrayVector result;
    result.reserve(arrays.size());

    for (const auto& array : arrays) {
        auto type_id = array->type_id();
        if (type_id == arrow::Type::FIXED_SIZE_BINARY) {
            ValidateFixedSizeBinaryVectorWidth(
                array, data_type, dim, field_meta);
            result.push_back(array);
            continue;
        }

        int64_t num_rows = array->length();
        auto buffer_result = arrow::AllocateBuffer(num_rows * byte_width);
        AssertInfo(buffer_result.ok(),
                   "Failed to allocate buffer for vector normalization");
        auto buffer = std::move(*buffer_result);
        auto dst = buffer->mutable_data();
        // Zero-fill for null rows (nullable vectors from external Parquet)
        if (array->null_count() > 0) {
            memset(dst, 0, num_rows * byte_width);
        }

        if (type_id == arrow::Type::LIST) {
            auto list_array = std::static_pointer_cast<arrow::ListArray>(array);
            auto values = list_array->values();
            ValidateVectorListElementType(
                values->type(), data_type, field_meta);
            // Computed per array: the expected length depends on the actual
            // element type (dim elements for float lists, byte-width for raw
            // uint8 byte lists), so it cannot be hoisted out of the loop.
            int expected_list_length =
                ExpectedVectorListLength(data_type, dim, values->type());
            int elem_bit_width = values->type()->bit_width();
            AssertInfo(elem_bit_width > 0 && elem_bit_width % 8 == 0,
                       "vector list element{} must be fixed-width "
                       "byte-aligned type, got bit_width={}",
                       FieldErrorSuffix(field_meta),
                       elem_bit_width);
            int elem_byte_size = elem_bit_width / 8;
            auto raw = reinterpret_cast<const uint8_t*>(
                values->data()->buffers[1]->data());
            for (int64_t i = 0; i < num_rows; i++) {
                if (array->IsValid(i)) {
                    auto offset = list_array->value_offset(i);
                    auto actual_length =
                        list_array->value_offset(i + 1) - offset;
                    AssertInfo(actual_length == expected_list_length,
                               "vector list length mismatch{}, expected {}, "
                               "actual {} at row {}",
                               FieldErrorSuffix(field_meta),
                               expected_list_length,
                               actual_length,
                               i);
                    ValidateNoNullValuesInRange(
                        values, offset, offset + actual_length, "vector list");
                    milvus::fastmem::FastMemcpy(dst + i * byte_width,
                                                raw + offset * elem_byte_size,
                                                byte_width);
                }
            }
        } else if (type_id == arrow::Type::FIXED_SIZE_LIST) {
            auto fsl_array =
                std::static_pointer_cast<arrow::FixedSizeListArray>(array);
            auto values = fsl_array->values();
            ValidateVectorListElementType(
                values->type(), data_type, field_meta);
            // Computed per array: the expected length depends on the actual
            // element type (dim elements for float lists, byte-width for raw
            // uint8 byte lists), so it cannot be hoisted out of the loop.
            int expected_list_length =
                ExpectedVectorListLength(data_type, dim, values->type());
            int elem_bit_width = values->type()->bit_width();
            AssertInfo(elem_bit_width > 0 && elem_bit_width % 8 == 0,
                       "vector list element{} must be fixed-width "
                       "byte-aligned type, got bit_width={}",
                       FieldErrorSuffix(field_meta),
                       elem_bit_width);
            int elem_byte_size = elem_bit_width / 8;
            AssertInfo(fsl_array->value_length() == expected_list_length,
                       "vector list length mismatch{}, expected {}, actual {}",
                       FieldErrorSuffix(field_meta),
                       expected_list_length,
                       fsl_array->value_length());
            auto raw = reinterpret_cast<const uint8_t*>(
                values->data()->buffers[1]->data());
            for (int64_t i = 0; i < num_rows; i++) {
                if (array->IsValid(i)) {
                    auto offset = fsl_array->value_offset(i);
                    ValidateNoNullValuesInRange(values,
                                                offset,
                                                offset + expected_list_length,
                                                "vector list");
                    milvus::fastmem::FastMemcpy(dst + i * byte_width,
                                                raw + offset * elem_byte_size,
                                                byte_width);
                }
            }
        } else {
            ThrowInfo(ErrorCode::Unsupported,
                      "unsupported arrow type for vector normalization{}: {}",
                      FieldErrorSuffix(field_meta),
                      array->type()->ToString());
        }

        // Preserve null bitmap from the source array
        std::shared_ptr<arrow::Buffer> null_bitmap;
        if (array->null_count() > 0 && array->data()->buffers[0]) {
            null_bitmap = array->data()->buffers[0];
        }
        auto fsb_data = arrow::ArrayData::Make(fsb_type,
                                               num_rows,
                                               {null_bitmap, std::move(buffer)},
                                               array->null_count());
        result.push_back(
            std::make_shared<arrow::FixedSizeBinaryArray>(fsb_data));
    }
    return result;
}

// ============================================================
// Arrow type normalization helpers for external table loading
// ============================================================

arrow::ArrayVector
ConvertFixedSizeBinaryToBinary(const arrow::ArrayVector& arrays) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        if (arr->type_id() == arrow::Type::BINARY) {
            result.push_back(arr);
            continue;
        }
        auto fsb = std::static_pointer_cast<arrow::FixedSizeBinaryArray>(arr);
        int byte_width = fsb->byte_width();
        int64_t n = fsb->length();

        // Build offsets: null rows occupy no data space
        auto offsets_result = arrow::AllocateBuffer((n + 1) * sizeof(int32_t));
        AssertInfo(offsets_result.ok(), "Failed to allocate offsets buffer");
        auto offsets_buf = std::move(*offsets_result);
        auto* offsets = reinterpret_cast<int32_t*>(offsets_buf->mutable_data());
        int32_t offset = 0;
        offsets[0] = 0;
        for (int64_t i = 0; i < n; i++) {
            offset += fsb->IsValid(i) ? byte_width : 0;
            offsets[i + 1] = offset;
        }

        // Build data: copy only valid rows
        auto data_result = arrow::AllocateBuffer(offset);
        AssertInfo(data_result.ok(), "Failed to allocate data buffer");
        auto data_buf = std::move(*data_result);
        auto* dst = data_buf->mutable_data();
        for (int64_t i = 0; i < n; i++) {
            if (fsb->IsValid(i)) {
                milvus::fastmem::FastMemcpy(dst, fsb->Value(i), byte_width);
                dst += byte_width;
            }
        }

        // Reuse the original null bitmap
        auto bin_data = arrow::ArrayData::Make(arrow::binary(),
                                               n,
                                               {fsb->data()->buffers[0],
                                                std::move(offsets_buf),
                                                std::move(data_buf)},
                                               fsb->null_count(),
                                               /*offset=*/0);
        result.push_back(std::make_shared<arrow::BinaryArray>(bin_data));
    }
    return result;
}

arrow::ArrayVector
ConvertStringArrayToBinary(const arrow::ArrayVector& arrays) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        if (arr->type_id() != arrow::Type::STRING) {
            result.push_back(arr);
            continue;
        }
        // Zero-copy: StringArray and BinaryArray share identical layout
        auto d = arr->data();
        auto bin = arrow::ArrayData::Make(
            arrow::binary(), d->length, d->buffers, d->null_count, d->offset);
        result.push_back(std::make_shared<arrow::BinaryArray>(bin));
    }
    return result;
}

// Coerce any binary-like Arrow array to canonical BinaryArray.
// Source readers (e.g. vortex) may produce LARGE_BINARY / BINARY_VIEW /
// LARGE_STRING / STRING_VIEW for the same logical bytes; downstream
// ChunkWriters dynamic_cast to BinaryArray and would null-deref.
arrow::ArrayVector
CoerceToBinary(const arrow::ArrayVector& arrays) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        const auto tid = arr->type_id();
        if (tid == arrow::Type::BINARY) {
            result.push_back(arr);
            continue;
        }
        if (tid == arrow::Type::STRING) {
            // Zero-copy: identical buffer layout
            auto d = arr->data();
            auto bin = arrow::ArrayData::Make(arrow::binary(),
                                              d->length,
                                              d->buffers,
                                              d->null_count,
                                              d->offset);
            result.push_back(std::make_shared<arrow::BinaryArray>(bin));
            continue;
        }
        if (tid == arrow::Type::LARGE_BINARY ||
            tid == arrow::Type::LARGE_STRING ||
            tid == arrow::Type::BINARY_VIEW ||
            tid == arrow::Type::STRING_VIEW) {
            // Different offset/buffer layout -> rebuild via builder.
            arrow::BinaryBuilder builder;
            auto status = builder.Reserve(arr->length());
            AssertInfo(status.ok(),
                       "BinaryBuilder reserve failed: " + status.ToString());
            switch (tid) {
                case arrow::Type::LARGE_BINARY: {
                    auto src =
                        std::static_pointer_cast<arrow::LargeBinaryArray>(arr);
                    for (int64_t i = 0; i < src->length(); ++i) {
                        if (src->IsNull(i)) {
                            status = builder.AppendNull();
                        } else {
                            auto v = src->GetView(i);
                            status = builder.Append(
                                reinterpret_cast<const uint8_t*>(v.data()),
                                v.size());
                        }
                        AssertInfo(status.ok(),
                                   "BinaryBuilder append failed: " +
                                       status.ToString());
                    }
                    break;
                }
                case arrow::Type::LARGE_STRING: {
                    auto src =
                        std::static_pointer_cast<arrow::LargeStringArray>(arr);
                    for (int64_t i = 0; i < src->length(); ++i) {
                        if (src->IsNull(i)) {
                            status = builder.AppendNull();
                        } else {
                            auto v = src->GetView(i);
                            status = builder.Append(
                                reinterpret_cast<const uint8_t*>(v.data()),
                                v.size());
                        }
                        AssertInfo(status.ok(),
                                   "BinaryBuilder append failed: " +
                                       status.ToString());
                    }
                    break;
                }
                case arrow::Type::BINARY_VIEW: {
                    auto src =
                        std::static_pointer_cast<arrow::BinaryViewArray>(arr);
                    for (int64_t i = 0; i < src->length(); ++i) {
                        if (src->IsNull(i)) {
                            status = builder.AppendNull();
                        } else {
                            auto v = src->GetView(i);
                            status = builder.Append(
                                reinterpret_cast<const uint8_t*>(v.data()),
                                v.size());
                        }
                        AssertInfo(status.ok(),
                                   "BinaryBuilder append failed: " +
                                       status.ToString());
                    }
                    break;
                }
                case arrow::Type::STRING_VIEW: {
                    auto src =
                        std::static_pointer_cast<arrow::StringViewArray>(arr);
                    for (int64_t i = 0; i < src->length(); ++i) {
                        if (src->IsNull(i)) {
                            status = builder.AppendNull();
                        } else {
                            auto v = src->GetView(i);
                            status = builder.Append(
                                reinterpret_cast<const uint8_t*>(v.data()),
                                v.size());
                        }
                        AssertInfo(status.ok(),
                                   "BinaryBuilder append failed: " +
                                       status.ToString());
                    }
                    break;
                }
                default:
                    break;
            }
            std::shared_ptr<arrow::Array> out;
            status = builder.Finish(&out);
            AssertInfo(status.ok(),
                       "BinaryBuilder finish failed: " + status.ToString());
            result.push_back(out);
            continue;
        }
        result.push_back(arr);
    }
    return result;
}

// Forward decl: defined alongside CanonicalizeArrowVariants below.
static std::shared_ptr<arrow::Buffer>
RebuildNullBitmap(const std::shared_ptr<arrow::Array>& array);

// Coerce LARGE_LIST / LIST_VIEW to canonical (32-bit offset) ListArray.
// Vortex schemaless mode may emit list variants for the same logical
// List<T>; downstream code (ConvertListToProtobufBinary,
// NormalizeVectorArrayInner, ArrowListToScalarFieldProto) expects
// arrow::ListArray and would static_cast-fail otherwise.
arrow::ArrayVector
CoerceToList(const arrow::ArrayVector& arrays) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        const auto tid = arr->type_id();
        if (tid != arrow::Type::LARGE_LIST && tid != arrow::Type::LIST_VIEW) {
            result.push_back(arr);
            continue;
        }

        std::shared_ptr<arrow::Array> values;
        std::vector<std::pair<int64_t, int64_t>> ranges;
        ranges.reserve(arr->length());
        if (tid == arrow::Type::LARGE_LIST) {
            auto la = std::static_pointer_cast<arrow::LargeListArray>(arr);
            values = la->values();
            for (int64_t i = 0; i < la->length(); ++i) {
                int64_t s = la->value_offset(i);
                int64_t e = s + la->value_length(i);
                ranges.emplace_back(s, e);
            }
        } else {
            auto lv = std::static_pointer_cast<arrow::ListViewArray>(arr);
            values = lv->values();
            for (int64_t i = 0; i < lv->length(); ++i) {
                int64_t s = lv->value_offset(i);
                int64_t e = s + lv->value_length(i);
                ranges.emplace_back(s, e);
            }
        }

        arrow::Int32Builder offset_builder;
        auto status = offset_builder.Reserve(arr->length() + 1);
        AssertInfo(status.ok(),
                   "CoerceToList: offset reserve failed: " + status.ToString());
        std::vector<std::shared_ptr<arrow::Array>> value_slices;
        int32_t cur = 0;
        status = offset_builder.Append(0);
        AssertInfo(status.ok(), "CoerceToList: offset append failed");
        for (int64_t i = 0; i < arr->length(); ++i) {
            if (!arr->IsNull(i)) {
                auto [s, e] = ranges[i];
                int64_t len = e - s;
                AssertInfo(cur + len <= INT32_MAX,
                           "CoerceToList: offset overflows int32");
                value_slices.push_back(values->Slice(s, len));
                cur += static_cast<int32_t>(len);
            }
            status = offset_builder.Append(cur);
            AssertInfo(status.ok(), "CoerceToList: offset append failed");
        }
        std::shared_ptr<arrow::Array> offsets_arr;
        status = offset_builder.Finish(&offsets_arr);
        AssertInfo(status.ok(),
                   "CoerceToList: offset finish failed: " + status.ToString());

        std::shared_ptr<arrow::Array> concat_values;
        if (value_slices.empty()) {
            auto empty = arrow::MakeArrayOfNull(values->type(), 0);
            AssertInfo(empty.ok(), "CoerceToList: empty values failed");
            concat_values = *empty;
        } else {
            auto concat = arrow::Concatenate(value_slices);
            AssertInfo(
                concat.ok(),
                "CoerceToList: concat failed: " + concat.status().ToString());
            concat_values = *concat;
        }

        auto offsets_buf =
            std::static_pointer_cast<arrow::Int32Array>(offsets_arr)->values();
        auto list_arr =
            std::make_shared<arrow::ListArray>(arrow::list(values->type()),
                                               arr->length(),
                                               offsets_buf,
                                               concat_values,
                                               RebuildNullBitmap(arr),
                                               arr->null_count(),
                                               0);
        result.push_back(list_arr);
    }
    return result;
}

// Rebuild null bitmap as a fresh buffer at offset=0.
// Required when constructing a new arrow::Array that does not share the
// input's logical offset - direct reuse of `array->null_bitmap()` is
// incorrect for sliced inputs (offset != 0) because callers will read the
// bitmap starting at bit 0 instead of bit `array->offset()`.
static std::shared_ptr<arrow::Buffer>
RebuildNullBitmap(const std::shared_ptr<arrow::Array>& array) {
    if (array->null_count() == 0) {
        return nullptr;
    }
    arrow::TypedBufferBuilder<bool> bb;
    auto status = bb.Reserve(array->length());
    AssertInfo(status.ok(),
               "RebuildNullBitmap: reserve failed: " + status.ToString());
    for (int64_t i = 0; i < array->length(); ++i) {
        bb.UnsafeAppend(!array->IsNull(i));
    }
    std::shared_ptr<arrow::Buffer> buf;
    status = bb.Finish(&buf);
    AssertInfo(status.ok(),
               "RebuildNullBitmap: finish failed: " + status.ToString());
    return buf;
}

// Single-source-of-truth view-variant elimination. Callers that ingest
// vortex (or any schemaless reader that may emit *_VIEW / LARGE_* layouts)
// route arrays through this helper before any type-id dispatch. Pure
// type/layout normalization - no semantic conversion (no WKT->WKB,
// no Timestamp->Int64, etc).
//
//   STRING_VIEW / LARGE_STRING -> STRING
//   BINARY_VIEW / LARGE_BINARY -> BINARY
//   LARGE_LIST  / LIST_VIEW    -> LIST  (recurses into inner values)
//   LIST                       -> LIST  (recurses into inner values)
//   FIXED_SIZE_LIST            -> FIXED_SIZE_LIST (recurses into inner values)
//   anything else              -> unchanged
std::shared_ptr<arrow::Array>
CanonicalizeArrowVariants(const std::shared_ptr<arrow::Array>& array) {
    const auto tid = array->type_id();

    // Variable-length string variants -> STRING
    if (tid == arrow::Type::STRING_VIEW || tid == arrow::Type::LARGE_STRING) {
        arrow::StringBuilder builder;
        auto status = builder.Reserve(array->length());
        AssertInfo(status.ok(),
                   "CanonicalizeArrowVariants: string reserve failed");
        if (tid == arrow::Type::STRING_VIEW) {
            auto src = std::static_pointer_cast<arrow::StringViewArray>(array);
            for (int64_t i = 0; i < src->length(); ++i) {
                if (src->IsNull(i)) {
                    AssertInfo(builder.AppendNull().ok(), "appendnull");
                } else {
                    auto v = src->GetView(i);
                    AssertInfo(builder.Append(v.data(), v.size()).ok(),
                               "append str");
                }
            }
        } else {
            auto src = std::static_pointer_cast<arrow::LargeStringArray>(array);
            for (int64_t i = 0; i < src->length(); ++i) {
                if (src->IsNull(i)) {
                    AssertInfo(builder.AppendNull().ok(), "appendnull");
                } else {
                    auto v = src->GetView(i);
                    AssertInfo(builder.Append(v.data(), v.size()).ok(),
                               "append str");
                }
            }
        }
        std::shared_ptr<arrow::Array> out;
        AssertInfo(builder.Finish(&out).ok(), "string finish");
        return out;
    }

    // Variable-length binary variants -> BINARY
    if (tid == arrow::Type::BINARY_VIEW || tid == arrow::Type::LARGE_BINARY) {
        arrow::BinaryBuilder builder;
        auto status = builder.Reserve(array->length());
        AssertInfo(status.ok(),
                   "CanonicalizeArrowVariants: binary reserve failed");
        if (tid == arrow::Type::BINARY_VIEW) {
            auto src = std::static_pointer_cast<arrow::BinaryViewArray>(array);
            for (int64_t i = 0; i < src->length(); ++i) {
                if (src->IsNull(i)) {
                    AssertInfo(builder.AppendNull().ok(), "appendnull");
                } else {
                    auto v = src->GetView(i);
                    AssertInfo(
                        builder
                            .Append(reinterpret_cast<const uint8_t*>(v.data()),
                                    v.size())
                            .ok(),
                        "append bin");
                }
            }
        } else {
            auto src = std::static_pointer_cast<arrow::LargeBinaryArray>(array);
            for (int64_t i = 0; i < src->length(); ++i) {
                if (src->IsNull(i)) {
                    AssertInfo(builder.AppendNull().ok(), "appendnull");
                } else {
                    auto v = src->GetView(i);
                    AssertInfo(
                        builder
                            .Append(reinterpret_cast<const uint8_t*>(v.data()),
                                    v.size())
                            .ok(),
                        "append bin");
                }
            }
        }
        std::shared_ptr<arrow::Array> out;
        AssertInfo(builder.Finish(&out).ok(), "binary finish");
        return out;
    }

    // List variants -> LIST (recursively canonicalize values)
    if (tid == arrow::Type::LARGE_LIST || tid == arrow::Type::LIST_VIEW ||
        tid == arrow::Type::LIST) {
        std::shared_ptr<arrow::Array> values;
        std::vector<std::pair<int64_t, int64_t>> ranges;
        ranges.reserve(array->length());
        if (tid == arrow::Type::LIST) {
            auto la = std::static_pointer_cast<arrow::ListArray>(array);
            values = la->values();
            for (int64_t i = 0; i < la->length(); ++i) {
                int64_t s = la->value_offset(i);
                ranges.emplace_back(s, s + la->value_length(i));
            }
        } else if (tid == arrow::Type::LARGE_LIST) {
            auto la = std::static_pointer_cast<arrow::LargeListArray>(array);
            values = la->values();
            for (int64_t i = 0; i < la->length(); ++i) {
                int64_t s = la->value_offset(i);
                ranges.emplace_back(s, s + la->value_length(i));
            }
        } else {
            auto lv = std::static_pointer_cast<arrow::ListViewArray>(array);
            values = lv->values();
            for (int64_t i = 0; i < lv->length(); ++i) {
                int64_t s = lv->value_offset(i);
                ranges.emplace_back(s, s + lv->value_length(i));
            }
        }

        auto canonical_values = CanonicalizeArrowVariants(values);

        // Fast path: already canonical LIST and inner unchanged -> return as-is
        if (tid == arrow::Type::LIST &&
            canonical_values.get() == values.get()) {
            return array;
        }

        // Rebuild offsets + slice canonical values per row
        arrow::Int32Builder offset_builder;
        AssertInfo(offset_builder.Reserve(array->length() + 1).ok(),
                   "offset reserve");
        std::vector<std::shared_ptr<arrow::Array>> slices;
        int32_t cur = 0;
        AssertInfo(offset_builder.Append(0).ok(), "offset append");
        for (int64_t i = 0; i < array->length(); ++i) {
            if (!array->IsNull(i)) {
                auto [s, e] = ranges[i];
                int64_t len = e - s;
                AssertInfo(cur + len <= INT32_MAX,
                           "CanonicalizeArrowVariants: int32 offset overflow");
                slices.push_back(canonical_values->Slice(s, len));
                cur += static_cast<int32_t>(len);
            }
            AssertInfo(offset_builder.Append(cur).ok(), "offset append");
        }
        std::shared_ptr<arrow::Array> offsets_arr;
        AssertInfo(offset_builder.Finish(&offsets_arr).ok(), "offset finish");

        std::shared_ptr<arrow::Array> concat_values;
        if (slices.empty()) {
            auto empty = arrow::MakeArrayOfNull(canonical_values->type(), 0);
            AssertInfo(empty.ok(), "empty values");
            concat_values = *empty;
        } else {
            auto concat = arrow::Concatenate(slices);
            AssertInfo(concat.ok(), "concat: " + concat.status().ToString());
            concat_values = *concat;
        }

        auto offsets_buf =
            std::static_pointer_cast<arrow::Int32Array>(offsets_arr)->values();
        return std::make_shared<arrow::ListArray>(
            arrow::list(canonical_values->type()),
            array->length(),
            offsets_buf,
            concat_values,
            RebuildNullBitmap(array),
            array->null_count(),
            0);
    }

    // FIXED_SIZE_LIST: recurse into values only (layout already canonical)
    if (tid == arrow::Type::FIXED_SIZE_LIST) {
        auto fsl = std::static_pointer_cast<arrow::FixedSizeListArray>(array);
        auto inner = fsl->values();
        auto canonical_inner = CanonicalizeArrowVariants(inner);
        if (canonical_inner.get() == inner.get()) {
            return array;
        }
        auto fsl_type =
            std::static_pointer_cast<arrow::FixedSizeListType>(fsl->type());
        return std::make_shared<arrow::FixedSizeListArray>(
            arrow::fixed_size_list(canonical_inner->type(),
                                   fsl_type->list_size()),
            fsl->length(),
            canonical_inner,
            RebuildNullBitmap(fsl),
            fsl->null_count(),
            0);
    }

    return array;
}

arrow::ArrayVector
ConvertWKTStringArrayToWKBBinary(const arrow::ArrayVector& arrays) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());

    GEOSContextHandle_t ctx = GEOS_init_r();
    AssertInfo(ctx != nullptr, "Failed to initialize GEOS context");

    for (const auto& arr : arrays) {
        const auto tid = arr->type_id();
        if (tid != arrow::Type::STRING && tid != arrow::Type::LARGE_STRING &&
            tid != arrow::Type::STRING_VIEW) {
            result.push_back(arr);
            continue;
        }
        auto get_string = [&](int64_t i) -> std::string {
            if (tid == arrow::Type::STRING) {
                return std::static_pointer_cast<arrow::StringArray>(arr)
                    ->GetString(i);
            }
            if (tid == arrow::Type::LARGE_STRING) {
                return std::static_pointer_cast<arrow::LargeStringArray>(arr)
                    ->GetString(i);
            }
            auto sv =
                std::static_pointer_cast<arrow::StringViewArray>(arr)->GetView(
                    i);
            return std::string(sv.data(), sv.size());
        };
        arrow::BinaryBuilder builder;
        auto status = builder.Reserve(arr->length());
        AssertInfo(status.ok(),
                   "BinaryBuilder reserve failed: " + status.ToString());
        for (int64_t i = 0; i < arr->length(); ++i) {
            if (arr->IsNull(i)) {
                status = builder.AppendNull();
                AssertInfo(status.ok(), "AppendNull failed");
                continue;
            }
            auto wkt = get_string(i);
            Geometry geom(ctx, wkt.c_str());
            auto wkb = geom.to_wkb_string();
            status = builder.Append(wkb);
            AssertInfo(status.ok(), "Append WKB failed");
        }
        std::shared_ptr<arrow::Array> wkb_array;
        status = builder.Finish(&wkb_array);
        AssertInfo(status.ok(), "BinaryBuilder finish failed");
        result.push_back(wkb_array);
    }

    GEOS_finish_r(ctx);
    return result;
}

arrow::ArrayVector
ConvertTimestampToInt64(const arrow::ArrayVector& arrays) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        if (arr->type_id() != arrow::Type::TIMESTAMP) {
            result.push_back(arr);
            continue;
        }
        auto ts_arr = std::static_pointer_cast<arrow::TimestampArray>(arr);
        auto ts_type =
            std::static_pointer_cast<arrow::TimestampType>(arr->type());
        auto unit = ts_type->unit();
        arrow::Int64Builder builder;
        auto status = builder.Reserve(ts_arr->length());
        AssertInfo(status.ok(), "Int64Builder reserve failed");
        for (int64_t i = 0; i < ts_arr->length(); i++) {
            if (ts_arr->IsNull(i)) {
                status = builder.AppendNull();
            } else {
                status = builder.Append(
                    ConvertToMicroseconds(ts_arr->Value(i), unit));
            }
            AssertInfo(status.ok(), "Int64Builder append failed");
        }
        std::shared_ptr<arrow::Array> int_array;
        status = builder.Finish(&int_array);
        AssertInfo(status.ok(), "Int64Builder finish failed");
        result.push_back(std::move(int_array));
    }
    return result;
}

DataType
ArrowListElementTypeToMilvus(const std::shared_ptr<arrow::Array>& values,
                             const FieldMeta& field_meta) {
    switch (values->type_id()) {
        case arrow::Type::BOOL:
            return DataType::BOOL;
        case arrow::Type::INT8:
            return DataType::INT8;
        case arrow::Type::INT16:
            return DataType::INT16;
        case arrow::Type::INT32:
            return DataType::INT32;
        case arrow::Type::INT64:
            return DataType::INT64;
        case arrow::Type::FLOAT:
            return DataType::FLOAT;
        case arrow::Type::DOUBLE:
            return DataType::DOUBLE;
        case arrow::Type::STRING:
        case arrow::Type::LARGE_STRING:
        case arrow::Type::STRING_VIEW:
            return DataType::STRING;
        default:
            ThrowInfo(ErrorCode::Unsupported,
                      "unsupported array element arrow type{}: {}",
                      FieldErrorSuffix(field_meta),
                      values->type()->ToString());
    }
}

bool
IsCompatibleArrayElementType(DataType actual_type, DataType expected_type) {
    if (actual_type == expected_type) {
        return true;
    }
    return actual_type == DataType::STRING && IsStringDataType(expected_type);
}

arrow::ArrayVector
ConvertListToProtobufBinary(const arrow::ArrayVector& arrays,
                            DataType element_type,
                            const FieldMeta& field_meta) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        if (arr->type_id() != arrow::Type::LIST) {
            result.push_back(arr);
            continue;
        }
        auto list_arr = std::static_pointer_cast<arrow::ListArray>(arr);
        auto actual_element_type =
            ArrowListElementTypeToMilvus(list_arr->values(), field_meta);
        AssertInfo(
            IsCompatibleArrayElementType(actual_element_type, element_type),
            "array element type mismatch{}, expected {}, actual {}",
            FieldErrorSuffix(field_meta),
            element_type,
            actual_element_type);
        arrow::BinaryBuilder builder;
        auto status = builder.Reserve(list_arr->length());
        AssertInfo(status.ok(), "BinaryBuilder reserve failed");
        for (int64_t i = 0; i < list_arr->length(); i++) {
            if (list_arr->IsNull(i)) {
                status = builder.AppendNull();
            } else {
                auto start = list_arr->value_offset(i);
                auto end = list_arr->value_offset(i + 1);
                ValidateNoNullValuesInRange(
                    list_arr->values(), start, end, "array list");
                auto proto = ArrowListToScalarFieldProto(list_arr, i);
                std::string serialized;
                proto.SerializeToString(&serialized);
                status = builder.Append(serialized);
            }
            AssertInfo(status.ok(), "BinaryBuilder append failed");
        }
        std::shared_ptr<arrow::Array> bin_array;
        status = builder.Finish(&bin_array);
        AssertInfo(status.ok(), "BinaryBuilder finish failed");
        result.push_back(std::move(bin_array));
    }
    return result;
}

arrow::ArrayVector
NormalizeVectorArrays(const arrow::ArrayVector& arrays,
                      DataType data_type,
                      int64_t dim,
                      bool nullable,
                      const FieldMeta& field_meta) {
    if (arrays.empty()) {
        return arrays;
    }

    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& array : arrays) {
        auto type_id = array->type_id();
        if (type_id == arrow::Type::BINARY && nullable) {
            ValidateBinaryVectorWidth(
                array, data_type, static_cast<int>(dim), field_meta);
            result.push_back(array);
            continue;
        }

        arrow::ArrayVector fsb;
        if (type_id == arrow::Type::FIXED_SIZE_BINARY) {
            ValidateFixedSizeBinaryVectorWidth(
                array, data_type, static_cast<int>(dim), field_meta);
            fsb.push_back(array);
        } else {
            fsb = NormalizeVectorArraysToFixedSizeBinary(
                {array}, data_type, static_cast<int>(dim), field_meta);
        }

        if (nullable) {
            auto binary = ConvertFixedSizeBinaryToBinary(fsb);
            result.push_back(binary[0]);
        } else {
            result.push_back(fsb[0]);
        }
    }
    return result;
}

arrow::ArrayVector
NormalizeVectorArrayInner(const arrow::ArrayVector& arrays,
                          DataType element_type,
                          int64_t dim,
                          const FieldMeta& field_meta) {
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        if (arr->type_id() != arrow::Type::LIST) {
            result.push_back(arr);
            continue;
        }
        auto list_arr = std::static_pointer_cast<arrow::ListArray>(arr);
        AssertInfo(field_meta.is_nullable() || list_arr->null_count() == 0,
                   "VECTOR_ARRAY does not support null rows");
        if (list_arr->values()->type_id() == arrow::Type::FIXED_SIZE_BINARY) {
            ValidateFixedSizeBinaryVectorWidth(list_arr->values(),
                                               element_type,
                                               static_cast<int>(dim),
                                               field_meta);
            result.push_back(arr);
            continue;
        }
        auto normalized =
            NormalizeVectorArraysToFixedSizeBinary({list_arr->values()},
                                                   element_type,
                                                   static_cast<int>(dim),
                                                   field_meta);
        auto rebuilt_data = arrow::ArrayData::Make(
            arrow::list(normalized[0]->type()),
            list_arr->length(),
            {list_arr->null_bitmap(), list_arr->value_offsets()},
            {normalized[0]->data()},
            list_arr->null_count(),
            list_arr->offset());
        result.push_back(arrow::MakeArray(rebuilt_data));
    }
    return result;
}

// --- Tier 2: Per-consumer entry points ---

// FieldMeta overload: extracts parameters and delegates.
std::shared_ptr<arrow::Array>
NormalizeExternalArrow(const std::shared_ptr<arrow::Array>& array,
                       const FieldMeta& field_meta) {
    auto dt = field_meta.get_data_type();
    int64_t dim = (IsVectorDataType(dt) && !IsSparseFloatVectorDataType(dt))
                      ? field_meta.get_dim()
                      : 0;
    auto element_type = IsVectorArrayDataType(dt) || IsArrayDataType(dt)
                            ? field_meta.get_element_type()
                            : DataType::NONE;
    return NormalizeExternalArrowByType(
        array, dt, dim, field_meta.is_nullable(), element_type, field_meta);
}

// Load path: batch wrapper.
arrow::ArrayVector
NormalizeArrowForChunkWriter(const arrow::ArrayVector& arrays,
                             const FieldMeta& field_meta) {
    if (arrays.empty()) {
        return arrays;
    }
    arrow::ArrayVector result;
    result.reserve(arrays.size());
    for (const auto& arr : arrays) {
        result.push_back(NormalizeExternalArrow(arr, field_meta));
    }
    return result;
}

std::shared_ptr<arrow::DataType>
ExpectedScalarArrowType(DataType data_type) {
    switch (data_type) {
        case DataType::BOOL:
            return arrow::boolean();
        case DataType::INT8:
            return arrow::int8();
        case DataType::INT16:
            return arrow::int16();
        case DataType::INT32:
            return arrow::int32();
        case DataType::INT64:
            return arrow::int64();
        case DataType::FLOAT:
            return arrow::float32();
        case DataType::DOUBLE:
            return arrow::float64();
        default:
            return nullptr;
    }
}

void
ValidateScalarArrowType(DataType data_type,
                        const std::shared_ptr<arrow::Array>& array,
                        const FieldMeta& field_meta) {
    auto expected_type = ExpectedScalarArrowType(data_type);
    if (expected_type == nullptr) {
        return;
    }
    AssertInfo(array->type()->Equals(*expected_type),
               "field type mismatch{}, expected Arrow {}, actual Arrow {}",
               FieldErrorSuffix(field_meta),
               expected_type->ToString(),
               array->type()->ToString());
}

void
AssertExternalArrowType(const std::shared_ptr<arrow::Array>& array,
                        const std::string& expected,
                        const FieldMeta& field_meta) {
    ThrowInfo(ErrorCode::Unsupported,
              "field type mismatch{}, expected Arrow {}, actual Arrow {}",
              FieldErrorSuffix(field_meta),
              expected,
              array->type()->ToString());
}

static std::shared_ptr<arrow::Array>
NormalizeExternalArrowByType(const std::shared_ptr<arrow::Array>& array_in,
                             DataType data_type,
                             int64_t dim,
                             bool nullable,
                             DataType element_type,
                             const FieldMeta& field_meta) {
    // Single view-variant elimination pass: STRING_VIEW/LARGE_STRING -> STRING,
    // BINARY_VIEW/LARGE_BINARY -> BINARY, LARGE_LIST/LIST_VIEW -> LIST
    // (recursive into list inner). Downstream branches only need to dispatch
    // on canonical type_ids.
    auto array = CanonicalizeArrowVariants(array_in);
    auto type_id = array->type_id();

    if (data_type == DataType::TIMESTAMPTZ) {
        if (type_id == arrow::Type::TIMESTAMP) {
            auto result = ConvertTimestampToInt64({array});
            return result[0];
        }
        if (type_id == arrow::Type::INT64) {
            return array;
        }
        AssertExternalArrowType(array, "timestamp or int64", field_meta);
    }

    ValidateScalarArrowType(data_type, array, field_meta);

    if (IsSparseFloatVectorDataType(data_type)) {
        if (type_id == arrow::Type::BINARY) {
            return array;
        }
        AssertExternalArrowType(array, "binary", field_meta);
    }

    // Dense vectors
    if (IsVectorDataType(data_type) &&
        !IsSparseFloatVectorDataType(data_type) &&
        !IsVectorArrayDataType(data_type)) {
        if (type_id == arrow::Type::FIXED_SIZE_BINARY && !nullable) {
            ValidateFixedSizeBinaryVectorWidth(
                array, data_type, static_cast<int>(dim), field_meta);
            return array;
        }
        if (type_id == arrow::Type::BINARY && nullable) {
            ValidateBinaryVectorWidth(
                array, data_type, static_cast<int>(dim), field_meta);
            return array;
        }
        auto result = NormalizeVectorArrays(
            {array}, data_type, dim, nullable, field_meta);
        return result[0];
    }
    // VectorArray inner: List<List<scalar>> -> List<FixedSizeBinary>
    if (IsVectorArrayDataType(data_type)) {
        AssertInfo(element_type != DataType::NONE,
                   "element_type must be specified for VECTOR_ARRAY");
        if (type_id == arrow::Type::LIST) {
            auto result = NormalizeVectorArrayInner(
                {array}, element_type, dim, field_meta);
            return result[0];
        }
        AssertExternalArrowType(array, "list", field_meta);
    }
    // Geometry STRING input -> WKT, convert to WKB.
    // (View/large-string variants already canonicalized to STRING above.)
    if (data_type == DataType::GEOMETRY && type_id == arrow::Type::STRING) {
        auto result = ConvertWKTStringArrayToWKBBinary({array});
        return result[0];
    }
    // Geometry BINARY input -> already WKB, no-op (issue #49353).
    if (data_type == DataType::GEOMETRY && type_id == arrow::Type::BINARY) {
        return array;
    }
    if (data_type == DataType::GEOMETRY) {
        AssertExternalArrowType(array, "string or binary", field_meta);
    }
    // JSON/VARCHAR/STRING/TEXT: String -> Binary (canonical internal repr).
    // (View/large variants already canonicalized to STRING above; issue #49352.)
    if ((IsStringDataType(data_type) || data_type == DataType::JSON) &&
        type_id == arrow::Type::STRING) {
        auto result = ConvertStringArrayToBinary({array});
        return result[0];
    }
    // JSON/VARCHAR/STRING/TEXT: BINARY input is already canonical.
    if ((IsStringDataType(data_type) || data_type == DataType::JSON) &&
        type_id == arrow::Type::BINARY) {
        return array;
    }
    if (IsStringDataType(data_type) || data_type == DataType::JSON) {
        AssertExternalArrowType(array, "string or binary", field_meta);
    }
    // Array: List -> Protobuf Binary
    if (data_type == DataType::ARRAY && type_id == arrow::Type::LIST) {
        auto result =
            ConvertListToProtobufBinary({array}, element_type, field_meta);
        return result[0];
    }
    if (data_type == DataType::ARRAY && type_id == arrow::Type::BINARY) {
        return array;
    }
    if (data_type == DataType::ARRAY) {
        AssertExternalArrowType(array, "list or binary", field_meta);
    }
    return array;
}

// ============================================================

proto::schema::ScalarField
ArrowListToScalarFieldProto(const std::shared_ptr<arrow::ListArray>& list_array,
                            int64_t row_index) {
    proto::schema::ScalarField sf;
    int64_t start = list_array->value_offset(row_index);
    int64_t end = list_array->value_offset(row_index + 1);
    auto values = list_array->values();

    switch (values->type_id()) {
        case arrow::Type::BOOL: {
            auto typed = std::static_pointer_cast<arrow::BooleanArray>(values);
            auto* d = sf.mutable_bool_data();
            for (int64_t j = start; j < end; j++) d->add_data(typed->Value(j));
            break;
        }
        case arrow::Type::INT8: {
            auto typed = std::static_pointer_cast<arrow::Int8Array>(values);
            auto* d = sf.mutable_int_data();
            for (int64_t j = start; j < end; j++)
                d->add_data(static_cast<int32_t>(typed->Value(j)));
            break;
        }
        case arrow::Type::INT16: {
            auto typed = std::static_pointer_cast<arrow::Int16Array>(values);
            auto* d = sf.mutable_int_data();
            for (int64_t j = start; j < end; j++)
                d->add_data(static_cast<int32_t>(typed->Value(j)));
            break;
        }
        case arrow::Type::INT32: {
            auto typed = std::static_pointer_cast<arrow::Int32Array>(values);
            auto* d = sf.mutable_int_data();
            for (int64_t j = start; j < end; j++) d->add_data(typed->Value(j));
            break;
        }
        case arrow::Type::INT64: {
            auto typed = std::static_pointer_cast<arrow::Int64Array>(values);
            auto* d = sf.mutable_long_data();
            for (int64_t j = start; j < end; j++) d->add_data(typed->Value(j));
            break;
        }
        case arrow::Type::FLOAT: {
            auto typed = std::static_pointer_cast<arrow::FloatArray>(values);
            auto* d = sf.mutable_float_data();
            for (int64_t j = start; j < end; j++) d->add_data(typed->Value(j));
            break;
        }
        case arrow::Type::DOUBLE: {
            auto typed = std::static_pointer_cast<arrow::DoubleArray>(values);
            auto* d = sf.mutable_double_data();
            for (int64_t j = start; j < end; j++) d->add_data(typed->Value(j));
            break;
        }
        case arrow::Type::STRING: {
            auto typed = std::static_pointer_cast<arrow::StringArray>(values);
            auto* d = sf.mutable_string_data();
            for (int64_t j = start; j < end; j++)
                d->add_data(typed->GetString(j));
            break;
        }
        case arrow::Type::LARGE_STRING: {
            auto typed =
                std::static_pointer_cast<arrow::LargeStringArray>(values);
            auto* d = sf.mutable_string_data();
            for (int64_t j = start; j < end; j++)
                d->add_data(typed->GetString(j));
            break;
        }
        case arrow::Type::STRING_VIEW: {
            auto typed =
                std::static_pointer_cast<arrow::StringViewArray>(values);
            auto* d = sf.mutable_string_data();
            for (int64_t j = start; j < end; j++) {
                auto sv = typed->GetView(j);
                d->add_data(std::string(sv.data(), sv.size()));
            }
            break;
        }
        default:
            ThrowInfo(
                ErrorCode::Unsupported,
                "Unsupported element type for ArrowListToScalarFieldProto: {}",
                values->type()->ToString());
    }
    return sf;
}

}  // namespace milvus::storage
