sql: populate pg_catalog.pg_trigger

This commit populates the pg_catalog.pg_trigger virtual table,
which provides metadata about database triggers.

The implementation includes:
- Correct calculation of the tgtype bitmap field according to PostgreSQL's
specification, where bits represent:
 - Bit 0: FOR EACH ROW (1) or FOR EACH STATEMENT (0)
 - Bit 1: BEFORE timing
 - Bits 2-4: Event types (INSERT=4, DELETE=8, UPDATE=16)
 - Bit 5: TRUNCATE (32) - reserved for future use
 - Bit 6: INSTEAD OF (64) - reserved for future use
- Support for trigger function arguments stored in tgargs as a null-separated
  bytea array.
- WHEN clause conditions stored in tgqual

The tgenabled field always returns 'A' (always enabled) as CockroachDB does
not yet support trigger enable/disable states.

Tests are added to verify the pg_trigger contents, including the tgtype
bitmap calculation and triggers with function arguments.

Fixes cockroachdb#143534

Release note (sql change): The pg_catalog.pg_trigger table now
returns metadata about database triggers.

Author: rafiss

pkg/ccl/logictestccl/testdata/logic_test/triggers

@@ -4501,12 +4501,12 @@ DROP FUNCTION f1;
 subtest end
 
 # ==============================================================================
-# Test information_schema.triggers
+# Test information_schema.triggers and pg_catalog.pg_trigger
 # ==============================================================================
 
-subtest information_schema_triggers
+subtest trigger_introspection
 
