fix subscription plan caching for client-specific views

Author: joshua-spacetime

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