diff --git a/internal/db/schema/migrations/oss/postgres/81/01_deleted_tables_and_triggers.up.sql b/internal/db/schema/migrations/oss/postgres/81/01_deleted_tables_and_triggers.up.sql
index 6d47514eab..844625ac08 100644
--- a/internal/db/schema/migrations/oss/postgres/81/01_deleted_tables_and_triggers.up.sql
+++ b/internal/db/schema/migrations/oss/postgres/81/01_deleted_tables_and_triggers.up.sql
@@ -308,6 +308,7 @@ begin;
     'affected by the trigger and the current timestamp. It is used to populate rows '
     'of the deleted tables.';
 
+  -- Removed in 91/05_deletion_tables_view and replaced with a view.
   create function get_deletion_tables() returns setof name
   as $$
     select c.relname
diff --git a/internal/db/schema/migrations/oss/postgres/91/05_deletion_tables_view.up.sql b/internal/db/schema/migrations/oss/postgres/91/05_deletion_tables_view.up.sql
new file mode 100644
index 0000000000..ccdf35647c
--- /dev/null
+++ b/internal/db/schema/migrations/oss/postgres/91/05_deletion_tables_view.up.sql
@@ -0,0 +1,18 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+  -- Originially added in 81/01_deleted_tables_and_triggers.up.sql
+  -- This is being replaced with a view.
+  drop function get_deletion_tables;
+
+  -- This view uses the pg_catalog to find all tables that end in _deleted and are visibile.
+  -- See: https://www.postgresql.org/docs/current/catalog-pg-class.html
+  --      https://www.postgresql.org/docs/current/functions-info.html#FUNCTIONS-INFO-SCHEMA
+  create view deletion_table as
+    select c.relname as tablename
+      from pg_catalog.pg_class c
+     where c.relkind in ('r') -- r = ordinary table
+       and c.relname operator(pg_catalog.~) '^(.+_deleted)$' collate pg_catalog.default
+       and pg_catalog.pg_table_is_visible(c.oid);
+commit;
diff --git a/internal/db/sqltest/tests/purge/deleted_table_tests.sql b/internal/db/sqltest/tests/purge/deleted_table_tests.sql
index 9d303f2a04..b0bb36752a 100644
--- a/internal/db/sqltest/tests/purge/deleted_table_tests.sql
+++ b/internal/db/sqltest/tests/purge/deleted_table_tests.sql
@@ -21,6 +21,15 @@ begin;
     );
   $$ language sql;
 
+  -- tests that the deletion table has the bulk insert trigger
+  create function has_bulk_insert_trigger(deletion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_trigger(op_table(deletion_table_name), 'bulk_insert_deleted_ids'),
+      trigger_is(op_table(deletion_table_name), 'bulk_insert_deleted_ids', 'bulk_insert_deleted_ids')
+    );
+  $$ language sql;
+
   -- tests the public_id column
   create function has_public_id(deletion_table_name name) returns text
   as $$
@@ -72,15 +81,32 @@ begin;
     );
   $$ language sql;
 
+  -- like above, but using the bulk delete trigger
+  create function test_bulk_deletion_table(deletion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_correct_tables(deletion_table_name),
+      has_public_id(deletion_table_name),
+      has_delete_time(deletion_table_name),
+      has_delete_time_index(deletion_table_name),
+      has_bulk_insert_trigger(deletion_table_name)
+    );
+  $$ language sql;
+
   -- 11 tests for each deletion table
   select plan(a.table_count::integer)
     from (
       select 11 * count(*) as table_count
-        from get_deletion_tables()
+        from deletion_table
     ) as a;
 
-    select test_deletion_table(a)
-      from get_deletion_tables() a;
+    select test_deletion_table(a.tablename)
+      from deletion_table a
+     where a.tablename not in ('session_deleted');
+
+    select test_bulk_deletion_table(a.tablename)
+      from deletion_table a
+     where a.tablename in ('session_deleted');
 
   select * from finish();
 rollback;
diff --git a/internal/pagination/purge/purge_test.go b/internal/pagination/purge/purge_test.go
index 8b1c41bb42..6ef25ba4fb 100644
--- a/internal/pagination/purge/purge_test.go
+++ b/internal/pagination/purge/purge_test.go
@@ -26,7 +26,7 @@ func TestPurgeTables(t *testing.T) {
 		t.Errorf("error getting db connection %s", err)
 	}
 
-	rows, err := db.Query("select get_deletion_tables()")
+	rows, err := db.Query("select tablename from deletion_table")
 	if err != nil {
 		t.Errorf("unable to query for deletion tables %s", err)
 	}
diff --git a/internal/pagination/purge/query.go b/internal/pagination/purge/query.go
index 966016ccd1..daccd014a9 100644
--- a/internal/pagination/purge/query.go
+++ b/internal/pagination/purge/query.go
@@ -5,7 +5,8 @@ package purge
 
 const (
 	getDeletionTablesQuery = `
-select get_deletion_tables();
+select tablename
+  from deletion_table;
 `
 	deleteQueryTemplate = `
 delete from %s where delete_time < now() - interval '30 days'