velesdb-core 7.0.0

//! Bulk CRUD operations for Collection (`upsert_bulk`).
//!
//! Extracted from `crud.rs` (Issue #425) to keep each file under 500 NLOC.
//! These methods are optimized for high-throughput import with parallel I/O.
//! Raw import path (`upsert_bulk_from_raw`) is in `crud_bulk_raw.rs`.
//!
//! When `async_index_builder` is configured, `upsert_bulk` uses an optimized
//! V2 path: `DirectVectorWriter` places each vector in the graph's
//! `ContiguousVectors`, and `AsyncIndexBuilder` defers linking it into the
//! graph for higher throughput.

use crate::collection::types::Collection;
use crate::error::{Error, Result};
use crate::index::hnsw::direct_writer::DirectVectorWriter;
use crate::point::Point;
use crate::storage::VectorStorage;

use std::collections::BTreeMap;

impl Collection {
    /// Bulk insert optimized for high-throughput import.
    ///
    /// # Performance
    ///
    /// This method is optimized for bulk loading:
    /// - Uses parallel HNSW insertion (rayon)
    /// - Parallel payload + vector I/O via `rayon::join` (Issue #424)
    /// - Single flush at the end (not per-point)
    /// - No HNSW index save (deferred for performance)
    /// - Faster than the previous sequential approach on large batches (5000+)
    ///
    /// # Errors
    ///
    /// Returns an error if any point has a mismatched dimension.
    pub fn upsert_bulk(&self, points: &[Point]) -> Result<usize> {
        self.upsert_bulk_inner(points, true)
    }

    /// Bulk insert without forcing WAL fsync at the end.
    ///
    /// Identical to [`upsert_bulk`](Self::upsert_bulk) except the WAL
    /// buffer is flushed to the OS kernel (ensuring data is out of the
    /// process) but **not** fsynced to disk. This eliminates the 1-5ms
    /// per-batch fsync overhead on Windows.
    ///
    /// # Safety Contract
    ///
    /// The caller **must** call [`flush()`](Self::flush) after the final
    /// batch to establish a durability barrier. Without that final fsync,
    /// data since the last sync point may be lost on power failure.
    ///
    /// # When to Use
    ///
    /// Use this for intermediate batches in a streaming bulk import.
    /// The final batch should use [`upsert_bulk`](Self::upsert_bulk) or be
    /// followed by an explicit [`flush()`](Self::flush).
    ///
    /// # Errors
    ///
    /// Returns an error if any point has a mismatched dimension.
    #[allow(dead_code)] // Reserved for future streaming ingestion surface.
    pub(crate) fn upsert_bulk_deferred_sync(&self, points: &[Point]) -> Result<usize> {
        self.upsert_bulk_inner(points, false)
    }

    /// Shared implementation for bulk insert with configurable fsync.
    fn upsert_bulk_inner(&self, points: &[Point], fsync: bool) -> Result<usize> {
        if points.is_empty() {
            return Ok(0);
        }

        // Parity item E + dimension validation at the cold boundary, before any
        // storage lock / WAL write (shared with the single-upsert path).
        let dimension = self.storage.config.read().dimension;
        self.validate_vector_upsert_batch(points, dimension)?;

        let vector_refs: Vec<(u64, &[f32])> =
            points.iter().map(|p| (p.id, p.vector.as_slice())).collect();
        let sparse_batch = Self::collect_sparse_batch(points);

        // The V2 fast path writes vectors directly into the graph store and
        // bypasses the quantized backends' insert — on a RaBitQ or SQ8
        // backend that would desynchronize the positional encoding store
        // from the node ids. Quantized collections always take the standard
        // path.
        let use_v2 = self.streaming.async_index_builder.is_some()
            && !self.storage.index.is_quantized_backend();
        let count = if use_v2 {
            self.upsert_bulk_v2_path(&vector_refs, points, &sparse_batch, fsync)?
        } else {
            self.upsert_bulk_standard_path(&vector_refs, points, &sparse_batch, fsync)?
        };

        // Wave 3 Commit 9 — wire `AutoReindexManager` into the bulk hot
        // path. No-op when no manager is attached; emits a `tracing::info!`
        // event when the attached manager reports divergence. Actual
        // reindex reconstruction is out of scope for runtime-only
        // attachment and is left to the external consumer.
        self.notify_auto_reindex_after_bulk();

        Ok(count)
    }

