PostgreSQL触发器设计及Hello World测试问题排查

背景需求 #

我需要在PostgreSQL中实现触发器（优先通过SQLAlchemy ORM），基础测试目标是：每次插入employee表行时，自动将字符串"Hello world"写入该行的test列。

表结构定义（SQLAlchemy代码） #

我用以下代码创建了带继承结构的表：

from __future__ import annotations
from typing import List
import sqlalchemy
from sqlalchemy import create_engine, ForeignKey, Column, Integer, String, CheckConstraint, MetaData
from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, mapped_column, relationship
from sqlalchemy.schema import DDL
import datetime

dbUrl = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'

engine = create_engine(dbUrl, echo=True)

metadata = MetaData()

class Base(MappedAsDataclass, DeclarativeBase):
    pass

class Company(Base):
    __tablename__ = "company"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    managers: Mapped[List[Manager]] = relationship(back_populates="company")

Company.__table__
Base.metadata.create_all(engine)

class Employee(Base):
    __tablename__ = "employee"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    type: Mapped[str]
    test: Mapped[str] = mapped_column(nullable=True)

    __mapper_args__ = {
        "polymorphic_identity": "employee",
        "polymorphic_on": "type",
    }

Employee.__table__
Base.metadata.create_all(engine)

class Manager(Employee):
    __tablename__ = "manager"
    id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
    manager_name: Mapped[str]
    CheckConstraint("manager_name == employee.name", name="check1")

    company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
    company: Mapped[Company] = relationship(back_populates="managers")

    __mapper_args__ = {
        "polymorphic_identity": "manager",
    }

class Engineer(Employee):
    __tablename__ = "engineer"
    id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
    engineer_name: Mapped[str]

    company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
    company: Mapped[Company] = relationship(back_populates="engineers")

    __mapper_args__ = {
        "polymorphic_identity": "engineer",
    }  
Engineer.__table__
Base.metadata.create_all(engine)

初始触发器尝试的问题 #

1. SQLAlchemy中添加触发器无报错但未生效 #

我尝试用SQLAlchemy的事件监听添加触发器：

sqlalchemy.event.listen(metadata,
                       "before_create",
                       DDL("""CREATE TRIGGER autoCreateEmployee
                       AFTER INSERT
                       ON employee
                       FOR EACH ROW
                       INSERT INTO employee(test)
                       VALUES("Hello world"),
                       0)  
                       """))

表创建成功无报错，但触发器并未被添加到数据库中。

2. 直接在DBeaver中执行触发器语句报错 #

我把触发器语句拿到DBeaver中测试：

CREATE TRIGGER autoCreateEmployee
                       AFTER INSERT
                       ON employee
                       FOR EACH ROW
                       INSERT INTO employee(test)
                       VALUES("Hello world")

得到语法错误：

SQL Error [42601]: ERROR: syntax error at or near "INSERT"
Position: 165
Error position: line: 5 pos: 164

疑问：触发器语句到底哪里有问题？为什么SQLAlchemy中运行没报错？

调整后的问题：触发器执行逻辑错误 #

后来我改用PL/pgSQL函数+触发器的方式，能成功创建函数和触发器，但逻辑不符合预期：

当前PL/pgSQL函数 #

CREATE FUNCTION hello_world()
RETURNS trigger
LANGUAGE PLPGSQL
AS $hello_world$
    BEGIN
        INSERT INTO employee(test) VALUES ('HELLO WORLD!');
    END;
$hello_world$

触发器创建语句 #

CREATE TRIGGER hello
    AFTER INSERT ON employee
    FOR EACH ROW
    EXECUTE FUNCTION hello_world();

执行问题 #

当插入数据：

INSERT INTO employee (name,type) VALUES ('Merv', 'manager');

触发报错：

SQL Error [23502]: ERROR: null value in column "name" of relation "employee" violates not-null constraint
Detail: Failing row contains (31, null, null, HELLO WORLD!).
Where: SQL statement "INSERT INTO employee(test) VALUES ('HELLO WORLD!')"
PL/pgSQL function hello_world() line 3 at SQL statement

显然触发器执行时新建了一条只有test列有值的空行，而不是更新刚插入的目标行。我误解了FOR EACH ROW的作用，现在需要解决：如何让触发函数定位到刚插入的那一行并更新它的test列。

内容的提问来源于stack exchange，提问作者Logos Masters