/*--------------------------------------------------------------------
 * FILE:
 *     replicate.c
 *
 * NOTE:
 *     This file is composed of the functions to call with the source
 *     at backend for the replication.
 *     Low level I/O functions that called by in these functions are 
 *     contained in 'replicate_com.c'.
 *
 * Portions Copyright (c) 2003-2006, Atsushi Mitani
 *--------------------------------------------------------------------
 */

/*--------------------------------------
 * INTERFACE ROUTINES
 *
 * setup/teardown:
 *      PGR_Init_Replicate_Server_Data
 *      PGR_Set_Replicate_Server_Socket
 *      PGR_delete_shm
 * I/O call:
 *      PGR_Send_Replicate_Command
 * table handling:
 *      PGR_get_replicate_server_info
 * status distinction:
 *      PGR_Is_Replicated_Command
 *      Xlog_Check_Replicatec
 * replicateion main:
 *      PGR_replication 
 *-------------------------------------
 */
#ifdef USE_REPLICATION

#include "postgres.h"

#include <stdio.h>
#include <strings.h>
#include <signal.h>
#include <errno.h>
#include <fcntl.h>
#include <grp.h>
#include <unistd.h>
#include <ctype.h>
#include <time.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <netdb.h>
#include <netinet/in.h>
#ifdef HAVE_NETINET_TCP_H
#include <netinet/tcp.h>
#endif
#include <arpa/inet.h>
#include <sys/file.h>
#include <netdb.h>

#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "miscadmin.h"
#include "nodes/nodes.h"
#include "nodes/print.h"
#include "utils/guc.h"
#include "parser/parser.h"
#include "access/xact.h"
#include "storage/proc.h"
#include "replicate.h"

PGR_ReplicationLog_Info ReplicationLog_Info;

/*--------------------------------------
 * PROTOTYPE DECLARATION
 *--------------------------------------
 */
static int set_command_args(char argv[PGR_CMD_ARG_NUM][256],char *str);
static bool is_same_replication_server(ReplicateServerInfo * sp1, ReplicateServerInfo * sp2 );
static ReplicateServerInfo * search_new_replication_server ( ReplicateServerInfo * sp , int socket_type );

static int close_replicate_server_socket ( ReplicateServerInfo * sp , int socket_type );
static int recv_replicate_result(int sock,char * result,int user_timeout);
static int recv_message(int sock,char * buf,int flag);
static int send_replicate_packet(int sock,ReplicateHeader * header, char * query_string);
static int get_replicate_server_socket ( ReplicateServerInfo * sp , int socket_type );
static bool is_copy_from(char * query);
//static int is_session_authorization(char * query);
static int get_words( char words[MAX_WORDS][MAX_WORD_LETTERS] ,char * string,int length,int upper);
static int get_table_name(char * table_name, char * query, int position );
static bool is_not_replication_query(char * query_string, int query_len, char cmdType);
static int Comp_Not_Replicate(PGR_Not_Replicate_Type * nrp1,PGR_Not_Replicate_Type* nrp2);
static bool is_serial_control_query(char cmdType,char * query);
static bool is_select_into_query(char cmdType,char * query);
static int send_response_to_replication_server(const char * notice);
static bool is_transaction_command(const char * commandTag);
static bool do_not_replication_command(const char * commandTag);
static bool is_create_temp_table(char * query);
static int add_replication_server(char * hostname,char * port, char * recovery_port);
static int change_replication_server(char * hostname,char * port, char * recovery_port);
static int get_new_replication_socket( ReplicateServerInfo * base, ReplicateServerInfo * sp, int socket_type);
static char * get_hostName(char * str);
static void set_response_mode(char * mode);
static void PGR_Set_Current_Replication_Query_ID(char *id);
#ifdef CONTROL_LOCK_CONFLICT
static int wait_lock_answer(void);
static int read_trigger(char * result, int buf_size);
#endif /* CONTROL_LOCK_CONFLICT */


static unsigned int get_next_request_id(void);
static bool is_this_query_replicated(char * id);
static int set_replication_id(char * id);

extern ssize_t secure_read(Port *, void *, size_t);

bool autocommit = true;
/*--------------------------------------------------------------------
 * SYMBOL
 *    PGR_Init_Replicate_Server_Data()
 * NOTES
 *    Read Configuration file and create ReplicateServerData table
 * ARGS
 *    void
 * RETURN
 *    OK: STATUS_OK
 *    NG: STATUS_ERROR
 *--------------------------------------------------------------------
 */
int
PGR_Init_Replicate_Server_Data(void)
{
	int table_size,str_size;
	ReplicateServerInfo  *sp;
	PGR_Not_Replicate_Type * nrp;
	ConfDataType * conf;
	int rec_no,cnt;
	unsigned int ip;
	char HostName[128];


	if (ConfData_Top == (ConfDataType *)NULL)
	{
		return STATUS_ERROR;
	}

	/* allocate replication server information table */
	table_size = sizeof(ReplicateServerInfo) * MAX_SERVER_NUM;
	ReplicateServerShmid = shmget(IPC_PRIVATE,table_size,IPC_CREAT | IPC_EXCL | 0600);
	if (ReplicateServerShmid < 0)
	{
		return STATUS_ERROR;
	}
	ReplicateServerData = (ReplicateServerInfo *)shmat(ReplicateServerShmid,0,0);
	if (ReplicateServerData == (ReplicateServerInfo *)-1)
	{
		return STATUS_ERROR;
	}
	memset(ReplicateServerData,0,table_size);
	sp = ReplicateServerData;

	/* allocate cluster db information table */
	ClusterDBShmid = shmget(IPC_PRIVATE,sizeof(ClusterDBInfo),IPC_CREAT | IPC_EXCL | 0600);
	if (ClusterDBShmid < 0)
	{
		return STATUS_ERROR;
	}
	ClusterDBData = (ClusterDBInfo *)shmat(ClusterDBShmid,0,0);
	if (ClusterDBData == (ClusterDBInfo *)-1)
	{
		return STATUS_ERROR;
	}
	memset(ClusterDBData,0,sizeof(ClusterDBInfo));
	PGR_Set_Cluster_Status(STATUS_REPLICATE);

	/* allocate partial replicate table */
	table_size = sizeof(PGR_Not_Replicate_Type) * MAX_SERVER_NUM;
	PGR_Not_Replicate = malloc(table_size);
	if (PGR_Not_Replicate == (PGR_Not_Replicate_Type*)NULL)
	{
		return STATUS_ERROR;
	}
	memset(PGR_Not_Replicate, 0, table_size);
	nrp = PGR_Not_Replicate;
	cnt = 0;
	conf = ConfData_Top;
	while ((conf != (ConfDataType *)NULL) && (cnt < MAX_SERVER_NUM))
	{
		/* set replication server table */
		if (!strcmp(conf->table,REPLICATION_SERVER_INFO_TAG))
		{
			rec_no = conf->rec_no;
			cnt = rec_no;
			if (!strcmp(conf->key,HOST_NAME_TAG))
			{
				strncpy((sp + rec_no)->hostName,conf->value,sizeof(sp->hostName));
				conf = (ConfDataType *)conf->next;
				continue;
			}
			if (!strcmp(conf->key,PORT_TAG))
			{
				(sp + rec_no)->portNumber = atoi(conf->value);
				(sp + rec_no)->sock = -1;
				if ((sp + rec_no)->useFlag != DATA_USE)
				{
					PGR_Set_Replication_Server_Status((sp+rec_no), DATA_INIT);
				}
				memset((sp + rec_no + 1)->hostName,0,sizeof(sp->hostName));
				(sp + rec_no + 1)->useFlag = DATA_END;
				conf = (ConfDataType *)conf->next;
				continue;
			}
			if (!strcmp(conf->key,RECOVERY_PORT_TAG))
			{
				(sp + rec_no)->recoveryPortNumber = atoi(conf->value);
				if ((sp + rec_no)->useFlag != DATA_USE)
				{
					PGR_Set_Replication_Server_Status((sp+rec_no), DATA_INIT);
				}
				memset((sp + rec_no + 1)->hostName,0,sizeof(sp->hostName));
				(sp + rec_no + 1)->useFlag = DATA_END;
				conf = (ConfDataType *)conf->next;
				continue;
			}
		}
		/* set part replication table */
		if (!strcmp(conf->table,NOT_REPLICATE_INFO_TAG))
		{
			rec_no = conf->rec_no;
			cnt = rec_no;
			if (PGR_Not_Replicate_Rec_Num < rec_no +1)
			{
				PGR_Not_Replicate_Rec_Num = rec_no +1;
			}
			if (!strcmp(conf->key,DB_NAME_TAG))
			{
				strncpy((nrp + rec_no)->db_name,conf->value,sizeof(nrp->db_name));
				conf = (ConfDataType *)conf->next;
				continue;
			}
			if (!strcmp(conf->key,TABLE_NAME_TAG))
			{
				strncpy((nrp + rec_no)->table_name,conf->value,sizeof(nrp->table_name));
				conf = (ConfDataType *)conf->next;
				continue;
			}
		}
		if (!strcmp(conf->key,RECOVERY_PORT_TAG))
		{
			RecoveryPortNumber = atoi(conf->value);
		}
		if (!strcmp(conf->key,RSYNC_PATH_TAG))
		{
			str_size = strlen(conf->value) ;
			RsyncPath = malloc(str_size + 1);
			if (RsyncPath == NULL)
			{
				return STATUS_ERROR;
			}
			memset(RsyncPath,0,str_size + 1);
			strncpy(RsyncPath,conf->value,str_size);
		}
		if (!strcmp(conf->key,RSYNC_OPTION_TAG))
		{
			str_size = strlen(conf->value) ;
			RsyncOption = malloc(str_size + 1);
			if (RsyncOption == NULL)
			{
				return STATUS_ERROR;
			}
			memset(RsyncOption,0,str_size + 1);
			strncpy(RsyncOption,conf->value,str_size);
		}
		if (!strcmp(conf->key,RSYNC_COMPRESS_TAG))
		{
			if (!strcmp(conf->value, "yes"))
				RsyncCompress = true;
			else if (!strcmp(conf->value, "no"))
				RsyncCompress = false;
		}
		if (!strcmp(conf->key,STAND_ALONE_TAG))
		{
			PGR_Stand_Alone = (PGR_Stand_Alone_Type*)malloc(sizeof(PGR_Stand_Alone_Type));
			if (PGR_Stand_Alone == (PGR_Stand_Alone_Type *)NULL)
			{
				return STATUS_ERROR;
			}
			PGR_Stand_Alone->is_stand_alone = false;
			if (!strcmp(conf->value,READ_WRITE_IF_STAND_ALONE))
			{
				PGR_Stand_Alone->permit = PERMIT_READ_WRITE;
			}
			else
			{
				PGR_Stand_Alone->permit = PERMIT_READ_ONLY;
			}
		}
		conf = (ConfDataType *)conf->next;
	}
	TransactionSock = -1;
	ReplicateCurrentTime = (ReplicateNow *)malloc(sizeof(ReplicateNow));
	if (ReplicateCurrentTime == (ReplicateNow *)NULL)
	{
		return STATUS_ERROR;
	}
	memset(ReplicateCurrentTime,0,sizeof(ReplicateNow));

	PGRCopyData = (CopyData *)malloc(sizeof(CopyData));
	if (PGRCopyData == (CopyData *)NULL)
	{
		return STATUS_ERROR;
	}
	memset(PGRCopyData,0,sizeof(CopyData));

	if (PGR_Not_Replicate_Rec_Num  == 0)
	{
		free(PGR_Not_Replicate);
		PGR_Not_Replicate = NULL;
	}
	else
	{
		qsort((char *)PGR_Not_Replicate,PGR_Not_Replicate_Rec_Num,sizeof(PGR_Not_Replicate_Type), (int (*)(const void*,const void*))Comp_Not_Replicate);
	}

	PGRSelfHostName = malloc(HOSTNAME_MAX_LENGTH);
	if (PGRSelfHostName == NULL)
	{
		return STATUS_ERROR;
	}
	memset(PGRSelfHostName,0,HOSTNAME_MAX_LENGTH);
	if (gethostname(HostName,HOSTNAME_MAX_LENGTH) < 0)
	{
		return STATUS_ERROR;
	}
	ip=PGRget_ip_by_name(HostName);

        sprintf(PGRSelfHostName,
					 "%d.%d.%d.%d",
					 (ip      ) & 0xff ,
					 (ip >>  8) & 0xff ,
					 (ip >> 16) & 0xff ,
					 (ip >> 24) & 0xff );

	return	STATUS_OK;
}

