relcache.c

关系型数据库 Postgresql 6.5.2

/*-------------------------------------------------------------------------
 *
 * relcache.c
 *	  POSTGRES relation descriptor cache code
 *
 * Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  $Header: /usr/local/cvsroot/pgsql/src/backend/utils/cache/relcache.c,v 1.62.2.2 1999/09/02 04:07:18 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
/*
 * INTERFACE ROUTINES
 *		RelationInitialize				- initialize relcache
 *		RelationIdCacheGetRelation		- get a reldesc from the cache (id)
 *		RelationNameCacheGetRelation	- get a reldesc from the cache (name)
 *		RelationIdGetRelation			- get a reldesc by relation id
 *		RelationNameGetRelation			- get a reldesc by relation name
 *		RelationClose					- close an open relation
 *		RelationFlushRelation			- flush relation information
 *
 * NOTES
 *		This file is in the process of being cleaned up
 *		before I add system attribute indexing.  -cim 1/13/91
 *
 *		The following code contains many undocumented hacks.  Please be
 *		careful....
 *
 */
#include <sys/types.h>
#include <errno.h>
#include <sys/file.h>
#include <fcntl.h>

#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/istrat.h"
#include "catalog/catalog.h"
#include "catalog/catname.h"
#include "catalog/index.h"
#include "catalog/indexing.h"
#include "catalog/pg_attrdef.h"
#include "catalog/pg_log.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_relcheck.h"
#include "catalog/pg_rewrite.h"
#include "catalog/pg_type.h"
#include "catalog/pg_variable.h"
#include "lib/hasht.h"
#include "miscadmin.h"
#include "storage/smgr.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
#include "utils/relcache.h"


static void RelationFlushRelation(Relation *relationPtr,
					  bool onlyFlushReferenceCountZero);
static Relation RelationNameCacheGetRelation(char *relationName);
static void init_irels(void);
static void write_irels(void);

/* ----------------
 *		externs
 * ----------------
 */
extern bool AMI_OVERRIDE;		/* XXX style */
extern GlobalMemory CacheCxt;	/* from utils/cache/catcache.c */

/* ----------------
 *		hardcoded tuple descriptors.  see lib/backend/catalog/pg_attribute.h
 * ----------------
 */
FormData_pg_attribute Desc_pg_class[Natts_pg_class] = {Schema_pg_class};
FormData_pg_attribute Desc_pg_attribute[Natts_pg_attribute] = {Schema_pg_attribute};
FormData_pg_attribute Desc_pg_proc[Natts_pg_proc] = {Schema_pg_proc};
FormData_pg_attribute Desc_pg_type[Natts_pg_type] = {Schema_pg_type};
FormData_pg_attribute Desc_pg_variable[Natts_pg_variable] = {Schema_pg_variable};
FormData_pg_attribute Desc_pg_log[Natts_pg_log] = {Schema_pg_log};

/* ----------------
 *		global variables
 *
 *		Relations are cached two ways, by name and by id,
 *		thus there are two hash tables for referencing them.
 * ----------------
 */
HTAB	   *RelationNameCache;
HTAB	   *RelationIdCache;

/* ----------------
 *		RelationBuildDescInfo exists so code can be shared
 *		between RelationIdGetRelation() and RelationNameGetRelation()
 * ----------------
 */
typedef struct RelationBuildDescInfo
{
	int			infotype;		/* lookup by id or by name */
#define INFO_RELID 1
#define INFO_RELNAME 2
	union
	{
		Oid			info_id;	/* relation object id */
		char	   *info_name;	/* relation name */
	}			i;
} RelationBuildDescInfo;

typedef struct relidcacheent
{
	Oid			reloid;
	Relation	reldesc;
} RelIdCacheEnt;

typedef struct relnamecacheent
{
	NameData	relname;
	Relation	reldesc;
} RelNameCacheEnt;

/* -----------------
 *		macros to manipulate name cache and id cache
 * -----------------
 */
