sql: add RLS policy support for UPSERT

Previously, row-level security (RLS) was not enforced for UPSERT
statements. This change introduces full RLS support for the following
UPSERT variants:
- UPSERT
- INSERT … ON CONFLICT DO UPDATE
- INSERT … ON CONFLICT DO NOTHING

The specific RLS policies enforced during an UPSERT depend on whether a
row conflict occurs:
- No Conflict (row is inserted):
  - SELECT/ALL: used to scan for potential conflicts.
  - INSERT/ALL: enforced on the newly inserted row.
- Conflict (row is updated):
  - SELECT/ALL: enforced to verify access to the conflicting row.
  - UPDATE/ALL: enforced on the row being updated.

Importantly, SELECT/ALL policies used in conflict detection are not
applied as filters. Filtering out unauthorized rows would permit
inserting duplicates. Instead, we enforce these policies via explicit
checks: if the user is not authorized to see the conflicting row, the
UPSERT fails.

To support this behavior, we reused the single synthetic check
constraint per table. It now has a combined expression that captures all
relevant RLS checks for UPSERTs:
1. Enforce SELECT/ALL on existing rows during conflict detection.
2. Enforce UPDATE/ALL on conflicting rows that are updated.
3. Enforce INSERT/ALL on new rows that are inserted.

This implementation differs from PostgreSQL in one key way. In Postgres,
the VALUES clause is always subject to constraint and policy checks—even
if the row is not written due to a conflict. This is a known limitation
discussed in #35370.

As a result, we allow statements like the following to succeed if a
conflict occurs, even when the VALUES clause violates RLS:
```
INSERT INTO t VALUES (...violates RLS...) ON CONFLICT DO UPDATE SET ...complies with RLS...;
INSERT INTO t VALUES (...violates RLS...) ON CONFLICT DO NOTHING;
```

Closes #141998

Epic: CRDB-11724
Release note: none

Author: spilchen

pkg/sql/logictest/testdata/logic_test/row_level_security

@@ -27,6 +27,9 @@ const (
 	PolicyScopeUpdate
 	// PolicyScopeDelete indicates that the policy applies to DELETE operations.
 	PolicyScopeDelete
+	// PolicyScopeUpsert is used to indicate it's an INSERT ... ON CONFLICT
+	// statement.
+	PolicyScopeUpsert
 	// PolicyScopeExempt indicates that the operation is exempt from row-level security policies.
 	PolicyScopeExempt
 )

pkg/sql/opt/cat/policy.go

@@ -421,3 +421,94 @@ select count(*) from t1,t2 where t1.c1 = t2.c1;
           table: t2@t2_pkey
           spans: FULL SCAN
           policies: t2_pol_1
+
+# Show that policies are shown for UPSERT (and its variations).
+# ----------------------------------------------------------------------
+
+exec-ddl
+CREATE TABLE ups (pk int not null primary key, comment text);
+----
+
+exec-ddl
+ALTER TABLE ups OWNER TO rls_accessor
+----
+
+exec-ddl
+ALTER TABLE ups ENABLE ROW LEVEL SECURITY, FORCE ROW LEVEL SECURITY;
+----
+
+exec-ddl
+CREATE POLICY p_ins ON ups FOR INSERT USING (comment = 'insert');
+----
+
+exec-ddl
+CREATE POLICY p_sel ON ups FOR SELECT USING (comment = 'select');
+----
+
+exec-ddl
+CREATE POLICY p_upd ON ups FOR UPDATE USING (comment = 'update');
+----
+
+plan
+UPSERT INTO ups VALUES (1, 'first value');
+----
+• upsert
+│ into: ups(pk, comment)
+│ auto commit
+│ arbiter indexes: ups_pkey
+│ policies: p_ins, p_sel, p_upd
+│
+└── • render
+    │
+    └── • cross join (left outer)
+        │
+        ├── • values
+        │     size: 2 columns, 1 row
+        │
+        └── • scan
+              table: ups@ups_pkey
+              spans: 1+ spans
+              locking strength: for update
+
+plan
+INSERT INTO ups VALUES (1, 'first value') ON CONFLICT (pk) DO UPDATE SET comment = 'second value';
+----
+• upsert
+│ into: ups(pk, comment)
+│ auto commit
+│ arbiter indexes: ups_pkey
+│ policies: p_ins, p_sel, p_upd
+│
+└── • render
+    │
+    └── • render
+        │
+        └── • cross join (left outer)
+            │
+            ├── • values
+            │     size: 2 columns, 1 row
+            │
+            └── • scan
+                  table: ups@ups_pkey
+                  spans: 1+ spans
+                  locking strength: for update
+
+plan
+INSERT INTO ups VALUES (1, 'first value') ON CONFLICT (pk) DO NOTHING;
+----
+• insert
+│ into: ups(pk, comment)
+│ auto commit
+│ arbiter indexes: ups_pkey
+│ policies: p_ins, p_sel
+│
+└── • render
+    │
+    └── • cross join (anti)
+        │
+        ├── • values
+        │     size: 2 columns, 1 row
+        │
+        └── • scan
+              table: ups@ups_pkey
+              spans: 1+ spans