/*--------------------------------------------------------------------
 * SYMBOL
 *    PGR_Set_Replicate_Server_Socket()
 * NOTES
 *    Create new socket and set ReplicateServerData table
 * ARGS
 *    void
 * RETURN
 *    OK: STATUS_OK
 *    NG: STATUS_ERROR
 *--------------------------------------------------------------------
 */
int
PGR_Set_Replicate_Server_Socket(void)
{
	ReplicateServerInfo * sp;
	if (ReplicateServerData == NULL)
	{
		return STATUS_ERROR;
	}
	sp = ReplicateServerData;
	while (sp->useFlag != DATA_END){
		sp->sock = -1;
		PGR_Create_Socket_Connect(&(sp->sock),sp->hostName,sp->portNumber);
		sp ++;
	}
	return	STATUS_OK;
}

/*--------------------------------------------------------------------
 * SYMBOL
 *    get_replicate_server_socket()
 * NOTES
 *    search or create a socket to connect with the replication server
 * ARGS
 *    ReplicateServerInfo * sp: replication server data (I)
 *    int socket_type: socket type (I)
 *                       -PGR_TRANSACTION_SOCKET:
 *                       -PGR_QUERY_SOCKET:
 * RETURN
 *    OK: >0(socket)
 *    NG: -1
 *--------------------------------------------------------------------
 */
static int
get_replicate_server_socket ( ReplicateServerInfo * sp , int socket_type )
{
	ReplicateServerInfo * tmp;
	tmp = sp;
	if (tmp == (ReplicateServerInfo *) NULL)
	{
		return -1;
	}
	if (tmp->hostName[0] == '\0')
	{
		return -1;
	}

	if (TransactionSock != -1)
	{
		return TransactionSock;
	}

	while(PGR_Create_Socket_Connect(&TransactionSock,tmp->hostName,tmp->portNumber) != STATUS_OK)
	{
		close(TransactionSock);
		TransactionSock = -1;
		PGR_Set_Replication_Server_Status(tmp, DATA_ERR);
		usleep(20);
		tmp = PGR_get_replicate_server_info();
		if (tmp == (ReplicateServerInfo *)NULL)
		{
			return -1;
		}
		PGR_Set_Replication_Server_Status(tmp, DATA_USE);
		usleep(10);
	}
	return TransactionSock;
}

/*--------------------------------------------------------------------
 * SYMBOL
 *    close_replicate_server_socket()
 * NOTES
 *    close the socket connected with the replication server
 * ARGS
 *    ReplicateServerInfo * sp: replication server data (I)
 *    int socket_type: socket type (I)
 *                       -PGR_TRANSACTION_SOCKET:
 *                       -PGR_QUERY_SOCKET:
 * RETURN
 *    OK: STATUS_OK
 *    NG: STATUS_ERROR
 *--------------------------------------------------------------------
 */
static int
close_replicate_server_socket ( ReplicateServerInfo * sp , int socket_type )
{
	if (sp == (ReplicateServerInfo *)NULL )
	{
		return STATUS_ERROR;
	}
	if (sp->hostName[0] == '\0')
	{
		return STATUS_ERROR;
	}
	if (TransactionSock != -1)
	{
		PGR_Close_Sock(&(TransactionSock));
		TransactionSock = -1;
	}
	switch (socket_type)
	{
		case PGR_TRANSACTION_SOCKET:
			if (TransactionSock != -1)
			{
				PGR_Close_Sock(&(TransactionSock));
			}
			TransactionSock = -1;
			sp->sock = -1;
			break;
		case PGR_QUERY_SOCKET:
			if (sp->sock != -1)
			{
				PGR_Close_Sock(&(sp->sock));
			}
			sp->sock = -1;
			break;
	}
	PGR_Set_Replication_Server_Status(sp, DATA_INIT);
	return STATUS_OK;
}

static bool
is_same_replication_server(ReplicateServerInfo * sp1, ReplicateServerInfo * sp2 )
{
	if ((sp1 == NULL) || (sp2 == NULL))
	{
		return false;
	}
	if ((!strcmp(sp1->hostName,sp2->hostName)) &&
		(sp1->portNumber == sp2->portNumber) &&
		(sp1->recoveryPortNumber == sp2->recoveryPortNumber))
	{
		return true;
	}
	return false;
}

static ReplicateServerInfo *
search_new_replication_server ( ReplicateServerInfo * sp , int socket_type )
{
	ReplicateHeader dummy_header;
	ReplicateServerInfo * rs_tbl;
	char command[256];
	int sock = -1;
	int cnt = 0;

	if ((ReplicateServerData == NULL) || ( sp == NULL))
	{
		return NULL;
	}
	rs_tbl = sp;
	close_replicate_server_socket ( sp , socket_type);
	sp ++;
	while (is_same_replication_server(sp,rs_tbl) != true)
	{
		if (sp->useFlag == DATA_END)
		{
			sp = ReplicateServerData;
		}
		sock = get_replicate_server_socket( sp , socket_type);
		if (sock < 0 )
		{
			if (is_same_replication_server(sp,rs_tbl) == true)
			{
				return NULL;
			}
			else
			{
				sp++;
			}
			continue;
		}
		memset(&dummy_header, 0, sizeof(ReplicateHeader));
		memset(command,0,sizeof(command));
		snprintf(command,sizeof(command)-1,"SELECT %s(%d,%s,%d,%d)",
				PGR_SYSTEM_COMMAND_FUNC,
				PGR_CHANGE_REPLICATION_SERVER_FUNC_NO,
				sp->hostName,
				sp->portNumber,
				sp->recoveryPortNumber);
		dummy_header.cmdSys = CMD_SYS_CALL;
		dummy_header.cmdSts = CMD_STS_NOTICE;
		dummy_header.query_size = htonl(strlen(command));
		if (send_replicate_packet(sock,&dummy_header,command) != STATUS_OK)
		{
			cnt ++;
			close_replicate_server_socket ( sp , socket_type);
			PGR_Set_Replication_Server_Status(sp, DATA_ERR);
		}
		else
		{
			PGR_Set_Replication_Server_Status(sp, DATA_USE);
			return sp;
		}
		if (cnt > MAX_RETRY_TIMES )
		{
			sp++;
			cnt = 0;
		}
		else
		{
			continue;
		}
	}
	return NULL;
}

static int
get_table_name(char * table_name, char * query, int position )
{
	
	int i,wc;
	char * p;
	char * sp;
	int length;

	if ((table_name == NULL) || (query == NULL) || (position < 1))
	{
		return STATUS_ERROR;
	}
	length = strlen(query);
	p = query;
	wc = 1;
	sp = table_name;
	for (i = 0 ; i < length ; i ++)
	{
		while(isspace(*p))
		{
			p++;
			i++;
		}
		while((*p != '\0') && (! isspace(*p)))
		{
			if ((*p == ';') || (*p == '('))
				break;
			if (wc == position)
			{
				*sp = *p;
				sp++;
			}
			p++;
			i++;
		}
		if (wc == position)
		{
			*sp = '\0';
			break;
		}
		wc++;
	}
	return STATUS_OK;
}

static bool 
is_not_replication_query(char * query_string, int query_len, char cmdType)
{
	PGR_Not_Replicate_Type key;
	PGR_Not_Replicate_Type * ptr = NULL;

	if (PGR_Not_Replicate_Rec_Num <= 0)
		return false;
	if (query_string == NULL)
		return true;
	memset(&key,0,sizeof(PGR_Not_Replicate_Type));
	strncpy(key.db_name ,(char *)(MyProcPort->database_name),sizeof(key.db_name)-1);
	switch (cmdType)
	{
		case CMD_TYPE_INSERT:
			get_table_name(key.table_name,query_string,3);
			break;
		case CMD_TYPE_UPDATE:
			get_table_name(key.table_name,query_string,2);
			break;
		case CMD_TYPE_DELETE:
			get_table_name(key.table_name,query_string,3);
			break;
		case CMD_TYPE_COPY:
			get_table_name(key.table_name,query_string,2);
			break;
		default:
			return false;
	}
	ptr = (PGR_Not_Replicate_Type*)bsearch((void*)&key,(void*)PGR_Not_Replicate,PGR_Not_Replicate_Rec_Num,sizeof(PGR_Not_Replicate_Type), (int (*)(const void*,const void*))Comp_Not_Replicate);
	if (ptr == NULL)
	{
		return false;
	}
	return true;

}

/*--------------------------------------------------------------------
 * SYMBOL
 *    PGR_Send_Replicate_Command()
 * NOTES
 *    create new socket
 * ARGS
 *    char * query_string: query strings (I)
 *    char cmdSts: 
 *    char cmdType:
 * RETURN
 *    OK: result
 *    NG: NULL
 *--------------------------------------------------------------------
 */
