compute: allow SUBSCRIBE AS OF MAX UP TO MAX

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

src/compute-types/src/dataflows.rs

@@ -87,19 +87,20 @@ impl<P, S> DataflowDescription<P, S> {
         // 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<P> {
     /// 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<Timestamp>,
+        until: Antichain<Timestamp>,
+    ) -> DataflowDescription<OptimizedMirRelationExpr, ()> {
+        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()
+        );
+    }
+}

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=<TIMESTAMP>
+$ set-regex match=\d{13,} replacement=<TIMESTAMP>
 
 > 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');
+<TIMESTAMP> true <null> <null>
+
+> COMMIT