fix: handle nullable vector compact data paths

Author: marcelo-cjl

internal/core/src/common/FieldData.cpp

@@ -128,7 +128,9 @@ FieldDataImpl<Type, is_type_entire_row>::FillFieldData(
     if (element_count == 0) {
         return;
     }
-    null_count_ = array->null_count();
+    if (!IsVectorDataType(data_type_)) {
+        null_count_ = array->null_count();
+    }
     switch (data_type_) {
         case DataType::BOOL: {
             AssertInfo(array->type()->id() == arrow::Type::type::BOOL,
@@ -687,6 +689,9 @@ FieldDataVectorImpl<Type, is_type_entire_row>::FillFieldData(
             this->valid_data_.data(),
             this->length_,
             total_element_count);
+    } else {
+        bitset::detail::ElementWiseBitsetPolicy<uint8_t>::op_fill(
+            this->valid_data_.data(), this->length_, total_element_count, true);
     }
 
     // update logical to physical offset mapping
@@ -703,7 +708,7 @@ FieldDataVectorImpl<Type, is_type_entire_row>::FillFieldData(
         this->valid_count_ += valid_count;
     }
 
-    this->null_count_ = total_element_count - valid_count;
+    this->null_count_ += total_element_count - valid_count;
     this->length_ += total_element_count;
 }
 

internal/core/src/index/VectorMemIndex.cpp

@@ -78,6 +78,79 @@
 
 namespace milvus::index {
 
+namespace {
+
+bool
+IsValidDataBinary(const std::string& name) {
+    return name == VALID_DATA_COUNT_KEY || name == VALID_DATA_KEY;
+}
+
+bool
+ContainsOnlyValidData(const BinarySet& binary_set) {
+    if (!binary_set.Contains(VALID_DATA_COUNT_KEY) ||
+        !binary_set.Contains(VALID_DATA_KEY)) {
+        return false;
+    }
+    for (const auto& [name, _] : binary_set.binary_map_) {
+        if (!IsValidDataBinary(name)) {
+            return false;
+        }
+    }
+    return true;
+}
+
+void
+AppendValidDataToBinarySet(const OffsetMapping& offset_mapping,
+                           BinarySet& binary_set) {
+    if (!offset_mapping.IsEnabled()) {
+        return;
+    }
+
+    auto total_count = offset_mapping.GetTotalCount();
+
+    std::shared_ptr<uint8_t[]> count_buf(new uint8_t[sizeof(size_t)]);
+    size_t count = static_cast<size_t>(total_count);
+    std::memcpy(count_buf.get(), &count, sizeof(size_t));
+    binary_set.Append(VALID_DATA_COUNT_KEY, count_buf, sizeof(size_t));
+
+    size_t byte_size = (count + 7) / 8;
+    std::shared_ptr<uint8_t[]> data(new uint8_t[byte_size]);
+    std::memset(data.get(), 0, byte_size);
+    for (size_t i = 0; i < count; ++i) {
+        if (offset_mapping.IsValid(i)) {
+            data[i / 8] |= (1 << (i % 8));
+        }
+    }
+    binary_set.Append(VALID_DATA_KEY, data, byte_size);
+}
+
+bool
+LoadValidDataFromBinarySet(const BinarySet& binary_set,
+                           VectorIndex* vector_index) {
+    bool has_count = binary_set.Contains(VALID_DATA_COUNT_KEY);
+    bool has_data = binary_set.Contains(VALID_DATA_KEY);
+    if (!has_count && !has_data) {
+        return false;
+    }
+    AssertInfo(has_count && has_data,
+               "nullable vector index valid_data files are incomplete");
+
+    auto count_ptr = binary_set.GetByName(VALID_DATA_COUNT_KEY);
+    size_t count;
+    std::memcpy(&count, count_ptr->data.get(), sizeof(size_t));
+
+    auto data_ptr = binary_set.GetByName(VALID_DATA_KEY);
+    std::unique_ptr<bool[]> valid_data(new bool[count]);
+    auto bitmap = data_ptr->data.get();
+    for (size_t i = 0; i < count; ++i) {
+        valid_data[i] = (bitmap[i / 8] >> (i % 8)) & 1;
+    }
+    vector_index->BuildValidData(valid_data.get(), count);
+    return true;
+}
+
+}  // namespace
+
 template <typename T>
 VectorMemIndex<T>::VectorMemIndex(
     DataType elem_type,
@@ -166,32 +239,18 @@ template <typename T>
 BinarySet
 VectorMemIndex<T>::Serialize(const Config& config) {
     knowhere::BinarySet ret;
-    auto stat = index_.Serialize(ret);
-    if (stat != knowhere::Status::success)
-        ThrowInfo(ErrorCode::UnexpectedError,
-                  "failed to serialize index: {}",
-                  KnowhereStatusString(stat));
-
-    // Serialize valid_data from offset_mapping if enabled
-    if (offset_mapping_.IsEnabled()) {
-        auto total_count = offset_mapping_.GetTotalCount();
-
-        std::shared_ptr<uint8_t[]> count_buf(new uint8_t[sizeof(size_t)]);
-        size_t count = static_cast<size_t>(total_count);
-        std::memcpy(count_buf.get(), &count, sizeof(size_t));
-        ret.Append(VALID_DATA_COUNT_KEY, count_buf, sizeof(size_t));
-
-        size_t byte_size = (count + 7) / 8;
-        std::shared_ptr<uint8_t[]> data(new uint8_t[byte_size]);
-        std::memset(data.get(), 0, byte_size);
-        for (size_t i = 0; i < count; ++i) {
-            if (offset_mapping_.IsValid(i)) {
-                data[i / 8] |= (1 << (i % 8));
-            }
-        }
-        ret.Append(VALID_DATA_KEY, data, byte_size);
+    bool all_null_nullable =
+        offset_mapping_.IsEnabled() && offset_mapping_.GetTotalCount() > 0 &&
+        offset_mapping_.GetValidCount() == 0;
+    if (!all_null_nullable) {
+        auto stat = index_.Serialize(ret);
+        if (stat != knowhere::Status::success)
+            ThrowInfo(ErrorCode::UnexpectedError,
+                      "failed to serialize index: {}",
+                      KnowhereStatusString(stat));
     }
 
+    AppendValidDataToBinarySet(offset_mapping_, ret);
     Disassemble(ret);
 
     return ret;
@@ -201,31 +260,20 @@ template <typename T>
 void
 VectorMemIndex<T>::LoadWithoutAssemble(const BinarySet& binary_set,
                                        const Config& config) {
-    auto stat = index_.Deserialize(binary_set, config);
-    if (stat != knowhere::Status::success)
-        ThrowInfo(ErrorCode::UnexpectedError,
-                  "failed to Deserialize index: {}",
-                  KnowhereStatusString(stat));
-
-    // Deserialize valid_data bitmap and rebuild offset_mapping
-    if (binary_set.Contains(VALID_DATA_COUNT_KEY) &&
-        binary_set.Contains(VALID_DATA_KEY)) {
-        knowhere::BinaryPtr ptr;
-        ptr = binary_set.GetByName(VALID_DATA_COUNT_KEY);
-        size_t count;
-        std::memcpy(&count, ptr->data.get(), sizeof(size_t));
-
-        ptr = binary_set.GetByName(VALID_DATA_KEY);
-        // Convert bitmap to bool array
-        std::unique_ptr<bool[]> valid_data(new bool[count]);
-        auto bitmap = ptr->data.get();
-        for (size_t i = 0; i < count; ++i) {
-            valid_data[i] = (bitmap[i / 8] >> (i % 8)) & 1;
+    if (ContainsOnlyValidData(binary_set)) {
+        if (config.contains(DIM_KEY)) {
+            SetDim(GetDimFromConfig(config));
         }
-        BuildValidData(valid_data.get(), count);
+    } else {
+        auto stat = index_.Deserialize(binary_set, config);
+        if (stat != knowhere::Status::success)
+            ThrowInfo(ErrorCode::UnexpectedError,
+                      "failed to Deserialize index: {}",
+                      KnowhereStatusString(stat));
+        SetDim(index_.Dim());
     }
 
-    SetDim(index_.Dim());
+    LoadValidDataFromBinarySet(binary_set, this);
 }
 
 template <typename T>
@@ -451,6 +499,11 @@ VectorMemIndex<T>::Build(const Config& config) {
                 total_size += data->Size();
             }
         }
+        if (nullable && total_valid_rows == 0) {
+            SetDim(dim);
+            BuildValidData(valid_data.get(), total_num_rows);
+            return;
+        }
         auto buf = std::shared_ptr<uint8_t[]>(new uint8_t[total_size]);
 
         size_t lim_offset = 0;
@@ -523,6 +576,11 @@ VectorMemIndex<T>::Build(const Config& config) {
                     field_data)
                     ->Dim());
         }
+        if (nullable && total_valid_rows == 0) {
+            SetDim(dim);
+            BuildValidData(valid_data.get(), total_num_rows);
+            return;
+        }
         std::vector<knowhere::sparse::SparseRow<SparseValueType>> vec(
             total_valid_rows);
         int64_t offset = 0;
@@ -775,6 +833,7 @@ void VectorMemIndex<T>::LoadFromFile(const Config& config) {
     std::chrono::duration<double> write_disk_duration_sum;
     std::unique_ptr<storage::DataCodec> valid_data_count_codec;
     std::unique_ptr<storage::DataCodec> valid_data_codec;
+    bool wrote_index_data = false;
     // load files in two parts:
     // 1. EMB_LIST_META: Written separately to embedding_list_meta_writer_ptr (if embedding list type)
     // 2. All other binaries: Merged and written to file_writer, forming a unified index file for knowhere
@@ -818,6 +877,7 @@ void VectorMemIndex<T>::LoadFromFile(const Config& config) {
                     } else {
                         file_writer.Write(data->PayloadData(),
                                           data->PayloadSize());
+                        wrote_index_data = true;
                     }
                     write_disk_duration_sum +=
                         (std::chrono::system_clock::now() - start_write_file);
@@ -863,6 +923,7 @@ void VectorMemIndex<T>::LoadFromFile(const Config& config) {
             } else {
                 file_writer.Write(index_data->PayloadData(),
                                   index_data->PayloadSize());
+                wrote_index_data = true;
             }
         }
         write_disk_duration_sum +=
@@ -880,29 +941,38 @@ void VectorMemIndex<T>::LoadFromFile(const Config& config) {
         embedding_list_meta_writer_ptr->Finish();
     }
 
-    LOG_INFO("load index into Knowhere...");
     auto conf = config;
     conf.erase(MMAP_FILE_PATH);
     conf[ENABLE_MMAP] = true;
     if (is_embedding_list) {
         conf["emb_list_meta_file_path"] = embedding_list_meta_path.value();
     }
     auto start_deserialize = std::chrono::system_clock::now();
-    auto stat = index_.DeserializeFromFile(local_filepath.value(), conf);
-    auto deserialize_duration =
-        std::chrono::system_clock::now() - start_deserialize;
-    if (stat != knowhere::Status::success) {
-        ThrowInfo(ErrorCode::UnexpectedError,
-                  "failed to Deserialize index: {}",
-                  KnowhereStatusString(stat));
+    std::chrono::duration<double> deserialize_duration{};
+    if (wrote_index_data) {
+        LOG_INFO("load index into Knowhere...");
+        auto stat = index_.DeserializeFromFile(local_filepath.value(), conf);
+        deserialize_duration = std::chrono::system_clock::now() -
+                               start_deserialize;
+        if (stat != knowhere::Status::success) {
+            ThrowInfo(ErrorCode::UnexpectedError,
+                      "failed to Deserialize index: {}",
+                      KnowhereStatusString(stat));
+        }
+        this->SetDim(index_.Dim());
+    } else {
+        LOG_INFO("load all-null nullable vector index valid data only...");
+        AssertInfo(valid_data_count_codec && valid_data_codec,
+                   "nullable vector index valid_data files are incomplete");
+        if (conf.contains(DIM_KEY)) {
+            this->SetDim(GetDimFromConfig(conf));
+        }
     }
     milvus::monitor::internal_storage_deserialize_duration.Observe(
         std::chrono::duration_cast<std::chrono::milliseconds>(
             deserialize_duration)
             .count());
 
-    this->SetDim(index_.Dim());
-
     // Restore valid_data for nullable vector support
     if (valid_data_count_codec && valid_data_codec) {
         size_t count;

internal/core/src/indexbuilder/VecIndexCreator.cpp

@@ -70,12 +70,19 @@ void
 VecIndexCreator::Build(const milvus::DatasetPtr& dataset,
                        const bool* valid_data,
                        const int64_t valid_data_len) {
-    index_->BuildWithDataset(dataset, config_);
     if (valid_data && valid_data_len > 0) {
         auto vec_index = dynamic_cast<index::VectorIndex*>(index_.get());
         AssertInfo(vec_index != nullptr, "failed to cast index to VectorIndex");
+        if (dataset->GetRows() == 0) {
+            vec_index->SetDim(dataset->GetDim());
+            vec_index->BuildValidData(valid_data, valid_data_len);
+            return;
+        }
+        index_->BuildWithDataset(dataset, config_);
         vec_index->BuildValidData(valid_data, valid_data_len);
+        return;
     }
+    index_->BuildWithDataset(dataset, config_);
 }
 
 void

internal/core/src/query/Utils.h

@@ -55,9 +55,12 @@ ApplyElementIDMapping(const std::vector<int64_t>& element_ids,
 inline TargetBitmap
 TransformBitset(const BitsetView& bitset,
                 const milvus::OffsetMapping& mapping) {
+    // bit=true means filtered out. The logical bitset may be shorter than the
+    // full mapping on growing segments because it is sized by query timestamp
+    // visibility. Physical rows outside that logical view are not visible yet.
     TargetBitmap result;
     auto count = mapping.GetValidCount();
-    result.resize(count);
+    result.resize(count, true);
     for (int64_t physical_idx = 0; physical_idx < count; physical_idx++) {
         auto logical_idx = mapping.GetLogicalOffset(physical_idx);
         if (logical_idx >= 0 &&