Re: System column support for partitioned tables using heap

On Tue, Jul 19, 2022 at 10:38 PM Robert Haas <robertmhaas(at)gmail(dot)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;