Subject: POC PATCH: copy from ... exceptions to: (was Re: VLDB Features)

From: Alex Shulgin <[email-masked]>
Date: Fri, 26 Dec 2014 00:23:19 +0300

Trent Shipley <[email-masked]> writes:

> On Friday 2007-12-14 16:22, Tom Lane wrote:
>> Neil Conway <[email-masked]> writes:
>> > By modifying COPY: COPY IGNORE ERRORS or some such would instruct COPY
>> > to drop (and log) rows that contain malformed data. That is, rows with
>> > too many or too few columns, rows that result in constraint violations,
>> > and rows containing columns where the data type's input function raises
>> > an error. The last case is the only thing that would be a bit tricky to
>> > implement, I think: you could use PG_TRY() around the InputFunctionCall,
>> > but I guess you'd need a subtransaction to ensure that you reset your
>> > state correctly after catching an error.
>>
>> Yeah.  It's the subtransaction per row that's daunting --- not only the
>> cycles spent for that, but the ensuing limitation to 4G rows imported
>> per COPY.
>
> You could extend the COPY FROM syntax with a COMMIT EVERY n clause.  This 
> would help with the 4G subtransaction limit.  The cost to the ETL process is 
> that a simple rollback would not be guaranteed send the process back to it's 
> initial state.  There are easy ways to deal with the rollback issue though.  
>
> A {NO} RETRY {USING algorithm} clause might be useful.   If the NO RETRY 
> option is selected then the COPY FROM can run without subtransactions and in 
> excess of the 4G per transaction limit.  NO RETRY should be the default since 
> it preserves the legacy behavior of COPY FROM.
>
> You could have an EXCEPTIONS TO {filename|STDERR} clause. I would not give the 
> option of sending exceptions to a table since they are presumably malformed, 
> otherwise they would not be exceptions.  (Users should re-process exception 
> files if they want an if good then table a else exception to table b ...)
>
> EXCEPTIONS TO and NO RETRY would be mutually exclusive.
>
>
>> If we could somehow only do a subtransaction per failure, things would
>> be much better, but I don't see how.

Hello,

Attached is a proof of concept patch for this TODO item.  There is no
docs yet, I just wanted to know if approach is sane.

The added syntax is like the following:

  COPY [table] FROM [file/program/stdin] EXCEPTIONS TO [file or stdout]

The way it's done it is abusing Copy Both mode and from my limited
testing, that seems to just work.  The error trapping itself is done
using PG_TRY/PG_CATCH and can only catch formatting or before-insert
trigger errors, no attempt is made to recover from a failed unique
constraint, etc.

Example in action:

postgres=# \d test_copy2
  Table "public.test_copy2"
 Column |  Type   | Modifiers 
--------+---------+-----------
 id     | integer | 
 val    | integer | 

postgres=# copy test_copy2 from program 'seq 3' exceptions to stdout;
1
NOTICE:  missing data for column "val"
CONTEXT:  COPY test_copy2, line 1: "1"
2
NOTICE:  missing data for column "val"
CONTEXT:  COPY test_copy2, line 2: "2"
3
NOTICE:  missing data for column "val"
CONTEXT:  COPY test_copy2, line 3: "3"
NOTICE:  total exceptions ignored: 3

postgres=# \d test_copy1
  Table "public.test_copy1"
 Column |  Type   | Modifiers 
--------+---------+-----------
 id     | integer | not null

postgres=# set client_min_messages to warning;
SET
postgres=# copy test_copy1 from program 'ls /proc' exceptions to stdout;
...
vmstat
zoneinfo
postgres=# 

Limited performance testing shows no significant difference between
error-catching and plain code path.  For example, timing

  copy test_copy1 from program 'seq 1000000' [exceptions to stdout]

shows similar numbers with or without the added "exceptions to" clause.

Now that I'm sending this I wonder if the original comment about the
need for subtransaction around every loaded line still holds.  Any
example of what would be not properly rolled back by just PG_TRY?

Happy hacking!
--
Alex

>From 50f7ab0a503a0d61776add8a138abf2622fc6c35 Mon Sep 17 00:00:00 2001
From: Alex Shulgin <[email-masked]>
Date: Fri, 19 Dec 2014 18:21:31 +0300
Subject: [PATCH] POC: COPY FROM ... EXCEPTIONS TO

---
 contrib/file_fdw/file_fdw.c             |   4 +-
 src/backend/commands/copy.c             | 251 +++++++++++++++++++++++++++++---
 src/backend/parser/gram.y               |  26 +++-
 src/bin/psql/common.c                   |  14 +-
 src/bin/psql/copy.c                     | 119 ++++++++++++++-
 src/bin/psql/settings.h                 |   1 +
 src/bin/psql/startup.c                  |   1 +
 src/bin/psql/tab-complete.c             |  12 +-
 src/include/commands/copy.h             |   3 +-
 src/include/nodes/parsenodes.h          |   1 +
 src/include/parser/kwlist.h             |   1 +
 src/interfaces/ecpg/preproc/ecpg.addons |   2 +-
 12 files changed, 396 insertions(+), 39 deletions(-)

diff --git a/contrib/file_fdw/file_fdw.c b/contrib/file_fdw/file_fdw.c
new file mode 100644
index 5a4d5aa..0df02f7
*** a/contrib/file_fdw/file_fdw.c
--- b/contrib/file_fdw/file_fdw.c
*************** fileBeginForeignScan(ForeignScanState *n
*** 624,629 ****
--- 624,630 ----
  	cstate = BeginCopyFrom(node->ss.ss_currentRelation,
  						   filename,
  						   false,
+ 						   NULL,
  						   NIL,
  						   options);
  
*************** fileReScanForeignScan(ForeignScanState *
*** 697,702 ****
--- 698,704 ----
  	festate->cstate = BeginCopyFrom(node->ss.ss_currentRelation,
  									festate->filename,
  									false,
+ 									NULL,
  									NIL,
  									festate->options);
  }
*************** file_acquire_sample_rows(Relation onerel
*** 1030,1036 ****
  	/*
  	 * Create CopyState from FDW options.
  	 */
! 	cstate = BeginCopyFrom(onerel, filename, false, NIL, options);
  
  	/*
  	 * Use per-tuple memory context to prevent leak of memory used to read
--- 1032,1038 ----
  	/*
  	 * Create CopyState from FDW options.
  	 */
! 	cstate = BeginCopyFrom(onerel, filename, false, NULL, NIL, options);
  
  	/*
  	 * Use per-tuple memory context to prevent leak of memory used to read
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
new file mode 100644
index 08abe14..4f59c63
*** a/src/backend/commands/copy.c
--- b/src/backend/commands/copy.c
*************** typedef enum EolType
*** 96,102 ****
  typedef struct CopyStateData
  {
  	/* low-level state data */
! 	CopyDest	copy_dest;		/* type of copy source/destination */
  	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
  	StringInfo	fe_msgbuf;		/* used for all dests during COPY TO, only for
  								 * dest == COPY_NEW_FE in COPY FROM */
--- 96,103 ----
  typedef struct CopyStateData
  {
  	/* low-level state data */
! 	CopyDest	copy_src;		/* type of copy source */
! 	CopyDest	copy_dest;		/* type of copy destination */
  	FILE	   *copy_file;		/* used if copy_dest == COPY_FILE */
  	StringInfo	fe_msgbuf;		/* used for all dests during COPY TO, only for
  								 * dest == COPY_NEW_FE in COPY FROM */
*************** typedef struct CopyStateData
*** 105,110 ****
--- 106,114 ----
  	int			file_encoding;	/* file or remote side's character encoding */
  	bool		need_transcoding;		/* file encoding diff from server? */
  	bool		encoding_embeds_ascii;	/* ASCII can be non-first byte? */
+ 	bool		ignore_exceptions;		/* should we trap and ignore exceptions? */
+ 	FILE	   *exc_file;		/* file stream to write erroring lines to */
+ 	uint64		exceptions;		/* total number of exceptions ignored */
  
  	/* parameters from the COPY command */
  	Relation	rel;			/* relation to copy to or from */
*************** typedef struct CopyStateData
*** 112,117 ****
--- 116,122 ----
  	List	   *attnumlist;		/* integer list of attnums to copy */
  	char	   *filename;		/* filename, or NULL for STDIN/STDOUT */
  	bool		is_program;		/* is 'filename' a program to popen? */
+ 	char	   *exc_filename;	/* filename for exceptions or NULL for STDOUT */
  	bool		binary;			/* binary format? */
  	bool		oids;			/* include OIDs? */
  	bool		freeze;			/* freeze rows on loading? */
*************** SendCopyBegin(CopyState cstate)
*** 347,352 ****
--- 352,366 ----
  		int16		format = (cstate->binary ? 1 : 0);
  		int			i;
  
+ 		/*
+ 		 * Check if we might need to stream exceptions to the frontend.  If
+ 		 * so, this must be a "COPY FROM file/program EXCEPTIONS TO STDOUT".
+ 		 *
+ 		 * We need to create the frontend message buffer now.
+ 		 */
+ 		if (cstate->ignore_exceptions)
+ 			cstate->fe_msgbuf = makeStringInfo();
+ 
  		pq_beginmessage(&buf, 'H');
  		pq_sendbyte(&buf, format);		/* overall format */
  		pq_sendint(&buf, natts, 2);
*************** ReceiveCopyBegin(CopyState cstate)
*** 388,404 ****
  	{
  		/* new way */
  		StringInfoData buf;
  		int			natts = list_length(cstate->attnumlist);
  		int16		format = (cstate->binary ? 1 : 0);
  		int			i;
  
! 		pq_beginmessage(&buf, 'G');
  		pq_sendbyte(&buf, format);		/* overall format */
  		pq_sendint(&buf, natts, 2);
  		for (i = 0; i < natts; i++)
  			pq_sendint(&buf, format, 2);		/* per-column formats */
  		pq_endmessage(&buf);
! 		cstate->copy_dest = COPY_NEW_FE;
  		cstate->fe_msgbuf = makeStringInfo();
  	}
  	else if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2)
--- 402,428 ----
  	{
  		/* new way */
  		StringInfoData buf;
+ 		char		msgid = 'G';		/* receiving from client only */
  		int			natts = list_length(cstate->attnumlist);
  		int16		format = (cstate->binary ? 1 : 0);
  		int			i;
  
! 		/*
! 		 * Check if we also need to pipe exceptions back to the frontend.
! 		 */
! 		if (cstate->ignore_exceptions && cstate->exc_filename == NULL)
! 		{
! 			msgid = 'W';		/* copying in both directions */
! 			cstate->copy_dest = COPY_NEW_FE;
! 		}
! 
! 		pq_beginmessage(&buf, msgid);
  		pq_sendbyte(&buf, format);		/* overall format */
  		pq_sendint(&buf, natts, 2);
  		for (i = 0; i < natts; i++)
  			pq_sendint(&buf, format, 2);		/* per-column formats */
  		pq_endmessage(&buf);
! 		cstate->copy_src = COPY_NEW_FE;
  		cstate->fe_msgbuf = makeStringInfo();
  	}
  	else if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2)
*************** ReceiveCopyBegin(CopyState cstate)
*** 409,415 ****
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  			errmsg("COPY BINARY is not supported to stdout or from stdin")));
  		pq_putemptymessage('G');
! 		cstate->copy_dest = COPY_OLD_FE;
  	}
  	else
  	{
--- 433,439 ----
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  			errmsg("COPY BINARY is not supported to stdout or from stdin")));
  		pq_putemptymessage('G');
! 		cstate->copy_src = COPY_OLD_FE;
  	}
  	else
  	{
*************** ReceiveCopyBegin(CopyState cstate)
*** 419,425 ****
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  			errmsg("COPY BINARY is not supported to stdout or from stdin")));
  		pq_putemptymessage('D');
! 		cstate->copy_dest = COPY_OLD_FE;
  	}
  	/* We *must* flush here to ensure FE knows it can send. */
  	pq_flush();
--- 443,449 ----
  					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
  			errmsg("COPY BINARY is not supported to stdout or from stdin")));
  		pq_putemptymessage('D');
! 		cstate->copy_src = COPY_OLD_FE;
  	}
  	/* We *must* flush here to ensure FE knows it can send. */
  	pq_flush();
*************** CopySendChar(CopyState cstate, char c)
*** 472,486 ****
  	appendStringInfoCharMacro(cstate->fe_msgbuf, c);
  }
  
  static void
  CopySendEndOfRow(CopyState cstate)
  {
  	StringInfo	fe_msgbuf = cstate->fe_msgbuf;
  
  	switch (cstate->copy_dest)
  	{
  		case COPY_FILE:
! 			if (!cstate->binary)
  			{
  				/* Default line termination depends on platform */
  #ifndef WIN32
--- 496,560 ----
  	appendStringInfoCharMacro(cstate->fe_msgbuf, c);
  }
  
+ /*
+  * This should be called from PG_CATCH() after switching to appropriate
+  * MemoryContext.
+  */
+ static void
+ CopySendException(CopyState cstate)
+ {
+ 	ErrorData	*error;
+ 
+ 	++cstate->exceptions;
+ 
+ 	/*
+ 	 * When reading from the frontend, we reuse the current line held in the
+ 	 * message buffer to send the exception line back, otherwise we need to
+ 	 * copy the line over from the line buffer.
+ 	 */
+ 	if (cstate->copy_src == COPY_FILE)
+ 		CopySendData(cstate, cstate->line_buf.data, cstate->line_buf.len);
+ 
+ 	/* this flushes the message buffer */
+ 	CopySendEndOfRow(cstate);
+ 
+ 	error = CopyErrorData();
+ 	FlushErrorState();
+ 
+ 	/* report error as a harmless notice */
+ 	ereport(NOTICE,
+ 			(errmsg("%s", error->message)));
+ 	FreeErrorData(error);
+ }
+ 
  static void
  CopySendEndOfRow(CopyState cstate)
  {
  	StringInfo	fe_msgbuf = cstate->fe_msgbuf;
+ 	FILE	   *file;
+ 	bool		should_add_newline;
+ 
+ 	/* determine where are we writing to */
+ 	if (cstate->ignore_exceptions)
+ 	{
+ 		file = cstate->exc_file;
+ 		/*
+ 		 * We should only add a newline if we're not sending the frontend what
+ 		 * it has just sent us and in any case we shouldn't do this for binary
+ 		 * copy.
+ 		 */
+ 		should_add_newline = (cstate->copy_src == COPY_FILE && !cstate->binary);
+ 	}
+ 	else
+ 	{
+ 		file = cstate->copy_file;
+ 		should_add_newline = !(cstate->binary);
+ 	}
  
  	switch (cstate->copy_dest)
  	{
  		case COPY_FILE:
! 			if (should_add_newline)
  			{
  				/* Default line termination depends on platform */
  #ifndef WIN32
*************** CopySendEndOfRow(CopyState cstate)
*** 490,498 ****
  #endif
  			}
  
! 			if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1,
! 					   cstate->copy_file) != 1 ||
! 				ferror(cstate->copy_file))
  			{
  				if (cstate->is_program)
  				{
--- 564,571 ----
  #endif
  			}
  
! 			if (fwrite(fe_msgbuf->data, fe_msgbuf->len, 1, file) != 1 ||
! 				ferror(file))
  			{
  				if (cstate->is_program)
  				{
*************** CopySendEndOfRow(CopyState cstate)
*** 525,531 ****
  			break;
  		case COPY_OLD_FE:
  			/* The FE/BE protocol uses \n as newline for all platforms */
! 			if (!cstate->binary)
  				CopySendChar(cstate, '\n');
  
  			if (pq_putbytes(fe_msgbuf->data, fe_msgbuf->len))
--- 598,604 ----
  			break;
  		case COPY_OLD_FE:
  			/* The FE/BE protocol uses \n as newline for all platforms */
! 			if (should_add_newline)
  				CopySendChar(cstate, '\n');
  
  			if (pq_putbytes(fe_msgbuf->data, fe_msgbuf->len))
*************** CopySendEndOfRow(CopyState cstate)
*** 538,544 ****
  			break;
  		case COPY_NEW_FE:
  			/* The FE/BE protocol uses \n as newline for all platforms */
! 			if (!cstate->binary)
  				CopySendChar(cstate, '\n');
  
  			/* Dump the accumulated row as one CopyData message */
--- 611,617 ----
  			break;
  		case COPY_NEW_FE:
  			/* The FE/BE protocol uses \n as newline for all platforms */
! 			if (should_add_newline)
  				CopySendChar(cstate, '\n');
  
  			/* Dump the accumulated row as one CopyData message */
*************** CopySendEndOfRow(CopyState cstate)
*** 546,552 ****
  			break;
  	}
  
! 	resetStringInfo(fe_msgbuf);
  }
  
  /*
--- 619,630 ----
  			break;
  	}
  
! 	/*
! 	 * Avoid resetting the buffer we reused to send the exception line back to
! 	 * the frontend.
! 	 */
! 	if (!cstate->ignore_exceptions || cstate->copy_src == COPY_FILE)
! 		resetStringInfo(fe_msgbuf);
  }
  
  /*
*************** CopyGetData(CopyState cstate, void *data
*** 567,573 ****
  {
  	int			bytesread = 0;
  
! 	switch (cstate->copy_dest)
  	{
  		case COPY_FILE:
  			bytesread = fread(databuf, 1, maxread, cstate->copy_file);
--- 645,651 ----
  {
  	int			bytesread = 0;
  
! 	switch (cstate->copy_src)
  	{
  		case COPY_FILE:
  			bytesread = fread(databuf, 1, maxread, cstate->copy_file);
*************** DoCopy(const CopyStmt *stmt, const char
*** 919,930 ****
  			PreventCommandIfReadOnly("COPY FROM");
  
  		cstate = BeginCopyFrom(rel, stmt->filename, stmt->is_program,
! 							   stmt->attlist, stmt->options);
  		*processed = CopyFrom(cstate);	/* copy from file to database */
  		EndCopyFrom(cstate);
  	}
  	else
  	{
  		cstate = BeginCopyTo(rel, query, queryString, relid,
  							 stmt->filename, stmt->is_program,
  							 stmt->attlist, stmt->options);
--- 997,1018 ----
  			PreventCommandIfReadOnly("COPY FROM");
  
  		cstate = BeginCopyFrom(rel, stmt->filename, stmt->is_program,
! 							   stmt->exc_filename, stmt->attlist, stmt->options);
  		*processed = CopyFrom(cstate);	/* copy from file to database */
+ 		if (cstate->exceptions)
+ 			ereport(NOTICE,
+ 					(errmsg("total exceptions ignored: " UINT64_FORMAT,
+ 							cstate->exceptions)));
  		EndCopyFrom(cstate);
  	}
  	else
  	{
+ 		if (stmt->exc_filename != NULL)
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_SYNTAX_ERROR),
+ 					 errmsg("EXCEPTIONS TO not allowed with COPY ... TO"),
+ 					 errhint("see COPY ... FROM")));
+ 
  		cstate = BeginCopyTo(rel, query, queryString, relid,
  							 stmt->filename, stmt->is_program,
  							 stmt->attlist, stmt->options);
*************** BeginCopy(bool is_from,
*** 1561,1566 ****
--- 1649,1655 ----
  	/* See Multibyte encoding comment above */
  	cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding);
  
+ 	cstate->copy_src  = COPY_FILE;		/* default */
  	cstate->copy_dest = COPY_FILE;		/* default */
  
  	MemoryContextSwitchTo(oldcontext);
*************** EndCopy(CopyState cstate)
*** 1608,1613 ****
--- 1697,1708 ----
  							cstate->filename)));
  	}
  
+ 	if (cstate->exc_filename != NULL && FreeFile(cstate->exc_file))
+ 		ereport(ERROR,
+ 				(errcode_for_file_access(),
+ 				 errmsg("could not close file \"%s\": %m",
+ 						cstate->exc_filename)));
+ 
  	MemoryContextDelete(cstate->copycontext);
  	pfree(cstate);
  }
