Don't access pg_dist_partition->partkey directly, use heap_getattr().

Text datums can't be directly accessed via the struct equivalence trick
used to access catalogs. That's because, as an optimization, they're
sometimes aligned to 1 byte ("text"'s alignment), and sometimes to 4
bytes. That depends on it being a short
varlena (cf. VARATT_NOT_PAD_BYTE) or not.

In the case at hand here, partkey became longer than 127 characters -
the boundary for short varlenas (cf. VARATT_CAN_MAKE_SHORT()). Thus it
became 4 byte/int aligned. Which lead to the direct struct access
accessing the wrong data.

The fix is simply to never access partkey that way - to enforce that,
hide partkey ehind the usual ifdef.

Fixes: #674

src/backend/distributed/utils/metadata_cache.c

@@ -78,7 +78,7 @@ static bool HasUninitializedShardInterval(ShardInterval **sortedShardIntervalArr
 static void InitializeDistTableCache(void);
 static void ResetDistTableCacheEntry(DistTableCacheEntry *cacheEntry);
 static void InvalidateDistRelationCacheCallback(Datum argument, Oid relationId);
-static HeapTuple LookupDistPartitionTuple(Oid relationId);
+static HeapTuple LookupDistPartitionTuple(Relation pgDistPartition, Oid relationId);
 static List * LookupDistShardTuples(Oid relationId);
 static void GetPartitionTypeInputInfo(char *partitionKeyString, char partitionMethod,
 									  Oid *intervalTypeId, int32 *intervalTypeMod);
@@ -227,6 +227,7 @@ LookupDistTableCacheEntry(Oid relationId)
 	bool hasUninitializedShardInterval = false;
 	bool hasUniformHashDistribution = false;
 	void *hashKey = (void *) &relationId;
+	Relation pgDistPartition = NULL;
 
 	if (DistTableCacheHash == NULL)
 	{
@@ -247,15 +248,23 @@ LookupDistTableCacheEntry(Oid relationId)
 		ResetDistTableCacheEntry(cacheEntry);
 	}
 
-	distPartitionTuple = LookupDistPartitionTuple(relationId);
+	pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
+	distPartitionTuple = LookupDistPartitionTuple(pgDistPartition, relationId);
 	if (distPartitionTuple != NULL)
 	{
 		Form_pg_dist_partition partitionForm =
 			(Form_pg_dist_partition) GETSTRUCT(distPartitionTuple);
-		Datum partitionKeyDatum = PointerGetDatum(&partitionForm->partkey);
+		Datum partitionKeyDatum = 0;
+		MemoryContext oldContext = NULL;
+		bool isNull = false;
 
-		MemoryContext oldContext = MemoryContextSwitchTo(CacheMemoryContext);
+		partitionKeyDatum = heap_getattr(distPartitionTuple,
+										 Anum_pg_dist_partition_partkey,
+										 RelationGetDescr(pgDistPartition),
+										 &isNull);
+		Assert(!isNull);
 
+		oldContext = MemoryContextSwitchTo(CacheMemoryContext);
 		partitionKeyString = TextDatumGetCString(partitionKeyDatum);
 		partitionMethod = partitionForm->partmethod;
 
@@ -264,6 +273,8 @@ LookupDistTableCacheEntry(Oid relationId)
 		heap_freetuple(distPartitionTuple);
 	}
 
+	heap_close(pgDistPartition, NoLock);
+
 	distShardTupleList = LookupDistShardTuples(relationId);
 	shardIntervalArrayLength = list_length(distShardTupleList);
 	if (shardIntervalArrayLength > 0)
@@ -1019,16 +1030,13 @@ InvalidateDistRelationCacheCallback(Datum argument, Oid relationId)
  * and returns that or, if no matching entry was found, NULL.
  */
 static HeapTuple
-LookupDistPartitionTuple(Oid relationId)
+LookupDistPartitionTuple(Relation pgDistPartition, Oid relationId)
 {
-	Relation pgDistPartition = NULL;
 	HeapTuple distPartitionTuple = NULL;
 	HeapTuple currentPartitionTuple = NULL;
 	SysScanDesc scanDescriptor;
 	ScanKeyData scanKey[1];
 
-	pgDistPartition = heap_open(DistPartitionRelationId(), AccessShareLock);
-
 	/* copy scankey to local copy, it will be modified during the scan */
 	memcpy(scanKey, DistPartitionScanKey, sizeof(DistPartitionScanKey));
 
@@ -1049,8 +1057,6 @@ LookupDistPartitionTuple(Oid relationId)
 
 	systable_endscan(scanDescriptor);
 
-	heap_close(pgDistPartition, NoLock);
-
 	return distPartitionTuple;
 }
 

src/include/distributed/pg_dist_partition.h

@@ -23,7 +23,9 @@ typedef struct FormData_pg_dist_partition
 {
 	Oid logicalrelid;              /* logical relation id; references pg_class oid */
 	char partmethod;               /* partition method; see codes below */
+#ifdef CATALOG_VARLEN              /* variable-length fields start here */
 	text partkey;                  /* partition key expression */
+#endif
 } FormData_pg_dist_partition;
 
 /* ----------------