#define RelationCacheInsert(RELATION)	\
do { \
	RelIdCacheEnt *idhentry; RelNameCacheEnt *namehentry; \
	char *relname; Oid reloid; bool found; \
	relname = (RELATION->rd_rel->relname).data; \
	namehentry = (RelNameCacheEnt*)hash_search(RelationNameCache, \
											   relname, \
											   HASH_ENTER, \
											   &found); \
	if (namehentry == NULL) \
		elog(FATAL, "can't insert into relation descriptor cache"); \
	if (found && !IsBootstrapProcessingMode()) \
		/* used to give notice -- now just keep quiet */ ; \
	namehentry->reldesc = RELATION; \
	reloid = RELATION->rd_id; \
	idhentry = (RelIdCacheEnt*)hash_search(RelationIdCache, \
										   (char *)&reloid, \
										   HASH_ENTER, \
										   &found); \
	if (idhentry == NULL) \
		elog(FATAL, "can't insert into relation descriptor cache"); \
	if (found && !IsBootstrapProcessingMode()) \
		/* used to give notice -- now just keep quiet */ ; \
	idhentry->reldesc = RELATION; \
} while(0)

#define RelationNameCacheLookup(NAME, RELATION) \
do { \
	RelNameCacheEnt *hentry; bool found; \
	hentry = (RelNameCacheEnt*)hash_search(RelationNameCache, \
										   (char *)NAME,HASH_FIND,&found); \
	if (hentry == NULL) \
		elog(FATAL, "error in CACHE"); \
	if (found) \
		RELATION = hentry->reldesc; \
	else \
		RELATION = NULL; \
} while(0)

#define RelationIdCacheLookup(ID, RELATION) \
do { \
	RelIdCacheEnt *hentry; \
	bool found; \
	hentry = (RelIdCacheEnt*)hash_search(RelationIdCache, \
										 (char *)&(ID),HASH_FIND, &found); \
	if (hentry == NULL) \
		elog(FATAL, "error in CACHE"); \
	if (found) \
		RELATION = hentry->reldesc; \
	else \
		RELATION = NULL; \
} while(0)

#define RelationCacheDelete(RELATION) \
do { \
	RelNameCacheEnt *namehentry; RelIdCacheEnt *idhentry; \
	char *relname; Oid reloid; bool found; \
	relname = (RELATION->rd_rel->relname).data; \
	namehentry = (RelNameCacheEnt*)hash_search(RelationNameCache, \
											   relname, \
											   HASH_REMOVE, \
											   &found); \
	if (namehentry == NULL) \
		elog(FATAL, "can't delete from relation descriptor cache"); \
	if (!found) \
		elog(NOTICE, "trying to delete a reldesc that does not exist."); \
	reloid = RELATION->rd_id; \
	idhentry = (RelIdCacheEnt*)hash_search(RelationIdCache, \
										   (char *)&reloid, \
										   HASH_REMOVE, &found); \
	if (idhentry == NULL) \
		elog(FATAL, "can't delete from relation descriptor cache"); \
	if (!found) \
		elog(NOTICE, "trying to delete a reldesc that does not exist."); \
} while(0)

/* non-export function prototypes */
static void formrdesc(char *relationName, u_int natts,
		  FormData_pg_attribute *att);

#ifdef NOT_USED					/* See comments at line 1304 */
static void RelationFlushIndexes(Relation *r, Oid accessMethodId);

#endif

static HeapTuple ScanPgRelation(RelationBuildDescInfo buildinfo);
static HeapTuple scan_pg_rel_seq(RelationBuildDescInfo buildinfo);
static HeapTuple scan_pg_rel_ind(RelationBuildDescInfo buildinfo);
static Relation AllocateRelationDesc(u_int natts, Form_pg_class relp);
static void RelationBuildTupleDesc(RelationBuildDescInfo buildinfo,
					   Relation relation, u_int natts);
static void build_tupdesc_seq(RelationBuildDescInfo buildinfo,
				  Relation relation, u_int natts);
static void build_tupdesc_ind(RelationBuildDescInfo buildinfo,
				  Relation relation, u_int natts);
static Relation RelationBuildDesc(RelationBuildDescInfo buildinfo);
static void IndexedAccessMethodInitialize(Relation relation);
static void AttrDefaultFetch(Relation relation);
static void RelCheckFetch(Relation relation);

extern void RelationBuildTriggers(Relation relation);
extern void FreeTriggerDesc(Relation relation);

/*
 * newlyCreatedRelns -
 *	  relations created during this transaction. We need to keep track of
 *	  these.
 */
static List *newlyCreatedRelns = NULL;

