opclasscmds.c

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

	/*
	 * The subtype is "default" (0) if second input type matches the operator
	 * class, otherwise it is the second input type.
	 */
	if (procform->proargtypes.values[1] == typeoid)
		subtype = InvalidOid;
	else
		subtype = procform->proargtypes.values[1];
	ReleaseSysCache(proctup);
	return subtype;
}

/*
 * Add a new class member to the appropriate list, after checking for
 * duplicated strategy or proc number.
 */
static void
addClassMember(List **list, OpClassMember *member, bool isProc)
{
	ListCell   *l;

	foreach(l, *list)
	{
		OpClassMember *old = (OpClassMember *) lfirst(l);

		if (old->number == member->number &&
			old->subtype == member->subtype)
		{
			if (isProc)
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
						 errmsg("procedure number %d appears more than once",
								member->number)));
			else
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
						 errmsg("operator number %d appears more than once",
								member->number)));
		}
	}
	*list = lappend(*list, member);
}

/*
 * Dump the operators to pg_amop
 */
static void
storeOperators(Oid opclassoid, List *operators)
{
	Relation	rel;
	Datum		values[Natts_pg_amop];
	char		nulls[Natts_pg_amop];
	HeapTuple	tup;
	ListCell   *l;
	int			i;

	rel = heap_open(AccessMethodOperatorRelationId, RowExclusiveLock);

	foreach(l, operators)
	{
		OpClassMember *op = (OpClassMember *) lfirst(l);

		for (i = 0; i < Natts_pg_amop; ++i)
		{
			nulls[i] = ' ';
			values[i] = (Datum) NULL;
		}

		i = 0;
		values[i++] = ObjectIdGetDatum(opclassoid);		/* amopclaid */
		values[i++] = ObjectIdGetDatum(op->subtype);	/* amopsubtype */
		values[i++] = Int16GetDatum(op->number);		/* amopstrategy */
		values[i++] = BoolGetDatum(op->recheck);		/* amopreqcheck */
		values[i++] = ObjectIdGetDatum(op->object);		/* amopopr */

		tup = heap_formtuple(rel->rd_att, values, nulls);

		simple_heap_insert(rel, tup);

		CatalogUpdateIndexes(rel, tup);

		heap_freetuple(tup);
	}

	heap_close(rel, RowExclusiveLock);
}

/*
 * Dump the procedures (support routines) to pg_amproc
 */
static void
storeProcedures(Oid opclassoid, List *procedures)
{
	Relation	rel;
	Datum		values[Natts_pg_amproc];
	char		nulls[Natts_pg_amproc];
	HeapTuple	tup;
	ListCell   *l;
	int			i;

	rel = heap_open(AccessMethodProcedureRelationId, RowExclusiveLock);

	foreach(l, procedures)
	{
		OpClassMember *proc = (OpClassMember *) lfirst(l);

		for (i = 0; i < Natts_pg_amproc; ++i)
		{
			nulls[i] = ' ';
			values[i] = (Datum) NULL;
		}

		i = 0;
		values[i++] = ObjectIdGetDatum(opclassoid);		/* amopclaid */
		values[i++] = ObjectIdGetDatum(proc->subtype);	/* amprocsubtype */
		values[i++] = Int16GetDatum(proc->number);		/* amprocnum */
		values[i++] = ObjectIdGetDatum(proc->object);	/* amproc */

		tup = heap_formtuple(rel->rd_att, values, nulls);

		simple_heap_insert(rel, tup);

		CatalogUpdateIndexes(rel, tup);

		heap_freetuple(tup);
	}

	heap_close(rel, RowExclusiveLock);
}


/*
 * RemoveOpClass
 *		Deletes an opclass.
 */
