fix plan caching for client-specific views

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,79 @@ def test_recovery_from_trapped_views_auto_migration(self):
  id | level
 ----+-------
 """)
+
+class SubscribeViews(Smoketest):
+    MODULE_CODE = """
+use spacetimedb::{Identity, ReducerContext, Table, ViewContext};
+
+#[derive(Copy, Clone)]
+#[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 assertSql(self, sql, expected):
+        self.maxDiff = None
+        sql_out = self.spacetime("sql", self.database_identity, sql)
+        sql_out = "\n".join([line.rstrip() for line in sql_out.splitlines()])
+        expected = "\n".join([line.rstrip() for line in expected.splitlines()])
+
+    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")
+
+        # 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=1)
+        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,
+        )