/* ----------------------------------------------------------------
 *		RelationIdGetRelation() and RelationNameGetRelation()
 *						support functions
 * ----------------------------------------------------------------
 */


#if NOT_USED					/* XXX This doesn't seem to be used
								 * anywhere */
/* --------------------------------
 *		BuildDescInfoError returns a string appropriate to
 *		the buildinfo passed to it
 * --------------------------------
 */
static char *
BuildDescInfoError(RelationBuildDescInfo buildinfo)
{
	static char errBuf[64];

	MemSet(errBuf, 0, (int) sizeof(errBuf));
	switch (buildinfo.infotype)
	{
		case INFO_RELID:
			sprintf(errBuf, "(relation id %u)", buildinfo.i.info_id);
			break;
		case INFO_RELNAME:
			sprintf(errBuf, "(relation name %s)", buildinfo.i.info_name);
			break;
	}

	return errBuf;
}

#endif

/* --------------------------------
 *		ScanPgRelation
 *
 *		this is used by RelationBuildDesc to find a pg_class
 *		tuple matching either a relation name or a relation id
 *		as specified in buildinfo.
 * --------------------------------
 */
static HeapTuple
ScanPgRelation(RelationBuildDescInfo buildinfo)
{

	/*
	 * If this is bootstrap time (initdb), then we can't use the system
	 * catalog indices, because they may not exist yet.  Otherwise, we
	 * can, and do.
	 */

	if (IsBootstrapProcessingMode())
		return scan_pg_rel_seq(buildinfo);
	else
		return scan_pg_rel_ind(buildinfo);
}

static HeapTuple
scan_pg_rel_seq(RelationBuildDescInfo buildinfo)
{
	HeapTuple	pg_class_tuple;
	HeapTuple	return_tuple;
	Relation	pg_class_desc;
	HeapScanDesc pg_class_scan;
	ScanKeyData key;

	/* ----------------
	 *	form a scan key
	 * ----------------
	 */
	switch (buildinfo.infotype)
	{
		case INFO_RELID:
			ScanKeyEntryInitialize(&key, 0,
								   ObjectIdAttributeNumber,
								   F_OIDEQ,
								   ObjectIdGetDatum(buildinfo.i.info_id));
			break;

		case INFO_RELNAME:
			ScanKeyEntryInitialize(&key, 0,
								   Anum_pg_class_relname,
								   F_NAMEEQ,
								   NameGetDatum(buildinfo.i.info_name));
			break;

		default:
			elog(ERROR, "ScanPgRelation: bad buildinfo");
			return NULL;
	}

	/* ----------------
	 *	open pg_class and fetch a tuple
	 * ----------------
	 */
	pg_class_desc = heap_openr(RelationRelationName);
	pg_class_scan = heap_beginscan(pg_class_desc, 0, SnapshotNow, 1, &key);
	pg_class_tuple = heap_getnext(pg_class_scan, 0);

	/* ----------------
	 *	get set to return tuple
	 * ----------------
	 */
	if (!HeapTupleIsValid(pg_class_tuple))
		return_tuple = pg_class_tuple;
	else
	{
		/* ------------------
		 *	a satanic bug used to live here: pg_class_tuple used to be
		 *	returned here without having the corresponding buffer pinned.
		 *	so when the buffer gets replaced, all hell breaks loose.
		 *	this bug is discovered and killed by wei on 9/27/91.
		 * -------------------
		 */
		return_tuple = heap_copytuple(pg_class_tuple);
	}

	/* all done */
	heap_endscan(pg_class_scan);
	heap_close(pg_class_desc);

	return return_tuple;
}

static HeapTuple
scan_pg_rel_ind(RelationBuildDescInfo buildinfo)
{
	Relation	pg_class_desc;
	HeapTuple	return_tuple;

	pg_class_desc = heap_openr(RelationRelationName);
	if (!IsInitProcessingMode())
		LockRelation(pg_class_desc, AccessShareLock);

	switch (buildinfo.infotype)
	{
		case INFO_RELID:
			return_tuple = ClassOidIndexScan(pg_class_desc, buildinfo.i.info_id);
			break;

		case INFO_RELNAME:
			return_tuple = ClassNameIndexScan(pg_class_desc,
											  buildinfo.i.info_name);
			break;

		default:
			elog(ERROR, "ScanPgRelation: bad buildinfo");

			/*
			 * XXX I hope this is right.  It seems better than returning
			 * an uninitialized value
			 */
			return_tuple = NULL;
	}

	/* all done */
	if (!IsInitProcessingMode())
		UnlockRelation(pg_class_desc, AccessShareLock);
	heap_close(pg_class_desc);

	return return_tuple;
}