    /// V2 optimized path: `DirectVectorWriter` + `AsyncIndexBuilder`.
    ///
    /// Places each vector in `ContiguousVectors` and maps its id there, then
    /// queues the id for the builder to link where it is: each vector is
    /// written once (#2264).
    fn upsert_bulk_v2_path(
        &self,
        vector_refs: &[(u64, &[f32])],
        points: &[Point],
        sparse_batch: &BTreeMap<String, Vec<(u64, crate::index::sparse::SparseVector)>>,
        fsync: bool,
    ) -> Result<usize> {
        let Some(aib) = self.streaming.async_index_builder.as_ref() else {
            return Err(Error::Config(
                "bulk v2 path requires async index builder".to_string(),
            ));
        };

        // Collect pre-batch payloads before overwriting — used for histogram decrements.
        let old_payloads = {
            let storage = self.storage.payload_storage.read();
            Self::collect_old_payloads(points, &storage)
        };

        // WAL + payload write (same durability guarantees as standard path).
        self.store_vectors_and_payloads_inner(vector_refs, points, &old_payloads, fsync)?;

        // Place each vector in the graph's ContiguousVectors, where rerank and
        // brute force see it at once, and queue its id for the builder to link
        // that slot. An index with its exact-distance features off gives the
        // writer no slot to fill: the builder then places the vectors itself.
        let writer = DirectVectorWriter::new(&self.storage.index);
        let needs_flush = match writer.write_batch_direct(vector_refs)? {
            Some(_) => aib.enqueue_placed(vector_refs.iter().map(|&(id, _)| id)),
            None => aib.enqueue(points.iter().map(|p| (p.id, p.vector.clone())).collect()),
        };

        if needs_flush {
            // Buffer reached merge_threshold — flush synchronously.
            aib.flush_sync(&self.storage.index)?;
        }

        let count = vector_refs.len();
        self.finalize_bulk_upsert(points, &old_payloads, sparse_batch)?;

        // Track inserts for periodic HNSW save (Issue #423 Component 3).
        #[allow(clippy::cast_possible_truncation)]
        self.generations
            .inserts_since_last_hnsw_save
            .fetch_add(count as u64, std::sync::atomic::Ordering::Relaxed);

        tracing::debug!(
            "upsert_bulk V2 path: inserted {count} vectors via DirectVectorWriter + AsyncIndexBuilder"
        );

        Ok(count)
    }

    /// Standard path: synchronous HNSW insertion.
    fn upsert_bulk_standard_path(
        &self,
        vector_refs: &[(u64, &[f32])],
        points: &[Point],
        sparse_batch: &BTreeMap<String, Vec<(u64, crate::index::sparse::SparseVector)>>,
        fsync: bool,
    ) -> Result<usize> {
        // Collect pre-batch payloads before overwriting — used for histogram decrements.
        let old_payloads = {
            let storage = self.storage.payload_storage.read();
            Self::collect_old_payloads(points, &storage)
        };

        self.store_vectors_and_payloads_inner(vector_refs, points, &old_payloads, fsync)?;

        let inserted = self.bulk_index_or_defer(vector_refs);
        self.finalize_bulk_upsert(points, &old_payloads, sparse_batch)?;

        Ok(inserted)
    }

    /// Post-write bookkeeping shared by the V2 and standard bulk paths:
    /// refreshes the point count, applies the sparse batch, replaces payload
    /// histograms (dedup by id), and invalidates caches.
    fn finalize_bulk_upsert(
        &self,
        points: &[Point],
        old_payloads: &[Option<serde_json::Value>],
        sparse_batch: &BTreeMap<String, Vec<(u64, crate::index::sparse::SparseVector)>>,
    ) -> Result<()> {
        // Two statements so the guards never overlap: the one-liner form
        // evaluates the RHS first, holding vector_storage (rank 2) while
        // acquiring config (rank 1) — a decreed-order inversion.
        let point_count = self.storage.vector_storage.read().len();
        self.storage.config.write().point_count = point_count;
        self.apply_sparse_batch_bulk(sparse_batch)?;
        // Incremental histogram maintenance (Bug #47 + Bug #49): dedup by id
        // so only the final payload counts, then atomic decrement + increment.
        self.apply_histogram_replace_dedup(points, old_payloads);
        self.bump_generation_with_mirror_upserts(points);
        Ok(())
    }

