import.c

/*
 * import.c
 *
 * Copyright (c) 2012-2016, NIPPON TELEGRAPH AND TELEPHONE CORPORATION
 */
#include "postgres.h"

#include "access/xact.h"
#include "catalog/namespace.h"
#include "catalog/pg_type.h"
#include "commands/copy.h"
#include "executor/spi.h"
#include "libpq/pqformat.h"
#include "mb/pg_wchar.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"
#if PG_VERSION_NUM >= 90200
#include "catalog/pg_class.h"
#endif
#if PG_VERSION_NUM >= 90300
#include "access/htup_details.h"
#endif

#include "pg_dbms_stats.h"

#if PG_VERSION_NUM >= 90200
#define RELATION_PARAM_NUM	9
#else
#define RELATION_PARAM_NUM	8
#endif

extern PGDLLIMPORT bool standard_conforming_strings;

PG_FUNCTION_INFO_V1(dbms_stats_import);

Datum	dbms_stats_import(PG_FUNCTION_ARGS);

static void get_args(FunctionCallInfo fcinfo, char **nspname, char **relname,
				char **attname, char **filename);
static void spi_exec_utility(const char *query);
static void spi_exec_query(const char *query, int nargs, Oid *argtypes,
				SPIPlanPtr *plan, Datum *values, const char *nulls, int result);
static void import_stats_from_file(char *filename, char *nspname, char *relname,
				char *attname);

/*
 * dbms_stats_import
 *   Import exported statistics from stdin or a file.
 *
 *   Order of arguments:
 *     1) schema name
 *     2) relation oid
 *     3) attribute name
 *     4) absolute path of source file, or 'stdin' (case insensitive)
 */
