Add nullable vector support for segment search and indexing

This commit adds nullable vector support for both growing and sealed segments,
including search, retrieval, and index building operations. It introduces
OffsetMapping for efficient logical-to-physical offset translation.

Core infrastructure changes:
- Add OffsetMapping class with Build()/BuildIncremental() for bitmap-based
  offset mapping, supporting both dense lookup and efficient valid count tracking
- Add TransformBitset() and TransformOffset() utilities for converting between
  logical and physical coordinate spaces during search operations
- Update SearchResult to include has_raw_data_ flag and physical offset handling

Growing segment changes:
- Update ConcurrentVector to support nullable vectors with OffsetMapping
- Add get_offset_mapping() and valid data tracking in AckSeal()
- Update FieldIndexing to handle nullable vectors with physical offset storage
- Modify SearchOnGrowing to transform bitset/offsets for nullable vector search
- Add nullable vector tests in SegmentGrowingTest covering search and retrieval

Sealed segment changes:
- Update ChunkedColumn/ChunkedColumnGroup to support OffsetMapping
- Modify ChunkedSegmentSealedImpl to handle nullable vector field data loading
- Update SearchOnSealedIndex/SearchOnSealedColumn with early return for 100% null
  case to prevent crash when searching empty index
- Add FilterVectorValidOffsets() for output field retrieval with nullable vectors
- Expand ChunkedSegmentSealedBinlogIndexTest with comprehensive nullable test cases

Index building changes:
- Update VectorMemIndex/VectorDiskIndex to store and serialize OffsetMapping
- Add BuildValidData()/UpdateValidData() for nullable vector index building
- Update VecIndexCreator and index_c API with valid data parameter support
- Modify InterimSealedIndexTranslator to handle nullable vector binlog index

Search and reduce changes:
- Update Reduce.cpp to transform physical offsets back to logical for output
- Update StreamReduce with nullable vector offset transformation support
- Add early return in SearchOnSealed when offset_mapping has zero valid count

Signed-off-by: marcelo-cjl <[email protected]>

Author: marcelo-cjl

internal/core/src/common/Chunk.h

@@ -126,6 +126,11 @@ class Chunk {
         return data_;
     }
 
