CI fix and reindent

niupre/DeferredDrop
Nitish Upreti 2022-08-13 21:24:27 -07:00
parent 96b389d8ef
commit b165be402f
15 changed files with 121 additions and 93 deletions

View File

@ -162,7 +162,8 @@ PreprocessGrantStmt(Node *node, const char *queryString,
{ {
/* Propogate latest policies issue on deleted shards to avoid any potential issues */ /* Propogate latest policies issue on deleted shards to avoid any potential issues */
bool includeOrphanedShards = true; bool includeOrphanedShards = true;
ddlJob->taskList = DDLTaskList(relationId, ddlString.data, includeOrphanedShards); ddlJob->taskList = DDLTaskList(relationId, ddlString.data,
includeOrphanedShards);
} }
ddlJobs = lappend(ddlJobs, ddlJob); ddlJobs = lappend(ddlJobs, ddlJob);

View File

@ -481,8 +481,8 @@ PreprocessAlterStatisticsOwnerStmt(Node *node, const char *queryString,
ddlJob->metadataSyncCommand = ddlCommand; ddlJob->metadataSyncCommand = ddlCommand;
/* Propogate owner changes on deleted shards to avoid any potential issues */ /* Propogate owner changes on deleted shards to avoid any potential issues */
// TODO(niupre): Can this cause failure when we try to drop orphaned table? /* TODO(niupre): Can this cause failure when we try to drop orphaned table? */
// If this is the case, do we need to allow CREATE STATS as well? /* If this is the case, do we need to allow CREATE STATS as well? */
bool includeOrphanedShards = true; bool includeOrphanedShards = true;
ddlJob->taskList = DDLTaskList(relationId, ddlCommand, includeOrphanedShards); ddlJob->taskList = DDLTaskList(relationId, ddlCommand, includeOrphanedShards);

View File

@ -1165,7 +1165,8 @@ PreprocessAlterTableStmt(Node *node, const char *alterTableCommand,
/* Propogate latest updates issue on deleted shards to avoid any potential issues */ /* Propogate latest updates issue on deleted shards to avoid any potential issues */
bool includeOrphanedShards = true; bool includeOrphanedShards = true;
ddlJob->taskList = DDLTaskList(leftRelationId, sqlForTaskList, includeOrphanedShards); ddlJob->taskList = DDLTaskList(leftRelationId, sqlForTaskList,
includeOrphanedShards);
if (!propagateCommandToWorkers) if (!propagateCommandToWorkers)
{ {
ddlJob->taskList = NIL; ddlJob->taskList = NIL;
@ -1841,7 +1842,8 @@ PreprocessAlterTableSchemaStmt(Node *node, const char *queryString,
/* Propogate latest schema on deleted shards to avoid any potential issues */ /* Propogate latest schema on deleted shards to avoid any potential issues */
bool includeOrphanedShards = true; bool includeOrphanedShards = true;
ddlJob->taskList = DDLTaskList(relationId, ddlJob->metadataSyncCommand, includeOrphanedShards); ddlJob->taskList = DDLTaskList(relationId, ddlJob->metadataSyncCommand,
includeOrphanedShards);
return list_make1(ddlJob); return list_make1(ddlJob);
} }

View File

@ -289,7 +289,8 @@ static void CachedRelationNamespaceLookupExtended(const char *relationName,
bool missing_ok); bool missing_ok);
static ShardPlacement * ResolveGroupShardPlacement( static ShardPlacement * ResolveGroupShardPlacement(
GroupShardPlacement *groupShardPlacement, CitusTableCacheEntry *tableEntry, GroupShardPlacement *groupShardPlacement, CitusTableCacheEntry *tableEntry,
ShardIntervalArrayType arrayType, int shardIndex); ShardIntervalArrayType arrayType, int
shardIndex);
static Oid LookupEnumValueId(Oid typeId, char *valueName); static Oid LookupEnumValueId(Oid typeId, char *valueName);
static void InvalidateCitusTableCacheEntrySlot(CitusTableCacheEntrySlot *cacheSlot); static void InvalidateCitusTableCacheEntrySlot(CitusTableCacheEntrySlot *cacheSlot);
static void InvalidateDistTableCache(void); static void InvalidateDistTableCache(void);
@ -298,13 +299,18 @@ static void InitializeTableCacheEntry(int64 shardId);
static bool IsCitusTableTypeInternal(char partitionMethod, char replicationModel, static bool IsCitusTableTypeInternal(char partitionMethod, char replicationModel,
CitusTableType tableType); CitusTableType tableType);
static bool RefreshTableCacheEntryIfInvalid(ShardIdCacheEntry *shardEntry); static bool RefreshTableCacheEntryIfInvalid(ShardIdCacheEntry *shardEntry);
static void BuildShardIdCacheAndPlacementList(ShardInterval **shardIntervalArray, int arrayLength, static void BuildShardIdCacheAndPlacementList(ShardInterval **shardIntervalArray, int
ShardIntervalArrayType arrayType, CitusTableCacheEntry *cacheEntry); arrayLength,
ShardIntervalArrayType arrayType,
CitusTableCacheEntry *cacheEntry);
static void FreeCitusTableCacheShardAndPlacementEntryFromArray( static void FreeCitusTableCacheShardAndPlacementEntryFromArray(
ShardInterval **shardIntervalArray, ShardInterval **shardIntervalArray,
int shardIntervalArrayLength, int
GroupShardPlacement **arrayOfPlacementArrays, shardIntervalArrayLength,
int *arrayOfPlacementArrayLengths); GroupShardPlacement **
arrayOfPlacementArrays,
int *
arrayOfPlacementArrayLengths);
static Oid DistAuthinfoRelationId(void); static Oid DistAuthinfoRelationId(void);
static Oid DistAuthinfoIndexId(void); static Oid DistAuthinfoIndexId(void);
@ -851,7 +857,8 @@ LoadShardPlacement(uint64 shardId, uint64 placementId)
GroupShardPlacement *groupPlacement = LoadGroupShardPlacement(shardId, placementId); GroupShardPlacement *groupPlacement = LoadGroupShardPlacement(shardId, placementId);
ShardPlacement *nodePlacement = ResolveGroupShardPlacement(groupPlacement, ShardPlacement *nodePlacement = ResolveGroupShardPlacement(groupPlacement,
tableEntry, arrayType, shardIndex); tableEntry, arrayType,
shardIndex);
return nodePlacement; return nodePlacement;
} }
@ -940,7 +947,6 @@ ResolveGroupShardPlacement(GroupShardPlacement *groupShardPlacement,
ShardIntervalArrayType arrayType, ShardIntervalArrayType arrayType,
int shardIndex) int shardIndex)
{ {
ShardInterval *shardInterval = NULL; ShardInterval *shardInterval = NULL;
if (arrayType == ACTIVE_SHARD_ARRAY) if (arrayType == ACTIVE_SHARD_ARRAY)
{ {
@ -1134,7 +1140,8 @@ ShardPlacementListIncludingOrphanedPlacements(uint64 shardId)
/* the offset better be in a valid range */ /* the offset better be in a valid range */
Assert(shardIndex < tableEntry->orphanedShardIntervalArrayLength); Assert(shardIndex < tableEntry->orphanedShardIntervalArrayLength);
placementArray = tableEntry->arrayOfOrphanedShardsPlacementArrays[shardIndex]; placementArray = tableEntry->arrayOfOrphanedShardsPlacementArrays[shardIndex];
numberOfPlacements = tableEntry->arrayOfOrphanedShardsPlacementArrayLengths[shardIndex]; numberOfPlacements =
tableEntry->arrayOfOrphanedShardsPlacementArrayLengths[shardIndex];
} }
for (int i = 0; i < numberOfPlacements; i++) for (int i = 0; i < numberOfPlacements; i++)
@ -1819,9 +1826,11 @@ BuildCachedShardList(CitusTableCacheEntry *cacheEntry)
varcollid, varcollid,
shardIntervalCompareFunction); shardIntervalCompareFunction);
sortedOrphanedShardIntervalArray = SortShardIntervalArray(shardIntervalToDeleteArray, sortedOrphanedShardIntervalArray = SortShardIntervalArray(
shardIntervalToDeleteArray,
orphanedShardCount, orphanedShardCount,
cacheEntry->partitionColumn-> cacheEntry->
partitionColumn->
varcollid, varcollid,
shardIntervalCompareFunction); shardIntervalCompareFunction);
@ -1857,7 +1866,8 @@ BuildCachedShardList(CitusTableCacheEntry *cacheEntry)
BuildShardIdCacheAndPlacementList(sortedShardIntervalArray, activeShardCount, BuildShardIdCacheAndPlacementList(sortedShardIntervalArray, activeShardCount,
ACTIVE_SHARD_ARRAY, cacheEntry); ACTIVE_SHARD_ARRAY, cacheEntry);
BuildShardIdCacheAndPlacementList(sortedOrphanedShardIntervalArray, orphanedShardCount, BuildShardIdCacheAndPlacementList(sortedOrphanedShardIntervalArray,
orphanedShardCount,
TO_DELETE_SHARD_ARRAY, cacheEntry); TO_DELETE_SHARD_ARRAY, cacheEntry);
cacheEntry->shardColumnCompareFunction = shardColumnCompareFunction; cacheEntry->shardColumnCompareFunction = shardColumnCompareFunction;
@ -1870,7 +1880,8 @@ BuildCachedShardList(CitusTableCacheEntry *cacheEntry)
*/ */
void void
BuildShardIdCacheAndPlacementList(ShardInterval **shardIntervalArray, int arrayLength, BuildShardIdCacheAndPlacementList(ShardInterval **shardIntervalArray, int arrayLength,
ShardIntervalArrayType arrayType, CitusTableCacheEntry *cacheEntry) ShardIntervalArrayType arrayType,
CitusTableCacheEntry *cacheEntry)
{ {
/* maintain shardId->(table,ShardInterval) cache */ /* maintain shardId->(table,ShardInterval) cache */
for (int shardIndex = 0; shardIndex < arrayLength; shardIndex++) for (int shardIndex = 0; shardIndex < arrayLength; shardIndex++)
@ -1928,7 +1939,8 @@ BuildShardIdCacheAndPlacementList(ShardInterval **shardIntervalArray, int arrayL
else else
{ {
cacheEntry->arrayOfOrphanedShardsPlacementArrays[shardIndex] = placementArray; cacheEntry->arrayOfOrphanedShardsPlacementArrays[shardIndex] = placementArray;
cacheEntry->arrayOfOrphanedShardsPlacementArrayLengths[shardIndex] = numberOfPlacements; cacheEntry->arrayOfOrphanedShardsPlacementArrayLengths[shardIndex] =
numberOfPlacements;
} }
/* store the shard index in the ShardInterval */ /* store the shard index in the ShardInterval */
@ -4139,10 +4151,10 @@ RegisterAuthinfoCacheCallbacks(void)
* *
*/ */
static void static void
FreeCitusTableCacheShardAndPlacementEntryFromArray( FreeCitusTableCacheShardAndPlacementEntryFromArray(ShardInterval **shardIntervalArray,
ShardInterval **shardIntervalArray,
int shardIntervalArrayLength, int shardIntervalArrayLength,
GroupShardPlacement **arrayOfPlacementArrays, GroupShardPlacement **
arrayOfPlacementArrays,
int *arrayOfPlacementArrayLengths) int *arrayOfPlacementArrayLengths)
{ {
for (int shardIndex = 0; shardIndex < shardIntervalArrayLength; for (int shardIndex = 0; shardIndex < shardIntervalArrayLength;
@ -4215,8 +4227,11 @@ ResetCitusTableCacheEntry(CitusTableCacheEntry *cacheEntry)
} }
/* clean up ShardIdCacheHash */ /* clean up ShardIdCacheHash */
RemoveStaleShardIdCacheEntries(cacheEntry, cacheEntry->sortedShardIntervalArray, cacheEntry->shardIntervalArrayLength); RemoveStaleShardIdCacheEntries(cacheEntry, cacheEntry->sortedShardIntervalArray,
RemoveStaleShardIdCacheEntries(cacheEntry, cacheEntry->sortedOrphanedShardIntervalArray, cacheEntry->orphanedShardIntervalArrayLength); cacheEntry->shardIntervalArrayLength);
RemoveStaleShardIdCacheEntries(cacheEntry,
cacheEntry->sortedOrphanedShardIntervalArray,
cacheEntry->orphanedShardIntervalArrayLength);
FreeCitusTableCacheShardAndPlacementEntryFromArray( FreeCitusTableCacheShardAndPlacementEntryFromArray(
cacheEntry->sortedShardIntervalArray, cacheEntry->sortedShardIntervalArray,
@ -4288,8 +4303,7 @@ ResetCitusTableCacheEntry(CitusTableCacheEntry *cacheEntry)
* we leave it in place. * we leave it in place.
*/ */
static void static void
RemoveStaleShardIdCacheEntries( RemoveStaleShardIdCacheEntries(CitusTableCacheEntry *invalidatedTableEntry,
CitusTableCacheEntry *invalidatedTableEntry,
ShardInterval **sortedShardIntervalArray, ShardInterval **sortedShardIntervalArray,
int shardIntervalArrayLength) int shardIntervalArrayLength)
{ {

View File

@ -3233,11 +3233,13 @@ citus_internal_add_shard_metadata(PG_FUNCTION_ARGS)
* not sane, the user can only affect its own tables. Given that the * not sane, the user can only affect its own tables. Given that the
* user is owner of the table, we should allow. * user is owner of the table, we should allow.
*/ */
EnsureShardMetadataIsSane(relationId, shardId, storageType, shardState, shardMinValue, EnsureShardMetadataIsSane(relationId, shardId, storageType, shardState,
shardMinValue,
shardMaxValue); shardMaxValue);
} }
InsertShardRow(relationId, shardId, storageType, shardState, shardMinValue, shardMaxValue); InsertShardRow(relationId, shardId, storageType, shardState, shardMinValue,
shardMaxValue);
PG_RETURN_VOID(); PG_RETURN_VOID();
} }

View File

@ -1068,6 +1068,7 @@ LoadShardIntervalList(Oid relationId)
return shardList; return shardList;
} }
/* /*
* LoadShardIntervalListWithOrphanedShards returns a list of shard intervals related for a given * LoadShardIntervalListWithOrphanedShards returns a list of shard intervals related for a given
* distributed table. This includes both ACTIVE and TO_DELETE shards. * distributed table. This includes both ACTIVE and TO_DELETE shards.
@ -1076,7 +1077,8 @@ LoadShardIntervalList(Oid relationId)
* list where elements are sorted on shardminvalue. Shard intervals with uninitialized * list where elements are sorted on shardminvalue. Shard intervals with uninitialized
* shard min/max values are placed in the end of the list. * shard min/max values are placed in the end of the list.
*/ */
List * LoadShardIntervalListWithOrphanedShards(Oid relationId) List *
LoadShardIntervalListWithOrphanedShards(Oid relationId)
{ {
CitusTableCacheEntry *cacheEntry = GetCitusTableCacheEntry(relationId); CitusTableCacheEntry *cacheEntry = GetCitusTableCacheEntry(relationId);
List *shardList = NIL; List *shardList = NIL;
@ -1088,10 +1090,12 @@ List * LoadShardIntervalListWithOrphanedShards(Oid relationId)
.collation = cacheEntry->partitionColumn->varcollid .collation = cacheEntry->partitionColumn->varcollid
}; };
while(indexOne < cacheEntry->shardIntervalArrayLength && indexTwo < cacheEntry->orphanedShardIntervalArrayLength) while (indexOne < cacheEntry->shardIntervalArrayLength && indexTwo <
cacheEntry->orphanedShardIntervalArrayLength)
{ {
ShardInterval *leftInterval = cacheEntry->sortedShardIntervalArray[indexOne]; ShardInterval *leftInterval = cacheEntry->sortedShardIntervalArray[indexOne];
ShardInterval *rightInterval = cacheEntry->sortedOrphanedShardIntervalArray[indexTwo]; ShardInterval *rightInterval =
cacheEntry->sortedOrphanedShardIntervalArray[indexTwo];
int cmp = CompareShardIntervals(leftInterval, rightInterval, &sortContext); int cmp = CompareShardIntervals(leftInterval, rightInterval, &sortContext);
@ -1115,7 +1119,8 @@ List * LoadShardIntervalListWithOrphanedShards(Oid relationId)
while (indexTwo < cacheEntry->orphanedShardIntervalArrayLength) while (indexTwo < cacheEntry->orphanedShardIntervalArrayLength)
{ {
shardList = lappend(shardList, cacheEntry->sortedOrphanedShardIntervalArray[indexTwo]); shardList = lappend(shardList,
cacheEntry->sortedOrphanedShardIntervalArray[indexTwo]);
indexTwo++; indexTwo++;
} }

View File

@ -265,7 +265,7 @@ CreateColocatedShards(Oid targetRelationId, Oid sourceRelationId, bool
LockRelationOid(sourceRelationId, AccessShareLock); LockRelationOid(sourceRelationId, AccessShareLock);
/* prevent placement changes of the source relation until we colocate with them */ /* prevent placement changes of the source relation until we colocate with them */
// TODO(niupre): We should only return ACTIVE shards, not ALL shards. /* TODO(niupre): We should only return ACTIVE shards, not ALL shards. */
List *sourceShardIntervalList = LoadShardIntervalList(sourceRelationId); List *sourceShardIntervalList = LoadShardIntervalList(sourceRelationId);
LockShardListMetadata(sourceShardIntervalList, ShareLock); LockShardListMetadata(sourceShardIntervalList, ShareLock);
@ -378,7 +378,8 @@ CreateReferenceTableShard(Oid distributedTableId)
/* shard state is active by default */ /* shard state is active by default */
ShardState shardState = SHARD_STATE_ACTIVE; ShardState shardState = SHARD_STATE_ACTIVE;
InsertShardRow(distributedTableId, shardId, shardStorageType, shardState, shardMinValue, InsertShardRow(distributedTableId, shardId, shardStorageType, shardState,
shardMinValue,
shardMaxValue); shardMaxValue);
List *insertedShardPlacements = InsertShardPlacementRows(distributedTableId, shardId, List *insertedShardPlacements = InsertShardPlacementRows(distributedTableId, shardId,

View File

@ -1147,6 +1147,7 @@ CreateForeignKeyConstraints(List *shardGroupSplitIntervalListList,
} }
} }
/* /*
* MarkShardListForDrop drops shards and their metadata from both the coordinator and * MarkShardListForDrop drops shards and their metadata from both the coordinator and
* mx nodes. * mx nodes.

View File

@ -187,7 +187,8 @@ master_create_empty_shard(PG_FUNCTION_ARGS)
candidateNodeIndex++; candidateNodeIndex++;
} }
InsertShardRow(relationId, shardId, storageType, shardState, nullMinValue, nullMaxValue); InsertShardRow(relationId, shardId, storageType, shardState, nullMinValue,
nullMaxValue);
CreateAppendDistributedShardPlacements(relationId, shardId, candidateNodeList, CreateAppendDistributedShardPlacements(relationId, shardId, candidateNodeList,
ShardReplicationFactor); ShardReplicationFactor);

View File

@ -3982,7 +3982,7 @@ ShardIntervalsOverlapWithParams(Datum firstMin, Datum firstMax,
int firstComparison = DatumGetInt32(firstDatum); int firstComparison = DatumGetInt32(firstDatum);
int secondComparison = DatumGetInt32(secondDatum); int secondComparison = DatumGetInt32(secondDatum);
// If one of the shards are marked as TO_DELETED, ignore the overlap. /* If one of the shards are marked as TO_DELETED, ignore the overlap. */
bool markedForDelete = (firstState == SHARD_STATE_TO_DELETE || bool markedForDelete = (firstState == SHARD_STATE_TO_DELETE ||
secondState == SHARD_STATE_TO_DELETE); secondState == SHARD_STATE_TO_DELETE);

View File

@ -1,10 +1,10 @@
CREATE OR REPLACE FUNCTION pg_catalog.citus_internal_add_shard_metadata( CREATE OR REPLACE FUNCTION pg_catalog.citus_internal_add_shard_metadata(
relation_id regclass, shard_id bigint, relation_id regclass, shard_id bigint,
storage_type "char", shard_min_value text, storage_type "char", shardstate integer,
shard_max_value text shard_min_value text, shard_max_value text
) )
RETURNS void RETURNS void
LANGUAGE C LANGUAGE C
AS 'MODULE_PATHNAME'; AS 'MODULE_PATHNAME';
COMMENT ON FUNCTION pg_catalog.citus_internal_add_shard_metadata(regclass, bigint, "char", text, text) IS COMMENT ON FUNCTION pg_catalog.citus_internal_add_shard_metadata(regclass, bigint, "char", integer, text, text) IS
'Inserts into pg_dist_shard with user checks'; 'Inserts into pg_dist_shard with user checks';

View File

@ -94,7 +94,8 @@ extern void ProcessUtilityParseTree(Node *node, const char *queryString,
); );
extern void MarkInvalidateForeignKeyGraph(void); extern void MarkInvalidateForeignKeyGraph(void);
extern void InvalidateForeignKeyGraphForDDL(void); extern void InvalidateForeignKeyGraphForDDL(void);
extern List * DDLTaskList(Oid relationId, const char *commandString, bool includeOrphanedShards); extern List * DDLTaskList(Oid relationId, const char *commandString, bool
includeOrphanedShards);
extern List * NodeDDLTaskList(TargetWorkerSet targets, List *commands); extern List * NodeDDLTaskList(TargetWorkerSet targets, List *commands);
extern bool AlterTableInProgress(void); extern bool AlterTableInProgress(void);
extern bool DropSchemaOrDBInProgress(void); extern bool DropSchemaOrDBInProgress(void);