pkg/sql/opt/exec/explain/testdata/row_level_security

@@ -310,7 +310,7 @@ func (b *Builder) buildInsert(ins *tree.Insert, inScope *scope) (outScope *scope
 		mb.buildRowLevelBeforeTriggers(tree.TriggerEventInsert, false /* cascade */)
 
 		// Build the final insert statement, including any returned expressions.
-		mb.buildInsert(returning, ins.VectorInsert())
+		mb.buildInsert(returning, ins.VectorInsert(), false /* hasOnConflict */)
 
 	// Case 2: INSERT..ON CONFLICT DO NOTHING.
 	case ins.OnConflict.DoNothing:
@@ -325,7 +325,7 @@ func (b *Builder) buildInsert(ins *tree.Insert, inScope *scope) (outScope *scope
 
 		// Since buildInputForDoNothing filters out rows with conflicts, always
 		// insert rows that are not filtered.
-		mb.buildInsert(returning, false /* vectorInsert */)
+		mb.buildInsert(returning, false /* vectorInsert */, true /* hasOnConflict */)
 
 	// Case 3: UPSERT statement.
 	case ins.OnConflict.IsUpsertAlias():
@@ -425,6 +425,8 @@ func (b *Builder) buildInsert(ins *tree.Insert, inScope *scope) (outScope *scope
 //  4. Each update value is the same as the corresponding insert value.
 //  5. There are no inbound foreign keys containing non-key columns.
 //  6. There are no UPDATE triggers on the target table.
+//  7. Row-level security is disabled for the table. RLS may need to check for
+//     policy violations on the old rows, so we always need to fetch them.
 //
 // TODO(andyk): The fast path is currently only enabled when the UPSERT alias
 // is explicitly selected by the user. It's possible to fast path some queries
@@ -490,7 +492,9 @@ func (mb *mutationBuilder) needExistingRows() bool {
 		}
 	}
 
-	return false
+	// If RLS is enabled, we need to fetch the old rows to check for policy
+	// violations.
+	return mb.tab.IsRowLevelSecurityEnabled()
 }
 
 // addTargetNamedColsForInsert adds a list of user-specified column names to the
@@ -749,13 +753,16 @@ func (mb *mutationBuilder) addSynthesizedColsForInsert() {
 
 // buildInsert constructs an Insert operator, possibly wrapped by a Project
 // operator that corresponds to the given RETURNING clause.
-func (mb *mutationBuilder) buildInsert(returning *tree.ReturningExprs, vectorInsert bool) {
+func (mb *mutationBuilder) buildInsert(
+	returning *tree.ReturningExprs, vectorInsert bool, hasOnConflict bool,
+) {
 	// Disambiguate names so that references in any expressions, such as a
 	// check constraint, refer to the correct columns.
 	mb.disambiguateColumns()
 
 	// Add any check constraint boolean columns to the input.
-	mb.addCheckConstraintCols(false /* isUpdate */)
+	mb.addCheckConstraintCols(false, /* isUpdate */
+		cat.PolicyScopeInsert, returning != nil || hasOnConflict /* includeSelectOnInsert */)
 
 	// Project partial index PUT boolean columns.
 	mb.projectPartialIndexPutCols()
@@ -960,7 +967,8 @@ func (mb *mutationBuilder) buildUpsert(returning *tree.ReturningExprs) {
 	mb.disambiguateColumns()
 
 	// Add any check constraint boolean columns to the input.
-	mb.addCheckConstraintCols(false /* isUpdate */)
+	mb.addCheckConstraintCols(false, /* isUpdate */
+		cat.PolicyScopeUpsert, false /* includeSelectOnInsert */)
 
 	// Add the partial index predicate expressions to the table metadata.
 	// These expressions are used to prune fetch columns during