/*
 * Dump user-specific configuration
 */
static void
dumpUserConfig(PGconn *conn, const char *username)
{
	PQExpBuffer buf = createPQExpBuffer();
	int			count = 1;

	for (;;)
	{
		PGresult   *res;

		if (server_version >= 80100)
			printfPQExpBuffer(buf, "SELECT rolconfig[%d] FROM pg_authid WHERE rolname = ", count);
		else
			printfPQExpBuffer(buf, "SELECT useconfig[%d] FROM pg_shadow WHERE usename = ", count);
		appendStringLiteral(buf, username, true);

		res = executeQuery(conn, buf->data);
		if (PQntuples(res) == 1 &&
			!PQgetisnull(res, 0, 0))
		{
			makeAlterConfigCommand(PQgetvalue(res, 0, 0), "ROLE", username);
			PQclear(res);
			count++;
		}
		else
		{
			PQclear(res);
			break;
		}
	}

	destroyPQExpBuffer(buf);
}

/*
 * Run pg_dump on dbname.
 */
static int
runPgDump(const char *dbname)
{
	PQExpBuffer cmd = createPQExpBuffer();
	int			ret;

	appendPQExpBuffer(cmd, SYSTEMQUOTE "\"%s\" %s", pg_dump_bin,
					  pgdumpopts->data);

	/*
	 * If we have a filename, use the undocumented plain-append pg_dump
	 * format.
	 */
	if (filename)
		appendPQExpBuffer(cmd, " -Fa ");
	else
		appendPQExpBuffer(cmd, " -Fp ");

	doShellQuoting(cmd, dbname);

	appendPQExpBuffer(cmd, "%s", SYSTEMQUOTE);

	if (verbose)
		fprintf(stderr, _("%s: running \"%s\"\n"), progname, cmd->data);

	fflush(stdout);
	fflush(stderr);

	ret = system(cmd->data);

	destroyPQExpBuffer(cmd);

	return ret;
}

/*
 * Dump database-specific configuration
 */
static void
dumpDatabaseConfig(PGconn *conn, const char *dbname)
{
	PQExpBuffer buf = createPQExpBuffer();
	int			count = 1;

	for (;;)
	{
		PGresult   *res;

		printfPQExpBuffer(buf, "SELECT datconfig[%d] FROM pg_database WHERE datname = ", count);
		appendStringLiteral(buf, dbname, true);
		appendPQExpBuffer(buf, ";");

		res = executeQuery(conn, buf->data);
		if (!PQgetisnull(res, 0, 0))
		{
			makeAlterConfigCommand(PQgetvalue(res, 0, 0), "DATABASE", dbname);
			PQclear(res);
			count++;
		}
		else
		{
			PQclear(res);
			break;
		}
	}

	destroyPQExpBuffer(buf);
}

/*
 * Convert a string value to a dollar quoted literal and append it to
 * the given buffer. If the dqprefix parameter is not NULL then the
 * dollar quote delimiter will begin with that (after the opening $).
 *
 * No escaping is done at all on str, in compliance with the rules
 * for parsing dollar quoted strings.  Also, we need not worry about
 * encoding issues.
 */
void
appendStringLiteralDQ(PQExpBuffer buf, const char *str, const char *dqprefix)
{
	static const char suffixes[] = "_XXXXXXX";
	int			nextchar = 0;
	PQExpBuffer delimBuf = createPQExpBuffer();

	/* start with $ + dqprefix if not NULL */
	appendPQExpBufferChar(delimBuf, '$');
	if (dqprefix)
		appendPQExpBufferStr(delimBuf, dqprefix);

	/*
	 * Make sure we choose a delimiter which (without the trailing $) is not
	 * present in the string being quoted. We don't check with the trailing $
	 * because a string ending in $foo must not be quoted with $foo$.
	 */
	while (strstr(str, delimBuf->data) != NULL)
	{
		appendPQExpBufferChar(delimBuf, suffixes[nextchar++]);
		nextchar %= sizeof(suffixes) - 1;
	}

	/* add trailing $ */
	appendPQExpBufferChar(delimBuf, '$');

	/* quote it and we are all done */
	appendPQExpBufferStr(buf, delimBuf->data);
	appendPQExpBufferStr(buf, str);
	appendPQExpBufferStr(buf, delimBuf->data);

	destroyPQExpBuffer(delimBuf);
}

