external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use table_openXXX methods in the codebase 

With PG13 heap_* (heap_open, heap_close etc) are replaced with table_*
(table_open, table_close etc).

It is better to use the new table access methods in the codebase and
define the macros for the previous versions as we can easily remove the
macro without having to change the codebase when we drop the support for
the old version.

Commits that introduced this change on Postgres:
f25968c49697db673f6cd2a07b3f7626779f1827
e0c4ec07284db817e1f8d9adfb3fffc952252db0
4b21acf522d751ba5b6679df391d5121b6c4a35f

Command to see relevant commits on Postgres side:
git log --all --grep="heap_open"
pull/3900/head
Sait Talha Nisanci 2020-05-18 12:19:29 +03:00
parent 0819b79631
commit bf831d2e59
30 changed files with 191 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/backend/distributed/commands/create_distributed_table.c
View File

@ -571,7 +571,7 @@ ColocationIdForNewTable(Oid relationId, Var *distributionColumn,
*/
Assert(distributionMethod == DISTRIBUTE_BY_HASH);
Relation pgDistColocation = heap_open(DistColocationRelationId(), ExclusiveLock);
Relation pgDistColocation = table_open(DistColocationRelationId(), ExclusiveLock);
Oid distributionColumnType = distributionColumn->vartype;
Oid distributionColumnCollation = get_typcollation(distributionColumnType);
@ -618,12 +618,12 @@ ColocationIdForNewTable(Oid relationId, Var *distributionColumn,
if (createdColocationGroup)
{
/* keep the exclusive lock */
heap_close(pgDistColocation, NoLock);
table_close(pgDistColocation, NoLock);
}
else
{
/* release the exclusive lock */
heap_close(pgDistColocation, ExclusiveLock);
table_close(pgDistColocation, ExclusiveLock);
}
}
@ -1266,7 +1266,7 @@ static void
CopyLocalDataIntoShards(Oid distributedRelationId)
{
/* take an ExclusiveLock to block all operations except SELECT */
Relation distributedRelation = heap_open(distributedRelationId, ExclusiveLock);
Relation distributedRelation = table_open(distributedRelationId, ExclusiveLock);
/*
* Skip copying from partitioned tables, we will copy the data from
@ -1274,7 +1274,7 @@ CopyLocalDataIntoShards(Oid distributedRelationId)
*/
if (PartitionedTable(distributedRelationId))
{
heap_close(distributedRelation, NoLock);
table_close(distributedRelation, NoLock);
return;
}
@ -1330,7 +1330,7 @@ CopyLocalDataIntoShards(Oid distributedRelationId)
/* free memory and close the relation */
ExecDropSingleTupleTableSlot(slot);
FreeExecutorState(estate);
heap_close(distributedRelation, NoLock);
table_close(distributedRelation, NoLock);
PopActiveSnapshot();
}

8
src/backend/distributed/commands/foreign_constraint.c
View File

@ -388,7 +388,7 @@ ColumnAppearsInForeignKeyToReferenceTable(char *columnName, Oid relationId)
int scanKeyCount = 1;
bool foreignKeyToReferenceTableIncludesGivenColumn = false;
Relation pgConstraint = heap_open(ConstraintRelationId, AccessShareLock);
Relation pgConstraint = table_open(ConstraintRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_constraint_contype, BTEqualStrategyNumber,
F_CHAREQ, CharGetDatum(CONSTRAINT_FOREIGN));
@ -446,7 +446,7 @@ ColumnAppearsInForeignKeyToReferenceTable(char *columnName, Oid relationId)
/* clean up scan and close system catalog */
systable_endscan(scanDescriptor);
heap_close(pgConstraint, NoLock);
table_close(pgConstraint, NoLock);
return foreignKeyToReferenceTableIncludesGivenColumn;
}
@ -720,7 +720,7 @@ GetForeignKeyOids(Oid relationId, int flags)
ScanKeyData scanKey[1];
int scanKeyCount = 1;
Relation pgConstraint = heap_open(ConstraintRelationId, AccessShareLock);
Relation pgConstraint = table_open(ConstraintRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], pgConstraintTargetAttrNumber,
BTEqualStrategyNumber, F_OIDEQ, relationId);
SysScanDesc scanDescriptor = systable_beginscan(pgConstraint, indexOid, useIndex,
@ -770,7 +770,7 @@ GetForeignKeyOids(Oid relationId, int flags)
* on pg_constraint to make sure that caller will process valid foreign key
* constraints through the transaction.
*/
heap_close(pgConstraint, NoLock);
table_close(pgConstraint, NoLock);
return foreignKeyOids;
}

9
src/backend/distributed/commands/function.c
View File

@ -47,6 +47,7 @@
#include "distributed/multi_executor.h"
#include "distributed/namespace_utils.h"
#include "distributed/relation_access_tracking.h"
#include "distributed/version_compat.h"
#include "distributed/worker_create_or_replace.h"
#include "distributed/worker_transaction.h"
#include "nodes/makefuncs.h"
@ -352,7 +353,7 @@ GetFunctionColocationId(Oid functionOid, char *colocateWithTableName,
Oid distributionArgumentOid)
{
int colocationId = INVALID_COLOCATION_ID;
Relation pgDistColocation = heap_open(DistColocationRelationId(), ShareLock);
Relation pgDistColocation = table_open(DistColocationRelationId(), ShareLock);
if (pg_strncasecmp(colocateWithTableName, "default", NAMEDATALEN) == 0)
{
@ -400,7 +401,7 @@ GetFunctionColocationId(Oid functionOid, char *colocateWithTableName,
}
/* keep the lock */
heap_close(pgDistColocation, NoLock);
table_close(pgDistColocation, NoLock);
return colocationId;
}
@ -489,7 +490,7 @@ UpdateFunctionDistributionInfo(const ObjectAddress *distAddress,
bool isnull[Natts_pg_dist_object];
bool replace[Natts_pg_dist_object];
Relation pgDistObjectRel = heap_open(DistObjectRelationId(), RowExclusiveLock);
Relation pgDistObjectRel = table_open(DistObjectRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistObjectRel);
/* scan pg_dist_object for classid = $1 AND objid = $2 AND objsubid = $3 via index */
@ -549,7 +550,7 @@ UpdateFunctionDistributionInfo(const ObjectAddress *distAddress,
systable_endscan(scanDescriptor);
heap_close(pgDistObjectRel, NoLock);
table_close(pgDistObjectRel, NoLock);
}

16
src/backend/distributed/commands/index.c
View File

@ -138,7 +138,7 @@ PreprocessIndexStmt(Node *node, const char *createIndexCommand)
* checked permissions, and will only fail when executing the actual
* index statements.
*/
Relation relation = heap_openrv(createIndexStatement->relation, lockmode);
Relation relation = table_openrv(createIndexStatement->relation, lockmode);
Oid relationId = RelationGetRelid(relation);
bool isCitusRelation = IsCitusTable(relationId);
@ -160,7 +160,7 @@ PreprocessIndexStmt(Node *node, const char *createIndexCommand)
relationContext, namespaceName);
}
heap_close(relation, NoLock);
table_close(relation, NoLock);
if (isCitusRelation)
{
@ -246,7 +246,7 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand)
RangeVarGetRelidExtended(reindexStatement->relation, lockmode, 0,
RangeVarCallbackOwnsTable, NULL);
relation = heap_openrv(reindexStatement->relation, NoLock);
relation = table_openrv(reindexStatement->relation, NoLock);
relationId = RelationGetRelid(relation);
}
@ -275,7 +275,7 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand)
}
else
{
heap_close(relation, NoLock);
table_close(relation, NoLock);
}
if (isCitusRelation)
@ -426,13 +426,13 @@ PostprocessIndexStmt(Node *node, const char *queryString)
StartTransactionCommand();
/* get the affected relation and index */
Relation relation = heap_openrv(indexStmt->relation, ShareUpdateExclusiveLock);
Relation relation = table_openrv(indexStmt->relation, ShareUpdateExclusiveLock);
Oid indexRelationId = get_relname_relid(indexStmt->idxname,
schemaId);
Relation indexRelation = index_open(indexRelationId, RowExclusiveLock);
/* close relations but retain locks */
heap_close(relation, NoLock);
table_close(relation, NoLock);
index_close(indexRelation, NoLock);
/* mark index as invalid, in-place (cannot be rolled back) */
@ -443,7 +443,7 @@ PostprocessIndexStmt(Node *node, const char *queryString)
StartTransactionCommand();
/* now, update index's validity in a way that can roll back */
Relation pg_index = heap_open(IndexRelationId, RowExclusiveLock);
Relation pg_index = table_open(IndexRelationId, RowExclusiveLock);
HeapTuple indexTuple = SearchSysCacheCopy1(INDEXRELID, ObjectIdGetDatum(
indexRelationId));
@ -457,7 +457,7 @@ PostprocessIndexStmt(Node *node, const char *queryString)
/* clean up; index now marked valid, but ROLLBACK will mark invalid */
heap_freetuple(indexTuple);
heap_close(pg_index, RowExclusiveLock);
table_close(pg_index, RowExclusiveLock);
return NIL;
}