-# Create test tables and trigger functions
+# Create test tables and trigger functions.
 statement ok
 CREATE TABLE test_triggers (
     id INT PRIMARY KEY,
@@ -4601,15 +4601,15 @@ CREATE FUNCTION trigger_func3() RETURNS TRIGGER AS $$
   END;
 $$ LANGUAGE PLpgSQL;
 
-# TODO(sql-queries): Enable tests for statement-level triggers when supported.
+# TODO(#126362, #135655): Enable tests for statement-level triggers when supported.
 # statement ok
 # CREATE TRIGGER after_update_transition
 # AFTER UPDATE ON test_triggers
 # REFERENCING OLD TABLE AS old_data NEW TABLE AS new_data
 # FOR EACH STATEMENT
 # EXECUTE FUNCTION trigger_func3();
 
-# TODO(sql-queries): Enable tests for statement-level triggers when supported.
+# TODO(#126362, #135655): Enable tests for statement-level triggers when supported.
 # query TTTTT colnames
 # SELECT 
 #     trigger_name,
@@ -4671,6 +4671,99 @@ trigger_schema  trigger_count
 other_schema    1
 public          7
 
+# Create a trigger with arguments to test tgnargs and tgargs.
+statement ok
+CREATE TRIGGER trigger_with_args
+AFTER INSERT ON test_triggers
+FOR EACH ROW
+EXECUTE FUNCTION trigger_func1('arg1', 'arg2', 'test value with spaces');
+
+# Test pg_catalog.pg_trigger.
+query TIBITTT colnames
+SELECT 
+    tgname,
+    tgtype,
+    tgfoid > 0 AS has_func_oid,
+    tgnargs,
+    tgenabled,
+    tgoldtable,
+    tgnewtable
+FROM pg_catalog.pg_trigger
+WHERE tgrelid = 'test_triggers'::regclass
+ORDER BY tgname;
+----
+tgname              tgtype  has_func_oid  tgnargs  tgenabled  tgoldtable  tgnewtable
+after_delete_row    9       true          0        A          NULL        NULL
+after_insert_row    5       true          0        A          NULL        NULL
+before_update_row   19      true          0        A          NULL        NULL
+multi_event_trigger 31      true          0        A          NULL        NULL
+trigger_with_args   5       true          3        A          NULL        NULL
+
+# Test tgtype bitmap calculation
+# Bit 0: FOR EACH ROW (1) or FOR EACH STATEMENT (0)
+# Bit 1: BEFORE (2) or AFTER (0)
+# Bits 2-4: INSERT (4), DELETE (8), UPDATE (16)
+# TODO(#126363, #135657): Add tests for INSTEAD OF and TRUNCATE triggers here when supported.
+query TBBBBB colnames
+SELECT 
+    tgname,
+    (tgtype::INT & 1) > 0 AS is_row_level,
+    (tgtype::INT & 2) > 0 AS is_before,
+    (tgtype::INT & 4) > 0 AS has_insert,
+    (tgtype::INT & 8) > 0 AS has_delete,
+    (tgtype::INT & 16) > 0 AS has_update
+FROM pg_catalog.pg_trigger
+WHERE tgrelid = 'test_triggers'::regclass
+ORDER BY tgname;
+----
+tgname               is_row_level  is_before  has_insert  has_delete  has_update
+after_delete_row     true          false      false       true        false
+after_insert_row     true          false      true        false       false
+before_update_row    true          true       false       false       true
+multi_event_trigger  true          true       true        true        true
+trigger_with_args    true          false      true        false       false
+
+# Test trigger with arguments.
+query TIT colnames
+SELECT 
+    tgname,
+    tgnargs,
+    encode(tgargs, 'escape') AS tgargs
+FROM pg_catalog.pg_trigger
+WHERE tgrelid = 'test_triggers'::regclass
+ORDER BY tgname;
+----
+tgname               tgnargs  tgargs
+after_delete_row     0        ·
+after_insert_row     0        ·
+before_update_row    0        ·
+multi_event_trigger  0        ·
+trigger_with_args    3        arg1\000arg2\000test value with spaces\000
+
+# Test trigger with WHEN condition.
+query TTT colnames
+SELECT 
+    tgname,
+    tgqual,
+    tgattr
+FROM pg_catalog.pg_trigger
+WHERE tgrelid = 'test_triggers'::regclass AND tgqual IS NOT NULL;
+----
+tgname             tgqual               tgattr
+before_update_row  ((new).value > 100)  ·
+
+# Test cross-schema trigger.
+query TTT colnames
+SELECT 
+    tgname,
+    tgrelid::regclass::text AS table_name,
+    tgfoid::regproc::text AS trigger_func_name
+FROM pg_catalog.pg_trigger
+WHERE tgrelid = 'other_schema.test_table'::regclass;
+----
+tgname                table_name  trigger_func_name
+other_schema_trigger  test_table  trigger_func1
+
 # Clean up
 statement ok
 DROP TABLE test_triggers CASCADE;

pkg/sql/pg_catalog.go

@@ -3302,14 +3302,127 @@ https://www.postgresql.org/docs/9.5/catalog-pg-tablespace.html`,
 }
 
 var pgCatalogTriggerTable = virtualSchemaTable{
-	comment: `triggers (empty - feature does not exist)
-https://www.postgresql.org/docs/9.5/catalog-pg-trigger.html`,
+	comment: `trigger definitions
+https://www.postgresql.org/docs/16/catalog-pg-trigger.html`,
 	schema: vtable.PGCatalogTrigger,
 	populate: func(ctx context.Context, p *planner, dbContext catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) error {
-		// Triggers are unsupported.
-		return nil
+		h := makeOidHasher()
+		opts := forEachTableDescOptions{virtualOpts: hideVirtual} /* virtual schemas have no triggers */
+		return forEachTableDesc(ctx, p, dbContext, opts,
+			func(ctx context.Context, descCtx tableDescContext) error {
+				tableOid := tableOid(descCtx.table.GetID())
+
+				triggers := descCtx.table.GetTriggers()
+				for i := range triggers {
+					trigger := &triggers[i]
+
+					// Generate a unique OID for the trigger.
+					h.writeTable(descCtx.table.GetID())
+					h.writeUInt32(uint32(trigger.ID))
+					triggerOid := h.getOid()
+
+					// Calculate tgtype bitmap. The bits are defined in Postgres source:
+					// https://github.com/postgres/postgres/blob/44ce4e1593b1821005b29ffaa19d9cbdd80747b2/src/include/catalog/pg_trigger.h#L92-L99
+					// Bit 0: BEFORE (set) or AFTER (unset)
+					// Bit 1: FOR EACH ROW (set) or FOR EACH STATEMENT (unset)
+					// Bits 2-4: One bit for each event type (INSERT=4, DELETE=8, UPDATE=16)
+					// Bit 5: TRUNCATE (64)
+					// Bit 6: INSTEAD OF (128)
+					const tgtypeRow = 1
+					const tgtypeBefore = 1 << 1
+					const tgtypeInsert = 1 << 2
+					const tgtypeDelete = 1 << 3
+					const tgtypeUpdate = 1 << 4
+					const tgtypeTruncate = 1 << 5
+					const tgtypeInstead = 1 << 6
+
+					tgtype := int16(0)
+
+					// Timing bits.
+					switch trigger.ActionTime {
+					case semenumpb.TriggerActionTime_BEFORE:
+						tgtype |= tgtypeBefore
+					case semenumpb.TriggerActionTime_INSTEAD_OF:
+						tgtype |= tgtypeInstead
+					}
+
+					// Row/Statement bit.
+					if trigger.ForEachRow {
+						tgtype |= tgtypeRow
+					}
+
+					// Event type bits.
+					for _, event := range trigger.Events {
+						switch event.Type {
+						case semenumpb.TriggerEventType_INSERT:
+							tgtype |= tgtypeInsert
+						case semenumpb.TriggerEventType_DELETE:
+							tgtype |= tgtypeDelete
+						case semenumpb.TriggerEventType_UPDATE:
+							tgtype |= tgtypeUpdate
+						case semenumpb.TriggerEventType_TRUNCATE:
+							tgtype |= tgtypeTruncate
+						}
+					}
+
+					// tgenabled: O = origin and local, D = disabled, R = replica, A = always
+					tgenabled := tree.NewDString("A")
+					if !trigger.Enabled {
+						tgenabled = tree.NewDString("D")
+					}
+
+					// tgargs: Function arguments as a bytea array.
+					// Format: arg1\000arg2\000...argN\000
+					var tgargs []byte
+					for _, arg := range trigger.FuncArgs {
+						tgargs = append(tgargs, []byte(arg)...)
+						tgargs = append(tgargs, 0) // null terminator
+					}
+
+					// tgattr: Column numbers for UPDATE OF - not implemented.
+					tgattr := tree.NewDIntVectorFromDArray(tree.NewDArray(types.Int2))
+					// tgoldtable/tgnewtable: Transition table names.
+					var oldTableName, newTableName tree.Datum = tree.DNull, tree.DNull
+					if trigger.OldTransitionAlias != "" {
+						oldTableName = tree.NewDName(trigger.OldTransitionAlias)
+					}
+					if trigger.NewTransitionAlias != "" {
+						newTableName = tree.NewDName(trigger.NewTransitionAlias)
+					}
+
+					// tgqual: WHEN condition expression (internal format).
+					var tgqual tree.Datum = tree.DNull
+					if trigger.WhenExpr != "" {
+						tgqual = tree.NewDString(trigger.WhenExpr)
+					}
+
+					if err := addRow(
+						triggerOid,                  // oid
+						tableOid,                    // tgrelid
+						oidZero,                     // tgparentid (partitioning not supported)
+						tree.NewDName(trigger.Name), // tgname
+						tree.NewDOid(catid.FuncIDToOID(trigger.FuncID)), // tgfoid
+						tree.NewDInt(tree.DInt(tgtype)),                 // tgtype
+						tgenabled,                                       // tgenabled
+						tree.DBoolFalse,                                 // tgisinternal
+						oidZero,                                         // tgconstrrelid (foreign key table)
+						oidZero,                                         // tgconstrindid (constraint index)
+						oidZero,                                         // tgconstraint (constraint oid)
+						tree.DBoolFalse,                                 // tgdeferrable
+						tree.DBoolFalse,                                 // tginitdeferred
+						tree.NewDInt(tree.DInt(len(trigger.FuncArgs))), // tgnargs
+						tgattr,                              // tgattr
+						tree.NewDBytes(tree.DBytes(tgargs)), // tgargs
+						tgqual,                              // tgqual
+						oldTableName,                        // tgoldtable
+						newTableName,                        // tgnewtable
+					); err != nil {
+						return err
+					}
+				}
+				return nil
+			})
 	},
-	unimplemented: true,
 }
 
 var (

pkg/sql/vtable/pg_catalog.go

@@ -759,11 +759,12 @@ CREATE TABLE pg_catalog.pg_tablespace (
 )`
 
 // PGCatalogTrigger describes the schema of the pg_catalog.pg_trigger table.
-// https://www.postgresql.org/docs/9.5/catalog-pg-trigger.html,
+// https://https://www.postgresql.org/docs/16/catalog-pg-trigger.html
 const PGCatalogTrigger = `
 CREATE TABLE pg_catalog.pg_trigger (
 	oid OID,
 	tgrelid OID,
+	tgparentid OID,
 	tgname NAME,
 	tgfoid OID,
 	tgtype INT2,
@@ -779,8 +780,7 @@ CREATE TABLE pg_catalog.pg_trigger (
 	tgargs BYTEA,
 	tgqual TEXT,
 	tgoldtable NAME,
-	tgnewtable NAME,
-	tgparentid OID
+	tgnewtable NAME
 )`
 
 // PGCatalogType describes the schema of the pg_catalog.pg_type table.