#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/xact.h"
#ifndef POSTGRES81
 #include "catalog/catname.h"
#endif
#include "catalog/indexing.h"
#include "catalog/pg_operator.h"
#include "catalog/pg_trigger.h"
#include "commands/trigger.h"
#include "commands/async.h"
#include "executor/spi.h"
#include "fmgr.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "parser/parse_type.h"
#include "storage/lwlock.h"
#include "storage/proc.h"
#ifdef POSTGRES81
 #include "storage/procarray.h"
 #include "storage/shmem.h"
#else
 #include "storage/sinval.h"
 #include "storage/sinvaladt.h"
#endif
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/memutils.h"
#ifdef POSTGRES81
 #include "utils/nabstime.h"
#endif
#include "utils/syscache.h"

#ifdef HAVE_TYPCACHE
#include "utils/typcache.h"
#else
#include "parser/parse_oper.h"
#ifdef POSTGRES81
 #include "pgstat.h"
#endif
#endif
#include "mb/pg_wchar.h"

#ifdef POSTGRES81
 #define RelationRelationName "pg_class"
 #define TriggerRelationName "pg_trigger"
 #define TriggerConstrNameIndex "pg_trigger_tgconstrname_index"
#endif

#include <errno.h>
#include <signal.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#ifdef POSTGRES81

typedef struct ProcArrayStruct
{
  int numProcs;
  int maxProcs;
  PGPROC     *procs[1];
} ProcArrayStruct;

static ProcArrayStruct *procArray;

#endif

PG_FUNCTION_INFO_V1(logger);
PG_FUNCTION_INFO_V1(current_transaction_id);
PG_FUNCTION_INFO_V1(get_active_transactions);
PG_FUNCTION_INFO_V1(prepare_recovery_logging);
PG_FUNCTION_INFO_V1(begin_recovery_logging);
PG_FUNCTION_INFO_V1(stop_recovery_logging);
PG_FUNCTION_INFO_V1(cleanup_recovery_logging);
PG_FUNCTION_INFO_V1(create_transaction_quiet_point);

PG_FUNCTION_INFO_V1(disable_fk);
PG_FUNCTION_INFO_V1(enable_fk);

PG_FUNCTION_INFO_V1(file_size);
PG_FUNCTION_INFO_V1(file_read);

static int GetNumberOfRunningTransactions(void);
static bool changeNumberOfTriggers(Oid tgrelid, int diff);

Datum logger(PG_FUNCTION_ARGS);
Datum current_transaction_id(PG_FUNCTION_ARGS);
Datum get_active_transactions(PG_FUNCTION_ARGS);
Datum prepare_recovery_logging(PG_FUNCTION_ARGS);
Datum begin_recovery_logging(PG_FUNCTION_ARGS);
Datum stop_recovery_logging(PG_FUNCTION_ARGS);
Datum cleanup_recovery_logging(PG_FUNCTION_ARGS);
Datum create_transaction_quiet_point(PG_FUNCTION_ARGS);

Datum disable_fk(PG_FUNCTION_ARGS);
Datum enable_fk(PG_FUNCTION_ARGS);

extern Datum file_size(PG_FUNCTION_ARGS);
extern Datum file_read(PG_FUNCTION_ARGS);

#define LOG_INSERT_SQL "INSERT INTO postgresforest.log" \
                       "(tableid,xid,cid,oper,arg,ltime)" \
                       "VALUES($1,$2,$3,$4,$5,$6)"

#define CREATE_LOG_TABLE_SQL "CREATE TABLE postgresforest.log (" \
                             "  ltime timestamp, tableid oid, \"xid\" int4," \
                             "  \"cid\" cid, oper char, arg text) with oids; CREATE INDEX log_ltime_idx on postgresforest.log(ltime);"

#define COUNT_LOG_TABLE_SQL "SELECT COUNT(c.relname)" \
                            " FROM pg_class c, pg_namespace n" \
                            " WHERE c.relname='log'" \
                            " AND c.relnamespace=n.oid" \
                            " AND n.nspname='postgresforest'"

#define DROP_LOG_TABLE_SQL "DROP TABLE postgresforest.log"

struct varbuf {
	char *ptr;
	size_t len;
	size_t pos;
} varbuf;

static void varbuf_alloc(struct varbuf *buf, size_t size);
static void varbuf_expand(struct varbuf *buf, size_t newsize);
static void varbuf_free(struct varbuf *buf);
static void varbuf_strcat(struct varbuf *buf, const char *str);