5
src/backend/distributed/commands/local_multi_copy.c
View File

@ -34,6 +34,7 @@
#include "distributed/local_executor.h"
#include "distributed/local_multi_copy.h"
#include "distributed/shard_utils.h"
#include "distributed/version_compat.h"
static int ReadFromLocalBufferCallback(void *outBuf, int minRead, int maxRead);
static void AddSlotToBuffer(TupleTableSlot *slot, CitusCopyDestReceiver *copyDest,
@ -160,7 +161,7 @@ DoLocalCopy(StringInfo buffer, Oid relationId, int64 shardId, CopyStmt *copyStat
LocalCopyBuffer = buffer;
Oid shardOid = GetTableLocalShardOid(relationId, shardId);
Relation shard = heap_open(shardOid, RowExclusiveLock);
Relation shard = table_open(shardOid, RowExclusiveLock);
ParseState *pState = make_parsestate(NULL);
/* p_rtable of pState is set so that we can check constraints. */
@ -172,7 +173,7 @@ DoLocalCopy(StringInfo buffer, Oid relationId, int64 shardId, CopyStmt *copyStat
CopyFrom(cstate);
EndCopyFrom(cstate);
heap_close(shard, NoLock);
table_close(shard, NoLock);
free_parsestate(pState);
}

28
src/backend/distributed/commands/multi_copy.c
View File

@ -410,7 +410,7 @@ CopyToExistingShards(CopyStmt *copyStatement, char *completionTag)
ErrorContextCallback errorCallback;
/* allocate column values and nulls arrays */
Relation distributedRelation = heap_open(tableId, RowExclusiveLock);
Relation distributedRelation = table_open(tableId, RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(distributedRelation);
uint32 columnCount = tupleDescriptor->natts;
Datum *columnValues = palloc0(columnCount * sizeof(Datum));
@ -545,7 +545,7 @@ CopyToExistingShards(CopyStmt *copyStatement, char *completionTag)
ExecDropSingleTupleTableSlot(tupleTableSlot);
FreeExecutorState(executorState);
heap_close(distributedRelation, NoLock);
table_close(distributedRelation, NoLock);
/* mark failed placements as inactive */
MarkFailedShardPlacements();
@ -568,7 +568,7 @@ static void
CopyToNewShards(CopyStmt *copyStatement, char *completionTag, Oid relationId)
{
/* allocate column values and nulls arrays */
Relation distributedRelation = heap_open(relationId, RowExclusiveLock);
Relation distributedRelation = table_open(relationId, RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(distributedRelation);
uint32 columnCount = tupleDescriptor->natts;
Datum *columnValues = palloc0(columnCount * sizeof(Datum));
@ -732,7 +732,7 @@ CopyToNewShards(CopyStmt *copyStatement, char *completionTag, Oid relationId)
}
EndCopyFrom(copyState);
heap_close(distributedRelation, NoLock);
table_close(distributedRelation, NoLock);
/* check for cancellation one last time before returning */
CHECK_FOR_INTERRUPTS();
@ -2138,7 +2138,7 @@ CitusCopyDestReceiverStartup(DestReceiver *dest, int operation,
const char *nullPrintCharacter = "\\N";
/* look up table properties */
Relation distributedRelation = heap_open(tableId, RowExclusiveLock);
Relation distributedRelation = table_open(tableId, RowExclusiveLock);
CitusTableCacheEntry *cacheEntry = GetCitusTableCacheEntry(tableId);
partitionMethod = cacheEntry->partitionMethod;
@ -2626,7 +2626,7 @@ CitusCopyDestReceiverShutdown(DestReceiver *destReceiver)
}
PG_END_TRY();
heap_close(distributedRelation, NoLock);
table_close(distributedRelation, NoLock);
}
@ -2801,7 +2801,7 @@ ProcessCopyStmt(CopyStmt *copyStatement, char *completionTag, const char *queryS
bool isFrom = copyStatement->is_from;
/* consider using RangeVarGetRelidExtended to check perms before locking */
Relation copiedRelation = heap_openrv(copyStatement->relation,
Relation copiedRelation = table_openrv(copyStatement->relation,
isFrom ? RowExclusiveLock :
AccessShareLock);
@ -2816,7 +2816,7 @@ ProcessCopyStmt(CopyStmt *copyStatement, char *completionTag, const char *queryS
schemaName = MemoryContextStrdup(relationContext, schemaName);
copyStatement->relation->schemaname = schemaName;
heap_close(copiedRelation, NoLock);
table_close(copiedRelation, NoLock);
if (isCitusRelation)
{
@ -2875,7 +2875,7 @@ CitusCopySelect(CopyStmt *copyStatement)
SelectStmt *selectStmt = makeNode(SelectStmt);
selectStmt->fromClause = list_make1(copyObject(copyStatement->relation));
Relation distributedRelation = heap_openrv(copyStatement->relation, AccessShareLock);
Relation distributedRelation = table_openrv(copyStatement->relation, AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(distributedRelation);
List *targetList = NIL;
@ -2905,7 +2905,7 @@ CitusCopySelect(CopyStmt *copyStatement)
targetList = lappend(targetList, selectTarget);
}
heap_close(distributedRelation, NoLock);
table_close(distributedRelation, NoLock);
selectStmt->targetList = targetList;
return selectStmt;
@ -2922,7 +2922,7 @@ CitusCopyTo(CopyStmt *copyStatement, char *completionTag)
ListCell *shardIntervalCell = NULL;
int64 tuplesSent = 0;
Relation distributedRelation = heap_openrv(copyStatement->relation, AccessShareLock);
Relation distributedRelation = table_openrv(copyStatement->relation, AccessShareLock);
Oid relationId = RelationGetRelid(distributedRelation);
TupleDesc tupleDescriptor = RelationGetDescr(distributedRelation);
@ -3004,7 +3004,7 @@ CitusCopyTo(CopyStmt *copyStatement, char *completionTag)
SendCopyEnd(copyOutState);
heap_close(distributedRelation, AccessShareLock);
table_close(distributedRelation, AccessShareLock);
if (completionTag != NULL)
{
@ -3079,7 +3079,7 @@ CheckCopyPermissions(CopyStmt *copyStatement)
List *attnums;
ListCell *cur;
rel = heap_openrv(copyStatement->relation,
rel = table_openrv(copyStatement->relation,
is_from ? RowExclusiveLock : AccessShareLock);
range_table = CreateRangeTable(rel, required_access);
@ -3105,7 +3105,7 @@ CheckCopyPermissions(CopyStmt *copyStatement)
/* TODO: Perform RLS checks once supported */
heap_close(rel, NoLock);
table_close(rel, NoLock);
/* *INDENT-ON* */
}

9
src/backend/distributed/commands/role.c
View File

@ -33,6 +33,7 @@
#include "distributed/coordinator_protocol.h"
#include "distributed/metadata/distobject.h"
#include "distributed/metadata_sync.h"
#include "distributed/version_compat.h"
#include "distributed/worker_transaction.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
@ -315,7 +316,7 @@ CreateCreateOrAlterRoleCommand(const char *roleName,
static const char *
ExtractEncryptedPassword(Oid roleOid)
{
Relation pgAuthId = heap_open(AuthIdRelationId, AccessShareLock);
Relation pgAuthId = table_open(AuthIdRelationId, AccessShareLock);
TupleDesc pgAuthIdDescription = RelationGetDescr(pgAuthId);
HeapTuple tuple = SearchSysCache1(AUTHOID, roleOid);
bool isNull = true;
@ -328,7 +329,7 @@ ExtractEncryptedPassword(Oid roleOid)
Datum passwordDatum = heap_getattr(tuple, Anum_pg_authid_rolpassword,
pgAuthIdDescription, &isNull);
heap_close(pgAuthId, AccessShareLock);
table_close(pgAuthId, AccessShareLock);
ReleaseSysCache(tuple);
if (isNull)
@ -527,7 +528,7 @@ GenerateCreateOrAlterRoleCommand(Oid roleOid)
List *
GenerateAlterRoleSetCommandForRole(Oid roleid)
{
Relation DbRoleSetting = heap_open(DbRoleSettingRelationId, AccessShareLock);
Relation DbRoleSetting = table_open(DbRoleSettingRelationId, AccessShareLock);
TupleDesc DbRoleSettingDescription = RelationGetDescr(DbRoleSetting);
HeapTuple tuple = NULL;
List *commands = NIL;
@ -561,7 +562,7 @@ GenerateAlterRoleSetCommandForRole(Oid roleid)
}
heap_endscan(scan);
heap_close(DbRoleSetting, AccessShareLock);
table_close(DbRoleSetting, AccessShareLock);
return commands;
}

7
src/backend/distributed/commands/schema.c
View File

@ -30,6 +30,7 @@
#include "distributed/resource_lock.h"
#include <distributed/remote_commands.h>
#include <distributed/remote_commands.h>
#include "distributed/version_compat.h"
#include "nodes/parsenodes.h"
#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
@ -71,7 +72,7 @@ PreprocessDropSchemaStmt(Node *node, const char *queryString)
continue;
}
pgClass = heap_open(RelationRelationId, AccessShareLock);
pgClass = table_open(RelationRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_class_relnamespace, BTEqualStrategyNumber,
F_OIDEQ, namespaceOid);
@ -105,7 +106,7 @@ PreprocessDropSchemaStmt(Node *node, const char *queryString)
MarkInvalidateForeignKeyGraph();
systable_endscan(scanDescriptor);
heap_close(pgClass, NoLock);
table_close(pgClass, NoLock);
return NIL;
}
@ -113,7 +114,7 @@ PreprocessDropSchemaStmt(Node *node, const char *queryString)
}
systable_endscan(scanDescriptor);
heap_close(pgClass, NoLock);
table_close(pgClass, NoLock);
}
return NIL;

4
src/backend/distributed/commands/trigger.c
View File

@ -70,7 +70,7 @@ GetExplicitTriggerIdList(Oid relationId)
{
List *triggerIdList = NIL;
Relation pgTrigger = heap_open(TriggerRelationId, AccessShareLock);
Relation pgTrigger = table_open(TriggerRelationId, AccessShareLock);
int scanKeyCount = 1;
ScanKeyData scanKey[1];
@ -103,7 +103,7 @@ GetExplicitTriggerIdList(Oid relationId)
}
systable_endscan(scanDescriptor);
heap_close(pgTrigger, NoLock);
table_close(pgTrigger, NoLock);
return triggerIdList;
}

5
src/backend/distributed/commands/type.c
View File

@ -64,6 +64,7 @@
#include "distributed/remote_commands.h"
#include "distributed/transaction_management.h"
#include "distributed/worker_create_or_replace.h"
#include "distributed/version_compat.h"
#include "distributed/worker_manager.h"
#include "distributed/worker_transaction.h"
#include "miscadmin.h"
@ -791,7 +792,7 @@ EnumValsList(Oid typeOid)
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(typeOid));
Relation enum_rel = heap_open(EnumRelationId, AccessShareLock);
Relation enum_rel = table_open(EnumRelationId, AccessShareLock);
SysScanDesc enum_scan = systable_beginscan(enum_rel,
EnumTypIdSortOrderIndexId,
true, NULL,
@ -805,7 +806,7 @@ EnumValsList(Oid typeOid)
}
systable_endscan(enum_scan);
heap_close(enum_rel, AccessShareLock);
table_close(enum_rel, AccessShareLock);
return vals;
}

8
src/backend/distributed/deparser/citus_ruleutils.c
View File

@ -128,7 +128,7 @@ get_extension_schema(Oid ext_oid)
HeapTuple tuple;
ScanKeyData entry[1];
rel = heap_open(ExtensionRelationId, AccessShareLock);
rel = table_open(ExtensionRelationId, AccessShareLock);
ScanKeyInit(&entry[0],
#if PG_VERSION_NUM >= PG_VERSION_12
@ -152,7 +152,7 @@ get_extension_schema(Oid ext_oid)
systable_endscan(scandesc);
heap_close(rel, AccessShareLock);
table_close(rel, AccessShareLock);
return result;
/* *INDENT-ON* */
@ -1174,7 +1174,7 @@ pg_get_replica_identity_command(Oid tableRelationId)
{
StringInfo buf = makeStringInfo();
Relation relation = heap_open(tableRelationId, AccessShareLock);
Relation relation = table_open(tableRelationId, AccessShareLock);
char replicaIdentity = relation->rd_rel->relreplident;
@ -1202,7 +1202,7 @@ pg_get_replica_identity_command(Oid tableRelationId)
relationName);
}
heap_close(relation, AccessShareLock);
table_close(relation, AccessShareLock);
return (buf->len > 0) ? buf->data : NULL;
}

1
src/backend/distributed/executor/insert_select_executor.c
View File

@ -34,6 +34,7 @@
#include "distributed/shardinterval_utils.h"
#include "distributed/subplan_execution.h"
#include "distributed/transaction_management.h"
#include "distributed/version_compat.h"
#include "executor/executor.h"
#include "nodes/execnodes.h"
#include "nodes/makefuncs.h"

9
src/backend/distributed/metadata/dependency.c
View File

@ -25,6 +25,7 @@
#include "distributed/metadata/dependency.h"
#include "distributed/metadata/distobject.h"
#include "distributed/metadata_cache.h"
#include "distributed/version_compat.h"
#include "miscadmin.h"
#include "utils/fmgroids.h"
#include "utils/hsearch.h"
@ -304,7 +305,7 @@ DependencyDefinitionFromPgDepend(ObjectAddress target)
/*
* iterate the actual pg_depend catalog
*/
Relation depRel = heap_open(DependRelationId, AccessShareLock);
Relation depRel = table_open(DependRelationId, AccessShareLock);
/* scan pg_depend for classid = $1 AND objid = $2 using pg_depend_depender_index */
ScanKeyInit(&key[0], Anum_pg_depend_classid, BTEqualStrategyNumber, F_OIDEQ,
@ -346,7 +347,7 @@ DependencyDefinitionFromPgShDepend(ObjectAddress target)
/*
* iterate the actual pg_shdepend catalog
*/
Relation shdepRel = heap_open(SharedDependRelationId, AccessShareLock);
Relation shdepRel = table_open(SharedDependRelationId, AccessShareLock);
/*
* Scan pg_shdepend for dbid = $1 AND classid = $2 AND objid = $3 using
@ -621,7 +622,7 @@ IsObjectAddressOwnedByExtension(const ObjectAddress *target,
HeapTuple depTup = NULL;
bool result = false;
Relation depRel = heap_open(DependRelationId, AccessShareLock);
Relation depRel = table_open(DependRelationId, AccessShareLock);
/* scan pg_depend for classid = $1 AND objid = $2 using pg_depend_depender_index */
ScanKeyInit(&key[0], Anum_pg_depend_classid, BTEqualStrategyNumber, F_OIDEQ,
@ -647,7 +648,7 @@ IsObjectAddressOwnedByExtension(const ObjectAddress *target,
}
systable_endscan(depScan);
heap_close(depRel, AccessShareLock);
table_close(depRel, AccessShareLock);
return result;
}

11
src/backend/distributed/metadata/distobject.c
View File

@ -31,6 +31,7 @@
#include "distributed/metadata/distobject.h"
#include "distributed/metadata/pg_dist_object.h"
#include "distributed/metadata_cache.h"
#include "distributed/version_compat.h"
#include "executor/spi.h"
#include "nodes/makefuncs.h"
#include "nodes/pg_list.h"
@ -103,7 +104,7 @@ ObjectExists(const ObjectAddress *address)
if (is_objectclass_supported(address->classId))
{
HeapTuple objtup;
Relation catalog = heap_open(address->classId, AccessShareLock);
Relation catalog = table_open(address->classId, AccessShareLock);
#if PG_VERSION_NUM >= PG_VERSION_12
objtup = get_catalog_object_by_oid(catalog, get_object_attnum_oid(
@ -111,7 +112,7 @@ ObjectExists(const ObjectAddress *address)
#else
objtup = get_catalog_object_by_oid(catalog, address->objectId);
#endif
heap_close(catalog, AccessShareLock);
table_close(catalog, AccessShareLock);
if (objtup != NULL)
{
return true;
@ -257,7 +258,7 @@ IsObjectDistributed(const ObjectAddress *address)
ScanKeyData key[3];
bool result = false;
Relation pgDistObjectRel = heap_open(DistObjectRelationId(), AccessShareLock);
Relation pgDistObjectRel = table_open(DistObjectRelationId(), AccessShareLock);
/* scan pg_dist_object for classid = $1 AND objid = $2 AND objsubid = $3 via index */
ScanKeyInit(&key[0], Anum_pg_dist_object_classid, BTEqualStrategyNumber, F_OIDEQ,
@ -295,7 +296,7 @@ ClusterHasDistributedFunctionWithDistArgument(void)
HeapTuple pgDistObjectTup = NULL;
Relation pgDistObjectRel = heap_open(DistObjectRelationId(), AccessShareLock);
Relation pgDistObjectRel = table_open(DistObjectRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistObjectRel);
@ -340,7 +341,7 @@ GetDistributedObjectAddressList(void)
HeapTuple pgDistObjectTup = NULL;
List *objectAddressList = NIL;
Relation pgDistObjectRel = heap_open(DistObjectRelationId(), AccessShareLock);
Relation pgDistObjectRel = table_open(DistObjectRelationId(), AccessShareLock);
SysScanDesc pgDistObjectScan = systable_beginscan(pgDistObjectRel, InvalidOid, false,
NULL, 0,
NULL);

52
src/backend/distributed/metadata/metadata_cache.c
View File

@ -323,7 +323,7 @@ IsCitusTableViaCatalog(Oid relationId)
ScanKeyData scanKey[1];
bool indexOK = true;
Relation pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId));
@ -334,7 +334,7 @@ IsCitusTableViaCatalog(Oid relationId)
HeapTuple partitionTuple = systable_getnext(scanDescriptor);
systable_endscan(scanDescriptor);
heap_close(pgDistPartition, AccessShareLock);
table_close(pgDistPartition, AccessShareLock);
return HeapTupleIsValid(partitionTuple);
}
@ -1006,7 +1006,7 @@ LookupDistObjectCacheEntry(Oid classid, Oid objid, int32 objsubid)
cacheEntry->key.objid = objid;
cacheEntry->key.objsubid = objsubid;
Relation pgDistObjectRel = heap_open(DistObjectRelationId(), AccessShareLock);
Relation pgDistObjectRel = table_open(DistObjectRelationId(), AccessShareLock);
TupleDesc pgDistObjectTupleDesc = RelationGetDescr(pgDistObjectRel);
ScanKeyInit(&pgDistObjectKey[0], Anum_pg_dist_object_classid,
@ -1059,14 +1059,14 @@ LookupDistObjectCacheEntry(Oid classid, Oid objid, int32 objsubid)
static CitusTableCacheEntry *
BuildCitusTableCacheEntry(Oid relationId)
{
Relation pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
HeapTuple distPartitionTuple =
LookupDistPartitionTuple(pgDistPartition, relationId);
if (distPartitionTuple == NULL)
{
/* not a distributed table, done */
heap_close(pgDistPartition, NoLock);
table_close(pgDistPartition, NoLock);
return NULL;
}
@ -1166,7 +1166,7 @@ BuildCitusTableCacheEntry(Oid relationId)
MemoryContextSwitchTo(oldContext);
heap_close(pgDistPartition, NoLock);
table_close(pgDistPartition, NoLock);
cacheEntry->isValid = true;
@ -1201,7 +1201,7 @@ BuildCachedShardList(CitusTableCacheEntry *cacheEntry)
int shardIntervalArrayLength = list_length(distShardTupleList);
if (shardIntervalArrayLength > 0)
{
Relation distShardRelation = heap_open(DistShardRelationId(), AccessShareLock);
Relation distShardRelation = table_open(DistShardRelationId(), AccessShareLock);
TupleDesc distShardTupleDesc = RelationGetDescr(distShardRelation);
int arrayIndex = 0;
@ -1236,7 +1236,7 @@ BuildCachedShardList(CitusTableCacheEntry *cacheEntry)
arrayIndex++;
}
heap_close(distShardRelation, AccessShareLock);
table_close(distShardRelation, AccessShareLock);
}
/* look up value comparison function */
@ -1847,7 +1847,7 @@ InstalledExtensionVersion(void)
InitializeCaches();
Relation relation = heap_open(ExtensionRelationId, AccessShareLock);
Relation relation = table_open(ExtensionRelationId, AccessShareLock);
ScanKeyInit(&entry[0], Anum_pg_extension_extname, BTEqualStrategyNumber, F_NAMEEQ,
CStringGetDatum("citus"));
@ -1889,7 +1889,7 @@ InstalledExtensionVersion(void)
systable_endscan(scandesc);
heap_close(relation, AccessShareLock);
table_close(relation, AccessShareLock);
return installedExtensionVersion;
}
@ -2400,7 +2400,7 @@ CitusExtensionOwner(void)
return MetadataCache.extensionOwner;
}
Relation relation = heap_open(ExtensionRelationId, AccessShareLock);
Relation relation = table_open(ExtensionRelationId, AccessShareLock);
ScanKeyInit(&entry[0],
Anum_pg_extension_extname,
@ -2440,7 +2440,7 @@ CitusExtensionOwner(void)
systable_endscan(scandesc);
heap_close(relation, AccessShareLock);
table_close(relation, AccessShareLock);
return MetadataCache.extensionOwner;
}
@ -3228,7 +3228,7 @@ GetLocalGroupId(void)
return 0;
}
Relation pgDistLocalGroupId = heap_open(localGroupTableOid, AccessShareLock);
Relation pgDistLocalGroupId = table_open(localGroupTableOid, AccessShareLock);
SysScanDesc scanDescriptor = systable_beginscan(pgDistLocalGroupId,
InvalidOid, false,
@ -3260,7 +3260,7 @@ GetLocalGroupId(void)
}
systable_endscan(scanDescriptor);
heap_close(pgDistLocalGroupId, AccessShareLock);
table_close(pgDistLocalGroupId, AccessShareLock);
return groupId;
}
@ -3671,7 +3671,7 @@ DistTableOidList(void)
int scanKeyCount = 0;
List *distTableOidList = NIL;
Relation pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
SysScanDesc scanDescriptor = systable_beginscan(pgDistPartition,
InvalidOid, false,
@ -3693,7 +3693,7 @@ DistTableOidList(void)
}
systable_endscan(scanDescriptor);
heap_close(pgDistPartition, AccessShareLock);
table_close(pgDistPartition, AccessShareLock);
return distTableOidList;
}
@ -3713,7 +3713,7 @@ ReferenceTableOidList()
int scanKeyCount = 0;
List *referenceTableOidList = NIL;
Relation pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
SysScanDesc scanDescriptor = systable_beginscan(pgDistPartition,
InvalidOid, false,
@ -3742,7 +3742,7 @@ ReferenceTableOidList()
}
systable_endscan(scanDescriptor);
heap_close(pgDistPartition, AccessShareLock);
table_close(pgDistPartition, AccessShareLock);
return referenceTableOidList;
}
@ -3856,7 +3856,7 @@ LookupDistShardTuples(Oid relationId)
List *distShardTupleList = NIL;
ScanKeyData scanKey[1];
Relation pgDistShard = heap_open(DistShardRelationId(), AccessShareLock);
Relation pgDistShard = table_open(DistShardRelationId(), AccessShareLock);
/* copy scankey to local copy, it will be modified during the scan */
scanKey[0] = DistShardScanKey[0];
@ -3878,7 +3878,7 @@ LookupDistShardTuples(Oid relationId)
}
systable_endscan(scanDescriptor);
heap_close(pgDistShard, AccessShareLock);
table_close(pgDistShard, AccessShareLock);
return distShardTupleList;
}
@ -3896,7 +3896,7 @@ LookupShardRelationFromCatalog(int64 shardId, bool missingOk)
ScanKeyData scanKey[1];
int scanKeyCount = 1;
Form_pg_dist_shard shardForm = NULL;
Relation pgDistShard = heap_open(DistShardRelationId(), AccessShareLock);
Relation pgDistShard = table_open(DistShardRelationId(), AccessShareLock);
Oid relationId = InvalidOid;
ScanKeyInit(&scanKey[0], Anum_pg_dist_shard_shardid,
@ -3924,7 +3924,7 @@ LookupShardRelationFromCatalog(int64 shardId, bool missingOk)
}
systable_endscan(scanDescriptor);
heap_close(pgDistShard, NoLock);
table_close(pgDistShard, NoLock);
return relationId;
}
@ -4206,7 +4206,7 @@ CitusInvalidateRelcacheByShardId(int64 shardId)
ScanKeyData scanKey[1];
int scanKeyCount = 1;
Form_pg_dist_shard shardForm = NULL;
Relation pgDistShard = heap_open(DistShardRelationId(), AccessShareLock);
Relation pgDistShard = table_open(DistShardRelationId(), AccessShareLock);
/*
* Load shard, to find the associated relation id. Can't use
@ -4249,7 +4249,7 @@ CitusInvalidateRelcacheByShardId(int64 shardId)
}
systable_endscan(scanDescriptor);
heap_close(pgDistShard, NoLock);
table_close(pgDistShard, NoLock);
/* bump command counter, to force invalidation to take effect */
CommandCounterIncrement();
@ -4274,7 +4274,7 @@ DistNodeMetadata(void)
ereport(ERROR, (errmsg("pg_dist_node_metadata was not found")));
}
Relation pgDistNodeMetadata = heap_open(metadataTableOid, AccessShareLock);
Relation pgDistNodeMetadata = table_open(metadataTableOid, AccessShareLock);
SysScanDesc scanDescriptor = systable_beginscan(pgDistNodeMetadata,
InvalidOid, false,
NULL, scanKeyCount, scanKey);
@ -4295,7 +4295,7 @@ DistNodeMetadata(void)
}
systable_endscan(scanDescriptor);
heap_close(pgDistNodeMetadata, AccessShareLock);
table_close(pgDistNodeMetadata, AccessShareLock);
return metadata;
}

