Add new COPY option SAVE_ERROR_TO

Currently, when source data contains unexpected data regarding data type or
range, the entire COPY fails. However, in some cases, such data can be ignored
and just copying normal data is preferable.

This commit adds a new option SAVE_ERROR_TO, which specifies where to save the
error information. When this option is specified, COPY skips soft errors and
continues copying.

Currently, SAVE_ERROR_TO only supports "none". This indicates error information
is not saved and COPY just skips the unexpected data and continues running.

Later works are expected to add more choices, such as 'log' and 'table'.

Author: Damir Belyalov, Atsushi Torikoshi, Alex Shulgin, Jian He
Discussion: https://postgr.es/m/87k31ftoe0.fsf_-_%40commandprompt.com
Reviewed-by: Pavel Stehule, Andres Freund, Tom Lane, Daniel Gustafsson,
Reviewed-by: Alena Rybakina, Andy Fan, Andrei Lepikhov, Masahiko Sawada
Reviewed-by: Vignesh C, Atsushi Torikoshi

1 files changed, 48 insertions, 0 deletions

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index 37836a769c7..46b23e345b8 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -42,6 +42,7 @@
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
 #include "miscadmin.h"
+#include "nodes/miscnodes.h"
 #include "optimizer/optimizer.h"
 #include "pgstat.h"
 #include "rewrite/rewriteHandler.h"
@@ -656,6 +657,9 @@ CopyFrom(CopyFromState cstate)
 	Assert(cstate->rel);
 	Assert(list_length(cstate->range_table) == 1);
 
+	if (cstate->opts.save_error_to != COPY_SAVE_ERROR_TO_ERROR)
+		Assert(cstate->escontext);
+
 	/*
 	 * The target must be a plain, foreign, or partitioned relation, or have
 	 * an INSTEAD OF INSERT row trigger.  (Currently, such triggers are only
@@ -992,6 +996,25 @@ CopyFrom(CopyFromState cstate)
 		if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
 			break;
 
+		if (cstate->opts.save_error_to != COPY_SAVE_ERROR_TO_ERROR &&
+			cstate->escontext->error_occurred)
+		{
+			/*
+			 * Soft error occured, skip this tuple and save error information
+			 * according to SAVE_ERROR_TO.
+			 */
+			if (cstate->opts.save_error_to == COPY_SAVE_ERROR_TO_NONE)
+
+				/*
+				 * Just make ErrorSaveContext ready for the next NextCopyFrom.
+				 * Since we don't set details_wanted and error_data is not to
+				 * be filled, just resetting error_occurred is enough.
+				 */
+				cstate->escontext->error_occurred = false;
+
+			continue;
+		}
+
 		ExecStoreVirtualTuple(myslot);
 
 		/*
@@ -1284,6 +1307,14 @@ CopyFrom(CopyFromState cstate)
 	/* Done, clean up */
 	error_context_stack = errcallback.previous;
 
+	if (cstate->opts.save_error_to != COPY_SAVE_ERROR_TO_ERROR &&
+		cstate->num_errors > 0)
+		ereport(NOTICE,
+				errmsg_plural("%zd row were skipped due to data type incompatibility",
+							  "%zd rows were skipped due to data type incompatibility",
+							  cstate->num_errors,
+							  cstate->num_errors));
+
 	if (bistate != NULL)
 		FreeBulkInsertState(bistate);
 
@@ -1419,6 +1450,23 @@ BeginCopyFrom(ParseState *pstate,
 		}
 	}
 
+	/* Set up soft error handler for SAVE_ERROR_TO */
+	if (cstate->opts.save_error_to != COPY_SAVE_ERROR_TO_ERROR)
+	{
+		cstate->escontext = makeNode(ErrorSaveContext);
+		cstate->escontext->type = T_ErrorSaveContext;
+		cstate->escontext->error_occurred = false;
+
+		/*
+		 * Currently we only support COPY_SAVE_ERROR_TO_NONE. We'll add other
+		 * options later
+		 */
+		if (cstate->opts.save_error_to == COPY_SAVE_ERROR_TO_NONE)
+			cstate->escontext->details_wanted = false;
+	}
+	else
+		cstate->escontext = NULL;
+
 	/* Convert FORCE_NULL name list to per-column flags, check validity */
 	cstate->opts.force_null_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
 	if (cstate->opts.force_null_all)