static void *plan_insert_log;

static void
prepare_plan()
{
	Oid plan_types[6];
	int j=0;

	plan_types[j++] = OIDOID;   /* tableid */
	plan_types[j++] = INT4OID;  /* xid */
	plan_types[j++] = CIDOID;   /* cid */
	plan_types[j++] = CHAROID;  /* oper */
	plan_types[j++] = TEXTOID;  /* arg */
	plan_types[j++] = TIMESTAMPTZOID;  /* ltime */

	plan_insert_log = SPI_saveplan(SPI_prepare(LOG_INSERT_SQL, j, plan_types));

	if( plan_insert_log==NULL )
		elog(ERROR, "SPI_prepare() failed.");
}


static void
varbuf_alloc(struct varbuf *buf, size_t size)
{
	buf->len = size;
	buf->ptr = palloc(buf->len);
	buf->pos = 0;
	memset(buf->ptr, 0, buf->len);
}


static void
varbuf_expand(struct varbuf *buf, size_t newsize)
{
	void *newptr = palloc(newsize);

	memset(newptr, 0, newsize);
	memcpy(newptr, buf->ptr, buf->len);

	buf->len = newsize;

	pfree(buf->ptr);
	buf->ptr = newptr;
}

static void
varbuf_free(struct varbuf *buf)
{
	buf->len = buf->pos = 0;
	pfree(buf->ptr);
}


static void
varbuf_strcat(struct varbuf *buf, const char *str)
{
	int i;

	for (i=0 ; *(str+i)!='\0' ; i++)
	{
		if ( buf->len==buf->pos )
			varbuf_expand(buf, buf->len*2);

		*(buf->ptr+(buf->pos++)) = *(str+i);
	}
}

#define NEED_QUOTE(typ) (1)

#define IS_BYTEA(typ) ( (typ)->atttypid==BYTEAOID )

/* --------------------------------------------------------------
 * escape_bytea
 *
 * bytea型のデータをシングルクオートでくくり、
 * 8進表記のデータ（表示不可文字）の '\' を '\' でエスケープする。
 *
 * 例：
 *   入力：a\001b\000c\003d
 *   出力：'a\\001b\\000c\\003d'
 * --------------------------------------------------------------
 */
static void
escape_bytea(struct varbuf *buf, const char *str)
{
	int i = 0;
	char ch[2];
	  
	ch[1] = '\0';
	varbuf_strcat(buf, "'");

	for ( i=0 ; i<strlen(str) ; i++ )
	{
		/*
		 * single quote must be escaped by another single quote
		 */
		if ( str[i]=='\'' )
		{
		  varbuf_strcat(buf, "''");
		}
		/*
		 * '\\' expresses a single backslash,
		 * which should be represented by '\\\\' or '\\134'.
		 */
		else if ( i<strlen(str)-1 && str[i]=='\\' && str[i+1]=='\\' )
		{
		  varbuf_strcat(buf, "\\\\134");
		  i += 1;
		}
		/*
		 * non-printable charactor '\xxx'
		 */
		else if ( i<strlen(str)-3 && str[i]=='\\'
				  && isdigit(str[i+1]) && isdigit(str[i+2]) && isdigit(str[i+3]) )
		{
		  char cc[5];

		  cc[0] = '\\';	/* single backslash must be escaped by another backslash */
		  cc[1] = '\\';
		  cc[2] = str[i+1]; /* digit */
		  cc[3] = str[i+2]; /* digit */
		  cc[4] = str[i+3]; /* digit */
		  cc[5] = '\0';

		  varbuf_strcat(buf, cc);
		  i += 3;
		}
		/*
		 * printable charactor
		 */
		else
		{
		  ch[0] = str[i];
		  varbuf_strcat(buf, ch);
		}
	}
	varbuf_strcat(buf, "'");
}

/* --------------------------------------------------------------
 * build_quoted_string
 *
 * 更新ログに記録するために、更新データのPostgreSQL内部表現を
 * SQL入力形式に変換する（但し、bytea型は別処理）。
 *
 * 変換ルール
 *   1.) 文字列全体をシングルクオートでくくる。
 *   2.) 文字列内のシングルクオートは、シングルクオート2つに変換。
 *   3.) 文字列内のバックスラッシュは、バックスラッシュ2つに変換。
 *
 * 例：
 *   入力：ab'CD'\nef
 *   出力：'ab''CD''\\nef'
 * --------------------------------------------------------------
 */