4
src/backend/distributed/metadata/metadata_sync.c
View File

@ -989,7 +989,7 @@ UpdateDistNodeBoolAttr(const char *nodeName, int32 nodePort, int attrNum, bool v
bool isnull[Natts_pg_dist_node];
bool replace[Natts_pg_dist_node];
Relation pgDistNode = heap_open(DistNodeRelationId(), RowExclusiveLock);
Relation pgDistNode = table_open(DistNodeRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistNode);
ScanKeyInit(&scanKey[0], Anum_pg_dist_node_nodename,
@ -1022,7 +1022,7 @@ UpdateDistNodeBoolAttr(const char *nodeName, int32 nodePort, int attrNum, bool v
CommandCounterIncrement();
systable_endscan(scanDescriptor);
heap_close(pgDistNode, NoLock);
table_close(pgDistNode, NoLock);
}

42
src/backend/distributed/metadata/metadata_utility.c
View File

@ -190,7 +190,7 @@ DistributedTableSize(Oid relationId, char *sizeQuery)
totalRelationSize += relationSizeOnNode;
}
heap_close(relation, AccessShareLock);
table_close(relation, AccessShareLock);
return totalRelationSize;
}
@ -633,7 +633,7 @@ NodeGroupHasShardPlacements(int32 groupId, bool onlyConsiderActivePlacements)
ScanKeyData scanKey[2];
Relation pgPlacement = heap_open(DistPlacementRelationId(),
Relation pgPlacement = table_open(DistPlacementRelationId(),
AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_placement_groupid,
@ -654,7 +654,7 @@ NodeGroupHasShardPlacements(int32 groupId, bool onlyConsiderActivePlacements)
bool hasActivePlacements = HeapTupleIsValid(heapTuple);
systable_endscan(scanDescriptor);
heap_close(pgPlacement, NoLock);
table_close(pgPlacement, NoLock);
return hasActivePlacements;
}
@ -731,7 +731,7 @@ BuildShardPlacementList(ShardInterval *shardInterval)
int scanKeyCount = 1;
bool indexOK = true;
Relation pgPlacement = heap_open(DistPlacementRelationId(), AccessShareLock);
Relation pgPlacement = table_open(DistPlacementRelationId(), AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_placement_shardid,
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(shardId));
@ -755,7 +755,7 @@ BuildShardPlacementList(ShardInterval *shardInterval)
}
systable_endscan(scanDescriptor);
heap_close(pgPlacement, NoLock);
table_close(pgPlacement, NoLock);
return shardPlacementList;
}
@ -774,7 +774,7 @@ AllShardPlacementsOnNodeGroup(int32 groupId)
int scanKeyCount = 1;
bool indexOK = true;
Relation pgPlacement = heap_open(DistPlacementRelationId(), AccessShareLock);
Relation pgPlacement = table_open(DistPlacementRelationId(), AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_placement_groupid,
BTEqualStrategyNumber, F_INT4EQ, Int32GetDatum(groupId));
@ -798,7 +798,7 @@ AllShardPlacementsOnNodeGroup(int32 groupId)
}
systable_endscan(scanDescriptor);
heap_close(pgPlacement, NoLock);
table_close(pgPlacement, NoLock);
return shardPlacementList;
}
@ -879,7 +879,7 @@ InsertShardRow(Oid relationId, uint64 shardId, char storageType,
}
/* open shard relation and insert new tuple */
Relation pgDistShard = heap_open(DistShardRelationId(), RowExclusiveLock);
Relation pgDistShard = table_open(DistShardRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistShard);
HeapTuple heapTuple = heap_form_tuple(tupleDescriptor, values, isNulls);
@ -890,7 +890,7 @@ InsertShardRow(Oid relationId, uint64 shardId, char storageType,
CitusInvalidateRelcacheByRelid(relationId);
CommandCounterIncrement();
heap_close(pgDistShard, NoLock);
table_close(pgDistShard, NoLock);
}
@ -923,7 +923,7 @@ InsertShardPlacementRow(uint64 shardId, uint64 placementId,
values[Anum_pg_dist_placement_groupid - 1] = Int32GetDatum(groupId);
/* open shard placement relation and insert new tuple */
Relation pgDistPlacement = heap_open(DistPlacementRelationId(), RowExclusiveLock);
Relation pgDistPlacement = table_open(DistPlacementRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPlacement);
HeapTuple heapTuple = heap_form_tuple(tupleDescriptor, values, isNulls);
@ -933,7 +933,7 @@ InsertShardPlacementRow(uint64 shardId, uint64 placementId,
CitusInvalidateRelcacheByShardId(shardId);
CommandCounterIncrement();
heap_close(pgDistPlacement, NoLock);
table_close(pgDistPlacement, NoLock);
return placementId;
}
@ -953,7 +953,7 @@ InsertIntoPgDistPartition(Oid relationId, char distributionMethod,
bool newNulls[Natts_pg_dist_partition];
/* open system catalog and insert new tuple */
Relation pgDistPartition = heap_open(DistPartitionRelationId(), RowExclusiveLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
/* form new tuple for pg_dist_partition */
memset(newValues, 0, sizeof(newValues));
@ -991,7 +991,7 @@ InsertIntoPgDistPartition(Oid relationId, char distributionMethod,
RecordDistributedRelationDependencies(relationId);
CommandCounterIncrement();
heap_close(pgDistPartition, NoLock);
table_close(pgDistPartition, NoLock);
}
@ -1038,7 +1038,7 @@ DeletePartitionRow(Oid distributedRelationId)
ScanKeyData scanKey[1];
int scanKeyCount = 1;
Relation pgDistPartition = heap_open(DistPartitionRelationId(), RowExclusiveLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(distributedRelationId));
@ -1064,7 +1064,7 @@ DeletePartitionRow(Oid distributedRelationId)
/* increment the counter so that next command can see the row */
CommandCounterIncrement();
heap_close(pgDistPartition, NoLock);
table_close(pgDistPartition, NoLock);
}
@ -1079,7 +1079,7 @@ DeleteShardRow(uint64 shardId)
int scanKeyCount = 1;
bool indexOK = true;
Relation pgDistShard = heap_open(DistShardRelationId(), RowExclusiveLock);
Relation pgDistShard = table_open(DistShardRelationId(), RowExclusiveLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_shard_shardid,
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(shardId));
@ -1106,7 +1106,7 @@ DeleteShardRow(uint64 shardId)
CitusInvalidateRelcacheByRelid(distributedRelationId);
CommandCounterIncrement();
heap_close(pgDistShard, NoLock);
table_close(pgDistShard, NoLock);
}
@ -1122,7 +1122,7 @@ DeleteShardPlacementRow(uint64 placementId)
bool indexOK = true;
bool isNull = false;
Relation pgDistPlacement = heap_open(DistPlacementRelationId(), RowExclusiveLock);
Relation pgDistPlacement = table_open(DistPlacementRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPlacement);
ScanKeyInit(&scanKey[0], Anum_pg_dist_placement_placementid,
@ -1154,7 +1154,7 @@ DeleteShardPlacementRow(uint64 placementId)
CitusInvalidateRelcacheByShardId(shardId);
CommandCounterIncrement();
heap_close(pgDistPlacement, NoLock);
table_close(pgDistPlacement, NoLock);
}
@ -1251,7 +1251,7 @@ UpdateShardPlacementState(uint64 placementId, char shardState)
bool replace[Natts_pg_dist_placement];
bool colIsNull = false;
Relation pgDistPlacement = heap_open(DistPlacementRelationId(), RowExclusiveLock);
Relation pgDistPlacement = table_open(DistPlacementRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPlacement);
ScanKeyInit(&scanKey[0], Anum_pg_dist_placement_placementid,
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(placementId));
@ -1288,7 +1288,7 @@ UpdateShardPlacementState(uint64 placementId, char shardState)
CommandCounterIncrement();
systable_endscan(scanDescriptor);
heap_close(pgDistPlacement, NoLock);
table_close(pgDistPlacement, NoLock);
}