void
RemoveOpClass(RemoveOpClassStmt *stmt)
{
	Oid			amID,
				opcID;
	char	   *schemaname;
	char	   *opcname;
	HeapTuple	tuple;
	ObjectAddress object;

	/*
	 * Get the access method's OID.
	 */
	amID = GetSysCacheOid(AMNAME,
						  CStringGetDatum(stmt->amname),
						  0, 0, 0);
	if (!OidIsValid(amID))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("access method \"%s\" does not exist",
						stmt->amname)));

	/*
	 * Look up the opclass.
	 */

	/* deconstruct the name list */
	DeconstructQualifiedName(stmt->opclassname, &schemaname, &opcname);

	if (schemaname)
	{
		/* Look in specific schema only */
		Oid			namespaceId;

		namespaceId = LookupExplicitNamespace(schemaname);
		tuple = SearchSysCache(CLAAMNAMENSP,
							   ObjectIdGetDatum(amID),
							   PointerGetDatum(opcname),
							   ObjectIdGetDatum(namespaceId),
							   0);
	}
	else
	{
		/* Unqualified opclass name, so search the search path */
		opcID = OpclassnameGetOpcid(amID, opcname);
		if (!OidIsValid(opcID))
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("operator class \"%s\" does not exist for access method \"%s\"",
							opcname, stmt->amname)));
		tuple = SearchSysCache(CLAOID,
							   ObjectIdGetDatum(opcID),
							   0, 0, 0);
	}

	if (!HeapTupleIsValid(tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("operator class \"%s\" does not exist for access method \"%s\"",
						NameListToString(stmt->opclassname), stmt->amname)));

	opcID = HeapTupleGetOid(tuple);

	/* Permission check: must own opclass or its namespace */
	if (!pg_opclass_ownercheck(opcID, GetUserId()) &&
		!pg_namespace_ownercheck(((Form_pg_opclass) GETSTRUCT(tuple))->opcnamespace,
								 GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_OPCLASS,
					   NameListToString(stmt->opclassname));

	ReleaseSysCache(tuple);

	/*
	 * Do the deletion
	 */
	object.classId = OperatorClassRelationId;
	object.objectId = opcID;
	object.objectSubId = 0;

	performDeletion(&object, stmt->behavior);
}

/*
 * Guts of opclass deletion.
 */
void
RemoveOpClassById(Oid opclassOid)
{
	Relation	rel;
	HeapTuple	tup;
	ScanKeyData skey[1];
	SysScanDesc scan;

	/*
	 * First remove the pg_opclass entry itself.
	 */
	rel = heap_open(OperatorClassRelationId, RowExclusiveLock);

	tup = SearchSysCache(CLAOID,
						 ObjectIdGetDatum(opclassOid),
						 0, 0, 0);
	if (!HeapTupleIsValid(tup)) /* should not happen */
		elog(ERROR, "cache lookup failed for opclass %u", opclassOid);

	simple_heap_delete(rel, &tup->t_self);

	ReleaseSysCache(tup);

	heap_close(rel, RowExclusiveLock);

	/*
	 * Remove associated entries in pg_amop.
	 */
	ScanKeyInit(&skey[0],
				Anum_pg_amop_amopclaid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(opclassOid));

	rel = heap_open(AccessMethodOperatorRelationId, RowExclusiveLock);

	scan = systable_beginscan(rel, AccessMethodStrategyIndexId, true,
							  SnapshotNow, 1, skey);

	while (HeapTupleIsValid(tup = systable_getnext(scan)))
		simple_heap_delete(rel, &tup->t_self);

	systable_endscan(scan);
	heap_close(rel, RowExclusiveLock);

	/*
	 * Remove associated entries in pg_amproc.
	 */
	ScanKeyInit(&skey[0],
				Anum_pg_amproc_amopclaid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(opclassOid));

	rel = heap_open(AccessMethodProcedureRelationId, RowExclusiveLock);

	scan = systable_beginscan(rel, AccessMethodProcedureIndexId, true,
							  SnapshotNow, 1, skey);

	while (HeapTupleIsValid(tup = systable_getnext(scan)))
		simple_heap_delete(rel, &tup->t_self);

	systable_endscan(scan);
	heap_close(rel, RowExclusiveLock);
}


/*
 * Rename opclass
 */
void
RenameOpClass(List *name, const char *access_method, const char *newname)
{
	Oid			opcOid;
	Oid			amOid;
	Oid			namespaceOid;
	char	   *schemaname;
	char	   *opcname;
	HeapTuple	tup;
	Relation	rel;
	AclResult	aclresult;

	amOid = GetSysCacheOid(AMNAME,
						   CStringGetDatum(access_method),
						   0, 0, 0);
	if (!OidIsValid(amOid))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("access method \"%s\" does not exist",
						access_method)));

	rel = heap_open(OperatorClassRelationId, RowExclusiveLock);

	/*
	 * Look up the opclass
	 */
	DeconstructQualifiedName(name, &schemaname, &opcname);

	if (schemaname)
	{
		namespaceOid = LookupExplicitNamespace(schemaname);

		tup = SearchSysCacheCopy(CLAAMNAMENSP,
								 ObjectIdGetDatum(amOid),
								 PointerGetDatum(opcname),
								 ObjectIdGetDatum(namespaceOid),
								 0);
		if (!HeapTupleIsValid(tup))
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("operator class \"%s\" does not exist for access method \"%s\"",
							opcname, access_method)));

		opcOid = HeapTupleGetOid(tup);
	}
	else
	{
		opcOid = OpclassnameGetOpcid(amOid, opcname);
		if (!OidIsValid(opcOid))
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("operator class \"%s\" does not exist for access method \"%s\"",
							opcname, access_method)));

		tup = SearchSysCacheCopy(CLAOID,
								 ObjectIdGetDatum(opcOid),
								 0, 0, 0);
		if (!HeapTupleIsValid(tup))		/* should not happen */
			elog(ERROR, "cache lookup failed for opclass %u", opcOid);

		namespaceOid = ((Form_pg_opclass) GETSTRUCT(tup))->opcnamespace;
	}

	/* make sure the new name doesn't exist */
	if (SearchSysCacheExists(CLAAMNAMENSP,
							 ObjectIdGetDatum(amOid),
							 CStringGetDatum(newname),
							 ObjectIdGetDatum(namespaceOid),
							 0))
	{
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("operator class \"%s\" for access method \"%s\" already exists in schema \"%s\"",
						newname, access_method,
						get_namespace_name(namespaceOid))));
	}

	/* must be owner */
	if (!pg_opclass_ownercheck(opcOid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_OPCLASS,
					   NameListToString(name));

	/* must have CREATE privilege on namespace */
	aclresult = pg_namespace_aclcheck(namespaceOid, GetUserId(), ACL_CREATE);
	if (aclresult != ACLCHECK_OK)
		aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
					   get_namespace_name(namespaceOid));

	/* rename */
	namestrcpy(&(((Form_pg_opclass) GETSTRUCT(tup))->opcname), newname);
	simple_heap_update(rel, &tup->t_self, tup);
	CatalogUpdateIndexes(rel, tup);

	heap_close(rel, NoLock);
	heap_freetuple(tup);
}