static void
build_value_string(struct varbuf *buf, const char *str, bool needQuote, bool isBytea)
{
	if ( isBytea )
	{
		escape_bytea(buf, str);
		return;
	}

	if ( needQuote )
	{
		int i = 0;
		char ch[2];
	  
		ch[1] = '\0';
		varbuf_strcat(buf, "'");
		for ( i=0 ; i<strlen(str) ; i++ )
		{
			if ( str[i]=='\'' )
				varbuf_strcat(buf, "'");
		  
			if ( str[i]=='\\' )
				varbuf_strcat(buf, "\\"); /* normal text */
		  
			ch[0] = str[i];
			varbuf_strcat(buf, ch);
		}
		varbuf_strcat(buf, "'");
 	}
	else
	{
		varbuf_strcat(buf, str);
 	}
}
 

Datum
logger(PG_FUNCTION_ARGS)
{
	TriggerData	   *tg;
	Datum			argv[6];
	int rc;

	struct varbuf buf, buf2;

	tg = (TriggerData *)(fcinfo->context);

	if ( (rc = SPI_connect())<0 )
		elog(ERROR, "SPI_connect() failed.");

#ifdef NOT_USED
	if ( plan_insert_log==NULL )
		prepare_plan();
#endif
	prepare_plan();

	varbuf_alloc(&buf, 256);
	varbuf_alloc(&buf2, 256);

	argv[0] = ObjectIdGetDatum(tg->tg_relation->rd_id);
	argv[1] = Int32GetDatum(GetCurrentTransactionId());
	argv[2] = CommandIdGetDatum(GetCurrentCommandId());

	if (TRIGGER_FIRED_BY_INSERT(tg->tg_event))
	{
		HeapTuple	new_row = tg->tg_trigtuple;
		TupleDesc	tupdesc = tg->tg_relation->rd_att;

		int i;

//		elog(NOTICE, "natts=%d", tg->tg_relation->rd_att->natts);
		varbuf_strcat(&buf, "(");
//		elog(NOTICE, " %s", buf.ptr);

		varbuf_strcat(&buf2, "(");
		for (i=0 ; i<tg->tg_relation->rd_att->natts ; i++)
		{
			if ( i!=0 )
			{
				varbuf_strcat(&buf, ",");
				varbuf_strcat(&buf2, ",");
			}
			varbuf_strcat(&buf, SPI_fname(tupdesc, i + 1));

			if ( SPI_getvalue(new_row, tupdesc, i + 1)==NULL )
			{
				varbuf_strcat(&buf2, "null");
			}
			else
			{
				build_value_string(&buf2,
						   SPI_getvalue(new_row, tupdesc, i + 1),
						   NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]),
						   IS_BYTEA(tg->tg_relation->rd_att->attrs[i]));
#ifdef NOT_USED
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf2, "'");
				varbuf_strcat(&buf2, SPI_getvalue(new_row, tupdesc, i + 1));
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf2, "'");
#endif
			}
		}
		varbuf_strcat(&buf, ")");
		varbuf_strcat(&buf2, ")");

		varbuf_strcat(&buf, " values ");
		varbuf_strcat(&buf, buf2.ptr);

		argv[3] = CharGetDatum('I');
		argv[4] = DirectFunctionCall1(textin, CStringGetDatum(buf.ptr));
	}
	else if (TRIGGER_FIRED_BY_UPDATE(tg->tg_event))
	{
		TupleDesc	tupdesc = tg->tg_relation->rd_att;
		HeapTuple	old_row = tg->tg_trigtuple;
		HeapTuple	new_row = tg->tg_newtuple;
		Datum		old_value;
		Datum		new_value;
		bool		old_isnull;
		bool		new_isnull;
		bool		first_elem;
		int i;

		first_elem = true;
		for (i = 0; i < tg->tg_relation->rd_att->natts; i++)
		{
			if (tupdesc->attrs[i]->attisdropped)
				continue;

			old_value = SPI_getbinval(old_row, tupdesc, i + 1, &old_isnull);
			new_value = SPI_getbinval(new_row, tupdesc, i + 1, &new_isnull);

			/*
			 * If old and new value are NULL, the column is unchanged
			 */
			if (old_isnull && new_isnull)
				continue;

			if (!old_isnull && !new_isnull)
			{
				char   *old_strval = SPI_getvalue(old_row, tupdesc, i + 1);
				char   *new_strval = SPI_getvalue(new_row, tupdesc, i + 1);

				if (strcmp(old_strval, new_strval) == 0)
					continue; /* not changed. */
			}

			/* changed column */
			if ( first_elem )
				varbuf_strcat(&buf, "SET ");
			else
				varbuf_strcat(&buf, ",");

			varbuf_strcat(&buf, SPI_fname(tupdesc, i + 1));
			varbuf_strcat(&buf, "=");
			if ( new_isnull )
			{
				varbuf_strcat(&buf, "null");
			}
			else
			{
				build_value_string(&buf,
						   SPI_getvalue(new_row, tupdesc, i + 1),
						   NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]),
						   IS_BYTEA(tg->tg_relation->rd_att->attrs[i]));
#ifdef NOT_USED
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf, "'");
				varbuf_strcat(&buf, SPI_getvalue(new_row, tupdesc, i + 1));
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf, "'");
#endif
			}
			first_elem = false;
		}

		if ( buf.ptr[0]=='\0' )
			goto finish;

		varbuf_strcat(&buf, " WHERE ");

		first_elem = true;

		for (i = 0; i < tg->tg_relation->rd_att->natts; i++)
		{
			if (tupdesc->attrs[i]->attisdropped)
				continue;

			if ( !first_elem )
				varbuf_strcat(&buf, " AND ");

			varbuf_strcat(&buf, SPI_fname(tupdesc, i + 1));
			if ( SPI_getvalue(old_row, tupdesc, i + 1)==NULL )
			{
				varbuf_strcat(&buf, " IS NULL");
			}
			else
			{
				varbuf_strcat(&buf, "=");
				build_value_string(&buf,
						   SPI_getvalue(old_row, tupdesc, i + 1),
						   NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]),
						   IS_BYTEA(tg->tg_relation->rd_att->attrs[i]));
