pg_dump.c

PostgreSQL 8.1.4的源码 适用于Linux下的开源数据库系统

		/* Process the tuples, if any */
		for (i = 0; i < PQntuples(res); i++)
		{
			Oid			blobOid;
			int			loFd;

			blobOid = atooid(PQgetvalue(res, i, 0));
			/* Open the BLOB */
			loFd = lo_open(g_conn, blobOid, INV_READ);
			if (loFd == -1)
			{
				write_msg(NULL, "dumpBlobs(): could not open large object: %s",
						  PQerrorMessage(g_conn));
				exit_nicely();
			}

			StartBlob(AH, blobOid);

			/* Now read it in chunks, sending data to archive */
			do
			{
				cnt = lo_read(g_conn, loFd, buf, LOBBUFSIZE);
				if (cnt < 0)
				{
					write_msg(NULL, "dumpBlobs(): error reading large object: %s",
							  PQerrorMessage(g_conn));
					exit_nicely();
				}

				WriteData(AH, buf, cnt);
			} while (cnt > 0);

			lo_close(g_conn, loFd);

			EndBlob(AH, blobOid);
		}
	} while (PQntuples(res) > 0);

	PQclear(res);

	return 1;
}

/*
 * dumpBlobComments
 *	dump all blob comments
 *
 * Since we don't provide any way to be selective about dumping blobs,
 * there's no need to be selective about their comments either.  We put
 * all the comments into one big TOC entry.
 */
static int
dumpBlobComments(Archive *AH, void *arg)
{
	const char *blobQry;
	const char *blobFetchQry;
	PQExpBuffer commentcmd = createPQExpBuffer();
	PGresult   *res;
	int			i;

	if (g_verbose)
		write_msg(NULL, "saving large object comments\n");

	/* Make sure we are in proper schema */
	selectSourceSchema("pg_catalog");

	/* Cursor to get all BLOB comments */
	if (AH->remoteVersion >= 70200)
		blobQry = "DECLARE blobcmt CURSOR FOR SELECT DISTINCT loid, obj_description(loid, 'pg_largeobject') FROM pg_largeobject";
	else if (AH->remoteVersion >= 70100)
		blobQry = "DECLARE blobcmt CURSOR FOR SELECT DISTINCT loid, obj_description(loid) FROM pg_largeobject";
	else
		blobQry = "DECLARE blobcmt CURSOR FOR SELECT oid, (SELECT description FROM pg_description pd WHERE pd.objoid=pc.oid) FROM pg_class pc WHERE relkind = 'l'";

	res = PQexec(g_conn, blobQry);
	check_sql_result(res, g_conn, blobQry, PGRES_COMMAND_OK);

	/* Command to fetch from cursor */
	blobFetchQry = "FETCH 100 IN blobcmt";

	do
	{
		PQclear(res);

		/* Do a fetch */
		res = PQexec(g_conn, blobFetchQry);
		check_sql_result(res, g_conn, blobFetchQry, PGRES_TUPLES_OK);

		/* Process the tuples, if any */
		for (i = 0; i < PQntuples(res); i++)
		{
			Oid			blobOid;
			char	   *comment;

			/* ignore blobs without comments */
			if (PQgetisnull(res, i, 1))
				continue;

			blobOid = atooid(PQgetvalue(res, i, 0));
			comment = PQgetvalue(res, i, 1);

			printfPQExpBuffer(commentcmd, "COMMENT ON LARGE OBJECT %u IS ",
							  blobOid);
			appendStringLiteral(commentcmd, comment, false);
			appendPQExpBuffer(commentcmd, ";\n");

			archputs(commentcmd->data, AH);
		}
	} while (PQntuples(res) > 0);

	PQclear(res);

	archputs("\n", AH);

	destroyPQExpBuffer(commentcmd);

	return 1;
}

/*
 * getNamespaces:
 *	  read all namespaces in the system catalogs and return them in the
 * NamespaceInfo* structure
 *
 *	numNamespaces is set to the number of namespaces read in
 */
