text: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   * 
   *   http://www.apache.org/licenses/LICENSE-2.0
   * 
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  
  
  #include "postgres.h"
  #include "fmgr.h"
  #include "funcapi.h"
  
  #include "access/formatter.h"
  #include "catalog/pg_proc.h"
  #include "utils/builtins.h"
  #include "utils/memutils.h"
  #include "utils/typcache.h"
  #include "utils/syscache.h"
  #include "utils/lsyscache.h"
  #include "commands/copy.h"
  #include <unistd.h>
  
  /* Do the module magic dance */
  PG_MODULE_MAGIC;
  PG_FUNCTION_INFO_V1(fixedwidth_out);
  PG_FUNCTION_INFO_V1(fixedwidth_in);
  Datum fixedwidth_out(PG_FUNCTION_ARGS);
  Datum fixedwidth_in(PG_FUNCTION_ARGS);
  
  typedef struct formatConfig
  {
  	/*
  	 * Normaly we would have only one list of structs, each struct containing three fields:
  	 * name, size, index. The reason we use three lists here is because we work with the infrastructure
  	 * function CopyGetAttnums, which expects as input a list of names and returns a list of indexes.
  	 * fldIndexes - holds the index of each field fetched from the file, into the fields description array
  	 * tupdesc->attr[...]
  	 */
  	List       *fldNames;
  	List       *fldSizes;
  	List       *fldIndexes;
  	List       *fldNullsWithBlanks;
  	int         fields_tot_size;
  	
  	/*
  	 * formating parameters
  	 */
  	int         preserve_blanks;
  	char       *null_value;
  	char       *line_delimiter;
  	int         line_delimiter_length;
  	
  	/*
  	 * infrastructure variables required by postgres "type resolution" methods
  	 */
  	FmgrInfo   *conv_functions;
  	Oid        *typioparams;
  	
  } FormatConfig;
  
  typedef struct {
  	int            ncols;
  	Datum         *values;
  	bool          *nulls;
  	int            buflen;
  	bytea         *buffer;
  	StringInfoData one_val;
  	StringInfoData one_field;
  	int            lineno;
  	bool		   convert; 	/* true - perform conversion on column value. false - don't */
  } format_t;
  
  static void 
  init_format_t(format_t** data, int ncolumns, FunctionCallInfo fcinfo)
  {
  	*data            = palloc(sizeof(format_t));
  	(*data)->ncols   = ncolumns;
  	(*data)->values  = palloc(sizeof(Datum) * ncolumns);
  	(*data)->nulls   = palloc(sizeof(bool) * ncolumns);
  	(*data)->lineno  = 1;
  	(*data)->convert = false;
  	initStringInfo( &((*data)->one_val) );
  	initStringInfo( &((*data)->one_field) );
  	
  	FORMATTER_SET_USER_CTX(fcinfo, *data);	
  	
  }
  
  /*
   * extract_field
   *
   * extract a field value from a character string 'data_cursor'. If we
   * preserve blanks, then the entire field_total_length is extracted.
   * Otherwise, we extract all bytes except the trailing blanks. The field
   * value is then stored inside 'output'.
   */
  static void
  extract_field(char *data_cursor, int field_total_length, bool preserve_blanks, StringInfo output)
  {
  	int actual_length;
  	
  	/*
  	 * the actual length of the string we will restore into the database depends whether
  	 * we preserve_blanks or not.
  	 */
  	if (preserve_blanks)
  	{
  		actual_length = field_total_length;
  	}
  	else 
  	{
  		/*
  		 * assume all field characters are blanks
  		 */
  		char *tail = data_cursor + field_total_length - 1;
  		actual_length = 0;
  		
  		while (tail != data_cursor)
  		{
  			if (*tail != ' ')
  			{
  				actual_length = tail - data_cursor + 1;
  				break;
  			}
  			tail--;
  		}
  		
  		if ( (tail == data_cursor) && (*data_cursor != ' ') )
  		{
  			actual_length  = 1;
  		}
  	}
  	
  	/* store the extracted field value */
  	appendBinaryStringInfo(output, data_cursor, actual_length);
  }
  
  static void
  reset_format_in_config(FormatConfig *format_config)
  {
  	format_config->preserve_blanks = 0;
  	format_config->null_value = NULL;
  	format_config->line_delimiter = "\n";
  	format_config->line_delimiter_length = strlen(format_config->line_delimiter);
  	format_config->fldNames = NIL;
  	format_config->fldSizes = NIL;
  	format_config->fldIndexes = NIL;
  	format_config->fldNullsWithBlanks = NIL;
  	format_config->fields_tot_size = 0;
  }
  
  /*
   * load_format_config
   *
   * parse the user specified fixed width keywords. Currently supported
   * keywords are: 'preserve_blanks', 'line_delim' and 'null'. any other
   * unrecognized keyword is treated as a column name (and later on gets
   * verified as a valid column).
   */
  static void
  load_format_config(FormatConfig *format_config, FunctionCallInfo fcinfo)
  {
  	int   i;
  	char *key;
  	char *val;
  	int   args_num = FORMATTER_GET_NUM_ARGS(fcinfo);
  	
  	reset_format_in_config(format_config);
  	
  	for (i = 1; i <= args_num; i++)
  	{
  		key = FORMATTER_GET_NTH_ARG_KEY(fcinfo, i);
  		val = FORMATTER_GET_NTH_ARG_VAL(fcinfo, i);
  		
  		if ( strcasecmp("preserve_blanks", key) == 0)
  		{
  			if ( strcasecmp("on", val) == 0)
  			{
  				format_config->preserve_blanks = 1;
  			}
  		}
  		else if ( strcasecmp("line_delim", key) == 0)
  		{
  			format_config->line_delimiter = val;
  			format_config->line_delimiter_length = strlen(val);
  		}
  		else if ( strcasecmp("null", key) == 0)
  		{
  			format_config->null_value = val;
  		}
  		else
  		{
  			int size = atoi(val);
  			format_config->fldNames = lappend(format_config->fldNames, makeString(key));
  			format_config->fldSizes = lappend_int(format_config->fldSizes, size);
  			format_config->fields_tot_size += size;
  		}		
  	}
  }
  
  /*
   * encoding_check_str
   *
   * for a given string 'str' of length 'len', check if performing
   * an encoding conversion will modify the original string or not
   * and return the answer. The input string remains *unmodified*.
   * While at it, the encoding converter also verifies that the
   * input string is valid in the clinet (external table) encoding.
   */
  static bool
  encoding_check_str(FunctionCallInfo fcinfo, char *str, int len, bool is_import)
  {
  	char	*cvt = NULL;
  
  	FORMATTER_ENCODE_STRING(fcinfo, str, len, cvt, is_import);
  
  	if (cvt != NULL && cvt != str)
  	{
  		pfree(cvt);
  		return true;
  	}
  
  	return false;
  }
  
  /*
   * encoding_encode_strinfo
   *
   * convert a given stringinfo 'strinfo' to the appropriate (pre-defined)
   * encoding (encoding will only be done if really needed).
   */
  static void
  encoding_encode_strinfo(FunctionCallInfo fcinfo, StringInfo strinfo, bool is_import)
  {
  	char	*cvt = NULL;
  
  	FORMATTER_ENCODE_STRING(fcinfo, strinfo->data, strinfo->len, cvt, is_import);
  
  	if (cvt != NULL && cvt != strinfo->data)
  	{
  		/* transfer converted data back to strinfo */
  		resetStringInfo(strinfo);
  		appendBinaryStringInfo(strinfo, cvt, strlen(cvt));
  		pfree(cvt);
  	}
  }
  
  
  static char*
  make_null_val_with_blanks(char *value, int field_size)
  {
  	char *ret;
  	char *cur;
  	int actual_size = field_size + 1;
  	int size = strlen(value);
  	
  	if ( size > field_size)
  	{
  		ereport(ERROR,
  				(errcode(ERRCODE_STRING_DATA_LENGTH_MISMATCH),
  				 errmsg("The size of the NULL value cannot be bigger than the field size")));
  	}
  	
  	ret = (char*)palloc(actual_size);
  	strcpy(ret, value);
  	cur = ret + size;
  	memset(cur, ' ', actual_size - size);
  	ret[actual_size - 1] = '\0';
  	
  	return ret;
  }
  
  /*
   * make_val_with_blanks
   *
   * Pad one string value with blanks, so the size will corespond to the fixedwidth 
   * required by the format. Make sure to encode the string into external table
   * encoding before writing it out (if conversion is needed).
   *
   * Arguments:
   *  value           - the field value in string format 
   *  field_size      - the fixedwidth field size, that is required for the value, for it to be added on the output line
   *  buf             - The temporary field buffer used for field value expansion - so it will reach the fixedwidth size
   *
   * Returns:
   * 	blank padded    - padded value of size field_size, in case value is NULL the return string will contain only blanks
   *  value
   */
  static char*
  make_val_with_blanks(FunctionCallInfo fcinfo, char *value, int field_size, StringInfo buf)
  {
  	int   	sz = 0;
  	
  	if (value)
  	{
  		sz = strlen(value);
  		
  		if (sz > field_size)
  			ereport(ERROR,
  					(errcode(ERRCODE_STRING_DATA_LENGTH_MISMATCH),
  					 errmsg("The size of the value cannot be bigger than the field size value: %s, size: %d, field_size %d",
  							value, sz, field_size)));			
  		
  		appendBinaryStringInfo(buf, value, sz);
  		appendStringInfoFill(buf, field_size - sz, ' ');
  	}
  	else 
  	{
  		appendStringInfoFill(buf, field_size, ' ');
  	}
  	
  	/* assert */
  	if(buf->len != field_size)
  		ereport(ERROR,
  				(errcode(ERRCODE_GP_INTERNAL_ERROR),
  				 errmsg("Internal error in fixed width formatter. size mismatch in field export")));
  
  	/*
  	 * convert value from server encoding to external table encoding. Since
  	 * it is possible that the encoded string will vary in size from the
  	 * original string we must re-adjust if necessary
  	 */
  	encoding_encode_strinfo(fcinfo, buf, false /* export */);
  
  	if(buf->len != field_size)
  	{
  		/* encoded string width was changed. fix it */
  		if(buf->len < field_size)
  		{
  			/* pad missing bytes with blanks */
  			appendStringInfoFill(buf, field_size - buf->len, ' ');
  		}
  		else
  		{
  			/* truncate extra bytes */
  			if (buf->data[field_size - 1] != ' ') /* oh oh... we are truncating user data. don't allow */
  				ereport(ERROR,
  						(errcode(ERRCODE_STRING_DATA_LENGTH_MISMATCH),
  						 errmsg("The size of the value after conversion to external table encoding became bigger than the field size value: %s, size: %d, field_size %d",
  								value, sz, field_size),
  						 errhint("Set the width of this column to a larger value")));
  
  
  			truncateStringInfo(buf, field_size);
  		}
  	}
  
  	return buf->data;
  }
  
  static void
  fill_null_with_blanks_list(FormatConfig *format_in_config)
  {
  	ListCell *curSize;
  	int field_size;
  	
  	foreach(curSize, format_in_config->fldSizes)
  	{
  		field_size = lfirst_int(curSize);
  		format_in_config->fldNullsWithBlanks = lappend(format_in_config->fldNullsWithBlanks, 
  													   makeString(make_null_val_with_blanks(format_in_config->null_value, field_size)));
  	}
  }
  
  /*
   * validate_format_params
   *
   * verifies that every field specified in the table creation list is also present in the formatting string
   * and vice versa
   */
  static void
  validate_format_params(FormatConfig *format_in_config, TupleDesc tupdesc)
  {
  	ListCell   *l;
  	int num_fields_in_format_string = list_length(format_in_config->fldNames);
  	int num_fields_in_table_list = tupdesc->natts;
  	
  	if (num_fields_in_format_string != num_fields_in_table_list)
  	{
  		ereport(ERROR,
  				(errcode(ERRCODE_UNDEFINED_COLUMN),
  				errmsg("the fixed width formatter requires a length specification for each one of the"
  					   " external table columns being used (currently <%d>, however format string has <%d>",
  					   num_fields_in_table_list, num_fields_in_format_string)));
  	}
  	
  	foreach(l, format_in_config->fldNames)
  	{
  		int i;
  		bool is_in_both_lists = false;
  		char *name = strVal(lfirst(l));
  		for (i = 0; i < num_fields_in_table_list; i++)
  		{
  			if (namestrcmp(&(tupdesc->attrs[i]->attname), name) == 0)
  			{
  				is_in_both_lists = true;
  				break;
  			}			
  		}
  		
  		if (is_in_both_lists == false)
  		{
  			ereport(ERROR,
  					(errcode(ERRCODE_UNDEFINED_COLUMN),
  					 errmsg("the fixed width formatter requires a length specification for each one of the "
  							"external table columns being used (missing field <%s>", name)));			
  		}
  	}
  }
  
  static void
  init_format_in_config(FormatConfig *format_in_config, int ncolumns, TupleDesc tupdesc, FunctionCallInfo fcinfo)
  {
  	load_format_config(format_in_config, fcinfo);
  	validate_format_params(format_in_config, tupdesc);
  	
  	if (format_in_config->null_value != NULL)
  	{
  		fill_null_with_blanks_list(format_in_config);
  	}
  	
  	format_in_config->conv_functions = FORMATTER_GET_CONVERSION_FUNCS(fcinfo);
  	format_in_config->typioparams = FORMATTER_GET_TYPIOPARAMS(fcinfo);	
  	format_in_config->fldIndexes = CopyGetAttnums(tupdesc, FORMATTER_GET_RELATION(fcinfo), format_in_config->fldNames);	
  }
  
  static void
  init_format_out_config(FormatConfig *format_out_config, int ncolumns, TupleDesc tupdesc, FunctionCallInfo fcinfo)
  {
  	load_format_config(format_out_config, fcinfo);
  	validate_format_params(format_out_config, tupdesc);
  	
  	if (format_out_config->null_value != NULL)
  	{
  		fill_null_with_blanks_list(format_out_config);
  	}
  	
  	format_out_config->conv_functions = FORMATTER_GET_CONVERSION_FUNCS(fcinfo);	
  	format_out_config->fldIndexes = CopyGetAttnums(tupdesc, FORMATTER_GET_RELATION(fcinfo), format_out_config->fldNames);	
  }
  
  static void 
  get_tuple_info(TupleDesc tupdesc, int *r_ncolumns, format_t **r_myData, char **data, 
  			   FunctionCallInfo fcinfo, FormatConfig *format_out_config)
  {
  	HeapTupleData		tuple;	
  	HeapTupleHeader		rec	= PG_GETARG_HEAPTUPLEHEADER(0);	
  	/* Get our internal description of the formatter */
  	*r_ncolumns = tupdesc->natts;
  	int ncolumns = *r_ncolumns;	
  	*r_myData = (format_t *) FORMATTER_GET_USER_CTX(fcinfo);
  	format_t *myData = *r_myData;
  	
  	if (myData == NULL)
  	{
  		myData          = palloc(sizeof(format_t));
  		*r_myData       = myData;
  		
  		myData->ncols   = ncolumns;
  		myData->values  = palloc(sizeof(Datum) * ncolumns);
  		myData->nulls   = palloc(sizeof(bool) * ncolumns);
  		initStringInfo( &(myData->one_field) );
  		
  		
  		init_format_out_config(format_out_config, ncolumns, tupdesc, fcinfo);
  		
  		/* Determine required buffer size */
  		myData->buflen = format_out_config->fields_tot_size + strlen(format_out_config->line_delimiter);
  		myData->buflen = Max(128, myData->buflen);  /* allocate at least 128 bytes */
  		myData->buffer = palloc(myData->buflen + VARHDRSZ);
  		
  		FORMATTER_SET_USER_CTX(fcinfo, myData);
  	}
  	if (myData->ncols != ncolumns)
  		ereport(ERROR,
  				(errcode(ERRCODE_GP_INTERNAL_ERROR),
  				 errmsg("formatter_export: unexpected change of output record type")));	
  	
  	/* break the input tuple into fields */
  	tuple.t_len = HeapTupleHeaderGetDatumLength(rec);
  	ItemPointerSetInvalid(&(tuple.t_self));
  	tuple.t_data = rec;
  	heap_deform_tuple(&tuple, tupdesc, myData->values, myData->nulls);
  	*data = VARDATA(myData->buffer);
  	
  }
  
  static int
  get_actual_line_size(FormatConfig *format_in_config, char *line_start, int cur_size, int tot_size, PG_FUNCTION_ARGS)
  {
  	int   row_size;
  	int   actual_fields_size;
  	int   remaining;
  	char *line_end; 
  	char *expected_delim_loc = line_start + format_in_config->fields_tot_size;
  	
  	/*
  	 * the case where there is no line delimiter
  	 */
  	if ( 0 == format_in_config->line_delimiter_length )
  	{
  		return format_in_config->fields_tot_size;
  	}
  
  	if ( 1 == format_in_config->line_delimiter_length )
  	{
  		char delim = format_in_config->line_delimiter[0];
  				
  		if ( *expected_delim_loc == delim )
  			line_end = expected_delim_loc;
  		else
  			line_end = strchr(line_start, delim);
  		
  	}
  	else /* > 1 */
  	{
  		int i;
  		bool as_expected = true;
  		
  		for ( i = 0; i < format_in_config->line_delimiter_length; i++)
  		{
  			if ( expected_delim_loc[i] != format_in_config->line_delimiter[i] )
  			{
  				as_expected = false;
  				break;
  			}
  		}
  		
  		if ( as_expected )
  			line_end = expected_delim_loc;
  		else
  			line_end = strstr(line_start, format_in_config->line_delimiter);
  	}
  	
  	/*
  	 * line_end will be 0, if strchr or strstr did not find the delimiter.
  	 * In this case we throw an exception  ( unless this is the last line in the buffer )--> The line delimiter specified in
  	 * FormatConfig must be present in the file.
  	 */
  	if ( 0 == line_end /*did not find delimiter*/ )
  	{
  		remaining = tot_size - cur_size - format_in_config->fields_tot_size;
  		if (1 == remaining) /* we are at the last line so we cannot find a custom delimiter - we have an OS line delimiter here */
  		{
  			return (format_in_config->fields_tot_size + 1);
  		}
  		
  		/*
  		 * this is the case where the last line in the buffer is incomplete, that's why the end of line was not found.
  		 * the rest of the line is in the next buffer
  		 */  
  		if ( (tot_size - cur_size) < format_in_config->fields_tot_size )
  		{
  			return tot_size - cur_size;
  		}
  		
  		/* 
  		 * if we are here, it means the file simply does not contain the line delimiter specified in the formatter string.
  		 * so we throw an exception
  		 */
  		ereport(ERROR,
  				(errcode(ERRCODE_DATA_EXCEPTION),
  	             errmsg("The line delimiter specified in the Formatter arguments: <%s> is not located in the data file", 
  						format_in_config->line_delimiter)));			
  	}
  
  	actual_fields_size = line_end - line_start; 
  	if ( actual_fields_size != format_in_config->fields_tot_size )
  	{
  		int total_actual_field_size = actual_fields_size + format_in_config->line_delimiter_length;
  		
  		FORMATTER_SET_BAD_ROW_DATA(fcinfo, line_start, total_actual_field_size);
  		FORMATTER_SET_BYTE_NUMBER(fcinfo, total_actual_field_size);
  		
  		ereport(ERROR,
  				(errcode(ERRCODE_DATA_EXCEPTION),
  	             errmsg("Expected line size from the formatting string: %d, but the actual size is: %d", 
  						format_in_config->fields_tot_size, actual_fields_size)));			
  	}
  	
  	row_size = actual_fields_size + format_in_config->line_delimiter_length;
  	return row_size;
  }
  
  
  Datum 
  fixedwidth_out(PG_FUNCTION_ARGS)
  {
  	TupleDesc           tupdesc;
  	MemoryContext 		m, oldcontext;
  	int                 ncolumns = 0;
  	format_t           *myData;
  	char               *data;
  	int                 datlen = 0;
  	ListCell           *curIdx; 
  	ListCell           *curSize;
  	int                 field_size;
  	char               *mapped_val;
  	char               *mapped_val_with_blanks;
  	bool		        isnull;
  	Datum		        value;
  	int			        idx;
  	static FormatConfig format_out_config;
  	
  	/* Must be called via the external table format manager */
  	if (!CALLED_AS_FORMATTER(fcinfo))
  		ereport(ERROR,
  				(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
  				 errmsg("fixedwidth_out: not called by format manager")));	
  	
  	
  	tupdesc = FORMATTER_GET_TUPDESC(fcinfo);
  	get_tuple_info(tupdesc, &ncolumns, &myData, &data, fcinfo, &format_out_config);
  	
  	/* =======================================================================
  	 *                            MAIN FORMATTING CODE
  	 * ======================================================================= */
  	m = FORMATTER_GET_PER_ROW_MEM_CTX(fcinfo); 
  	oldcontext = MemoryContextSwitchTo(m); 
  	
  	forboth(curIdx, format_out_config.fldIndexes, curSize, format_out_config.fldSizes)
  	{
  		field_size = lfirst_int(curSize);
  		idx  = lfirst_int(curIdx) - 1;		
  		isnull = myData->nulls[idx];
  		value = myData->values[idx];
  		
  		resetStringInfo(&(myData->one_field));
  		
  		if ( isnull )
  		{
  			mapped_val_with_blanks = make_val_with_blanks(fcinfo, format_out_config.null_value, field_size, &(myData->one_field));
  		}
  		else 
  		{
  			mapped_val = OutputFunctionCall(&format_out_config.conv_functions[idx], value);
  			mapped_val_with_blanks = make_val_with_blanks(fcinfo, mapped_val, field_size, &(myData->one_field));
  		}
  
  		memcpy(&data[datlen], mapped_val_with_blanks, field_size);
  		datlen += field_size;
  	}
  	
  	memcpy(&data[datlen], format_out_config.line_delimiter, format_out_config.line_delimiter_length);
  	datlen += format_out_config.line_delimiter_length;
  	
  	MemoryContextSwitchTo(oldcontext);
  	/* ======================================================================= */
  	
  	SET_VARSIZE(myData->buffer, datlen + VARHDRSZ);
  		
  	PG_RETURN_BYTEA_P(myData->buffer);
  }
  
  /*
   * fixedwidth_in
   * each time this function is called, it builds one tuple from the input data buffer
   */
  Datum 
  fixedwidth_in(PG_FUNCTION_ARGS)
  {	
  	HeapTuple			tuple;
  	TupleDesc           tupdesc;
  	MemoryContext 		m, oldcontext;
  	format_t           *myData;
  	char               *data_buf;
  	int                 ncolumns = 0;
  	int			  		data_cur;
  	int                 data_len;
  	bool                saw_eof;
  	bool				eof_is_lf;
  	ListCell           *curIdx; 
  	ListCell           *curSize;
  	ListCell           *cur_null_with_blanks = NULL;	
  	int		            remaining;
  	int                 field_size;
  	int   				row_size;
  	char               *nullval;
  	int			        idx;
  	char               *null_val_with_blanks;
  	static FormatConfig format_in_config;
  	
  	/* Must be called via the external table format manager */
  	if (!CALLED_AS_FORMATTER(fcinfo))
  		ereport(ERROR,
  				(errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION),
  				 errmsg("fixedwidth_in: not called by format manager")));
  		
  	tupdesc = FORMATTER_GET_TUPDESC(fcinfo);
  	
  	/* Get our internal description of the formatter */
  	ncolumns = tupdesc->natts;	
  	myData = (format_t *) FORMATTER_GET_USER_CTX(fcinfo);
  	
  	if (myData == NULL)
  	{
  		init_format_in_config(&format_in_config, ncolumns, tupdesc, fcinfo);
  		init_format_t(&myData, ncolumns, fcinfo);			
  	}
  	
  	/* start clean */
  	MemSet(myData->values, 0, ncolumns * sizeof(Datum));
  	MemSet(myData->nulls, true, ncolumns * sizeof(bool));
  
  	/* get our input data buf and number of valid bytes in it */
  	data_buf = FORMATTER_GET_DATABUF(fcinfo);
  	data_len = FORMATTER_GET_DATALEN(fcinfo); 
  	data_cur = FORMATTER_GET_DATACURSOR(fcinfo);
  	saw_eof  = FORMATTER_GET_SAW_EOF(fcinfo);
  
  	eof_is_lf = (format_in_config.line_delimiter[0] == '\n' ? true : false);
  
  	/* =======================================================================
  	 *                            MAIN FORMATTING CODE
  	 * ======================================================================= */	
  	/*
  	 * tuple data extraction is done in a separate memory context
  	 */
  	m = FORMATTER_GET_PER_ROW_MEM_CTX(fcinfo); 
  	oldcontext = MemoryContextSwitchTo(m); 
  		
  	/*
  	 * if data_cur == data_len, it means we finished the current buffer, we will not do any formatting,
  	 * instead inside forboth loop we will fall inside "if (remaining < field_size)", so there is NO need to
  	 * set the BAD_ROW_DATA error string ---> there will be no formatting errors that throw exceptions
  	 */
  	if (data_cur < data_len)
  	{
  		/* setting the line number for "line size" exceptions that might be thrown in get_actual_line_size */
  		FORMATTER_SET_BAD_ROW_NUM(fcinfo, myData->lineno); 
  		/*
  		 * myData->lineno represents the line number in the datafile, when the file was opened
  		 * with a conventional editor, so we increase the lineno only when the delimiter is LF
  		 */
  		if (eof_is_lf)
  			myData->lineno++;	
  		row_size = get_actual_line_size(&format_in_config, data_buf + data_cur, data_cur, data_len, fcinfo);
  
  		FORMATTER_SET_BAD_ROW_DATA(fcinfo, data_buf + data_cur, row_size);
  		FORMATTER_SET_BYTE_NUMBER(fcinfo, row_size);
  	}
  	else 
  	{
  		/*
  		 * This line is not finish. Next buffer will bring the remaining of the line.
  		 * So the line number shpuld not grow.
  		 */
  		if (eof_is_lf)
  			myData->lineno--;			
  		
  		MemoryContextSwitchTo(oldcontext);
  		FORMATTER_RETURN_NOTIFICATION(fcinfo, FMT_NEED_MORE_DATA);				
  	}
  	
  	/*
  	 * Encoding of client data to server encoding.
  	 *
  	 * Ideally we would run a conversion over a line of data and be done.
  	 * However, this may change the byte offsets and mess up with the fixed
  	 * width of the input data.
  	 *
  	 * As we want to avoid encoding conversion when necessary (for performance)
  	 * We first run a test on a whole line and see if it passes input encoding
  	 * validation. if not, an error is emitted. if yes, we make a note whether
  	 * the input string was actually modified or not and take note of it in the
  	 * convert boolean. In most cases 'convert' will remain false and we're done.
  	 * In cases where it is true we postpone the actual conversion of values to
  	 * a later stage (per attribute) in order to keep the formatter clean.
  	 */
  	myData->convert = encoding_check_str(fcinfo, data_buf + data_cur, row_size, true);
  
  
  	if (format_in_config.null_value != NULL)
  		cur_null_with_blanks = list_head(format_in_config.fldNullsWithBlanks);
  	
  	forboth(curIdx, format_in_config.fldIndexes, curSize, format_in_config.fldSizes)
  	{
  		remaining	= 0;
  		field_size = lfirst_int(curSize);
  		nullval = format_in_config.null_value;		
  		remaining = data_len - data_cur;
  		
  		if (remaining < field_size)
  		{
  			/*
  			 * we will get here only in the case we are working without a line delimiter. Because "remaining smaller then fieldsize"
  			 * means that our actual line is smaller than expected size, and if we have a line delimiter this problem will be discovered
  			 * in function  get_actual_line_size which is called above.
  			 */
  			
  			if (saw_eof && (remaining > 1))
  			{
  				data_cur += remaining;
  				FORMATTER_SET_DATACURSOR(fcinfo, data_cur);
  				ereport(ERROR,
  						(errcode(ERRCODE_DATA_EXCEPTION),
  						 errmsg("Last line in the file contains an incomplete tuple")));
  				
  			}
  			else if (saw_eof && (remaining == 1))
  			{
  				/* we are in a case of no line delimiter, but the end of the file contains one EOL */
  				data_cur += remaining;
  				FORMATTER_SET_DATACURSOR(fcinfo, data_cur);
  				MemoryContextSwitchTo(oldcontext);
  				FORMATTER_RETURN_NOTIFICATION(fcinfo, FMT_NEED_MORE_DATA);				
  				
  			}			 
  			else 
  			{
  				/*
  				 * This line is not finish. Next buffer will bring the remaining of the line.
  				 * So the line number shpuld not grow.
  				 */
  				if (eof_is_lf)
  					myData->lineno--;			
  				
  				MemoryContextSwitchTo(oldcontext);
  				FORMATTER_RETURN_NOTIFICATION(fcinfo, FMT_NEED_MORE_DATA);				
  			}
  
  		}
  		
  		resetStringInfo(&(myData->one_val));
  		
  		idx  = lfirst_int(curIdx) - 1;
  		
  		if (format_in_config.preserve_blanks == 0)
  		{
  			/* extract field value while ignoring blanks */
  			extract_field(data_buf + data_cur, field_size, false, &(myData->one_val));
  
  			/*
  			 * there are two (2) cases when we set value to null:
  			 * a. there is a null value defined in the formatter arguments, and this value was found in the field
  			 * b. there is no null value defined and the field contained only blanks
  			 */ 
  			if ( !( (nullval != NULL) && (strcmp(myData->one_val.data, nullval) == 0) )  &&
  				 /* we are not in case a */
  				 !( (nullval == NULL) && (myData->one_val.data[0] == '\0') ) )
  				 /* and also not in case b */
  			{			
  				/* perform encoding conversion on field value if needed */
  				if(myData->convert)
  					encoding_encode_strinfo(fcinfo, &(myData->one_val), true);
  
  				myData->values[idx] = InputFunctionCall(&format_in_config.conv_functions[idx],
  											  myData->one_val.data,
  											  format_in_config.typioparams[idx],
  											  tupdesc->attrs[idx]->atttypmod);
  				myData->nulls[idx] = false;
  			}
  		}
  		else 
  		{
  			if (nullval == NULL || cur_null_with_blanks == NULL)
  				ereport(ERROR,
  						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
  						 errmsg("A null_value was not defined. When preserve_blanks is on, a null_value \
  								 must be defined in the formatter arguments string")));
  			
  			/* extract field value while treating blanks as data */
  			extract_field(data_buf + data_cur, field_size, true, &(myData->one_val));
  
  			null_val_with_blanks = strVal(lfirst(cur_null_with_blanks));
  			cur_null_with_blanks = lnext(cur_null_with_blanks);
  			
  			if (strcmp(myData->one_val.data, null_val_with_blanks) != 0)
  			{
  				/* perform encoding conversion on field value if needed */
  				if(myData->convert)
  					encoding_encode_strinfo(fcinfo, &(myData->one_val), true);
  
  				myData->values[idx] = InputFunctionCall(&format_in_config.conv_functions[idx],
  													  myData->one_val.data,
  													  format_in_config.typioparams[idx],
  													  tupdesc->attrs[idx]->atttypmod);
  				myData->nulls[idx] = false;
  			}
  		}
  		data_cur += field_size;
  	}	
  	
  	/*
  	 * go over the line delimiter
  	 */
  	remaining = data_len - data_cur;
  	if (remaining > 1)
  	{
  		data_cur += format_in_config.line_delimiter_length;
  	}
  	else if (remaining == 1)
  	{
  		data_cur += 1;
  	}
  
  	/*
  	 * wrapping up
  	 */
  	MemoryContextSwitchTo(oldcontext);
  	/* ======================================================================= */
  	
  	FORMATTER_SET_DATACURSOR(fcinfo, data_cur);
  	tuple = heap_form_tuple(tupdesc, myData->values, myData->nulls);
  	FORMATTER_SET_TUPLE(fcinfo, tuple);
  	FORMATTER_RETURN_TUPLE(tuple);
  }
  