sql/rls: refactor rls filter injection

This does a refactor of the logic that adds the filter for row-level
security (RLS) policies. The intent is to match what we did for the
check constraint. Namely a single function (addRowLevelSecurityFilter)
that you can quickly see at a glance what policies are applied to the
various policy command scope.

There is a slight update to the optbuilder tests as we include redundant
clauses that didn't exist before. These will get optimized out.

Epic: none
Release note: none

Author: spilchen

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

@@ -293,7 +293,7 @@ UPDATE writer SET key = key * 10;
 └── • render
     │
     └── • norows
-          policies: row-level security enabled, no policies applied.
+          policies: applied (filtered all rows)
 
 exec-ddl
 CREATE POLICY p_update1 ON writer FOR UPDATE USING (key = 5)  WITH CHECK (value like 'updater: %')
@@ -396,6 +396,11 @@ UPDATE writer SET key = 10 WHERE true;
 # Show policy information for delete
 # ----------------------------------------------------------------------
 
+# First ensure no policies apply.
+exec-ddl
+DROP POLICY p_select ON writer;
+----
+
 plan
 DELETE FROM writer WHERE key = 1;
 ----
@@ -406,6 +411,22 @@ DELETE FROM writer WHERE key = 1;
 └── • norows
       policies: row-level security enabled, no policies applied.
 
+# Create a SELECT policy again. But still no delete policy, so everything is
+# filtered.
+exec-ddl
+CREATE POLICY p_select ON writer FOR SELECT USING (true);
+----
+
+plan
+DELETE FROM writer WHERE key = 1;
+----
+• delete
+│ from: writer
+│ auto commit
+│
+└── • norows
+      policies: applied (filtered all rows)
+
 exec-ddl
 CREATE POLICY p_delete ON writer FOR DELETE USING (key = 1);
 ----

pkg/sql/opt/optbuilder/row_level_security.go

