fix ignore uppers for empty batch sets

Author: ptravers

src/persist-client/src/lib.rs

@@ -1187,6 +1187,7 @@ mod tests {
                     append_upper: Antichain::from_elem(5),
                 },
             );
+
             let batch = write0
                 .batch(&ts3, Antichain::from_elem(3), Antichain::from_elem(4))
                 .await
@@ -1203,6 +1204,7 @@ mod tests {
                     append_upper: Antichain::from_elem(3),
                 },
             );
+
             let batch = write0
                 .batch(&ts3, Antichain::from_elem(3), Antichain::from_elem(4))
                 .await

src/persist-client/src/write.rs

@@ -463,7 +463,26 @@ where
     where
         D: Send + Sync,
     {
+        let mut max_upper: Option<&Antichain<T>> = None;
+        let mut min_lower: Option<&Antichain<T>> = None;
+
         for batch in batches.iter() {
+            if let Some(current_max_upper) = max_upper {
+                if batch.upper().gt(current_max_upper) {
+                    max_upper = Some(batch.upper());
+                }
+            } else {
+                max_upper = Some(batch.upper());
+            }
+
+            if let Some(current_min_lower) = min_lower {
+                if batch.lower().lt(current_min_lower) {
+                    min_lower = Some(batch.lower());
+                }
+            } else {
+                min_lower = Some(batch.lower());
+            }
+
             if self.machine.shard_id() != batch.shard_id() {
                 return Err(InvalidUsage::BatchNotFromThisShard {
                     batch_shard: batch.shard_id(),
@@ -484,6 +503,18 @@ where
 
         let lower = expected_upper.clone();
         let upper = new_upper;
+
+        if let (Some(max_upper), Some(min_lower)) = (max_upper, min_lower) {
+            if max_upper.lt(&upper) || min_lower.gt(&lower) {
+                return Err(InvalidUsage::InvalidBatchBounds {
+                    batch_lower: min_lower.clone(),
+                    batch_upper: max_upper.clone(),
+                    append_lower: lower.clone(),
+                    append_upper: upper.clone(),
+                });
+            }
+        }
+
         let since = Antichain::from_elem(T::minimum());
         let desc = Description::new(lower, upper, since);
 

src/storage/src/render/persist_sink.rs

@@ -107,7 +107,7 @@ use mz_persist_client::cache::PersistClientCache;
 use mz_persist_client::error::UpperMismatch;
 use mz_persist_types::codec_impls::UnitSchema;
 use mz_persist_types::{Codec, Codec64};
-use mz_repr::{Diff, GlobalId, Row};
+use mz_repr::{Diff, GlobalId, Row, TimestampManipulation};
 use mz_storage_types::controller::CollectionMetadata;
 use mz_storage_types::errors::DataflowError;
 use mz_storage_types::sources::SourceData;
@@ -116,12 +116,12 @@ use mz_timely_util::builder_async::{
     Event, OperatorBuilder as AsyncOperatorBuilder, PressOnDropButton,
 };
 use serde::{Deserialize, Serialize};
+use timely::PartialOrder;
 use timely::container::CapacityContainerBuilder;
 use timely::dataflow::channels::pact::{Exchange, Pipeline};
 use timely::dataflow::operators::{Broadcast, Capability, CapabilitySet, Inspect};
 use timely::dataflow::{Scope, Stream};
 use timely::progress::{Antichain, Timestamp};
-use timely::{Container, PartialOrder};
 use tokio::sync::Semaphore;
 use tracing::trace;
 
@@ -1111,21 +1111,6 @@ where
                 }
             });
 
-            let mut done_batches_iter = done_batches.iter();
-
-            let Some(first_batch_description) = done_batches_iter.next() else {
-                upper_cap_set.downgrade(current_upper.borrow().iter());
-                continue;
-            };
-
-            let batch_upper = first_batch_description.1.clone();
-
-            let mut batch_lower = if let Some(last_batch_description) = done_batches_iter.last() {
-                last_batch_description.0.clone()
-            } else {
-                first_batch_description.0.clone()
-            };
-
             let mut batches: Vec<FinishedBatch> = vec![];
             let mut batch_metrics: Vec<BatchMetrics> = vec![];
 
@@ -1136,7 +1121,23 @@ where
                 batch_metrics.push(batch.batch_metrics);
             }
 
-            let mut to_append = batches.iter_mut().map(|b| &mut b.batch).collect::<Vec<_>>();
+            let mut batch_upper = Antichain::from_elem(mz_repr::Timestamp::minimum());
+            let mut batch_lower = Antichain::from_elem(mz_repr::Timestamp::maximum());
+
+            let mut to_append = batches
+                .iter_mut()
+                .map(|b| {
+                    if batch_upper.lt(b.batch.upper()) {
+                        batch_upper = b.batch.upper().clone();
+                    }
+
+                    if batch_lower.gt(b.batch.lower()) {
+                        batch_lower = b.batch.lower().clone();
+                    }
+
+                    &mut b.batch
+                })
+                .collect::<Vec<_>>();
 
             loop {
                 let result = {