Re: System column support for partitioned tables using heap - Mailing list pgsql-hackers

On Tue, Jul 19, 2022 at 10:38 PM Robert Haas <robertmhaas@gmail.com> wrote:

> For MERGE itself, I wonder if some information about this should be
> included in the command tag. It looks like MERGE already includes some
> sort of row count in the command tag, but I guess perhaps it doesn't
> distinguish between inserts and updates. I don't know why we couldn't
> expose multiple values this way, though.

It would be great to get some sort of feedback from MERGE accessible through SQL results, even if that doesn't come in the form of a RETURNING list.

> I wonder whether you could just have the CTEs bubble up 1 or 0 and
> then sum them at some stage, instead of relying on xmax. Presumably
> your UPSERT simulation knows which thing it did in each case.

It might help if I show a sample insert handling function. The issue is with the line at the end of the top CTE, insert_rows:

        returning xmax as inserted_transaction_id),

That's what fails on partitions. Is there an alternative way to test what happened to the row(s)? here's the full function. . I wrote a code generator, so I don't have to hand-code all of these bits for each table+version:

-- Create a function to accept an array of rows formatted as item_type_v1 for UPSERT into item_type.
DROP FUNCTION IF EXISTS types_plus.insert_item_type_v1 (types_plus.item_type_v1[]);

CREATE OR REPLACE FUNCTION types_plus.insert_item_type_v1 (data_in types_plus.item_type_v1[])

RETURNS TABLE (
   insert_count            integer,
   estimated_update_count  integer,
   transaction_id          text)

LANGUAGE SQL

BEGIN ATOMIC

-- The CTE below is a roundabout way of returning an insertion count from a pure SQL function in Postgres.
WITH
inserted_rows as (
        INSERT INTO item_type (
id,
marked_for_deletion,
name_)

        SELECT
rows_in.id,
rows_in.marked_for_deletion,
rows_in.name_

        FROM unnest(data_in) as rows_in

        ON CONFLICT(id) DO UPDATE SET
marked_for_deletion = EXCLUDED.marked_for_deletion,
name_               = EXCLUDED.name_

        returning xmax as inserted_transaction_id),

status_data AS (
 select count(*) FILTER (where inserted_transaction_id  = 0) AS insert_count,
        count(*) FILTER (where inserted_transaction_id != 0) AS estimated_update_count,
        pg_current_xact_id_if_assigned()::text               AS transaction_id

   from inserted_rows),

insert_log_entry AS (
       INSERT INTO insert_log (
           data_file_id,
           ib_version,
           job_run_id,

  schema_name,
  table_name,
  records_submitted,
  insert_count,
  estimated_update_count)

SELECT
   coalesce_session_variable(
         'data_file_id',
         '00000000000000000000000000000000')::uuid,

   coalesce_session_variable('ib_version'),  -- Default result is ''

   coalesce_session_variable(
         'job_run_id',
         '00000000000000000000000000000000')::uuid,

   'ascendco',
   'item_type',
   (select cardinality(data_in)),
   insert_count,
   estimated_update_count

  FROM status_data
)

-- Final output/result.
   select insert_count,
          estimated_update_count,
          transaction_id

      from status_data;

END;