/*
 * pg_get_serverdef_string finds the foreign server that corresponds to the
 * given foreign tableId, and returns this server's definition.
 */
char *
pg_get_serverdef_string(Oid tableRelationId)
{
	ForeignTable *foreignTable = GetForeignTable(tableRelationId);
	ForeignServer *server = GetForeignServer(foreignTable->serverid);
	ForeignDataWrapper *foreignDataWrapper = GetForeignDataWrapper(server->fdwid);

	StringInfoData buffer = { NULL, 0, 0, 0 };
	initStringInfo(&buffer);

	appendStringInfo(&buffer, "CREATE SERVER %s", quote_identifier(server->servername));
	if (server->servertype != NULL)
	{
		appendStringInfo(&buffer, " TYPE %s",
						 quote_literal_cstr(server->servertype));
	}
	if (server->serverversion != NULL)
	{
		appendStringInfo(&buffer, " VERSION %s",
						 quote_literal_cstr(server->serverversion));
	}

	appendStringInfo(&buffer, " FOREIGN DATA WRAPPER %s",
					 quote_identifier(foreignDataWrapper->fdwname));

	/* append server options, if any */
	AppendOptionListToString(&buffer, server->options);

	return (buffer.data);
}

/*
 * pg_get_extensiondef_string finds the foreign data wrapper that corresponds to
 * the given foreign tableId, and checks if an extension owns this foreign data
 * wrapper. If it does, the function returns the extension's definition. If not,
 * the function returns null.
 */
char *
pg_get_extensiondef_string(Oid tableRelationId)
{
	ForeignTable *foreignTable = GetForeignTable(tableRelationId);
	ForeignServer *server = GetForeignServer(foreignTable->serverid);
	ForeignDataWrapper *foreignDataWrapper = GetForeignDataWrapper(server->fdwid);
	StringInfoData buffer = { NULL, 0, 0, 0 };

	Oid classId = ForeignDataWrapperRelationId;
	Oid objectId = server->fdwid;

	Oid extensionId = getExtensionOfObject(classId, objectId);
	if (OidIsValid(extensionId))
	{
		char *extensionName = get_extension_name(extensionId);
		Oid extensionSchemaId = get_extension_schema(extensionId);
		char *extensionSchema = get_namespace_name(extensionSchemaId);

		initStringInfo(&buffer);
		appendStringInfo(&buffer, "CREATE EXTENSION IF NOT EXISTS %s WITH SCHEMA %s",
						 quote_identifier(extensionName),
						 quote_identifier(extensionSchema));
	}
	else
	{
		ereport(NOTICE, (errmsg("foreign-data wrapper \"%s\" does not have an "
								"extension defined", foreignDataWrapper->fdwname)));
	}

	return (buffer.data);
}

/*
 * Copied from src/backend/utils/adt/ruleutils.c, not exported.
 *
 * get_opclass_name			- fetch name of an index operator class
 *
 * The opclass name is appended (after a space) to buf.
 *
 * Output is suppressed if the opclass is the default for the given
 * actual_datatype.  (If you don't want this behavior, just pass
 * InvalidOid for actual_datatype.)
 */
void
get_opclass_name(Oid opclass, Oid actual_datatype,
				 StringInfo buf)
{
	HeapTuple	ht_opc;
	Form_pg_opclass opcrec;
	char	   *opcname;
	char	   *nspname;

	ht_opc = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclass));
	if (!HeapTupleIsValid(ht_opc))
		elog(ERROR, "cache lookup failed for opclass %u", opclass);
	opcrec = (Form_pg_opclass) GETSTRUCT(ht_opc);

	if (!OidIsValid(actual_datatype) ||
		GetDefaultOpClass(actual_datatype, opcrec->opcmethod) != opclass)
	{
		/* Okay, we need the opclass name.  Do we need to qualify it? */
		opcname = NameStr(opcrec->opcname);
		if (OpclassIsVisible(opclass))
			appendStringInfo(buf, " %s", quote_identifier(opcname));
		else
		{
			nspname = get_namespace_name(opcrec->opcnamespace);
			appendStringInfo(buf, " %s.%s",
							 quote_identifier(nspname),
							 quote_identifier(opcname));
		}
	}
	ReleaseSysCache(ht_opc);
}

