Introduce overflowing diff type.

The change introduced an overflowing type, which overflows in release
builds, too. It implements all relevant traits to be usable where we
would use a number normally, including differential and a few other
traits defined by external crates.

It uses the overflowing type in all parts of Materialize where possible.
Persist still requires the old plain `i64` type, so we introduce a
`StorageDiff` type alias, and at the boundary to persist convert to/from
the overflowing diff type.

This isn't ideal because storage can overflow, too, on consolidation,
but changing the type there seems harder and is out of my expertise.

The diff is quite large to account for boilerplate changes, everywhere
where one would write 1 previously, one now has to convert type or
mention `Diff::ONE` directly.

Signed-off-by: Moritz Hoffmann <mh@materialize.com>

Author: antiguru

Cargo.lock

@@ -22,7 +22,7 @@ use mz_expr::compare_columns;
 use mz_ore::cast::CastFrom;
 use mz_ore::now::EpochMillis;
 use mz_repr::adt::numeric;
-use mz_repr::{CatalogItemId, Datum, GlobalId, IntoRowIterator, Row, Timestamp};
+use mz_repr::{CatalogItemId, Datum, Diff, GlobalId, IntoRowIterator, Row, Timestamp};
 use mz_sql::plan::SubscribeOutput;
 use mz_storage_types::instances::StorageInstanceId;
 use timely::progress::Antichain;