35
src/backend/distributed/metadata/node_metadata.c
View File

@ -41,6 +41,7 @@
#include "distributed/resource_lock.h"
#include "distributed/shardinterval_utils.h"
#include "distributed/shared_connection_stats.h"
#include "distributed/version_compat.h"
#include "distributed/worker_manager.h"
#include "distributed/worker_transaction.h"
#include "lib/stringinfo.h"
@ -798,7 +799,7 @@ UpdateNodeLocation(int32 nodeId, char *newNodeName, int32 newNodePort)
bool isnull[Natts_pg_dist_node];
bool replace[Natts_pg_dist_node];
Relation pgDistNode = heap_open(DistNodeRelationId(), RowExclusiveLock);
Relation pgDistNode = table_open(DistNodeRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistNode);
ScanKeyInit(&scanKey[0], Anum_pg_dist_node_nodeid,
@ -834,7 +835,7 @@ UpdateNodeLocation(int32 nodeId, char *newNodeName, int32 newNodePort)
CommandCounterIncrement();
systable_endscan(scanDescriptor);
heap_close(pgDistNode, NoLock);
table_close(pgDistNode, NoLock);
}
@ -978,7 +979,7 @@ FindWorkerNodeAnyCluster(const char *nodeName, int32 nodePort)
{
WorkerNode *workerNode = NULL;
Relation pgDistNode = heap_open(DistNodeRelationId(), AccessShareLock);
Relation pgDistNode = table_open(DistNodeRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistNode);
HeapTuple heapTuple = GetNodeTuple(nodeName, nodePort);
@ -987,7 +988,7 @@ FindWorkerNodeAnyCluster(const char *nodeName, int32 nodePort)
workerNode = TupleToWorkerNode(tupleDescriptor, heapTuple);
}
heap_close(pgDistNode, NoLock);
table_close(pgDistNode, NoLock);
return workerNode;
}
@ -1007,7 +1008,7 @@ ReadDistNode(bool includeNodesFromOtherClusters)
int scanKeyCount = 0;
List *workerNodeList = NIL;
Relation pgDistNode = heap_open(DistNodeRelationId(), AccessShareLock);
Relation pgDistNode = table_open(DistNodeRelationId(), AccessShareLock);
SysScanDesc scanDescriptor = systable_beginscan(pgDistNode,
InvalidOid, false,
@ -1031,7 +1032,7 @@ ReadDistNode(bool includeNodesFromOtherClusters)
}
systable_endscan(scanDescriptor);
heap_close(pgDistNode, NoLock);
table_close(pgDistNode, NoLock);
return workerNodeList;
}
@ -1208,7 +1209,7 @@ AddNodeMetadata(char *nodeName, int32 nodePort,
static WorkerNode *
SetWorkerColumn(WorkerNode *workerNode, int columnIndex, Datum value)
{
Relation pgDistNode = heap_open(DistNodeRelationId(), RowExclusiveLock);
Relation pgDistNode = table_open(DistNodeRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistNode);
HeapTuple heapTuple = GetNodeTuple(workerNode->workerName, workerNode->workerPort);
@ -1261,7 +1262,7 @@ SetWorkerColumn(WorkerNode *workerNode, int columnIndex, Datum value)
WorkerNode *newWorkerNode = TupleToWorkerNode(tupleDescriptor, heapTuple);
heap_close(pgDistNode, NoLock);
table_close(pgDistNode, NoLock);
/* we also update the column at worker nodes */
SendCommandToWorkersWithMetadata(metadataSyncCommand);
@ -1305,7 +1306,7 @@ SetNodeState(char *nodeName, int nodePort, bool isActive)
static HeapTuple
GetNodeTuple(const char *nodeName, int32 nodePort)
{
Relation pgDistNode = heap_open(DistNodeRelationId(), AccessShareLock);
Relation pgDistNode = table_open(DistNodeRelationId(), AccessShareLock);
const int scanKeyCount = 2;
const bool indexOK = false;
@ -1326,7 +1327,7 @@ GetNodeTuple(const char *nodeName, int32 nodePort)
}
systable_endscan(scanDescriptor);
heap_close(pgDistNode, NoLock);
table_close(pgDistNode, NoLock);
return nodeTuple;
}
@ -1448,7 +1449,7 @@ InsertNodeRow(int nodeid, char *nodeName, int32 nodePort, NodeMetadata *nodeMeta
values[Anum_pg_dist_node_shouldhaveshards - 1] = BoolGetDatum(
nodeMetadata->shouldHaveShards);
Relation pgDistNode = heap_open(DistNodeRelationId(), RowExclusiveLock);
Relation pgDistNode = table_open(DistNodeRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistNode);
HeapTuple heapTuple = heap_form_tuple(tupleDescriptor, values, isNulls);
@ -1461,7 +1462,7 @@ InsertNodeRow(int nodeid, char *nodeName, int32 nodePort, NodeMetadata *nodeMeta
CommandCounterIncrement();
/* close relation */
heap_close(pgDistNode, NoLock);
table_close(pgDistNode, NoLock);
}
@ -1475,7 +1476,7 @@ DeleteNodeRow(char *nodeName, int32 nodePort)
bool indexOK = false;
ScanKeyData scanKey[2];
Relation pgDistNode = heap_open(DistNodeRelationId(), RowExclusiveLock);
Relation pgDistNode = table_open(DistNodeRelationId(), RowExclusiveLock);
/*
* simple_heap_delete() expects that the caller has at least an
@ -1510,8 +1511,8 @@ DeleteNodeRow(char *nodeName, int32 nodePort)
/* increment the counter so that next command won't see the row */
CommandCounterIncrement();
heap_close(replicaIndex, AccessShareLock);
heap_close(pgDistNode, NoLock);
table_close(replicaIndex, AccessShareLock);
table_close(pgDistNode, NoLock);
}
@ -1628,7 +1629,7 @@ UnsetMetadataSyncedForAll(void)
* pg_dist_node in different orders. To protect against deadlock, we
* get an exclusive lock here.
*/
Relation relation = heap_open(DistNodeRelationId(), ExclusiveLock);
Relation relation = table_open(DistNodeRelationId(), ExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(relation);
ScanKeyInit(&scanKey[0], Anum_pg_dist_node_hasmetadata,
BTEqualStrategyNumber, F_BOOLEQ, BoolGetDatum(true));
@ -1676,7 +1677,7 @@ UnsetMetadataSyncedForAll(void)
systable_endscan(scanDescriptor);
CatalogCloseIndexes(indstate);
heap_close(relation, NoLock);
table_close(relation, NoLock);
return updatedAtLeastOne;
}

5
src/backend/distributed/operations/node_protocol.c
View File

@ -48,6 +48,7 @@
#include "distributed/metadata_sync.h"
#include "distributed/namespace_utils.h"
#include "distributed/pg_dist_shard.h"
#include "distributed/version_compat.h"
#include "distributed/worker_manager.h"
#include "foreign/foreign.h"
#include "lib/stringinfo.h"
@ -648,7 +649,7 @@ GetTableIndexAndConstraintCommands(Oid relationId)
PushOverrideEmptySearchPath(CurrentMemoryContext);
/* open system catalog and scan all indexes that belong to this table */
Relation pgIndex = heap_open(IndexRelationId, AccessShareLock);
Relation pgIndex = table_open(IndexRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_index_indrelid,
BTEqualStrategyNumber, F_OIDEQ, relationId);
@ -696,7 +697,7 @@ GetTableIndexAndConstraintCommands(Oid relationId)
/* clean up scan and close system catalog */
systable_endscan(scanDescriptor);
heap_close(pgIndex, AccessShareLock);
table_close(pgIndex, AccessShareLock);
/* revert back to original search_path */
PopOverrideSearchPath();

4
src/backend/distributed/planner/deparse_shard_query.c
View File

@ -358,7 +358,7 @@ UpdateRelationsToLocalShardTables(Node *node, List *relationShardList)
static void
ConvertRteToSubqueryWithEmptyResult(RangeTblEntry *rte)
{
Relation relation = heap_open(rte->relid, NoLock);
Relation relation = table_open(rte->relid, NoLock);
TupleDesc tupleDescriptor = RelationGetDescr(relation);
int columnCount = tupleDescriptor->natts;
List *targetList = NIL;
@ -388,7 +388,7 @@ ConvertRteToSubqueryWithEmptyResult(RangeTblEntry *rte)
targetList = lappend(targetList, targetEntry);
}
heap_close(relation, NoLock);
table_close(relation, NoLock);
FromExpr *joinTree = makeNode(FromExpr);
joinTree->quals = makeBoolConst(false, false);

4
src/backend/distributed/planner/insert_select_planner.c
View File

@ -1494,7 +1494,7 @@ AddInsertSelectCasts(List *insertTargetList, List *selectTargetList,
*/
Assert(list_length(insertTargetList) <= list_length(selectTargetList));
Relation distributedRelation = heap_open(targetRelationId, RowExclusiveLock);
Relation distributedRelation = table_open(targetRelationId, RowExclusiveLock);
TupleDesc destTupleDescriptor = RelationGetDescr(distributedRelation);
int targetEntryIndex = 0;
@ -1579,7 +1579,7 @@ AddInsertSelectCasts(List *insertTargetList, List *selectTargetList,
selectTargetEntry->resno = entryResNo++;
}
heap_close(distributedRelation, NoLock);
table_close(distributedRelation, NoLock);
return selectTargetList;
}

4
src/backend/distributed/planner/multi_logical_optimizer.c
View File

@ -3564,7 +3564,7 @@ AggregateFunctionOid(const char *functionName, Oid inputType)
ScanKeyData scanKey[1];
int scanKeyCount = 1;
Relation procRelation = heap_open(ProcedureRelationId, AccessShareLock);
Relation procRelation = table_open(ProcedureRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_proc_proname,
BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(functionName));
@ -3605,7 +3605,7 @@ AggregateFunctionOid(const char *functionName, Oid inputType)
}
systable_endscan(scanDescriptor);
heap_close(procRelation, AccessShareLock);
table_close(procRelation, AccessShareLock);
return functionOid;
}

