zeppelin-embed 0.6.0

In-process vector, lexical, and hybrid search for macOS and Apple silicon
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

//! Explicit same-tokenizer reindex using immutable segment replacement.
use super::{PublishedSnapshot, Store, StoreError, StoreState};
use crate::manifest::io::{MANIFEST_FILE, commit_manifest, load_manifest};
use crate::segment::SegmentId;
use std::sync::Arc;

impl Store {
    /// Rebuilds every sealed text index from stored text under the current tokenizer.
    /// Active text is already indexed; seal it first. Publication replaces every
    /// sealed segment in one manifest commit, preserving all non-postings regions.
    pub fn reindex_text(&self) -> Result<u64, StoreError> {
        self.seal()?;
        let _maintenance = self
            .maintenance
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "maintenance",
            })?;
        let state = self
            .state
            .lock()
            .map_err(|_| StoreError::Synchronization { component: "state" })?;
        match *state {
            StoreState::Open => {}
            StoreState::Closing => return Err(StoreError::Closing),
            StoreState::Closed => return Err(StoreError::Closed),
        }
        let writer = self
            .writer_lock
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "writer lock",
            })?;
        if writer.is_none() {
            return Err(StoreError::ReadOnly);
        }
        let wal = self
            .wal_writer
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "WAL writer",
            })?;
        let durable_end = wal.as_ref().ok_or(StoreError::ReadOnly)?.durable_end();
        let mut active = self
            .active
            .lock()
            .map_err(|_| StoreError::Synchronization {
                component: "active segment",
            })?;
        let current = active.as_mut().ok_or(StoreError::Closed)?;
        let snapshot = self
            .snapshot
            .read()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })?
            .clone()
            .ok_or(StoreError::Closed)?;
        if snapshot.all_segments().is_empty() {
            return Ok(current.generation);
        }
        let mut manifest = load_manifest(
            self.vfs.as_ref(),
            &self.directory.join(MANIFEST_FILE),
            durable_end,
        )
        .map_err(StoreError::Manifest)?;
        for epoch in &manifest.epochs {
            super::validate_tokenizer_epoch(
                Some(crate::epoch::EpochIdentity {
                    embedding: epoch.id,
                    tokenizer: epoch.tokenizer,
                }),
                &self.tokenizer,
            )?;
        }
        let generation = manifest
            .generation
            .max(current.generation)
            .checked_add(1)
            .ok_or(StoreError::GenerationOverflow)?;
        let mut replacements = Vec::new();
        for segment in snapshot.all_segments() {
            let mut bytes = *segment.meta().id.as_bytes();
            let hash = xxhash_rust::xxh3::xxh3_64_with_seed(&bytes, generation ^ 0x5245494e444558);
            bytes
                .get_mut(..8)
                .ok_or(StoreError::GenerationOverflow)?
                .copy_from_slice(&generation.to_be_bytes());
            bytes
                .get_mut(8..)
                .ok_or(StoreError::GenerationOverflow)?
                .copy_from_slice(&hash.to_be_bytes());
            replacements.push(
                crate::segment::writer::reindex_segment(
                    self.vfs.as_ref(),
                    &self.directory,
                    segment,
                    SegmentId::from_bytes(bytes),
                    &self.tokenizer,
                    self.durability_policy,
                )
                .map_err(StoreError::Segment)?,
            );
        }
        manifest.segments = replacements;
        manifest.generation = generation;
        commit_manifest(
            self.vfs.as_ref(),
            &self.directory,
            &manifest,
            self.durability_policy,
        )
        .map_err(StoreError::Manifest)?;
        let remapped =
            PublishedSnapshot::load_on_vfs(&self.directory, &self.accounting, self.vfs.as_ref())?;
        *self
            .snapshot
            .write()
            .map_err(|_| StoreError::Synchronization {
                component: "published snapshot",
            })? = Some(Arc::new(remapped));
        current.generation = generation;
        // Keep old files for pinned readers; ordinary open reachability cleanup reclaims them.
        Ok(generation)
    }
}