sql: implement row_security session variable

Complete the implementation of the `row_security` session variable,
which was previously defined but non-functional.

When set to `'off'`, non-admin queries that would be affected by
row-level security (RLS) fail with an error instead of silently
filtering rows. This matches postgres' behavior.

Admin users and table owners (without FORCE) remain exempt from RLS
regardless of this setting.

The variable defaults to `'on'` and is tracked in the memo to invalidate
cached plans when changed.

Closes #138916.

Epic: none

Release note (sql change): The `row_security` session variable now
behaves like in postgres, allowing users to detect when RLS is applied.

Author: spilchen

pkg/sql/exec_util.go

@@ -3836,6 +3836,10 @@ func (m *sessionDataMutator) SetAllowPrepareAsOptPlan(val bool) {
 	m.data.AllowPrepareAsOptPlan = val
 }
 
+func (m *sessionDataMutator) SetRowSecurity(val bool) {
+	m.data.RowSecurity = val
+}
+
 func (m *sessionDataMutator) SetSaveTablesPrefix(prefix string) {
 	m.data.SaveTablesPrefix = prefix
 }

pkg/sql/logictest/testdata/logic_test/information_schema

@@ -4243,7 +4243,7 @@ reorder_joins_limit                                              8
 require_explicit_primary_keys                                    off
 results_buffer_size                                              524288
 role                                                             none
-row_security                                                     off
+row_security                                                     on
 search_path                                                      "$user", public
 serial_normalization                                             rowid
 server_encoding                                                  UTF8

pkg/sql/logictest/testdata/logic_test/pg_catalog

@@ -3145,7 +3145,7 @@ reorder_joins_limit                                              8
 require_explicit_primary_keys                                    off                 NULL      NULL        NULL        string
 results_buffer_size                                              524288              NULL      NULL        NULL        string
 role                                                             none                NULL      NULL        NULL        string
-row_security                                                     off                 NULL      NULL        NULL        string
+row_security                                                     on                  NULL      NULL        NULL        string
 search_path                                                      "$user", public     NULL      NULL        NULL        string
 serial_normalization                                             rowid               NULL      NULL        NULL        string
 server_encoding                                                  UTF8                NULL      NULL        NULL        string
@@ -3390,7 +3390,7 @@ reorder_joins_limit                                              8
 require_explicit_primary_keys                                    off                 NULL  user     NULL      off                 off
 results_buffer_size                                              524288              NULL  user     NULL      524288              524288
 role                                                             none                NULL  user     NULL      none                none
-row_security                                                     off                 NULL  user     NULL      off                 off
+row_security                                                     on                  NULL  user     NULL      on                  on
 search_path                                                      "$user", public     NULL  user     NULL      "$user", public     "$user", public
 serial_normalization                                             rowid               NULL  user     NULL      rowid               rowid
 server_encoding                                                  UTF8                NULL  user     NULL      UTF8                UTF8

pkg/sql/logictest/testdata/logic_test/row_level_security

@@ -5728,3 +5728,206 @@ statement ok
 SET allow_view_with_security_invoker_clause = off
 
 subtest end