char *
PGR_Send_Replicate_Command(char * query_string, int query_len, char cmdSts ,char cmdType)
{
	int sock = -1;
	int cnt = 0;
	ReplicateHeader header;
	char * serverName = NULL;
	int portNumber=0;
	char * result = NULL;
	ReplicateServerInfo * sp = NULL;
	ReplicateServerInfo * base = NULL;
	int socket_type = 0;
	char argv[ PGR_CMD_ARG_NUM ][256];
	int argc = 0;
	int func_no = 0;
	int check_flag =0;


	/*
	 * check query string
	 */
	if ((query_string == NULL)  ||
		(query_len < 0))
	{
		return NULL;
	}
	/* check not replication query */
	if (is_not_replication_query(query_string, query_len, cmdType) == true)
	{
		PGR_Copy_Data_Need_Replicate = false;
		return NULL;
	}

	if ((cmdSts == CMD_STS_TRANSACTION ) ||
		(cmdSts == CMD_STS_SET_SESSION_AUTHORIZATION ) ||
		(cmdSts == CMD_STS_TEMP_TABLE ))
	{
		socket_type = PGR_TRANSACTION_SOCKET ;
	}
	else
	{
		socket_type = PGR_QUERY_SOCKET ;
	}
	sp = PGR_get_replicate_server_info();
	if (sp == NULL)
	{
		if (Debug_pretty_print)
			elog(DEBUG1,"PGR_get_replicate_server_info get error");
		return NULL;
	}
	sock = get_replicate_server_socket( sp , socket_type);
	if (sock < 0)
	{
		if (Debug_pretty_print)
			elog(DEBUG1,"get_replicate_server_socket fail");
		return NULL;
	}
	result = malloc(PGR_MESSAGE_BUFSIZE + 4);
	if (result == NULL)
	{
		return NULL;
	}

	serverName = sp->hostName;
	portNumber = (int)sp->portNumber;
	memset(&header,0,sizeof(ReplicateHeader));

	header.cmdSts = cmdSts;
	header.cmdType = cmdType;
	header.port = htons(PostPortNumber);
	header.pid = htons(getpid());
	header.query_size = htonl(query_len); 

	strncpy(header.dbName ,(char *)(MyProcPort->database_name),sizeof(header.dbName)-1);
	strncpy(header.userName , (char *)(MyProcPort->user_name),sizeof(header.userName)-1);
	header.request_id = htonl(get_next_request_id());
	header.rlog = 0;

	if (PGRSelfHostName != NULL)
	{
		strncpy(header.from_host, PGRSelfHostName, HOSTNAME_MAX_LENGTH);
	}

	base = sp;
	PGR_Sock_To_Replication_Server = sock;

retry_send_prereplicate_packet:

	memset(result,0,PGR_MESSAGE_BUFSIZE + 4);
	cnt = 0;
	header.cmdSys=CMD_SYS_PREREPLICATE;

	while (send_replicate_packet(sock,&header,query_string) != STATUS_OK)
	{
		cnt++;
		if (cnt >= MAX_RETRY_TIMES )
		{
			sock = get_new_replication_socket( base, sp, socket_type);
			if (sock < 0)
			{
				if (Debug_pretty_print)
					elog(DEBUG1,"all replication servers may be down");
				PGR_Stand_Alone->is_stand_alone = true;
				if (cmdSts == CMD_STS_TRANSACTION )
				{
					strcpy(result,PGR_REPLICATION_ABORT_MSG);
					return result;
				}
				free(result);
				result = NULL;
				return NULL;
				
			}
			if(cmdSts==CMD_STS_TRANSACTION ) 
			{
				if (cmdType!=CMD_TYPE_BEGIN 
				    && cmdType!=CMD_TYPE_ROLLBACK)  
				{							
					elog(ERROR,"replicate server down during replicating transaction. aborted.");
					free(result);
					return NULL;
				}
			}					 
			PGR_Sock_To_Replication_Server = sock;
			cnt = 0;
		}
	}

	memset(result,0,PGR_MESSAGE_BUFSIZE);
	if (recv_replicate_result(sock,result,0) < 0)
	{

			sock = get_new_replication_socket( base, sp, socket_type);
			if (sock < 0)
			{
					if (Debug_pretty_print)
							elog(DEBUG1,"all replication servers may be down");
					PGR_Stand_Alone->is_stand_alone = true;

					if (cmdSts == CMD_STS_TRANSACTION )
					{
							strcpy(result,PGR_REPLICATION_ABORT_MSG);
							return result;
					}
					if(result!=NULL) {
							free(result);
							result = NULL;
					}
					return NULL;
			}
			PGR_Sock_To_Replication_Server = sock;
			/* replication server should be down */

			if(cmdSts==CMD_STS_TRANSACTION 
			   && (cmdType!=CMD_TYPE_BEGIN && cmdType!=CMD_TYPE_ROLLBACK))
			{
					elog(ERROR,"replicate server down during replicating transaction. aborted.");
					free(result);
					return NULL;
			}

			goto retry_send_prereplicate_packet;
	}


	argc = set_command_args(argv,result);
	func_no=atoi(argv[0]);
	if(func_no==0) {
			/* this server is not primary replicate server*/
			sock=-1;
			goto retry_send_prereplicate_packet;
	}
retry_send_replicate_packet:

	memset(result,0,PGR_MESSAGE_BUFSIZE + 4);
	cnt = 0;
	header.cmdSys = CMD_SYS_REPLICATE;
	while (send_replicate_packet(sock,&header,query_string) != STATUS_OK)
	{
		if (cnt > MAX_RETRY_TIMES )
		{
			sock = get_new_replication_socket( base, sp, socket_type);
			if (sock < 0)
			{
				if (Debug_pretty_print)
				  elog(DEBUG1,"all replication servers may be down");
				PGR_Stand_Alone->is_stand_alone = true;
				if (cmdSts == CMD_STS_TRANSACTION )
				{
				        strcpy(result,PGR_REPLICATION_ABORT_MSG);
				        return result;
				}
				free(result);
				result = NULL;
				return NULL;

			}
			PGR_Sock_To_Replication_Server = sock;
			header.rlog = CONNECTION_SUSPENDED_TYPE;
			cnt = 0;
		}
		cnt ++;
	}

	memset(result,0,PGR_MESSAGE_BUFSIZE);
	if (recv_replicate_result(sock,result,0) < 0)
	{
		/* replication server should be down */
		sock = get_new_replication_socket( base, sp, socket_type);
		if (sock < 0)
		{
			if (Debug_pretty_print)
				elog(DEBUG1,"all replication servers may be down");
			PGR_Stand_Alone->is_stand_alone = true;

			if (cmdSts == CMD_STS_TRANSACTION )
			{
			        strcpy(result,PGR_REPLICATION_ABORT_MSG);
				return result;
			}
			if(result!=NULL) {
			        free(result);
			        result = NULL;
			}
			return NULL;
		}
		PGR_Sock_To_Replication_Server = sock;
		header.rlog = CONNECTION_SUSPENDED_TYPE;

		goto retry_send_replicate_packet;
	}

	argc = set_command_args(argv,result);
	if (argc >= 1)
	{
		func_no = atoi(argv[0]);
		if (func_no == PGR_SET_CURRENT_TIME_FUNC_NO)
		{
			PGR_Set_Current_Time(argv[1],argv[2]);
			set_replication_id(argv[3]);
			set_response_mode(argv[4]);
         		PGR_Set_Current_Replication_Query_ID(argv[5]);
		}
		else if (func_no == PGR_NOTICE_DEADLOCK_DETECTION_FUNC_NO)
		{
			memset(result,0,PGR_MESSAGE_BUFSIZE);
			strcpy(result,PGR_DEADLOCK_DETECTION_MSG);
		}
                else if (func_no == PGR_SET_CURRENT_REPLICATION_QUERY_ID_NO) 
		{
		  PGR_Set_Current_Replication_Query_ID(argv[1]);
		}  
		else if (func_no == PGR_QUERY_CONFIRM_ANSWER_FUNC_NO)
		{
			check_flag = atoi(argv[1]);
			if (check_flag == PGR_ALREADY_COMMITTED )
			{
				PGR_Set_Current_Time(argv[2],argv[3]);
				set_replication_id(argv[4]);
			}
			else
			{
				PGR_Set_Current_Time(argv[1],argv[2]);
				set_replication_id(argv[3]);
				/* this query is not replicated */
				/*
				free(result);
				return NULL;
				*/
			}
		}
	}
	return result;
}


static int
set_replication_id(char * id)
{
        uint32_t rid=0;
		uint32_t saved_id;
		if (id == NULL)
		{
				return STATUS_ERROR;
		}

		rid=(uint32_t)atol(id);
		if(rid==0)
				return STATUS_OK;

		needToUpdateReplicateIdOnNextQueryIsDone=true;
		saved_id=ReplicationLog_Info.PGR_Replicate_ID;

		ReplicationLog_Info.PGR_Replicate_ID =rid;


		/*set replicate id in this process */


		if (CurrentReplicateServer == NULL)
		{
				PGR_get_replicate_server_info();
		}
		if (CurrentReplicateServer != NULL)
		{
				/* set replicate id in this system */
				saved_id=CurrentReplicateServer->replicate_id;
				elog(DEBUG1, "replication id set from %d to %d", saved_id, rid); 

				CurrentReplicateServer->replicate_id = (uint32_t)(atol(id));
		}
	
		return STATUS_OK;
}


static unsigned int
get_next_request_id(void)
{
	if (ReplicationLog_Info.PGR_Request_ID +1 < PGR_MAX_COUNTER)
	{
		ReplicationLog_Info.PGR_Request_ID ++;
	}
	else
	{
		ReplicationLog_Info.PGR_Request_ID = 0;
	}
	return ReplicationLog_Info.PGR_Request_ID ;
		
}

static bool
is_this_query_replicated(char * id)
{
	uint32_t replicate_id = 0;
	uint32_t saved_id = 0;
	int32_t diff=0;
	ReplicateServerInfo * replicate_server_info = NULL;

	if (id == NULL)
	{
		return false;
	}
	replicate_id = (uint32_t)atol(id);
	elog(DEBUG1, "check for replication id , input=%u", replicate_id);

	if (CurrentReplicateServer == NULL)
	{
		PGR_get_replicate_server_info();
	}

	if (CurrentReplicateServer != NULL)
	{
		replicate_server_info = CurrentReplicateServer;
	}
        else if (LastReplicateServer != NULL)
	{	 
		replicate_server_info = LastReplicateServer;
	}
	if (replicate_server_info != NULL)
	{

	        saved_id=replicate_server_info->replicate_id;
		saved_id = saved_id < ReplicationLog_Info.PGR_Replicate_ID
		  ? ReplicationLog_Info.PGR_Replicate_ID
		  : saved_id;

		elog(DEBUG1, "check for replication id , now=%u", saved_id);
		/* check replicate_id < saved_id logically 
		 * 
		 * see also:
		 *  backend/transam/transam.c#TransactionIdPrecedes
		 */

	        diff = (int32) (saved_id-replicate_id);
		return (diff > 0);
	}
	elog(DEBUG1, "check for replication id check failed. no replication server");
	return false;
}


static int
get_new_replication_socket( ReplicateServerInfo * base, ReplicateServerInfo * sp, int socket_type)
{
	int sock = -1;

	if (( base == NULL) ||
		( sp == NULL))
	{
		return -1;
	}
	close_replicate_server_socket ( sp , socket_type);
	PGR_Set_Replication_Server_Status(sp, DATA_ERR);
	sp = search_new_replication_server(base, socket_type);
	if (sp == NULL)
	{
		if (Debug_pretty_print)
			elog(DEBUG1,"all replication servers may be down");
		PGR_Stand_Alone->is_stand_alone = true;
		return -1;
	}
	sock = get_replicate_server_socket( sp , socket_type);
	return sock;
}