#ifdef NOT_USED
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf, "'");
				varbuf_strcat(&buf, SPI_getvalue(old_row, tupdesc, i + 1));
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf, "'");
#endif
			}
			
			first_elem = false;
		}

		argv[3] = CharGetDatum('U');
		argv[4] = DirectFunctionCall1(textin, CStringGetDatum(buf.ptr));
	}
	else if (TRIGGER_FIRED_BY_DELETE(tg->tg_event))
	{
		HeapTuple	old_row = tg->tg_trigtuple;
		TupleDesc	tupdesc = tg->tg_relation->rd_att;
		bool first_elem;
		int i;

		first_elem = true;
		for (i = 0; i<tg->tg_relation->rd_att->natts; i++)
		{
			if (tupdesc->attrs[i]->attisdropped)
				continue;

			if (!first_elem)
				varbuf_strcat(&buf, " AND ");

			if ( SPI_getvalue(old_row, tupdesc, i + 1)==NULL )
			{
				varbuf_strcat(&buf, SPI_fname(tupdesc, i+1));
				varbuf_strcat(&buf, " IS NULL");
			}
			else
			{
				varbuf_strcat(&buf, SPI_fname(tupdesc, i+1));
				varbuf_strcat(&buf, "=");
				build_value_string(&buf,
						   SPI_getvalue(old_row, tupdesc, i + 1),
						   NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]),
						   IS_BYTEA(tg->tg_relation->rd_att->attrs[i]));
#ifdef NOT_USED
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf, "'");
				varbuf_strcat(&buf, SPI_getvalue(old_row, tupdesc, i + 1));
				if ( NEED_QUOTE(tg->tg_relation->rd_att->attrs[i]) )
					varbuf_strcat(&buf, "'");
#endif
			}

			first_elem = false;
		}

//		elog(NOTICE, " D=%s", buf.ptr);

		argv[3] = CharGetDatum('D');
		argv[4] = DirectFunctionCall1(textin, CStringGetDatum(buf.ptr));
	}

	{
#ifdef POSTGRES81
		argv[5] = TimestampTzGetDatum(GetCurrentTimestamp());
#else
		int usec;
		AbsoluteTime sec;
		sec = GetCurrentAbsoluteTimeUsec(&usec);

		argv[5] = TimestampTzGetDatum(AbsoluteTimeUsecToTimestampTz(sec, usec));
#endif
	}

	SPI_execp(plan_insert_log, argv, NULL, 0);

//	elog(NOTICE, "logger: invoked.");

finish:
	SPI_finish();

	varbuf_free(&buf);
	varbuf_free(&buf2);

	return PointerGetDatum(NULL);
}

Datum
current_transaction_id(PG_FUNCTION_ARGS)
{
	PG_RETURN_DATUM(TransactionIdGetDatum(GetCurrentTransactionId()));
}