@@ -182,10 +182,10 @@ impl ActiveSubscribe {
                     |(left_time, left_row, left_diff), (right_time, right_row, right_diff)| {
                         left_time.cmp(right_time).then_with(|| {
                             let mut left_datums = left_datum_vec.borrow();
-                            left_datums.extend(&[Datum::Int64(*left_diff)]);
+                            left_datums.extend(&[Datum::Int64(**left_diff)]);
                             left_datums.extend(left_row.iter());
                             let mut right_datums = right_datum_vec.borrow();
-                            right_datums.extend(&[Datum::Int64(*right_diff)]);
+                            right_datums.extend(&[Datum::Int64(**right_diff)]);
                             right_datums.extend(right_row.iter());
                             compare_columns(order_by, &left_datums, &right_datums, || {
                                 left_row.cmp(right_row).then(left_diff.cmp(right_diff))
@@ -238,7 +238,7 @@ impl ActiveSubscribe {
                     let value_columns = self.arity - order_by_keys.len();
                     let mut packer = row_buf.packer();
                     new_rows.push(match &group[..] {
-                        [(_, row, 1)] => {
+                        [(_, row, Diff::ONE)] => {
                             packer.push(if debezium {
                                 Datum::String("insert")
                             } else {
@@ -258,9 +258,9 @@ impl ActiveSubscribe {
                                     packer.push(datums[idx]);
                                 }
                             }
-                            (start.0, row_buf.clone(), 0)
+                            (start.0, row_buf.clone(), Diff::ZERO)
                         }
-                        [(_, _, -1)] => {
+                        [(_, _, Diff::MINUS_ONE)] => {
                             packer.push(Datum::String("delete"));
                             let datums = datum_vec.borrow_with(&start.1);
                             for column_order in order_by_keys {
@@ -276,9 +276,9 @@ impl ActiveSubscribe {
                             for _ in 0..self.arity - order_by_keys.len() {
                                 packer.push(Datum::Null);
                             }
-                            (start.0, row_buf.clone(), 0)
+                            (start.0, row_buf.clone(), Diff::ZERO)
                         }
-                        [(_, old_row, -1), (_, row, 1)] => {
+                        [(_, old_row, Diff::MINUS_ONE), (_, row, Diff::ONE)] => {
                             packer.push(Datum::String("upsert"));
                             let datums = datum_vec.borrow_with(row);
                             let old_datums = old_datum_vec.borrow_with(old_row);
@@ -298,7 +298,7 @@ impl ActiveSubscribe {
                                     packer.push(datums[idx]);
                                 }
                             }
-                            (start.0, row_buf.clone(), 0)
+                            (start.0, row_buf.clone(), Diff::ZERO)
                         }
                         _ => {
                             packer.push(Datum::String("key_violation"));
@@ -314,7 +314,7 @@ impl ActiveSubscribe {
                             for _ in 0..(self.arity - order_by_keys.len()) {
                                 packer.push(Datum::Null);
                             }
-                            (start.0, row_buf.clone(), 0)
+                            (start.0, row_buf.clone(), Diff::ZERO)
                         }
                     });
                 }
@@ -342,7 +342,7 @@ impl ActiveSubscribe {
                     SubscribeOutput::EnvelopeUpsert { .. }
                     | SubscribeOutput::EnvelopeDebezium { .. } => {}
                     SubscribeOutput::Diffs | SubscribeOutput::WithinTimestampOrderBy { .. } => {
-                        packer.push(Datum::Int64(diff));
+                        packer.push(Datum::Int64(*diff));
                     }
                 }
 

src/adapter/src/active_compute_sink.rs

@@ -42,7 +42,7 @@ use mz_ore::{instrument, soft_assert_no_log};
 use mz_pgrepr::oid::INVALID_OID;
 use mz_repr::adt::mz_acl_item::{MzAclItem, PrivilegeMap};
 use mz_repr::role_id::RoleId;
-use mz_repr::{CatalogItemId, GlobalId, RelationVersion, Timestamp, VersionedRelationDesc};
+use mz_repr::{CatalogItemId, Diff, GlobalId, RelationVersion, Timestamp, VersionedRelationDesc};
 use mz_sql::catalog::CatalogError as SqlCatalogError;
 use mz_sql::catalog::{CatalogItem as SqlCatalogItem, CatalogItemType, CatalogSchema, CatalogType};
 use mz_sql::names::{
@@ -1319,7 +1319,7 @@ impl CatalogState {
             // catalog.
             let item_id = entry.id();
             state.insert_entry(entry);
-            builtin_table_updates.extend(state.pack_item_update(item_id, 1));
+            builtin_table_updates.extend(state.pack_item_update(item_id, Diff::ONE));
         }
 
         let mut handles = Vec::new();
@@ -1523,7 +1523,7 @@ impl CatalogState {
         builtin_table_updates.extend(
             item_ids
                 .into_iter()
-                .flat_map(|id| state.pack_item_update(id, 1)),
+                .flat_map(|id| state.pack_item_update(id, Diff::ONE)),
         );
 
         builtin_table_updates

src/adapter/src/catalog/apply.rs

@@ -1960,7 +1960,7 @@ impl CatalogState {
             Datum::Int32(ip.prefix_len().into()),
             Datum::String(&format!("{}/{}", addr, ip.prefix_len())),
         ]);
-        Ok(BuiltinTableUpdate::row(id, row, 1))
+        Ok(BuiltinTableUpdate::row(id, row, Diff::ONE))
     }
 
     pub fn pack_replica_metric_updates(
@@ -2034,7 +2034,7 @@ impl CatalogState {
                         disk_bytes.into(),
                         (*credits_per_hour).into(),
                     ]);
-                    BuiltinTableUpdate::row(id, row, 1)
+                    BuiltinTableUpdate::row(id, row, Diff::ONE)
                 },
             )
             .collect();

src/adapter/src/catalog/builtin_table_updates.rs

@@ -16,7 +16,7 @@ use mz_catalog::memory::objects::{BootstrapStateUpdateKind, StateUpdate};
 use mz_ore::collections::CollectionExt;
 use mz_ore::now::NowFn;
 use mz_persist_types::ShardId;
-use mz_repr::{CatalogItemId, Timestamp};
+use mz_repr::{CatalogItemId, Diff, Timestamp};
 use mz_sql::ast::display::AstDisplay;
 use mz_sql::ast::CreateSinkOptionName;
 use mz_sql::names::FullItemName;
@@ -83,7 +83,7 @@ where
 
 pub(crate) struct MigrateResult {
     pub(crate) builtin_table_updates: Vec<BuiltinTableUpdate<&'static BuiltinTable>>,
-    pub(crate) post_item_updates: Vec<(BootstrapStateUpdateKind, Timestamp, i64)>,
+    pub(crate) post_item_updates: Vec<(BootstrapStateUpdateKind, Timestamp, Diff)>,
 }
 
 /// Migrates all user items and loads them into `state`.

src/adapter/src/catalog/migrate.rs

@@ -253,7 +253,7 @@ impl Catalog {
         let mut updates = into_consolidatable_updates_startup(updates, config.boot_ts);
         differential_dataflow::consolidation::consolidate_updates(&mut updates);
         soft_assert_no_log!(
-            updates.iter().all(|(_, _, diff)| *diff == 1),
+            updates.iter().all(|(_, _, diff)| **diff == 1),
             "consolidated updates should be positive during startup: {updates:?}"
         );
 
@@ -530,7 +530,7 @@ impl Catalog {
                     mz_sql::func::Func::Scalar(impls) => {
                         for imp in impls {
                             builtin_table_updates.push(catalog.state.resolve_builtin_table_update(
-                                catalog.state.pack_op_update(op, imp.details(), 1),
+                                catalog.state.pack_op_update(op, imp.details(), Diff::ONE),
                             ));
                         }
                     }

src/adapter/src/catalog/open.rs

@@ -35,6 +35,7 @@ use mz_persist_types::ShardId;
 use mz_repr::{CatalogItemId, Diff, GlobalId, Timestamp};
 use mz_sql::catalog::CatalogItem as _;
 use mz_storage_client::controller::StorageTxn;
+use mz_storage_types::StorageDiff;
 use timely::progress::{Antichain, Timestamp as TimelyTimestamp};
 use tracing::{debug, error};
 
@@ -199,19 +200,20 @@ async fn migrate_builtin_items_0dt(
             "builtin table migration shard for org {organization_id:?} version {build_version:?}"
         ),
     };
-    let mut since_handle: SinceHandle<TableKey, ShardId, Timestamp, Diff, i64> = persist_client
-        .open_critical_since(
-            shard_id,
-            // TODO: We may need to use a different critical reader
-            // id for this if we want to be able to introspect it via SQL.
-            PersistClient::CONTROLLER_CRITICAL_SINCE,
-            diagnostics.clone(),
-        )
-        .await
-        .expect("invalid usage");
+    let mut since_handle: SinceHandle<TableKey, ShardId, Timestamp, StorageDiff, i64> =
+        persist_client
+            .open_critical_since(
+                shard_id,
+                // TODO: We may need to use a different critical reader
+                // id for this if we want to be able to introspect it via SQL.
+                PersistClient::CONTROLLER_CRITICAL_SINCE,
+                diagnostics.clone(),
+            )
+            .await
+            .expect("invalid usage");
     let (mut write_handle, mut read_handle): (
-        WriteHandle<TableKey, ShardId, Timestamp, Diff>,
-        ReadHandle<TableKey, ShardId, Timestamp, Diff>,
+        WriteHandle<TableKey, ShardId, Timestamp, StorageDiff>,
+        ReadHandle<TableKey, ShardId, Timestamp, StorageDiff>,
     ) = persist_client
         .open(
             shard_id,
@@ -223,7 +225,7 @@ async fn migrate_builtin_items_0dt(
         .await
         .expect("invalid usage");
     // Commit an empty write at the minimum timestamp so the shard is always readable.
-    const EMPTY_UPDATES: &[((TableKey, ShardId), Timestamp, Diff)] = &[];
+    const EMPTY_UPDATES: &[((TableKey, ShardId), Timestamp, StorageDiff)] = &[];
     let res = write_handle
         .compare_and_append(
             EMPTY_UPDATES,
@@ -323,7 +325,7 @@ async fn migrate_builtin_items_0dt(
             if storage_collection_metadata.get(&GlobalId::System(table_key.global_id))
                 == Some(&shard_id)
             {
-                migrated_shard_updates.push(((table_key, shard_id.clone()), upper, -1));
+                migrated_shard_updates.push(((table_key, shard_id.clone()), upper, -Diff::ONE));
             } else {
                 migration_shards_to_finalize.insert((table_key, shard_id));
             }
@@ -343,7 +345,7 @@ async fn migrate_builtin_items_0dt(
                 global_id,
                 build_version: build_version.clone(),
             };
-            migrated_shard_updates.push(((table_key, shard_id), upper, 1));
+            migrated_shard_updates.push(((table_key, shard_id), upper, Diff::ONE));
         }
     }
 
@@ -412,7 +414,7 @@ async fn migrate_builtin_items_0dt(
         if !read_only {
             let updates: Vec<_> = migration_shards_to_finalize
                 .into_iter()
-                .map(|(table_key, shard_id)| ((table_key, shard_id), upper, -1))
+                .map(|(table_key, shard_id)| ((table_key, shard_id), upper, -Diff::ONE))
                 .collect();
             if !updates.is_empty() {
                 // Ignore any errors, these shards will get cleaned up in the next upgrade.
@@ -437,7 +439,7 @@ async fn migrate_builtin_items_0dt(
 }
 
 async fn fetch_upper(
-    write_handle: &mut WriteHandle<TableKey, ShardId, Timestamp, Diff>,
+    write_handle: &mut WriteHandle<TableKey, ShardId, Timestamp, StorageDiff>,
 ) -> Timestamp {
     write_handle
         .fetch_recent_upper()
@@ -450,8 +452,8 @@ async fn fetch_upper(
 async fn write_to_migration_shard(
     updates: Vec<((TableKey, ShardId), Timestamp, Diff)>,
     upper: Timestamp,
-    write_handle: &mut WriteHandle<TableKey, ShardId, Timestamp, Diff>,
-    since_handle: &mut SinceHandle<TableKey, ShardId, Timestamp, Diff, i64>,
+    write_handle: &mut WriteHandle<TableKey, ShardId, Timestamp, StorageDiff>,
+    since_handle: &mut SinceHandle<TableKey, ShardId, Timestamp, StorageDiff, i64>,
 ) -> Result<Timestamp, Error> {
     let next_upper = upper.step_forward();
     // Lag the shard's upper by 1 to keep it readable.

src/adapter/src/catalog/open/builtin_item_migration.rs

@@ -41,7 +41,7 @@ use mz_persist_types::ShardId;
 use mz_repr::adt::mz_acl_item::{merge_mz_acl_items, AclMode, MzAclItem, PrivilegeMap};
 use mz_repr::network_policy_id::NetworkPolicyId;
 use mz_repr::role_id::RoleId;
-use mz_repr::{strconv, CatalogItemId, ColumnName, ColumnType, GlobalId};
+use mz_repr::{strconv, CatalogItemId, ColumnName, ColumnType, Diff, GlobalId};
 use mz_sql::ast::RawDataType;
 use mz_sql::catalog::{
     CatalogDatabase, CatalogError as SqlCatalogError, CatalogItem as SqlCatalogItem, CatalogRole,
@@ -462,7 +462,7 @@ impl Catalog {
             self.state().pack_optimizer_notices(
                 &mut builtin_table_updates,
                 dropped_notices.iter(),
-                -1,
+                -Diff::ONE,
             );
         }
 
@@ -2347,7 +2347,7 @@ impl Catalog {
                 let id = tx.allocate_storage_usage_ids()?;
                 let metric =
                     VersionedStorageUsage::new(id, object_id, size_bytes, collection_timestamp);
-                let builtin_table_update = state.pack_storage_usage_update(metric, 1);
+                let builtin_table_update = state.pack_storage_usage_update(metric, Diff::ONE);
                 let builtin_table_update = state.resolve_builtin_table_update(builtin_table_update);
                 weird_builtin_table_update = Some(builtin_table_update);
             }

src/adapter/src/catalog/transact.rs

@@ -131,7 +131,7 @@ use mz_repr::explain::{ExplainConfig, ExplainFormat};
 use mz_repr::global_id::TransientIdGen;
 use mz_repr::optimize::OptimizerFeatures;
 use mz_repr::role_id::RoleId;
-use mz_repr::{CatalogItemId, GlobalId, RelationDesc, Timestamp};
+use mz_repr::{CatalogItemId, Diff, GlobalId, RelationDesc, Timestamp};
 use mz_secrets::cache::CachingSecretsReader;
 use mz_secrets::{SecretsController, SecretsReader};
 use mz_sql::ast::{Raw, Statement};
@@ -1843,10 +1843,13 @@ impl Coordinator {
         for replica_statuses in self.cluster_replica_statuses.0.values() {
             for (replica_id, processes_statuses) in replica_statuses {
                 for (process_id, status) in processes_statuses {
-                    let builtin_table_update = self
-                        .catalog()
-                        .state()
-                        .pack_cluster_replica_status_update(*replica_id, *process_id, status, 1);
+                    let builtin_table_update =
+                        self.catalog().state().pack_cluster_replica_status_update(
+                            *replica_id,
+                            *process_id,
+                            status,
+                            Diff::ONE,
+                        );
                     let builtin_table_update = self
                         .catalog()
                         .state()
@@ -2018,7 +2021,7 @@ impl Coordinator {
                             self.catalog().state().pack_optimizer_notices(
                                 &mut builtin_table_updates,
                                 df_meta.optimizer_notices.iter(),
-                                1,
+                                Diff::ONE,
                             );
                         }
 
@@ -2074,7 +2077,7 @@ impl Coordinator {
                         self.catalog().state().pack_optimizer_notices(
                             &mut builtin_table_updates,
                             df_meta.optimizer_notices.iter(),
-                            1,
+                            Diff::ONE,
                         );
                     }
 
@@ -2125,7 +2128,7 @@ impl Coordinator {
                         self.catalog().state().pack_optimizer_notices(
                             &mut builtin_table_updates,
                             df_meta.optimizer_notices.iter(),
-                            1,
+                            Diff::ONE,
                         );
                     }
 
@@ -3807,7 +3810,7 @@ impl Coordinator {
                     .expect("all collections happen after Jan 1 1970");
                 if collection_timestamp < cutoff_ts {
                     debug!("pruning storage event {row:?}");
-                    let builtin_update = BuiltinTableUpdate::row(item_id, row, -1);
+                    let builtin_update = BuiltinTableUpdate::row(item_id, row, -Diff::ONE);
                     expired.push(builtin_update);
                 }
             }