static int
recv_replicate_result(int sock,char * result,int user_timeout)
{
	fd_set      rmask;
	struct timeval timeout;
	int rtn;

	if (result == NULL)
	{
		return -1;
	}

	/*
	 * Wait for something to happen.
	 */
	for (;;)
	{
		if (user_timeout == 0)
			timeout.tv_sec = PGR_RECV_TIMEOUT * 3;
		else
			timeout.tv_sec = user_timeout;

		timeout.tv_usec = 0;

		FD_ZERO(&rmask);
		FD_SET(sock,&rmask);
		rtn = select(sock+1, &rmask, (fd_set *)NULL, (fd_set *)NULL, &timeout);
		if (rtn < 0)
		{
			if (errno != EINTR)
				return -1;
		}

		else if ((rtn > 0) && (FD_ISSET(sock, &rmask)))
		{
			return (recv_message(sock, result,0));
		}
	}
	return -1;
}

static int
recv_message(int sock,char * buf,int flag)
{
	int cnt = 0;
	int r = 0;
	char * read_ptr;
	int read_size = 0;
	cnt = 0;
	read_ptr = buf;

	for (;;)
	{
		r = recv(sock,read_ptr + read_size ,PGR_MESSAGE_BUFSIZE - read_size, flag); 
		if (r < 0) {
			if (errno == EINTR || errno == EAGAIN) {
				continue;
			} else {
			  elog(DEBUG1, "recv_message():recv failed");
			  return -1;
			}
		} else if (r == 0) {
			elog(DEBUG1, "recv_message():unexpected EOF");
			return -1;
		} else /*if (r > 0)*/ {
			read_size += r;
			if (read_size == PGR_MESSAGE_BUFSIZE)
			{
				return read_size;
			}
		}
	}
	return -1;
}

static int
send_replicate_packet(int sock,ReplicateHeader * header, char * query_string)
{
	int s = 0;
	char * send_ptr = NULL;
	char * buf = NULL;
	int send_size = 0;
	int buf_size = 0;
	int header_size = 0;
	int rtn = 0;
	fd_set      wmask;
	struct timeval timeout;
	int query_size = 0;

	/* check parameter */
	if ((sock < 0) || (header == NULL))
	{
		return STATUS_ERROR;
	}

	query_size = ntohl(header->query_size);
	header_size = sizeof(ReplicateHeader);
	buf_size = header_size + query_size + 4;
	buf = malloc(buf_size);
	if (buf == NULL)
	{
		return STATUS_ERROR;
	}
	memset(buf,0,buf_size);
	buf_size -= 4;
	memcpy(buf,header,header_size);
	if (query_string != NULL)
	{
		memcpy((char *)(buf+header_size),query_string,query_size+1);
	}
	send_ptr = buf;

	/*
	 * Wait for something to happen.
	 */
	rtn = 1;
	for (;;)
	{
		timeout.tv_sec = PGR_SEND_TIMEOUT;
		timeout.tv_usec = 0;

		FD_ZERO(&wmask);
		FD_SET(sock,&wmask);
		rtn = select(sock+1, (fd_set *)NULL, &wmask, (fd_set *)NULL, &timeout);
		if (rtn < 0)
		{
			if (errno == EINTR)
				continue;
			else
			{
				elog(DEBUG1, "send_replicate_packet():select() failed");
				return STATUS_ERROR;
			}
		}
		else if (rtn && FD_ISSET(sock, &wmask))
		{


			s = send(sock,send_ptr + send_size,buf_size - send_size ,0);
			if (s < 0){
				if (errno == EINTR || errno == EAGAIN)
				{
					continue;
				}
				elog(DEBUG1, "send_replicate_packet():send error");

				/* EPIPE || ENCONNREFUSED || ENSOCK || EHOSTUNREACH */
				return STATUS_ERROR;
			} else if (s == 0) {
				free(buf);
				buf = NULL;
				elog(DEBUG1, "send_replicate_packet():unexpected EOF");
				return STATUS_ERROR;
			} else /*if (s > 0)*/ {
				send_size += s;
				if (send_size == buf_size)
				{
					free(buf);
					buf = NULL;
					return STATUS_OK;
				}
			}
		}
	}
	if (buf != NULL)
	{
		free(buf);
		buf = NULL;
	}
	return STATUS_ERROR;
}

bool
PGR_Is_Replicated_Command(char * query)
{

	return (PGR_Is_System_Command(query));
}

int
Xlog_Check_Replicate(int operation)
{
	if ((PGR_Get_Cluster_Status() == STATUS_RECOVERY)	&&
		(Transaction_Mode == false ) )
	{
		elog(WARNING, "This query is not permitted while recovery db ");
	}
	else if ((operation == CMD_UTILITY ) ||
		(operation == CMD_INSERT )  ||
		(operation == CMD_UPDATE )  ||
		(operation == CMD_DELETE ))
	{
		return (PGR_Replicate_Function_Call());
	}
	return STATUS_OK;
}

int 
PGR_Replicate_Function_Call(void)
{
	char *result = NULL;
	int status = STATUS_OK;

	if (PGR_Stand_Alone == NULL)
	{
		 return STATUS_OK;
	}
    if (Query_String != NULL)
    {
		if (PGR_Is_Stand_Alone() == true)
		{
			if (PGR_Stand_Alone->permit == PERMIT_READ_ONLY)
			{
				Query_String = NULL;
				return STATUS_ERROR;
			}
		}
		PGR_Need_Notice = true;
		PGR_Check_Lock.check_lock_conflict = true;
        result = PGR_Send_Replicate_Command(Query_String,strlen(Query_String), CMD_STS_QUERY,CMD_TYPE_SELECT);
		if (result != NULL)
		{
			PGR_Reload_Start_Time();
			if (!strncmp(result,PGR_DEADLOCK_DETECTION_MSG,strlen(PGR_DEADLOCK_DETECTION_MSG)))
			{
				status = STATUS_DEADLOCK_DETECT;
			}
			free(result);
			result = NULL;
		}
		else
		{
			status = STATUS_ERROR;
		}
		Query_String = NULL;
    }
	return status;
}

void
PGR_delete_shm(void)
{

	if (ReplicateServerData != NULL)
	{
		shmdt(ReplicateServerData);
		ReplicateServerData = NULL;
		shmctl(ReplicateServerShmid,IPC_RMID,(struct shmid_ds *)NULL);
	}
	if (ClusterDBData != NULL)
	{
		shmdt(ClusterDBData);
		ClusterDBData = NULL;
		shmctl(ClusterDBShmid,IPC_RMID,(struct shmid_ds *)NULL);
	}

	if (TransactionSock != -1)
	{
		close(TransactionSock);
	}
	
	if (RsyncPath != NULL)
	{
		free(RsyncPath);
		RsyncPath = NULL;
	}
	if (RsyncOption != NULL)
	{
		free(RsyncOption);
		RsyncOption = NULL;
	}

	if (ReplicateCurrentTime != NULL)
	{
		free(ReplicateCurrentTime);
		ReplicateCurrentTime = NULL;
	}

	if (PGRCopyData != NULL)
	{
		free (PGRCopyData);
		PGRCopyData = NULL;
	}

	if (PGR_Stand_Alone != NULL)
	{
		free(PGR_Stand_Alone);
		PGR_Stand_Alone = NULL;
	}

	if (PGR_Not_Replicate != NULL)
	{
		free(PGR_Not_Replicate);
		PGR_Not_Replicate = NULL;
	}
	if (PGRSelfHostName != NULL)
	{
		free(PGRSelfHostName);
		PGRSelfHostName = NULL;
	}
}

ReplicateServerInfo * 
PGR_get_replicate_server_info(void)
{

	ReplicateServerInfo * sp;

	if (ReplicateServerData == NULL)
	{
		return (ReplicateServerInfo *)NULL;
	}
	sp = ReplicateServerData;
	while (sp->useFlag != DATA_END)
	{
		if (sp->useFlag == DATA_USE )
		{
			if (CurrentReplicateServer != NULL)
			{
				LastReplicateServer = CurrentReplicateServer;
				CurrentReplicateServer->replicate_id = LastReplicateServer->replicate_id;
			}
			CurrentReplicateServer = sp;
			return sp;
		}
		sp++;
	}
	sp = ReplicateServerData;
	while (sp->useFlag != DATA_END)
	{
		if (sp->useFlag != DATA_ERR )
		{
			if (CurrentReplicateServer != NULL)
			{
				LastReplicateServer = CurrentReplicateServer;
				CurrentReplicateServer->replicate_id = LastReplicateServer-> replicate_id;
			}
			CurrentReplicateServer = sp;
			PGR_Set_Replication_Server_Status(sp, DATA_USE);
			return sp;
		}
		sp++;
	}
	PGR_Stand_Alone->is_stand_alone = true;
	if (CurrentReplicateServer != NULL)
	{
	  LastReplicateServer = CurrentReplicateServer;
	  CurrentReplicateServer->replicate_id = LastReplicateServer-> replicate_id;
	}
	CurrentReplicateServer = NULL;
	return (ReplicateServerInfo *)NULL;
}



int
PGR_Send_Copy(CopyData * copy,int end )
{

	char cmdSts,cmdType;
	char * p = NULL;
	char *result = NULL;
	char term[8];
	int i;
	/*int status = 0; */

	if (copy == NULL)
	{
		return STATUS_ERROR;
	}

	cmdSts = CMD_STS_COPY;

	if (Transaction_Mode)
	{
		cmdSts = CMD_STS_TRANSACTION ;
	}
	if (Session_Authorization_Mode)
	{
		cmdSts = CMD_STS_SET_SESSION_AUTHORIZATION ;
	}
	cmdType = CMD_TYPE_COPY_DATA;

	if (end)
	{
		memset(term,0,sizeof(term));
		cmdType = CMD_TYPE_COPY_DATA_END;
		i = copy->cnt-1;
		if (copy->copy_data[i] == EOF)
		{
			p = &(copy->copy_data[i]);
			strcpy(term,"\\.\n");
			strncpy(p,term,sizeof(term));
			copy->cnt += 2;
		}
	}
	result = PGR_Send_Replicate_Command(copy->copy_data, copy->cnt, cmdSts, cmdType);
	memset(copy,0,sizeof(CopyData));

	if (result != NULL)
	{
		PGR_Reload_Start_Time();
		free(result);
		result = NULL;
		return STATUS_OK;
	}
	else
	{
		return STATUS_ERROR;
	}
}

