feat: pg_replication support Enum column types

[bar-bitkover-lendbuzz]

Source name

pg_replication

Describe the data you'd like to see

We have tables in our PostgreSQL database that include columns of the Enum type. When DLT receives a DML command (such as INSERT) via pgoutput for a table with an Enum column, the operation ID (op id) is 89. Currently, this operation type is not supported by DLT (see the process_msg function), which leads to replication failures for these tables.

It would be helpful if DLT could recognize and correctly process operation ID 89 for Enum columns.

Are you a dlt user?

I'm considering using dlt, but this bug is preventing this.

Do you ready to contribute this extension?

No.

dlt destination

Snowflake

Additional information

When attempting to replicate an INSERT operation into a table with an Enum column, we encounter the following error log:

extraction of resource `sync_replication_slot` in `generator` `replication_resource` caused an exception: Unknown replication op 89

In order to test it locally I created an Enum type and altered an existing table:

-- Create enum type for status
CREATE TYPE enum_status AS ENUM ('created', 'in_progress', 'done');

-- Add status column to search_presets table
ALTER TABLE test_table 
ADD COLUMN status enum_status NOT NULL DEFAULT 'created';

-- Insert a new row with the status enum
INSERT INTO test_table (status) 
VALUES ('in_progress');

If you need more details or example schemas, please let me know.

[idonissim-lendbuzz]

Following up on this feature request.

We did some digging in the PG replication message spec and even wrote the PgType decoder in decoders.py, but ultimately understood it's not required.

If the destination doesn't need to enforce the constraint (enum in our case), then the message can be safely ignored as the enum value is sent as cleartext in the INSERT message later.

Tiny code change to helpers.py that solves the enum breakage and follows similar convention you already used for Truncate messages:

    def process_msg(self, msg: ReplicationMessage) -> None:
        """Processes encoded replication message.

        Identifies message type and decodes accordingly.
        Message treatment is different for various message types.
        Breaks out of stream with StopReplication exception when
        - `upto_lsn` is reached
        - `target_batch_size` is reached
        - a table's schema has changed
        """
        op = msg.payload[0]
        if op == 73:  # ASCII for 'I'
            self.process_change(Insert(msg.payload), msg.data_start)
        elif op == 85:  # ASCII for 'U'
            self.process_change(Update(msg.payload), msg.data_start)
        elif op == 68:  # ASCII for 'D'
            self.process_change(Delete(msg.payload), msg.data_start)
        elif op == 66:  # ASCII for 'B'
            self.last_commit_ts = Begin(msg.payload).commit_ts  # type: ignore[assignment]
        elif op == 67:  # ASCII for 'C'
            self.process_commit(msg)
        elif op == 82:  # ASCII for 'R'
            self.process_relation(Relation(msg.payload))
        elif op == 89:  # ASCII for 'Y'
            logger.warning(
                "The Type operation is currently not supported. "
                "Type replication messages are ignored."
            )
        elif op == 84:  # ASCII for 'T'
            logger.warning(
                "The truncate operation is currently not supported. "
                "Truncate replication messages are ignored."
            )
        else:
            raise ValueError(f"Unknown replication op {op}")

@bar-bitkover-lendbuzz @VioletM I think this feature request can safely be closed or pushed back for when there is an actual requirement to enforce ENUM values in the destination.