/* ----------------
 *		AllocateRelationDesc
 *
 *		This is used to allocate memory for a new relation descriptor
 *		and initialize the rd_rel field.
 * ----------------
 */
static Relation
AllocateRelationDesc(u_int natts, Form_pg_class relp)
{
	Relation	relation;
	Size		len;
	Form_pg_class relationForm;

	/* ----------------
	 *	allocate space for the relation tuple form
	 * ----------------
	 */
	relationForm = (Form_pg_class)
		palloc((Size) (sizeof(FormData_pg_class)));

	memmove((char *) relationForm, (char *) relp, CLASS_TUPLE_SIZE);

	/* ----------------
	 *	allocate space for new relation descriptor
	 */
	len = sizeof(RelationData) + 10;	/* + 10 is voodoo XXX mao */

	relation = (Relation) palloc(len);

	/* ----------------
	 *	clear new reldesc
	 * ----------------
	 */
	MemSet((char *) relation, 0, len);

	/* initialize attribute tuple form */
	relation->rd_att = CreateTemplateTupleDesc(natts);

	/* and initialize relation tuple form */
	relation->rd_rel = relationForm;

	return relation;
}

/* --------------------------------
 *		RelationBuildTupleDesc
 *
 *		Form the relation's tuple descriptor from information in
 *		the pg_attribute, pg_attrdef & pg_relcheck system cataloges.
 * --------------------------------
 */
static void
RelationBuildTupleDesc(RelationBuildDescInfo buildinfo,
					   Relation relation,
					   u_int natts)
{

	/*
	 * If this is bootstrap time (initdb), then we can't use the system
	 * catalog indices, because they may not exist yet.  Otherwise, we
	 * can, and do.
	 */

	if (IsBootstrapProcessingMode())
		build_tupdesc_seq(buildinfo, relation, natts);
	else
		build_tupdesc_ind(buildinfo, relation, natts);
}

static void
build_tupdesc_seq(RelationBuildDescInfo buildinfo,
				  Relation relation,
				  u_int natts)
{
	HeapTuple	pg_attribute_tuple;
	Relation	pg_attribute_desc;
	HeapScanDesc pg_attribute_scan;
	Form_pg_attribute attp;
	ScanKeyData key;
	int			need;

	/* ----------------
	 *	form a scan key
	 * ----------------
	 */
	ScanKeyEntryInitialize(&key, 0,
						   Anum_pg_attribute_attrelid,
						   F_OIDEQ,
						   ObjectIdGetDatum(RelationGetRelid(relation)));

	/* ----------------
	 *	open pg_attribute and begin a scan
	 * ----------------
	 */
	pg_attribute_desc = heap_openr(AttributeRelationName);
	pg_attribute_scan = heap_beginscan(pg_attribute_desc, 0, SnapshotNow, 1, &key);

	/* ----------------
	 *	add attribute data to relation->rd_att
	 * ----------------
	 */
	need = natts;

	pg_attribute_tuple = heap_getnext(pg_attribute_scan, 0);
	while (HeapTupleIsValid(pg_attribute_tuple) && need > 0)
	{
		attp = (Form_pg_attribute) GETSTRUCT(pg_attribute_tuple);

		if (attp->attnum > 0)
		{
			relation->rd_att->attrs[attp->attnum - 1] =
				(Form_pg_attribute) palloc(ATTRIBUTE_TUPLE_SIZE);

			memmove((char *) (relation->rd_att->attrs[attp->attnum - 1]),
					(char *) attp,
					ATTRIBUTE_TUPLE_SIZE);
			need--;
		}
		pg_attribute_tuple = heap_getnext(pg_attribute_scan, 0);
	}

	if (need > 0)
		elog(ERROR, "catalog is missing %d attribute%s for relid %u",
			 need, (need == 1 ? "" : "s"), RelationGetRelid(relation));