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

added some missing object address callbacks (#6056)

pull/6090/head
aykut-bozkurt 2022-07-27 17:36:04 +03:00 committed by GitHub
parent 1259d83511
commit b08e5ec29d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 210 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
src/backend/distributed/commands/common.c
View File

@ -14,6 +14,8 @@
#include "postgres.h"
#include "catalog/objectaddress.h"
#include "catalog/pg_ts_config.h"
#include "catalog/pg_ts_dict.h"
#include "nodes/parsenodes.h"
#include "tcop/utility.h"
@ -287,3 +289,53 @@ PreprocessDropDistributedObjectStmt(Node *node, const char *queryString,
return NodeDDLTaskList(NON_COORDINATOR_NODES, commands);
}
/*
* DropTextSearchDictObjectAddress returns list of object addresses in
* the drop tsdict statement.
*/
List *
DropTextSearchDictObjectAddress(Node *node, bool missing_ok)
{
DropStmt *stmt = castNode(DropStmt, node);
List *objectAddresses = NIL;
List *objNameList = NIL;
foreach_ptr(objNameList, stmt->objects)
{
Oid tsdictOid = get_ts_dict_oid(objNameList, missing_ok);
ObjectAddress *objectAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*objectAddress, TSDictionaryRelationId, tsdictOid);
objectAddresses = lappend(objectAddresses, objectAddress);
}
return objectAddresses;
}
/*
* DropTextSearchConfigObjectAddress returns list of object addresses in
* the drop tsconfig statement.
*/
List *
DropTextSearchConfigObjectAddress(Node *node, bool missing_ok)
{
DropStmt *stmt = castNode(DropStmt, node);
List *objectAddresses = NIL;
List *objNameList = NIL;
foreach_ptr(objNameList, stmt->objects)
{
Oid tsconfigOid = get_ts_config_oid(objNameList, missing_ok);
ObjectAddress *objectAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*objectAddress, TSConfigRelationId, tsconfigOid);
objectAddresses = lappend(objectAddresses, objectAddress);
}
return objectAddresses;
}

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