Datum
get_active_transactions(PG_FUNCTION_ARGS)
{
	PG_RETURN_DATUM(Int32GetDatum(GetNumberOfRunningTransactions()));
}


#ifdef POSTGRES81
static void
read_proc_array()
{
  bool found;

  procArray = (ProcArrayStruct *)ShmemInitStruct("Proc Array", ProcArrayShmemSize(), &found);

  if ( !found )
    elog(ERROR, "ProcArray is not found in the shared memory.");
}
#endif

static PGPROC *
get_proc(int i)
{
#ifdef POSTGRES81
  read_proc_array();

  return procArray->procs[i];
#else
  SISeg   *segP = shmInvalBuffer;
  ProcState  *stateP = segP->procState;

  SHMEM_OFFSET pOffset = stateP[i].procStruct;
  return (PGPROC *) MAKE_PTR(pOffset);
#endif
}

static int
get_number_of_backends()
{
  int nbackends;

#ifdef POSTGRES81
  int found;

  read_proc_array();

  elog(NOTICE, "ProcArray=%p, numProcs=%d", procArray,
       procArray->numProcs);

  nbackends = procArray->numProcs;
#else
  SISeg   *segP = shmInvalBuffer;
  ProcState  *stateP = segP->procState;

  nbackends = segP->lastBackend;
#endif

  return nbackends;
}

static int
GetNumberOfRunningTransactions()
{
	int nBackends = 0;
	int nActiveTransactions = 0;
	int i;

	LWLockAcquire(SInvalLock, LW_SHARED);

	nBackends = get_number_of_backends();

	for (i=0 ; i<nBackends; i++)
	{
	  PGPROC *proc = get_proc(i);
	  elog(NOTICE, "xid = %d", proc->xid);
	  
	  if ( TransactionIdIsValid(proc->xid) )
	    nActiveTransactions++;
	}

	LWLockRelease(SInvalLock);

	return nActiveTransactions;
}

/* ------------------------------------------------------------
 * Before entering this function, SPI connection must be opened.
 * ------------------------------------------------------------
 */
static bool
checkLogTableExists()
{
	if ( SPI_exec(COUNT_LOG_TABLE_SQL, 1) == SPI_OK_SELECT )
	{
		char *tmp
			= SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);

		if ( strcmp(tmp, "0")==0 )
			return false;
	}

	return true;
}

Datum
prepare_recovery_logging(PG_FUNCTION_ARGS)
{
	text  *relname = PG_GETARG_TEXT_P(0);

	bool found = false;

	SPI_connect();

	found = checkLogTableExists();

	if ( found )
	{
		elog(NOTICE, "postgresforest.log table already exists.");

		/* --- FIXME: need truncate or delete records? --- */
	}
	else
	{
		if ( SPI_exec(CREATE_LOG_TABLE_SQL, 1) != SPI_OK_UTILITY )
			elog(ERROR, "SPI_exec() failed.");
	}

	SPI_finish();

	PG_RETURN_BOOL(true);
}

static char *
text2cstring(text *t)
{
	char  *cstr = NULL;
	int len = VARSIZE(t) - VARHDRSZ;

	cstr = palloc(len + 1);
	memset(cstr, 0, len + 1);
	memcpy(cstr, VARDATA(t), len);

	return cstr;
}

static void
DumpAllBackends()
{
	int nBackends = 0;
	int i;

	LWLockAcquire(SInvalLock, LW_SHARED);

	nBackends = get_number_of_backends();

	for (i=0 ; i<nBackends ; i++)
	{
	  PGPROC   *proc = get_proc(i);

	  elog(NOTICE, "mypid=%d, myxid=%d, pid=%d, xid=%d, isvalid=%d",
	       MyProc->pid, MyProc->xid,
	       proc->pid, proc->xid,
	       TransactionIdIsValid(proc->xid));

//		if ( MyProc->pid != proc->pid && TransactionIdIsValid(proc->pid) )
//			kill(proc->pid, SIGTERM);
	}

	LWLockRelease(SInvalLock);
}


static int
GetRunningTransactionIds(uint32 *ids, int len)
{
  int nBackends = 0;
  int i,j;

  LWLockAcquire(SInvalLock, LW_SHARED);
  
  nBackends = get_number_of_backends();
  
  for (i=0, j=0 ; i<nBackends && j<len ; i++)
  {
    PGPROC *proc = get_proc(i);
      
    ids[j++] = proc->xid;
  }
  
  LWLockRelease(SInvalLock);

  nBackends = j;
  
  return nBackends;
}

