fix: support nullable vector array growing flush (#50540)

issue: #50536

## What changed

This fixes growing-source flush for nullable `VECTOR_ARRAY` fields.
`BuildVectorArrayForChunk` now preserves null logical rows as Arrow null
list entries and maps valid logical rows to compact physical offsets
before reading `ConcurrentVector<VectorArray>`.

A regression test covers nullable vector-array round-trip flushing with
`valid, null, valid, null` rows.

## Why

With StorageV3/Loon and growing-source flush enabled, dropping a
collection can flush growing segment data through
`FlushGrowingSegmentData`. Nullable StructArray ArrayOfVector subfields
can contain null logical rows, but the previous vector-array builder
rejected any `valid_data`, causing `VECTOR_ARRAY does not support null
rows` and a server panic during drop collection.

## Verification

- `git diff --check`
- `milvus-dev-cli cpp-ut local . -b master --allow-dirty --target
all_tests --case
FlushGrowingSegmentTest.FlushNullableVectorArrayRoundTrip -f`
  - UT ID: `ut-cpp-local-zhuwenxi-zhuwenxing-b-3e34efd-705e9e04`
  - Status: succeeded
- `milvus-dev-cli build local -b master --allow-dirty -f`
  - Build ID: `build-local-zhuwenxing-zhuwenxing-bug-m-3e34efd`
- Image:
`harbor.milvus.io/manta/milvus:local-zhuwenxing-zhuwenxing-bug-milvus-panics-when-dropping-nullable-structa-3e34efd`
- Deployed with:
  - `common.storage.useLoonFFI=true`
  - `common.storage.enableGrowingSourceFlush=true`
  - `--mq rocksmq`
- Ran the issue repro with `pymilvus==3.1.0rc39` from TestPyPI:
  - `drop_collection succeeded`
  - `has_collection_after_drop=False`
  - standalone pod restart count stayed `0`
- no `panic: flush growing source data` or `VECTOR_ARRAY does not
support null rows` in standalone logs

Signed-off-by: zhuwenxing <wenxing.zhu@zilliz.com>

internal/core/src/segcore/flush_growing_segment_test.cpp

@@ -1061,6 +1061,94 @@ TEST_F(FlushGrowingSegmentTest, FlushVectorArrayRoundTrip) {
     FreeFlushResult(&result);
 }
 
+TEST_F(FlushGrowingSegmentTest, FlushNullableVectorArrayRoundTrip) {
+    auto schema = std::make_shared<Schema>();
+    auto pk_fid = schema->AddDebugField("pk", DataType::INT64);
+    auto vec_array_fid = schema->AddDebugVectorArrayField(
+        "emb_list", DataType::VECTOR_FLOAT, 4, knowhere::metric::L2, true);
+    schema->set_primary_field_id(pk_fid);
+
+    auto segment = CreateGrowingSegment(schema, empty_index_meta);
+    ASSERT_NE(segment, nullptr);
+
+    constexpr int N = 4;
+    std::vector<int64_t> row_ids = {10, 11, 12, 13};
+    std::vector<Timestamp> timestamps = {100, 101, 102, 103};
+    std::vector<int64_t> pks = {1000, 1001, 1002, 1003};
+    std::vector<VectorFieldProto> vec_arrays(N);
+
+    auto fill_float_vectors = [](VectorFieldProto& vector_array,
+                                 std::initializer_list<float> values) {
+        vector_array.set_dim(4);
+        vector_array.mutable_float_vector()->mutable_data()->Add(values.begin(),
+                                                                 values.end());
+    };
+
+    fill_float_vectors(vec_arrays[0], {1.0F, 2.0F, 3.0F, 4.0F});
+    vec_arrays[1].set_dim(4);
+    vec_arrays[1].mutable_float_vector();
+    fill_float_vectors(vec_arrays[2],
+                       {5.0F, 6.0F, 7.0F, 8.0F, 9.0F, 10.0F, 11.0F, 12.0F});
+    vec_arrays[3].set_dim(4);
+    vec_arrays[3].mutable_float_vector();
+    bool valid_data[N] = {true, false, true, false};
+
+    auto insert_data = std::make_unique<InsertRecordProto>();
+    insert_data->set_num_rows(N);
+    insert_data->mutable_fields_data()->AddAllocated(
+        CreateDataArrayFrom(pks.data(), nullptr, N, (*schema)[pk_fid])
+            .release());
+    insert_data->mutable_fields_data()->AddAllocated(
+        CreateDataArrayFrom(
+            vec_arrays.data(), valid_data, N, (*schema)[vec_array_fid])
+            .release());
+
+    segment->PreInsert(N);
+    segment->Insert(0, N, row_ids.data(), timestamps.data(), insert_data.get());
+
+    CFlushConfig config{};
+    std::string segment_path = test_dir_ + "/segment_nullable_vector_array";
+    config.segment_path = segment_path.c_str();
+    config.read_version = -1;
+    config.retry_limit = 3;
+    config.text_field_ids = nullptr;
+    config.text_lob_paths = nullptr;
+    config.num_text_columns = 0;
+
+    CFlushResult result{};
+    auto status =
+        FlushGrowingSegmentData(segment.get(), 0, N, &config, &result);
+
+    ASSERT_EQ(status.error_code, Success) << status.error_msg;
+    ASSERT_EQ(result.num_rows, N);
+
+    auto field_datas = ReadFlushedFieldData(segment_path,
+                                            result,
+                                            vec_array_fid,
+                                            DataType::VECTOR_ARRAY,
+                                            true,
+                                            4,
+                                            DataType::VECTOR_FLOAT);
+    ASSERT_EQ(field_datas.size(), 1);
+    ASSERT_EQ(field_datas[0]->get_num_rows(), N);
+    ASSERT_EQ(field_datas[0]->get_valid_rows(), 2);
+    EXPECT_TRUE(field_datas[0]->is_valid(0));
+    EXPECT_FALSE(field_datas[0]->is_valid(1));
+    EXPECT_TRUE(field_datas[0]->is_valid(2));
+    EXPECT_FALSE(field_datas[0]->is_valid(3));
+
+    auto vector_array_data =
+        std::dynamic_pointer_cast<milvus::FieldData<milvus::VectorArray>>(
+            field_datas[0]);
+    ASSERT_NE(vector_array_data, nullptr);
+    ASSERT_EQ(vector_array_data->value_at(0)->output_data().SerializeAsString(),
+              vec_arrays[0].SerializeAsString());
+    ASSERT_EQ(vector_array_data->value_at(1)->output_data().SerializeAsString(),
+              vec_arrays[2].SerializeAsString());
+
+    FreeFlushResult(&result);
+}
+
 TEST_F(FlushGrowingSegmentTest, FlushVectorArrayElementTypesRoundTrip) {
     struct Case {
         DataType element_type;

internal/core/src/segcore/segment_c.cpp

@@ -1431,11 +1431,6 @@ arrow::Result<std::shared_ptr<arrow::Array>>
 BuildVectorArrayForChunk(const FieldInfo& field_info,
                          int64_t start_offset,
                          int64_t num_rows) {
-    if (field_info.valid_data) {
-        return arrow::Status::Invalid(
-            "VECTOR_ARRAY does not support null rows");
-    }
-
     auto vector_array_vec = dynamic_cast<
         const milvus::segcore::ConcurrentVector<milvus::VectorArray>*>(
         field_info.vec_base);
@@ -1451,7 +1446,21 @@ BuildVectorArrayForChunk(const FieldInfo& field_info,
     ARROW_RETURN_NOT_OK(builder.Reserve(num_rows));
 
     for (int64_t i = 0; i < num_rows; i++) {
-        const auto& vector_array = (*vector_array_vec)[start_offset + i];
+        auto logical_offset = start_offset + i;
+        if (field_info.valid_data &&
+            !field_info.valid_data->is_valid(logical_offset)) {
+            ARROW_RETURN_NOT_OK(builder.AppendNull());
+            continue;
+        }
+
+        auto physical_offset =
+            field_info.vec_base->get_physical_offset(logical_offset);
+        if (physical_offset < 0) {
+            return arrow::Status::Invalid(
+                "valid nullable vector array row missing physical data");
+        }
+
+        const auto& vector_array = (*vector_array_vec)[physical_offset];
         if (vector_array.get_element_type() != field_info.element_type) {
             return arrow::Status::Invalid("VECTOR_ARRAY element type mismatch");
         }