+
+subtest row_security_off
+
+statement ok
+CREATE USER alice;
+
+statement ok
+CREATE TABLE accounts (
+    id INT PRIMARY KEY,
+    owner TEXT,
+    balance INT
+);
+
+statement ok
+INSERT INTO accounts VALUES
+  (1, 'alice', 100),
+  (2, 'bob', 200);
+
+statement ok
+ALTER TABLE accounts ENABLE ROW LEVEL SECURITY;
+
+statement ok
+GRANT ALL ON accounts TO alice;
+
+statement ok
+SET ROLE alice;
+
+# RLS enabled but no policies (deny all) - row_security='off' should fail.
+query ITI
+SELECT * FROM accounts;
+----
+
+statement ok
+SET SESSION row_security = 'off';
+
+statement error pgcode 42501 pq: query would be affected by row-level security policy for table "accounts"
+SELECT * FROM accounts;
+
+statement ok
+RESET row_security;
+
+statement ok
+SET ROLE root;
+
+# Add policy and test with alice as non-owner.
+statement ok
+CREATE POLICY user_is_owner ON accounts
+  USING (owner = current_user());
+
+statement ok
+SET ROLE alice;
+
+query ITI
+SELECT * FROM accounts;
+----
+1  alice  100
+
+statement ok
+SET SESSION row_security = 'off';
+
+statement error pgcode 42501 pq: query would be affected by row-level security policy for table "accounts"
+SELECT * FROM accounts;
+
+statement ok
+RESET row_security;
+
+# Test INSERT and UPDATE.
+statement ok
+INSERT INTO accounts VALUES (3, 'alice', 300);
+
+statement ok
+UPDATE accounts SET balance = balance + 50 WHERE id = 3;
+
+statement ok
+SET SESSION row_security = 'off';
+
+statement error pgcode 42501 pq: query would be affected by row-level security policy for table "accounts"
+INSERT INTO accounts VALUES (5, 'alice', 500);
+
+statement error pgcode 42501 pq: query would be affected by row-level security policy for table "accounts"
+UPDATE accounts SET balance = balance + 50 WHERE id = 3;
+
+statement ok
+RESET row_security;
+
+# Test DELETE and UPSERT.
+statement ok
+DELETE FROM accounts WHERE id = 1;
+
+statement ok
+UPSERT INTO accounts VALUES (3, 'alice', 400);
+
+statement ok
+SET SESSION row_security = 'off';
+
+statement error pgcode 42501 pq: query would be affected by row-level security policy for table "accounts"
+DELETE FROM accounts WHERE id = 3;
+
+statement error pgcode 42501 pq: query would be affected by row-level security policy for table "accounts"
+UPSERT INTO accounts VALUES (6, 'alice', 600);
+
+statement ok
+RESET row_security;
+
+statement ok
+SET ROLE root;
+
+# Table owner bypasses RLS without FORCE.
+statement ok
+ALTER TABLE accounts OWNER TO alice;
+
+statement ok
+SET ROLE alice;
+
+statement ok
+DROP POLICY user_is_owner ON accounts;
+
+statement ok
+CREATE POLICY deny_all ON accounts USING (false);
+
+query ITI rowsort
+SELECT * FROM accounts;
+----
+2  bob    200
+3  alice  400
+
+statement ok
+SET SESSION row_security = 'off';
+
+query ITI rowsort
+SELECT * FROM accounts;
+----
+2  bob    200
+3  alice  400
+
+statement ok
+RESET row_security;
+
+# With FORCE, table owner is restricted.
+statement ok
+ALTER TABLE accounts FORCE ROW LEVEL SECURITY;
+
+query ITI
+SELECT * FROM accounts;
+----
+
+statement ok
+SET SESSION row_security = 'off';
+
+statement error pgcode 42501 pq: query would be affected by row-level security policy for table "accounts"
+SELECT * FROM accounts;
+
+statement ok
+RESET row_security;
+
+statement ok
+SET ROLE root;
+
+# Verify setting doesn't apply to admin.
+statement ok
+SET SESSION row_security = 'off';
+
+query ITI rowsort
+SELECT * FROM accounts;
+----
+2  bob    200
+3  alice  400
+
+statement ok
+RESET row_security;
+
+# Non-RLS table allows row_security='off'.
+statement ok
+CREATE TABLE non_rls (id INT PRIMARY KEY);
+
+statement ok
+INSERT INTO non_rls VALUES (1), (2);
+
+statement ok
+GRANT ALL ON non_rls TO alice;
+
+statement ok
+SET ROLE alice;
+
+statement ok
+SET SESSION row_security = 'off';
+
+query I rowsort
+SELECT * FROM non_rls;
+----
+1
+2
+
+statement ok
+SET ROLE root;
+
+statement ok
+DROP TABLE accounts, non_rls;
+
+statement ok
+DROP USER alice;
+
+subtest end

pkg/sql/logictest/testdata/logic_test/show_source

@@ -209,7 +209,7 @@ reorder_joins_limit                                              8
 require_explicit_primary_keys                                    off
 results_buffer_size                                              524288
 role                                                             none
-row_security                                                     off
+row_security                                                     on
 search_path                                                      "$user", public
 serial_normalization                                             rowid
 server_encoding                                                  UTF8

pkg/sql/opt/memo/memo.go

