sql: implement pg_catalog.pg_policies table

These code changes underneath the hood run a query on
`pg_policy`, `pg_class`, `pg_namespace` and `pg_roles`
to fill all the data in the pg_policies table.

Fixes: #143444
Epic: CRDB-11724
Release note: none

Author: Dedej-Bergin

pkg/cli/clisqlshell/testdata/describe

@@ -559,7 +559,8 @@ pg_catalog,pg_operator,table,node,permanent,prefix,"operators (incomplete)
 https://www.postgresql.org/docs/9.5/catalog-pg-operator.html"
 pg_catalog,pg_opfamily,table,node,permanent,prefix,pg_opfamily was created for compatibility and is currently unimplemented
 pg_catalog,pg_partitioned_table,table,node,permanent,prefix,pg_partitioned_table was created for compatibility and is currently unimplemented
-pg_catalog,pg_policies,table,node,permanent,prefix,pg_policies was created for compatibility and is currently unimplemented
+pg_catalog,pg_policies,table,node,permanent,prefix,"pg_policies provides a user-friendly view of row-level security policies
+https://www.postgresql.org/docs/17/view-pg-policies.html"
 pg_catalog,pg_policy,table,node,permanent,prefix,"stores row-level security policies for tables
 https://www.postgresql.org/docs/17/catalog-pg-policy.html"
 pg_catalog,pg_prepared_statements,table,node,permanent,prefix,"prepared statements

pkg/sql/delegate/show_policies.go