static StringInfo
get_dymanic_record_fields(PlxFn *plx_fn, FunctionCallInfo fcinfo)
{
    StringInfo buf;
    Oid        oid;
    TupleDesc  tuple_desc;
    int        i;

    get_call_result_type(fcinfo, &oid, &tuple_desc);

    buf = makeStringInfo();
    for (i = 0; i < tuple_desc->natts; i++)
    {
        Form_pg_attribute a;
        HeapTuple         type_tuple;
        Form_pg_type      type_struct;

        a = tuple_desc->attrs[i];
        type_tuple = SearchSysCache(TYPEOID, ObjectIdGetDatum(a->atttypid), 0, 0, 0);
        if (!HeapTupleIsValid(type_tuple))
            plx_error(plx_fn, "cache lookup failed for type %u", a->atttypid);
        type_struct = (Form_pg_type) GETSTRUCT(type_tuple);
        {
            appendStringInfo(
                buf,
                "%s%s %s",
                ((i > 0) ? ", " : ""),
                quote_identifier(NameStr(a->attname)),
                quote_identifier(NameStr(type_struct->typname)));
        }
        ReleaseSysCache(type_tuple);
    }
    return buf;
}

/*
 * DropErrorTable
 *
 * Drop the error table from the database. This function will be called from
 * destroyCdbSreh when an autogenerated error table was not used in the COPY
 * operation granted KEEP wasn't specified.
 *
 */
static
void DropErrorTable(CdbSreh *cdbsreh)
{
	StringInfoData dropstmt;
	RangeVar *errtbl_rv;
	
	Insist(Gp_role == GP_ROLE_DISPATCH);

	ereport(NOTICE,
			(errcode(ERRCODE_SUCCESSFUL_COMPLETION),
			 errmsg("Dropping the auto-generated unused error table"),
			 errhint("Use KEEP in LOG INTO clause to force keeping the error table alive")));								
	
	initStringInfo(&dropstmt); 
	
	appendStringInfo(&dropstmt, "DROP TABLE %s.%s",
					 quote_identifier(get_namespace_name(RelationGetNamespace(cdbsreh->errtbl))),
					 quote_identifier(RelationGetRelationName(cdbsreh->errtbl)));
	
	errtbl_rv = makeRangeVar(get_namespace_name(RelationGetNamespace(cdbsreh->errtbl)),
							 RelationGetRelationName(cdbsreh->errtbl), -1);

	/* DROP the relation on the QD */
	RemoveRelation(errtbl_rv,DROP_RESTRICT, NULL, RELKIND_RELATION);
				   
	/* dispatch the DROP to the QEs */
	CdbDoCommand(dropstmt.data, false, /*no txn */ false);

	pfree(dropstmt.data);
}

/*
 * Creates and populates a sample table for a PXF table.
 * The actual queried table is not the original PXF table but a copy of it
 * with additional attributes to enable sampling.
 *
 * The results are stored in sampleTableRelTuples.
 */