@@ -211,6 +211,7 @@ type Memo struct {
 	useExistsFilterHoistRule                   bool
 	disableSlowCascadeFastPathForRBRTables     bool
 	useImprovedHoistJoinProject                bool
+	rowSecurity                                bool
 
 	// txnIsoLevel is the isolation level under which the plan was created. This
 	// affects the planning of some locking operations, so it must be included in
@@ -318,6 +319,7 @@ func (m *Memo) Init(ctx context.Context, evalCtx *eval.Context) {
 		useExistsFilterHoistRule:                   evalCtx.SessionData().OptimizerUseExistsFilterHoistRule,
 		disableSlowCascadeFastPathForRBRTables:     evalCtx.SessionData().OptimizerDisableCrossRegionCascadeFastPathForRBRTables,
 		useImprovedHoistJoinProject:                evalCtx.SessionData().OptimizerUseImprovedHoistJoinProject,
+		rowSecurity:                                evalCtx.SessionData().RowSecurity,
 		txnIsoLevel:                                evalCtx.TxnIsoLevel,
 	}
 	m.metadata.Init()
@@ -493,6 +495,7 @@ func (m *Memo) IsStale(
 		m.useExistsFilterHoistRule != evalCtx.SessionData().OptimizerUseExistsFilterHoistRule ||
 		m.disableSlowCascadeFastPathForRBRTables != evalCtx.SessionData().OptimizerDisableCrossRegionCascadeFastPathForRBRTables ||
 		m.useImprovedHoistJoinProject != evalCtx.SessionData().OptimizerUseImprovedHoistJoinProject ||
+		m.rowSecurity != evalCtx.SessionData().RowSecurity ||
 		m.txnIsoLevel != evalCtx.TxnIsoLevel {
 		return true, nil
 	}

pkg/sql/opt/memo/memo_test.go

@@ -674,6 +674,12 @@ func TestMemoIsStale(t *testing.T) {
 	notStale()
 	o.Memo().Metadata().ClearRLSEnabled()
 	notStale()
+
+	// Stale row_security.
+	evalCtx.SessionData().RowSecurity = true
+	stale()
+	evalCtx.SessionData().RowSecurity = false
+	notStale()
 }
 
 // TestStatsAvailable tests that the statisticsBuilder correctly identifies

pkg/sql/opt/optbuilder/row_level_security.go

@@ -12,6 +12,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/cat"
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/memo"
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -85,6 +87,11 @@ func (b *Builder) isExemptFromRLSPolicies(
 	if isAdmin || isOwnerAndNotForced || bypassRLS {
 		return true
 	}
+	// RLS applies. Reject the query if row-level security is disabled in the session.
+	if !b.evalCtx.SessionData().RowSecurity {
+		panic(pgerror.Newf(pgcode.InsufficientPrivilege,
+			"query would be affected by row-level security policy for table %q", tabMeta.Table.Name()))
+	}
 	return false
 }
 

pkg/sql/sessiondatapb/local_only_session_data.proto

@@ -741,6 +741,8 @@ message LocalOnlySessionData {
   bool optimizer_use_improved_hoist_join_project = 186;
   // EnableInspectCommand controls use of the INSPECT command.
   bool enable_inspect_command = 187;
+  // RowSecurity controls whether row-level security is enabled.
+  bool row_security = 188;
 
   ///////////////////////////////////////////////////////////////////////////
   // WARNING: consider whether a session parameter you're adding needs to  //

pkg/sql/vars.go

@@ -1684,15 +1684,21 @@ var varGen = map[string]sessionVar{
 	// results received by clients, we accept both values.
 	`synchronize_seqscans`: makeCompatBoolVar(`synchronize_seqscans`, true, true /* anyAllowed */),
 
-	// See https://www.postgresql.org/docs/10/static/runtime-config-client.html#GUC-ROW-SECURITY
-	// The default in pg is "on" but row security is not supported in CockroachDB.
-	// We blindly accept both values because as long as there are now row security policies defined,
-	// either value produces the same query results in PostgreSQL. That is, as long as CockroachDB
-	// does not support row security, accepting either "on" and "off" but ignoring the result
-	// is postgres-compatible.
-	// If/when CockroachDB is extended to support row security, the default and allowed values
-	// should be modified accordingly.
-	`row_security`: makeCompatBoolVar(`row_security`, false, true /* anyAllowed */),
+	`row_security`: {
+		GetStringVal: makePostgresBoolGetStringValFn("row_security"),
+		Get: func(evalCtx *extendedEvalContext, _ *kv.Txn) (string, error) {
+			return formatBoolAsPostgresSetting(evalCtx.SessionData().RowSecurity), nil
+		},
+		Set: func(_ context.Context, m sessionDataMutator, s string) error {
+			b, err := paramparse.ParseBoolVar("row_security", s)
+			if err != nil {
+				return err
+			}
+			m.SetRowSecurity(b)
+			return nil
+		},
+		GlobalDefault: globalTrue,
+	},
 
 	`statement_timeout`: {
 		GetStringVal: makeTimeoutVarGetter(`statement_timeout`),