flow/batching_mode/
eval_schedule.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Helpers for stable `EVAL INTERVAL` scheduled times.
16
17pub use common_meta::key::flow::flow_info::{FlowMissedTickPolicy, FlowScheduleConfig};
18use snafu::ensure;
19
20use crate::error::{InvalidQuerySnafu, Result, UnexpectedSnafu};
21
22/// Schedule for an `EVAL INTERVAL` flow.
23#[derive(Debug, Clone, PartialEq)]
24pub struct EvalSchedule {
25    /// Interval between scheduled times in seconds.
26    pub interval_secs: i64,
27    /// Anchor timestamp as seconds since Unix epoch.
28    pub anchor_secs: i64,
29    /// First scheduled time as seconds since Unix epoch.
30    pub start_secs: i64,
31    /// Policy for handling missed scheduled times.
32    pub missed_tick_policy: FlowMissedTickPolicy,
33    /// Maximum number of due scheduled times to catch up.
34    pub max_runs: u32,
35    /// Maximum age of a due scheduled time to keep for catch-up.
36    pub max_lag_secs: i64,
37}
38
39impl EvalSchedule {
40    pub fn from_config(
41        eval_interval_secs: Option<i64>,
42        config: Option<&FlowScheduleConfig>,
43    ) -> Result<Option<Self>> {
44        let Some(interval_secs) = eval_interval_secs else {
45            return Ok(None);
46        };
47        ensure!(
48            interval_secs > 0,
49            InvalidQuerySnafu {
50                reason: format!(
51                    "Invalid eval_interval_secs: must be positive, got {interval_secs}"
52                )
53            }
54        );
55
56        Ok(Some(match config {
57            Some(c) => {
58                ensure!(
59                    c.catchup_max_runs > 0,
60                    InvalidQuerySnafu {
61                        reason:
62                            "Invalid FlowScheduleConfig.catchup_max_runs: must be positive, got 0"
63                                .to_string()
64                    }
65                );
66                ensure!(
67                    c.catchup_max_lag_secs > 0,
68                    InvalidQuerySnafu {
69                        reason: format!(
70                            "Invalid FlowScheduleConfig.catchup_max_lag_secs: must be positive, got {}",
71                            c.catchup_max_lag_secs
72                        )
73                    }
74                );
75                // The anchor defines the epoch phase `anchor + k * interval`; it
76                // must be a valid offset within one interval.
77                ensure!(
78                    c.anchor_secs >= 0 && c.anchor_secs < interval_secs,
79                    InvalidQuerySnafu {
80                        reason: format!(
81                            "Invalid FlowScheduleConfig.anchor_secs: must be in [0, {interval_secs}), got {}",
82                            c.anchor_secs
83                        )
84                    }
85                );
86                // The start must be phase-consistent with the anchor (on an
87                // `anchor + k * interval` boundary) and not before the anchor.
88                ensure!(
89                    c.start_secs >= c.anchor_secs
90                        && (c.start_secs - c.anchor_secs) % interval_secs == 0,
91                    InvalidQuerySnafu {
92                        reason: format!(
93                            "Invalid FlowScheduleConfig.start_secs: must be on an anchor + k * interval boundary and >= anchor, got start={}, anchor={}, interval={}",
94                            c.start_secs, c.anchor_secs, interval_secs
95                        )
96                    }
97                );
98
99                Self {
100                    interval_secs,
101                    anchor_secs: c.anchor_secs,
102                    start_secs: c.start_secs,
103                    missed_tick_policy: c.missed_tick_policy,
104                    max_runs: c.catchup_max_runs,
105                    max_lag_secs: c.catchup_max_lag_secs,
106                }
107            }
108            None => {
109                let c = FlowScheduleConfig::default_with_start(0, interval_secs);
110                Self {
111                    interval_secs,
112                    anchor_secs: c.anchor_secs,
113                    start_secs: c.start_secs,
114                    missed_tick_policy: c.missed_tick_policy,
115                    max_runs: c.catchup_max_runs,
116                    max_lag_secs: c.catchup_max_lag_secs,
117                }
118            }
119        }))
120    }
121
122    /// Returns the next scheduled time strictly after `cursor_secs`, on the
123    /// `anchor + k * interval` lattice.
124    ///
125    /// Fallible: a non-positive interval or a next boundary that does not fit
126    /// in `i64` yields an explicit error instead of a saturated non-phase
127    /// value such as `i64::MAX`.
128    pub fn next_scheduled_time_after(&self, cursor_secs: i64) -> Result<i64> {
129        next_in_sequence(cursor_secs, self.start_secs, self.interval_secs)
130    }
131}
132
133/// The smallest `start + k * interval` value that is strictly after `cursor`
134/// (`start` itself lies on the `anchor + k * interval` lattice, so every
135/// result is phase-consistent with the anchor). All arithmetic happens in
136/// `i128`: `cursor - start` cannot overflow and the result is either exactly
137/// on the lattice or an explicit error.
138fn next_in_sequence(cursor: i64, start: i64, interval: i64) -> Result<i64> {
139    ensure!(
140        interval > 0,
141        InvalidQuerySnafu {
142            reason: format!("Invalid eval interval: must be positive, got {interval}")
143        }
144    );
145    let interval = i128::from(interval);
146    let start = i128::from(start);
147    let cursor = i128::from(cursor);
148
149    let next = if cursor < start {
150        start
151    } else {
152        let k = (cursor - start) / interval;
153        start + (k + 1) * interval
154    };
155
156    i64::try_from(next).map_err(|_| {
157        UnexpectedSnafu {
158            reason: format!(
159                "Cannot advance the eval schedule past cursor {cursor}: the next scheduled time {next} does not fit in i64 (start={start}, interval={interval})"
160            ),
161        }
162        .build()
163    })
164}
165
166fn first_due_in_sequence(cursor: i64, start: i64, interval: i64) -> Result<i64> {
167    if cursor < start {
168        Ok(start)
169    } else {
170        next_in_sequence(cursor, start, interval)
171    }
172}
173
174/// Scheduled times selected for execution in one scheduler pass.
175///
176/// A scheduled time is the logical evaluation timestamp for one flow run. When
177/// executing a timestamp from `scheduled_times_secs`, SQL/TQL `now()` is bound
178/// to that timestamp instead of the wall-clock execution time.
179#[derive(Debug, Clone, PartialEq)]
180pub struct DueScheduledTimes {
181    /// Scheduled times to execute, ordered oldest to newest.
182    pub scheduled_times_secs: Vec<i64>,
183    /// Number of due scheduled times skipped by lag or max-runs limits.
184    pub skipped: u64,
185}
186
187/// Select due scheduled times `<= wall_now_secs` without materializing all missed ticks.
188///
189/// Fallible: a non-positive interval or a scheduled time that does not fit in
190/// `i64` yields an explicit error instead of silently producing saturated
191/// non-phase timestamps.
192pub fn select_due_scheduled_times(
193    schedule: &EvalSchedule,
194    cursor_secs: i64,
195    wall_now_secs: i64,
196) -> Result<DueScheduledTimes> {
197    let interval = schedule.interval_secs;
198    ensure!(
199        interval > 0,
200        InvalidQuerySnafu {
201            reason: format!("Invalid eval interval: must be positive, got {interval}")
202        }
203    );
204
205    let first_due = first_due_in_sequence(cursor_secs, schedule.start_secs, interval)?;
206    if first_due > wall_now_secs {
207        return Ok(DueScheduledTimes {
208            scheduled_times_secs: vec![],
209            skipped: 0,
210        });
211    }
212
213    // Count and select due scheduled times in i128 so every value stays
214    // exactly on the `anchor + k * interval` lattice; a value beyond `i64` is
215    // an explicit error, never a saturated non-phase timestamp.
216    let first_due = i128::from(first_due);
217    let wall_now = i128::from(wall_now_secs);
218    let interval = i128::from(interval);
219
220    let total_count = (wall_now - first_due) / interval + 1;
221    // `first_due >= 0` and `wall_now <= i64::MAX`, so this always fits in u64.
222    let total_count = u64::try_from(total_count).map_err(|_| {
223        UnexpectedSnafu {
224            reason: format!(
225                "Cannot count due eval scheduled times up to {wall_now}: {total_count} does not fit in u64"
226            ),
227        }
228        .build()
229    })?;
230
231    match schedule.missed_tick_policy {
232        FlowMissedTickPolicy::Skip => {
233            // Keep only the latest due scheduled time; it is still on-lattice
234            // and `<= wall_now`.
235            let last = i64::try_from(first_due + i128::from(total_count - 1) * interval)
236                .map_err(|_| {
237                    UnexpectedSnafu {
238                        reason: format!(
239                            "Cannot compute the latest due eval scheduled time (first_due={first_due}, interval={interval}, count={total_count}): result does not fit in i64"
240                        ),
241                    }
242                    .build()
243                })?;
244            Ok(DueScheduledTimes {
245                scheduled_times_secs: vec![last],
246                skipped: total_count - 1,
247            })
248        }
249        FlowMissedTickPolicy::BoundedCatchUp => {
250            // The cutoff is computed in i128: `wall_now - max_lag` may
251            // legitimately underflow i64 (a cutoff before the Unix epoch) and
252            // must not saturate to a wrong value.
253            let cutoff = wall_now - i128::from(schedule.max_lag_secs);
254            let skipped_by_cutoff = if first_due >= cutoff {
255                0
256            } else {
257                // ceil((cutoff - first_due) / interval), capped at u64::MAX
258                // before the `.min(total_count)` below.
259                let skipped = (cutoff - first_due + interval - 1) / interval;
260                u64::try_from(skipped).unwrap_or(u64::MAX)
261            }
262            .min(total_count);
263
264            let remaining = total_count - skipped_by_cutoff;
265            if remaining == 0 {
266                return Ok(DueScheduledTimes {
267                    scheduled_times_secs: vec![],
268                    skipped: total_count,
269                });
270            }
271
272            // max_lag decides which missed scheduled times are recent enough to
273            // run; max_runs caps how many of those times execute back-to-back
274            // in one scheduler pass.
275            let keep_count = remaining.min(u64::from(schedule.max_runs));
276            let keep_start = skipped_by_cutoff + remaining - keep_count;
277            let mut scheduled_times_secs = Vec::with_capacity(keep_count as usize);
278            for i in 0..keep_count {
279                let t = i64::try_from(
280                    first_due + (i128::from(keep_start) + i128::from(i)) * interval,
281                )
282                .map_err(|_| {
283                    UnexpectedSnafu {
284                        reason: format!(
285                            "Cannot compute a due eval scheduled time (first_due={first_due}, interval={interval}, index={i}): result does not fit in i64"
286                        ),
287                    }
288                    .build()
289                })?;
290                scheduled_times_secs.push(t);
291            }
292
293            Ok(DueScheduledTimes {
294                scheduled_times_secs,
295                skipped: total_count - keep_count,
296            })
297        }
298    }
299}
300
301/// Ceils `time` to the next `anchor + k * interval` boundary.
302///
303/// Fallible: if the next boundary does not fit in `i64`, an explicit error is
304/// returned instead of clamping to a non-phase value such as `i64::MAX`.
305pub fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> Result<i64> {
306    if interval <= 0 {
307        return Ok(time);
308    }
309    if time <= anchor {
310        return Ok(anchor);
311    }
312
313    let diff = i128::from(time) - i128::from(anchor);
314    let interval = i128::from(interval);
315    let k = (diff + interval - 1) / interval;
316    let boundary = i128::from(anchor) + k * interval;
317
318    i64::try_from(boundary).map_err(|_| {
319        crate::error::UnexpectedSnafu {
320            reason: format!(
321                "Cannot align time {time} to the next `anchor + k * interval` boundary (anchor={anchor}, interval={interval}): result {boundary} does not fit in i64"
322            ),
323        }
324        .build()
325    })
326}
327
328#[cfg(test)]
329mod test {
330    use super::*;
331
332    fn schedule(
333        start: i64,
334        policy: FlowMissedTickPolicy,
335        max_runs: u32,
336        max_lag_secs: i64,
337    ) -> EvalSchedule {
338        EvalSchedule {
339            interval_secs: 60,
340            anchor_secs: 0,
341            start_secs: start,
342            missed_tick_policy: policy,
343            max_runs,
344            max_lag_secs,
345        }
346    }
347
348    fn config(policy: FlowMissedTickPolicy) -> FlowScheduleConfig {
349        FlowScheduleConfig {
350            anchor_secs: 10,
351            // phase-consistent: 310 = anchor(10) + 1 * interval(300)
352            start_secs: 310,
353            missed_tick_policy: policy,
354            catchup_max_runs: 4,
355            catchup_max_lag_secs: 600,
356        }
357    }
358
359    #[test]
360    fn ceil_to_boundary_handles_anchor_and_interval_edges() {
361        assert_eq!(ceil_to_boundary(-10, 0, 60).unwrap(), 0);
362        assert_eq!(ceil_to_boundary(0, 0, 60).unwrap(), 0);
363        assert_eq!(ceil_to_boundary(1, 0, 60).unwrap(), 60);
364        assert_eq!(ceil_to_boundary(60, 0, 60).unwrap(), 60);
365        assert_eq!(ceil_to_boundary(101, 100, 60).unwrap(), 160);
366        assert_eq!(ceil_to_boundary(50, 0, 0).unwrap(), 50);
367        // Never clamp to the non-phase i64::MAX: the next boundary does not fit.
368        assert!(ceil_to_boundary(i64::MAX, 0, 60).is_err());
369        assert!(ceil_to_boundary(i64::MAX - 1, i64::MIN, 60).is_err());
370    }
371
372    #[test]
373    fn from_config_maps_typed_config_and_defaults() {
374        assert!(EvalSchedule::from_config(None, None).unwrap().is_none());
375        assert!(EvalSchedule::from_config(Some(0), None).is_err());
376
377        let from_typed =
378            EvalSchedule::from_config(Some(300), Some(&config(FlowMissedTickPolicy::Skip)))
379                .unwrap()
380                .unwrap();
381        assert_eq!(from_typed.interval_secs, 300);
382        assert_eq!(from_typed.anchor_secs, 10);
383        assert_eq!(from_typed.start_secs, 310);
384        assert_eq!(from_typed.missed_tick_policy, FlowMissedTickPolicy::Skip);
385        assert_eq!(from_typed.max_runs, 4);
386        assert_eq!(from_typed.max_lag_secs, 600);
387
388        let defaulted = EvalSchedule::from_config(Some(300), None).unwrap().unwrap();
389        assert_eq!(defaulted.start_secs, 0);
390        assert_eq!(defaulted.max_runs, 3);
391        assert_eq!(defaulted.max_lag_secs, 900);
392    }
393
394    #[test]
395    fn from_config_rejects_invalid_catchup_limits() {
396        let mut c = config(FlowMissedTickPolicy::BoundedCatchUp);
397        c.catchup_max_runs = 0;
398        assert!(EvalSchedule::from_config(Some(300), Some(&c)).is_err());
399
400        let mut c = config(FlowMissedTickPolicy::BoundedCatchUp);
401        c.catchup_max_lag_secs = 0;
402        assert!(EvalSchedule::from_config(Some(300), Some(&c)).is_err());
403    }
404
405    #[test]
406    fn nonzero_anchor_due_selection_follows_phase() {
407        // anchor=120 (i.e. `EVAL OFFSET '2 minutes'`), interval=3600:
408        // boundaries at :02 every hour. start=3720 (120 + 3600).
409        let s = EvalSchedule {
410            interval_secs: 3600,
411            anchor_secs: 120,
412            start_secs: 3720,
413            missed_tick_policy: FlowMissedTickPolicy::BoundedCatchUp,
414            max_runs: 3,
415            max_lag_secs: 3600,
416        };
417        assert_eq!(
418            select_due_scheduled_times(&s, 0, 100)
419                .unwrap()
420                .scheduled_times_secs,
421            Vec::<i64>::new()
422        );
423        // From 3720 on, every selected time must be on the :02 phase.
424        let due = select_due_scheduled_times(&s, 0, 3720).unwrap();
425        assert_eq!(due.scheduled_times_secs, vec![3720]);
426        let due = select_due_scheduled_times(&s, 3720, 7320).unwrap();
427        assert_eq!(due.scheduled_times_secs, vec![7320]);
428        for t in &due.scheduled_times_secs {
429            assert_eq!((t - 120) % 3600, 0);
430        }
431        assert_eq!(s.next_scheduled_time_after(3720).unwrap(), 7320);
432        assert_eq!(s.next_scheduled_time_after(7300).unwrap(), 7320);
433    }
434
435    #[test]
436    fn next_scheduled_time_after_respects_start_sequence() {
437        let s = schedule(50, FlowMissedTickPolicy::BoundedCatchUp, 3, 300);
438        assert_eq!(s.next_scheduled_time_after(0).unwrap(), 50);
439        assert_eq!(s.next_scheduled_time_after(50).unwrap(), 110);
440        assert_eq!(s.next_scheduled_time_after(100).unwrap(), 110);
441    }
442
443    #[test]
444    fn near_i64_max_advancement_is_exact_or_explicit_error() {
445        // anchor=0, interval=60: the next boundary after i64::MAX - 60 is
446        // 9223372036854775800, still in range and exactly on the lattice.
447        let s = schedule(0, FlowMissedTickPolicy::Skip, 5, 3600);
448        let cursor = i64::MAX - 60;
449        let next = s.next_scheduled_time_after(cursor).unwrap();
450        assert_eq!(next, 9223372036854775800);
451        assert_eq!(next % 60, 0);
452
453        // Advancing past the last representable boundary is an explicit error,
454        // never a saturated non-phase value like i64::MAX.
455        let err = s
456            .next_scheduled_time_after(9223372036854775800)
457            .unwrap_err();
458        assert!(err.to_string().contains("does not fit in i64"));
459
460        // A non-positive interval is an explicit error, not a saturating
461        // `cursor + 1` result.
462        let invalid = EvalSchedule {
463            interval_secs: 0,
464            anchor_secs: 0,
465            start_secs: 0,
466            missed_tick_policy: FlowMissedTickPolicy::Skip,
467            max_runs: 3,
468            max_lag_secs: 900,
469        };
470        assert!(invalid.next_scheduled_time_after(0).is_err());
471    }
472
473    #[test]
474    fn due_scheduled_time_selection_handles_empty_and_start_boundary() {
475        let s = schedule(120, FlowMissedTickPolicy::BoundedCatchUp, 10, 3600);
476        assert_eq!(
477            select_due_scheduled_times(&s, 0, 100)
478                .unwrap()
479                .scheduled_times_secs,
480            Vec::<i64>::new()
481        );
482        assert_eq!(
483            select_due_scheduled_times(&s, 0, 300)
484                .unwrap()
485                .scheduled_times_secs,
486            vec![120, 180, 240, 300]
487        );
488    }
489
490    #[test]
491    fn bounded_catch_up_applies_lag_and_max_runs() {
492        let s = schedule(0, FlowMissedTickPolicy::BoundedCatchUp, 2, 180);
493        let due = select_due_scheduled_times(&s, 0, 600).unwrap();
494        assert_eq!(due.scheduled_times_secs, vec![540, 600]);
495        assert_eq!(due.skipped, 8);
496    }
497
498    #[test]
499    fn bounded_catch_up_can_skip_all_due_scheduled_times() {
500        let s = schedule(0, FlowMissedTickPolicy::BoundedCatchUp, 3, 30);
501        let due = select_due_scheduled_times(&s, 0, 100).unwrap();
502        assert!(due.scheduled_times_secs.is_empty());
503        assert_eq!(due.skipped, 1);
504    }
505
506    #[test]
507    fn skip_policy_keeps_only_latest_due_scheduled_time() {
508        let s = schedule(0, FlowMissedTickPolicy::Skip, 5, 3600);
509        let due = select_due_scheduled_times(&s, 0, 300).unwrap();
510        assert_eq!(due.scheduled_times_secs, vec![300]);
511        assert_eq!(due.skipped, 4);
512    }
513
514    #[test]
515    fn huge_missed_gap_allocates_only_kept_scheduled_times() {
516        let s = schedule(0, FlowMissedTickPolicy::BoundedCatchUp, 5, 3600);
517        let due = select_due_scheduled_times(&s, 0, 86400).unwrap();
518        assert_eq!(
519            due.scheduled_times_secs,
520            vec![86160, 86220, 86280, 86340, 86400]
521        );
522        assert_eq!(due.skipped, 1435);
523    }
524}
flow/batching_mode/eval_schedule.rs

flow/batching_mode/
eval_schedule.rs