+    FixedVector<bool>&
+    Valid() {
+        return valid_;
+    }
+
     virtual bool
     isValid(int offset) const {
         if (nullable_) {
@@ -559,17 +564,32 @@ class SparseFloatVectorChunk : public Chunk {
                            bool nullable,
                            std::shared_ptr<ChunkMmapGuard> chunk_mmap_guard)
         : Chunk(row_nums, data, size, nullable, chunk_mmap_guard) {
-        vec_.resize(row_nums);
         auto null_bitmap_bytes_num = nullable ? (row_nums + 7) / 8 : 0;
         auto offsets_ptr =
             reinterpret_cast<uint64_t*>(data + null_bitmap_bytes_num);
-        for (int i = 0; i < row_nums; i++) {
-            vec_[i] = {(offsets_ptr[i + 1] - offsets_ptr[i]) /
-                           knowhere::sparse::SparseRow<
-                               SparseValueType>::element_size(),
-                       reinterpret_cast<uint8_t*>(data + offsets_ptr[i]),
-                       false};
-            dim_ = std::max(dim_, vec_[i].dim());
+
+        if (nullable_) {
+            for (int i = 0; i < row_nums; i++) {
+                if (isValid(i)) {
+                    vec_.emplace_back(
+                        (offsets_ptr[i + 1] - offsets_ptr[i]) /
+                            knowhere::sparse::SparseRow<
+                                SparseValueType>::element_size(),
+                        reinterpret_cast<uint8_t*>(data + offsets_ptr[i]),
+                        false);
+                    dim_ = std::max(dim_, vec_.back().dim());
+                }
+            }
+        } else {
+            vec_.resize(row_nums);
+            for (int i = 0; i < row_nums; i++) {
+                vec_[i] = {(offsets_ptr[i + 1] - offsets_ptr[i]) /
+                               knowhere::sparse::SparseRow<
+                                   SparseValueType>::element_size(),
+                           reinterpret_cast<uint8_t*>(data + offsets_ptr[i]),
+                           false};
+                dim_ = std::max(dim_, vec_[i].dim());
+            }
         }
     }
 

internal/core/src/common/ChunkWriter.cpp

@@ -435,8 +435,10 @@ SparseFloatVectorChunkWriter::calculate_size(
     for (const auto& data : array_vec) {
         auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
         for (int64_t i = 0; i < array->length(); ++i) {
-            auto str = array->GetView(i);
-            size += str.size();
+            if (!nullable_ || !array->IsNull(i)) {
+                auto str = array->GetView(i);
+                size += str.size();
+            }
         }
         row_nums_ += array->length();
     }
@@ -459,8 +461,10 @@ SparseFloatVectorChunkWriter::write_to_target(
     for (const auto& data : array_vec) {
         auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
         for (int64_t i = 0; i < array->length(); ++i) {
-            auto str = array->GetView(i);
-            strs.emplace_back(str);
+            if (!nullable_ || !array->IsNull(i)) {
+                auto str = array->GetView(i);
+                strs.emplace_back(str);
+            }
         }
         if (nullable_) {
             null_bitmaps.emplace_back(
@@ -478,9 +482,23 @@ SparseFloatVectorChunkWriter::write_to_target(
     std::vector<uint64_t> offsets;
     offsets.reserve(offset_num);
 
-    for (const auto& str : strs) {
-        offsets.push_back(offset_start_pos);
-        offset_start_pos += str.size();
+    if (nullable_) {
+        size_t str_idx = 0;
+        for (const auto& data : array_vec) {
+            auto array = std::dynamic_pointer_cast<arrow::BinaryArray>(data);
+            for (int i = 0; i < array->length(); i++) {
+                offsets.push_back(offset_start_pos);
+                if (!array->IsNull(i)) {
+                    offset_start_pos += strs[str_idx].size();
+                    str_idx++;
+                }
+            }
+        }
+    } else {
+        for (const auto& str : strs) {
+            offsets.push_back(offset_start_pos);
+            offset_start_pos += str.size();
+        }
     }
     offsets.push_back(offset_start_pos);
 
@@ -524,22 +542,43 @@ create_chunk_writer(const FieldMeta& field_meta) {
             return std::make_shared<ChunkWriter<arrow::Int64Array, int64_t>>(
                 dim, nullable);
         case milvus::DataType::VECTOR_FLOAT:
+            if (nullable) {
+                return std::make_shared<
+                    NullableVectorChunkWriter<knowhere::fp32>>(dim, nullable);
+            }
             return std::make_shared<
                 ChunkWriter<arrow::FixedSizeBinaryArray, knowhere::fp32>>(
                 dim, nullable);
         case milvus::DataType::VECTOR_BINARY:
+            if (nullable) {
+                return std::make_shared<
+                    NullableVectorChunkWriter<knowhere::bin1>>(dim / 8,
+                                                               nullable);
+            }
             return std::make_shared<
                 ChunkWriter<arrow::FixedSizeBinaryArray, knowhere::bin1>>(
                 dim / 8, nullable);
         case milvus::DataType::VECTOR_FLOAT16:
+            if (nullable) {
+                return std::make_shared<
+                    NullableVectorChunkWriter<knowhere::fp16>>(dim, nullable);
+            }
             return std::make_shared<
                 ChunkWriter<arrow::FixedSizeBinaryArray, knowhere::fp16>>(
                 dim, nullable);
         case milvus::DataType::VECTOR_BFLOAT16:
+            if (nullable) {
+                return std::make_shared<
+                    NullableVectorChunkWriter<knowhere::bf16>>(dim, nullable);
+            }
             return std::make_shared<
                 ChunkWriter<arrow::FixedSizeBinaryArray, knowhere::bf16>>(
                 dim, nullable);
         case milvus::DataType::VECTOR_INT8:
+            if (nullable) {
+                return std::make_shared<
+                    NullableVectorChunkWriter<knowhere::int8>>(dim, nullable);
+            }
             return std::make_shared<
                 ChunkWriter<arrow::FixedSizeBinaryArray, knowhere::int8>>(
                 dim, nullable);

internal/core/src/common/ChunkWriter.h

@@ -129,6 +129,55 @@ class ChunkWriter final : public ChunkWriterBase {
     const int64_t dim_;
 };
 
+template <typename T>
+class NullableVectorChunkWriter final : public ChunkWriterBase {
+ public:
+    NullableVectorChunkWriter(int64_t dim, bool nullable)
+        : ChunkWriterBase(nullable), dim_(dim) {
+    }
+
+    std::pair<size_t, size_t>
+    calculate_size(const arrow::ArrayVector& array_vec) override {
+        size_t size = 0;
+        size_t row_nums = 0;
+
+        for (const auto& data : array_vec) {
+            row_nums += data->length();
+            auto binary_array =
+                std::static_pointer_cast<arrow::BinaryArray>(data);
+            int64_t valid_count = data->length() - binary_array->null_count();
+            size += valid_count * dim_ * sizeof(T);
+        }
+
+        // null bitmap size
+        size += (row_nums + 7) / 8;
+        row_nums_ = row_nums;
+        return {size, row_nums};
+    }
+
+    void
+    write_to_target(const arrow::ArrayVector& array_vec,
+                    const std::shared_ptr<ChunkTarget>& target) override {
+        std::vector<std::tuple<const uint8_t*, int64_t, int64_t>> null_bitmaps;
+        for (const auto& data : array_vec) {
+            null_bitmaps.emplace_back(
+                data->null_bitmap_data(), data->length(), data->offset());
+        }
+        write_null_bit_maps(null_bitmaps, target);
+
+        for (const auto& data : array_vec) {
+            auto binary_array =
+                std::static_pointer_cast<arrow::BinaryArray>(data);
+            auto data_ptr = binary_array->value_data()->data();
+            int64_t valid_count = data->length() - binary_array->null_count();
+            target->write(data_ptr, valid_count * dim_ * sizeof(T));
+        }
+    }
+
+ private:
+    const int64_t dim_;
+};
+
 template <>
 inline void
 ChunkWriter<arrow::BooleanArray, bool>::write_to_target(

internal/core/src/common/FieldMeta.cpp

@@ -183,12 +183,17 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) {
                              data_type,
                              dim,
                              std::nullopt,
-                             false,
+                             nullable,
                              default_value};
         }
         auto metric_type = index_map.at("metric_type");
-        return FieldMeta{
-            name, field_id, data_type, dim, metric_type, false, default_value};
+        return FieldMeta{name,
+                         field_id,
+                         data_type,
+                         dim,
+                         metric_type,
+                         nullable,
+                         default_value};
     }
 
     if (IsStringDataType(data_type)) {

internal/core/src/common/FieldMeta.h

@@ -125,7 +125,6 @@ class FieldMeta {
           vector_info_(VectorInfo{dim, std::move(metric_type)}),
           default_value_(std::move(default_value)) {
         Assert(IsVectorDataType(type_));
-        Assert(!nullable);
     }
 
     // array of vector type