Skip to main content

journal/
explorer.rs

1use super::*;
2use journal_core::file::{DataObject, offset_array::InlinedCursor};
3use std::collections::{HashMap, HashSet};
4use std::time::{Duration, Instant};
5
6const DEFAULT_HISTOGRAM_TARGET_BUCKETS: usize = 150;
7const DEFAULT_TIME_SLACK_USEC: u64 = 120_000_000;
8const EXPLORER_CONTROL_CHECK_EVERY_ROWS: u64 = 8192;
9const DEFAULT_ROWS_FULL_CHECK_EVERY_ROWS: u64 = 1;
10const EXPLORER_SAMPLING_SLOTS_MAX: usize = 1000;
11const EXPLORER_SAMPLING_RECALIBRATE_ROWS: u64 = 10_000;
12const EXPLORER_SAMPLING_ESTIMATE_AFTER_PROGRESS: f64 = 0.01;
13const SOURCE_REALTIME_FIELD: &[u8] = b"_SOURCE_REALTIME_TIMESTAMP";
14const UNSET_VALUE: &[u8] = b"-";
15const EXPLORER_UNSAMPLED_VALUE: &[u8] = b"[unsampled]";
16const EXPLORER_ESTIMATED_VALUE: &[u8] = b"[estimated]";
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub enum ExplorerAnchor {
20    Auto,
21    Head,
22    Tail,
23    Realtime(u64),
24}
25
26impl Default for ExplorerAnchor {
27    fn default() -> Self {
28        Self::Auto
29    }
30}
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub enum ExplorerFieldMode {
34    AllValues,
35    FirstValue,
36}
37
38impl Default for ExplorerFieldMode {
39    fn default() -> Self {
40        Self::FirstValue
41    }
42}
43
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45#[non_exhaustive]
46pub enum ExplorerStrategy {
47    Traversal,
48    Index,
49    Compare,
50}
51
52impl Default for ExplorerStrategy {
53    fn default() -> Self {
54        Self::Traversal
55    }
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct ExplorerFilter {
60    pub field: Vec<u8>,
61    pub values: Vec<Vec<u8>>,
62}
63
64impl ExplorerFilter {
65    pub fn new(
66        field: impl Into<Vec<u8>>,
67        values: impl IntoIterator<Item = impl Into<Vec<u8>>>,
68    ) -> Self {
69        Self {
70            field: field.into(),
71            values: values.into_iter().map(Into::into).collect(),
72        }
73    }
74}
75
76#[derive(Debug, Clone)]
77pub struct ExplorerQuery {
78    pub after_realtime_usec: Option<u64>,
79    pub before_realtime_usec: Option<u64>,
80    pub anchor: ExplorerAnchor,
81    pub direction: Direction,
82    pub limit: usize,
83    pub filters: Vec<ExplorerFilter>,
84    pub facets: Vec<Vec<u8>>,
85    pub histogram: Option<Vec<u8>>,
86    pub histogram_after_realtime_usec: Option<u64>,
87    pub histogram_before_realtime_usec: Option<u64>,
88    pub histogram_target_buckets: usize,
89    pub fts_terms: Vec<ExplorerFtsPattern>,
90    pub fts_patterns: Vec<Vec<u8>>,
91    pub fts_negative_patterns: Vec<Vec<u8>>,
92    pub field_mode: ExplorerFieldMode,
93    pub exclude_facet_field_filters: bool,
94    pub use_source_realtime: bool,
95    pub realtime_slack_usec: u64,
96    pub stop_when_rows_full: bool,
97    pub stop_when_rows_full_check_every: u64,
98    pub sampling: Option<ExplorerSampling>,
99    /// Debug-only discrepancy tool. Production explorer callers must never set
100    /// this; column catalogs belong to FIELD indexes, not row traversal.
101    #[doc(hidden)]
102    pub debug_collect_column_fields_by_row_traversal: bool,
103}
104
105impl Default for ExplorerQuery {
106    fn default() -> Self {
107        Self {
108            after_realtime_usec: None,
109            before_realtime_usec: None,
110            anchor: ExplorerAnchor::Auto,
111            direction: Direction::Forward,
112            limit: 200,
113            filters: Vec::new(),
114            facets: Vec::new(),
115            histogram: None,
116            histogram_after_realtime_usec: None,
117            histogram_before_realtime_usec: None,
118            histogram_target_buckets: DEFAULT_HISTOGRAM_TARGET_BUCKETS,
119            fts_terms: Vec::new(),
120            fts_patterns: Vec::new(),
121            fts_negative_patterns: Vec::new(),
122            field_mode: ExplorerFieldMode::FirstValue,
123            exclude_facet_field_filters: true,
124            use_source_realtime: true,
125            realtime_slack_usec: DEFAULT_TIME_SLACK_USEC,
126            stop_when_rows_full: false,
127            stop_when_rows_full_check_every: DEFAULT_ROWS_FULL_CHECK_EVERY_ROWS,
128            sampling: None,
129            debug_collect_column_fields_by_row_traversal: false,
130        }
131    }
132}
133
134#[derive(Debug, Clone, Copy, PartialEq, Eq)]
135pub struct ExplorerSampling {
136    pub budget: u64,
137    pub matched_files: u64,
138    pub file_head_realtime_usec: u64,
139    pub file_tail_realtime_usec: u64,
140    pub file_head_seqnum: u64,
141    pub file_tail_seqnum: u64,
142    pub file_entries: u64,
143}
144
145#[derive(Debug, Clone, PartialEq, Eq)]
146pub struct ExplorerFtsPattern {
147    pub parts: Vec<Vec<u8>>,
148    pub negative: bool,
149}
150
151impl ExplorerFtsPattern {
152    pub fn substring(pattern: impl Into<Vec<u8>>, negative: bool) -> Self {
153        let pattern = pattern.into();
154        let parts = pattern
155            .split(|byte| *byte == b'*')
156            .filter(|part| !part.is_empty())
157            .map(|part| part.to_vec())
158            .collect();
159        Self { parts, negative }
160    }
161
162    fn matches(&self, value: &[u8]) -> bool {
163        if value.is_empty() {
164            return false;
165        }
166        if self.parts.is_empty() {
167            return true;
168        }
169
170        let mut haystack = value;
171        for part in &self.parts {
172            let Some(index) = find_ascii_case_insensitive(haystack, part) else {
173                return false;
174            };
175            haystack = &haystack[index.saturating_add(part.len())..];
176        }
177        true
178    }
179}
180
181impl ExplorerQuery {
182    pub fn with_filter(
183        mut self,
184        field: impl Into<Vec<u8>>,
185        values: impl IntoIterator<Item = impl Into<Vec<u8>>>,
186    ) -> Self {
187        self.filters.push(ExplorerFilter::new(field, values));
188        self
189    }
190
191    pub fn with_facet(mut self, field: impl Into<Vec<u8>>) -> Self {
192        self.facets.push(field.into());
193        self
194    }
195
196    pub fn with_histogram(mut self, field: impl Into<Vec<u8>>) -> Self {
197        self.histogram = Some(field.into());
198        self
199    }
200
201    pub fn with_fts_pattern(mut self, pattern: impl Into<Vec<u8>>) -> Self {
202        let pattern = pattern.into();
203        self.fts_terms
204            .push(ExplorerFtsPattern::substring(pattern.clone(), false));
205        self.fts_patterns.push(pattern);
206        self
207    }
208
209    pub fn with_fts_negative_pattern(mut self, pattern: impl Into<Vec<u8>>) -> Self {
210        let pattern = pattern.into();
211        self.fts_terms
212            .push(ExplorerFtsPattern::substring(pattern.clone(), true));
213        self.fts_negative_patterns.push(pattern);
214        self
215    }
216}
217
218#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
219pub struct ExplorerStats {
220    pub rows_examined: u64,
221    pub rows_matched: u64,
222    pub facet_rows_matched: u64,
223    pub rows_returned: u64,
224    pub rows_unsampled: u64,
225    pub rows_estimated: u64,
226    pub sampling_sampled: u64,
227    pub sampling_unsampled: u64,
228    pub sampling_estimated: u64,
229    pub last_realtime_usec: u64,
230    pub max_source_realtime_delta_usec: u64,
231    pub data_refs_seen: u64,
232    pub data_refs_skipped: u64,
233    pub data_payloads_loaded: u64,
234    pub data_objects_classified: u64,
235    pub data_cache_hits: u64,
236    pub data_cache_misses: u64,
237    pub payloads_decompressed: u64,
238    pub fts_scans: u64,
239    pub facet_updates: u64,
240    pub histogram_updates: u64,
241    pub returned_row_expansions: u64,
242    pub early_stop_opportunities: u64,
243    pub early_stops: u64,
244}
245
246#[derive(Debug, Clone)]
247pub struct ExplorerRow {
248    pub realtime_usec: u64,
249    pub cursor: String,
250    pub payloads: Vec<Vec<u8>>,
251}
252
253#[derive(Debug, Clone, Copy, PartialEq, Eq)]
254pub(crate) enum ExplorerRowPayloadMode {
255    Expand,
256    CursorOnly,
257}
258
259#[derive(Debug, Clone)]
260pub struct ExplorerHistogramBucket {
261    pub start_realtime_usec: u64,
262    pub end_realtime_usec: u64,
263    pub values: HashMap<Vec<u8>, u64>,
264}
265
266#[derive(Debug, Clone)]
267pub struct ExplorerHistogram {
268    pub field: Vec<u8>,
269    pub buckets: Vec<ExplorerHistogramBucket>,
270}
271
272#[derive(Debug, Clone, Default)]
273pub struct ExplorerComparison {
274    pub traversal_duration: Duration,
275    pub index_duration: Duration,
276    pub traversal_stats: ExplorerStats,
277    pub index_stats: ExplorerStats,
278}
279
280#[derive(Debug, Clone, Default)]
281pub struct ExplorerResult {
282    pub rows: Vec<ExplorerRow>,
283    pub facets: HashMap<Vec<u8>, HashMap<Vec<u8>, u64>>,
284    pub histogram: Option<ExplorerHistogram>,
285    pub column_fields: HashSet<Vec<u8>>,
286    pub stats: ExplorerStats,
287    pub comparison: Option<ExplorerComparison>,
288}
289
290#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
291pub enum ExplorerStopReason {
292    TimedOut,
293    Cancelled,
294}
295
296#[derive(Debug, Clone)]
297pub struct ExplorerProgress {
298    pub stats: ExplorerStats,
299    pub elapsed: Duration,
300}
301
302pub struct ExplorerControl<'a> {
303    deadline: Option<Instant>,
304    cancellation: Option<&'a dyn Fn() -> bool>,
305    progress: Option<&'a mut dyn FnMut(ExplorerProgress)>,
306    candidate_row: Option<&'a mut dyn FnMut(u64) -> bool>,
307    matched_row: Option<&'a mut dyn FnMut(u64, u64) -> bool>,
308    sampling: Option<&'a mut ExplorerSamplingState>,
309    progress_interval: Duration,
310    started: Instant,
311    last_progress: Instant,
312    next_check_rows: u64,
313    stop_reason: Option<ExplorerStopReason>,
314}
315
316impl<'a> ExplorerControl<'a> {
317    pub fn new() -> Self {
318        let now = Instant::now();
319        Self {
320            deadline: None,
321            cancellation: None,
322            progress: None,
323            candidate_row: None,
324            matched_row: None,
325            sampling: None,
326            progress_interval: Duration::from_millis(250),
327            started: now,
328            last_progress: now,
329            next_check_rows: EXPLORER_CONTROL_CHECK_EVERY_ROWS,
330            stop_reason: None,
331        }
332    }
333
334    pub fn set_deadline(&mut self, deadline: Option<Instant>) {
335        self.deadline = deadline;
336    }
337
338    pub fn set_cancellation_callback(&mut self, cancellation: Option<&'a dyn Fn() -> bool>) {
339        self.cancellation = cancellation;
340    }
341
342    pub fn set_progress_callback(&mut self, progress: Option<&'a mut dyn FnMut(ExplorerProgress)>) {
343        self.progress = progress;
344    }
345
346    pub(crate) fn set_candidate_row_callback(
347        &mut self,
348        candidate_row: Option<&'a mut dyn FnMut(u64) -> bool>,
349    ) {
350        self.candidate_row = candidate_row;
351    }
352
353    pub fn set_matched_row_callback(
354        &mut self,
355        matched_row: Option<&'a mut dyn FnMut(u64, u64) -> bool>,
356    ) {
357        self.matched_row = matched_row;
358    }
359
360    pub(crate) fn set_sampling_state(&mut self, sampling: Option<&'a mut ExplorerSamplingState>) {
361        self.sampling = sampling;
362    }
363
364    pub fn set_progress_interval(&mut self, interval: Duration) {
365        self.progress_interval = interval;
366    }
367
368    pub fn stop_reason(&self) -> Option<ExplorerStopReason> {
369        self.stop_reason
370    }
371
372    fn should_stop_after_rows(&mut self, rows_seen: u64, stats: &ExplorerStats) -> bool {
373        if self.stop_reason.is_some() {
374            return true;
375        }
376        if rows_seen < self.next_check_rows {
377            return false;
378        }
379        self.next_check_rows = rows_seen.saturating_add(EXPLORER_CONTROL_CHECK_EVERY_ROWS);
380        self.check(stats)
381    }
382
383    fn check(&mut self, stats: &ExplorerStats) -> bool {
384        let now = Instant::now();
385        if now.duration_since(self.last_progress) >= self.progress_interval {
386            self.emit_progress(stats, now);
387        }
388        if self.cancellation.is_some_and(|is_cancelled| is_cancelled()) {
389            self.stop_reason = Some(ExplorerStopReason::Cancelled);
390            self.emit_progress(stats, now);
391            return true;
392        }
393        if self.deadline.is_some_and(|deadline| now >= deadline) {
394            self.stop_reason = Some(ExplorerStopReason::TimedOut);
395            self.emit_progress(stats, now);
396            return true;
397        }
398        false
399    }
400
401    fn emit_progress(&mut self, stats: &ExplorerStats, now: Instant) {
402        self.last_progress = now;
403        if let Some(progress) = self.progress.as_deref_mut() {
404            progress(ExplorerProgress {
405                stats: stats.clone(),
406                elapsed: now.duration_since(self.started),
407            });
408        }
409    }
410
411    fn emit_matched_row(&mut self, realtime_usec: u64, rows_matched: u64) -> bool {
412        if let Some(matched_row) = self.matched_row.as_deref_mut() {
413            return matched_row(realtime_usec, rows_matched);
414        }
415        false
416    }
417}
418
419impl Default for ExplorerControl<'_> {
420    fn default() -> Self {
421        Self::new()
422    }
423}
424
425#[derive(Debug, Clone, Copy, PartialEq, Eq)]
426enum ExplorerSamplingDecision {
427    Full {
428        sampled: bool,
429    },
430    SkipFields,
431    StopAndEstimate {
432        remaining_rows: u64,
433        from_realtime_usec: u64,
434        to_realtime_usec: u64,
435    },
436}
437
438#[derive(Debug)]
439pub(crate) struct ExplorerSamplingState {
440    start_realtime_usec: u64,
441    end_realtime_usec: u64,
442    file_head_realtime_usec: u64,
443    file_tail_realtime_usec: u64,
444    file_head_seqnum: u64,
445    file_tail_seqnum: u64,
446    file_entries: u64,
447    first_realtime_usec: Option<u64>,
448    step_realtime_usec: u64,
449    enable_after_samples: u64,
450    per_file_enable_after_samples: u64,
451    per_slot_enable_after_samples: u64,
452    sampled: u64,
453    per_file_sampled: u64,
454    per_file_unsampled: u64,
455    per_file_every: u64,
456    per_file_skipped: u64,
457    per_file_recalibrate: u64,
458    per_slot_sampled: Vec<u64>,
459    per_slot_unsampled: Vec<u64>,
460    matched_files: u64,
461    direction: Direction,
462}
463
464impl ExplorerSamplingState {
465    pub(crate) fn for_query(
466        query: &ExplorerQuery,
467        histogram_bucket_count: Option<usize>,
468    ) -> Option<Self> {
469        let sampling = query.sampling?;
470        let start_realtime_usec = query.after_realtime_usec?;
471        let end_realtime_usec = query.before_realtime_usec?;
472        if sampling.budget == 0
473            || sampling.matched_files == 0
474            || start_realtime_usec >= end_realtime_usec
475        {
476            return None;
477        }
478
479        let slots = histogram_bucket_count
480            .unwrap_or(query.histogram_target_buckets)
481            .clamp(2, EXPLORER_SAMPLING_SLOTS_MAX);
482        let delta = end_realtime_usec.saturating_sub(start_realtime_usec);
483        let step_realtime_usec = (delta / slots as u64).saturating_sub(1).max(1);
484        let per_file_enable_after_samples =
485            ((sampling.budget / 4) / sampling.matched_files.max(1)).max(query.limit as u64);
486        let per_slot_enable_after_samples =
487            ((sampling.budget / 4) / slots as u64).max(query.limit as u64);
488
489        Some(Self {
490            start_realtime_usec,
491            end_realtime_usec,
492            file_head_realtime_usec: sampling.file_head_realtime_usec,
493            file_tail_realtime_usec: sampling.file_tail_realtime_usec,
494            file_head_seqnum: sampling.file_head_seqnum,
495            file_tail_seqnum: sampling.file_tail_seqnum,
496            file_entries: sampling.file_entries,
497            first_realtime_usec: None,
498            step_realtime_usec,
499            enable_after_samples: sampling.budget / 2,
500            per_file_enable_after_samples,
501            per_slot_enable_after_samples,
502            sampled: 0,
503            per_file_sampled: 0,
504            per_file_unsampled: 0,
505            per_file_every: 0,
506            per_file_skipped: 0,
507            per_file_recalibrate: 0,
508            per_slot_sampled: vec![0; slots],
509            per_slot_unsampled: vec![0; slots],
510            matched_files: sampling.matched_files.max(1),
511            direction: query.direction,
512        })
513    }
514
515    fn begin_file(&mut self, sampling: ExplorerSampling) {
516        self.file_head_realtime_usec = sampling.file_head_realtime_usec;
517        self.file_tail_realtime_usec = sampling.file_tail_realtime_usec;
518        self.file_head_seqnum = sampling.file_head_seqnum;
519        self.file_tail_seqnum = sampling.file_tail_seqnum;
520        self.file_entries = sampling.file_entries;
521        self.first_realtime_usec = None;
522        self.per_file_sampled = 0;
523        self.per_file_unsampled = 0;
524        self.per_file_every = 0;
525        self.per_file_skipped = 0;
526        self.per_file_recalibrate = 0;
527    }
528
529    fn decide(
530        &mut self,
531        realtime_usec: u64,
532        seqnum: u64,
533        candidate_to_keep: bool,
534    ) -> ExplorerSamplingDecision {
535        if self.first_realtime_usec.is_none() {
536            self.first_realtime_usec = Some(realtime_usec);
537        }
538        if candidate_to_keep {
539            return ExplorerSamplingDecision::Full { sampled: false };
540        }
541
542        let slot = self.slot_for_realtime(realtime_usec);
543        let should_sample = if self.sampled < self.enable_after_samples
544            || self.per_file_sampled < self.per_file_enable_after_samples
545            || self.per_slot_sampled[slot] < self.per_slot_enable_after_samples
546        {
547            true
548        } else if self.per_file_recalibrate >= EXPLORER_SAMPLING_RECALIBRATE_ROWS
549            || self.per_file_every == 0
550        {
551            self.recalibrate(realtime_usec, seqnum);
552            true
553        } else if self.per_file_skipped >= self.per_file_every {
554            self.per_file_skipped = 0;
555            true
556        } else {
557            self.per_file_skipped = self.per_file_skipped.saturating_add(1);
558            false
559        };
560
561        if should_sample {
562            self.sampled = self.sampled.saturating_add(1);
563            self.per_file_sampled = self.per_file_sampled.saturating_add(1);
564            self.per_slot_sampled[slot] = self.per_slot_sampled[slot].saturating_add(1);
565            return ExplorerSamplingDecision::Full { sampled: true };
566        }
567
568        self.per_file_recalibrate = self.per_file_recalibrate.saturating_add(1);
569        self.per_file_unsampled = self.per_file_unsampled.saturating_add(1);
570        self.per_slot_unsampled[slot] = self.per_slot_unsampled[slot].saturating_add(1);
571
572        if self.per_file_unsampled > self.per_file_sampled
573            && self.progress_by_time(realtime_usec) > EXPLORER_SAMPLING_ESTIMATE_AFTER_PROGRESS
574        {
575            let remaining_rows = self.estimate_remaining_rows(realtime_usec, seqnum);
576            let (from_realtime_usec, to_realtime_usec) = self.remaining_range(realtime_usec);
577            return ExplorerSamplingDecision::StopAndEstimate {
578                remaining_rows,
579                from_realtime_usec,
580                to_realtime_usec,
581            };
582        }
583
584        ExplorerSamplingDecision::SkipFields
585    }
586
587    fn slot_for_realtime(&self, realtime_usec: u64) -> usize {
588        let clamped = realtime_usec.clamp(self.start_realtime_usec, self.end_realtime_usec);
589        let slot =
590            (clamped.saturating_sub(self.start_realtime_usec) / self.step_realtime_usec) as usize;
591        slot.min(self.per_slot_sampled.len().saturating_sub(1))
592    }
593
594    fn recalibrate(&mut self, realtime_usec: u64, seqnum: u64) {
595        let remaining_rows = self.estimate_remaining_rows(realtime_usec, seqnum);
596        let wanted_samples = (self.enable_after_samples / self.matched_files).max(1);
597        self.per_file_every = (remaining_rows / wanted_samples).max(1);
598        self.per_file_recalibrate = 0;
599    }
600
601    fn estimate_remaining_rows(&self, realtime_usec: u64, seqnum: u64) -> u64 {
602        if let Some(remaining) = self.estimate_remaining_rows_by_seqnum(seqnum) {
603            return remaining;
604        }
605        self.estimate_remaining_rows_by_time(realtime_usec)
606    }
607
608    fn estimate_remaining_rows_by_seqnum(&self, seqnum: u64) -> Option<u64> {
609        self.validate_seqnum_estimate_inputs(seqnum)?;
610        let scanned_rows = self.scanned_file_rows();
611        let seqnum_span = self.seqnum_span_so_far(seqnum)?;
612        if seqnum_span == 0 {
613            return None;
614        }
615        let proportion_of_all_lines_so_far =
616            bounded_positive_proportion(scanned_rows as f64 / seqnum_span as f64)?;
617        let expected_matching_logs =
618            (proportion_of_all_lines_so_far * self.file_entries as f64) as u64;
619        if expected_matching_logs == 0 {
620            return None;
621        }
622        // Match Netdata systemd-journal.plugin sampling_running_file_query_estimate_remaining_lines():
623        // remaining_logs_by_seqnum = expected_matching_logs_by_seqnum - scanned_lines,
624        // clamped to one when the subtraction reaches zero.
625        Some(expected_matching_logs.saturating_sub(scanned_rows).max(1))
626    }
627
628    fn validate_seqnum_estimate_inputs(&self, seqnum: u64) -> Option<()> {
629        (self.file_entries != 0
630            && self.file_head_seqnum != 0
631            && self.file_tail_seqnum != 0
632            && seqnum != 0)
633            .then_some(())
634    }
635
636    fn scanned_file_rows(&self) -> u64 {
637        self.per_file_sampled
638            .saturating_add(self.per_file_unsampled)
639            .max(1)
640    }
641
642    fn seqnum_span_so_far(&self, seqnum: u64) -> Option<u64> {
643        match self.direction {
644            Direction::Forward => seqnum.checked_sub(self.file_head_seqnum),
645            Direction::Backward => self.file_tail_seqnum.checked_sub(seqnum),
646        }
647    }
648
649    fn estimate_remaining_rows_by_time(&self, realtime_usec: u64) -> u64 {
650        let scanned_rows = self.scanned_file_rows();
651        let (after, before) = self.overlapping_timeframe(realtime_usec);
652        let total_time = self
653            .remaining_time_bounds(realtime_usec, after, before)
654            .0
655            .max(1);
656        let remaining_time = self.remaining_time_bounds(realtime_usec, after, before).1;
657        let elapsed = total_time.saturating_sub(remaining_time).max(1);
658        let mut proportion_by_time = elapsed as f64 / total_time as f64;
659        if proportion_by_time == 0.0 || proportion_by_time > 1.0 || !proportion_by_time.is_finite()
660        {
661            proportion_by_time = 1.0;
662        }
663        let mut expected_total = (scanned_rows as f64 / proportion_by_time) as u64;
664        if self.file_entries != 0 && expected_total > self.file_entries {
665            expected_total = self.file_entries;
666        }
667        expected_total.saturating_sub(scanned_rows).max(1)
668    }
669
670    fn progress_by_time(&self, realtime_usec: u64) -> f64 {
671        let (after, before) = self.overlapping_timeframe(realtime_usec);
672        let total_time = before.saturating_sub(after).max(1);
673        let elapsed = match self.direction {
674            Direction::Forward => realtime_usec.saturating_sub(after),
675            Direction::Backward => before.saturating_sub(realtime_usec),
676        }
677        .min(total_time);
678        elapsed as f64 / total_time as f64
679    }
680
681    fn overlapping_timeframe(&self, realtime_usec: u64) -> (u64, u64) {
682        match self.direction {
683            Direction::Forward => {
684                let mut oldest = self
685                    .first_realtime_usec
686                    .or((self.file_head_realtime_usec != 0).then_some(self.file_head_realtime_usec))
687                    .unwrap_or(self.start_realtime_usec);
688                let mut newest = if self.file_tail_realtime_usec != 0 {
689                    self.end_realtime_usec.min(self.file_tail_realtime_usec)
690                } else {
691                    self.end_realtime_usec
692                };
693                if newest <= oldest {
694                    newest = oldest.saturating_add(1);
695                }
696                if realtime_usec < oldest {
697                    oldest = realtime_usec.saturating_sub(1);
698                }
699                (oldest, newest)
700            }
701            Direction::Backward => {
702                let mut newest = self
703                    .first_realtime_usec
704                    .or((self.file_tail_realtime_usec != 0).then_some(self.file_tail_realtime_usec))
705                    .unwrap_or(self.end_realtime_usec);
706                let oldest = if self.file_head_realtime_usec != 0 {
707                    self.start_realtime_usec.max(self.file_head_realtime_usec)
708                } else {
709                    self.start_realtime_usec
710                };
711                if newest <= oldest {
712                    newest = oldest.saturating_add(1);
713                }
714                if newest < realtime_usec {
715                    newest = realtime_usec.saturating_add(1);
716                }
717                (oldest, newest)
718            }
719        }
720    }
721
722    fn remaining_range(&self, realtime_usec: u64) -> (u64, u64) {
723        let (after, before) = self.overlapping_timeframe(realtime_usec);
724        let (_, _, remaining_start, remaining_end) =
725            self.remaining_time_details(realtime_usec, after, before);
726        (remaining_start, remaining_end)
727    }
728
729    fn remaining_time_bounds(&self, realtime_usec: u64, after: u64, before: u64) -> (u64, u64) {
730        let (total, remaining, _, _) = self.remaining_time_details(realtime_usec, after, before);
731        (total, remaining)
732    }
733
734    fn remaining_time_details(
735        &self,
736        realtime_usec: u64,
737        mut after: u64,
738        mut before: u64,
739    ) -> (u64, u64, u64, u64) {
740        if realtime_usec <= after {
741            after = realtime_usec.saturating_sub(1);
742        }
743        if realtime_usec >= before {
744            before = realtime_usec.saturating_add(1);
745        }
746        if before <= after {
747            before = after.saturating_add(1);
748        }
749        let (remaining_start, remaining_end) = match self.direction {
750            Direction::Forward => (realtime_usec, before),
751            Direction::Backward => (after, realtime_usec),
752        };
753        (
754            before.saturating_sub(after).max(1),
755            remaining_end.saturating_sub(remaining_start),
756            remaining_start,
757            remaining_end,
758        )
759    }
760}
761
762pub(crate) fn histogram_bucket_count_for_query(query: &ExplorerQuery) -> Option<usize> {
763    query
764        .histogram
765        .as_deref()
766        .map(|field| new_histogram(field, query).buckets.len())
767}
768
769#[derive(Default)]
770struct RowScan {
771    timestamp: Option<u64>,
772    fts_matches: bool,
773    fts_negative_match: bool,
774    column_fields: Vec<Vec<u8>>,
775}
776
777const FACET_PUBLIC: u8 = 0x01;
778const FACET_HISTOGRAM: u8 = 0x02;
779const FACET_SOURCE_REALTIME: u8 = 0x04;
780
781#[derive(Clone, Copy, Debug, PartialEq, Eq)]
782enum OffsetClass {
783    Irrelevant,
784    FtsMatch,
785    FtsNegativeMatch,
786    Value(usize),
787}
788
789impl OffsetClass {
790    const IRRELEVANT_RAW: usize = 1;
791    const FTS_MATCH_RAW: usize = 2;
792    const FTS_NEGATIVE_MATCH_RAW: usize = 3;
793    const VALUE_BASE: usize = 4;
794
795    fn to_raw(self) -> usize {
796        match self {
797            Self::Irrelevant => Self::IRRELEVANT_RAW,
798            Self::FtsMatch => Self::FTS_MATCH_RAW,
799            Self::FtsNegativeMatch => Self::FTS_NEGATIVE_MATCH_RAW,
800            Self::Value(index) => Self::VALUE_BASE.saturating_add(index),
801        }
802    }
803
804    fn from_raw(raw: usize) -> Self {
805        match raw {
806            Self::IRRELEVANT_RAW => Self::Irrelevant,
807            Self::FTS_MATCH_RAW => Self::FtsMatch,
808            Self::FTS_NEGATIVE_MATCH_RAW => Self::FtsNegativeMatch,
809            raw => Self::Value(raw.saturating_sub(Self::VALUE_BASE)),
810        }
811    }
812}
813
814#[derive(Clone, Copy, Debug, Default)]
815struct OffsetClassSlot {
816    offset: u64,
817    class: usize,
818}
819
820#[derive(Debug)]
821struct OffsetClassCache {
822    slots: Vec<OffsetClassSlot>,
823    len: usize,
824}
825
826impl Default for OffsetClassCache {
827    fn default() -> Self {
828        Self {
829            slots: vec![OffsetClassSlot::default(); 256],
830            len: 0,
831        }
832    }
833}
834
835impl OffsetClassCache {
836    fn lookup(&self, offset: NonZeroU64) -> Option<OffsetClass> {
837        let mask = self.slots.len().saturating_sub(1);
838        let mut index = offset_slot(offset.get()) & mask;
839        for _ in 0..self.slots.len() {
840            let slot = self.slots[index];
841            if slot.offset == 0 {
842                return None;
843            }
844            if slot.offset == offset.get() {
845                return Some(OffsetClass::from_raw(slot.class));
846            }
847            index = (index + 1) & mask;
848        }
849        None
850    }
851
852    fn insert(&mut self, offset: NonZeroU64, class: OffsetClass) {
853        if (self.len + 1).saturating_mul(4) >= self.slots.len().saturating_mul(3) {
854            self.grow();
855        }
856        self.insert_raw(offset.get(), class.to_raw());
857    }
858
859    fn grow(&mut self) {
860        let new_len = self.slots.len().saturating_mul(2).max(256);
861        let old = std::mem::replace(&mut self.slots, vec![OffsetClassSlot::default(); new_len]);
862        self.len = 0;
863        for slot in old {
864            if slot.offset != 0 {
865                self.insert_raw(slot.offset, slot.class);
866            }
867        }
868    }
869
870    fn insert_raw(&mut self, offset: u64, class: usize) {
871        let mask = self.slots.len().saturating_sub(1);
872        let mut index = offset_slot(offset) & mask;
873        loop {
874            if self.slots[index].offset == 0 {
875                self.slots[index] = OffsetClassSlot { offset, class };
876                self.len += 1;
877                return;
878            }
879            if self.slots[index].offset == offset {
880                self.slots[index].class = class;
881                return;
882            }
883            index = (index + 1) & mask;
884        }
885    }
886}
887
888fn offset_slot(offset: u64) -> usize {
889    let mut value = offset >> 3;
890    value ^= value >> 33;
891    value = value.wrapping_mul(0xff51afd7ed558ccd);
892    value ^= value >> 33;
893    value as usize
894}
895
896struct ExplorerAccumulator {
897    field_lookup: HashMap<Vec<u8>, usize>,
898    fields: Vec<Vec<u8>>,
899    flags: Vec<u8>,
900    last_seen_row_ids: Vec<u64>,
901    unset_counts: Vec<u64>,
902    values_by_field: Vec<Vec<usize>>,
903    value_counts: Vec<u64>,
904    value_field_indices: Vec<usize>,
905    value_labels: Vec<Vec<u8>>,
906    value_fts_matches: Vec<bool>,
907    value_source_realtime: Vec<Option<u64>>,
908    value_histogram_buckets: Vec<Option<Vec<u64>>>,
909    field_histogram_unset_buckets: Vec<Option<Vec<u64>>>,
910    offset_cache: OffsetClassCache,
911    histogram_start_realtime_usec: u64,
912    histogram_bucket_width_usec: u64,
913    histogram_bucket_count: usize,
914    required_identity_count: usize,
915}
916
917impl ExplorerAccumulator {
918    fn for_main(query: &ExplorerQuery, histogram: Option<&ExplorerHistogram>) -> Self {
919        Self::for_combined(query, &[], histogram)
920    }
921
922    fn for_facets(
923        query: &ExplorerQuery,
924        facet_indices: &[usize],
925        include_source_realtime: bool,
926    ) -> Self {
927        let mut out = Self::new(None);
928        for facet_index in facet_indices {
929            if let Some(field) = query.facets.get(*facet_index) {
930                out.add_field(field, FACET_PUBLIC);
931            }
932        }
933        if include_source_realtime {
934            out.add_field(SOURCE_REALTIME_FIELD, FACET_SOURCE_REALTIME);
935        }
936        out
937    }
938
939    fn for_combined(
940        query: &ExplorerQuery,
941        facet_indices: &[usize],
942        histogram: Option<&ExplorerHistogram>,
943    ) -> Self {
944        let mut out = Self::new(histogram);
945        if let Some(field) = &query.histogram {
946            out.add_field(field, FACET_HISTOGRAM);
947        }
948        for facet_index in facet_indices {
949            if let Some(field) = query.facets.get(*facet_index) {
950                out.add_field(field, FACET_PUBLIC);
951            }
952        }
953        if query_needs_source_realtime_main(query) || facet_pass_needs_source_realtime(query) {
954            out.add_field(SOURCE_REALTIME_FIELD, FACET_SOURCE_REALTIME);
955        }
956        out
957    }
958
959    fn new(histogram: Option<&ExplorerHistogram>) -> Self {
960        Self {
961            field_lookup: HashMap::new(),
962            fields: Vec::new(),
963            flags: Vec::new(),
964            last_seen_row_ids: Vec::new(),
965            unset_counts: Vec::new(),
966            values_by_field: Vec::new(),
967            value_counts: Vec::new(),
968            value_field_indices: Vec::new(),
969            value_labels: Vec::new(),
970            value_fts_matches: Vec::new(),
971            value_source_realtime: Vec::new(),
972            value_histogram_buckets: Vec::new(),
973            field_histogram_unset_buckets: Vec::new(),
974            offset_cache: OffsetClassCache::default(),
975            histogram_start_realtime_usec: histogram
976                .and_then(|histogram| histogram.buckets.first())
977                .map(|bucket| bucket.start_realtime_usec)
978                .unwrap_or_default(),
979            histogram_bucket_width_usec: histogram
980                .and_then(|histogram| histogram.buckets.first())
981                .map(|bucket| {
982                    bucket
983                        .end_realtime_usec
984                        .saturating_sub(bucket.start_realtime_usec)
985                        .max(1)
986                })
987                .unwrap_or(1),
988            histogram_bucket_count: histogram
989                .map(|histogram| histogram.buckets.len())
990                .unwrap_or_default(),
991            required_identity_count: 0,
992        }
993    }
994
995    fn add_field(&mut self, field: &[u8], flags: u8) {
996        if let Some(index) = self.field_lookup.get(field).copied() {
997            let had_required = self.flags[index] != 0;
998            self.flags[index] |= flags;
999            if flags & FACET_HISTOGRAM != 0 && self.field_histogram_unset_buckets[index].is_none() {
1000                self.field_histogram_unset_buckets[index] =
1001                    Some(vec![0; self.histogram_bucket_count]);
1002            }
1003            if !had_required && self.flags[index] != 0 {
1004                self.required_identity_count += 1;
1005            }
1006            return;
1007        }
1008
1009        let index = self.fields.len();
1010        self.field_lookup.insert(field.to_vec(), index);
1011        self.fields.push(field.to_vec());
1012        self.flags.push(flags);
1013        self.last_seen_row_ids.push(0);
1014        self.unset_counts.push(0);
1015        self.values_by_field.push(Vec::new());
1016        self.field_histogram_unset_buckets
1017            .push((flags & FACET_HISTOGRAM != 0).then(|| vec![0; self.histogram_bucket_count]));
1018        if flags != 0 {
1019            self.required_identity_count += 1;
1020        }
1021    }
1022
1023    fn add_value(
1024        &mut self,
1025        field_index: usize,
1026        _data_offset: NonZeroU64,
1027        value: &[u8],
1028        fts_matches: bool,
1029    ) -> usize {
1030        let value_index = self.value_counts.len();
1031        let flags = self.flags[field_index];
1032        self.value_counts.push(0);
1033        self.value_field_indices.push(field_index);
1034        self.value_labels.push(value.to_vec());
1035        self.value_fts_matches.push(fts_matches);
1036        self.value_source_realtime
1037            .push(if flags & FACET_SOURCE_REALTIME != 0 {
1038                parse_source_realtime(value)
1039            } else {
1040                None
1041            });
1042        self.value_histogram_buckets
1043            .push((flags & FACET_HISTOGRAM != 0).then(|| vec![0; self.histogram_bucket_count]));
1044        self.values_by_field[field_index].push(value_index);
1045        value_index
1046    }
1047
1048    fn mark_field_seen(&mut self, field_index: usize, row_id: u64) -> bool {
1049        // Duplicate values for one field must not satisfy another required
1050        // field identity in first-value mode.
1051        if self.last_seen_row_ids[field_index] == row_id {
1052            return false;
1053        }
1054        self.last_seen_row_ids[field_index] = row_id;
1055        true
1056    }
1057
1058    fn apply_value(
1059        &mut self,
1060        value_index: usize,
1061        realtime_usec: Option<u64>,
1062        stats: &mut ExplorerStats,
1063    ) {
1064        let field_index = self.value_field_indices[value_index];
1065        let flags = self.flags[field_index];
1066        if flags & FACET_PUBLIC != 0 {
1067            self.value_counts[value_index] = self.value_counts[value_index].saturating_add(1);
1068            stats.facet_updates = stats.facet_updates.saturating_add(1);
1069        }
1070        if flags & FACET_HISTOGRAM != 0 {
1071            if let (Some(realtime_usec), Some(buckets)) = (
1072                realtime_usec,
1073                self.value_histogram_buckets[value_index].as_mut(),
1074            ) {
1075                if let Some(bucket_index) = histogram_bucket_index_from_bounds(
1076                    realtime_usec,
1077                    self.histogram_start_realtime_usec,
1078                    self.histogram_bucket_width_usec,
1079                    buckets.len(),
1080                ) {
1081                    buckets[bucket_index] = buckets[bucket_index].saturating_add(1);
1082                    stats.histogram_updates = stats.histogram_updates.saturating_add(1);
1083                }
1084            }
1085        }
1086    }
1087
1088    fn finish_facet_row(&mut self, row_id: u64, stats: &mut ExplorerStats) {
1089        for field_index in 0..self.fields.len() {
1090            if self.flags[field_index] & FACET_PUBLIC == 0 {
1091                continue;
1092            }
1093            if self.last_seen_row_ids[field_index] != row_id {
1094                self.unset_counts[field_index] = self.unset_counts[field_index].saturating_add(1);
1095                stats.facet_updates = stats.facet_updates.saturating_add(1);
1096            }
1097        }
1098    }
1099
1100    fn finish_histogram_row(&mut self, row_id: u64, realtime_usec: u64, stats: &mut ExplorerStats) {
1101        for field_index in 0..self.fields.len() {
1102            if self.flags[field_index] & FACET_HISTOGRAM == 0 {
1103                continue;
1104            }
1105            if self.last_seen_row_ids[field_index] == row_id {
1106                continue;
1107            }
1108            let Some(buckets) = self.field_histogram_unset_buckets[field_index].as_mut() else {
1109                continue;
1110            };
1111            if let Some(bucket_index) = histogram_bucket_index_from_bounds(
1112                realtime_usec,
1113                self.histogram_start_realtime_usec,
1114                self.histogram_bucket_width_usec,
1115                buckets.len(),
1116            ) {
1117                buckets[bucket_index] = buckets[bucket_index].saturating_add(1);
1118                stats.histogram_updates = stats.histogram_updates.saturating_add(1);
1119            }
1120        }
1121    }
1122
1123    fn finish_facets(&self, result: &mut ExplorerResult) {
1124        for field_index in 0..self.fields.len() {
1125            if self.flags[field_index] & FACET_PUBLIC == 0 {
1126                continue;
1127            }
1128            let mut values = HashMap::new();
1129            for value_index in &self.values_by_field[field_index] {
1130                let count = self.value_counts[*value_index];
1131                if count != 0 {
1132                    increment_counter_by(&mut values, &self.value_labels[*value_index], count);
1133                }
1134            }
1135            if self.unset_counts[field_index] != 0 {
1136                increment_counter_by(&mut values, UNSET_VALUE, self.unset_counts[field_index]);
1137            }
1138            result
1139                .facets
1140                .insert(self.fields[field_index].clone(), values);
1141        }
1142    }
1143
1144    fn finish_histogram(&self, histogram: Option<&mut ExplorerHistogram>) {
1145        let Some(histogram) = histogram else {
1146            return;
1147        };
1148        for buckets in self.field_histogram_unset_buckets.iter().flatten() {
1149            for (bucket_index, count) in buckets.iter().enumerate() {
1150                if *count == 0 {
1151                    continue;
1152                }
1153                if let Some(bucket) = histogram.buckets.get_mut(bucket_index) {
1154                    increment_counter_by(&mut bucket.values, UNSET_VALUE, *count);
1155                }
1156            }
1157        }
1158        for value_index in 0..self.value_histogram_buckets.len() {
1159            let Some(buckets) = &self.value_histogram_buckets[value_index] else {
1160                continue;
1161            };
1162            for (bucket_index, count) in buckets.iter().enumerate() {
1163                if *count == 0 {
1164                    continue;
1165                }
1166                if let Some(bucket) = histogram.buckets.get_mut(bucket_index) {
1167                    increment_counter_by(
1168                        &mut bucket.values,
1169                        &self.value_labels[value_index],
1170                        *count,
1171                    );
1172                }
1173            }
1174        }
1175    }
1176}
1177
1178fn bounded_positive_proportion(value: f64) -> Option<f64> {
1179    if value <= 0.0 || !value.is_finite() {
1180        return None;
1181    }
1182    Some(value.min(1.0))
1183}
1184
1185impl FileReader {
1186    pub fn explore(&mut self, query: &ExplorerQuery) -> Result<ExplorerResult> {
1187        self.explore_with_strategy(query, ExplorerStrategy::Traversal)
1188    }
1189
1190    pub fn explore_with_strategy(
1191        &mut self,
1192        query: &ExplorerQuery,
1193        strategy: ExplorerStrategy,
1194    ) -> Result<ExplorerResult> {
1195        self.explore_with_strategy_and_payload_mode(query, strategy, ExplorerRowPayloadMode::Expand)
1196    }
1197
1198    pub fn explore_with_strategy_and_control(
1199        &mut self,
1200        query: &ExplorerQuery,
1201        strategy: ExplorerStrategy,
1202        control: &mut ExplorerControl<'_>,
1203    ) -> Result<ExplorerResult> {
1204        validate_no_debug_column_collection(query)?;
1205        self.explore_with_strategy_and_payload_mode_unchecked(
1206            query,
1207            strategy,
1208            ExplorerRowPayloadMode::Expand,
1209            Some(control),
1210        )
1211    }
1212
1213    #[cfg(test)]
1214    pub(crate) fn explore_with_strategy_cursor_rows(
1215        &mut self,
1216        query: &ExplorerQuery,
1217        strategy: ExplorerStrategy,
1218    ) -> Result<ExplorerResult> {
1219        self.explore_with_strategy_and_payload_mode(
1220            query,
1221            strategy,
1222            ExplorerRowPayloadMode::CursorOnly,
1223        )
1224    }
1225
1226    pub(crate) fn explore_with_strategy_cursor_rows_controlled(
1227        &mut self,
1228        query: &ExplorerQuery,
1229        strategy: ExplorerStrategy,
1230        control: &mut ExplorerControl<'_>,
1231    ) -> Result<ExplorerResult> {
1232        validate_no_debug_column_collection(query)?;
1233        self.explore_with_strategy_and_payload_mode_unchecked(
1234            query,
1235            strategy,
1236            ExplorerRowPayloadMode::CursorOnly,
1237            Some(control),
1238        )
1239    }
1240
1241    fn explore_with_strategy_and_payload_mode(
1242        &mut self,
1243        query: &ExplorerQuery,
1244        strategy: ExplorerStrategy,
1245        row_payload_mode: ExplorerRowPayloadMode,
1246    ) -> Result<ExplorerResult> {
1247        validate_no_debug_column_collection(query)?;
1248        self.explore_with_strategy_and_payload_mode_unchecked(
1249            query,
1250            strategy,
1251            row_payload_mode,
1252            None,
1253        )
1254    }
1255
1256    fn explore_with_strategy_and_payload_mode_unchecked(
1257        &mut self,
1258        query: &ExplorerQuery,
1259        strategy: ExplorerStrategy,
1260        row_payload_mode: ExplorerRowPayloadMode,
1261        mut control: Option<&mut ExplorerControl<'_>>,
1262    ) -> Result<ExplorerResult> {
1263        match strategy {
1264            ExplorerStrategy::Traversal => {
1265                self.explore_traversal(query, row_payload_mode, control.as_deref_mut())
1266            }
1267            ExplorerStrategy::Index => {
1268                self.explore_indexed(query, row_payload_mode, control.as_deref_mut())
1269            }
1270            ExplorerStrategy::Compare => self.explore_compare(query, row_payload_mode),
1271        }
1272    }
1273
1274    fn explore_traversal(
1275        &mut self,
1276        query: &ExplorerQuery,
1277        row_payload_mode: ExplorerRowPayloadMode,
1278        mut control: Option<&mut ExplorerControl<'_>>,
1279    ) -> Result<ExplorerResult> {
1280        validate_query(query)?;
1281        let mut result = explorer_result_for_query(query);
1282        let facet_groups = facet_pass_groups(query);
1283        if can_run_combined_explorer_pass(&facet_groups) {
1284            self.explore_traversal_combined(
1285                query,
1286                row_payload_mode,
1287                &mut result,
1288                &facet_groups,
1289                control.as_deref_mut(),
1290            )?;
1291        } else {
1292            self.explore_traversal_split(
1293                query,
1294                row_payload_mode,
1295                &mut result,
1296                facet_groups,
1297                control.as_deref_mut(),
1298            )?;
1299        }
1300        self.configure_explorer_filters(query, None)?;
1301        Ok(result)
1302    }
1303
1304    fn explore_traversal_combined(
1305        &mut self,
1306        query: &ExplorerQuery,
1307        row_payload_mode: ExplorerRowPayloadMode,
1308        result: &mut ExplorerResult,
1309        facet_groups: &[FacetPassGroup],
1310        control: Option<&mut ExplorerControl<'_>>,
1311    ) -> Result<()> {
1312        let facet_indices = combined_facet_indices(facet_groups);
1313        if query_needs_main_pass(query) || !facet_indices.is_empty() {
1314            self.configure_explorer_filters(query, None)?;
1315            let mut accumulator =
1316                ExplorerAccumulator::for_combined(query, &facet_indices, result.histogram.as_ref());
1317            self.scan_explorer_combined(
1318                query,
1319                &mut accumulator,
1320                result,
1321                !facet_indices.is_empty(),
1322                row_payload_mode,
1323                control,
1324            )?;
1325            accumulator.finish_facets(result);
1326            accumulator.finish_histogram(result.histogram.as_mut());
1327        }
1328        Ok(())
1329    }
1330
1331    fn explore_traversal_split(
1332        &mut self,
1333        query: &ExplorerQuery,
1334        row_payload_mode: ExplorerRowPayloadMode,
1335        result: &mut ExplorerResult,
1336        facet_groups: Vec<FacetPassGroup>,
1337        mut control: Option<&mut ExplorerControl<'_>>,
1338    ) -> Result<()> {
1339        if query_needs_main_pass(query) {
1340            self.configure_explorer_filters(query, None)?;
1341            let mut accumulator = ExplorerAccumulator::for_main(query, result.histogram.as_ref());
1342            self.scan_explorer_main(
1343                query,
1344                &mut accumulator,
1345                result,
1346                row_payload_mode,
1347                control.as_deref_mut(),
1348            )?;
1349            accumulator.finish_histogram(result.histogram.as_mut());
1350        }
1351
1352        for group in facet_groups {
1353            if explorer_control_stopped(control.as_deref()) {
1354                break;
1355            }
1356            self.configure_explorer_filters(query, group.excluded_field.as_deref())?;
1357            let mut accumulator = ExplorerAccumulator::for_facets(
1358                query,
1359                &group.facet_indices,
1360                facet_pass_needs_source_realtime(query),
1361            );
1362            self.scan_explorer_facet(
1363                query,
1364                &mut accumulator,
1365                &mut result.stats,
1366                control.as_deref_mut(),
1367            )?;
1368            accumulator.finish_facets(result);
1369        }
1370        Ok(())
1371    }
1372
1373    fn explore_compare(
1374        &mut self,
1375        query: &ExplorerQuery,
1376        row_payload_mode: ExplorerRowPayloadMode,
1377    ) -> Result<ExplorerResult> {
1378        let traversal_started = Instant::now();
1379        let traversal = self.explore_traversal(query, row_payload_mode, None)?;
1380        let traversal_duration = traversal_started.elapsed();
1381
1382        let index_started = Instant::now();
1383        let mut indexed = self.explore_indexed(query, row_payload_mode, None)?;
1384        let index_duration = index_started.elapsed();
1385
1386        if !explorer_outputs_match(&traversal, &indexed) {
1387            return Err(SdkError::VerificationError(
1388                "indexed explorer output differs from traversal explorer output".to_string(),
1389            ));
1390        }
1391        indexed.comparison = Some(ExplorerComparison {
1392            traversal_duration,
1393            index_duration,
1394            traversal_stats: traversal.stats,
1395            index_stats: indexed.stats.clone(),
1396        });
1397        Ok(indexed)
1398    }
1399
1400    fn explore_indexed(
1401        &mut self,
1402        query: &ExplorerQuery,
1403        row_payload_mode: ExplorerRowPayloadMode,
1404        mut control: Option<&mut ExplorerControl<'_>>,
1405    ) -> Result<ExplorerResult> {
1406        validate_query(query)?;
1407        validate_indexed_query(query)?;
1408        let mut result = explorer_result_for_query(query);
1409        self.indexed_collect_rows(query, row_payload_mode, &mut result, control.as_deref_mut())?;
1410        self.indexed_collect_facets(query, &mut result, control.as_deref())?;
1411        self.indexed_collect_histogram(query, &mut result, control.as_deref())?;
1412        self.configure_explorer_filters(query, None)?;
1413        Ok(result)
1414    }
1415
1416    fn indexed_collect_rows(
1417        &mut self,
1418        query: &ExplorerQuery,
1419        row_payload_mode: ExplorerRowPayloadMode,
1420        result: &mut ExplorerResult,
1421        control: Option<&mut ExplorerControl<'_>>,
1422    ) -> Result<()> {
1423        if query.limit == 0 {
1424            return Ok(());
1425        }
1426        let mut row_query = query.clone();
1427        row_query.facets.clear();
1428        row_query.histogram = None;
1429        self.configure_explorer_filters(&row_query, None)?;
1430        let mut accumulator = ExplorerAccumulator::for_main(&row_query, None);
1431        self.scan_explorer_main(
1432            &row_query,
1433            &mut accumulator,
1434            result,
1435            row_payload_mode,
1436            control,
1437        )
1438    }
1439
1440    fn indexed_collect_facets(
1441        &mut self,
1442        query: &ExplorerQuery,
1443        result: &mut ExplorerResult,
1444        control: Option<&ExplorerControl<'_>>,
1445    ) -> Result<()> {
1446        if explorer_control_stopped(control) {
1447            return Ok(());
1448        }
1449        for group in facet_pass_groups(query) {
1450            let candidates = self.indexed_candidate_set(query, group.excluded_field.as_deref())?;
1451            self.inner.with_file(|file| {
1452                indexed_count_facet_group(file, query, &group, &candidates, result)
1453            })?;
1454        }
1455        Ok(())
1456    }
1457
1458    fn indexed_collect_histogram(
1459        &mut self,
1460        query: &ExplorerQuery,
1461        result: &mut ExplorerResult,
1462        control: Option<&ExplorerControl<'_>>,
1463    ) -> Result<()> {
1464        if query.histogram.is_none() || explorer_control_stopped(control) {
1465            return Ok(());
1466        }
1467        let candidates = self.indexed_candidate_set(query, None)?;
1468        self.inner
1469            .with_file(|file| indexed_count_histogram(file, query, &candidates, result))
1470    }
1471
1472    fn indexed_candidate_set(
1473        &mut self,
1474        query: &ExplorerQuery,
1475        excluded_field: Option<&[u8]>,
1476    ) -> Result<IndexedCandidateSet> {
1477        if query.filters.is_empty()
1478            && query.after_realtime_usec.is_none()
1479            && query.before_realtime_usec.is_none()
1480        {
1481            let count = self
1482                .inner
1483                .with_file(|file| file.journal_header_ref().n_entries);
1484            return Ok(IndexedCandidateSet::All { count });
1485        }
1486
1487        self.configure_explorer_filters(query, excluded_field)?;
1488        self.seek_for_explorer(query);
1489        let mut offsets = HashSet::new();
1490        while self.step_for_explorer(query.direction)? {
1491            let Some(metadata) = self.row.metadata() else {
1492                continue;
1493            };
1494            let commit_realtime = metadata.realtime;
1495            if stop_by_commit_time(query, commit_realtime) {
1496                break;
1497            }
1498            if !timestamp_in_range(query, commit_realtime) {
1499                continue;
1500            }
1501            if let Some(entry_offset) = self.row.entry_offset() {
1502                offsets.insert(entry_offset);
1503            }
1504        }
1505        Ok(IndexedCandidateSet::Set {
1506            count: offsets.len() as u64,
1507            offsets,
1508        })
1509    }
1510
1511    fn configure_explorer_filters(
1512        &mut self,
1513        query: &ExplorerQuery,
1514        excluded_field: Option<&[u8]>,
1515    ) -> Result<()> {
1516        self.flush_matches();
1517        for filter in &query.filters {
1518            if excluded_field.is_some_and(|field| field == filter.field.as_slice()) {
1519                continue;
1520            }
1521            if filter.values.is_empty() {
1522                continue;
1523            }
1524            for value in &filter.values {
1525                let payload = payload_from_parts(&filter.field, value);
1526                self.add_match(&payload);
1527            }
1528        }
1529        Ok(())
1530    }
1531
1532    fn next_explorer_row_frame(
1533        &mut self,
1534        query: &ExplorerQuery,
1535        rows_seen: &mut u64,
1536        stats: &ExplorerStats,
1537        control: Option<&mut ExplorerControl<'_>>,
1538    ) -> Result<ExplorerLoopStep> {
1539        if !self.step_for_explorer(query.direction)? {
1540            return Ok(ExplorerLoopStep::Stop);
1541        }
1542        *rows_seen = rows_seen.saturating_add(1);
1543        if control.is_some_and(|control| control.should_stop_after_rows(*rows_seen, stats)) {
1544            return Ok(ExplorerLoopStep::Stop);
1545        }
1546        let Some(metadata) = self.row.metadata() else {
1547            return Ok(ExplorerLoopStep::Skip);
1548        };
1549        if stop_by_commit_time(query, metadata.realtime) {
1550            return Ok(ExplorerLoopStep::Stop);
1551        }
1552        if skip_by_commit_time(query, metadata.realtime) {
1553            return Ok(ExplorerLoopStep::Skip);
1554        }
1555        Ok(ExplorerLoopStep::Row(ExplorerRowFrame {
1556            commit_realtime: metadata.realtime,
1557            seqnum: metadata.seqnum,
1558        }))
1559    }
1560
1561    fn scan_row_data_or_default(
1562        &mut self,
1563        query: &ExplorerQuery,
1564        accumulator: &mut ExplorerAccumulator,
1565        row_id: &mut u64,
1566        deferred_values: &mut Vec<usize>,
1567        stats: &mut ExplorerStats,
1568    ) -> Result<RowScan> {
1569        if accumulator.required_identity_count == 0 && !query_has_fts(query) {
1570            stats.rows_examined = stats.rows_examined.saturating_add(1);
1571            return Ok(RowScan::default());
1572        }
1573        *row_id = row_id.saturating_add(1);
1574        deferred_values.clear();
1575        self.scan_current_row(
1576            query,
1577            accumulator,
1578            *row_id,
1579            ScanApply::Deferred(deferred_values),
1580            stats,
1581        )
1582    }
1583
1584    fn accepted_effective_realtime(
1585        query: &ExplorerQuery,
1586        scan: &RowScan,
1587        commit_realtime: u64,
1588        stats: &mut ExplorerStats,
1589        control: Option<&mut ExplorerControl<'_>>,
1590    ) -> Option<u64> {
1591        let effective_realtime = effective_realtime_from_scan(scan.timestamp, commit_realtime);
1592        record_source_realtime_delta(stats, scan.timestamp, commit_realtime);
1593        let _ = control;
1594        (timestamp_in_range(query, effective_realtime) && !row_rejected_by_fts(query, scan))
1595            .then_some(effective_realtime)
1596    }
1597
1598    fn push_explorer_row_if_wanted(
1599        &mut self,
1600        query: &ExplorerQuery,
1601        result: &mut ExplorerResult,
1602        row_payload_mode: ExplorerRowPayloadMode,
1603        effective_realtime: u64,
1604    ) -> Result<()> {
1605        if row_within_anchor(query, effective_realtime) && result.rows.len() < query.limit {
1606            result.rows.push(self.current_explorer_row(
1607                effective_realtime,
1608                &mut result.stats,
1609                row_payload_mode,
1610            )?);
1611        }
1612        Ok(())
1613    }
1614
1615    fn apply_main_scanned_row(
1616        &mut self,
1617        query: &ExplorerQuery,
1618        accumulator: &mut ExplorerAccumulator,
1619        result: &mut ExplorerResult,
1620        row_payload_mode: ExplorerRowPayloadMode,
1621        scanned: MainScannedRow<'_>,
1622        control: Option<&mut ExplorerControl<'_>>,
1623    ) -> Result<bool> {
1624        if query.debug_collect_column_fields_by_row_traversal {
1625            result.column_fields.extend(scanned.scan.column_fields);
1626        }
1627        record_last_realtime(&mut result.stats, scanned.commit_realtime);
1628        result.stats.rows_matched = result.stats.rows_matched.saturating_add(1);
1629        let stop_after_matched_row = control.is_some_and(|control| {
1630            control.emit_matched_row(scanned.effective_realtime, result.stats.rows_matched)
1631        });
1632        for value_index in scanned.deferred_values {
1633            accumulator.apply_value(
1634                *value_index,
1635                Some(scanned.effective_realtime),
1636                &mut result.stats,
1637            );
1638        }
1639        accumulator.finish_histogram_row(
1640            scanned.row_id,
1641            scanned.effective_realtime,
1642            &mut result.stats,
1643        );
1644        self.push_explorer_row_if_wanted(
1645            query,
1646            result,
1647            row_payload_mode,
1648            scanned.effective_realtime,
1649        )?;
1650        Ok(stop_after_matched_row
1651            || should_stop_when_rows_full(
1652                query,
1653                &result.rows,
1654                scanned.effective_realtime,
1655                result.stats.rows_matched,
1656            ))
1657    }
1658
1659    fn sampling_state_for_combined(
1660        query: &ExplorerQuery,
1661        result: &ExplorerResult,
1662        control: Option<&mut ExplorerControl<'_>>,
1663    ) -> Option<ExplorerSamplingState> {
1664        let sampling = ExplorerSamplingState::for_query(
1665            query,
1666            result
1667                .histogram
1668                .as_ref()
1669                .map(|histogram| histogram.buckets.len()),
1670        );
1671        if let Some(control) = control {
1672            if let (Some(shared_sampling), Some(file_sampling)) =
1673                (control.sampling.as_deref_mut(), query.sampling)
1674            {
1675                shared_sampling.begin_file(file_sampling);
1676            }
1677        }
1678        sampling
1679    }
1680
1681    fn combined_sampling_decision(
1682        query: &ExplorerQuery,
1683        rows: &[ExplorerRow],
1684        frame: ExplorerRowFrame,
1685        sampling: &mut Option<ExplorerSamplingState>,
1686        mut control: Option<&mut ExplorerControl<'_>>,
1687    ) -> Option<ExplorerSamplingDecision> {
1688        let candidate_to_keep = if let Some(control) = control.as_deref_mut() {
1689            control.candidate_row.as_deref_mut().map_or_else(
1690                || row_candidate_to_keep(query, rows, frame.commit_realtime),
1691                |candidate_row| candidate_row(frame.commit_realtime),
1692            )
1693        } else {
1694            row_candidate_to_keep(query, rows, frame.commit_realtime)
1695        };
1696        if let Some(control) = control {
1697            if let Some(shared_sampling) = control.sampling.as_deref_mut() {
1698                return Some(shared_sampling.decide(
1699                    frame.commit_realtime,
1700                    frame.seqnum,
1701                    candidate_to_keep,
1702                ));
1703            }
1704        }
1705        sampling
1706            .as_mut()
1707            .map(|sampling| sampling.decide(frame.commit_realtime, frame.seqnum, candidate_to_keep))
1708    }
1709
1710    fn apply_combined_sampling_decision(
1711        decision: ExplorerSamplingDecision,
1712        mode: CombinedScanMode,
1713        result: &mut ExplorerResult,
1714        frame: ExplorerRowFrame,
1715    ) -> SamplingRowAction {
1716        match decision {
1717            ExplorerSamplingDecision::Full { sampled } => {
1718                if sampled {
1719                    result.stats.sampling_sampled = result.stats.sampling_sampled.saturating_add(1);
1720                }
1721                SamplingRowAction::Scan
1722            }
1723            ExplorerSamplingDecision::SkipFields => {
1724                record_combined_unsampled_row(
1725                    &mut result.stats,
1726                    mode,
1727                    frame.commit_realtime,
1728                    1,
1729                    true,
1730                );
1731                add_special_histogram_value(
1732                    result.histogram.as_mut(),
1733                    frame.commit_realtime,
1734                    EXPLORER_UNSAMPLED_VALUE,
1735                    1,
1736                    &mut result.stats,
1737                );
1738                SamplingRowAction::Skip
1739            }
1740            ExplorerSamplingDecision::StopAndEstimate {
1741                remaining_rows,
1742                from_realtime_usec,
1743                to_realtime_usec,
1744            } => {
1745                record_combined_unsampled_row(
1746                    &mut result.stats,
1747                    mode,
1748                    frame.commit_realtime,
1749                    remaining_rows,
1750                    false,
1751                );
1752                result.stats.rows_estimated =
1753                    result.stats.rows_estimated.saturating_add(remaining_rows);
1754                result.stats.sampling_estimated = result
1755                    .stats
1756                    .sampling_estimated
1757                    .saturating_add(remaining_rows);
1758                add_estimated_histogram_range(
1759                    result.histogram.as_mut(),
1760                    from_realtime_usec,
1761                    to_realtime_usec,
1762                    remaining_rows,
1763                    &mut result.stats,
1764                );
1765                SamplingRowAction::Stop
1766            }
1767        }
1768    }
1769
1770    fn apply_combined_scanned_row(
1771        &mut self,
1772        query: &ExplorerQuery,
1773        accumulator: &mut ExplorerAccumulator,
1774        result: &mut ExplorerResult,
1775        row_payload_mode: ExplorerRowPayloadMode,
1776        mode: CombinedScanMode,
1777        scanned: MainScannedRow<'_>,
1778        control: Option<&mut ExplorerControl<'_>>,
1779    ) -> Result<bool> {
1780        if query.debug_collect_column_fields_by_row_traversal {
1781            result.column_fields.extend(scanned.scan.column_fields);
1782        }
1783        record_last_realtime(&mut result.stats, scanned.commit_realtime);
1784        let stop_after_matched_row = update_combined_matched_stats(
1785            &mut result.stats,
1786            mode,
1787            scanned.effective_realtime,
1788            control,
1789        );
1790        let value_realtime = query
1791            .histogram
1792            .is_some()
1793            .then_some(scanned.effective_realtime);
1794        for value_index in scanned.deferred_values {
1795            accumulator.apply_value(*value_index, value_realtime, &mut result.stats);
1796        }
1797        if query.histogram.is_some() {
1798            accumulator.finish_histogram_row(
1799                scanned.row_id,
1800                scanned.effective_realtime,
1801                &mut result.stats,
1802            );
1803        }
1804        if mode.include_facets {
1805            accumulator.finish_facet_row(scanned.row_id, &mut result.stats);
1806        }
1807        self.push_explorer_row_if_wanted(
1808            query,
1809            result,
1810            row_payload_mode,
1811            scanned.effective_realtime,
1812        )?;
1813        Ok(stop_after_matched_row
1814            || should_stop_when_rows_full(
1815                query,
1816                &result.rows,
1817                scanned.effective_realtime,
1818                result.stats.rows_matched,
1819            ))
1820    }
1821
1822    fn scan_explorer_main(
1823        &mut self,
1824        query: &ExplorerQuery,
1825        accumulator: &mut ExplorerAccumulator,
1826        result: &mut ExplorerResult,
1827        row_payload_mode: ExplorerRowPayloadMode,
1828        mut control: Option<&mut ExplorerControl<'_>>,
1829    ) -> Result<()> {
1830        self.seek_for_explorer(query);
1831        let mut row_id = 0u64;
1832        let mut rows_seen = 0u64;
1833        let mut deferred_values = Vec::new();
1834        loop {
1835            let frame = match self.next_explorer_row_frame(
1836                query,
1837                &mut rows_seen,
1838                &result.stats,
1839                control.as_deref_mut(),
1840            )? {
1841                ExplorerLoopStep::Stop => break,
1842                ExplorerLoopStep::Skip => continue,
1843                ExplorerLoopStep::Row(frame) => frame,
1844            };
1845            let scan = self.scan_row_data_or_default(
1846                query,
1847                accumulator,
1848                &mut row_id,
1849                &mut deferred_values,
1850                &mut result.stats,
1851            )?;
1852            let Some(effective_realtime) = Self::accepted_effective_realtime(
1853                query,
1854                &scan,
1855                frame.commit_realtime,
1856                &mut result.stats,
1857                control.as_deref_mut(),
1858            ) else {
1859                continue;
1860            };
1861            let scanned = MainScannedRow {
1862                row_id,
1863                commit_realtime: frame.commit_realtime,
1864                effective_realtime,
1865                scan,
1866                deferred_values: &deferred_values,
1867            };
1868            if self.apply_main_scanned_row(
1869                query,
1870                accumulator,
1871                result,
1872                row_payload_mode,
1873                scanned,
1874                control.as_deref_mut(),
1875            )? {
1876                break;
1877            }
1878        }
1879        result.stats.rows_returned = result.rows.len() as u64;
1880        Ok(())
1881    }
1882
1883    fn scan_explorer_combined(
1884        &mut self,
1885        query: &ExplorerQuery,
1886        accumulator: &mut ExplorerAccumulator,
1887        result: &mut ExplorerResult,
1888        include_facets: bool,
1889        row_payload_mode: ExplorerRowPayloadMode,
1890        mut control: Option<&mut ExplorerControl<'_>>,
1891    ) -> Result<()> {
1892        self.seek_for_explorer(query);
1893        let mode = CombinedScanMode {
1894            include_main: query_needs_main_pass(query),
1895            include_facets,
1896        };
1897        let mut row_id = 0u64;
1898        let mut rows_seen = 0u64;
1899        let mut deferred_values = Vec::new();
1900        let mut sampling = Self::sampling_state_for_combined(query, result, control.as_deref_mut());
1901        loop {
1902            let frame = match self.next_explorer_row_frame(
1903                query,
1904                &mut rows_seen,
1905                &result.stats,
1906                control.as_deref_mut(),
1907            )? {
1908                ExplorerLoopStep::Stop => break,
1909                ExplorerLoopStep::Skip => continue,
1910                ExplorerLoopStep::Row(frame) => frame,
1911            };
1912            if let Some(decision) = Self::combined_sampling_decision(
1913                query,
1914                &result.rows,
1915                frame,
1916                &mut sampling,
1917                control.as_deref_mut(),
1918            ) {
1919                match Self::apply_combined_sampling_decision(decision, mode, result, frame) {
1920                    SamplingRowAction::Scan => {}
1921                    SamplingRowAction::Skip => continue,
1922                    SamplingRowAction::Stop => break,
1923                }
1924            }
1925            let scan = self.scan_row_data_or_default(
1926                query,
1927                accumulator,
1928                &mut row_id,
1929                &mut deferred_values,
1930                &mut result.stats,
1931            )?;
1932            let Some(effective_realtime) = Self::accepted_effective_realtime(
1933                query,
1934                &scan,
1935                frame.commit_realtime,
1936                &mut result.stats,
1937                control.as_deref_mut(),
1938            ) else {
1939                continue;
1940            };
1941            let scanned = MainScannedRow {
1942                row_id,
1943                commit_realtime: frame.commit_realtime,
1944                effective_realtime,
1945                scan,
1946                deferred_values: &deferred_values,
1947            };
1948            if self.apply_combined_scanned_row(
1949                query,
1950                accumulator,
1951                result,
1952                row_payload_mode,
1953                mode,
1954                scanned,
1955                control.as_deref_mut(),
1956            )? {
1957                break;
1958            }
1959        }
1960        result.stats.rows_returned = result.rows.len() as u64;
1961        Ok(())
1962    }
1963
1964    fn scan_explorer_facet(
1965        &mut self,
1966        query: &ExplorerQuery,
1967        accumulator: &mut ExplorerAccumulator,
1968        stats: &mut ExplorerStats,
1969        mut control: Option<&mut ExplorerControl<'_>>,
1970    ) -> Result<()> {
1971        self.seek_for_explorer(query);
1972        let defer_apply = query.after_realtime_usec.is_some()
1973            || query.before_realtime_usec.is_some()
1974            || query_has_fts(query);
1975        let mut row_id = 0u64;
1976        let mut rows_seen = 0u64;
1977        let mut deferred_values = Vec::new();
1978        loop {
1979            let frame = match self.next_explorer_row_frame(
1980                query,
1981                &mut rows_seen,
1982                stats,
1983                control.as_deref_mut(),
1984            )? {
1985                ExplorerLoopStep::Stop => break,
1986                ExplorerLoopStep::Skip => continue,
1987                ExplorerLoopStep::Row(frame) => frame,
1988            };
1989            row_id = row_id.saturating_add(1);
1990            deferred_values.clear();
1991            let scan = if defer_apply {
1992                self.scan_current_row(
1993                    query,
1994                    accumulator,
1995                    row_id,
1996                    ScanApply::Deferred(&mut deferred_values),
1997                    stats,
1998                )?
1999            } else {
2000                self.scan_current_row(query, accumulator, row_id, ScanApply::Immediate, stats)?
2001            };
2002            if Self::accepted_effective_realtime(query, &scan, frame.commit_realtime, stats, None)
2003                .is_none()
2004            {
2005                continue;
2006            }
2007            record_last_realtime(stats, frame.commit_realtime);
2008            stats.facet_rows_matched = stats.facet_rows_matched.saturating_add(1);
2009            if defer_apply {
2010                for value_index in &deferred_values {
2011                    accumulator.apply_value(*value_index, None, stats);
2012                }
2013            }
2014            accumulator.finish_facet_row(row_id, stats);
2015        }
2016        Ok(())
2017    }
2018
2019    fn scan_current_row(
2020        &mut self,
2021        query: &ExplorerQuery,
2022        accumulator: &mut ExplorerAccumulator,
2023        row_id: u64,
2024        mut apply: ScanApply<'_>,
2025        stats: &mut ExplorerStats,
2026    ) -> Result<RowScan> {
2027        stats.rows_examined = stats.rows_examined.saturating_add(1);
2028        let mut out = RowScan::default();
2029        let mut state = RowScanState::new(query, accumulator);
2030
2031        let inner = &mut self.inner;
2032        let row = &mut self.row;
2033        inner.with_mut(|fields| {
2034            fields.reader.release_object_guards();
2035            row.restart_data()?;
2036            let result = (|| {
2037                for index in 0..row.data_offset_count() {
2038                    let Some(data_offset) = row.data_offset_at(index) else {
2039                        break;
2040                    };
2041                    stats.data_refs_seen = stats.data_refs_seen.saturating_add(1);
2042                    let class = classify_data_for_accumulator(
2043                        fields.file,
2044                        row,
2045                        data_offset,
2046                        accumulator,
2047                        state.needs_fts,
2048                        query,
2049                        query
2050                            .debug_collect_column_fields_by_row_traversal
2051                            .then_some(&mut out.column_fields),
2052                        stats,
2053                    )?;
2054
2055                    handle_row_offset_class(
2056                        class,
2057                        accumulator,
2058                        row_id,
2059                        &mut state,
2060                        &mut out,
2061                        &mut apply,
2062                        stats,
2063                    );
2064                    if state.should_stop_row_scan() {
2065                        record_row_scan_early_stop(stats);
2066                        break;
2067                    }
2068                }
2069                Ok::<_, SdkError>(())
2070            })();
2071            row.reset_data_state(fields.file)?;
2072            result
2073        })?;
2074        Ok(out)
2075    }
2076
2077    fn seek_for_explorer(&mut self, query: &ExplorerQuery) {
2078        let anchor = if query.stop_when_rows_full {
2079            query.anchor
2080        } else {
2081            ExplorerAnchor::Auto
2082        };
2083        match query.direction {
2084            Direction::Forward => match anchor {
2085                ExplorerAnchor::Auto => {
2086                    if let Some(after) = query.after_realtime_usec {
2087                        self.seek_realtime(after.saturating_sub(query.realtime_slack_usec));
2088                    } else {
2089                        self.seek_head();
2090                    }
2091                }
2092                ExplorerAnchor::Realtime(usec) => self.seek_realtime(usec),
2093                ExplorerAnchor::Tail => self.seek_tail(),
2094                ExplorerAnchor::Head => {
2095                    if let Some(after) = query.after_realtime_usec {
2096                        self.seek_realtime(after.saturating_sub(query.realtime_slack_usec));
2097                    } else {
2098                        self.seek_head();
2099                    }
2100                }
2101            },
2102            Direction::Backward => match anchor {
2103                ExplorerAnchor::Auto => {
2104                    if let Some(before) = query.before_realtime_usec {
2105                        self.seek_realtime(before.saturating_add(query.realtime_slack_usec));
2106                    } else {
2107                        self.seek_tail();
2108                    }
2109                }
2110                ExplorerAnchor::Realtime(usec) => self.seek_realtime(usec),
2111                ExplorerAnchor::Head => self.seek_head(),
2112                ExplorerAnchor::Tail => {
2113                    if let Some(before) = query.before_realtime_usec {
2114                        self.seek_realtime(before.saturating_add(query.realtime_slack_usec));
2115                    } else {
2116                        self.seek_tail();
2117                    }
2118                }
2119            },
2120        }
2121    }
2122
2123    fn step_for_explorer(&mut self, direction: Direction) -> Result<bool> {
2124        match direction {
2125            Direction::Forward => self.next(),
2126            Direction::Backward => self.previous(),
2127        }
2128    }
2129
2130    fn current_explorer_row(
2131        &mut self,
2132        realtime_usec: u64,
2133        stats: &mut ExplorerStats,
2134        row_payload_mode: ExplorerRowPayloadMode,
2135    ) -> Result<ExplorerRow> {
2136        let cursor = self.get_cursor()?;
2137        let mut payloads = Vec::new();
2138        if row_payload_mode == ExplorerRowPayloadMode::Expand {
2139            self.collect_entry_payloads(&mut payloads)?;
2140            stats.returned_row_expansions = stats.returned_row_expansions.saturating_add(1);
2141        }
2142        Ok(ExplorerRow {
2143            realtime_usec,
2144            cursor,
2145            payloads,
2146        })
2147    }
2148}
2149
2150enum ScanApply<'a> {
2151    Immediate,
2152    Deferred(&'a mut Vec<usize>),
2153}
2154
2155#[derive(Debug, Clone, Copy)]
2156struct ExplorerRowFrame {
2157    commit_realtime: u64,
2158    seqnum: u64,
2159}
2160
2161enum ExplorerLoopStep {
2162    Stop,
2163    Skip,
2164    Row(ExplorerRowFrame),
2165}
2166
2167#[derive(Debug, Clone, Copy)]
2168struct CombinedScanMode {
2169    include_main: bool,
2170    include_facets: bool,
2171}
2172
2173struct MainScannedRow<'a> {
2174    row_id: u64,
2175    commit_realtime: u64,
2176    effective_realtime: u64,
2177    scan: RowScan,
2178    deferred_values: &'a [usize],
2179}
2180
2181struct RowScanState {
2182    use_first_value: bool,
2183    needs_fts: bool,
2184    collect_column_fields: bool,
2185    fields_missing_from_row: usize,
2186}
2187
2188impl RowScanState {
2189    fn new(query: &ExplorerQuery, accumulator: &ExplorerAccumulator) -> Self {
2190        let use_first_value = query.field_mode == ExplorerFieldMode::FirstValue;
2191        Self {
2192            use_first_value,
2193            needs_fts: query_has_fts(query),
2194            collect_column_fields: query.debug_collect_column_fields_by_row_traversal,
2195            fields_missing_from_row: if use_first_value {
2196                accumulator.required_identity_count
2197            } else {
2198                0
2199            },
2200        }
2201    }
2202
2203    fn should_stop_row_scan(&self) -> bool {
2204        self.use_first_value
2205            && !self.needs_fts
2206            && !self.collect_column_fields
2207            && self.fields_missing_from_row == 0
2208    }
2209}
2210
2211enum SamplingRowAction {
2212    Scan,
2213    Skip,
2214    Stop,
2215}
2216
2217enum IndexedCandidateSet {
2218    All {
2219        count: u64,
2220    },
2221    Set {
2222        count: u64,
2223        offsets: HashSet<NonZeroU64>,
2224    },
2225}
2226
2227impl IndexedCandidateSet {
2228    fn count(&self) -> u64 {
2229        match self {
2230            Self::All { count } | Self::Set { count, .. } => *count,
2231        }
2232    }
2233
2234    fn contains(&self, entry_offset: NonZeroU64) -> bool {
2235        match self {
2236            Self::All { .. } => true,
2237            Self::Set { offsets, .. } => offsets.contains(&entry_offset),
2238        }
2239    }
2240}
2241
2242struct FacetPassGroup {
2243    excluded_field: Option<Vec<u8>>,
2244    facet_indices: Vec<usize>,
2245}
2246
2247fn facet_pass_groups(query: &ExplorerQuery) -> Vec<FacetPassGroup> {
2248    let filter_fields: HashSet<&[u8]> = query
2249        .filters
2250        .iter()
2251        .map(|filter| filter.field.as_slice())
2252        .collect();
2253    let mut groups: Vec<FacetPassGroup> = Vec::new();
2254
2255    for (index, facet) in query.facets.iter().enumerate() {
2256        let excluded_field = (query.exclude_facet_field_filters
2257            && filter_fields.contains(facet.as_slice()))
2258        .then(|| facet.clone());
2259        if let Some(existing) = groups
2260            .iter_mut()
2261            .find(|group| group.excluded_field.as_deref() == excluded_field.as_deref())
2262        {
2263            existing.facet_indices.push(index);
2264        } else {
2265            groups.push(FacetPassGroup {
2266                excluded_field,
2267                facet_indices: vec![index],
2268            });
2269        }
2270    }
2271
2272    groups
2273}
2274
2275fn indexed_count_facet_group(
2276    file: &JournalFile<Mmap>,
2277    query: &ExplorerQuery,
2278    group: &FacetPassGroup,
2279    candidates: &IndexedCandidateSet,
2280    result: &mut ExplorerResult,
2281) -> Result<()> {
2282    result.stats.facet_rows_matched = result
2283        .stats
2284        .facet_rows_matched
2285        .saturating_add(candidates.count());
2286
2287    for facet_index in &group.facet_indices {
2288        let Some(field) = query.facets.get(*facet_index) else {
2289            continue;
2290        };
2291        let mut values = HashMap::new();
2292        let mut rows_with_field = HashSet::new();
2293        let mut decompressed = Vec::new();
2294
2295        for item in file.field_data_objects_with_offsets(field)? {
2296            let (_, data) = item?;
2297            let Some((value, cursor)) =
2298                indexed_value_and_cursor(&data, field, &mut decompressed, &mut result.stats)?
2299            else {
2300                continue;
2301            };
2302            drop(data);
2303
2304            let count = indexed_count_facet_entries(
2305                file,
2306                cursor,
2307                candidates,
2308                &mut rows_with_field,
2309                &mut result.stats,
2310            )?;
2311            if count == 0 {
2312                continue;
2313            }
2314            increment_counter_by(&mut values, &value, count);
2315            result.stats.facet_updates = result.stats.facet_updates.saturating_add(count);
2316        }
2317
2318        let unset = candidates
2319            .count()
2320            .saturating_sub(rows_with_field.len() as u64);
2321        if unset != 0 {
2322            increment_counter_by(&mut values, UNSET_VALUE, unset);
2323            result.stats.facet_updates = result.stats.facet_updates.saturating_add(unset);
2324        }
2325        result.facets.insert(field.clone(), values);
2326    }
2327
2328    Ok(())
2329}
2330
2331fn indexed_count_histogram(
2332    file: &JournalFile<Mmap>,
2333    query: &ExplorerQuery,
2334    candidates: &IndexedCandidateSet,
2335    result: &mut ExplorerResult,
2336) -> Result<()> {
2337    let Some(histogram) = result.histogram.as_mut() else {
2338        return Ok(());
2339    };
2340    let field = histogram.field.clone();
2341    let mut decompressed = Vec::new();
2342    let mut rows_with_field = HashSet::new();
2343
2344    for item in file.field_data_objects_with_offsets(&field)? {
2345        let (_, data) = item?;
2346        let Some((value, cursor)) =
2347            indexed_value_and_cursor(&data, &field, &mut decompressed, &mut result.stats)?
2348        else {
2349            continue;
2350        };
2351        drop(data);
2352
2353        indexed_count_histogram_entries(
2354            file,
2355            cursor,
2356            candidates,
2357            &value,
2358            histogram,
2359            query,
2360            &mut rows_with_field,
2361            &mut result.stats,
2362        )?;
2363    }
2364
2365    indexed_count_histogram_unset_entries(
2366        file,
2367        candidates,
2368        &rows_with_field,
2369        histogram,
2370        query,
2371        &mut result.stats,
2372    )?;
2373
2374    Ok(())
2375}
2376
2377fn indexed_value_and_cursor(
2378    data: &DataObject<&[u8]>,
2379    field: &[u8],
2380    decompressed: &mut Vec<u8>,
2381    stats: &mut ExplorerStats,
2382) -> Result<Option<(Vec<u8>, Option<InlinedCursor>)>> {
2383    stats.data_objects_classified = stats.data_objects_classified.saturating_add(1);
2384    stats.data_payloads_loaded = stats.data_payloads_loaded.saturating_add(1);
2385    let payload = if data.is_compressed() {
2386        decompressed.clear();
2387        let len = data.decompress(decompressed)?;
2388        stats.payloads_decompressed = stats.payloads_decompressed.saturating_add(1);
2389        &decompressed[..len]
2390    } else {
2391        data.raw_payload()
2392    };
2393
2394    let Some((payload_field, value)) = split_payload_bytes(payload) else {
2395        return Ok(None);
2396    };
2397    if payload_field != field {
2398        return Ok(None);
2399    }
2400    Ok(Some((value.to_vec(), data.inlined_cursor())))
2401}
2402
2403fn indexed_count_facet_entries(
2404    file: &JournalFile<Mmap>,
2405    cursor: Option<InlinedCursor>,
2406    candidates: &IndexedCandidateSet,
2407    rows_with_field: &mut HashSet<NonZeroU64>,
2408    stats: &mut ExplorerStats,
2409) -> Result<u64> {
2410    let mut count = 0u64;
2411    indexed_visit_entries(file, cursor, |entry_offset| {
2412        stats.data_refs_seen = stats.data_refs_seen.saturating_add(1);
2413        if candidates.contains(entry_offset) {
2414            count = count.saturating_add(1);
2415            rows_with_field.insert(entry_offset);
2416        }
2417        Ok(())
2418    })?;
2419    Ok(count)
2420}
2421
2422fn indexed_count_histogram_entries(
2423    file: &JournalFile<Mmap>,
2424    cursor: Option<InlinedCursor>,
2425    candidates: &IndexedCandidateSet,
2426    value: &[u8],
2427    histogram: &mut ExplorerHistogram,
2428    query: &ExplorerQuery,
2429    rows_with_field: &mut HashSet<NonZeroU64>,
2430    stats: &mut ExplorerStats,
2431) -> Result<()> {
2432    let histogram_start = histogram
2433        .buckets
2434        .first()
2435        .map(|bucket| bucket.start_realtime_usec)
2436        .unwrap_or_default();
2437    let histogram_bucket_width = histogram
2438        .buckets
2439        .first()
2440        .map(|bucket| {
2441            bucket
2442                .end_realtime_usec
2443                .saturating_sub(bucket.start_realtime_usec)
2444                .max(1)
2445        })
2446        .unwrap_or(1);
2447    let histogram_bucket_count = histogram.buckets.len();
2448
2449    indexed_visit_entries(file, cursor, |entry_offset| {
2450        stats.data_refs_seen = stats.data_refs_seen.saturating_add(1);
2451        if !candidates.contains(entry_offset) {
2452            return Ok(());
2453        }
2454        let entry = file.entry_ref(entry_offset)?;
2455        let realtime = entry.header.realtime;
2456        drop(entry);
2457        rows_with_field.insert(entry_offset);
2458        if !timestamp_in_range(query, realtime) {
2459            return Ok(());
2460        }
2461        let Some(bucket_index) = histogram_bucket_index_from_bounds(
2462            realtime,
2463            histogram_start,
2464            histogram_bucket_width,
2465            histogram_bucket_count,
2466        ) else {
2467            return Ok(());
2468        };
2469        if let Some(bucket) = histogram.buckets.get_mut(bucket_index) {
2470            increment_counter_by(&mut bucket.values, value, 1);
2471            stats.histogram_updates = stats.histogram_updates.saturating_add(1);
2472        }
2473        Ok(())
2474    })
2475}
2476
2477fn indexed_count_histogram_unset_entries(
2478    file: &JournalFile<Mmap>,
2479    candidates: &IndexedCandidateSet,
2480    rows_with_field: &HashSet<NonZeroU64>,
2481    histogram: &mut ExplorerHistogram,
2482    query: &ExplorerQuery,
2483    stats: &mut ExplorerStats,
2484) -> Result<()> {
2485    let histogram_start = histogram
2486        .buckets
2487        .first()
2488        .map(|bucket| bucket.start_realtime_usec)
2489        .unwrap_or_default();
2490    let histogram_bucket_width = histogram
2491        .buckets
2492        .first()
2493        .map(|bucket| {
2494            bucket
2495                .end_realtime_usec
2496                .saturating_sub(bucket.start_realtime_usec)
2497                .max(1)
2498        })
2499        .unwrap_or(1);
2500    let histogram_bucket_count = histogram.buckets.len();
2501
2502    let mut visit = |entry_offset: NonZeroU64| -> Result<()> {
2503        if rows_with_field.contains(&entry_offset) {
2504            return Ok(());
2505        }
2506        let entry = file.entry_ref(entry_offset)?;
2507        let realtime = entry.header.realtime;
2508        drop(entry);
2509        if !timestamp_in_range(query, realtime) {
2510            return Ok(());
2511        }
2512        let Some(bucket_index) = histogram_bucket_index_from_bounds(
2513            realtime,
2514            histogram_start,
2515            histogram_bucket_width,
2516            histogram_bucket_count,
2517        ) else {
2518            return Ok(());
2519        };
2520        if let Some(bucket) = histogram.buckets.get_mut(bucket_index) {
2521            increment_counter_by(&mut bucket.values, UNSET_VALUE, 1);
2522            stats.histogram_updates = stats.histogram_updates.saturating_add(1);
2523        }
2524        Ok(())
2525    };
2526
2527    match candidates {
2528        IndexedCandidateSet::All { .. } => {
2529            let mut entry_offsets = Vec::new();
2530            file.entry_offsets(&mut entry_offsets)?;
2531            for entry_offset in entry_offsets {
2532                visit(entry_offset)?;
2533            }
2534        }
2535        IndexedCandidateSet::Set { offsets, .. } => {
2536            for entry_offset in offsets {
2537                visit(*entry_offset)?;
2538            }
2539        }
2540    }
2541
2542    Ok(())
2543}
2544
2545fn indexed_visit_entries<F>(
2546    file: &JournalFile<Mmap>,
2547    cursor: Option<InlinedCursor>,
2548    mut visitor: F,
2549) -> Result<()>
2550where
2551    F: FnMut(NonZeroU64) -> Result<()>,
2552{
2553    let Some(mut cursor) = cursor.map(|cursor| cursor.head()) else {
2554        return Ok(());
2555    };
2556    let mut needle = NonZeroU64::MIN;
2557    while let Some(entry_offset) = cursor.next_until(file, needle)? {
2558        visitor(entry_offset)?;
2559        let Some(next) = entry_offset.get().checked_add(1).and_then(NonZeroU64::new) else {
2560            break;
2561        };
2562        needle = next;
2563    }
2564    Ok(())
2565}
2566
2567fn handle_row_offset_class(
2568    class: OffsetClass,
2569    accumulator: &mut ExplorerAccumulator,
2570    row_id: u64,
2571    state: &mut RowScanState,
2572    out: &mut RowScan,
2573    apply: &mut ScanApply<'_>,
2574    stats: &mut ExplorerStats,
2575) {
2576    match class {
2577        OffsetClass::Irrelevant => {
2578            stats.data_refs_skipped = stats.data_refs_skipped.saturating_add(1);
2579        }
2580        OffsetClass::FtsNegativeMatch => {
2581            out.fts_negative_match = true;
2582        }
2583        OffsetClass::FtsMatch => {
2584            out.fts_matches = true;
2585        }
2586        OffsetClass::Value(value_index) => {
2587            handle_row_value_class(value_index, accumulator, row_id, state, out, apply, stats)
2588        }
2589    }
2590}
2591
2592fn handle_row_value_class(
2593    value_index: usize,
2594    accumulator: &mut ExplorerAccumulator,
2595    row_id: u64,
2596    state: &mut RowScanState,
2597    out: &mut RowScan,
2598    apply: &mut ScanApply<'_>,
2599    stats: &mut ExplorerStats,
2600) {
2601    if accumulator.value_fts_matches[value_index] {
2602        out.fts_matches = true;
2603    }
2604    let field_index = accumulator.value_field_indices[value_index];
2605    let first_for_field = if state.use_first_value
2606        || accumulator.flags[field_index] & (FACET_PUBLIC | FACET_HISTOGRAM) != 0
2607    {
2608        accumulator.mark_field_seen(field_index, row_id)
2609    } else {
2610        true
2611    };
2612    if state.use_first_value && first_for_field {
2613        state.fields_missing_from_row = state.fields_missing_from_row.saturating_sub(1);
2614    }
2615    if !state.use_first_value || first_for_field {
2616        if let Some(timestamp) = accumulator.value_source_realtime[value_index] {
2617            out.timestamp = Some(timestamp);
2618        }
2619        match apply {
2620            ScanApply::Immediate => accumulator.apply_value(value_index, None, stats),
2621            ScanApply::Deferred(values) => values.push(value_index),
2622        }
2623    }
2624}
2625
2626fn record_row_scan_early_stop(stats: &mut ExplorerStats) {
2627    stats.early_stop_opportunities = stats.early_stop_opportunities.saturating_add(1);
2628    stats.early_stops = stats.early_stops.saturating_add(1);
2629}
2630
2631fn cached_offset_class_for_accumulator(
2632    file: &JournalFile<Mmap>,
2633    row: &mut CurrentRowView,
2634    data_offset: NonZeroU64,
2635    accumulator: &ExplorerAccumulator,
2636    column_fields: Option<&mut Vec<Vec<u8>>>,
2637    stats: &mut ExplorerStats,
2638) -> Result<Option<OffsetClass>> {
2639    let Some(class) = accumulator.offset_cache.lookup(data_offset) else {
2640        return Ok(None);
2641    };
2642    if let Some(column_fields) = column_fields {
2643        if let Some((field, _)) = read_payload_field(file, row, data_offset, stats)? {
2644            column_fields.push(field);
2645        }
2646    }
2647    stats.data_cache_hits = stats.data_cache_hits.saturating_add(1);
2648    Ok(Some(class))
2649}
2650
2651fn payload_for_classification<'a>(
2652    file: &JournalFile<Mmap>,
2653    row: &'a mut CurrentRowView,
2654    data_offset: NonZeroU64,
2655    stats: &mut ExplorerStats,
2656) -> Result<&'a [u8]> {
2657    stats.data_cache_misses = stats.data_cache_misses.saturating_add(1);
2658    stats.data_payloads_loaded = stats.data_payloads_loaded.saturating_add(1);
2659    let was_compressed = file.data_ref(data_offset)?.is_compressed();
2660    let payload = row.read_payload_at(file, data_offset)?;
2661    if was_compressed {
2662        stats.payloads_decompressed = stats.payloads_decompressed.saturating_add(1);
2663    }
2664    Ok(row.payload_slice(payload))
2665}
2666
2667fn fts_flags_for_value(
2668    value: &[u8],
2669    needs_fts: bool,
2670    query: &ExplorerQuery,
2671    stats: &mut ExplorerStats,
2672) -> (bool, bool) {
2673    if !needs_fts {
2674        return (false, false);
2675    }
2676    stats.fts_scans = stats.fts_scans.saturating_add(1);
2677    match match_fts_query(value, query) {
2678        FtsTermMatch::Positive => (true, false),
2679        FtsTermMatch::Negative => (false, true),
2680        FtsTermMatch::None => (false, false),
2681    }
2682}
2683
2684fn structured_payload_class(
2685    field: &[u8],
2686    value: &[u8],
2687    data_offset: NonZeroU64,
2688    accumulator: &mut ExplorerAccumulator,
2689    fts_matches: bool,
2690    fts_negative_match: bool,
2691) -> OffsetClass {
2692    if fts_negative_match {
2693        OffsetClass::FtsNegativeMatch
2694    } else if let Some(field_index) = accumulator.field_lookup.get(field).copied() {
2695        OffsetClass::Value(accumulator.add_value(field_index, data_offset, value, fts_matches))
2696    } else if fts_matches {
2697        OffsetClass::FtsMatch
2698    } else {
2699        OffsetClass::Irrelevant
2700    }
2701}
2702
2703fn classify_data_for_accumulator(
2704    file: &JournalFile<Mmap>,
2705    row: &mut CurrentRowView,
2706    data_offset: NonZeroU64,
2707    accumulator: &mut ExplorerAccumulator,
2708    needs_fts: bool,
2709    query: &ExplorerQuery,
2710    mut column_fields: Option<&mut Vec<Vec<u8>>>,
2711    stats: &mut ExplorerStats,
2712) -> Result<OffsetClass> {
2713    if let Some(class) = cached_offset_class_for_accumulator(
2714        file,
2715        row,
2716        data_offset,
2717        accumulator,
2718        column_fields.as_mut().map(|fields| &mut **fields),
2719        stats,
2720    )? {
2721        return Ok(class);
2722    }
2723
2724    let payload = payload_for_classification(file, row, data_offset, stats)?;
2725    let Some((field, value)) = split_payload_bytes(payload) else {
2726        let class = classify_unstructured_payload(payload, needs_fts, query, stats);
2727        accumulator.offset_cache.insert(data_offset, class);
2728        stats.data_objects_classified = stats.data_objects_classified.saturating_add(1);
2729        return Ok(class);
2730    };
2731    if let Some(column_fields) = column_fields {
2732        column_fields.push(field.to_vec());
2733    }
2734
2735    let (fts_matches, fts_negative_match) = fts_flags_for_value(value, needs_fts, query, stats);
2736    let class = structured_payload_class(
2737        field,
2738        value,
2739        data_offset,
2740        accumulator,
2741        fts_matches,
2742        fts_negative_match,
2743    );
2744    accumulator.offset_cache.insert(data_offset, class);
2745    stats.data_objects_classified = stats.data_objects_classified.saturating_add(1);
2746    Ok(class)
2747}
2748
2749fn read_payload_field(
2750    file: &JournalFile<Mmap>,
2751    row: &mut CurrentRowView,
2752    data_offset: NonZeroU64,
2753    stats: &mut ExplorerStats,
2754) -> Result<Option<(Vec<u8>, Vec<u8>)>> {
2755    let was_compressed = file.data_ref(data_offset)?.is_compressed();
2756    let payload = row.read_payload_at(file, data_offset)?;
2757    if was_compressed {
2758        stats.payloads_decompressed = stats.payloads_decompressed.saturating_add(1);
2759    }
2760    let payload = row.payload_slice(payload);
2761    Ok(split_payload_bytes(payload).map(|(field, value)| (field.to_vec(), value.to_vec())))
2762}
2763
2764fn classify_unstructured_payload(
2765    payload: &[u8],
2766    needs_fts: bool,
2767    query: &ExplorerQuery,
2768    stats: &mut ExplorerStats,
2769) -> OffsetClass {
2770    if !needs_fts {
2771        return OffsetClass::Irrelevant;
2772    }
2773    stats.fts_scans = stats.fts_scans.saturating_add(1);
2774    match match_fts_query(payload, query) {
2775        FtsTermMatch::Positive => OffsetClass::FtsMatch,
2776        FtsTermMatch::Negative => OffsetClass::FtsNegativeMatch,
2777        FtsTermMatch::None => OffsetClass::Irrelevant,
2778    }
2779}
2780
2781fn histogram_bucket_index_from_bounds(
2782    realtime_usec: u64,
2783    start_realtime_usec: u64,
2784    bucket_width_usec: u64,
2785    bucket_count: usize,
2786) -> Option<usize> {
2787    if bucket_count == 0 {
2788        return None;
2789    }
2790    realtime_usec
2791        .saturating_sub(start_realtime_usec)
2792        .checked_div(bucket_width_usec.max(1))
2793        .map(|index| (index as usize).min(bucket_count - 1))
2794}
2795
2796fn validate_query(query: &ExplorerQuery) -> Result<()> {
2797    if query
2798        .after_realtime_usec
2799        .zip(query.before_realtime_usec)
2800        .is_some_and(|(after, before)| after > before)
2801    {
2802        return Err(SdkError::InvalidPath(
2803            "after_realtime_usec must be <= before_realtime_usec".to_string(),
2804        ));
2805    }
2806    for filter in &query.filters {
2807        if filter.field.is_empty() || filter.field.contains(&b'=') {
2808            return Err(SdkError::InvalidPath(
2809                "filter field must be non-empty and must not contain '='".to_string(),
2810            ));
2811        }
2812    }
2813    for field in query.facets.iter().chain(query.histogram.iter()) {
2814        if field.is_empty() || field.contains(&b'=') {
2815            return Err(SdkError::InvalidPath(
2816                "facet and histogram fields must be non-empty and must not contain '='".to_string(),
2817            ));
2818        }
2819    }
2820    let mut seen_facets: HashSet<&[u8]> = HashSet::new();
2821    for facet in &query.facets {
2822        if !seen_facets.insert(facet) {
2823            return Err(SdkError::InvalidPath(
2824                "facet fields must not be duplicated".to_string(),
2825            ));
2826        }
2827    }
2828    Ok(())
2829}
2830
2831fn validate_no_debug_column_collection(query: &ExplorerQuery) -> Result<()> {
2832    if query.debug_collect_column_fields_by_row_traversal {
2833        return Err(SdkError::Unsupported(
2834            "debug_collect_column_fields_by_row_traversal is a debug-only discrepancy tool; production explorer queries must use FIELD-index column catalogs instead",
2835        ));
2836    }
2837    Ok(())
2838}
2839
2840fn validate_indexed_query(query: &ExplorerQuery) -> Result<()> {
2841    if query.field_mode != ExplorerFieldMode::AllValues {
2842        return Err(SdkError::Unsupported(
2843            "indexed explorer strategy requires ExplorerFieldMode::AllValues",
2844        ));
2845    }
2846    if query_has_fts(query) {
2847        return Err(SdkError::Unsupported(
2848            "indexed explorer strategy does not support FTS",
2849        ));
2850    }
2851    if query.use_source_realtime
2852        && (query.after_realtime_usec.is_some()
2853            || query.before_realtime_usec.is_some()
2854            || query.histogram.is_some())
2855    {
2856        return Err(SdkError::Unsupported(
2857            "indexed explorer strategy requires commit realtime for time-bounded facets and histograms",
2858        ));
2859    }
2860    Ok(())
2861}
2862
2863fn explorer_outputs_match(left: &ExplorerResult, right: &ExplorerResult) -> bool {
2864    if left.rows.len() != right.rows.len() {
2865        return false;
2866    }
2867    if left.rows.iter().zip(&right.rows).any(|(left, right)| {
2868        left.realtime_usec != right.realtime_usec
2869            || left.cursor != right.cursor
2870            || left.payloads != right.payloads
2871    }) {
2872        return false;
2873    }
2874    if left.facets != right.facets {
2875        return false;
2876    }
2877    explorer_histograms_match(left.histogram.as_ref(), right.histogram.as_ref())
2878}
2879
2880fn explorer_histograms_match(
2881    left: Option<&ExplorerHistogram>,
2882    right: Option<&ExplorerHistogram>,
2883) -> bool {
2884    match (left, right) {
2885        (None, None) => true,
2886        (Some(left), Some(right)) => {
2887            left.field == right.field
2888                && left.buckets.len() == right.buckets.len()
2889                && left
2890                    .buckets
2891                    .iter()
2892                    .zip(&right.buckets)
2893                    .all(|(left, right)| {
2894                        left.start_realtime_usec == right.start_realtime_usec
2895                            && left.end_realtime_usec == right.end_realtime_usec
2896                            && left.values == right.values
2897                    })
2898        }
2899        _ => false,
2900    }
2901}
2902
2903fn query_needs_source_realtime_main(query: &ExplorerQuery) -> bool {
2904    query.use_source_realtime
2905        && (query.after_realtime_usec.is_some()
2906            || query.before_realtime_usec.is_some()
2907            || query.histogram.is_some()
2908            || query.limit > 0)
2909}
2910
2911fn facet_pass_needs_source_realtime(query: &ExplorerQuery) -> bool {
2912    query.use_source_realtime
2913        && (query.after_realtime_usec.is_some() || query.before_realtime_usec.is_some())
2914}
2915
2916fn query_needs_main_pass(query: &ExplorerQuery) -> bool {
2917    query.limit > 0 || query.histogram.is_some()
2918}
2919
2920fn explorer_result_for_query(query: &ExplorerQuery) -> ExplorerResult {
2921    ExplorerResult {
2922        histogram: query
2923            .histogram
2924            .as_ref()
2925            .map(|field| new_histogram(field, query)),
2926        ..ExplorerResult::default()
2927    }
2928}
2929
2930fn explorer_control_stopped(control: Option<&ExplorerControl<'_>>) -> bool {
2931    control.and_then(ExplorerControl::stop_reason).is_some()
2932}
2933
2934fn can_run_combined_explorer_pass(facet_groups: &[FacetPassGroup]) -> bool {
2935    facet_groups
2936        .iter()
2937        .all(|group| group.excluded_field.is_none())
2938}
2939
2940fn combined_facet_indices(facet_groups: &[FacetPassGroup]) -> Vec<usize> {
2941    facet_groups
2942        .iter()
2943        .flat_map(|group| group.facet_indices.iter().copied())
2944        .collect()
2945}
2946
2947fn record_combined_unsampled_row(
2948    stats: &mut ExplorerStats,
2949    mode: CombinedScanMode,
2950    commit_realtime: u64,
2951    row_count: u64,
2952    count_rows_unsampled: bool,
2953) {
2954    record_last_realtime(stats, commit_realtime);
2955    if mode.include_main {
2956        stats.rows_matched = stats.rows_matched.saturating_add(row_count);
2957    }
2958    if mode.include_facets {
2959        stats.facet_rows_matched = stats.facet_rows_matched.saturating_add(row_count);
2960    }
2961    if count_rows_unsampled {
2962        stats.rows_unsampled = stats.rows_unsampled.saturating_add(row_count);
2963    }
2964    stats.sampling_unsampled = stats.sampling_unsampled.saturating_add(1);
2965}
2966
2967fn update_combined_matched_stats(
2968    stats: &mut ExplorerStats,
2969    mode: CombinedScanMode,
2970    effective_realtime: u64,
2971    control: Option<&mut ExplorerControl<'_>>,
2972) -> bool {
2973    let mut stop_after_matched_row = false;
2974    if mode.include_main {
2975        stats.rows_matched = stats.rows_matched.saturating_add(1);
2976        stop_after_matched_row = control
2977            .map(|control| control.emit_matched_row(effective_realtime, stats.rows_matched))
2978            .unwrap_or(false);
2979    }
2980    if mode.include_facets {
2981        stats.facet_rows_matched = stats.facet_rows_matched.saturating_add(1);
2982    }
2983    stop_after_matched_row
2984}
2985
2986fn should_stop_when_rows_full(
2987    query: &ExplorerQuery,
2988    rows: &[ExplorerRow],
2989    effective_realtime: u64,
2990    rows_matched: u64,
2991) -> bool {
2992    if !query.stop_when_rows_full || query.limit == 0 || rows.len() < query.limit {
2993        return false;
2994    }
2995    let every = query.stop_when_rows_full_check_every.max(1);
2996    if rows_matched == 0 || rows_matched % every != 0 {
2997        return false;
2998    }
2999    match query.direction {
3000        Direction::Backward => {
3001            rows.iter()
3002                .map(|row| row.realtime_usec)
3003                .min()
3004                .is_some_and(|oldest| {
3005                    effective_realtime < oldest.saturating_sub(query.realtime_slack_usec)
3006                })
3007        }
3008        Direction::Forward => {
3009            rows.iter()
3010                .map(|row| row.realtime_usec)
3011                .max()
3012                .is_some_and(|newest| {
3013                    effective_realtime > newest.saturating_add(query.realtime_slack_usec)
3014                })
3015        }
3016    }
3017}
3018
3019fn row_candidate_to_keep(query: &ExplorerQuery, rows: &[ExplorerRow], realtime_usec: u64) -> bool {
3020    if query.limit == 0 {
3021        return false;
3022    }
3023    if !row_within_anchor(query, realtime_usec) {
3024        return false;
3025    }
3026    if rows.len() < query.limit {
3027        return true;
3028    }
3029    match query.direction {
3030        Direction::Backward => rows
3031            .iter()
3032            .map(|row| row.realtime_usec)
3033            .min()
3034            .is_some_and(|oldest| realtime_usec >= oldest),
3035        Direction::Forward => rows
3036            .iter()
3037            .map(|row| row.realtime_usec)
3038            .max()
3039            .is_some_and(|newest| realtime_usec <= newest),
3040    }
3041}
3042
3043fn row_within_anchor(query: &ExplorerQuery, realtime_usec: u64) -> bool {
3044    match (query.direction, query.anchor) {
3045        (Direction::Forward, ExplorerAnchor::Realtime(anchor)) => realtime_usec > anchor,
3046        (Direction::Backward, ExplorerAnchor::Realtime(anchor)) => realtime_usec <= anchor,
3047        _ => true,
3048    }
3049}
3050
3051fn add_special_histogram_value(
3052    histogram: Option<&mut ExplorerHistogram>,
3053    realtime_usec: u64,
3054    value: &[u8],
3055    count: u64,
3056    stats: &mut ExplorerStats,
3057) {
3058    let Some(histogram) = histogram else {
3059        return;
3060    };
3061    let Some(bucket_index) = histogram_bucket_index(histogram, realtime_usec) else {
3062        return;
3063    };
3064    if let Some(bucket) = histogram.buckets.get_mut(bucket_index) {
3065        increment_counter_by(&mut bucket.values, value, count);
3066        stats.histogram_updates = stats.histogram_updates.saturating_add(1);
3067    }
3068}
3069
3070fn add_estimated_histogram_range(
3071    histogram: Option<&mut ExplorerHistogram>,
3072    from_realtime_usec: u64,
3073    to_realtime_usec: u64,
3074    entries: u64,
3075    stats: &mut ExplorerStats,
3076) {
3077    let Some(histogram) = histogram else {
3078        return;
3079    };
3080    if entries == 0 || from_realtime_usec >= to_realtime_usec {
3081        return;
3082    }
3083
3084    let Some(first) = histogram.buckets.first() else {
3085        return;
3086    };
3087    let Some(last) = histogram.buckets.last() else {
3088        return;
3089    };
3090    let from_realtime_usec = from_realtime_usec.max(first.start_realtime_usec);
3091    let to_realtime_usec = to_realtime_usec.min(last.end_realtime_usec);
3092    if from_realtime_usec >= to_realtime_usec {
3093        return;
3094    }
3095
3096    let total = to_realtime_usec.saturating_sub(from_realtime_usec).max(1);
3097    let mut touched = 0u64;
3098    for bucket in &mut histogram.buckets {
3099        if bucket.start_realtime_usec > to_realtime_usec {
3100            break;
3101        }
3102        let overlap_start = bucket.start_realtime_usec.max(from_realtime_usec);
3103        let overlap_end = bucket.end_realtime_usec.min(to_realtime_usec);
3104        if overlap_start >= overlap_end {
3105            continue;
3106        }
3107        let bucket_entries = ((overlap_end.saturating_sub(overlap_start) as u128 * entries as u128)
3108            / total as u128) as u64;
3109        if bucket_entries != 0 {
3110            increment_counter_by(&mut bucket.values, EXPLORER_ESTIMATED_VALUE, bucket_entries);
3111        }
3112        touched = touched.saturating_add(1);
3113    }
3114    stats.histogram_updates = stats.histogram_updates.saturating_add(touched);
3115}
3116
3117fn histogram_bucket_index(histogram: &ExplorerHistogram, realtime_usec: u64) -> Option<usize> {
3118    let first = histogram.buckets.first()?;
3119    let width = first
3120        .end_realtime_usec
3121        .saturating_sub(first.start_realtime_usec)
3122        .max(1);
3123    histogram_bucket_index_from_bounds(
3124        realtime_usec,
3125        first.start_realtime_usec,
3126        width,
3127        histogram.buckets.len(),
3128    )
3129}
3130
3131fn payload_from_parts(field: &[u8], value: &[u8]) -> Vec<u8> {
3132    let mut out = Vec::with_capacity(field.len() + 1 + value.len());
3133    out.extend_from_slice(field);
3134    out.push(b'=');
3135    out.extend_from_slice(value);
3136    out
3137}
3138
3139fn split_payload_bytes(payload: &[u8]) -> Option<(&[u8], &[u8])> {
3140    let eq = payload.iter().position(|byte| *byte == b'=')?;
3141    Some((&payload[..eq], &payload[eq + 1..]))
3142}
3143
3144fn parse_source_realtime(value: &[u8]) -> Option<u64> {
3145    std::str::from_utf8(value).ok()?.parse().ok()
3146}
3147
3148fn effective_realtime_from_scan(source_realtime: Option<u64>, commit_realtime: u64) -> u64 {
3149    match source_realtime {
3150        Some(source_realtime) if source_realtime != 0 && source_realtime < commit_realtime => {
3151            source_realtime
3152        }
3153        _ => commit_realtime,
3154    }
3155}
3156
3157fn record_last_realtime(stats: &mut ExplorerStats, commit_realtime: u64) {
3158    if commit_realtime > stats.last_realtime_usec {
3159        stats.last_realtime_usec = commit_realtime;
3160    }
3161}
3162
3163fn record_source_realtime_delta(
3164    stats: &mut ExplorerStats,
3165    source_realtime: Option<u64>,
3166    commit_realtime: u64,
3167) {
3168    let Some(source_realtime) = source_realtime else {
3169        return;
3170    };
3171    if source_realtime == 0 || source_realtime >= commit_realtime {
3172        return;
3173    }
3174    let delta = commit_realtime.saturating_sub(source_realtime);
3175    if delta > stats.max_source_realtime_delta_usec {
3176        stats.max_source_realtime_delta_usec = delta;
3177    }
3178}
3179
3180fn query_has_fts(query: &ExplorerQuery) -> bool {
3181    !query.fts_terms.is_empty()
3182        || !query.fts_patterns.is_empty()
3183        || !query.fts_negative_patterns.is_empty()
3184}
3185
3186fn query_has_positive_fts(query: &ExplorerQuery) -> bool {
3187    if !query.fts_terms.is_empty() {
3188        query.fts_terms.iter().any(|term| !term.negative)
3189    } else {
3190        !query.fts_patterns.is_empty()
3191    }
3192}
3193
3194fn row_rejected_by_fts(query: &ExplorerQuery, scan: &RowScan) -> bool {
3195    query_has_fts(query)
3196        && (scan.fts_negative_match || query_has_positive_fts(query) && !scan.fts_matches)
3197}
3198
3199#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3200enum FtsTermMatch {
3201    None,
3202    Positive,
3203    Negative,
3204}
3205
3206fn match_fts_query(value: &[u8], query: &ExplorerQuery) -> FtsTermMatch {
3207    if !query.fts_terms.is_empty() {
3208        for term in &query.fts_terms {
3209            if term.matches(value) {
3210                return if term.negative {
3211                    FtsTermMatch::Negative
3212                } else {
3213                    FtsTermMatch::Positive
3214                };
3215            }
3216        }
3217        return FtsTermMatch::None;
3218    }
3219
3220    if matches_fts(value, &query.fts_negative_patterns) {
3221        FtsTermMatch::Negative
3222    } else if matches_fts(value, &query.fts_patterns) {
3223        FtsTermMatch::Positive
3224    } else {
3225        FtsTermMatch::None
3226    }
3227}
3228
3229fn matches_fts(value: &[u8], patterns: &[Vec<u8>]) -> bool {
3230    patterns
3231        .iter()
3232        .filter(|pattern| !pattern.is_empty())
3233        .any(|pattern| contains_ascii_case_insensitive(value, pattern))
3234}
3235
3236fn contains_ascii_case_insensitive(haystack: &[u8], needle: &[u8]) -> bool {
3237    if needle.is_empty() {
3238        return true;
3239    }
3240    if haystack.len() < needle.len() {
3241        return false;
3242    }
3243    haystack.windows(needle.len()).any(|window| {
3244        window
3245            .iter()
3246            .zip(needle)
3247            .all(|(left, right)| left.eq_ignore_ascii_case(right))
3248    })
3249}
3250
3251fn find_ascii_case_insensitive(haystack: &[u8], needle: &[u8]) -> Option<usize> {
3252    if needle.is_empty() {
3253        return Some(0);
3254    }
3255    if haystack.len() < needle.len() {
3256        return None;
3257    }
3258    haystack.windows(needle.len()).position(|window| {
3259        window
3260            .iter()
3261            .zip(needle)
3262            .all(|(left, right)| left.eq_ignore_ascii_case(right))
3263    })
3264}
3265
3266fn timestamp_in_range(query: &ExplorerQuery, timestamp: u64) -> bool {
3267    if query
3268        .after_realtime_usec
3269        .is_some_and(|after| timestamp < after)
3270    {
3271        return false;
3272    }
3273    if query
3274        .before_realtime_usec
3275        .is_some_and(|before| timestamp > before)
3276    {
3277        return false;
3278    }
3279    true
3280}
3281
3282fn stop_by_commit_time(query: &ExplorerQuery, commit_realtime: u64) -> bool {
3283    // Netdata expands the seek/stop side by the learned journal-vs-source
3284    // realtime delta, then still uses commit realtime for the fast boundary
3285    // checks before row DATA is scanned for _SOURCE_REALTIME_TIMESTAMP.
3286    match query.direction {
3287        Direction::Forward => query.before_realtime_usec.is_some_and(|before| {
3288            commit_realtime > before.saturating_add(query.realtime_slack_usec)
3289        }),
3290        Direction::Backward => query
3291            .after_realtime_usec
3292            .is_some_and(|after| commit_realtime < after),
3293    }
3294}
3295
3296fn skip_by_commit_time(query: &ExplorerQuery, commit_realtime: u64) -> bool {
3297    match query.direction {
3298        Direction::Forward => query
3299            .after_realtime_usec
3300            .is_some_and(|after| commit_realtime < after),
3301        Direction::Backward => query.before_realtime_usec.is_some_and(|before| {
3302            commit_realtime > before.saturating_add(query.realtime_slack_usec)
3303        }),
3304    }
3305}
3306
3307fn new_histogram(field: &[u8], query: &ExplorerQuery) -> ExplorerHistogram {
3308    let (start, end) = histogram_bounds(query);
3309    let target_buckets = query.histogram_target_buckets.max(1);
3310    let mut width = histogram_bar_width_usec(start, end, target_buckets);
3311    let start = histogram_slot_baseline_usec(start, width);
3312    let mut end = histogram_slot_baseline_usec(end, width).saturating_add(width);
3313    let mut bucket_count = end
3314        .saturating_sub(start)
3315        .checked_div(width)
3316        .unwrap_or(0)
3317        .saturating_add(1) as usize;
3318    if bucket_count > 1001 {
3319        bucket_count = 1001;
3320        width = end
3321            .saturating_sub(start)
3322            .checked_div(1000)
3323            .unwrap_or(0)
3324            .max(1);
3325        end = start.saturating_add(width.saturating_mul(1000));
3326    }
3327    let mut buckets = Vec::with_capacity(bucket_count);
3328    for index in 0..bucket_count {
3329        let bucket_start = start.saturating_add(width.saturating_mul(index as u64));
3330        let bucket_end = if index + 1 == bucket_count {
3331            end.saturating_add(1)
3332        } else {
3333            bucket_start.saturating_add(width)
3334        };
3335        buckets.push(ExplorerHistogramBucket {
3336            start_realtime_usec: bucket_start,
3337            end_realtime_usec: bucket_end,
3338            values: HashMap::new(),
3339        });
3340    }
3341    ExplorerHistogram {
3342        field: field.to_vec(),
3343        buckets,
3344    }
3345}
3346
3347pub(crate) fn empty_histogram_for_query(field: &[u8], query: &ExplorerQuery) -> ExplorerHistogram {
3348    new_histogram(field, query)
3349}
3350
3351fn histogram_bar_width_usec(after: u64, before: u64, target_buckets: usize) -> u64 {
3352    const USEC_PER_SEC: u64 = 1_000_000;
3353    const VALID_DURATIONS_SECONDS: &[u64] = &[
3354        1, 2, 5, 10, 15, 30, 60, 120, 180, 300, 600, 900, 1800, 3600, 7200, 21600, 28800, 43200,
3355        86400, 172800, 259200, 432000, 604800, 1209600, 2592000,
3356    ];
3357    let duration = before.saturating_sub(after);
3358    for seconds in VALID_DURATIONS_SECONDS.iter().rev() {
3359        let width = seconds.saturating_mul(USEC_PER_SEC);
3360        if width != 0 && duration / width >= target_buckets as u64 {
3361            return width;
3362        }
3363    }
3364    USEC_PER_SEC
3365}
3366
3367fn histogram_slot_baseline_usec(value: u64, width: u64) -> u64 {
3368    value.saturating_sub(value % width.max(1))
3369}
3370
3371fn histogram_bounds(query: &ExplorerQuery) -> (u64, u64) {
3372    let start = query
3373        .histogram_after_realtime_usec
3374        .or(query.after_realtime_usec)
3375        .unwrap_or(0);
3376    let end = query
3377        .histogram_before_realtime_usec
3378        .or(query.before_realtime_usec)
3379        .unwrap_or_else(|| start.saturating_add(3_600_000_000));
3380    if end <= start {
3381        (start, start.saturating_add(1))
3382    } else {
3383        (start, end)
3384    }
3385}
3386
3387fn increment_counter_by(map: &mut HashMap<Vec<u8>, u64>, value: &[u8], delta: u64) {
3388    if let Some(count) = map.get_mut(value) {
3389        *count = count.saturating_add(delta);
3390    } else {
3391        map.insert(value.to_vec(), delta);
3392    }
3393}
3394
3395#[cfg(test)]
3396mod tests {
3397    use super::*;
3398    use journal_core::file::{JournalFileOptions, JournalWriter, MmapMut};
3399    use journal_core::repository::File as RepoFile;
3400    use tempfile::TempDir;
3401
3402    fn test_uuid(seed: u8) -> uuid::Uuid {
3403        uuid::Uuid::from_bytes([seed; 16])
3404    }
3405
3406    fn create_writer(
3407        path: &std::path::Path,
3408        compression: Option<(Compression, usize)>,
3409    ) -> (JournalFile<MmapMut>, JournalWriter) {
3410        if let Some(parent) = path.parent() {
3411            std::fs::create_dir_all(parent).expect("create journal parent");
3412        }
3413        let repo_file = RepoFile::from_path(path).expect("repo file");
3414        let mut options = JournalFileOptions::new(test_uuid(1), test_uuid(2), test_uuid(3));
3415        if let Some((compression, threshold)) = compression {
3416            options = options
3417                .with_compression(compression)
3418                .with_compress_threshold(threshold);
3419        }
3420        let mut file = JournalFile::<MmapMut>::create(&repo_file, options).expect("create journal");
3421        let writer = if let Some((compression, threshold)) = compression {
3422            JournalWriter::new_with_compression(&mut file, 1, test_uuid(4), compression, threshold)
3423                .expect("writer")
3424        } else {
3425            JournalWriter::new(&mut file, 1, test_uuid(4)).expect("writer")
3426        };
3427        (file, writer)
3428    }
3429
3430    fn write_entries(
3431        path: &std::path::Path,
3432        compression: Option<(Compression, usize)>,
3433        entries: &[(&[&[u8]], u64)],
3434    ) {
3435        let (mut file, mut writer) = create_writer(path, compression);
3436        for (payloads, realtime) in entries {
3437            writer
3438                .add_entry(&mut file, payloads, *realtime, *realtime)
3439                .expect("write entry");
3440        }
3441        file.sync().expect("sync journal");
3442    }
3443
3444    fn write_many_entries(path: &std::path::Path, count: usize) {
3445        let (mut file, mut writer) = create_writer(path, None);
3446        for index in 0..count {
3447            let message = format!("MESSAGE=row-{index}");
3448            let service = if index % 2 == 0 {
3449                b"SERVICE=even".as_slice()
3450            } else {
3451                b"SERVICE=odd".as_slice()
3452            };
3453            let payloads: [&[u8]; 2] = [message.as_bytes(), service];
3454            let realtime = 1_700_000_000_000_000u64.saturating_add(index as u64);
3455            writer
3456                .add_entry(&mut file, &payloads, realtime, realtime)
3457                .expect("write entry");
3458        }
3459        file.sync().expect("sync journal");
3460    }
3461
3462    #[test]
3463    fn explorer_control_reports_progress_during_large_scan() {
3464        let dir = TempDir::new().expect("tempdir");
3465        let path = dir.path().join("progress.journal");
3466        write_many_entries(&path, 9_000);
3467
3468        let mut reports = Vec::new();
3469        let mut progress = |progress: ExplorerProgress| {
3470            reports.push(progress.stats.rows_examined);
3471        };
3472        let mut control = ExplorerControl::new();
3473        control.set_progress_interval(Duration::ZERO);
3474        control.set_progress_callback(Some(&mut progress));
3475        let mut reader = FileReader::open(&path).expect("open reader");
3476        let query = ExplorerQuery {
3477            facets: vec![b"SERVICE".to_vec()],
3478            limit: 0,
3479            ..ExplorerQuery::default()
3480        };
3481
3482        let result = reader
3483            .explore_with_strategy_and_control(&query, ExplorerStrategy::Traversal, &mut control)
3484            .expect("explore");
3485
3486        assert_eq!(control.stop_reason(), None);
3487        assert_eq!(result.stats.rows_examined, 9_000);
3488        assert!(!reports.is_empty());
3489        assert!(reports.iter().any(|rows| *rows >= 8_191));
3490    }
3491
3492    #[test]
3493    fn explorer_control_cancels_inside_large_scan() {
3494        let dir = TempDir::new().expect("tempdir");
3495        let path = dir.path().join("cancel.journal");
3496        write_many_entries(&path, 9_000);
3497
3498        let is_cancelled = || true;
3499        let mut control = ExplorerControl::new();
3500        control.set_cancellation_callback(Some(&is_cancelled));
3501        let mut reader = FileReader::open(&path).expect("open reader");
3502        let query = ExplorerQuery {
3503            facets: vec![b"SERVICE".to_vec()],
3504            limit: 0,
3505            ..ExplorerQuery::default()
3506        };
3507
3508        let result = reader
3509            .explore_with_strategy_and_control(&query, ExplorerStrategy::Traversal, &mut control)
3510            .expect("explore");
3511
3512        assert_eq!(control.stop_reason(), Some(ExplorerStopReason::Cancelled));
3513        assert!(result.stats.rows_examined < 9_000);
3514    }
3515
3516    #[test]
3517    fn explorer_filters_with_or_values_and_and_fields() {
3518        let dir = TempDir::new().expect("tempdir");
3519        let path = dir.path().join("filter.journal");
3520        write_entries(
3521            &path,
3522            None,
3523            &[
3524                (&[b"SERVICE=a", b"PRIORITY=3"], 1_000),
3525                (&[b"SERVICE=b", b"PRIORITY=3"], 2_000),
3526                (&[b"SERVICE=b", b"PRIORITY=4"], 3_000),
3527            ],
3528        );
3529
3530        let mut reader = FileReader::open(&path).expect("open reader");
3531        let query = ExplorerQuery {
3532            filters: vec![
3533                ExplorerFilter::new(b"SERVICE".to_vec(), [b"a".to_vec(), b"b".to_vec()]),
3534                ExplorerFilter::new(b"PRIORITY".to_vec(), [b"3".to_vec()]),
3535            ],
3536            facets: vec![b"SERVICE".to_vec()],
3537            limit: 10,
3538            ..ExplorerQuery::default()
3539        };
3540
3541        let result = reader.explore(&query).expect("explore");
3542        assert_eq!(result.rows.len(), 2);
3543        let service = result
3544            .facets
3545            .get(b"SERVICE".as_slice())
3546            .expect("service facet");
3547        assert_eq!(service.get(b"a".as_slice()), Some(&1));
3548        assert_eq!(service.get(b"b".as_slice()), Some(&1));
3549        assert!(result.stats.data_cache_misses > 0);
3550    }
3551
3552    #[test]
3553    fn explorer_rejects_debug_row_traversal_column_collection() {
3554        let dir = TempDir::new().expect("tempdir");
3555        let path = dir.path().join("debug-column-collection.journal");
3556        write_entries(&path, None, &[(&[b"PRIORITY=3", b"MESSAGE=hello"], 1_000)]);
3557
3558        let query = ExplorerQuery {
3559            facets: vec![b"PRIORITY".to_vec()],
3560            debug_collect_column_fields_by_row_traversal: true,
3561            ..ExplorerQuery::default()
3562        };
3563
3564        let mut reader = FileReader::open(&path).expect("open reader");
3565        let err = reader
3566            .explore(&query)
3567            .expect_err("debug-only column collection is rejected");
3568        assert!(matches!(err, SdkError::Unsupported(_)));
3569        assert!(
3570            err.to_string()
3571                .contains("debug_collect_column_fields_by_row_traversal")
3572        );
3573
3574        let mut reader = FileReader::open(&path).expect("reopen reader");
3575        let err = reader
3576            .explore_with_strategy_cursor_rows(&query, ExplorerStrategy::Traversal)
3577            .expect_err("cursor-row explorer also rejects debug-only column collection");
3578        assert!(matches!(err, SdkError::Unsupported(_)));
3579    }
3580
3581    #[test]
3582    fn explorer_skips_irrelevant_compressed_data_for_facets() {
3583        let dir = TempDir::new().expect("tempdir");
3584        let path = dir.path().join("compressed.journal");
3585        let large_message = b"MESSAGE=abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz";
3586        write_entries(
3587            &path,
3588            Some((Compression::Zstd, 32)),
3589            &[(&[b"PRIORITY=3", large_message], 1_000)],
3590        );
3591
3592        let mut reader = FileReader::open(&path).expect("open reader");
3593        let query = ExplorerQuery {
3594            facets: vec![b"PRIORITY".to_vec()],
3595            limit: 0,
3596            ..ExplorerQuery::default()
3597        };
3598
3599        let result = reader.explore(&query).expect("explore");
3600        let priority = result
3601            .facets
3602            .get(b"PRIORITY".as_slice())
3603            .expect("priority facet");
3604        assert_eq!(priority.get(b"3".as_slice()), Some(&1));
3605        assert_eq!(result.stats.payloads_decompressed, 0);
3606        assert_eq!(result.stats.data_refs_seen, 1);
3607        assert_eq!(result.stats.early_stops, 1);
3608    }
3609
3610    #[test]
3611    fn explorer_reuses_classified_data_objects() {
3612        let dir = TempDir::new().expect("tempdir");
3613        let path = dir.path().join("reuse.journal");
3614        write_entries(
3615            &path,
3616            None,
3617            &[
3618                (&[b"PRIORITY=3"], 1_000),
3619                (&[b"PRIORITY=3"], 2_000),
3620                (&[b"PRIORITY=3"], 3_000),
3621            ],
3622        );
3623
3624        let mut reader = FileReader::open(&path).expect("open reader");
3625        let query = ExplorerQuery {
3626            facets: vec![b"PRIORITY".to_vec()],
3627            limit: 0,
3628            ..ExplorerQuery::default()
3629        };
3630
3631        let result = reader.explore(&query).expect("explore");
3632        let priority = result
3633            .facets
3634            .get(b"PRIORITY".as_slice())
3635            .expect("priority facet");
3636        assert_eq!(priority.get(b"3".as_slice()), Some(&3));
3637        assert!(result.stats.data_cache_hits >= 2);
3638    }
3639
3640    #[test]
3641    fn explorer_groups_facets_with_same_filter_set() {
3642        let dir = TempDir::new().expect("tempdir");
3643        let path = dir.path().join("grouped-facets.journal");
3644        write_entries(
3645            &path,
3646            None,
3647            &[
3648                (&[b"SERVICE=a", b"PRIORITY=3"], 1_000),
3649                (&[b"SERVICE=b", b"PRIORITY=4"], 2_000),
3650            ],
3651        );
3652
3653        let mut reader = FileReader::open(&path).expect("open reader");
3654        let query = ExplorerQuery {
3655            facets: vec![b"SERVICE".to_vec(), b"PRIORITY".to_vec()],
3656            limit: 0,
3657            ..ExplorerQuery::default()
3658        };
3659
3660        let result = reader.explore(&query).expect("explore");
3661        assert_eq!(result.stats.rows_examined, 2);
3662        assert_eq!(result.stats.facet_rows_matched, 2);
3663        assert_eq!(
3664            result
3665                .facets
3666                .get(b"SERVICE".as_slice())
3667                .and_then(|values| values.get(b"a".as_slice())),
3668            Some(&1)
3669        );
3670        assert_eq!(
3671            result
3672                .facets
3673                .get(b"PRIORITY".as_slice())
3674                .and_then(|values| values.get(b"4".as_slice())),
3675            Some(&1)
3676        );
3677    }
3678
3679    #[test]
3680    fn explorer_combines_rows_histogram_and_facets_in_one_pass() {
3681        let dir = TempDir::new().expect("tempdir");
3682        let path = dir.path().join("combined-pass.journal");
3683        write_entries(
3684            &path,
3685            None,
3686            &[
3687                (&[b"SERVICE=a", b"PRIORITY=3"], 1_000),
3688                (&[b"SERVICE=b", b"PRIORITY=4"], 2_000),
3689            ],
3690        );
3691
3692        let mut reader = FileReader::open(&path).expect("open reader");
3693        let query = ExplorerQuery {
3694            facets: vec![b"SERVICE".to_vec()],
3695            histogram: Some(b"PRIORITY".to_vec()),
3696            histogram_target_buckets: 2,
3697            limit: 2,
3698            ..ExplorerQuery::default()
3699        };
3700
3701        let result = reader.explore(&query).expect("explore");
3702        assert_eq!(result.rows.len(), 2);
3703        assert_eq!(result.stats.rows_examined, 2);
3704        assert_eq!(result.stats.rows_matched, 2);
3705        assert_eq!(result.stats.facet_rows_matched, 2);
3706        assert_eq!(
3707            result
3708                .facets
3709                .get(b"SERVICE".as_slice())
3710                .and_then(|values| values.get(b"a".as_slice())),
3711            Some(&1)
3712        );
3713        let histogram_total = result
3714            .histogram
3715            .as_ref()
3716            .expect("histogram")
3717            .buckets
3718            .iter()
3719            .flat_map(|bucket| bucket.values.values())
3720            .sum::<u64>();
3721        assert_eq!(histogram_total, 2);
3722    }
3723
3724    #[test]
3725    fn explorer_sampling_uses_actual_histogram_bucket_count() {
3726        let query = ExplorerQuery {
3727            after_realtime_usec: Some(1_733_494_460_000_000),
3728            before_realtime_usec: Some(1_735_656_412_000_000),
3729            histogram: Some(b"PRIORITY".to_vec()),
3730            histogram_target_buckets: 300,
3731            sampling: Some(ExplorerSampling {
3732                budget: 20_000,
3733                matched_files: 200,
3734                file_head_realtime_usec: 1_733_494_460_000_000,
3735                file_tail_realtime_usec: 1_735_656_412_000_000,
3736                file_head_seqnum: 1,
3737                file_tail_seqnum: 2,
3738                file_entries: 2,
3739            }),
3740            ..ExplorerQuery::default()
3741        };
3742
3743        let bucket_count = histogram_bucket_count_for_query(&query).expect("bucket count");
3744        let sampling =
3745            ExplorerSamplingState::for_query(&query, Some(bucket_count)).expect("sampling");
3746
3747        assert_eq!(bucket_count, 302);
3748        assert_eq!(sampling.per_slot_sampled.len(), bucket_count);
3749    }
3750
3751    #[test]
3752    fn explorer_sampling_seqnum_estimate_clamps_over_scanned_to_one() {
3753        let query = ExplorerQuery {
3754            after_realtime_usec: Some(1),
3755            before_realtime_usec: Some(100),
3756            direction: Direction::Forward,
3757            sampling: Some(ExplorerSampling {
3758                budget: 20,
3759                matched_files: 1,
3760                file_head_realtime_usec: 1,
3761                file_tail_realtime_usec: 100,
3762                file_head_seqnum: 1,
3763                file_tail_seqnum: 100,
3764                file_entries: 3,
3765            }),
3766            ..ExplorerQuery::default()
3767        };
3768        let mut sampling = ExplorerSamplingState::for_query(&query, None).expect("sampling");
3769        sampling.per_file_sampled = 10;
3770
3771        assert_eq!(sampling.estimate_remaining_rows_by_seqnum(5), Some(1));
3772    }
3773
3774    #[test]
3775    fn explorer_estimated_histogram_distribution_matches_netdata_integer_math() {
3776        let mut histogram = ExplorerHistogram {
3777            field: b"PRIORITY".to_vec(),
3778            buckets: vec![
3779                ExplorerHistogramBucket {
3780                    start_realtime_usec: 0,
3781                    end_realtime_usec: 10,
3782                    values: HashMap::new(),
3783                },
3784                ExplorerHistogramBucket {
3785                    start_realtime_usec: 10,
3786                    end_realtime_usec: 20,
3787                    values: HashMap::new(),
3788                },
3789                ExplorerHistogramBucket {
3790                    start_realtime_usec: 20,
3791                    end_realtime_usec: 30,
3792                    values: HashMap::new(),
3793                },
3794            ],
3795        };
3796        let mut stats = ExplorerStats::default();
3797
3798        add_estimated_histogram_range(Some(&mut histogram), 0, 30, 10, &mut stats);
3799
3800        let counts = histogram
3801            .buckets
3802            .iter()
3803            .map(|bucket| {
3804                bucket
3805                    .values
3806                    .get(EXPLORER_ESTIMATED_VALUE)
3807                    .copied()
3808                    .unwrap_or_default()
3809            })
3810            .collect::<Vec<_>>();
3811        assert_eq!(counts, vec![3, 3, 3]);
3812        assert_eq!(counts.iter().sum::<u64>(), 9);
3813    }
3814
3815    #[test]
3816    fn explorer_filters_then_combines_outputs_in_one_candidate_pass() {
3817        let dir = TempDir::new().expect("tempdir");
3818        let path = dir.path().join("filtered-combined-pass.journal");
3819        write_entries(
3820            &path,
3821            None,
3822            &[
3823                (&[b"SERVICE=a", b"PRIORITY=3"], 1_000),
3824                (&[b"SERVICE=b", b"PRIORITY=4"], 2_000),
3825                (&[b"SERVICE=c", b"PRIORITY=3"], 3_000),
3826            ],
3827        );
3828
3829        let mut reader = FileReader::open(&path).expect("open reader");
3830        let query = ExplorerQuery {
3831            filters: vec![ExplorerFilter::new(b"PRIORITY".to_vec(), [b"3".to_vec()])],
3832            facets: vec![b"SERVICE".to_vec()],
3833            histogram: Some(b"SERVICE".to_vec()),
3834            histogram_target_buckets: 2,
3835            limit: 10,
3836            ..ExplorerQuery::default()
3837        };
3838
3839        let result = reader.explore(&query).expect("explore");
3840        assert_eq!(result.rows.len(), 2);
3841        assert_eq!(result.stats.rows_examined, 2);
3842        assert_eq!(result.stats.rows_matched, 2);
3843        assert_eq!(result.stats.facet_rows_matched, 2);
3844        let service = result
3845            .facets
3846            .get(b"SERVICE".as_slice())
3847            .expect("service facet");
3848        assert_eq!(service.get(b"a".as_slice()), Some(&1));
3849        assert_eq!(service.get(b"c".as_slice()), Some(&1));
3850        assert_eq!(service.get(b"b".as_slice()), None);
3851    }
3852
3853    #[test]
3854    fn explorer_cursor_rows_defer_payload_expansion() {
3855        let dir = TempDir::new().expect("tempdir");
3856        let path = dir.path().join("cursor-only-row.journal");
3857        write_entries(
3858            &path,
3859            None,
3860            &[(&[b"SERVICE=a", b"PRIORITY=3", b"MESSAGE=hello"], 1_000)],
3861        );
3862
3863        let query = ExplorerQuery {
3864            limit: 1,
3865            ..ExplorerQuery::default()
3866        };
3867        let mut reader = FileReader::open(&path).expect("open reader");
3868        let result = reader
3869            .explore_with_strategy_cursor_rows(&query, ExplorerStrategy::Traversal)
3870            .expect("explore cursor rows");
3871
3872        assert_eq!(result.rows.len(), 1);
3873        assert!(result.rows[0].payloads.is_empty());
3874        assert_eq!(result.stats.returned_row_expansions, 0);
3875
3876        let cursor = result.rows[0].cursor.clone();
3877        let mut reader = FileReader::open(&path).expect("reopen reader");
3878        reader.seek_cursor(&cursor).expect("seek cursor");
3879        assert!(reader.test_cursor(&cursor).expect("test cursor"));
3880
3881        let mut payloads = Vec::new();
3882        reader
3883            .collect_entry_payloads(&mut payloads)
3884            .expect("collect payloads");
3885        assert!(payloads.iter().any(|payload| payload == b"MESSAGE=hello"));
3886    }
3887
3888    #[test]
3889    fn explorer_same_field_filter_exclusion_counts_filtered_out_facet_values() {
3890        let dir = TempDir::new().expect("tempdir");
3891        let path = dir.path().join("same-field-filter-facet.journal");
3892        write_entries(
3893            &path,
3894            None,
3895            &[
3896                (&[b"SERVICE=a", b"PRIORITY=3"], 1_000),
3897                (&[b"SERVICE=b", b"PRIORITY=3"], 2_000),
3898                (&[b"SERVICE=a", b"PRIORITY=4"], 3_000),
3899            ],
3900        );
3901
3902        let mut reader = FileReader::open(&path).expect("open reader");
3903        let query = ExplorerQuery {
3904            filters: vec![
3905                ExplorerFilter::new(b"SERVICE".to_vec(), [b"a".to_vec()]),
3906                ExplorerFilter::new(b"PRIORITY".to_vec(), [b"3".to_vec()]),
3907            ],
3908            facets: vec![b"SERVICE".to_vec(), b"PRIORITY".to_vec()],
3909            limit: 0,
3910            ..ExplorerQuery::default()
3911        };
3912
3913        let result = reader.explore(&query).expect("explore");
3914        let service = result
3915            .facets
3916            .get(b"SERVICE".as_slice())
3917            .expect("service facet");
3918        assert_eq!(service.get(b"a".as_slice()), Some(&1));
3919        assert_eq!(service.get(b"b".as_slice()), Some(&1));
3920
3921        let priority = result
3922            .facets
3923            .get(b"PRIORITY".as_slice())
3924            .expect("priority facet");
3925        assert_eq!(priority.get(b"3".as_slice()), Some(&1));
3926        assert_eq!(priority.get(b"4".as_slice()), Some(&1));
3927    }
3928
3929    #[test]
3930    fn explorer_index_strategy_matches_traversal_for_all_values() {
3931        let dir = TempDir::new().expect("tempdir");
3932        let path = dir.path().join("indexed-all-values.journal");
3933        write_entries(
3934            &path,
3935            None,
3936            &[
3937                (&[b"SERVICE=a", b"PRIORITY=3", b"TAG=x"], 1_000),
3938                (&[b"SERVICE=b", b"PRIORITY=3", b"TAG=x"], 2_000),
3939                (&[b"SERVICE=a", b"PRIORITY=4", b"TAG=y", b"TAG=z"], 3_000),
3940                (&[b"PRIORITY=3"], 4_000),
3941            ],
3942        );
3943
3944        let query = ExplorerQuery {
3945            after_realtime_usec: Some(0),
3946            before_realtime_usec: Some(5_000),
3947            filters: vec![ExplorerFilter::new(b"PRIORITY".to_vec(), [b"3".to_vec()])],
3948            facets: vec![b"SERVICE".to_vec(), b"TAG".to_vec()],
3949            histogram: Some(b"SERVICE".to_vec()),
3950            histogram_target_buckets: 2,
3951            limit: 2,
3952            field_mode: ExplorerFieldMode::AllValues,
3953            use_source_realtime: false,
3954            ..ExplorerQuery::default()
3955        };
3956
3957        let mut reader = FileReader::open(&path).expect("open reader");
3958        let result = reader
3959            .explore_with_strategy(&query, ExplorerStrategy::Compare)
3960            .expect("compare");
3961
3962        let comparison = result.comparison.as_ref().expect("comparison diagnostics");
3963        assert_eq!(comparison.index_stats, result.stats);
3964        assert_eq!(comparison.traversal_stats.rows_returned, 2);
3965        assert_eq!(comparison.index_stats.rows_returned, 2);
3966
3967        assert_eq!(result.rows.len(), 2);
3968        let service = result
3969            .facets
3970            .get(b"SERVICE".as_slice())
3971            .expect("service facet");
3972        assert_eq!(service.get(b"a".as_slice()), Some(&1));
3973        assert_eq!(service.get(b"b".as_slice()), Some(&1));
3974        assert_eq!(service.get(UNSET_VALUE), Some(&1));
3975        let histogram = result.histogram.as_ref().expect("histogram");
3976        assert_eq!(histogram.buckets.len(), 2);
3977        assert_eq!(histogram.buckets[0].values.get(b"a".as_slice()), Some(&1));
3978        assert_eq!(histogram.buckets[0].values.get(b"b".as_slice()), Some(&1));
3979        assert_eq!(histogram.buckets[0].values.get(UNSET_VALUE), Some(&1));
3980    }
3981
3982    #[test]
3983    fn explorer_index_strategy_preserves_same_field_filter_exclusion() {
3984        let dir = TempDir::new().expect("tempdir");
3985        let path = dir.path().join("indexed-same-field-filter.journal");
3986        write_entries(
3987            &path,
3988            None,
3989            &[
3990                (&[b"SERVICE=a", b"PRIORITY=3"], 1_000),
3991                (&[b"SERVICE=b", b"PRIORITY=3"], 2_000),
3992                (&[b"SERVICE=a", b"PRIORITY=4"], 3_000),
3993            ],
3994        );
3995
3996        let query = ExplorerQuery {
3997            filters: vec![
3998                ExplorerFilter::new(b"SERVICE".to_vec(), [b"a".to_vec()]),
3999                ExplorerFilter::new(b"PRIORITY".to_vec(), [b"3".to_vec()]),
4000            ],
4001            facets: vec![b"SERVICE".to_vec(), b"PRIORITY".to_vec()],
4002            field_mode: ExplorerFieldMode::AllValues,
4003            use_source_realtime: false,
4004            ..ExplorerQuery::default()
4005        };
4006
4007        let mut reader = FileReader::open(&path).expect("open reader");
4008        let result = reader
4009            .explore_with_strategy(&query, ExplorerStrategy::Compare)
4010            .expect("compare");
4011        let service = result
4012            .facets
4013            .get(b"SERVICE".as_slice())
4014            .expect("service facet");
4015        assert_eq!(service.get(b"a".as_slice()), Some(&1));
4016        assert_eq!(service.get(b"b".as_slice()), Some(&1));
4017    }
4018
4019    #[test]
4020    fn explorer_index_strategy_rejects_first_value_semantics() {
4021        let dir = TempDir::new().expect("tempdir");
4022        let path = dir.path().join("indexed-first-value.journal");
4023        write_entries(&path, None, &[(&[b"TAG=one", b"TAG=two"], 1_000)]);
4024
4025        let mut reader = FileReader::open(&path).expect("open reader");
4026        let err = reader
4027            .explore_with_strategy(
4028                &ExplorerQuery {
4029                    facets: vec![b"TAG".to_vec()],
4030                    field_mode: ExplorerFieldMode::FirstValue,
4031                    ..ExplorerQuery::default()
4032                },
4033                ExplorerStrategy::Index,
4034            )
4035            .expect_err("first-value index strategy should be rejected");
4036
4037        assert!(matches!(err, SdkError::Unsupported(_)));
4038    }
4039
4040    #[test]
4041    fn explorer_first_value_counts_one_value_per_selected_field() {
4042        let dir = TempDir::new().expect("tempdir");
4043        let path = dir.path().join("first-value.journal");
4044        write_entries(
4045            &path,
4046            None,
4047            &[(&[b"TAG=one", b"TAG=two", b"SERVICE=a"], 1_000)],
4048        );
4049
4050        let mut all_values_reader = FileReader::open(&path).expect("open all-values reader");
4051        let all_values = all_values_reader
4052            .explore(&ExplorerQuery {
4053                facets: vec![b"TAG".to_vec()],
4054                limit: 0,
4055                field_mode: ExplorerFieldMode::AllValues,
4056                ..ExplorerQuery::default()
4057            })
4058            .expect("all-values explore");
4059        let all_tag = all_values
4060            .facets
4061            .get(b"TAG".as_slice())
4062            .expect("all-values tag facet");
4063        assert_eq!(all_tag.values().sum::<u64>(), 2);
4064        assert_eq!(all_tag.len(), 2);
4065
4066        let mut first_value_reader = FileReader::open(&path).expect("open first-value reader");
4067        let first_value = first_value_reader
4068            .explore(&ExplorerQuery {
4069                facets: vec![b"TAG".to_vec()],
4070                limit: 0,
4071                field_mode: ExplorerFieldMode::FirstValue,
4072                ..ExplorerQuery::default()
4073            })
4074            .expect("first-value explore");
4075        let first_tag = first_value
4076            .facets
4077            .get(b"TAG".as_slice())
4078            .expect("first-value tag facet");
4079        assert_eq!(first_tag.values().sum::<u64>(), 1);
4080        assert_eq!(first_tag.len(), 1);
4081        assert_eq!(first_value.stats.early_stops, 1);
4082    }
4083
4084    #[test]
4085    fn explorer_first_value_does_not_double_count_duplicate_facets_or_histogram() {
4086        let dir = TempDir::new().expect("tempdir");
4087        let path = dir.path().join("first-value-no-double-count.journal");
4088        write_entries(
4089            &path,
4090            None,
4091            &[(
4092                &[
4093                    b"_SOURCE_REALTIME_TIMESTAMP=1000",
4094                    b"TAG=one",
4095                    b"TAG=two",
4096                    b"MESSAGE=after-tag",
4097                ],
4098                1_000,
4099            )],
4100        );
4101
4102        let mut reader = FileReader::open(&path).expect("open reader");
4103        let result = reader
4104            .explore(&ExplorerQuery {
4105                facets: vec![b"TAG".to_vec()],
4106                histogram: Some(b"TAG".to_vec()),
4107                histogram_target_buckets: 1,
4108                limit: 0,
4109                ..ExplorerQuery::default()
4110            })
4111            .expect("explore");
4112
4113        assert_eq!(
4114            result
4115                .facets
4116                .get(b"TAG".as_slice())
4117                .expect("tag facet")
4118                .values()
4119                .sum::<u64>(),
4120            1
4121        );
4122        assert_eq!(
4123            result
4124                .histogram
4125                .as_ref()
4126                .expect("histogram")
4127                .buckets
4128                .iter()
4129                .flat_map(|bucket| bucket.values.values())
4130                .sum::<u64>(),
4131            1
4132        );
4133
4134        let mut all_values_reader = FileReader::open(&path).expect("open all-values reader");
4135        let all_values = all_values_reader
4136            .explore(&ExplorerQuery {
4137                facets: vec![b"TAG".to_vec()],
4138                histogram: Some(b"TAG".to_vec()),
4139                histogram_target_buckets: 1,
4140                limit: 0,
4141                field_mode: ExplorerFieldMode::AllValues,
4142                ..ExplorerQuery::default()
4143            })
4144            .expect("all-values explore");
4145
4146        assert_eq!(
4147            all_values
4148                .facets
4149                .get(b"TAG".as_slice())
4150                .expect("tag facet")
4151                .values()
4152                .sum::<u64>(),
4153            2
4154        );
4155        assert_eq!(
4156            all_values
4157                .histogram
4158                .as_ref()
4159                .expect("histogram")
4160                .buckets
4161                .iter()
4162                .flat_map(|bucket| bucket.values.values())
4163                .sum::<u64>(),
4164            2
4165        );
4166    }
4167
4168    #[test]
4169    fn explorer_first_value_tracks_required_field_identities() {
4170        let dir = TempDir::new().expect("tempdir");
4171        let path = dir.path().join("first-value-identities.journal");
4172        write_entries(
4173            &path,
4174            None,
4175            &[(&[b"TAG=one", b"TAG=two", b"SERVICE=a"], 1_000)],
4176        );
4177
4178        let mut reader = FileReader::open(&path).expect("open reader");
4179        let result = reader
4180            .explore(&ExplorerQuery {
4181                facets: vec![b"TAG".to_vec(), b"SERVICE".to_vec()],
4182                limit: 0,
4183                field_mode: ExplorerFieldMode::FirstValue,
4184                ..ExplorerQuery::default()
4185            })
4186            .expect("explore");
4187
4188        assert_eq!(
4189            result
4190                .facets
4191                .get(b"TAG".as_slice())
4192                .expect("tag facet")
4193                .values()
4194                .sum::<u64>(),
4195            1
4196        );
4197        assert_eq!(
4198            result
4199                .facets
4200                .get(b"SERVICE".as_slice())
4201                .and_then(|values| values.get(b"a".as_slice())),
4202            Some(&1)
4203        );
4204        assert_eq!(result.stats.early_stops, 1);
4205    }
4206
4207    #[test]
4208    fn explorer_rejects_duplicate_facet_fields() {
4209        let dir = TempDir::new().expect("tempdir");
4210        let path = dir.path().join("duplicate-facets.journal");
4211        write_entries(&path, None, &[(&[b"SERVICE=a"], 1_000)]);
4212
4213        let mut reader = FileReader::open(&path).expect("open reader");
4214        let err = reader
4215            .explore(&ExplorerQuery {
4216                facets: vec![b"SERVICE".to_vec(), b"SERVICE".to_vec()],
4217                limit: 0,
4218                ..ExplorerQuery::default()
4219            })
4220            .expect_err("duplicate facets rejected");
4221
4222        assert!(err.to_string().contains("must not be duplicated"));
4223    }
4224
4225    #[test]
4226    fn explorer_empty_result_keeps_requested_facet_with_no_values() {
4227        let dir = TempDir::new().expect("tempdir");
4228        let path = dir.path().join("empty-result.journal");
4229        write_entries(&path, None, &[(&[b"SERVICE=a", b"PRIORITY=3"], 1_000)]);
4230
4231        let mut reader = FileReader::open(&path).expect("open reader");
4232        let result = reader
4233            .explore(&ExplorerQuery {
4234                after_realtime_usec: Some(10_000),
4235                before_realtime_usec: Some(20_000),
4236                facets: vec![b"SERVICE".to_vec()],
4237                limit: 10,
4238                realtime_slack_usec: 0,
4239                ..ExplorerQuery::default()
4240            })
4241            .expect("explore");
4242
4243        assert!(result.rows.is_empty());
4244        assert_eq!(result.stats.rows_matched, 0);
4245        assert!(
4246            result
4247                .facets
4248                .get(b"SERVICE".as_slice())
4249                .expect("service facet")
4250                .is_empty()
4251        );
4252    }
4253
4254    #[test]
4255    fn explorer_facet_time_bounds_do_not_count_slack_rows_without_source_realtime() {
4256        let dir = TempDir::new().expect("tempdir");
4257        let path = dir.path().join("facet-time-bound.journal");
4258        write_entries(
4259            &path,
4260            None,
4261            &[
4262                (&[b"SERVICE=before"], 340_000_000),
4263                (&[b"SERVICE=inside"], 360_000_000),
4264                (&[b"SERVICE=after"], 400_000_000),
4265            ],
4266        );
4267
4268        let mut reader = FileReader::open(&path).expect("open reader");
4269        let result = reader
4270            .explore(&ExplorerQuery {
4271                after_realtime_usec: Some(350_000_000),
4272                before_realtime_usec: Some(370_000_000),
4273                facets: vec![b"SERVICE".to_vec()],
4274                limit: 0,
4275                realtime_slack_usec: 20_000_000,
4276                use_source_realtime: false,
4277                ..ExplorerQuery::default()
4278            })
4279            .expect("explore");
4280
4281        let service = result
4282            .facets
4283            .get(b"SERVICE".as_slice())
4284            .expect("service facet");
4285        assert_eq!(service.get(b"inside".as_slice()), Some(&1));
4286        assert_eq!(service.get(b"before".as_slice()), None);
4287        assert_eq!(service.get(b"after".as_slice()), None);
4288        assert_eq!(result.stats.facet_rows_matched, 1);
4289    }
4290
4291    #[test]
4292    fn explorer_fts_disables_first_value_early_stop() {
4293        let dir = TempDir::new().expect("tempdir");
4294        let path = dir.path().join("fts-no-early-stop.journal");
4295        write_entries(&path, None, &[(&[b"TAG=one", b"MESSAGE=needle"], 1_000)]);
4296
4297        let mut reader = FileReader::open(&path).expect("open reader");
4298        let result = reader
4299            .explore(&ExplorerQuery {
4300                facets: vec![b"TAG".to_vec()],
4301                fts_patterns: vec![b"needle".to_vec()],
4302                limit: 0,
4303                ..ExplorerQuery::default()
4304            })
4305            .expect("explore");
4306
4307        assert_eq!(result.stats.early_stops, 0);
4308        assert_eq!(result.stats.data_refs_seen, 2);
4309        assert_eq!(
4310            result
4311                .facets
4312                .get(b"TAG".as_slice())
4313                .and_then(|values| values.get(b"one".as_slice())),
4314            Some(&1)
4315        );
4316    }
4317
4318    #[test]
4319    fn explorer_fts_or_terms_and_negative_terms_filter_rows() {
4320        let dir = TempDir::new().expect("tempdir");
4321        let path = dir.path().join("fts-negative.journal");
4322        write_entries(
4323            &path,
4324            None,
4325            &[
4326                (&[b"TAG=alpha", b"MESSAGE=alpha keep"], 1_000),
4327                (&[b"TAG=beta", b"MESSAGE=beta keep"], 2_000),
4328                (&[b"TAG=debug", b"MESSAGE=alpha debug"], 3_000),
4329                (&[b"TAG=other", b"MESSAGE=other"], 4_000),
4330                (&[b"TAG=wild", b"MESSAGE=start middle end"], 5_000),
4331            ],
4332        );
4333
4334        let mut reader = FileReader::open(&path).expect("open reader");
4335        let result = reader
4336            .explore(&ExplorerQuery {
4337                facets: vec![b"TAG".to_vec()],
4338                fts_terms: vec![
4339                    ExplorerFtsPattern::substring(b"alpha".to_vec(), false),
4340                    ExplorerFtsPattern::substring(b"beta".to_vec(), false),
4341                    ExplorerFtsPattern::substring(b"debug".to_vec(), true),
4342                    ExplorerFtsPattern::substring(b"start*end".to_vec(), false),
4343                ],
4344                limit: 10,
4345                ..ExplorerQuery::default()
4346            })
4347            .expect("explore");
4348
4349        let tag = result.facets.get(b"TAG".as_slice()).expect("TAG facet");
4350        assert_eq!(result.rows.len(), 3);
4351        assert_eq!(tag.get(b"alpha".as_slice()), Some(&1));
4352        assert_eq!(tag.get(b"beta".as_slice()), Some(&1));
4353        assert_eq!(tag.get(b"wild".as_slice()), Some(&1));
4354        assert_eq!(tag.get(b"debug".as_slice()), None);
4355        assert_eq!(tag.get(b"other".as_slice()), None);
4356    }
4357
4358    #[test]
4359    fn explorer_auto_anchor_scans_backward_from_tail() {
4360        let dir = TempDir::new().expect("tempdir");
4361        let path = dir.path().join("backward.journal");
4362        write_entries(
4363            &path,
4364            None,
4365            &[
4366                (&[b"SERVICE=a", b"PRIORITY=3"], 1_000),
4367                (&[b"SERVICE=b", b"PRIORITY=4"], 2_000),
4368            ],
4369        );
4370
4371        let mut reader = FileReader::open(&path).expect("open reader");
4372        let query = ExplorerQuery {
4373            direction: Direction::Backward,
4374            limit: 2,
4375            ..ExplorerQuery::default()
4376        };
4377
4378        let result = reader.explore(&query).expect("explore");
4379        assert_eq!(result.rows.len(), 2);
4380        assert_eq!(result.rows[0].realtime_usec, 2_000);
4381        assert_eq!(result.rows[1].realtime_usec, 1_000);
4382    }
4383
4384    #[test]
4385    fn explorer_backward_time_bound_stops_after_slack_window() {
4386        let dir = TempDir::new().expect("tempdir");
4387        let path = dir.path().join("backward-time-bound.journal");
4388        write_entries(
4389            &path,
4390            None,
4391            &[
4392                (&[b"SERVICE=a"], 100_000_000),
4393                (&[b"SERVICE=b"], 200_000_000),
4394                (&[b"SERVICE=c"], 300_000_000),
4395                (&[b"SERVICE=d"], 400_000_000),
4396                (&[b"SERVICE=e"], 500_000_000),
4397            ],
4398        );
4399
4400        let mut reader = FileReader::open(&path).expect("open reader");
4401        let query = ExplorerQuery {
4402            after_realtime_usec: Some(350_000_000),
4403            direction: Direction::Backward,
4404            limit: 10,
4405            realtime_slack_usec: 10_000_000,
4406            ..ExplorerQuery::default()
4407        };
4408
4409        let result = reader.explore(&query).expect("explore");
4410        assert_eq!(result.rows.len(), 2);
4411        assert_eq!(result.rows[0].realtime_usec, 500_000_000);
4412        assert_eq!(result.rows[1].realtime_usec, 400_000_000);
4413        assert_eq!(result.stats.rows_examined, 2);
4414    }
4415
4416    #[test]
4417    fn explorer_histogram_and_fts_are_opt_in() {
4418        let dir = TempDir::new().expect("tempdir");
4419        let path = dir.path().join("histogram.journal");
4420        write_entries(
4421            &path,
4422            None,
4423            &[
4424                (&[b"MESSAGE=alpha", b"PRIORITY=3"], 1_000),
4425                (&[b"MESSAGE=beta", b"PRIORITY=4"], 2_000),
4426            ],
4427        );
4428
4429        let mut reader = FileReader::open(&path).expect("open reader");
4430        let query = ExplorerQuery {
4431            after_realtime_usec: Some(0),
4432            before_realtime_usec: Some(3_000),
4433            histogram: Some(b"PRIORITY".to_vec()),
4434            histogram_target_buckets: 2,
4435            fts_patterns: vec![b"alp".to_vec()],
4436            limit: 10,
4437            ..ExplorerQuery::default()
4438        };
4439
4440        let result = reader.explore(&query).expect("explore");
4441        assert_eq!(result.rows.len(), 1);
4442        assert!(result.stats.fts_scans > 0);
4443        assert_eq!(
4444            result
4445                .histogram
4446                .as_ref()
4447                .expect("histogram")
4448                .buckets
4449                .iter()
4450                .flat_map(|bucket| bucket.values.values())
4451                .sum::<u64>(),
4452            1
4453        );
4454    }
4455
4456    #[test]
4457    fn explorer_first_value_stops_after_same_data_satisfies_multiple_roles() {
4458        let dir = TempDir::new().expect("tempdir");
4459        let path = dir.path().join("same-data-multiple-roles.journal");
4460        write_entries(
4461            &path,
4462            None,
4463            &[(
4464                &[b"_SOURCE_REALTIME_TIMESTAMP=1000", b"MESSAGE=after-source"],
4465                1_000,
4466            )],
4467        );
4468
4469        let mut reader = FileReader::open(&path).expect("open reader");
4470        let result = reader
4471            .explore(&ExplorerQuery {
4472                histogram: Some(SOURCE_REALTIME_FIELD.to_vec()),
4473                histogram_target_buckets: 1,
4474                limit: 0,
4475                field_mode: ExplorerFieldMode::FirstValue,
4476                ..ExplorerQuery::default()
4477            })
4478            .expect("explore");
4479
4480        assert_eq!(result.stats.histogram_updates, 1);
4481        assert_eq!(result.stats.early_stops, 1);
4482        assert_eq!(
4483            result
4484                .histogram
4485                .as_ref()
4486                .expect("histogram")
4487                .buckets
4488                .iter()
4489                .flat_map(|bucket| bucket.values.values())
4490                .sum::<u64>(),
4491            1
4492        );
4493    }
4494}