NamespaceInfo *
getNamespaces(int *numNamespaces)
{
	PGresult   *res;
	int			ntups;
	int			i;
	PQExpBuffer query;
	NamespaceInfo *nsinfo;
	int			i_tableoid;
	int			i_oid;
	int			i_nspname;
	int			i_rolname;
	int			i_nspacl;

	/*
	 * Before 7.3, there are no real namespaces; create two dummy entries, one
	 * for user stuff and one for system stuff.
	 */
	if (g_fout->remoteVersion < 70300)
	{
		nsinfo = (NamespaceInfo *) malloc(2 * sizeof(NamespaceInfo));

		nsinfo[0].dobj.objType = DO_NAMESPACE;
		nsinfo[0].dobj.catId.tableoid = 0;
		nsinfo[0].dobj.catId.oid = 0;
		AssignDumpId(&nsinfo[0].dobj);
		nsinfo[0].dobj.name = strdup("public");
		nsinfo[0].rolname = strdup("");
		nsinfo[0].nspacl = strdup("");

		selectDumpableNamespace(&nsinfo[0]);

		nsinfo[1].dobj.objType = DO_NAMESPACE;
		nsinfo[1].dobj.catId.tableoid = 0;
		nsinfo[1].dobj.catId.oid = 1;
		AssignDumpId(&nsinfo[1].dobj);
		nsinfo[1].dobj.name = strdup("pg_catalog");
		nsinfo[1].rolname = strdup("");
		nsinfo[1].nspacl = strdup("");

		selectDumpableNamespace(&nsinfo[1]);

		g_namespaces = nsinfo;
		g_numNamespaces = *numNamespaces = 2;

		return nsinfo;
	}

	query = createPQExpBuffer();

	/* Make sure we are in proper schema */
	selectSourceSchema("pg_catalog");

	/*
	 * we fetch all namespaces including system ones, so that every object we
	 * read in can be linked to a containing namespace.
	 */
	appendPQExpBuffer(query, "SELECT tableoid, oid, nspname, "
					  "(%s nspowner) as rolname, "
					  "nspacl FROM pg_namespace",
					  username_subquery);

	res = PQexec(g_conn, query->data);
	check_sql_result(res, g_conn, query->data, PGRES_TUPLES_OK);

	ntups = PQntuples(res);

	nsinfo = (NamespaceInfo *) malloc(ntups * sizeof(NamespaceInfo));

	i_tableoid = PQfnumber(res, "tableoid");
	i_oid = PQfnumber(res, "oid");
	i_nspname = PQfnumber(res, "nspname");
	i_rolname = PQfnumber(res, "rolname");
	i_nspacl = PQfnumber(res, "nspacl");

	for (i = 0; i < ntups; i++)
	{
		nsinfo[i].dobj.objType = DO_NAMESPACE;
		nsinfo[i].dobj.catId.tableoid = atooid(PQgetvalue(res, i, i_tableoid));
		nsinfo[i].dobj.catId.oid = atooid(PQgetvalue(res, i, i_oid));
		AssignDumpId(&nsinfo[i].dobj);
		nsinfo[i].dobj.name = strdup(PQgetvalue(res, i, i_nspname));
		nsinfo[i].rolname = strdup(PQgetvalue(res, i, i_rolname));
		nsinfo[i].nspacl = strdup(PQgetvalue(res, i, i_nspacl));

		/* Decide whether to dump this namespace */
		selectDumpableNamespace(&nsinfo[i]);

		if (strlen(nsinfo[i].rolname) == 0)
			write_msg(NULL, "WARNING: owner of schema \"%s\" appears to be invalid\n",
					  nsinfo[i].dobj.name);
	}

	/*
	 * If the user attempted to dump a specific namespace, check to ensure
	 * that the specified namespace actually exists.
	 */
	if (selectSchemaName)
	{
		for (i = 0; i < ntups; i++)
			if (strcmp(nsinfo[i].dobj.name, selectSchemaName) == 0)
				break;

		/* Didn't find a match */
		if (i == ntups)
		{
			write_msg(NULL, "specified schema \"%s\" does not exist\n",
					  selectSchemaName);
			exit_nicely();
		}
	}

	PQclear(res);
	destroyPQExpBuffer(query);

	g_namespaces = nsinfo;
	g_numNamespaces = *numNamespaces = ntups;

	return nsinfo;
}

