pg_backup_archiver.c

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

/*-------------------------------------------------------------------------
 *
 * pg_backup_archiver.c
 *
 *	Private implementation of the archiver routines.
 *
 *	See the headers to pg_restore for more details.
 *
 * Copyright (c) 2000, Philip Warner
 *	Rights are granted to use this software in any way so long
 *	as this notice is not removed.
 *
 *	The author is not responsible for loss or damages that may
 *	result from its use.
 *
 *
 * IDENTIFICATION
 *		$PostgreSQL: pgsql/src/bin/pg_dump/pg_backup_archiver.c,v 1.117.2.3 2006/04/12 22:19:01 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */

#include "pg_backup.h"
#include "pg_dump.h"
#include "pg_backup_archiver.h"
#include "pg_backup_db.h"
#include "dumputils.h"

#include <ctype.h>
#include <unistd.h>

#ifdef WIN32
#include <io.h>
#endif

#include "pqexpbuffer.h"
#include "libpq/libpq-fs.h"


const char *progname;

static char *modulename = gettext_noop("archiver");


static ArchiveHandle *_allocAH(const char *FileSpec, const ArchiveFormat fmt,
		 const int compression, ArchiveMode mode);
static void _getObjectDescription(PQExpBuffer buf, TocEntry *te,
					  ArchiveHandle *AH);
static void _printTocEntry(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt, bool isData, bool acl_pass);


static void _doSetFixedOutputState(ArchiveHandle *AH);
static void _doSetSessionAuth(ArchiveHandle *AH, const char *user);
static void _doSetWithOids(ArchiveHandle *AH, const bool withOids);
static void _reconnectToDB(ArchiveHandle *AH, const char *dbname);
static void _becomeUser(ArchiveHandle *AH, const char *user);
static void _becomeOwner(ArchiveHandle *AH, TocEntry *te);
static void _selectOutputSchema(ArchiveHandle *AH, const char *schemaName);
static void _selectTablespace(ArchiveHandle *AH, const char *tablespace);

static teReqs _tocEntryRequired(TocEntry *te, RestoreOptions *ropt, bool include_acls);
static void _disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
static void _enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt);
static TocEntry *getTocEntryByDumpId(ArchiveHandle *AH, DumpId id);
static void _moveAfter(ArchiveHandle *AH, TocEntry *pos, TocEntry *te);
static int	_discoverArchiveFormat(ArchiveHandle *AH);

static void dump_lo_buf(ArchiveHandle *AH);
static void _write_msg(const char *modulename, const char *fmt, va_list ap);
static void _die_horribly(ArchiveHandle *AH, const char *modulename, const char *fmt, va_list ap);

static void dumpTimestamp(ArchiveHandle *AH, const char *msg, time_t tim);


/*
 *	Wrapper functions.
 *
 *	The objective it to make writing new formats and dumpers as simple
 *	as possible, if necessary at the expense of extra function calls etc.
 *
 */


/* Create a new archive */
/* Public */
Archive *
CreateArchive(const char *FileSpec, const ArchiveFormat fmt,
			  const int compression)

{
	ArchiveHandle *AH = _allocAH(FileSpec, fmt, compression, archModeWrite);

	return (Archive *) AH;
}

/* Open an existing archive */
/* Public */
Archive *
OpenArchive(const char *FileSpec, const ArchiveFormat fmt)
{
	ArchiveHandle *AH = _allocAH(FileSpec, fmt, 0, archModeRead);

	return (Archive *) AH;
}

/* Public */
void
CloseArchive(Archive *AHX)
{
	int			res = 0;
	ArchiveHandle *AH = (ArchiveHandle *) AHX;

	(*AH->ClosePtr) (AH);

	/* Close the output */
	if (AH->gzOut)
		res = GZCLOSE(AH->OF);
	else if (AH->OF != stdout)
		res = fclose(AH->OF);

	if (res != 0)
		die_horribly(AH, modulename, "could not close output archive file\n");
}