@@ -10,32 +10,14 @@ import "github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 func (d *delegator) delegateShowPolicies(stmt *tree.ShowPolicies) (tree.Statement, error) {
 	query := `
 			SELECT 
-			p.polname AS name,
-			CASE p.polcmd::text
-					WHEN '*' THEN 'ALL'
-					WHEN 'a' THEN 'INSERT'
-					WHEN 'w' THEN 'UPDATE'
-					WHEN 'd' THEN 'DELETE'
-					WHEN 'r' THEN 'SELECT'
-			END AS cmd,
-			CASE p.polpermissive
-					WHEN true THEN 'permissive'
-					ELSE 'restrictive'
-			END AS type,
-			array_agg(
-					CASE 
-							WHEN role_id.uid = 0 THEN 'public'
-							ELSE r.rolname
-					END
-					ORDER BY r.rolname
-			) AS roles,
-			COALESCE(p.polqual::text, '') AS using_expr,
-			COALESCE(p.polwithcheck::text, '') AS with_check_expr
-			FROM pg_policy p
-			LEFT JOIN LATERAL unnest(p.polroles) AS role_id(uid) ON true
-			LEFT JOIN pg_catalog.pg_roles r ON r.oid = role_id.uid
-			WHERE p.polrelid = %[6]d
-			GROUP BY p.polname, p.polcmd, p.polpermissive, p.polqual, p.polwithcheck`
+			policyname AS name,
+			cmd,
+			permissive AS type,
+			roles,
+			COALESCE(qual, '') AS using_expr,
+			COALESCE(with_check, '') AS with_check_expr
+			FROM pg_catalog.pg_policies
+			WHERE schemaname = %[5]s AND tablename = %[2]s`
 
 	return d.showTableDetails(stmt.Table, query)
 }

pkg/sql/logictest/testdata/logic_test/crdb_internal

@@ -123,7 +123,7 @@ pg_opclass                       true
 pg_operator                      false
 pg_opfamily                      true
 pg_partitioned_table             true
-pg_policies                      true
+pg_policies                      false
 pg_policy                        false
 pg_prepared_statements           false
 pg_prepared_xacts                false

pkg/sql/logictest/testdata/logic_test/row_level_security

@@ -142,7 +142,7 @@ statement ok
 CREATE POLICY "policy7" ON multi_pol_tab1 FOR SELECT
 
 statement ok
-CREATE USER papa_roach;
+CREATE USER papa_roach
 
 statement ok
 CREATE POLICY "policy8" ON multi_pol_tab1 FOR ALL TO papa_roach, public
@@ -176,6 +176,19 @@ oid  polname  polrelid  polcmd  polpermissive  polroles       polqual  polwithch
 7    policy7  110       r       true           {0}            NULL     NULL
 8    policy8  110       *       true           {0,835509264}  NULL     NULL
 
+query TTTTTTTT colnames,rowsort
+select schemaname, tablename, policyname, permissive, roles, cmd, qual, with_check from pg_catalog.pg_policies
+----
+schemaname  tablename       policyname  permissive   roles                cmd     qual  with_check
+public      multi_pol_tab1  policy1     permissive   {public}             ALL     NULL  NULL
+public      multi_pol_tab1  policy2     restrictive  {public}             ALL     NULL  NULL
+public      multi_pol_tab1  policy3     permissive   {public}             ALL     NULL  NULL
+public      multi_pol_tab1  policy4     permissive   {public}             INSERT  NULL  NULL
+public      multi_pol_tab1  policy5     permissive   {public}             UPDATE  NULL  NULL
+public      multi_pol_tab1  policy6     permissive   {public}             DELETE  NULL  NULL
+public      multi_pol_tab1  policy7     permissive   {public}             SELECT  NULL  NULL
+public      multi_pol_tab1  policy8     permissive   {public,papa_roach}  ALL     NULL  NULL
+
 query TTTTTT colnames,rowsort
 SHOW POLICIES FOR multi_pol_tab1
 ----
@@ -189,6 +202,23 @@ policy6  DELETE  permissive   {public}             ·           ·
 policy7  SELECT  permissive   {public}             ·           ·
 policy8  ALL     permissive   {public,papa_roach}  ·           ·
 
+statement ok
+CREATE DATABASE roachdb
+
+statement ok
+USE roachdb
+
+query TTTTTTTT colnames,rowsort
+select schemaname, tablename, policyname, permissive, roles, cmd, qual, with_check from pg_catalog.pg_policies
+----
+schemaname  tablename  policyname  permissive  roles  cmd  qual  with_check
+
+statement ok
+USE db1
+
+statement ok
+DROP DATABASE roachdb
+
 statement ok
 CREATE TABLE multi_pol_tab2 (c1 INT NOT NULL PRIMARY KEY)
 
@@ -224,15 +254,29 @@ query ITITBTTT colnames,rowsort
 select oid::INT, polname, polrelid::INT, polcmd, polpermissive, polroles::string, polqual, polwithcheck from pg_catalog.pg_policy WHERE polrelid = 'multi_pol_tab2'::regclass
 ----
 oid  polname   polrelid  polcmd  polpermissive  polroles  polqual  polwithcheck
-1    policy9   111       *       true           {0}       NULL     NULL
-2    policy10  111       *       true           {0}       NULL     NULL
+1    policy9   113       *       true           {0}       NULL     NULL
+2    policy10  113       *       true           {0}       NULL     NULL
+
+query TTTTTTTT colnames,rowsort
+select schemaname, tablename, policyname, permissive, roles, cmd, qual, with_check from pg_catalog.pg_policies where tablename = 'multi_pol_tab2'
+----
+schemaname  tablename       policyname  permissive  roles     cmd  qual  with_check
+public      multi_pol_tab2  policy9     permissive  {public}  ALL  NULL  NULL
+public      multi_pol_tab2  policy10    permissive  {public}  ALL  NULL  NULL
 
 query ITITBTTT colnames,rowsort
 select oid::INT, polname, polrelid::INT, polcmd, polpermissive, polroles::string, polqual, polwithcheck from pg_catalog.pg_policy WHERE polrelid = 'multi_pol_tab3'::regclass
 ----
 oid  polname   polrelid  polcmd  polpermissive  polroles  polqual  polwithcheck
-1    policy11  112       *       true           {0}       NULL     NULL
-2    policy12  112       *       true           {0}       NULL     NULL
+1    policy11  114       *       true           {0}       NULL     NULL
+2    policy12  114       *       true           {0}       NULL     NULL
+
+query TTTTTTTT colnames,rowsort
+select schemaname, tablename, policyname, permissive, roles, cmd, qual, with_check from pg_catalog.pg_policies where tablename = 'multi_pol_tab3'
+----
+schemaname  tablename       policyname  permissive  roles     cmd  qual  with_check
+public      multi_pol_tab3  policy11    permissive  {public}  ALL  NULL  NULL
+public      multi_pol_tab3  policy12    permissive  {public}  ALL  NULL  NULL
 
 query TTTTTT colnames,rowsort
 SHOW POLICIES FOR multi_pol_tab2
@@ -500,6 +544,12 @@ select using_expr from [SHOW POLICIES FOR funcref];
 ----
 public.is_valid(c1)
 
+query TT colnames,rowsort
+select qual, with_check from pg_catalog.pg_policies where tablename = 'funcref'
+----
+qual                 with_check
+public.is_valid(c1)  public.is_valid(c1)
+
 statement error pq: cannot drop function "is_valid" because other objects \(\[db1.public.funcref\]\) still depend on it
 DROP FUNCTION is_valid;
 

pkg/sql/pg_catalog.go

@@ -4098,12 +4098,89 @@ var pgCatalogStatioAllSequencesTable = virtualSchemaTable{
 }
 
 var pgCatalogPoliciesTable = virtualSchemaTable{
-	comment: "pg_policies was created for compatibility and is currently unimplemented",
-	schema:  vtable.PgCatalogPolicies,
-	populate: func(ctx context.Context, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) error {
+	comment: `pg_policies provides a user-friendly view of row-level security policies
+https://www.postgresql.org/docs/17/view-pg-policies.html`,
+	schema: vtable.PgCatalogPolicies,
+	populate: func(ctx context.Context, p *planner, db catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) error {
+		query := `
+			SELECT 
+				n.nspname,
+				c.relname,
+				pol.polname,
+				pol.polpermissive,
+				array_agg(
+					CASE 
+						WHEN role_id.uid = 0 THEN 'public'
+						ELSE r.rolname
+					END
+					ORDER BY r.rolname
+				) AS roles,
+				CASE pol.polcmd::text
+					WHEN '*' THEN 'ALL'
+					WHEN 'a' THEN 'INSERT'
+					WHEN 'w' THEN 'UPDATE'
+					WHEN 'd' THEN 'DELETE'
+					WHEN 'r' THEN 'SELECT'
+				END AS cmd,
+				pol.polqual,
+				pol.polwithcheck
+			FROM pg_catalog.pg_policy pol
+			JOIN pg_catalog.pg_class c ON pol.polrelid = c.oid
+			JOIN pg_catalog.pg_namespace n ON c.relnamespace = n.oid
+			LEFT JOIN LATERAL unnest(pol.polroles) AS role_id(uid) ON true
+			LEFT JOIN pg_catalog.pg_roles r ON r.oid = role_id.uid
+			GROUP BY n.nspname, c.relname, pol.polname, pol.polpermissive, pol.polcmd, pol.polqual, pol.polwithcheck
+		`
+
+		rows, err := p.InternalSQLTxn().QueryBufferedEx(
+			ctx, "read-policies", p.txn,
+			sessiondata.NodeUserSessionDataOverride,
+			query,
+		)
+		if err != nil {
+			return err
+		}
+
+		for _, row := range rows {
+			schemaName := tree.MustBeDString(row[0])
+			tableName := tree.MustBeDString(row[1])
+			policyName := tree.MustBeDString(row[2])
+			isPermissive := tree.MustBeDBool(row[3])
+			roles := tree.MustBeDArray(row[4]) // This is now already a string array of role names
+			cmd := tree.MustBeDString(row[5])
+			qual := row[6]
+			withCheck := row[7]
+
+			// Convert permissive to string
+			permissive := "permissive"
+			if !isPermissive {
+				permissive = "restrictive"
+			}
+
+			// Create a NAME array for roles
+			roleNames := tree.NewDArray(types.Name)
+			for _, role := range roles.Array {
+				roleName := tree.MustBeDString(role)
+				if err := roleNames.Append(tree.NewDName(string(roleName))); err != nil {
+					return err
+				}
+			}
+
+			if err := addRow(
+				tree.NewDName(string(schemaName)), // schemaname
+				tree.NewDName(string(tableName)),  // tablename
+				tree.NewDName(string(policyName)), // policyname
+				tree.NewDString(permissive),       // permissive
+				roleNames,                         // roles
+				tree.NewDString(string(cmd)),      // cmd (already in correct format from query)
+				qual,                              // qual
+				withCheck,                         // with_check
+			); err != nil {
+				return err
+			}
+		}
 		return nil
 	},
-	unimplemented: true,
 }
 
 var pgCatalogStatsExtTable = virtualSchemaTable{

pkg/sql/vtable/pg_catalog.go

@@ -1249,7 +1249,8 @@ CREATE TABLE pg_catalog.pg_stat_gssapi (
 	encrypted BOOL
 )`
 
-// PgCatalogPolicies is an empty table in the pg_catalog that is not implemented yet
+// PgCatalogPolicies provides a user-friendly view of row-level security policies.
+// https://www.postgresql.org/docs/17/view-pg-policies.html
 const PgCatalogPolicies = `
 CREATE TABLE pg_catalog.pg_policies (
 	schemaname NAME,