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