@@ -187,7 +187,7 @@ export function readSql(sql: string, dialect?: string): string {
       body,
     ) ||
     /\bREPLACE\b(?!\s*\()/i.test(body) ||
-    /\b(nextval|setval|set_config|load_extension|writefile|dblink_exec|lo_import|lo_export|lo_create|lo_unlink|lo_put|pg_write_file|pg_append_file|pg_file_write|pg_file_rename|pg_file_unlink|pg_advisory_lock|pg_advisory_xact_lock|pg_terminate_backend|pg_cancel_backend)\s*\(/i.test(
+    /\b(nextval|setval|set_config|load_extension|writefile|dblink_exec|dblink_send_query|dblink_connect|dblink_connect_u|lo_import|lo_export|lo_create|lo_unlink|lo_put|pg_write_file|pg_append_file|pg_file_write|pg_file_rename|pg_file_unlink|pg_advisory_lock|pg_advisory_xact_lock|pg_terminate_backend|pg_cancel_backend|pg_promote|pg_reload_conf)\s*\(/i.test(
       functions,
     )
   ) {
@@ -328,6 +328,7 @@ export async function checkSql(input: SqlCheckPublicInput) {
   async function run(query: string): Promise<Row[]> {
     const response = await connector.execute(query, undefined, undefined, {
       noLimit: true,
+      readOnly: true,
       queryTimeoutMs: args.query_timeout_ms,
     })
     if (response.truncated) throw new Error("sql_check received truncated aggregates; no verdict is available")
 
@@ -24,12 +24,16 @@ afterEach(() => {
   Registry.reset()
 })
 
-function postgres(relkind = "r", reltuples = 5, translate = (sql: string) => sql) {
+function postgres(
+  relkind = "r",
+  reltuples = 5,
+  translate = (sql: string, _options?: Parameters<Connector["execute"]>[3]) => sql,
+) {
   database.run("INSERT INTO pg_class VALUES (1, ?, ?)", [relkind, reltuples])
-  const execute = mock(async (...[sql]: Parameters<Connector["execute"]>) => {
+  const execute = mock(async (...[sql, , , options]: Parameters<Connector["execute"]>) => {
     // Keep the generated WHERE and FILTER clauses executable: removing either
     // guard condition must change results, rather than leave a canned mock row.
-    const query = translate(sql)
+    const query = translate(sql, options)
       .replace("pg_catalog.pg_class", "pg_class")
       .replace(/pg_catalog\.to_regclass\(E'(?:[^']|'')*'\)/, "1")
     const statement = database.query(query)
@@ -41,6 +45,96 @@ function postgres(relkind = "r", reltuples = 5, translate = (sql: string) => sql
 }
 
 describe("predicate_effect with mocked Postgres", () => {
+  test.each([
+    { name: "predicates", base_sql: "orders", predicates: ["custom_write_udf() > 0"] as string[], calls: 2 },
+    {
+      name: "populations",
+      base_sql: "SELECT * FROM orders WHERE custom_write_udf() > 0",
+      predicates: ["id > 0"] as string[],
+      calls: 1,
+    },
+  ])("uses read-only execution to reject writing UDFs in $name", async ({ base_sql, predicates, calls }) => {
+    let writes = 0
+    const execute = postgres("r", 5, (sql, options) => {
+      if (!sql.includes("custom_write_udf()")) return sql
+      // Model the connector's transaction boundary for a function unknown to the denylist.
+      if (options?.readOnly) throw new Error("cannot execute INSERT in a read-only transaction")
+      writes++
+      return sql.replaceAll("custom_write_udf()", "1")
+    })
+    await rejects(
+      checkSql({ warehouse, check: "predicate_effect", base_sql, predicates, query_timeout_ms: 4321 }),
+      /read-only transaction/,
+    )
+    expect(writes).toBe(0)
+    expect(execute).toHaveBeenCalledTimes(calls)
+    for (const call of execute.mock.calls) {
+      expect(call[3]).toMatchObject({ noLimit: true, queryTimeoutMs: 4321, readOnly: true })
+    }
+  })
+
+  for (const [kind, calls] of [
+    [
+      "server-control",
+      [
+        "pg_promote()",
+        'pg_catalog."pg_promote" /* promote */ ()',
+        "PG_RELOAD_CONF()",
+        'pg_catalog."pg_reload_conf" -- reload\n()',
+      ],
+    ],
+    [
+      "dblink-write",
+      [
+        "dblink_send_query('remote', 'DELETE FROM important_table')",
+        `public."dblink_send_query" /* send */ ('remote', 'DELETE FROM important_table')`,
+        "DBLINK_CONNECT('remote', 'dbname=remote')",
+        `public."dblink_connect" -- connect\n('remote', 'dbname=remote')`,
+        "dblink_connect_u('remote', 'dbname=remote')",
+        `public."dblink_connect_u" /* connect */ ('remote', 'dbname=remote')`,
+      ],
+    ],
+  ] satisfies [string, string[]][]) {
+    test.each(calls)(`rejects ${kind} predicate %s before query execution`, async (call) => {
+      const predicate = `${call} IS NOT NULL`
+      const execute = postgres("r", 5, (sql) => sql.replaceAll(call, "1"))
+      await rejects(
+        checkSql({ warehouse, check: "predicate_effect", base_sql: "orders", predicates: [predicate] }),
+        /sql_check only accepts a single read-only SELECT/,
+      )
+      expect(execute).not.toHaveBeenCalled()
+    })
+
+    test.each(calls)(`rejects ${kind} population %s before query execution`, async (call) => {
+      const execute = postgres("r", 5, (sql) => sql.replaceAll(call, "1"))
+      await rejects(
+        checkSql({
+          warehouse,
+          check: "predicate_effect",
+          base_sql: `SELECT ${call} AS value FROM orders`,
+          predicates: ["value IS NOT NULL"],
+        }),
+        /sql_check only accepts a single read-only SELECT/,
+      )
+      expect(execute).not.toHaveBeenCalled()
+    })
+  }
+
+  test.each(["pg_promote", "pg_reload_conf", "dblink_send_query", "dblink_connect", "dblink_connect_u"])(
+    "accepts %s in literals, comments and column identifiers",
+    async (name) => {
+      const execute = postgres()
+      const result = await checkSql({
+        warehouse,
+        check: "predicate_effect",
+        base_sql: `SELECT id AS "${name}" FROM orders`,
+        predicates: [`"${name}" > 2 -- ${name}()`, `length('${name}()') > 0 /* ${name}() */`],
+      })
+      expect(result.metadata).toMatchObject({ total_rows: 5, combined_kept_rows: 3 })
+      expect(execute).toHaveBeenCalledTimes(1)
+    },
+  )
+
   test.each([
     "pg_terminate_backend(id)",
     "PG_TERMINATE_BACKEND(id)",