    /// Writes vectors and payloads with configurable fsync behavior.
    ///
    /// When `fsync` is `false`, WAL data is written and the buffer is
    /// flushed to the OS kernel, but `sync_all()` is skipped. This
    /// eliminates the 1-5ms per-batch overhead on Windows for
    /// intermediate streaming batches.
    fn store_vectors_and_payloads_inner(
        &self,
        vector_refs: &[(u64, &[f32])],
        points: &[Point],
        old_payloads: &[Option<serde_json::Value>],
        fsync: bool,
    ) -> Result<()> {
        #[cfg(feature = "persistence")]
        {
            let (vec_result, pay_result) = rayon::join(
                || self.bulk_store_vectors_inner(vector_refs, fsync),
                || self.bulk_store_payloads_inner(points, old_payloads, fsync),
            );
            vec_result?;
            pay_result?;
        }

        #[cfg(not(feature = "persistence"))]
        {
            self.bulk_store_vectors_inner(vector_refs, fsync)?;
            self.bulk_store_payloads_inner(points, old_payloads, fsync)?;
        }

        Ok(())
    }

    /// Collects sparse vectors grouped by index name for batch insert.
    fn collect_sparse_batch(
        points: &[Point],
    ) -> BTreeMap<String, Vec<(u64, crate::index::sparse::SparseVector)>> {
        let mut batch: BTreeMap<String, Vec<(u64, crate::index::sparse::SparseVector)>> =
            BTreeMap::new();
        for point in points {
            if let Some(sv_map) = &point.sparse_vectors {
                for (name, sv) in sv_map {
                    batch
                        .entry(name.clone())
                        .or_default()
                        .push((point.id, sv.clone()));
                }
            }
        }
        batch
    }

    /// Stores vectors in bulk via batch WAL + mmap write.
    pub(super) fn bulk_store_vectors(&self, vectors: &[(u64, &[f32])]) -> Result<()> {
        self.bulk_store_vectors_inner(vectors, true)
    }

    /// Stores vectors with configurable fsync behavior.
    ///
    /// When `fsync` is `false`, `store_batch()` writes WAL entries to the
    /// `BufWriter` but `flush()` is skipped entirely. The mmap write is
    /// still performed so the data is immediately readable in-process.
    fn bulk_store_vectors_inner(&self, vectors: &[(u64, &[f32])], fsync: bool) -> Result<()> {
        let mut storage = self.storage.vector_storage.write();
        storage.store_batch(vectors)?;
        if fsync {
            storage.flush()?;
        }
        drop(storage);
        Ok(())
    }

    /// Stores payloads and updates BM25 text index + label index in bulk.
    ///
    /// Uses `LogPayloadStorage::store_batch()` for a single WAL sync instead
    /// of per-point fsync: a batch pays one sync, not one per point.
    ///
    /// When `fsync` is `false`, WAL entries are written and the buffer is
    /// flushed to the OS kernel, but `sync_all()` is skipped.
    fn bulk_store_payloads_inner(
        &self,
        points: &[Point],
        old_payloads: &[Option<serde_json::Value>],
        fsync: bool,
    ) -> Result<()> {
        let entries: Vec<(u64, &serde_json::Value)> = points
            .iter()
            .filter_map(|p| p.payload.as_ref().map(|pl| (p.id, pl)))
            .collect();

        if fsync {
            self.storage.payload_storage.write().store_batch(&entries)?;
        } else {
            self.storage
                .payload_storage
                .write()
                .store_batch_deferred(&entries)?;
        }

        // Issue #425: BM25 skip — when no point has a payload AND the BM25
        // index is empty, skip the text index work entirely. The bulk path
        // inserts fresh points (no old documents to remove), so it would be a
        // no-op for every point.
        //
        // Issue #1797: this used to call `update_text_index` PER POINT, and
        // that helper opens the BM25 WAL, writes one frame and `sync_all()`s it
        // on every call — one durability barrier per document, which capped
        // bulk insertion of text-bearing points at roughly one fsync each. The
        // batch below writes every frame under a single open + flush + fsync.
        if !entries.is_empty() || !self.storage.text_index.is_empty() {
            self.bulk_update_text_index(points, old_payloads)?;
        }

        // Issue #486: Update label index for bulk-inserted points.
        // The bulk path previously skipped label indexing (handled in
        // per_point_updates for the single-upsert path). Without this,
        // MATCH queries with label patterns (e.g., `(d:Doc)`) return
        // empty results for points inserted via upsert_bulk / REST API.
        self.update_label_index_bulk(points, old_payloads);

        Ok(())
    }

