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

Fix downgrade and following upgrade, fix review comments

pull/7950/head
Maksim Melnikov 2025-04-07 15:46:53 +03:00
parent 458e6abe6d
commit f5c419487d
6 changed files with 89 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/backend/distributed/cdc/cdc_decoder_utils.c
View File

@ -349,8 +349,8 @@ CdcIsReferenceTableViaCatalog(Oid relationId)
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock); Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray);

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

@ -320,9 +320,10 @@ static void CachedRelationNamespaceLookup(const char *relationName, Oid relnames
static void CachedRelationNamespaceLookupExtended(const char *relationName, static void CachedRelationNamespaceLookupExtended(const char *relationName,
Oid renamespace, Oid *cachedOid, Oid renamespace, Oid *cachedOid,
bool missing_ok); bool missing_ok);
static ShardPlacement * ResolveGroupShardPlacement( static ShardPlacement * ResolveGroupShardPlacement(GroupShardPlacement *
GroupShardPlacement *groupShardPlacement, CitusTableCacheEntry *tableEntry, groupShardPlacement,
int shardIndex); CitusTableCacheEntry *tableEntry,
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);
@ -733,8 +734,8 @@ PartitionMethodViaCatalog(Oid relationId)
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray);
@ -776,8 +777,8 @@ PartitionColumnViaCatalog(Oid relationId)
Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock); Relation pgDistPartition = table_open(DistPartitionRelationId(), AccessShareLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray);
@ -826,8 +827,8 @@ ColocationIdViaCatalog(Oid relationId)
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(partitionTuple, tupleDescriptor, datumArray, isNullArray);
@ -1757,8 +1758,8 @@ BuildCitusTableCacheEntry(Oid relationId)
MemoryContext oldContext = NULL; MemoryContext oldContext = NULL;
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(distPartitionTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(distPartitionTuple, tupleDescriptor, datumArray, isNullArray);
@ -5029,10 +5030,12 @@ CitusTableTypeIdList(CitusTableType citusTableType)
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
HeapTuple heapTuple = systable_getnext(scanDescriptor); HeapTuple heapTuple = systable_getnext(scanDescriptor);
Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
while (HeapTupleIsValid(heapTuple)) while (HeapTupleIsValid(heapTuple))
{ {
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); memset(datumArray, 0, tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); memset(isNullArray, 0, tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray);
Datum partMethodDatum = datumArray[Anum_pg_dist_partition_partmethod - 1]; Datum partMethodDatum = datumArray[Anum_pg_dist_partition_partmethod - 1];
@ -5054,9 +5057,9 @@ CitusTableTypeIdList(CitusTableType citusTableType)
} }
heapTuple = systable_getnext(scanDescriptor); heapTuple = systable_getnext(scanDescriptor);
pfree(datumArray);
pfree(isNullArray);
} }
pfree(datumArray);
pfree(isNullArray);
systable_endscan(scanDescriptor); systable_endscan(scanDescriptor);
table_close(pgDistPartition, AccessShareLock); table_close(pgDistPartition, AccessShareLock);

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

