/***************************************************************************
 * PostgreSQL read
 *
 *	select FileAddress from shc_data fs_data table
 *	mmap_ptr becomes the pointer to the return from SELECT.
 *	core_blk is also teh pointer to the return from SELECT
 *
 * call this routine with get_tpf_rw(), tpfasm.h has a #if to specify SQL path or FILE path
 ***************************************************************************/
int SQL_get_tpf_rw( int32_t fa, unsigned char **core_blk, size_t *st_size, unsigned char *bid, unsigned char *rcc, int *fd, bool hold, unsigned char **mmap_ptr)
{
char *p;
char *c;
int rc;
struct stat statbuf;

PGresult *res;
uint32_t fa2 = htonl(fa);
char *fa_val = (char *)&fa2;
bool hold2 = hold;
char *hold_val = (char *)&hold2;


const char *paramValues[2]  	= {fa_val, hold_val};
int   paramLengths[2]       	= {sizeof(fa2), sizeof(hold2)};
int   paramFormats[2]       	= {1, 1};       // identify parameter as binary

if(SQL_vsn10) {
	paramValues[0]  	= fa_val;
	paramValues[1]  	= NULL; 
	paramLengths[0]       	= sizeof(fa2);
	paramLengths[1]       	= 0; 
	paramFormats[0]       	= 1;  
	paramFormats[1]       	= 0; 
}
int   resultFormat          	= 1;		// binary 
int   nFields = 0;
int   nTuples = 0;
int   blk_size = 0;

	if(bid)
	{
		LOG_TRACE("%s() SQL read data from File Address fa(%i) fa(%08X) fa2 htonl(%i) bid (%02X%02X)",__func__,fa,fa,fa2,*bid,*(bid+1));
	}
	else
	{
		LOG_TRACE("%s() SQL read data from File Address fa(%i) fa(%08X) fa2 htonl(%i) bid (0)",__func__,fa,fa,fa2);
	}

	//LOG_DEBUG("%s() CALL SQL_select_data_procedure1(%i, %i, NULL);",__func__,fa2,hold2);
        //LOG_DEBUG("%s() SELECT data FROM fs_data WHERE file_address = %i","SQL_select_data_procedure1",fa2);

        bool is_data_found = false;

	LOG_DEBUG("%s() SelectDataCommand = %s %i",__func__,SelectDataCommand,fa);
	LOG_DEBUG("%s() SelectDataNParams = %i",__func__,SelectDataNParams);
	if(is_shadow_db) 
	{
		res = PQexecPrepared(shadow_db_conn,
				     SelectDataName,
				     SelectDataNParams,
				     paramValues,
				     paramLengths,
				     paramFormats,
				     resultFormat);
                if (PQresultStatus(res) == PGRES_TUPLES_OK && ( PQnfields(res) != 0 || PQnfields(res) != 0) )
		{
                    is_data_found = true;
                    LOG_DEBUG("SELECT operation succeeded on Shadow DB"); 
                }
                else
                    LOG_ERROR("SELECT failed on Shadow DB: %s", PQerrorMessage(shadow_db_conn ));
	}
	if(is_data_found !=true) 
	{
	        res = PQexecPrepared(conn,
			     SelectDataName,
			     SelectDataNParams,
			     paramValues,
			     paramLengths,
			     paramFormats,
			     resultFormat);
        }
	LOG_DEBUG("%s() res = %s",__func__,PQresStatus(PQresultStatus(res)));

	if ( hold )
	{
		if ( SQL_vsn10 )
		{
			SQL_lock_row(fa);
		}
		printSQL(0);
	}

        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
                LOG_ERROR("SELECT failed: %s", PQerrorMessage(conn));
		LOG_DEBUG("ROLLBACK TRANSACTION AND CHAIN");
                PQclear(res);
		res = PQexec(conn,"ROLLBACK TRANSACTION AND CHAIN");
		LOG_ERROR("INVALID_FILE_ADDRESS %08X",fa);
		rc = ERR_INVALID_FILE_ADDRESS;
		if ( ((uint32_t)fa)>>24 != 0x94) {
				if ( bid != NULL ) {
					LOG_INFO("missing data: %08x %02x %02x", fa, *bid,*(bid+1));
					char str1[8];
					int16_t a=*bid;
					int16_t b=*(bid+1);
					sprintf(str1,"%02x%02x",a,b);
					FA[COUNT]=fa;
					memcpy(REC_ID1[COUNT],str1,5);
					COUNT++;
				} else {
					bid=0x0000;
					LOG_INFO("missing data: %08x", fa);
					FA[COUNT++]=fa;
					REC_ID1[COUNT][0]=0x00;
				}
			}
		PQclear(res);
        }
	else
	{
		nFields = PQnfields(res);	// number of columns
		nTuples = PQntuples(res);	// number of rows
		//LOG_DEBUG("%s() nFields(cols)=%i nTuples(rows)=%i",__func__,nFields,nTuples);
		if( nFields == 0 || nTuples == 0) 
		{
                	LOG_ERROR("SELECT failed: NO ROWS RETURNED");
                        PQclear(res);
			res = PQexec(conn,"ROLLBACK TRANSACTION AND CHAIN");
			LOG_ERROR("INVALID_FILE_ADDRESS %08X",fa);
			rc = ERR_INVALID_FILE_ADDRESS;
			if ( ((uint32_t)fa)>>24 != 0x94) {
				if ( bid != NULL ) {
					LOG_INFO("missing data: %08x %02x %02x", fa, *bid,*(bid+1));
					char str1[8];
					int16_t a=*bid;
					int16_t b=*(bid+1);
					sprintf(str1,"%02x%02x",a,b);
					FA[COUNT]=fa;
					memcpy(REC_ID1[COUNT],str1,5);
					COUNT++;
				} else {
					bid=0x0000;
					LOG_INFO("missing data: %08x", fa);
					FA[COUNT++]=fa;
					REC_ID1[COUNT][0]=0x00;
				}
			}
			PQclear(res);
		}
		else
		{
			blk_size = ntohl(*(uint32_t *)PQgetvalue (res, 0, 0));
			//LOG_DEBUG("%s() blk_size returned is %i",__func__,blk_size);
			if(blk_size < 127 || blk_size > 4096)
			{
                		LOG_ERROR("SELECT failed: NO ROWS RETURNED");
                                PQclear(res);
				res = PQexec(conn,"ROLLBACK TRANSACTION AND CHAIN");
				LOG_ERROR("INVALID_FILE_ADDRESS %08X",fa);
				rc = ERR_INVALID_FILE_ADDRESS;
				if ( ((uint32_t)fa)>>24 != 0x94) {
					if ( bid != NULL ) {
						LOG_INFO("missing data: %08x %02x %02x", fa, *bid,*(bid+1));
						char str1[8];
						int16_t a=*bid;
						int16_t b=*(bid+1);
						sprintf(str1,"%02x%02x",a,b);
						FA[COUNT]=fa;
						memcpy(REC_ID1[COUNT],str1,5);
						COUNT++;
					} else {
						bid=0x0000;
						LOG_INFO("missing data: %08x", fa);
						FA[COUNT++]=fa;
						REC_ID1[COUNT][0]=0x00;
					}
				}
				PQclear(res);
			}
			else
			{
				*fd = fa;				// use File Address as File Descriptor 
				c = PQgetvalue(res, 0, 1);  		// SELECT has data as second field 
				//dump_hex(c,16);
				//printf(" PQgetlength of data  = %i\n",PQgetlength(res,0,1));
				p = c;	// use same pointer for both mmap and core_blk
	
				//LOG_DEBUG("Core block address %p for FA %08X (RW)", p, fa);
				//dump_hex(p,32);
				/* probably need some validation of block size? */
				if (bid != NULL && memcmp(bid, p, 2) != 0 && memcmp(bid, "\x00\x00", 2) != 0) {
					rc = ERR_RID_MISMATCH;
					int b1 = *(unsigned char *)p;
					int b2 = *(unsigned char *)(p+1);
					int a1 = *(unsigned char *)(bid);
					int a2 = *(unsigned char *)(bid+1);
					LOG_ERROR("Error record id mismatch %02X%02X != %02X%02X FA %08X", b1, b2, a1, a2, fa);
				} else if (rcc != NULL && *rcc != 0 && memcmp(rcc, p+2, 1) != 0) {
					rc = ERR_RCC_MISMATCH;
					int b1 = *(unsigned char *)(p+2);
					int a1 = *(unsigned char *)(rcc);
					LOG_ERROR("Error rcc mismatch %02X != %02X FA %08X", b1, a1, fa);
				} else {
					rc = 0;
				}
				*mmap_ptr = c;
				*core_blk = p;
				*st_size = blk_size;
				//dump_hex(*mmap_ptr,1055);
			}
		}
	}
	//LOG_DEBUG("%s() rc = %i",__func__,rc);
	return rc;
}