/*
 * Helper function for dumpXXXConfig().
 */
static void
makeAlterConfigCommand(const char *arrayitem, const char *type, const char *name)
{
	char	   *pos;
	char	   *mine;
	PQExpBuffer buf = createPQExpBuffer();

	mine = strdup(arrayitem);
	pos = strchr(mine, '=');
	if (pos == NULL)
		return;

	*pos = 0;
	appendPQExpBuffer(buf, "ALTER %s %s ", type, fmtId(name));
	appendPQExpBuffer(buf, "SET %s TO ", fmtId(mine));

	/*
	 * Some GUC variable names are 'LIST' type and hence must not be quoted.
	 */
	if (pg_strcasecmp(mine, "DateStyle") == 0
		|| pg_strcasecmp(mine, "search_path") == 0)
		appendPQExpBuffer(buf, "%s", pos + 1);
	else
		appendStringLiteral(buf, pos + 1, false);
	appendPQExpBuffer(buf, ";\n");

	printf("%s", buf->data);
	destroyPQExpBuffer(buf);
	free(mine);
}

/*
 * Dump group memberships from a pre-8.1 server.  It's annoying that we
 * can't share any useful amount of code with the post-8.1 case, but
 * the catalog representations are too different.
 *
 * Note: we expect dumpRoles already created all the roles, but there is
 * no membership yet.
 */
static void
dumpGroups(PGconn *conn)
{
	PQExpBuffer buf = createPQExpBuffer();
	PGresult   *res;
	int			i;

	res = executeQuery(conn,
					   "SELECT groname, grolist FROM pg_group ORDER BY 1");

	if (PQntuples(res) > 0)
		printf("--\n-- Role memberships\n--\n\n");

	for (i = 0; i < PQntuples(res); i++)
	{
		char	   *groname = PQgetvalue(res, i, 0);
		char	   *grolist = PQgetvalue(res, i, 1);
		PGresult   *res2;
		int			j;

		/*
		 * Array representation is {1,2,3} ... convert to (1,2,3)
		 */
		if (strlen(grolist) < 3)
			continue;

		grolist = strdup(grolist);
		grolist[0] = '(';
		grolist[strlen(grolist) - 1] = ')';
		printfPQExpBuffer(buf,
						  "SELECT usename FROM pg_shadow "
						  "WHERE usesysid IN %s ORDER BY 1",
						  grolist);
		free(grolist);

		res2 = executeQuery(conn, buf->data);

		for (j = 0; j < PQntuples(res2); j++)
		{
			char	   *usename = PQgetvalue(res2, j, 0);

			/*
			 * Don't try to grant a role to itself; can happen if old
			 * installation has identically named user and group.
			 */
			if (strcmp(groname, usename) == 0)
				continue;

			printf("GRANT %s", fmtId(groname));
			printf(" TO %s;\n", fmtId(usename));
		}

		PQclear(res2);
	}

	PQclear(res);
	destroyPQExpBuffer(buf);

	printf("\n\n");
}

/*
 * Dump user-and-database-specific configuration
 */
static void
dumpDbRoleConfig(PGconn *conn)
{
	PQExpBuffer buf = createPQExpBuffer();
	PGresult   *res;
	int			i;

	printfPQExpBuffer(buf, "SELECT rolname, datname, unnest(setconfig) "
					  "FROM pg_db_role_setting, pg_authid, pg_database "
		  "WHERE setrole = pg_authid.oid AND setdatabase = pg_database.oid");
	res = executeQuery(conn, buf->data);

	if (PQntuples(res) > 0)
	{
		fprintf(OPF, "--\n-- Per-Database Role Settings \n--\n\n");

		for (i = 0; i < PQntuples(res); i++)
		{
			makeAlterConfigCommand(conn, PQgetvalue(res, i, 2),
								   "ROLE", PQgetvalue(res, i, 0),
								   "DATABASE", PQgetvalue(res, i, 1));
		}

		fprintf(OPF, "\n\n");
	}

	PQclear(res);
	destroyPQExpBuffer(buf);
}

/*
 * This function takes a function description, e.g. "x" or "x(int)", and
 * issues a query on the given connection to retrieve the function's OID
 * using a cast to regproc or regprocedure (as appropriate). The result,
 * if there is one, is returned at *foid.  Note that we'll fail if the
 * function doesn't exist OR if there are multiple matching candidates
 * OR if there's something syntactically wrong with the function description;
 * unfortunately it can be hard to tell the difference.
 */