4
src/backend/distributed/planner/tdigest_extension.c
View File

@ -32,7 +32,7 @@ TDigestExtensionSchema()
Form_pg_extension extensionForm = NULL;
Oid tdigestExtensionSchema = InvalidOid;
Relation relation = heap_open(ExtensionRelationId, AccessShareLock);
Relation relation = table_open(ExtensionRelationId, AccessShareLock);
ScanKeyInit(&entry[0],
Anum_pg_extension_extname,
@ -57,7 +57,7 @@ TDigestExtensionSchema()
systable_endscan(scandesc);
heap_close(relation, AccessShareLock);
table_close(relation, AccessShareLock);
return tdigestExtensionSchema;
}

8
src/backend/distributed/transaction/transaction_recovery.c
View File

@ -95,7 +95,7 @@ LogTransactionRecord(int32 groupId, char *transactionName)
values[Anum_pg_dist_transaction_gid - 1] = CStringGetTextDatum(transactionName);
/* open transaction relation and insert new tuple */
Relation pgDistTransaction = heap_open(DistTransactionRelationId(), RowExclusiveLock);
Relation pgDistTransaction = table_open(DistTransactionRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistTransaction);
HeapTuple heapTuple = heap_form_tuple(tupleDescriptor, values, isNulls);
@ -105,7 +105,7 @@ LogTransactionRecord(int32 groupId, char *transactionName)
CommandCounterIncrement();
/* close relation and invalidate previous cache entry */
heap_close(pgDistTransaction, NoLock);
table_close(pgDistTransaction, NoLock);
}
@ -171,7 +171,7 @@ RecoverWorkerTransactions(WorkerNode *workerNode)
MemoryContext oldContext = MemoryContextSwitchTo(localContext);
/* take table lock first to avoid running concurrently */
Relation pgDistTransaction = heap_open(DistTransactionRelationId(),
Relation pgDistTransaction = table_open(DistTransactionRelationId(),
ShareUpdateExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistTransaction);
@ -344,7 +344,7 @@ RecoverWorkerTransactions(WorkerNode *workerNode)
}
systable_endscan(scanDescriptor);
heap_close(pgDistTransaction, NoLock);
table_close(pgDistTransaction, NoLock);
if (!recoveryFailed)
{

35
src/backend/distributed/utils/colocation_utils.c
View File

@ -29,6 +29,7 @@
#include "distributed/pg_dist_colocation.h"
#include "distributed/resource_lock.h"
#include "distributed/shardinterval_utils.h"
#include "distributed/version_compat.h"
#include "distributed/worker_protocol.h"
#include "distributed/worker_transaction.h"
#include "storage/lmgr.h"
@ -154,7 +155,7 @@ BreakColocation(Oid sourceRelationId)
* can be sure that there will no modifications on the colocation table
* until this transaction is committed.
*/
Relation pgDistColocation = heap_open(DistColocationRelationId(), ExclusiveLock);
Relation pgDistColocation = table_open(DistColocationRelationId(), ExclusiveLock);
uint32 newColocationId = GetNextColocationId();
UpdateRelationColocationGroup(sourceRelationId, newColocationId);
@ -162,7 +163,7 @@ BreakColocation(Oid sourceRelationId)
/* if there is not any remaining table in the colocation group, delete it */
DeleteColocationGroupIfNoTablesBelong(sourceRelationId);
heap_close(pgDistColocation, NoLock);
table_close(pgDistColocation, NoLock);
}
@ -248,7 +249,7 @@ MarkTablesColocated(Oid sourceRelationId, Oid targetRelationId)
* can be sure that there will no modifications on the colocation table
* until this transaction is committed.
*/
Relation pgDistColocation = heap_open(DistColocationRelationId(), ExclusiveLock);
Relation pgDistColocation = table_open(DistColocationRelationId(), ExclusiveLock);
/* check if shard placements are colocated */
ErrorIfShardPlacementsNotColocated(sourceRelationId, targetRelationId);
@ -271,7 +272,7 @@ MarkTablesColocated(Oid sourceRelationId, Oid targetRelationId)
/* if there is not any remaining table in the colocation group, delete it */
DeleteColocationGroupIfNoTablesBelong(targetColocationId);
heap_close(pgDistColocation, NoLock);
table_close(pgDistColocation, NoLock);
}
@ -514,7 +515,7 @@ ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType,
ScanKeyData scanKey[4];
bool indexOK = true;
Relation pgDistColocation = heap_open(DistColocationRelationId(), AccessShareLock);
Relation pgDistColocation = table_open(DistColocationRelationId(), AccessShareLock);
/* set scan arguments */
ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_distributioncolumntype,
@ -541,7 +542,7 @@ ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType,
}
systable_endscan(scanDescriptor);
heap_close(pgDistColocation, AccessShareLock);
table_close(pgDistColocation, AccessShareLock);
return colocationId;
}
@ -574,7 +575,7 @@ CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionCol
ObjectIdGetDatum(distributionColumnCollation);
/* open colocation relation and insert the new tuple */
Relation pgDistColocation = heap_open(DistColocationRelationId(), RowExclusiveLock);
Relation pgDistColocation = table_open(DistColocationRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistColocation);
HeapTuple heapTuple = heap_form_tuple(tupleDescriptor, values, isNulls);
@ -583,7 +584,7 @@ CreateColocationGroup(int shardCount, int replicationFactor, Oid distributionCol
/* increment the counter so that next command can see the row */
CommandCounterIncrement();
heap_close(pgDistColocation, RowExclusiveLock);
table_close(pgDistColocation, RowExclusiveLock);
return colocationId;
}
@ -716,7 +717,7 @@ UpdateRelationColocationGroup(Oid distributedRelationId, uint32 colocationId)
bool isNull[Natts_pg_dist_partition];
bool replace[Natts_pg_dist_partition];
Relation pgDistPartition = heap_open(DistPartitionRelationId(), RowExclusiveLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid,
@ -753,7 +754,7 @@ UpdateRelationColocationGroup(Oid distributedRelationId, uint32 colocationId)
CommandCounterIncrement();
systable_endscan(scanDescriptor);
heap_close(pgDistPartition, NoLock);
table_close(pgDistPartition, NoLock);
bool shouldSyncMetadata = ShouldSyncTableMetadata(distributedRelationId);
if (shouldSyncMetadata)
@ -882,7 +883,7 @@ ColocationGroupTableList(Oid colocationId)
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_colocationid,
BTEqualStrategyNumber, F_INT4EQ, ObjectIdGetDatum(colocationId));
Relation pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
SysScanDesc scanDescriptor = systable_beginscan(pgDistPartition,
DistPartitionColocationidIndexId(),
@ -901,7 +902,7 @@ ColocationGroupTableList(Oid colocationId)
}
systable_endscan(scanDescriptor);
heap_close(pgDistPartition, AccessShareLock);
table_close(pgDistPartition, AccessShareLock);
return colocatedTableList;
}
@ -997,7 +998,7 @@ ColocatedTableId(Oid colocationId)
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_colocationid,
BTEqualStrategyNumber, F_INT4EQ, ObjectIdGetDatum(colocationId));
Relation pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
SysScanDesc scanDescriptor = systable_beginscan(pgDistPartition,
DistPartitionColocationidIndexId(),
@ -1034,7 +1035,7 @@ ColocatedTableId(Oid colocationId)
}
systable_endscan(scanDescriptor);
heap_close(pgDistPartition, AccessShareLock);
table_close(pgDistPartition, AccessShareLock);
return colocatedTableId;
}
@ -1085,7 +1086,7 @@ DeleteColocationGroup(uint32 colocationId)
ScanKeyData scanKey[1];
bool indexOK = false;
Relation pgDistColocation = heap_open(DistColocationRelationId(), RowExclusiveLock);
Relation pgDistColocation = table_open(DistColocationRelationId(), RowExclusiveLock);
ScanKeyInit(&scanKey[0], Anum_pg_dist_colocation_colocationid,
BTEqualStrategyNumber, F_INT4EQ, UInt32GetDatum(colocationId));
@ -1108,9 +1109,9 @@ DeleteColocationGroup(uint32 colocationId)
CitusInvalidateRelcacheByRelid(DistColocationRelationId());
CommandCounterIncrement();
heap_close(replicaIndex, AccessShareLock);
table_close(replicaIndex, AccessShareLock);
}
systable_endscan(scanDescriptor);
heap_close(pgDistColocation, RowExclusiveLock);
table_close(pgDistColocation, RowExclusiveLock);
}