Datum
dbms_stats_import(PG_FUNCTION_ARGS)
{
	char		   *nspname;
	char		   *relname;
	char		   *attname;
	char		   *filename;	/* filename, or NULL for STDIN */
	int				ret;
	int				i;
	uint32			r_num;
	HeapTuple	   *r_tups;
	TupleDesc		r_tupdesc;
	SPIPlanPtr		r_upd_plan = NULL;
	SPIPlanPtr		r_ins_plan = NULL;
	SPIPlanPtr		c_sel_plan = NULL;
	SPIPlanPtr		c_del_plan = NULL;
	SPIPlanPtr		c_ins_plan = NULL;

	/* get validated arguments */
	get_args(fcinfo, &nspname, &relname, &attname, &filename);

	/* for debug use */
	elog(DEBUG3, "%s() f=%s n=%s r=%s a=%s", __FUNCTION__,
		 filename ? filename : "(null)",
		 nspname ? nspname : "(null)",
		 relname ? relname : "(null)",
		 attname ? attname : "(null)");

	/* connect to SPI */
	ret = SPI_connect();
	if (ret != SPI_OK_CONNECT)
		elog(ERROR, "pg_dbms_stats: SPI_connect => %d", ret);

	/* lock dummy statistics tables. */
	spi_exec_utility("LOCK dbms_stats.relation_stats_locked"
						" IN SHARE UPDATE EXCLUSIVE MODE");
	spi_exec_utility("LOCK dbms_stats.column_stats_locked"
						" IN SHARE UPDATE EXCLUSIVE MODE");

	/*
	 * Create a temp table to save the statistics to import.
	 * This table should fit with the content of export files.
	 */
	spi_exec_utility("CREATE TEMP TABLE dbms_stats_work_stats ("
					 "nspname          name   NOT NULL,"
					 "relname          name   NOT NULL,"
					 "relpages         int4   NOT NULL,"
					 "reltuples        float4 NOT NULL,"
#if PG_VERSION_NUM >= 90200
					 "relallvisible    int4   NOT NULL,"
#endif
					 "curpages         int4   NOT NULL,"
					 "last_analyze     timestamp with time zone,"
					 "last_autoanalyze timestamp with time zone,"
					 "attname          name,"
					 "nspname_of_typename name,"
					 "typname name,"
					 "atttypmod int4,"
					 "stainherit       bool,"
					 "stanullfrac      float4,"
					 "stawidth         int4,"
					 "stadistinct      float4,"
					 "stakind1         int2,"
					 "stakind2         int2,"
					 "stakind3         int2,"
					 "stakind4         int2,"
#if PG_VERSION_NUM >= 90200
					 "stakind5         int2,"
#endif
					 "staop1           oid,"
					 "staop2           oid,"
					 "staop3           oid,"
					 "staop4           oid,"
#if PG_VERSION_NUM >= 90200
					 "staop5           oid,"
#endif
					 "stanumbers1      float4[],"
					 "stanumbers2      float4[],"
					 "stanumbers3      float4[],"
					 "stanumbers4      float4[],"
#if PG_VERSION_NUM >= 90200
					 "stanumbers5      float4[],"
#endif
					 "stavalues1       dbms_stats.anyarray,"
					 "stavalues2       dbms_stats.anyarray,"
					 "stavalues3       dbms_stats.anyarray,"
					 "stavalues4       dbms_stats.anyarray"
#if PG_VERSION_NUM >= 90200
					",stavalues5       dbms_stats.anyarray"
#endif
					 ")");

	/* load the statistics from export file to the temp table */
	import_stats_from_file(filename, nspname, relname, attname);

	/* Determine the Oid of local table from the tablename and schemaname. */
	ret = SPI_execute("SELECT DISTINCT w.nspname, w.relname, c.oid, "
							 "w.relpages, w.reltuples, "
							 "w.curpages, w.last_analyze, w.last_autoanalyze "
#if PG_VERSION_NUM >= 90200
							 ",w.relallvisible "
#endif
						"FROM pg_catalog.pg_class c "
						"JOIN pg_catalog.pg_namespace n "
						  "ON (c.relnamespace = n.oid) "
					   "RIGHT JOIN dbms_stats_work_stats w "
						  "ON (w.relname = c.relname AND w.nspname = n.nspname) "
					   "ORDER BY 1, 2", false, 0);
	if (ret != SPI_OK_SELECT)
		elog(ERROR, "pg_dbms_stats: SPI_execute => %d", ret);

	/*
	 * If there is no record in the staging table after loading source and
	 * deleting unnecessary records, we treat it as an error.
	 */
	if (SPI_processed == 0)
		elog(ERROR, "no per-table statistic data to be imported");

	/* */
	r_num = SPI_processed;
	r_tups = SPI_tuptable->vals;
	r_tupdesc = SPI_tuptable->tupdesc;
	for (i = 0; i < r_num; i++)
	{
		bool	isnull;
		Datum	w_nspname;
		Datum	w_relname;
		Datum	w_relid;
		Datum	values[9];
		char	nulls[9] = {'t', 't', 't', 't', 't', 't', 't', 't', 't'};
		Oid		r_types[9] = {NAMEOID, NAMEOID, INT4OID, FLOAT4OID, INT4OID,
							  TIMESTAMPTZOID, TIMESTAMPTZOID, OIDOID, INT4OID};
		Oid		c_types[5] = {OIDOID, INT2OID, NAMEOID, NAMEOID,
							  NAMEOID};
		uint32		c_num;
		TupleDesc	c_tupdesc;
		HeapTuple  *c_tups;
		int			j;

		values[0] = w_nspname = SPI_getbinval(r_tups[i], r_tupdesc, 1, &isnull);
		values[1] = w_relname = SPI_getbinval(r_tups[i], r_tupdesc, 2, &isnull);
		values[7] = w_relid = SPI_getbinval(r_tups[i], r_tupdesc, 3, &isnull);
		if (isnull)
		{
			elog(WARNING, "relation \"%s.%s\" does not exist",
					DatumGetName(w_nspname)->data,
					DatumGetName(w_relname)->data);
			continue;
		}

		values[2] = SPI_getbinval(r_tups[i], r_tupdesc, 4, &isnull);
		values[3] = SPI_getbinval(r_tups[i], r_tupdesc, 5, &isnull);
		values[4] = SPI_getbinval(r_tups[i], r_tupdesc, 6, &isnull);
		values[5] = SPI_getbinval(r_tups[i], r_tupdesc, 7, &isnull);
		nulls[5] = isnull ? 'n' : 't';
		values[6] = SPI_getbinval(r_tups[i], r_tupdesc, 8, &isnull);
		nulls[6] = isnull ? 'n' : 't';
		values[8] = SPI_getbinval(r_tups[i], r_tupdesc, 9, &isnull);

		/*
		 * First we try UPDATE with the oid.  When no record matched, try
		 * INSERT.  We can't use DELETE-then-INSERT method because we have FK
		 * on relation_stats_locked so DELETE would delete child records in
		 * column_stats_locked undesirably.
		 */
		spi_exec_query("UPDATE dbms_stats.relation_stats_locked SET "
				"relname = quote_ident($1) || '.' || quote_ident($2), "
				"relpages = $3, reltuples = $4, "
#if PG_VERSION_NUM >= 90200
				"relallvisible = $9, "
#endif
				"curpages = $5, last_analyze = $6, last_autoanalyze = $7 "
				"WHERE relid = $8",
				RELATION_PARAM_NUM, r_types, &r_upd_plan, values, nulls,
				SPI_OK_UPDATE);
		if (SPI_processed == 0)
		{
			spi_exec_query("INSERT INTO dbms_stats.relation_stats_locked "
					"(relname, relpages, reltuples, curpages, "
					"last_analyze, last_autoanalyze, relid"
#if PG_VERSION_NUM >= 90200
					", relallvisible"
#endif
					") VALUES (quote_ident($1) || '.' || quote_ident($2), "
					"$3, $4, $5, $6, $7, $8"
#if PG_VERSION_NUM >= 90200
					", $9"
#endif
					")",
					RELATION_PARAM_NUM, r_types, &r_ins_plan, values, nulls,
					SPI_OK_INSERT);
			/*  If we failed to insert, we can't proceed. */
			if (SPI_processed != 1)
				elog(ERROR, "failed to insert import data");
		}

		elog(DEBUG2, "\"%s.%s\" relation statistic import",
			DatumGetName(w_nspname)->data, DatumGetName(w_relname)->data);

		/*
		 * Determine the attnum of the attribute with given name, and load
		 * statistics from temp table into dbms.column_stats_locked.
		 */
		spi_exec_query("SELECT w.stainherit, w.attname, a.attnum, "
							  "w.nspname_of_typename, tn.nspname, "
							  "w.typname, t.typname, w.atttypmod, a.atttypmod "
						 "FROM pg_catalog.pg_class c "
						 "JOIN pg_catalog.pg_namespace cn "
						   "ON (cn.oid = c.relnamespace) "
						 "JOIN pg_catalog.pg_attribute a "
						   "ON (a.attrelid = c.oid) "
						 "JOIN pg_catalog.pg_type t "
						   "ON (t.oid = a.atttypid) "
						 "JOIN pg_catalog.pg_namespace tn "
						   "ON (tn.oid = t.typnamespace) "
						"RIGHT JOIN dbms_stats_work_stats w "
						   "ON (w.nspname = cn.nspname AND w.relname = c.relname "
							   "AND (w.attname = a.attname OR w.attname = '')) "
						"WHERE w.nspname = $1 AND w.relname = $2 "
						  "AND a.attnum > 0"
						"ORDER BY 1, 3, 2",
				2, r_types, &c_sel_plan, values, NULL, SPI_OK_SELECT);

		/* This query ought to return at least one record. */
		if (SPI_processed == 0)
			elog(ERROR, "no per-column statistic data to be imported");

		values[0] = w_relid;
		values[2] = w_nspname;
		values[3] = w_relname;

		c_num = SPI_processed;
		c_tups = SPI_tuptable->vals;
		c_tupdesc = SPI_tuptable->tupdesc;
		for (j = 0; j < c_num; j++)
		{
			char   *w_typnamespace;
			char   *a_typnamespace;
			char   *w_typname;
			char   *a_typname;
			int		w_typmod;
			int		a_typmod;

			/*
			 * If we have only per-relation statistics in source, all of
			 * column_stats_effective for per-column statistics are NULL.
			 */
			(void) SPI_getbinval(c_tups[j], c_tupdesc, 1, &isnull);
			if (isnull)
				continue;

			/*
			 * If there is no column with given name, we skip the rest of
			 * import process.
			 */
			values[4] = SPI_getbinval(c_tups[j], c_tupdesc, 2, &isnull);
			values[1] = SPI_getbinval(c_tups[j], c_tupdesc, 3, &isnull);
			if (isnull)
			{
				elog(WARNING, "column \"%s\" of \"%s.%s\" does not exist",
					DatumGetName(values[4])->data,
						DatumGetName(w_nspname)->data,
						DatumGetName(w_relname)->data);
				continue;
			}

			/*
			 * If the destination column has different data type from source
			 * column, we stop importing to avoid corrupted statistics.
			 */
			w_typnamespace = DatumGetName(SPI_getbinval(c_tups[j], c_tupdesc, 4,
						&isnull))->data;
			a_typnamespace = DatumGetName(SPI_getbinval(c_tups[j], c_tupdesc, 5,
						&isnull))->data;
			w_typname = DatumGetName(SPI_getbinval(c_tups[j], c_tupdesc, 6,
						&isnull))->data;
			a_typname = DatumGetName(SPI_getbinval(c_tups[j], c_tupdesc, 7,
						&isnull))->data;
			if (strcmp(w_typnamespace, a_typnamespace) != 0 ||
				strcmp(w_typname, a_typname) != 0)
			{
				ereport(WARNING,
						(errcode(ERRCODE_DATATYPE_MISMATCH),
						 errmsg("column \"%s\" is of type \"%s.%s\""
								" but import data is of type \"%s.%s\"",
								DatumGetName(values[4])->data,
								a_typnamespace, a_typname,
								w_typnamespace, w_typname)));
				continue;
			}

			/*
			 * If the atttypmod of the destination column is different from the
			 * one of source, column, we stop importing to avoid corrupted
			 * statistics.
			 */
			w_typmod = DatumGetInt32(SPI_getbinval(c_tups[j], c_tupdesc, 8,
						&isnull));
			a_typmod = DatumGetInt32(SPI_getbinval(c_tups[j], c_tupdesc, 9,
						&isnull));
			if (w_typmod != a_typmod)
			{
				ereport(WARNING,
						(errcode(ERRCODE_DATATYPE_MISMATCH),
						 errmsg("column \"%s\" is of atttypmod %d"
								" but import data is of atttypmod %d",
								DatumGetName(values[4])->data,
								a_typmod, a_typmod)));
				continue;
			}

			/*
			 * First delete old dummy statistics, and import new one.  We use
			 * DELETE-then-INSERT method here to simplify codes.
			 */
			spi_exec_query("DELETE FROM dbms_stats.column_stats_locked "
					"WHERE starelid = $1 AND staattnum = $2", 2, c_types,
					&c_del_plan, values, NULL, SPI_OK_DELETE);

			spi_exec_query("INSERT INTO dbms_stats.column_stats_locked "
				"SELECT $1, $2, "
				"stainherit, stanullfrac, stawidth, stadistinct, "
				"stakind1, stakind2, stakind3, stakind4, "
#if PG_VERSION_NUM >= 90200
				"stakind5, "
#endif
				"staop1, staop2, staop3, staop4, "
#if PG_VERSION_NUM >= 90200
				"staop5, "
#endif
				"stanumbers1, stanumbers2, stanumbers3, stanumbers4, "
#if PG_VERSION_NUM >= 90200
				"stanumbers5, "
#endif
				"stavalues1, stavalues2, stavalues3, stavalues4 "
#if PG_VERSION_NUM >= 90200
				", stavalues5 "
#endif
				"FROM dbms_stats_work_stats "
				"WHERE nspname = $3 AND relname = $4 "
				"AND attname = $5 "
				"ORDER BY 3",
				5, c_types, &c_ins_plan, values, NULL, SPI_OK_INSERT);

			elog(DEBUG2, "\"%s.%s.%s\" column statistic import",
				DatumGetName(w_nspname)->data,
				DatumGetName(w_relname)->data, DatumGetName(values[4])->data);
		}

		if (c_num == 0)
			elog(DEBUG2, "\"%s.%s\" column statistic no data",
				DatumGetName(w_nspname)->data, DatumGetName(w_relname)->data);
	}

	/* release the cached plan */
	SPI_freeplan(r_upd_plan);
	SPI_freeplan(r_ins_plan);
	SPI_freeplan(c_sel_plan);
	SPI_freeplan(c_del_plan);
	SPI_freeplan(c_ins_plan);

	/* delete the temp table */
	spi_exec_utility("DROP TABLE dbms_stats_work_stats");

	/* disconnect SPI */
	ret = SPI_finish();
	if (ret != SPI_OK_FINISH)
		elog(ERROR, "pg_dbms_stats: SPI_finish => %d", ret);

	/*
	 * Recover the protocol state because it has been invalidated by our
	 * COPY-from-stdin.
	 */
	if (filename == NULL)
		pq_puttextmessage('C', "dbms_stats_import");

	PG_RETURN_VOID();
}

