fix plan caching for client-specific views (#3672)

# Description of Changes

Fixes a subscription plan caching bug related to client-specific views.

Before this fix, you could define a client-specific view:
```rust
fn my_view(ctx: &ViewContext) -> Option<Player> {
    ctx.db.player().identity().find(ctx.sender)
}
```

And subscribe to it as follows:
```sql
SELECT * FROM my_view
```

Note this view is implicitly parameterized by `:sender`, however when
generating a query hash for this subscription, this fact would not be
taken into account which would result in this query being cached and
reused for all callers.

After this fix, a query hash is generated for this subscription as
though it were given as:
```sql
SELECT * FROM my_view WHERE identity = :sender
```

# API and ABI breaking changes

None

**Note for CLI code owners:**

I had to touch the `subscribe` cli command file. No updates to the api.
It just needed to be updated to look for views in the module def.

# Expected complexity level and risk

1

# Testing

- [x] Added a regression smoketest

Author: joshua-spacetime

crates/cli/src/subcommands/subscribe.rs

@@ -86,12 +86,11 @@ fn reformat_update<'a>(
     msg.tables
         .iter()
         .map(|upd| {
-            let table_schema = schema
-                .tables
-                .iter()
-                .find(|tbl| tbl.name == upd.table_name)
-                .context("table not found in schema")?;
-            let table_ty = schema.typespace.resolve(table_schema.product_type_ref);
+            let table_ty = schema.typespace.resolve(
+                schema
+                    .type_ref_for_table_like(&upd.table_name)
+                    .context("table not found in schema")?,
+            );
 
             let reformat_row = |row: &str| -> anyhow::Result<Value> {
                 // TODO: can the following two calls be merged into a single call to reduce allocations?

crates/lib/src/db/raw_def/v9.rs

@@ -97,6 +97,50 @@ pub struct RawModuleDefV9 {
     pub row_level_security: Vec<RawRowLevelSecurityDefV9>,
 }
 
+impl RawModuleDefV9 {
+    /// Find a [`RawTableDefV9`] by name in this raw module def
+    fn find_table_def(&self, table_name: &str) -> Option<&RawTableDefV9> {
+        self.tables
+            .iter()
+            .find(|table_def| table_def.name.as_ref() == table_name)
+    }
+
+    /// Find a [`RawViewDefV9`] by name in this raw module def
+    fn find_view_def(&self, view_name: &str) -> Option<&RawViewDefV9> {
+        self.misc_exports.iter().find_map(|misc_export| match misc_export {
+            RawMiscModuleExportV9::View(view_def) if view_def.name.as_ref() == view_name => Some(view_def),
+            _ => None,
+        })
+    }
+
+    /// Find and return the product type ref for a table in this module def
+    fn type_ref_for_table(&self, table_name: &str) -> Option<AlgebraicTypeRef> {
+        self.find_table_def(table_name)
+            .map(|table_def| table_def.product_type_ref)
+    }
+
+    /// Find and return the product type ref for a view in this module def
+    fn type_ref_for_view(&self, view_name: &str) -> Option<AlgebraicTypeRef> {
+        self.find_view_def(view_name)
+            .map(|view_def| &view_def.return_type)
+            .and_then(|return_type| {
+                return_type
+                    .as_option()
+                    .and_then(|inner| inner.clone().into_ref().ok())
+                    .or_else(|| {
+                        return_type
+                            .as_array()
+                            .and_then(|inner| inner.elem_ty.clone().into_ref().ok())
+                    })
+            })
+    }
+
+    /// Find and return the product type ref for a table or view in this module def
+    pub fn type_ref_for_table_like(&self, name: &str) -> Option<AlgebraicTypeRef> {
+        self.type_ref_for_table(name).or_else(|| self.type_ref_for_view(name))
+    }
+}
+
 /// The definition of a database table.
 ///
 /// This struct holds information about the table, including its name, columns, indexes,

crates/physical-plan/src/plan.rs

@@ -115,6 +115,13 @@ impl ProjectPlan {
     pub fn returns_view_table(&self) -> bool {
         self.return_table().is_some_and(|schema| schema.is_view())
     }
+
+    /// Does this plan read from an (anonymous) view?
+    pub fn reads_from_view(&self, anonymous: bool) -> bool {
+        match self {
+            Self::None(plan) | Self::Name(plan, ..) => plan.reads_from_view(anonymous),
+        }
+    }
 }
 
 /// Physical plans always terminate with a projection.
@@ -213,6 +220,15 @@ impl ProjectListPlan {
     pub fn returns_view_table(&self) -> bool {
         self.return_table().is_some_and(|schema| schema.is_view())
     }
+
+    /// Does this plan read from an (anonymous) view?
+    pub fn reads_from_view(&self, anonymous: bool) -> bool {
+        match self {
+            Self::Limit(plan, _) => plan.reads_from_view(anonymous),
+            Self::Name(plans) => plans.iter().any(|plan| plan.reads_from_view(anonymous)),
+            Self::List(plans, ..) | Self::Agg(plans, ..) => plans.iter().any(|plan| plan.reads_from_view(anonymous)),
+        }
+    }
 }
 
 /// Query operators return tuples of rows.
@@ -1124,6 +1140,19 @@ impl PhysicalPlan {
     pub fn returns_view_table(&self) -> bool {
         self.return_table().is_some_and(|schema| schema.is_view())
     }
+
+    /// Does this plan read from an (anonymous) view?
+    pub fn reads_from_view(&self, anonymous: bool) -> bool {
+        self.any(&|plan| match plan {
+            Self::TableScan(scan, _) if anonymous => scan.schema.is_anonymous_view(),
+            Self::TableScan(scan, _) => scan.schema.is_view() && !scan.schema.is_anonymous_view(),
+            Self::IxScan(scan, _) if anonymous => scan.schema.is_anonymous_view(),
+            Self::IxScan(scan, _) => scan.schema.is_view() && !scan.schema.is_anonymous_view(),
+            Self::IxJoin(join, _) if anonymous => join.rhs.is_anonymous_view(),
+            Self::IxJoin(join, _) => join.rhs.is_view() && !join.rhs.is_anonymous_view(),
+            _ => false,
+        })
+    }
 }
 
 /// Scan a table row by row, returning row ids

crates/query/src/lib.rs

@@ -45,9 +45,14 @@ pub fn compile_subscription(
     let plan_fragments = resolve_views_for_sub(tx, plan, auth, &mut has_param)?
         .into_iter()
         .map(compile_select)
-        .collect();
+        .collect::<Vec<_>>();
 
-    Ok((plan_fragments, return_id, return_name, has_param))
+    // Does this subscription read from a client-specific view?
+    // If so, it is as if the view is parameterized by `:sender`.
+    // We must know this in order to generate the correct query hash.
+    let reads_view = plan_fragments.iter().any(|plan| plan.reads_from_view(false));
+
+    Ok((plan_fragments, return_id, return_name, has_param || reads_view))
 }
 
 /// A utility for parsing and type checking a sql statement

smoketests/tests/views.py

@@ -416,3 +416,72 @@ def test_recovery_from_trapped_views_auto_migration(self):
  id | level
 ----+-------
 """)
+
+class SubscribeViews(Smoketest):
+    MODULE_CODE = """
+use spacetimedb::{Identity, ReducerContext, Table, ViewContext};
+
+#[spacetimedb::table(name = player_state)]
+pub struct PlayerState {
+    #[primary_key]
+    identity: Identity,
+    #[unique]
+    name: String,
+}
+
+#[spacetimedb::view(name = my_player, public)]
+pub fn my_player(ctx: &ViewContext) -> Option<PlayerState> {
+    ctx.db.player_state().identity().find(ctx.sender)
+}
+
+#[spacetimedb::reducer]
+pub fn insert_player(ctx: &ReducerContext, name: String) {
+    ctx.db.player_state().insert(PlayerState { name, identity: ctx.sender });
+}
+"""
+
+    def _test_subscribing_with_different_identities(self):
+        """Tests different clients subscribing to a client-specific view"""
+
+        # Insert an identity for Alice
+        self.call("insert_player", "Alice")
+
+        # Generate and insert a new identity for Bob
+        self.reset_config()
+        self.new_identity()
+        self.call("insert_player", "Bob")
+
+        # Subscribe to `my_player` as Bob
+        sub = self.subscribe("select * from my_player", n=0)
+        events = sub()
+
+        # Project out the identity field.
+        # TODO: Eventually we should be able to do this directly in the sql.
+        # But for now we implement it in python.
+        projection = [
+            {
+                'my_player': {
+                    'deletes': [
+                        {'name': row['name']}
+                        for row in event['my_player']['deletes']
+                    ],
+                    'inserts': [
+                        {'name': row['name']}
+                        for row in event['my_player']['inserts']
+                    ],
+                }
+            }
+            for event in events
+        ]
+
+        self.assertEqual(
+            [
+                {
+                    'my_player': {
+                        'deletes': [],
+                        'inserts': [{'name': 'Bob'}],
+                    }
+                },
+            ],
+            projection,
+        )