@ -573,8 +573,8 @@ FetchRelationIdFromPgPartitionHeapTuple(HeapTuple heapTuple, TupleDesc tupleDesc
{ {
Assert(heapTuple->t_tableOid == DistPartitionRelationId()); Assert(heapTuple->t_tableOid == DistPartitionRelationId());
Datum* datumArray = (Datum *) palloc0(tupleDesc->natts * sizeof(Datum)); Datum *datumArray = (Datum *) palloc0(tupleDesc->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc0(tupleDesc->natts * sizeof(bool)); bool *isNullArray = (bool *) palloc0(tupleDesc->natts * sizeof(bool));
heap_deform_tuple(heapTuple, tupleDesc, datumArray, isNullArray); heap_deform_tuple(heapTuple, tupleDesc, datumArray, isNullArray);

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

@ -127,11 +127,11 @@ static bool SetFieldText(int attno, Datum values[], bool isnull[], bool replace[
static bool SetFieldNull(int attno, Datum values[], bool isnull[], bool replace[]); static bool SetFieldNull(int attno, Datum values[], bool isnull[], bool replace[]);
#define InitFieldValue(attno, values, isnull, initValue) \ #define InitFieldValue(attno, values, isnull, initValue) \
(void) SetFieldValue((attno), (values), (isnull), NULL, (initValue)) (void) SetFieldValue((attno), (values), (isnull), NULL, (initValue))
#define InitFieldText(attno, values, isnull, initValue) \ #define InitFieldText(attno, values, isnull, initValue) \
(void) SetFieldText((attno), (values), (isnull), NULL, (initValue)) (void) SetFieldText((attno), (values), (isnull), NULL, (initValue))
#define InitFieldNull(attno, values, isnull) \ #define InitFieldNull(attno, values, isnull) \
(void) SetFieldNull((attno), (values), (isnull), NULL) (void) SetFieldNull((attno), (values), (isnull), NULL)
/* exports for SQL callable functions */ /* exports for SQL callable functions */
PG_FUNCTION_INFO_V1(citus_local_disk_space_stats); PG_FUNCTION_INFO_V1(citus_local_disk_space_stats);
@ -822,7 +822,8 @@ GenerateSizeQueryOnMultiplePlacements(List *shardIntervalList,
/* SELECT SUM(worker_partitioned_...) FROM VALUES (...) */ /* SELECT SUM(worker_partitioned_...) FROM VALUES (...) */
char *subqueryForPartitionedShards = char *subqueryForPartitionedShards =
GenerateSizeQueryForRelationNameList(partitionedShardNames, GenerateSizeQueryForRelationNameList(partitionedShardNames,
GetWorkerPartitionedSizeUDFNameBySizeQueryType( GetWorkerPartitionedSizeUDFNameBySizeQueryType
(
sizeQueryType)); sizeQueryType));
/* SELECT SUM(pg_..._size) FROM VALUES (...) */ /* SELECT SUM(pg_..._size) FROM VALUES (...) */
@ -1923,8 +1924,8 @@ InsertIntoPgDistPartition(Oid relationId, char distributionMethod,
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock); Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* newValues = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum)); Datum *newValues = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum));
bool* newNulls = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *newNulls = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
/* form new tuple for pg_dist_partition */ /* form new tuple for pg_dist_partition */
newValues[Anum_pg_dist_partition_logicalrelid - 1] = newValues[Anum_pg_dist_partition_logicalrelid - 1] =
@ -1933,7 +1934,8 @@ InsertIntoPgDistPartition(Oid relationId, char distributionMethod,
CharGetDatum(distributionMethod); CharGetDatum(distributionMethod);
newValues[Anum_pg_dist_partition_colocationid - 1] = UInt32GetDatum(colocationId); newValues[Anum_pg_dist_partition_colocationid - 1] = UInt32GetDatum(colocationId);
newValues[Anum_pg_dist_partition_repmodel - 1] = CharGetDatum(replicationModel); newValues[Anum_pg_dist_partition_repmodel - 1] = CharGetDatum(replicationModel);
newValues[GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor)] = BoolGetDatum(autoConverted); newValues[GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor)] = BoolGetDatum(
autoConverted);
/* set partkey column to NULL for reference tables */ /* set partkey column to NULL for reference tables */
if (distributionMethod != DISTRIBUTE_BY_NONE) if (distributionMethod != DISTRIBUTE_BY_NONE)
@ -2159,9 +2161,9 @@ UpdatePlacementGroupId(uint64 placementId, int groupId)
Relation pgDistPlacement = table_open(DistPlacementRelationId(), RowExclusiveLock); Relation pgDistPlacement = table_open(DistPlacementRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPlacement); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPlacement);
Datum* values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum)); Datum *values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum));
bool* isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
bool* replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
ScanKeyInit(&scanKey[0], Anum_pg_dist_placement_placementid, ScanKeyInit(&scanKey[0], Anum_pg_dist_placement_placementid,
BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(placementId)); BTEqualStrategyNumber, F_INT8EQ, Int64GetDatum(placementId));
@ -2213,13 +2215,12 @@ UpdatePgDistPartitionAutoConverted(Oid citusTableId, bool autoConverted)
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
int scanKeyCount = 1; int scanKeyCount = 1;
bool indexOK = true; bool indexOK = true;
int autoconvertedindex;
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock); Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum)); Datum *values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum));
bool* isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
bool* replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid, ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(citusTableId)); BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(citusTableId));
@ -2236,7 +2237,7 @@ UpdatePgDistPartitionAutoConverted(Oid citusTableId, bool autoConverted)
citusTableId))); citusTableId)));
} }
autoconvertedindex = GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor); int autoconvertedindex = GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor);
values[autoconvertedindex] = BoolGetDatum(autoConverted); values[autoconvertedindex] = BoolGetDatum(autoConverted);
isnull[autoconvertedindex] = false; isnull[autoconvertedindex] = false;
replace[autoconvertedindex] = true; replace[autoconvertedindex] = true;
@ -2294,13 +2295,12 @@ UpdateDistributionColumn(Oid relationId, char distributionMethod, Var *distribut
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
int scanKeyCount = 1; int scanKeyCount = 1;
bool indexOK = true; bool indexOK = true;
int autoconvertedindex;
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock); Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum)); Datum *values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum));
bool* isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
bool* replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid, ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId)); BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId));
@ -2325,7 +2325,7 @@ UpdateDistributionColumn(Oid relationId, char distributionMethod, Var *distribut
values[Anum_pg_dist_partition_colocationid - 1] = UInt32GetDatum(colocationId); values[Anum_pg_dist_partition_colocationid - 1] = UInt32GetDatum(colocationId);
isnull[Anum_pg_dist_partition_colocationid - 1] = false; isnull[Anum_pg_dist_partition_colocationid - 1] = false;
autoconvertedindex = GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor); int autoconvertedindex = GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor);
replace[autoconvertedindex] = true; replace[autoconvertedindex] = true;
values[autoconvertedindex] = BoolGetDatum(false); values[autoconvertedindex] = BoolGetDatum(false);
isnull[autoconvertedindex] = false; isnull[autoconvertedindex] = false;
@ -2393,13 +2393,12 @@ UpdateNoneDistTableMetadata(Oid relationId, char replicationModel, uint32 coloca
ScanKeyData scanKey[1]; ScanKeyData scanKey[1];
int scanKeyCount = 1; int scanKeyCount = 1;
bool indexOK = true; bool indexOK = true;
int autoconvertedindex;
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock); Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum)); Datum *values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum));
bool* isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *isnull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
bool* replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid, ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId)); BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(relationId));
@ -2424,7 +2423,7 @@ UpdateNoneDistTableMetadata(Oid relationId, char replicationModel, uint32 coloca
isnull[Anum_pg_dist_partition_repmodel - 1] = false; isnull[Anum_pg_dist_partition_repmodel - 1] = false;
replace[Anum_pg_dist_partition_repmodel - 1] = true; replace[Anum_pg_dist_partition_repmodel - 1] = true;
autoconvertedindex = GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor); int autoconvertedindex = GetAutoConvertedAttrIndexInPgDistPartition(tupleDescriptor);
values[autoconvertedindex] = BoolGetDatum(autoConverted); values[autoconvertedindex] = BoolGetDatum(autoConverted);
isnull[autoconvertedindex] = false; isnull[autoconvertedindex] = false;
replace[autoconvertedindex] = true; replace[autoconvertedindex] = true;
@ -3166,8 +3165,8 @@ ScheduleBackgroundTask(int64 jobId, Oid owner, char *command, int dependingTaskC
values[Anum_pg_dist_background_task_nodes_involved - 1] = values[Anum_pg_dist_background_task_nodes_involved - 1] =
IntArrayToDatum(nodesInvolvedCount, nodesInvolved); IntArrayToDatum(nodesInvolvedCount, nodesInvolved);
nulls[Anum_pg_dist_background_task_nodes_involved - 1] = (nodesInvolvedCount == nulls[Anum_pg_dist_background_task_nodes_involved - 1] = (nodesInvolvedCount == 0)
0); ;
HeapTuple newTuple = heap_form_tuple(RelationGetDescr(pgDistBackgroundTask), HeapTuple newTuple = heap_form_tuple(RelationGetDescr(pgDistBackgroundTask),
values, nulls); values, nulls);
@ -4243,7 +4242,8 @@ CancelTasksForJob(int64 jobid)
const bool indexOK = true; const bool indexOK = true;
SysScanDesc scanDescriptor = systable_beginscan(pgDistBackgroundTasks, SysScanDesc scanDescriptor = systable_beginscan(pgDistBackgroundTasks,
DistBackgroundTaskJobIdTaskIdIndexId(), DistBackgroundTaskJobIdTaskIdIndexId()
,
indexOK, NULL, indexOK, NULL,
lengthof(scanKey), scanKey); lengthof(scanKey), scanKey);
@ -4438,11 +4438,17 @@ UnblockDependingBackgroundTasks(BackgroundTask *task)
table_close(pgDistBackgroundTasksDepend, NoLock); table_close(pgDistBackgroundTasksDepend, NoLock);
} }
/* /*
* AutoConverted attr was added to table pg_dist_partition using alter operation, so * GetAutoConvertedAttrIndexInPgDistPartition returns attrnum for autoconverted attr.
* in case of downgrade we can get additional tuple field in tuple desc with dropped flag *
* and in case of futher upgrade we can get number of fields > Natts_pg_dist_partition. * autoconverted attr was added to table pg_dist_partition using alter operation after
* So we should properly address AutoConverted attr in arrays. * the version where Citus started supporting downgrades, and it's only column that we've
* introduced to pg_dist_partition since then.
*
* And in case of a downgrade + upgrade, tupleDEsc->natts becomes greater than
* Natts_pg_dist_partition and when this happens, then we know that attrnum autoconverted is
* not Anum_pg_dist_partition_autoconverted anymore but tupleDEsc->natts - 1.
*/ */
int int
GetAutoConvertedAttrIndexInPgDistPartition(TupleDesc tupleDEsc) GetAutoConvertedAttrIndexInPgDistPartition(TupleDesc tupleDEsc)
@ -4450,4 +4456,4 @@ GetAutoConvertedAttrIndexInPgDistPartition(TupleDesc tupleDEsc)
return TupleDescSize(tupleDEsc) == Natts_pg_dist_partition return TupleDescSize(tupleDEsc) == Natts_pg_dist_partition
? (Anum_pg_dist_partition_autoconverted - 1) ? (Anum_pg_dist_partition_autoconverted - 1)
: tupleDEsc->natts - 1; : tupleDEsc->natts - 1;
} }

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