/*
 * spi_exec_utility
 *   Execute given utility command via SPI.
 */
static void
spi_exec_utility(const char *query)
{
	int	ret;

	ret = SPI_exec(query, 0);
	if (ret != SPI_OK_UTILITY)
		elog(ERROR, "pg_dbms_stats: SPI_exec => %d", ret);
}

/*
 * spi_exec_query
 *   Execute given SQL command via SPI.
 *   The plan will be cached by SPI_prepare if it hasn't been.
 */
static void
spi_exec_query(const char *query, int nargs, Oid *argtypes, SPIPlanPtr *plan,
				Datum *values, const char *nulls, int result)
{
	int	ret;

	if (*plan == NULL)
		*plan = SPI_prepare(query, nargs, argtypes);

	ret = SPI_execute_plan(*plan, values, nulls, false, 0);
	if (ret != result)
		elog(ERROR, "pg_dbms_stats: SPI_execute_plan => %d", ret);
}

static char *
get_text_arg(FunctionCallInfo fcinfo, int n, bool is_name)
{
	text   *arg;
	char   *s;
	int		len;
	char   *result;

	arg = PG_GETARG_TEXT_PP(n);
	s = text_to_cstring(arg);
	PG_FREE_IF_COPY(arg, n);

	if (!is_name)
		return s;

	len = strlen(s);

	/* Truncate oversize input */
	if (len >= NAMEDATALEN)
		len = pg_mbcliplen(s, len, NAMEDATALEN - 1);

	/* We use palloc0 here to ensure result is zero-padded */
	result = (char *) palloc0(NAMEDATALEN);
	memcpy(result, s, len);
	pfree(s);

	return result;
}

