From aab8e6b32429e47d32490a0d8453b6bebfe13271 Mon Sep 17 00:00:00 2001
From: Andres Freund <andres@anarazel.de>
Date: Fri, 29 Jul 2016 10:02:36 -0700
Subject: [PATCH] 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
---
 .../distributed/utils/metadata_cache.c        | 26 ++++++++++++-------
 src/include/distributed/pg_dist_partition.h   |  2 ++
 2 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/src/backend/distributed/utils/metadata_cache.c b/src/backend/distributed/utils/metadata_cache.c
index d798ec4a0..9ed32e6bb 100644
--- a/src/backend/distributed/utils/metadata_cache.c
+++ b/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;
 }
 
diff --git a/src/include/distributed/pg_dist_partition.h b/src/include/distributed/pg_dist_partition.h
index de7568465..56799ac20 100644
--- a/src/include/distributed/pg_dist_partition.h
+++ b/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;
 
 /* ----------------