/*
 * Change opclass owner
 */
void
AlterOpClassOwner(List *name, const char *access_method, Oid newOwnerId)
{
	Oid			opcOid;
	Oid			amOid;
	Oid			namespaceOid;
	char	   *schemaname;
	char	   *opcname;
	HeapTuple	tup;
	Relation	rel;
	AclResult	aclresult;
	Form_pg_opclass opcForm;

	amOid = GetSysCacheOid(AMNAME,
						   CStringGetDatum(access_method),
						   0, 0, 0);
	if (!OidIsValid(amOid))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("access method \"%s\" does not exist",
						access_method)));

	rel = heap_open(OperatorClassRelationId, RowExclusiveLock);

	/*
	 * Look up the opclass
	 */
	DeconstructQualifiedName(name, &schemaname, &opcname);

	if (schemaname)
	{
		namespaceOid = LookupExplicitNamespace(schemaname);

		tup = SearchSysCacheCopy(CLAAMNAMENSP,
								 ObjectIdGetDatum(amOid),
								 PointerGetDatum(opcname),
								 ObjectIdGetDatum(namespaceOid),
								 0);
		if (!HeapTupleIsValid(tup))
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("operator class \"%s\" does not exist for access method \"%s\"",
							opcname, access_method)));

		opcOid = HeapTupleGetOid(tup);
	}
	else
	{
		opcOid = OpclassnameGetOpcid(amOid, opcname);
		if (!OidIsValid(opcOid))
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("operator class \"%s\" does not exist for access method \"%s\"",
							opcname, access_method)));

		tup = SearchSysCacheCopy(CLAOID,
								 ObjectIdGetDatum(opcOid),
								 0, 0, 0);
		if (!HeapTupleIsValid(tup))		/* should not happen */
			elog(ERROR, "cache lookup failed for opclass %u", opcOid);
		namespaceOid = ((Form_pg_opclass) GETSTRUCT(tup))->opcnamespace;
	}
	opcForm = (Form_pg_opclass) GETSTRUCT(tup);

	/*
	 * If the new owner is the same as the existing owner, consider the
	 * command to have succeeded.  This is for dump restoration purposes.
	 */
	if (opcForm->opcowner != newOwnerId)
	{
		/* Superusers can always do it */
		if (!superuser())
		{
			/* Otherwise, must be owner of the existing object */
			if (!pg_opclass_ownercheck(HeapTupleGetOid(tup), GetUserId()))
				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_OPCLASS,
							   NameListToString(name));

			/* Must be able to become new owner */
			check_is_member_of_role(GetUserId(), newOwnerId);

			/* New owner must have CREATE privilege on namespace */
			aclresult = pg_namespace_aclcheck(namespaceOid, newOwnerId,
											  ACL_CREATE);
			if (aclresult != ACLCHECK_OK)
				aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
							   get_namespace_name(namespaceOid));
		}

		/*
		 * Modify the owner --- okay to scribble on tup because it's a copy
		 */
		opcForm->opcowner = newOwnerId;

		simple_heap_update(rel, &tup->t_self, tup);

		CatalogUpdateIndexes(rel, tup);

		/* Update owner dependency reference */
		changeDependencyOnOwner(OperatorClassRelationId, opcOid, newOwnerId);
	}

	heap_close(rel, NoLock);
	heap_freetuple(tup);
}