/* Public */
void
RestoreArchive(Archive *AHX, RestoreOptions *ropt)
{
	ArchiveHandle *AH = (ArchiveHandle *) AHX;
	TocEntry   *te;
	teReqs		reqs;
	OutputContext sav;
	bool		defnDumped;

	AH->ropt = ropt;
	AH->stage = STAGE_INITIALIZING;

	/*
	 * Check for nonsensical option combinations.
	 *
	 * NB: create+dropSchema is useless because if you're creating the DB,
	 * there's no need to drop individual items in it.  Moreover, if we tried
	 * to do that then we'd issue the drops in the database initially
	 * connected to, not the one we will create, which is very bad...
	 */
	if (ropt->create && ropt->dropSchema)
		die_horribly(AH, modulename, "-C and -c are incompatible options\n");

	/*
	 * If we're using a DB connection, then connect it.
	 */
	if (ropt->useDB)
	{
		ahlog(AH, 1, "connecting to database for restore\n");
		if (AH->version < K_VERS_1_3)
			die_horribly(AH, modulename, "direct database connections are not supported in pre-1.3 archives\n");

		/* XXX Should get this from the archive */
		AHX->minRemoteVersion = 070100;
		AHX->maxRemoteVersion = 999999;

		ConnectDatabase(AHX, ropt->dbname,
						ropt->pghost, ropt->pgport, ropt->username,
						ropt->requirePassword, ropt->ignoreVersion);

		/*
		 * If we're talking to the DB directly, don't send comments since they
		 * obscure SQL when displaying errors
		 */
		AH->noTocComments = 1;
	}

	/*
	 * Work out if we have an implied data-only restore. This can happen if
	 * the dump was data only or if the user has used a toc list to exclude
	 * all of the schema data. All we do is look for schema entries - if none
	 * are found then we set the dataOnly flag.
	 *
	 * We could scan for wanted TABLE entries, but that is not the same as
	 * dataOnly. At this stage, it seems unnecessary (6-Mar-2001).
	 */
	if (!ropt->dataOnly)
	{
		int			impliedDataOnly = 1;

		for (te = AH->toc->next; te != AH->toc; te = te->next)
		{
			reqs = _tocEntryRequired(te, ropt, true);
			if ((reqs & REQ_SCHEMA) != 0)
			{					/* It's schema, and it's wanted */
				impliedDataOnly = 0;
				break;
			}
		}
		if (impliedDataOnly)
		{
			ropt->dataOnly = impliedDataOnly;
			ahlog(AH, 1, "implied data-only restore\n");
		}
	}

	/*
	 * Setup the output file if necessary.
	 */
	if (ropt->filename || ropt->compression)
		sav = SetOutput(AH, ropt->filename, ropt->compression);

	ahprintf(AH, "--\n-- PostgreSQL database dump\n--\n\n");

	if (AH->public.verbose)
		dumpTimestamp(AH, "Started on", AH->createDate);

	/*
	 * Establish important parameter values right away.
	 */
	_doSetFixedOutputState(AH);

	AH->stage = STAGE_PROCESSING;

	/*
	 * Drop the items at the start, in reverse order
	 */
	if (ropt->dropSchema)
	{
		for (te = AH->toc->prev; te != AH->toc; te = te->prev)
		{
			AH->currentTE = te;

			reqs = _tocEntryRequired(te, ropt, false /* needn't drop ACLs */ );
			if (((reqs & REQ_SCHEMA) != 0) && te->dropStmt)
			{
				/* We want the schema */
				ahlog(AH, 1, "dropping %s %s\n", te->desc, te->tag);
				/* Select owner and schema as necessary */
				_becomeOwner(AH, te);
				_selectOutputSchema(AH, te->namespace);
				/* Drop it */
				ahprintf(AH, "%s", te->dropStmt);
			}
		}
	}

	/*
	 * Now process each non-ACL TOC entry
	 */
	for (te = AH->toc->next; te != AH->toc; te = te->next)
	{
		AH->currentTE = te;

		/* Work out what, if anything, we want from this entry */
		reqs = _tocEntryRequired(te, ropt, false);

		/* Dump any relevant dump warnings to stderr */
		if (!ropt->suppressDumpWarnings && strcmp(te->desc, "WARNING") == 0)
		{
			if (!ropt->dataOnly && te->defn != NULL && strlen(te->defn) != 0)
				write_msg(modulename, "warning from original dump file: %s\n", te->defn);
			else if (te->copyStmt != NULL && strlen(te->copyStmt) != 0)
				write_msg(modulename, "warning from original dump file: %s\n", te->copyStmt);
		}

		defnDumped = false;

		if ((reqs & REQ_SCHEMA) != 0)	/* We want the schema */
		{
			ahlog(AH, 1, "creating %s %s\n", te->desc, te->tag);

			_printTocEntry(AH, te, ropt, false, false);
			defnDumped = true;

			/* If we created a DB, connect to it... */
			if (strcmp(te->desc, "DATABASE") == 0)
			{
				ahlog(AH, 1, "connecting to new database \"%s\"\n", te->tag);
				_reconnectToDB(AH, te->tag);
			}
		}

		/*
		 * If we have a data component, then process it
		 */
		if ((reqs & REQ_DATA) != 0)
		{
			/*
			 * hadDumper will be set if there is genuine data component for
			 * this node. Otherwise, we need to check the defn field for
			 * statements that need to be executed in data-only restores.
			 */
			if (te->hadDumper)
			{
				/*
				 * If we can output the data, then restore it.
				 */
				if (AH->PrintTocDataPtr !=NULL && (reqs & REQ_DATA) != 0)
				{
#ifndef HAVE_LIBZ
					if (AH->compression != 0)
						die_horribly(AH, modulename, "cannot restore from compressed archive (not configured for compression support)\n");
#endif

					_printTocEntry(AH, te, ropt, true, false);

					if (strcmp(te->desc, "BLOBS") == 0)
					{
						ahlog(AH, 1, "restoring large object data\n");

						_selectOutputSchema(AH, "pg_catalog");

						(*AH->PrintTocDataPtr) (AH, te, ropt);
					}
					else
					{
						_disableTriggersIfNecessary(AH, te, ropt);

						/* Select owner and schema as necessary */
						_becomeOwner(AH, te);
						_selectOutputSchema(AH, te->namespace);

						ahlog(AH, 1, "restoring data for table \"%s\"\n",
							  te->tag);

						/*
						 * If we have a copy statement, use it. As of V1.3,
						 * these are separate to allow easy import from
						 * withing a database connection. Pre 1.3 archives can
						 * not use DB connections and are sent to output only.
						 *
						 * For V1.3+, the table data MUST have a copy
						 * statement so that we can go into appropriate mode
						 * with libpq.
						 */
						if (te->copyStmt && strlen(te->copyStmt) > 0)
						{
							ahprintf(AH, "%s", te->copyStmt);
							AH->writingCopyData = true;
						}

						(*AH->PrintTocDataPtr) (AH, te, ropt);

						AH->writingCopyData = false;

						_enableTriggersIfNecessary(AH, te, ropt);
					}
				}
			}
			else if (!defnDumped)
			{
				/* If we haven't already dumped the defn part, do so now */
				ahlog(AH, 1, "executing %s %s\n", te->desc, te->tag);
				_printTocEntry(AH, te, ropt, false, false);
			}
		}
	}							/* end loop over TOC entries */

	/*
	 * Scan TOC again to output ownership commands and ACLs
	 */
	for (te = AH->toc->next; te != AH->toc; te = te->next)
	{
		AH->currentTE = te;

		/* Work out what, if anything, we want from this entry */
		reqs = _tocEntryRequired(te, ropt, true);

		if ((reqs & REQ_SCHEMA) != 0)	/* We want the schema */
		{
			ahlog(AH, 1, "setting owner and privileges for %s %s\n",
				  te->desc, te->tag);
			_printTocEntry(AH, te, ropt, false, true);
		}
	}

	if (AH->public.verbose)
		dumpTimestamp(AH, "Completed on", time(NULL));

	ahprintf(AH, "--\n-- PostgreSQL database dump complete\n--\n\n");

	/*
	 * Clean up & we're done.
	 */
	AH->stage = STAGE_FINALIZING;

	if (ropt->filename || ropt->compression)
		ResetOutput(AH, sav);

	if (ropt->useDB)
	{
		PQfinish(AH->connection);
		AH->connection = NULL;
	}
}