/****
sprintf(SelectDataName,"%s","SelectData");
if(SQL_vsn10) {
        sprintf(SelectDataCommand,"%s","SELECT blk_size,data FROM fs_data WHERE file_address = $1");
        SelectDataNParams       = 1;
        SelectDataParamTypes[0] = 23;           // {int}
} else {
        sprintf(SelectDataCommand,"%s","CALL SQL_select_data_procedure($1, $2, NULL, NULL)");
        SelectDataNParams       = 2;
        SelectDataParamTypes[0] = 23;             // {int}
        SelectDataParamTypes[1] = 16;             // {bool}
}
***/

/***********************************************************
 * Connect to PostgreSQL database
 ***********************************************************/
void SQL_init_db_connection(void)
{
PGresult *res;

	LOG_DEBUG("%s() conninfo=%s",__func__,conninfo);	

	if(is_shadow_db)
	{
		shadow_db_conn = PQconnectdb(shadow_db_conn_info);
		if ( PQstatus(shadow_db_conn ) != CONNECTION_OK )
		{
		   LOG_ERROR("Connection to shadow database failed! %s", PQerrorMessage(conn));
		   PQfinish(shadow_db_conn);
		   exit(1);
		}
		res = PQexec(shadow_db_conn, "SET bytea_output = 'escape'");
		LOG_DEBUG("%s() Connection to shadow_shc_data database SUCCESSFUL",__func__);
		// execute_stored_procedure(shadow_db_conn);
	}

	conn = PQconnectdb(conninfo);
	if ( PQstatus(conn) != CONNECTION_OK ) {
		LOG_ERROR("Connection to database failed! %s", PQerrorMessage(conn));
		PQfinish(conn);
		exit(1);
	} else {
		res = PQexec(conn, "SET bytea_output = 'escape'");
		LOG_DEBUG("%s() Connection to shc_data database SUCCESSFUL",__func__);
	}

	res = PQexec(conn, "START TRANSACTION");
	if (PQresultStatus(res) != PGRES_COMMAND_OK)
	{
		LOG_ERROR("START TRANSACTION failed: %s", PQerrorMessage(conn));
		SQL_exit_nicely(conn,res);
	}
	PQclear(res);


	LOG_INFO("PostgreSQL Server Version = %d.%d protocol %i",PQserverVersion(conn)/100,PQserverVersion(conn)%100,PQprotocolVersion(conn));
	if(PQserverVersion(conn)/100 < 1100)
		SQL_vsn10 = 1;

	/*------------------------------------------------------------------------------------------------------------------------------------*/
	sprintf(SelectDataName,"%s","SelectData");
	if(SQL_vsn10) {
		sprintf(SelectDataCommand,"%s","SELECT blk_size,data FROM fs_data WHERE file_address = $1");
		SelectDataNParams    	= 1;
		SelectDataParamTypes[0]	= 23;          	// {int}
	} else {
		sprintf(SelectDataCommand,"%s","CALL SQL_select_data_procedure($1, $2, NULL, NULL)");
		SelectDataNParams    	= 2;
		SelectDataParamTypes[0]	= 23;             // {int}
		SelectDataParamTypes[1]	= 16;             // {bool}
	}

	//-------------------------------------------	
	sprintf(InsertDataName,"%s","InsertData");
	if(SQL_vsn10) {
		sprintf(InsertDataCommand,"%s","INSERT INTO fs_data (file_address, face_type, ordinal, xaddr, recid, blk_size, data) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING data"); 
	} else {
		sprintf(InsertDataCommand,"%s","CALL SQL_insert_data_procedure($1, $2, $3, $4, $5, $6, $7, NULL)");
	}
	InsertDataNParams   	= 7;
	InsertDataParamTypes[0]	= 23;   // {int}
	InsertDataParamTypes[1]	= 23;   // {int}
	InsertDataParamTypes[2]	= 23;   // {int}
	InsertDataParamTypes[3]	= 25;	// {text}
	InsertDataParamTypes[4]	= 25;	// {text}
	InsertDataParamTypes[5]	= 23;	// {int}
	InsertDataParamTypes[6]	= 17;	// {bytea}
	
	//-------------------------------------------	
	sprintf(UpdateDataName,"%s","UpdateData");
	if(SQL_vsn10) {
		sprintf(UpdateDataCommand,"%s","UPDATE fs_data SET data = $1 WHERE file_address = $2"); 
		UpdateDataNParams   	= 2;
		UpdateDataParamTypes[0]	= 17;   // {bytea}
		UpdateDataParamTypes[1]	= 23;   // {int}
	} else {
		sprintf(UpdateDataCommand,"%s","CALL SQL_update_data_procedure($1, $2, $3)");
		UpdateDataNParams   	= 3;
		UpdateDataParamTypes[0]	= 17;   // {bytea}
		UpdateDataParamTypes[1]	= 16;   // {bool}
		UpdateDataParamTypes[2]	= 23;   // {int}
	}
	
	//-------------------------------------------	
	sprintf(SelectSizeName,"%s","SelectSize");
	if(SQL_vsn10) {
		sprintf(SelectSizeCommand,"%s","SELECT size FROM riat WHERE id = $1 AND rtp = $2");
	} else {
		sprintf(SelectSizeCommand,"%s","CALL SQL_select_size_procedure($1, $2, NULL)");
	}
	SelectSizeNParams       = 2;
	SelectSizeParamTypes[0] = 25;  // {text}
	SelectSizeParamTypes[1] = 23;  // {int}

	/*------------------------------------------------------------------------------------------------------------------------------------*/

	// SELECT advisory_lock(FileAddress)
        if(is_shadow_db)
        {
            res = PQprepare(shadow_db_conn,
                        AdvisoryLockDataName,
                        AdvisoryLockDataCommand,
                        AdvisoryLockDataNParams,
                        AdvisoryLockDataParamTypes);
            LOG_DEBUG("%s() PREPARE AdvisoryLockData for shadow db PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
            if(PQresultStatus(res) != PGRES_COMMAND_OK)
            {
                LOG_ERROR("PREPARE failed for shadow db! %s", PQerrorMessage(conn));
                SQL_exit_nicely(conn,res);
            }
            PQclear(res);

        }

	res = PQprepare(conn,
			AdvisoryLockDataName,
			AdvisoryLockDataCommand,
			AdvisoryLockDataNParams,
			AdvisoryLockDataParamTypes);
	LOG_DEBUG("%s() PREPARE AdvisoryLockData PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
	if(PQresultStatus(res) != PGRES_COMMAND_OK)
	{
		LOG_ERROR("PREPARE failed! %s", PQerrorMessage(conn));
		SQL_exit_nicely(conn,res);
	}
	PQclear(res);

	// SELECT advisory_unlock(FileAddress)
        if(is_shadow_db)
        {
            res = PQprepare(shadow_db_conn,
                        AdvisoryUnLockDataName,
                        AdvisoryUnLockDataCommand,
                        AdvisoryUnLockDataNParams,
                        AdvisoryUnLockDataParamTypes);
            LOG_DEBUG("%s() PREPARE AdvisoryUnLockData for shadow db PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
            if(PQresultStatus(res) != PGRES_COMMAND_OK)
            {
                LOG_ERROR("PREPARE failed for shadow db! %s", PQerrorMessage(conn));
                SQL_exit_nicely(conn,res);
            }
            PQclear(res);

        }

	res = PQprepare(conn,
			AdvisoryUnLockDataName,
			AdvisoryUnLockDataCommand,
			AdvisoryUnLockDataNParams,
			AdvisoryUnLockDataParamTypes);
	LOG_DEBUG("%s() PREPARE AdvisoryUnLockData PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
	if(PQresultStatus(res) != PGRES_COMMAND_OK)
	{
		LOG_ERROR("PREPARE failed! %s", PQerrorMessage(conn));
		SQL_exit_nicely(conn,res);
	}
	PQclear(res);

	if ( sql_db )
	{
		/*
		CREATE OR REPLACE PROCEDURE sql_select_size_procedure(id text, rtp integer, INOUT size_data text) 
		LANGUAGE plpgsql 
		AS $$ 
		BEGIN 
		SELECT size FROM riat WHERE id = id AND rtp = rtp INTO size_data; 
		END; 
		$$;
		*/
	
		if(!SQL_vsn10)
		{
                  /***
                   if(is_shadow_db)
                   {
		      res = PQexec(shadow_db_conn," CREATE OR REPLACE PROCEDURE sql_select_size_procedure(hexid text, rtp_in integer, INOUT size_data text) LANGUAGE plpgsql AS $$ BEGIN SELECT size FROM riat WHERE id = hexid AND rtp = rtp_in INTO size_data; END; $$;");
		      LOG_DEBUG("%s() CREATE sql_select_size_procedure for shadow db PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		      if(PQresultStatus(res) != PGRES_COMMAND_OK)
		      {
			LOG_ERROR("CREATE sql_select_size_procedure failed for shadow db! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		      }
		      PQclear(res);

                   }
                   ***/
		res = PQexec(conn," CREATE OR REPLACE PROCEDURE sql_select_size_procedure(hexid text, rtp_in integer, INOUT size_data text) LANGUAGE plpgsql AS $$ BEGIN SELECT size FROM riat WHERE id = hexid AND rtp = rtp_in INTO size_data; END; $$;");
		LOG_DEBUG("%s() CREATE sql_select_size_procedure PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("CREATE sql_select_size_procedure failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);
		}
	
		// SELECT size FROM riat WHERE id = $1 AND rtp = $2;
                /**
                if(is_shadow_db)
                {
                   res = PQprepare(shadow_db_conn,
                                SelectSizeName,
                                SelectSizeCommand,
                                SelectSizeNParams,
                                SelectSizeParamTypes);
                   LOG_DEBUG("%s() PREPARE SelectSize PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
                   if(PQresultStatus(res) != PGRES_COMMAND_OK)
                   {
                        LOG_ERROR("PREPARE failed for RIAT! %s", PQerrorMessage(conn));
                        SQL_exit_nicely(conn,res);
                   }
                   PQclear(res);

                }
                **/
		res = PQprepare(conn, 
				SelectSizeName, 
				SelectSizeCommand,
				SelectSizeNParams, 
				SelectSizeParamTypes);
		LOG_DEBUG("%s() PREPARE SelectSize PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("PREPARE failed for RIAT! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);

		/*
		// SQL commands to create the stored procedure for select and hold
		CREATE OR REPLACE PROCEDURE sql_select_data_procedure(fa integer, hold bool, INOUT blksize integer, INOUT fadata bytea)
		LANGUAGE plpgsql
		AS $$
		BEGIN
		IF (hold) THEN
			PERFORM pg_advisory_lock(fa);
		END IF;
		SELECT blk_size,data FROM fs_data WHERE file_address = fa INTO blksize,fadata;
		END;
		$$;
		*/

		if(!SQL_vsn10)
		{

                  if(is_shadow_db)
                  {
		      res = PQexec(shadow_db_conn," CREATE OR REPLACE PROCEDURE sql_select_data_procedure(fa integer, hold bool, INOUT blksize integer, INOUT fadata bytea) LANGUAGE plpgsql AS $$ BEGIN IF (hold) THEN PERFORM pg_advisory_lock(fa); END IF; SELECT blk_size,data FROM fs_data WHERE file_address = fa INTO blksize,fadata; END; $$;");
		      LOG_DEBUG("%s() CREATE sql_select_data_procedure for shadow db PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		      if(PQresultStatus(res) != PGRES_COMMAND_OK)
		      {
			LOG_ERROR("CREATE sql_select_data_procedure for shadow db failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		      }
		      PQclear(res);

                  }
                
		res = PQexec(conn," CREATE OR REPLACE PROCEDURE sql_select_data_procedure(fa integer, hold bool, INOUT blksize integer, INOUT fadata bytea) LANGUAGE plpgsql AS $$ BEGIN IF (hold) THEN PERFORM pg_advisory_lock(fa); END IF; SELECT blk_size,data FROM fs_data WHERE file_address = fa INTO blksize,fadata; END; $$;");
		LOG_DEBUG("%s() CREATE sql_select_data_procedure PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("CREATE sql_select_data_procedure failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);
		}

		// SELECT data FROM fs_data WHERE file_address = $1;
                if(is_shadow_db)
                {
                   res = PQprepare(shadow_db_conn,
                                SelectDataName,
                                SelectDataCommand,
                                SelectDataNParams,
                                SelectDataParamTypes);
                   LOG_DEBUG("%s() PREPARE SelectData PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
                   if(PQresultStatus(res) != PGRES_COMMAND_OK)
                   {
                        LOG_ERROR("Prepare failed! from shadow_db %s", PQerrorMessage(conn));
                        SQL_exit_nicely(conn,res);
                   }
                   PQclear(res);

                }

		res = PQprepare(conn,
				SelectDataName,
				SelectDataCommand,
				SelectDataNParams,
				SelectDataParamTypes);
		LOG_DEBUG("%s() PREPARE SelectData PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("CREATE failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);


		/* INSERT data 
		CREATE PROCEDURE sql_insert_data_procedure(fa integer, ft integer, ord integer, xaddr text, recid text, blk_size integer, data bytea, INOUT data) 
		LANGUAGE plpgsql 
		AS $$ 
		BEGIN 
			INSERT INTO fs_data (file_address, face_type, ordinal, xaddr, recid, blk_size, data) VALUES (fa,ft,ord,xaddr,recid,blk_size,data) RETURNING data; 
		END; 
		$$;
		*/

		if(!SQL_vsn10)
		{
                   if(is_shadow_db)
                   {
		      res = PQexec(shadow_db_conn,"CREATE OR REPLACE PROCEDURE sql_insert_data_procedure(fa integer, ft integer, ord integer, xaddr text, recid text, blk_size integer, indata bytea, INOUT outdata bytea) LANGUAGE plpgsql AS $$ BEGIN INSERT INTO fs_data (file_address, face_type, ordinal, xaddr, recid, blk_size, data) VALUES (fa, ft, ord, xaddr, recid, blk_size, indata) RETURNING data INTO outdata; END; $$;");
		      LOG_DEBUG("%s() CREATE sql_insert_data_procedure for shadow db PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		      if(PQresultStatus(res) != PGRES_COMMAND_OK)
		      {
			LOG_ERROR("CREATE sql_insert_data_procedure for shadow db failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		      }
		      PQclear(res);

                   }
		res = PQexec(conn,"CREATE OR REPLACE PROCEDURE sql_insert_data_procedure(fa integer, ft integer, ord integer, xaddr text, recid text, blk_size integer, indata bytea, INOUT outdata bytea) LANGUAGE plpgsql AS $$ BEGIN INSERT INTO fs_data (file_address, face_type, ordinal, xaddr, recid, blk_size, data) VALUES (fa, ft, ord, xaddr, recid, blk_size, indata) RETURNING data INTO outdata; END; $$;");
		LOG_DEBUG("%s() CREATE sql_insert_data_procedure PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("CREATE sql_insert_data_procedure failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);
		}

                if(is_shadow_db)
                {
                    res = PQprepare(shadow_db_conn,
                                InsertDataName,
                                InsertDataCommand,
                                InsertDataNParams,
                                InsertDataParamTypes);
                    LOG_DEBUG("%s() PREPARE InsertData PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
                    if(PQresultStatus(res) != PGRES_COMMAND_OK)
                    {
                        LOG_ERROR("PREPARE failed from shadow_db! %s", PQerrorMessage(conn));
                        SQL_exit_nicely(conn,res);
                    }
                    PQclear(res);

                }
	
		res = PQprepare(conn,
				InsertDataName,
				InsertDataCommand,
				InsertDataNParams,
				InsertDataParamTypes);
		LOG_DEBUG("%s() PREPARE InsertData PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("PREPARE failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);

		/*
		// SQL commands to create the stored procedure for unpdate and unhold
		CREATE OR REPLACE PROCEDURE sql_update_data_procedure(indata bytea, unhold bool, fa integer)
		LANGUAGE plpgsql
		AS $$
		BEGIN
		UPDATE fs_data SET data = indata WHERE file_address = fa;
		IF (unhold) THEN
			PERFORM pg_advisory_unlock(fa);
		END IF;
		END; -- COMMIT TRANSACTION
		$$;
		*/
		if(!SQL_vsn10)
		{
                   if(is_shadow_db)
                   {
		      res = PQexec(shadow_db_conn,"CREATE OR REPLACE PROCEDURE sql_update_data_procedure(indata bytea, unhold bool, fa integer) LANGUAGE plpgsql AS $$ BEGIN UPDATE fs_data SET data = indata WHERE file_address = fa; IF (unhold) THEN PERFORM pg_advisory_unlock(fa); END IF; END; -- COMMIT TRANSACTION $$;");
		      LOG_DEBUG("%s() CREATE sql_update_data_procedure for shadow db PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		      if(PQresultStatus(res) != PGRES_COMMAND_OK)
		      {
			LOG_ERROR("CREATE sql_update_data_procedure for shadow db failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		      }
		     PQclear(res);

                   }
		res = PQexec(conn,"CREATE OR REPLACE PROCEDURE sql_update_data_procedure(indata bytea, unhold bool, fa integer) LANGUAGE plpgsql AS $$ BEGIN UPDATE fs_data SET data = indata WHERE file_address = fa; IF (unhold) THEN PERFORM pg_advisory_unlock(fa); END IF; END; -- COMMIT TRANSACTION $$;");
		LOG_DEBUG("%s() CREATE sql_update_data_procedure PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("CREATE sql_update_data_procedure failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);
		}

		// UPDATE fs_data SET data = $1 WHERE file_address = $2
                if(is_shadow_db)
                {
                   res = PQprepare(shadow_db_conn,
                                UpdateDataName,
                                UpdateDataCommand,
                                UpdateDataNParams,
                                UpdateDataParamTypes);
                   LOG_DEBUG("%s() PREPARE UpdateData from shadow_db PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
                   if(PQresultStatus(res) != PGRES_COMMAND_OK)
                   {
                        LOG_ERROR("PREPARE failed! for shadow db %s", PQerrorMessage(conn));
                        SQL_exit_nicely(conn,res);
                   }
                   PQclear(res);

                }

		res = PQprepare(conn,
				UpdateDataName,
				UpdateDataCommand,
				UpdateDataNParams,
				UpdateDataParamTypes);
		LOG_DEBUG("%s() PREPARE UpdateData PQresultStatus = %s",__func__,PQresStatus(PQresultStatus(res)));
		if(PQresultStatus(res) != PGRES_COMMAND_OK)
		{
			LOG_ERROR("PREPARE failed! %s", PQerrorMessage(conn));
			SQL_exit_nicely(conn,res);
		}
		PQclear(res);

	}
	res = PQexec(conn, "END TRANSACTION");
	PQclear(res);
}