v11-0001-Copy-code-readjustment-to-support-parallel-copy.patch

From 663a3cbfa0ac84774e2e1a6c0e55da2312badcea Mon Sep 17 00:00:00 2001
From: Vignesh C <vignesh21@gmail.com>
Date: Wed, 2 Dec 2020 08:47:27 +0530
Subject: [PATCH v11 1/6] Copy code readjustment to support parallel copy.

This patch has the copy code slightly readjusted so that the common code is
separated to functions/macros, these functions/macros will be used by the
workers in the parallel copy code of the upcoming patches. EOL removal is moved
from CopyReadLine to CopyReadLineText, this change was required because in case
of parallel copy the record identification and record updation is done in
CopyReadLineText, before record information is updated in shared memory the new
line characters should be removed.
---
 src/backend/commands/copyfrom.c          | 474 +++++++++++++++++--------------
 src/backend/commands/copyfromparse.c     |  90 +++---
 src/include/commands/copyfrom_internal.h |   2 +
 3 files changed, 317 insertions(+), 249 deletions(-)

diff --git a/src/backend/commands/copyfrom.c b/src/backend/commands/copyfrom.c
index 1b14e9a..7f7f720 100644
--- a/src/backend/commands/copyfrom.c
+++ b/src/backend/commands/copyfrom.c
@@ -96,6 +96,8 @@ typedef struct CopyMultiInsertInfo
 static char *limit_printout_length(const char *str);
 
 static void ClosePipeFromProgram(CopyFromState cstate);
+static void PopulateCommonCStateInfo(CopyFromState cstate, TupleDesc tup_desc,
+									 List *attnamelist);
 
 /*
  * error context callback for COPY FROM
@@ -516,32 +518,13 @@ CopyMultiInsertInfoStore(CopyMultiInsertInfo *miinfo, ResultRelInfo *rri,
 }
 
 /*
- * Copy FROM file to relation.
+ * CheckTargetRelValidity
+ *
+ * Check if the relation specified in copy from is valid.
  */
-uint64
-CopyFrom(CopyFromState cstate)
+static void
+CheckTargetRelValidity(CopyFromState cstate)
 {
-	ResultRelInfo *resultRelInfo;
-	ResultRelInfo *target_resultRelInfo;
-	ResultRelInfo *prevResultRelInfo = NULL;
-	EState	   *estate = CreateExecutorState(); /* for ExecConstraints() */
-	ModifyTableState *mtstate;
-	ExprContext *econtext;
-	TupleTableSlot *singleslot = NULL;
-	MemoryContext oldcontext = CurrentMemoryContext;
-
-	PartitionTupleRouting *proute = NULL;
-	ErrorContextCallback errcallback;
-	CommandId	mycid = GetCurrentCommandId(true);
-	int			ti_options = 0; /* start with default options for insert */
-	BulkInsertState bistate = NULL;
-	CopyInsertMethod insertMethod;
-	CopyMultiInsertInfo multiInsertInfo = {0};	/* pacify compiler */
-	uint64		processed = 0;
-	bool		has_before_insert_row_trig;
-	bool		has_instead_insert_row_trig;
-	bool		leafpart_use_multi_insert = false;
-
 	Assert(cstate->rel);
 	Assert(list_length(cstate->range_table) == 1);
 
@@ -579,27 +562,6 @@ CopyFrom(CopyFromState cstate)
 							RelationGetRelationName(cstate->rel))));
 	}
 
