rushdb_query/
point_fields.rs

1//! Record properties read by point reads instead of rebuilding the record
2//!: the ordering properties of an ordered page, the properties a
3//! residual predicate names, and the columns a SQL label scan reads.
4
5use crate::aggregate_rows::Cell;
6use crate::planner::{family_id, pids_of};
7use crate::{access, normalize, reconstruct, Expr, FieldPred, Json, QueryError, QuerySnapshot};
8use rushdb_lmpg::reader::TryRead;
9use rushdb_types::value::ID_FIELD;
10use serde_json::Map;
11
12/// Record properties read with one point read per typed definition instead
13/// of rebuilding the record (READ-1: sorting 667K matches, or checking an
14/// unindexed `age < 25` on them, rebuilt every one). Covers scalar record
15/// properties and the record id (`__id`, read from the record header);
16/// [`Self::resolve`] answers `None` for another system field or a property
17/// stored as an array anywhere, and the caller rebuilds the record.
18///
19/// A record holds one value per property: a write deletes the value of
20/// every typed definition of the property before it stores the new one
21/// (also when one batch writes the property twice). So the first definition
22/// holding a value is the only one, and definitions are probed in the order
23/// most records hold them: by descending owner support (under the read's
24/// labels). A CSV column inferred as both Int64 and Float64 is read with one
25/// point read per row for most rows, not two.
26pub struct PointFields(Vec<(String, Source)>);
27
28/// Where one point field is read from.
29enum Source {
30    /// The record id, from the record header.
31    Id,
32    /// The typed definitions of a scalar property, in probe order.
33    Pids(Vec<u64>),
34}
35
36impl PointFields {
37    /// Point reads for `names`, each row keyed by the name as given; `None`
38    /// when one of them needs the whole record. Definitions are probed by
39    /// their tenant-wide support.
40    ///
41    /// # Errors
42    /// Returns storage read errors from the property catalog.
43    pub fn resolve<'n>(
44        snap: &QuerySnapshot,
45        tenant: rushdb_types::ids::TenantId,
46        names: impl IntoIterator<Item = &'n str>,
47    ) -> Result<Option<Self>, QueryError> {
48        Self::resolve_in(snap, tenant, &[], names)
49    }
50
51    /// [`Self::resolve`] for records of `labels` (every label when empty):
52    /// definitions are probed by their support under those labels.
53    ///
54    /// # Errors
55    /// Returns storage read errors from the property catalog.
56    pub fn resolve_in<'n>(
57        snap: &QuerySnapshot,
58        tenant: rushdb_types::ids::TenantId,
59        labels: &[String],
60        names: impl IntoIterator<Item = &'n str>,
61    ) -> Result<Option<Self>, QueryError> {
62        let mut fields = Vec::new();
63        for name in names {
64            let source = if normalize(name) == ID_FIELD {
65                Source::Id
66            } else {
67                let Some(pids) = scalar_pids(snap, tenant, labels, name)? else {
68                    return Ok(None);
69                };
70                Source::Pids(pids)
71            };
72            fields.push((name.to_owned(), source));
73        }
74        Ok(Some(Self(fields)))
75    }
76
77    /// The properties `predicate` reads, or `None` when it needs the whole
78    /// record.
79    pub(super) fn for_predicate(
80        snap: &QuerySnapshot,
81        tenant: rushdb_types::ids::TenantId,
82        labels: &[String],
83        predicate: &Expr,
84    ) -> Result<Option<Self>, QueryError> {
85        let mut names = std::collections::BTreeSet::new();
86        if !predicate_fields(predicate, &mut names) {
87            return Ok(None);
88        }
89        Self::resolve_in(snap, tenant, labels, names.iter().map(String::as_str))
90    }
91
92    /// The fields of record `ord`: each one's value, as the rebuilt record
93    /// would carry it.
94    ///
95    /// # Errors
96    /// Returns storage read errors and corrupt-occurrence errors.
97    pub fn row(
98        &self,
99        snap: &QuerySnapshot,
100        tenant: rushdb_types::ids::TenantId,
101        ord: u64,
102    ) -> Result<Map<String, Json>, QueryError> {
103        let mut row = Map::new();
104        for (field, source) in &self.0 {
105            match source {
106                Source::Id => {
107                    if let Some(id) = record_id(snap, tenant, ord)? {
108                        row.insert(field.clone(), Json::String(id));
109                    }
110                }
111                Source::Pids(pids) => {
112                    if let Some(occ) = held(snap, tenant, ord, pids)? {
113                        row.insert(
114                            field.clone(),
115                            reconstruct::decode_value(occ.value, &occ.payload)?,
116                        );
117                    }
118                }
119            }
120        }
121        Ok(row)
122    }
123
124    /// The fields of record `ord` as typed cells, one per field in the order
125    /// the fields were resolved (`None`: the record does not hold it).
126    pub(crate) fn cells(
127        &self,
128        snap: &QuerySnapshot,
129        tenant: rushdb_types::ids::TenantId,
130        ord: u64,
131        out: &mut Vec<Option<Cell>>,
132    ) -> Result<(), QueryError> {
133        out.clear();
134        for (_, source) in &self.0 {
135            out.push(match source {
136                Source::Id => record_id(snap, tenant, ord)?.map(Cell::Str),
137                Source::Pids(pids) => match held(snap, tenant, ord, pids)? {
138                    Some(occ) => Some(Cell::decode(occ.value, &occ.payload)?),
139                    None => None,
140                },
141            });
142        }
143        Ok(())
144    }
145}
146
147/// The id of record `ord`, as the rebuilt record carries it.
148fn record_id(
149    snap: &QuerySnapshot,
150    tenant: rushdb_types::ids::TenantId,
151    ord: u64,
152) -> Result<Option<String>, QueryError> {
153    Ok(access::record_by_ord(snap, tenant, ord)?
154        .map(|header| reconstruct::id_string(&header.record_id)))
155}
156
157/// The occurrence record `ord` holds under one of `pids` (probed in order).
158fn held(
159    snap: &QuerySnapshot,
160    tenant: rushdb_types::ids::TenantId,
161    ord: u64,
162    pids: &[u64],
163) -> Result<Option<access::Occurrence>, QueryError> {
164    for &pid in pids {
165        if let Some(occ) = access::occurrence(
166            snap,
167            tenant,
168            rushdb_types::ids::OwnerKind::Record,
169            ord,
170            pid,
171            0,
172        )? {
173            return Ok(Some(occ));
174        }
175    }
176    Ok(None)
177}
178
179/// Collect the property names `expr` reads into `out`; `false` when it needs
180/// the whole record (`$id`). A traversal reads none of the record's
181/// properties: it starts from the record's ordinal.
182pub(super) fn predicate_fields(expr: &Expr, out: &mut std::collections::BTreeSet<String>) -> bool {
183    match expr {
184        Expr::And(items) | Expr::Or(items) | Expr::Nor(items) => {
185            items.iter().all(|item| predicate_fields(item, out))
186        }
187        Expr::Xor(a, b) => predicate_fields(a, out) && predicate_fields(b, out),
188        Expr::Not(inner) => predicate_fields(inner, out),
189        Expr::Related(_) => true,
190        Expr::Field(FieldPred::Id(_)) => false,
191        Expr::Field(
192            FieldPred::Eq(name, _)
193            | FieldPred::Ne(name, _)
194            | FieldPred::Gt(name, _)
195            | FieldPred::Gte(name, _)
196            | FieldPred::Lt(name, _)
197            | FieldPred::Lte(name, _)
198            | FieldPred::Contains(name, _)
199            | FieldPred::StartsWith(name, _)
200            | FieldPred::EndsWith(name, _)
201            | FieldPred::Exists(name, _)
202            | FieldPred::IsNull(name, _)
203            | FieldPred::In(name, _)
204            | FieldPred::Nin(name, _)
205            | FieldPred::Type(name, _)
206            | FieldPred::DatetimeComponent { field: name, .. },
207        ) => {
208            out.insert(name.clone());
209            true
210        }
211    }
212}
213
214/// Typed definitions holding record property `name` as a scalar, in probe
215/// order: most owners (under `labels`, every label when empty) first, ties
216/// by id. `None` when the name is a system field or any label stores the
217/// property as an array: those need the whole record.
218fn scalar_pids<R: TryRead>(
219    snap: &R,
220    tenant: rushdb_types::ids::TenantId,
221    labels: &[String],
222    name: &str,
223) -> Result<Option<Vec<u64>>, QueryError> {
224    let norm = normalize(name);
225    if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
226        return Ok(None);
227    }
228    let Some(fid) = family_id(snap, tenant, &norm)? else {
229        return Ok(Some(Vec::new()));
230    };
231    let kind = rushdb_types::ids::OwnerKind::Record;
232    let mut by_support = Vec::new();
233    for pid in pids_of(snap, tenant, fid)? {
234        let (arrays, _) = rushdb_lmpg::derived::array_markers(snap, tenant, kind, pid)?;
235        if arrays {
236            return Ok(None);
237        }
238        let support = if labels.is_empty() {
239            rushdb_lmpg::derived::prop_support_total(snap, tenant, kind, pid)?
240        } else {
241            let mut support = 0u64;
242            let distinct: std::collections::BTreeSet<&String> = labels.iter().collect();
243            for label in distinct {
244                support = support.saturating_add(rushdb_lmpg::derived::prop_support_label(
245                    snap, tenant, kind, pid, label,
246                )?);
247            }
248            support
249        };
250        by_support.push((std::cmp::Reverse(support), pid));
251    }
252    by_support.sort_unstable();
253    Ok(Some(by_support.into_iter().map(|(_, pid)| pid).collect()))
254}
255
256/// The properties a collect with a `fields` projection reads of each child:
257/// the projected sources, the ordering fields and the filter's fields, by
258/// either spelling. `None` (whole child) without `fields`, or when the
259/// filter needs the whole record (`$id`, a traversal).
260pub(super) fn collect_selection(
261    spec: &crate::CollectSpec,
262) -> Option<std::collections::HashSet<String>> {
263    if spec.fields.is_empty() {
264        return None;
265    }
266    let mut names = std::collections::BTreeSet::new();
267    if let Some(filter) = &spec.filter {
268        if !predicate_fields(filter, &mut names) {
269            return None;
270        }
271    }
272    for (_, source) in &spec.fields {
273        names.insert(source.strip_prefix("$record.").unwrap_or(source).to_owned());
274    }
275    for (field, _) in &spec.order_by {
276        names.insert(field.clone());
277    }
278    Some(
279        names
280            .into_iter()
281            .flat_map(|name| [normalize(&name), name])
282            .collect(),
283    )
284}
rushdb_query/point_fields.rs

rushdb_query/
point_fields.rs
