split WebsocketFormat, adding BuildableWebsocketFormat

Author: Centril

crates/client-api-messages/src/websocket.rs

@@ -98,6 +98,12 @@ pub trait WebsocketFormat: Sized {
         + Clone
         + Default;
 
+    /// The type used to encode query updates.
+    /// This type exists so that some formats, e.g., BSATN, can compress an update.
+    type QueryUpdate: SpacetimeType + for<'de> Deserialize<'de> + Serialize + Debug + Clone + Send;
+}
+
+pub trait BuildableWebsocketFormat: WebsocketFormat {
     /// The builder for [`Self::List`].
     type ListBuilder: RowListBuilder<FinishedList = Self::List>;
 
@@ -112,10 +118,6 @@ pub trait WebsocketFormat: Sized {
         (list.finish(), num_rows)
     }
 
-    /// The type used to encode query updates.
-    /// This type exists so that some formats, e.g., BSATN, can compress an update.
-    type QueryUpdate: SpacetimeType + for<'de> Deserialize<'de> + Serialize + Debug + Clone + Send;
-
     /// Convert a `QueryUpdate` into `Self::QueryUpdate`.
     /// This allows some formats to e.g., compress the update.
     fn into_query_update(qu: QueryUpdate<Self>, compression: Compression) -> Self::QueryUpdate;
@@ -779,11 +781,12 @@ pub struct JsonFormat;
 
 impl WebsocketFormat for JsonFormat {
     type Single = ByteString;
-
     type List = Vec<ByteString>;
-    type ListBuilder = Self::List;
-
     type QueryUpdate = QueryUpdate<Self>;
+}
+
+impl BuildableWebsocketFormat for JsonFormat {
+    type ListBuilder = Self::List;
 
     fn into_query_update(qu: QueryUpdate<Self>, _: Compression) -> Self::QueryUpdate {
         qu
@@ -807,11 +810,12 @@ pub struct BsatnFormat;
 
 impl WebsocketFormat for BsatnFormat {
     type Single = Box<[u8]>;
-
     type List = BsatnRowList;
-    type ListBuilder = BsatnRowListBuilder;
-
     type QueryUpdate = CompressableQueryUpdate<Self>;
+}
+
+impl BuildableWebsocketFormat for BsatnFormat {
+    type ListBuilder = BsatnRowListBuilder;
 
     fn into_query_update(qu: QueryUpdate<Self>, compression: Compression) -> Self::QueryUpdate {
         let qu_len_would_have_been = bsatn::to_len(&qu).unwrap();

crates/core/src/host/module_host.rs

@@ -25,7 +25,9 @@ use derive_more::From;
 use indexmap::IndexSet;
 use itertools::Itertools;
 use prometheus::{Histogram, IntGauge};
-use spacetimedb_client_api_messages::websocket::{ByteListLen, Compression, OneOffTable, QueryUpdate, WebsocketFormat};
+use spacetimedb_client_api_messages::websocket::{
+    BuildableWebsocketFormat, ByteListLen, Compression, OneOffTable, QueryUpdate,
+};
 use spacetimedb_data_structures::error_stream::ErrorStream;
 use spacetimedb_data_structures::map::{HashCollectionExt as _, IntMap};
 use spacetimedb_datastore::execution_context::{ExecutionContext, ReducerContext, Workload, WorkloadType};
@@ -134,7 +136,7 @@ impl UpdatesRelValue<'_> {
         !(self.deletes.is_empty() && self.inserts.is_empty())
     }
 
-    pub fn encode<F: WebsocketFormat>(&self) -> (F::QueryUpdate, u64, usize) {
+    pub fn encode<F: BuildableWebsocketFormat>(&self) -> (F::QueryUpdate, u64, usize) {
         let (deletes, nr_del) = F::encode_list(self.deletes.iter());
         let (inserts, nr_ins) = F::encode_list(self.inserts.iter());
         let num_rows = nr_del + nr_ins;
@@ -1096,7 +1098,7 @@ impl ModuleHost {
     /// This only returns an error if there is a db-level problem.
     /// An error with the query itself will be sent to the client.
     #[tracing::instrument(level = "trace", skip_all)]
-    pub async fn one_off_query<F: WebsocketFormat>(
+    pub async fn one_off_query<F: BuildableWebsocketFormat>(
         &self,
         caller_identity: Identity,
         query: String,

crates/core/src/subscription/execution_unit.rs

@@ -8,7 +8,7 @@ use crate::messages::websocket::TableUpdate;
 use crate::util::slow::SlowQueryLogger;
 use crate::vm::{build_query, TxMode};
 use spacetimedb_client_api_messages::websocket::{
-    Compression, QueryUpdate, RowListLen as _, SingleQueryUpdate, WebsocketFormat,
+    BuildableWebsocketFormat, Compression, QueryUpdate, RowListLen as _, SingleQueryUpdate,
 };
 use spacetimedb_datastore::locking_tx_datastore::TxId;
 use spacetimedb_lib::Identity;
@@ -236,7 +236,7 @@ impl ExecutionUnit {
 
     /// Evaluate this execution unit against the database using the specified format.
     #[tracing::instrument(level = "trace", skip_all)]
-    pub fn eval<F: WebsocketFormat>(
+    pub fn eval<F: BuildableWebsocketFormat>(
         &self,
         db: &RelationalDB,
         tx: &Tx,

crates/core/src/subscription/mod.rs

@@ -5,8 +5,8 @@ use module_subscription_manager::Plan;
 use prometheus::IntCounter;
 use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use spacetimedb_client_api_messages::websocket::{
-    ByteListLen, Compression, DatabaseUpdate, QueryUpdate, RowListBuilder as _, SingleQueryUpdate, TableUpdate,
-    WebsocketFormat,
+    BuildableWebsocketFormat, ByteListLen, Compression, DatabaseUpdate, QueryUpdate, RowListBuilder as _,
+    SingleQueryUpdate, TableUpdate,
 };
 use spacetimedb_execution::{pipelined::PipelinedProject, Datastore, DeltaStore};
 use spacetimedb_lib::{metrics::ExecutionMetrics, Identity};
@@ -96,7 +96,7 @@ impl MetricsRecorder for ExecutionCounters {
 pub fn execute_plan<Tx, F>(plan_fragments: &[PipelinedProject], tx: &Tx) -> Result<(F::List, u64, ExecutionMetrics)>
 where
     Tx: Datastore + DeltaStore,
-    F: WebsocketFormat,
+    F: BuildableWebsocketFormat,
 {
     let mut count = 0;
     let mut list = F::ListBuilder::default();
@@ -134,7 +134,7 @@ pub fn collect_table_update<Tx, F>(
 ) -> Result<(TableUpdate<F>, ExecutionMetrics)>
 where
     Tx: Datastore + DeltaStore,
-    F: WebsocketFormat,
+    F: BuildableWebsocketFormat,
 {
     execute_plan::<Tx, F>(plan_fragments, tx).map(|(rows, num_rows, metrics)| {
         let empty = F::List::default();
@@ -167,7 +167,7 @@ pub fn execute_plans<Tx, F>(
 ) -> Result<(DatabaseUpdate<F>, ExecutionMetrics), DBError>
 where
     Tx: Datastore + DeltaStore + Sync,
-    F: WebsocketFormat,
+    F: BuildableWebsocketFormat,
 {
     plans
         .par_iter()

crates/core/src/subscription/module_subscription_manager.rs

@@ -16,8 +16,8 @@ use hashbrown::{HashMap, HashSet};
 use parking_lot::RwLock;
 use prometheus::IntGauge;
 use spacetimedb_client_api_messages::websocket::{
-    BsatnFormat, CompressableQueryUpdate, FormatSwitch, JsonFormat, QueryId, QueryUpdate, SingleQueryUpdate,
-    WebsocketFormat,
+    BsatnFormat, BuildableWebsocketFormat, CompressableQueryUpdate, FormatSwitch, JsonFormat, QueryId, QueryUpdate,
+    SingleQueryUpdate,
 };
 use spacetimedb_data_structures::map::{Entry, IntMap};
 use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
@@ -1187,7 +1187,7 @@ impl SubscriptionManager {
                 let mut ops_bin_uncompressed: Option<(CompressableQueryUpdate<BsatnFormat>, _, _)> = None;
                 let mut ops_json: Option<(QueryUpdate<JsonFormat>, _, _)> = None;
 
-                fn memo_encode<F: WebsocketFormat>(
+                fn memo_encode<F: BuildableWebsocketFormat>(
                     updates: &UpdatesRelValue<'_>,
                     memory: &mut Option<(F::QueryUpdate, u64, usize)>,
                     metrics: &mut ExecutionMetrics,

crates/core/src/subscription/subscription.rs

@@ -31,7 +31,7 @@ use crate::sql::ast::SchemaViewer;
 use crate::vm::{build_query, TxMode};
 use anyhow::Context;
 use itertools::Either;
-use spacetimedb_client_api_messages::websocket::{Compression, WebsocketFormat};
+use spacetimedb_client_api_messages::websocket::{BuildableWebsocketFormat, Compression};
 use spacetimedb_data_structures::map::HashSet;
 use spacetimedb_datastore::locking_tx_datastore::TxId;
 use spacetimedb_lib::db::auth::{StAccess, StTableType};
@@ -512,7 +512,7 @@ pub struct ExecutionSet {
 }
 
 impl ExecutionSet {
-    pub fn eval<F: WebsocketFormat>(
+    pub fn eval<F: BuildableWebsocketFormat>(
         &self,
         db: &RelationalDB,
         tx: &Tx,