-	/*
-	 * If the target file is new-in-transaction, we assume that checking FSM
-	 * for free space is a waste of time.  This could possibly be wrong, but
-	 * it's unlikely.
-	 */
-	if (RELKIND_HAS_STORAGE(cstate->rel->rd_rel->relkind) &&
-		(cstate->rel->rd_createSubid != InvalidSubTransactionId ||
-		 cstate->rel->rd_firstRelfilenodeSubid != InvalidSubTransactionId))
-		ti_options |= TABLE_INSERT_SKIP_FSM;
-
-	/*
-	 * Optimize if new relfilenode was created in this subxact or one of its
-	 * committed children and we won't see those rows later as part of an
-	 * earlier scan or command. The subxact test ensures that if this subxact
-	 * aborts then the frozen rows won't be visible after xact cleanup.  Note
-	 * that the stronger test of exactly which subtransaction created it is
-	 * crucial for correctness of this optimization. The test for an earlier
-	 * scan or command tolerates false negatives. FREEZE causes other sessions
-	 * to see rows they would not see under MVCC, and a false negative merely
-	 * spreads that anomaly to the current session.
-	 */
 	if (cstate->opts.freeze)
 	{
 		/*
@@ -637,7 +599,61 @@ CopyFrom(CopyFromState cstate)
 			ereport(ERROR,
 					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 					 errmsg("cannot perform COPY FREEZE because the table was not created or truncated in the current subtransaction")));
+	}
+}
+
+/*
+ * Copy FROM file to relation.
+ */
+uint64
+CopyFrom(CopyFromState cstate)
+{
+	ResultRelInfo *resultRelInfo;
+	ResultRelInfo *target_resultRelInfo;
+	ResultRelInfo *prevResultRelInfo = NULL;
+	EState	   *estate = CreateExecutorState(); /* for ExecConstraints() */
+	ModifyTableState *mtstate;
+	ExprContext *econtext;
+	TupleTableSlot *singleslot = NULL;
+	MemoryContext oldcontext = CurrentMemoryContext;
 
+	PartitionTupleRouting *proute = NULL;
+	ErrorContextCallback errcallback;
+	CommandId	mycid = GetCurrentCommandId(true);
+	int			ti_options = 0; /* start with default options for insert */
+	BulkInsertState bistate = NULL;
+	CopyInsertMethod insertMethod;
+	CopyMultiInsertInfo multiInsertInfo = {0};	/* pacify compiler */
+	uint64		processed = 0;
+	bool		has_before_insert_row_trig;
+	bool		has_instead_insert_row_trig;
+	bool		leafpart_use_multi_insert = false;
+
+	CheckTargetRelValidity(cstate);
+
+	/*
+	 * If the target file is new-in-transaction, we assume that checking FSM
+	 * for free space is a waste of time.  This could possibly be wrong, but
+	 * it's unlikely.
+	 */
+	if (RELKIND_HAS_STORAGE(cstate->rel->rd_rel->relkind) &&
+		(cstate->rel->rd_createSubid != InvalidSubTransactionId ||
+		 cstate->rel->rd_firstRelfilenodeSubid != InvalidSubTransactionId))
+		ti_options |= TABLE_INSERT_SKIP_FSM;
+
+	/*
+	 * Optimize if new relfilenode was created in this subxact or one of its
+	 * committed children and we won't see those rows later as part of an
+	 * earlier scan or command. The subxact test ensures that if this subxact
+	 * aborts then the frozen rows won't be visible after xact cleanup.  Note
+	 * that the stronger test of exactly which subtransaction created it is
+	 * crucial for correctness of this optimization. The test for an earlier
+	 * scan or command tolerates false negatives. FREEZE causes other sessions
+	 * to see rows they would not see under MVCC, and a false negative merely
+	 * spreads that anomaly to the current session.
+	 */
+	if (cstate->opts.freeze)
+	{
 		ti_options |= TABLE_INSERT_FROZEN;
 	}
 
@@ -1160,30 +1176,13 @@ CopyFrom(CopyFromState cstate)
 }
 
 /*
- * Setup to read tuples from a file for COPY FROM.
- *
- * 'rel': Used as a template for the tuples
- * 'whereClause': WHERE clause from the COPY FROM command
- * 'filename': Name of server-local file to read, NULL for STDIN
- * 'is_program': true if 'filename' is program to execute
- * 'data_source_cb': callback that provides the input data
- * 'attnamelist': List of char *, columns to include. NIL selects all cols.
- * 'options': List of DefElem. See copy_opt_item in gram.y for selections.
+ * PopulateCStateCatalogInfo
  *
- * Returns a CopyFromState, to be passed to NextCopyFrom and related functions.
+ * Populate the cstate catalog information.
  */
-CopyFromState
-BeginCopyFrom(ParseState *pstate,
-			  Relation rel,
-			  Node *whereClause,
-			  const char *filename,
-			  bool is_program,
-			  copy_data_source_cb data_source_cb,
-			  List *attnamelist,
-			  List *options)
+static void
+PopulateCStateCatalogInfo(CopyFromState cstate)
 {
-	CopyFromState	cstate;
-	bool		pipe = (filename == NULL);
 	TupleDesc	tupDesc;
 	AttrNumber	num_phys_attrs,
 				num_defaults;
@@ -1193,158 +1192,8 @@ BeginCopyFrom(ParseState *pstate,
 	Oid			in_func_oid;
 	int		   *defmap;
 	ExprState **defexprs;
-	MemoryContext oldcontext;
 	bool		volatile_defexprs;
 
-	/* Allocate workspace and zero all fields */
-	cstate = (CopyFromStateData *) palloc0(sizeof(CopyFromStateData));
-
-	/*
-	 * We allocate everything used by a cstate in a new memory context. This
-	 * avoids memory leaks during repeated use of COPY in a query.
-	 */
-	cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext,
-												"COPY",
-												ALLOCSET_DEFAULT_SIZES);
-
-	oldcontext = MemoryContextSwitchTo(cstate->copycontext);
-
-	/* Extract options from the statement node tree */
-	ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */, options);
-
-	/* Process the target relation */
-	cstate->rel = rel;
-
-	tupDesc = RelationGetDescr(cstate->rel);
-
-	/* process commmon options or initialization */
-
-	/* Generate or convert list of attributes to process */
-	cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
-
-	num_phys_attrs = tupDesc->natts;
-
-	/* Convert FORCE_NOT_NULL name list to per-column flags, check validity */
-	cstate->opts.force_notnull_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
-	if (cstate->opts.force_notnull)
-	{
-		List	   *attnums;
-		ListCell   *cur;
-
-		attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_notnull);
-
-		foreach(cur, attnums)
-		{
-			int			attnum = lfirst_int(cur);
-			Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
-
-			if (!list_member_int(cstate->attnumlist, attnum))
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
-						 errmsg("FORCE_NOT_NULL column \"%s\" not referenced by COPY",
-								NameStr(attr->attname))));
-			cstate->opts.force_notnull_flags[attnum - 1] = true;
-		}
-	}
-
-	/* Convert FORCE_NULL name list to per-column flags, check validity */
-	cstate->opts.force_null_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
-	if (cstate->opts.force_null)
-	{
-		List	   *attnums;
-		ListCell   *cur;
-
-		attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_null);
-
-		foreach(cur, attnums)
-		{
-			int			attnum = lfirst_int(cur);
-			Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
-
-			if (!list_member_int(cstate->attnumlist, attnum))
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
-						 errmsg("FORCE_NULL column \"%s\" not referenced by COPY",
-								NameStr(attr->attname))));
-			cstate->opts.force_null_flags[attnum - 1] = true;
-		}
-	}
-
-	/* Convert convert_selectively name list to per-column flags */
-	if (cstate->opts.convert_selectively)
-	{
-		List	   *attnums;
-		ListCell   *cur;
-
-		cstate->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
-
-		attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.convert_select);
-
-		foreach(cur, attnums)
-		{
-			int			attnum = lfirst_int(cur);
-			Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
-
-			if (!list_member_int(cstate->attnumlist, attnum))
-				ereport(ERROR,
-						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
-						 errmsg_internal("selected column \"%s\" not referenced by COPY",
-										 NameStr(attr->attname))));
-			cstate->convert_select_flags[attnum - 1] = true;
-		}
-	}
-
-	/* Use client encoding when ENCODING option is not specified. */
-	if (cstate->opts.file_encoding < 0)
-		cstate->file_encoding = pg_get_client_encoding();
-	else
-		cstate->file_encoding = cstate->opts.file_encoding;
-
-	/*
-	 * Set up encoding conversion info.  Even if the file and server encodings
-	 * are the same, we must apply pg_any_to_server() to validate data in
-	 * multibyte encodings.
-	 */
-	cstate->need_transcoding =
-		(cstate->file_encoding != GetDatabaseEncoding() ||
-		 pg_database_encoding_max_length() > 1);
-	/* See Multibyte encoding comment above */
-	cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding);
-
-	cstate->copy_src = COPY_FILE;	/* default */
-
-	cstate->whereClause = whereClause;
-
-	MemoryContextSwitchTo(oldcontext);
-
-	oldcontext = MemoryContextSwitchTo(cstate->copycontext);
-
-	/* Initialize state variables */
-	cstate->reached_eof = false;
-	cstate->eol_type = EOL_UNKNOWN;
-	cstate->cur_relname = RelationGetRelationName(cstate->rel);
-	cstate->cur_lineno = 0;
-	cstate->cur_attname = NULL;
-	cstate->cur_attval = NULL;
-
-	/*
-	 * Set up variables to avoid per-attribute overhead.  attribute_buf and
-	 * raw_buf are used in both text and binary modes, but we use line_buf
-	 * only in text mode.
-	 */
-	initStringInfo(&cstate->attribute_buf);
-	cstate->raw_buf = (char *) palloc(RAW_BUF_SIZE + 1);
-	cstate->raw_buf_index = cstate->raw_buf_len = 0;
-	if (!cstate->opts.binary)
-	{
-		initStringInfo(&cstate->line_buf);
-		cstate->line_buf_converted = false;
-	}
-
-	/* Assign range table, we'll need it in CopyFrom. */
-	if (pstate)
-		cstate->range_table = pstate->p_rtable;
-
 	tupDesc = RelationGetDescr(cstate->rel);
 	num_phys_attrs = tupDesc->natts;
 	num_defaults = 0;