    /// Batch-updates the label index for bulk-upserted points.
    ///
    /// This used to index the incoming payloads and nothing else, justified by
    /// "for the bulk path, points are always new inserts (no old payload to
    /// remove from the label index)". That was never true: both bulk paths
    /// call [`collect_old_payloads`](Self::collect_old_payloads) precisely
    /// because a bulk upsert overwrites existing points, and the pre-batch
    /// payload it returns is the one whose labels have to come back out.
    /// Re-upserting a point with `_labels` changed from `Doc` to `Archived`
    /// left it indexed under both, so `MATCH (d:Doc)` returned a point that
    /// is no longer a `Doc`.
    ///
    /// It now runs the same remove-then-index sequence as the single-upsert
    /// path, through the same helpers, so the two cannot drift again:
    ///
    /// - [`needs_label_updates`](Self::needs_label_updates) also asks whether
    ///   an *old* payload carried labels. Asking only about the incoming
    ///   points made a payload that dropped its labels return before the loop
    ///   and keep the stale entry forever.
    /// - [`resolve_effective_old`](Self::resolve_effective_old) resolves the
    ///   old payload for a duplicate id within one batch to the previous
    ///   occurrence rather than the pre-batch value.
    /// - [`apply_label_updates`](Self::apply_label_updates) removes before it
    ///   indexes, which is what keeps a label carried on *both* sides of the
    ///   update from being added and then taken straight back out.
    ///
    /// LOCK ORDER: label_index(7) — after payload_storage(3).
    fn update_label_index_bulk(
        &self,
        points: &[Point],
        old_payloads: &[Option<serde_json::Value>],
    ) {
        if !Self::needs_label_updates(points, old_payloads) {
            return;
        }
        let mut seen: std::collections::HashMap<u64, Option<&serde_json::Value>> =
            std::collections::HashMap::with_capacity(points.len());
        let mut label_updates = Vec::with_capacity(points.len());
        for (point, pre_batch_old) in points.iter().zip(old_payloads) {
            let effective_old =
                Self::resolve_effective_old(&seen, point.id, pre_batch_old.as_ref());
            label_updates.push((point.id, effective_old.cloned(), point.payload.clone()));
            seen.insert(point.id, point.payload.as_ref());
        }
        Self::apply_label_updates(&self.graph.label_index, &label_updates);
    }

    /// Applies sparse batch with WAL-before-apply for bulk insert.
    // Exclusive across WAL append and in-memory apply, so compaction cannot
    // snapshot between the two and then discard the record (WAL-before-apply).
    #[expect(clippy::significant_drop_tightening)]
    fn apply_sparse_batch_bulk(
        &self,
        sparse_batch: &BTreeMap<String, Vec<(u64, crate::index::sparse::SparseVector)>>,
    ) -> Result<()> {
        if sparse_batch.is_empty() {
            return Ok(());
        }
        // Serialize WAL-before-apply with snapshot commit and WAL reset.
        let mut indexes = self.query.sparse_indexes.write();
        #[cfg(feature = "persistence")]
        {
            self.append_sparse_wal_entries(sparse_batch.iter().flat_map(|(name, docs)| {
                docs.iter()
                    .map(move |(point_id, sv)| (name.as_str(), *point_id, sv))
            }))?;
        }
        for (name, docs) in sparse_batch {
            let idx = indexes.entry(name.clone()).or_default();
            idx.insert_batch_chunk(docs);
        }
        Ok(())
    }
}