CopyData * 
PGR_Set_Copy_Data(CopyData * copy, char *str, int len,int end)
{
	int copy_size, send_size;
	char * ep;
	int rest_len;
	CopyData rest_data;
	int status = STATUS_OK;

	if ((PGR_Copy_Data_Need_Replicate == false) ||
		(copy == NULL))
	{
		return (CopyData *)NULL;
	}
	if (len > 0)
	{
		copy_size = COPYBUFSIZ - copy->cnt -4;
		send_size = 0;
		if (copy_size < len)
		{
			memcpy(&(copy->copy_data[copy->cnt]) ,str + send_size,copy_size);
			copy->cnt  += copy_size;
			send_size += copy_size;
			len -= copy_size;
			ep = strrchr(copy->copy_data,'\n');
			if (ep != NULL)
			{
				*ep = '\0';
				rest_len = copy->cnt - strlen(copy->copy_data) -1;
				memset(&rest_data,0,sizeof(CopyData));
				memcpy(rest_data.copy_data,(ep +1),rest_len);
				copy->cnt -= (rest_len );
				*ep = '\n';
				*(ep+1) = '\0';
				status = PGR_Send_Copy(copy,0);
				memcpy(copy->copy_data,rest_data.copy_data,rest_len);
				copy->cnt = rest_len;
			}
			status = PGR_Send_Copy(copy,0);
			/*copy_size = COPYBUFSIZ - copy->cnt -1;*/
		}
		memcpy(&(copy->copy_data[copy->cnt]) ,str + send_size,len);
		copy->cnt  += len;
	}
	if (end)
	{
		status = PGR_Send_Copy(copy,end);
	}

	if (status != STATUS_OK)
	{
		return (CopyData *)NULL;
	}

	return copy;
}

int
PGR_replication(char * query_string, CommandDest dest, Node *parsetree, const char * commandTag)
{
	char *result = NULL;
	char cmdSts = CMD_STS_OTHER;
	char cmdType = CMD_TYPE_OTHER;
	int query_len = 0;

	if ((query_string == NULL) ||
		(commandTag == NULL))
	{
		return STATUS_ERROR;
	}

	Query_String = NULL;
	query_len = strlen(query_string);

	/* save query data for retry */
	PGR_Retry_Query.query_string = query_string;
	PGR_Retry_Query.query_len = query_len;
	PGR_Retry_Query.cmdSts = cmdSts;
	PGR_Retry_Query.cmdType = cmdType;
	PGR_Retry_Query.useFlag = DATA_USE;
	/* set cmdType */
	if (!strcmp(commandTag,"BEGIN")) cmdType = CMD_TYPE_BEGIN ;
	else if (!strcmp(commandTag,"COMMIT")) cmdType = CMD_TYPE_COMMIT ;
	else if (!strcmp(commandTag,"SELECT")) cmdType = CMD_TYPE_SELECT ;
	else if (!strcmp(commandTag,"INSERT")) cmdType = CMD_TYPE_INSERT ;
	else if (!strcmp(commandTag,"UPDATE")) cmdType = CMD_TYPE_UPDATE ;
	else if (!strcmp(commandTag,"DELETE")) cmdType = CMD_TYPE_DELETE ;
	else if (!strcmp(commandTag,"VACUUM")) cmdType = CMD_TYPE_VACUUM ;
	else if (!strcmp(commandTag,"ANALYZE")) cmdType = CMD_TYPE_ANALYZE ;
	else if (!strcmp(commandTag,"REINDEX")) cmdType = CMD_TYPE_REINDEX ;
	else if (!strcmp(commandTag,"ROLLBACK")) cmdType = CMD_TYPE_ROLLBACK ;
	else if (!strcmp(commandTag,"RESET")) cmdType = CMD_TYPE_RESET ;
	else if (!strcmp(commandTag,"START TRANSACTION")) cmdType = CMD_TYPE_BEGIN ;

	/* only "replication_server" statement-name is replicated for SHOW. */
	/*   see CreateCommandTag() @ backend/tcop/postgres.c      */
	/*
	else if (!strcmp(commandTag,"SHOW"))
	{
	}
	*/

	else if (!strcmp(commandTag,"COPY"))
	{
		cmdType = CMD_TYPE_COPY ;
		if (is_copy_from(query_string))
		{
			PGR_Copy_Data_Need_Replicate = true;
		}
		else
		{
			PGR_Copy_Data_Need_Replicate = false;
			return STATUS_NOT_REPLICATE;
		}
	}
	else if (!strcmp(commandTag,"SET")) 
	{
		VariableSetStmt *stmt = (VariableSetStmt *)parsetree;
		cmdType = CMD_TYPE_SET;

		if (strcmp(stmt->name, "TRANSACTION ISOLATION LEVEL") &&
			strcmp(stmt->name, "datestyle") &&
			strcmp(stmt->name, "autocommit") &&
			strcmp(stmt->name, "client_encoding") &&
			strcmp(stmt->name, "password_encryption") &&
			strcmp(stmt->name, "search_path") &&
			strcmp(stmt->name, "session_authorization") &&
			strcmp(stmt->name, "timezone"))

			return STATUS_NOT_REPLICATE;
	}
	else if (!strcmp(commandTag,"CREATE TABLE")) 
	{
		if (is_create_temp_table(query_string))
		{
			Create_Temp_Table_Mode = true;
		}
	}
	if (Create_Temp_Table_Mode)
	{
		cmdSts = CMD_STS_TEMP_TABLE ;
	}

	if ((is_transaction_command(commandTag) == true) ||
		(autocommit == false ))
	{
		Transaction_Mode = true;
	}
	if (Transaction_Mode)
	{
		cmdSts = CMD_STS_TRANSACTION ;
		if ((cmdType == CMD_TYPE_COMMIT ) ||
			(cmdType == CMD_TYPE_ROLLBACK ))
		{
			if (autocommit == true)
				Transaction_Mode = false;
			if (ReplicateCurrentTime != NULL)
			{
				ReplicateCurrentTime->useFlag = DATA_INIT;
				ReplicateCurrentTime->use_seed = 0;
			}
		}
	}
	else
	{
		if ((cmdType == CMD_TYPE_COMMIT ) ||
			(cmdType == CMD_TYPE_ROLLBACK ))
		{
			return STATUS_NOT_REPLICATE;
		}
	}
	if (Session_Authorization_Mode)
	{
		cmdSts = CMD_STS_SET_SESSION_AUTHORIZATION ;
		if (cmdType == CMD_TYPE_SESSION_AUTHORIZATION_END)
		{
			Session_Authorization_Mode = false;
		}
	}
	if ((cmdSts == CMD_STS_TRANSACTION ) ||
		(cmdSts == CMD_STS_SET_SESSION_AUTHORIZATION ) ||
		(cmdSts == CMD_STS_TEMP_TABLE ))
	{
		/* check partitional replication table */
		if (is_not_replication_query(query_string, query_len, cmdType)== true )
		{
			PGR_Copy_Data_Need_Replicate = false;
			return STATUS_NOT_REPLICATE;
		}
		Query_String = NULL;
		if (( do_not_replication_command(commandTag) == true) &&
			(strcmp(commandTag,"SELECT")))
		{
			return STATUS_NOT_REPLICATE;
		}

		if (Debug_pretty_print)
			elog(DEBUG1,"transaction query send :%s",(char *)query_string);
		PGR_Retry_Query.cmdSts = cmdSts;
		PGR_Retry_Query.cmdType = cmdType;
		result = PGR_Send_Replicate_Command(query_string,query_len, cmdSts,cmdType);
		if (result != NULL)
		{
			if (!strncmp(result,PGR_DEADLOCK_DETECTION_MSG,strlen(PGR_DEADLOCK_DETECTION_MSG)))
			{
				/*
				PGR_Send_Message_To_Frontend(result);
				*/
				free(result);
				result = NULL;
				return STATUS_DEADLOCK_DETECT;
			}
			else if (!strncmp(result,PGR_REPLICATION_ABORT_MSG,strlen(PGR_REPLICATION_ABORT_MSG)))
			{
				free(result);
				result = NULL;
				return STATUS_REPLICATION_ABORT;
			}
			free(result);
			result = NULL;
			return STATUS_CONTINUE;
		}
		else
		{
			return STATUS_ERROR;
		}
	}
	else
	{
		cmdSts = CMD_STS_QUERY ;
		if ( do_not_replication_command(commandTag) == false)
		{
			Query_String = NULL;
			/* check partitional replication table */
			if (is_not_replication_query(query_string, query_len, cmdType)== true )
			{
				PGR_Copy_Data_Need_Replicate = false;
				return STATUS_NOT_REPLICATE;
			}
			result = PGR_Send_Replicate_Command(query_string,query_len,cmdSts,cmdType);
			if (result != NULL)
			{
				if (!strncmp(result,PGR_DEADLOCK_DETECTION_MSG,strlen(PGR_DEADLOCK_DETECTION_MSG)))
				{
					free(result);
					result = NULL;
					return STATUS_DEADLOCK_DETECT;
				}
				else if (!strncmp(result,PGR_REPLICATION_ABORT_MSG,strlen(PGR_REPLICATION_ABORT_MSG)))
				{
					free(result);
					result = NULL;
					return STATUS_REPLICATION_ABORT;
				}
				/*
				PGR_Send_Message_To_Frontend(result);
				*/
				free(result);
				result = NULL;
				return STATUS_CONTINUE;
			}
			else
			{
				return STATUS_ERROR;
			}
		}
		else
		{
			if (( is_serial_control_query(cmdType,query_string) == true) ||
				( is_select_into_query(cmdType,query_string) == true))
			{
				Query_String = NULL;
				PGR_Need_Notice = true;
				PGR_Check_Lock.check_lock_conflict = true;
				result = PGR_Send_Replicate_Command(query_string,query_len,cmdSts,cmdType);
				if (result != NULL)
				{
					/*
					PGR_Send_Message_To_Frontend(result);
					*/
					if (!strncmp(result,PGR_DEADLOCK_DETECTION_MSG,strlen(PGR_DEADLOCK_DETECTION_MSG)))
					{
						free(result);
						return STATUS_DEADLOCK_DETECT;
					}
					free(result);
					result = NULL;
					return STATUS_CONTINUE;
				}
				else
				{
					return STATUS_ERROR;
				}
			}
			else
			{
				Query_String = query_string;
				/*PGR_Sock_To_Replication_Server = -1;*/
			}
			return STATUS_CONTINUE_SELECT;
		}
	}
	return STATUS_CONTINUE;
}


bool
PGR_Is_System_Command(char * query)
{
	char * ptr;

	if (query == NULL)
	{
		return false;
	}
	ptr = strstr(query,PGR_SYSTEM_COMMAND_FUNC);
	if (ptr != NULL)
	{
		ptr = strchr(ptr,'(');
		if (ptr == NULL)
			return false;
		return true;
	}
	return false;
}

static int
set_command_args(char argv[ PGR_CMD_ARG_NUM ][256],char *str)
{
	int i,j,cnt,len;
	char * ptr = str;

	if (str == NULL)
	{
		return 0;
	}
	len = strlen(str);
	cnt = j = 0;
	for ( i = 0 ; i < len ; i++,ptr++)
	{
		if (cnt >= PGR_CMD_ARG_NUM)
			break;
		if (( *ptr == ',') || (*ptr == ')'))
		{
			argv[cnt][j] = '\0';
			cnt ++;
			j = 0;
			continue;
		}
		argv[cnt][j] = *ptr;
		j++;
	}
	if (cnt < PGR_CMD_ARG_NUM)
		argv[cnt][j] = '\0';
	cnt ++;

	return cnt;
}