@@ -28,11 +28,28 @@ func (b *Builder) addRowLevelSecurityFilter(
 	if b.isExemptFromRLSPolicies(tabMeta, cmdScope) {
 		return
 	}
-	scalar := b.buildRowLevelSecurityUsingExpression(tabMeta, tableScope, cmdScope)
-	if scalar != nil {
-		tableScope.expr = b.factory.ConstructSelect(tableScope.expr,
-			memo.FiltersExpr{b.factory.ConstructFiltersItem(scalar)})
+
+	var scalar opt.ScalarExpr
+	switch cmdScope {
+	case cat.PolicyScopeSelect:
+		scalar = b.genPolicyUsingExpr(tabMeta, tableScope, cat.PolicyScopeSelect)
+	case cat.PolicyScopeUpdate:
+		scalar = b.factory.ConstructAnd(
+			b.genPolicyUsingExpr(tabMeta, tableScope, cat.PolicyScopeSelect),
+			b.genPolicyUsingExpr(tabMeta, tableScope, cat.PolicyScopeUpdate),
+		)
+	case cat.PolicyScopeDelete:
+		scalar = b.factory.ConstructAnd(
+			b.genPolicyUsingExpr(tabMeta, tableScope, cat.PolicyScopeSelect),
+			b.genPolicyUsingExpr(tabMeta, tableScope, cat.PolicyScopeDelete),
+		)
+	default:
+		panic(errors.AssertionFailedf("unsupported policy command scope for filter: %v", cmdScope))
 	}
+
+	tableScope.expr = b.factory.ConstructSelect(tableScope.expr,
+		memo.FiltersExpr{b.factory.ConstructFiltersItem(scalar)})
+	b.factory.Metadata().GetRLSMeta().RefreshNoPoliciesAppliedForTable(tabMeta.MetaID)
 }
 
 // isExemptFromRLSPolicies will check if the given user is exempt from RLS policies.
@@ -65,52 +82,11 @@ func (b *Builder) isExemptFromRLSPolicies(
 	return false
 }
 
-// buildRowLevelSecurityUsingExpression generates a scalar expression for read
-// operations by combining all applicable RLS policies. An expression is always
-// returned; if no policies apply, a 'false' expression is returned.
-func (b *Builder) buildRowLevelSecurityUsingExpression(
-	tabMeta *opt.TableMeta, tableScope *scope, cmdScope cat.PolicyCommandScope,
-) opt.ScalarExpr {
-	combinedExpr := b.combinePolicyUsingExpressionForCommand(tabMeta, tableScope, cmdScope)
-	if combinedExpr == nil {
-		// No policies, filter out all rows by adding a "false" expression.
-		b.factory.Metadata().GetRLSMeta().NoPoliciesApplied = true
-		return memo.FalseSingleton
-	}
-	return b.buildScalar(combinedExpr, tableScope, nil, nil, nil)
-}
-
-// combinePolicyUsingExpressionForCommand generates a `tree.TypedExpr` command
-// for use in the scan. Depending on the policy command scope, it may simply
-// pass through to `buildRowLevelSecurityUsingExpressionForCommand`.
-// For other commands, the process is more complex and may involve multiple
-// calls to that function to generate the expression.
-func (b *Builder) combinePolicyUsingExpressionForCommand(
-	tabMeta *opt.TableMeta, tableScope *scope, cmdScope cat.PolicyCommandScope,
-) tree.TypedExpr {
-	// For DELETE and UPDATE, we always apply SELECT/ALL policies because
-	// reading the existing row is necessary before performing the write.
-	switch cmdScope {
-	case cat.PolicyScopeUpdate, cat.PolicyScopeDelete:
-		selectExpr := b.buildRowLevelSecurityUsingExpressionForCommand(tabMeta, tableScope, cat.PolicyScopeSelect)
-		if selectExpr == nil {
-			return selectExpr
-		}
-		updateExpr := b.buildRowLevelSecurityUsingExpressionForCommand(tabMeta, tableScope, cmdScope)
-		if updateExpr == nil {
-			return nil
-		}
-		return tree.NewTypedAndExpr(selectExpr, updateExpr)
-	default:
-		return b.buildRowLevelSecurityUsingExpressionForCommand(tabMeta, tableScope, cmdScope)
-	}
-}
-
-// buildRowLevelSecurityUsingExpressionForCommand will generate a tree.TypedExpr
+// genPolicyUsingExpr will generate a opt.ScalarExpr
 // for all policies that apply to the given policy command scope.
-func (b *Builder) buildRowLevelSecurityUsingExpressionForCommand(
+func (b *Builder) genPolicyUsingExpr(
 	tabMeta *opt.TableMeta, tableScope *scope, cmdScope cat.PolicyCommandScope,
-) tree.TypedExpr {
+) opt.ScalarExpr {
 	var policiesUsed opt.PolicyIDSet
 	var combinedExpr tree.TypedExpr
 	policies := tabMeta.Table.Policies()
@@ -147,8 +123,8 @@ func (b *Builder) buildRowLevelSecurityUsingExpressionForCommand(
 		buildForPolicy(policy, false /* restrictive */)
 	}
 	if combinedExpr == nil {
-		// No permissive policies. Return an empty expr to force the caller to generate a deny-all expression.
-		return nil
+		// No permissive policies. Return the false expr as a deny-all expression.
+		return memo.FalseSingleton
 	}
 	for _, policy := range policies.Restrictive {
 		buildForPolicy(policy, true /* restrictive */)
@@ -159,7 +135,7 @@ func (b *Builder) buildRowLevelSecurityUsingExpressionForCommand(
 		panic(errors.AssertionFailedf("at least one applicable policy should have been found"))
 	}
 	b.factory.Metadata().GetRLSMeta().AddPoliciesUsed(tabMeta.MetaID, policiesUsed, true /* applyFilterExpr */)
-	return combinedExpr
+	return b.buildScalar(combinedExpr, tableScope, nil, nil, nil)
 }
 
 // policyAppliesToCommandScope checks whether a given PolicyCommandScope applies

pkg/sql/opt/optbuilder/testdata/row_level_security

@@ -115,7 +115,7 @@ update t1
       │    │    │    │    ├── columns: c1:7 c2:8 c3:9 rowid:10!null crdb_internal_mvcc_timestamp:11 tableoid:12
       │    │    │    │    └── flags: avoid-full-scan
       │    │    │    └── filters
-      │    │    │         └── false
+      │    │    │         └── ((c1:7 % 2) = 0) AND false
       │    │    └── filters
       │    │         └── c1:7 > 0
       │    └── projections
@@ -516,7 +516,7 @@ delete t1
       │    │    ├── columns: c1:7 c2:8 c3:9 rowid:10!null crdb_internal_mvcc_timestamp:11 tableoid:12
       │    │    └── flags: avoid-full-scan
       │    └── filters
-      │         └── false
+      │         └── ((c1:7 % 2) = 0) AND false
       └── filters
            └── (c1:7 >= 0) AND (c1:7 <= 20)
 
@@ -781,7 +781,7 @@ update t1
       │    │    │    ├── columns: c1:7 c2:8 c3:9 rowid:10!null crdb_internal_mvcc_timestamp:11 tableoid:12
       │    │    │    └── flags: avoid-full-scan
       │    │    └── filters
-      │    │         └── false
+      │    │         └── true AND false
       │    └── projections
       │         └── 'new val' [as=c2_new:13]
       └── projections

pkg/sql/opt/row_level_security.go

@@ -99,8 +99,10 @@ func (r *RowLevelSecurityMeta) GetPoliciesUsed(
 // is complete for a given table. If no policies were recorded for the table,
 // it updates the 'NoPoliciesApplied' flag to indicate this.
 func (r *RowLevelSecurityMeta) RefreshNoPoliciesAppliedForTable(tableID TableID) {
-	if _, ok := r.PoliciesApplied[tableID]; !ok {
+	if a, ok := r.PoliciesApplied[tableID]; !ok {
 		r.NoPoliciesApplied = true
+	} else {
+		r.NoPoliciesApplied = a.Filter.Len() == 0 && a.Check.Len() == 0
 	}
 }