/*
 * Allocate a new RestoreOptions block.
 * This is mainly so we can initialize it, but also for future expansion,
 */
RestoreOptions *
NewRestoreOptions(void)
{
	RestoreOptions *opts;

	opts = (RestoreOptions *) calloc(1, sizeof(RestoreOptions));

	opts->format = archUnknown;
	opts->suppressDumpWarnings = false;
	opts->exit_on_error = false;

	return opts;
}

static void
_disableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt)
{
	/* This hack is only needed in a data-only restore */
	if (!ropt->dataOnly || !ropt->disable_triggers)
		return;

	ahlog(AH, 1, "disabling triggers for %s\n", te->tag);

	/*
	 * Become superuser if possible, since they are the only ones who can
	 * disable constraint triggers.  If -S was not given, assume the initial
	 * user identity is a superuser.  (XXX would it be better to become the
	 * table owner?)
	 */
	_becomeUser(AH, ropt->superuser);

	/*
	 * Disable them.
	 */
	_selectOutputSchema(AH, te->namespace);

	ahprintf(AH, "ALTER TABLE %s DISABLE TRIGGER ALL;\n\n",
			 fmtId(te->tag));
}

static void
_enableTriggersIfNecessary(ArchiveHandle *AH, TocEntry *te, RestoreOptions *ropt)
{
	/* This hack is only needed in a data-only restore */
	if (!ropt->dataOnly || !ropt->disable_triggers)
		return;

	ahlog(AH, 1, "enabling triggers for %s\n", te->tag);

	/*
	 * Become superuser if possible, since they are the only ones who can
	 * disable constraint triggers.  If -S was not given, assume the initial
	 * user identity is a superuser.  (XXX would it be better to become the
	 * table owner?)
	 */
	_becomeUser(AH, ropt->superuser);

	/*
	 * Enable them.
	 */
	_selectOutputSchema(AH, te->namespace);

	ahprintf(AH, "ALTER TABLE %s ENABLE TRIGGER ALL;\n\n",
			 fmtId(te->tag));
}