static void buildSampleFromPxf(const char* sampleSchemaName,
		const char* sampleTableName,
		const char* pxfSampleTable,
		List *lAttributeNames,
		float4 *sampleTableRelTuples)
{
	int nAttributes = -1;
	int i = 0;
	ListCell *le = NULL;
	StringInfoData str;

	initStringInfo(&str);

	appendStringInfo(&str, "create table %s.%s as (select ",
			quote_identifier(sampleSchemaName), quote_identifier(sampleTableName));

	nAttributes = list_length(lAttributeNames);

	foreach_with_count(le, lAttributeNames, i)
	{
		appendStringInfo(&str, "Ta.%s", quote_identifier((const char *) lfirst(le)));
		if (i < nAttributes - 1)
		{
			appendStringInfo(&str, ", ");
		}
		else
		{
			appendStringInfo(&str, " ");
		}
	}

/*
 * Get the max size of the relation across segments
 */
int64
cdbRelMaxSegSize(Relation rel)
{
	int64 size = 0;
	int i;
	CdbPgResults cdb_pgresults = {NULL, 0};
	StringInfoData buffer;

	char *schemaName;
	char *relName;

	/*
	 * Let's ask the QEs for the size of the relation
	 */
	initStringInfo(&buffer);

	schemaName = get_namespace_name(RelationGetNamespace(rel));
	if (schemaName == NULL)
		elog(ERROR, "cache lookup failed for namespace %d",
			 RelationGetNamespace(rel));
	relName = RelationGetRelationName(rel);

	/*
	 * Safer to pass names than oids, just in case they get out of sync between QD and QE,
	 * which might happen with a toast table or index, I think (but maybe not)
	 */
	appendStringInfo(&buffer, "select pg_relation_size('%s.%s')",
					 quote_identifier(schemaName), quote_identifier(relName));

	CdbDispatchCommand(buffer.data, DF_WITH_SNAPSHOT, &cdb_pgresults);

	for (i = 0; i < cdb_pgresults.numResults; i++)
	{
		struct pg_result * pgresult = cdb_pgresults.pg_results[i];
		if (PQresultStatus(pgresult) != PGRES_TUPLES_OK)
		{
			cdbdisp_clearCdbPgResults(&cdb_pgresults);
			elog(ERROR,"cdbRelMaxSegSize: resultStatus not tuples_Ok: %s %s",
				 PQresStatus(PQresultStatus(pgresult)),PQresultErrorMessage(pgresult));
		}
		else
		{
			Assert(PQntuples(pgresult) == 1);
			int64 tempsize = 0;
			(void) scanint8(PQgetvalue(pgresult, 0, 0), false, &tempsize);
			if (tempsize > size)
				size = tempsize;
		}
	}

	pfree(buffer.data);

	cdbdisp_clearCdbPgResults(&cdb_pgresults);

	return size;
}

/*
 * Append a psql meta-command that connects to the given database with the
 * then-current connection's user, host and port.
 */
void
appendPsqlMetaConnect(PQExpBuffer buf, const char *dbname)
{
	const char *s;
	bool		complex;

	/*
	 * If the name is plain ASCII characters, emit a trivial "\connect "foo"".
	 * For other names, even many not technically requiring it, skip to the
	 * general case.  No database has a zero-length name.
	 */
	complex = false;
	for (s = dbname; *s; s++)
	{
		if (*s == '\n' || *s == '\r')
		{
			fprintf(stderr,
					_("database name contains a newline or carriage return: \"%s\"\n"),
					dbname);
			exit(EXIT_FAILURE);
		}

		if (!((*s >= 'a' && *s <= 'z') || (*s >= 'A' && *s <= 'Z') ||
			  (*s >= '0' && *s <= '9') || *s == '_' || *s == '.'))
		{
			complex = true;
		}
	}

	appendPQExpBufferStr(buf, "\\connect ");
	if (complex)
	{
		PQExpBufferData connstr;

		initPQExpBuffer(&connstr);
		appendPQExpBuffer(&connstr, "dbname=");
		appendConnStrVal(&connstr, dbname);

		appendPQExpBuffer(buf, "-reuse-previous=on ");

		/*
		 * As long as the name does not contain a newline, SQL identifier
		 * quoting satisfies the psql meta-command parser.  Prefer not to
		 * involve psql-interpreted single quotes, which behaved differently
		 * before PostgreSQL 9.2.
		 */
		appendPQExpBufferStr(buf, quote_identifier(connstr.data));

		termPQExpBuffer(&connstr);
	}
	else
		appendPQExpBufferStr(buf, quote_identifier(dbname));
	appendPQExpBufferChar(buf, '\n');
}

/*
 * It is possible for there to be a mismatch in the need for TOAST tables
 * between the old and new servers, e.g. some pre-9.1 tables didn't need
 * TOAST tables but will need them in 9.1+.  (There are also opposite cases,
 * but these are handled by setting binary_upgrade_next_toast_pg_class_oid.)
 *
 * We can't allow the TOAST table to be created by pg_dump with a
 * pg_dump-assigned oid because it might conflict with a later table that
 * uses that oid, causing a "file exists" error for pg_class conflicts, and
 * a "duplicate oid" error for pg_type conflicts.  (TOAST tables need pg_type
 * entries.)
 *
 * Therefore, a backend in binary-upgrade mode will not create a TOAST
 * table unless an OID as passed in via pg_upgrade_support functions.
 * This function is called after the restore and uses ALTER TABLE to
 * auto-create any needed TOAST tables which will not conflict with
 * restored oids.
 */
void
optionally_create_toast_tables(void)
{
	int			dbnum;

	prep_status("Creating newly-required TOAST tables");

	for (dbnum = 0; dbnum < new_cluster.dbarr.ndbs; dbnum++)
	{
		PGresult   *res;
		int			ntups;
		int			rowno;
		int			i_nspname,
					i_relname;
		DbInfo	   *active_db = &new_cluster.dbarr.dbs[dbnum];
		PGconn	   *conn = connectToServer(&new_cluster, active_db->db_name);

		res = executeQueryOrDie(conn,
								"SELECT n.nspname, c.relname "
								"FROM	pg_catalog.pg_class c, "
								"		pg_catalog.pg_namespace n "
								"WHERE	c.relnamespace = n.oid AND "
							  "		n.nspname NOT IN ('pg_catalog', 'information_schema') AND "
								"c.relkind IN ('r', 'm') AND "
								"c.reltoastrelid = 0");

		ntups = PQntuples(res);
		i_nspname = PQfnumber(res, "nspname");
		i_relname = PQfnumber(res, "relname");
		for (rowno = 0; rowno < ntups; rowno++)
		{
			/* enable auto-oid-numbered TOAST creation if needed */
			PQclear(executeQueryOrDie(conn, "SELECT binary_upgrade.set_next_toast_pg_class_oid('%d'::pg_catalog.oid);",
					OPTIONALLY_CREATE_TOAST_OID));

			/* dummy command that also triggers check for required TOAST table */
			PQclear(executeQueryOrDie(conn, "ALTER TABLE %s.%s RESET (binary_upgrade_dummy_option);",
					quote_identifier(PQgetvalue(res, rowno, i_nspname)),
					quote_identifier(PQgetvalue(res, rowno, i_relname))));
		}

		PQclear(res);

		PQfinish(conn);
	}

	check_ok();
}

/*
 * Print a WHERE clause item
 */
static void
print_where_clause_item(StringInfo s,
						Relation relation,
						HeapTuple tuple,
						int natt,
						bool *first_column)
{
	Form_pg_attribute	attr;
	Datum				origval;
	bool				isnull;
	TupleDesc			tupdesc = RelationGetDescr(relation);

	attr = tupdesc->attrs[natt - 1];

	/* Skip dropped columns and system columns */
	if (attr->attisdropped || attr->attnum < 0)
		return;

	/* Skip comma for first colums */
	if (!*first_column)
		appendStringInfoString(s, " AND ");
	else
		*first_column = false;

	/* Print attribute name */
	appendStringInfo(s, "%s = ", quote_identifier(NameStr(attr->attname)));

	/* Get Datum from tuple */
	origval = heap_getattr(tuple, natt, tupdesc, &isnull);

	/* Get output function */
	print_value(s, origval, attr->atttypid, isnull);
}

void InstallHelper_groundwork()
{
	Invocation ctx;
	Invocation_pushInvocation(&ctx, false);
	ctx.function = Function_INIT_WRITER;
	PG_TRY();
	{
		char const *lpt = LOADPATH_TBL_NAME;
		char const *lptq = quote_identifier(lpt);
		jstring pljlp = String_createJavaStringFromNTS(pljavaLoadPath);
		jstring jlpt = String_createJavaStringFromNTS(lpt);
		jstring jlptq = String_createJavaStringFromNTS(lptq);
		if ( lptq != lpt )
			pfree((void *)lptq);
		JNI_callStaticVoidMethod(
			s_InstallHelper_class, s_InstallHelper_groundwork,
			pljlp, jlpt, jlptq,
			pljavaLoadingAsExtension ? JNI_TRUE : JNI_FALSE,
			extensionExNihilo ? JNI_TRUE : JNI_FALSE);
		JNI_deleteLocalRef(pljlp);
		JNI_deleteLocalRef(jlpt);
		JNI_deleteLocalRef(jlptq);
		Invocation_popInvocation(false);
	}
	PG_CATCH();
	{
		Invocation_popInvocation(true);
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * AppendOptionListToString converts the option list to its textual format, and
 * appends this text to the given string buffer.
 */
void
AppendOptionListToString(StringInfo stringBuffer, List *optionList)
{
	if (optionList != NIL)
	{
		ListCell *optionCell = NULL;
		bool firstOptionPrinted = false;

		appendStringInfo(stringBuffer, " OPTIONS (");

		foreach(optionCell, optionList)
		{
			DefElem *option = (DefElem *) lfirst(optionCell);
			char *optionName = option->defname;
			char *optionValue = defGetString(option);

			if (firstOptionPrinted)
			{
				appendStringInfo(stringBuffer, ", ");
			}
			firstOptionPrinted = true;

			appendStringInfo(stringBuffer, "%s ", quote_identifier(optionName));
			appendStringInfo(stringBuffer, "%s", quote_literal_cstr(optionValue));
		}

/*
 * format_operator		- converts operator OID to "opr_name(args)"
 *
 * This exports the useful functionality of regoperatorout for use
 * in other backend modules.  The result is a palloc'd string.
 */
char *
format_operator(Oid operator_oid)
{
	char	   *result;
	HeapTuple	opertup;

	opertup = SearchSysCache1(OPEROID, ObjectIdGetDatum(operator_oid));

	if (HeapTupleIsValid(opertup))
	{
		Form_pg_operator operform = (Form_pg_operator) GETSTRUCT(opertup);
		char	   *oprname = NameStr(operform->oprname);
		char	   *nspname;
		StringInfoData buf;

		/* XXX no support here for bootstrap mode */

		initStringInfo(&buf);

		/*
		 * Would this oper be found (given the right args) by regoperatorin?
		 * If not, we need to qualify it.
		 */
		if (!OperatorIsVisible(operator_oid))
		{
			nspname = get_namespace_name(operform->oprnamespace);
			appendStringInfo(&buf, "%s.",
							 quote_identifier(nspname));
		}

		appendStringInfo(&buf, "%s(", oprname);

		if (operform->oprleft)
			appendStringInfo(&buf, "%s,",
							 format_type_be(operform->oprleft));
		else
			appendStringInfo(&buf, "NONE,");

		if (operform->oprright)
			appendStringInfo(&buf, "%s)",
							 format_type_be(operform->oprright));
		else
			appendStringInfo(&buf, "NONE)");

		result = buf.data;

		ReleaseSysCache(opertup);
	}
	else
	{
		/*
		 * If OID doesn't match any pg_operator entry, return it numerically
		 */
		result = (char *) palloc(NAMEDATALEN);
		snprintf(result, NAMEDATALEN, "%u", operator_oid);
	}

	return result;
}

/*
 *	split_old_dump
 *
 *	This function splits pg_dumpall output into global values and
 *	database creation, and per-db schemas.	This allows us to create
 *	the support functions between restoring these two parts of the
 *	dump.  We split on the first "\connect " after a CREATE ROLE
 *	username match;  this is where the per-db restore starts.
 *
 *	We suppress recreation of our own username so we don't generate
 *	an error during restore
 */
void
split_old_dump(void)
{
	FILE	   *all_dump,
			   *globals_dump,
			   *db_dump;
	FILE	   *current_output;
	char		line[LINE_ALLOC];
	bool		start_of_line = true;
	char		create_role_str[MAX_STRING];
	char		create_role_str_quote[MAX_STRING];
	char		filename[MAXPGPATH];
	bool		suppressed_username = false;

	snprintf(filename, sizeof(filename), "%s", ALL_DUMP_FILE);
	if ((all_dump = fopen(filename, "r")) == NULL)
		pg_log(PG_FATAL, "Could not open dump file \"%s\": %s\n", filename, getErrorText(errno));
	snprintf(filename, sizeof(filename), "%s", GLOBALS_DUMP_FILE);
	if ((globals_dump = fopen_priv(filename, "w")) == NULL)
		pg_log(PG_FATAL, "Could not write to dump file \"%s\": %s\n", filename, getErrorText(errno));
	snprintf(filename, sizeof(filename), "%s", DB_DUMP_FILE);
	if ((db_dump = fopen_priv(filename, "w")) == NULL)
		pg_log(PG_FATAL, "Could not write to dump file \"%s\": %s\n", filename, getErrorText(errno));

	current_output = globals_dump;

	/* patterns used to prevent our own username from being recreated */
	snprintf(create_role_str, sizeof(create_role_str),
			 "CREATE ROLE %s;", os_info.user);
	snprintf(create_role_str_quote, sizeof(create_role_str_quote),
			 "CREATE ROLE %s;", quote_identifier(os_info.user));

	while (fgets(line, sizeof(line), all_dump) != NULL)
	{
		/* switch to db_dump file output? */
		if (current_output == globals_dump && start_of_line &&
			suppressed_username &&
			strncmp(line, "\\connect ", strlen("\\connect ")) == 0)
			current_output = db_dump;

		/* output unless we are recreating our own username */
		if (current_output != globals_dump || !start_of_line ||
			(strncmp(line, create_role_str, strlen(create_role_str)) != 0 &&
			 strncmp(line, create_role_str_quote, strlen(create_role_str_quote)) != 0))
			fputs(line, current_output);
		else
			suppressed_username = true;

		if (strlen(line) > 0 && line[strlen(line) - 1] == '\n')
			start_of_line = true;
		else
			start_of_line = false;
	}

	fclose(all_dump);
	fclose(globals_dump);
	fclose(db_dump);
}

/*
 * Decode an INSERT entry
 */
static void
decoder_raw_insert(StringInfo s,
				   Relation relation,
				   HeapTuple tuple)
{
	TupleDesc		tupdesc = RelationGetDescr(relation);
	int				natt;
	bool			first_column = true;
	StringInfo		values = makeStringInfo();

	/* Initialize string info for values */
	initStringInfo(values);

	/* Query header */
	appendStringInfo(s, "INSERT INTO ");
	print_relname(s, relation);
	appendStringInfo(s, " (");

	/* Build column names and values */
	for (natt = 0; natt < tupdesc->natts; natt++)
	{
		Form_pg_attribute	attr;
		Datum				origval;
		bool				isnull;

		attr = tupdesc->attrs[natt];

		/* Skip dropped columns and system columns */
		if (attr->attisdropped || attr->attnum < 0)
			continue;

		/* Skip comma for first colums */
		if (!first_column)
		{
			appendStringInfoString(s, ", ");
			appendStringInfoString(values, ", ");
		}
		else
			first_column = false;

		/* Print attribute name */
		appendStringInfo(s, "%s", quote_identifier(NameStr(attr->attname)));

		/* Get Datum from tuple */
		origval = heap_getattr(tuple, natt + 1, tupdesc, &isnull);

		/* Get output function */
		print_value(values, origval, attr->atttypid, isnull);
	}

	/* Append values  */
	appendStringInfo(s, ") VALUES (%s);", values->data);

	/* Clean up */
	resetStringInfo(values);
}

/*
 * quote_ident -
 *	  returns a properly quoted identifier
 */
Datum
quote_ident(PG_FUNCTION_ARGS)
{
	text	   *t = PG_GETARG_TEXT_PP(0);
	const char *qstr;
	char	   *str;

	str = text_to_cstring(t);
	qstr = quote_identifier(str);
	PG_RETURN_TEXT_P(cstring_to_text(qstr));
}

/*
 * Decode an UPDATE entry
 */
static void
decoder_raw_update(StringInfo s,
				   Relation relation,
				   HeapTuple oldtuple,
				   HeapTuple newtuple)
{
	TupleDesc		tupdesc = RelationGetDescr(relation);
	int				natt;
	bool			first_column = true;

	/* If there are no new values, simply leave as there is nothing to do */
	if (newtuple == NULL)
		return;

	appendStringInfo(s, "UPDATE ");
	print_relname(s, relation);

	/* Build the SET clause with the new values */
	appendStringInfo(s, " SET ");
	for (natt = 0; natt < tupdesc->natts; natt++)
	{
		Form_pg_attribute	attr;
		Datum				origval;
		bool				isnull;

		attr = tupdesc->attrs[natt];

		/* Skip dropped columns and system columns */
		if (attr->attisdropped || attr->attnum < 0)
			continue;

		/* Skip comma for first colums */
		if (!first_column)
		{
			appendStringInfoString(s, ", ");
		}
		else
			first_column = false;

		/* Print attribute name */
		appendStringInfo(s, "%s = ", quote_identifier(NameStr(attr->attname)));

		/* Get Datum from tuple */
		origval = heap_getattr(newtuple, natt + 1, tupdesc, &isnull);

		/* Get output function */
		print_value(s, origval, attr->atttypid, isnull);
	}

	/* Print WHERE clause */
	print_where_clause(s, relation, oldtuple, newtuple);

	appendStringInfoString(s, ";");
}

/*
 * quote_object_names
 *
 * It tries to quote the supplied identifiers
 */
static char *
quote_object_name(const char *src1, const char *src2,
				  const char *src3, const char *src4)
{
	StringInfoData	result;
	const char	   *temp;

	initStringInfo(&result);

	if (src1)
	{
		temp = quote_identifier(src1);
		appendStringInfo(&result, "%s", temp);
		if (src1 != temp)
			pfree((void *)temp);
	}
	if (src2)
	{
		temp = quote_identifier(src2);
		appendStringInfo(&result, ".%s", temp);
		if (src2 != temp)
			pfree((void *)temp);
	}
	if (src3)
	{
		temp = quote_identifier(src3);
		appendStringInfo(&result, ".%s", temp);
		if (src3 != temp)
			pfree((void *)temp);
	}
	if (src4)
	{
		temp = quote_identifier(src4);
		appendStringInfo(&result, ".%s", temp);
		if (src4 != temp)
			pfree((void *)temp);
	}
	return result.data;
}

/* attach worker to the shared memory segment, read the job structure */
static void
initialize_worker(uint32 segment)
{
	dsm_segment   *seg;
	ResourceOwner old,
				  tmp;
	/* Connect to dynamic shared memory segment.
	 *
	 * In order to attach a dynamic shared memory segment, we need a
	 * resource owner. We cannot to StartTransactionCommand here, since
	 * we haven't yet attached to the database: to do this, we need to
	 * fetch information about connection properties from the shared
	 * memory segment.
	 */
	 old = CurrentResourceOwner;
	 CurrentResourceOwner = ResourceOwnerCreate(NULL, "Background Worker");
	 seg = dsm_attach(segment);
	 if (seg == NULL)
	 	ereport(ERROR,
	 			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
	 			 errmsg("unable to map dynamic shared memory segment")));
	 dsm_pin_mapping(seg);
	 tmp = CurrentResourceOwner;
	 CurrentResourceOwner = old;
	 ResourceOwnerDelete(tmp);

	 job = palloc(sizeof(JobDesc));
	 /* copy the arguments from shared memory segment */
	 memcpy(job, dsm_segment_address(seg), sizeof(JobDesc));

	 /* and detach it right away */
	 dsm_detach(seg);
	 Assert(job->magic == JOB_MAGIC);

	 job_run_function.schema = quote_identifier(job->schemaname);
	 job_run_function.name = quote_identifier("run_job");
}

static PyObject *
PLy_quote_ident(PyObject *self, PyObject *args)
{
	const char *str;
	const char *quoted;
	PyObject   *ret;

	if (!PyArg_ParseTuple(args, "s:quote_ident", &str))
		return NULL;

	quoted = quote_identifier(str);
	ret = PyString_FromString(quoted);

	return ret;
}