static bool
lookup_function_oid(PGconn *conn, const char *desc, Oid *foid)
{
	bool		result = true;
	PQExpBuffer query;
	PGresult   *res;

	query = createPQExpBuffer();
	printfPQExpBuffer(query, "SELECT ");
	appendStringLiteralConn(query, desc, conn);
	appendPQExpBuffer(query, "::pg_catalog.%s::pg_catalog.oid",
					  strchr(desc, '(') ? "regprocedure" : "regproc");

	res = PQexec(conn, query->data);
	if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) == 1)
		*foid = atooid(PQgetvalue(res, 0, 0));
	else
	{
		minimal_error_message(res);
		result = false;
	}

	PQclear(res);
	destroyPQExpBuffer(query);

	return result;
}

/*
 * Fetches the "CREATE OR REPLACE FUNCTION ..." command that describes the
 * function with the given OID.  If successful, the result is stored in buf.
 */
static bool
get_create_function_cmd(PGconn *conn, Oid oid, PQExpBuffer buf)
{
	bool		result = true;
	PQExpBuffer query;
	PGresult   *res;

	query = createPQExpBuffer();
	printfPQExpBuffer(query, "SELECT pg_catalog.pg_get_functiondef(%u)", oid);

	res = PQexec(conn, query->data);
	if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) == 1)
	{
		resetPQExpBuffer(buf);
		appendPQExpBufferStr(buf, PQgetvalue(res, 0, 0));
	}
	else
	{
		minimal_error_message(res);
		result = false;
	}

	PQclear(res);
	destroyPQExpBuffer(query);

	return result;
}

/*
 * Build ALTER DEFAULT PRIVILEGES command(s) for single pg_default_acl entry.
 *
 *	type: the object type (TABLES, FUNCTIONS, etc)
 *	nspname: schema name, or NULL for global default privileges
 *	acls: the ACL string fetched from the database
 *	owner: username of privileges owner (will be passed through fmtId)
 *	remoteVersion: version of database
 *
 * Returns TRUE if okay, FALSE if could not parse the acl string.
 * The resulting commands (if any) are appended to the contents of 'sql'.
 */
bool
buildDefaultACLCommands(const char *type, const char *nspname,
						const char *acls, const char *owner,
						int remoteVersion,
						PQExpBuffer sql)
{
	bool		result;
	PQExpBuffer prefix;

	prefix = createPQExpBuffer();

	/*
	 * We incorporate the target role directly into the command, rather than
	 * playing around with SET ROLE or anything like that.	This is so that a
	 * permissions error leads to nothing happening, rather than changing
	 * default privileges for the wrong user.
	 */
	appendPQExpBuffer(prefix, "ALTER DEFAULT PRIVILEGES FOR ROLE %s ",
					  fmtId(owner));
	if (nspname)
		appendPQExpBuffer(prefix, "IN SCHEMA %s ", fmtId(nspname));

	result = buildACLCommands("", NULL,
							  type, acls, owner,
							  prefix->data, remoteVersion,
							  sql);

	destroyPQExpBuffer(prefix);

	return result;
}

/*
 * DoCancelNotifyListen: This function executes a LISTEN or a NOTIFY command, with
 * name in the format N%s_%d_%d, where the %s is replaced by the CDBDumpKey,
 * and the 2 integers are the contentid and dbid.
 */
void
DoCancelNotifyListen(PGconn *pConn,
					 bool bListen,
					 const char *pszCDBDumpKey,
					 int role_id,
					 int db_id,
					 int target_db_id,
					 const char *pszSuffix)
{
	PGresult   *pRes;
	PQExpBuffer q = createPQExpBuffer();
	char	   *pszCmd = bListen ? "LISTEN" : "NOTIFY";

	appendPQExpBuffer(q, "%s N%s_%d_%d",
					  pszCmd, pszCDBDumpKey, role_id, db_id);

	/* this is valid only for restore operations */
	if (target_db_id != -1)
		appendPQExpBuffer(q, "_T%d", target_db_id);

	if (pszSuffix != NULL)
		appendPQExpBuffer(q, "_%s", pszSuffix);

	pRes = PQexec(pConn, q->data);
	if (pRes == NULL || PQresultStatus(pRes) != PGRES_COMMAND_OK)
	{
		mpp_err_msg_cache("%s command failed for for backup key %s, instid %d, segid %d failed : %s",
			   pszCmd, pszCDBDumpKey, role_id, db_id, PQerrorMessage(pConn));
	}

	PQclear(pRes);
	destroyPQExpBuffer(q);
}