@ -291,7 +291,7 @@ static DistributeObjectOps Any_Reindex = {
.qualify = NULL,
.preprocess = PreprocessReindexStmt,
.postprocess = NULL,
.address = NULL,
.address = ReindexStmtObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps Any_Rename = {
@ -531,7 +531,7 @@ static DistributeObjectOps View_Drop = {
.qualify = QualifyDropViewStmt,
.preprocess = PreprocessDropViewStmt,
.postprocess = NULL,
.address = NULL,
.address = DropViewStmtObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps Function_Rename = {
@ -655,7 +655,7 @@ static DistributeObjectOps Sequence_Drop = {
.qualify = QualifyDropSequenceStmt,
.preprocess = PreprocessDropSequenceStmt,
.postprocess = NULL,
.address = NULL,
.address = SequenceDropStmtObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps Sequence_Grant = {
@ -724,7 +724,7 @@ static DistributeObjectOps TextSearchConfig_Drop = {
.qualify = QualifyDropTextSearchConfigurationStmt,
.preprocess = PreprocessDropDistributedObjectStmt,
.postprocess = NULL,
.address = NULL,
.address = DropTextSearchConfigObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps TextSearchConfig_Rename = {
@ -786,7 +786,7 @@ static DistributeObjectOps TextSearchDict_Drop = {
.qualify = QualifyDropTextSearchDictionaryStmt,
.preprocess = PreprocessDropDistributedObjectStmt,
.postprocess = NULL,
.address = NULL,
.address = DropTextSearchDictObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps TextSearchDict_Rename = {
@ -903,7 +903,7 @@ static DistributeObjectOps Statistics_Drop = {
.qualify = QualifyDropStatisticsStmt,
.preprocess = PreprocessDropStatisticsStmt,
.postprocess = NULL,
.address = NULL,
.address = DropStatisticsObjectAddress,
.markDistributed = false,
};
static DistributeObjectOps Statistics_Rename = {

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

@ -75,7 +75,7 @@ static void RangeVarCallbackForReindexIndex(const RangeVar *rel, Oid relOid, Oid
static void ErrorIfUnsupportedIndexStmt(IndexStmt *createIndexStatement);
static void ErrorIfUnsupportedDropIndexStmt(DropStmt *dropIndexStatement);
static List * DropIndexTaskList(Oid relationId, Oid indexId, DropStmt *dropStmt);
static Oid ReindexStmtFindRelationOid(ReindexStmt *reindexStmt, bool missingOk);
/*
* This struct defines the state for the callback for drop statements.
@ -522,6 +522,49 @@ GetCreateIndexRelationLockMode(IndexStmt *createIndexStatement)
}
/*
* ReindexStmtFindRelationOid returns the oid of the relation on which the index exist
* if the object is an index in the reindex stmt. It returns the oid of the relation
* if the object is a table in the reindex stmt. It also acquires the relevant lock
* for the statement.
*/
static Oid
ReindexStmtFindRelationOid(ReindexStmt *reindexStmt, bool missingOk)
{
Assert(reindexStmt->relation != NULL);
Assert(reindexStmt->kind == REINDEX_OBJECT_INDEX ||
reindexStmt->kind == REINDEX_OBJECT_TABLE);
Oid relationId = InvalidOid;
LOCKMODE lockmode = IsReindexWithParam_compat(reindexStmt, "concurrently") ?
ShareUpdateExclusiveLock : AccessExclusiveLock;
if (reindexStmt->kind == REINDEX_OBJECT_INDEX)
{
struct ReindexIndexCallbackState state;
state.concurrent = IsReindexWithParam_compat(reindexStmt,
"concurrently");
state.locked_table_oid = InvalidOid;
Oid indOid = RangeVarGetRelidExtended(reindexStmt->relation, lockmode,
(missingOk) ? RVR_MISSING_OK : 0,
RangeVarCallbackForReindexIndex,
&state);
relationId = IndexGetRelation(indOid, missingOk);
}
else
{
relationId = RangeVarGetRelidExtended(reindexStmt->relation, lockmode,
(missingOk) ? RVR_MISSING_OK : 0,
RangeVarCallbackOwnsTable, NULL);
}
return relationId;
}
/*
* PreprocessReindexStmt determines whether a given REINDEX statement involves
* a distributed table. If so (and if the statement does not use unsupported
@ -544,36 +587,17 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand,
*/
if (reindexStatement->relation != NULL)
{
Relation relation = NULL;
Oid relationId = InvalidOid;
LOCKMODE lockmode = IsReindexWithParam_compat(reindexStatement, "concurrently") ?
ShareUpdateExclusiveLock : AccessExclusiveLock;
Oid relationId = ReindexStmtFindRelationOid(reindexStatement, false);
MemoryContext relationContext = NULL;
Assert(reindexStatement->kind == REINDEX_OBJECT_INDEX ||
reindexStatement->kind == REINDEX_OBJECT_TABLE);
Relation relation = NULL;
if (reindexStatement->kind == REINDEX_OBJECT_INDEX)
{
struct ReindexIndexCallbackState state;
state.concurrent = IsReindexWithParam_compat(reindexStatement,
"concurrently");
state.locked_table_oid = InvalidOid;
Oid indOid = RangeVarGetRelidExtended(reindexStatement->relation,
lockmode, 0,
RangeVarCallbackForReindexIndex,
&state);
Oid indOid = RangeVarGetRelid(reindexStatement->relation, NoLock, 0);
relation = index_open(indOid, NoLock);
relationId = IndexGetRelation(indOid, false);
}
else
{
RangeVarGetRelidExtended(reindexStatement->relation, lockmode, 0,
RangeVarCallbackOwnsTable, NULL);
relation = table_openrv(reindexStatement->relation, NoLock);
relationId = RelationGetRelid(relation);
}
bool isCitusRelation = IsCitusTable(relationId);
@ -628,6 +652,23 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand,
}
/*
* ReindexStmtObjectAddress returns list of object addresses in the reindex
* statement.
*/
List *
ReindexStmtObjectAddress(Node *stmt, bool missing_ok)
{
ReindexStmt *reindexStatement = castNode(ReindexStmt, stmt);
Oid relationId = ReindexStmtFindRelationOid(reindexStatement, missing_ok);
ObjectAddress *objectAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*objectAddress, RelationRelationId, relationId);
return list_make1(objectAddress);
}
/*
* PreprocessDropIndexStmt determines whether a given DROP INDEX statement involves
* a distributed table. If so (and if the statement does not use unsupported

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

@ -313,6 +313,34 @@ PreprocessDropSequenceStmt(Node *node, const char *queryString,
}
/*
* SequenceDropStmtObjectAddress returns list of object addresses in the drop sequence
* statement.
*/
List *
SequenceDropStmtObjectAddress(Node *stmt, bool missing_ok)
{
DropStmt *dropSeqStmt = castNode(DropStmt, stmt);
List *objectAddresses = NIL;
List *droppingSequencesList = dropSeqStmt->objects;
List *objectNameList = NULL;
foreach_ptr(objectNameList, droppingSequencesList)
{
RangeVar *seq = makeRangeVarFromNameList(objectNameList);
Oid seqOid = RangeVarGetRelid(seq, AccessShareLock, missing_ok);
ObjectAddress *objectAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*objectAddress, SequenceRelationId, seqOid);
objectAddresses = lappend(objectAddresses, objectAddress);
}
return objectAddresses;
}
/*
* PreprocessRenameSequenceStmt is called when the user is renaming a sequence. The invocation
* happens before the statement is applied locally.

27
src/backend/distributed/commands/statistics.c
View File

@ -210,6 +210,33 @@ PreprocessDropStatisticsStmt(Node *node, const char *queryString,
}
/*
* DropStatisticsObjectAddress returns list of object addresses in the drop statistics
* statement.
*/
List *
DropStatisticsObjectAddress(Node *node, bool missing_ok)
{
DropStmt *dropStatisticsStmt = castNode(DropStmt, node);
Assert(dropStatisticsStmt->removeType == OBJECT_STATISTIC_EXT);
List *objectAddresses = NIL;
List *objectNameList = NULL;
foreach_ptr(objectNameList, dropStatisticsStmt->objects)
{
Oid statsOid = get_statistics_object_oid(objectNameList,
dropStatisticsStmt->missing_ok);
ObjectAddress *objectAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*objectAddress, StatisticExtRelationId, statsOid);
objectAddresses = lappend(objectAddresses, objectAddress);
}
return objectAddresses;
}
/*
* PreprocessAlterStatisticsRenameStmt is called during the planning phase for
* ALTER STATISTICS RENAME.

27
src/backend/distributed/commands/view.c
View File

@ -221,6 +221,33 @@ PreprocessDropViewStmt(Node *node, const char *queryString, ProcessUtilityContex
}
/*
* DropViewStmtObjectAddress returns list of object addresses in the drop view
* statement.
*/
List *
DropViewStmtObjectAddress(Node *stmt, bool missing_ok)
{
DropStmt *dropStmt = castNode(DropStmt, stmt);
List *objectAddresses = NIL;
List *possiblyQualifiedViewName = NULL;
foreach_ptr(possiblyQualifiedViewName, dropStmt->objects)
{
RangeVar *viewRangeVar = makeRangeVarFromNameList(possiblyQualifiedViewName);
Oid viewOid = RangeVarGetRelid(viewRangeVar, AccessShareLock,
missing_ok);
ObjectAddress *objectAddress = palloc0(sizeof(ObjectAddress));
ObjectAddressSet(*objectAddress, RelationRelationId, viewOid);
objectAddresses = lappend(objectAddresses, objectAddress);
}
return objectAddresses;
}
/*
* FilterNameListForDistributedViews takes a list of view names and filters against the
* views that are distributed.

6
src/include/distributed/commands.h
View File

@ -147,6 +147,8 @@ extern List * PostprocessAlterDistributedObjectStmt(Node *stmt, const char *quer
extern List * PreprocessDropDistributedObjectStmt(Node *node, const char *queryString,
ProcessUtilityContext
processUtilityContext);
extern List * DropTextSearchConfigObjectAddress(Node *node, bool missing_ok);
extern List * DropTextSearchDictObjectAddress(Node *node, bool missing_ok);
/* index.c */
typedef void (*PGIndexProcessor)(Form_pg_index, List **, int);
@ -328,6 +330,7 @@ extern LOCKMODE GetCreateIndexRelationLockMode(IndexStmt *createIndexStatement);
extern List * PreprocessReindexStmt(Node *ReindexStatement,
const char *ReindexCommand,
ProcessUtilityContext processUtilityContext);
extern List * ReindexStmtObjectAddress(Node *stmt, bool missing_ok);
extern List * PreprocessDropIndexStmt(Node *dropIndexStatement,
const char *dropIndexCommand,
ProcessUtilityContext processUtilityContext);
@ -417,6 +420,7 @@ extern List * PreprocessAlterSequenceOwnerStmt(Node *node, const char *queryStri
extern List * PostprocessAlterSequenceOwnerStmt(Node *node, const char *queryString);
extern List * PreprocessDropSequenceStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext);
extern List * SequenceDropStmtObjectAddress(Node *stmt, bool missing_ok);
extern List * PreprocessRenameSequenceStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext);
extern List * PreprocessGrantOnSequenceStmt(Node *node, const char *queryString,
@ -439,6 +443,7 @@ extern List * PostprocessCreateStatisticsStmt(Node *node, const char *queryStrin
extern List * CreateStatisticsStmtObjectAddress(Node *node, bool missingOk);
extern List * PreprocessDropStatisticsStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext);
extern List * DropStatisticsObjectAddress(Node *node, bool missing_ok);
extern List * PreprocessAlterStatisticsRenameStmt(Node *node, const char *queryString,
ProcessUtilityContext
processUtilityContext);
@ -569,6 +574,7 @@ extern List * ViewStmtObjectAddress(Node *node, bool missing_ok);
extern List * AlterViewStmtObjectAddress(Node *node, bool missing_ok);
extern List * PreprocessDropViewStmt(Node *node, const char *queryString,
ProcessUtilityContext processUtilityContext);
extern List * DropViewStmtObjectAddress(Node *node, bool missing_ok);
extern char * CreateViewDDLCommand(Oid viewOid);
extern List * GetViewCreationCommandsOfTable(Oid relationId);
extern List * GetViewCreationTableDDLCommandsOfTable(Oid relationId);