static int
add_replication_server(char * hostname,char * port, char * recovery_port)
{
	int cnt;
	int portNumber;
	int recoveryPortNumber;
	ReplicateServerInfo * sp;

	if ((hostname == NULL) ||
		(port == NULL ) ||
		(recovery_port == NULL ))
	{
		return STATUS_ERROR;
	}
	if (ReplicateServerData == NULL)
	{
		return STATUS_ERROR;
	}
	portNumber = atoi(port);
	recoveryPortNumber = atoi(recovery_port);
	cnt = 0;
	sp = ReplicateServerData;
	while (sp->useFlag != DATA_END){
		if((!strncmp(sp->hostName,hostname,sizeof(sp->hostName))) &&
			(sp->portNumber == portNumber) &&
			(sp->recoveryPortNumber == recoveryPortNumber))
		{
			if (sp->useFlag != DATA_USE)
			{
				PGR_Set_Replication_Server_Status(sp, DATA_INIT);
			}
			return STATUS_OK;
		}
		sp ++;
		cnt ++;
	}
	if (cnt < MAX_SERVER_NUM)
	{
		strncpy(sp->hostName,hostname,sizeof(sp->hostName));
		sp->portNumber = portNumber;
		sp->recoveryPortNumber = recoveryPortNumber;
		PGR_Set_Replication_Server_Status(sp, DATA_INIT);
		memset((sp+1),0,sizeof(ReplicateServerInfo));
		(sp + 1)->useFlag = DATA_END;
	}
	else
	{
		return STATUS_ERROR;
	}
	return	STATUS_OK;
}

static int
change_replication_server(char * hostname,char * port, char * recovery_port)
{
	int cnt;
	int portNumber;
	int recoveryPortNumber;
	ReplicateServerInfo * sp;

	if ((hostname == NULL) ||
		(port == NULL ) ||
		(recovery_port == NULL ))
	{
		return STATUS_ERROR;
	}
	if (ReplicateServerData == NULL)
	{
		return STATUS_ERROR;
	}
	portNumber = atoi(port);
	recoveryPortNumber = atoi(recovery_port);
	cnt = 0;
	sp = ReplicateServerData;
	while (sp->useFlag != DATA_END){
		if((!strcmp(sp->hostName,hostname)) &&
			(sp->portNumber == portNumber) &&
			(sp->recoveryPortNumber == recoveryPortNumber))
		{
			PGR_Set_Replication_Server_Status(sp, DATA_USE);
		}
		else
		{
			if (sp->useFlag == DATA_USE)
			{
				PGR_Set_Replication_Server_Status(sp, DATA_INIT);
			}
		}
		sp ++;
		cnt ++;
	}
	return	STATUS_OK;
}

int
PGR_Set_Current_Time(char * sec, char * usec)
{
	int rtn = 0;
	struct timeval local_tp;
	struct timezone local_tpz;
	struct timeval tv;

	if ((sec == NULL) ||
		(usec == NULL))
	{
		return STATUS_ERROR;
	}
	rtn = gettimeofday(&local_tp, &local_tpz);
	tv.tv_sec = atol(sec);
	tv.tv_usec = atol(usec);
	ReplicateCurrentTime->offset_sec = local_tp.tv_sec - tv.tv_sec;
	ReplicateCurrentTime->offset_usec = local_tp.tv_usec - tv.tv_usec;
	ReplicateCurrentTime->tp.tv_sec = tv.tv_sec;
	ReplicateCurrentTime->tp.tv_usec = tv.tv_usec;
	ReplicateCurrentTime->useFlag = DATA_USE;
	ReplicateCurrentTime->use_seed = 0;

	return	STATUS_OK;
}

static void
PGR_Set_Current_Replication_Query_ID(char *id) {
      MyProc->replicationId=atol(id);
      return;
}

static void
set_response_mode(char * mode)
{
	int response_mode = 0;

	if (mode == NULL)
		return;
	response_mode = atoi(mode);
	if (response_mode < 0)
		return;
	if (CurrentReplicateServer == NULL)
	{
		PGR_get_replicate_server_info();
		if (CurrentReplicateServer == NULL)
		{
			return;
		}
	}
	if (CurrentReplicateServer->response_mode != response_mode)
	{
		CurrentReplicateServer->response_mode = response_mode;
	}
}

int
PGR_Call_System_Command(char * command)
{
	char * ptr;
	char * args;
	char argv[ PGR_CMD_ARG_NUM ][256];
	int argc = 0;
	int func_no;
	char * hostName = NULL;

	if ((command == NULL) || (ReplicateCurrentTime == NULL))
	{
		return STATUS_ERROR;
	}
	ptr = strstr(command,PGR_SYSTEM_COMMAND_FUNC);
	if (ptr == NULL)
		return STATUS_ERROR;
	ptr = strchr(ptr,'(');
	if (ptr == NULL)
		return STATUS_ERROR;
	args = ptr+1;
	ptr = strchr(ptr,')');
	if (ptr == NULL)
		return STATUS_ERROR;
	*ptr = '\0';
	argc = set_command_args(argv,args);
	if (argc < 1)
		return STATUS_ERROR;
	func_no = atoi(argv[0]);
	switch (func_no)
	{
		/* set current system time */
		case PGR_SET_CURRENT_TIME_FUNC_NO:
			if (atol(argv[1]) == 0)
			{
				CreateCheckPoint(false,true);
			}
			else
			{
			  /*
			  if ((atoi(argv[3]) > 0) &&
				(is_this_query_replicated(argv[3]) == true))
			    {				
			      return STATUS_SKIP_QUERY;
			    }
			  */
				PGR_Set_Current_Time(argv[1],argv[2]);
				set_replication_id(argv[3]);
				set_response_mode(argv[4]);
				PGR_Set_Current_Replication_Query_ID(argv[5]);

			}
			break;
		/* add new replication server data */
		case PGR_STARTUP_REPLICATION_SERVER_FUNC_NO:
			hostName = get_hostName(argv[1]);
			add_replication_server(hostName,argv[2],argv[3]);
			break;
		/* change new replication server */
		case PGR_CHANGE_REPLICATION_SERVER_FUNC_NO:
			hostName = get_hostName(argv[1]);
			change_replication_server(hostName,argv[2],argv[3]);
			break;
        	case PGR_SET_CURRENT_REPLICATION_QUERY_ID_NO:
		  PGR_Set_Current_Replication_Query_ID(argv[1]);
			break;
		case PGR_QUERY_CONFIRM_ANSWER_FUNC_NO:
			if ((atoi(argv[3]) > 0) &&
				(is_this_query_replicated(argv[3]) == true))
			{
				/* skip this query */
			  return STATUS_SKIP_QUERY;
			}
			else
			{
				PGR_Set_Current_Time(argv[1],argv[2]);
				set_replication_id(argv[3]);
			}
			break;
	}
	return STATUS_OK;
}

int
PGR_GetTimeOfDay(struct timeval *tp, struct timezone *tpz)
{

	int rtn;

	rtn = gettimeofday(tp, tpz);
	if (ReplicateCurrentTime == NULL)
	{
		return rtn;
	}
	if (ReplicateCurrentTime->useFlag == DATA_USE)
	{
		if (ReplicateCurrentTime->use_seed > 0)
		{
			tp->tv_sec -= ReplicateCurrentTime->offset_sec;
			if (tp->tv_usec < ReplicateCurrentTime->offset_usec)
			{
				tp->tv_usec += (1000000 -  ReplicateCurrentTime->offset_usec);
				tp->tv_sec -= 1;
			}
			else
			{
				tp->tv_usec -= ReplicateCurrentTime->offset_usec;
			}
		}
		else
		{
			tp->tv_sec = ReplicateCurrentTime->tp.tv_sec;
			tp->tv_usec = ReplicateCurrentTime->tp.tv_usec;
			ReplicateCurrentTime->use_seed = 1;
		}
		rtn = 0;
	}
	return rtn;
}

long
PGR_Random(void)
{
	double rtn;
	if (ReplicateCurrentTime != NULL)
	{
		if ( ReplicateCurrentTime->use_seed == 0)
		{
			srand( ReplicateCurrentTime->tp.tv_usec );
			ReplicateCurrentTime->use_seed = 1;
		}
	}
	rtn = random();
	return rtn;
}

char *
PGR_scan_terminate( char * str)
{
	char * p;
	int sflag = 0;
	int dflag = 0;

	if (str == NULL)
		return NULL;
	p = str;
	while ( *p != '\0' )
	{
		if ((!strncmp(p,"--",2)) ||
			(!strncmp(p,"//",2)))
		{
			while (( *p != '\n') && (*p != '\0'))
			{
				p++;
			}
			continue;
		}

		switch (*p)
		{
			case '\'':
				sflag ^= 1;
				break;
			case '\"':
				dflag ^= 1;
				break;
			case '\\':
				p +=2;
				continue;
				break;
			case ';':
				if ((!sflag) && (!dflag))
					return p;
				break;
		}
		p++;
	}
	return NULL;
}

static bool
is_copy_from(char * query)
{
	char * p;
	int i;
	char buf[12];
	int c_flag = 0;
	if (query == NULL)
		return false;
	p = query;
	for ( i = 0 ; i <= 1 ; i ++)
	{
		/* get 'copy table_name' string */
		while(isspace(*p))
			p++;
		while ((*p != '\0') && (*p  != '(') && (!isspace(*p)))
			p++;
	}
	while(isspace(*p))
		p++;
	/* skip table column */
	if (*p == '(')
	{
		c_flag = 1;
		p++;
		while (*p != '\0') 
		{
			if (*p == '(')
				c_flag ++;
			if (*p == ')')
				c_flag --;
			if (c_flag == 0)
			{
				p++;
				break;
			}
			p++;
		}
		while(isspace(*p))
			p++;
	}
	/* get 'from' or 'to' */
	i = 0;
	memset(buf,0,sizeof(buf));
	while ((*p != '\0') && (!isspace(*p)) && ( i < sizeof(buf)-1))
	{
		buf[i] = (char)toupper(*p);
		p++;
		i++;
	}
	if (!strcmp(buf,"FROM"))
	{
		return true;
	}
	else
	{
		return false;
	}
}

static bool
is_create_temp_table(char * query)
{
	int len,wc;
	char buf[MAX_WORDS][MAX_WORD_LETTERS];

	if (query == NULL)
		return false;
	len = strlen(query);
	wc = get_words(buf,query,len,1);
	if (wc < 4)
		return false;
	if ((!strncmp(buf[0],"CREATE", strlen("CREATE"))) &&
		(!strncmp(buf[1],"TEMP",strlen("TEMP"))) &&
		(!strncmp(buf[2],"TABLE",strlen("TABLE"))))
	{
		return true;
	}
	return false;
}