/*
 * ReadBackendBackupFile: This function calls the backend function gp_read_backup_file
 * which reads the contents out of the appropriate file on the database server.
 * If the call succeeds/fails, it returns status code 0/-1, an appropriate error string
 * is inserted into the buffer of pszRtn.
 */
int
ReadBackendBackupFileError(PGconn *pConn, const char *pszBackupDirectory, const char *pszKey,
		BackupFileType fileType, const char *progName, PQExpBuffer pszRtn)
{
	char       *pszFileType;
	PQExpBuffer Qry;
	PGresult   *pRes;
	int status = 0;

	switch (fileType)
	{
		case BFT_BACKUP:
			pszFileType = "0";
			break;
		case BFT_BACKUP_STATUS:
			pszFileType = "1";
			break;
		case BFT_RESTORE_STATUS:
			pszFileType = "2";
			break;
		default:
			appendPQExpBuffer(pszRtn, "Unknown file type passed to ReadBackendBackupFile");
			mpp_err_msg("ERROR", progName, " %s: %d\n", pszRtn->data, fileType);
			return -1;
	}

	Qry = createPQExpBuffer();

	appendPQExpBuffer(Qry, "SELECT * FROM gp_read_backup_file('%s', '%s', %s)",
			StringNotNull(pszBackupDirectory, ""),
			StringNotNull(pszKey, ""),
			pszFileType);

	pRes = PQexec(pConn, Qry->data);
	if (!pRes || PQresultStatus(pRes) != PGRES_TUPLES_OK || PQntuples(pRes) == 0)
	{
		appendPQExpBuffer(pszRtn, "Error executing query %s : %s\n", Qry->data, PQerrorMessage(pConn));
		mpp_err_msg_cache("ERROR", progName, pszRtn->data);
		status = -1;
	}
	else
	{
		char *res = PQgetvalue(pRes, 0, 0);
		appendPQExpBufferStr(pszRtn, res);
		if (strstr(res, "ERROR:") || strstr(res, "[ERROR]"))
		{
			status = -1;
		}
	}

	PQclear(pRes);
	destroyPQExpBuffer(Qry);

	return status;
}

/*
 * Run pg_dump on dbname.
 */