/*
 * get_args
 *   Retrieve arguments from FunctionCallInfo and validate them.  We assume
 *   that order of arguments is:
 *     1) schema name
 *     2) relation oid
 *     3) attribute name
 *     4) absolute path of source file, or 'stdin' (case insensitive)
 */
static void
get_args(FunctionCallInfo fcinfo, char **nspname, char **relname,
		char **attname, char **filename)
{
	Oid				nspid;
	Oid				relid;
	AttrNumber		attnum;
	HeapTuple		tp;
	Form_pg_class	reltup;
	char			relkind;

	*nspname = *relname = *attname = *filename = NULL;

	/*
	 * First of all, check whether combination of arguments is consistent.
	 *
	 * 1) relid and attname can't be used with schemaname.
	 * 2) relid is required when attname is given.
	 */
	if (!PG_ARGISNULL(0) && (!PG_ARGISNULL(1) || !PG_ARGISNULL(2)))
		elog(ERROR, "relid and attnum can not be used with schemaname");
	else if (PG_ARGISNULL(1) && !PG_ARGISNULL(2))
		elog(ERROR, "relation is required");

	/* filepath validation */
	if (!PG_ARGISNULL(3))
	{
		*filename = get_text_arg(fcinfo, 3, false);

		/*
		 * If given filepath is "stdin", clear filename to tell caller to
		 * import from standard input.  Note that we accept only absolute path
		 * for security reason.
		 */
		if (pg_strcasecmp(*filename, "stdin") == 0)
			*filename = NULL;
		else if (!is_absolute_path(*filename))
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_NAME),
					 errmsg("relative path not allowed for dbms_stats_export"
							" to file")));
	}

	/* schemaname validation */
	if (!PG_ARGISNULL(0))
	{
		*nspname = get_text_arg(fcinfo, 0, true);

		/* check that a schema with given name exists */
		get_namespace_oid(*nspname, false);

		/* check that given schema is not one of system schemas */
		if (dbms_stats_is_system_schema_internal(*nspname))
			elog(ERROR, "\"%s\" is a system catalog", *nspname);
	}

	/* table oid validation */
	if (!PG_ARGISNULL(1))
	{
		relid = PG_GETARG_OID(1);
		tp = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
		if (!HeapTupleIsValid(tp))
			elog(ERROR, "relid %d does not exist", relid);

		/* check that the target is an ordinary table or an index */
		reltup = (Form_pg_class) GETSTRUCT(tp);
		*relname = pstrdup(reltup->relname.data);
		relkind = reltup->relkind;
		nspid = reltup->relnamespace;
		ReleaseSysCache(tp);

		if (relkind != RELKIND_RELATION && relkind != RELKIND_INDEX
#if PG_VERSION_NUM >= 90200
			&& relkind != RELKIND_FOREIGN_TABLE
#endif
#if PG_VERSION_NUM >= 90300
			&& relkind != RELKIND_MATVIEW
#endif
		)
			elog(ERROR, "relkind of \"%s\" is \"%c\", can not import",
				get_rel_name(relid), relkind);

		/* check that the relation is not in one of system schemas */
		*nspname = get_namespace_name(nspid);
		if (dbms_stats_is_system_schema_internal(*nspname))
			elog(ERROR, "\"%s\" is a system catalog", *nspname);

		/* attribute name validation */
		if (!PG_ARGISNULL(2))
		{
			*attname = get_text_arg(fcinfo, 2, true);
			attnum = get_attnum(relid, *attname);
			if (!AttributeNumberIsValid(attnum))
				elog(ERROR, "column \"%s\" of \"%s.%s\" does not exist", *attname, *nspname, *relname);
		}
	}
}

