1use 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 serde_json::Map;
10
11pub struct PointFields(Vec<(String, Vec<u64>)>);
25
26impl PointFields {
27 pub fn resolve<'n>(
34 snap: &QuerySnapshot,
35 tenant: rushdb_types::ids::TenantId,
36 names: impl IntoIterator<Item = &'n str>,
37 ) -> Result<Option<Self>, QueryError> {
38 Self::resolve_in(snap, tenant, &[], names)
39 }
40
41 pub fn resolve_in<'n>(
47 snap: &QuerySnapshot,
48 tenant: rushdb_types::ids::TenantId,
49 labels: &[String],
50 names: impl IntoIterator<Item = &'n str>,
51 ) -> Result<Option<Self>, QueryError> {
52 let mut fields = Vec::new();
53 for name in names {
54 let Some(pids) = scalar_pids(snap, tenant, labels, name)? else {
55 return Ok(None);
56 };
57 fields.push((name.to_owned(), pids));
58 }
59 Ok(Some(Self(fields)))
60 }
61
62 pub(super) fn for_predicate(
65 snap: &QuerySnapshot,
66 tenant: rushdb_types::ids::TenantId,
67 labels: &[String],
68 predicate: &Expr,
69 ) -> Result<Option<Self>, QueryError> {
70 let mut names = std::collections::BTreeSet::new();
71 if !predicate_fields(predicate, &mut names) {
72 return Ok(None);
73 }
74 Self::resolve_in(snap, tenant, labels, names.iter().map(String::as_str))
75 }
76
77 pub fn row(
83 &self,
84 snap: &QuerySnapshot,
85 tenant: rushdb_types::ids::TenantId,
86 ord: u64,
87 ) -> Result<Map<String, Json>, QueryError> {
88 let mut row = Map::new();
89 for (field, pids) in &self.0 {
90 if let Some(occ) = held(snap, tenant, ord, pids)? {
91 row.insert(
92 field.clone(),
93 reconstruct::decode_value(occ.value, &occ.payload)?,
94 );
95 }
96 }
97 Ok(row)
98 }
99
100 pub(crate) fn cells(
103 &self,
104 snap: &QuerySnapshot,
105 tenant: rushdb_types::ids::TenantId,
106 ord: u64,
107 out: &mut Vec<Option<Cell>>,
108 ) -> Result<(), QueryError> {
109 out.clear();
110 for (_, pids) in &self.0 {
111 out.push(match held(snap, tenant, ord, pids)? {
112 Some(occ) => Some(Cell::decode(occ.value, &occ.payload)?),
113 None => None,
114 });
115 }
116 Ok(())
117 }
118}
119
120fn held(
122 snap: &QuerySnapshot,
123 tenant: rushdb_types::ids::TenantId,
124 ord: u64,
125 pids: &[u64],
126) -> Result<Option<access::Occurrence>, QueryError> {
127 for &pid in pids {
128 if let Some(occ) = access::occurrence(
129 snap,
130 tenant,
131 rushdb_types::ids::OwnerKind::Record,
132 ord,
133 pid,
134 0,
135 )? {
136 return Ok(Some(occ));
137 }
138 }
139 Ok(None)
140}
141
142pub(super) fn predicate_fields(expr: &Expr, out: &mut std::collections::BTreeSet<String>) -> bool {
146 match expr {
147 Expr::And(items) | Expr::Or(items) | Expr::Nor(items) => {
148 items.iter().all(|item| predicate_fields(item, out))
149 }
150 Expr::Xor(a, b) => predicate_fields(a, out) && predicate_fields(b, out),
151 Expr::Not(inner) => predicate_fields(inner, out),
152 Expr::Related(_) => true,
153 Expr::Field(FieldPred::Id(_)) => false,
154 Expr::Field(
155 FieldPred::Eq(name, _)
156 | FieldPred::Ne(name, _)
157 | FieldPred::Gt(name, _)
158 | FieldPred::Gte(name, _)
159 | FieldPred::Lt(name, _)
160 | FieldPred::Lte(name, _)
161 | FieldPred::Contains(name, _)
162 | FieldPred::StartsWith(name, _)
163 | FieldPred::EndsWith(name, _)
164 | FieldPred::Exists(name, _)
165 | FieldPred::IsNull(name, _)
166 | FieldPred::In(name, _)
167 | FieldPred::Nin(name, _)
168 | FieldPred::Type(name, _)
169 | FieldPred::DatetimeComponent { field: name, .. },
170 ) => {
171 out.insert(name.clone());
172 true
173 }
174 }
175}
176
177fn scalar_pids<R: TryRead>(
182 snap: &R,
183 tenant: rushdb_types::ids::TenantId,
184 labels: &[String],
185 name: &str,
186) -> Result<Option<Vec<u64>>, QueryError> {
187 let norm = normalize(name);
188 if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
189 return Ok(None);
190 }
191 let Some(fid) = family_id(snap, tenant, &norm)? else {
192 return Ok(Some(Vec::new()));
193 };
194 let kind = rushdb_types::ids::OwnerKind::Record;
195 let mut by_support = Vec::new();
196 for pid in pids_of(snap, tenant, fid)? {
197 let (arrays, _) = rushdb_lmpg::derived::array_markers(snap, tenant, kind, pid)?;
198 if arrays {
199 return Ok(None);
200 }
201 let support = if labels.is_empty() {
202 rushdb_lmpg::derived::prop_support_total(snap, tenant, kind, pid)?
203 } else {
204 let mut support = 0u64;
205 let distinct: std::collections::BTreeSet<&String> = labels.iter().collect();
206 for label in distinct {
207 support = support.saturating_add(rushdb_lmpg::derived::prop_support_label(
208 snap, tenant, kind, pid, label,
209 )?);
210 }
211 support
212 };
213 by_support.push((std::cmp::Reverse(support), pid));
214 }
215 by_support.sort_unstable();
216 Ok(Some(by_support.into_iter().map(|(_, pid)| pid).collect()))
217}
218
219pub(super) fn collect_selection(
224 spec: &crate::CollectSpec,
225) -> Option<std::collections::HashSet<String>> {
226 if spec.fields.is_empty() {
227 return None;
228 }
229 let mut names = std::collections::BTreeSet::new();
230 if let Some(filter) = &spec.filter {
231 if !predicate_fields(filter, &mut names) {
232 return None;
233 }
234 }
235 for (_, source) in &spec.fields {
236 names.insert(source.strip_prefix("$record.").unwrap_or(source).to_owned());
237 }
238 for (field, _) in &spec.order_by {
239 names.insert(field.clone());
240 }
241 Some(
242 names
243 .into_iter()
244 .flat_map(|name| [normalize(&name), name])
245 .collect(),
246 )
247}