Merge #143587

143587: sql: add RLS policy support for UPSERT r=spilchen a=spilchen

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 behaviour, 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
Closes #144802 

Epic: CRDB-11724
Release note: none

Co-authored-by: Matt Spilchen <[email protected]>

Author: craig[bot]

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