/*
 * appendLiteral - Format a string as a SQL literal, append to buf
 *
 * This function was copied from simple_quote_literal() in
 * src/backend/utils/adt/ruleutils.c
 */
static void
appendLiteral(StringInfo buf, const char *val)
{
	const char *valptr;

	/*
	 * We form the string literal according to the prevailing setting of
	 * standard_conforming_strings; we never use E''. User is responsible for
	 * making sure result is used correctly.
	 */
	appendStringInfoChar(buf, '\'');
	for (valptr = val; *valptr; valptr++)
	{
		char		ch = *valptr;

		if (SQL_STR_DOUBLE(ch, !standard_conforming_strings))
			appendStringInfoChar(buf, ch);
		appendStringInfoChar(buf, ch);
	}
	appendStringInfoChar(buf, '\'');
}

/*
 * import_stats_from_file
 *	 load data from file or stdin into work table, and delete unnecessary
 *	 records.
 */
static void
import_stats_from_file(char *filename, char *nspname, char *relname,
	char *attname)
{
	StringInfoData	buf;
	List		   *parsetree_list;
	uint64			processed;
	Datum			values[3];
	Oid				argtypes[3] = { CSTRINGOID, CSTRINGOID, CSTRINGOID };
	char			nulls[3] = { 'n', 'n', 'n' };
	int				nargs;
	int				ret;

	/* for debug use */
	elog(DEBUG3, "%s() f=%s n=%s r=%s a=%s", __FUNCTION__,
		 filename ? filename : "(null)",
		 nspname ? nspname : "(null)",
		 relname ? relname : "(null)",
		 attname ? attname : "(null)");

	/*
	 * Construct COPY statement.  NULL for filename indicates that source is
	 * stdin.
	 */
	initStringInfo(&buf);
	appendStringInfoString(&buf, "COPY dbms_stats_work_stats FROM ");
	if (filename == NULL)
		appendStringInfoString(&buf, "stdin");
	else
		appendLiteral(&buf, filename);

	appendStringInfoString(&buf, " (FORMAT 'binary')");

	/* Execute COPY FROM command. */
	parsetree_list = pg_parse_query(buf.data);
#if PG_VERSION_NUM >= 90300
	DoCopy((CopyStmt *)linitial(parsetree_list), buf.data, &processed);
#else
	processed = DoCopy((CopyStmt *)linitial(parsetree_list), buf.data);
#endif

	if (processed == 0)
		elog(ERROR, "no data to be imported");

	/*
	 * Delete the statistics other than the specified object's statistic from
	 * the temp table.  We can skip DELETEing staging data when schemaname is
	 * NULL, because it means database-wise import.
	 */
	if (nspname == NULL)
		return;

	resetStringInfo(&buf);
	appendStringInfoString(&buf,
						   "DELETE FROM dbms_stats_work_stats "
						   " WHERE nspname <> $1::text ");
	values[0] = CStringGetDatum(nspname);
	nulls[0] = 't';
	nargs = 1;

	if (relname != NULL)
	{
		values[1] = CStringGetDatum(relname);
		nulls[1] = 't';
		nargs++;
		appendStringInfoString(&buf, " OR (relname <> $2::text) ");

		if (attname != NULL)
		{
			values[2] = CStringGetDatum(attname);
			nulls[2] = 't';
			nargs++;
			appendStringInfoString(&buf, " OR (attname <> $3::text) ");
		}
	}

	ret = SPI_execute_with_args(buf.data, nargs, argtypes, values, nulls,
								 false, 0);
	if (ret != SPI_OK_DELETE)
		elog(ERROR, "pg_dbms_stats: SPI_execute_with_args => %d", ret);
}