#if 0
static int
is_session_authorization(char * query)
{

	int len,wc;
	int cnt,rtn;
	char buf[MAX_WORDS][MAX_WORD_LETTERS];

	if (query == NULL)
		return NOT_SESSION_AUTHORIZATION;

	rtn = NOT_SESSION_AUTHORIZATION;
	len = strlen(query);
	wc = get_words(buf,query,len,1);
	if (wc < 4)
	{
		return NOT_SESSION_AUTHORIZATION;
	}
	if ((!strncmp(buf[0],"RESET", strlen("RESET"))) &&
		(!strncmp(buf[1],"SESSION",strlen("SESSION"))) &&
		(!strncmp(buf[2],"AUTHORIZATION",strlen("AUTHORIZATION"))))
	{
		return SESSION_AUTHORIZATION_END;
	}
	cnt = 1;
	if (!strcmp(buf[cnt],"SESSION"))
	{
		cnt ++;
		if (!strcmp(buf[cnt],"AUTHORIZATION"))
		{
			rtn = SESSION_AUTHORIZATION_BEGIN;
		}
		else
		{
			cnt ++;
			if (!strcmp(buf[cnt],"AUTHORIZATION"))
			{
				rtn = SESSION_AUTHORIZATION_BEGIN;
			}
		}
		cnt ++;
		if (!strcmp(buf[cnt],"DEFAULT"))
		{
			rtn = SESSION_AUTHORIZATION_END;
		}
	}
	return rtn;
}
#endif

static int
get_words( char words[MAX_WORDS][MAX_WORD_LETTERS] ,char * string,int length,int upper)
{
	int i,wc,lc;
	char * p = NULL;
	char * buf = NULL;

	if (string == NULL)
		return STATUS_ERROR;
	buf = malloc(length);
	if (buf == NULL)
		return STATUS_ERROR;

	memset(buf,0,length);
	p = string;
	wc = 0;
	for (i = 0 ; i < length ; i ++)
	{
		if ((*p == '\0') || (wc >= MAX_WORDS))
			break;
		while (isspace(*p))
		{
			p++;
			i++;
		}
		lc = 0;
		while ((*p != '\0') && (! isspace(*p)))
		{
			if (upper)
				*(buf+lc) = (char)toupper(*p);
			else
				*(buf+lc) = *p;

			p++;
			i++;
			lc++;
		}
		memset(words[wc],0,MAX_WORD_LETTERS);
		memcpy(words[wc],buf,lc);
		memset(buf,0,length);
		wc++;
	}
	free(buf);
	buf = NULL;
	return wc;
}

static int
Comp_Not_Replicate(PGR_Not_Replicate_Type * nrp1,PGR_Not_Replicate_Type* nrp2)
{
	int rtn;

	if ((nrp1 == NULL) ||
		(nrp2 == NULL))
	{
		return 0;
	}
	rtn = strcasecmp(nrp1->table_name,nrp2->table_name);
	if (rtn == 0)
	{
		rtn = strcasecmp(nrp1->db_name,nrp2->db_name);
	}
	return rtn;
}

bool
PGR_Is_Stand_Alone(void)
{
	ReplicateServerInfo * sp = NULL;

	if (PGR_Stand_Alone == NULL)
		return true;
	if (PGR_Stand_Alone->is_stand_alone == true)
	{
		sp = PGR_get_replicate_server_info();
		if (sp == NULL)
		{
			return true;
		}
	}
	return false;
}

void
PGR_Send_Message_To_Frontend(char * msg)
{
	StringInfoData buf;

	if (msg == NULL)
		return;
	switch (*msg)
	{
		/* normal message */
		case 'C':
			pq_puttextmessage(*msg,msg+1);
			break;
		/* error message */
		case 'E':
		case 'N':
			initStringInfo(&buf);
			pq_sendbyte(&buf,*msg);
			pq_sendstring(&buf, msg+1);
			pq_endmessage(&buf);
			break;
	}
	pq_flush();
}

static bool
is_serial_control_query(char cmdType,char * query)
{
	char * buf = NULL;
	int len = 0;
	int i = 0;
	char * p = NULL;

	if ((cmdType != CMD_TYPE_SELECT ) ||
		( query == NULL))
	{
		return false;
	}

	p = query;
	len = strlen(query) +1;
	buf = malloc(len);
	if (buf == NULL)
		return false;

	memset(buf,0,len);
	for ( i = 0 ; i < len ; i ++)
	{
		*(buf+i) = toupper(*(query+i));
	}
	if ((strstr(buf,"NEXTVAL") != NULL) ||
		(strstr(buf,"SETVAL") != NULL))
	{
		free(buf);
		buf = NULL;
		return true;
	}
	free(buf);
	buf = NULL;
	return false;
}

static bool
is_select_into_query(char cmdType,char * query)
{
	char * buf = NULL;
	int len = 0;
	int i = 0;
	char * p = NULL;

	if ((cmdType != CMD_TYPE_SELECT ) ||
		( query == NULL))
	{
		return false;
	}

	p = query;
	len = strlen(query) +1;
	buf = malloc(len);
	if (buf == NULL)
		return false;

	memset(buf,0,len);
	for ( i = 0 ; i < len ; i ++)
	{
		*(buf+i) = toupper(*(query+i));
	}
	if (strstr(buf,"INTO") != NULL)
	{
		free(buf);
		buf = NULL;
		return true;
	}
	if (strstr(buf,"CREATE") != NULL)
	{
		free(buf);
		buf = NULL;
		return true;
	}
	free(buf);
	buf = NULL;
	return false;
}

static int
send_response_to_replication_server(const char * notice)
{
	ReplicateHeader header;
	int status;

	if (PGR_Lock_Noticed)
	{
		return STATUS_OK;
	}
	if ((notice == NULL) ||
		(PGR_Sock_To_Replication_Server < 0))
	{
		return STATUS_ERROR;
	}

	memset(&header,0,sizeof(ReplicateHeader));
	header.cmdSys = CMD_SYS_CALL;
	header.cmdSts = CMD_STS_RESPONSE;
	if (!strcmp(notice,PGR_QUERY_ABORTED_NOTICE_CMD))
	{
		header.cmdType = CMD_TYPE_FRONTEND_CLOSED;
	}
	header.query_size = htonl(strlen(notice));
	status = send_replicate_packet(PGR_Sock_To_Replication_Server,&header,(char *)notice);
	return status;
}

void
PGR_Notice_Transaction_Query_Done(void)
{
	send_response_to_replication_server(PGR_QUERY_DONE_NOTICE_CMD);
}

void
PGR_Notice_Transaction_Query_Aborted(void)
{
	send_response_to_replication_server(PGR_QUERY_ABORTED_NOTICE_CMD);
}

int
PGR_Notice_Conflict(void)
{
	const char * msg = NULL ;
	int rtn = STATUS_OK;

	msg = PGR_LOCK_CONFLICT_NOTICE_CMD ;
	if (PGR_Check_Lock.deadlock == true)
	{
		msg = PGR_DEADLOCK_DETECT_NOTICE_CMD ;
	}
	if (PGR_Check_Lock.dest == TO_FRONTEND)
	{
		pq_puttextmessage('C',msg);
		pq_putbytes("Z",1);
		pq_flush();
#ifdef CONTROL_LOCK_CONFLICT
		rtn = wait_lock_answer();
#endif /* CONTROL_LOCK_CONFLICT */
	}
	else
	{
		send_response_to_replication_server(msg);
#ifdef CONTROL_LOCK_CONFLICT
		rtn = PGR_Recv_Trigger (PGR_RECV_TIMEOUT);
#endif /* CONTROL_LOCK_CONFLICT */
	}
	return rtn;
}

#ifdef CONTROL_LOCK_CONFLICT
static int
wait_lock_answer(void)
{
	char result[PGR_MESSAGE_BUFSIZE+4];
	int rtn = 0;

	memset(result,0,sizeof(result));
	rtn = read_trigger(result, PGR_MESSAGE_BUFSIZE);
	if (rtn < 0)
		return STATUS_ERROR;
	return STATUS_OK;
}

static int
read_trigger(char * result, int buf_size)
{
	int i = 0;
	char c;
	int r = 0;

	if ((result == NULL) || (buf_size <= 0 ))
	{
		return EOF;
	}
	/*
	pq_getbytes(result,buf_size);
	*/
	while ((r = pq_getbytes(&c,1)) == 0)
	{
		if (i < buf_size -1)
		{
			*(result + i) = c;
		}
		else
		{
			break;
		}
		if (c == '\0')
			break;
		i++;
	}

	return r;
}
#endif /* CONTROL_LOCK_CONFLICT */

int
PGR_Recv_Trigger (int user_timeout)
{
	char result[PGR_MESSAGE_BUFSIZE];
	int rtn = 0;
	int func_no = 0;

	
	if (PGR_Lock_Noticed)
	{
		return STATUS_OK;
	}
	if (PGR_Sock_To_Replication_Server < 0)
		return STATUS_ERROR;
	memset(result,0,sizeof(result));
	rtn = recv_replicate_result(PGR_Sock_To_Replication_Server,result,user_timeout);
	if (rtn > 0)
	{
		func_no = atoi(result);
		if (func_no  <= 0)
		{
			func_no = STATUS_OK;
		}
		return func_no;
	}
	else 
	{
		if (user_timeout == 0)
		{
			PGR_Set_Replication_Server_Status(CurrentReplicateServer, DATA_ERR);
		}
		return STATUS_ERROR;
	}
	return STATUS_OK;
}

bool
PGR_Set_Transaction_Mode(bool mode,const char * commandTag)
{
	if (commandTag == NULL)
	{
		return mode;
	}
	if (autocommit == false)
	{
		return true;
	}
	if (mode == false)
	{
		if ((!strcmp(commandTag,"BEGIN")) ||
			(!strcmp(commandTag,"START TRANSACTION")) )
		{
			return true;
		}
	}
	if (mode == true)
	{
		if ((!strcmp(commandTag,"COMMIT")) ||
			(!strcmp(commandTag,"ROLLBACK")))
		{
			return false;
		}
	}
	return mode;
}


static bool
is_transaction_command(const char * commandTag)
{
	if (commandTag == NULL)
	{
		return false;
	}
	if ((!strcmp(commandTag,"BEGIN")) ||
		(!strcmp(commandTag,"START TRANSACTION")))
		/*
		(!strcmp(commandTag,"COMMIT")) ||
		(!strcmp(commandTag,"ROLLBACK")))
		*/
	{
		return true;
	}
	else
	{
		return false;
	}
}

static bool
do_not_replication_command(const char * commandTag)
{
	if (commandTag == NULL)
	{
		return true;
	}
	if ((!strcmp(commandTag,"SELECT")) ||
		(!strcmp(commandTag,"CLOSE CURSOR")) ||
		(!strcmp(commandTag,"MOVE")) ||
		(!strcmp(commandTag,"FETCH")) ||
		(!strcmp(commandTag,"EXPLAIN")))
	{
		return true;
	}
	else
	{
		return false;
	}
}