@@ -1422,6 +1271,95 @@ BeginCopyFrom(ParseState *pstate,
 	cstate->defexprs = defexprs;
 	cstate->volatile_defexprs = volatile_defexprs;
 	cstate->num_defaults = num_defaults;
+}
+
+/*
+ * Setup to read tuples from a file for COPY FROM.
+ *
+ * 'rel': Used as a template for the tuples
+ * 'whereClause': WHERE clause from the COPY FROM command
+ * 'filename': Name of server-local file to read, NULL for STDIN
+ * 'is_program': true if 'filename' is program to execute
+ * 'data_source_cb': callback that provides the input data
+ * 'attnamelist': List of char *, columns to include. NIL selects all cols.
+ * 'options': List of DefElem. See copy_opt_item in gram.y for selections.
+ *
+ * Returns a CopyFromState, to be passed to NextCopyFrom and related functions.
+ */
+CopyFromState
+BeginCopyFrom(ParseState *pstate,
+			  Relation rel,
+			  Node *whereClause,
+			  const char *filename,
+			  bool is_program,
+			  copy_data_source_cb data_source_cb,
+			  List *attnamelist,
+			  List *options)
+{
+	CopyFromState cstate;
+	bool		pipe = (filename == NULL);
+	TupleDesc	tupDesc;
+	MemoryContext oldcontext;
+
+	/* Allocate workspace and zero all fields */
+	cstate = (CopyFromStateData *) palloc0(sizeof(CopyFromStateData));
+
+	/*
+	 * We allocate everything used by a cstate in a new memory context. This
+	 * avoids memory leaks during repeated use of COPY in a query.
+	 */
+	cstate->copycontext = AllocSetContextCreate(CurrentMemoryContext,
+												"COPY",
+												ALLOCSET_DEFAULT_SIZES);
+
+	oldcontext = MemoryContextSwitchTo(cstate->copycontext);
+
+	/* Extract options from the statement node tree */
+	ProcessCopyOptions(pstate, &cstate->opts, true /* is_from */ , options);
+
+	/* Process the target relation */
+	cstate->rel = rel;
+
+	tupDesc = RelationGetDescr(cstate->rel);
+
+	/* process commmon options or initialization */
+	PopulateCommonCStateInfo(cstate, tupDesc, attnamelist);
+
+	cstate->copy_src = COPY_FILE;	/* default */
+
+	cstate->whereClause = whereClause;
+
+	MemoryContextSwitchTo(oldcontext);
+
+	oldcontext = MemoryContextSwitchTo(cstate->copycontext);
+
+	/* Initialize state variables */
+	cstate->reached_eof = false;
+	cstate->eol_type = EOL_UNKNOWN;
+	cstate->cur_relname = RelationGetRelationName(cstate->rel);
+	cstate->cur_lineno = 0;
+	cstate->cur_attname = NULL;
+	cstate->cur_attval = NULL;
+
+	/*
+	 * Set up variables to avoid per-attribute overhead.  attribute_buf and
+	 * raw_buf are used in both text and binary modes, but we use line_buf
+	 * only in text mode.
+	 */
+	initStringInfo(&cstate->attribute_buf);
+	cstate->raw_buf = (char *) palloc(RAW_BUF_SIZE + 1);
+	cstate->raw_buf_index = cstate->raw_buf_len = 0;
+	if (!cstate->opts.binary)
+	{
+		initStringInfo(&cstate->line_buf);
+		cstate->line_buf_converted = false;
+	}
+
+	/* Assign range table, we'll need it in CopyFrom. */
+	if (pstate)
+		cstate->range_table = pstate->p_rtable;
+
+	PopulateCStateCatalogInfo(cstate);
 	cstate->is_program = is_program;
 
 	if (data_source_cb)
@@ -1503,6 +1441,110 @@ BeginCopyFrom(ParseState *pstate,
 }
 
 /*
+ * PopulateCommonCStateInfo
+ *
+ * Populates the common variables required for copy from operation. This is a
+ * helper function for BeginCopy function.
+ */
+static void
+PopulateCommonCStateInfo(CopyFromState cstate, TupleDesc tupDesc, List *attnamelist)
+{
+	int			num_phys_attrs;
+
+	/* Generate or convert list of attributes to process */
+	cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
+
+	num_phys_attrs = tupDesc->natts;
+
+	/* Convert FORCE_NOT_NULL name list to per-column flags, check validity */
+	cstate->opts.force_notnull_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
+	if (cstate->opts.force_notnull)
+	{
+		List	   *attnums;
+		ListCell   *cur;
+
+		attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_notnull);
+
+		foreach(cur, attnums)
+		{
+			int			attnum = lfirst_int(cur);
+			Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
+
+			if (!list_member_int(cstate->attnumlist, attnum))
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+						 errmsg("FORCE_NOT_NULL column \"%s\" not referenced by COPY",
+								NameStr(attr->attname))));
+			cstate->opts.force_notnull_flags[attnum - 1] = true;
+		}
+	}
+
+	/* Convert FORCE_NULL name list to per-column flags, check validity */
+	cstate->opts.force_null_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
+	if (cstate->opts.force_null)
+	{
+		List	   *attnums;
+		ListCell   *cur;
+
+		attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.force_null);
+
+		foreach(cur, attnums)
+		{
+			int			attnum = lfirst_int(cur);
+			Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
+
+			if (!list_member_int(cstate->attnumlist, attnum))
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+						 errmsg("FORCE_NULL column \"%s\" not referenced by COPY",
+								NameStr(attr->attname))));
+			cstate->opts.force_null_flags[attnum - 1] = true;
+		}
+	}
+
+	/* Convert convert_selectively name list to per-column flags */
+	if (cstate->opts.convert_selectively)
+	{
+		List	   *attnums;
+		ListCell   *cur;
+
+		cstate->convert_select_flags = (bool *) palloc0(num_phys_attrs * sizeof(bool));
+
+		attnums = CopyGetAttnums(tupDesc, cstate->rel, cstate->opts.convert_select);
+
+		foreach(cur, attnums)
+		{
+			int			attnum = lfirst_int(cur);
+			Form_pg_attribute attr = TupleDescAttr(tupDesc, attnum - 1);
+
+			if (!list_member_int(cstate->attnumlist, attnum))
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
+						 errmsg_internal("selected column \"%s\" not referenced by COPY",
+										 NameStr(attr->attname))));
+			cstate->convert_select_flags[attnum - 1] = true;
+		}
+	}
+
+	/* Use client encoding when ENCODING option is not specified. */
+	if (cstate->opts.file_encoding < 0)
+		cstate->file_encoding = pg_get_client_encoding();
+	else
+		cstate->file_encoding = cstate->opts.file_encoding;
+
+	/*
+	 * Set up encoding conversion info.  Even if the file and server encodings
+	 * are the same, we must apply pg_any_to_server() to validate data in
+	 * multibyte encodings.
+	 */
+	cstate->need_transcoding =
+		(cstate->file_encoding != GetDatabaseEncoding() ||
+		 pg_database_encoding_max_length() > 1);
+	/* See Multibyte encoding comment above */
+	cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding);
+}
+
+/*
  * Clean up storage and release resources for COPY FROM.
  */
 void
diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c
index 34ed3cf..6762f3c 100644
--- a/src/backend/commands/copyfromparse.c
+++ b/src/backend/commands/copyfromparse.c
@@ -118,6 +118,9 @@ static inline bool CopyGetInt32(CopyFromState cstate, int32 *val);
 static inline bool CopyGetInt16(CopyFromState cstate, int16 *val);
 static bool CopyLoadRawBuf(CopyFromState cstate);
 static int	CopyReadBinaryData(CopyFromState cstate, char *dest, int nbytes);
+static void ClearEOLFromCopiedData(CopyFromState cstate, char *copy_line_data,
+								   int copy_line_pos, int *copy_line_size);
+static void ConvertToServerEncoding(CopyFromState cstate);
 
 void
 ReceiveCopyBegin(CopyFromState cstate)
@@ -716,40 +719,60 @@ CopyReadLine(CopyFromState cstate)
 			} while (CopyLoadRawBuf(cstate));
 		}
 	}
-	else
+
+	ConvertToServerEncoding(cstate);
+	return result;
+}
+
+/*
+ * ClearEOLFromCopiedData
+ *
+ * Clear EOL from the copied data.
+ */
+static void
+ClearEOLFromCopiedData(CopyFromState cstate, char *copy_line_data,
+					   int copy_line_pos, int *copy_line_size)
+{
+	/*
+	 * If we didn't hit EOF, then we must have transferred the EOL marker to
+	 * line_buf along with the data.  Get rid of it.
+	 */
+	switch (cstate->eol_type)
 	{
-		/*
-		 * If we didn't hit EOF, then we must have transferred the EOL marker
-		 * to line_buf along with the data.  Get rid of it.
-		 */
-		switch (cstate->eol_type)
-		{
-			case EOL_NL:
-				Assert(cstate->line_buf.len >= 1);
-				Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\n');
-				cstate->line_buf.len--;
-				cstate->line_buf.data[cstate->line_buf.len] = '\0';
-				break;
-			case EOL_CR:
-				Assert(cstate->line_buf.len >= 1);
-				Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\r');
-				cstate->line_buf.len--;
-				cstate->line_buf.data[cstate->line_buf.len] = '\0';
-				break;
-			case EOL_CRNL:
-				Assert(cstate->line_buf.len >= 2);
-				Assert(cstate->line_buf.data[cstate->line_buf.len - 2] == '\r');
-				Assert(cstate->line_buf.data[cstate->line_buf.len - 1] == '\n');
-				cstate->line_buf.len -= 2;
-				cstate->line_buf.data[cstate->line_buf.len] = '\0';
-				break;
-			case EOL_UNKNOWN:
-				/* shouldn't get here */
-				Assert(false);
-				break;
-		}
+		case EOL_NL:
+			Assert(*copy_line_size >= 1);
+			Assert(copy_line_data[copy_line_pos - 1] == '\n');
+			copy_line_data[copy_line_pos - 1] = '\0';
+			(*copy_line_size)--;
+			break;
+		case EOL_CR:
+			Assert(*copy_line_size >= 1);
+			Assert(copy_line_data[copy_line_pos - 1] == '\r');
+			copy_line_data[copy_line_pos - 1] = '\0';
+			(*copy_line_size)--;
+			break;
+		case EOL_CRNL:
+			Assert(*copy_line_size >= 2);
+			Assert(copy_line_data[copy_line_pos - 2] == '\r');
+			Assert(copy_line_data[copy_line_pos - 1] == '\n');
+			copy_line_data[copy_line_pos - 2] = '\0';
+			*copy_line_size -= 2;
+			break;
+		case EOL_UNKNOWN:
+			/* shouldn't get here */
+			Assert(false);
+			break;
 	}
+}
 