4
src/backend/distributed/utils/foreign_key_relationship.c
View File

@ -296,7 +296,7 @@ PopulateAdjacencyLists(void)
Oid prevReferencedOid = InvalidOid;
List *frelEdgeList = NIL;
Relation pgConstraint = heap_open(ConstraintRelationId, AccessShareLock);
Relation pgConstraint = table_open(ConstraintRelationId, AccessShareLock);
ScanKeyInit(&scanKey[0], Anum_pg_constraint_contype, BTEqualStrategyNumber, F_CHAREQ,
CharGetDatum(CONSTRAINT_FOREIGN));
@ -345,7 +345,7 @@ PopulateAdjacencyLists(void)
}
systable_endscan(scanDescriptor);
heap_close(pgConstraint, AccessShareLock);
table_close(pgConstraint, AccessShareLock);
}

21
src/backend/distributed/utils/multi_partitioning_utils.c
View File

@ -21,6 +21,7 @@
#include "distributed/coordinator_protocol.h"
#include "distributed/multi_partitioning_utils.h"
#include "distributed/shardinterval_utils.h"
#include "distributed/version_compat.h"
#include "lib/stringinfo.h"
#include "nodes/pg_list.h"
#include "pgstat.h"
@ -58,7 +59,7 @@ PartitionedTable(Oid relationId)
}
/* keep the lock */
heap_close(rel, NoLock);
table_close(rel, NoLock);
return partitionedTable;
}
@ -87,7 +88,7 @@ PartitionedTableNoLock(Oid relationId)
}
/* keep the lock */
heap_close(rel, NoLock);
table_close(rel, NoLock);
return partitionedTable;
}
@ -110,7 +111,7 @@ PartitionTable(Oid relationId)
bool partitionTable = rel->rd_rel->relispartition;
/* keep the lock */
heap_close(rel, NoLock);
table_close(rel, NoLock);
return partitionTable;
}
@ -135,7 +136,7 @@ PartitionTableNoLock(Oid relationId)
bool partitionTable = rel->rd_rel->relispartition;
/* keep the lock */
heap_close(rel, NoLock);
table_close(rel, NoLock);
return partitionTable;
}
@ -185,7 +186,7 @@ IsChildTable(Oid relationId)
HeapTuple inheritsTuple = NULL;
bool tableInherits = false;
Relation pgInherits = heap_open(InheritsRelationId, AccessShareLock);
Relation pgInherits = table_open(InheritsRelationId, AccessShareLock);
ScanKeyInit(&key[0], Anum_pg_inherits_inhrelid,
BTEqualStrategyNumber, F_OIDEQ,
@ -207,7 +208,7 @@ IsChildTable(Oid relationId)
}
systable_endscan(scan);
heap_close(pgInherits, AccessShareLock);
table_close(pgInherits, AccessShareLock);
if (tableInherits && PartitionTable(relationId))
{
@ -229,7 +230,7 @@ IsParentTable(Oid relationId)
ScanKeyData key[1];
bool tableInherited = false;
Relation pgInherits = heap_open(InheritsRelationId, AccessShareLock);
Relation pgInherits = table_open(InheritsRelationId, AccessShareLock);
ScanKeyInit(&key[0], Anum_pg_inherits_inhparent,
BTEqualStrategyNumber, F_OIDEQ,
@ -243,7 +244,7 @@ IsParentTable(Oid relationId)
tableInherited = true;
}
systable_endscan(scan);
heap_close(pgInherits, AccessShareLock);
table_close(pgInherits, AccessShareLock);
if (tableInherited && PartitionedTable(relationId))
{
@ -277,7 +278,7 @@ PartitionParentOid(Oid partitionOid)
List *
PartitionList(Oid parentRelationId)
{
Relation rel = heap_open(parentRelationId, AccessShareLock);
Relation rel = table_open(parentRelationId, AccessShareLock);
List *partitionList = NIL;
@ -298,7 +299,7 @@ PartitionList(Oid parentRelationId)
}
/* keep the lock */
heap_close(rel, NoLock);
table_close(rel, NoLock);
return partitionList;
}

4
src/backend/distributed/utils/statistics_collection.c
View File

@ -197,14 +197,14 @@ DistributedTablesSize(List *distTableOids)
if (PartitionMethod(relationId) == DISTRIBUTE_BY_HASH &&
!SingleReplicatedTable(relationId))
{
heap_close(relation, AccessShareLock);
table_close(relation, AccessShareLock);
continue;
}
Datum tableSizeDatum = DirectFunctionCall1(citus_table_size,
ObjectIdGetDatum(relationId));
totalSize += DatumGetInt64(tableSizeDatum);
heap_close(relation, AccessShareLock);
table_close(relation, AccessShareLock);
}
return totalSize;