static bool
checkTriggerExists(const char *tableName)
{
	char *qbuf;
	char *tmp;

	qbuf = palloc(1024);

	snprintf(qbuf, 1024, "SELECT COUNT(*) FROM pg_trigger t,pg_class c"
			             " WHERE t.tgrelid=c.oid "
				         "   AND t.tgname='%s_logger' "
				         "   AND c.relname='%s'", tableName, tableName);

	if ( SPI_exec(qbuf, 1) != SPI_OK_SELECT )
		elog(ERROR, "SPI_exec() failed.");

	tmp = SPI_getvalue(SPI_tuptable->vals[0], SPI_tuptable->tupdesc, 1);

	pfree(qbuf);

	if ( strcmp(tmp, "0")==0 )
		return false;

	return true;
}

Datum
begin_recovery_logging(PG_FUNCTION_ARGS)
{
	text  *relname = PG_GETARG_TEXT_P(0);
//	int retry_max  = PG_GETARG_INT32(1);
	char  *relname_c = text2cstring(relname);

	TransactionId borderXid = InvalidTransactionId;

	if ( SPI_connect() != SPI_OK_CONNECT )
		elog(ERROR, "SPI_connect() failed.");

	/*
	 * Set a log trigger
	 */
	if ( !checkTriggerExists(relname_c) )
	{
		char *qbuf;

		qbuf = palloc(1024);

		snprintf(qbuf, 1024,
				 "CREATE TRIGGER %s_logger AFTER INSERT OR UPDATE OR DELETE"
				 " ON %s FOR EACH ROW"
				 " EXECUTE PROCEDURE postgresforest.logger()",
				 relname_c, relname_c);

		if ( SPI_exec(qbuf, 1) != SPI_OK_UTILITY )
			elog(ERROR, "SPI_exec() failed.");

		pfree(qbuf);
		pfree(relname_c);
	}
	else
		elog(NOTICE, "log trigger on %s already exists.", relname_c);

	if ( SPI_finish() != SPI_OK_FINISH )
		elog(ERROR, "SPI_finish() faield.");

	CommitTransactionCommand();
	StartTransactionCommand();

	/*
	 * To garantee all transactions are running after the log trigger set,
	 * wait for current running transactions finishing.
	 */
	{
		int nTx;
		int i;
		uint32 ids[1024];

//		LWLockAcquire(XidGenLock, LW_SHARED);
		nTx = GetRunningTransactionIds(ids, 1024);
//		LWLockRelease(XidGenLock);

		for (i=0 ; i<nTx ; i++)
		{
			elog(NOTICE, "ids[%d]: xid=%d", i, ids[i]);
			XactLockTableWait(ids[i]);
		}

		borderXid = ReadNewTransactionId();
	}

	PG_RETURN_DATUM( TransactionIdGetDatum(borderXid) );
}

Datum
create_transaction_quiet_point(PG_FUNCTION_ARGS)
{
  int nTx;
  int i;
  uint32 ids[1024];
  TransactionId borderXid;

//		LWLockAcquire(XidGenLock, LW_SHARED);
  nTx = GetRunningTransactionIds(ids, 1024);
//		LWLockRelease(XidGenLock);

  elog(NOTICE, "nTx=%d", nTx);

  for (i=0 ; i<nTx ; i++)
  {
    elog(NOTICE, "ids[%d]: xid=%d", i, ids[i]);
    XactLockTableWait(ids[i]);
  }

  borderXid = ReadNewTransactionId();

  PG_RETURN_DATUM( TransactionIdGetDatum(borderXid) );
}

Datum
stop_recovery_logging(PG_FUNCTION_ARGS)
{
	text *relname = PG_GETARG_TEXT_P(0);
	char  *relname_c = text2cstring(relname);
	char *qbuf;

	qbuf = palloc(1024);

	if ( SPI_connect() != SPI_OK_CONNECT )
		elog(ERROR, "SPI_connect() failed.");

	if ( checkTriggerExists(relname_c) )
	{
		snprintf(qbuf, 1024, "DROP TRIGGER %s_logger ON %s", relname_c, relname_c);

		if ( SPI_exec(qbuf, 1) != SPI_OK_UTILITY )
			elog(ERROR, "SPI_exec() failed.");
	}

	if ( SPI_finish() != SPI_OK_FINISH )
		elog(ERROR, "SPI_finish() faield.");

	pfree(relname_c);
	pfree(qbuf);

	PG_RETURN_BOOL(true);
}