/*
 * This is a routine that is part of the dumper interface, hence the 'Archive*' parameter.
 */

/* Public */
size_t
WriteData(Archive *AHX, const void *data, size_t dLen)
{
	ArchiveHandle *AH = (ArchiveHandle *) AHX;

	if (!AH->currToc)
		die_horribly(AH, modulename, "internal error -- WriteData cannot be called outside the context of a DataDumper routine\n");

	return (*AH->WriteDataPtr) (AH, data, dLen);
}

/*
 * Create a new TOC entry. The TOC was designed as a TOC, but is now the
 * repository for all metadata. But the name has stuck.
 */

/* Public */
void
ArchiveEntry(Archive *AHX,
			 CatalogId catalogId, DumpId dumpId,
			 const char *tag,
			 const char *namespace,
			 const char *tablespace,
			 const char *owner, bool withOids,
			 const char *desc, const char *defn,
			 const char *dropStmt, const char *copyStmt,
			 const DumpId *deps, int nDeps,
			 DataDumperPtr dumpFn, void *dumpArg)
{
	ArchiveHandle *AH = (ArchiveHandle *) AHX;
	TocEntry   *newToc;

	newToc = (TocEntry *) calloc(1, sizeof(TocEntry));
	if (!newToc)
		die_horribly(AH, modulename, "out of memory\n");

	AH->tocCount++;
	if (dumpId > AH->maxDumpId)
		AH->maxDumpId = dumpId;

	newToc->prev = AH->toc->prev;
	newToc->next = AH->toc;
	AH->toc->prev->next = newToc;
	AH->toc->prev = newToc;

	newToc->catalogId = catalogId;
	newToc->dumpId = dumpId;

	newToc->tag = strdup(tag);
	newToc->namespace = namespace ? strdup(namespace) : NULL;
	newToc->tablespace = tablespace ? strdup(tablespace) : NULL;
	newToc->owner = strdup(owner);
	newToc->withOids = withOids;
	newToc->desc = strdup(desc);
	newToc->defn = strdup(defn);
	newToc->dropStmt = strdup(dropStmt);
	newToc->copyStmt = copyStmt ? strdup(copyStmt) : NULL;

	if (nDeps > 0)
	{
		newToc->dependencies = (DumpId *) malloc(nDeps * sizeof(DumpId));
		memcpy(newToc->dependencies, deps, nDeps * sizeof(DumpId));
		newToc->nDeps = nDeps;
	}
	else
	{
		newToc->dependencies = NULL;