bug: Error saying 'enum already exists' when trying to load pre-existing table using stream-maps

[edgarrmondragon]

Target Version

0.0.14

Python Version

3.12

PostgreSQL Version

Do not know

Operating System

NA

Description

hello team! I'm using stream_maps (tap-postgres) to drop a column and sync the data into target-postgres. Config:

- name: tap-postgress
  config:
    stream_maps:
      public-table:
        column1_text: column1
        column1: __NULL__
  metadata:
    public-*:
      replication-method: FULL_TABLE
  select:
    - public-table.*

On the target side the table has been created manually (not from meltano) and has all the columns (column1 and column1_text)
When I run the sync meltano run tap-postgres target-postgres it fails with the error that enum already exist (column1 is type ENUM). If I drop the column1 (on target side) the sync works correctly and in the Inserting with SQL logs I can see that the data doesn't have any info regarding column1. column1_text has the data of column1 as it's setup on stream_maps tap side. Is there any configuration I'm missing in order to have column1 column on target side as well or no?

Link to Slack/Linen

https://meltano.slack.com/archives/C069CQNHDNF/p1724772642004549

[tbroka]

The trace is

File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context 
     self.dialect.do_execute(   
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 941, 
     cursor.execute(statement, parameters) 
 psycopg2.errors.DuplicateObject: type "companycategory" already exists 
 The above exception was the direct cause of the following exception: 
                                
 Traceback (most recent call last): 
   File "/project/.meltano/loaders/target-postgres/venv/bin/target-postgres", line 8, in <module> 
     sys.exit(TargetPostgres.cli()) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/click/core.py", line 1157, in __call__ 
     return self.main(*args, **kwargs) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/click/core.py", line 1078, in main 
     rv = self.invoke(ctx)      
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/singer_sdk/plugin_base.py", line 80, in invoke 
     return super().invoke(ctx) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/click/core.py", line 1434, in invoke 
     return ctx.invoke(self.callback, **ctx.params) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/click/core.py", line 783, in invoke 
     return __callback(*args, **kwargs) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/singer_sdk/target_base.py", line 567, in invoke 
     target.listen(file_input)  
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/singer_sdk/io_base.py", line 36, in listen 
     self._process_lines(file_input) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/singer_sdk/target_base.py", line 307, in _process_lines 
     counter = super()._process_lines(file_input) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/singer_sdk/io_base.py", line 95, in _process_lines 
     self._process_record_message(line_dict) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/singer_sdk/target_base.py", line 371, in _process_record_message 
     self.drain_one(sink)       
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/singer_sdk/target_base.py", line 512, in drain_one 
     sink.process_batch(draining_status) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/target_postgres/sinks.py", line 85, in process_batch 
     temp_table: sa.Table = self.connector.copy_table_structure( 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/target_postgres/connector.py", line 185, in copy_table_structure 
     new_table.create(bind=connection) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/sql/schema.py", line 1284, in create 
     bind._run_ddl_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 2459, in _run_ddl_visitor 
     visitorcallable(self.dialect, self, **kwargs).traverse_single(element) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/sql/visitors.py", line 664, in traverse_single 
     return meth(obj, **kw)     
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/sql/ddl.py", line 938, in visit_table 
     with self.with_ddl_events( 
   File "/pyenv/versions/3.8.15/lib/python3.8/contextlib.py", line 113, in __enter__ 
     return next(self.gen)      
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/sql/ddl.py", line 822, in with_ddl_events 
     target.dispatch.before_create( 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/event/attr.py", line 497, in __call__ 
     fn(*args, **kw)            
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 852, in __call__ 
     return getattr(self.target, self.name)(*arg, **kw) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/dialects/postgresql/named_types.py", line 98, in _on_table_create 
     self.create(bind=bind, checkfirst=checkfirst) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/dialects/postgresql/named_types.py", line 338, in create 
     super().create(bind, checkfirst=checkfirst) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/dialects/postgresql/named_types.py", line 51, in create 
     bind._run_ddl_visitor(self.DDLGenerator, self, checkfirst=checkfirst) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 2459, in _run_ddl_visitor 
     visitorcallable(self.dialect, self, **kwargs).traverse_single(element) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/sql/visitors.py", line 664, in traverse_single 
     return meth(obj, **kw)     
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/dialects/postgresql/named_types.py", line 153, in visit_enum 
     self.connection.execute(CreateEnumType(enum)) 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1418, in execute 
     return meth(               
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/sql/ddl.py", line 180, in _execute_on_connection 
     return connection._execute_ddl( 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1529, in _execute_ddl 
     ret = self._execute_context( 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1846, in _execute_context 
     return self._exec_single_context( 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1986, in _exec_single_context 
     self._handle_dbapi_exception( 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 2355, in _handle_dbapi_exception 
     raise sqlalchemy_exception.with_traceback(exc_info[2]) from e 
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context 
     self.dialect.do_execute(   
   File "/project/.meltano/loaders/target-postgres/venv/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 941, 
     cursor.execute(statement, parameters) 
 sqlalchemy.exc.ProgrammingError: (psycopg2.errors.DuplicateObject) type "companycategory" already exists 
                                
 [SQL: CREATE TYPE companycategory AS ENUM ('ONE', 'TWO', 'THREE', 'FOUR')] 
 (Background on this error at: https://sqlalche.me/e/20/f405)

The type companycategory has been created like

CREATE TYPE public.companycategory AS ENUM (
    'ONE',
    'TWO',
    'THREE',
    'FOUR'
);

column1 in target postgres is companycategory type

[grigi]

I have the same issue. Diagnosing it, when it creates a temp table, sqlalchemy doesn't know that the enum does not already exist, and so will try to re-create it. This causes the error.

It works when I simply comment out the code that creates the enum type out of sqlalchemy. This is not a solution.

Changing the instance of sqlalchemy.dialects.postgresql.named_types.ENUM to have create_type=False (Edit: typo) now changes the type to varchar instead and generate a different type issue.

I'm still investigating what path I'll ultimately end up taking. Unfortunately this is a case where we are fighting SqlAlchemy's smarts being dumb.

[grigi]

Turns out that native enum types in postgres breaks on some of our apps, so I'm "fixing" both this and that issue by defining it as a varchar with constraint instead.
Not ideal, but this isn't a large table.

[grigi]

FYI: the changes I did to move the needle past creating an already created enum was:

in copy_table_structure() of target_postgres/connector.py (Approx line 230)

        columns = [column._copy() for column in from_table.columns]
        for column in columns:
            if isinstance(column.type, sqlalchemy.dialects.postgresql.named_types.ENUM):
                column.type.create_type = False

This would bypass creating the duplicate type in the DB, but would fail because the cast was never applied to convert the string data we get from JSON to the correct type.

I think that technically target-postgres has enough data to do this, It knows the target schema had an enum type, and it should define the cast on insert into the temp table.

This would not apply if it had to create the table itself as the singer schema only exposes the source as a string, it would just create a varchar instead.

I'm not going to spend any more effort on this right now as we don't need this. I'm just dumping my findings if someone else wants to continue with it.