siderite_backends/mongodb/compile/
mod.rs

1//! Pure compiler: plans to MongoDB filters, update documents and pipelines.
2//!
3//! Nothing here touches a server; every function returns an error for a plan
4//! the backend cannot express, so unsupported plans fail before any I/O.
5
6use super::value::to_bson;
7use ::mongodb::bson::{Bson, Document, doc};
8use siderite_orm::expr::{Aggregate, Column};
9use siderite_orm::{
10    BackendCapabilities, BackendCapabilityError, BackendKind, DeletePlan, Expr, Feature,
11    InsertPlan, OrmError, QueryError, QueryPlan, UpdatePlan, Value, WritePlan,
12};
13use std::collections::HashMap;
14
15mod aggregate;
16mod expr;
17mod filter;
18mod pipeline;
19
20use expr::expr_bson;
21use filter::filter_doc;
22use pipeline::build;
23
24/// Which column of a collection is stored as `_id`.
25///
26/// Plans carry no model metadata, so the adapter is told: the default is a
27/// column named `id`, overridable per collection.
28#[derive(Debug, Clone)]
29pub struct Keys {
30    default: String,
31    per_collection: HashMap<String, String>,
32}
33
34impl Default for Keys {
35    fn default() -> Self {
36        Self {
37            default: "id".into(),
38            per_collection: HashMap::new(),
39        }
40    }
41}
42
43impl Keys {
44    /// Use `column` as the primary key of `collection`.
45    #[must_use]
46    pub fn with(mut self, collection: impl Into<String>, column: impl Into<String>) -> Self {
47        self.per_collection.insert(collection.into(), column.into());
48        self
49    }
50
51    /// Use `column` for every collection without an explicit entry.
52    #[must_use]
53    pub fn with_default(mut self, column: impl Into<String>) -> Self {
54        self.default = column.into();
55        self
56    }
57
58    /// Primary-key column of `collection`.
59    pub fn pk(&self, collection: &str) -> &str {
60        self.per_collection
61            .get(collection)
62            .map_or(self.default.as_str(), String::as_str)
63    }
64}
65
66/// A compiled read: run `pipeline` on `collection`.
67#[derive(Debug, Clone, PartialEq)]
68pub struct CompiledQuery {
69    /// Collection the pipeline runs on.
70    pub collection: String,
71    /// Aggregation pipeline.
72    pub pipeline: Vec<Document>,
73    /// Output column names, in order. Empty means "every field of the
74    /// document" (`_id` is renamed to [`pk`](Self::pk)).
75    pub columns: Vec<String>,
76    /// Primary-key column of the root collection.
77    pub pk: String,
78    /// The single row an ungrouped aggregate produces over no input (SQL
79    /// returns one row, `$group` returns none).
80    pub empty_row: Option<Vec<Value>>,
81}
82
83/// A compiled insert.
84#[derive(Debug, Clone, PartialEq)]
85pub struct CompiledInsert {
86    /// Target collection.
87    pub collection: String,
88    /// Primary-key column, stored as `_id`.
89    pub pk: String,
90    /// Documents in input order. `_id` is present only when the plan
91    /// supplied the key column; otherwise the executor assigns keys.
92    pub docs: Vec<Document>,
93    /// Columns to return.
94    pub returning: Vec<String>,
95}
96
97/// How an update changes documents.
98#[derive(Debug, Clone, PartialEq)]
99pub enum UpdateSpec {
100    /// `{"$set": {..}}` with literal values.
101    Set(Document),
102    /// An update pipeline (`$set` stage) for F-expression assignments.
103    Pipeline(Vec<Document>),
104}
105
106/// A compiled update.
107#[derive(Debug, Clone, PartialEq)]
108pub struct CompiledUpdate {
109    /// Target collection.
110    pub collection: String,
111    /// Primary-key column, stored as `_id`.
112    pub pk: String,
113    /// Filter for `update_many`.
114    pub filter: Document,
115    /// Update to apply.
116    pub update: UpdateSpec,
117    /// Columns to return.
118    pub returning: Vec<String>,
119}
120
121/// A compiled delete.
122#[derive(Debug, Clone, PartialEq)]
123pub struct CompiledDelete {
124    /// Target collection.
125    pub collection: String,
126    /// Primary-key column, stored as `_id`.
127    pub pk: String,
128    /// Filter for `delete_many`.
129    pub filter: Document,
130    /// Columns to return.
131    pub returning: Vec<String>,
132}
133
134/// A compiled write.
135#[derive(Debug, Clone, PartialEq)]
136pub enum CompiledWrite {
137    /// Insert.
138    Insert(CompiledInsert),
139    /// Update.
140    Update(CompiledUpdate),
141    /// Delete.
142    Delete(CompiledDelete),
143}
144
145fn invalid(message: impl Into<String>) -> OrmError {
146    QueryError::InvalidPlan(message.into()).into()
147}
148
149fn unsupported(feature: Feature) -> OrmError {
150    BackendCapabilityError::from_feature(BackendKind::MongoDb, feature).into()
151}
152
153type Res<T> = Result<T, OrmError>;
154
155/// Compile a read plan.
156///
157/// # Errors
158/// A capability error for features MongoDB lacks, or
159/// [`QueryError::InvalidPlan`] for plans it cannot express.
160pub fn compile_query(plan: &QueryPlan, keys: &Keys) -> Res<CompiledQuery> {
161    plan.check(&BackendCapabilities::mongodb())?;
162    let built = build(plan, keys)?;
163    Ok(CompiledQuery {
164        collection: built.collection,
165        pipeline: built.pipeline,
166        columns: built.columns,
167        pk: built.pk,
168        empty_row: built.empty_row,
169    })
170}
171
172/// Compile a write plan.
173///
174/// # Errors
175/// As [`compile_query`].
176pub fn compile_write(plan: &WritePlan, keys: &Keys) -> Res<CompiledWrite> {
177    plan.check(&BackendCapabilities::mongodb())?;
178    Ok(match plan {
179        WritePlan::Insert(p) => CompiledWrite::Insert(compile_insert(p, keys)?),
180        WritePlan::Update(p) => CompiledWrite::Update(compile_update(p, keys)?),
181        WritePlan::Delete(p) => CompiledWrite::Delete(compile_delete(p, keys)?),
182    })
183}
184
185fn returning_columns(columns: &[siderite_orm::expr::Ident]) -> Vec<String> {
186    columns.iter().map(ToString::to_string).collect()
187}
188
189fn compile_insert(plan: &InsertPlan, keys: &Keys) -> Res<CompiledInsert> {
190    let pk = keys.pk(&plan.table).to_owned();
191    check_name(&plan.table)?;
192    let mut docs = Vec::with_capacity(plan.rows.len());
193    for row in &plan.rows {
194        if row.len() != plan.columns.len() {
195            return Err(invalid(format!(
196                "insert row has {} values for {} columns",
197                row.len(),
198                plan.columns.len()
199            )));
200        }
201        let mut doc = Document::new();
202        for (column, value) in plan.columns.iter().zip(row) {
203            let name = if *column == pk {
204                "_id".to_owned()
205            } else {
206                check_name(column)?;
207                column.to_string()
208            };
209            let value = to_bson(value)?;
210            if name == "_id" && value == Bson::Null {
211                return Err(invalid("the primary key cannot be NULL"));
212            }
213            doc.insert(name, value);
214        }
215        docs.push(doc);
216    }
217    Ok(CompiledInsert {
218        collection: plan.table.to_string(),
219        pk,
220        docs,
221        returning: returning_columns(&plan.returning),
222    })
223}
224
225fn compile_update(plan: &UpdatePlan, keys: &Keys) -> Res<CompiledUpdate> {
226    check_name(&plan.table)?;
227    let pk = keys.pk(&plan.table).to_owned();
228    let ctx = Ctx {
229        pk: Some(&pk),
230        root: plan.table.as_ref(),
231    };
232    if plan.assignments.is_empty() {
233        return Err(invalid("an update needs at least one assignment"));
234    }
235    let filter = match &plan.filter {
236        Some(f) => filter_doc(f, false, &ctx)?,
237        None => Document::new(),
238    };
239    let literal_only = plan
240        .assignments
241        .iter()
242        .all(|(_, e)| matches!(e, Expr::Value(_)));
243    let mut set = Document::new();
244    for (column, expr) in &plan.assignments {
245        if *column == pk {
246            return Err(invalid("the primary key cannot be updated"));
247        }
248        check_name(column)?;
249        let value = if literal_only {
250            match expr {
251                Expr::Value(v) => to_bson(v)?,
252                _ => Bson::Null,
253            }
254        } else {
255            expr_bson(expr, &Scope::row(&ctx))?
256        };
257        set.insert(column.to_string(), value);
258    }
259    let update = if literal_only {
260        UpdateSpec::Set(doc! { "$set": set })
261    } else {
262        UpdateSpec::Pipeline(vec![doc! { "$set": set }])
263    };
264    Ok(CompiledUpdate {
265        collection: plan.table.to_string(),
266        pk,
267        filter,
268        update,
269        returning: returning_columns(&plan.returning),
270    })
271}
272
273fn compile_delete(plan: &DeletePlan, keys: &Keys) -> Res<CompiledDelete> {
274    check_name(&plan.table)?;
275    let pk = keys.pk(&plan.table).to_owned();
276    let ctx = Ctx {
277        pk: Some(&pk),
278        root: plan.table.as_ref(),
279    };
280    let filter = match &plan.filter {
281        Some(f) => filter_doc(f, false, &ctx)?,
282        None => Document::new(),
283    };
284    Ok(CompiledDelete {
285        collection: plan.table.to_string(),
286        pk,
287        filter,
288        returning: returning_columns(&plan.returning),
289    })
290}
291
292// ---------------------------------------------------------------------
293// Names and contexts
294// ---------------------------------------------------------------------
295
296fn check_name(name: &str) -> Res<()> {
297    if name.is_empty() || name.starts_with('$') || name.contains('.') || name.contains('\0') {
298        return Err(invalid(format!(
299            "`{name}` is not a valid MongoDB field or collection name for this backend"
300        )));
301    }
302    Ok(())
303}
304
305/// Naming context of one pipeline level.
306struct Ctx<'a> {
307    /// Column stored as `_id`; `None` for the output of a derived table.
308    pk: Option<&'a str>,
309    /// Name the plan uses for its source (table or alias).
310    root: &'a str,
311}
312
313impl Ctx<'_> {
314    fn field(&self, column: &Column) -> Res<String> {
315        if let Some(source) = &column.source
316            && source.as_ref() != self.root
317        {
318            return Err(invalid(format!(
319                "column `{source}.{}` refers to a source other than `{}`",
320                column.name, self.root
321            )));
322        }
323        if self.pk == Some(column.name.as_ref()) {
324            return Ok("_id".into());
325        }
326        check_name(&column.name)?;
327        Ok(column.name.to_string())
328    }
329}
330
331/// Post-`$group` naming: group keys and aggregates replace source columns.
332struct GroupScope {
333    keys: Vec<Expr>,
334    aggs: Vec<Aggregate>,
335    aliases: Vec<(String, Expr)>,
336}
337
338struct Scope<'a> {
339    ctx: &'a Ctx<'a>,
340    group: Option<&'a GroupScope>,
341}
342
343impl<'a> Scope<'a> {
344    fn row(ctx: &'a Ctx<'a>) -> Self {
345        Self { ctx, group: None }
346    }
347}
siderite_backends/mongodb/compile/mod.rs

siderite_backends/mongodb/compile/
mod.rs
