feat: COPY ON CONFLICT (DO NOTHING / DO UPDATE) — AnalyticDB-compatible syntax

doc/src/sgml/ref/copy.sgml

@@ -48,6 +48,7 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
     REJECT_LIMIT <replaceable class="parameter">maxerror</replaceable>
     ENCODING '<replaceable class="parameter">encoding_name</replaceable>'
     LOG_VERBOSITY <replaceable class="parameter">verbosity</replaceable>
+    DO ON CONFLICT { DO NOTHING | DO UPDATE }
 </synopsis>
  </refsynopsisdiv>
 
@@ -479,6 +480,56 @@ COPY (SELECT j FROM (VALUES ('null'::json), (NULL::json)) v(j))
     </listitem>
    </varlistentry>
 
+   <varlistentry id="sql-copy-params-on-conflict">
+    <term><literal>ON CONFLICT</literal></term>
+    <listitem>
+     <para>
+      Specifies how to behave when an input row conflicts with a unique or
+      exclusion constraint on the target table, instead of aborting the whole
+      <command>COPY</command> command.  This clause is compatible with
+      <literal>COPY ON CONFLICT</literal> in Alibaba Cloud AnalyticDB for
+      PostgreSQL and is only applicable to <command>COPY FROM</command>.
+     </para>
+     <para>
+      <literal>DO ON CONFLICT DO NOTHING</literal> discards the conflicting
+      input row and continues with the next one.  The number of skipped rows
+      is reported in a <literal>NOTICE</literal> message at the end of the
+      command.
+     </para>
+     <para>
+      <literal>DO ON CONFLICT DO UPDATE</literal> overwrites the conflicting
+      row with the input row (a full-row overwrite, matching the AnalyticDB
+      semantics where no <literal>SET</literal> targets are allowed).  Note
+      that for a column subset <literal>COPY (column [, ...]) FROM ...</literal>,
+      columns not listed in the input are filled with their default values,
+      so a volatile default expression (e.g. <literal>nextval</literal>) will
+      produce a new value for each imported row and overwrite the stored one.
+     </para>
+     <para>
+      The number of updated rows is reported in a <literal>NOTICE</literal>
+      message at the end of the command.
+     </para>
+     <para>
+      The target table must have at least one unique or exclusion constraint
+      (primary key, unique index, or <literal>EXCLUDE</literal> constraint)
+      for the conflict check to be meaningful; otherwise the
+      <command>COPY</command> command fails at startup.  For partitioned
+      tables, uniqueness is enforced per leaf partition.
+     </para>
+     <para>
+      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
+      <literal>DO ON CONFLICT DO UPDATE</literal>, a row that is concurrently
+      modified is locked and the update is retried once the concurrent
+      transaction completes.  Duplicate constrained values within a single
+      <command>COPY</command> raise the same error as
+      <command>INSERT ... ON CONFLICT</command>.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry id="sql-copy-params-reject-limit">
     <term><literal>REJECT_LIMIT</literal></term>
     <listitem>

src/backend/commands/copy.c

@@ -503,6 +503,35 @@ defGetCopyOnErrorChoice(DefElem *def, ParseState *pstate, bool is_from)
 	return COPY_ON_ERROR_STOP;	/* keep compiler quiet */
 }
 
+/*
+ * Extract a CopyOnConflictChoice value from a DefElem.
+ */
+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)
+		return COPY_ON_CONFLICT_UPDATE;
+
+	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 +616,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 +790,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)