0001-poc-copy-from-.-exceptions-to.patch

>From 50f7ab0a503a0d61776add8a138abf2622fc6c35 Mon Sep 17 00:00:00 2001
From: Alex Shulgin <ash@commandprompt.com>
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