Merge pull request #30989 from teskje/fix-mv-sink-persist-frontier

mv-sink: stop discarding valid batch descriptions

Author: teskje

src/compute/src/sink/correction.rs

@@ -200,11 +200,6 @@ impl<D: Data> Correction<D> {
         count
     }
 
-    /// Return the current since frontier.
-    pub fn since(&self) -> &Antichain<Timestamp> {
-        &self.since
-    }
-
     /// Advance the since frontier.
     ///
     /// # Panics

src/compute/src/sink/materialized_view_v2.rs

@@ -794,12 +794,15 @@ mod write {
         /// The frontiers of the `desired` inputs.
         desired_frontiers: OkErr<Antichain<Timestamp>, Antichain<Timestamp>>,
         /// The frontiers of the `persist` inputs.
+        ///
+        /// Note that this is _not_ the same as the write frontier of the output persist shard! It
+        /// usually is, but during snapshot processing, these frontiers will start at the shard's
+        /// read frontier, so they can be beyond its write frontier. This is important as it means
+        /// we must not discard batch descriptions based on these persist frontiers: A batch
+        /// description might still be valid even if its `lower` is before the persist frontiers we
+        /// observe.
         persist_frontiers: OkErr<Antichain<Timestamp>, Antichain<Timestamp>>,
         /// The current valid batch description and associated output capability, if any.
-        ///
-        /// Note that "valid" here implies that if a batch description is set, it must be true that
-        /// its `lower` is >= the `persist_frontier`. Otherwise the described batch couldn't be
-        /// appended anymore, rendering the batch description invalid.
         batch_description: Option<(BatchDescription, Capability<Timestamp>)>,
         /// A request to force a consolidation of `corrections` once both `desired_frontiers` and
         /// `persist_frontiers` become greater than the given frontier.
@@ -894,15 +897,6 @@ mod write {
             self.corrections.ok.advance_since(frontier.clone());
             self.corrections.err.advance_since(frontier.clone());
 
-            // If the `persist` frontier is greater than the `lower` of the current batch
-            // description, we won't be able to append the batch, so the batch description is not
-            // valid anymore.
-            if let Some((desc, _)) = &self.batch_description {
-                if PartialOrder::less_than(&desc.lower, frontier) {
-                    self.batch_description = None;
-                }
-            }
-
             self.maybe_force_consolidation();
         }
 
@@ -927,16 +921,18 @@ mod write {
         }
 
         fn absorb_batch_description(&mut self, desc: BatchDescription, cap: Capability<Timestamp>) {
-            // The incoming batch description is outdated if either:
-            // * we already have a batch description with a greater `lower`, or
-            // * its `lower` is less than the persist frontier
-            let validity_frontier = match &self.batch_description {
-                Some((prev, _)) => &prev.lower,
-                None => self.persist_frontiers.frontier(),
-            };
-            if PartialOrder::less_than(&desc.lower, validity_frontier) {
-                self.trace(format!("skipping outdated batch description: {desc:?}"));
-                return;
+            // The incoming batch description is outdated if we already have a batch description
+            // with a greater `lower`.
+            //
+            // Note that we cannot assume a description is outdated based on the comparison of its
+            // `lower` with the `persist_frontier`. The persist frontier observed by the `write`
+            // operator is initialized with the shard's read frontier, so it can be greater than
+            // the shard's write frontier.
+            if let Some((prev, _)) = &self.batch_description {
+                if PartialOrder::less_than(&desc.lower, &prev.lower) {
+                    self.trace(format!("skipping outdated batch description: {desc:?}"));
+                    return;
+                }
             }
 
             self.batch_description = Some((desc, cap));
@@ -948,7 +944,8 @@ mod write {
 
             // We can write a new batch if we have seen all `persist` updates before `lower` and
             // all `desired` updates up to `upper`.
-            let persist_complete = desc.lower == *self.persist_frontiers.frontier();
+            let persist_complete =
+                PartialOrder::less_equal(&desc.lower, self.persist_frontiers.frontier());
             let desired_complete =
                 PartialOrder::less_equal(&desc.upper, self.desired_frontiers.frontier());
             if !persist_complete || !desired_complete {
@@ -957,9 +954,6 @@ mod write {
 
             let (desc, cap) = self.batch_description.take()?;
 
-            assert_eq!(desc.lower, *self.corrections.ok.since());
-            assert_eq!(desc.lower, *self.corrections.err.since());
-
             let ok_updates = self.corrections.ok.updates_before(&desc.upper);
             let err_updates = self.corrections.err.updates_before(&desc.upper);
 

test/sqllogictest/github-8867.slt

@@ -0,0 +1,43 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Regression test for database-issues#8867.
+
+mode cockroach
+
+simple conn=mz_system,user=mz_system
+ALTER SYSTEM SET enable_refresh_every_mvs = true
+----
+COMPLETE 0
+
+statement ok
+CREATE TABLE t (x int)
+
+statement ok
+INSERT INTO t VALUES (1)
+
+statement ok
+CREATE MATERIALIZED VIEW mv1
+WITH (REFRESH EVERY '10s' ALIGNED TO mz_now()::text::int8 + 10000)
+AS SELECT * FROM t
+
+statement ok
+CREATE MATERIALIZED VIEW mv2
+WITH (REFRESH EVERY '10 s' ALIGNED TO mz_now()::text::int8 + 10000)
+AS SELECT * FROM mv1
+
+query I
+SELECT * FROM mv1;
+----
+1
+
+query I
+SELECT * FROM mv2;
+----
+1