/*
 * findNamespace:
 *		given a namespace OID and an object OID, look up the info read by
 *		getNamespaces
 *
 * NB: for pre-7.3 source database, we use object OID to guess whether it's
 * a system object or not.	In 7.3 and later there is no guessing.
 */
static NamespaceInfo *
findNamespace(Oid nsoid, Oid objoid)
{
	int			i;

	if (g_fout->remoteVersion >= 70300)
	{
		for (i = 0; i < g_numNamespaces; i++)
		{
			NamespaceInfo *nsinfo = &g_namespaces[i];

			if (nsoid == nsinfo->dobj.catId.oid)
				return nsinfo;
		}
		write_msg(NULL, "schema with OID %u does not exist\n", nsoid);
		exit_nicely();
	}
	else
	{
		/* This code depends on the layout set up by getNamespaces. */
		if (objoid > g_last_builtin_oid)
			i = 0;				/* user object */
		else
			i = 1;				/* system object */
		return &g_namespaces[i];
	}

	return NULL;				/* keep compiler quiet */
}

/*
 * getTypes:
 *	  read all types in the system catalogs and return them in the
 * TypeInfo* structure
 *
 *	numTypes is set to the number of types read in
 *
 * NB: this must run after getFuncs() because we assume we can do
 * findFuncByOid().
 */
