sql: add security barrier views POC

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/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.
@@ -288,9 +295,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 +366,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/mz-deploy/src/cli/commands/test/lower.rs

@@ -800,6 +800,7 @@ fn create_target_view_sql(stmt: &Statement, fqn: &FullyQualifiedName) -> Result<
             definition: ViewDefinition {
                 name: mv.name,
                 columns: mv.columns,
+                with_options: vec![],
                 query: mv.query,
             },
         },

src/sql-lexer/src/keywords.txt

@@ -60,6 +60,7 @@ Auto
 Availability
 Avro
 Aws
+Barrier
 Batch
 Begin
 Between

src/sql-parser/src/ast/defs/ddl.rs

@@ -28,6 +28,40 @@ use crate::ast::{
     AstInfo, ColumnName, Expr, Ident, OrderByExpr, UnresolvedItemName, Version, WithOptionValue,
 };
 
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum ViewOptionName {
+    /// The `SECURITY BARRIER [=] <bool>` option.
+    SecurityBarrier,
+}
+
+impl AstDisplay for ViewOptionName {
+    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
+        match self {
+            ViewOptionName::SecurityBarrier => f.write_str("SECURITY BARRIER"),
+        }
+    }
+}
+
+impl WithOptionName for ViewOptionName {
+    /// # WARNING
+    ///
+    /// Whenever implementing this trait consider very carefully whether or not
+    /// this value could contain sensitive user data. If you're uncertain, err
+    /// on the conservative side and return `true`.
+    fn redact_value(&self) -> bool {
+        match self {
+            ViewOptionName::SecurityBarrier => false,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct ViewOption<T: AstInfo> {
+    pub name: ViewOptionName,
+    pub value: Option<WithOptionValue<T>>,
+}
+impl_display_for_with_option!(ViewOption);
+
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub enum MaterializedViewOptionName {
     /// The `ASSERT NOT NULL [=] <ident>` option.

src/sql-parser/src/ast/defs/statement.rs

@@ -33,7 +33,7 @@ use crate::ast::{
     Ident, IntervalValue, KeyConstraint, MaterializedViewOption, Query, SelectItem, SinkEnvelope,
     SourceEnvelope, SourceIncludeMetadata, SubscribeOutput, TableAlias, TableConstraint,
     TableWithJoins, UnresolvedDatabaseName, UnresolvedItemName, UnresolvedObjectName,
-    UnresolvedSchemaName, Value,
+    UnresolvedSchemaName, Value, ViewOption,
 };
 
 /// A top-level statement (SELECT, INSERT, CREATE, etc.)
@@ -1513,6 +1513,7 @@ pub struct ViewDefinition<T: AstInfo> {
     /// View name
     pub name: UnresolvedItemName,
     pub columns: Vec<Ident>,
+    pub with_options: Vec<ViewOption<T>>,
     pub query: Query<T>,
 }
 
@@ -1526,6 +1527,12 @@ impl<T: AstInfo> AstDisplay for ViewDefinition<T> {
             f.write_str(")");
         }
 
+        if !self.with_options.is_empty() {
+            f.write_str(" WITH (");
+            f.write_node(&display::comma_separated(&self.with_options));
+            f.write_str(")");
+        }
+
         f.write_str(" AS ");
         f.write_node(&self.query);
     }

src/sql-parser/src/parser.rs

@@ -4172,17 +4172,34 @@ impl<'a> Parser<'a> {
         // ANSI SQL and Postgres support RECURSIVE here, but we don't.
         let name = self.parse_item_name()?;
         let columns = self.parse_parenthesized_column_list(Optional)?;
-        // Postgres supports WITH options here, but we don't.
+        let with_options = if self.parse_keyword(WITH) {
+            self.expect_token(&Token::LParen)?;
+            let options = self.parse_comma_separated(Parser::parse_view_option)?;
+            self.expect_token(&Token::RParen)?;
+            options
+        } else {
+            vec![]
+        };
         self.expect_keyword(AS)?;
         let query = self.parse_query()?;
         // Optional `WITH [ CASCADED | LOCAL ] CHECK OPTION` is widely supported here.
         Ok(ViewDefinition {
             name,
             columns,
+            with_options,
             query,
         })
     }
 
+    fn parse_view_option(&mut self) -> Result<ViewOption<Raw>, ParserError> {
+        self.expect_keywords(&[SECURITY, BARRIER])?;
+        let value = self.parse_optional_option_value()?;
+        Ok(ViewOption {
+            name: ViewOptionName::SecurityBarrier,
+            value,
+        })
+    }
+
     fn parse_create_materialized_view(&mut self) -> Result<Statement<Raw>, ParserError> {
         let mut if_exists = if self.parse_keyword(OR) {
             self.expect_keyword(REPLACE)?;