hello, I’m having trouble getting the upsert loading to work with a tap-postgres (transferwise variant) with target-redshift (ticketswap variant). What I mean by this is that records are sinking into the right schema/table but for some reason its working like “append-only” giving multiple versions of the record instead of just upserting one entry per record. I’m wondering what are the requirements to get upsert to work on target-redshift. Here is one snippet of my meltano.yml:

environments:
- name: prod
  config:
    plugins:
      loaders:
      - name: target-redshift
        config: 
          load_method: upsert
plugins:
  extractors:
  - name: tap-postgres
    variant: transferwise
    pip_url: 
      git+<https://github.com/transferwise/pipelinewise.git#subdirectory=singer-connectors/tap-postgres>
    config:
      dbname: ${src_dbname}
      default_replication_method: LOG_BASED
      filter_schemas: public
      host: ${src_host}
      port: ${src_port}
      user: ${src_username}
      password: ${src_password}
    metadata:
      public-new_risk_decision:
        replication-method: LOG_BASED
        replication-key: _sdc_lsn
        table-key-properties: ["id"]
  loaders:
  - name: target-redshift
    variant: ticketswap
    pip_url: git+<https://github.com/TicketSwap/target-redshift.git>
    config:
      dbname: ${DB_ENV}
      default_target_schema: ...
      host: ${REDSHIFT_HOST}
      load_method: upsert
      user: ${tgt_username}
      password: ${tgt_password}
      s3_bucket: ...
      s3_key_prefix: ...
      aws_redshift_copy_role_arn:  ...
      port: '5439'
      s3_region: us-east-1
      ssl_enable: true
      ssl_mode: require
      validate_records: false
      batch_size_rows: 1000000
      activate_version: true
      add_record_metadata: true

I've seen this too, for mssql log_based into target-redshift. I had to fork to fix this. Basically, it is attempting to upsert, but the problem is the messages coming from the tap are not unique by the key property. since it is log_based replication method, the messages can include multiple messages for one key property in the same batch, which are not reduced to one message before upsert, and the upsert does not remove duplicates either. This is a full explanation of the issue and fix on my end

## Problem
The resulting tables are not unique by the key property field.

## Why
This is due to the style of upserts done in target-redshift. It works as expected when the tap data is unique by the primary key, but with log-based CDC into target-redshift, we are sending non-unique ids when the target is expecting uniqueness.

more than one record is produced by the target for a given id in the source, one for each change made since the latest run (each change is captured, with CDC - change data capture)
we still configure a primary key of id implying uniqueness
records are not loaded one by one like other targets often are, but in bulk
In the target, Redshift's MERGE INTO is set to use REMOVE DUPLICATES, however Redshift's documentation notes that

> Rows in target_table can't match multiple rows in source_table.

source_table here means the records we are loading and target_table means the redshift table we want to upsert to.

Fix
We need target-redshift to apply only the final change from the change data capture records, but we still want all the records to land in S3 for our reference. Each record contains the state of the db row at the point of that change, and a few columns noting what the type of change was: insert, update, delete. The latest row is the current state, and what we want in the redshift table.

Meltano numbers the records that come through the tap/target in _sdc_sequence. We choose the record with the highest _sdc_sequence within each primary key partition, which ensures only one source row matching each target row in the merge command.

With this, I also had to increase the precision of

_sdc_sequence

to actually be unique.

We deal with multiple changes to one primary key in the same run by ordering on _sdc_sequence. However, _sdc_sequence is time-based and can have the same value for two entries if they are processed in the same millisecond. Increasing this to nanosecond makes _sdc_sequence unique

oh wow. Ok

Without forking, you could try a different replication_method. incremental or full_table

I actually want to try to fork the repo, possibly contribute. Never had to before so this would be cool if its ok

yeah go for it! it's great to contribute

got the code to work. The changes were perfect

yes the incremental upsert does not have this issue since it's pulling from the database table as opposed to the CDC table. the latter is a list of changes which is not unique by the primary key whereas the former would be unique record metadata is not optional AFAIK when you want it to upsert instead of append. it has to know what makes the rows unique