feat: optimize() resues the first segment if possible

src/core/interface/index.cc

@@ -12,10 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <filesystem>
 #include <magic_enum/magic_enum.hpp>
+#include <zvec/ailego/io/file.h>
 #include <zvec/core/framework/index_error.h>
 #include <zvec/core/framework/index_storage.h>
 #include <zvec/core/interface/index.h>
+#include <zvec/core/interface/index_factory.h>
 #include "mixed_reducer/mixed_reducer_params.h"
 
 namespace zvec::core_interface {
@@ -305,6 +308,7 @@ int Index::Open(const std::string &file_path, StorageOptions storage_options) {
 
   is_open_ = true;
   is_read_only_ = storage_options.read_only;
+  storage_path_ = file_path;
   return 0;
 }
 
@@ -851,4 +855,91 @@ std::string Index::get_metric_name(MetricType metric_type, bool is_sparse) {
   }
 }
 
+int Index::MergeAll(const std::string &output_path,
+                    const BaseIndexParam &target_index_param,
+                    const std::vector<Index::Pointer> &all_indexes,
+                    const IndexFilter &filter, const MergeOptions &options,
+                    StorageOptions::StorageType storage_type,
+                    Index::Pointer *result) {
+  if (all_indexes.empty()) {
+    LOG_ERROR("MergeAll: no indexes provided");
+    return core::IndexError_InvalidArgument;
+  }
+  if (result == nullptr) {
+    LOG_ERROR("MergeAll: result pointer is null");
+    return core::IndexError_InvalidArgument;
+  }
+
+  // Check reuse feasibility
+  // Current: only reuse the first index (offset math is trivial)
+  // Must have zero deletions & no builder (IVF has builder, needs full rebuild)
+  auto first_param = all_indexes[0]->GetParam();
+  bool can_reuse =
+      (!filter.is_valid()) && (all_indexes[0]->builder_ == nullptr) &&
+      first_param != nullptr &&
+      target_index_param.SerializeToJson() == first_param->SerializeToJson();
+
+  auto target = IndexFactory::CreateAndInitIndex(target_index_param);
+  if (target == nullptr) {
+    LOG_ERROR("MergeAll: failed to create target index");
+    return core::IndexError_Runtime;
+  }
+
+  int ret = 0;
+  if (can_reuse) {
+    auto &first_index = all_indexes[0];
+    if (first_index->is_open_ && !first_index->is_read_only_) {
+      first_index->Flush();
+    }
+    const std::string &src_path = first_index->storage_path_;
+    // TODO: use FileHelper
+    std::error_code ec;
+    std::filesystem::copy_file(
+        src_path, output_path,
+        std::filesystem::copy_options::overwrite_existing, ec);
+    if (ec) {
+      LOG_ERROR("MergeAll: failed to copy file from %s to %s: %s",
+                src_path.c_str(), output_path.c_str(), ec.message().c_str());
+      return core::IndexError_Runtime;
+    }
+    ret = target->Open(output_path, {storage_type, false, false});
+    if (ret != 0) {
+      LOG_ERROR("MergeAll: failed to open copied target at %s",
+                output_path.c_str());
+      return ret;
+    }
+    if (all_indexes.size() > 1) {
+      std::vector<Index::Pointer> remaining(all_indexes.begin() + 1,
+                                            all_indexes.end());
+      ret = target->Merge(remaining, filter, options);
+      if (ret != 0) {
+        LOG_ERROR("MergeAll: merge of remaining indexes failed");
+        return ret;
+      }
+    }
+    LOG_INFO("MergeAll: reused first index (%u docs) as target",
+             first_index->GetDocCount());
+  } else {
+    ret = target->Open(output_path, {storage_type, true, false});
+    if (ret != 0) {
+      LOG_ERROR("MergeAll: failed to open new target at %s",
+                output_path.c_str());
+      return ret;
+    }
+    ret = target->Merge(all_indexes, filter, options);
+    if (ret != 0) {
+      LOG_ERROR("MergeAll: full merge failed");
+      return ret;
+    }
+  }
+
+  ret = target->Flush();
+  if (ret != 0) {
+    LOG_ERROR("MergeAll: flush failed");
+    return ret;
+  }
+  *result = std::move(target);
+  return 0;
+}
+
 }  // namespace zvec::core_interface

src/core/mixed_reducer/mixed_streamer_reducer.cc

