qdrant_edge/edge/update_only/
lifecycle.rs

1use std::path::{Path, PathBuf};
2
3use crate::common::universal_io::{MmapFile, MmapFs, UniversalRead, UniversalReadFs};
4use parking_lot::RwLock;
5use rayon::prelude::*;
6use crate::segment::common::operation_error::OperationResult;
7use crate::segment::segment::update_only::UpdateOnlySegment;
8use uuid::Uuid;
9
10use crate::edge::read_only::{LocalSegmentEnumerator, SegmentEnumerator};
11use crate::edge::read_view::build_segment_pool;
12use crate::edge::update_only::UpdateOnlyEdgeShard;
13use crate::edge::update_only::holder::UpdateOnlySegmentHolder;
14
15impl UpdateOnlyEdgeShard<MmapFile> {
16    /// Open a writer over local memory-mapped files, discovering segments by
17    /// scanning the `segments/` directory — the writer owns the directory it
18    /// writes to, so there is no manifest to agree with.
19    pub fn open_mmap(path: &Path) -> OperationResult<Self> {
20        Self::open(MmapFs, path, LocalSegmentEnumerator::new(path))
21    }
22}
23
24impl<S: UniversalRead + 'static> UpdateOnlyEdgeShard<S> {
25    /// Open a writer over the shard directory at `path`, using `fs` as the
26    /// read backend and `enumerator` to discover the segments.
27    ///
28    /// Segments are opened in parallel on the shard's thread pool, each over
29    /// its own prefetching [`CachedFs`](common::universal_io::CachedFs) (see
30    /// [`UpdateOnlySegment::open`]) — the same shape as the read-only
31    /// follower's load — and entirely cold: no point data is fetched until a
32    /// batch reads a point. A segment that fails to load is an error, not a
33    /// skip — a writer that misses a segment would resolve a point against a
34    /// stale copy of itself, or duplicate it.
35    pub fn open(
36        fs: S::Fs,
37        path: &Path,
38        enumerator: impl SegmentEnumerator + 'static,
39    ) -> OperationResult<Self>
40    where
41        S::Fs: UniversalReadFs<File = S>,
42    {
43        // Sized like the search pools: over-provisioned relative to the CPU
44        // count, since on a remote backend the threads mostly wait on IO.
45        let pool = build_segment_pool(
46            "edge-update",
47            crate::common::defaults::search_thread_count(0),
48            None,
49        )?;
50
51        let segments: Vec<(Uuid, PathBuf)> = enumerator.list_segments()?.into_iter().collect();
52        let opened: Vec<(Uuid, UpdateOnlySegment<S>)> = pool.install(|| {
53            segments
54                .into_par_iter()
55                .map(|(uuid, segment_path)| {
56                    // No deferred threshold yet: it belongs to the coordination
57                    // with an external rebuilder, which does not exist in this
58                    // iteration.
59                    let segment = UpdateOnlySegment::<S>::open(&fs, &segment_path, uuid, None)?;
60                    Ok((uuid, segment))
61                })
62                .collect::<OperationResult<Vec<_>>>()
63        })?;
64
65        let mut holder = UpdateOnlySegmentHolder::default();
66        for (uuid, segment) in opened {
67            holder.insert(uuid, segment);
68        }
69
70        if holder.is_empty() {
71            // Creating the first appendable segment needs the append-only
72            // components the writer cannot build yet, so an empty directory is
73            // not something this iteration can bootstrap.
74            todo!("creating the initial appendable segment needs the append-only components");
75        }
76
77        Ok(Self {
78            path: path.to_path_buf(),
79            fs,
80            segments: RwLock::new(holder),
81            pool,
82        })
83    }
84}
qdrant_edge/edge/update_only/lifecycle.rs

qdrant_edge/edge/update_only/
lifecycle.rs
