nodedb_lite/query/document_ops/
indexes.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Collection registration and secondary-index DDL for the Document engine.
3
4use std::sync::Arc;
5
6use nodedb_types::Namespace;
7use nodedb_types::result::QueryResult;
8
9use crate::error::LiteError;
10use crate::index::rebuild::build_index;
11use crate::index::{
12    IndexDef, IndexEngine, IndexPredicate, canonical_field, default_index_name, field_spec,
13};
14use crate::nodedb::collection::CollectionMeta;
15use crate::query::engine::LiteQueryEngine;
16use crate::storage::engine::StorageEngine;
17
18/// Register: initialize a collection in the appropriate engine.
19///
20/// For strict collections (`StorageMode::Strict`), the schema is persisted to
21/// the strict engine. For schemaless collections, this is a no-op — CRDT
22/// collections are discovered on first write.
23pub async fn register<S: StorageEngine>(
24    engine: &LiteQueryEngine<S>,
25    collection: &str,
26    storage_mode: &nodedb_physical::physical_plan::document::types::StorageMode,
27) -> Result<QueryResult, LiteError> {
28    use nodedb_physical::physical_plan::document::types::StorageMode;
29    match storage_mode {
30        StorageMode::Strict { schema } => {
31            engine
32                .strict
33                .create_collection(collection, schema.clone())
34                .await?;
35        }
36        StorageMode::Schemaless => {
37            // Schemaless collections are auto-discovered — no registration needed.
38        }
39    }
40    Ok(affected(0))
41}
42
43/// A `CREATE INDEX`, however it was spelled.
44pub struct CreateIndexRequest<'a> {
45    /// Index name; `None` derives `idx_<collection>_<field>`.
46    pub name: Option<&'a str>,
47    pub collection: &'a str,
48    /// The field as written: `email`, `$.a.b`, or `tags[]` for an array index.
49    pub field: &'a str,
50    pub unique: bool,
51    pub case_insensitive: bool,
52    /// The `WHERE` body of a partial index.
53    pub predicate: Option<&'a str>,
54    /// An index that already exists makes the statement a no-op.
55    pub if_not_exists: bool,
56}
57
58fn affected(n: u64) -> QueryResult {
59    QueryResult {
60        columns: Vec::new(),
61        rows: Vec::new(),
62        rows_affected: n,
63        command: None,
64    }
65}
66
67/// The engine whose rows an index on `collection` covers.
68async fn index_engine_for<S: StorageEngine>(
69    engine: &LiteQueryEngine<S>,
70    collection: &str,
71) -> Result<IndexEngine, LiteError> {
72    if engine.strict.schema(collection).is_some() {
73        return Ok(IndexEngine::Strict);
74    }
75    if engine.columnar.schema(collection).is_some() {
76        return Err(LiteError::Unsupported {
77            detail: format!(
78                "CREATE INDEX on columnar collection '{collection}': columnar collections \
79                 are filtered by block statistics, not secondary indexes"
80            ),
81        });
82    }
83    let meta_key = format!("collection:{collection}");
84    if let Some(bytes) = engine
85        .storage
86        .get(Namespace::Meta, meta_key.as_bytes())
87        .await?
88    {
89        let meta: CollectionMeta =
90            sonic_rs::from_slice(&bytes).map_err(|e| LiteError::Serialization {
91                detail: format!("metadata of collection '{collection}' does not decode: {e}"),
92            })?;
93        return Ok(match meta.collection_type.as_str() {
94            "kv" => IndexEngine::KeyValue,
95            "columnar" | "timeseries" | "spatial" => {
96                return Err(LiteError::Unsupported {
97                    detail: format!(
98                        "CREATE INDEX on {} collection '{collection}' is not supported",
99                        meta.collection_type
100                    ),
101                });
102            }
103            _ => IndexEngine::Document,
104        });
105    }
106    let known = engine
107        .crdt
108        .lock()
109        .map_err(|_| LiteError::LockPoisoned)?
110        .collection_names()
111        .iter()
112        .any(|n| n == collection);
113    if known {
114        Ok(IndexEngine::Document)
115    } else {
116        Err(LiteError::BadRequest {
117            detail: format!("CREATE INDEX: collection '{collection}' does not exist"),
118        })
119    }
120}
121
122/// `CREATE INDEX`: declare the index and build its entries from the rows the
123/// collection holds. A unique index over rows that already share a value is
124/// refused. Answers with the number of entries built.
125pub async fn create_index<S: StorageEngine>(
126    engine: &LiteQueryEngine<S>,
127    req: CreateIndexRequest<'_>,
128) -> Result<QueryResult, LiteError> {
129    let index_engine = index_engine_for(engine, req.collection).await?;
130    declare_index(engine, req, index_engine).await
131}
132
133/// Declare an index over `index_engine` rows and build it: `CREATE INDEX`
134/// once the engine is known.
135pub(crate) async fn declare_index<S: StorageEngine>(
136    engine: &LiteQueryEngine<S>,
137    req: CreateIndexRequest<'_>,
138    index_engine: IndexEngine,
139) -> Result<QueryResult, LiteError> {
140    let (path, is_array) = canonical_field(req.field);
141    let name = req
142        .name
143        .map(str::to_string)
144        .unwrap_or_else(|| default_index_name(req.collection, req.field));
145    if let Some(existing) = engine.indexes.def_named(&name) {
146        if req.if_not_exists {
147            return Ok(affected(0));
148        }
149        return Err(LiteError::BadRequest {
150            detail: format!("index '{name}' already exists on '{}'", existing.collection),
151        });
152    }
153    let spec = field_spec(&path, is_array);
154    if let Some(existing) = engine.indexes.def_on_field(req.collection, &spec) {
155        if req.if_not_exists {
156            return Ok(affected(0));
157        }
158        return Err(LiteError::BadRequest {
159            detail: format!(
160                "collection '{}' already has index '{}' on {spec}: \
161                 drop it before indexing the field again",
162                req.collection, existing.name
163            ),
164        });
165    }
166    let def = Arc::new(IndexDef {
167        name,
168        collection: req.collection.to_string(),
169        path,
170        unique: req.unique,
171        case_insensitive: req.case_insensitive,
172        is_array,
173        predicate: req.predicate.map(IndexPredicate::parse).transpose()?,
174        engine: index_engine,
175    });
176    Ok(affected(rebuild_index(engine, def).await?))
177}
178
179/// `DROP INDEX`: remove the index and its entries. A missing index is an
180/// error unless `if_exists`.
181pub async fn drop_index<S: StorageEngine>(
182    engine: &LiteQueryEngine<S>,
183    name: &str,
184    if_exists: bool,
185) -> Result<QueryResult, LiteError> {
186    match engine.indexes.drop_index(&*engine.storage, name).await? {
187        Some(_) => Ok(affected(0)),
188        None if if_exists => Ok(affected(0)),
189        None => Err(LiteError::BadRequest {
190            detail: format!("index '{name}' does not exist"),
191        }),
192    }
193}
194
195/// The physical `DropIndex` op: drop the index on `field` of `collection`,
196/// or the index named `field`. Dropping an index that does not exist is a
197/// no-op.
198pub async fn drop_field_index<S: StorageEngine>(
199    engine: &LiteQueryEngine<S>,
200    collection: &str,
201    field: &str,
202) -> Result<QueryResult, LiteError> {
203    let (path, is_array) = canonical_field(field);
204    let name = engine
205        .indexes
206        .def_on_field(collection, &field_spec(&path, is_array))
207        .map(|def| def.name.clone())
208        .unwrap_or_else(|| field.to_string());
209    drop_index(engine, &name, true).await
210}
211
212/// Rebuild the entries of one index from its collection's rows.
213pub async fn rebuild_index<S: StorageEngine>(
214    engine: &LiteQueryEngine<S>,
215    def: Arc<IndexDef>,
216) -> Result<u64, LiteError> {
217    build_index(
218        &engine.indexes,
219        &*engine.storage,
220        &engine.crdt,
221        &engine.strict,
222        def,
223    )
224    .await
225}
226
227/// `REINDEX`: rebuild the index named `name`, or every index on
228/// `collection` when no name is given. Answers with the entries built.
229pub async fn reindex<S: StorageEngine>(
230    engine: &LiteQueryEngine<S>,
231    collection: &str,
232    name: Option<&str>,
233) -> Result<QueryResult, LiteError> {
234    let defs: Vec<Arc<IndexDef>> = match name {
235        Some(name) => {
236            let def = engine
237                .indexes
238                .def_named(name)
239                .ok_or_else(|| LiteError::BadRequest {
240                    detail: format!("index '{name}' does not exist"),
241                })?;
242            vec![def]
243        }
244        None => engine
245            .indexes
246            .defs()
247            .into_iter()
248            .filter(|d| d.collection == collection)
249            .collect(),
250    };
251    let mut built = 0;
252    for def in defs {
253        built += rebuild_index(engine, def).await?;
254    }
255    Ok(affected(built))
256}
257
258/// The physical `BackfillIndex` op: rebuild the index already declared on the
259/// field, or declare one under the derived name and build it.
260pub async fn backfill_index<S: StorageEngine>(
261    engine: &LiteQueryEngine<S>,
262    req: CreateIndexRequest<'_>,
263) -> Result<QueryResult, LiteError> {
264    let (path, is_array) = canonical_field(req.field);
265    match engine
266        .indexes
267        .def_on_field(req.collection, &field_spec(&path, is_array))
268    {
269        Some(def) => Ok(affected(rebuild_index(engine, def).await?)),
270        None => create_index(engine, req).await,
271    }
272}
nodedb_lite/query/document_ops/indexes.rs

nodedb_lite/query/document_ops/
indexes.rs