@@ -148,6 +148,19 @@ int MixedStreamerReducer::reduce(const IndexFilter &filter) {
   // TODO: use id instead of key
   uint32_t id_offset = 0, next_id = 0;
 
+  // If the target already has data (reuse path from MergeAll), skip its range
+  if (target_builder_ == nullptr) {
+    uint32_t existing_count =
+        is_sparse_ ? target_streamer_->create_sparse_provider()->count()
+                   : target_streamer_->create_provider()->count();
+    if (existing_count > 0) {
+      next_id = existing_count;
+      id_offset = existing_count;
+      LOG_DEBUG("MixedStreamerReducer: reuse mode, starting IDs from %u",
+                existing_count);
+    }
+  }
+
   if (is_sparse_) {
     for (size_t i = 0; i < num_of_add_threads_; i++) {
       add_group->submit(ailego::Closure::New(

src/db/index/column/vector_column/vector_column_indexer.cc

@@ -133,6 +133,57 @@ Status VectorColumnIndexer::Merge(
   return Status::OK();
 }
 
+Result<VectorColumnIndexer::Ptr> VectorColumnIndexer::MergeAll(
+    const std::string &output_path, const FieldSchema &field_schema,
+    const vector_column_params::ReadOptions &read_options,
+    const std::vector<VectorColumnIndexer::Ptr> &all_indexers,
+    const IndexFilter::Ptr &filter,
+    const vector_column_params::MergeOptions &merge_options) {
+  if (all_indexers.empty()) {
+    return tl::make_unexpected(Status::InvalidArgument("No indexers to merge"));
+  }
+
+  auto index_param_result =
+      ProximaEngineHelper::convert_to_engine_index_param(field_schema);
+  if (!index_param_result.has_value()) {
+    return tl::make_unexpected(
+        Status::InvalidArgument(index_param_result.error().message()));
+  }
+  auto &index_param = index_param_result.value();
+
+  std::vector<core_interface::Index::Pointer> engine_indexes;
+  engine_indexes.reserve(all_indexers.size());
+  for (auto &indexer : all_indexers) {
+    engine_indexes.push_back(indexer->index);
+  }
+
+  auto engine_filter =
+      ProximaEngineHelper::convert_to_engine_filter(filter.get());
+  if (engine_filter == nullptr) {
+    return tl::make_unexpected(
+        Status::InvalidArgument("Failed to convert filter"));
+  }
+
+  auto storage_type =
+      read_options.use_mmap
+          ? core_interface::StorageOptions::StorageType::kMMAP
+          : core_interface::StorageOptions::StorageType::kBufferPool;
+
+  core_interface::Index::Pointer merged_index;
+  int ret = core_interface::Index::MergeAll(
+      output_path, *index_param, engine_indexes, *engine_filter,
+      {merge_options.write_concurrency, merge_options.pool}, storage_type,
+      &merged_index);
+  if (ret != 0) {
+    return tl::make_unexpected(Status::InternalError("Index::MergeAll failed"));
+  }
+
+  auto merged_indexer =
+      std::make_shared<VectorColumnIndexer>(output_path, field_schema);
+  merged_indexer->index = std::move(merged_index);
+  return merged_indexer;
+}
+
 Status VectorColumnIndexer::Insert(
     const vector_column_params::VectorData &vector_data, uint32_t doc_id) {
   if (index == nullptr) {

src/db/index/column/vector_column/vector_column_indexer.h

@@ -63,14 +63,20 @@ class VectorColumnIndexer {
   // Destroy will call Close() and remove index file
   Status Destroy();
 
-
   // If HNSWIndexer.merge([FlatIndexer1, FlatIndexer2])
   // then the merged indexer is a HNSWIndexer
   Status Merge(const std::vector<VectorColumnIndexer::Ptr> &indexers,
                const IndexFilter::Ptr &filter = nullptr,
                const vector_column_params::MergeOptions &merge_options = {});
-  // TODO: should we use this function? or a Reducer?
-  //  TODO: sstatic reduce, optimize; iterator/scan
+
+  //! Merge all indexers into a new one at output_path, automatically reusing
+  //! the first segment to avoid rebuilding the graph from scratch if possible.
+  static Result<VectorColumnIndexer::Ptr> MergeAll(
+      const std::string &output_path, const FieldSchema &field_schema,
+      const vector_column_params::ReadOptions &read_options,
+      const std::vector<VectorColumnIndexer::Ptr> &all_indexers,
+      const IndexFilter::Ptr &filter,
+      const vector_column_params::MergeOptions &merge_options);
 
 
   //! Insert vector

src/db/index/segment/segment.cc

@@ -1572,13 +1572,6 @@ Status SegmentImpl::create_all_vector_index(
 Result<VectorColumnIndexer::Ptr> SegmentImpl::merge_vector_indexer(
     const std::string &index_file_path, const std::string &column,
     const FieldSchema &field, int concurrency) {
-  VectorColumnIndexer::Ptr vector_indexer =
-      std::make_shared<VectorColumnIndexer>(index_file_path, field);
-
-  vector_column_params::ReadOptions options{options_.enable_mmap_, true};
-
-  auto s = vector_indexer->Open(options);
-  CHECK_RETURN_STATUS_EXPECTED(s);
   std::vector<VectorColumnIndexer::Ptr> to_merge_indexers =
       vector_indexers_[column];
   vector_column_params::MergeOptions merge_options;
@@ -1589,12 +1582,20 @@ Result<VectorColumnIndexer::Ptr> SegmentImpl::merge_vector_indexer(
   } else {
     merge_options.write_concurrency = concurrency;
   }
-  s = vector_indexer->Merge(to_merge_indexers, filter_, merge_options);
-  CHECK_RETURN_STATUS_EXPECTED(s);
-  s = vector_indexer->Flush();
+
+
+  auto merged_result = VectorColumnIndexer::MergeAll(
+      index_file_path, field,
+      vector_column_params::ReadOptions{options_.enable_mmap_, true},
+      to_merge_indexers, filter_, merge_options);
+  if (!merged_result.has_value()) {
+    return tl::make_unexpected(merged_result.error());
+  }
+  auto merged = std::move(merged_result.value());
+  auto s = merged->Flush();
   CHECK_RETURN_STATUS_EXPECTED(s);
 
-  return vector_indexer;
+  return merged;
 }
 
 Status SegmentImpl::create_vector_index(
@@ -4462,4 +4463,4 @@ Result<Segment::Ptr> Segment::Open(const std::string &path,
   return segment;
 }
 
-}  // namespace zvec
+}  // namespace zvec