static int
runPgDump(const char *dbname)
{
	PQExpBuffer cmd = createPQExpBuffer();
	const char *p;
	int			ret;

	/*
	 * Win32 has to use double-quotes for args, rather than single quotes.
	 * Strangely enough, this is the only place we pass a database name on the
	 * command line, except "postgres" which doesn't need quoting.
	 */
#ifndef WIN32
	appendPQExpBuffer(cmd, "%s\"%s\" %s -Fp '", SYSTEMQUOTE, pg_dump_bin,
#else
	appendPQExpBuffer(cmd, "%s\"%s\" %s -Fp \"", SYSTEMQUOTE, pg_dump_bin,
#endif
					  pgdumpopts->data);

	/* Shell quoting is not quite like SQL quoting, so can't use fmtId */
	for (p = dbname; *p; p++)
	{
#ifndef WIN32
		if (*p == '\'')
			appendPQExpBuffer(cmd, "'\"'\"'");
#else
		if (*p == '"')
			appendPQExpBuffer(cmd, "\\\"");
#endif
		else
			appendPQExpBufferChar(cmd, *p);
	}

#ifndef WIN32
	appendPQExpBufferChar(cmd, '\'');
#else
	appendPQExpBufferChar(cmd, '"');
#endif

	appendPQExpBuffer(cmd, "%s", SYSTEMQUOTE);

	if (verbose)
		fprintf(stderr, _("%s: running \"%s\"\n"), progname, cmd->data);

	fflush(stdout);
	fflush(stderr);

	ret = system(cmd->data);

	destroyPQExpBuffer(cmd);

	return ret;
}

/* 
 * Exit closing active database connections
 */
static void
exit_gracefuly(int status)
{
	destroyPQExpBuffer(dbQry);
	if(lastDbConn)
		PQfinish(lastDbConn);
	if(conn)
		PQfinish(conn);

	close(logFd);
	exit(status);
}

/*
 * Issue a query on a catalog table, and produce calls to a preassign support
 * function from the result set.
 *
 * The output is a string, containing SQL calls like:
 *
 * SELECT binary_upgrade.preassign_*_oid(<oid>, <other args);
 *
 * 'funcname' is the "preassign_*_oid" function to use.
 * 'sql' is the query to issue. The columns of the result set are passed as
 * arguments to the preassign-support function.
 *
 */
static void
dump_rows(migratorContext *ctx, PQExpBuffer buf, FILE *file, PGconn *conn,
		  const char *sql, const char *funcname)
{
	int			ntups;
	int			ncols;
	int			row;
	int			col;
	PGresult   *res;

	if (file != NULL)
		buf = createPQExpBuffer();

	/*
	 * Add a WHERE or AND clause to filter out built-in objects.
	 *
	 * If the query contains "UNION ALL", then it's the caller's
	 * responsibility to do the filtering. This special case is for the
	 * one more complicated query in get_old_oids() function; all the
	 * other queries are very simple ones.
	 */
	if (strstr(sql, "WHERE ") == NULL)
		res = executeQueryOrDie(ctx, conn, "%s WHERE oid >= %u", sql, FirstNormalObjectId);
	else if (strstr(sql, "UNION ALL") == NULL)
		res = executeQueryOrDie(ctx, conn, "%s AND oid >= %u", sql, FirstNormalObjectId);
	else
		res = executeQueryOrDie(ctx, conn, "%s", sql);

	ntups = PQntuples(res);
	ncols = PQnfields(res);

	for (row = 0; row < ntups; row++)
	{
		appendPQExpBuffer(buf, "SELECT binary_upgrade.%s('%s'",
						  funcname,
						  simple_escape_literal(ctx, conn, PQgetvalue(res, row, 0)));

		for (col = 1; col < ncols; col++)
			appendPQExpBuffer(buf, ", '%s'",
							  simple_escape_literal(ctx, conn, PQgetvalue(res, row, col)));
		appendPQExpBuffer(buf, ");\n");

		if (file)
		{
			fwrite(buf->data, buf->len, 1, file);
			resetPQExpBuffer(buf);
		}
	}
	PQclear(res);

	if (file != NULL)
		destroyPQExpBuffer(buf);
}

/*
 * Destroy a CdbDispatchResult object.
 */
void
cdbdisp_termResult(CdbDispatchResult *dispatchResult)
{
	PQExpBuffer trash;

	dispatchResult->segdbDesc = NULL;

	/*
	 * Free the PGresult objects.
	 */
	cdbdisp_resetResult(dispatchResult);

	/*
	 * Free the error message buffer and result buffer.
	 */
	trash = dispatchResult->resultbuf;
	dispatchResult->resultbuf = NULL;
	destroyPQExpBuffer(trash);

	trash = dispatchResult->error_message;
	dispatchResult->error_message = NULL;
	destroyPQExpBuffer(trash);
}

/*
 * ReadBackendBackupFile: This function calls the backend function gp_read_backup_file
 * which reads the contents out of the appropriate file on the database server.
 * If the call fails, it returns NULL.	The returned pointer must be freed by the caller.
 */
char *
ReadBackendBackupFile(PGconn *pConn, const char *pszBackupDirectory, const char *pszKey, BackupFileType fileType, const char *progName)
{
	char	   *pszRtn = NULL;
	char	   *pszFileType;
	PQExpBuffer Qry;
	PGresult   *pRes;

	switch (fileType)
	{
		case BFT_BACKUP:
			pszFileType = "0";
			break;
		case BFT_BACKUP_STATUS:
			pszFileType = "1";
			break;
		case BFT_RESTORE_STATUS:
			pszFileType = "2";
			break;
		default:
			mpp_err_msg("ERROR", progName, "Unknown file type passed to ReadBackendBackupFile : %d\n", fileType);
			return NULL;
	}

	Qry = createPQExpBuffer();
	appendPQExpBuffer(Qry, "SELECT * FROM gp_read_backup_file('%s', '%s', %s)",
					  StringNotNull(pszBackupDirectory, ""),
					  StringNotNull(pszKey, ""),
					  pszFileType);

	pRes = PQexec(pConn, Qry->data);
	if (!pRes || PQresultStatus(pRes) != PGRES_TUPLES_OK || PQntuples(pRes) == 0)
	{
		mpp_err_msg_cache("ERROR", progName, "Error executing query %s : %s\n",
						  Qry->data,
						  PQerrorMessage(pConn));
	}
	else
	{
		pszRtn = strdup(PQgetvalue(pRes, 0, 0));
	}

	PQclear(pRes);
	destroyPQExpBuffer(Qry);

	return pszRtn;
}

/*
 * Called by dumper via archiver from within a data dump routine
 * We substitute this for _WriteData while emitting a BLOB
 */
static size_t
_WriteBlobData(ArchiveHandle *AH, const void *data, size_t dLen)
{
	if (dLen > 0)
	{
		PQExpBuffer buf = createPQExpBuffer();

		appendByteaLiteralAHX(buf,
							  (const unsigned char *) data,
							  dLen,
							  AH);

		ahprintf(AH, "SELECT pg_catalog.lowrite(0, %s);\n", buf->data);

		destroyPQExpBuffer(buf);
	}
	return dLen;
}

/*
 * Dump database-specific configuration
 */
static void
dumpDatabaseConfig(PGconn *conn, const char *dbname)
{
	PQExpBuffer buf = createPQExpBuffer();
	int			count = 1;

	for (;;)
	{
		PGresult   *res;

		if (server_version >= 90000)
			printfPQExpBuffer(buf, "SELECT setconfig[%d] FROM pg_db_role_setting WHERE "
							  "setrole = 0 AND setdatabase = (SELECT oid FROM pg_database WHERE datname = ", count);
		else
			printfPQExpBuffer(buf, "SELECT datconfig[%d] FROM pg_database WHERE datname = ", count);
		appendStringLiteralConn(buf, dbname, conn);

		if (server_version >= 90000)
			appendPQExpBuffer(buf, ")");

		appendPQExpBuffer(buf, ";");

		res = executeQuery(conn, buf->data);
		if (PQntuples(res) == 1 &&
			!PQgetisnull(res, 0, 0))
		{
			makeAlterConfigCommand(conn, PQgetvalue(res, 0, 0),
								   "DATABASE", dbname, NULL, NULL);
			PQclear(res);
			count++;
		}
		else
		{
			PQclear(res);
			break;
		}
	}

	destroyPQExpBuffer(buf);
}

/* Initiates the non-blocking capture of a consistent snapshot of the database,
 * using the exported snapshot context->repl.snapshot_name. */
int snapshot_start(client_context_t context) {
    if (!context->repl.snapshot_name || context->repl.snapshot_name[0] == '\0') {
        client_error(context, "snapshot_name must be set in client context");
        return EINVAL;
    }

    int err = 0;
    check(err, exec_sql(context, "BEGIN"));
    check(err, exec_sql(context, "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ"));

    PQExpBuffer query = createPQExpBuffer();
    appendPQExpBuffer(query, "SET TRANSACTION SNAPSHOT '%s'", context->repl.snapshot_name);
    check(err, exec_sql(context, query->data));
    destroyPQExpBuffer(query);

    Oid argtypes[] = { 25, 16 }; // 25 == TEXTOID, 16 == BOOLOID
    const char *args[] = { "%", context->allow_unkeyed ? "t" : "f" };

    if (!PQsendQueryParams(context->sql_conn,
                "SELECT bottledwater_export(table_pattern := $1, allow_unkeyed := $2)",
                2, argtypes, args, NULL, NULL, 1)) { // The final 1 requests results in binary format
        client_error(context, "Could not dispatch snapshot fetch: %s",
                PQerrorMessage(context->sql_conn));
        return EIO;
    }

    if (!PQsetSingleRowMode(context->sql_conn)) {
        client_error(context, "Could not activate single-row mode");
        return EIO;
    }

    // Invoke the begin-transaction callback with xid==0 to indicate start of snapshot
    begin_txn_cb begin_txn = context->repl.frame_reader->on_begin_txn;
    void *cb_context = context->repl.frame_reader->cb_context;
    if (begin_txn) {
        check(err, begin_txn(cb_context, context->repl.start_lsn, 0));
    }
    return 0;
}