poc: security barrier views via security levels, with property-checked ordering

src/adapter/src/catalog/migrate.rs

@@ -470,6 +470,7 @@ fn rewrite_sources_to_tables(
                         definition: ViewDefinition {
                             name: progress_name,
                             columns: vec![],
+                            with_options: vec![],
                             query: Query {
                                 ctes: CteBlock::Simple(vec![]),
                                 body: SetExpr::Table(RawItemName::Id(

src/adapter/src/catalog/state.rs

@@ -1575,6 +1575,7 @@ impl CatalogState {
                     conn_id: None,
                     resolved_ids,
                     dependencies: DependencyIds(dependencies),
+                    security_barrier: view.security_barrier,
                 })
             }
             Plan::CreateMaterializedView(CreateMaterializedViewPlan {

src/adapter/src/coord/peek.rs

@@ -340,6 +340,7 @@ impl FastPathPlan {
                     *ctx.as_mut() += 1;
                 }
                 if !filter.is_empty() {
+                    let filter = filter.iter().map(|p| p.expr.clone()).collect::<Vec<_>>();
                     let predicates = separated(" AND ", mode.seq(&filter, cols.as_ref()));
                     writeln!(f, "{}Filter {}", ctx.as_mut(), predicates)?;
                     *ctx.as_mut() += 1;
@@ -388,6 +389,7 @@ impl FastPathPlan {
                     *ctx.as_mut() += 1;
                 }
                 if !filter.is_empty() {
+                    let filter = filter.iter().map(|p| p.expr.clone()).collect::<Vec<_>>();
                     let predicates = separated(" AND ", mode.seq(&filter, cols.as_ref()));
                     writeln!(f, "{}Filter {}", ctx.as_mut(), predicates)?;
                     *ctx.as_mut() += 1;

src/adapter/src/coord/sequencer/inner/create_view.rs

@@ -391,6 +391,7 @@ impl Coordinator {
                             dependencies,
                             column_names,
                             temporary,
+                            security_barrier,
                         },
                     drop_ids,
                     if_not_exists,
@@ -425,6 +426,7 @@ impl Coordinator {
                     },
                     resolved_ids: resolved_ids.clone(),
                     dependencies: dependencies.clone(),
+                    security_barrier,
                 }),
                 owner_id: *session.current_role_id(),
             },

src/adapter/src/optimize/copy_to.rs

@@ -334,6 +334,7 @@ impl<'s> Optimize<LocalMirPlan<Resolved<'s>>> for Optimizer {
             &self.typecheck_ctx,
             &mut df_meta,
             Some(&mut self.metrics),
+            df_builder.security_barriers(),
         );
         // Run global optimization.
         mz_transform::optimize_dataflow(&mut df_desc, &mut transform_ctx, false)?;

src/adapter/src/optimize/dataflows.rs

@@ -128,6 +128,13 @@ pub struct DataflowBuilder<'a> {
     pub replan: Option<GlobalId>,
     /// A guard for recursive operations in this [`DataflowBuilder`] instance.
     recursion_guard: RecursionGuard,
+    /// Views imported so far that are declared security barriers.
+    ///
+    /// Collected during import because that is the only point at which the
+    /// optimizer has the catalog entry in hand. Handed to
+    /// [`mz_transform::TransformCtx::global`], which is what the barrier gates
+    /// read. See `doc/developer/design/20260828_security_barrier_views.md`.
+    security_barriers: BTreeSet<GlobalId>,
 }
 
 /// Behavior to prepare relation and scalar expressions for use in a dataflow.
@@ -163,7 +170,8 @@ impl ExprPrep for ExprPrepMaintained {
         expr.0.try_visit_mut_post(&mut |e| {
             // Carefully test filter expressions, which may represent temporal filters.
             if let MirRelationExpr::Filter { input, predicates } = &*e {
-                let mfp = MapFilterProject::new(input.arity()).filter(predicates.iter().cloned());
+                let mfp = MapFilterProject::new(input.arity())
+                    .filter(predicates.iter().map(|p| p.expr.clone()));
                 match mfp.into_plan() {
                     Err(e) => Err(OptimizerError::UnsupportedTemporalExpression(e)),
                     Ok(mut mfp) => {
@@ -288,9 +296,15 @@ impl<'a> DataflowBuilder<'a> {
             compute,
             replan: None,
             recursion_guard: RecursionGuard::with_limit(RECURSION_LIMIT),
+            security_barriers: BTreeSet::new(),
         }
     }
 
+    /// The security barriers among the views imported into the dataflow so far.
+    pub fn security_barriers(&self) -> &BTreeSet<GlobalId> {
+        &self.security_barriers
+    }
+
     // TODO(aalexandrov): strictly speaking it should be better if we can make
     // `config: &OptimizerConfig` a field in the enclosing builder. However,
     // before we can do that we should make sure that nobody outside of the
@@ -353,6 +367,9 @@ impl<'a> DataflowBuilder<'a> {
                         dataflow.import_source(*id, source.desc.typ().clone(), monotonic);
                     }
                     CatalogItem::View(view) => {
+                        if view.security_barrier {
+                            self.security_barriers.insert(*id);
+                        }
                         let expr = view.locally_optimized_expr.as_ref();
                         self.import_view_into_dataflow(id, expr, dataflow, features)?;
                     }

src/adapter/src/optimize/index.rs

@@ -179,6 +179,7 @@ impl Optimize<Index> for Optimizer {
             &self.typecheck_ctx,
             &mut df_meta,
             Some(&mut self.metrics),
+            df_builder.security_barriers(),
         );
         // Run global optimization.
         mz_transform::optimize_dataflow(&mut df_desc, &mut transform_ctx, false)?;

src/adapter/src/optimize/materialized_view.rs

@@ -277,6 +277,7 @@ impl Optimize<LocalMirPlan> for Optimizer {
             &self.typecheck_ctx,
             &mut df_meta,
             Some(&mut self.metrics),
+            df_builder.security_barriers(),
         );
         // Run global optimization.
         mz_transform::optimize_dataflow(&mut df_desc, &mut transform_ctx, false)?;

src/adapter/src/optimize/metric_sink.rs

@@ -215,6 +215,7 @@ impl Optimize<MetricSink> for Optimizer {
             &self.typecheck_ctx,
             &mut df_meta,
             Some(&mut self.metrics),
+            df_builder.security_barriers(),
         );
         // Run global optimization.
         mz_transform::optimize_dataflow(&mut df_desc, &mut transform_ctx, false)?;

src/adapter/src/optimize/peek.rs

@@ -330,6 +330,7 @@ impl<'s> Optimize<LocalMirPlan<Resolved<'s>>> for Optimizer {
             &self.typecheck_ctx,
             &mut df_meta,
             Some(&mut self.metrics),
+            df_builder.security_barriers(),
         );
 
         // Let's already try creating a fast path plan. If successful, we don't need to run the

src/adapter/src/optimize/subscribe.rs

@@ -214,6 +214,7 @@ impl Optimizer {
             &self.typecheck_ctx,
             &mut df_meta,
             Some(&mut self.metrics),
+            df_builder.security_barriers(),
         );
         // Run global optimization.
         mz_transform::optimize_dataflow(&mut df_desc, &mut transform_ctx, false)?;

src/catalog/src/memory/objects.rs

@@ -1401,6 +1401,14 @@ pub struct View {
     pub resolved_ids: ResolvedIds,
     /// All of the catalog objects that are referenced by this view.
     pub dependencies: DependencyIds,
+    /// Whether this view is a security barrier.
+    ///
+    /// A barrier view is never inlined into a reading dataflow, and only
+    /// leakproof predicates from a reader are pushed into its plan, so no
+    /// reader-supplied expression is evaluated against a row the view's own
+    /// filters would have excluded. See
+    /// `doc/developer/design/20260828_security_barrier_views.md`.
+    pub security_barrier: bool,
 }
 
 impl View {

src/clusterd-test-driver/src/dataflow.rs

@@ -508,13 +508,17 @@ impl DataflowBuilder {
             let indexes = ImportedIndexOracle::new(&mir.index_imports);
             let typecheck_ctx = empty_typechecking_context();
             let mut df_meta = DataflowMetainfo::default();
+            // The driver builds dataflows from a MIR spec with no catalog
+            // behind it, so no import can be a security barrier.
+            let security_barriers = std::collections::BTreeSet::new();
             let mut ctx = TransformCtx::global(
                 &indexes,
                 &EmptyStatisticsOracle,
                 features,
                 &typecheck_ctx,
                 &mut df_meta,
                 None,
+                &security_barriers,
             );
             optimize_dataflow(&mut mir, &mut ctx, false)
                 .map_err(|e| anyhow::anyhow!("optimizing dataflow failed: {e}"))?;

src/compute-types/src/plan.rs

@@ -781,7 +781,7 @@ impl LirRelationExpr {
                                                 (row_arity..row_arity + key_len).chain(thinning),
                                             )
                                             .map(m)
-                                            .filter(f)
+                                            .filter_leveled(f)
                                             .project(p);
                                         closure.before =
                                             mfp.into_plan().unwrap().into_nontemporal().unwrap();

src/compute-types/src/plan/reduce.rs

@@ -584,7 +584,7 @@ impl ReducePlan {
             let (m, f, p) = non_temporal.into_map_filter_project();
             mfp_push = MapFilterProject::new(input_arity)
                 .map(m.clone())
-                .filter(f)
+                .filter_leveled(f)
                 .project(0..input_arity);
             output_arity = input_arity;
 

src/compute-types/src/plan/scalar.rs

@@ -776,10 +776,10 @@ pub fn mfp_mir_to_lir(mfp: MapFilterProject<MirScalarExpr>) -> MapFilterProject<
         .predicates
         .iter()
         .map(|(pos, pred)| {
-            (
-                *pos,
-                LirScalarExpr::try_from(pred).expect("unmaterializable in MFP predicate"),
-            )
+            let pred = pred.clone().map_expr(|expr| {
+                LirScalarExpr::try_from(&expr).expect("unmaterializable in MFP predicate")
+            });
+            (*pos, pred)
         })
         .collect();
     MapFilterProject::<LirScalarExpr> {
@@ -830,7 +830,7 @@ pub fn mfp_plan_lir_to_mir(plan: MfpPlan<LirScalarExpr>) -> MfpPlan<MirScalarExp
     let predicates = mfp
         .predicates
         .iter()
-        .map(|(pos, pred)| (*pos, MirScalarExpr::from(pred)))
+        .map(|(pos, pred)| (*pos, pred.clone().map_expr(|e| MirScalarExpr::from(&e))))
         .collect();
     let mir_mfp = MapFilterProject::<MirScalarExpr> {
         expressions,

src/expr-parser/src/parser.rs

@@ -477,6 +477,10 @@ mod relation {
         let parse_input = ParseChildren::new(input, filter.span().start());
         let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);
 
+        let predicates = predicates
+            .into_iter()
+            .map(mz_expr::Predicate::unconstrained)
+            .collect();
         Ok(MirRelationExpr::Filter { input, predicates })
     }
 

src/expr/src/explain.rs

@@ -113,8 +113,8 @@ impl<'a> ExplainSource<'a> {
                 let pushdown = op
                     .predicates
                     .iter()
-                    .filter(|(_, e)| mfp_mapped.expr(e).pushdownable())
-                    .map(|(_, e)| e)
+                    .filter(|(_, p)| mfp_mapped.expr(&p.expr).pushdownable())
+                    .map(|(_, p)| &p.expr)
                     .collect();
                 PushdownInfo { pushdown }
             })

src/expr/src/explain/text.rs

@@ -244,7 +244,7 @@ where
         }
         // render `filter` field iff predicates are present
         if !predicates.is_empty() {
-            let predicates = predicates.iter().map(|(_, p)| self.child(p));
+            let predicates = predicates.iter().map(|(_, p)| self.child(&p.expr));
             let predicates = separated(" AND ", predicates);
             writeln!(f, "{}filter=({})", ctx.as_mut(), predicates)?;
         }
@@ -286,7 +286,11 @@ impl<'a, M: HumanizerMode, E: OptimizableExpr + HumanizeDisplay>
 
         // render `filter` field iff predicates are present
         if !self.expr.predicates.is_empty() {
-            let predicates = self.expr.predicates.iter().map(|(_, p)| self.child(p));
+            let predicates = self
+                .expr
+                .predicates
+                .iter()
+                .map(|(_, p)| self.child(&p.expr));
             let predicates = separated(" AND ", predicates);
             writeln!(f, "{}Filter: {predicates}", ctx.indent)?;
         }
@@ -576,9 +580,22 @@ impl MirRelationExpr {
                             write!(f, "{}Filter", ctx.indent)?;
                         } else {
                             let cols = input.column_names(ctx);
-                            let predicates = mode.seq(predicates, cols);
+                            let exprs = predicates
+                                .iter()
+                                .map(|p| p.expr.clone())
+                                .collect::<Vec<_>>();
+                            let predicates_levels =
+                                predicates.iter().map(|p| p.level()).collect::<Vec<_>>();
+                            let predicates = mode.seq(&exprs, cols);
                             let predicates = separated(" AND ", predicates);
                             write!(f, "{}Filter {}", ctx.indent, predicates)?;
+                            // Surface the ordering constraint when there is one,
+                            // so that a security barrier is visible in a plan.
+                            // Surface the ordering constraint when there is
+                            // one, so a security barrier is visible in a plan.
+                            if predicates_levels.iter().any(|l| *l > 0) {
+                                write!(f, " [levels: {:?}]", predicates_levels)?;
+                            }
                         }
                         self.fmt_analyses(f, ctx)
                     },

src/expr/src/interpret.rs

@@ -1363,6 +1363,7 @@ impl Interpreter for Trace {
 
 #[cfg(test)]
 mod tests {
+    use crate::Predicate;
     use itertools::Itertools;
     use mz_repr::adt::datetime::DateTimeUnits;
     use mz_repr::{Datum, PropDatum, RowArena, SqlScalarType};
@@ -2507,7 +2508,10 @@ mod tests {
             }],
             predicates: vec![(
                 1,
-                MirScalarExpr::literal_ok(Datum::True, ReprScalarType::Bool),
+                Predicate::unconstrained(MirScalarExpr::literal_ok(
+                    Datum::True,
+                    ReprScalarType::Bool,
+                )),
             )],
             projection: vec![0, 1],
             input_arity: 1,
@@ -2596,26 +2600,26 @@ mod tests {
                 // Always fails on the known input range
                 (
                     1,
-                    CallUnary {
+                    Predicate::unconstrained(CallUnary {
                         func: UnaryFunc::IsNull(IsNull),
                         expr: Box::new(CallBinary {
                             func: MulInt32.into(),
                             expr1: Box::new(MirScalarExpr::column(0)),
                             expr2: Box::new(MirScalarExpr::column(0)),
                         }),
-                    },
+                    }),
                 ),
                 // Always returns false on the known input range
                 (
                     1,
-                    CallBinary {
+                    Predicate::unconstrained(CallBinary {
                         func: Eq.into(),
                         expr1: Box::new(MirScalarExpr::column(0)),
                         expr2: Box::new(MirScalarExpr::literal_ok(
                             Datum::Int32(1727694505),
                             ReprScalarType::Int32,
                         )),
-                    },
+                    }),
                 ),
             ],
             projection: vec![],

src/expr/src/lib.rs

@@ -24,6 +24,7 @@ mod relation;
 mod scalar;
 
 pub mod explain;
+pub mod predicate;
 pub mod row;
 pub mod virtual_syntax;
 pub mod visit;
@@ -33,6 +34,7 @@ pub use interpret::{ColumnSpec, ColumnSpecs, Interpreter, ResultSpec, Trace, Tra
 pub use linear::plan::{MfpPlan, SafeMfpPlan};
 pub use linear::util::{join_permutations, permutation_for_arrangement};
 pub use linear::{MapFilterProject, memoize_expr};
+pub use predicate::{Predicate, SecurityLevel};
 pub use relation::func::REPEAT_ROW_NAME;
 pub use relation::func::order_aggregate_datums as order_aggregate_datums_exported_for_benchmarking;
 pub use relation::func::{
@@ -44,7 +46,7 @@ pub use relation::{
     AccessStrategy, AggregateExpr, CollectionPlan, ColumnOrder, JoinImplementation,
     JoinInputCharacteristics, LetRecLimit, MirRelationExpr, RECURSION_LIMIT, RowComparator,
     RowSetFinishing, RowSetFinishingIncremental, WindowFrame, WindowFrameBound, WindowFrameUnits,
-    canonicalize, compare_columns, non_nullable_columns,
+    canonicalize, compare_columns, non_nullable_columns, non_nullable_columns_of,
 };
 pub use scalar::func::{self, BinaryFunc, UnaryFunc, UnmaterializableFunc, VariadicFunc};
 pub use scalar::{