upsert: refactor struct-looking enum into struct

Signed-off-by: Petros Angelatos <[email protected]>

Author: petrosagg

src/storage/src/upsert.rs

@@ -48,7 +48,7 @@ use crate::metrics::upsert::UpsertMetrics;
 use crate::storage_state::StorageInstanceContext;
 use crate::upsert_continual_feedback;
 use types::{
-    BincodeOpts, StateValue, UpsertState, UpsertStateBackend, Value, consolidating_merge_function,
+    BincodeOpts, StateValue, UpsertState, UpsertStateBackend, consolidating_merge_function,
     upsert_bincode_opts,
 };
 
@@ -607,15 +607,15 @@ async fn drain_staged_input<S, G, T, FromTime, E>(
                 if let Some(old_value) =
                     existing_value.replace(StateValue::finalized_value(value.clone()))
                 {
-                    if let Value::FinalizedValue(old_value) = old_value.into_decoded() {
+                    if let Some(old_value) = old_value.into_decoded().finalized {
                         output_updates.push((old_value, ts.clone(), Diff::MINUS_ONE));
                     }
                 }
                 output_updates.push((value, ts, Diff::ONE));
             }
             None => {
                 if let Some(old_value) = existing_value.take() {
-                    if let Value::FinalizedValue(old_value) = old_value.into_decoded() {
+                    if let Some(old_value) = old_value.into_decoded().finalized {
                         output_updates.push((old_value, ts, Diff::MINUS_ONE));
                     }
                 }

src/storage/src/upsert/memory.rs

@@ -17,6 +17,7 @@
 use std::collections::HashMap;
 
 use itertools::Itertools;
+use mz_ore::cast::CastFrom;
 
 use super::UpsertKey;
 use super::types::{
@@ -93,7 +94,7 @@ where
             stats.processed_gets += 1;
             let value = self.state.get(&key).cloned();
             let metadata = value.as_ref().map(|v| ValueMetadata {
-                size: v.memory_size(),
+                size: u64::cast_from(v.memory_size()),
                 is_tombstone: v.is_tombstone(),
             });
             stats.processed_gets_size += metadata.map_or(0, |m| m.size);