Datum
cleanup_recovery_logging(PG_FUNCTION_ARGS)
{
	bool found = false;
	bool rc;

	SPI_connect();

	found = checkLogTableExists();

	if ( found )
	{
		if ( SPI_exec(DROP_LOG_TABLE_SQL, 1) != SPI_OK_UTILITY )
			elog(ERROR, "SPI_exec() failed.");
		rc = true;
	}
	else
		rc = false;

	SPI_finish();

	PG_RETURN_BOOL(rc);
}

static bool
changeTriggerStatus(char *tgconstrname, bool flag)
{
	SysScanDesc tgscan;
	Relation tgrel;
	ScanKeyData skey[2];
	HeapTuple tup;
	bool changed = false;
	Oid changedRelid[128];
	int p = 0;

	/*
	 * Enable/Disable a trigger
	 */
#ifdef POSTGRES81
	tgrel = heap_open(TriggerRelationId, AccessExclusiveLock);

	ScanKeyInit(&skey[0],
		    Anum_pg_trigger_tgconstrname,
		    BTEqualStrategyNumber,
		    F_NAMEEQ,
		    CStringGetDatum(tgconstrname));

	tgscan = systable_beginscan(tgrel, TriggerConstrNameIndexId, true,
				    SnapshotNow, 1, skey);

	while ( HeapTupleIsValid(tup = systable_getnext(tgscan)) )
	  {
	    Form_pg_trigger oldtrig = (Form_pg_trigger) GETSTRUCT(tup);

	    if (oldtrig->tgenabled != flag)
	    {
	      /* need to change this one ... make a copy to scribble on */
	      HeapTuple   newtup = heap_copytuple(tup);
	      Form_pg_trigger newtrig = (Form_pg_trigger) GETSTRUCT(newtup);

	      newtrig->tgenabled = flag;

	      simple_heap_update(tgrel, &newtup->t_self, newtup);

	      /* Keep catalog indexes current */
	      CatalogUpdateIndexes(tgrel, newtup);

	      heap_freetuple(newtup);
	      
	      changed = true;
	      changedRelid[p] = oldtrig->tgrelid;
	      p++;

	      //	      CacheInvalidateRelcacheByRelid(oldtrig->tgrelid);
	      //	      CacheInvalidateRelcacheByRelid(oldtrig->tgconstrrelid);
	    }
#ifdef NOT_USED
		int trig = 0;
		Oid tgrelid = InvalidOid;
		HeapTuple newtup = heap_copytuple(tup);
			
		Form_pg_trigger tgform = (Form_pg_trigger)GETSTRUCT(newtup);

		Relation targetrel = RelationIdGetRelation(tgform->tgrelid);

		elog(NOTICE, "targetrel->rd_rel->reltriggers = %d", targetrel->rd_rel->reltriggers);

		//		EnableDisableTrigger(targetrel, tgconstrname, flag, false);

		/*
		if ( tgform->tgenabled==flag ) /* already in this state */
		{
			changed = false;
		}
		else if ( tgform->tgenabled==true )
		{
			tgform->tgenabled = false;
			changed = true;

			changeNumberOfTriggers(tgform->tgrelid, -1);
		}
		else if ( tgform->tgenabled==false )
		{
			tgform->tgenabled = true;
			changed = true;

			changeNumberOfTriggers(tgform->tgrelid, +1);
		}
*/
		simple_heap_update(tgrel, &tup->t_self, newtup);

		CatalogUpdateIndexes(tgrel, newtup);

		//		CacheInvalidateRelcacheByRelid(tgform->tgrelid);
		//		CacheInvalidateRelcacheByRelid(tgform->tgconstrrelid);
		heap_freetuple(newtup);
#endif
	  }
	
	systable_endscan(tgscan);
	heap_close(tgrel, AccessExclusiveLock);

	if (changed)
	{
	  int i;
	  for (i=0 ; i<p ; i++)
	    CacheInvalidateRelcache(RelationIdGetRelation(changedRelid[i]));
	}
#else
	tgrel = heap_openr(TriggerRelationName, AccessExclusiveLock);

	ScanKeyEntryInitialize(&skey[0], 0x0,
			       Anum_pg_trigger_tgconstrname,
			       F_NAMEEQ,
			       CStringGetDatum(tgconstrname));

	tgscan = systable_beginscan(tgrel, TriggerConstrNameIndex, true,
				    SnapshotNow, 1, skey);

	while ( HeapTupleIsValid(tup = systable_getnext(tgscan)) )
	  {
	    int trig = 0;
	    Oid tgrelid = InvalidOid;
		HeapTuple newtup = heap_copytuple(tup);
			
		Form_pg_trigger tgform = (Form_pg_trigger)GETSTRUCT(newtup);
			
		if ( tgform->tgenabled==flag ) /* already in this state */
		{
			changed = false;
		}
		else if ( tgform->tgenabled==true )
		{
			tgform->tgenabled = false;
			changed = true;

			changeNumberOfTriggers(tgform->tgrelid, -1);
		}
		else if ( tgform->tgenabled==false )
		{
			tgform->tgenabled = true;
			changed = true;

			changeNumberOfTriggers(tgform->tgrelid, +1);
		}

		simple_heap_update(tgrel, &tup->t_self, newtup);
		CatalogUpdateIndexes(tgrel, newtup);

		ReleaseSysCache(tup);
		heap_freetuple(newtup);
		CommandCounterIncrement();
	}

	systable_endscan(tgscan);
	heap_close(tgrel, AccessExclusiveLock);
#endif

	return changed;
}