@ -557,8 +557,8 @@ ColocationId(int shardCount, int replicationFactor, Oid distributionColumnType,
continue; continue;
} }
if (colocationId == INVALID_COLOCATION_ID || colocationId > if (colocationId == INVALID_COLOCATION_ID || colocationId > colocationForm->
colocationForm->colocationid) colocationid)
{ {
/* /*
* We assign the smallest colocation id among all the matches so that we * We assign the smallest colocation id among all the matches so that we
@ -819,9 +819,9 @@ UpdateRelationColocationGroup(Oid distributedRelationId, uint32 colocationId,
Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock); Relation pgDistPartition = table_open(DistPartitionRelationId(), RowExclusiveLock);
TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition); TupleDesc tupleDescriptor = RelationGetDescr(pgDistPartition);
Datum* values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum)); Datum *values = (Datum *) palloc0(tupleDescriptor->natts * sizeof(Datum));
bool* isNull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *isNull = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
bool* replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool)); bool *replace = (bool *) palloc0(tupleDescriptor->natts * sizeof(bool));
ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid, ScanKeyInit(&scanKey[0], Anum_pg_dist_partition_logicalrelid,
BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(distributedRelationId)); BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(distributedRelationId));
@ -999,18 +999,18 @@ ColocationGroupTableList(uint32 colocationId, uint32 count)
indexOK, NULL, scanKeyCount, scanKey); indexOK, NULL, scanKeyCount, scanKey);
HeapTuple heapTuple = systable_getnext(scanDescriptor); HeapTuple heapTuple = systable_getnext(scanDescriptor);
Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
while (HeapTupleIsValid(heapTuple)) while (HeapTupleIsValid(heapTuple))
{ {
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); memset(datumArray, 0, tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); memset(isNullArray, 0, tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray);
Oid colocatedTableId = DatumGetObjectId( Oid colocatedTableId = DatumGetObjectId(
datumArray[Anum_pg_dist_partition_logicalrelid - 1]); datumArray[Anum_pg_dist_partition_logicalrelid - 1]);
colocatedTableList = lappend_oid(colocatedTableList, colocatedTableId); colocatedTableList = lappend_oid(colocatedTableList, colocatedTableId);
heapTuple = systable_getnext(scanDescriptor); heapTuple = systable_getnext(scanDescriptor);
pfree(datumArray);
pfree(isNullArray);
if (count == 0) if (count == 0)
{ {
@ -1023,6 +1023,8 @@ ColocationGroupTableList(uint32 colocationId, uint32 count)
break; break;
} }
} }
pfree(datumArray);
pfree(isNullArray);
systable_endscan(scanDescriptor); systable_endscan(scanDescriptor);
table_close(pgDistPartition, AccessShareLock); table_close(pgDistPartition, AccessShareLock);
@ -1075,8 +1077,8 @@ ColocatedShardIntervalList(ShardInterval *shardInterval)
* Since we iterate over co-located tables, shard count of each table should be * Since we iterate over co-located tables, shard count of each table should be
* same and greater than shardIntervalIndex. * same and greater than shardIntervalIndex.
*/ */
Assert(cacheEntry->shardIntervalArrayLength == Assert(cacheEntry->shardIntervalArrayLength == colocatedTableCacheEntry->
colocatedTableCacheEntry->shardIntervalArrayLength); shardIntervalArrayLength);
ShardInterval *colocatedShardInterval = ShardInterval *colocatedShardInterval =
colocatedTableCacheEntry->sortedShardIntervalArray[shardIntervalIndex]; colocatedTableCacheEntry->sortedShardIntervalArray[shardIntervalIndex];
@ -1146,8 +1148,8 @@ ColocatedNonPartitionShardIntervalList(ShardInterval *shardInterval)
* Since we iterate over co-located tables, shard count of each table should be * Since we iterate over co-located tables, shard count of each table should be
* same and greater than shardIntervalIndex. * same and greater than shardIntervalIndex.
*/ */
Assert(cacheEntry->shardIntervalArrayLength == Assert(cacheEntry->shardIntervalArrayLength == colocatedTableCacheEntry->
colocatedTableCacheEntry->shardIntervalArrayLength); shardIntervalArrayLength);
ShardInterval *colocatedShardInterval = ShardInterval *colocatedShardInterval =
colocatedTableCacheEntry->sortedShardIntervalArray[shardIntervalIndex]; colocatedTableCacheEntry->sortedShardIntervalArray[shardIntervalIndex];
@ -1195,10 +1197,12 @@ ColocatedTableId(int32 colocationId)
indexOK, NULL, scanKeyCount, scanKey); indexOK, NULL, scanKeyCount, scanKey);
HeapTuple heapTuple = systable_getnext(scanDescriptor); HeapTuple heapTuple = systable_getnext(scanDescriptor);
Datum *datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum));
bool *isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool));
while (HeapTupleIsValid(heapTuple)) while (HeapTupleIsValid(heapTuple))
{ {
Datum* datumArray = (Datum *) palloc(tupleDescriptor->natts * sizeof(Datum)); memset(datumArray, 0, tupleDescriptor->natts * sizeof(Datum));
bool* isNullArray = (bool *) palloc(tupleDescriptor->natts * sizeof(bool)); memset(isNullArray, 0, tupleDescriptor->natts * sizeof(bool));
heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray); heap_deform_tuple(heapTuple, tupleDescriptor, datumArray, isNullArray);
colocatedTableId = DatumGetObjectId( colocatedTableId = DatumGetObjectId(
datumArray[Anum_pg_dist_partition_logicalrelid - 1]); datumArray[Anum_pg_dist_partition_logicalrelid - 1]);
@ -1225,9 +1229,9 @@ ColocatedTableId(int32 colocationId)
colocatedTableId = InvalidOid; colocatedTableId = InvalidOid;
heapTuple = systable_getnext(scanDescriptor); heapTuple = systable_getnext(scanDescriptor);
pfree(datumArray);
pfree(isNullArray);
} }
pfree(datumArray);
pfree(isNullArray);
systable_endscan(scanDescriptor); systable_endscan(scanDescriptor);
table_close(pgDistPartition, AccessShareLock); table_close(pgDistPartition, AccessShareLock);