#ifdef UNIT_TEST
void test_import(int *passed, int *total);
static void test_spi_exec_query(int *passed, int *total);
static void test_spi_exec_utility(int *passed, int *total);
static void test_appendLiteral(int *passed, int *total);

#define StringEq(actual, expected)	\
		(strcmp((actual), (expected)) == 0 ? 1 : 0)

/*
 * Test appendLiteral function
 */
static void
test_appendLiteral(int *passed, int *total)
{
	bool			org_standard_conforming_strings;
	int				caseno = 0;
	StringInfoData	buf;

	/* Backup current GUC parameters */
	NewGUCNestLevel();
	org_standard_conforming_strings = standard_conforming_strings;

	/* Initialize resources for tests */
	initStringInfo(&buf);

	/*
	 * *-*-1:
	 *   - no special char
	 */
	caseno++;
	resetStringInfo(&buf);
	appendStringInfoString(&buf, "BEFORE");
	appendLiteral(&buf, "\"abc 123\tあいう\n\"");
	if (StringEq(buf.data, "BEFORE'\"abc 123\tあいう\n\"'"))
	{
		elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
		(*passed)++;
	}
	else
	{
		elog(WARNING, "%s-%d failed: [%s]", __FUNCTION__, caseno, buf.data);
	}

	/*
	 * *-*-2:
	 *   - contains special chars (single quote, back slash),
	 *   - standard_conforming_strings is true
	 */
	caseno++;
	resetStringInfo(&buf);
	appendStringInfoString(&buf, "BEFORE");
	standard_conforming_strings = true;
	appendLiteral(&buf, "'abc 123\tあいう\n\\");
	if (StringEq(buf.data, "BEFORE'''abc 123\tあいう\n\\'"))
	{
		elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
		(*passed)++;
	}
	else
	{
		elog(WARNING, "%s-%d failed: [%s]", __FUNCTION__, caseno, buf.data);
	}

	/*
	 * *-*-3:
	 *   - contains special chars (single quote, back slash),
	 *   - standard_conforming_strings is false
	 */
	caseno++;
	resetStringInfo(&buf);
	appendStringInfoString(&buf, "BEFORE");
	standard_conforming_strings = false;
	appendLiteral(&buf, "'abc 123\tあいう\n\\");
	if (StringEq(buf.data, "BEFORE'''abc 123\tあいう\n\\\\'"))
	{
		elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
		(*passed)++;
	}
	else
	{
		elog(WARNING, "%s-%d failed: [%s]", __FUNCTION__, caseno, buf.data);
	}

	/*
	 * *-*-4:
	 *   - empty string
	 */
	caseno++;
	resetStringInfo(&buf);
	appendStringInfoString(&buf, "BEFORE");
	appendLiteral(&buf, "");
	if (StringEq(buf.data, "BEFORE''"))
	{
		elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
		(*passed)++;
	}
	else
	{
		elog(WARNING, "%s-%d failed: [%s]", __FUNCTION__, caseno, buf.data);
	}

	/* report # of tests */
	*total += caseno;

	/* Restore GUC parameters */
	standard_conforming_strings = org_standard_conforming_strings;
}

