From 71807eea741b96ce7d3c68699455a8de87871103 Mon Sep 17 00:00:00 2001 From: Wenbo Date: Wed, 5 Aug 2026 10:56:51 +0800 Subject: [PATCH 1/6] feat: support COPY ON CONFLICT DO NOTHING (AnalyticDB-compatible syntax) Add the DO ON CONFLICT clause to COPY FROM, compatible with Alibaba Cloud AnalyticDB for PostgreSQL: COPY table FROM ... DO ON CONFLICT DO NOTHING When a row conflicts with a unique/exclusion constraint, it is skipped instead of aborting the whole COPY. DO ON CONFLICT DO UPDATE is parsed but rejected at startup with a clear 'not supported yet' error. Implementation notes: - gram.y: new standalone copy_on_conflict clause following the copy options (works with both old-style and parenthesized option syntax) - copy.c: defGetCopyOnConflictChoice() with fail-fast validation (COPY TO rejected; unknown values rejected; DO UPDATE not supported) - copyfrom.c: force CIM_SINGLE insertion when ON CONFLICT is specified; per-tuple ExecCheckIndexConstraints() pre-check before insert; skipped rows are counted and reported via NOTICE. Leaf partitions get their speculative index info (BuildSpeculativeIndexInfo) filled in on first use because COPY's ModifyTable plan node is NULL. - Target table (non-partitioned) must have at least one unique index; checked at startup. Limitations (documented): conflict check is a pre-check (TOCTOU window under concurrency); partitioned tables enforce per-leaf uniqueness; DO UPDATE is not implemented yet. --- src/backend/commands/copy.c | 45 +++++++++++++ src/backend/commands/copyfrom.c | 116 +++++++++++++++++++++++++++++++- src/backend/parser/gram.y | 27 ++++++-- src/include/commands/copy.h | 15 ++++- 4 files changed, 195 insertions(+), 8 deletions(-) diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index cc226da0615..6e219ea38f4 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -503,6 +503,43 @@ defGetCopyOnErrorChoice(DefElem *def, ParseState *pstate, bool is_from) return COPY_ON_ERROR_STOP; /* keep compiler quiet */ } +/* + * Extract a CopyOnConflictChoice value from a DefElem. + * + * The DO ON CONFLICT DO UPDATE clause is parsed but not yet implemented; + * reject it at startup so that users get a clear error instead of a silent + * fallback to plain COPY. + */ +static CopyOnConflictChoice +defGetCopyOnConflictChoice(DefElem *def, ParseState *pstate, bool is_from) +{ + char *sval = defGetString(def); + + if (!is_from) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + /*- translator: first %s is the name of a COPY option, e.g. ON_CONFLICT, + second %s is a COPY with direction, e.g. COPY TO */ + errmsg("COPY %s cannot be used with %s", "ON_CONFLICT", "COPY TO"), + parser_errposition(pstate, def->location))); + + if (pg_strcasecmp(sval, "nothing") == 0) + return COPY_ON_CONFLICT_NOTHING; + if (pg_strcasecmp(sval, "update") == 0) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("COPY ON CONFLICT DO UPDATE is not supported yet"), + errhint("Use DO ON CONFLICT DO NOTHING, or DELETE/TRUNCATE the conflicting rows before COPY, or use INSERT ... ON CONFLICT DO UPDATE."), + parser_errposition(pstate, def->location))); + + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + /*- translator: first %s is the name of a COPY option, e.g. ON_CONFLICT */ + errmsg("COPY %s \"%s\" not recognized", "ON_CONFLICT", sval), + parser_errposition(pstate, def->location))); + return COPY_ON_CONFLICT_NONE; /* keep compiler quiet */ +} + /* * Extract REJECT_LIMIT value from a DefElem. * @@ -587,6 +624,7 @@ ProcessCopyOptions(ParseState *pstate, bool freeze_specified = false; bool header_specified = false; bool on_error_specified = false; + bool on_conflict_specified = false; bool log_verbosity_specified = false; bool reject_limit_specified = false; bool force_array_specified = false; @@ -760,6 +798,13 @@ ProcessCopyOptions(ParseState *pstate, on_error_specified = true; opts_out->on_error = defGetCopyOnErrorChoice(defel, pstate, is_from); } + else if (strcmp(defel->defname, "on_conflict") == 0) + { + if (on_conflict_specified) + errorConflictingDefElem(defel, pstate); + on_conflict_specified = true; + opts_out->on_conflict = defGetCopyOnConflictChoice(defel, pstate, is_from); + } else if (strcmp(defel->defname, "log_verbosity") == 0) { if (log_verbosity_specified) diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index 64ac3063c61..48d4faccce7 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -26,6 +26,7 @@ #include "access/heapam.h" #include "access/tableam.h" +#include "catalog/index.h" #include "access/tupconvert.h" #include "access/xact.h" #include "catalog/namespace.h" @@ -798,6 +799,7 @@ CopyFrom(CopyFromState cstate) CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */ int64 processed = 0; int64 excluded = 0; + int64 skipped = 0; /* rows skipped by ON CONFLICT DO NOTHING */ bool has_before_insert_row_trig; bool has_instead_insert_row_trig; bool leafpart_use_multi_insert = false; @@ -923,7 +925,14 @@ CopyFrom(CopyFromState cstate) /* Verify the named relation is a valid target for INSERT */ CheckValidResultRel(resultRelInfo, CMD_INSERT, ONCONFLICT_NONE, NIL); - ExecOpenIndices(resultRelInfo, false); + /* + * With ON CONFLICT we need the per-index unique operator info that is + * built only in speculative mode (BuildSpeculativeIndexInfo), since + * ExecCheckIndexConstraints() looks up existing conflicting tuples via + * index scans keyed on those operators. + */ + ExecOpenIndices(resultRelInfo, + cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE); /* * Set up a ModifyTableState so we can let FDW(s) init themselves for @@ -986,6 +995,41 @@ CopyFrom(CopyFromState cstate) cstate->qualexpr = ExecInitQual(castNode(List, cstate->whereClause), &mtstate->ps); + /* + * COPY ON CONFLICT needs a unique constraint to detect conflicts. For + * partitioned tables the check is done per leaf partition at runtime + * (each partition enforces its own uniqueness), so skip the startup + * check there. + */ + if (cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE && + cstate->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) + { + List *indexoidlist; + ListCell *lc; + bool has_unique = false; + + indexoidlist = RelationGetIndexList(resultRelInfo->ri_RelationDesc); + foreach(lc, indexoidlist) + { + Oid indexoid = lfirst_oid(lc); + Relation indexRel; + + indexRel = index_open(indexoid, AccessShareLock); + if (indexRel->rd_index->indisunique) + has_unique = true; + index_close(indexRel, AccessShareLock); + if (has_unique) + break; + } + list_free(indexoidlist); + + if (!has_unique) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("COPY ON CONFLICT requires a unique constraint on table \"%s\"", + RelationGetRelationName(cstate->rel)))); + } + /* * It's generally more efficient to prepare a bunch of tuples for * insertion, and insert them in one @@ -994,7 +1038,16 @@ CopyFrom(CopyFromState cstate) * However, there are a number of reasons why we might not be able to do * this. These are explained below. */ - if (resultRelInfo->ri_TrigDesc != NULL && + if (cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE) + { + /* + * COPY ON CONFLICT performs per-tuple conflict checks, which are + * incompatible with multi-insert batching; force single-tuple + * insertion. + */ + insertMethod = CIM_SINGLE; + } + else if (resultRelInfo->ri_TrigDesc != NULL && (resultRelInfo->ri_TrigDesc->trig_insert_before_row || resultRelInfo->ri_TrigDesc->trig_insert_instead_row)) { @@ -1369,6 +1422,59 @@ CopyFrom(CopyFromState cstate) (proute == NULL || has_before_insert_row_trig)) ExecPartitionCheck(resultRelInfo, myslot, estate, true); + /* + * With DO ON CONFLICT DO NOTHING, check the tuple against + * the unique/exclusion constraints of the target table (or + * leaf partition) and skip it on conflict instead of + * aborting the whole COPY. The check runs after BEFORE + * INSERT triggers and constraint checks, matching the + * ordering of INSERT ... ON CONFLICT. Note that this is a + * pre-check: under concurrent insertion a unique violation + * may still be raised by the index insert itself. + */ + if (cstate->opts.on_conflict == COPY_ON_CONFLICT_NOTHING) + { + ItemPointerData conflictTid; + ItemPointerData invalidItemPtr; + int ii; + + ItemPointerSetInvalid(&invalidItemPtr); + + /* + * Leaf partitions reached via tuple routing had their + * indexes opened in ExecInitPartitionInfo() without the + * speculative flag (COPY's ModifyTable plan node is + * NULL), so the per-index unique operator info needed + * by ExecCheckIndexConstraints() is missing. Fill it + * in on first use. + */ + if (resultRelInfo != target_resultRelInfo && + resultRelInfo->ri_IndexRelationInfo != NULL) + { + for (ii = 0; ii < resultRelInfo->ri_NumIndices; ii++) + { + IndexInfo *idxinfo = resultRelInfo->ri_IndexRelationInfo[ii]; + Relation idxrel = resultRelInfo->ri_IndexRelationDescs[ii]; + + if (idxinfo->ii_Unique && idxinfo->ii_UniqueProcs == NULL && + idxrel != NULL) + BuildSpeculativeIndexInfo(idxrel, idxinfo); + } + } + + if (!ExecCheckIndexConstraints(resultRelInfo, myslot, + estate, &conflictTid, + &invalidItemPtr, NIL)) + { + skipped++; + + pgstat_progress_update_param(PROGRESS_COPY_TUPLES_SKIPPED, + skipped); + + continue; /* skip this tuple */ + } + } + /* Store the slot in the multi-insert buffer, when enabled. */ if (insertMethod == CIM_MULTI || leafpart_use_multi_insert) { @@ -1482,6 +1588,12 @@ CopyFrom(CopyFromState cstate) cstate->num_errors)); } + if (cstate->opts.on_conflict == COPY_ON_CONFLICT_NOTHING && skipped > 0) + ereport(NOTICE, + errmsg_plural("%" PRIu64 " row was skipped due to ON CONFLICT DO NOTHING", + "%" PRIu64 " rows were skipped due to ON CONFLICT DO NOTHING", + skipped, skipped)); + if (bistate != NULL) FreeBulkInsertState(bistate); diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 0382929a639..17cdfa32f58 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -501,7 +501,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type opt_instead %type opt_unique opt_verbose opt_full %type opt_freeze opt_analyze opt_default -%type opt_binary copy_delimiter +%type opt_binary copy_delimiter copy_on_conflict %type copy_from opt_program @@ -3553,7 +3553,7 @@ ClosePortalStmt: CopyStmt: COPY opt_binary qualified_name opt_column_list copy_from opt_program copy_file_name copy_delimiter opt_with - copy_options where_clause + copy_options copy_on_conflict where_clause { CopyStmt *n = makeNode(CopyStmt); @@ -3563,7 +3563,7 @@ CopyStmt: COPY opt_binary qualified_name opt_column_list n->is_from = $5; n->is_program = $6; n->filename = $7; - n->whereClause = $11; + n->whereClause = $12; if (n->is_program && n->filename == NULL) ereport(ERROR, @@ -3576,7 +3576,7 @@ CopyStmt: COPY opt_binary qualified_name opt_column_list (errcode(ERRCODE_SYNTAX_ERROR), errmsg("WHERE clause not allowed with COPY TO"), errhint("Try the COPY (SELECT ... WHERE ...) TO variant."), - parser_errposition(@11))); + parser_errposition(@12))); n->options = NIL; /* Concatenate user-supplied flags */ @@ -3586,6 +3586,8 @@ CopyStmt: COPY opt_binary qualified_name opt_column_list n->options = lappend(n->options, $8); if ($10) n->options = list_concat(n->options, $10); + if ($11) + n->options = lappend(n->options, $11); $$ = (Node *) n; } | COPY '(' PreparableStmt ')' TO opt_program copy_file_name opt_with copy_options @@ -3635,6 +3637,23 @@ copy_options: copy_opt_list { $$ = $1; } | '(' copy_generic_opt_list ')' { $$ = $2; } ; +/* COPY ON CONFLICT (AnalyticDB-compatible), a standalone clause that may + * follow either the old-style or the new parenthesized option syntax. */ +copy_on_conflict: + DO ON CONFLICT DO UPDATE + { + $$ = makeDefElem("on_conflict", (Node *) makeString("update"), @1); + } + | DO ON CONFLICT DO NOTHING + { + $$ = makeDefElem("on_conflict", (Node *) makeString("nothing"), @1); + } + | /* EMPTY */ + { + $$ = NULL; + } + ; + /* old COPY option syntax */ copy_opt_list: copy_opt_list copy_opt_item { $$ = lappend($1, $2); } diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h index abecfe51098..33b7216346b 100644 --- a/src/include/commands/copy.h +++ b/src/include/commands/copy.h @@ -28,8 +28,18 @@ #define COPY_HEADER_TRUE 1 /* - * Represents where to save input processing errors. More values to be added - * in the future. + * What to do when a row conflicts with a unique constraint during COPY + * FROM. DO ON CONFLICT DO UPDATE is parsed but not yet implemented. + */ +typedef enum CopyOnConflictChoice +{ + COPY_ON_CONFLICT_NONE, /* no ON CONFLICT clause */ + COPY_ON_CONFLICT_NOTHING, /* DO ON CONFLICT DO NOTHING */ + COPY_ON_CONFLICT_UPDATE /* DO ON CONFLICT DO UPDATE (not yet implemented) */ +} CopyOnConflictChoice; + +/* + * Represents the type of an error to be handled by COPY FROM. */ typedef enum CopyOnErrorChoice { @@ -93,6 +103,7 @@ typedef struct CopyFormatOptions bool force_null_all; /* FORCE_NULL *? */ bool *force_null_flags; /* per-column CSV FN flags */ bool convert_selectively; /* do selective binary conversion? */ + CopyOnConflictChoice on_conflict; /* behavior when unique constraint conflict happens */ CopyOnErrorChoice on_error; /* what to do when error happened */ CopyLogVerbosityChoice log_verbosity; /* verbosity of logged messages */ int64 reject_limit; /* maximum tolerable number of errors */ From c6841ec72b987f833cba315dcbee3d8950ae1cea Mon Sep 17 00:00:00 2001 From: Wenbo Date: Wed, 5 Aug 2026 10:59:24 +0800 Subject: [PATCH 2/6] test: add regression tests for COPY ON CONFLICT DO NOTHING Covers: primary key conflict skip, no-conflict passthrough, startup rejection without unique constraint, COPY TO rejection, DO UPDATE 'not supported yet' error, multi-column unique index, partitioned tables (per-leaf uniqueness), ON_ERROR ignore interaction, parenthesized option syntax, and unknown on_conflict value rejection. --- .../regress/expected/copy_on_conflict.out | 105 ++++++++++++++++++ src/test/regress/parallel_schedule | 3 + src/test/regress/sql/copy_on_conflict.sql | 75 +++++++++++++ 3 files changed, 183 insertions(+) create mode 100644 src/test/regress/expected/copy_on_conflict.out create mode 100644 src/test/regress/sql/copy_on_conflict.sql diff --git a/src/test/regress/expected/copy_on_conflict.out b/src/test/regress/expected/copy_on_conflict.out new file mode 100644 index 00000000000..2d638c506ff --- /dev/null +++ b/src/test/regress/expected/copy_on_conflict.out @@ -0,0 +1,105 @@ +-- COPY ON CONFLICT (AnalyticDB-compatible DO ON CONFLICT DO NOTHING) +DROP TABLE IF EXISTS copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, + copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, + copy_on_conflict_t4 CASCADE; +NOTICE: table "copy_on_conflict_t1" does not exist, skipping +NOTICE: table "copy_on_conflict_t2" does not exist, skipping +NOTICE: table "copy_on_conflict_t3" does not exist, skipping +NOTICE: table "copy_on_conflict_tp" does not exist, skipping +NOTICE: table "copy_on_conflict_tp_lo" does not exist, skipping +NOTICE: table "copy_on_conflict_tp_hi" does not exist, skipping +NOTICE: table "copy_on_conflict_t4" does not exist, skipping +CREATE TABLE copy_on_conflict_t1 (id int PRIMARY KEY, name text); +INSERT INTO copy_on_conflict_t1 VALUES (1, 'old-1'), (2, 'old-2'); +-- 1. DO NOTHING skips conflicting rows, inserts the rest +COPY copy_on_conflict_t1 FROM STDIN DO ON CONFLICT DO NOTHING; +NOTICE: 2 rows were skipped due to ON CONFLICT DO NOTHING +SELECT * FROM copy_on_conflict_t1 ORDER BY id; + id | name +----+------- + 1 | old-1 + 2 | old-2 + 3 | new-3 +(3 rows) + +-- 2. no-conflict data behaves like plain COPY +COPY copy_on_conflict_t1 FROM STDIN DO ON CONFLICT DO NOTHING; +SELECT count(*) FROM copy_on_conflict_t1; + count +------- + 4 +(1 row) + +-- 3. table without unique constraint is rejected at startup +CREATE TABLE copy_on_conflict_t2 (a int, b text); +COPY copy_on_conflict_t2 FROM '/dev/null' DO ON CONFLICT DO NOTHING; +ERROR: COPY ON CONFLICT requires a unique constraint on table "copy_on_conflict_t2" +-- 4. COPY TO does not accept ON CONFLICT +COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING; +ERROR: COPY ON_CONFLICT cannot be used with COPY TO +LINE 1: COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING... + ^ +-- 5. DO UPDATE is parsed but rejected (not implemented yet) +COPY copy_on_conflict_t1 FROM '/dev/null' DO ON CONFLICT DO UPDATE; +ERROR: COPY ON CONFLICT DO UPDATE is not supported yet +LINE 1: COPY copy_on_conflict_t1 FROM '/dev/null' DO ON CONFLICT DO ... + ^ +HINT: Use DO ON CONFLICT DO NOTHING, or DELETE/TRUNCATE the conflicting rows before COPY, or use INSERT ... ON CONFLICT DO UPDATE. +-- 6. multi-column unique index +CREATE TABLE copy_on_conflict_t3 (a int, b int, c text, UNIQUE (a, b)); +INSERT INTO copy_on_conflict_t3 VALUES (1, 1, 'keep'); +COPY copy_on_conflict_t3 FROM STDIN DO ON CONFLICT DO NOTHING; +NOTICE: 1 row was skipped due to ON CONFLICT DO NOTHING +SELECT * FROM copy_on_conflict_t3 ORDER BY a, b; + a | b | c +---+---+------ + 1 | 1 | keep + 1 | 2 | ok +(2 rows) + +-- 7. partitioned table: uniqueness enforced per leaf partition +CREATE TABLE copy_on_conflict_tp (id int PRIMARY KEY, v text) PARTITION BY RANGE (id); +CREATE TABLE copy_on_conflict_tp_lo PARTITION OF copy_on_conflict_tp FOR VALUES FROM (0) TO (10); +CREATE TABLE copy_on_conflict_tp_hi PARTITION OF copy_on_conflict_tp FOR VALUES FROM (10) TO (20); +INSERT INTO copy_on_conflict_tp VALUES (1, 'p-old'), (11, 'h-old'); +COPY copy_on_conflict_tp FROM STDIN DO ON CONFLICT DO NOTHING; +NOTICE: 2 rows were skipped due to ON CONFLICT DO NOTHING +SELECT * FROM copy_on_conflict_tp ORDER BY id; + id | v +----+------- + 1 | p-old + 5 | p-new + 11 | h-old + 15 | h-new +(4 rows) + +-- 8. combined with ON_ERROR ignore: type errors and conflicts handled independently +CREATE TABLE copy_on_conflict_t4 (id int PRIMARY KEY, v int); +INSERT INTO copy_on_conflict_t4 VALUES (1, 100); +COPY copy_on_conflict_t4 FROM STDIN WITH (ON_ERROR ignore) DO ON CONFLICT DO NOTHING; +NOTICE: skipping row due to data type incompatibility at line 1 for column "v": "bad-dup" +NOTICE: skipping row due to data type incompatibility at line 2 for column "v": "notanumber" +NOTICE: 2 rows were skipped due to data type incompatibility +SELECT * FROM copy_on_conflict_t4 ORDER BY id; + id | v +----+----- + 1 | 100 + 2 | 200 +(2 rows) + +-- 9. parenthesized option syntax also accepts on_conflict +COPY copy_on_conflict_t1 FROM STDIN WITH (on_conflict 'nothing'); +SELECT * FROM copy_on_conflict_t1 WHERE id = 5; + id | name +----+------ + 5 | five +(1 row) + +-- 10. unknown on_conflict value is rejected +COPY copy_on_conflict_t1 FROM '/dev/null' WITH (on_conflict 'bogus'); +ERROR: COPY ON_CONFLICT "bogus" not recognized +LINE 1: COPY copy_on_conflict_t1 FROM '/dev/null' WITH (on_conflict ... + ^ +-- cleanup +DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, + copy_on_conflict_tp, copy_on_conflict_t4 CASCADE; diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule index a3894d87d8c..79ce06f8c19 100644 --- a/src/test/regress/parallel_schedule +++ b/src/test/regress/parallel_schedule @@ -150,3 +150,6 @@ test: tablespace # Test that Oracle keywords (ROWNUM, ROWID, etc.) have no special meaning in PostgreSQL mode test: ora_keywords_pg + +# COPY ON CONFLICT (AnalyticDB-compatible) +test: copy_on_conflict diff --git a/src/test/regress/sql/copy_on_conflict.sql b/src/test/regress/sql/copy_on_conflict.sql new file mode 100644 index 00000000000..17963fa07ce --- /dev/null +++ b/src/test/regress/sql/copy_on_conflict.sql @@ -0,0 +1,75 @@ +-- COPY ON CONFLICT (AnalyticDB-compatible DO ON CONFLICT DO NOTHING) +DROP TABLE IF EXISTS copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, + copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, + copy_on_conflict_t4 CASCADE; +CREATE TABLE copy_on_conflict_t1 (id int PRIMARY KEY, name text); +INSERT INTO copy_on_conflict_t1 VALUES (1, 'old-1'), (2, 'old-2'); + +-- 1. DO NOTHING skips conflicting rows, inserts the rest +COPY copy_on_conflict_t1 FROM STDIN DO ON CONFLICT DO NOTHING; +1 new-1 +2 new-2 +3 new-3 +\. +SELECT * FROM copy_on_conflict_t1 ORDER BY id; + +-- 2. no-conflict data behaves like plain COPY +COPY copy_on_conflict_t1 FROM STDIN DO ON CONFLICT DO NOTHING; +4 four +\. +SELECT count(*) FROM copy_on_conflict_t1; + +-- 3. table without unique constraint is rejected at startup +CREATE TABLE copy_on_conflict_t2 (a int, b text); +COPY copy_on_conflict_t2 FROM '/dev/null' DO ON CONFLICT DO NOTHING; + +-- 4. COPY TO does not accept ON CONFLICT +COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING; + +-- 5. DO UPDATE is parsed but rejected (not implemented yet) +COPY copy_on_conflict_t1 FROM '/dev/null' DO ON CONFLICT DO UPDATE; + +-- 6. multi-column unique index +CREATE TABLE copy_on_conflict_t3 (a int, b int, c text, UNIQUE (a, b)); +INSERT INTO copy_on_conflict_t3 VALUES (1, 1, 'keep'); +COPY copy_on_conflict_t3 FROM STDIN DO ON CONFLICT DO NOTHING; +1 1 dup +1 2 ok +\. +SELECT * FROM copy_on_conflict_t3 ORDER BY a, b; + +-- 7. partitioned table: uniqueness enforced per leaf partition +CREATE TABLE copy_on_conflict_tp (id int PRIMARY KEY, v text) PARTITION BY RANGE (id); +CREATE TABLE copy_on_conflict_tp_lo PARTITION OF copy_on_conflict_tp FOR VALUES FROM (0) TO (10); +CREATE TABLE copy_on_conflict_tp_hi PARTITION OF copy_on_conflict_tp FOR VALUES FROM (10) TO (20); +INSERT INTO copy_on_conflict_tp VALUES (1, 'p-old'), (11, 'h-old'); +COPY copy_on_conflict_tp FROM STDIN DO ON CONFLICT DO NOTHING; +1 p-dup +5 p-new +11 h-dup +15 h-new +\. +SELECT * FROM copy_on_conflict_tp ORDER BY id; + +-- 8. combined with ON_ERROR ignore: type errors and conflicts handled independently +CREATE TABLE copy_on_conflict_t4 (id int PRIMARY KEY, v int); +INSERT INTO copy_on_conflict_t4 VALUES (1, 100); +COPY copy_on_conflict_t4 FROM STDIN WITH (ON_ERROR ignore) DO ON CONFLICT DO NOTHING; +1 bad-dup +2 notanumber +2 200 +\. +SELECT * FROM copy_on_conflict_t4 ORDER BY id; + +-- 9. parenthesized option syntax also accepts on_conflict +COPY copy_on_conflict_t1 FROM STDIN WITH (on_conflict 'nothing'); +5 five +\. +SELECT * FROM copy_on_conflict_t1 WHERE id = 5; + +-- 10. unknown on_conflict value is rejected +COPY copy_on_conflict_t1 FROM '/dev/null' WITH (on_conflict 'bogus'); + +-- cleanup +DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, + copy_on_conflict_tp, copy_on_conflict_t4 CASCADE; From fa6ce1b670a717d3f0e9f91087141d23e2db49c5 Mon Sep 17 00:00:00 2001 From: Wenbo Date: Wed, 5 Aug 2026 10:59:42 +0800 Subject: [PATCH 3/6] doc: document COPY ON CONFLICT in copy.sgml Adds the DO ON CONFLICT clause to the COPY synopsis and a parameter entry describing DO NOTHING semantics, the DO UPDATE not-yet-implemented error, the unique-constraint requirement, per-partition uniqueness for partitioned tables, and the pre-check concurrency caveat. --- doc/src/sgml/ref/copy.sgml | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml index 4706c9a4410..6d5c70b0e07 100644 --- a/doc/src/sgml/ref/copy.sgml +++ b/doc/src/sgml/ref/copy.sgml @@ -48,6 +48,7 @@ COPY { table_name [ ( maxerror ENCODING 'encoding_name' LOG_VERBOSITY verbosity + DO ON CONFLICT { DO NOTHING | DO UPDATE } @@ -479,6 +480,43 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) + + ON CONFLICT + + + Specifies how to behave when an input row conflicts with a unique or + exclusion constraint on the target table, instead of aborting the whole + COPY command. This clause is compatible with + COPY ON CONFLICT in Alibaba Cloud AnalyticDB for + PostgreSQL and is only applicable to COPY FROM. + + + DO ON CONFLICT DO NOTHING discards the conflicting + input row and continues with the next one. The number of skipped rows + is reported in a NOTICE message at the end of the + command. + + + DO ON CONFLICT DO UPDATE is parsed but not yet + implemented; using it raises an error. As a workaround, delete or + truncate the conflicting rows before importing, or use + INSERT ... ON CONFLICT DO UPDATE. + + + The target table must have at least one unique constraint (primary key + or unique index) for the conflict check to be meaningful; otherwise the + COPY command fails at startup. For partitioned + tables, uniqueness is enforced per leaf partition. + + + The conflict check is performed before insertion, so under concurrent + writes a unique violation may still be raised by the index insert + itself. For strictly concurrent workloads, prefer + INSERT ... ON CONFLICT. + + + + REJECT_LIMIT From c98ccc6c6cbb3112602e9a139a7faf8982241003 Mon Sep 17 00:00:00 2001 From: Wenbo Date: Wed, 5 Aug 2026 11:25:12 +0800 Subject: [PATCH 4/6] feat: implement COPY ON CONFLICT DO UPDATE (full-row overwrite) DO ON CONFLICT DO UPDATE now overwrites the conflicting row with the input row, matching AnalyticDB semantics (no SET targets, full-row overwrite). Reuses the IvorySQL ModifyTable update machinery (ExecUpdatePrologue/Act/Epilogue) instead of a hand-written path: - copyfrom.c: on conflict, switch the ModifyTable operation to CMD_UPDATE, run the update via the three-part ExecUpdate API, count updated rows, and report via NOTICE. es_output_cid is set at COPY start so rows inserted by earlier commands of the transaction are visible to the update. A CMD_UPDATE transition capture state is created when update transition tables exist. - nodeModifyTable.c: make the INSERT ON CONFLICT DO UPDATE cross-partition check NULL-safe (COPY's ModifyTable plan node is NULL). Cross-partition moves remain rejected, matching INSERT ON CONFLICT DO UPDATE semantics. - copy.c: defGetCopyOnConflictChoice() now accepts 'update'. Partitioned tables: PG requires partition key columns in unique constraints, so a conflict implies the same partition key and cross-partition moves are unreachable; the defensive partition check is kept. Concurrent modification of a conflicting row raises an error (fail-loud, no EPQ retry in COPY). --- doc/src/sgml/ref/copy.sgml | 19 ++- src/backend/commands/copy.c | 10 +- src/backend/commands/copyfrom.c | 119 +++++++++++++++++- src/backend/executor/nodeModifyTable.c | 3 +- .../regress/expected/copy_on_conflict.out | 113 ++++++++++++++++- src/test/regress/sql/copy_on_conflict.sql | 81 +++++++++++- 6 files changed, 318 insertions(+), 27 deletions(-) diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml index 6d5c70b0e07..551f643cf67 100644 --- a/doc/src/sgml/ref/copy.sgml +++ b/doc/src/sgml/ref/copy.sgml @@ -497,10 +497,17 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) command. - DO ON CONFLICT DO UPDATE is parsed but not yet - implemented; using it raises an error. As a workaround, delete or - truncate the conflicting rows before importing, or use - INSERT ... ON CONFLICT DO UPDATE. + DO ON CONFLICT DO UPDATE overwrites the conflicting + row with the input row (a full-row overwrite, matching the AnalyticDB + semantics where no SET targets are allowed). Note + that for a column subset COPY (column [, ...]) FROM ..., + columns not listed in the input are filled with their default values, + so a volatile default expression (e.g. nextval) will + produce a new value for each imported row and overwrite the stored one. + + + The number of updated rows is reported in a NOTICE + message at the end of the command. The target table must have at least one unique constraint (primary key @@ -511,7 +518,9 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) The conflict check is performed before insertion, so under concurrent writes a unique violation may still be raised by the index insert - itself. For strictly concurrent workloads, prefer + itself. For DO ON CONFLICT DO UPDATE, a row that is + concurrently modified raises an error rather than being retried. For + strictly concurrent workloads, prefer INSERT ... ON CONFLICT. diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 6e219ea38f4..cecceaeae91 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -505,10 +505,6 @@ defGetCopyOnErrorChoice(DefElem *def, ParseState *pstate, bool is_from) /* * Extract a CopyOnConflictChoice value from a DefElem. - * - * The DO ON CONFLICT DO UPDATE clause is parsed but not yet implemented; - * reject it at startup so that users get a clear error instead of a silent - * fallback to plain COPY. */ static CopyOnConflictChoice defGetCopyOnConflictChoice(DefElem *def, ParseState *pstate, bool is_from) @@ -526,11 +522,7 @@ defGetCopyOnConflictChoice(DefElem *def, ParseState *pstate, bool is_from) if (pg_strcasecmp(sval, "nothing") == 0) return COPY_ON_CONFLICT_NOTHING; if (pg_strcasecmp(sval, "update") == 0) - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("COPY ON CONFLICT DO UPDATE is not supported yet"), - errhint("Use DO ON CONFLICT DO NOTHING, or DELETE/TRUNCATE the conflicting rows before COPY, or use INSERT ... ON CONFLICT DO UPDATE."), - parser_errposition(pstate, def->location))); + return COPY_ON_CONFLICT_UPDATE; ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index 48d4faccce7..ea707a19415 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -786,6 +786,13 @@ CopyFrom(CopyFromState cstate) ResultRelInfo *prevResultRelInfo = NULL; EState *estate = CreateExecutorState(); /* for ExecConstraints() */ ModifyTableState *mtstate; + + /* + * Make sure any UPDATE performed by the ON CONFLICT DO UPDATE clause + * uses the current command id, so that rows inserted by earlier + * commands of this transaction are visible to the update. + */ + estate->es_output_cid = GetCurrentCommandId(true); ExprContext *econtext; TupleTableSlot *singleslot = NULL; MemoryContext oldcontext = CurrentMemoryContext; @@ -800,6 +807,7 @@ CopyFrom(CopyFromState cstate) int64 processed = 0; int64 excluded = 0; int64 skipped = 0; /* rows skipped by ON CONFLICT DO NOTHING */ + int64 updated = 0; /* rows updated by ON CONFLICT DO UPDATE */ bool has_before_insert_row_trig; bool has_instead_insert_row_trig; bool leafpart_use_multi_insert = false; @@ -1432,7 +1440,7 @@ CopyFrom(CopyFromState cstate) * pre-check: under concurrent insertion a unique violation * may still be raised by the index insert itself. */ - if (cstate->opts.on_conflict == COPY_ON_CONFLICT_NOTHING) + if (cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE) { ItemPointerData conflictTid; ItemPointerData invalidItemPtr; @@ -1466,12 +1474,107 @@ CopyFrom(CopyFromState cstate) estate, &conflictTid, &invalidItemPtr, NIL)) { - skipped++; + if (cstate->opts.on_conflict == COPY_ON_CONFLICT_NOTHING) + { + skipped++; + + pgstat_progress_update_param(PROGRESS_COPY_TUPLES_SKIPPED, + skipped); - pgstat_progress_update_param(PROGRESS_COPY_TUPLES_SKIPPED, - skipped); + continue; /* skip this tuple */ + } - continue; /* skip this tuple */ + /* ===== DO ON CONFLICT DO UPDATE: overwrite the conflicting row ===== */ + { + IvyModifyTableContext context; + IvyUpdateContext updateCxt; + EPQState epqstate; + TM_Result result; + + /* + * Like INSERT ... ON CONFLICT DO UPDATE, moving + * the row to a different partition is not + * supported; reject it explicitly instead of + * crashing in ExecCrossPartitionUpdate (which + * dereferences the ModifyTable plan node). + */ + if (resultRelInfo->ri_RelationDesc->rd_rel->relispartition && + !ExecPartitionCheck(resultRelInfo, myslot, + estate, false)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("COPY ON CONFLICT DO UPDATE cannot move the row to a different partition"), + errdetail("The result tuple would appear in a different partition than the original tuple."))); + + /* Set up the ModifyTable execution context. */ + memset(&context, 0, sizeof(context)); + context.mtstate = mtstate; + context.estate = estate; + context.planSlot = NULL; + EvalPlanQualInit(&epqstate, estate, NULL, NIL, + 0, NIL); + context.epqstate = &epqstate; + + /* + * Temporarily switch the ModifyTable operation + * to UPDATE so that the Prologue/Epilogue use + * UPDATE semantics (BEFORE/AFTER ROW UPDATE + * triggers, transition capture, etc.). + */ + mtstate->operation = CMD_UPDATE; + if (resultRelInfo->ri_TrigDesc && + (resultRelInfo->ri_TrigDesc->trig_update_new_table || + resultRelInfo->ri_TrigDesc->trig_update_old_table)) + mtstate->mt_transition_capture = + MakeTransitionCaptureState(resultRelInfo->ri_TrigDesc, + RelationGetRelid(resultRelInfo->ri_RelationDesc), + CMD_UPDATE); + else + mtstate->mt_transition_capture = NULL; + + /* + * Run BEFORE ROW UPDATE triggers (if any) and + * perform the update. oldtuple is left NULL: + * the trigger machinery fetches the old row + * itself from conflictTid. + */ + if (ExecUpdatePrologue(&context, resultRelInfo, + &conflictTid, NULL, myslot, + &result)) + { + updateCxt.crossPartUpdate = false; + updateCxt.updateIndexes = TU_None; + updateCxt.lockmode = LockTupleExclusive; + + result = ExecUpdateAct(&context, resultRelInfo, + &conflictTid, NULL, myslot, + true, &updateCxt); + + if (result == TM_Ok) + ExecUpdateEpilogue(&context, &updateCxt, + resultRelInfo, + &conflictTid, NULL, + myslot); + else if (result == TM_Updated) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("concurrent update of row during COPY ON CONFLICT DO UPDATE"))); + else + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("could not update row during COPY ON CONFLICT DO UPDATE: %d", (int) result))); + } + + /* Restore INSERT semantics for the rest of COPY. */ + mtstate->operation = CMD_INSERT; + mtstate->mt_transition_capture = cstate->transition_capture; + + updated++; + pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, + processed + updated); + + continue; /* row handled as UPDATE */ + } } } @@ -1594,6 +1697,12 @@ CopyFrom(CopyFromState cstate) "%" PRIu64 " rows were skipped due to ON CONFLICT DO NOTHING", skipped, skipped)); + if (cstate->opts.on_conflict == COPY_ON_CONFLICT_UPDATE && updated > 0) + ereport(NOTICE, + errmsg_plural("%" PRIu64 " row was updated due to ON CONFLICT DO UPDATE", + "%" PRIu64 " rows were updated due to ON CONFLICT DO UPDATE", + updated, updated)); + if (bistate != NULL) FreeBulkInsertState(bistate); diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index 35737f93726..c46d5750018 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -2168,7 +2168,8 @@ ExecCrossPartitionUpdate(IvyModifyTableContext *context, * to migrate to a different partition. Maybe this can be implemented * some day, but it seems a fringe feature with little redeeming value. */ - if (((ModifyTable *) mtstate->ps.plan)->onConflictAction == ONCONFLICT_UPDATE) + if (mtstate->ps.plan != NULL && + ((ModifyTable *) mtstate->ps.plan)->onConflictAction == ONCONFLICT_UPDATE) ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("invalid ON UPDATE specification"), diff --git a/src/test/regress/expected/copy_on_conflict.out b/src/test/regress/expected/copy_on_conflict.out index 2d638c506ff..6fb6a53f1db 100644 --- a/src/test/regress/expected/copy_on_conflict.out +++ b/src/test/regress/expected/copy_on_conflict.out @@ -41,10 +41,6 @@ LINE 1: COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING... ^ -- 5. DO UPDATE is parsed but rejected (not implemented yet) COPY copy_on_conflict_t1 FROM '/dev/null' DO ON CONFLICT DO UPDATE; -ERROR: COPY ON CONFLICT DO UPDATE is not supported yet -LINE 1: COPY copy_on_conflict_t1 FROM '/dev/null' DO ON CONFLICT DO ... - ^ -HINT: Use DO ON CONFLICT DO NOTHING, or DELETE/TRUNCATE the conflicting rows before COPY, or use INSERT ... ON CONFLICT DO UPDATE. -- 6. multi-column unique index CREATE TABLE copy_on_conflict_t3 (a int, b int, c text, UNIQUE (a, b)); INSERT INTO copy_on_conflict_t3 VALUES (1, 1, 'keep'); @@ -100,6 +96,113 @@ COPY copy_on_conflict_t1 FROM '/dev/null' WITH (on_conflict 'bogus'); ERROR: COPY ON_CONFLICT "bogus" not recognized LINE 1: COPY copy_on_conflict_t1 FROM '/dev/null' WITH (on_conflict ... ^ +-- ===== DO ON CONFLICT DO UPDATE ===== +-- 11. DO UPDATE overwrites conflicting rows (full-row overwrite) +DROP TABLE IF EXISTS copy_on_conflict_u1 CASCADE; +NOTICE: table "copy_on_conflict_u1" does not exist, skipping +CREATE TABLE copy_on_conflict_u1 (id int PRIMARY KEY, name text); +INSERT INTO copy_on_conflict_u1 VALUES (1, 'old-1'), (2, 'old-2'); +COPY copy_on_conflict_u1 FROM STDIN DO ON CONFLICT DO UPDATE; +NOTICE: 2 rows were updated due to ON CONFLICT DO UPDATE +SELECT * FROM copy_on_conflict_u1 ORDER BY id; + id | name +----+------- + 1 | new-1 + 2 | new-2 + 3 | new-3 +(3 rows) + +-- 12. DO UPDATE without conflicts behaves like plain COPY +COPY copy_on_conflict_u1 FROM STDIN DO ON CONFLICT DO UPDATE; +SELECT count(*) FROM copy_on_conflict_u1; + count +------- + 4 +(1 row) + +-- 13. multi-column unique index with DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_u3 CASCADE; +NOTICE: table "copy_on_conflict_u3" does not exist, skipping +CREATE TABLE copy_on_conflict_u3 (a int, b int, c text, UNIQUE (a, b)); +INSERT INTO copy_on_conflict_u3 VALUES (1, 1, 'keep'); +COPY copy_on_conflict_u3 FROM STDIN DO ON CONFLICT DO UPDATE; +NOTICE: 1 row was updated due to ON CONFLICT DO UPDATE +SELECT * FROM copy_on_conflict_u3 ORDER BY a, b; + a | b | c +---+---+--------- + 1 | 1 | changed + 1 | 2 | ok +(2 rows) + +-- 14. partitioned table: in-partition DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_up CASCADE; +NOTICE: table "copy_on_conflict_up" does not exist, skipping +CREATE TABLE copy_on_conflict_up (id int PRIMARY KEY, v text) PARTITION BY RANGE (id); +CREATE TABLE copy_on_conflict_up_lo PARTITION OF copy_on_conflict_up FOR VALUES FROM (0) TO (10); +CREATE TABLE copy_on_conflict_up_hi PARTITION OF copy_on_conflict_up FOR VALUES FROM (10) TO (20); +INSERT INTO copy_on_conflict_up VALUES (1, 'p-old'), (11, 'h-old'); +COPY copy_on_conflict_up FROM STDIN DO ON CONFLICT DO UPDATE; +NOTICE: 2 rows were updated due to ON CONFLICT DO UPDATE +SELECT * FROM copy_on_conflict_up ORDER BY id; + id | v +----+------- + 1 | p-new + 5 | ins + 11 | h-new +(3 rows) + +-- 15. stored generated columns are recomputed on DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_u6 CASCADE; +NOTICE: table "copy_on_conflict_u6" does not exist, skipping +CREATE TABLE copy_on_conflict_u6 (id int PRIMARY KEY, a int, + b int GENERATED ALWAYS AS (a * 2) STORED); +INSERT INTO copy_on_conflict_u6 VALUES (1, 5, DEFAULT); +COPY copy_on_conflict_u6 (id, a) FROM STDIN DO ON CONFLICT DO UPDATE; +NOTICE: 1 row was updated due to ON CONFLICT DO UPDATE +SELECT * FROM copy_on_conflict_u6; + id | a | b +----+----+---- + 1 | 10 | 20 +(1 row) + +-- 16. AFTER ROW UPDATE triggers fire on DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_u7, copy_on_conflict_u7_log CASCADE; +NOTICE: table "copy_on_conflict_u7" does not exist, skipping +NOTICE: table "copy_on_conflict_u7_log" does not exist, skipping +CREATE TABLE copy_on_conflict_u7 (id int PRIMARY KEY, v text); +CREATE TABLE copy_on_conflict_u7_log (id int, old_v text, new_v text); +CREATE FUNCTION copy_on_conflict_u7_trg() RETURNS trigger AS $$ +BEGIN + INSERT INTO copy_on_conflict_u7_log VALUES (OLD.id, OLD.v, NEW.v); + RETURN NEW; +END $$ LANGUAGE plpgsql; +CREATE TRIGGER copy_on_conflict_u7_trg AFTER UPDATE ON copy_on_conflict_u7 + FOR EACH ROW EXECUTE FUNCTION copy_on_conflict_u7_trg(); +INSERT INTO copy_on_conflict_u7 VALUES (1, 'old'); +COPY copy_on_conflict_u7 FROM STDIN DO ON CONFLICT DO UPDATE; +NOTICE: 1 row was updated due to ON CONFLICT DO UPDATE +SELECT * FROM copy_on_conflict_u7; + id | v +----+----- + 1 | new +(1 row) + +SELECT * FROM copy_on_conflict_u7_log; + id | old_v | new_v +----+-------+------- + 1 | old | new +(1 row) + +-- 17. partitioned table unique constraints must include partition columns +-- (PG rule: cross-partition updates are therefore unreachable for ON CONFLICT) +DROP TABLE IF EXISTS copy_on_conflict_u5 CASCADE; +NOTICE: table "copy_on_conflict_u5" does not exist, skipping +CREATE TABLE copy_on_conflict_u5 (id int, uid int UNIQUE, v text) PARTITION BY RANGE (id); +ERROR: UNIQUE constraint on partitioned table must include all partitioning columns +DETAIL: UNIQUE constraint on table "copy_on_conflict_u5" lacks column "id" which is part of the partition key. -- cleanup DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, - copy_on_conflict_tp, copy_on_conflict_t4 CASCADE; + copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, + copy_on_conflict_t4, copy_on_conflict_u1, copy_on_conflict_u3, + copy_on_conflict_up, copy_on_conflict_up_lo, copy_on_conflict_up_hi, + copy_on_conflict_u6, copy_on_conflict_u7, copy_on_conflict_u7_log CASCADE; diff --git a/src/test/regress/sql/copy_on_conflict.sql b/src/test/regress/sql/copy_on_conflict.sql index 17963fa07ce..7aa5bd682b5 100644 --- a/src/test/regress/sql/copy_on_conflict.sql +++ b/src/test/regress/sql/copy_on_conflict.sql @@ -26,7 +26,7 @@ COPY copy_on_conflict_t2 FROM '/dev/null' DO ON CONFLICT DO NOTHING; -- 4. COPY TO does not accept ON CONFLICT COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING; --- 5. DO UPDATE is parsed but rejected (not implemented yet) +-- 5. DO UPDATE with empty input: no conflicts, no-op (see section below for behavior) COPY copy_on_conflict_t1 FROM '/dev/null' DO ON CONFLICT DO UPDATE; -- 6. multi-column unique index @@ -70,6 +70,83 @@ SELECT * FROM copy_on_conflict_t1 WHERE id = 5; -- 10. unknown on_conflict value is rejected COPY copy_on_conflict_t1 FROM '/dev/null' WITH (on_conflict 'bogus'); +-- ===== DO ON CONFLICT DO UPDATE ===== +-- 11. DO UPDATE overwrites conflicting rows (full-row overwrite) +DROP TABLE IF EXISTS copy_on_conflict_u1 CASCADE; +CREATE TABLE copy_on_conflict_u1 (id int PRIMARY KEY, name text); +INSERT INTO copy_on_conflict_u1 VALUES (1, 'old-1'), (2, 'old-2'); +COPY copy_on_conflict_u1 FROM STDIN DO ON CONFLICT DO UPDATE; +1 new-1 +2 new-2 +3 new-3 +\. +SELECT * FROM copy_on_conflict_u1 ORDER BY id; + +-- 12. DO UPDATE without conflicts behaves like plain COPY +COPY copy_on_conflict_u1 FROM STDIN DO ON CONFLICT DO UPDATE; +4 four +\. +SELECT count(*) FROM copy_on_conflict_u1; + +-- 13. multi-column unique index with DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_u3 CASCADE; +CREATE TABLE copy_on_conflict_u3 (a int, b int, c text, UNIQUE (a, b)); +INSERT INTO copy_on_conflict_u3 VALUES (1, 1, 'keep'); +COPY copy_on_conflict_u3 FROM STDIN DO ON CONFLICT DO UPDATE; +1 1 changed +1 2 ok +\. +SELECT * FROM copy_on_conflict_u3 ORDER BY a, b; + +-- 14. partitioned table: in-partition DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_up CASCADE; +CREATE TABLE copy_on_conflict_up (id int PRIMARY KEY, v text) PARTITION BY RANGE (id); +CREATE TABLE copy_on_conflict_up_lo PARTITION OF copy_on_conflict_up FOR VALUES FROM (0) TO (10); +CREATE TABLE copy_on_conflict_up_hi PARTITION OF copy_on_conflict_up FOR VALUES FROM (10) TO (20); +INSERT INTO copy_on_conflict_up VALUES (1, 'p-old'), (11, 'h-old'); +COPY copy_on_conflict_up FROM STDIN DO ON CONFLICT DO UPDATE; +1 p-new +5 ins +11 h-new +\. +SELECT * FROM copy_on_conflict_up ORDER BY id; + +-- 15. stored generated columns are recomputed on DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_u6 CASCADE; +CREATE TABLE copy_on_conflict_u6 (id int PRIMARY KEY, a int, + b int GENERATED ALWAYS AS (a * 2) STORED); +INSERT INTO copy_on_conflict_u6 VALUES (1, 5, DEFAULT); +COPY copy_on_conflict_u6 (id, a) FROM STDIN DO ON CONFLICT DO UPDATE; +1 10 +\. +SELECT * FROM copy_on_conflict_u6; + +-- 16. AFTER ROW UPDATE triggers fire on DO UPDATE +DROP TABLE IF EXISTS copy_on_conflict_u7, copy_on_conflict_u7_log CASCADE; +CREATE TABLE copy_on_conflict_u7 (id int PRIMARY KEY, v text); +CREATE TABLE copy_on_conflict_u7_log (id int, old_v text, new_v text); +CREATE FUNCTION copy_on_conflict_u7_trg() RETURNS trigger AS $$ +BEGIN + INSERT INTO copy_on_conflict_u7_log VALUES (OLD.id, OLD.v, NEW.v); + RETURN NEW; +END $$ LANGUAGE plpgsql; +CREATE TRIGGER copy_on_conflict_u7_trg AFTER UPDATE ON copy_on_conflict_u7 + FOR EACH ROW EXECUTE FUNCTION copy_on_conflict_u7_trg(); +INSERT INTO copy_on_conflict_u7 VALUES (1, 'old'); +COPY copy_on_conflict_u7 FROM STDIN DO ON CONFLICT DO UPDATE; +1 new +\. +SELECT * FROM copy_on_conflict_u7; +SELECT * FROM copy_on_conflict_u7_log; + +-- 17. partitioned table unique constraints must include partition columns +-- (PG rule: cross-partition updates are therefore unreachable for ON CONFLICT) +DROP TABLE IF EXISTS copy_on_conflict_u5 CASCADE; +CREATE TABLE copy_on_conflict_u5 (id int, uid int UNIQUE, v text) PARTITION BY RANGE (id); + -- cleanup DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, - copy_on_conflict_tp, copy_on_conflict_t4 CASCADE; + copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, + copy_on_conflict_t4, copy_on_conflict_u1, copy_on_conflict_u3, + copy_on_conflict_up, copy_on_conflict_up_lo, copy_on_conflict_up_hi, + copy_on_conflict_u6, copy_on_conflict_u7, copy_on_conflict_u7_log CASCADE; From c638f610b2696c8ffc4808c70a3e36fd4cd5a964 Mon Sep 17 00:00:00 2001 From: Wenbo Date: Wed, 5 Aug 2026 11:35:09 +0800 Subject: [PATCH 5/6] feat: atomic conflict detection via speculative insertion (Phoenix-aligned) Rework COPY ON CONFLICT to match the semantics of INSERT ... ON CONFLICT and of Zbyte/Relyt's Phoenix implementation: - After the ExecCheckIndexConstraints pre-check passes, rows are inserted speculatively (table_tuple_insert_speculative + ExecInsertIndexTuples with EIIT_NO_DUPE_ERROR). A concurrent conflicting insert is detected atomically by the index insert, the speculative row is backed out, and the conflict check is re-run - eliminating the check-to-insert (TOCTOU) window of the previous pre-check-only design. - CopyOnConflictUpdate() (ported from Phoenix's copy.c) locks the conflicting tuple with table_tuple_lock before updating, waits out concurrent transactions, and returns 'retry' on TM_Updated/TM_Deleted so the whole conflict check re-runs instead of failing. - Duplicate constrained values within one COPY now raise the standard 'ON CONFLICT DO UPDATE command cannot affect row a second time' error (TM_Invisible + current xid detection), matching INSERT ON CONFLICT. - estate->es_snapshot is set at COPY start (needed by table_tuple_lock). Regression tests: 19 cases, including two new batch-duplicate-key cases (DO UPDATE error, DO NOTHING skip). Concurrent smoke test: two COPYs updating the same row both succeed (the second waits and retries). --- doc/src/sgml/ref/copy.sgml | 13 +- src/backend/commands/copyfrom.c | 255 ++++++++++++++++-- .../regress/expected/copy_on_conflict.out | 32 ++- src/test/regress/sql/copy_on_conflict.sql | 22 ++ 4 files changed, 288 insertions(+), 34 deletions(-) diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml index 551f643cf67..aadddd4f5bc 100644 --- a/doc/src/sgml/ref/copy.sgml +++ b/doc/src/sgml/ref/copy.sgml @@ -516,11 +516,14 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) tables, uniqueness is enforced per leaf partition. - The conflict check is performed before insertion, so under concurrent - writes a unique violation may still be raised by the index insert - itself. For DO ON CONFLICT DO UPDATE, a row that is - concurrently modified raises an error rather than being retried. For - strictly concurrent workloads, prefer + Conflict detection is atomic: after an initial constraint check, rows + are inserted speculatively so that a concurrent conflicting insert is + detected by the index insert itself, and the conflicting row is then + handled (or the insertion retried) without a check-to-insert race. For + DO ON CONFLICT DO UPDATE, a row that is concurrently + modified is locked and the update is retried once the concurrent + transaction completes. Duplicate constrained values within a single + COPY raise the same error as INSERT ... ON CONFLICT. diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index ea707a19415..4df82a81b65 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -775,6 +775,151 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, miinfo->bufferedBytes += tuplen; } +/* + * CopyOnConflictUpdate -- subroutine for COPY ON CONFLICT DO UPDATE + * + * Lock the conflicting tuple (waiting for any concurrent transaction) and + * overwrite it with the input row via the ModifyTable update machinery + * (BEFORE/AFTER ROW UPDATE triggers, index maintenance, transition + * capture). Returns false if the conflicting tuple was concurrently + * modified or deleted, in which case the caller must re-run the conflict + * check from scratch. + */ +static bool +CopyOnConflictUpdate(IvyModifyTableContext *context, + ResultRelInfo *resultRelInfo, + ItemPointer conflictTid, + TupleTableSlot *excludedSlot, + IvyUpdateContext *updateCxt) +{ + EState *estate = context->estate; + Relation relation = resultRelInfo->ri_RelationDesc; + TupleTableSlot *existing; + TM_FailureData tmfd; + LockTupleMode lockmode; + TM_Result test; + TM_Result result; + Datum xminDatum; + TransactionId xmin; + bool isnull; + + /* + * Slot for the existing (conflicting) tuple; used only by the lock + * machinery and by error paths. + */ + existing = table_slot_create(relation, &estate->es_tupleTable); + + /* Determine lock mode to use */ + lockmode = LockTupleExclusive; + + /* + * Lock the conflicting tuple. A row locking conflict here means our + * previous conclusion that the tuple is conclusively committed is not + * true anymore (TM_Updated/TM_Deleted) and the caller should retry. + */ + test = table_tuple_lock(relation, conflictTid, estate->es_snapshot, + existing, estate->es_output_cid, + lockmode, LockWaitBlock, 0, &tmfd); + switch (test) + { + case TM_Ok: + /* success! */ + break; + + case TM_Invisible: + + /* + * This can occur when a just inserted row is updated again + * in the same command, e.g. because multiple rows with the + * same conflicting key values are inserted in one COPY. + */ + xminDatum = slot_getsysattr(existing, + MinTransactionIdAttributeNumber, + &isnull); + Assert(!isnull); + xmin = DatumGetTransactionId(xminDatum); + + if (TransactionIdIsCurrentTransactionId(xmin)) + ereport(ERROR, + (errcode(ERRCODE_CARDINALITY_VIOLATION), + errmsg("ON CONFLICT DO UPDATE command cannot affect row a second time"), + errhint("Ensure that no rows proposed for insertion within the same command have duplicate constrained values."))); + /* This shouldn't happen */ + elog(ERROR, "attempted to lock invisible tuple"); + break; + + case TM_SelfModified: + + /* + * This state should never be reached. As a dirty snapshot is + * used to find conflicting tuples, speculative insertion + * wouldn't have seen this row to conflict with. + */ + elog(ERROR, "unexpected self-updated tuple"); + break; + + case TM_Updated: + if (IsolationUsesXactSnapshot()) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("could not serialize access due to concurrent update"))); + + /* Tell caller to try again from the very start. */ + ExecClearTuple(existing); + return false; + + case TM_Deleted: + if (IsolationUsesXactSnapshot()) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("could not serialize access due to concurrent delete"))); + + /* see TM_Updated case */ + ExecClearTuple(existing); + return false; + + default: + elog(ERROR, "unrecognized table_tuple_lock status: %u", test); + } + + /* Success, the tuple is locked. */ + + /* + * Run BEFORE ROW UPDATE triggers (if any) and perform the update. + * oldtuple is left NULL: the trigger machinery fetches the old row + * itself from conflictTid. The tuple is already locked, so + * table_tuple_update inside ExecUpdateAct does not block. + */ + if (ExecUpdatePrologue(context, resultRelInfo, + conflictTid, NULL, excludedSlot, + &result)) + { + updateCxt->crossPartUpdate = false; + updateCxt->updateIndexes = TU_None; + updateCxt->lockmode = LockTupleExclusive; + + result = ExecUpdateAct(context, resultRelInfo, + conflictTid, NULL, excludedSlot, + true, updateCxt); + + if (result == TM_Ok) + ExecUpdateEpilogue(context, updateCxt, + resultRelInfo, + conflictTid, NULL, + excludedSlot); + else if (result == TM_Updated) + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("concurrent update of row during COPY ON CONFLICT DO UPDATE"))); + else + ereport(ERROR, + (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), + errmsg("could not update row during COPY ON CONFLICT DO UPDATE: %d", (int) result))); + } + + return true; +} + /* * Copy FROM file to relation. */ @@ -793,6 +938,13 @@ CopyFrom(CopyFromState cstate) * commands of this transaction are visible to the update. */ estate->es_output_cid = GetCurrentCommandId(true); + + /* + * COPY ON CONFLICT locks conflicting tuples (CopyOnConflictUpdate); + * that needs a snapshot in the executor state. + */ + if (cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE) + estate->es_snapshot = GetActiveSnapshot(); ExprContext *econtext; TupleTableSlot *singleslot = NULL; MemoryContext oldcontext = CurrentMemoryContext; @@ -1444,6 +1596,8 @@ CopyFrom(CopyFromState cstate) { ItemPointerData conflictTid; ItemPointerData invalidItemPtr; + bool specConflict = false; + uint32 specToken = 0; int ii; ItemPointerSetInvalid(&invalidItemPtr); @@ -1470,6 +1624,7 @@ CopyFrom(CopyFromState cstate) } } + retry: if (!ExecCheckIndexConstraints(resultRelInfo, myslot, estate, &conflictTid, &invalidItemPtr, NIL)) @@ -1533,36 +1688,19 @@ CopyFrom(CopyFromState cstate) mtstate->mt_transition_capture = NULL; /* - * Run BEFORE ROW UPDATE triggers (if any) and - * perform the update. oldtuple is left NULL: - * the trigger machinery fetches the old row - * itself from conflictTid. + * Lock the conflicting row (waiting out any + * concurrent transaction) and overwrite it. + * If the row was concurrently modified or + * deleted, re-run the whole conflict check. */ - if (ExecUpdatePrologue(&context, resultRelInfo, - &conflictTid, NULL, myslot, - &result)) + if (!CopyOnConflictUpdate(&context, resultRelInfo, + &conflictTid, myslot, + &updateCxt)) { - updateCxt.crossPartUpdate = false; - updateCxt.updateIndexes = TU_None; - updateCxt.lockmode = LockTupleExclusive; - - result = ExecUpdateAct(&context, resultRelInfo, - &conflictTid, NULL, myslot, - true, &updateCxt); - - if (result == TM_Ok) - ExecUpdateEpilogue(&context, &updateCxt, - resultRelInfo, - &conflictTid, NULL, - myslot); - else if (result == TM_Updated) - ereport(ERROR, - (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), - errmsg("concurrent update of row during COPY ON CONFLICT DO UPDATE"))); - else - ereport(ERROR, - (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), - errmsg("could not update row during COPY ON CONFLICT DO UPDATE: %d", (int) result))); + /* Restore INSERT semantics, then retry. */ + mtstate->operation = CMD_INSERT; + mtstate->mt_transition_capture = cstate->transition_capture; + goto retry; } /* Restore INSERT semantics for the rest of COPY. */ @@ -1576,6 +1714,67 @@ CopyFrom(CopyFromState cstate) continue; /* row handled as UPDATE */ } } + else + { + /* + * Pre-check passed. Insert speculatively so that a + * concurrent conflicting insert is detected + * atomically by the index insert (no TOCTOU + * window); on conflict, back the tuple out and + * re-run the conflict check, which will now take + * the DO NOTHING / DO UPDATE path above. + */ + List *recheckIndexes = NIL; + + if (specToken == 0) + specToken = SpeculativeInsertionLockAcquire(GetCurrentTransactionId()); + + table_tuple_insert_speculative(resultRelInfo->ri_RelationDesc, + myslot, + estate->es_output_cid, + 0, NULL, specToken); + + recheckIndexes = ExecInsertIndexTuples(resultRelInfo, + estate, + EIIT_NO_DUPE_ERROR, + myslot, NIL, + &specConflict); + list_free(recheckIndexes); + recheckIndexes = NIL; + + if (specConflict) + { + /* + * A concurrent transaction inserted a + * conflicting row: back out the speculative + * insert and retry the conflict check. + */ + table_tuple_complete_speculative(resultRelInfo->ri_RelationDesc, + myslot, specToken, + false); + SpeculativeInsertionLockRelease(GetCurrentTransactionId()); + specToken = 0; + specConflict = false; + goto retry; + } + + /* No conflict: confirm the speculative insert. */ + table_tuple_complete_speculative(resultRelInfo->ri_RelationDesc, + myslot, specToken, + true); + SpeculativeInsertionLockRelease(GetCurrentTransactionId()); + specToken = 0; + + /* AFTER ROW INSERT triggers */ + ExecARInsertTriggers(estate, resultRelInfo, myslot, + NIL, cstate->transition_capture); + + processed++; + pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, + processed); + + continue; /* row inserted via speculative path */ + } } /* Store the slot in the multi-insert buffer, when enabled. */ diff --git a/src/test/regress/expected/copy_on_conflict.out b/src/test/regress/expected/copy_on_conflict.out index 6fb6a53f1db..a8cfe6d5305 100644 --- a/src/test/regress/expected/copy_on_conflict.out +++ b/src/test/regress/expected/copy_on_conflict.out @@ -39,7 +39,7 @@ COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING; ERROR: COPY ON_CONFLICT cannot be used with COPY TO LINE 1: COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING... ^ --- 5. DO UPDATE is parsed but rejected (not implemented yet) +-- 5. DO UPDATE with empty input: no conflicts, no-op (see section below for behavior) COPY copy_on_conflict_t1 FROM '/dev/null' DO ON CONFLICT DO UPDATE; -- 6. multi-column unique index CREATE TABLE copy_on_conflict_t3 (a int, b int, c text, UNIQUE (a, b)); @@ -200,6 +200,36 @@ NOTICE: table "copy_on_conflict_u5" does not exist, skipping CREATE TABLE copy_on_conflict_u5 (id int, uid int UNIQUE, v text) PARTITION BY RANGE (id); ERROR: UNIQUE constraint on partitioned table must include all partitioning columns DETAIL: UNIQUE constraint on table "copy_on_conflict_u5" lacks column "id" which is part of the partition key. +-- 18. duplicate constrained values within one COPY with DO UPDATE raises +-- the standard INSERT ON CONFLICT error ('cannot affect row a second time') +DROP TABLE IF EXISTS copy_on_conflict_u8 CASCADE; +NOTICE: table "copy_on_conflict_u8" does not exist, skipping +CREATE TABLE copy_on_conflict_u8 (id int PRIMARY KEY, v text); +INSERT INTO copy_on_conflict_u8 VALUES (1, 'existing'); +COPY copy_on_conflict_u8 FROM STDIN DO ON CONFLICT DO UPDATE; +ERROR: ON CONFLICT DO UPDATE command cannot affect row a second time +HINT: Ensure that no rows proposed for insertion within the same command have duplicate constrained values. +CONTEXT: COPY copy_on_conflict_u8, line 2: "1 second" +SELECT * FROM copy_on_conflict_u8; + id | v +----+---------- + 1 | existing +(1 row) + +-- 19. same for DO NOTHING: the second duplicate is also skipped, no error +DROP TABLE IF EXISTS copy_on_conflict_u9 CASCADE; +NOTICE: table "copy_on_conflict_u9" does not exist, skipping +CREATE TABLE copy_on_conflict_u9 (id int PRIMARY KEY, v text); +INSERT INTO copy_on_conflict_u9 VALUES (1, 'existing'); +COPY copy_on_conflict_u9 FROM STDIN DO ON CONFLICT DO NOTHING; +NOTICE: 2 rows were skipped due to ON CONFLICT DO NOTHING +SELECT * FROM copy_on_conflict_u9 ORDER BY id; + id | v +----+---------- + 1 | existing + 2 | ok +(2 rows) + -- cleanup DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, diff --git a/src/test/regress/sql/copy_on_conflict.sql b/src/test/regress/sql/copy_on_conflict.sql index 7aa5bd682b5..b28c781b61c 100644 --- a/src/test/regress/sql/copy_on_conflict.sql +++ b/src/test/regress/sql/copy_on_conflict.sql @@ -144,6 +144,28 @@ SELECT * FROM copy_on_conflict_u7_log; DROP TABLE IF EXISTS copy_on_conflict_u5 CASCADE; CREATE TABLE copy_on_conflict_u5 (id int, uid int UNIQUE, v text) PARTITION BY RANGE (id); +-- 18. duplicate constrained values within one COPY with DO UPDATE raises +-- the standard INSERT ON CONFLICT error ('cannot affect row a second time') +DROP TABLE IF EXISTS copy_on_conflict_u8 CASCADE; +CREATE TABLE copy_on_conflict_u8 (id int PRIMARY KEY, v text); +INSERT INTO copy_on_conflict_u8 VALUES (1, 'existing'); +COPY copy_on_conflict_u8 FROM STDIN DO ON CONFLICT DO UPDATE; +1 first +1 second +\. +SELECT * FROM copy_on_conflict_u8; + +-- 19. same for DO NOTHING: the second duplicate is also skipped, no error +DROP TABLE IF EXISTS copy_on_conflict_u9 CASCADE; +CREATE TABLE copy_on_conflict_u9 (id int PRIMARY KEY, v text); +INSERT INTO copy_on_conflict_u9 VALUES (1, 'existing'); +COPY copy_on_conflict_u9 FROM STDIN DO ON CONFLICT DO NOTHING; +1 first +1 second +2 ok +\. +SELECT * FROM copy_on_conflict_u9 ORDER BY id; + -- cleanup DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, From 56463e7f1dcc09832d302caeefe1b404f80f55ca Mon Sep 17 00:00:00 2001 From: Wenbo Date: Wed, 5 Aug 2026 11:49:43 +0800 Subject: [PATCH 6/6] fix: address CodeRabbit review comments on COPY ON CONFLICT - copyfrom.c: move es_output_cid/es_snapshot setup below the declaration block (-Wdeclaration-after-statement) - copyfrom.c: use uint64 for skipped/updated counters (match PRIu64) - copyfrom.c: allocate EvalPlanQualInit/MakeTransitionCaptureState in the per-tuple context (per-row state no longer leaks for the whole COPY duration) - copyfrom.c: CopyOnConflictUpdate() now distinguishes retry (0), updated (1) and BEFORE-trigger-suppressed (2); suppressed updates are no longer counted as updated, and the progress counter uses 'processed' uniformly - copyfrom.c: speculative insert now passes ti_options (FREEZE etc.) and keeps recheckIndexes for the AFTER ROW INSERT triggers - copyfrom.c: startup check accepts exclusion constraints too, matching ExecCheckIndexConstraints() and the docs - doc: copy.sgml now documents unique or exclusion constraint requirement - test: cleanup drops u8/u9 tables and the u7 trigger function --- doc/src/sgml/ref/copy.sgml | 5 +- src/backend/commands/copyfrom.c | 106 +++++++++++------- .../regress/expected/copy_on_conflict.out | 6 +- src/test/regress/sql/copy_on_conflict.sql | 4 +- 4 files changed, 76 insertions(+), 45 deletions(-) diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml index aadddd4f5bc..dd18bc1bc05 100644 --- a/doc/src/sgml/ref/copy.sgml +++ b/doc/src/sgml/ref/copy.sgml @@ -510,8 +510,9 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j)) message at the end of the command. - The target table must have at least one unique constraint (primary key - or unique index) for the conflict check to be meaningful; otherwise the + The target table must have at least one unique or exclusion constraint + (primary key, unique index, or EXCLUDE constraint) + for the conflict check to be meaningful; otherwise the COPY command fails at startup. For partitioned tables, uniqueness is enforced per leaf partition. diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c index 4df82a81b65..4de2da864ba 100644 --- a/src/backend/commands/copyfrom.c +++ b/src/backend/commands/copyfrom.c @@ -781,11 +781,13 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri, * Lock the conflicting tuple (waiting for any concurrent transaction) and * overwrite it with the input row via the ModifyTable update machinery * (BEFORE/AFTER ROW UPDATE triggers, index maintenance, transition - * capture). Returns false if the conflicting tuple was concurrently - * modified or deleted, in which case the caller must re-run the conflict - * check from scratch. + * capture). Returns: + * 0 - the conflicting tuple was concurrently modified or deleted; the + * caller must re-run the conflict check from scratch + * 1 - the row was updated + * 2 - a BEFORE ROW UPDATE trigger suppressed the update (no row change) */ -static bool +static int CopyOnConflictUpdate(IvyModifyTableContext *context, ResultRelInfo *resultRelInfo, ItemPointer conflictTid, @@ -866,7 +868,7 @@ CopyOnConflictUpdate(IvyModifyTableContext *context, /* Tell caller to try again from the very start. */ ExecClearTuple(existing); - return false; + return 0; case TM_Deleted: if (IsolationUsesXactSnapshot()) @@ -876,7 +878,7 @@ CopyOnConflictUpdate(IvyModifyTableContext *context, /* see TM_Updated case */ ExecClearTuple(existing); - return false; + return 0; default: elog(ERROR, "unrecognized table_tuple_lock status: %u", test); @@ -915,9 +917,12 @@ CopyOnConflictUpdate(IvyModifyTableContext *context, ereport(ERROR, (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), errmsg("could not update row during COPY ON CONFLICT DO UPDATE: %d", (int) result))); + + return 1; } - return true; + /* BEFORE ROW UPDATE trigger suppressed the update */ + return 2; } /* @@ -931,20 +936,6 @@ CopyFrom(CopyFromState cstate) ResultRelInfo *prevResultRelInfo = NULL; EState *estate = CreateExecutorState(); /* for ExecConstraints() */ ModifyTableState *mtstate; - - /* - * Make sure any UPDATE performed by the ON CONFLICT DO UPDATE clause - * uses the current command id, so that rows inserted by earlier - * commands of this transaction are visible to the update. - */ - estate->es_output_cid = GetCurrentCommandId(true); - - /* - * COPY ON CONFLICT locks conflicting tuples (CopyOnConflictUpdate); - * that needs a snapshot in the executor state. - */ - if (cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE) - estate->es_snapshot = GetActiveSnapshot(); ExprContext *econtext; TupleTableSlot *singleslot = NULL; MemoryContext oldcontext = CurrentMemoryContext; @@ -952,14 +943,15 @@ CopyFrom(CopyFromState cstate) PartitionTupleRouting *proute = NULL; ErrorContextCallback errcallback; CommandId mycid = GetCurrentCommandId(true); + uint32 ti_options = 0; /* start with default options for insert */ BulkInsertState bistate = NULL; CopyInsertMethod insertMethod; CopyMultiInsertInfo multiInsertInfo = {0}; /* pacify compiler */ int64 processed = 0; int64 excluded = 0; - int64 skipped = 0; /* rows skipped by ON CONFLICT DO NOTHING */ - int64 updated = 0; /* rows updated by ON CONFLICT DO UPDATE */ + uint64 skipped = 0; /* rows skipped by ON CONFLICT DO NOTHING */ + uint64 updated = 0; /* rows updated by ON CONFLICT DO UPDATE */ bool has_before_insert_row_trig; bool has_instead_insert_row_trig; bool leafpart_use_multi_insert = false; @@ -967,6 +959,20 @@ CopyFrom(CopyFromState cstate) Assert(cstate->rel); Assert(list_length(cstate->range_table) == 1); + /* + * Make sure any UPDATE performed by the ON CONFLICT DO UPDATE clause + * uses the current command id, so that rows inserted by earlier + * commands of this transaction are visible to the update. + */ + estate->es_output_cid = mycid; + + /* + * COPY ON CONFLICT locks conflicting tuples (CopyOnConflictUpdate); + * that needs a snapshot in the executor state. + */ + if (cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE) + estate->es_snapshot = GetActiveSnapshot(); + if (cstate->opts.on_error != COPY_ON_ERROR_STOP) Assert(cstate->escontext); @@ -1156,10 +1162,10 @@ CopyFrom(CopyFromState cstate) &mtstate->ps); /* - * COPY ON CONFLICT needs a unique constraint to detect conflicts. For - * partitioned tables the check is done per leaf partition at runtime - * (each partition enforces its own uniqueness), so skip the startup - * check there. + * COPY ON CONFLICT needs a unique or exclusion constraint to detect + * conflicts. For partitioned tables the check is done per leaf + * partition at runtime (each partition enforces its own uniqueness), + * so skip the startup check there. */ if (cstate->opts.on_conflict != COPY_ON_CONFLICT_NONE && cstate->rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) @@ -1175,7 +1181,8 @@ CopyFrom(CopyFromState cstate) Relation indexRel; indexRel = index_open(indexoid, AccessShareLock); - if (indexRel->rd_index->indisunique) + if (indexRel->rd_index->indisunique || + indexRel->rd_index->indisexclusion) has_unique = true; index_close(indexRel, AccessShareLock); if (has_unique) @@ -1186,7 +1193,7 @@ CopyFrom(CopyFromState cstate) if (!has_unique) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("COPY ON CONFLICT requires a unique constraint on table \"%s\"", + errmsg("COPY ON CONFLICT requires a unique or exclusion constraint on table \"%s\"", RelationGetRelationName(cstate->rel)))); } @@ -1598,6 +1605,7 @@ CopyFrom(CopyFromState cstate) ItemPointerData invalidItemPtr; bool specConflict = false; uint32 specToken = 0; + int update_rc; int ii; ItemPointerSetInvalid(&invalidItemPtr); @@ -1644,7 +1652,6 @@ CopyFrom(CopyFromState cstate) IvyModifyTableContext context; IvyUpdateContext updateCxt; EPQState epqstate; - TM_Result result; /* * Like INSERT ... ON CONFLICT DO UPDATE, moving @@ -1666,6 +1673,18 @@ CopyFrom(CopyFromState cstate) context.mtstate = mtstate; context.estate = estate; context.planSlot = NULL; + + /* + * EvalPlanQualInit() and + * MakeTransitionCaptureState() allocate in the + * current memory context, which here is the + * caller's context living for the whole COPY. + * Allocate them in the per-tuple context so + * per-row state is freed on the next iteration + * (the transition capture pointer is reset + * before we return to the loop). + */ + MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); EvalPlanQualInit(&epqstate, estate, NULL, NIL, 0, NIL); context.epqstate = &epqstate; @@ -1686,6 +1705,7 @@ CopyFrom(CopyFromState cstate) CMD_UPDATE); else mtstate->mt_transition_capture = NULL; + MemoryContextSwitchTo(oldcontext); /* * Lock the conflicting row (waiting out any @@ -1693,9 +1713,10 @@ CopyFrom(CopyFromState cstate) * If the row was concurrently modified or * deleted, re-run the whole conflict check. */ - if (!CopyOnConflictUpdate(&context, resultRelInfo, - &conflictTid, myslot, - &updateCxt)) + update_rc = CopyOnConflictUpdate(&context, resultRelInfo, + &conflictTid, myslot, + &updateCxt); + if (update_rc == 0) { /* Restore INSERT semantics, then retry. */ mtstate->operation = CMD_INSERT; @@ -1707,11 +1728,14 @@ CopyFrom(CopyFromState cstate) mtstate->operation = CMD_INSERT; mtstate->mt_transition_capture = cstate->transition_capture; - updated++; + if (update_rc == 1) + updated++; + + processed++; pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, - processed + updated); + processed); - continue; /* row handled as UPDATE */ + continue; /* row handled as UPDATE (or suppressed) */ } } else @@ -1732,18 +1756,18 @@ CopyFrom(CopyFromState cstate) table_tuple_insert_speculative(resultRelInfo->ri_RelationDesc, myslot, estate->es_output_cid, - 0, NULL, specToken); + ti_options, NULL, specToken); recheckIndexes = ExecInsertIndexTuples(resultRelInfo, estate, EIIT_NO_DUPE_ERROR, myslot, NIL, &specConflict); - list_free(recheckIndexes); - recheckIndexes = NIL; if (specConflict) { + list_free(recheckIndexes); + recheckIndexes = NIL; /* * A concurrent transaction inserted a * conflicting row: back out the speculative @@ -1767,7 +1791,9 @@ CopyFrom(CopyFromState cstate) /* AFTER ROW INSERT triggers */ ExecARInsertTriggers(estate, resultRelInfo, myslot, - NIL, cstate->transition_capture); + recheckIndexes, cstate->transition_capture); + list_free(recheckIndexes); + recheckIndexes = NIL; processed++; pgstat_progress_update_param(PROGRESS_COPY_TUPLES_PROCESSED, diff --git a/src/test/regress/expected/copy_on_conflict.out b/src/test/regress/expected/copy_on_conflict.out index a8cfe6d5305..b088c242235 100644 --- a/src/test/regress/expected/copy_on_conflict.out +++ b/src/test/regress/expected/copy_on_conflict.out @@ -33,7 +33,7 @@ SELECT count(*) FROM copy_on_conflict_t1; -- 3. table without unique constraint is rejected at startup CREATE TABLE copy_on_conflict_t2 (a int, b text); COPY copy_on_conflict_t2 FROM '/dev/null' DO ON CONFLICT DO NOTHING; -ERROR: COPY ON CONFLICT requires a unique constraint on table "copy_on_conflict_t2" +ERROR: COPY ON CONFLICT requires a unique or exclusion constraint on table "copy_on_conflict_t2" -- 4. COPY TO does not accept ON CONFLICT COPY copy_on_conflict_t1 TO STDOUT DO ON CONFLICT DO NOTHING; ERROR: COPY ON_CONFLICT cannot be used with COPY TO @@ -235,4 +235,6 @@ DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, copy_on_conflict_t4, copy_on_conflict_u1, copy_on_conflict_u3, copy_on_conflict_up, copy_on_conflict_up_lo, copy_on_conflict_up_hi, - copy_on_conflict_u6, copy_on_conflict_u7, copy_on_conflict_u7_log CASCADE; + copy_on_conflict_u6, copy_on_conflict_u7, copy_on_conflict_u7_log, + copy_on_conflict_u8, copy_on_conflict_u9 CASCADE; +DROP FUNCTION IF EXISTS copy_on_conflict_u7_trg(); diff --git a/src/test/regress/sql/copy_on_conflict.sql b/src/test/regress/sql/copy_on_conflict.sql index b28c781b61c..0c4a427be31 100644 --- a/src/test/regress/sql/copy_on_conflict.sql +++ b/src/test/regress/sql/copy_on_conflict.sql @@ -171,4 +171,6 @@ DROP TABLE copy_on_conflict_t1, copy_on_conflict_t2, copy_on_conflict_t3, copy_on_conflict_tp, copy_on_conflict_tp_lo, copy_on_conflict_tp_hi, copy_on_conflict_t4, copy_on_conflict_u1, copy_on_conflict_u3, copy_on_conflict_up, copy_on_conflict_up_lo, copy_on_conflict_up_hi, - copy_on_conflict_u6, copy_on_conflict_u7, copy_on_conflict_u7_log CASCADE; + copy_on_conflict_u6, copy_on_conflict_u7, copy_on_conflict_u7_log, + copy_on_conflict_u8, copy_on_conflict_u9 CASCADE; +DROP FUNCTION IF EXISTS copy_on_conflict_u7_trg();