`sql_raw` output should support batched/transactional sends when possible

I have a processing pipeline that takes a few Kafka topic inputs and upserts the records in a Postgres database. There is a switch output config that selects the output resource by the input topic. When using Benthos 3.65 this pipeline is very fast, it upserts more than one million Kafka records in the DB in a few minutes. After upgrading to Benthos 4.11 the processing became much slower and I also see errors after a few tens of thousands records are processed.

The input for Benthos 3.65 is configured like this:

resources:
  inputs:
    input-topics:
      kafka:
      ...
        topics: T1, T2
        checkpoint_limit: 1
        batching:
          count: 1500
          period: 100ms      

The output for Benthos 3.65 is configured using a sql configuration like this:

  outputs:
    output-table1:
      retry:
        max_retries: 2
        backoff:
          initial_interval: 500ms
          max_interval: 3s
        output:
          sql:
            max_in_flight: 1
            driver: postgres
            data_source_name: postgres://...
           query: INSERT INTO output_table1 (..) VALUES (..) ON CONFLICT(pk) DO UPDATE SET(..) = (..)
     output-table2:

After upgrading to Benthos 4.11 we had to change the output config since sql output config is not supported anymore:

  outputs:
    output-table1:
      retry:
        max_retries: 2
        backoff:
          initial_interval: 500ms
          max_interval: 3s
        output:
          sql_raw:
            max_in_flight: 1
            batching:
              count: 1500
              period: 1s
            driver: postgres
            dsn: postgres://...
            query: INSERT INTO output_table (..) VALUES (..) ON CONFLICT(pk) DO UPDATE SET(..) = (..)

    output-table2:
         ...

I've set max_in_flight=1 to ensure the upserts are correctly ordered. I also tried using the sql_insert configuration but I don't know how to correctly configure an ON CONFLICT.. DO UPDATE SET with params and also I am seeing errors when multiple records with the same key are part of a batch (since it's not possible to have a batch insert in a DB tx for records having the same key).

What am I doing wrong? Is there a config option that could bring this config to the 3.65 performance level?
Thanks!
florin

Start by tracing the sql_raw output's batching and transaction behavior and comparing it with the former sql output, using the supplied Kafka and Postgres configurations as reproductions. Check the sql_insert behavior for duplicate keys and ON CONFLICT handling. Done means batched transactional upserts preserve ordering and handle duplicate keys without the reported errors while restoring comparable throughput.