*************** CopyFrom(CopyState cstate)
*** 2331,2336 ****
--- 2426,2432 ----
  	{
  		TupleTableSlot *slot;
  		bool		skip_tuple;
+ 		bool		depleted;
  		Oid			loaded_oid = InvalidOid;
  
  		CHECK_FOR_INTERRUPTS();
*************** CopyFrom(CopyState cstate)
*** 2348,2356 ****
  		/* Switch into its memory context */
  		MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
  
! 		if (!NextCopyFrom(cstate, econtext, values, nulls, &loaded_oid))
  			break;
  
  		/* And now we can form the input tuple. */
  		tuple = heap_form_tuple(tupDesc, values, nulls);
  
--- 2444,2475 ----
  		/* Switch into its memory context */
  		MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
  
! 		skip_tuple = false;
! 		depleted = false;
! 
! 		PG_TRY();
! 		{
! 			if (!NextCopyFrom(cstate, econtext, values, nulls, &loaded_oid))
! 				/* can't break right here due to PG_TRY using do/while(0) */
! 				depleted = true;
! 		}
! 		PG_CATCH();
! 		{
! 			if (!cstate->ignore_exceptions)
! 				PG_RE_THROW();
! 
! 			skip_tuple = true;
! 			MemoryContextSwitchTo(oldcontext);
! 			CopySendException(cstate);
! 		}
! 		PG_END_TRY();
! 
! 		if (depleted)
  			break;
  
+ 		if (skip_tuple)
+ 			continue;
+ 
  		/* And now we can form the input tuple. */
  		tuple = heap_form_tuple(tupDesc, values, nulls);
  
*************** CopyFrom(CopyState cstate)
*** 2376,2382 ****
  		if (resultRelInfo->ri_TrigDesc &&
  			resultRelInfo->ri_TrigDesc->trig_insert_before_row)
  		{
! 			slot = ExecBRInsertTriggers(estate, resultRelInfo, slot);
  
  			if (slot == NULL)	/* "do nothing" */
  				skip_tuple = true;
--- 2495,2514 ----
  		if (resultRelInfo->ri_TrigDesc &&
  			resultRelInfo->ri_TrigDesc->trig_insert_before_row)
  		{
! 			PG_TRY();
! 			{
! 				slot = ExecBRInsertTriggers(estate, resultRelInfo, slot);
! 			}
! 			PG_CATCH();
! 			{
! 				if (!cstate->ignore_exceptions)
! 					PG_RE_THROW();
! 
! 				slot = NULL;
! 				MemoryContextSwitchTo(oldcontext);
! 				CopySendException(cstate);
! 			}
! 			PG_END_TRY();
  
  			if (slot == NULL)	/* "do nothing" */
  				skip_tuple = true;
*************** CopyFrom(CopyState cstate)
*** 2384,2395 ****
  				tuple = ExecMaterializeSlot(slot);
  		}
  
! 		if (!skip_tuple)
  		{
! 			/* Check the constraints of the tuple */
! 			if (cstate->rel->rd_att->constr)
  				ExecConstraints(resultRelInfo, slot, estate);
  
  			if (useHeapMultiInsert)
  			{
  				/* Add this tuple to the tuple buffer */
--- 2516,2545 ----
  				tuple = ExecMaterializeSlot(slot);
  		}
  
! 		if (skip_tuple)
! 			continue;
! 
! 		/* Check the constraints of the tuple */
! 		if (cstate->rel->rd_att->constr)
  		{
! 			PG_TRY();
! 			{
  				ExecConstraints(resultRelInfo, slot, estate);
+ 			}
+ 			PG_CATCH();
+ 			{
+ 				if (!cstate->ignore_exceptions)
+ 					PG_RE_THROW();
  
+ 				skip_tuple = true;
+ 				MemoryContextSwitchTo(oldcontext);
+ 				CopySendException(cstate);
+ 			}
+ 			PG_END_TRY();
+ 		}
+ 
+ 		if (!skip_tuple)
+ 		{
  			if (useHeapMultiInsert)
  			{
  				/* Add this tuple to the tuple buffer */
*************** CopyState
*** 2573,2583 ****
--- 2723,2736 ----
  BeginCopyFrom(Relation rel,
  			  const char *filename,
  			  bool is_program,
+ 			  const char *exc_filename,
  			  List *attnamelist,
  			  List *options)
  {
  	CopyState	cstate;
  	bool		pipe = (filename == NULL);
+ 	bool		ignore_exceptions = (exc_filename != NULL);
+ 	bool		exc_pipe = (exc_filename != NULL && *exc_filename == 0);
  	TupleDesc	tupDesc;
  	Form_pg_attribute *attr;
  	AttrNumber	num_phys_attrs,
*************** BeginCopyFrom(Relation rel,
*** 2686,2691 ****
--- 2839,2900 ----
  	cstate->volatile_defexprs = volatile_defexprs;
  	cstate->num_defaults = num_defaults;
  	cstate->is_program = is_program;
+ 	cstate->ignore_exceptions = ignore_exceptions;
+ 
+ 	if (ignore_exceptions)
+ 	{
+ 		if (exc_pipe)
+ 		{
+ 			if (whereToSendOutput == DestRemote)
+ 			{
+ 				if (!pipe)
+ 					SendCopyBegin(cstate);
+ 				else
+ 					; /* handled by ReceiveCopyBegin() call below */
+ 			}
+ 			else
+ 			{
+ 				cstate->exc_file = stdout;
+ 			}
+ 		}
+ 		else
+ 		{
+ 			if (!is_absolute_path(exc_filename))
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_INVALID_NAME),
+ 						 errmsg("relative path not allowed for EXCEPTIONS TO file")));
+ 
+ 			if (!pipe)
+ 			{
+ 				struct stat		stbuf_exc;
+ 				struct stat		stbuf_from;
+ 
+ 				/* check if both FROM and EXCEPTIONS TO are the same file */
+ 				if (stat(exc_filename, &stbuf_exc) == 0 &&
+ 					stat(filename, &stbuf_from) == 0 &&
+ 					stbuf_exc.st_dev == stbuf_from.st_dev &&
+ 					stbuf_exc.st_ino == stbuf_from.st_ino)
+ 				{
+ 					ereport(ERROR,
+ 							(errcode(ERRCODE_INVALID_NAME),
+ 							 errmsg("cannot specify the same file for COPY FROM and EXCEPTIONS TO")));
+ 				}
+ 				/*
+ 				 * We won't send or receive data via frontend, but we still
+ 				 * need to the buffer for CopySendException() to work with.
+ 				 */
+ 				cstate->fe_msgbuf = makeStringInfo();
+ 			}
+ 
+ 			cstate->exc_filename = pstrdup(exc_filename);
+ 			cstate->exc_file = AllocateFile(cstate->exc_filename, PG_BINARY_W);
+ 			if (cstate->exc_file == NULL)
+ 				ereport(ERROR,
+ 						(errcode_for_file_access(),
+ 						 errmsg("could not open file \"%s\" for writing: %m",
+ 								cstate->exc_filename)));
+ 		}
+ 	}
  
  	if (pipe)
  	{
*************** NextCopyFrom(CopyState cstate, ExprConte
*** 3019,3025 ****
  			 */
  			char		dummy;
  
! 			if (cstate->copy_dest != COPY_OLD_FE &&
  				CopyGetData(cstate, &dummy, 1, 1) > 0)
  				ereport(ERROR,
  						(errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
--- 3228,3234 ----
  			 */
  			char		dummy;
  
! 			if (cstate->copy_src != COPY_OLD_FE &&
  				CopyGetData(cstate, &dummy, 1, 1) > 0)
  				ereport(ERROR,
  						(errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
*************** CopyReadLine(CopyState cstate)
*** 3133,3139 ****
  		 * after \. up to the protocol end of copy data.  (XXX maybe better
  		 * not to treat \. as special?)
  		 */
! 		if (cstate->copy_dest == COPY_NEW_FE)
  		{
  			do
  			{
--- 3342,3348 ----
  		 * after \. up to the protocol end of copy data.  (XXX maybe better
  		 * not to treat \. as special?)
  		 */
! 		if (cstate->copy_src == COPY_NEW_FE)
  		{
  			do
  			{
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
new file mode 100644
index 6431601..47f2be2
*** a/src/backend/parser/gram.y
--- b/src/backend/parser/gram.y
*************** static Node *makeRecursiveViewSelect(cha
*** 307,313 ****
  %type <defelt>	event_trigger_when_item
  %type <chr>		enable_trigger
  
! %type <str>		copy_file_name
  				database_name access_method_clause access_method attr_name
  				name cursor_name file_name
  				index_name opt_index_name cluster_index_specification
--- 307,313 ----
  %type <defelt>	event_trigger_when_item
  %type <chr>		enable_trigger
  
! %type <str>		copy_file_name opt_copy_exceptions
  				database_name access_method_clause access_method attr_name
  				name cursor_name file_name
  				index_name opt_index_name cluster_index_specification
*************** static Node *makeRecursiveViewSelect(cha
*** 561,567 ****
  	DEFERRABLE DEFERRED DEFINER DELETE_P DELIMITER DELIMITERS DESC
  	DICTIONARY DISABLE_P DISCARD DISTINCT DO DOCUMENT_P DOMAIN_P DOUBLE_P DROP
  
! 	EACH ELSE ENABLE_P ENCODING ENCRYPTED END_P ENUM_P ESCAPE EVENT EXCEPT
  	EXCLUDE EXCLUDING EXCLUSIVE EXECUTE EXISTS EXPLAIN
  	EXTENSION EXTERNAL EXTRACT
  
--- 561,568 ----
  	DEFERRABLE DEFERRED DEFINER DELETE_P DELIMITER DELIMITERS DESC
  	DICTIONARY DISABLE_P DISCARD DISTINCT DO DOCUMENT_P DOMAIN_P DOUBLE_P DROP
  
! 	EACH ELSE ENABLE_P ENCODING ENCRYPTED END_P ENUM_P ESCAPE EVENT
! 	EXCEPT EXCEPTIONS
  	EXCLUDE EXCLUDING EXCLUSIVE EXECUTE EXISTS EXPLAIN
  	EXTENSION EXTERNAL EXTRACT
  
*************** ClosePortalStmt:
*** 2515,2521 ****
  /*****************************************************************************
   *
   *		QUERY :
!  *				COPY relname [(columnList)] FROM/TO file [WITH] [(options)]
   *				COPY ( SELECT ... ) TO file	[WITH] [(options)]
   *
   *				where 'file' can be one of:
--- 2516,2523 ----
  /*****************************************************************************
   *
   *		QUERY :
!  *				COPY relname [(columnList)] FROM/TO file [EXCEPTIONS TO file]
!  *					[WITH] [(options)]
   *				COPY ( SELECT ... ) TO file	[WITH] [(options)]
   *
   *				where 'file' can be one of:
*************** ClosePortalStmt:
*** 2534,2540 ****
   *****************************************************************************/
  
  CopyStmt:	COPY opt_binary qualified_name opt_column_list opt_oids
! 			copy_from opt_program copy_file_name copy_delimiter opt_with copy_options
  				{
  					CopyStmt *n = makeNode(CopyStmt);
  					n->relation = $3;
--- 2536,2543 ----
   *****************************************************************************/
  
  CopyStmt:	COPY opt_binary qualified_name opt_column_list opt_oids
! 			copy_from opt_program copy_file_name opt_copy_exceptions copy_delimiter
! 			opt_with copy_options
  				{
  					CopyStmt *n = makeNode(CopyStmt);
  					n->relation = $3;
*************** CopyStmt:	COPY opt_binary qualified_name
*** 2543,2548 ****
--- 2546,2552 ----
  					n->is_from = $6;
  					n->is_program = $7;
  					n->filename = $8;
+ 					n->exc_filename = $9;
  
  					if (n->is_program && n->filename == NULL)
  						ereport(ERROR,
*************** CopyStmt:	COPY opt_binary qualified_name
*** 2556,2565 ****
  						n->options = lappend(n->options, $2);
  					if ($5)
  						n->options = lappend(n->options, $5);
! 					if ($9)
! 						n->options = lappend(n->options, $9);
! 					if ($11)
! 						n->options = list_concat(n->options, $11);
  					$$ = (Node *)n;
  				}
  			| COPY select_with_parens TO opt_program copy_file_name opt_with copy_options
--- 2560,2569 ----
  						n->options = lappend(n->options, $2);
  					if ($5)
  						n->options = lappend(n->options, $5);
! 					if ($10)
! 						n->options = lappend(n->options, $10);
! 					if ($12)
! 						n->options = list_concat(n->options, $12);
  					$$ = (Node *)n;
  				}
  			| COPY select_with_parens TO opt_program copy_file_name opt_with copy_options
*************** copy_file_name:
*** 2604,2609 ****
--- 2608,2618 ----
  			| STDOUT								{ $$ = NULL; }
  		;
  
+ opt_copy_exceptions:
+ 			EXCEPTIONS TO copy_file_name			{ $$ = ($3 ? $3 : ""); }
+ 			| /* EMPTY */							{ $$ = NULL; }
+ 		;
+ 
  copy_options: copy_opt_list							{ $$ = $1; }
  			| '(' copy_generic_opt_list ')'			{ $$ = $2; }
  		;
*************** unreserved_keyword:
*** 13142,13147 ****
--- 13151,13157 ----
  			| ENUM_P
  			| ESCAPE
  			| EVENT
+ 			| EXCEPTIONS
  			| EXCLUDE
  			| EXCLUDING
  			| EXCLUSIVE
diff --git a/src/bin/psql/common.c b/src/bin/psql/common.c
new file mode 100644
index 66d80b5..d308286
*** a/src/bin/psql/common.c
--- b/src/bin/psql/common.c
*************** AcceptResult(const PGresult *result)
*** 388,393 ****
--- 388,394 ----
  			case PGRES_EMPTY_QUERY:
  			case PGRES_COPY_IN:
  			case PGRES_COPY_OUT:
+ 			case PGRES_COPY_BOTH:
  				/* Fine, do nothing */
  				OK = true;
  				break;
*************** ProcessResult(PGresult **results)
*** 751,756 ****
--- 752,758 ----
  
  			case PGRES_COPY_OUT:
  			case PGRES_COPY_IN:
+ 			case PGRES_COPY_BOTH:
  				is_copy = true;
  				break;
  
*************** ProcessResult(PGresult **results)
*** 777,784 ****
  			SetCancelConn();
  			if (result_status == PGRES_COPY_OUT)
  			{
! 				if (!copystream)
  					copystream = pset.queryFout;
  				success = handleCopyOut(pset.db,
  										copystream,
  										&copy_result) && success;
--- 779,793 ----
  			SetCancelConn();
  			if (result_status == PGRES_COPY_OUT)
  			{
! 				/*
! 				 * If we have the stream for exceptions, then this must be the
! 				 * capture phase: use it.
! 				 */
! 				if (pset.copyExcStream)
! 					copystream = pset.copyExcStream;
! 				else if (!copystream)
  					copystream = pset.queryFout;
+ 
  				success = handleCopyOut(pset.db,
  										copystream,
  										&copy_result) && success;
*************** ProcessResult(PGresult **results)
*** 794,800 ****
  					copy_result = NULL;
  				}
  			}
! 			else
  			{
  				if (!copystream)
  					copystream = pset.cur_cmd_source;
--- 803,809 ----
  					copy_result = NULL;
  				}
  			}
! 			else	/* PGRES_COPY_IN or PGRES_COPY_BOTH */
  			{
  				if (!copystream)
  					copystream = pset.cur_cmd_source;
*************** PrintQueryResults(PGresult *results)
*** 913,918 ****
--- 922,928 ----
  
  		case PGRES_COPY_OUT:
  		case PGRES_COPY_IN:
+ 		case PGRES_COPY_BOTH:
  			/* nothing to do here */
  			success = true;
  			break;
diff --git a/src/bin/psql/copy.c b/src/bin/psql/copy.c
new file mode 100644
index 010a593..cc36071
*** a/src/bin/psql/copy.c
--- b/src/bin/psql/copy.c
*************** struct copy_options
*** 58,63 ****
--- 58,66 ----
  	bool		program;		/* is 'file' a program to popen? */
  	bool		psql_inout;		/* true = use psql stdin/stdout */
  	bool		from;			/* true = FROM, false = TO */
+ 	bool		exceptions;		/* has EXCEPTIONS TO */
+ 	char	   *exc_file;		/* NULL = stdin/stdout */
+ 	bool		exc_psql_inout;	/* true = use psql stdout for exceptions */
  };

*************** free_copy_options(struct copy_options *
*** 69,74 ****
--- 72,78 ----
  	free(ptr->before_tofrom);
  	free(ptr->after_tofrom);
  	free(ptr->file);
+ 	free(ptr->exc_file);
  	free(ptr);
  }
  
*************** parse_slash_copy(const char *args)
*** 240,250 ****
  		expand_tilde(&result->file);
  	}
  
  	/* Collect the rest of the line (COPY options) */
  	token = strtokx(NULL, "", NULL, NULL,
  					0, false, false, pset.encoding);
  	if (token)
! 		result->after_tofrom = pg_strdup(token);
  
  	return result;
  
--- 244,302 ----
  		expand_tilde(&result->file);
  	}
  
+ 	result->after_tofrom = pg_strdup("");		/* initialize for appending */
+ 
+ 	/* check for COPY FROM ... EXCEPTIONS TO */
+ 	if (result->from)
+ 	{
+ 		token = strtokx(NULL, whitespace, NULL, NULL,
+ 						0, false, false, pset.encoding);
+ 		if (token)
+ 		{
+ 			if (pg_strcasecmp(token, "exceptions") == 0)
+ 			{
+ 				result->exceptions = true;
+ 
+ 				token = strtokx(NULL, whitespace, NULL, NULL,
+ 								0, false, false, pset.encoding);
+ 				if (!token || pg_strcasecmp(token, "to") != 0)
+ 					goto error;
+ 
+ 				token = strtokx(NULL, whitespace, ";", "'",
+ 								0, false, false, pset.encoding);
+ 				if (!token)
+ 					goto error;
+ 
+ 				if (pg_strcasecmp(token, "stdin") == 0 ||
+ 					pg_strcasecmp(token, "stdout") == 0)
+ 				{
+ 					result->exc_file = NULL;
+ 				}
+ 				else if (pg_strcasecmp(token, "pstdin") == 0 ||
+ 						 pg_strcasecmp(token, "pstdout") == 0)
+ 				{
+ 					result->exc_psql_inout = true;
+ 					result->exc_file = NULL;
+ 				}
+ 				else
+ 				{
+ 					strip_quotes(token, '\'', 0, pset.encoding);
+ 					result->exc_file = pg_strdup(token);
+ 					expand_tilde(&result->exc_file);
+ 				}
+ 			}
+ 			else
+ 			{
+ 				xstrcat(&result->after_tofrom, token);
+ 			}
+ 		}
+ 	}
+ 
  	/* Collect the rest of the line (COPY options) */
  	token = strtokx(NULL, "", NULL, NULL,
  					0, false, false, pset.encoding);
  	if (token)
! 		xstrcat(&result->after_tofrom, token);
  
  	return result;
  
*************** do_copy(const char *args)
*** 269,274 ****
--- 321,327 ----
  {
  	PQExpBufferData query;
  	FILE	   *copystream;
+ 	FILE	   *excstream;
  	struct copy_options *options;
  	bool		success;
  
*************** do_copy(const char *args)
*** 278,287 ****
  	if (!options)
  		return false;
  
! 	/* prepare to read or write the target file */
  	if (options->file && !options->program)
  		canonicalize_path(options->file);
  
  	if (options->from)
  	{
  		if (options->file)
--- 331,346 ----
  	if (!options)
  		return false;
  
! 	/* prepare to read or write the target file(s) */
  	if (options->file && !options->program)
  		canonicalize_path(options->file);
  
+ 	if (options->exc_file)
+ 		canonicalize_path(options->exc_file);
+ 
+ 	copystream = NULL;
+ 	excstream = NULL;
+ 
  	if (options->from)
  	{
  		if (options->file)
*************** do_copy(const char *args)
*** 294,305 ****
--- 353,394 ----
  				copystream = popen(options->file, PG_BINARY_R);
  			}
  			else
+ 			{
+ 				if (options->exc_file)
+ 				{
+ 					struct stat		stbuf_exc;
+ 					struct stat		stbuf_from;
+ 
+ 					/* check if both FROM and EXCEPTIONS TO are the same file */
+ 					if (stat(options->exc_file, &stbuf_exc) == 0 &&
+ 						stat(options->file, &stbuf_from) == 0 &&
+ 						stbuf_exc.st_dev == stbuf_from.st_dev &&
+ 						stbuf_exc.st_ino == stbuf_from.st_ino)
+ 					{
+ 						psql_error("COPY FROM and EXCEPTIONS TO cannot point to the same file\n");
+ 						free_copy_options(options);
+ 						return false;
+ 					}
+ 				}
  				copystream = fopen(options->file, PG_BINARY_R);
+ 			}
  		}
  		else if (!options->psql_inout)
  			copystream = pset.cur_cmd_source;
  		else
  			copystream = stdin;
+ 
+ 		if (options->exceptions)
+ 		{
+ 			if (options->exc_file)
+ 			{
+ 				excstream = fopen(options->exc_file, PG_BINARY_W);
+ 			}
+ 			else if (!options->exc_psql_inout)
+ 				excstream = pset.queryFout;
+ 			else
+ 				excstream = stdout;
+ 		}
  	}
  	else
  	{
*************** do_copy(const char *args)
*** 332,337 ****
--- 421,438 ----
  		else
  			psql_error("%s: %s\n",
  					   options->file, strerror(errno));
+ 		if (options->exc_file && excstream)
+ 			fclose(excstream);
+ 		free_copy_options(options);
+ 		return false;
+ 	}
+ 
+ 	if (options->exceptions && !excstream)
+ 	{
+ 		psql_error("%s: %s\n",
+ 				   options->exc_file, strerror(errno));
+ 		if (options->file)
+ 			fclose(copystream);
  		free_copy_options(options);
  		return false;
  	}
*************** do_copy(const char *args)
*** 353,358 ****
--- 454,461 ----
  		if (result < 0 || S_ISDIR(st.st_mode))
  		{
  			fclose(copystream);
+ 			if (options->exc_file && excstream)
+ 				fclose(excstream);
  			free_copy_options(options);
  			return false;
  		}
*************** do_copy(const char *args)
*** 366,378 ****
--- 469,485 ----
  		appendPQExpBufferStr(&query, " FROM STDIN ");
  	else
  		appendPQExpBufferStr(&query, " TO STDOUT ");
+ 	if (options->exceptions)
+ 		appendPQExpBufferStr(&query, " EXCEPTIONS TO STDOUT ");
  	if (options->after_tofrom)
  		appendPQExpBufferStr(&query, options->after_tofrom);
  
  	/* run it like a user command, but with copystream as data source/sink */
  	pset.copyStream = copystream;
+ 	pset.copyExcStream = excstream;
  	success = SendQuery(query.data);
  	pset.copyStream = NULL;
+ 	pset.copyExcStream = NULL;
  	termPQExpBuffer(&query);
  
  	if (options->file != NULL)
*************** do_copy(const char *args)
*** 410,415 ****
--- 517,530 ----
  			}
  		}
  	}
+ 	if (options->exc_file != NULL)
+ 	{
+ 		if (fclose(excstream) != 0)
+ 		{
+ 			psql_error("%s: %s\n", options->exc_file, strerror(errno));
+ 			success = false;
+ 		}
+ 	}
  	free_copy_options(options);
  	return success;
  }
diff --git a/src/bin/psql/settings.h b/src/bin/psql/settings.h
new file mode 100644
index ef24a4e..36c9300
*** a/src/bin/psql/settings.h
--- b/src/bin/psql/settings.h
*************** typedef struct _psqlSettings
*** 72,77 ****
--- 72,78 ----
  	bool		queryFoutPipe;	/* queryFout is from a popen() */
  
  	FILE	   *copyStream;		/* Stream to read/write for \copy command */
+ 	FILE	   *copyExcStream;	/* Stream to read exceptions for \copy command */
  
  	printQueryOpt popt;
  
diff --git a/src/bin/psql/startup.c b/src/bin/psql/startup.c
new file mode 100644
index 11a159a..f1f65df
*** a/src/bin/psql/startup.c
--- b/src/bin/psql/startup.c
*************** main(int argc, char *argv[])
*** 122,127 ****
--- 122,128 ----
  	pset.queryFout = stdout;
  	pset.queryFoutPipe = false;
  	pset.copyStream = NULL;
+ 	pset.copyExcStream = NULL;
  	pset.cur_cmd_source = stdin;
  	pset.cur_cmd_interactive = false;
  
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
new file mode 100644
index 82c926d..361cd80
*** a/src/bin/psql/tab-complete.c
--- b/src/bin/psql/tab-complete.c
*************** psql_completion(const char *text, int st
*** 2169,2175 ****
  		completion_charp = "";
  		matches = completion_matches(text, complete_from_files);
  	}
- 
  	/* Handle COPY|BINARY <sth> FROM|TO filename */
  	else if ((pg_strcasecmp(prev4_wd, "COPY") == 0 ||
  			  pg_strcasecmp(prev4_wd, "\\copy") == 0 ||
--- 2169,2174 ----
*************** psql_completion(const char *text, int st
*** 2178,2187 ****
  			  pg_strcasecmp(prev2_wd, "TO") == 0))
  	{
  		static const char *const list_COPY[] =
! 		{"BINARY", "OIDS", "DELIMITER", "NULL", "CSV", "ENCODING", NULL};
! 
  		COMPLETE_WITH_LIST(list_COPY);
  	}
  
  	/* Handle COPY|BINARY <sth> FROM|TO filename CSV */
  	else if (pg_strcasecmp(prev_wd, "CSV") == 0 &&
--- 2177,2193 ----
  			  pg_strcasecmp(prev2_wd, "TO") == 0))
  	{
  		static const char *const list_COPY[] =
! 		{"BINARY", "OIDS", "DELIMITER", "NULL", "CSV", "ENCODING", "EXCEPTIONS TO", NULL};
  		COMPLETE_WITH_LIST(list_COPY);
  	}
+ 	/* If we have [COPY...] FROM <sth> EXCEPTIONS TO, complete with filename */
+ 	else if (pg_strcasecmp(prev4_wd, "FROM") == 0 &&
+ 			 pg_strcasecmp(prev2_wd, "EXCEPTIONS") == 0 &&
+ 			 pg_strcasecmp(prev_wd, "TO") == 0)
+ 	{
+ 		completion_charp = "";
+ 		matches = completion_matches(text, complete_from_files);
+ 	}
  
  	/* Handle COPY|BINARY <sth> FROM|TO filename CSV */
  	else if (pg_strcasecmp(prev_wd, "CSV") == 0 &&
diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h
new file mode 100644
index ba0f1b3..7137dfc
*** a/src/include/commands/copy.h
--- b/src/include/commands/copy.h
*************** extern Oid DoCopy(const CopyStmt *stmt,
*** 26,32 ****
  
  extern void ProcessCopyOptions(CopyState cstate, bool is_from, List *options);
  extern CopyState BeginCopyFrom(Relation rel, const char *filename,
! 			  bool is_program, List *attnamelist, List *options);
  extern void EndCopyFrom(CopyState cstate);
  extern bool NextCopyFrom(CopyState cstate, ExprContext *econtext,
  			 Datum *values, bool *nulls, Oid *tupleOid);
--- 26,33 ----
  
  extern void ProcessCopyOptions(CopyState cstate, bool is_from, List *options);
  extern CopyState BeginCopyFrom(Relation rel, const char *filename,
! 			  bool is_program, const char *exc_filename,
! 			  List *attnamelist, List *options);
  extern void EndCopyFrom(CopyState cstate);
  extern bool NextCopyFrom(CopyState cstate, ExprContext *econtext,
  			 Datum *values, bool *nulls, Oid *tupleOid);
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
new file mode 100644
index 9141c30..93f50cb
*** a/src/include/nodes/parsenodes.h
--- b/src/include/nodes/parsenodes.h
*************** typedef struct CopyStmt
*** 1513,1518 ****
--- 1513,1519 ----
  	bool		is_from;		/* TO or FROM */
  	bool		is_program;		/* is 'filename' a program to popen? */
  	char	   *filename;		/* filename, or NULL for STDIN/STDOUT */
+ 	char	   *exc_filename;	/* filename for exceptions or NULL, empty string for STDOUT */
  	List	   *options;		/* List of DefElem nodes */
  } CopyStmt;
  
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
new file mode 100644
index e14dc9a..c44f02b
*** a/src/include/parser/kwlist.h
--- b/src/include/parser/kwlist.h
*************** PG_KEYWORD("enum", ENUM_P, UNRESERVED_KE
*** 143,148 ****
--- 143,149 ----
  PG_KEYWORD("escape", ESCAPE, UNRESERVED_KEYWORD)
  PG_KEYWORD("event", EVENT, UNRESERVED_KEYWORD)
  PG_KEYWORD("except", EXCEPT, RESERVED_KEYWORD)
+ PG_KEYWORD("exceptions", EXCEPTIONS, UNRESERVED_KEYWORD)
  PG_KEYWORD("exclude", EXCLUDE, UNRESERVED_KEYWORD)
  PG_KEYWORD("excluding", EXCLUDING, UNRESERVED_KEYWORD)
  PG_KEYWORD("exclusive", EXCLUSIVE, UNRESERVED_KEYWORD)
diff --git a/src/interfaces/ecpg/preproc/ecpg.addons b/src/interfaces/ecpg/preproc/ecpg.addons
new file mode 100644
index b3b36cf..0a415e6
*** a/src/interfaces/ecpg/preproc/ecpg.addons
--- b/src/interfaces/ecpg/preproc/ecpg.addons
*************** ECPG: where_or_current_clauseWHERECURREN
*** 192,198 ****
  		char *cursor_marker = $4[0] == ':' ? mm_strdup("$0") : $4;
  		$$ = cat_str(2,mm_strdup("where current of"), cursor_marker);
  	}
! ECPG: CopyStmtCOPYopt_binaryqualified_nameopt_column_listopt_oidscopy_fromopt_programcopy_file_namecopy_delimiteropt_withcopy_options addon
  			if (strcmp($6, "from") == 0 &&
  			   (strcmp($7, "stdin") == 0 || strcmp($7, "stdout") == 0))
  				mmerror(PARSE_ERROR, ET_WARNING, "COPY FROM STDIN is not implemented");
--- 192,198 ----
  		char *cursor_marker = $4[0] == ':' ? mm_strdup("$0") : $4;
  		$$ = cat_str(2,mm_strdup("where current of"), cursor_marker);
  	}
! ECPG: CopyStmtCOPYopt_binaryqualified_nameopt_column_listopt_oidscopy_fromopt_programcopy_file_nameopt_copy_exceptionscopy_delimiteropt_withcopy_options addon
  			if (strcmp($6, "from") == 0 &&
  			   (strcmp($7, "stdin") == 0 || strcmp($7, "stdout") == 0))
  				mmerror(PARSE_ERROR, ET_WARNING, "COPY FROM STDIN is not implemented");
-- 
2.1.0

-- 
Sent via pgsql-hackers mailing list ([email-masked])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers

---

From: Pavel Stehule <[email-masked]>
Date: Fri, 26 Dec 2014 11:41:54 +0100

>  Column |  Type   | Modifiers

this method is unsafe .. exception handlers doesn't free memory usually -
there is risk of memory leaks, source leaks

you can enforce same performance with block subtransactions - when you use
subtransaction for 1000 rows, then impact of subtransactions is minimal

when block fails, then you can use row level subtransaction - it works well
when you expect almost correct data.

Regards

Pavel

---

From: Pavel Stehule <[email-masked]>
Date: Fri, 26 Dec 2014 11:49:05 +0100

2014-12-26 11:41 GMT+01:00 Pavel Stehule <[email-masked]>:

> when block fails, then you can use row level subtransaction - it works
> well when you expect almost correct data.
>

Two years ago I wrote a extension that did it - but I have not time to
finish it and push to upstream.

Regards

Pavel

---

From: Damir Belyalov <[email-masked]>
Date: Sat, 18 Dec 2021 11:55:24 +0300

Hello.

Wrote a patch implementing COPY with ignoring errors in rows using block
subtransactions.

Syntax: COPY [table] FROM [file/stdin] WITH IGNORE_ERROS;

Examples:
CREATE TABLE check_ign_err (n int, m int, k int);
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
1 1 1
2 2 2 2
3 3 3
\.
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
SELECT * FROM check_ign_err;
 n | m | k
---+---+---
 1 | 1 | 1
 3 | 3 | 3
(2 rows)

##################################################

TRUNCATE check_ign_err;
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
1 1 1
2 2
3 3 3
\.
WARNING:  COPY check_ign_err, line 2: "2 2"
SELECT * FROM check_ign_err;
 n | m | k
---+---+---
 1 | 1 | 1
 3 | 3 | 3
(2 rows)

##################################################

TRUNCATE check_ign_err;
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
1 1 1
2 a 2
3 3 3
\.
WARNING:  COPY check_ign_err, line 2, column m: "a"
SELECT * FROM check_ign_err;
 n | m | k
---+---+---
 1 | 1 | 1
 3 | 3 | 3
(2 rows)

Regards, Damir

пт, 10 дек. 2021 г. в 21:48, Pavel Stehule <[email-masked]>:

>> you can enforce same performance with block subtransactions - when you
>> use subtransaction for 1000 rows, then impact of subtransactions is minimal
>>
From b2dbe11f103655bf845aabc3d7af3697d1441b43 Mon Sep 17 00:00:00 2001
From: Damir Belyalov <[email-masked]>
Date: Fri, 15 Oct 2021 11:55:18 +0300
Subject: [PATCH] COPY IGNORE_ERRORS option

---
 doc/src/sgml/ref/copy.sgml               |  13 +
 src/backend/commands/copy.c              |   8 +
 src/backend/commands/copyfrom.c          |  72 ++++-
 src/backend/commands/copyfromparse.c     |  13 +-
 src/backend/parser/gram.y                |   8 +-
 src/backend/utils/.gitignore             |   1 -
 src/backend/utils/errcodes.h             | 354 +++++++++++++++++++++++
 src/bin/psql/tab-complete.c              |   3 +-
 src/include/commands/copy.h              |   1 +
 src/include/commands/copyfrom_internal.h |   1 +
 src/include/parser/kwlist.h              |   1 +
 src/test/regress/expected/copy2.out      |  32 ++
 src/test/regress/sql/copy2.sql           |  26 ++
 13 files changed, 525 insertions(+), 8 deletions(-)
 create mode 100644 src/backend/utils/errcodes.h

diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml
index 4624c8f4c9..5ca8ff876d 100644
--- a/doc/src/sgml/ref/copy.sgml
+++ b/doc/src/sgml/ref/copy.sgml
@@ -34,6 +34,7 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
 
     FORMAT <replaceable class="parameter">format_name</replaceable>
     FREEZE [ <replaceable class="parameter">boolean</replaceable> ]
+    IGNORE_ERRORS [ <replaceable class="parameter">boolean</replaceable> ]
     DELIMITER '<replaceable class="parameter">delimiter_character</replaceable>'
     NULL '<replaceable class="parameter">null_string</replaceable>'
     HEADER [ <replaceable class="parameter">boolean</replaceable> ]
@@ -233,6 +234,18 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><literal>IGNORE_ERRORS</literal></term>
+    <listitem>
+     <para>
+      Drop rows that contain malformed data while copying. That is rows
+      containing syntax errors in data, rows with too many or too few columns,
+      rows that result in constraint violations, rows containing columns where
+      the data type's input function raises an error.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>DELIMITER</literal></term>
     <listitem>
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 53f4853141..9ddc8c3d96 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -337,6 +337,7 @@ ProcessCopyOptions(ParseState *pstate,
 {
 	bool		format_specified = false;
 	bool		freeze_specified = false;
+	bool		ignore_errors_specified = false;
 	bool		header_specified = false;
 	ListCell   *option;
 
@@ -377,6 +378,13 @@ ProcessCopyOptions(ParseState *pstate,
 			freeze_specified = true;
 			opts_out->freeze = defGetBoolean(defel);
 		}
+		else if (strcmp(defel->defname, "ignore_errors") == 0)
+		{
+			if (ignore_errors_specified)
+				errorConflictingDefElem(defel, pstate);
+			ignore_errors_specified = true;
+			opts_out->ignore_errors = defGetBoolean(defel);
+		}
 		else if (strcmp(defel->defname, "delimiter") == 0)
 		{
 			if (opts_out->delim)
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index f366a818a1..d416222faf 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -852,9 +852,75 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);
 
-		/* Directly store the values/nulls array in the slot */
-		if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
-			break;
+		/*
+		 * If option IGNORE_ERRORS is enabled, COPY skip rows with errors.
+		 * NextCopyFrom() directly store the values/nulls array in the slot.
+		 */
+		if (cstate->opts.ignore_errors)
+		{
+			bool break_for = false;
+			bool skip_tuple_ignore_errors = false;
+			MemoryContext ccxt = CurrentMemoryContext;
+			ResourceOwner oldowner = CurrentResourceOwner;
+
+			BeginInternalSubTransaction(NULL);
+			MemoryContextSwitchTo(ccxt);
+
+			PG_TRY();
+			{
+				if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+				{
+					// can't do break in PG_TRY
+					break_for = true;
+				}
+
+				ReleaseCurrentSubTransaction();
+				MemoryContextSwitchTo(ccxt);
+				CurrentResourceOwner = oldowner;
+			}
+			PG_CATCH();
+			{
+				MemoryContext ecxt = MemoryContextSwitchTo(ccxt);
+				ErrorData *errdata = CopyErrorData();
+
+				switch (errdata->sqlerrcode)
+				{
+					case ERRCODE_BAD_COPY_FILE_FORMAT:
+					case ERRCODE_INVALID_TEXT_REPRESENTATION:
+						skip_tuple_ignore_errors = true;
+						elog(WARNING, errdata->context);
+
+						RollbackAndReleaseCurrentSubTransaction();
+						MemoryContextSwitchTo(ccxt);
+						CurrentResourceOwner = oldowner;
+
+						ExecClearTuple(myslot);
+						MemSet(myslot->tts_values, 0, cstate->attr_count * sizeof(Datum));
+						MemSet(myslot->tts_isnull, true, cstate->attr_count * sizeof(bool));
+
+						break;
+					default:
+						MemoryContextSwitchTo(ecxt);
+						PG_RE_THROW();
+				}
+
+				FlushErrorState();
+				FreeErrorData(errdata);
+				errdata = NULL;
+			}
+			PG_END_TRY();
+
+			if (break_for)
+				break;
+
+			if (skip_tuple_ignore_errors)
+				continue;
+		}
+		else
+		{
+			if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+				break;
+		}
 
 		ExecStoreVirtualTuple(myslot);
 
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index aac10165ec..702b70861d 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -817,6 +817,7 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 	tupDesc = RelationGetDescr(cstate->rel);
 	num_phys_attrs = tupDesc->natts;
 	attr_count = list_length(cstate->attnumlist);
+	cstate->attr_count = attr_count;
 
 	/* Initialize all values for row to NULL */
 	MemSet(values, 0, num_phys_attrs * sizeof(Datum));
@@ -893,6 +894,7 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 										  string,
 										  typioparams[m],
 										  att->atttypmod);
+
 			if (string != NULL)
 				nulls[m] = false;
 			cstate->cur_attname = NULL;
@@ -1446,10 +1448,17 @@ CopyReadAttributesText(CopyFromState cstate)
 	if (cstate->max_fields <= 0)
 	{
 		if (cstate->line_buf.len != 0)
-			ereport(ERROR,
-					(errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
+		{
+			if (cstate->opts.ignore_errors)
+				ereport(ERROR,
+					(errcode(ERRCODE_FOR_IGNORE_ERRORS),
 					 errmsg("extra data after last expected column")));
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_BAD_COPY_FILE_FORMAT),
+						errmsg("extra data after last expected column")));
 		return 0;
+		}
 	}
 
 	resetStringInfo(&cstate->attribute_buf);
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 86ce33bd97..5f6863e5e1 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -674,7 +674,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 
 	HANDLER HAVING HEADER_P HOLD HOUR_P
 
-	IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
+	IDENTITY_P IF_P IGNORE_ERRORS ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
 	INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
 	INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
 	INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -3160,6 +3160,10 @@ copy_opt_item:
 				{
 					$$ = makeDefElem("freeze", (Node *)makeInteger(true), @1);
 				}
+			| IGNORE_ERRORS
+				{
+					$$ = makeDefElem("ignore_errors", (Node *)makeInteger(true), @1);
+				}
 			| DELIMITER opt_as Sconst
 				{
 					$$ = makeDefElem("delimiter", (Node *)makeString($3), @1);
@@ -15659,6 +15663,7 @@ unreserved_keyword:
 			| HOUR_P
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| IMMEDIATE
 			| IMMUTABLE
 			| IMPLICIT_P
@@ -16209,6 +16214,7 @@ bare_label_keyword:
 			| HOLD
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| ILIKE
 			| IMMEDIATE
 			| IMMUTABLE
diff --git a/src/backend/utils/.gitignore b/src/backend/utils/.gitignore
index 0685556959..3c679d07ff 100644
--- a/src/backend/utils/.gitignore
+++ b/src/backend/utils/.gitignore
@@ -3,4 +3,3 @@
 /fmgrprotos.h
 /fmgr-stamp
 /probes.h
-/errcodes.h
diff --git a/src/backend/utils/errcodes.h b/src/backend/utils/errcodes.h
new file mode 100644
index 0000000000..b187599bdd
--- /dev/null
+++ b/src/backend/utils/errcodes.h
@@ -0,0 +1,354 @@
+/* autogenerated from src/backend/utils/errcodes.txt, do not edit */
+/* there is deliberately not an #ifndef ERRCODES_H here */
+
+/* Class 00 - Successful Completion */
+#define ERRCODE_SUCCESSFUL_COMPLETION MAKE_SQLSTATE('0','0','0','0','0')
+
+/* Class 01 - Warning */
+#define ERRCODE_WARNING MAKE_SQLSTATE('0','1','0','0','0')
+#define ERRCODE_WARNING_DYNAMIC_RESULT_SETS_RETURNED MAKE_SQLSTATE('0','1','0','0','C')
+#define ERRCODE_WARNING_IMPLICIT_ZERO_BIT_PADDING MAKE_SQLSTATE('0','1','0','0','8')
+#define ERRCODE_WARNING_NULL_VALUE_ELIMINATED_IN_SET_FUNCTION MAKE_SQLSTATE('0','1','0','0','3')
+#define ERRCODE_WARNING_PRIVILEGE_NOT_GRANTED MAKE_SQLSTATE('0','1','0','0','7')
+#define ERRCODE_WARNING_PRIVILEGE_NOT_REVOKED MAKE_SQLSTATE('0','1','0','0','6')
+#define ERRCODE_WARNING_STRING_DATA_RIGHT_TRUNCATION MAKE_SQLSTATE('0','1','0','0','4')
+#define ERRCODE_WARNING_DEPRECATED_FEATURE MAKE_SQLSTATE('0','1','P','0','1')
+
+/* Class 02 - No Data (this is also a warning class per the SQL standard) */
+#define ERRCODE_NO_DATA MAKE_SQLSTATE('0','2','0','0','0')
+#define ERRCODE_NO_ADDITIONAL_DYNAMIC_RESULT_SETS_RETURNED MAKE_SQLSTATE('0','2','0','0','1')
+
+/* Class 03 - SQL Statement Not Yet Complete */
+#define ERRCODE_SQL_STATEMENT_NOT_YET_COMPLETE MAKE_SQLSTATE('0','3','0','0','0')
+
+/* Class 08 - Connection Exception */
+#define ERRCODE_CONNECTION_EXCEPTION MAKE_SQLSTATE('0','8','0','0','0')
+#define ERRCODE_CONNECTION_DOES_NOT_EXIST MAKE_SQLSTATE('0','8','0','0','3')
+#define ERRCODE_CONNECTION_FAILURE MAKE_SQLSTATE('0','8','0','0','6')
+#define ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION MAKE_SQLSTATE('0','8','0','0','1')
+#define ERRCODE_SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION MAKE_SQLSTATE('0','8','0','0','4')
+#define ERRCODE_TRANSACTION_RESOLUTION_UNKNOWN MAKE_SQLSTATE('0','8','0','0','7')
+#define ERRCODE_PROTOCOL_VIOLATION MAKE_SQLSTATE('0','8','P','0','1')
+
+/* Class 09 - Triggered Action Exception */
+#define ERRCODE_TRIGGERED_ACTION_EXCEPTION MAKE_SQLSTATE('0','9','0','0','0')
+
+/* Class 0A - Feature Not Supported */
+#define ERRCODE_FEATURE_NOT_SUPPORTED MAKE_SQLSTATE('0','A','0','0','0')
+
+/* Class 0B - Invalid Transaction Initiation */
+#define ERRCODE_INVALID_TRANSACTION_INITIATION MAKE_SQLSTATE('0','B','0','0','0')
+
+/* Class 0F - Locator Exception */
+#define ERRCODE_LOCATOR_EXCEPTION MAKE_SQLSTATE('0','F','0','0','0')
+#define ERRCODE_L_E_INVALID_SPECIFICATION MAKE_SQLSTATE('0','F','0','0','1')
+
+/* Class 0L - Invalid Grantor */
+#define ERRCODE_INVALID_GRANTOR MAKE_SQLSTATE('0','L','0','0','0')
+#define ERRCODE_INVALID_GRANT_OPERATION MAKE_SQLSTATE('0','L','P','0','1')
+
+/* Class 0P - Invalid Role Specification */
+#define ERRCODE_INVALID_ROLE_SPECIFICATION MAKE_SQLSTATE('0','P','0','0','0')
+
+/* Class 0Z - Diagnostics Exception */
+#define ERRCODE_DIAGNOSTICS_EXCEPTION MAKE_SQLSTATE('0','Z','0','0','0')
+#define ERRCODE_STACKED_DIAGNOSTICS_ACCESSED_WITHOUT_ACTIVE_HANDLER MAKE_SQLSTATE('0','Z','0','0','2')
+
+/* Class 20 - Case Not Found */
+#define ERRCODE_CASE_NOT_FOUND MAKE_SQLSTATE('2','0','0','0','0')
+
+/* Class 21 - Cardinality Violation */
+#define ERRCODE_CARDINALITY_VIOLATION MAKE_SQLSTATE('2','1','0','0','0')
+
+/* Class 22 - Data Exception */
+#define ERRCODE_DATA_EXCEPTION MAKE_SQLSTATE('2','2','0','0','0')
+#define ERRCODE_ARRAY_ELEMENT_ERROR MAKE_SQLSTATE('2','2','0','2','E')
+#define ERRCODE_ARRAY_SUBSCRIPT_ERROR MAKE_SQLSTATE('2','2','0','2','E')
+#define ERRCODE_CHARACTER_NOT_IN_REPERTOIRE MAKE_SQLSTATE('2','2','0','2','1')
+#define ERRCODE_DATETIME_FIELD_OVERFLOW MAKE_SQLSTATE('2','2','0','0','8')
+#define ERRCODE_DATETIME_VALUE_OUT_OF_RANGE MAKE_SQLSTATE('2','2','0','0','8')
+#define ERRCODE_DIVISION_BY_ZERO MAKE_SQLSTATE('2','2','0','1','2')
+#define ERRCODE_ERROR_IN_ASSIGNMENT MAKE_SQLSTATE('2','2','0','0','5')
+#define ERRCODE_ESCAPE_CHARACTER_CONFLICT MAKE_SQLSTATE('2','2','0','0','B')
+#define ERRCODE_INDICATOR_OVERFLOW MAKE_SQLSTATE('2','2','0','2','2')
+#define ERRCODE_INTERVAL_FIELD_OVERFLOW MAKE_SQLSTATE('2','2','0','1','5')
+#define ERRCODE_INVALID_ARGUMENT_FOR_LOG MAKE_SQLSTATE('2','2','0','1','E')
+#define ERRCODE_INVALID_ARGUMENT_FOR_NTILE MAKE_SQLSTATE('2','2','0','1','4')
+#define ERRCODE_INVALID_ARGUMENT_FOR_NTH_VALUE MAKE_SQLSTATE('2','2','0','1','6')
+#define ERRCODE_INVALID_ARGUMENT_FOR_POWER_FUNCTION MAKE_SQLSTATE('2','2','0','1','F')
+#define ERRCODE_INVALID_ARGUMENT_FOR_WIDTH_BUCKET_FUNCTION MAKE_SQLSTATE('2','2','0','1','G')
+#define ERRCODE_INVALID_CHARACTER_VALUE_FOR_CAST MAKE_SQLSTATE('2','2','0','1','8')
+#define ERRCODE_INVALID_DATETIME_FORMAT MAKE_SQLSTATE('2','2','0','0','7')
+#define ERRCODE_INVALID_ESCAPE_CHARACTER MAKE_SQLSTATE('2','2','0','1','9')
+#define ERRCODE_INVALID_ESCAPE_OCTET MAKE_SQLSTATE('2','2','0','0','D')
+#define ERRCODE_INVALID_ESCAPE_SEQUENCE MAKE_SQLSTATE('2','2','0','2','5')
+#define ERRCODE_NONSTANDARD_USE_OF_ESCAPE_CHARACTER MAKE_SQLSTATE('2','2','P','0','6')
+#define ERRCODE_INVALID_INDICATOR_PARAMETER_VALUE MAKE_SQLSTATE('2','2','0','1','0')
+#define ERRCODE_INVALID_PARAMETER_VALUE MAKE_SQLSTATE('2','2','0','2','3')
+#define ERRCODE_INVALID_PRECEDING_OR_FOLLOWING_SIZE MAKE_SQLSTATE('2','2','0','1','3')
+#define ERRCODE_INVALID_REGULAR_EXPRESSION MAKE_SQLSTATE('2','2','0','1','B')
+#define ERRCODE_INVALID_ROW_COUNT_IN_LIMIT_CLAUSE MAKE_SQLSTATE('2','2','0','1','W')
+#define ERRCODE_INVALID_ROW_COUNT_IN_RESULT_OFFSET_CLAUSE MAKE_SQLSTATE('2','2','0','1','X')
+#define ERRCODE_INVALID_TABLESAMPLE_ARGUMENT MAKE_SQLSTATE('2','2','0','2','H')
+#define ERRCODE_INVALID_TABLESAMPLE_REPEAT MAKE_SQLSTATE('2','2','0','2','G')
+#define ERRCODE_INVALID_TIME_ZONE_DISPLACEMENT_VALUE MAKE_SQLSTATE('2','2','0','0','9')
+#define ERRCODE_INVALID_USE_OF_ESCAPE_CHARACTER MAKE_SQLSTATE('2','2','0','0','C')
+#define ERRCODE_MOST_SPECIFIC_TYPE_MISMATCH MAKE_SQLSTATE('2','2','0','0','G')
+#define ERRCODE_NULL_VALUE_NOT_ALLOWED MAKE_SQLSTATE('2','2','0','0','4')
+#define ERRCODE_NULL_VALUE_NO_INDICATOR_PARAMETER MAKE_SQLSTATE('2','2','0','0','2')
+#define ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE MAKE_SQLSTATE('2','2','0','0','3')
+#define ERRCODE_SEQUENCE_GENERATOR_LIMIT_EXCEEDED MAKE_SQLSTATE('2','2','0','0','H')
+#define ERRCODE_STRING_DATA_LENGTH_MISMATCH MAKE_SQLSTATE('2','2','0','2','6')
+#define ERRCODE_STRING_DATA_RIGHT_TRUNCATION MAKE_SQLSTATE('2','2','0','0','1')
+#define ERRCODE_SUBSTRING_ERROR MAKE_SQLSTATE('2','2','0','1','1')
+#define ERRCODE_TRIM_ERROR MAKE_SQLSTATE('2','2','0','2','7')
+#define ERRCODE_UNTERMINATED_C_STRING MAKE_SQLSTATE('2','2','0','2','4')
+#define ERRCODE_ZERO_LENGTH_CHARACTER_STRING MAKE_SQLSTATE('2','2','0','0','F')
+#define ERRCODE_FLOATING_POINT_EXCEPTION MAKE_SQLSTATE('2','2','P','0','1')
+#define ERRCODE_INVALID_TEXT_REPRESENTATION MAKE_SQLSTATE('2','2','P','0','2')
+#define ERRCODE_INVALID_BINARY_REPRESENTATION MAKE_SQLSTATE('2','2','P','0','3')
+#define ERRCODE_BAD_COPY_FILE_FORMAT MAKE_SQLSTATE('2','2','P','0','4')
+#define ERRCODE_UNTRANSLATABLE_CHARACTER MAKE_SQLSTATE('2','2','P','0','5')
+#define ERRCODE_NOT_AN_XML_DOCUMENT MAKE_SQLSTATE('2','2','0','0','L')
+#define ERRCODE_INVALID_XML_DOCUMENT MAKE_SQLSTATE('2','2','0','0','M')
+#define ERRCODE_INVALID_XML_CONTENT MAKE_SQLSTATE('2','2','0','0','N')
+#define ERRCODE_INVALID_XML_COMMENT MAKE_SQLSTATE('2','2','0','0','S')
+#define ERRCODE_INVALID_XML_PROCESSING_INSTRUCTION MAKE_SQLSTATE('2','2','0','0','T')
+#define ERRCODE_DUPLICATE_JSON_OBJECT_KEY_VALUE MAKE_SQLSTATE('2','2','0','3','0')
+#define ERRCODE_INVALID_ARGUMENT_FOR_SQL_JSON_DATETIME_FUNCTION MAKE_SQLSTATE('2','2','0','3','1')
+#define ERRCODE_INVALID_JSON_TEXT MAKE_SQLSTATE('2','2','0','3',   '2')
+#define ERRCODE_INVALID_SQL_JSON_SUBSCRIPT MAKE_SQLSTATE('2','2','0','3','3')
+#define ERRCODE_MORE_THAN_ONE_SQL_JSON_ITEM MAKE_SQLSTATE('2','2','0','3','4')
+#define ERRCODE_NO_SQL_JSON_ITEM MAKE_SQLSTATE('2','2','0','3','5')
+#define ERRCODE_NON_NUMERIC_SQL_JSON_ITEM MAKE_SQLSTATE('2','2','0','3','6')
+#define ERRCODE_NON_UNIQUE_KEYS_IN_A_JSON_OBJECT MAKE_SQLSTATE('2','2','0','3','7')
+#define ERRCODE_SINGLETON_SQL_JSON_ITEM_REQUIRED MAKE_SQLSTATE('2','2','0','3','8')
+#define ERRCODE_SQL_JSON_ARRAY_NOT_FOUND MAKE_SQLSTATE('2','2','0','3','9')
+#define ERRCODE_SQL_JSON_MEMBER_NOT_FOUND MAKE_SQLSTATE('2','2','0','3','A')
+#define ERRCODE_SQL_JSON_NUMBER_NOT_FOUND MAKE_SQLSTATE('2','2','0','3','B')
+#define ERRCODE_SQL_JSON_OBJECT_NOT_FOUND MAKE_SQLSTATE('2','2','0','3','C')
+#define ERRCODE_TOO_MANY_JSON_ARRAY_ELEMENTS MAKE_SQLSTATE('2','2','0','3','D')
+#define ERRCODE_TOO_MANY_JSON_OBJECT_MEMBERS MAKE_SQLSTATE('2','2','0','3','E')
+#define ERRCODE_SQL_JSON_SCALAR_REQUIRED MAKE_SQLSTATE('2','2','0','3','F')
+#define ERRCODE_FOR_IGNORE_ERRORS MAKE_SQLSTATE('2','2','0','4','0')
+
+/* Class 23 - Integrity Constraint Violation */
+#define ERRCODE_INTEGRITY_CONSTRAINT_VIOLATION MAKE_SQLSTATE('2','3','0','0','0')
+#define ERRCODE_RESTRICT_VIOLATION MAKE_SQLSTATE('2','3','0','0','1')
+#define ERRCODE_NOT_NULL_VIOLATION MAKE_SQLSTATE('2','3','5','0','2')
+#define ERRCODE_FOREIGN_KEY_VIOLATION MAKE_SQLSTATE('2','3','5','0','3')
+#define ERRCODE_UNIQUE_VIOLATION MAKE_SQLSTATE('2','3','5','0','5')
+#define ERRCODE_CHECK_VIOLATION MAKE_SQLSTATE('2','3','5','1','4')
+#define ERRCODE_EXCLUSION_VIOLATION MAKE_SQLSTATE('2','3','P','0','1')
+
+/* Class 24 - Invalid Cursor State */
+#define ERRCODE_INVALID_CURSOR_STATE MAKE_SQLSTATE('2','4','0','0','0')
+
+/* Class 25 - Invalid Transaction State */
+#define ERRCODE_INVALID_TRANSACTION_STATE MAKE_SQLSTATE('2','5','0','0','0')
+#define ERRCODE_ACTIVE_SQL_TRANSACTION MAKE_SQLSTATE('2','5','0','0','1')
+#define ERRCODE_BRANCH_TRANSACTION_ALREADY_ACTIVE MAKE_SQLSTATE('2','5','0','0','2')
+#define ERRCODE_HELD_CURSOR_REQUIRES_SAME_ISOLATION_LEVEL MAKE_SQLSTATE('2','5','0','0','8')
+#define ERRCODE_INAPPROPRIATE_ACCESS_MODE_FOR_BRANCH_TRANSACTION MAKE_SQLSTATE('2','5','0','0','3')
+#define ERRCODE_INAPPROPRIATE_ISOLATION_LEVEL_FOR_BRANCH_TRANSACTION MAKE_SQLSTATE('2','5','0','0','4')
+#define ERRCODE_NO_ACTIVE_SQL_TRANSACTION_FOR_BRANCH_TRANSACTION MAKE_SQLSTATE('2','5','0','0','5')
+#define ERRCODE_READ_ONLY_SQL_TRANSACTION MAKE_SQLSTATE('2','5','0','0','6')
+#define ERRCODE_SCHEMA_AND_DATA_STATEMENT_MIXING_NOT_SUPPORTED MAKE_SQLSTATE('2','5','0','0','7')
+#define ERRCODE_NO_ACTIVE_SQL_TRANSACTION MAKE_SQLSTATE('2','5','P','0','1')
+#define ERRCODE_IN_FAILED_SQL_TRANSACTION MAKE_SQLSTATE('2','5','P','0','2')
+#define ERRCODE_IDLE_IN_TRANSACTION_SESSION_TIMEOUT MAKE_SQLSTATE('2','5','P','0','3')
+
+/* Class 26 - Invalid SQL Statement Name */
+#define ERRCODE_INVALID_SQL_STATEMENT_NAME MAKE_SQLSTATE('2','6','0','0','0')
+
+/* Class 27 - Triggered Data Change Violation */
+#define ERRCODE_TRIGGERED_DATA_CHANGE_VIOLATION MAKE_SQLSTATE('2','7','0','0','0')
+
+/* Class 28 - Invalid Authorization Specification */
+#define ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION MAKE_SQLSTATE('2','8','0','0','0')
+#define ERRCODE_INVALID_PASSWORD MAKE_SQLSTATE('2','8','P','0','1')
+
+/* Class 2B - Dependent Privilege Descriptors Still Exist */
+#define ERRCODE_DEPENDENT_PRIVILEGE_DESCRIPTORS_STILL_EXIST MAKE_SQLSTATE('2','B','0','0','0')
+#define ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST MAKE_SQLSTATE('2','B','P','0','1')
+
+/* Class 2D - Invalid Transaction Termination */
+#define ERRCODE_INVALID_TRANSACTION_TERMINATION MAKE_SQLSTATE('2','D','0','0','0')
+
+/* Class 2F - SQL Routine Exception */
+#define ERRCODE_SQL_ROUTINE_EXCEPTION MAKE_SQLSTATE('2','F','0','0','0')
+#define ERRCODE_S_R_E_FUNCTION_EXECUTED_NO_RETURN_STATEMENT MAKE_SQLSTATE('2','F','0','0','5')
+#define ERRCODE_S_R_E_MODIFYING_SQL_DATA_NOT_PERMITTED MAKE_SQLSTATE('2','F','0','0','2')
+#define ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED MAKE_SQLSTATE('2','F','0','0','3')
+#define ERRCODE_S_R_E_READING_SQL_DATA_NOT_PERMITTED MAKE_SQLSTATE('2','F','0','0','4')
+
+/* Class 34 - Invalid Cursor Name */
+#define ERRCODE_INVALID_CURSOR_NAME MAKE_SQLSTATE('3','4','0','0','0')
+
+/* Class 38 - External Routine Exception */
+#define ERRCODE_EXTERNAL_ROUTINE_EXCEPTION MAKE_SQLSTATE('3','8','0','0','0')
+#define ERRCODE_E_R_E_CONTAINING_SQL_NOT_PERMITTED MAKE_SQLSTATE('3','8','0','0','1')
+#define ERRCODE_E_R_E_MODIFYING_SQL_DATA_NOT_PERMITTED MAKE_SQLSTATE('3','8','0','0','2')
+#define ERRCODE_E_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED MAKE_SQLSTATE('3','8','0','0','3')
+#define ERRCODE_E_R_E_READING_SQL_DATA_NOT_PERMITTED MAKE_SQLSTATE('3','8','0','0','4')
+
+/* Class 39 - External Routine Invocation Exception */
+#define ERRCODE_EXTERNAL_ROUTINE_INVOCATION_EXCEPTION MAKE_SQLSTATE('3','9','0','0','0')
+#define ERRCODE_E_R_I_E_INVALID_SQLSTATE_RETURNED MAKE_SQLSTATE('3','9','0','0','1')
+#define ERRCODE_E_R_I_E_NULL_VALUE_NOT_ALLOWED MAKE_SQLSTATE('3','9','0','0','4')
+#define ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED MAKE_SQLSTATE('3','9','P','0','1')
+#define ERRCODE_E_R_I_E_SRF_PROTOCOL_VIOLATED MAKE_SQLSTATE('3','9','P','0','2')
+#define ERRCODE_E_R_I_E_EVENT_TRIGGER_PROTOCOL_VIOLATED MAKE_SQLSTATE('3','9','P','0','3')
+
+/* Class 3B - Savepoint Exception */
+#define ERRCODE_SAVEPOINT_EXCEPTION MAKE_SQLSTATE('3','B','0','0','0')
+#define ERRCODE_S_E_INVALID_SPECIFICATION MAKE_SQLSTATE('3','B','0','0','1')
+
+/* Class 3D - Invalid Catalog Name */
+#define ERRCODE_INVALID_CATALOG_NAME MAKE_SQLSTATE('3','D','0','0','0')
+
+/* Class 3F - Invalid Schema Name */
+#define ERRCODE_INVALID_SCHEMA_NAME MAKE_SQLSTATE('3','F','0','0','0')
+
+/* Class 40 - Transaction Rollback */
+#define ERRCODE_TRANSACTION_ROLLBACK MAKE_SQLSTATE('4','0','0','0','0')
+#define ERRCODE_T_R_INTEGRITY_CONSTRAINT_VIOLATION MAKE_SQLSTATE('4','0','0','0','2')
+#define ERRCODE_T_R_SERIALIZATION_FAILURE MAKE_SQLSTATE('4','0','0','0','1')
+#define ERRCODE_T_R_STATEMENT_COMPLETION_UNKNOWN MAKE_SQLSTATE('4','0','0','0','3')
+#define ERRCODE_T_R_DEADLOCK_DETECTED MAKE_SQLSTATE('4','0','P','0','1')
+
+/* Class 42 - Syntax Error or Access Rule Violation */
+#define ERRCODE_SYNTAX_ERROR_OR_ACCESS_RULE_VIOLATION MAKE_SQLSTATE('4','2','0','0','0')
+#define ERRCODE_SYNTAX_ERROR MAKE_SQLSTATE('4','2','6','0','1')
+#define ERRCODE_INSUFFICIENT_PRIVILEGE MAKE_SQLSTATE('4','2','5','0','1')
+#define ERRCODE_CANNOT_COERCE MAKE_SQLSTATE('4','2','8','4','6')
+#define ERRCODE_GROUPING_ERROR MAKE_SQLSTATE('4','2','8','0','3')
+#define ERRCODE_WINDOWING_ERROR MAKE_SQLSTATE('4','2','P','2','0')
+#define ERRCODE_INVALID_RECURSION MAKE_SQLSTATE('4','2','P','1','9')
+#define ERRCODE_INVALID_FOREIGN_KEY MAKE_SQLSTATE('4','2','8','3','0')
+#define ERRCODE_INVALID_NAME MAKE_SQLSTATE('4','2','6','0','2')
+#define ERRCODE_NAME_TOO_LONG MAKE_SQLSTATE('4','2','6','2','2')
+#define ERRCODE_RESERVED_NAME MAKE_SQLSTATE('4','2','9','3','9')
+#define ERRCODE_DATATYPE_MISMATCH MAKE_SQLSTATE('4','2','8','0','4')
+#define ERRCODE_INDETERMINATE_DATATYPE MAKE_SQLSTATE('4','2','P','1','8')
+#define ERRCODE_COLLATION_MISMATCH MAKE_SQLSTATE('4','2','P','2','1')
+#define ERRCODE_INDETERMINATE_COLLATION MAKE_SQLSTATE('4','2','P','2','2')
+#define ERRCODE_WRONG_OBJECT_TYPE MAKE_SQLSTATE('4','2','8','0','9')
+#define ERRCODE_GENERATED_ALWAYS MAKE_SQLSTATE('4','2','8','C','9')
+#define ERRCODE_UNDEFINED_COLUMN MAKE_SQLSTATE('4','2','7','0','3')
+#define ERRCODE_UNDEFINED_CURSOR MAKE_SQLSTATE('3','4','0','0','0')
+#define ERRCODE_UNDEFINED_DATABASE MAKE_SQLSTATE('3','D','0','0','0')
+#define ERRCODE_UNDEFINED_FUNCTION MAKE_SQLSTATE('4','2','8','8','3')
+#define ERRCODE_UNDEFINED_PSTATEMENT MAKE_SQLSTATE('2','6','0','0','0')
+#define ERRCODE_UNDEFINED_SCHEMA MAKE_SQLSTATE('3','F','0','0','0')
+#define ERRCODE_UNDEFINED_TABLE MAKE_SQLSTATE('4','2','P','0','1')
+#define ERRCODE_UNDEFINED_PARAMETER MAKE_SQLSTATE('4','2','P','0','2')
+#define ERRCODE_UNDEFINED_OBJECT MAKE_SQLSTATE('4','2','7','0','4')
+#define ERRCODE_DUPLICATE_COLUMN MAKE_SQLSTATE('4','2','7','0','1')
+#define ERRCODE_DUPLICATE_CURSOR MAKE_SQLSTATE('4','2','P','0','3')
+#define ERRCODE_DUPLICATE_DATABASE MAKE_SQLSTATE('4','2','P','0','4')
+#define ERRCODE_DUPLICATE_FUNCTION MAKE_SQLSTATE('4','2','7','2','3')
+#define ERRCODE_DUPLICATE_PSTATEMENT MAKE_SQLSTATE('4','2','P','0','5')
+#define ERRCODE_DUPLICATE_SCHEMA MAKE_SQLSTATE('4','2','P','0','6')
+#define ERRCODE_DUPLICATE_TABLE MAKE_SQLSTATE('4','2','P','0','7')
+#define ERRCODE_DUPLICATE_ALIAS MAKE_SQLSTATE('4','2','7','1','2')
+#define ERRCODE_DUPLICATE_OBJECT MAKE_SQLSTATE('4','2','7','1','0')
+#define ERRCODE_AMBIGUOUS_COLUMN MAKE_SQLSTATE('4','2','7','0','2')
+#define ERRCODE_AMBIGUOUS_FUNCTION MAKE_SQLSTATE('4','2','7','2','5')
+#define ERRCODE_AMBIGUOUS_PARAMETER MAKE_SQLSTATE('4','2','P','0','8')
+#define ERRCODE_AMBIGUOUS_ALIAS MAKE_SQLSTATE('4','2','P','0','9')
+#define ERRCODE_INVALID_COLUMN_REFERENCE MAKE_SQLSTATE('4','2','P','1','0')
+#define ERRCODE_INVALID_COLUMN_DEFINITION MAKE_SQLSTATE('4','2','6','1','1')
+#define ERRCODE_INVALID_CURSOR_DEFINITION MAKE_SQLSTATE('4','2','P','1','1')
+#define ERRCODE_INVALID_DATABASE_DEFINITION MAKE_SQLSTATE('4','2','P','1','2')
+#define ERRCODE_INVALID_FUNCTION_DEFINITION MAKE_SQLSTATE('4','2','P','1','3')
+#define ERRCODE_INVALID_PSTATEMENT_DEFINITION MAKE_SQLSTATE('4','2','P','1','4')
+#define ERRCODE_INVALID_SCHEMA_DEFINITION MAKE_SQLSTATE('4','2','P','1','5')
+#define ERRCODE_INVALID_TABLE_DEFINITION MAKE_SQLSTATE('4','2','P','1','6')
+#define ERRCODE_INVALID_OBJECT_DEFINITION MAKE_SQLSTATE('4','2','P','1','7')
+
+/* Class 44 - WITH CHECK OPTION Violation */
+#define ERRCODE_WITH_CHECK_OPTION_VIOLATION MAKE_SQLSTATE('4','4','0','0','0')
+
+/* Class 53 - Insufficient Resources */
+#define ERRCODE_INSUFFICIENT_RESOURCES MAKE_SQLSTATE('5','3','0','0','0')
+#define ERRCODE_DISK_FULL MAKE_SQLSTATE('5','3','1','0','0')
+#define ERRCODE_OUT_OF_MEMORY MAKE_SQLSTATE('5','3','2','0','0')
+#define ERRCODE_TOO_MANY_CONNECTIONS MAKE_SQLSTATE('5','3','3','0','0')
+#define ERRCODE_CONFIGURATION_LIMIT_EXCEEDED MAKE_SQLSTATE('5','3','4','0','0')
+
+/* Class 54 - Program Limit Exceeded */
+#define ERRCODE_PROGRAM_LIMIT_EXCEEDED MAKE_SQLSTATE('5','4','0','0','0')
+#define ERRCODE_STATEMENT_TOO_COMPLEX MAKE_SQLSTATE('5','4','0','0','1')
+#define ERRCODE_TOO_MANY_COLUMNS MAKE_SQLSTATE('5','4','0','1','1')
+#define ERRCODE_TOO_MANY_ARGUMENTS MAKE_SQLSTATE('5','4','0','2','3')
+
+/* Class 55 - Object Not In Prerequisite State */
+#define ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE MAKE_SQLSTATE('5','5','0','0','0')
+#define ERRCODE_OBJECT_IN_USE MAKE_SQLSTATE('5','5','0','0','6')
+#define ERRCODE_CANT_CHANGE_RUNTIME_PARAM MAKE_SQLSTATE('5','5','P','0','2')
+#define ERRCODE_LOCK_NOT_AVAILABLE MAKE_SQLSTATE('5','5','P','0','3')
+#define ERRCODE_UNSAFE_NEW_ENUM_VALUE_USAGE MAKE_SQLSTATE('5','5','P','0','4')
+
+/* Class 57 - Operator Intervention */
+#define ERRCODE_OPERATOR_INTERVENTION MAKE_SQLSTATE('5','7','0','0','0')
+#define ERRCODE_QUERY_CANCELED MAKE_SQLSTATE('5','7','0','1','4')
+#define ERRCODE_ADMIN_SHUTDOWN MAKE_SQLSTATE('5','7','P','0','1')
+#define ERRCODE_CRASH_SHUTDOWN MAKE_SQLSTATE('5','7','P','0','2')
+#define ERRCODE_CANNOT_CONNECT_NOW MAKE_SQLSTATE('5','7','P','0','3')
+#define ERRCODE_DATABASE_DROPPED MAKE_SQLSTATE('5','7','P','0','4')
+#define ERRCODE_IDLE_SESSION_TIMEOUT MAKE_SQLSTATE('5','7','P','0','5')
+
+/* Class 58 - System Error (errors external to PostgreSQL itself) */
+#define ERRCODE_SYSTEM_ERROR MAKE_SQLSTATE('5','8','0','0','0')
+#define ERRCODE_IO_ERROR MAKE_SQLSTATE('5','8','0','3','0')
+#define ERRCODE_UNDEFINED_FILE MAKE_SQLSTATE('5','8','P','0','1')
+#define ERRCODE_DUPLICATE_FILE MAKE_SQLSTATE('5','8','P','0','2')
+
+/* Class 72 - Snapshot Failure */
+#define ERRCODE_SNAPSHOT_TOO_OLD MAKE_SQLSTATE('7','2','0','0','0')
+
+/* Class F0 - Configuration File Error */
+#define ERRCODE_CONFIG_FILE_ERROR MAKE_SQLSTATE('F','0','0','0','0')
+#define ERRCODE_LOCK_FILE_EXISTS MAKE_SQLSTATE('F','0','0','0','1')
+
+/* Class HV - Foreign Data Wrapper Error (SQL/MED) */
+#define ERRCODE_FDW_ERROR MAKE_SQLSTATE('H','V','0','0','0')
+#define ERRCODE_FDW_COLUMN_NAME_NOT_FOUND MAKE_SQLSTATE('H','V','0','0','5')
+#define ERRCODE_FDW_DYNAMIC_PARAMETER_VALUE_NEEDED MAKE_SQLSTATE('H','V','0','0','2')
+#define ERRCODE_FDW_FUNCTION_SEQUENCE_ERROR MAKE_SQLSTATE('H','V','0','1','0')
+#define ERRCODE_FDW_INCONSISTENT_DESCRIPTOR_INFORMATION MAKE_SQLSTATE('H','V','0','2','1')
+#define ERRCODE_FDW_INVALID_ATTRIBUTE_VALUE MAKE_SQLSTATE('H','V','0','2','4')
+#define ERRCODE_FDW_INVALID_COLUMN_NAME MAKE_SQLSTATE('H','V','0','0','7')
+#define ERRCODE_FDW_INVALID_COLUMN_NUMBER MAKE_SQLSTATE('H','V','0','0','8')
+#define ERRCODE_FDW_INVALID_DATA_TYPE MAKE_SQLSTATE('H','V','0','0','4')
+#define ERRCODE_FDW_INVALID_DATA_TYPE_DESCRIPTORS MAKE_SQLSTATE('H','V','0','0','6')
+#define ERRCODE_FDW_INVALID_DESCRIPTOR_FIELD_IDENTIFIER MAKE_SQLSTATE('H','V','0','9','1')
+#define ERRCODE_FDW_INVALID_HANDLE MAKE_SQLSTATE('H','V','0','0','B')
+#define ERRCODE_FDW_INVALID_OPTION_INDEX MAKE_SQLSTATE('H','V','0','0','C')
+#define ERRCODE_FDW_INVALID_OPTION_NAME MAKE_SQLSTATE('H','V','0','0','D')
+#define ERRCODE_FDW_INVALID_STRING_LENGTH_OR_BUFFER_LENGTH MAKE_SQLSTATE('H','V','0','9','0')
+#define ERRCODE_FDW_INVALID_STRING_FORMAT MAKE_SQLSTATE('H','V','0','0','A')
+#define ERRCODE_FDW_INVALID_USE_OF_NULL_POINTER MAKE_SQLSTATE('H','V','0','0','9')
+#define ERRCODE_FDW_TOO_MANY_HANDLES MAKE_SQLSTATE('H','V','0','1','4')
+#define ERRCODE_FDW_OUT_OF_MEMORY MAKE_SQLSTATE('H','V','0','0','1')
+#define ERRCODE_FDW_NO_SCHEMAS MAKE_SQLSTATE('H','V','0','0','P')
+#define ERRCODE_FDW_OPTION_NAME_NOT_FOUND MAKE_SQLSTATE('H','V','0','0','J')
+#define ERRCODE_FDW_REPLY_HANDLE MAKE_SQLSTATE('H','V','0','0','K')
+#define ERRCODE_FDW_SCHEMA_NOT_FOUND MAKE_SQLSTATE('H','V','0','0','Q')
+#define ERRCODE_FDW_TABLE_NOT_FOUND MAKE_SQLSTATE('H','V','0','0','R')
+#define ERRCODE_FDW_UNABLE_TO_CREATE_EXECUTION MAKE_SQLSTATE('H','V','0','0','L')
+#define ERRCODE_FDW_UNABLE_TO_CREATE_REPLY MAKE_SQLSTATE('H','V','0','0','M')
+#define ERRCODE_FDW_UNABLE_TO_ESTABLISH_CONNECTION MAKE_SQLSTATE('H','V','0','0','N')
+
+/* Class P0 - PL/pgSQL Error */
+#define ERRCODE_PLPGSQL_ERROR MAKE_SQLSTATE('P','0','0','0','0')
+#define ERRCODE_RAISE_EXCEPTION MAKE_SQLSTATE('P','0','0','0','1')
+#define ERRCODE_NO_DATA_FOUND MAKE_SQLSTATE('P','0','0','0','2')
+#define ERRCODE_TOO_MANY_ROWS MAKE_SQLSTATE('P','0','0','0','3')
+#define ERRCODE_ASSERT_FAILURE MAKE_SQLSTATE('P','0','0','0','4')
+
+/* Class XX - Internal Error */
+#define ERRCODE_INTERNAL_ERROR MAKE_SQLSTATE('X','X','0','0','0')
+#define ERRCODE_DATA_CORRUPTED MAKE_SQLSTATE('X','X','0','0','1')
+#define ERRCODE_INDEX_CORRUPTED MAKE_SQLSTATE('X','X','0','0','2')
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 2f412ca3db..855cbeb401 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -2545,7 +2545,8 @@ psql_completion(const char *text, int start, int end)
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "("))
 		COMPLETE_WITH("FORMAT", "FREEZE", "DELIMITER", "NULL",
 					  "HEADER", "QUOTE", "ESCAPE", "FORCE_QUOTE",
-					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING");
+					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING",
+					  "IGNORE_ERRORS");
 
 	/* Complete COPY <sth> FROM|TO filename WITH (FORMAT */
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "(", "FORMAT"))
diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h
index 264895d278..ade74ce5a1 100644
--- a/src/include/commands/copy.h
+++ b/src/include/commands/copy.h
@@ -31,6 +31,7 @@ typedef struct CopyFormatOptions
 								 * -1 if not specified */
 	bool		binary;			/* binary format? */
 	bool		freeze;			/* freeze rows on loading? */
+	bool		ignore_errors;  /* ignore rows with errors */
 	bool		csv_mode;		/* Comma Separated Value format? */
 	bool		header_line;	/* CSV header line? */
 	char	   *null_print;		/* NULL marker string (server encoding!) */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index 4d68d9cceb..93a3b74999 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -68,6 +68,7 @@ typedef struct CopyFromStateData
 	/* parameters from the COPY command */
 	Relation	rel;			/* relation to copy from */
 	List	   *attnumlist;		/* integer list of attnums to copy */
+	int			attr_count;		/* length of attnumlist */
 	char	   *filename;		/* filename, or NULL for STDIN */
 	bool		is_program;		/* is 'filename' a program to popen? */
 	copy_data_source_cb data_source_cb; /* function for reading data */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index f836acf876..2bb3cccea9 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -196,6 +196,7 @@ PG_KEYWORD("hold", HOLD, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("hour", HOUR_P, UNRESERVED_KEYWORD, AS_LABEL)
 PG_KEYWORD("identity", IDENTITY_P, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("if", IF_P, UNRESERVED_KEYWORD, BARE_LABEL)
+PG_KEYWORD("ignore_errors", IGNORE_ERRORS, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("ilike", ILIKE, TYPE_FUNC_NAME_KEYWORD, BARE_LABEL)
 PG_KEYWORD("immediate", IMMEDIATE, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("immutable", IMMUTABLE, UNRESERVED_KEYWORD, BARE_LABEL)
diff --git a/src/test/regress/expected/copy2.out b/src/test/regress/expected/copy2.out
index 5f3685e9ef..4e04efcbba 100644
--- a/src/test/regress/expected/copy2.out
+++ b/src/test/regress/expected/copy2.out
@@ -649,6 +649,37 @@ SELECT * FROM instead_of_insert_tbl;
 (2 rows)
 
 COMMIT;
+-- test IGNORE_ERRORS option
+CREATE TABLE check_ign_err (n int, m int, k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	2	2	2"
+SELECT * FROM check_ign_err;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+ 3 | 3 | 3
+(2 rows)
+
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	2"
+SELECT * FROM check_ign_err;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+ 3 | 3 | 3
+(2 rows)
+
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2, column m: "a"
+SELECT * FROM check_ign_err;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+ 3 | 3 | 3
+(2 rows)
+
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
@@ -663,3 +694,4 @@ DROP TABLE instead_of_insert_tbl;
 DROP VIEW instead_of_insert_tbl_view;
 DROP VIEW instead_of_insert_tbl_view_2;
 DROP FUNCTION fun_instead_of_insert_tbl();
+DROP TABLE check_ign_err;
diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql
index b3c16af48e..3642c11f91 100644
--- a/src/test/regress/sql/copy2.sql
+++ b/src/test/regress/sql/copy2.sql
@@ -454,6 +454,31 @@ test1
 SELECT * FROM instead_of_insert_tbl;
 COMMIT;
 
+-- test IGNORE_ERRORS option
+CREATE TABLE check_ign_err (n int, m int, k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+1	1	1
+2	2	2	2
+3	3	3
+\.
+SELECT * FROM check_ign_err;
+
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+1	1	1
+2	2
+3	3	3
+\.
+SELECT * FROM check_ign_err;
+
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+1	1	1
+2	a	2
+3	3	3
+\.
+SELECT * FROM check_ign_err;
+
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
@@ -468,3 +493,4 @@ DROP TABLE instead_of_insert_tbl;
 DROP VIEW instead_of_insert_tbl_view;
 DROP VIEW instead_of_insert_tbl_view_2;
 DROP FUNCTION fun_instead_of_insert_tbl();
+DROP TABLE check_ign_err;
-- 
2.25.1

---

From: Pavel Stehule <[email-masked]>
Date: Sun, 19 Dec 2021 06:09:41 +0100

Hi

> Wrote a patch implementing COPY with ignoring errors in rows using block

It is great so you are working on this patch. Unfortunately, I am afraid
this simple design is not optimal. Using subtransaction for every row has
too big overhead. I think it should use subtransaction for blocks of rows
(1000 rows), and only when there is an exception, then it should replay
inserts in subtransaction per row. You should check performance overhead.

Regards

Pavel

>>> this method is unsafe .. exception handlers doesn't free memory usually
>>> - there is risk of memory leaks, source leaks
>>>

---

From: Damir Belyalov <[email-masked]>
Date: Tue, 19 Jul 2022 15:40:36 +0300

Hi!

Improved my patch by adding block subtransactions.
The block size is determined by the REPLAY_BUFFER_SIZE parameter.
I used the idea of a buffer for accumulating tuples in it.
If we read REPLAY_BUFFER_SIZE rows without errors, the subtransaction will
be committed.
If we find an error, the subtransaction will rollback and the buffer will
be replayed containing tuples.

In the patch REPLAY_BUFFER_SIZE equals 3, but it can be changed to any
other number (for example 1000).
There is an idea to create a GUC parameter for it.
Also maybe create a GUC parameter for the number of occurring WARNINGS by
rows with errors.

For CIM_MULTI and CIM_MULTI_CONDITIONAL cases the buffer is not needed.
It is needed for the CIM_SINGLE case.

Tests:

-- CIM_MULTI case
CREATE TABLE check_ign_err (n int, m int, k int);
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
SELECT * FROM check_ign_err;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
 n | m | k
---+---+---
 1 | 1 | 1
 7 | 7 | 7
(2 rows)

-- CIM_SINGLE case
-- BEFORE row trigger
CREATE TABLE trig_test(n int, m int);
CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
  BEGIN
    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
    RETURN NEW;
  END;
' LANGUAGE plpgsql;
CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
 7 | 7 | 7
(2 rows)

-- INSTEAD OF row trigger
CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
  BEGIN
    INSERT INTO check_ign_err VALUES(NEW.n, NEW.m, NEW.k);
    RETURN NEW;
  END;
' LANGUAGE plpgsql;
CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
SELECT * FROM check_ign_err;
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
SELECT * FROM check_ign_err_view;
 n | m | k
---+---+---
 1 | 1 | 1
 7 | 7 | 7
(2 rows)

-- foreign table case in postgres_fdw extension

-- volatile function in WHERE clause
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
  WHERE n = floor(random()*(1-1+1))+1; /* find values equal 1 */
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
SELECT * FROM check_ign_err;
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
(1 row)

-- CIM_MULTI_CONDITIONAL case
-- INSERT triggers for partition tables
CREATE TABLE check_ign_err (n int, m int, k int) PARTITION BY RANGE (n);
CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
  FOR VALUES FROM (1) TO (4);
CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
  FOR VALUES FROM (4) TO (8);
CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
  BEGIN
    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
    RETURN NEW;
  END;
' LANGUAGE plpgsql;
CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
WARNING:  COPY check_ign_err, line 2: "2 2 2 2"
WARNING:  COPY check_ign_err, line 3: "3 3"
WARNING:  COPY check_ign_err, line 4, column n: "a"
WARNING:  COPY check_ign_err, line 5, column m: "b"
WARNING:  COPY check_ign_err, line 6, column n: ""
SELECT * FROM check_ign_err;
1 1 1
2 2 2 2
3 3
a 4 4
5 b b

7 7 7
\.
 n | m | k
---+---+---
 1 | 1 | 1
 7 | 7 | 7
(2 rows)

Thanks for feedback.
Regards, Damir

---

From: torikoshia <[email-masked]>
Date: Mon, 15 Aug 2022 21:29:10 +0900

On 2022-07-19 21:40, Damir Belyalov wrote:
> If we read REPLAY_BUFFER_SIZE rows without errors, the subtransaction
> will be committed.
> If we find an error, the subtransaction will rollback and the buffer
> will be replayed containing tuples.

Thanks for working on this!

I tested 0002-COPY-IGNORE_ERRORS.patch and faced an unexpected behavior.

I loaded 10000 rows which contained 1 wrong row.
I expected I could see 9999 rows after COPY, but just saw 999 rows.

Since when I changed MAX_BUFFERED_TUPLES from 1000 to other values, the 
number of loaded rows also changed, I imagine MAX_BUFFERED_TUPLES might 
be giving influence of this behavior.

```sh
$ cat /tmp/test10000.dat

1   aaa
2   aaa
3   aaa
4   aaa
5   aaa
6   aaa
7   aaa
8   aaa
9   aaa
10  aaa
11  aaa
...
9994    aaa
9995    aaa
9996    aaa
9997    aaa
9998    aaa
9999    aaa
xxx aaa
```

```SQL
=# CREATE TABLE test (id int, data text);

=# COPY test FROM '/tmp/test10000.dat' WITH (IGNORE_ERRORS);
WARNING:  COPY test, line 10000, column i: "xxx"
COPY 9999

=# SELECT COUNT(*) FROM test;
  count
-------
    999
(1 row)
```

BTW I may be overlooking it, but have you submit this proposal to the 
next CommitFest?

https://commitfest.postgresql.org/39/

-- 
Regards,

--
Atsushi Torikoshi
NTT DATA CORPORATION

---

From: Damir Belyalov <[email-masked]>
Date: Mon, 15 Aug 2022 16:23:26 +0300

> Thank you for feedback.
I improved my patch recently and tested it on different sizes of
MAX_BUFFERED_TUPLES and REPLAY_BUFFER_SIZE.

> I expected I could see 9999 rows after COPY, but just saw 999 rows.
Also I implemented your case and it worked correctly.

> BTW I may be overlooking it, but have you submit this proposal to the
next CommitFest?
Good idea. Haven't done it yet.

Regards,
Damir
Postgres Professional
From fa6b99c129eb890b25f006bb7891a247c8a431a7 Mon Sep 17 00:00:00 2001
From: Damir Belyalov <[email-masked]>
Date: Fri, 15 Oct 2021 11:55:18 +0300
Subject: [PATCH] COPY_IGNORE_ERRORS without GUC with function
 safeNextCopyFrom() with struct SafeCopyFromState with refactoring

---
 doc/src/sgml/ref/copy.sgml               |  13 ++
 src/backend/commands/copy.c              |   8 ++
 src/backend/commands/copyfrom.c          | 162 ++++++++++++++++++++++-
 src/backend/parser/gram.y                |   8 +-
 src/bin/psql/tab-complete.c              |   3 +-
 src/include/commands/copy.h              |   1 +
 src/include/commands/copyfrom_internal.h |  21 +++
 src/include/parser/kwlist.h              |   1 +
 src/test/regress/expected/copy2.out      | 123 +++++++++++++++++
 src/test/regress/sql/copy2.sql           | 110 +++++++++++++++
 10 files changed, 445 insertions(+), 5 deletions(-)

index 8aae711b3b..7d20b1649e 100644
 
     HEADER [ <replaceable class="parameter">boolean</replaceable> | MATCH ]
 
+   <varlistentry>
index 3ac731803b..fead1aba46 100644
@@ -402,6 +402,7 @@ ProcessCopyOptions(ParseState *pstate,
 
@@ -442,6 +443,13 @@ ProcessCopyOptions(ParseState *pstate,
 			freeze_specified = true;
 			opts_out->freeze = defGetBoolean(defel);
 		}
+		else if (strcmp(defel->defname, "ignore_errors") == 0)
+		{
+			if (ignore_errors_specified)
+				errorConflictingDefElem(defel, pstate);
+			ignore_errors_specified = true;
+			opts_out->ignore_errors = defGetBoolean(defel);
+		}
 		else if (strcmp(defel->defname, "delimiter") == 0)
 		{
 			if (opts_out->delim)
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index a976008b3d..285c491ddd 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -106,6 +106,9 @@ static char *limit_printout_length(const char *str);
 
 static void ClosePipeFromProgram(CopyFromState cstate);
 
+static bool safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext,
+					         Datum *values, bool *nulls);
+
 /*
  * error context callback for COPY FROM
  *
@@ -521,6 +524,125 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
 	miinfo->bufferedBytes += tuplen;
 }
 
+/*
+ * Analog of NextCopyFrom() but ignore rows with errors while copying.
+ */
+static bool
+safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum *values, bool *nulls)
+{
+	SafeCopyFromState *safecstate = cstate->safecstate;
+	bool valid_row = true;
+
+	safecstate->skip_row = false;
+
+	PG_TRY();
+	{
+		if (!safecstate->replay_is_active)
+		{
+			if (safecstate->begin_subtransaction)
+			{
+				BeginInternalSubTransaction(NULL);
+				CurrentResourceOwner = safecstate->oldowner;
+
+				safecstate->begin_subtransaction = false;
+			}
+
+			if (safecstate->saved_tuples < REPLAY_BUFFER_SIZE)
+			{
+				valid_row = NextCopyFrom(cstate, econtext, values, nulls);
+				if (valid_row)
+				{
+					/* Fill replay_buffer in oldcontext*/
+					MemoryContextSwitchTo(safecstate->oldcontext);
+					safecstate->replay_buffer[safecstate->saved_tuples++] = heap_form_tuple(RelationGetDescr(cstate->rel), values, nulls);
+
+					safecstate->skip_row = true;
+				}
+				else if (!safecstate->processed_remaining_tuples)
+				{
+					ReleaseCurrentSubTransaction();
+					CurrentResourceOwner = safecstate->oldowner;
+					if (safecstate->replayed_tuples < safecstate->saved_tuples)
+					{
+						/* Prepare to replay remaining tuples if they exist */
+						safecstate->replay_is_active = true;
+						safecstate->processed_remaining_tuples = true;
+						safecstate->skip_row = true;
+						return true;
+					}
+				}
+			}
+			else
+			{
+				/* Buffer was filled, commit subtransaction and prepare to replay */
+				ReleaseCurrentSubTransaction();
+				CurrentResourceOwner = safecstate->oldowner;
+
+				safecstate->replay_is_active = true;
+				safecstate->begin_subtransaction = true;
+				safecstate->skip_row = true;
+			}
+		}
+		else
+		{
+			if (safecstate->replayed_tuples < safecstate->saved_tuples)
+			{
+				/* Replaying tuple */
+				heap_deform_tuple(safecstate->replay_buffer[safecstate->replayed_tuples++], RelationGetDescr(cstate->rel), values, nulls);
+			}
+			else
+			{
+				/* Clean up replay_buffer */
+				MemSet(safecstate->replay_buffer, 0, REPLAY_BUFFER_SIZE * sizeof(HeapTuple));
+				safecstate->saved_tuples = safecstate->replayed_tuples = 0;
+
+				safecstate->replay_is_active = false;
+				safecstate->skip_row = true;
+			}
+		}
+	}
+	PG_CATCH();
+	{
+		ErrorData *errdata;
+		MemoryContextSwitchTo(safecstate->oldcontext);
+		errdata = CopyErrorData();
+
+		switch (errdata->sqlerrcode)
+		{
+			case ERRCODE_BAD_COPY_FILE_FORMAT:
+			case ERRCODE_INVALID_TEXT_REPRESENTATION:
+				RollbackAndReleaseCurrentSubTransaction();
+				CurrentResourceOwner = safecstate->oldowner;
+
+				safecstate->errors++;
+				if (safecstate->errors <= 100)
+					ereport(WARNING,
+							(errcode(errdata->sqlerrcode),
+							errmsg("%s", errdata->context)));
+
+				safecstate->begin_subtransaction = true;
+				safecstate->skip_row = true;
+				break;
+			default:
+				PG_RE_THROW();
+		}
+
+		FlushErrorState();
+		FreeErrorData(errdata);
+		errdata = NULL;
+	}
+	PG_END_TRY();
+
+	if (!valid_row)
+	{
+		ereport(WARNING,
+				errmsg("FIND %d ERRORS", safecstate->errors));
+		return false;
+	}
+
+	return true;
+}
+
 /*
  * Copy FROM file to relation.
  */
@@ -535,6 +657,7 @@ CopyFrom(CopyFromState cstate)
 	ExprContext *econtext;
 	TupleTableSlot *singleslot = NULL;
 	MemoryContext oldcontext = CurrentMemoryContext;
+	ResourceOwner oldowner = CurrentResourceOwner;
 
 	PartitionTupleRouting *proute = NULL;
 	ErrorContextCallback errcallback;
@@ -819,6 +942,23 @@ CopyFrom(CopyFromState cstate)
 	errcallback.previous = error_context_stack;
 	error_context_stack = &errcallback;
 
+	/* Initialize safeCopyFromState for IGNORE_ERRORS option*/
+	if (cstate->opts.ignore_errors)
+	{
+		cstate->safecstate = palloc(sizeof(SafeCopyFromState));
+
+		cstate->safecstate->saved_tuples = 0;
+		cstate->safecstate->replayed_tuples = 0;
+		cstate->safecstate->errors = 0;
+		cstate->safecstate->replay_is_active = false;
+		cstate->safecstate->begin_subtransaction = true;
+		cstate->safecstate->processed_remaining_tuples = false;
+
+		cstate->safecstate->oldowner = oldowner;
+		cstate->safecstate->oldcontext = oldcontext;
+		cstate->safecstate->insertMethod = insertMethod;
+	}
+
 	for (;;)
 	{
 		TupleTableSlot *myslot;
@@ -855,9 +995,25 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);
 
-		/* Directly store the values/nulls array in the slot */
-		if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
-			break;
+		/*
+		 * If option IGNORE_ERRORS is enabled, COPY skips rows with errors.
+		 * NextCopyFrom() directly store the values/nulls array in the slot.
+		 */
+		if (cstate->safecstate)
+		{
+			bool valid_row = safeNextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull);
+
+			/* Cannot continue or break in PG_TRY in safeNextCopyFrom() */
+			if (cstate->safecstate->skip_row)
+				continue;
+			if (!valid_row)
+				break;
+		}
+		else
+		{
+			if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+				break;
+		}

diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index df5ceea910..3bb7235b34 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -800,7 +800,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);

-	IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
+	IDENTITY_P IF_P IGNORE_ERRORS ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
 	INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
 	INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
 	INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -3456,6 +3456,10 @@ copy_opt_item:
 				{
 					$$ = makeDefElem("freeze", (Node *) makeBoolean(true), @1);
 				}
+			| IGNORE_ERRORS
+				{
+					$$ = makeDefElem("ignore_errors", (Node *)makeInteger(true), @1);
+				}
 			| DELIMITER opt_as Sconst
 				{
 					$$ = makeDefElem("delimiter", (Node *) makeString($3), @1);
@@ -17814,6 +17818,7 @@ unreserved_keyword:
 			| HOUR_P
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| IMMEDIATE
 			| IMMUTABLE
 			| IMPLICIT_P
@@ -18393,6 +18398,7 @@ bare_label_keyword:
 			| HOLD
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| ILIKE
 			| IMMEDIATE
 			| IMMUTABLE
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index e572f585ef..feaf18b043 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -2742,7 +2742,8 @@ psql_completion(const char *text, int start, int end)
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "("))
 		COMPLETE_WITH("FORMAT", "FREEZE", "DELIMITER", "NULL",
 					  "HEADER", "QUOTE", "ESCAPE", "FORCE_QUOTE",
-					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING");
+					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING",
+					  "IGNORE_ERRORS");
 
 	/* Complete COPY <sth> FROM|TO filename WITH (FORMAT */
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "(", "FORMAT"))
diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h
index cb0096aeb6..2b696f99bc 100644
--- a/src/include/commands/copy.h
+++ b/src/include/commands/copy.h
@@ -42,6 +42,7 @@ typedef struct CopyFormatOptions
 								 * -1 if not specified */
 	bool		binary;			/* binary format? */
 	bool		freeze;			/* freeze rows on loading? */
+	bool		ignore_errors;  /* ignore rows with errors */
 	bool		csv_mode;		/* Comma Separated Value format? */
 	CopyHeaderChoice header_line;	/* header line? */
 	char	   *null_print;		/* NULL marker string (server encoding!) */
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index 3df1c5a97c..d9d3af1fb4 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -16,6 +16,8 @@
 
 #include "commands/copy.h"
 #include "commands/trigger.h"
+#include "utils/resowner.h"
+
 
 /*
  * Represents the different source cases we need to worry about at
@@ -49,6 +51,24 @@ typedef enum CopyInsertMethod
 	CIM_MULTI_CONDITIONAL		/* use table_multi_insert only if valid */
 } CopyInsertMethod;
 
+/* Struct that holding fields for ignore_errors option. */
+typedef struct SafeCopyFromState
+{
+#define			REPLAY_BUFFER_SIZE 1000
+	HeapTuple	    replay_buffer[REPLAY_BUFFER_SIZE]; /* accumulates tuples for replaying it after an error */
+	int				saved_tuples;				/* # of tuples in replay_buffer */
+	int 			replayed_tuples;			/* # of tuples was replayed from buffer */
+	int				errors;						/* total # of errors */
+	bool			replay_is_active;
+	bool			begin_subtransaction;
+	bool			processed_remaining_tuples;	/* for case of replaying last tuples */
+	bool			skip_row;
+
+	MemoryContext	oldcontext;
+	ResourceOwner	oldowner;
+	CopyInsertMethod insertMethod;
+} SafeCopyFromState;
+
 /*
  * This struct contains all the state variables used throughout a COPY FROM
  * operation.
@@ -71,6 +91,7 @@ typedef struct CopyFromStateData
 	char	   *filename;		/* filename, or NULL for STDIN */
 	bool		is_program;		/* is 'filename' a program to popen? */
 	copy_data_source_cb data_source_cb; /* function for reading data */
+	SafeCopyFromState *safecstate; /* struct for ignore_errors option */
 
 	CopyFormatOptions opts;
 	bool	   *convert_select_flags;	/* per-column CSV/TEXT CS flags */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index ae35f03251..2af11bd359 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -201,6 +201,7 @@ PG_KEYWORD("hold", HOLD, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("hour", HOUR_P, UNRESERVED_KEYWORD, AS_LABEL)
 PG_KEYWORD("identity", IDENTITY_P, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("if", IF_P, UNRESERVED_KEYWORD, BARE_LABEL)
+PG_KEYWORD("ignore_errors", IGNORE_ERRORS, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("ilike", ILIKE, TYPE_FUNC_NAME_KEYWORD, BARE_LABEL)
 PG_KEYWORD("immediate", IMMEDIATE, UNRESERVED_KEYWORD, BARE_LABEL)
 PG_KEYWORD("immutable", IMMUTABLE, UNRESERVED_KEYWORD, BARE_LABEL)
diff --git a/src/test/regress/expected/copy2.out b/src/test/regress/expected/copy2.out
index 5f3685e9ef..ab1f059a02 100644
--- a/src/test/regress/expected/copy2.out
+++ b/src/test/regress/expected/copy2.out
@@ -649,6 +649,129 @@ SELECT * FROM instead_of_insert_tbl;
 (2 rows)
 
 COMMIT;
+-- tests for IGNORE_ERRORS option
+-- CIM_MULTI case
+CREATE TABLE check_ign_err (n int, m int, k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	2	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	3"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column m: "b"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  FIND 5 ERRORS
+SELECT * FROM check_ign_err;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+ 7 | 7 | 7
+(2 rows)
+
+-- CIM_SINGLE case
+-- BEFORE row trigger
+TRUNCATE check_ign_err;
+CREATE TABLE trig_test(n int, m int);
+CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	2	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	3"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column m: "b"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  FIND 5 ERRORS
+SELECT * FROM check_ign_err;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+ 7 | 7 | 7
+(2 rows)
+
+DROP TRIGGER trig_before on check_ign_err;
+-- INSTEAD OF row trigger
+TRUNCATE check_ign_err;
+TRUNCATE trig_test;
+CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
+CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO check_ign_err VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
+COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err_view, line 2: "2	2	2	2"
+WARNING:  COPY check_ign_err_view, line 3: "3	3"
+WARNING:  COPY check_ign_err_view, line 4, column n: "a"
+WARNING:  COPY check_ign_err_view, line 5, column m: "b"
+WARNING:  COPY check_ign_err_view, line 6, column n: ""
+WARNING:  FIND 5 ERRORS
+SELECT * FROM check_ign_err_view;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+ 7 | 7 | 7
+(2 rows)
+
+DROP TRIGGER trig_instead_of ON check_ign_err_view;
+DROP VIEW check_ign_err_view;
+-- foreign table case in postgres_fdw extension
+-- volatile function in WHERE clause
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
+  WHERE n = floor(random()*(1-1+1))+1; /* find values equal 1 */
+WARNING:  COPY check_ign_err, line 2: "2	2	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	3"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column m: "b"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  FIND 5 ERRORS
+SELECT * FROM check_ign_err;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+(1 row)
+
+DROP TABLE check_ign_err;
+-- CIM_MULTI_CONDITIONAL case
+-- INSERT triggers for partition tables
+TRUNCATE trig_test;
+CREATE TABLE check_ign_err (n int, m int, k int) PARTITION BY RANGE (n);
+CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
+  FOR VALUES FROM (1) TO (4);
+CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
+  FOR VALUES FROM (4) TO (8);
+CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	2	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	3"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column m: "b"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  FIND 5 ERRORS
+SELECT * FROM check_ign_err;
+ n | m | k 
+---+---+---
+ 1 | 1 | 1
+ 7 | 7 | 7
+(2 rows)
+
+DROP TRIGGER trig_before_part on check_ign_err;
+DROP TABLE trig_test;
+DROP TABLE check_ign_err CASCADE;
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql
index b3c16af48e..7eee78bccd 100644
--- a/src/test/regress/sql/copy2.sql
+++ b/src/test/regress/sql/copy2.sql
@@ -454,6 +454,116 @@ test1
 SELECT * FROM instead_of_insert_tbl;
 COMMIT;
 
+-- tests for IGNORE_ERRORS option
+-- CIM_MULTI case
+CREATE TABLE check_ign_err (n int, m int, k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+1	1	1
+2	2	2	2
+3	3
+a	4	4
+5	b	b
+
+7	7	7
+\.
+SELECT * FROM check_ign_err;
+
+-- CIM_SINGLE case
+-- BEFORE row trigger
+TRUNCATE check_ign_err;
+CREATE TABLE trig_test(n int, m int);
+CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+1	1	1
+2	2	2	2
+3	3
+a	4	4
+5	b	b
+
+7	7	7
+\.
+SELECT * FROM check_ign_err;
+DROP TRIGGER trig_before on check_ign_err;
+
+-- INSTEAD OF row trigger
+TRUNCATE check_ign_err;
+TRUNCATE trig_test;
+CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
+CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO check_ign_err VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
+COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS;
+1	1	1
+2	2	2	2
+3	3
+a	4	4
+5	b	b
+
+7	7	7
+\.
+SELECT * FROM check_ign_err_view;
+DROP TRIGGER trig_instead_of ON check_ign_err_view;
+DROP VIEW check_ign_err_view;
+
+-- foreign table case in postgres_fdw extension
+
+-- volatile function in WHERE clause
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
+  WHERE n = floor(random()*(1-1+1))+1; /* find values equal 1 */
+1	1	1
+2	2	2	2
+3	3
+a	4	4
+5	b	b
+
+7	7	7
+\.
+SELECT * FROM check_ign_err;
+DROP TABLE check_ign_err;
+
+-- CIM_MULTI_CONDITIONAL case
+-- INSERT triggers for partition tables
+TRUNCATE trig_test;
+CREATE TABLE check_ign_err (n int, m int, k int) PARTITION BY RANGE (n);
+CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
+  FOR VALUES FROM (1) TO (4);
+CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
+  FOR VALUES FROM (4) TO (8);
+CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+1	1	1
+2	2	2	2
+3	3
+a	4	4
+5	b	b
+
+7	7	7
+\.
+SELECT * FROM check_ign_err;
+DROP TRIGGER trig_before_part on check_ign_err;
+DROP TABLE trig_test;
+DROP TABLE check_ign_err CASCADE;
+
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
-- 
2.25.1

---

From: torikoshia <[email-masked]>
Date: Mon, 22 Aug 2022 21:46:56 +0900

> Also I implemented your case and it worked correctly.

Thanks for the new patch!

Here are some comments on it.

> safecstate->replay_buffer[safecstate->saved_tuples++] = 
> heap_form_tuple(RelationGetDescr(cstate->rel), values, nulls);
> 

What is actually being committed by this ReleaseCurrentSubTransaction()?
It seems to me that just safecstate->replay_buffer is fulfilled before 
this commit.
As a test, I rewrote this ReleaseCurrentSubTransaction() to 
RollbackAndReleaseCurrentSubTransaction() and COPYed over 1000 rows of 
data, but same data were loaded.

> +#define            REPLAY_BUFFER_SIZE 1000

I feel it might be better to have it as a parameter rather than fixed at 
1000.

> +safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum 

NextCopyFrom() is in copyfromparse.c while safeNextCopyFrom() is in 
copyfrom.c.
Since safeNextCopyFrom() is analog of NextCopyFrom() as commented, would 
it be natural to put them in the same file?

> 188 +               safecstate->errors++;
> 189 +               if (safecstate->errors <= 100)
> 190 +                   ereport(WARNING,
> 191 +                           (errcode(errdata->sqlerrcode),
> 192 +                           errmsg("%s", errdata->context)));

It would be better to state in the manual that errors exceeding 100 are 
not displayed.
Or, it might be acceptable to output all errors that exceed 100.

> +   HeapTuple       replay_buffer[REPLAY_BUFFER_SIZE]; /* accumulates 
> tuples for replaying it after an error */

It would be helpful to add comments about skip_row, etc.

```
$ git apply ../patch/0003-COPY_IGNORE_ERRORS.patch

../patch/0003-COPY_IGNORE_ERRORS.patch:86: indent with spaces.
					         Datum *values, bool *nulls);
warning: 1 line adds whitespace errors.
```

There was a warning when applying the patch.

```
=# copy test from '/tmp/10000.data' with (ignore_errors);
WARNING:  FIND 0 ERRORS
COPY 1003
```

When there were no errors, this WARNING seems not necessary.

-- 
Regards,

--
Atsushi Torikoshi
NTT DATA CORPORATION

---

From: Damir Belyalov <[email-masked]>
Date: Wed, 24 Aug 2022 19:54:11 +0300

> It seems to me that just safecstate->replay_buffer is fulfilled before
All tuples are collected in replay_buffer, which is created in
''oldcontext'' of CopyFrom() (not context of a subtransaction). That's why
it didn't clean up when you used RollbackAndReleaseCurrentSubTransaction().
Subtransactions are needed for better safety. There is no error when
copying from a file to the replay_buffer, but an error may occur at the
next stage - when adding a tuple to the table. Also there may be other
errors that are not obvious at first glance.

Yes, I thought about it too. So I created another version with the GUC
parameter - replay_buffer_size. Attached it. But I think users won't need
to change replay_buffer_size.
Also replay_buffer does the same thing as MultiInsert buffer does and
MultiInsert buffer defined by const = 1000.

> it be natural to put them in the same file?
Sure, corrected it.

> Or, it might be acceptable to output all errors that exceed 100.
It'll be too complicated to create a new parameter just for showing the
given number of errors. I think 100 is an optimal size.

> It would be helpful to add comments about skip_row, etc.
Corrected it.

> When there were no errors, this WARNING seems not necessary.
Corrected it.

Add to this patch processing other errors and constraints and tests for
them.
I had to create another function safeExecConstraints() only for processing
constraints, because ExecConstraints() is after NextCopyFrom() and is not
in PG_TRY. This thing a little bit complicated the code.
Maybe it is a good approach to create a new function SafeCopyFrom() and do
all ''safe copying'' in PG_TRY, but it will almost duplicate the CopyFrom()
code.
Or maybe create a function only for loop for(;;). But we have the same
thing with duplicating code and passing a lot of variables (which are
created at the beginning of CopyFrom()) to this function.

---

From: Damir Belyalov <[email-masked]>
Date: Wed, 24 Aug 2022 19:57:13 +0300

From 09befdad45a6b1ae70d6c5abc90d1c2296e56ee1 Mon Sep 17 00:00:00 2001
From: Damir Belyalov <[email-masked]>
Date: Fri, 15 Oct 2021 11:55:18 +0300
Subject: [PATCH] COPY_IGNORE_ERRORS with GUC for replay_buffer size

---
 doc/src/sgml/config.sgml                      |  17 ++
 doc/src/sgml/ref/copy.sgml                    |  19 ++
 src/backend/commands/copy.c                   |   8 +
 src/backend/commands/copyfrom.c               | 114 +++++++++++-
 src/backend/commands/copyfromparse.c          | 169 ++++++++++++++++++
 src/backend/parser/gram.y                     |   8 +-
 src/backend/utils/misc/guc.c                  |  11 ++
 src/backend/utils/misc/postgresql.conf.sample |   2 +
 src/bin/psql/tab-complete.c                   |   3 +-
 src/include/commands/copy.h                   |   6 +
 src/include/commands/copyfrom_internal.h      |  19 ++
 src/include/parser/kwlist.h                   |   1 +
 src/test/regress/expected/copy2.out           | 130 ++++++++++++++
 src/test/regress/sql/copy2.sql                | 116 ++++++++++++
 14 files changed, 617 insertions(+), 6 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 37fd80388c..69373b8d8c 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -1961,6 +1961,23 @@ include_dir 'conf.d'
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-logical-decoding-work-mem" xreflabel="replay_buffer_size">
+      <term><varname>replay_buffer_size</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>replay_buffer_size</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Specifies the size of buffer for COPY FROM IGNORE_ERRORS option. This buffer
+        is created when subtransaction begins and accumulates tuples until an error
+        occurs. Then it starts replaying stored tuples. the buffer size is the size
+        of the subtransaction. Therefore, on large tables, in order to avoid the
+        error of the maximum number of subtransactions, it should be increased.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry id="guc-max-stack-depth" xreflabel="max_stack_depth">
       <term><varname>max_stack_depth</varname> (<type>integer</type>)
       <indexterm>
diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml
index 8aae711b3b..7ff6f6dea7 100644
--- a/doc/src/sgml/ref/copy.sgml
+++ b/doc/src/sgml/ref/copy.sgml
@@ -34,6 +34,7 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
 
@@ -233,6 +234,24 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><literal>IGNORE_ERRORS</literal></term>
+    <listitem>
+     <para>
+      Drop rows that contain malformed data while copying. These are rows
+      containing syntax errors in data, rows with too many or too few columns,
+      rows that result in constraint violations, rows containing columns where
+      the data type's input function raises an error.
+      Outputs warnings about rows with incorrect data (the number of warnings
+      is not more than 100) and the total number of errors.
+      The option is implemented with subtransactions and a buffer created in
+      them, which accumulates tuples until an error occures.
+      The size of buffer is the size of subtransaction block.
+      It is a GUC parameter "replay_buffer_size" and equals 1000 by default.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>DELIMITER</literal></term>
     <listitem>
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 3ac731803b..fead1aba46 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -402,6 +402,7 @@ ProcessCopyOptions(ParseState *pstate,
 {
 	bool		format_specified = false;
 	bool		freeze_specified = false;
+	bool		ignore_errors_specified = false;
 	bool		header_specified = false;
 	ListCell   *option;
 
@@ -442,6 +443,13 @@ ProcessCopyOptions(ParseState *pstate,
 			freeze_specified = true;
 			opts_out->freeze = defGetBoolean(defel);
 		}
+		else if (strcmp(defel->defname, "ignore_errors") == 0)
+		{
+			if (ignore_errors_specified)
+				errorConflictingDefElem(defel, pstate);
+			ignore_errors_specified = true;
+			opts_out->ignore_errors = defGetBoolean(defel);
+		}
 		else if (strcmp(defel->defname, "delimiter") == 0)
 		{
 			if (opts_out->delim)
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index a976008b3d..7e997d15c6 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -73,6 +73,11 @@
 /* Trim the list of buffers back down to this number after flushing */
 #define MAX_PARTITION_BUFFERS	32
 
+/*
+ * GUC parameters
+ */
+int		replay_buffer_size;
+
 /* Stores multi-insert data related to a single relation in CopyFrom. */
 typedef struct CopyMultiInsertBuffer
 {
@@ -100,12 +105,13 @@ typedef struct CopyMultiInsertInfo
 	int			ti_options;		/* table insert options */
 } CopyMultiInsertInfo;
 
-
 /* non-export function prototypes */
 static char *limit_printout_length(const char *str);

+static void safeExecConstraints(CopyFromState cstate, ResultRelInfo *resultRelInfo, TupleTableSlot *myslot, EState *estate);
+
 /*
  * error context callback for COPY FROM
  *
@@ -521,6 +527,61 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
 	miinfo->bufferedBytes += tuplen;
 }
 
+/*
+ * Ignore constraints if IGNORE_ERRORS is enabled
+ */
+static void
+safeExecConstraints(CopyFromState cstate, ResultRelInfo *resultRelInfo, TupleTableSlot *myslot, EState *estate)
+{
+	SafeCopyFromState *safecstate = cstate->safecstate;
+
+	safecstate->skip_row = false;
+
+	PG_TRY();
+		ExecConstraints(resultRelInfo, myslot, estate);
+	PG_CATCH();
+	{
+		ErrorData *errdata;
+		MemoryContext cxt;
+
+		cxt = MemoryContextSwitchTo(safecstate->oldcontext);
+		errdata = CopyErrorData();
+
+		RollbackAndReleaseCurrentSubTransaction();
+		CurrentResourceOwner = safecstate->oldowner;
+
+		switch (errdata->sqlerrcode)
+		{
+			/* Ignore Constraint Violation */
+			case ERRCODE_INTEGRITY_CONSTRAINT_VIOLATION:
+			case ERRCODE_RESTRICT_VIOLATION:
+			case ERRCODE_NOT_NULL_VIOLATION:
+			case ERRCODE_FOREIGN_KEY_VIOLATION:
+			case ERRCODE_UNIQUE_VIOLATION:
+			case ERRCODE_CHECK_VIOLATION:
+			case ERRCODE_EXCLUSION_VIOLATION:
+				safecstate->errors++;
+				if (cstate->safecstate->errors <= 100)
+					ereport(WARNING,
+							(errcode(errdata->sqlerrcode),
+							errmsg("%s", errdata->context)));
+
+				safecstate->begin_subtransaction = true;
+				safecstate->skip_row = true;
+				break;
+			default:
+				PG_RE_THROW();
+		}
+
+		FlushErrorState();
+		FreeErrorData(errdata);
+		errdata = NULL;
+
+		MemoryContextSwitchTo(cxt);
+	}
+	PG_END_TRY();
+}
+
 /*
  * Copy FROM file to relation.
  */
@@ -535,6 +596,7 @@ CopyFrom(CopyFromState cstate)
 	ExprContext *econtext;
 	TupleTableSlot *singleslot = NULL;
 	MemoryContext oldcontext = CurrentMemoryContext;
+	ResourceOwner oldowner = CurrentResourceOwner;
 
@@ -819,9 +881,30 @@ CopyFrom(CopyFromState cstate)
 
+	/* Initialize safeCopyFromState for IGNORE_ERRORS option */
+	if (cstate->opts.ignore_errors)
+	{
+		cstate->safecstate = palloc(sizeof(SafeCopyFromState));
+
+		/* Create replay_buffer in oldcontext */
+		cstate->safecstate->replay_buffer = (HeapTuple *) palloc(replay_buffer_size * sizeof(HeapTuple));
+
+		cstate->safecstate->saved_tuples = 0;
+		cstate->safecstate->replayed_tuples = 0;
+		cstate->safecstate->errors = 0;
+		cstate->safecstate->replay_is_active = false;
+		cstate->safecstate->begin_subtransaction = true;
+		cstate->safecstate->processed_remaining_tuples = false;
+
+		cstate->safecstate->oldowner = oldowner;
+		cstate->safecstate->oldcontext = oldcontext;
+		cstate->safecstate->insertMethod = insertMethod;
+	}
+
 	for (;;)
 	{
 		TupleTableSlot *myslot;
+		bool		valid_row;
 		bool		skip_tuple;
 
 		CHECK_FOR_INTERRUPTS();
@@ -855,8 +938,21 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);
 
-		/* Directly store the values/nulls array in the slot */
-		if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+		/*
+		 * NextCopyFrom() directly store the values/nulls array in the slot.
+		 * safeNextCopyFrom() ignores rows with errors if IGNORE_ERRORS is enabled.
+		 */
+		if (!cstate->safecstate)
+			valid_row = NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull);
+		else
+		{
+			valid_row = safeNextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull);
+
+			if (cstate->safecstate->skip_row)
+				continue;
+		}
+
+		if (!valid_row)
 			break;
 
 		ExecStoreVirtualTuple(myslot);
@@ -1035,7 +1131,17 @@ CopyFrom(CopyFromState cstate)
 				 */
 				if (resultRelInfo->ri_FdwRoutine == NULL &&
 					resultRelInfo->ri_RelationDesc->rd_att->constr)
-					ExecConstraints(resultRelInfo, myslot, estate);
+				{
+					if (cstate->opts.ignore_errors)
+					{
+						safeExecConstraints(cstate, resultRelInfo, myslot, estate);
+
+						if (cstate->safecstate->skip_row)
+							continue;
+					}
+					else
+						ExecConstraints(resultRelInfo, myslot, estate);
+				}
 
 				/*
 				 * Also check the tuple against the partition constraint, if
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 57813b3458..1aae27d80d 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -1026,6 +1026,175 @@ NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 	return true;
 }
 
+/*
+ * Analog of NextCopyFrom() but skips rows with errors while copying.
+ */
+bool
+safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum *values, bool *nulls)
+{
+	SafeCopyFromState *safecstate = cstate->safecstate;
+	bool valid_row = true;
+
+	safecstate->skip_row = false;
+
+	PG_TRY();
+	{
+		if (!safecstate->replay_is_active)
+		{
+			if (safecstate->begin_subtransaction)
+			{
+				BeginInternalSubTransaction(NULL);
+				CurrentResourceOwner = safecstate->oldowner;
+
+				safecstate->begin_subtransaction = false;
+			}
+
+			if (safecstate->saved_tuples < replay_buffer_size)
+			{
+				valid_row = NextCopyFrom(cstate, econtext, values, nulls);
+				if (valid_row)
+				{
+					/* Fill replay_buffer in CopyFrom() oldcontext */
+					MemoryContext cxt = MemoryContextSwitchTo(safecstate->oldcontext);
+
+					safecstate->replay_buffer[safecstate->saved_tuples++] = heap_form_tuple(RelationGetDescr(cstate->rel), values, nulls);
+					MemoryContextSwitchTo(cxt);
+
+					safecstate->skip_row = true;
+				}
+				else if (!safecstate->processed_remaining_tuples)
+				{
+					ReleaseCurrentSubTransaction();
+					CurrentResourceOwner = safecstate->oldowner;
+
+					if (safecstate->replayed_tuples < safecstate->saved_tuples)
+					{
+						/* Prepare to replay remaining tuples if they exist */
+						safecstate->replay_is_active = true;
+						safecstate->processed_remaining_tuples = true;
+						safecstate->skip_row = true;
+						return true;
+					}
+				}
+			}
+			else
+			{
+				/* Buffer was filled, commit subtransaction and prepare to replay */
+				ReleaseCurrentSubTransaction();
+				CurrentResourceOwner = safecstate->oldowner;
+
+				safecstate->replay_is_active = true;
+				safecstate->begin_subtransaction = true;
+				safecstate->skip_row = true;
+			}
+		}
+		else
+		{
+			if (safecstate->replayed_tuples < safecstate->saved_tuples)
+				/* Replaying tuple */
+				heap_deform_tuple(safecstate->replay_buffer[safecstate->replayed_tuples++], RelationGetDescr(cstate->rel), values, nulls);
+			else
+			{
+				/* Clean up replay_buffer */
+				MemSet(safecstate->replay_buffer, 0, replay_buffer_size * sizeof(HeapTuple));
+				safecstate->saved_tuples = safecstate->replayed_tuples = 0;
+
+				safecstate->replay_is_active = false;
+				safecstate->skip_row = true;
+			}
+		}
+	}
+	PG_CATCH();
+	{
+		ErrorData *errdata;
+		MemoryContext cxt;
+
+		cxt = MemoryContextSwitchTo(safecstate->oldcontext);
+		errdata = CopyErrorData();
+
+		RollbackAndReleaseCurrentSubTransaction();
+		CurrentResourceOwner = safecstate->oldowner;
+
+		switch (errdata->sqlerrcode)
+		{
+			/* Ignore malformed data */
+			case ERRCODE_DATA_EXCEPTION:
+			case ERRCODE_ARRAY_ELEMENT_ERROR:
+			case ERRCODE_DATETIME_VALUE_OUT_OF_RANGE:
+			case ERRCODE_INTERVAL_FIELD_OVERFLOW:
+			case ERRCODE_INVALID_CHARACTER_VALUE_FOR_CAST:
+			case ERRCODE_INVALID_DATETIME_FORMAT:
+			case ERRCODE_INVALID_ESCAPE_CHARACTER:
+			case ERRCODE_INVALID_ESCAPE_SEQUENCE:
+			case ERRCODE_NONSTANDARD_USE_OF_ESCAPE_CHARACTER:
+			case ERRCODE_INVALID_PARAMETER_VALUE:
+			case ERRCODE_INVALID_TABLESAMPLE_ARGUMENT:
+			case ERRCODE_INVALID_TIME_ZONE_DISPLACEMENT_VALUE:
+			case ERRCODE_NULL_VALUE_NOT_ALLOWED:
+			case ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE:
+			case ERRCODE_SEQUENCE_GENERATOR_LIMIT_EXCEEDED:
+			case ERRCODE_STRING_DATA_LENGTH_MISMATCH:
+			case ERRCODE_STRING_DATA_RIGHT_TRUNCATION:
+			case ERRCODE_INVALID_TEXT_REPRESENTATION:
+			case ERRCODE_INVALID_BINARY_REPRESENTATION:
+			case ERRCODE_BAD_COPY_FILE_FORMAT:
+			case ERRCODE_UNTRANSLATABLE_CHARACTER:
+			case ERRCODE_DUPLICATE_JSON_OBJECT_KEY_VALUE:
+			case ERRCODE_INVALID_ARGUMENT_FOR_SQL_JSON_DATETIME_FUNCTION:
+			case ERRCODE_INVALID_JSON_TEXT:
+			case ERRCODE_INVALID_SQL_JSON_SUBSCRIPT:
+			case ERRCODE_MORE_THAN_ONE_SQL_JSON_ITEM:
+			case ERRCODE_NO_SQL_JSON_ITEM:
+			case ERRCODE_NON_NUMERIC_SQL_JSON_ITEM:
+			case ERRCODE_NON_UNIQUE_KEYS_IN_A_JSON_OBJECT:
+			case ERRCODE_SINGLETON_SQL_JSON_ITEM_REQUIRED:
+			case ERRCODE_SQL_JSON_ARRAY_NOT_FOUND:
+			case ERRCODE_SQL_JSON_MEMBER_NOT_FOUND:
+			case ERRCODE_SQL_JSON_NUMBER_NOT_FOUND:
+			case ERRCODE_SQL_JSON_OBJECT_NOT_FOUND:
+			case ERRCODE_TOO_MANY_JSON_ARRAY_ELEMENTS:
+			case ERRCODE_TOO_MANY_JSON_OBJECT_MEMBERS:
+			case ERRCODE_SQL_JSON_SCALAR_REQUIRED:
+			case ERRCODE_SQL_JSON_ITEM_CANNOT_BE_CAST_TO_TARGET_TYPE:
+				safecstate->errors++;
+				if (safecstate->errors <= 100)
+					ereport(WARNING,
+							(errcode(errdata->sqlerrcode),
+							errmsg("%s", errdata->context)));
+
+				safecstate->begin_subtransaction = true;
+				safecstate->skip_row = true;
+				break;
+			default:
+				PG_RE_THROW();
+		}
+
+		FlushErrorState();
+		FreeErrorData(errdata);
+		errdata = NULL;
+
+		MemoryContextSwitchTo(cxt);
+	}
+	PG_END_TRY();
+
+	if (!valid_row)
+	{
+		if (safecstate->errors == 0)
+			ereport(NOTICE,
+					errmsg("FIND %d ERRORS", safecstate->errors));
+		else if (safecstate->errors == 1)
+			ereport(WARNING,
+					errmsg("FIND %d ERROR", safecstate->errors));
+		else
+			ereport(WARNING,
+					errmsg("FIND %d ERRORS", safecstate->errors));
+
+		return false;
+	}
+
+	return true;
+}
+
 /*
  * Read the next input line and stash it in line_buf.
  *
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index df5ceea910..3bb7235b34 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -800,7 +800,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);

-	IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
+	IDENTITY_P IF_P IGNORE_ERRORS ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
 	INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
 	INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
 	INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -3456,6 +3456,10 @@ copy_opt_item:
 				{
 					$$ = makeDefElem("freeze", (Node *) makeBoolean(true), @1);
 				}
+			| IGNORE_ERRORS
+				{
+					$$ = makeDefElem("ignore_errors", (Node *)makeInteger(true), @1);
+				}
 			| DELIMITER opt_as Sconst
 				{
 					$$ = makeDefElem("delimiter", (Node *) makeString($3), @1);
@@ -17814,6 +17818,7 @@ unreserved_keyword:
 			| HOUR_P
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| IMMEDIATE
 			| IMMUTABLE
 			| IMPLICIT_P
@@ -18393,6 +18398,7 @@ bare_label_keyword:
 			| HOLD
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| ILIKE
 			| IMMEDIATE
 			| IMMUTABLE
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 0328029d43..54209a4a3c 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -49,6 +49,7 @@
 #include "catalog/pg_parameter_acl.h"
 #include "catalog/storage.h"
 #include "commands/async.h"
+#include "commands/copy.h"
 #include "commands/prepare.h"
 #include "commands/tablespace.h"
 #include "commands/trigger.h"
@@ -2527,6 +2528,16 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"replay_buffer_size", PGC_USERSET, RESOURCES_MEM,
+			gettext_noop("Sets the size of replay buffer for COPY FROM IGNORE_ERRORS option"),
+			NULL
+		},
+		&replay_buffer_size,
+		1000, 1, INT_MAX,
+		NULL, NULL, NULL
+	},
+
 	/*
 	 * We use the hopefully-safely-small value of 100kB as the compiled-in
 	 * default for max_stack_depth.  InitializeGUCOptions will increase it if
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index b4bc06e5f5..f4e777a0a3 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -155,6 +155,8 @@
 					#   mmap
 					# (change requires restart)
 #min_dynamic_shared_memory = 0MB	# (change requires restart)
+#replay_buffer_size = 1000		# min 1
+					# (change requires restart)
 
 # - Disk -

 	/* Complete COPY <sth> FROM|TO filename WITH (FORMAT */
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "(", "FORMAT"))
diff --git a/src/include/commands/copy.h b/src/include/commands/copy.h
index cb0096aeb6..fc9f559efe 100644
--- a/src/include/commands/copy.h
+++ b/src/include/commands/copy.h
@@ -19,6 +19,9 @@
 #include "parser/parse_node.h"
 #include "tcop/dest.h"
 
+/* User-settable GUC parameters */
+extern PGDLLIMPORT int replay_buffer_size;
+
 /*
  * Represents whether a header line should be present, and whether it must
  * match the actual names (which implies "true").
@@ -42,6 +45,7 @@ typedef struct CopyFormatOptions
 								 * -1 if not specified */
 	bool		binary;			/* binary format? */
 	bool		freeze;			/* freeze rows on loading? */
+	bool		ignore_errors;  /* ignore rows with errors */
 	bool		csv_mode;		/* Comma Separated Value format? */
 	CopyHeaderChoice header_line;	/* header line? */
 	char	   *null_print;		/* NULL marker string (server encoding!) */
@@ -78,6 +82,8 @@ extern CopyFromState BeginCopyFrom(ParseState *pstate, Relation rel, Node *where
 extern void EndCopyFrom(CopyFromState cstate);
 extern bool NextCopyFrom(CopyFromState cstate, ExprContext *econtext,
 						 Datum *values, bool *nulls);
+extern bool safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext,
+							 Datum *values, bool *nulls);
 extern bool NextCopyFromRawFields(CopyFromState cstate,
 								  char ***fields, int *nfields);
 extern void CopyFromErrorCallback(void *arg);
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index 3df1c5a97c..4227a7babd 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -16,6 +16,7 @@

 /*
  * Represents the different source cases we need to worry about at
@@ -49,6 +50,23 @@ typedef enum CopyInsertMethod
 	CIM_MULTI_CONDITIONAL		/* use table_multi_insert only if valid */
 } CopyInsertMethod;
 
+ /* Struct that holding fields for COPY FROM IGNORE_ERRORS option. */
+typedef struct SafeCopyFromState
+{
+	HeapTuple	   *replay_buffer;		/* accumulates tuples for replaying them after an error */
+	int				saved_tuples;		/* # of tuples in replay_buffer */
+	int 			replayed_tuples;	/* # of tuples was replayed from buffer */
+	int				errors;				/* total # of errors */
+	bool			replay_is_active;	/* become active after an error */
+	bool			begin_subtransaction; /* if it's true, we can start subtransaction */
+	bool			processed_remaining_tuples;	/* for case of replaying last tuples */
+	bool			skip_row; 			/* if it's true, we should skip this row */
+
+	MemoryContext	oldcontext;			/* create repay_buffer in CopyFrom() context */
+	ResourceOwner	oldowner;			/* CopyFrom() resource owner */
+	CopyInsertMethod insertMethod;
+} SafeCopyFromState;
+
 /*
  * This struct contains all the state variables used throughout a COPY FROM
  * operation.
@@ -71,6 +89,7 @@ typedef struct CopyFromStateData
 	char	   *filename;		/* filename, or NULL for STDIN */
 	bool		is_program;		/* is 'filename' a program to popen? */
 	copy_data_source_cb data_source_cb; /* function for reading data */
+	SafeCopyFromState *safecstate; /* struct for ignore_errors option */
 
 	CopyFormatOptions opts;
index 5f3685e9ef..093c7958be 100644
@@ -649,6 +649,136 @@ SELECT * FROM instead_of_insert_tbl;
 (2 rows)
 
 COMMIT;
+-- tests for IGNORE_ERRORS option
+-- CIM_MULTI case
+CREATE TABLE check_ign_err (n int check (n < 8), m int[], k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column n: "5 {5} 5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  COPY check_ign_err, line 8, column n: "8 {8} 8"
+WARNING:  FIND 7 ERRORS
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+(1 row)
+
+-- CIM_SINGLE cases
+-- BEFORE row trigger
+TRUNCATE check_ign_err;
+CREATE TABLE trig_test(n int, m int[]);
+CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column n: "5 {5} 5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  COPY check_ign_err, line 8, column n: "8 {8} 8"
+WARNING:  FIND 7 ERRORS
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+(1 row)
+
+DROP TRIGGER trig_before on check_ign_err;
+-- INSTEAD OF row trigger
+TRUNCATE check_ign_err;
+TRUNCATE trig_test;
+CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
+CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO check_ign_err VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
+COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err_view, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err_view, line 3: "3	{3}"
+WARNING:  COPY check_ign_err_view, line 4, column n: "a"
+WARNING:  COPY check_ign_err_view, line 5, column n: "5 {5} 5555555555"
+WARNING:  COPY check_ign_err_view, line 6, column n: ""
+WARNING:  COPY check_ign_err_view, line 7, column m: "{a, 7}"
+WARNING:  COPY check_ign_err_view, line 8, column n: "8 {8} 8"
+WARNING:  FIND 7 ERRORS
+SELECT * FROM check_ign_err_view;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+(1 row)
+
+DROP TRIGGER trig_instead_of ON check_ign_err_view;
+DROP VIEW check_ign_err_view;
+-- foreign table case in postgres_fdw extension
+-- volatile function in WHERE clause
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
+  WHERE n = floor(random()*(1-1+1))+1; /* find values equal 1 */
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column n: "5 {5} 5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  COPY check_ign_err, line 8, column n: "8 {8} 8"
+WARNING:  FIND 7 ERRORS
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+(1 row)
+
+DROP TABLE check_ign_err;
+-- CIM_MULTI_CONDITIONAL case
+-- INSERT triggers for partition tables
+TRUNCATE trig_test;
+CREATE TABLE check_ign_err (n int check (n < 8), m int[], k int)
+  PARTITION BY RANGE (k);
+CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
+  FOR VALUES FROM (1) TO (4);
+CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
+  FOR VALUES FROM (4) TO (8);
+CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column n: "5 {5} 5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  COPY check_ign_err, line 8, column n: "8 {8} 8"
+WARNING:  FIND 7 ERRORS
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+(1 row)
+
+DROP TRIGGER trig_before_part on check_ign_err;
+DROP TABLE trig_test;
+DROP TABLE check_ign_err CASCADE;
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql
index b3c16af48e..c91122aa1e 100644
--- a/src/test/regress/sql/copy2.sql
+++ b/src/test/regress/sql/copy2.sql
@@ -454,6 +454,122 @@ test1
 SELECT * FROM instead_of_insert_tbl;
 COMMIT;
 
+CREATE TABLE check_ign_err (n int check (n < 8), m int[], k int);
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5 {5} 5555555555
+7	{a, 7}	7
+8 {8} 8
+-- CIM_SINGLE cases
+CREATE TABLE trig_test(n int, m int[]);
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5 {5} 5555555555
+7	{a, 7}	7
+8 {8} 8
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5 {5} 5555555555
+7	{a, 7}	7
+8 {8} 8
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5 {5} 5555555555
+7	{a, 7}	7
+8 {8} 8
+CREATE TABLE check_ign_err (n int check (n < 8), m int[], k int)
+  PARTITION BY RANGE (k);
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5 {5} 5555555555
+7	{a, 7}	7
+8 {8} 8

---

From: torikoshia <[email-masked]>
Date: Mon, 29 Aug 2022 22:02:38 +0900

On 2022-08-25 01:54, Damir Belyalov wrote:
> ''oldcontext'' of CopyFrom() (not context of a subtransaction). That's
> why it didn't clean up when you used
> RollbackAndReleaseCurrentSubTransaction().
> copying from a file to the replay_buffer, but an error may occur at
> the next stage - when adding a tuple to the table. Also there may be
> other errors that are not obvious at first glance.

Thanks for the explanation and updating patch.
I now understand that the data being COPYed are not the target of 
COMMIT.

Although in past discussions[1] it seems the data to be COPYed were also 
subject to COMMIT, but I understand this patch has adopted another 
design.

350 +               /* Buffer was filled, commit subtransaction and 
prepare to replay */
351 +               ReleaseCurrentSubTransaction();
352 +               CurrentResourceOwner = safecstate->oldowner;
353 +
354 +               safecstate->replay_is_active = true;

BTW in v4 patch, data are loaded into the buffer one by one, and when 
the buffer fills up, the data in the buffer are 'replayed' also one by 
one, right?
Wouldn't this have more overhead than a normal COPY?

As a test, I COPYed slightly larger data with and without ignore_errors 
option.
There might be other reasons, but I found a performance difference.

```
=# copy test from '/tmp/10000000.data' ;
COPY 10000000
Time: 6001.325 ms (00:06.001)

=# copy test from '/tmp/10000000.data' with (ignore_errors);
NOTICE:  FIND 0 ERRORS
COPY 10000000
Time: 7711.555 ms (00:07.712)
```

> parameter - replay_buffer_size. Attached it. But I think users won't
> need to change replay_buffer_size.

Yeah, when the data being COPYed are not the target of COMMIT,  I also 
think users won't neet to change it.
>> it be natural to put them in the same file?

Thanks.

> It'll be too complicated to create a new parameter just for showing
> the given number of errors. I think 100 is an optimal size.

Yeah, I may have misled you, but I also don't think this needs a new 
parameter.
I just thought 'either' of the following would be better:
- Put in the documentation that the warnings will not be output for more 
than 101 cases.
- Output all the warnings.

>> It would be helpful to add comments about skip_row, etc.

Thanks.

>> When there were no errors, this WARNING seems not necessary.

Thanks.

I applied v4 patch and when canceled the COPY, there was a case I found 
myself left in a transaction.
Should this situation be prevented from occurring?

```
=# copy test from '/tmp/10000000.data' with (ignore_errors );

^CCancel request sent
ERROR:  canceling statement due to user request

=# truncate test;
ERROR:  current transaction is aborted, commands ignored until end of 
transaction block
```

[1] 
https://www.postgresql.org/message-id/1197677930.1536.18.camel%40dell.linuxdev.us.dell.com

-- 
Regards,

--
Atsushi Torikoshi
NTT DATA CORPORATION

---

From: Damir Belyalov <[email-masked]>
Date: Wed, 21 Sep 2022 15:11:46 +0300

Thank you for reviewing.
In the previous patch there was an error when processing constraints. The
patch was fixed, but the code grew up and became more complicated
(0005-COPY_IGNORE_ERRORS). I also simplified the logic of
safeNextCopyFrom().
You asked why we need subtransactions, so the answer is in this patch. When
processing a row that does not satisfy constraints or INSTEAD OF triggers,
it is necessary to rollback the subtransaction and return the table to its
original state.
Cause of complexity, I had to abandon the constraints, triggers processing
in and handle only errors that occur when reading the file. Attaching
simplified patch (0006-COPY_IGNORE_ERRORS).
Checked out these patches on all COPY regress tests and it worked correctly.

> Wouldn't this have more overhead than a normal COPY?
The data is loaded into the buffer one by one, but in the "replay mode"
ignore_errors works as standard COPY. Tuples add to the slot and depending
on the type of the slot (CIM_SINGLE or CIM_MULTI) tuples are replayed in
the corresponding case.
For the 0006 patch you can imagine that we divide the loop for(;;) in 2
parts. The first part is adding tuples to the buffer and the second part is
inserting tuples to the table. These parts don't intersect with each other
and are completed sequentially.
The main idea of replay_buffer is that it is needed for the CIM_SINGLE
case. You can implement the CIM_SINGLE case and see that tuples before an
error occurring don't add to the table. Logic of the 0005 patch is similar
but with some differences.

> There might be other reasons, but I found a performance difference.

Tried to reduce performance difference with cleaning up replay_buffer with
resetting the new context for replay_buffer - replay_cxt.
```
Before:
Without ignore_errors:
COPY 10000000
Time: 15538,579 ms (00:15,539)
With ignore_errors:
COPY 10000000
Time: 21289,121 ms (00:21,289)

After:
Without ignore_errors:
COPY 10000000
Time: 15318,922 ms (00:15,319)
With ignore_errors:
COPY 10000000
Time: 19868,175 ms (00:19,868)
```

> than 101 cases.
Yeah, I point it out in the doc.

> ERROR:  current transaction is aborted, commands ignored until end of
Tried to implement your error and could not. The result was the same as
COPY FROM implements.
diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml
index c25b52d0cb..c99adabcc9 100644
--- a/doc/src/sgml/ref/copy.sgml
+++ b/doc/src/sgml/ref/copy.sgml
@@ -34,6 +34,7 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
 
@@ -233,6 +234,20 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+      Drops rows that contain malformed data while copying. These are rows
index 49924e476a..f41b25f26a 100644
@@ -406,6 +406,7 @@ ProcessCopyOptions(ParseState *pstate,
 				   bool is_from,
@@ -448,6 +449,13 @@ ProcessCopyOptions(ParseState *pstate,
index e8bb168aea..6474d47d29 100644
@@ -106,6 +106,12 @@ static char *limit_printout_length(const char *str);

+static bool safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext,
+							 Datum *values, bool *nulls);
+static bool safeExecConstraints(CopyFromState cstate, ResultRelInfo *resultRelInfo, TupleTableSlot *myslot, EState *estate);
+
+static void addToReplayBuffer(CopyFromState cstate, TupleTableSlot *myslot);
+
 /*
  * error context callback for COPY FROM
  *
@@ -521,6 +527,254 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
 	miinfo->bufferedBytes += tuplen;
 }
 
+/*
+ * Analog of NextCopyFrom() but ignores rows with errors while copying.
+ */
+bool
+safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum *values, bool *nulls)
+{
+	SafeCopyFromState *sfcstate = cstate->sfcstate;
+
+	sfcstate->tuple_is_valid = false;
+
+	PG_TRY();
+	{
+		if (sfcstate->begin_subxact)
+		{
+			BeginInternalSubTransaction(NULL);
+			CurrentResourceOwner = sfcstate->oldowner;
+
+			sfcstate->begin_subxact = false;
+		}
+
+		if (!sfcstate->replay_is_active)
+		{
+			if (sfcstate->saved_tuples < REPLAY_BUFFER_SIZE)
+			{
+				MemoryContext cxt = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
+				bool valid_row = NextCopyFrom(cstate, econtext, values, nulls);
+
+				CurrentMemoryContext = cxt;
+
+				sfcstate->tuple_is_valid = true;
+
+				if (!valid_row && sfcstate->replayed_tuples < sfcstate->saved_tuples)
+				{
+					/* Prepare for replaying remaining tuples if they exist */
+					sfcstate->replay_is_active = true;
+
+					/* If there are insteadof triggers we should rollback subtransaction */
+					if (sfcstate->has_instead_insert_row_trig)
+					{
+						RollbackAndReleaseCurrentSubTransaction();
+						CurrentResourceOwner = sfcstate->oldowner;
+
+						sfcstate->begin_subxact = true;
+					}
+				}
+				else if (!valid_row)
+				{
+					ReleaseCurrentSubTransaction();
+					CurrentResourceOwner = sfcstate->oldowner;
+
+					if (sfcstate->errors == 0)
+						ereport(NOTICE,
+								errmsg("%d errors", sfcstate->errors));
+					else if (sfcstate->errors == 1)
+						ereport(WARNING,
+								errmsg("%d error", sfcstate->errors));
+					else
+						ereport(WARNING,
+								errmsg("%d errors", sfcstate->errors));
+
+					return false;
+				}
+			}
+			else
+			{
+				/* Buffer was filled, prepare for replaying */
+				sfcstate->replay_is_active = true;
+
+				/* If there are insteadof triggers we should rollback subtransaction */
+				if (sfcstate->has_instead_insert_row_trig)
+				{
+					RollbackAndReleaseCurrentSubTransaction();
+					CurrentResourceOwner = sfcstate->oldowner;
+
+					sfcstate->begin_subxact = true;
+				}
+			}
+		}
+
+		if (sfcstate->replay_is_active)
+		{
+			if (sfcstate->replayed_tuples < sfcstate->saved_tuples)
+			{
+				/* Replaying the tuple */
+				MemoryContext cxt = MemoryContextSwitchTo(sfcstate->replay_cxt);
+
+				heap_deform_tuple(sfcstate->replay_buffer[sfcstate->replayed_tuples++], RelationGetDescr(cstate->rel), values, nulls);
+				MemoryContextSwitchTo(cxt);
+
+				sfcstate->tuple_is_valid = true;
+			}
+			else
+			{
+				/* Clean up replay_buffer */
+				MemoryContextReset(sfcstate->replay_cxt);
+				sfcstate->saved_tuples = sfcstate->replayed_tuples = 0;
+
+				cstate->sfcstate->replay_buffer = MemoryContextAlloc(cstate->sfcstate->replay_cxt,
+												  REPLAY_BUFFER_SIZE * sizeof(HeapTuple));
+
+				ReleaseCurrentSubTransaction();
+				CurrentResourceOwner = sfcstate->oldowner;
+
+				sfcstate->begin_subxact = true;
+				sfcstate->replay_is_active = false;
+			}
+		}
+	}
+	PG_CATCH();
+	{
+		MemoryContext ecxt = MemoryContextSwitchTo(sfcstate->oldcontext);
+		ErrorData *errdata = CopyErrorData();
+
+		RollbackAndReleaseCurrentSubTransaction();
+		CurrentResourceOwner = sfcstate->oldowner;
+
+		switch (errdata->sqlerrcode)
+		{
+			/* Ignore data exceptions */
+			case ERRCODE_CHARACTER_NOT_IN_REPERTOIRE:
+			case ERRCODE_DATA_EXCEPTION:
+			case ERRCODE_ARRAY_ELEMENT_ERROR:
+			case ERRCODE_DATETIME_VALUE_OUT_OF_RANGE:
+			case ERRCODE_INTERVAL_FIELD_OVERFLOW:
+			case ERRCODE_INVALID_CHARACTER_VALUE_FOR_CAST:
+			case ERRCODE_INVALID_DATETIME_FORMAT:
+			case ERRCODE_INVALID_ESCAPE_CHARACTER:
+			case ERRCODE_INVALID_ESCAPE_SEQUENCE:
+			case ERRCODE_NONSTANDARD_USE_OF_ESCAPE_CHARACTER:
+			case ERRCODE_INVALID_PARAMETER_VALUE:
+			case ERRCODE_INVALID_TABLESAMPLE_ARGUMENT:
+			case ERRCODE_INVALID_TIME_ZONE_DISPLACEMENT_VALUE:
+			case ERRCODE_NULL_VALUE_NOT_ALLOWED:
+			case ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE:
+			case ERRCODE_SEQUENCE_GENERATOR_LIMIT_EXCEEDED:
+			case ERRCODE_STRING_DATA_LENGTH_MISMATCH:
+			case ERRCODE_STRING_DATA_RIGHT_TRUNCATION:
+			case ERRCODE_INVALID_TEXT_REPRESENTATION:
+			case ERRCODE_INVALID_BINARY_REPRESENTATION:
+			case ERRCODE_BAD_COPY_FILE_FORMAT:
+			case ERRCODE_UNTRANSLATABLE_CHARACTER:
+			case ERRCODE_DUPLICATE_JSON_OBJECT_KEY_VALUE:
+			case ERRCODE_INVALID_ARGUMENT_FOR_SQL_JSON_DATETIME_FUNCTION:
+			case ERRCODE_INVALID_JSON_TEXT:
+			case ERRCODE_INVALID_SQL_JSON_SUBSCRIPT:
+			case ERRCODE_MORE_THAN_ONE_SQL_JSON_ITEM:
+			case ERRCODE_NO_SQL_JSON_ITEM:
+			case ERRCODE_NON_NUMERIC_SQL_JSON_ITEM:
+			case ERRCODE_NON_UNIQUE_KEYS_IN_A_JSON_OBJECT:
+			case ERRCODE_SINGLETON_SQL_JSON_ITEM_REQUIRED:
+			case ERRCODE_SQL_JSON_ARRAY_NOT_FOUND:
+			case ERRCODE_SQL_JSON_MEMBER_NOT_FOUND:
+			case ERRCODE_SQL_JSON_NUMBER_NOT_FOUND:
+			case ERRCODE_SQL_JSON_OBJECT_NOT_FOUND:
+			case ERRCODE_TOO_MANY_JSON_ARRAY_ELEMENTS:
+			case ERRCODE_TOO_MANY_JSON_OBJECT_MEMBERS:
+			case ERRCODE_SQL_JSON_SCALAR_REQUIRED:
+			case ERRCODE_SQL_JSON_ITEM_CANNOT_BE_CAST_TO_TARGET_TYPE:
+				sfcstate->errors++;
+				if (sfcstate->errors <= 100)
+					ereport(WARNING,
+							(errcode(errdata->sqlerrcode),
+							errmsg("%s", errdata->context)));
+
+				sfcstate->begin_subxact = true;
+
+				break;
+			default:
+				PG_RE_THROW();
+		}
+
+		FlushErrorState();
+		FreeErrorData(errdata);
+		errdata = NULL;
+
+		MemoryContextSwitchTo(ecxt);
+	}
+	PG_END_TRY();
+
+	return true;
+}
+
+/*
+ * Analog of ExecConstraints(), but ignores rows in constraint violations.
+ */
+bool
+safeExecConstraints(CopyFromState cstate, ResultRelInfo *resultRelInfo, TupleTableSlot *myslot, EState *estate)
+{
+	SafeCopyFromState *sfcstate = cstate->sfcstate;
+
+	PG_TRY();
+		ExecConstraints(resultRelInfo, myslot, estate);
+	PG_CATCH();
+	{
+		MemoryContext ecxt = MemoryContextSwitchTo(sfcstate->oldcontext);
+		ErrorData *errdata = CopyErrorData();
+
+		RollbackAndReleaseCurrentSubTransaction();
+		CurrentResourceOwner = sfcstate->oldowner;
+
+		switch (errdata->sqlerrcode)
+		{
+			/* Ignore constraint violations */
+			case ERRCODE_INTEGRITY_CONSTRAINT_VIOLATION:
+			case ERRCODE_RESTRICT_VIOLATION:
+			case ERRCODE_NOT_NULL_VIOLATION:
+			case ERRCODE_FOREIGN_KEY_VIOLATION:
+			case ERRCODE_UNIQUE_VIOLATION:
+			case ERRCODE_CHECK_VIOLATION:
+			case ERRCODE_EXCLUSION_VIOLATION:
+				sfcstate->errors++;
+				if (sfcstate->errors <= 100)
+					ereport(WARNING,
+							(errcode(errdata->sqlerrcode),
+							errmsg("%s %s", errdata->message, errdata->detail)));
+
+				sfcstate->begin_subxact = true;
+				sfcstate->tuple_is_valid = false;
+
+				break;
+			default:
+				PG_RE_THROW();
+		}
+
+		FlushErrorState();
+		FreeErrorData(errdata);
+		errdata = NULL;
+
+		MemoryContextSwitchTo(ecxt);
+	}
+	PG_END_TRY();
+
+	if (!sfcstate->tuple_is_valid)
+		return false;
+
+	return true;
+}
+
+void
+addToReplayBuffer(CopyFromState cstate, TupleTableSlot *myslot)
+{
+	MemoryContext cxt = MemoryContextSwitchTo(cstate->sfcstate->replay_cxt);
+	HeapTuple saved_tuple = heap_form_tuple(RelationGetDescr(cstate->rel), myslot->tts_values, myslot->tts_isnull);
+
+	cstate->sfcstate->replay_buffer[cstate->sfcstate->saved_tuples++] = saved_tuple;
+	MemoryContextSwitchTo(cxt);
+}
+
 /*
  * Copy FROM file to relation.
  */
@@ -535,6 +789,7 @@ CopyFrom(CopyFromState cstate)
 	ExprContext *econtext;
 	TupleTableSlot *singleslot = NULL;
 	MemoryContext oldcontext = CurrentMemoryContext;
+	ResourceOwner oldowner = CurrentResourceOwner;
 
@@ -819,6 +1074,27 @@ CopyFrom(CopyFromState cstate)
 
+	/* Initialize safeCopyFromState for IGNORE_ERRORS option */
+	if (cstate->opts.ignore_errors)
+	{
+		cstate->sfcstate = palloc(sizeof(SafeCopyFromState));
+
+		cstate->sfcstate->replay_cxt = AllocSetContextCreate(oldcontext,
+									   "Replay context",
+									   ALLOCSET_DEFAULT_SIZES);
+		cstate->sfcstate->replay_buffer = MemoryContextAlloc(cstate->sfcstate->replay_cxt,
+						   REPLAY_BUFFER_SIZE * sizeof(HeapTuple));
+		cstate->sfcstate->saved_tuples = 0;
+		cstate->sfcstate->replayed_tuples = 0;
+		cstate->sfcstate->errors = 0;
+		cstate->sfcstate->replay_is_active = false;
+		cstate->sfcstate->begin_subxact = true;
+		cstate->sfcstate->oldowner = oldowner;
+		cstate->sfcstate->oldcontext = oldcontext;
+		if (has_instead_insert_row_trig)
+			cstate->sfcstate->has_instead_insert_row_trig = true;
+	}
+
 	for (;;)
 	{
 		TupleTableSlot *myslot;
@@ -855,19 +1131,25 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);
 
-		/* Directly store the values/nulls array in the slot */
-		if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
-			break;
+		if (cstate->sfcstate)
+		{
+			/* If option IGNORE_ERRORS is enabled, COPY skips rows with errors */
+			if (!safeNextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+				break;
+			if (!cstate->sfcstate->tuple_is_valid)
+				continue;
+		}
+		else
+		{
+			/* Directly store the values/nulls array in the slot */
+			if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+				break;
+		}

-		/*
-		 * Constraints and where clause might reference the tableoid column,
-		 * so (re-)initialize tts_tableOid before evaluating them.
-		 */
 		myslot->tts_tableOid = RelationGetRelid(target_resultRelInfo->ri_RelationDesc);
 
-		/* Triggers and stuff need to be invoked in query context. */
 		MemoryContextSwitchTo(oldcontext);
 
 		if (cstate->whereClause)
@@ -1020,6 +1302,13 @@ CopyFrom(CopyFromState cstate)
 			if (has_instead_insert_row_trig)
 			{
 				ExecIRInsertTriggers(estate, resultRelInfo, myslot);
+
+				/* Add tuple to replay_buffer if IGNORE_ERRORS is enabled */
+				if (cstate->sfcstate && !cstate->sfcstate->replay_is_active)
+				{
+					addToReplayBuffer(cstate, myslot);
+					continue;
+				}
 			}
 			else
 			{
@@ -1035,7 +1324,15 @@ CopyFrom(CopyFromState cstate)
 				 */
 				if (resultRelInfo->ri_FdwRoutine == NULL &&
 					resultRelInfo->ri_RelationDesc->rd_att->constr)
-					ExecConstraints(resultRelInfo, myslot, estate);
+				{
+					if (cstate->sfcstate)
+					{
+						if (!safeExecConstraints(cstate, resultRelInfo, myslot, estate))
+							continue;
+					}
+					else
+						ExecConstraints(resultRelInfo, myslot, estate);
+				}
 
 				/*
 				 * Also check the tuple against the partition constraint, if
@@ -1047,6 +1344,13 @@ CopyFrom(CopyFromState cstate)
 					(proute == NULL || has_before_insert_row_trig))
 					ExecPartitionCheck(resultRelInfo, myslot, estate, true);
 
+				/* Add tuple to replay_buffer if IGNORE_ERRORS is enabled */
+				if (cstate->sfcstate && !cstate->sfcstate->replay_is_active)
+				{
+					addToReplayBuffer(cstate, myslot);
+					continue;
+				}
+
 				/* Store the slot in the multi-insert buffer, when enabled. */
 				if (insertMethod == CIM_MULTI || leafpart_use_multi_insert)
 				{
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index b5ab9d9c9a..b49954c0aa 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -808,7 +808,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);

-	IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
+	IDENTITY_P IF_P IGNORE_ERRORS ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
 	INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
 	INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
 	INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -3477,6 +3477,10 @@ copy_opt_item:
 				{
 					$$ = makeDefElem("freeze", (Node *) makeBoolean(true), @1);
 				}
+			| IGNORE_ERRORS
+				{
+					$$ = makeDefElem("ignore_errors", (Node *)makeBoolean(true), @1);
+				}
 			| DELIMITER opt_as Sconst
 				{
 					$$ = makeDefElem("delimiter", (Node *) makeString($3), @1);
@@ -17778,6 +17782,7 @@ unreserved_keyword:
 			| HOUR_P
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| IMMEDIATE
 			| IMMUTABLE
 			| IMPLICIT_P
@@ -18357,6 +18362,7 @@ bare_label_keyword:
 			| HOLD
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| ILIKE
 			| IMMEDIATE
 			| IMMUTABLE
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 62a39779b9..fe590ff7a8 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -2748,7 +2748,8 @@ psql_completion(const char *text, int start, int end)
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "("))
 		COMPLETE_WITH("FORMAT", "FREEZE", "DELIMITER", "NULL",
 					  "HEADER", "QUOTE", "ESCAPE", "FORCE_QUOTE",
-					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING");
+					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING",
+					  "IGNORE_ERRORS");
 
index e37c6032ae..5615fa55ef 100644

 /*
  * Represents the different source cases we need to worry about at
@@ -49,6 +50,24 @@ typedef enum CopyInsertMethod
 	CIM_MULTI_CONDITIONAL		/* use table_multi_insert only if valid */
 } CopyInsertMethod;
 
+/* Struct that holding fields for ignore_errors option. */
+typedef struct SafeCopyFromState
+{
+#define			REPLAY_BUFFER_SIZE 10
+	HeapTuple	   *replay_buffer; 			/* accumulates tuples for replaying it after an error */
+	int				saved_tuples;			/* # of tuples in replay_buffer */
+	int 			replayed_tuples;		/* # of tuples was replayed from buffer */
+	int				errors;					/* total # of errors */
+	bool			replay_is_active;		/* if true we replay tuples from buffer */
+	bool			begin_subxact;			/* if true we can begin subtransaction */
+	bool			tuple_is_valid;
+	bool			has_instead_insert_row_trig;
+
+	MemoryContext	replay_cxt;
+	MemoryContext	oldcontext;
+	ResourceOwner	oldowner;
+} SafeCopyFromState;
+
 /*
  * This struct contains all the state variables used throughout a COPY FROM
  * operation.
@@ -71,6 +90,7 @@ typedef struct CopyFromStateData
 	char	   *filename;		/* filename, or NULL for STDIN */
 	bool		is_program;		/* is 'filename' a program to popen? */
 	copy_data_source_cb data_source_cb; /* function for reading data */
+	SafeCopyFromState *sfcstate; /* struct for ignore_errors option */
 
 	CopyFormatOptions opts;
index 5f3685e9ef..d74575fd40 100644
@@ -649,6 +649,134 @@ SELECT * FROM instead_of_insert_tbl;
 (2 rows)
 
 COMMIT;
+-- tests for IGNORE_ERRORS option
+-- CIM_MULTI case
+CREATE TABLE check_ign_err (n int check (n != 6), m int[], k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  new row for relation "check_ign_err" violates check constraint "check_ign_err_n_check" Failing row contains (6, {6}, 6).
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 8 | {8} | 8
+(2 rows)
+
+-- CIM_SINGLE cases
+-- BEFORE row trigger
+TRUNCATE check_ign_err;
+CREATE TABLE trig_test(n int, m int[], k int);
+CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
+  FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  new row for relation "check_ign_err" violates check constraint "check_ign_err_n_check" Failing row contains (6, {6}, 6).
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 8 | {8} | 8
+(2 rows)
+
+DROP TRIGGER trig_before on check_ign_err;
+-- INSTEAD OF row trigger
+TRUNCATE check_ign_err;
+TRUNCATE trig_test;
+CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
+CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
+  FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
+COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err_view, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err_view, line 3: "3	{3}"
+WARNING:  COPY check_ign_err_view, line 4, column n: "a"
+WARNING:  COPY check_ign_err_view, line 5, column k: "5555555555"
+WARNING:  COPY check_ign_err_view, line 7, column m: "{a, 7}"
+WARNING:  5 errors
+SELECT * FROM trig_test;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 6 | {6} | 6
+ 8 | {8} | 8
+(3 rows)
+
+DROP TRIGGER trig_instead_of ON check_ign_err_view;
+DROP VIEW check_ign_err_view;
+-- foreign table case is in postgres_fdw extension
+-- volatile function in WHERE clause
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
+  WHERE n = floor(random()*(1-1+1))+1; /* finds values equal 1 */
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  5 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+(1 row)
+
+DROP TABLE check_ign_err;
+-- CIM_MULTI_CONDITIONAL case
+-- INSERT triggers for partition tables
+TRUNCATE trig_test;
+CREATE TABLE check_ign_err (n int check (n != 6), m int[], k int)
+  PARTITION BY RANGE (k);
+CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
+  FOR VALUES FROM (1) TO (4);
+CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
+  FOR VALUES FROM (4) TO (9);
+CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
+  FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  new row for relation "check_ign_err_part2" violates check constraint "check_ign_err_n_check" Failing row contains (6, {6}, 6).
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 8 | {8} | 8
+(2 rows)
+
+DROP TRIGGER trig_before_part on check_ign_err;
+DROP TABLE trig_test;
+DROP TABLE check_ign_err CASCADE;
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql
index b3c16af48e..8d29ceba26 100644
--- a/src/test/regress/sql/copy2.sql
+++ b/src/test/regress/sql/copy2.sql
@@ -454,6 +454,122 @@ test1
 SELECT * FROM instead_of_insert_tbl;
 COMMIT;
 
+-- tests for IGNORE_ERRORS option
+-- CIM_MULTI case
+CREATE TABLE check_ign_err (n int check (n != 6), m int[], k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+6	{6}	6
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+
+-- CIM_SINGLE cases
+-- BEFORE row trigger
+TRUNCATE check_ign_err;
+CREATE TABLE trig_test(n int, m int[], k int);
+CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
+  FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+6	{6}	6
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+DROP TRIGGER trig_before on check_ign_err;
+
+-- INSTEAD OF row trigger
+TRUNCATE check_ign_err;
+TRUNCATE trig_test;
+CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
+CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
+  FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
+COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+6	{6}	6
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM trig_test;
+DROP TRIGGER trig_instead_of ON check_ign_err_view;
+DROP VIEW check_ign_err_view;
+
+-- foreign table case is in postgres_fdw extension
+
+-- volatile function in WHERE clause
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
+  WHERE n = floor(random()*(1-1+1))+1; /* finds values equal 1 */
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+6	{6}	6
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+DROP TABLE check_ign_err;
+
+-- CIM_MULTI_CONDITIONAL case
+-- INSERT triggers for partition tables
+TRUNCATE trig_test;
+CREATE TABLE check_ign_err (n int check (n != 6), m int[], k int)
+  PARTITION BY RANGE (k);
+CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
+  FOR VALUES FROM (1) TO (4);
+CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
+  FOR VALUES FROM (4) TO (9);
+CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
+  FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+6	{6}	6
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+DROP TRIGGER trig_before_part on check_ign_err;
+DROP TABLE trig_test;
+DROP TABLE check_ign_err CASCADE;
+
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml
index c25b52d0cb..22c992e6f6 100644
--- a/doc/src/sgml/ref/copy.sgml
+++ b/doc/src/sgml/ref/copy.sgml
@@ -34,6 +34,7 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
 
@@ -233,6 +234,19 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><literal>IGNORE_ERRORS</literal></term>
+    <listitem>
+     <para>
+      Drops rows that contain malformed data while copying. These are rows
+      containing syntax errors in data, rows with too many or too few columns,
+      rows containing columns where the data type's input function raises an error.
+      Outputs warnings about rows with incorrect data (the number of warnings
+      is not more than 100) and the total number of errors.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>DELIMITER</literal></term>
     <listitem>
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 49924e476a..f41b25f26a 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -406,6 +406,7 @@ ProcessCopyOptions(ParseState *pstate,
 				   bool is_from,
 				   List *options)
 {
+	bool		ignore_errors_specified = false;
 	bool		format_specified = false;
 	bool		freeze_specified = false;
 	bool		header_specified = false;
@@ -448,6 +449,13 @@ ProcessCopyOptions(ParseState *pstate,
 			freeze_specified = true;
 			opts_out->freeze = defGetBoolean(defel);
 		}
+		else if (strcmp(defel->defname, "ignore_errors") == 0)
+		{
+			if (ignore_errors_specified)
+				errorConflictingDefElem(defel, pstate);
+			ignore_errors_specified = true;
+			opts_out->ignore_errors = defGetBoolean(defel);
+		}
 		else if (strcmp(defel->defname, "delimiter") == 0)
 		{
 			if (opts_out->delim)
diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index e8bb168aea..39f1dca084 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -535,6 +535,7 @@ CopyFrom(CopyFromState cstate)
 	ExprContext *econtext;
 	TupleTableSlot *singleslot = NULL;
 	MemoryContext oldcontext = CurrentMemoryContext;
+	ResourceOwner oldowner = CurrentResourceOwner;
 
@@ -819,6 +820,26 @@ CopyFrom(CopyFromState cstate)
 
+	/* Initialize safeCopyFromState for IGNORE_ERRORS option*/
+	if (cstate->opts.ignore_errors)
+	{
+		cstate->sfcstate = palloc(sizeof(SafeCopyFromState));
+
+		cstate->sfcstate->replay_cxt = AllocSetContextCreate(oldcontext,
+									   "Replay_context",
+									   ALLOCSET_DEFAULT_SIZES);
+		cstate->sfcstate->replay_buffer = MemoryContextAlloc(cstate->sfcstate->replay_cxt,
+										  REPLAY_BUFFER_SIZE * sizeof(HeapTuple));
+		cstate->sfcstate->saved_tuples = 0;
+		cstate->sfcstate->replayed_tuples = 0;
+		cstate->sfcstate->errors = 0;
+		cstate->sfcstate->replay_is_active = false;
+		cstate->sfcstate->begin_subxact = true;
+
+		cstate->sfcstate->oldowner = oldowner;
+		cstate->sfcstate->oldcontext = oldcontext;
+	}
+
 	for (;;)
 	{
 		TupleTableSlot *myslot;
@@ -855,9 +876,18 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);
 
-		/* Directly store the values/nulls array in the slot */
-		if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
-			break;
+		if (cstate->sfcstate)
+		{
+			/* If option IGNORE_ERRORS is enabled, COPY skips rows with errors */
+			if (!safeNextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+				break;
+			if (!cstate->sfcstate->replay_is_active)
+				continue;
+		}
+		else
+			/* Directly store the values/nulls array in the slot */
+			if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+				break;

@@ -1035,7 +1065,21 @@ CopyFrom(CopyFromState cstate)
 				 */
 				if (resultRelInfo->ri_FdwRoutine == NULL &&
 					resultRelInfo->ri_RelationDesc->rd_att->constr)
-					ExecConstraints(resultRelInfo, myslot, estate);
+				{
+					if (cstate->opts.ignore_errors)
+					{
+						PG_TRY();
+							ExecConstraints(resultRelInfo, myslot, estate);
+						PG_CATCH();
+							RollbackAndReleaseCurrentSubTransaction();
+							CurrentResourceOwner = cstate->sfcstate->oldowner;
+
+							PG_RE_THROW();
+						PG_END_TRY();
+					}
+					else
+						ExecConstraints(resultRelInfo, myslot, estate);
+				}
 
 				/*
 				 * Also check the tuple against the partition constraint, if
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 7cf3e865cf..6b92e781e3 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -839,6 +839,169 @@ NextCopyFromRawFields(CopyFromState cstate, char ***fields, int *nfields)
 	return true;
 }
 
+/*
+ * Analog of NextCopyFrom() but ignores rows with errors while copying.
+ */
+bool
+safeNextCopyFrom(CopyFromState cstate, ExprContext *econtext, Datum *values, bool *nulls)
+{
+	SafeCopyFromState *sfcstate = cstate->sfcstate;
+
+	PG_TRY();
+	{
+		if (sfcstate->begin_subxact)
+		{
+			BeginInternalSubTransaction(NULL);
+			CurrentResourceOwner = sfcstate->oldowner;
+
+			sfcstate->begin_subxact = false;
+		}
+
+		if (!sfcstate->replay_is_active)
+		{
+			if (sfcstate->saved_tuples < REPLAY_BUFFER_SIZE)
+			{
+				MemoryContext cxt = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
+				bool valid_row = NextCopyFrom(cstate, econtext, values, nulls);
+
+				CurrentMemoryContext = cxt;
+
+				if (valid_row)
+				{
+					/* Filling replay_buffer in Replay_context */
+					MemoryContext cxt = MemoryContextSwitchTo(sfcstate->replay_cxt);
+					HeapTuple saved_tuple = heap_form_tuple(RelationGetDescr(cstate->rel), values, nulls);
+
+					sfcstate->replay_buffer[sfcstate->saved_tuples++] = saved_tuple;
+					MemoryContextSwitchTo(cxt);
+				}
+				else if (sfcstate->replayed_tuples < sfcstate->saved_tuples)
+					/* Prepare for replaying remaining tuples if they exist */
+					sfcstate->replay_is_active = true;
+				else
+				{
+					ReleaseCurrentSubTransaction();
+					CurrentResourceOwner = sfcstate->oldowner;
+
+					if (sfcstate->errors == 0)
+						ereport(NOTICE,
+								errmsg("%d errors", sfcstate->errors));
+					else if (sfcstate->errors == 1)
+						ereport(WARNING,
+								errmsg("%d error", sfcstate->errors));
+					else
+						ereport(WARNING,
+								errmsg("%d errors", sfcstate->errors));
+
+					return false;
+				}
+			}
+			else
+				/* Buffer was filled, prepare for replaying */
+				sfcstate->replay_is_active = true;
+		}
+
+		if (sfcstate->replay_is_active)
+		{
+			if (sfcstate->replayed_tuples < sfcstate->saved_tuples)
+			{
+				/* Replaying the tuple */
+				MemoryContext cxt = MemoryContextSwitchTo(sfcstate->replay_cxt);
+
+				heap_deform_tuple(sfcstate->replay_buffer[sfcstate->replayed_tuples++], RelationGetDescr(cstate->rel), values, nulls);
+				MemoryContextSwitchTo(cxt);
+			}
+			else
+			{
+				/* Clean up replay_buffer */
+				MemoryContextReset(sfcstate->replay_cxt);
+				sfcstate->saved_tuples = sfcstate->replayed_tuples = 0;
+
+				cstate->sfcstate->replay_buffer = MemoryContextAlloc(cstate->sfcstate->replay_cxt,
+												  REPLAY_BUFFER_SIZE * sizeof(HeapTuple));
+
+				ReleaseCurrentSubTransaction();
+				CurrentResourceOwner = sfcstate->oldowner;
+
+				sfcstate->begin_subxact = true;
+				sfcstate->replay_is_active = false;
+			}
+		}
+	}
+	PG_CATCH();
+	{
+		MemoryContext ecxt = MemoryContextSwitchTo(sfcstate->oldcontext);
+		ErrorData *errdata = CopyErrorData();
+
+		RollbackAndReleaseCurrentSubTransaction();
+		CurrentResourceOwner = sfcstate->oldowner;
+
+		switch (errdata->sqlerrcode)
+		{
+			/* Ignore data exceptions */
+			case ERRCODE_CHARACTER_NOT_IN_REPERTOIRE:
+			case ERRCODE_DATA_EXCEPTION:
+			case ERRCODE_ARRAY_ELEMENT_ERROR:
+			case ERRCODE_DATETIME_VALUE_OUT_OF_RANGE:
+			case ERRCODE_INTERVAL_FIELD_OVERFLOW:
+			case ERRCODE_INVALID_CHARACTER_VALUE_FOR_CAST:
+			case ERRCODE_INVALID_DATETIME_FORMAT:
+			case ERRCODE_INVALID_ESCAPE_CHARACTER:
+			case ERRCODE_INVALID_ESCAPE_SEQUENCE:
+			case ERRCODE_NONSTANDARD_USE_OF_ESCAPE_CHARACTER:
+			case ERRCODE_INVALID_PARAMETER_VALUE:
+			case ERRCODE_INVALID_TABLESAMPLE_ARGUMENT:
+			case ERRCODE_INVALID_TIME_ZONE_DISPLACEMENT_VALUE:
+			case ERRCODE_NULL_VALUE_NOT_ALLOWED:
+			case ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE:
+			case ERRCODE_SEQUENCE_GENERATOR_LIMIT_EXCEEDED:
+			case ERRCODE_STRING_DATA_LENGTH_MISMATCH:
+			case ERRCODE_STRING_DATA_RIGHT_TRUNCATION:
+			case ERRCODE_INVALID_TEXT_REPRESENTATION:
+			case ERRCODE_INVALID_BINARY_REPRESENTATION:
+			case ERRCODE_BAD_COPY_FILE_FORMAT:
+			case ERRCODE_UNTRANSLATABLE_CHARACTER:
+			case ERRCODE_DUPLICATE_JSON_OBJECT_KEY_VALUE:
+			case ERRCODE_INVALID_ARGUMENT_FOR_SQL_JSON_DATETIME_FUNCTION:
+			case ERRCODE_INVALID_JSON_TEXT:
+			case ERRCODE_INVALID_SQL_JSON_SUBSCRIPT:
+			case ERRCODE_MORE_THAN_ONE_SQL_JSON_ITEM:
+			case ERRCODE_NO_SQL_JSON_ITEM:
+			case ERRCODE_NON_NUMERIC_SQL_JSON_ITEM:
+			case ERRCODE_NON_UNIQUE_KEYS_IN_A_JSON_OBJECT:
+			case ERRCODE_SINGLETON_SQL_JSON_ITEM_REQUIRED:
+			case ERRCODE_SQL_JSON_ARRAY_NOT_FOUND:
+			case ERRCODE_SQL_JSON_MEMBER_NOT_FOUND:
+			case ERRCODE_SQL_JSON_NUMBER_NOT_FOUND:
+			case ERRCODE_SQL_JSON_OBJECT_NOT_FOUND:
+			case ERRCODE_TOO_MANY_JSON_ARRAY_ELEMENTS:
+			case ERRCODE_TOO_MANY_JSON_OBJECT_MEMBERS:
+			case ERRCODE_SQL_JSON_SCALAR_REQUIRED:
+			case ERRCODE_SQL_JSON_ITEM_CANNOT_BE_CAST_TO_TARGET_TYPE:
+				sfcstate->errors++;
+				if (sfcstate->errors <= 100)
+					ereport(WARNING,
+							(errcode(errdata->sqlerrcode),
+							errmsg("%s", errdata->context)));
+
+				sfcstate->begin_subxact = true;
+
+				break;
+			default:
+				PG_RE_THROW();
+		}
+
+		FlushErrorState();
+		FreeErrorData(errdata);
+		errdata = NULL;
+
+		MemoryContextSwitchTo(ecxt);
+	}
+	PG_END_TRY();
+
+	return true;
+}
+
 /*
  * Read next tuple from file for COPY FROM. Return false if no more tuples.
  *
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index b5ab9d9c9a..b49954c0aa 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -808,7 +808,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);

-	IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
+	IDENTITY_P IF_P IGNORE_ERRORS ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
 	INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
 	INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
 	INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -3477,6 +3477,10 @@ copy_opt_item:
 				{
 					$$ = makeDefElem("freeze", (Node *) makeBoolean(true), @1);
 				}
+			| IGNORE_ERRORS
+				{
+					$$ = makeDefElem("ignore_errors", (Node *)makeBoolean(true), @1);
+				}
 			| DELIMITER opt_as Sconst
 				{
 					$$ = makeDefElem("delimiter", (Node *) makeString($3), @1);
@@ -17778,6 +17782,7 @@ unreserved_keyword:
 			| HOUR_P
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| IMMEDIATE
 			| IMMUTABLE
 			| IMPLICIT_P
@@ -18357,6 +18362,7 @@ bare_label_keyword:
 			| HOLD
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| ILIKE
 			| IMMEDIATE
 			| IMMUTABLE
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 62a39779b9..fe590ff7a8 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -2748,7 +2748,8 @@ psql_completion(const char *text, int start, int end)
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "("))
 		COMPLETE_WITH("FORMAT", "FREEZE", "DELIMITER", "NULL",
 					  "HEADER", "QUOTE", "ESCAPE", "FORCE_QUOTE",
-					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING");
+					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING",
+					  "IGNORE_ERRORS");
 
index cb0096aeb6..006e1024e1 100644
@@ -76,6 +77,8 @@ extern CopyFromState BeginCopyFrom(ParseState *pstate, Relation rel, Node *where
 								   bool is_program, copy_data_source_cb data_source_cb, List *attnamelist, List *options);
index e37c6032ae..9100d5f247 100644

 /*
  * Represents the different source cases we need to worry about at
@@ -49,6 +50,24 @@ typedef enum CopyInsertMethod
 	CIM_MULTI_CONDITIONAL		/* use table_multi_insert only if valid */
 } CopyInsertMethod;
 
+/*
+ * Struct that holding fields for ignore_errors option
+ */
+typedef struct SafeCopyFromState
+{
+#define			REPLAY_BUFFER_SIZE 1000
+	HeapTuple	   *replay_buffer; 			/* accumulates tuples for replaying it after an error */
+	int				saved_tuples;			/* # of tuples in replay_buffer */
+	int 			replayed_tuples;		/* # of tuples was replayed from buffer */
+	int				errors;					/* total # of errors */
+	bool			replay_is_active;		/* if true we replay tuples from buffer */
+	bool			begin_subxact;			/* if true we can begin subtransaction */
+
+	MemoryContext	replay_cxt;
+	MemoryContext	oldcontext;
+	ResourceOwner	oldowner;
+} SafeCopyFromState;
+
 /*
  * This struct contains all the state variables used throughout a COPY FROM
  * operation.
@@ -71,6 +90,7 @@ typedef struct CopyFromStateData
 	char	   *filename;		/* filename, or NULL for STDIN */
 	bool		is_program;		/* is 'filename' a program to popen? */
 	copy_data_source_cb data_source_cb; /* function for reading data */
+	SafeCopyFromState *sfcstate; /* struct for ignore_errors option */
 
 	CopyFormatOptions opts;
index 5f3685e9ef..acf4917e64 100644
@@ -649,6 +649,135 @@ SELECT * FROM instead_of_insert_tbl;
 (2 rows)
 
 COMMIT;
+-- tests for IGNORE_ERRORS option
+-- CIM_MULTI case
+CREATE TABLE check_ign_err (n int, m int[], k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 8 | {8} | 8
+(2 rows)
+
+-- CIM_SINGLE cases
+-- BEFORE row trigger
+TRUNCATE check_ign_err;
+CREATE TABLE trig_test(n int, m int[], k int);
+CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 8 | {8} | 8
+(2 rows)
+
+DROP TRIGGER trig_before on check_ign_err;
+-- INSTEAD OF row trigger
+TRUNCATE check_ign_err;
+TRUNCATE trig_test;
+CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
+CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
+COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err_view, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err_view, line 3: "3	{3}"
+WARNING:  COPY check_ign_err_view, line 4, column n: "a"
+WARNING:  COPY check_ign_err_view, line 5, column k: "5555555555"
+WARNING:  COPY check_ign_err_view, line 6, column n: ""
+WARNING:  COPY check_ign_err_view, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM trig_test;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 8 | {8} | 8
+(2 rows)
+
+DROP TRIGGER trig_instead_of ON check_ign_err_view;
+DROP VIEW check_ign_err_view;
+-- foreign table case is in postgres_fdw extension
+-- volatile function in WHERE clause
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
+  WHERE n = floor(random()*(1-1+1))+1; /* finds values equal 1 */
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+(1 row)
+
+DROP TABLE check_ign_err;
+-- CIM_MULTI_CONDITIONAL case
+-- INSERT triggers for partition tables
+TRUNCATE trig_test;
+CREATE TABLE check_ign_err (n int, m int[], k int)
+  PARTITION BY RANGE (k);
+CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
+  FOR VALUES FROM (1) TO (4);
+CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
+  FOR VALUES FROM (4) TO (9);
+CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+WARNING:  COPY check_ign_err, line 2: "2	{2}	2	2"
+WARNING:  COPY check_ign_err, line 3: "3	{3}"
+WARNING:  COPY check_ign_err, line 4, column n: "a"
+WARNING:  COPY check_ign_err, line 5, column k: "5555555555"
+WARNING:  COPY check_ign_err, line 6, column n: ""
+WARNING:  COPY check_ign_err, line 7, column m: "{a, 7}"
+WARNING:  6 errors
+SELECT * FROM check_ign_err;
+ n |  m  | k 
+---+-----+---
+ 1 | {1} | 1
+ 8 | {8} | 8
+(2 rows)
+
+DROP TRIGGER trig_before_part on check_ign_err;
+DROP TABLE trig_test;
+DROP TABLE check_ign_err CASCADE;
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;
diff --git a/src/test/regress/sql/copy2.sql b/src/test/regress/sql/copy2.sql
index b3c16af48e..b25b20182e 100644
--- a/src/test/regress/sql/copy2.sql
+++ b/src/test/regress/sql/copy2.sql
@@ -454,6 +454,122 @@ test1
 SELECT * FROM instead_of_insert_tbl;
 COMMIT;
 
+-- tests for IGNORE_ERRORS option
+-- CIM_MULTI case
+CREATE TABLE check_ign_err (n int, m int[], k int);
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+
+-- CIM_SINGLE cases
+-- BEFORE row trigger
+TRUNCATE check_ign_err;
+CREATE TABLE trig_test(n int, m int[], k int);
+CREATE FUNCTION fn_trig_before () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+DROP TRIGGER trig_before on check_ign_err;
+
+-- INSTEAD OF row trigger
+TRUNCATE check_ign_err;
+TRUNCATE trig_test;
+CREATE VIEW check_ign_err_view AS SELECT * FROM check_ign_err;
+CREATE FUNCTION fn_trig_instead_of () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m, NEW.k);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_instead_of INSTEAD OF INSERT ON check_ign_err_view
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_instead_of();
+COPY check_ign_err_view FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM trig_test;
+DROP TRIGGER trig_instead_of ON check_ign_err_view;
+DROP VIEW check_ign_err_view;
+
+-- foreign table case is in postgres_fdw extension
+
+-- volatile function in WHERE clause
+TRUNCATE check_ign_err;
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS
+  WHERE n = floor(random()*(1-1+1))+1; /* finds values equal 1 */
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+DROP TABLE check_ign_err;
+
+-- CIM_MULTI_CONDITIONAL case
+-- INSERT triggers for partition tables
+TRUNCATE trig_test;
+CREATE TABLE check_ign_err (n int, m int[], k int)
+  PARTITION BY RANGE (k);
+CREATE TABLE check_ign_err_part1 PARTITION OF check_ign_err
+  FOR VALUES FROM (1) TO (4);
+CREATE TABLE check_ign_err_part2 PARTITION OF check_ign_err
+  FOR VALUES FROM (4) TO (9);
+CREATE FUNCTION fn_trig_before_part () RETURNS TRIGGER AS '
+  BEGIN
+    INSERT INTO trig_test VALUES(NEW.n, NEW.m);
+    RETURN NEW;
+  END;
+' LANGUAGE plpgsql;
+CREATE TRIGGER trig_before_part BEFORE INSERT ON check_ign_err
+FOR EACH ROW EXECUTE PROCEDURE fn_trig_before_part();
+COPY check_ign_err FROM STDIN WITH IGNORE_ERRORS WHERE n < 9;
+1	{1}	1
+2	{2}	2	2
+3	{3}
+a	{4}	4
+5	{5}	5555555555
+
+7	{a, 7}	7
+8	{8}	8
+\.
+SELECT * FROM check_ign_err;
+DROP TRIGGER trig_before_part on check_ign_err;
+DROP TABLE trig_test;
+DROP TABLE check_ign_err CASCADE;
+
 -- clean up
 DROP TABLE forcetest;
 DROP TABLE vistest;

---

From: torikoshia <[email-masked]>
Date: Mon, 26 Sep 2022 22:04:36 +0900

On 2022-09-21 21:11, Damir Belyalov wrote:
Thanks for updating patch.

> In the previous patch there was an error when processing constraints.
> The patch was fixed, but the code grew up and became more complicated
> You asked why we need subtransactions, so the answer is in this patch.
> When processing a row that does not satisfy constraints or INSTEAD OF
> triggers, it is necessary to rollback the subtransaction and return
> the table to its original state.
> Cause of complexity, I had to abandon the constraints, triggers
> processing in and handle only errors that occur when reading the file.
> Attaching simplified patch (0006-COPY_IGNORE_ERRORS).

Do you mean you stop dealing with errors concerned with constraints and 
triggers and we should review 0006-COPY_IGNORE_ERRORS?

> Tried to implement your error and could not. The result was the same
> as COPY FROM implements.

Hmm, I applied v6 patch and when canceled COPY by sending SIGINT(ctrl + 
C), I faced the same situation as below.
I tested it on CentOS 8.4.

   =# COPY test FROM '/home/tori/pgsql/master/10000000.data' WITH 
(IGNORE_ERRORS);
   ^CCancel request sent
   ERROR:  canceling statement due to user request
   CONTEXT:  COPY test, line 628000: "628000       xxx"

   =# SELECT 1;

  =# ABORT;
   FATAL:  UserAbortTransactionBlock: unexpected state STARTED
   server closed the connection unexpectedly
           This probably means the server terminated abnormally
           before or while processing the request.
   The connection to the server was lost. Attempting reset: Succeeded.

I did the same procedure on COPY FROM without IGNORE_ERRORS and didn't 
face this situation.

-- 
Regards,

--
Atsushi Torikoshi
NTT DATA CORPORATION

---

From: Damir Belyalov <[email-masked]>
Date: Thu, 29 Sep 2022 16:18:51 +0300

> triggers and we should review 0006-COPY_IGNORE_ERRORS?
Yes, this patch is simpler and I think it's worth adding it first.

> I tested it on CentOS 8.4.
Thank you for pointing out this error. it really needs to be taken into
account. In the previous  0006 patch, there were 2 stages in one
subtransaction - filling the buffer and 'replay mode' (reading from the
buffer). Since only NextCopyFrom() is included in PG_TRY(), and the
ERRCODE_QUERY_CANCELED error can occur anywhere, it is impossible to catch
it and rollback the subtransaction.

I changed the 0006 patch and fixed this error and now only the 'replay
buffer filling' is made in the subtransaction.

Patch 0005 (that processed constraints) needs to be finalized, because it
requires subtransactions to rollback constraints, triggers. Therefore, it
is not possible to fix it yet. There is a decision to put for(;;) loop in
PG_TRY. It will solve the problem, but the code will be too complicated.
diff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml
index c25b52d0cb..22c992e6f6 100644
--- a/doc/src/sgml/ref/copy.sgml
+++ b/doc/src/sgml/ref/copy.sgml
@@ -34,6 +34,7 @@ COPY { <replaceable class="parameter">table_name</replaceable> [ ( <replaceable

+   <varlistentry>
index e8bb168aea..caa3375758 100644

+static bool FillReplayBuffer(CopyFromState cstate, ExprContext *econtext,
+							 TupleTableSlot *myslot);
+
 /*
  * error context callback for COPY FROM
  *
@@ -521,6 +524,177 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
 	miinfo->bufferedBytes += tuplen;
 }
 
+/*
+ * Fills replay_buffer for safe copying, enables by IGNORE_ERRORS option.
+ */
+bool
+FillReplayBuffer(CopyFromState cstate, ExprContext *econtext, TupleTableSlot *myslot)
+{
+		while (sfcstate->saved_tuples < REPLAY_BUFFER_SIZE)
+			bool tuple_is_valid = false;
+					tuple_is_valid = true;
+				Assert(IsSubTransaction());
+						/* If the error can be processed, begin a new subtransaction */
+			if (tuple_is_valid)
+				HeapTuple saved_tuple;
+				saved_tuple = heap_form_tuple(RelationGetDescr(cstate->rel), myslot->tts_values, myslot->tts_isnull);
+			MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
+		/* End of file or buffer was filled, prepare to replay remaining tuples from buffer */
+			heap_deform_tuple(sfcstate->replay_buffer[sfcstate->replayed_tuples++], RelationGetDescr(cstate->rel),
+							  myslot->tts_values, myslot->tts_isnull);
+			/* All tuples from buffer were replayed, clean it up */
+			sfcstate->replay_buffer = MemoryContextAlloc(cstate->sfcstate->replay_cxt,
+				/* All tuples were replayed */
@@ -855,9 +1029,19 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);
 
+			if (!FillReplayBuffer(cstate, econtext, myslot))

@@ -1550,6 +1734,25 @@ BeginCopyFrom(ParseState *pstate,
 		cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *));
 	}
 
+		cstate->sfcstate->oldowner = CurrentResourceOwner;
+		cstate->sfcstate->oldcontext = cstate->copycontext;
 
 	return cstate;

 
index e37c6032ae..d3f4c8d9df 100644

 /*
 
+/*
 
 	CopyFormatOptions opts;
 
 COMMIT;

---

From: Damir Belyalov <[email-masked]>
Date: Mon, 17 Oct 2022 13:17:38 +0300

Updated the patch due to conflicts when applying to master.

 
+   <varlistentry>
index db4c9dbc23..d04753a4c8 100644
index a079c70152..fa169d2cf4 100644
@@ -107,6 +107,9 @@ static char *limit_printout_length(const char *str);

@@ -625,6 +628,177 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
 
+/*
@@ -985,9 +1159,19 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);

 
@@ -1695,6 +1879,25 @@ BeginCopyFrom(ParseState *pstate,
 		cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *));
 	}

 	return cstate;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 737bd2d06d..b3a6c9931e 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -702,7 +702,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);

-	IDENTITY_P IF_P ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
+	IDENTITY_P IF_P IGNORE_ERRORS ILIKE IMMEDIATE IMMUTABLE IMPLICIT_P IMPORT_P IN_P INCLUDE
 	INCLUDING INCREMENT INDEX INDEXES INHERIT INHERITS INITIALLY INLINE_P
 	INNER_P INOUT INPUT_P INSENSITIVE INSERT INSTEAD INT_P INTEGER
 	INTERSECT INTERVAL INTO INVOKER IS ISNULL ISOLATION
@@ -3359,6 +3359,10 @@ copy_opt_item:
 				{
 					$$ = makeDefElem("freeze", (Node *) makeBoolean(true), @1);
 				}
+			| IGNORE_ERRORS
+				{
+					$$ = makeDefElem("ignore_errors", (Node *)makeBoolean(true), @1);
+				}
 			| DELIMITER opt_as Sconst
 				{
 					$$ = makeDefElem("delimiter", (Node *) makeString($3), @1);
@@ -16756,6 +16760,7 @@ unreserved_keyword:
 			| HOUR_P
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| IMMEDIATE
 			| IMMUTABLE
 			| IMPLICIT_P
@@ -17310,6 +17315,7 @@ bare_label_keyword:
 			| HOLD
 			| IDENTITY_P
 			| IF_P
+			| IGNORE_ERRORS
 			| ILIKE
 			| IMMEDIATE
 			| IMMUTABLE
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 584d9d5ae6..33d583a94c 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -2757,7 +2757,8 @@ psql_completion(const char *text, int start, int end)
 	else if (Matches("COPY|\\copy", MatchAny, "FROM|TO", MatchAny, "WITH", "("))
 		COMPLETE_WITH("FORMAT", "FREEZE", "DELIMITER", "NULL",
 					  "HEADER", "QUOTE", "ESCAPE", "FORCE_QUOTE",
-					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING");
+					  "FORCE_NOT_NULL", "FORCE_NULL", "ENCODING",
+					  "IGNORE_ERRORS");
 
index b77b935005..0bf9641b6e 100644
index 8d9cc5accd..3289d96872 100644

 /*
  * Represents the different source cases we need to worry about at
@@ -52,6 +53,23 @@ typedef enum CopyInsertMethod
 								 * ExecForeignBatchInsert only if valid */
 } CopyInsertMethod;
 
+/*
@@ -74,6 +92,7 @@ typedef struct CopyFromStateData
 
 	CopyFormatOptions opts;
index 957ee18d84..ed25a8c86c 100644
 
 COMMIT;

---

From: Damir Belyalov <[email-masked]>
Date: Wed, 2 Nov 2022 11:46:03 +0300

Updated the patch:
- Optimized and simplified logic of IGNORE_ERRORS
- Changed variable names to more understandable ones
- Added an analogue of MAX_BUFFERED_BYTES for safe buffer

Regards,
Damir Belyalov
Postgres Professional

 
+   <varlistentry>
index a079c70152..846eac022d 100644

+static bool SafeCopying(CopyFromState cstate, ExprContext *econtext,
+							 TupleTableSlot *myslot);
+
 /*
  * error context callback for COPY FROM
  *
@@ -625,6 +628,175 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
 	miinfo->bufferedBytes += tuplen;
 }
 
+/*
+ * Safely reads source data, converts to a tuple and fills tuple buffer.
+ * Skips some data in the case of failed conversion if data source for
+ * a next tuple can be surely read without a danger.
+ */
+bool
+SafeCopying(CopyFromState cstate, ExprContext *econtext, TupleTableSlot *myslot)
+{
+	SafeCopyFromState  *sfcstate = cstate->sfcstate;
+	bool 				valid_row = true;
+
+	/* Standard COPY if IGNORE_ERRORS is disabled */
+	if (!cstate->sfcstate)
+		/* Directly stores the values/nulls array in the slot */
+		return NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull);
+
+	if (sfcstate->replayed_tuples < sfcstate->saved_tuples)
+	{
+		Assert(sfcstate->saved_tuples > 0);
+
+		/* Prepare to replay the tuple */
+		heap_deform_tuple(sfcstate->safe_buffer[sfcstate->replayed_tuples++], RelationGetDescr(cstate->rel),
+						  myslot->tts_values, myslot->tts_isnull);
+		return true;
+	}
+	else
+	{
+		/* All tuples from buffer were replayed, clean it up */
+		MemoryContextReset(sfcstate->safe_cxt);
+
+		sfcstate->saved_tuples = sfcstate->replayed_tuples = 0;
+		sfcstate->safeBufferBytes = 0;
+	}
+
+	BeginInternalSubTransaction(NULL);
+	CurrentResourceOwner = sfcstate->oldowner;
+
+	while (sfcstate->saved_tuples < SAFE_BUFFER_SIZE &&
+		   sfcstate->safeBufferBytes < MAX_SAFE_BUFFER_BYTES)
+	{
+		bool	tuple_is_valid = true;
+
+		PG_TRY();
+		{
+			MemoryContext cxt = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
+
+			valid_row = NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull);
+			tuple_is_valid = valid_row;
+
+			if (valid_row)
+				sfcstate->safeBufferBytes += cstate->line_buf.len;
+
+			CurrentMemoryContext = cxt;
+		}
+		PG_CATCH();
+		{
+			MemoryContext ecxt = MemoryContextSwitchTo(sfcstate->oldcontext);
+			ErrorData 	 *errdata = CopyErrorData();
+
+			tuple_is_valid = false;
+
+			Assert(IsSubTransaction());
+
+			RollbackAndReleaseCurrentSubTransaction();
+			CurrentResourceOwner = sfcstate->oldowner;
+
+			switch (errdata->sqlerrcode)
+			{
+				/* Ignore data exceptions */
+				case ERRCODE_CHARACTER_NOT_IN_REPERTOIRE:
+				case ERRCODE_DATA_EXCEPTION:
+				case ERRCODE_ARRAY_ELEMENT_ERROR:
+				case ERRCODE_DATETIME_VALUE_OUT_OF_RANGE:
+				case ERRCODE_INTERVAL_FIELD_OVERFLOW:
+				case ERRCODE_INVALID_CHARACTER_VALUE_FOR_CAST:
+				case ERRCODE_INVALID_DATETIME_FORMAT:
+				case ERRCODE_INVALID_ESCAPE_CHARACTER:
+				case ERRCODE_INVALID_ESCAPE_SEQUENCE:
+				case ERRCODE_NONSTANDARD_USE_OF_ESCAPE_CHARACTER:
+				case ERRCODE_INVALID_PARAMETER_VALUE:
+				case ERRCODE_INVALID_TABLESAMPLE_ARGUMENT:
+				case ERRCODE_INVALID_TIME_ZONE_DISPLACEMENT_VALUE:
+				case ERRCODE_NULL_VALUE_NOT_ALLOWED:
+				case ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE:
+				case ERRCODE_SEQUENCE_GENERATOR_LIMIT_EXCEEDED:
+				case ERRCODE_STRING_DATA_LENGTH_MISMATCH:
+				case ERRCODE_STRING_DATA_RIGHT_TRUNCATION:
+				case ERRCODE_INVALID_TEXT_REPRESENTATION:
+				case ERRCODE_INVALID_BINARY_REPRESENTATION:
+				case ERRCODE_BAD_COPY_FILE_FORMAT:
+				case ERRCODE_UNTRANSLATABLE_CHARACTER:
+				case ERRCODE_DUPLICATE_JSON_OBJECT_KEY_VALUE:
+				case ERRCODE_INVALID_ARGUMENT_FOR_SQL_JSON_DATETIME_FUNCTION:
+				case ERRCODE_INVALID_JSON_TEXT:
+				case ERRCODE_INVALID_SQL_JSON_SUBSCRIPT:
+				case ERRCODE_MORE_THAN_ONE_SQL_JSON_ITEM:
+				case ERRCODE_NO_SQL_JSON_ITEM:
+				case ERRCODE_NON_NUMERIC_SQL_JSON_ITEM:
+				case ERRCODE_NON_UNIQUE_KEYS_IN_A_JSON_OBJECT:
+				case ERRCODE_SINGLETON_SQL_JSON_ITEM_REQUIRED:
+				case ERRCODE_SQL_JSON_ARRAY_NOT_FOUND:
+				case ERRCODE_SQL_JSON_MEMBER_NOT_FOUND:
+				case ERRCODE_SQL_JSON_NUMBER_NOT_FOUND:
+				case ERRCODE_SQL_JSON_OBJECT_NOT_FOUND:
+				case ERRCODE_TOO_MANY_JSON_ARRAY_ELEMENTS:
+				case ERRCODE_TOO_MANY_JSON_OBJECT_MEMBERS:
+				case ERRCODE_SQL_JSON_SCALAR_REQUIRED:
+				case ERRCODE_SQL_JSON_ITEM_CANNOT_BE_CAST_TO_TARGET_TYPE:
+					/* If the error can be processed, begin a new subtransaction */
+					BeginInternalSubTransaction(NULL);
+					CurrentResourceOwner = sfcstate->oldowner;
+
+					sfcstate->errors++;
+					if (sfcstate->errors <= 100)
+						ereport(WARNING,
+								(errcode(errdata->sqlerrcode),
+								errmsg("%s", errdata->context)));
+					break;
+				default:
+					MemoryContextSwitchTo(ecxt);
+
+					PG_RE_THROW();
+
+					break;
+			}
+
+			FlushErrorState();
+			FreeErrorData(errdata);
+			errdata = NULL;
+
+			MemoryContextSwitchTo(ecxt);
+		}
+		PG_END_TRY();
+
+		if (tuple_is_valid)
+		{
+			/* Add tuple to safe_buffer in Safe_context */
+			HeapTuple 	  saved_tuple;
+
+			MemoryContextSwitchTo(sfcstate->safe_cxt);
+
+			saved_tuple = heap_form_tuple(RelationGetDescr(cstate->rel), myslot->tts_values, myslot->tts_isnull);
+			sfcstate->safe_buffer[sfcstate->saved_tuples++] = saved_tuple;
+		}
+
+		ExecClearTuple(myslot);
+
+		if (!valid_row)
+			break;
+	}
+
+	ReleaseCurrentSubTransaction();
+	CurrentResourceOwner = sfcstate->oldowner;
+
+	/* Prepare to replay the first tuple from safe_buffer */
+	if (sfcstate->saved_tuples != 0)
+	{
+		heap_deform_tuple(sfcstate->safe_buffer[sfcstate->replayed_tuples++], RelationGetDescr(cstate->rel),
+						  myslot->tts_values, myslot->tts_isnull);
+		return true;
+	}
+
+	/* End of file and nothing to replay? */
+	if (!valid_row && sfcstate->replayed_tuples == sfcstate->saved_tuples)
+		return false;
+
+	return true;
+}
+
 /*
  * Copy FROM file to relation.
  */
@@ -985,8 +1157,8 @@ CopyFrom(CopyFromState cstate)
 
 		ExecClearTuple(myslot);
 
-		/* Directly store the values/nulls array in the slot */
-		if (!NextCopyFrom(cstate, econtext, myslot->tts_values, myslot->tts_isnull))
+		/* Standard copying with option "safe copying" enabled by IGNORE_ERRORS. */
+		if (!SafeCopying(cstate, econtext, myslot))
 			break;
 
 		ExecStoreVirtualTuple(myslot);
@@ -1270,6 +1442,10 @@ CopyFrom(CopyFromState cstate)
 		}
 	}
 
+	if (cstate->sfcstate && cstate->sfcstate->errors > 0)
+		ereport(WARNING,
+				errmsg("Errors: %d", cstate->sfcstate->errors));
+
 	/* Flush any remaining buffered tuples */
 	if (insertMethod != CIM_SINGLE)
 	{
@@ -1695,6 +1871,23 @@ BeginCopyFrom(ParseState *pstate,
 		cstate->raw_fields = (char **) palloc(attr_count * sizeof(char *));
 	}
 
+		cstate->sfcstate->safe_cxt = AllocSetContextCreate(oldcontext,
+									 "Safe_context",
+		cstate->sfcstate->safeBufferBytes = 0;
 
 	return cstate;

 
index 8d9cc5accd..7c65157866 100644

 /*
  * Represents the different source cases we need to worry about at
@@ -52,6 +53,25 @@ typedef enum CopyInsertMethod
 								 * ExecForeignBatchInsert only if valid */
 } CopyInsertMethod;
 
+/*
+ * Struct that holding fields for safe copying option enabled by IGNORE_ERRORS.
+ */
+typedef struct SafeCopyFromState
+{
+#define		SAFE_BUFFER_SIZE	1000
+#define		MAX_SAFE_BUFFER_BYTES	65535
+
+	HeapTuple	safe_buffer[SAFE_BUFFER_SIZE]; 	/* accumulates valid tuples */
+	int			saved_tuples;					/* # of tuples in safe_buffer */
+	int 		replayed_tuples;				/* # of tuples were replayed from buffer */
+	int			safeBufferBytes;				/* # of bytes from all buffered tuples */
+	int			errors;							/* total # of errors */
+
+	MemoryContext	safe_cxt;
+	MemoryContext	oldcontext;
+	ResourceOwner	oldowner;
+} SafeCopyFromState;
+
 /*
  * This struct contains all the state variables used throughout a COPY FROM
  * operation.
@@ -74,6 +94,7 @@ typedef struct CopyFromStateData
 	char	   *filename;		/* filename, or NULL for STDIN */
 	bool		is_program;		/* is 'filename' a program to popen? */
 	copy_data_source_cb data_source_cb; /* function for reading data */
+	SafeCopyFromState *sfcstate; /* struct for ignore_errors option */
 
 	CopyFormatOptions opts;
index 5f3685e9ef..cc6d572cf1 100644
 
 COMMIT;
+WARNING:  Errors: 6
+WARNING:  Errors: 6
+WARNING:  Errors: 6
+WARNING:  Errors: 6
+WARNING:  Errors: 6

---

From: Nikita Malakhov <[email-masked]>
Date: Wed, 7 Dec 2022 14:15:54 +0300

Hi Damir!

Your work looks like a very promising feature for production systems,
where data often needs to be loaded from external sources.

I've looked over the discussion and want to make a proposal -
when we load a bunch of records in database it does not make sense
to output errors to command output, and does not make sense to limit
error output to any number at all, because if we decided to load data
anyway - we would want to have a list (a file) with all records that were
discarded because of errors, with related error information, to, say,
deal with errors and process these records later. It looks like a reasonable
addition to your patch.

As a command output some limited number of error messages has much
less meaning than overall stats - records processed, records loaded,
records discarded, total number of errors.

For example you can look the Oracle SQL Loader feature, I hope this could
give some ideas for further improvements.

-- 
Regards,
Nikita Malakhov
Postgres Professional
https://postgrespro.ru/

---

From: Danil Anisimow <[email-masked]>
Date: Fri, 9 Dec 2022 21:25:34 +0700

Hi!

I have looked at your patch and have a few questions.

110: static bool SafeCopying(CopyFromState cstate, ExprContext *econtext,
111: TupleTableSlot *myslot);
---
636: bool
637: SafeCopying(CopyFromState cstate, ExprContext *econtext,
TupleTableSlot *myslot)

Why is there no static keyword in the definition of the SafeCopying()
function, but it is in the function declaration.

675: MemoryContext cxt =
MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
676:
677: valid_row = NextCopyFrom(cstate, econtext, myslot->tts_values,
myslot->tts_isnull);
678: tuple_is_valid = valid_row;
679:
680: if (valid_row)
681: sfcstate->safeBufferBytes += cstate->line_buf.len;
682:
683: CurrentMemoryContext = cxt;

Why are you using a direct assignment to CurrentMemoryContext instead of
using the MemoryContextSwitchTo function in the SafeCopying() routine?

1160: /* Standard copying with option "safe copying" enabled by
IGNORE_ERRORS. */
1161: if (!SafeCopying(cstate, econtext, myslot))
1162: break;

I checked with GDB that the CurrentMemoryContext changes when SafeCopying
returns. And the target context may be different each time you do a COPY in
psql.

1879: cstate->sfcstate->safe_cxt = AllocSetContextCreate(oldcontext,
1880: "Safe_context",
1881: ALLOCSET_DEFAULT_SIZES);

When you initialize the cstate->sfcstate structure, you create a
cstate->sfcstate->safe_cxt memory context that inherits from oldcontext.
Was it intended to use cstate->copycontext as the parent context here?

> - Added an analogue of MAX_BUFFERED_BYTES for safe buffer

Regards,
Daniil Anisimov
Postgres Professional