perf+refactor: read-model layer — single-statement writes, static schemas, one table vocabulary

distributed_macros/src/read_model.rs

@@ -119,6 +119,7 @@ fn expand_relational_read_model(
     let mut relationships = Vec::new();
     let mut hydrate_include_arms = Vec::new();
     let mut include_rows_arms = Vec::new();
+    let mut include_schema_arms = Vec::new();
 
     for (field, attrs) in fields.iter().zip(field_attrs) {
         let ident = field
@@ -129,10 +130,11 @@ fn expand_relational_read_model(
 
         if let Some(relationship) = attrs.relationship_tokens(&field_name)? {
             relationships.push(relationship);
-            let (hydrate_arm, include_rows_arm) =
+            let (hydrate_arm, include_rows_arm, include_schema_arm) =
                 attrs.relationship_include_tokens(field, &field_name)?;
             hydrate_include_arms.push(hydrate_arm);
             include_rows_arms.push(include_rows_arm);
+            include_schema_arms.push(include_schema_arm);
             row_fields.push(quote! { #ident: ::core::default::Default::default() });
             continue;
         }
@@ -162,7 +164,7 @@ fn expand_relational_read_model(
         let jsonb = attrs.jsonb;
 
         column_defs.push(quote! {
-            distributed::ColumnDef {
+            distributed::TableColumn {
                 field_name: #field_name.to_string(),
                 column_name: #column_name.to_string(),
                 column_type: #column_type,
@@ -231,34 +233,36 @@ fn expand_relational_read_model(
 
     Ok(quote! {
         impl distributed::RelationalReadModel for #name {
-            fn schema() -> distributed::ReadModelSchema {
-                distributed::ReadModelSchema {
-                    model_name: #model_name.to_string(),
-                    table_name: #table_name.to_string(),
-                    columns: vec![#(#column_defs),*],
-                    primary_key: distributed::PrimaryKey {
-                        columns: vec![#(#primary_key_columns),*],
-                    },
-                    version_column: Some(distributed::DEFAULT_READ_MODEL_VERSION_COLUMN.to_string()),
-                    foreign_keys: vec![#(#foreign_keys),*],
-                    indexes: vec![#(#indexes),*],
-                    relationships: vec![#(#relationships),*],
-                }
+            fn schema() -> &'static distributed::TableSchema {
+                static SCHEMA: ::std::sync::LazyLock<distributed::TableSchema> =
+                    ::std::sync::LazyLock::new(|| distributed::TableSchema {
+                        model_name: #model_name.to_string(),
+                        table_name: #table_name.to_string(),
+                        columns: vec![#(#column_defs),*],
+                        primary_key: distributed::PrimaryKey {
+                            columns: vec![#(#primary_key_columns),*],
+                        },
+                        version_column: Some(distributed::DEFAULT_TABLE_VERSION_COLUMN.to_string()),
+                        foreign_keys: vec![#(#foreign_keys),*],
+                        indexes: vec![#(#indexes),*],
+                        relationships: vec![#(#relationships),*],
+                    });
+                &SCHEMA
             }
 
-            fn primary_key(&self) -> Result<distributed::RowKey, distributed::ReadModelError> {
+            fn primary_key(&self) -> Result<distributed::RowKey, distributed::TableStoreError> {
                 let mut key = distributed::RowKey::default();
                 #(#key_inserts)*
                 Ok(key)
             }
 
-            fn to_row(&self) -> Result<distributed::RowValues, distributed::ReadModelError> {
+            fn to_row(&self) -> Result<distributed::RowValues, distributed::TableStoreError> {
                 let mut row = distributed::RowValues::new();
                 #(#row_inserts)*
                 Ok(row)
             }
 
-            fn from_row(row: distributed::RowValues) -> Result<Self, distributed::ReadModelError> {
+            fn from_row(row: distributed::RowValues) -> Result<Self, distributed::TableStoreError> {
                 Ok(Self {
                     #(#row_fields),*
                 })
@@ -270,10 +274,10 @@ fn expand_relational_read_model(
                 &mut self,
                 include: &str,
                 rows: Vec<distributed::RowValues>,
-            ) -> Result<(), distributed::ReadModelError> {
+            ) -> Result<(), distributed::TableStoreError> {
                 match include {
                     #(#hydrate_include_arms,)*
-                    _ => Err(distributed::ReadModelError::Metadata(format!(
+                    _ => Err(distributed::TableStoreError::Metadata(format!(
                         "read model `{}` has no hydratable relationship `{}`",
                         #model_name,
                         include
@@ -284,10 +288,23 @@ fn expand_relational_read_model(
             fn include_rows(
                 &self,
                 include: &str,
-            ) -> Result<Vec<distributed::RowValues>, distributed::ReadModelError> {
+            ) -> Result<Vec<distributed::RowValues>, distributed::TableStoreError> {
                 match include {
                     #(#include_rows_arms,)*
-                    _ => Err(distributed::ReadModelError::Metadata(format!(
+                    _ => Err(distributed::TableStoreError::Metadata(format!(
+                        "read model `{}` has no tracked relationship `{}`",
+                        #model_name,
+                        include
+                    ))),
+                }
+            }
+
+            fn include_target_schema(
+                include: &str,
+            ) -> Result<&'static distributed::TableSchema, distributed::TableStoreError> {
+                match include {
+                    #(#include_schema_arms,)*
+                    _ => Err(distributed::TableStoreError::Metadata(format!(
                         "read model `{}` has no tracked relationship `{}`",
                         #model_name,
                         include
@@ -356,7 +373,7 @@ fn index_def_tokens(
         .collect::<Vec<_>>();
 
     quote! {
-        distributed::IndexDef {
+        distributed::TableIndex {
             name: Some(#index_name.to_string()),
             columns: vec![#(#columns),*],
             unique: #unique,
@@ -661,7 +678,11 @@ impl FieldAttrs {
         &self,
         field: &Field,
         field_name: &str,
-    ) -> syn::Result<(proc_macro2::TokenStream, proc_macro2::TokenStream)> {
+    ) -> syn::Result<(
+        proc_macro2::TokenStream,
+        proc_macro2::TokenStream,
+        proc_macro2::TokenStream,
+    )> {
         let relationship = self.relationship.as_ref().ok_or_else(|| {
             syn::Error::new_spanned(field, "field is not a read-model relationship")
         })?;
@@ -689,7 +710,7 @@ impl FieldAttrs {
                         self.#ident = rows
                             .into_iter()
                             .map(<#inner as distributed::RelationalReadModel>::from_row)
-                            .collect::<Result<Vec<_>, distributed::ReadModelError>>()?;
+                            .collect::<Result<Vec<_>, distributed::TableStoreError>>()?;
                         Ok(())
                     }
                 };
@@ -698,9 +719,12 @@ impl FieldAttrs {
                         .#ident
                         .iter()
                         .map(distributed::RelationalReadModel::to_row)
-                        .collect::<Result<Vec<_>, distributed::ReadModelError>>()
+                        .collect::<Result<Vec<_>, distributed::TableStoreError>>()
+                };
+                let include_schema = quote! {
+                    #field_name => Ok(<#inner as distributed::RelationalReadModel>::schema())
                 };
-                Ok((hydrate, include_rows))
+                Ok((hydrate, include_rows, include_schema))
             }
             RelationshipKindAttr::BelongsTo => {
                 let inner = option_inner_type(&field.ty).ok_or_else(|| {
@@ -725,7 +749,7 @@ impl FieldAttrs {
                             None => None,
                         };
                         if rows.next().is_some() {
-                            return Err(distributed::ReadModelError::Metadata(format!(
+                            return Err(distributed::TableStoreError::Metadata(format!(
                                 "belongs_to relationship `{}` returned more than one row",
                                 #field_name
                             )));
@@ -742,7 +766,10 @@ impl FieldAttrs {
                         Ok(rows)
                     }
                 };
-                Ok((hydrate, include_rows))
+                let include_schema = quote! {
+                    #field_name => Ok(<#inner as distributed::RelationalReadModel>::schema())
+                };
+                Ok((hydrate, include_rows, include_schema))
             }
         }
     }

src/commit_builder/mod.rs

@@ -27,7 +27,8 @@
 use crate::aggregate::Aggregate;
 use crate::entity::Entity;
 use crate::outbox::OutboxMessage;
-use crate::read_model::{ReadModelWritePlan, ReadModelWritePlanBuilder};
+use crate::read_model::ReadModelWritePlanBuilder;
+use crate::table::TableWritePlan;
 use crate::repository::{
     CommitBatch, RepositoryError, StreamIdentity, StreamWrite, TransactionalCommit,
 };
@@ -97,7 +98,7 @@ pub struct CommitBuilder<'a, R> {
     streams: Vec<StreamWrite<'a>>,
     outbox_messages: Vec<OutboxMessage>,
     outbox_source: StagedOutboxSource,
-    read_model_plans: Vec<ReadModelWritePlan>,
+    read_model_plans: Vec<TableWritePlan>,
     error: Option<RepositoryError>,
 }
 

src/hashmap_repo/repository.rs

@@ -11,8 +11,7 @@ use crate::entity::{Entity, EventRecord};
 use crate::outbox::OutboxMessage;
 use crate::read_model::in_memory::apply_read_model_write_plan;
 use crate::read_model::{
-    InMemoryReadModelStore, ReadModelAdapterCapabilities, ReadModelCommitOutcome, ReadModelError,
-    ReadModelLoadGraph, ReadModelLoadRequest, ReadModelQueryCapabilities, ReadModelWritePlan,
+    InMemoryReadModelStore, ReadModelLoadGraph, ReadModelLoadRequest, ReadModelQueryCapabilities,
 };
 use crate::repository::{
     reject_duplicate_outbox_messages, reject_duplicate_streams,
@@ -22,6 +21,7 @@ use crate::repository::{
     TransactionalCommit,
 };
 use crate::snapshot::{InMemorySnapshotStore, SnapshotRecord};
+use crate::table::{TableAdapterCapabilities, TableCommitOutcome, TableStoreError, TableWritePlan};
 
 /// In-memory repository implementation using HashMap.
 ///
@@ -344,14 +344,14 @@ fn stored_stream_version(events: Option<&Vec<EventRecord>>) -> u64 {
 }
 
 impl ReadModelWritePlanStore for HashMapRepository {
-    fn read_model_capabilities(&self) -> ReadModelAdapterCapabilities {
+    fn read_model_capabilities(&self) -> TableAdapterCapabilities {
         self.model_store.read_model_capabilities()
     }
 
     fn commit_write_plan(
         &self,
-        plan: ReadModelWritePlan,
-    ) -> impl Future<Output = Result<ReadModelCommitOutcome, ReadModelError>> + Send + '_ {
+        plan: TableWritePlan,
+    ) -> impl Future<Output = Result<TableCommitOutcome, TableStoreError>> + Send + '_ {
         self.model_store.commit_write_plan(plan)
     }
 }
@@ -364,7 +364,7 @@ impl RelationalReadModelQueryStore for HashMapRepository {
     fn load_graph(
         &self,
         request: ReadModelLoadRequest,
-    ) -> impl Future<Output = Result<ReadModelLoadGraph, ReadModelError>> + Send + '_ {
+    ) -> impl Future<Output = Result<ReadModelLoadGraph, TableStoreError>> + Send + '_ {
         self.model_store.load_graph(request)
     }
 }

src/lib.rs

@@ -94,29 +94,26 @@ pub use queued_repo::{
 
 // Read models: projections and read-optimized views
 pub use read_model::{
-    ColumnDef, ColumnType, DeleteRowMutation, ExpectedVersion, ForeignKey, InMemoryReadModelStore,
-    IndexDef, PatchMode, PatchRowMutation, PrimaryKey, ReadModel, ReadModelAdapterCapabilities,
-    ReadModelCommitOutcome, ReadModelError, ReadModelIncludeRows, ReadModelLoadBuilder,
-    ReadModelLoadGraph, ReadModelLoadRequest, ReadModelMigrationArtifact, ReadModelMutation,
-    ReadModelQueryCapabilities, ReadModelSchema, ReadModelSchemaAdapter,
-    ReadModelSchemaAdapterCapabilities, ReadModelSchemaBootstrap, ReadModelSchemaIssue,
-    ReadModelSchemaIssueKind, ReadModelSchemaRegistry, ReadModelSchemaVerification,
-    ReadModelWorkspace, ReadModelWorkspaceExt, ReadModelWritePlan, ReadModelWritePlanBuilder,
-    RelationalReadModel, RelationalReadModelIncludes, RelationshipDef, RelationshipKind, RowKey,
-    RowMutation, RowPatch, RowValue, RowValues, RowWriteMode, Versioned,
-    DEFAULT_READ_MODEL_VERSION_COLUMN,
+    InMemoryReadModelStore, ReadModel, ReadModelIncludeRows, ReadModelLoadBuilder,
+    ReadModelLoadGraph, ReadModelLoadRequest, ReadModelQueryCapabilities, ReadModelWorkspace,
+    ReadModelWorkspaceExt, ReadModelWritePlanBuilder, RelationalReadModel,
+    RelationalReadModelIncludes, Versioned,
 };
 
-// Neutral table/row primitives shared by read models and operational tables stay
-// reachable under their module path (`distributed::table::*`). They are low-level
-// schema/adapter plumbing, not part of the quick-start surface, so they are not
-// re-exported at the crate root.
-//
-// Exception: `TableSchemaRegistry` is the entry point for registering operational
-// table schemas (it is what callers build before bootstrapping migrations), so it
-// is consumed directly by downstream users and integration tests. It stays at the
-// crate root as part of the public surface.
-pub use table::TableSchemaRegistry;
+// Neutral table/row primitives: the canonical schema, row, mutation, write-plan,
+// and error vocabulary shared by read models and operational tables (outbox,
+// inbox/checkpoint, and future operational tables). Read models build on these,
+// so they are part of the crate-root surface; SQL rendering helpers stay under
+// `distributed::table::*`.
+pub use table::{
+    ColumnType, DeleteTableRowMutation, ExpectedVersion, ForeignKey, PatchMode,
+    PatchTableRowMutation, PrimaryKey, RelationshipDef, RelationshipKind, RowKey, RowPatch,
+    RowValue, RowValues, RowWriteMode, TableAdapterCapabilities, TableColumn, TableCommitOutcome,
+    TableIndex, TableMigrationArtifact, TableModel, TableMutation, TableRowMutation, TableSchema,
+    TableSchemaAdapter, TableSchemaAdapterCapabilities, TableSchemaBootstrap, TableSchemaIssue,
+    TableSchemaIssueKind, TableSchemaRegistry, TableSchemaRegistryExt, TableSchemaVerification,
+    TableStoreError, TableWritePlan, DEFAULT_TABLE_VERSION_COLUMN,
+};
 
 pub use manifest::{
     DistributedManifestEnvelope, DistributedProjectManifest, MessageEndpointManifest,

src/manifest.rs

@@ -4,7 +4,7 @@ use crate::table::{
     generate_table_migration_artifacts, table_schema_statements, TableSchema, TableSchemaRegistry,
     TableSqlDialect,
 };
-use crate::{ReadModelError, ReadModelMigrationArtifact, RelationalReadModel};
+use crate::{RelationalReadModel, TableMigrationArtifact, TableStoreError};
 
 pub const DISTRIBUTED_MANIFEST_SCHEMA_VERSION: u32 = 1;
 
@@ -48,19 +48,19 @@ impl DistributedProjectManifest {
         self
     }
 
-    pub fn try_read_model<M>(mut self) -> Result<Self, ReadModelError>
+    pub fn try_read_model<M>(mut self) -> Result<Self, TableStoreError>
     where
         M: RelationalReadModel,
     {
         self.try_register_read_model::<M>()?;
         Ok(self)
     }
 
-    pub fn try_register_read_model<M>(&mut self) -> Result<&mut Self, ReadModelError>
+    pub fn try_register_read_model<M>(&mut self) -> Result<&mut Self, TableStoreError>
     where
         M: RelationalReadModel,
     {
-        self.try_register_table_schema(M::schema())
+        self.try_register_table_schema(M::schema().clone())
     }
 
     pub fn table_schema(mut self, schema: TableSchema) -> Self {
@@ -69,15 +69,15 @@ impl DistributedProjectManifest {
         self
     }
 
-    pub fn try_table_schema(mut self, schema: TableSchema) -> Result<Self, ReadModelError> {
+    pub fn try_table_schema(mut self, schema: TableSchema) -> Result<Self, TableStoreError> {
         self.try_register_table_schema(schema)?;
         Ok(self)
     }
 
     pub fn try_register_table_schema(
         &mut self,
         schema: TableSchema,
-    ) -> Result<&mut Self, ReadModelError> {
+    ) -> Result<&mut Self, TableStoreError> {
         let mut registry = self.table_registry()?;
         registry.register_schema(schema.clone())?;
         self.tables.push(schema);
@@ -89,22 +89,22 @@ impl DistributedProjectManifest {
         self
     }
 
-    pub fn table_registry(&self) -> Result<TableSchemaRegistry, ReadModelError> {
+    pub fn table_registry(&self) -> Result<TableSchemaRegistry, TableStoreError> {
         let mut registry = TableSchemaRegistry::new();
         for schema in &self.tables {
             registry.register_schema(schema.clone())?;
         }
         Ok(registry)
     }
 
-    pub fn sql_statements(&self, dialect: TableSqlDialect) -> Result<Vec<String>, ReadModelError> {
+    pub fn sql_statements(&self, dialect: TableSqlDialect) -> Result<Vec<String>, TableStoreError> {
         table_schema_statements(&self.table_registry()?, dialect)
     }
 
     pub fn sql_migration_artifacts(
         &self,
         dialect: TableSqlDialect,
-    ) -> Result<Vec<ReadModelMigrationArtifact>, ReadModelError> {
+    ) -> Result<Vec<TableMigrationArtifact>, TableStoreError> {
         generate_table_migration_artifacts(&self.table_registry()?, dialect)
     }
 
@@ -192,7 +192,7 @@ mod tests {
     fn manifest_collects_schema_service_metadata_and_renders_sql() {
         let manifest = DistributedProjectManifest::new("checkout")
             .read_model::<OrderView>()
-            .table_schema(outbox_message_schema())
+            .table_schema(outbox_message_schema().clone())
             .service(
                 ServiceManifest::new("checkout-saga")
                     .command("checkout.start")