1use super::{
4 access, canonical_id, hex_decode_16, normalize, scalar_equal, Expr, FieldPred, Json, Lmpg,
5 LogicalType, QueryError, QuerySnapshot, Scalar, SearchRequest, TypeState,
6};
7use rushdb_lmpg::reader::TryRead;
8
9mod explain;
10pub use explain::{explain_cardinality, explain_walk};
11
12pub(super) struct IndexPlan {
22 pub(super) ords: OwnerSet,
24 pub(super) exact: bool,
25 pub(super) known_total: Option<u64>,
29 pub(super) residual: Option<Expr>,
34}
35
36const PLANNER_MATERIALIZE_CAP: u64 = 2_000_000;
43
44pub const UNINDEXED_WALK_THRESHOLD: u64 = 10_000;
48
49static UNINDEXED_WALKS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
50
51#[must_use]
54pub fn unindexed_walks_total() -> u64 {
55 UNINDEXED_WALKS.load(std::sync::atomic::Ordering::Relaxed)
56}
57
58const PLANNER_BOUNDED_PAGE_CAP: usize = 100_000;
61
62#[expect(
63 clippy::too_many_lines,
64 reason = "the planner is a single cohesive access-path selection routine; splitting it would obscure the fall-through ordering"
65)]
66pub(super) fn index_candidates_with_budget(
67 lmpg: &Lmpg,
68 snap: &QuerySnapshot,
69 req: &SearchRequest,
70 budget: Option<usize>,
71 is_cancelled: &dyn Fn() -> bool,
72) -> Result<Option<IndexPlan>, QueryError> {
73 let owner_limit = budget.map_or(usize::MAX, |bytes| bytes / OWNER_SET_BYTES_PER_OWNER);
74 if req.force_scan {
75 return Ok(None);
76 }
77 let Some(expr) = &req.where_ else {
78 return Ok(None);
79 };
80 let mut leaves: Vec<&Expr> = Vec::new();
81 collect_leaves(expr, &mut leaves);
82 if leaves.is_empty() {
83 return Ok(None);
84 }
85 let mut eligible: Vec<(u64, &Expr)> = Vec::new();
86 let mut residual: Vec<Expr> = Vec::new();
87 let mut has_residual = false;
88 let mut exactable = true;
89 for leaf in leaves {
90 if is_cancelled() {
91 return Err(QueryError::Cancelled);
92 }
93 if let Some(estimate) = leaf_estimate(snap, lmpg.tenant, leaf, &req.labels)? {
94 if leaf_is_null_only(leaf) {
95 residual.push(leaf.clone());
96 }
97 eligible.push((estimate, leaf));
98 } else {
99 residual.push(leaf.clone());
100 has_residual = true;
101 if leaf_is_null_only(leaf) {
102 exactable = false;
103 }
104 }
105 }
106 if eligible.is_empty() {
107 return Ok(None);
108 }
109 if !has_residual
112 && exactable
113 && req.related.is_empty()
114 && req.order_by.is_empty()
115 && eligible.len() == 1
116 {
117 if let Expr::Field(FieldPred::Exists(name, true)) = eligible[0].1 {
118 if let Some(limit) = req.pagination.limit {
119 let page = req
120 .pagination
121 .skip
122 .saturating_add(limit)
123 .min(u64::try_from(PLANNER_BOUNDED_PAGE_CAP).unwrap_or(u64::MAX));
124 let page = usize::try_from(page).unwrap_or(PLANNER_BOUNDED_PAGE_CAP);
125 if let Some(total) = total_support(snap, lmpg.tenant, name, &req.labels)? {
126 let ords =
127 owners_with_presence_bounded(snap, lmpg.tenant, name, &req.labels, page)?;
128 return Ok(Some(IndexPlan {
129 ords: ords.into_iter().collect(),
130 exact: true,
131 known_total: Some(total),
132 residual: None,
133 }));
134 }
135 }
136 }
137 }
138 eligible.sort_by_key(|(estimate, _)| *estimate);
139 if eligible[0].0 != u64::MAX
140 && eligible[0].0 > PLANNER_MATERIALIZE_CAP
141 && !crate::index_access::answers(snap, lmpg.tenant, eligible[0].1, &req.labels)?
142 {
143 return Ok(None);
146 }
147 let label_scoped = req.labels.is_empty()
151 || eligible
152 .iter()
153 .any(|(_, leaf)| !is_id_leaf(leaf) && !matches!(leaf, Expr::Or(_)));
154 let mut acc: Option<OwnerSet> = None;
155 for (_, leaf) in eligible {
156 if is_cancelled() {
157 return Err(QueryError::Cancelled);
158 }
159 if acc.as_ref().is_some_and(OwnerSet::is_empty) {
160 break;
161 }
162 let scope = LeafScope {
165 labels: &req.labels,
166 owner_limit,
167 candidates: acc.as_ref(),
168 };
169 let Some(owners) = leaf_owners(snap, lmpg.tenant, leaf, &scope)? else {
170 return Ok(None);
172 };
173 acc = Some(match acc {
174 None => owners,
175 Some(prev) => prev & owners,
176 });
177 }
178 let mut out = acc.unwrap_or_default();
179 if !label_scoped {
180 let mut outside = Vec::new();
182 for ord in &out {
183 if is_cancelled() {
184 return Err(QueryError::Cancelled);
185 }
186 let mut allowed = false;
187 for label in &req.labels {
188 let key = rushdb_types::keys::label_member_key(lmpg.tenant, label, ord);
189 if snap.get(&key)?.is_some() {
190 allowed = true;
191 break;
192 }
193 }
194 if !allowed {
195 outside.push(ord);
196 }
197 }
198 for ord in outside {
199 out.remove(ord);
200 }
201 }
202 let exact = !has_residual && exactable && req.related.is_empty();
203 Ok(Some(IndexPlan {
204 ords: out,
205 exact,
206 known_total: None,
207 residual: match residual.len() {
208 0 => None,
209 1 => residual.pop(),
210 _ => Some(Expr::And(residual)),
211 },
212 }))
213}
214
215pub(super) fn collect_leaves<'a>(expr: &'a Expr, out: &mut Vec<&'a Expr>) {
219 match expr {
220 Expr::And(items) => {
221 for item in items {
222 collect_leaves(item, out);
223 }
224 }
225 _ => out.push(expr),
226 }
227}
228
229fn leaf_is_null_only(expr: &Expr) -> bool {
233 matches!(expr, Expr::Field(FieldPred::Eq(_, value)) if value.is_null())
234 || matches!(expr, Expr::Field(FieldPred::IsNull(_, true)))
235}
236
237fn leaf_estimate<R: TryRead>(
241 snap: &R,
242 tenant: rushdb_types::ids::TenantId,
243 expr: &Expr,
244 labels: &[String],
245) -> Result<Option<u64>, QueryError> {
246 if matches!(expr, Expr::Field(FieldPred::StartsWith(..))) {
248 return Ok(crate::index_access::answers(snap, tenant, expr, labels)?.then_some(u64::MAX));
249 }
250 Ok(match expr {
251 Expr::Field(FieldPred::Eq(_, value)) if value.is_null() => Some(u64::MAX),
252 Expr::Field(FieldPred::Eq(name, value)) => {
253 estimate_values(snap, tenant, name, std::slice::from_ref(value), labels)?
254 }
255 Expr::Field(FieldPred::In(name, values)) if values.iter().all(|v| !v.is_null()) => {
256 estimate_values(snap, tenant, name, values, labels)?
257 }
258 Expr::Field(FieldPred::Id(_)) => Some(1),
260 Expr::Or(items) if !items.is_empty() && items.iter().all(is_id_leaf) => {
263 Some(items.len() as u64)
264 }
265 Expr::Field(FieldPred::IsNull(_, true) | FieldPred::Exists(_, true)) => Some(u64::MAX),
266 Expr::Field(FieldPred::Gt(name, bound)) => {
269 range_leaf_estimate(snap, tenant, name, RangeOp::Gt, *bound)?
270 }
271 Expr::Field(FieldPred::Gte(name, bound)) => {
272 range_leaf_estimate(snap, tenant, name, RangeOp::Gte, *bound)?
273 }
274 Expr::Field(FieldPred::Lt(name, bound)) => {
275 range_leaf_estimate(snap, tenant, name, RangeOp::Lt, *bound)?
276 }
277 Expr::Field(FieldPred::Lte(name, bound)) => {
278 range_leaf_estimate(snap, tenant, name, RangeOp::Lte, *bound)?
279 }
280 _ => None,
281 })
282}
283
284fn is_range_leaf(expr: &Expr) -> bool {
290 matches!(
291 expr,
292 Expr::Field(
293 FieldPred::Gt(..) | FieldPred::Gte(..) | FieldPred::Lt(..) | FieldPred::Lte(..)
294 )
295 )
296}
297
298fn range_leaf_histogram_estimate<R: TryRead>(
304 snap: &R,
305 tenant: rushdb_types::ids::TenantId,
306 name: &str,
307 op: RangeOp,
308 bound: f64,
309) -> Result<Option<(u64, u64)>, QueryError> {
310 let norm = normalize(name);
311 if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
312 return Ok(None);
313 }
314 let Some(fid) = family_id(snap, tenant, &norm)? else {
315 return Ok(None);
316 };
317 let above = matches!(op, RangeOp::Gt | RangeOp::Gte);
318 for ty in [
319 LogicalType::Int64,
320 LogicalType::Float64,
321 LogicalType::DateTime,
322 ] {
323 let Some(pid) = property_id(snap, tenant, fid, ty)? else {
324 continue;
325 };
326 let Some(hist) = crate::stats_cache::distribution(
327 snap,
328 tenant,
329 rushdb_types::ids::OwnerKind::Record,
330 None,
331 pid,
332 )?
333 else {
334 continue;
335 };
336 let scalar = match ty {
337 LogicalType::Float64 => Scalar::Float64(bound),
338 LogicalType::Int64 => Scalar::Int64(round_bound(bound, above)),
339 LogicalType::DateTime => Scalar::DateTime(round_bound(bound, above)),
340 LogicalType::Boolean | LogicalType::Utf8 => continue,
341 };
342 let Ok(enc) = rushdb_types::encoding::encode_order_preserving(ty, &scalar) else {
343 continue;
344 };
345 if !hist.partial && hist.distinct <= LOW_CARDINALITY_EXACT_THRESHOLD {
348 let exact = exact_low_cardinality_range_count(snap, tenant, pid, ty, op, bound)?;
349 return Ok(Some((exact, hist.total())));
350 }
351 return Ok(Some((hist.estimate(&enc, above), hist.total())));
352 }
353 Ok(None)
354}
355
356const LOW_CARDINALITY_EXACT_THRESHOLD: u64 = 256;
358
359fn exact_low_cardinality_range_count<R: TryRead>(
362 snap: &R,
363 tenant: rushdb_types::ids::TenantId,
364 pid: u64,
365 ty: LogicalType,
366 op: RangeOp,
367 bound: f64,
368) -> Result<u64, QueryError> {
369 let tag = rushdb_types::encoding::logical_tag(ty);
370 let mut total = 0u64;
371 for (entry, count) in rushdb_lmpg::derived::dictionary(
372 snap,
373 tenant,
374 rushdb_types::ids::OwnerKind::Record,
375 pid,
376 &[],
377 )? {
378 let Some((&entry_tag, payload)) = entry.split_first() else {
379 continue;
380 };
381 if entry_tag != tag {
382 continue;
383 }
384 let Ok(scalar) = rushdb_types::encoding::decode_order_preserving(ty, payload) else {
385 continue;
386 };
387 if range_scalar_matches(&scalar, op, bound) {
388 total = total.saturating_add(count);
389 }
390 }
391 Ok(total)
392}
393
394fn range_leaf_estimate<R: TryRead>(
397 snap: &R,
398 tenant: rushdb_types::ids::TenantId,
399 name: &str,
400 op: RangeOp,
401 bound: f64,
402) -> Result<Option<u64>, QueryError> {
403 Ok(Some(
404 range_leaf_histogram_estimate(snap, tenant, name, op, bound)?
405 .map_or(u64::MAX, |(estimate, _total)| estimate),
406 ))
407}
408
409#[expect(
420 clippy::cast_precision_loss,
421 reason = "estimate/total feed a selectivity fraction, not an exact count; this is \
422 approximate by design"
423)]
424pub fn explain_range_selectivity<R: TryRead>(
425 snap: &R,
426 tenant: rushdb_types::ids::TenantId,
427 where_: &Expr,
428) -> Result<Option<f64>, QueryError> {
429 let mut leaves = Vec::new();
430 collect_leaves(where_, &mut leaves);
431 for leaf in leaves {
432 let Expr::Field(pred) = leaf else {
433 continue;
434 };
435 let (name, op, bound) = match pred {
436 FieldPred::Gt(name, bound) => (name, RangeOp::Gt, *bound),
437 FieldPred::Gte(name, bound) => (name, RangeOp::Gte, *bound),
438 FieldPred::Lt(name, bound) => (name, RangeOp::Lt, *bound),
439 FieldPred::Lte(name, bound) => (name, RangeOp::Lte, *bound),
440 _ => continue,
441 };
442 if let Some((estimate, total)) =
443 range_leaf_histogram_estimate(snap, tenant, name, op, bound)?
444 {
445 if total > 0 {
446 return Ok(Some((estimate as f64 / total as f64).clamp(0.0, 1.0)));
447 }
448 }
449 }
450 Ok(None)
451}
452
453pub(crate) type OwnerSet = roaring::RoaringTreemap;
457
458pub(crate) const OWNER_SET_BYTES_PER_OWNER: usize = 4;
462
463pub(super) fn extend_owners(
464 owners: &mut OwnerSet,
465 additions: impl IntoIterator<Item = u64>,
466 limit: usize,
467) -> Result<(), QueryError> {
468 let mut owners = BoundedOwners::new(owners, limit);
469 for owner in additions {
470 owners.insert(owner)?;
471 }
472 Ok(())
473}
474
475pub(super) struct BoundedOwners<'a> {
479 owners: &'a mut OwnerSet,
480 len: u64,
481 limit: u64,
482}
483
484impl<'a> BoundedOwners<'a> {
485 pub(super) fn new(owners: &'a mut OwnerSet, limit: usize) -> Self {
486 let len = owners.len();
487 Self {
488 owners,
489 len,
490 limit: u64::try_from(limit).unwrap_or(u64::MAX),
491 }
492 }
493
494 pub(super) fn contains(&self, owner: u64) -> bool {
495 self.owners.contains(owner)
496 }
497
498 pub(super) fn insert(&mut self, owner: u64) -> Result<(), QueryError> {
500 if self.owners.contains(owner) {
501 return Ok(());
502 }
503 if self.len >= self.limit {
504 return Err(QueryError::ResourceLimit(
505 "aggregate candidate budget exceeded".into(),
506 ));
507 }
508 self.owners.insert(owner);
509 self.len += 1;
510 Ok(())
511 }
512}
513
514fn leaf_owners<R: TryRead>(
520 snap: &R,
521 tenant: rushdb_types::ids::TenantId,
522 expr: &Expr,
523 scope: &LeafScope<'_>,
524) -> Result<Option<OwnerSet>, QueryError> {
525 let (labels, owner_limit) = (scope.labels, scope.owner_limit);
526 if let Some(owners) = crate::index_access::leaf_owners(snap, tenant, expr, labels, owner_limit)?
527 {
528 return Ok(Some(owners));
529 }
530 match expr {
531 Expr::Field(FieldPred::Eq(name, value)) if value.is_null() => {
532 owners_with_presence(snap, tenant, name, true, labels, owner_limit)
533 }
534 Expr::Field(FieldPred::Eq(name, value)) => {
535 owners_for_values(snap, tenant, name, std::slice::from_ref(value), scope)
536 }
537 Expr::Field(FieldPred::In(name, values)) if values.iter().all(|v| !v.is_null()) => {
538 owners_for_values(snap, tenant, name, values, scope)
539 }
540 Expr::Field(FieldPred::Id(id)) => Ok(Some(owners_for_id(snap, tenant, id)?)),
541 Expr::Or(items) if !items.is_empty() && items.iter().all(is_id_leaf) => {
542 let mut out = OwnerSet::new();
543 for item in items {
544 if let Expr::Field(FieldPred::Id(id)) = item {
545 extend_owners(&mut out, owners_for_id(snap, tenant, id)?, owner_limit)?;
546 }
547 }
548 Ok(Some(out))
549 }
550 Expr::Field(FieldPred::Exists(name, true)) => {
551 owners_with_presence(snap, tenant, name, false, labels, owner_limit)
552 }
553 Expr::Field(FieldPred::IsNull(name, true)) => {
554 owners_with_presence(snap, tenant, name, true, labels, owner_limit)
555 }
556 Expr::Field(FieldPred::Gt(name, bound)) => {
557 owners_with_range(snap, tenant, name, RangeOp::Gt, *bound, scope)
558 }
559 Expr::Field(FieldPred::Gte(name, bound)) => {
560 owners_with_range(snap, tenant, name, RangeOp::Gte, *bound, scope)
561 }
562 Expr::Field(FieldPred::Lt(name, bound)) => {
563 owners_with_range(snap, tenant, name, RangeOp::Lt, *bound, scope)
564 }
565 Expr::Field(FieldPred::Lte(name, bound)) => {
566 owners_with_range(snap, tenant, name, RangeOp::Lte, *bound, scope)
567 }
568 _ => Ok(None),
569 }
570}
571
572fn is_id_leaf(expr: &Expr) -> bool {
574 matches!(expr, Expr::Field(FieldPred::Id(_)))
575}
576
577fn owners_for_id<R: TryRead>(
580 snap: &R,
581 tenant: rushdb_types::ids::TenantId,
582 id: &str,
583) -> Result<OwnerSet, QueryError> {
584 let canonical = canonical_id(id);
585 let Some(bytes) = hex_decode_16(&canonical) else {
586 return Ok(OwnerSet::new());
587 };
588 let rid = rushdb_types::ids::RecordId(bytes);
589 Ok(access::record_by_id(snap, tenant, &rid)?
590 .map(|header| header.ordinal)
591 .into_iter()
592 .collect())
593}
594
595fn estimate_values<R: TryRead>(
598 snap: &R,
599 tenant: rushdb_types::ids::TenantId,
600 name: &str,
601 values: &[Json],
602 labels: &[String],
603) -> Result<Option<u64>, QueryError> {
604 let norm = normalize(name);
605 if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
606 return Ok(None);
607 }
608 let Some(fid) = family_id(snap, tenant, &norm)? else {
609 return Ok(Some(0));
610 };
611 let Some((pids, encodings)) = value_probes(snap, tenant, fid, values)? else {
612 return Ok(None);
613 };
614 let mut total = 0u64;
615 for (ty, enc) in &encodings {
616 let mut entry = vec![rushdb_types::encoding::logical_tag(*ty)];
617 entry.extend_from_slice(enc);
618 if entry.len() > rushdb_types::keys::DICT_MAX_VALUE_BYTES {
619 return Ok(Some(u64::MAX));
621 }
622 for &pid in &pids {
623 for label in labels_for(snap, tenant, pid, labels)? {
624 total = total.saturating_add(rushdb_lmpg::derived::value_count(
625 snap,
626 tenant,
627 rushdb_types::ids::OwnerKind::Record,
628 &label,
629 pid,
630 &entry,
631 )?);
632 }
633 }
634 }
635 Ok(Some(total))
636}
637
638fn labels_for<R: TryRead>(
641 snap: &R,
642 tenant: rushdb_types::ids::TenantId,
643 pid: u64,
644 labels: &[String],
645) -> Result<Vec<String>, QueryError> {
646 if labels.is_empty() {
647 Ok(rushdb_lmpg::derived::labels_of_pid(
648 snap,
649 tenant,
650 rushdb_types::ids::OwnerKind::Record,
651 pid,
652 )?)
653 } else {
654 Ok(labels.to_vec())
655 }
656}
657
658pub(super) struct LeafScope<'a> {
661 pub(super) labels: &'a [String],
662 pub(super) owner_limit: usize,
663 pub(super) candidates: Option<&'a OwnerSet>,
664}
665
666fn owners_matching<R: TryRead>(
672 snap: &R,
673 tenant: rushdb_types::ids::TenantId,
674 pid: u64,
675 scope: &LeafScope<'_>,
676 owners: &mut OwnerSet,
677 keep: &dyn Fn(&access::Occurrence) -> bool,
678) -> Result<bool, QueryError> {
679 let kind = rushdb_types::ids::OwnerKind::Record;
680 let labels = labels_for(snap, tenant, pid, scope.labels)?;
681 let mut population = 0u64;
682 for label in &labels {
683 population = population.saturating_add(rushdb_lmpg::derived::prop_support_label(
684 snap, tenant, kind, pid, label,
685 )?);
686 }
687 let (_, array_labels) = rushdb_lmpg::derived::array_markers(snap, tenant, kind, pid)?;
691 let matches = |owner: u64, scalar: bool| -> Result<bool, QueryError> {
692 Ok(if scalar {
693 access::occurrence(snap, tenant, kind, owner, pid, 0)?.is_some_and(|occ| keep(&occ))
694 } else {
695 access::occurrences_of_pid(snap, tenant, kind, owner, pid)?
696 .iter()
697 .any(keep)
698 })
699 };
700 let mut owners = BoundedOwners::new(owners, scope.owner_limit);
701 if let Some(candidates) = scope.candidates.filter(|c| c.len() < population) {
702 let scalar = array_labels.is_empty();
703 for owner in candidates {
704 if !owners.contains(owner) && matches(owner, scalar)? {
705 owners.insert(owner)?;
706 }
707 }
708 return Ok(true);
709 }
710 if population > PLANNER_MATERIALIZE_CAP {
711 return Ok(false);
712 }
713 if population > UNINDEXED_WALK_THRESHOLD {
714 UNINDEXED_WALKS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
715 }
716 for label in &labels {
717 let scalar = !array_labels.contains(label);
718 for owner in rushdb_lmpg::derived::presence_owners(
719 snap,
720 tenant,
721 kind,
722 label,
723 pid,
724 rushdb_types::keys::NULLNESS_VALUE,
725 )? {
726 if !owners.contains(owner) && matches(owner, scalar)? {
727 owners.insert(owner)?;
728 }
729 }
730 }
731 Ok(true)
732}
733
734fn owners_with_range<R: TryRead>(
737 snap: &R,
738 tenant: rushdb_types::ids::TenantId,
739 name: &str,
740 op: RangeOp,
741 bound: f64,
742 scope: &LeafScope<'_>,
743) -> Result<Option<OwnerSet>, QueryError> {
744 let norm = normalize(name);
745 if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
746 return Ok(None);
747 }
748 let Some(fid) = family_id(snap, tenant, &norm)? else {
749 return Ok(Some(OwnerSet::new()));
750 };
751 let mut owners = OwnerSet::new();
752 for ty in [
753 LogicalType::Int64,
754 LogicalType::Float64,
755 LogicalType::DateTime,
756 ] {
757 let Some(pid) = property_id(snap, tenant, fid, ty)? else {
758 continue;
759 };
760 let keep = |occ: &access::Occurrence| {
761 matches!(occ.value, rushdb_types::occurrence::OccurrenceValue::Stored { ty: stored } if stored == ty)
762 && rushdb_types::encoding::decode_order_preserving(ty, &occ.payload)
763 .is_ok_and(|scalar| range_scalar_matches(&scalar, op, bound))
764 };
765 if !owners_matching(snap, tenant, pid, scope, &mut owners, &keep)? {
766 return Ok(None);
767 }
768 }
769 Ok(Some(owners))
770}
771
772#[derive(Clone, Copy)]
773pub(super) enum RangeOp {
774 Gt,
775 Gte,
776 Lt,
777 Lte,
778}
779
780#[expect(
783 clippy::cast_possible_truncation,
784 reason = "floor/ceil already made the value integral; `as` saturates at the i64 edges"
785)]
786fn round_bound(bound: f64, lower: bool) -> i64 {
787 if lower {
788 bound.floor() as i64
789 } else {
790 bound.ceil() as i64
791 }
792}
793
794pub(super) fn range_scalar_matches(scalar: &Scalar, op: RangeOp, bound: f64) -> bool {
795 let scalar = match scalar {
799 Scalar::DateTime(value) => Scalar::Int64(*value),
800 other => other.clone(),
801 };
802 let Some(ordering) = rushdb_types::value::scalar_cmp(&scalar, &Scalar::Float64(bound)) else {
803 return false;
804 };
805 match op {
806 RangeOp::Gt => ordering == std::cmp::Ordering::Greater,
807 RangeOp::Gte => matches!(
808 ordering,
809 std::cmp::Ordering::Greater | std::cmp::Ordering::Equal
810 ),
811 RangeOp::Lt => ordering == std::cmp::Ordering::Less,
812 RangeOp::Lte => matches!(
813 ordering,
814 std::cmp::Ordering::Less | std::cmp::Ordering::Equal
815 ),
816 }
817}
818
819pub(super) fn family_id<R: TryRead>(
820 snap: &R,
821 tenant: rushdb_types::ids::TenantId,
822 normalized: &str,
823) -> Result<Option<u64>, QueryError> {
824 use rushdb_lmpg::reader::keys_ext as kx;
825 let sig = kx::family_sig_key(
826 tenant,
827 "default",
828 rushdb_types::ids::OwnerKind::Record,
829 normalized,
830 );
831 let Some(raw) = snap.get(&sig)? else {
832 return Ok(None);
833 };
834 Ok(Some(u64::from_be_bytes(
835 raw.as_slice()
836 .try_into()
837 .map_err(|_| rushdb_storage::StorageError::CorruptLog)?,
838 )))
839}
840
841fn property_id<R: TryRead>(
842 snap: &R,
843 tenant: rushdb_types::ids::TenantId,
844 fid: u64,
845 ty: LogicalType,
846) -> Result<Option<u64>, QueryError> {
847 use rushdb_lmpg::reader::keys_ext as kx;
848 let Some(raw) = snap.get(&kx::definition_key(tenant, fid, TypeState::Known(ty)))? else {
849 return Ok(None);
850 };
851 Ok(Some(u64::from_be_bytes(
852 raw.as_slice()
853 .try_into()
854 .map_err(|_| rushdb_storage::StorageError::CorruptLog)?,
855 )))
856}
857
858fn all_type_states() -> Vec<TypeState> {
860 let mut states = vec![TypeState::Unresolved];
861 states.extend(
862 [
863 LogicalType::Boolean,
864 LogicalType::Int64,
865 LogicalType::Float64,
866 LogicalType::Utf8,
867 LogicalType::DateTime,
868 ]
869 .into_iter()
870 .map(TypeState::Known),
871 );
872 states
873}
874
875pub(super) fn pids_of<R: TryRead>(
877 snap: &R,
878 tenant: rushdb_types::ids::TenantId,
879 fid: u64,
880) -> Result<Vec<u64>, QueryError> {
881 use rushdb_lmpg::reader::keys_ext as kx;
882 let mut pids = Vec::new();
883 for state in all_type_states() {
884 let Some(raw) = snap.get(&kx::definition_key(tenant, fid, state))? else {
885 continue;
886 };
887 pids.push(u64::from_be_bytes(
888 raw.as_slice()
889 .try_into()
890 .map_err(|_| rushdb_storage::StorageError::CorruptLog)?,
891 ));
892 }
893 Ok(pids)
894}
895
896fn presence_prefixes<R: TryRead>(
900 snap: &R,
901 tenant: rushdb_types::ids::TenantId,
902 name: &str,
903 null_only: bool,
904 labels: &[String],
905) -> Result<Option<Vec<Vec<u8>>>, QueryError> {
906 let norm = normalize(name);
907 if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
908 return Ok(None);
909 }
910 let Some(fid) = family_id(snap, tenant, &norm)? else {
911 return Ok(Some(Vec::new()));
912 };
913 let mut prefixes = Vec::new();
914 for pid in pids_of(snap, tenant, fid)? {
915 for label in labels_for(snap, tenant, pid, labels)? {
916 let nullness: &[u8] = if null_only {
917 &[rushdb_types::keys::NULLNESS_NULL]
918 } else {
919 &[
920 rushdb_types::keys::NULLNESS_VALUE,
921 rushdb_types::keys::NULLNESS_NULL,
922 ]
923 };
924 for &nullness in nullness {
925 prefixes.push(rushdb_types::keys::presence_nullness_prefix(
926 tenant,
927 rushdb_types::ids::OwnerKind::Record,
928 &label,
929 pid,
930 nullness,
931 ));
932 }
933 }
934 }
935 Ok(Some(prefixes))
936}
937
938pub(super) fn owners_with_presence<R: TryRead>(
942 snap: &R,
943 tenant: rushdb_types::ids::TenantId,
944 name: &str,
945 null_only: bool,
946 labels: &[String],
947 owner_limit: usize,
948) -> Result<Option<OwnerSet>, QueryError> {
949 let Some(prefixes) = presence_prefixes(snap, tenant, name, null_only, labels)? else {
950 return Ok(None);
951 };
952 let mut set = OwnerSet::new();
953 let mut owners = BoundedOwners::new(&mut set, owner_limit);
954 for prefix in prefixes {
955 let (start, end) = rushdb_storage::prefix_range(&prefix);
956 let mut failed = None;
957 snap.for_each_key(&start, &end, &mut |key| {
958 let result = rushdb_types::keys::read_ord_tail(key, prefix.len())
959 .ok_or_else(|| rushdb_storage::StorageError::CorruptLog.into())
960 .and_then(|owner| owners.insert(owner));
961 match result {
962 Ok(()) => true,
963 Err(error) => {
964 failed = Some(error);
965 false
966 }
967 }
968 })?;
969 if let Some(error) = failed {
970 return Err(error);
971 }
972 }
973 Ok(Some(set))
974}
975
976fn owners_for_values<R: TryRead>(
980 snap: &R,
981 tenant: rushdb_types::ids::TenantId,
982 name: &str,
983 values: &[Json],
984 scope: &LeafScope<'_>,
985) -> Result<Option<OwnerSet>, QueryError> {
986 let norm = normalize(name);
987 if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
989 return Ok(None);
990 }
991 let Some(fid) = family_id(snap, tenant, &norm)? else {
992 return Ok(Some(OwnerSet::new()));
993 };
994 let Some((pids, encodings)) = value_probes(snap, tenant, fid, values)? else {
995 return Ok(None);
996 };
997 let keep = |occ: &access::Occurrence| {
998 encodings.iter().any(|(ty, enc)| {
999 matches!(occ.value, rushdb_types::occurrence::OccurrenceValue::Stored { ty: stored } if stored == *ty)
1000 && occ.payload == *enc
1001 })
1002 };
1003 let mut owners = OwnerSet::new();
1004 for pid in pids {
1005 if !owners_matching(snap, tenant, pid, scope, &mut owners, &keep)? {
1006 return Ok(None);
1007 }
1008 }
1009 Ok(Some(owners))
1010}
1011
1012type ValueProbes = (std::collections::BTreeSet<u64>, Vec<(LogicalType, Vec<u8>)>);
1018
1019fn value_probes<R: TryRead>(
1020 snap: &R,
1021 tenant: rushdb_types::ids::TenantId,
1022 fid: u64,
1023 values: &[Json],
1024) -> Result<Option<ValueProbes>, QueryError> {
1025 let mut pids = std::collections::BTreeSet::new();
1026 let mut encodings = Vec::new();
1027 for value in values {
1028 let Some(variants) = scalar_variants(value) else {
1029 return Ok(None);
1030 };
1031 for (ty, scalar) in variants {
1032 let Ok(enc) = rushdb_types::encoding::encode_order_preserving(ty, &scalar) else {
1033 return Ok(None);
1034 };
1035 encodings.push((ty, enc));
1036 if let Some(pid) = property_id(snap, tenant, fid, ty)? {
1037 pids.insert(pid);
1038 }
1039 }
1040 }
1041 Ok(Some((pids, encodings)))
1042}
1043
1044pub(super) fn total_support<R: TryRead>(
1048 snap: &R,
1049 tenant: rushdb_types::ids::TenantId,
1050 name: &str,
1051 labels: &[String],
1052) -> Result<Option<u64>, QueryError> {
1053 let norm = normalize(name);
1054 if norm.is_empty() || rushdb_types::value::is_reserved_name(&norm) {
1055 return Ok(None);
1056 }
1057 let Some(fid) = family_id(snap, tenant, &norm)? else {
1058 return Ok(Some(0));
1059 };
1060 let kind = rushdb_types::ids::OwnerKind::Record;
1061 let mut total = 0u64;
1062 for pid in pids_of(snap, tenant, fid)? {
1063 if labels.is_empty() {
1064 total = total.saturating_add(rushdb_lmpg::derived::prop_support_total(
1065 snap, tenant, kind, pid,
1066 )?);
1067 } else {
1068 for label in labels {
1069 total = total.saturating_add(rushdb_lmpg::derived::prop_support_label(
1070 snap, tenant, kind, pid, label,
1071 )?);
1072 }
1073 }
1074 }
1075 Ok(Some(total))
1076}
1077
1078pub(super) fn owners_with_presence_bounded(
1082 snap: &rushdb_storage::ReadView,
1083 tenant: rushdb_types::ids::TenantId,
1084 name: &str,
1085 labels: &[String],
1086 k: usize,
1087) -> Result<Vec<u64>, QueryError> {
1088 if k == 0 {
1089 return Ok(Vec::new());
1090 }
1091 let Some(prefixes) = presence_prefixes(snap, tenant, name, false, labels)? else {
1092 return Ok(Vec::new());
1093 };
1094 let mut cursors: Vec<rushdb_storage::ScanCursor<'_>> = Vec::new();
1095 for prefix in &prefixes {
1096 cursors.push(snap.scan(
1097 rushdb_storage::ScanBounds::prefix(prefix),
1098 rushdb_storage::ScanProjection::Keys,
1099 )?);
1100 }
1101 let mut pending: Vec<Option<u64>> = Vec::with_capacity(cursors.len());
1102 for cursor in &mut cursors {
1103 pending.push(next_presence_owner(cursor)?);
1104 }
1105 let mut seen = std::collections::HashSet::new();
1106 let mut out = Vec::with_capacity(k.min(4096));
1107 while out.len() < k {
1108 let Some(i) = pending
1109 .iter()
1110 .enumerate()
1111 .filter_map(|(i, owner)| owner.map(|owner| (owner, i)))
1112 .min()
1113 .map(|(_, i)| i)
1114 else {
1115 break;
1116 };
1117 let owner = pending[i]
1118 .take()
1119 .ok_or(rushdb_storage::StorageError::CorruptLog)?;
1120 pending[i] = next_presence_owner(&mut cursors[i])?;
1121 if seen.insert(owner) {
1122 out.push(owner);
1123 }
1124 }
1125 Ok(out)
1126}
1127
1128fn next_presence_owner(
1129 cursor: &mut rushdb_storage::ScanCursor<'_>,
1130) -> Result<Option<u64>, QueryError> {
1131 let batch = cursor.next_batch(1, 1024 * 1024)?;
1132 batch
1133 .rows
1134 .into_iter()
1135 .next()
1136 .map(|row| {
1137 rushdb_types::keys::parse_presence_key(&row.key)
1138 .map(|(_, _, _, _, owner)| owner)
1139 .ok_or_else(|| rushdb_storage::StorageError::CorruptLog.into())
1140 })
1141 .transpose()
1142}
1143
1144#[expect(
1150 clippy::cast_precision_loss,
1151 clippy::cast_possible_truncation,
1152 reason = "the f64 range guards against i64 bounds before narrowing an integral JSON number back to i64"
1153)]
1154pub(super) fn scalar_variants(value: &Json) -> Option<Vec<(LogicalType, Scalar)>> {
1155 match value {
1156 Json::String(s) => {
1158 let mut out = vec![(LogicalType::Utf8, Scalar::Utf8(s.clone()))];
1159 if let Some(ms) = crate::datetime::parse_iso_8601(s) {
1160 out.push((LogicalType::DateTime, Scalar::DateTime(ms)));
1161 }
1162 Some(out)
1163 }
1164 Json::Bool(b) => Some(vec![(LogicalType::Boolean, Scalar::Boolean(*b))]),
1165 Json::Number(n) => {
1166 let mut out = Vec::new();
1167 let push_integer = |out: &mut Vec<(LogicalType, Scalar)>, i: i64| {
1170 out.push((LogicalType::Int64, Scalar::Int64(i)));
1171 out.push((LogicalType::DateTime, Scalar::DateTime(i)));
1172 };
1173 if let Some(i) = n.as_i64() {
1176 push_integer(&mut out, i);
1177 }
1178 if let Some(f) = n.as_f64() {
1179 let matches_integer = n
1180 .as_i64()
1181 .is_some_and(|i| scalar_equal(&Scalar::Int64(i), &Scalar::Float64(f)));
1182 let is_fractional = n.as_i64().is_none() && !n.is_u64();
1183 if f.is_finite() && (matches_integer || is_fractional) {
1184 out.push((LogicalType::Float64, Scalar::Float64(f)));
1185 }
1186 if f == 0.0 {
1190 if !out
1191 .iter()
1192 .any(|(_, scalar)| matches!(scalar, Scalar::Int64(0)))
1193 {
1194 push_integer(&mut out, 0);
1195 }
1196 out.retain(
1197 |(_, scalar)| !matches!(scalar, Scalar::Float64(value) if *value == 0.0),
1198 );
1199 out.push((LogicalType::Float64, Scalar::Float64(-0.0)));
1200 out.push((LogicalType::Float64, Scalar::Float64(0.0)));
1201 }
1202 if !n.is_u64() && n.as_i64().is_none() && f.fract() == 0.0 && f >= i64::MIN as f64 {
1203 let i = f as i64;
1207 if (i as f64).to_bits() == f.to_bits() && f < 2_f64.powi(63) {
1208 push_integer(&mut out, i);
1209 }
1210 }
1211 }
1212 if out.is_empty() {
1213 None
1214 } else {
1215 Some(out)
1216 }
1217 }
1218 _ => None,
1219 }
1220}