TypeInfo *
getTypes(int *numTypes)
{
	PGresult   *res;
	int			ntups;
	int			i;
	PQExpBuffer query = createPQExpBuffer();
	TypeInfo   *tinfo;
	int			i_tableoid;
	int			i_oid;
	int			i_typname;
	int			i_typnamespace;
	int			i_rolname;
	int			i_typinput;
	int			i_typoutput;
	int			i_typelem;
	int			i_typrelid;
	int			i_typrelkind;
	int			i_typtype;
	int			i_typisdefined;

	/*
	 * we include even the built-in types because those may be used as array
	 * elements by user-defined types
	 *
	 * we filter out the built-in types when we dump out the types
	 *
	 * same approach for undefined (shell) types
	 */

	/* Make sure we are in proper schema */
	selectSourceSchema("pg_catalog");

	if (g_fout->remoteVersion >= 70300)
	{
		appendPQExpBuffer(query, "SELECT tableoid, oid, typname, "
						  "typnamespace, "
						  "(%s typowner) as rolname, "
						  "typinput::oid as typinput, "
						  "typoutput::oid as typoutput, typelem, typrelid, "
						  "CASE WHEN typrelid = 0 THEN ' '::\"char\" "
						  "ELSE (SELECT relkind FROM pg_class WHERE oid = typrelid) END as typrelkind, "
						  "typtype, typisdefined "
						  "FROM pg_type",
						  username_subquery);
	}
	else if (g_fout->remoteVersion >= 70100)
	{
		appendPQExpBuffer(query, "SELECT tableoid, oid, typname, "
						  "0::oid as typnamespace, "
						  "(%s typowner) as rolname, "
						  "typinput::oid as typinput, "
						  "typoutput::oid as typoutput, typelem, typrelid, "
						  "CASE WHEN typrelid = 0 THEN ' '::\"char\" "
						  "ELSE (SELECT relkind FROM pg_class WHERE oid = typrelid) END as typrelkind, "
						  "typtype, typisdefined "
						  "FROM pg_type",
						  username_subquery);
	}
	else
	{
		appendPQExpBuffer(query, "SELECT "
		 "(SELECT oid FROM pg_class WHERE relname = 'pg_type') AS tableoid, "
						  "oid, typname, "
						  "0::oid as typnamespace, "
						  "(%s typowner) as rolname, "
						  "typinput::oid as typinput, "
						  "typoutput::oid as typoutput, typelem, typrelid, "
						  "CASE WHEN typrelid = 0 THEN ' '::\"char\" "
						  "ELSE (SELECT relkind FROM pg_class WHERE oid = typrelid) END as typrelkind, "
						  "typtype, typisdefined "
						  "FROM pg_type",
						  username_subquery);
	}

	res = PQexec(g_conn, query->data);
	check_sql_result(res, g_conn, query->data, PGRES_TUPLES_OK);

	ntups = PQntuples(res);

	tinfo = (TypeInfo *) malloc(ntups * sizeof(TypeInfo));

	i_tableoid = PQfnumber(res, "tableoid");
	i_oid = PQfnumber(res, "oid");
	i_typname = PQfnumber(res, "typname");
	i_typnamespace = PQfnumber(res, "typnamespace");
	i_rolname = PQfnumber(res, "rolname");
	i_typinput = PQfnumber(res, "typinput");
	i_typoutput = PQfnumber(res, "typoutput");
	i_typelem = PQfnumber(res, "typelem");
	i_typrelid = PQfnumber(res, "typrelid");
	i_typrelkind = PQfnumber(res, "typrelkind");
	i_typtype = PQfnumber(res, "typtype");
	i_typisdefined = PQfnumber(res, "typisdefined");

	for (i = 0; i < ntups; i++)
	{
		Oid			typoutput;
		FuncInfo   *funcInfo;

		tinfo[i].dobj.objType = DO_TYPE;
		tinfo[i].dobj.catId.tableoid = atooid(PQgetvalue(res, i, i_tableoid));
		tinfo[i].dobj.catId.oid = atooid(PQgetvalue(res, i, i_oid));
		AssignDumpId(&tinfo[i].dobj);
		tinfo[i].dobj.name = strdup(PQgetvalue(res, i, i_typname));
		tinfo[i].dobj.namespace = findNamespace(atooid(PQgetvalue(res, i, i_typnamespace)),
												tinfo[i].dobj.catId.oid);
		tinfo[i].rolname = strdup(PQgetvalue(res, i, i_rolname));
		tinfo[i].typinput = atooid(PQgetvalue(res, i, i_typinput));
		typoutput = atooid(PQgetvalue(res, i, i_typoutput));
		tinfo[i].typelem = atooid(PQgetvalue(res, i, i_typelem));
		tinfo[i].typrelid = atooid(PQgetvalue(res, i, i_typrelid));
		tinfo[i].typrelkind = *PQgetvalue(res, i, i_typrelkind);
		tinfo[i].typtype = *PQgetvalue(res, i, i_typtype);

		/*
		 * If it's a table's rowtype, use special type code to facilitate
		 * sorting into the desired order.	(We don't want to consider it an
		 * ordinary type because that would bring the table up into the
		 * datatype part of the dump order.)
		 */
		if (OidIsValid(tinfo[i].typrelid) && tinfo[i].typrelkind != 'c')
			tinfo[i].dobj.objType = DO_TABLE_TYPE;

		/*
		 * check for user-defined array types, omit system generated ones
		 */
		if (OidIsValid(tinfo[i].typelem) &&
			tinfo[i].dobj.name[0] != '_')
			tinfo[i].isArray = true;
		else
			tinfo[i].isArray = false;

		if (strcmp(PQgetvalue(res, i, i_typisdefined), "t") == 0)
			tinfo[i].isDefined = true;
		else
			tinfo[i].isDefined = false;

		/*
		 * If it's a domain, fetch info about its constraints, if any
		 */
		tinfo[i].nDomChecks = 0;
		tinfo[i].domChecks = NULL;
		if (tinfo[i].typtype == 'd')
			getDomainConstraints(&(tinfo[i]));

		/*
		 * Make sure there are dependencies from the type to its input and
		 * output functions.  (We don't worry about typsend, typreceive, or
		 * typanalyze since those are only valid in 7.4 and later, wherein the
		 * standard dependency mechanism will pick them up.)
		 */
		funcInfo = findFuncByOid(tinfo[i].typinput);
		if (funcInfo)
			addObjectDependency(&tinfo[i].dobj,
								funcInfo->dobj.dumpId);
		funcInfo = findFuncByOid(typoutput);
		if (funcInfo)
			addObjectDependency(&tinfo[i].dobj,
								funcInfo->dobj.dumpId);

		if (strlen(tinfo[i].rolname) == 0 && tinfo[i].isDefined)
			write_msg(NULL, "WARNING: owner of data type \"%s\" appears to be invalid\n",
					  tinfo[i].dobj.name);
	}

	*numTypes = ntups;

	PQclear(res);

	destroyPQExpBuffer(query);

	return tinfo;
}