13
src/include/distributed/metadata_utility.h
View File

@ -40,7 +40,7 @@
#define WORKER_PARTITIONED_TABLE_SIZE_FUNCTION "worker_partitioned_table_size(%s)" #define WORKER_PARTITIONED_TABLE_SIZE_FUNCTION "worker_partitioned_table_size(%s)"
#define WORKER_PARTITIONED_RELATION_SIZE_FUNCTION "worker_partitioned_relation_size(%s)" #define WORKER_PARTITIONED_RELATION_SIZE_FUNCTION "worker_partitioned_relation_size(%s)"
#define WORKER_PARTITIONED_RELATION_TOTAL_SIZE_FUNCTION \ #define WORKER_PARTITIONED_RELATION_TOTAL_SIZE_FUNCTION \
"worker_partitioned_relation_total_size(%s)" "worker_partitioned_relation_total_size(%s)"
#define SHARD_SIZES_COLUMN_COUNT (2) #define SHARD_SIZES_COLUMN_COUNT (2)
@ -302,12 +302,12 @@ typedef struct BackgroundTask
} BackgroundTask; } BackgroundTask;
#define SET_NULLABLE_FIELD(ptr, field, value) \ #define SET_NULLABLE_FIELD(ptr, field, value) \
(ptr)->__nullable_storage.field = (value); \ (ptr)->__nullable_storage.field = (value); \
(ptr)->field = &((ptr)->__nullable_storage.field) (ptr)->field = &((ptr)->__nullable_storage.field)
#define UNSET_NULLABLE_FIELD(ptr, field) \ #define UNSET_NULLABLE_FIELD(ptr, field) \
(ptr)->field = NULL; \ (ptr)->field = NULL; \
memset_struct_0((ptr)->__nullable_storage.field) memset_struct_0((ptr)->__nullable_storage.field)
/* Size functions */ /* Size functions */
extern Datum citus_table_size(PG_FUNCTION_ARGS); extern Datum citus_table_size(PG_FUNCTION_ARGS);
@ -345,7 +345,8 @@ extern bool IsDummyPlacement(ShardPlacement *taskPlacement);
extern StringInfo GenerateSizeQueryOnMultiplePlacements(List *shardIntervalList, extern StringInfo GenerateSizeQueryOnMultiplePlacements(List *shardIntervalList,
Oid indexId, Oid indexId,
SizeQueryType sizeQueryType, SizeQueryType sizeQueryType,
bool optimizePartitionCalculations); bool optimizePartitionCalculations
);
extern List * RemoveCoordinatorPlacementIfNotSingleNode(List *placementList); extern List * RemoveCoordinatorPlacementIfNotSingleNode(List *placementList);
/* Function declarations to modify shard and shard placement data */ /* Function declarations to modify shard and shard placement data */