bool
PGR_Check_DeadLock(
	LOCKMETHODTABLE *lockMethodTable,
	LOCKMODE lockmode,
	LOCK *lock,
	PROCLOCK *holder)
{
	PROC_QUEUE *waitQueue = &(lock->waitProcs);
	int			myHeldLocks = MyProc->heldLocks;
	bool		early_deadlock = false;
	PGPROC	   *proc;
	int			i;

	if (myHeldLocks != 0)
	{
		proc = (PGPROC *) MAKE_PTR(waitQueue->links.next);
		for (i = 0; i < waitQueue->size; i++)
		{
			/* Must he wait for me? */
			if (lockMethodTable->conflictTab[proc->waitLockMode] & myHeldLocks)
			{
				/* Must I wait for him ? */
				if (lockMethodTable->conflictTab[lockmode] & proc->heldLocks)
				{
					early_deadlock = true;
					return true;;
				}
			}
			/* Nope, so advance to next waiter */
			proc = (PGPROC *) MAKE_PTR(proc->links.next);
		}
	}
	else
	{
		/* I hold no locks, so I can't push in front of anyone. */
		proc = (PGPROC *) &(waitQueue->links);
	}

	/*
	 * Insert self into queue, ahead of the given proc (or at tail of
	 * queue).
	 */
	SHMQueueInsertBefore(&(proc->links), &(MyProc->links));
	waitQueue->size++;

	lock->waitMask |= (1 << lockmode);

	/* Set up wait information in PGPROC object, too */
	MyProc->waitLock = lock;
	MyProc->waitHolder = holder;
	MyProc->waitLockMode = lockmode;

	MyProc->errType = STATUS_OK;	/* initialize result for success */

	return false;
}

void
PGR_Set_Replication_Server_Status( ReplicateServerInfo * sp, int status)
{
	if (sp == NULL)
	{
		return;
	}
	if (sp->useFlag != status)
	{
		sp->useFlag = status;
	}
}

int
PGR_Is_Skip_Replication(char * query)
{
	char skip_2[256];

	if ((query == NULL) ||
		(MyProcPort == NULL))
	{
		return -1;
	}
	snprintf(skip_2,sizeof(skip_2),SKIP_QUERY_2,MyProcPort->user_name);
	if ((strncmp(query,SKIP_QUERY_1,strlen(SKIP_QUERY_1)) == 0) ||
		(strncmp(query,skip_2,strlen(skip_2)) == 0))
	{
		return 3;
	}
	if ((strncmp(query,SKIP_QUERY_3,strlen(SKIP_QUERY_3)) == 0) ||
		(strncmp(query,SKIP_QUERY_4,strlen(SKIP_QUERY_4)) == 0))
	{
		return 1;
	}
	return 0;
}

bool
PGR_Did_Commit_Transaction(void)
{

	int sock = -1;
	int cnt = 0;
	ReplicateHeader header;
	char * serverName = NULL;
	int portNumber=0;
	char * result = NULL;
	ReplicateServerInfo * sp = NULL;
	ReplicateServerInfo * base = NULL;
	int socket_type = 0;
	char argv[ PGR_CMD_ARG_NUM ][256];
	int argc = 0;
	int func_no = 0;

	if (ReplicateCurrentTime->useFlag != DATA_USE)
	{
		return false;
	}
	sp = PGR_get_replicate_server_info();
	if (sp == NULL)
	{
		if (Debug_pretty_print)
			elog(DEBUG1,"PGR_get_replicate_server_info get error");
		return false;
	}
	sock = get_replicate_server_socket( sp , PGR_QUERY_SOCKET);
	if (sock < 0)
	{
		if (Debug_pretty_print)
			elog(DEBUG1,"get_replicate_server_socket fail");
		return false;
	}
	result = malloc(PGR_MESSAGE_BUFSIZE);
	if (result == NULL)
	{
		return false;
	}
	memset(result,0,PGR_MESSAGE_BUFSIZE);

	serverName = sp->hostName;
	portNumber = (int)sp->portNumber;
	header.cmdSys = CMD_SYS_CALL;
	header.cmdSts = CMD_STS_TRANSACTION_ABORT;
	header.cmdType = CMD_TYPE_COMMIT_CONFIRM;
	header.port = htons(PostPortNumber);
	header.pid = htons(getpid());
	header.tv.tv_sec = htonl(ReplicateCurrentTime->tp.tv_sec);
	header.tv.tv_usec = htonl(ReplicateCurrentTime->tp.tv_usec);
	header.query_size = htonl(0); 
	strncpy(header.dbName ,(char *)(MyProcPort->database_name),sizeof(header.dbName)-1);
	strncpy(header.userName , (char *)(MyProcPort->user_name),sizeof(header.userName)-1);
	if (PGRSelfHostName != NULL)
	{
		strncpy(header.from_host, PGRSelfHostName, HOSTNAME_MAX_LENGTH);
	}
	header.replicate_id = htonl(ReplicationLog_Info.PGR_Replicate_ID);
	header.request_id = 0;

	base = sp;
	PGR_Sock_To_Replication_Server = sock;

retry_send_confirm_packet:

	cnt = 0;
	while (send_replicate_packet(sock,&header,"") != STATUS_OK)
	{
		if (cnt > MAX_RETRY_TIMES )
		{
			sock = get_new_replication_socket( base, sp, socket_type);
			if (sock < 0)
			{
				if (Debug_pretty_print)
					elog(DEBUG1,"all replication servers may be down");
				PGR_Stand_Alone->is_stand_alone = true;
				free(result);
				result = NULL;
				return false;
			}
			PGR_Sock_To_Replication_Server = sock;
			cnt = 0;
		}
		cnt ++;
	}

	if (recv_replicate_result(sock,result,0) < 0)
	{
		/* replication server should be down */
		sock = get_new_replication_socket( base, sp, socket_type);
		if (sock < 0)
		{
			if (Debug_pretty_print)
				elog(DEBUG1,"all replication servers may be down");
			PGR_Stand_Alone->is_stand_alone = true;
			free(result);
			result = NULL;
			return false;
		}
		PGR_Sock_To_Replication_Server = sock;
		goto retry_send_confirm_packet;
	}
	/* read answer */
	argc = set_command_args(argv,result);
	if (argc >= 1)
	{
		func_no = atoi(argv[0]);
		if (func_no == PGR_TRANSACTION_CONFIRM_ANSWER_FUNC_NO)
		{
			/* the transaction was commited in other server */
			if (atoi(argv[1]) == PGR_ALREADY_COMMITTED)
			{
				free(result);
				result = NULL;
				return true;
			}
		}
	}
	free(result);
	result = NULL;
	return false;
}

static char *
get_hostName(char * str)
{
	char * top = NULL;
	char * p = NULL;

	p = str;
	while ( *p != '\0')
	{
		if (*p == '\'')
		{
			*p = '\0';
			p++;
			if (top == NULL)
			{
				top = p;
			}
		}
		p++;
	}
	return top;
}

char *
PGR_Remove_Comment(char * str)
{
	char * p = NULL;
	p = str;
	while( *p != '\0')
	{
		while(isspace(*p))
		{
			p++;
		}
		if ((!memcmp(p,"--",2)) ||
			(!memcmp(p,"//",2)))
		{
			while((*p != '\n') && (*p != '\0'))
			{
				p++;
			}
			continue;
		}
		break;
	}
	return p;
}

void
PGR_Force_Replicate_Query(void)
{
	if (PGR_Retry_Query.useFlag == DATA_USE)
	{
		PGR_Send_Replicate_Command(PGR_Retry_Query.query_string,
			PGR_Retry_Query.query_len,
			PGR_Retry_Query.cmdSts,
			PGR_Retry_Query.cmdType);
	}
}

void
PGR_Notice_DeadLock(void)
{
	ReplicateHeader header;

	memset(&header,0,sizeof(ReplicateHeader));
	header.cmdSys = CMD_SYS_CALL;
	header.cmdSts = CMD_STS_NOTICE;
	header.cmdType = CMD_TYPE_DEADLOCK_DETECT;
	header.query_size = 0;
	send_replicate_packet(PGR_Sock_To_Replication_Server,&header,(char *)NULL);
}

void
PGR_Set_Cluster_Status(int status)
{
	if (ClusterDBData != NULL)
	{
		if (ClusterDBData->status != status)
		{
			ClusterDBData->status = status;
		}
	}
}

int
PGR_Get_Cluster_Status(void)
{
	if (ClusterDBData != NULL)
	{
		return (ClusterDBData->status);
	}
	return 0;
}

int
PGR_Check_Replicate_Server_Status(ReplicateServerInfo * sp)
{
	ReplicateHeader header;
	char * result = NULL;
	int status;
	int fdP;

	result = malloc(PGR_MESSAGE_BUFSIZE + 4);
	if (result == NULL)
	{
		if (Debug_pretty_print)
			elog(DEBUG1,"malloc failed in PGR_Check_Replicate_Server_Status()");
		return STATUS_ERROR;
	}

	memset(&header, 0, sizeof(ReplicateHeader));
	memset(result,  0, PGR_MESSAGE_BUFSIZE + 4);

	header.cmdSys = CMD_SYS_PREREPLICATE;
	header.cmdSts = CMD_STS_OTHER;
	header.cmdType = CMD_TYPE_OTHER;
	header.port = htons(PostPortNumber);
	header.pid = htons(getpid());
	header.query_size = 0;
	strncpy(header.dbName ,(char *)(MyProcPort->database_name),sizeof(header.dbName)-1);
	strncpy(header.userName , (char *)(MyProcPort->user_name),sizeof(header.userName)-1);
	header.request_id = htonl(get_next_request_id());
	header.rlog = 0;
	if (PGRSelfHostName != NULL) {
		strncpy(header.from_host, PGRSelfHostName, HOSTNAME_MAX_LENGTH);
	}

	/* open a new socket for lifecheck */
	if ((status = PGR_Create_Socket_Connect(&fdP, sp->hostName, sp->portNumber)) == STATUS_ERROR) {
		if (Debug_pretty_print) {
			elog(DEBUG1,"create socket failed in PGR_Create_Socket_Connect()");
		}
		
	/* status = STATUS_OK */
	} else {
		if ((status = send_replicate_packet(fdP, &header, (char *)NULL)) == STATUS_OK) {
			/* receive result to check for possible deadlock */
			status = (0 < recv_replicate_result(fdP, result ,0))
				? STATUS_OK : STATUS_ERROR;
		}
	}

	free(result);
	PGR_Close_Sock(&fdP);

	return status;
}

#if 0
void
PGR_Update_Replicate_Server_Status(ReplicateServerInfo * sp)
{
	ReplicateServerInfo * tmp;
	ReplicateServerInfo * base;

	tmp = PGR_get_replicate_server_info();
	base = tmp;
	get_new_replication_socket(base, tmp, PGR_QUERY_SOCKET);
	
/*
	PGR_Set_Replication_Server_Status(sp, DATA_ERR);
	search_new_replication_server(sp, PGR_QUERY_SOCKET);
	if (sp == NULL) {
		if (Debug_pretty_print) {
			elog(DEBUG1,"all replication servers may be down");
		}
		PGR_Stand_Alone->is_stand_alone = true;
	}
*/
}
#endif

#endif /* USE_REPLICATION */
