diff --git a/src/adapter/src/coord/sequencer/inner/subscribe.rs b/src/adapter/src/coord/sequencer/inner/subscribe.rs index 331398a83013e..a8167ba1347f4 100644 --- a/src/adapter/src/coord/sequencer/inner/subscribe.rs +++ b/src/adapter/src/coord/sequencer/inner/subscribe.rs @@ -573,6 +573,37 @@ impl Coordinator { }; active_subscribe.initialize(); + // A subscribe whose `as_of` equals its `up_to` covers the empty range + // `[as_of, up_to)` and is guaranteed to produce no data. The adapter + // reports this with the `EqualSubscribeBounds` notice. Skip installing a + // dataflow for it: the only observable output is the initial progress row + // that `initialize` already emitted (WITH PROGRESS only). Dropping + // `active_subscribe` closes the channel, which completes the subscribe on + // the client side. + let is_empty = df_desc + .as_of + .as_ref() + .is_some_and(|as_of| as_of == &df_desc.until); + if is_empty { + drop(active_subscribe); + drop(read_holds); + let resp = ExecuteResponse::Subscribing { + rx, + ctx_extra: std::mem::take(ctx_extra), + instance_id: cluster_id, + }; + let resp = match plan.copy_to { + None => resp, + Some(format) => ExecuteResponse::CopyTo { + format, + resp: Box::new(resp), + }, + }; + // No `mz_subscriptions` bookkeeping was deferred, so hand back an + // already-satisfied notify. + return Ok((resp, Box::pin(std::future::ready(())))); + } + // Register bookkeeping for the new SUBSCRIBE and ship its dataflow. The // `mz_subscriptions` write is deferred to a group commit (see // `add_active_compute_sink`) rather than committed inline, so it does not block diff --git a/src/compute-types/src/dataflows.rs b/src/compute-types/src/dataflows.rs index ccbc6190ca018..82a5c064759b9 100644 --- a/src/compute-types/src/dataflows.rs +++ b/src/compute-types/src/dataflows.rs @@ -87,19 +87,20 @@ impl DataflowDescription { // Ensure that as_of <= until. soft_assert_or_log!( timely::PartialOrder::less_equal(as_of, until), - "expected empty `as_of ≤ until`, got `{as_of:?} ≰ {until:?}`", + "expected `as_of ≤ until`, got `{as_of:?} ≰ {until:?}`", ); // IF `as_of` is not a single timestamp this can't be a single time dataflow. let Some(as_of) = as_of.as_option() else { return false; }; - // Ensure that `as_of = MAX` implies `until.is_empty()`. - soft_assert_or_log!( - as_of != &mz_repr::Timestamp::MAX || until.is_empty(), - "expected `until = {{}}` due to `as_of = MAX`, got `until = {until:?}`", - ); - // Note that the `(as_of = MAX, until = {})` case also returns `true` - // here (as expected) since we are going to compare two `None` values. + // With `as_of = MAX` the `as_of <= until` invariant leaves only two valid + // frontiers, and both are handled correctly by the comparison below: + // * `until = {}` (unbounded) yields `None == None`, so single time. `MAX` + // is the last timestamp, so an unbounded dataflow starting there sees + // exactly one time. + // * `until = {MAX}` is the empty range produced by e.g. + // `SUBSCRIBE ... AS OF MAX UP TO MAX`. `try_step_forward` on `MAX` + // overflows to `None`, so `None == Some(MAX)` is false: not single time. as_of.try_step_forward().as_ref() == until.as_option() } } @@ -629,3 +630,50 @@ pub struct BuildDesc

{ /// TODO(database-issues#7533): Add documentation. pub plan: P, } + +#[cfg(test)] +mod tests { + use super::*; + + /// Builds a minimal dataflow description carrying only the `as_of` and + /// `until` frontiers that `is_single_time` inspects. + fn desc_with_bounds( + as_of: Antichain, + until: Antichain, + ) -> DataflowDescription { + let mut desc = DataflowDescription::new("test".to_string()); + desc.set_as_of(as_of); + desc.until = until; + desc + } + + #[mz_ore::test] + fn is_single_time_classification() { + // Ordinary single-time dataflow: `until = as_of + 1`. + assert!( + desc_with_bounds( + Antichain::from_elem(5.into()), + Antichain::from_elem(6.into()) + ) + .is_single_time() + ); + + // `as_of = MAX` with an empty `until` is single time: `MAX` is the last + // timestamp, so an unbounded dataflow starting there sees exactly one time. + assert!( + desc_with_bounds(Antichain::from_elem(Timestamp::MAX), Antichain::new()) + .is_single_time() + ); + + // `as_of = MAX` with `until = {MAX}` is an empty range, produced by + // `SUBSCRIBE ... AS OF MAX UP TO MAX`. It must not be classified as single + // time, and it must not trip an invariant assertion (see CLU-169). + assert!( + !desc_with_bounds( + Antichain::from_elem(Timestamp::MAX), + Antichain::from_elem(Timestamp::MAX) + ) + .is_single_time() + ); + } +} diff --git a/test/testdrive/fetch-tail-as-of.td b/test/testdrive/fetch-tail-as-of.td index 7bd37a40654ca..20b193208f24b 100644 --- a/test/testdrive/fetch-tail-as-of.td +++ b/test/testdrive/fetch-tail-as-of.td @@ -11,7 +11,7 @@ # Make sure that the AS OF is observed for SUBSCRIBE # -$ set-regex match=\d{13} replacement= +$ set-regex match=\d{13,} replacement= > CREATE TABLE t1 (f1 INTEGER); > CREATE DEFAULT INDEX ON t1 @@ -59,3 +59,32 @@ contains:imestamp # No rows expected > FETCH 1 c WITH (timeout = '1s'); + +> COMMIT + +# Regression test for CLU-169. `AS OF MAX UP TO MAX` describes an empty range +# (`as_of` equals the exclusive `up_to`), so the subscribe is guaranteed empty. +# Optimization used to trip the `is_single_time` invariant assertion, which +# expected an empty `until` whenever `as_of` was `MAX`, and surfaced as an +# internal optimizer error. The empty subscribe is now short-circuited without +# installing a dataflow. +> BEGIN + +> DECLARE c CURSOR FOR SUBSCRIBE t1 AS OF 18446744073709551615 UP TO 18446744073709551615; + +# No rows expected. +> FETCH ALL c WITH (timeout = '5s'); + +> COMMIT + +# The short-circuited empty subscribe still honors `WITH (PROGRESS)`: it emits +# the initial progress row at the snapshot timestamp before completing, matching +# a subscribe that installs a dataflow. +> BEGIN + +> DECLARE c CURSOR FOR SUBSCRIBE t1 WITH (PROGRESS = TRUE) AS OF 18446744073709551615 UP TO 18446744073709551615; + +> FETCH ALL c WITH (timeout = '5s'); + true + +> COMMIT