+/*
+ * ConvertToServerEncoding
+ *
+ * Convert contents to server encoding.
+ */
+static void
+ConvertToServerEncoding(CopyFromState cstate)
+{
 	/* Done reading the line.  Convert it to server encoding. */
 	if (cstate->need_transcoding)
 	{
@@ -769,8 +792,6 @@ CopyReadLine(CopyFromState cstate)
 
 	/* Now it's safe to use the buffer in error messages */
 	cstate->line_buf_converted = true;
-
-	return result;
 }
 
 /*
@@ -1133,6 +1154,9 @@ not_end_of_copy:
 	 * Transfer any still-uncopied data to line_buf.
 	 */
 	REFILL_LINEBUF;
+	if (!result && !IsHeaderLine())
+		ClearEOLFromCopiedData(cstate, cstate->line_buf.data,
+							   cstate->line_buf.len, &cstate->line_buf.len);
 
 	return result;
 }
diff --git a/src/include/commands/copyfrom_internal.h b/src/include/commands/copyfrom_internal.h
index c15ea80..1cda8de 100644
--- a/src/include/commands/copyfrom_internal.h
+++ b/src/include/commands/copyfrom_internal.h
@@ -17,6 +17,8 @@
 #include "commands/copy.h"
 #include "commands/trigger.h"
 
+#define IsHeaderLine()			(cstate->opts.header_line && cstate->cur_lineno == 1)
+
 /*
  * Represents the different source cases we need to worry about at
  * the bottom level
-- 
1.8.3.1