sql: add security_invoker feature flag and grammar support

This change adds the `enable_view_security_invoker`
feature flag. It also adds grammar support for the
following two statements:
```
CREATE VIEW <view_name> WITH ( security_invoker = true ) AS SELECT * FROM <table_name>;
ALTER VIEW <view_name> SET ( security_invoker = false );
```

Informs: #138918
Epic: CRDB-48807
Release note: None

Author: Dedej-Bergin

docs/generated/sql/bnf/stmt_block.bnf

@@ -1459,6 +1459,7 @@ unreserved_keyword ::=
 	| 'SEARCH'
 	| 'SECOND'
 	| 'SECURITY'
+	| 'SECURITY_INVOKER'
 	| 'SECONDARY'
 	| 'SERIALIZABLE'
 	| 'SEQUENCE'
@@ -4385,6 +4386,7 @@ bare_label_keywords ::=
 	| 'SEARCH'
 	| 'SECONDARY'
 	| 'SECURITY'
+	| 'SECURITY_INVOKER'
 	| 'SELECT'
 	| 'SEQUENCE'
 	| 'SEQUENCES'

pkg/sql/create_view.go

@@ -77,6 +77,12 @@ func (n *createViewNode) startExec(params runParams) error {
 		return pgerror.Newf(pgcode.ReadOnlySQLTransaction, "schema changes are not allowed on a reader catalog")
 	}
 	createView := n.createView
+
+	if !params.SessionData().AllowViewWithSecurityInvokerClause && createView.Options != nil && createView.Options.SecurityInvoker {
+		return pgerror.Newf(pgcode.FeatureNotSupported,
+			"security invoker views are not supported")
+	}
+
 	tableType := tree.GetTableType(
 		false /* isSequence */, true /* isView */, createView.Materialized,
 	)

pkg/sql/exec_util.go

@@ -3720,6 +3720,10 @@ func (m *sessionDataMutator) SetAlterColumnTypeGeneral(val bool) {
 	m.data.AlterColumnTypeGeneralEnabled = val
 }
 
+func (m *sessionDataMutator) SetAllowViewWithSecurityInvokerClause(val bool) {
+	m.data.AllowViewWithSecurityInvokerClause = val
+}
+
 func (m *sessionDataMutator) SetEnableSuperRegions(val bool) {
 	m.data.EnableSuperRegions = val
 }

pkg/sql/logictest/testdata/logic_test/row_level_security

@@ -5634,4 +5634,27 @@ DROP FUNCTION fail, divbyzero;
 statement ok
 DROP ROLE alice;
 
+subtest unimplemented_alter_view_with_security_invoker
+
+statement ok
+CREATE TABLE roach_pg_table ( count INT )
+
+statement ok
+SET allow_view_with_security_invoker_clause = on
+
+statement ok
+CREATE VIEW security_invoker_view WITH ( security_invoker = true ) AS SELECT * FROM roach_pg_table
+
+statement error pq: at or near "\)": syntax error: unimplemented: this syntax\nHINT: You have attempted to use a feature that is not yet implemented\.\n\nPlease check the public issue tracker to check whether this problem is\nalready tracked\. If you cannot find it there, please report the error\nwith details by creating a new issue\.\n\nIf you would rather not post publicly, please contact us directly\nusing the support form\.\n\nWe appreciate your feedback\.\n\nDETAIL: source SQL:\nALTER VIEW security_invoker_view SET \( security_invoker = false \)
+ALTER VIEW security_invoker_view SET ( security_invoker = false )
+
+statement ok
+DROP VIEW security_invoker_view
+
+statement ok
+DROP TABLE roach_pg_table
+
+statement ok
+SET allow_view_with_security_invoker_clause = off
+
 subtest end

pkg/sql/logictest/testdata/logic_test/views

@@ -1933,9 +1933,21 @@ CREATE VIEW v AS (SELECT 1)
 statement error cross database type references are not supported: db106602a.public.e
 CREATE OR REPLACE VIEW v AS (SELECT 1 FROM (VALUES (1)) val(i) WHERE 'foo'::db106602a.e = 'foo'::db106602a.e)
 