/**
 *
 *
 */
static bool
changeNumberOfTriggers(Oid tgrelid, int diff)
{
	Relation crel;
	HeapTuple tup;
	bool result = false;
	
	/* -- decrease a number of triggers -- */
#ifdef POSTGRES81
	crel = heap_open(RelationRelationId, AccessExclusiveLock);
#else
	crel = heap_openr(RelationRelationName, AccessExclusiveLock);
#endif
	
	tup = SearchSysCacheCopy(RELOID, ObjectIdGetDatum(tgrelid), 0, 0, 0);

	if ( HeapTupleIsValid(tup) )
	{
		Form_pg_class cform = (Form_pg_class)GETSTRUCT(tup);
		
		cform->reltriggers += diff;
		
		simple_heap_update(crel, &tup->t_self, tup);
		CatalogUpdateIndexes(crel, tup);

		result = true;

		heap_freetuple(tup);
	}
	
	heap_close(crel, AccessExclusiveLock);

	return result;
}


Datum
disable_fk(PG_FUNCTION_ARGS)
{
	bool changed;
	char       *tgcname = NULL;

	{
		Name	 tgname = PG_GETARG_NAME(0);
		tgcname = NameStr(*tgname);
	}

	changed = changeTriggerStatus(tgcname, false);

	pfree(tgcname);

	PG_RETURN_BOOL(changed);
}

Datum
enable_fk(PG_FUNCTION_ARGS)
{
	bool changed;
	char       *tgcname = NULL;

	{
		Name	 tgname = PG_GETARG_NAME(0);
		tgcname = NameStr(*tgname);
	}

	changed = changeTriggerStatus(tgcname, true);

	pfree(tgcname);

	PG_RETURN_BOOL(changed);
}


/*
 *
 */
Datum
file_size(PG_FUNCTION_ARGS)
{
  text	   *filename = PG_GETARG_TEXT_P(0);
  struct stat st;
  char *fname = text2cstring(filename);

  if ( stat(fname, &st)!=0 )
	elog(ERROR, "file_size(): stat() failed. %s", strerror(errno));

  if ( !S_ISREG(st.st_mode) )
	elog(ERROR, "file_size(): it's not a regular file.");

  PG_RETURN_INT64(st.st_size);
}

Datum
file_read(PG_FUNCTION_ARGS)
{
  char   *fname = text2cstring(PG_GETARG_TEXT_P(0));
  size_t offset = PG_GETARG_INT64(1);
  size_t len    = PG_GETARG_INT64(2);
  FILE *fp;
  bytea  *result;

  result = (bytea *) palloc(len + VARHDRSZ);
  VARATT_SIZEP(result) = len + VARHDRSZ;

  fp = fopen(fname, "r");
  if ( fp==NULL )
	elog(ERROR, "file_read(): fopen() failed. %s", strerror(errno));

  if ( fseek(fp, offset, SEEK_SET)!=0 )
	elog(ERROR, "file_read(): fseek() failed. %s", strerror(errno));

  if ( fread(VARDATA(result), len, 1, fp)<1 && ferror(fp) )
	elog(ERROR, "file_read(): fread() failed. %s", strerror(errno));

  if ( fclose(fp)!=0 )
	elog(ERROR, "file_read(): fclose() failed. %s", strerror(errno));

  PG_RETURN_BYTEA_P(result);
}