static void
test_spi_exec_query(int *passed, int *total)
{
	int				rc;
	volatile int	caseno = 0;
	SPIPlanPtr		ptr = NULL;
	SPIPlanPtr		org_ptr;

	/* Initialize */
	rc = SPI_connect();
	if (rc != SPI_OK_CONNECT)
		elog(ERROR, "could not connect SPI: %s", SPI_result_code_string(rc));

	/*
	 * *-*-1
	 *   - plan is not cached
	 */
	caseno++;
	BeginInternalSubTransaction("test");
	PG_TRY();
	{
		spi_exec_query("SELECT 1", 0, NULL, &ptr, NULL, NULL, SPI_OK_SELECT);
		if (ptr != NULL && SPI_processed == 1)
		{
			elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
			(*passed)++;
		}
		ReleaseCurrentSubTransaction();
	}
	PG_CATCH();
	{
		elog(WARNING, "*-*-%d failed", caseno);
		RollbackAndReleaseCurrentSubTransaction();
		SPI_restore_connection();
	}
	PG_END_TRY();

	/*
	 * *-*-2
	 *   - plan is cached
	 */
	caseno++;
	BeginInternalSubTransaction("test");
	PG_TRY();
	{
		org_ptr = ptr;
		spi_exec_query(NULL, 0, NULL, &ptr, NULL, NULL, SPI_OK_SELECT);
		if (ptr == org_ptr && SPI_processed == 1)
		{
			elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
			(*passed)++;
		}
		ReleaseCurrentSubTransaction();
	}
	PG_CATCH();
	{
		elog(WARNING, "*-*-%d failed", caseno);
		RollbackAndReleaseCurrentSubTransaction();
		FlushErrorState();
		SPI_restore_connection();
	}
	PG_END_TRY();
	SPI_freeplan(ptr);
	ptr = NULL;

	/*
	 * *-*-3
	 *   - query error
	 */
	caseno++;
	BeginInternalSubTransaction("test");
	PG_TRY();
	{
		spi_exec_query("SELECT 1 / 0",
					   0, NULL, &ptr, NULL, NULL, SPI_OK_SELECT);
		elog(WARNING, "*-*-%d failed", caseno);
		ReleaseCurrentSubTransaction();
	}
	PG_CATCH();
	{
		elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
		(*passed)++;
		RollbackAndReleaseCurrentSubTransaction();
		FlushErrorState();
		SPI_restore_connection();
	}
	PG_END_TRY();
	SPI_freeplan(ptr);
	ptr = NULL;

	/*
	 * *-*-4
	 *   - query success
	 */
	caseno++;
	BeginInternalSubTransaction("test");
	PG_TRY();
	{
		spi_exec_query("SELECT 1", 0, NULL, &ptr, NULL, NULL, SPI_OK_SELECT);
		if (ptr != NULL && SPI_processed == 1)
		{
			elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
			(*passed)++;
		}
		ReleaseCurrentSubTransaction();
	}
	PG_CATCH();
	{
		elog(WARNING, "*-*-%d failed", caseno);
		PG_RE_THROW();
		RollbackAndReleaseCurrentSubTransaction();
		SPI_restore_connection();
	}
	PG_END_TRY();
	SPI_freeplan(ptr);
	ptr = NULL;

	/* report # of tests */
	(*total) += caseno;

	/* Cleanup */
	rc = SPI_finish();
	if (rc != SPI_OK_FINISH && rc != SPI_ERROR_UNCONNECTED)
		elog(ERROR, "could not finish SPI: %s", SPI_result_code_string(rc));
}