-subtest end
+subtest regression_test_128535
 
 # Regression test for hitting an assertion error when using an unimplemented
 # builtin in the view (#128535).
 statement error pgcode 0A000 unimplemented
 CREATE VIEW v128535 AS SELECT json_to_tsvector()
+
+subtest allow_view_with_security_invoker_clause_feature_flag
+
+statement ok
+SET allow_view_with_security_invoker_clause = on;
+
+query T
+show session allow_view_with_security_invoker_clause;
+----
+on
+
+subtest end

pkg/sql/parser/sql.y

@@ -270,6 +270,9 @@ func (u *sqlSymUnion) auditMode() tree.AuditMode {
 func (u *sqlSymUnion) bool() bool {
     return u.val.(bool)
 }
+func (u *sqlSymUnion) viewOptions() *tree.ViewOptions {
+    return u.val.(*tree.ViewOptions)
+}
 func (u *sqlSymUnion) strPtr() *string {
     return u.val.(*string)
 }
@@ -1056,7 +1059,7 @@ func (u *sqlSymUnion) doBlockOption() tree.DoBlockOption {
 %token <str> REVOKE RIGHT ROLE ROLES ROLLBACK ROLLUP ROUTINES ROW ROWS RSHIFT RULE RUNNING
 
 %token <str> SAVEPOINT SCANS SCATTER SCHEDULE SCHEDULES SCROLL SCHEMA SCHEMA_ONLY SCHEMAS SCRUB
-%token <str> SEARCH SECOND SECONDARY SECURITY SELECT SEQUENCE SEQUENCES
+%token <str> SEARCH SECOND SECONDARY SECURITY SECURITY_INVOKER SELECT SEQUENCE SEQUENCES
 %token <str> SERIALIZABLE SERVER SERVICE SESSION SESSIONS SESSION_USER SET SETOF SETS SETTING SETTINGS
 %token <str> SHARE SHARED SHOW SIMILAR SIMPLE SIZE SKIP SKIP_LOCALITIES_CHECK SKIP_MISSING_FOREIGN_KEYS
 %token <str> SKIP_MISSING_SEQUENCES SKIP_MISSING_SEQUENCE_OWNERS SKIP_MISSING_VIEWS SKIP_MISSING_UDFS SMALLINT SMALLSERIAL
@@ -1206,6 +1209,7 @@ func (u *sqlSymUnion) doBlockOption() tree.DoBlockOption {
 %type <tree.Statement> alter_rename_view_stmt
 %type <tree.Statement> alter_view_set_schema_stmt
 %type <tree.Statement> alter_view_owner_stmt
+%type <tree.Statement> alter_view_set_options_stmt
 
 // ALTER SEQUENCE
 %type <tree.Statement> alter_rename_sequence_stmt
@@ -1801,6 +1805,7 @@ func (u *sqlSymUnion) doBlockOption() tree.DoBlockOption {
 %type <*tree.TriggerTransition> trigger_transition
 %type <[]*tree.TriggerTransition> trigger_transition_list opt_trigger_transition_list
 %type <bool> transition_is_new transition_is_row
+%type <*tree.ViewOptions> opt_view_with
 %type <tree.TriggerForEach> trigger_for_each trigger_for_type
 %type <tree.Expr> trigger_when
 %type <str> trigger_func_arg opt_as function_or_procedure
@@ -2055,6 +2060,7 @@ alter_view_stmt:
   alter_rename_view_stmt
 | alter_view_set_schema_stmt
 | alter_view_owner_stmt
+| alter_view_set_options_stmt
 // ALTER VIEW has its error help token here because the ALTER VIEW
 // prefix is spread over multiple non-terminals.
 | ALTER VIEW error // SHOW HELP: ALTER VIEW
@@ -11901,44 +11907,50 @@ role_or_group_or_user:
 // %Help: CREATE VIEW - create a new view
 // %Category: DDL
 // %Text:
-// CREATE [TEMPORARY | TEMP] VIEW [IF NOT EXISTS] <viewname> [( <colnames...> )] AS <source>
+// CREATE [TEMPORARY | TEMP] VIEW [IF NOT EXISTS] <viewname> [( <colnames...> )] [WITH ( <option> [= <value>] [, ....] )] AS <source>
 // CREATE [TEMPORARY | TEMP] MATERIALIZED VIEW [IF NOT EXISTS] <viewname> [( <colnames...> )] AS <source> [WITH [NO] DATA]
+//
+// Options:
+//   security_invoker [= { true | false | 1 | 0 }]: controls view permissions (defaults to true if specified without value)
 // %SeeAlso: CREATE TABLE, SHOW CREATE, WEBDOCS/create-view.html
 create_view_stmt:
-  CREATE opt_temp opt_view_recursive VIEW view_name opt_column_list AS select_stmt
+  CREATE opt_temp opt_view_recursive VIEW view_name opt_column_list opt_view_with AS select_stmt
   {
     name := $5.unresolvedObjectName().ToTableName()
     $$.val = &tree.CreateView{
       Name: name,
       ColumnNames: $6.nameList(),
-      AsSource: $8.slct(),
+      AsSource: $9.slct(),
       Persistence: $2.persistence(),
+      Options: $7.viewOptions(),
       IfNotExists: false,
       Replace: false,
     }
   }
 // We cannot use a rule like opt_or_replace here as that would cause a conflict
 // with the opt_temp rule.
-| CREATE OR REPLACE opt_temp opt_view_recursive VIEW view_name opt_column_list AS select_stmt
+| CREATE OR REPLACE opt_temp opt_view_recursive VIEW view_name opt_column_list opt_view_with AS select_stmt
   {
     name := $7.unresolvedObjectName().ToTableName()
     $$.val = &tree.CreateView{
       Name: name,
       ColumnNames: $8.nameList(),
-      AsSource: $10.slct(),
+      AsSource: $11.slct(),
       Persistence: $4.persistence(),
+      Options: $9.viewOptions(),
       IfNotExists: false,
       Replace: true,
     }
   }
-| CREATE opt_temp opt_view_recursive VIEW IF NOT EXISTS view_name opt_column_list AS select_stmt
+| CREATE opt_temp opt_view_recursive VIEW IF NOT EXISTS view_name opt_column_list opt_view_with AS select_stmt
   {
     name := $8.unresolvedObjectName().ToTableName()
     $$.val = &tree.CreateView{
       Name: name,
       ColumnNames: $9.nameList(),
-      AsSource: $11.slct(),
+      AsSource: $12.slct(),
       Persistence: $2.persistence(),
+      Options: $10.viewOptions(),
       IfNotExists: true,
       Replace: false,
     }
@@ -12154,6 +12166,44 @@ opt_view_recursive:
   /* EMPTY */ { /* no error */ }
 | RECURSIVE { return unimplemented(sqllex, "create recursive view") }
 
+// View-specific WITH clause that only accepts security_invoker
+opt_view_with:
+  /* EMPTY */
+  {
+    $$.val = (*tree.ViewOptions)(nil)
+  }
+| WITH '(' SECURITY_INVOKER ')'
+  {
+    /* SKIP DOC */
+    // security_invoker without value defaults to true
+    $$.val = &tree.ViewOptions{SecurityInvoker: true}
+  }
+| WITH '(' SECURITY_INVOKER '=' TRUE ')'
+  {
+    /* SKIP DOC */
+    $$.val = &tree.ViewOptions{SecurityInvoker: true}
+  }
+| WITH '(' SECURITY_INVOKER '=' FALSE ')'
+  {
+    /* SKIP DOC */
+    $$.val = &tree.ViewOptions{SecurityInvoker: false}
+  }
+| WITH '(' SECURITY_INVOKER '=' ICONST ')'
+  {
+    /* SKIP DOC */
+    // Handle integer values: 1 = true, 0 = false
+    val, err := $5.numVal().AsInt64()
+    if err != nil {
+      return setErr(sqllex, err)
+    }
+    if val == 1 {
+      $$.val = &tree.ViewOptions{SecurityInvoker: true}
+    } else if val == 0 {
+      $$.val = &tree.ViewOptions{SecurityInvoker: false}
+    } else {
+      return setErr(sqllex, errors.New("security_invoker accepts only true/false or 1/0"))
+    }
+  }
 
 // %Help: CREATE TYPE - create a type
 // %Category: DDL
@@ -12768,6 +12818,16 @@ alter_view_owner_stmt:
     }
   }
 
+alter_view_set_options_stmt:
+  ALTER VIEW relation_expr SET '(' SECURITY_INVOKER '=' var_value ')'
+  {
+    return unimplemented(sqllex, "ALTER VIEW ... SET (security_invoker = ...) is not yet implemented.")
+  }
+| ALTER VIEW IF EXISTS relation_expr SET '(' SECURITY_INVOKER '=' var_value ')'
+  {
+    return unimplemented(sqllex, "ALTER VIEW ... IF EXISTS SET (security_invoker = ...) is not yet implemented.")
+  }
+
 alter_sequence_set_schema_stmt:
 	ALTER SEQUENCE relation_expr SET SCHEMA schema_name
 	 {
@@ -18493,6 +18553,7 @@ unreserved_keyword:
 | SEARCH
 | SECOND
 | SECURITY
+| SECURITY_INVOKER
 | SECONDARY
 | SERIALIZABLE
 | SEQUENCE
@@ -19074,6 +19135,7 @@ bare_label_keywords:
 | SEARCH
 | SECONDARY
 | SECURITY
+| SECURITY_INVOKER
 | SELECT
 | SEQUENCE
 | SEQUENCES

pkg/sql/parser/testdata/alter_view

@@ -61,3 +61,24 @@ ALTER MATERIALIZED VIEW IF EXISTS v RENAME TO v
 ALTER MATERIALIZED VIEW IF EXISTS v RENAME TO v -- fully parenthesized
 ALTER MATERIALIZED VIEW IF EXISTS v RENAME TO v -- literals removed
 ALTER MATERIALIZED VIEW IF EXISTS _ RENAME TO _ -- identifiers removed
+
+error
+ALTER VIEW v SET (security_invoker = true)
+----
+----
+at or near ")": syntax error: unimplemented: this syntax
+DETAIL: source SQL:
+ALTER VIEW v SET (security_invoker = true)
+                                         ^
+HINT: You have attempted to use a feature that is not yet implemented.
+
+Please check the public issue tracker to check whether this problem is
+already tracked. If you cannot find it there, please report the error
+with details by creating a new issue.
+
+If you would rather not post publicly, please contact us directly
+using the support form.
+
+We appreciate your feedback.
+----
+----

pkg/sql/parser/testdata/create_view

@@ -215,3 +215,84 @@ REFRESH MATERIALIZED VIEW a.b AS OF SYSTEM TIME '2025-01-01 11:11:11' WITH NO DA
 REFRESH MATERIALIZED VIEW a.b AS OF SYSTEM TIME ('2025-01-01 11:11:11') WITH NO DATA -- fully parenthesized
 REFRESH MATERIALIZED VIEW a.b AS OF SYSTEM TIME '_' WITH NO DATA -- literals removed
 REFRESH MATERIALIZED VIEW _._ AS OF SYSTEM TIME '2025-01-01 11:11:11' WITH NO DATA -- identifiers removed
+
+parse
+CREATE VIEW a WITH (SECURITY_INVOKER) AS SELECT * FROM b
+----
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- normalized!
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT (*) FROM b -- fully parenthesized
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- literals removed
+CREATE VIEW _ WITH ( security_invoker = true ) AS SELECT * FROM _ -- identifiers removed
+
+parse
+CREATE VIEW a WITH (SECURITY_INVOKER = TRUE) AS SELECT * FROM b
+----
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- normalized!
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT (*) FROM b -- fully parenthesized
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- literals removed
+CREATE VIEW _ WITH ( security_invoker = true ) AS SELECT * FROM _ -- identifiers removed
+
+parse
+CREATE VIEW a WITH (SECURITY_INVOKER = FALSE) AS SELECT * FROM b
+----
+CREATE VIEW a WITH ( security_invoker = false ) AS SELECT * FROM b -- normalized!
+CREATE VIEW a WITH ( security_invoker = false ) AS SELECT (*) FROM b -- fully parenthesized
+CREATE VIEW a WITH ( security_invoker = false ) AS SELECT * FROM b -- literals removed
+CREATE VIEW _ WITH ( security_invoker = false ) AS SELECT * FROM _ -- identifiers removed
+
+parse
+CREATE VIEW a WITH (SECURITY_INVOKER = 1) AS SELECT * FROM b
+----
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- normalized!
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT (*) FROM b -- fully parenthesized
+CREATE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- literals removed
+CREATE VIEW _ WITH ( security_invoker = true ) AS SELECT * FROM _ -- identifiers removed
+
+parse
+CREATE VIEW a WITH (SECURITY_INVOKER = 0) AS SELECT * FROM b
+----
+CREATE VIEW a WITH ( security_invoker = false ) AS SELECT * FROM b -- normalized!
+CREATE VIEW a WITH ( security_invoker = false ) AS SELECT (*) FROM b -- fully parenthesized
+CREATE VIEW a WITH ( security_invoker = false ) AS SELECT * FROM b -- literals removed
+CREATE VIEW _ WITH ( security_invoker = false ) AS SELECT * FROM _ -- identifiers removed
+
+parse
+CREATE VIEW a (x, y) WITH (SECURITY_INVOKER) AS SELECT c, d FROM b
+----
+CREATE VIEW a (x, y) WITH ( security_invoker = true ) AS SELECT c, d FROM b -- normalized!
+CREATE VIEW a (x, y) WITH ( security_invoker = true ) AS SELECT (c), (d) FROM b -- fully parenthesized
+CREATE VIEW a (x, y) WITH ( security_invoker = true ) AS SELECT c, d FROM b -- literals removed
+CREATE VIEW _ (_, _) WITH ( security_invoker = true ) AS SELECT _, _ FROM _ -- identifiers removed
+
+parse
+CREATE OR REPLACE VIEW a WITH (SECURITY_INVOKER) AS SELECT * FROM b
+----
+CREATE OR REPLACE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- normalized!
+CREATE OR REPLACE VIEW a WITH ( security_invoker = true ) AS SELECT (*) FROM b -- fully parenthesized
+CREATE OR REPLACE VIEW a WITH ( security_invoker = true ) AS SELECT * FROM b -- literals removed
+CREATE OR REPLACE VIEW _ WITH ( security_invoker = true ) AS SELECT * FROM _ -- identifiers removed
+
+parse
+CREATE VIEW IF NOT EXISTS a WITH (SECURITY_INVOKER) AS SELECT * FROM b
+----
+CREATE VIEW IF NOT EXISTS a WITH ( security_invoker = true ) AS SELECT * FROM b -- normalized!
+CREATE VIEW IF NOT EXISTS a WITH ( security_invoker = true ) AS SELECT (*) FROM b -- fully parenthesized
+CREATE VIEW IF NOT EXISTS a WITH ( security_invoker = true ) AS SELECT * FROM b -- literals removed
+CREATE VIEW IF NOT EXISTS _ WITH ( security_invoker = true ) AS SELECT * FROM _ -- identifiers removed
+
+error
+CREATE VIEW a WITH (SECURITY_INVOKER = 2) AS SELECT * FROM b
+----
+at or near ")": syntax error: security_invoker accepts only true/false or 1/0
+DETAIL: source SQL:
+CREATE VIEW a WITH (SECURITY_INVOKER = 2) AS SELECT * FROM b
+                                        ^
+
+error
+CREATE VIEW a WITH (SECURITY_INVOKER = 'invalid') AS SELECT * FROM b
+----
+at or near "invalid": syntax error
+DETAIL: source SQL:
+CREATE VIEW a WITH (SECURITY_INVOKER = 'invalid') AS SELECT * FROM b
+                                       ^
+HINT: try \h CREATE VIEW