4
src/backend/distributed/worker/worker_merge_protocol.c
View File

@ -278,7 +278,7 @@ worker_cleanup_job_schema_cache(PG_FUNCTION_ARGS)
CheckCitusVersion(ERROR);
pgNamespace = heap_open(NamespaceRelationId, AccessExclusiveLock);
pgNamespace = table_open(NamespaceRelationId, AccessExclusiveLock);
#if PG_VERSION_NUM >= PG_VERSION_12
scanDescriptor = table_beginscan_catalog(pgNamespace, scanKeyCount, scanKey);
#else
@ -304,7 +304,7 @@ worker_cleanup_job_schema_cache(PG_FUNCTION_ARGS)
}
heap_endscan(scanDescriptor);
heap_close(pgNamespace, AccessExclusiveLock);
table_close(pgNamespace, AccessExclusiveLock);
PG_RETURN_VOID();
}

4
src/include/distributed/version_compat.h
View File

@ -94,6 +94,10 @@ FileCompatFromFileStart(File fileDesc)
#else /* pre PG12 */
#define table_open(r, l) heap_open(r, l)
#define table_openrv(r, l) heap_openrv(r, l)
#define table_openrv_extended(r, l, m) heap_openrv_extended(r, l, m)
#define table_close(r, l) heap_close(r, l)
#define QTW_EXAMINE_RTES_BEFORE QTW_EXAMINE_RTES
#define MakeSingleTupleTableSlotCompat(tupleDesc, tts_opts) \
MakeSingleTupleTableSlot(tupleDesc)