static void
test_spi_exec_utility(int *passed, int *total)
{
	int				rc;
	volatile int	caseno = 0;

	/* Initialize */
	rc = SPI_connect();
	if (rc != SPI_OK_CONNECT)
		elog(ERROR, "could not connect SPI: %s", SPI_result_code_string(rc));

	/*
	 * *-*-1
	 *   - query error
	 */
	caseno++;
	BeginInternalSubTransaction("test");
	PG_TRY();
	{
		spi_exec_utility("RESET dummy_parameter");
		elog(WARNING, "*-*-%d failed", caseno);
		ReleaseCurrentSubTransaction();
	}
	PG_CATCH();
	{
		elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
		(*passed)++;
		RollbackAndReleaseCurrentSubTransaction();
		FlushErrorState();
		SPI_restore_connection();
	}
	PG_END_TRY();

	/*
	 * *-*-2
	 *   - query success
	 */
	caseno++;
	BeginInternalSubTransaction("test");
	PG_TRY();
	{
		spi_exec_utility("RESET client_min_messages");
		elog(WARNING, "%s-%d ok", __FUNCTION__, caseno);
		(*passed)++;
		ReleaseCurrentSubTransaction();
	}
	PG_CATCH();
	{
		elog(WARNING, "*-*-%d failed", caseno);
		RollbackAndReleaseCurrentSubTransaction();
		SPI_restore_connection();
	}
	PG_END_TRY();

	/* report # of tests */
	(*total) += caseno;

	/* Cleanup */
	rc = SPI_finish();
	if (rc != SPI_OK_FINISH && rc != SPI_ERROR_UNCONNECTED)
		elog(ERROR, "could not finish SPI: %s", SPI_result_code_string(rc));
}

/*
 * Unit test entry point for import.c.  This will be called by PG_init()
 * function, after initialization for this extension is completed .
 * This funciton should add the numbers of tests passed and the total number of
 * tests to parameter passed and total respectively.
 */
void
test_import(int *passed, int *total)
{
	int local_passed = 0;
	int local_total = 0;

	elog(WARNING, "==========");

	/* Do tests here */
	test_appendLiteral(&local_passed, &local_total);
	test_spi_exec_query(&local_passed, &local_total);
	test_spi_exec_utility(&local_passed, &local_total);

	elog(WARNING, "%s %d/%d passed", __FUNCTION__, local_passed, local_total);
	*passed += local_passed;
	*total += local_total;
}

#endif