Add a total order bound on persist

We de facto expect it currently, and it simplifies some logic to be able
to require it.

Author: bkirwi

src/persist-client/src/batch.rs

@@ -1414,7 +1414,6 @@ fn diffs_sum<D: Semigroup + Codec64>(updates: &Int64Array) -> Option<D> {
 #[cfg(test)]
 mod tests {
     use mz_dyncfg::ConfigUpdates;
-    use timely::order::Product;
 
     use super::*;
     use crate::PersistLocation;
@@ -1596,50 +1595,6 @@ mod tests {
         assert!(part_bytes.is_none());
     }
 
-    #[mz_ore::test(tokio::test)]
-    #[cfg_attr(miri, ignore)] // unsupported operation: returning ready events from epoll_wait is not yet implemented
-    async fn batch_builder_partial_order() {
-        let cache = PersistClientCache::new_no_metrics();
-        // Set blob_target_size to 0 so that each row gets forced into its own batch part
-        cache.cfg.set_config(&BLOB_TARGET_SIZE, 0);
-        // Otherwise fails: expected hollow part!
-        cache.cfg.set_config(&STRUCTURED_KEY_LOWER_LEN, 0);
-        cache.cfg.set_config(&INLINE_WRITES_SINGLE_MAX_BYTES, 0);
-        cache.cfg.set_config(&INLINE_WRITES_TOTAL_MAX_BYTES, 0);
-        let client = cache
-            .open(PersistLocation::new_in_mem())
-            .await
-            .expect("client construction failed");
-        let shard_id = ShardId::new();
-        let (mut write, _) = client
-            .expect_open::<String, String, Product<u32, u32>, i64>(shard_id)
-            .await;
-
-        let batch = write
-            .batch(
-                &[
-                    (("1".to_owned(), "one".to_owned()), Product::new(0, 10), 1),
-                    (("2".to_owned(), "two".to_owned()), Product::new(10, 0), 1),
-                ],
-                Antichain::from_elem(Product::new(0, 0)),
-                Antichain::from_iter([Product::new(0, 11), Product::new(10, 1)]),
-            )
-            .await
-            .expect("invalid usage");
-
-        assert_eq!(batch.batch.part_count(), 2);
-        for part in &batch.batch.parts {
-            let part = part.expect_hollow_part();
-            match BlobKey::parse_ids(&part.key.complete(&shard_id)) {
-                Ok((shard, PartialBlobKey::Batch(writer, _))) => {
-                    assert_eq!(shard.to_string(), shard_id.to_string());
-                    assert_eq!(writer, WriterKey::for_version(&cache.cfg.build_version));
-                }
-                _ => panic!("unparseable blob key"),
-            }
-        }
-    }
-
     #[mz_ore::test]
     fn untrimmable_columns() {
         let untrimmable = UntrimmableColumns {

src/persist-client/src/internal/compact.rs

@@ -942,7 +942,6 @@ mod tests {
     use mz_dyncfg::ConfigUpdates;
     use mz_ore::{assert_contains, assert_err};
     use mz_persist_types::codec_impls::StringSchema;
-    use timely::order::Product;
     use timely::progress::Antichain;
 
     use crate::PersistLocation;
@@ -1021,83 +1020,6 @@ mod tests {
         assert_eq!(updates, all_ok(&data, 10));
     }
 
-    #[mz_persist_proc::test(tokio::test)]
-    #[cfg_attr(miri, ignore)] // unsupported operation: returning ready events from epoll_wait is not yet implemented
-    async fn compaction_partial_order(dyncfgs: ConfigUpdates) {
-        let data = vec![
-            (("0".to_owned(), "zero".to_owned()), Product::new(0, 10), 1),
-            (("1".to_owned(), "one".to_owned()), Product::new(10, 0), 1),
-        ];
-
-        let cache = new_test_client_cache(&dyncfgs);
-        cache.cfg.set_config(&BLOB_TARGET_SIZE, 100);
-        let (mut write, _) = cache
-            .open(PersistLocation::new_in_mem())
-            .await
-            .expect("client construction failed")
-            .expect_open::<String, String, Product<u32, u32>, i64>(ShardId::new())
-            .await;
-        let b0 = write
-            .batch(
-                &data[..1],
-                Antichain::from_elem(Product::new(0, 0)),
-                Antichain::from_iter([Product::new(0, 11), Product::new(10, 0)]),
-            )
-            .await
-            .expect("invalid usage")
-            .into_hollow_batch();
-
-        let b1 = write
-            .batch(
-                &data[1..],
-                Antichain::from_iter([Product::new(0, 11), Product::new(10, 0)]),
-                Antichain::from_elem(Product::new(10, 1)),
-            )
-            .await
-            .expect("invalid usage")
-            .into_hollow_batch();
-
-        let req = CompactReq {
-            shard_id: write.machine.shard_id(),
-            desc: Description::new(
-                b0.desc.lower().clone(),
-                b1.desc.upper().clone(),
-                Antichain::from_elem(Product::new(10, 0)),
-            ),
-            inputs: vec![b0, b1],
-        };
-        let schemas = Schemas {
-            id: None,
-            key: Arc::new(StringSchema),
-            val: Arc::new(StringSchema),
-        };
-        let res = Compactor::<String, String, Product<u32, u32>, i64>::compact(
-            CompactConfig::new(&write.cfg, write.shard_id()),
-            Arc::clone(&write.blob),
-            Arc::clone(&write.metrics),
-            write.metrics.shards.shard(&write.machine.shard_id(), ""),
-            Arc::new(IsolatedRuntime::default()),
-            req.clone(),
-            schemas.clone(),
-        )
-        .await
-        .expect("compaction failed");
-
-        assert_eq!(res.output.desc, req.desc);
-        assert_eq!(res.output.len, 2);
-        assert_eq!(res.output.part_count(), 1);
-        let part = res.output.parts[0].expect_hollow_part();
-        let (part, updates) = expect_fetch_part(
-            write.blob.as_ref(),
-            &part.key.complete(&write.machine.shard_id()),
-            &write.metrics,
-            &schemas,
-        )
-        .await;
-        assert_eq!(part.desc, res.output.desc);
-        assert_eq!(updates, all_ok(&data, Product::new(10, 0)));
-    }
-
     #[mz_persist_proc::test(tokio::test)]
     #[cfg_attr(miri, ignore)] // unsupported operation: returning ready events from epoll_wait is not yet implemented
     async fn disable_compaction(dyncfgs: ConfigUpdates) {

src/persist-client/src/internal/state_diff.rs

@@ -1033,9 +1033,6 @@ fn sniff_compaction<'a, T: Timestamp + Lattice>(
 ///
 /// This can only happen when the batch needing to be split is empty, so error
 /// out if it isn't because that means something unexpected is going on.
-///
-/// TODO: This implementation is certainly not correct if T is actually only
-/// partially ordered.
 fn apply_compaction_lenient<'a, T: Timestamp + Lattice>(
     metrics: &Metrics,
     mut trace: Vec<HollowBatch<T>>,

src/persist-client/src/lib.rs

@@ -30,6 +30,7 @@ use mz_persist_types::schema::SchemaId;
 use mz_persist_types::{Codec, Codec64, Opaque};
 use mz_proto::{IntoRustIfSome, ProtoType};
 use semver::Version;
+use timely::order::TotalOrder;
 use timely::progress::{Antichain, Timestamp};
 
 use crate::async_runtime::IsolatedRuntime;
@@ -290,7 +291,7 @@ impl PersistClient {
     where
         K: Debug + Codec,
         V: Debug + Codec,
-        T: Timestamp + Lattice + Codec64 + Sync,
+        T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
         D: Semigroup + Ord + Codec64 + Send + Sync,
     {
         Ok((
@@ -332,7 +333,7 @@ impl PersistClient {
     where
         K: Debug + Codec,
         V: Debug + Codec,
-        T: Timestamp + Lattice + Codec64 + Sync,
+        T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
         D: Semigroup + Codec64 + Send + Sync,
     {
         let machine = self.make_machine(shard_id, diagnostics.clone()).await?;
@@ -504,7 +505,7 @@ impl PersistClient {
     where
         K: Debug + Codec,
         V: Debug + Codec,
-        T: Timestamp + Lattice + Codec64 + Sync,
+        T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
         D: Semigroup + Ord + Codec64 + Send + Sync,
     {
         let machine = self.make_machine(shard_id, diagnostics.clone()).await?;
@@ -566,7 +567,7 @@ impl PersistClient {
     where
         K: Debug + Codec,
         V: Debug + Codec,
-        T: Timestamp + Lattice + Codec64 + Sync,
+        T: Timestamp + Lattice + Codec64 + TotalOrder + Sync,
         D: Semigroup + Ord + Codec64 + Send + Sync,
     {
         WriteHandle::builder_inner(
@@ -652,7 +653,7 @@ impl PersistClient {
     where
         K: Debug + Codec + Ord,
         V: Debug + Codec + Ord,
-        T: Timestamp + Lattice + Codec64 + Sync,
+        T: Timestamp + Lattice + Codec64 + TotalOrder + Sync,
         D: Semigroup + Ord + Codec64 + Send + Sync,
     {
         let shard_metrics = self.metrics.shards.shard(&shard_id, "peek_stash");
@@ -839,7 +840,7 @@ impl PersistClient {
     where
         K: Debug + Codec,
         V: Debug + Codec,
-        T: Timestamp + Lattice + Codec64 + Sync,
+        T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
         D: Semigroup + Ord + Codec64 + Send + Sync,
         K::Schema: Default,
         V::Schema: Default,

src/persist-client/src/read.rs

@@ -33,6 +33,7 @@ use mz_persist_types::{Codec, Codec64};
 use proptest_derive::Arbitrary;
 use serde::{Deserialize, Serialize};
 use timely::PartialOrder;
+use timely::order::TotalOrder;
 use timely::progress::{Antichain, Timestamp};
 use tokio::runtime::Handle;
 use tracing::{Instrument, debug_span, warn};
@@ -113,7 +114,7 @@ impl<K, V, T, D> Subscribe<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     fn new(snapshot_parts: Vec<LeasedBatchPart<T>>, listen: Listen<K, V, T, D>) -> Self {
@@ -150,7 +151,7 @@ impl<K, V, T, D> Subscribe<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     /// Equivalent to `next`, but rather than returning a [`LeasedBatchPart`],
@@ -195,7 +196,7 @@ impl<K, V, T, D> Subscribe<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     /// Politely expires this subscribe, releasing its lease.
@@ -238,7 +239,7 @@ impl<K, V, T, D> Listen<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     async fn new(mut handle: ReadHandle<K, V, T, D>, as_of: Antichain<T>) -> Self {
@@ -328,8 +329,7 @@ where
         // batch the last time we called this) that are strictly less_than the
         // batch upper to compute a new since. For totally ordered times
         // (currently always the case in mz) self.frontier will always have a
-        // single element and it will be less_than upper, but the following
-        // logic is (hopefully) correct for partially order times as well. We
+        // single element and it will be less_than upper. We
         // could also abuse the fact that every time we actually emit is
         // guaranteed by definition to be less_than upper to be a bit more
         // prompt, but this would involve a lot more temporary antichains and
@@ -365,7 +365,7 @@ impl<K, V, T, D> Listen<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     /// Attempt to pull out the next values of this subscription.
@@ -440,7 +440,7 @@ impl<K, V, T, D> Listen<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     /// Fetches the contents of `part` and returns its lease.
@@ -525,7 +525,7 @@ impl<K, V, T, D> ReadHandle<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     pub(crate) async fn new(
@@ -939,7 +939,7 @@ impl<K, V, T, D> ReadHandle<K, V, T, D>
 where
     K: Debug + Codec + Ord,
     V: Debug + Codec + Ord,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Ord + Codec64 + Send + Sync,
 {
     /// Generates a [Self::snapshot], and fetches all of the batches it
@@ -1135,7 +1135,7 @@ impl<K, V, T, D> ReadHandle<K, V, T, D>
 where
     K: Debug + Codec + Ord,
     V: Debug + Codec + Ord,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Codec64 + Send + Sync,
 {
     /// Generates a [Self::snapshot], and streams out all of the updates
@@ -1188,7 +1188,7 @@ impl<K, V, T, D> ReadHandle<K, V, T, D>
 where
     K: Debug + Codec + Ord,
     V: Debug + Codec + Ord,
-    T: Timestamp + Lattice + Codec64 + Ord + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Ord + Codec64 + Send + Sync,
 {
     /// Test helper to generate a [Self::snapshot] call that is expected to

src/persist-client/src/write.rs

@@ -29,6 +29,7 @@ use proptest_derive::Arbitrary;
 use semver::Version;
 use serde::{Deserialize, Serialize};
 use timely::PartialOrder;
+use timely::order::TotalOrder;
 use timely::progress::{Antichain, Timestamp};
 use tokio::runtime::Handle;
 use tracing::{Instrument, debug_span, info, warn};
@@ -136,7 +137,7 @@ impl<K, V, T, D> WriteHandle<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
-    T: Timestamp + Lattice + Codec64 + Sync,
+    T: Timestamp + TotalOrder + Lattice + Codec64 + Sync,
     D: Semigroup + Ord + Codec64 + Send + Sync,
 {
     pub(crate) fn new(

src/persist-types/src/codec_impls.rs

@@ -18,7 +18,6 @@ use arrow::array::{
     StructArray,
 };
 use bytes::{BufMut, Bytes};
-use timely::order::Product;
 
 use crate::arrow::ArrayOrd;
 use crate::columnar::{ColumnDecoder, ColumnEncoder, Schema};
@@ -515,24 +514,6 @@ impl Opaque for u64 {
     }
 }
 
-impl Codec64 for Product<u32, u32> {
-    fn codec_name() -> String {
-        "Product<u32, u32>".to_owned()
-    }
-
-    fn encode(&self) -> [u8; 8] {
-        let o = self.outer.to_le_bytes();
-        let i = self.inner.to_le_bytes();
-        [o[0], o[1], o[2], o[3], i[0], i[1], i[2], i[3]]
-    }
-
-    fn decode(buf: [u8; 8]) -> Self {
-        let outer = [buf[0], buf[1], buf[2], buf[3]];
-        let inner = [buf[4], buf[5], buf[6], buf[7]];
-        Product::new(u32::from_le_bytes(outer), u32::from_le_bytes(inner))
-    }
-}
-
 // TODO: Remove this once we wrap coord epochs in an `Epoch` struct and impl
 // Opaque on `Epoch` instead.
 impl Opaque for i64 {

src/storage-client/src/storage_collections.rs

@@ -3194,7 +3194,7 @@ async fn finalize_shards_task<T>(
         read_only,
     }: FinalizeShardsTaskConfig,
 ) where
-    T: TimelyTimestamp + Lattice + Codec64 + Sync,
+    T: TimelyTimestamp + TotalOrder + Lattice + Codec64 + Sync,
 {
     if read_only {
         info!("disabling shard finalization in read only mode");

src/storage-controller/src/persist_handles.rs

@@ -95,7 +95,7 @@ impl<T: Timestamp + Lattice + Codec64> PersistTableWriteCmd<T> {
     }
 }
 
-async fn append_work<T2: Timestamp + Lattice + Codec64 + Sync>(
+async fn append_work<T2: Timestamp + TotalOrder + Lattice + Codec64 + Sync>(
     write_handles: &mut BTreeMap<GlobalId, WriteHandle<SourceData, (), T2, StorageDiff>>,
     mut commands: BTreeMap<
         GlobalId,