From 3a0e1fa7c7e4da46a86f7d5b9dd0392549f3b460 Mon Sep 17 00:00:00 2001
From: Alexander Korotkov <akorotkov@postgresql.org>
Date: Wed, 22 Jun 2022 18:11:26 +0300
Subject: [PATCH v2 5/6] Reorganize data structures

Reported-by:
Bug:
Discussion:
Author:
Reviewed-by:
Tested-by:
Backpatch-through:
---
 src/backend/utils/sort/tuplesort.c | 762 ++++++++++++++++-------------
 1 file changed, 432 insertions(+), 330 deletions(-)

diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
index 3bf990a1b34..e106e1ff9e2 100644
--- a/src/backend/utils/sort/tuplesort.c
+++ b/src/backend/utils/sort/tuplesort.c
@@ -126,8 +126,8 @@
 #define CLUSTER_SORT	3
 
 /* Sort parallel code from state for sort__start probes */
-#define PARALLEL_SORT(state)	((state)->shared == NULL ? 0 : \
-								 (state)->worker >= 0 ? 1 : 2)
+#define PARALLEL_SORT(coordinate)	((coordinate)->sharedsort == NULL ? 0 : \
+									 (coordinate)->isWorker >= 0 ? 1 : 2)
 
 /*
  * Initial size of memtuples array.  We're trying to select this size so that
@@ -236,37 +236,17 @@ typedef enum
 #define TAPE_BUFFER_OVERHEAD		BLCKSZ
 #define MERGE_BUFFER_SIZE			(BLCKSZ * 32)
 
+typedef struct TuplesortOps TuplesortOps;
+
 typedef int (*SortTupleComparator) (const SortTuple *a, const SortTuple *b,
 									Tuplesortstate *state);
 
-/*
- * Private state of a Tuplesort operation.
- */
-struct Tuplesortstate
+struct TuplesortOps
 {
-	TupSortStatus status;		/* enumerated value as shown above */
-	int			nKeys;			/* number of columns in sort key */
-	int			sortopt;		/* Bitmask of flags used to setup sort */
-	bool		bounded;		/* did caller specify a maximum number of
-								 * tuples to return? */
-	bool		boundUsed;		/* true if we made use of a bounded heap */
-	int			bound;			/* if bounded, the maximum number of tuples */
-	bool		tuples;			/* Can SortTuple.tuple ever be set? */
-	int64		availMem;		/* remaining memory available, in bytes */
-	int64		allowedMem;		/* total memory allowed, in bytes */
-	int			maxTapes;		/* max number of input tapes to merge in each
-								 * pass */
-	int64		maxSpace;		/* maximum amount of space occupied among sort
-								 * of groups, either in-memory or on-disk */
-	bool		isMaxSpaceDisk; /* true when maxSpace is value for on-disk
-								 * space, false when it's value for in-memory
-								 * space */
-	TupSortStatus maxSpaceStatus;	/* sort status when maxSpace was reached */
 	MemoryContext maincontext;	/* memory context for tuple sort metadata that
 								 * persists across multiple batches */
 	MemoryContext sortcontext;	/* memory context holding most sort data */
 	MemoryContext tuplecontext; /* sub-context of sortcontext for tuple data */
-	LogicalTapeSet *tapeset;	/* logtape.c object for tapes in a temp file */
 
 	/*
 	 * These function pointers decouple the routines that must know what kind
@@ -300,12 +280,116 @@ struct Tuplesortstate
 	void		(*readtup) (Tuplesortstate *state, SortTuple *stup,
 							LogicalTape *tape, unsigned int len);
 
+	void		(*freestate) (Tuplesortstate *state);
+
 	/*
 	 * Whether SortTuple's datum1 and isnull1 members are maintained by the
 	 * above routines.  If not, some sort specializations are disabled.
 	 */
 	bool		haveDatum1;
 
+	/*
+	 * The sortKeys variable is used by every case other than the hash index
+	 * case; it is set by tuplesort_begin_xxx.  tupDesc is only used by the
+	 * MinimalTuple and CLUSTER routines, though.
+	 */
+	int			nKeys;			/* number of columns in sort key */
+	SortSupport sortKeys;		/* array of length nKeys */
+
+	/*
+	 * This variable is shared by the single-key MinimalTuple case and the
+	 * Datum case (which both use qsort_ssup()).  Otherwise, it's NULL.  The
+	 * presence of a value in this field is also checked by various sort
+	 * specialization functions as an optimization when comparing the leading
+	 * key in a tiebreak situation to determine if there are any subsequent
+	 * keys to sort on.
+	 */
+	SortSupport onlyKey;
+
+	int			sortopt;		/* Bitmask of flags used to setup sort */
+
+	bool		tuples;			/* Can SortTuple.tuple ever be set? */
+
+	void *arg;
+};
+
+typedef struct
+{
+	TupleDesc	tupDesc;
+
+	/*
+	 * These variables are specific to the CLUSTER case; they are set by
+	 * tuplesort_begin_cluster.
+	 */
+	IndexInfo  *indexInfo;		/* info about index being used for reference */
+	EState	   *estate;			/* for evaluating index expressions */
+} TupleSortClusterArg;
+
+typedef struct
+{
+	/*
+	 * These variables are specific to the IndexTuple case; they are set by
+	 * tuplesort_begin_index_xxx and used only by the IndexTuple routines.
+	 */
+	Relation	heapRel;		/* table the index is being built on */
+	Relation	indexRel;		/* index being built */
+} TupleSortIndexArg;
+
+typedef struct
+{
+	TupleSortIndexArg	index;
+
+	/* These are specific to the index_btree subcase: */
+	bool		enforceUnique;	/* complain if we find duplicate tuples */
+	bool		uniqueNullsNotDistinct; /* unique constraint null treatment */
+} TupleSortIndexBTreeArg;
+
+typedef struct
+{
+	TupleSortIndexArg	index;
+
+	/* These are specific to the index_hash subcase: */
+	uint32		high_mask;		/* masks for sortable part of hash code */
+	uint32		low_mask;
+	uint32		max_buckets;
+} TupleSortIndexHashArg;
+
+typedef struct
+{
+	TupleSortIndexArg	index;
+
+	/*
+	 * These variables are specific to the Datum case; they are set by
+	 * tuplesort_begin_datum and used only by the DatumTuple routines.
+	 */
+	Oid			datumType;
+	/* we need typelen in order to know how to copy the Datums. */
+	int			datumTypeLen;
+} TupleSortDatumArg;
+
+/*
+ * Private state of a Tuplesort operation.
+ */
+struct Tuplesortstate
+{
+	TuplesortOps ops;
+	TupSortStatus status;		/* enumerated value as shown above */
+	bool		bounded;		/* did caller specify a maximum number of
+								 * tuples to return? */
+	bool		boundUsed;		/* true if we made use of a bounded heap */
+	int			bound;			/* if bounded, the maximum number of tuples */
+	int64		availMem;		/* remaining memory available, in bytes */
+	int64		allowedMem;		/* total memory allowed, in bytes */
+	int			maxTapes;		/* max number of input tapes to merge in each
+								 * pass */
+	int64		maxSpace;		/* maximum amount of space occupied among sort
+								 * of groups, either in-memory or on-disk */
+	bool		isMaxSpaceDisk; /* true when maxSpace is value for on-disk
+								 * space, false when it's value for in-memory
+								 * space */
+	TupSortStatus maxSpaceStatus;	/* sort status when maxSpace was reached */
+	LogicalTapeSet *tapeset;	/* logtape.c object for tapes in a temp file */
+
 	/*
 	 * This array holds the tuples now in sort memory.  If we are in state
 	 * INITIAL, the tuples are in no particular order; if we are in state
@@ -420,24 +504,6 @@ struct Tuplesortstate
 	Sharedsort *shared;
 	int			nParticipants;
 
-	/*
-	 * The sortKeys variable is used by every case other than the hash index
-	 * case; it is set by tuplesort_begin_xxx.  tupDesc is only used by the
-	 * MinimalTuple and CLUSTER routines, though.
-	 */
-	TupleDesc	tupDesc;
-	SortSupport sortKeys;		/* array of length nKeys */
-
-	/*
-	 * This variable is shared by the single-key MinimalTuple case and the
-	 * Datum case (which both use qsort_ssup()).  Otherwise, it's NULL.  The
-	 * presence of a value in this field is also checked by various sort
-	 * specialization functions as an optimization when comparing the leading
-	 * key in a tiebreak situation to determine if there are any subsequent
-	 * keys to sort on.
-	 */
-	SortSupport onlyKey;
-
 	/*
 	 * Additional state for managing "abbreviated key" sortsupport routines
 	 * (which currently may be used by all cases except the hash index case).
@@ -447,37 +513,6 @@ struct Tuplesortstate
 	int64		abbrevNext;		/* Tuple # at which to next check
 								 * applicability */
 
-	/*
-	 * These variables are specific to the CLUSTER case; they are set by
-	 * tuplesort_begin_cluster.
-	 */
-	IndexInfo  *indexInfo;		/* info about index being used for reference */
-	EState	   *estate;			/* for evaluating index expressions */
-
-	/*
-	 * These variables are specific to the IndexTuple case; they are set by
-	 * tuplesort_begin_index_xxx and used only by the IndexTuple routines.
-	 */
-	Relation	heapRel;		/* table the index is being built on */
-	Relation	indexRel;		/* index being built */
-
-	/* These are specific to the index_btree subcase: */
-	bool		enforceUnique;	/* complain if we find duplicate tuples */
-	bool		uniqueNullsNotDistinct; /* unique constraint null treatment */
-
-	/* These are specific to the index_hash subcase: */
-	uint32		high_mask;		/* masks for sortable part of hash code */
-	uint32		low_mask;
-	uint32		max_buckets;
-
-	/*
-	 * These variables are specific to the Datum case; they are set by
-	 * tuplesort_begin_datum and used only by the DatumTuple routines.
-	 */
-	Oid			datumType;
-	/* we need typelen in order to know how to copy the Datums. */
-	int			datumTypeLen;
-
 	/*
 	 * Resource snapshot for time of sort start.
 	 */
@@ -542,10 +577,13 @@ struct Sharedsort
 			pfree(buf); \
 	} while(0)
 
-#define GETDATUM1(state,stup)	((*(state)->getdatum1) (state, stup))
-#define COMPARETUP(state,a,b)	((*(state)->comparetup) (a, b, state))
-#define WRITETUP(state,tape,stup)	((*(state)->writetup) (state, tape, stup))
-#define READTUP(state,stup,tape,len) ((*(state)->readtup) (state, stup, tape, len))
+#define TuplesortstateGetOps(state) ((TuplesortOps *) state);
+
+#define GETDATUM1(state,stup)	((*(state)->ops.getdatum1) (state, stup))
+#define COMPARETUP(state,a,b)	((*(state)->ops.comparetup) (a, b, state))
+#define WRITETUP(state,tape,stup)	((*(state)->ops.writetup) (state, tape, stup))
+#define READTUP(state,stup,tape,len) ((*(state)->ops.readtup) (state, stup, tape, len))
+#define FREESTATE(state) ((state)->ops.freestate ? (*(state)->ops.freestate) (state) : (void) 0)
 #define LACKMEM(state)		((state)->availMem < 0 && !(state)->slabAllocatorUsed)
 #define USEMEM(state,amt)	((state)->availMem -= (amt))
 #define FREEMEM(state,amt)	((state)->availMem += (amt))
@@ -664,6 +702,7 @@ static void writetup_datum(Tuplesortstate *state, LogicalTape *tape,
 						   SortTuple *stup);
 static void readtup_datum(Tuplesortstate *state, SortTuple *stup,
 						  LogicalTape *tape, unsigned int len);
+static void freestate_cluster(Tuplesortstate *state);
 static int	worker_get_identifier(Tuplesortstate *state);
 static void worker_freeze_result_tape(Tuplesortstate *state);
 static void worker_nomergeruns(Tuplesortstate *state);
@@ -694,7 +733,7 @@ qsort_tuple_unsigned_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
 
 	compare = ApplyUnsignedSortComparator(a->datum1, a->isnull1,
 										  b->datum1, b->isnull1,
-										  &state->sortKeys[0]);
+										  &state->ops.sortKeys[0]);
 	if (compare != 0)
 		return compare;
 
@@ -702,10 +741,10 @@ qsort_tuple_unsigned_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
 	 * No need to waste effort calling the tiebreak function when there are no
 	 * other keys to sort on.
 	 */
-	if (state->onlyKey != NULL)
+	if (state->ops.onlyKey != NULL)
 		return 0;
 
-	return state->comparetup(a, b, state);
+	return state->ops.comparetup(a, b, state);
 }
 
 #if SIZEOF_DATUM >= 8
@@ -717,7 +756,7 @@ qsort_tuple_signed_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
 
 	compare = ApplySignedSortComparator(a->datum1, a->isnull1,
 										b->datum1, b->isnull1,
-										&state->sortKeys[0]);
+										&state->ops.sortKeys[0]);
 
 	if (compare != 0)
 		return compare;
@@ -726,10 +765,10 @@ qsort_tuple_signed_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
 	 * No need to waste effort calling the tiebreak function when there are no
 	 * other keys to sort on.
 	 */
-	if (state->onlyKey != NULL)
+	if (state->ops.onlyKey != NULL)
 		return 0;
 
-	return state->comparetup(a, b, state);
+	return state->ops.comparetup(a, b, state);
 }
 #endif
 
@@ -741,7 +780,7 @@ qsort_tuple_int32_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
 
 	compare = ApplyInt32SortComparator(a->datum1, a->isnull1,
 									   b->datum1, b->isnull1,
-									   &state->sortKeys[0]);
+									   &state->ops.sortKeys[0]);
 
 	if (compare != 0)
 		return compare;
@@ -750,10 +789,10 @@ qsort_tuple_int32_compare(SortTuple *a, SortTuple *b, Tuplesortstate *state)
 	 * No need to waste effort calling the tiebreak function when there are no
 	 * other keys to sort on.
 	 */
-	if (state->onlyKey != NULL)
+	if (state->ops.onlyKey != NULL)
 		return 0;
 
-	return state->comparetup(a, b, state);
+	return state->ops.comparetup(a, b, state);
 }
 
 /*
@@ -880,8 +919,9 @@ tuplesort_begin_common(int workMem, SortCoordinate coordinate, int sortopt)
 		pg_rusage_init(&state->ru_start);
 #endif
 
-	state->sortopt = sortopt;
-	state->tuples = true;
+	state->ops.sortopt = sortopt;
+	state->ops.tuples = true;
+	state->abbrevNext = 10;
 
 	/*
 	 * workMem is forced to be at least 64KB, the current minimum valid value
@@ -890,8 +930,8 @@ tuplesort_begin_common(int workMem, SortCoordinate coordinate, int sortopt)
 	 * with very little memory.
 	 */
 	state->allowedMem = Max(workMem, 64) * (int64) 1024;
-	state->sortcontext = sortcontext;
-	state->maincontext = maincontext;
+	state->ops.sortcontext = sortcontext;
+	state->ops.maincontext = maincontext;
 
 	/*
 	 * Initial size of array must be more than ALLOCSET_SEPARATE_THRESHOLD;
@@ -950,7 +990,7 @@ tuplesort_begin_batch(Tuplesortstate *state)
 {
 	MemoryContext oldcontext;
 
-	oldcontext = MemoryContextSwitchTo(state->maincontext);
+	oldcontext = MemoryContextSwitchTo(state->ops.maincontext);
 
 	/*
 	 * Caller tuple (e.g. IndexTuple) memory context.
@@ -965,12 +1005,12 @@ tuplesort_begin_batch(Tuplesortstate *state)
 	 * generation.c context as this keeps allocations more compact with less
 	 * wastage.  Allocations are also slightly more CPU efficient.
 	 */
-	if (state->sortopt & TUPLESORT_ALLOWBOUNDED)
-		state->tuplecontext = AllocSetContextCreate(state->sortcontext,
+	if (state->ops.sortopt & TUPLESORT_ALLOWBOUNDED)
+		state->ops.tuplecontext = AllocSetContextCreate(state->ops.sortcontext,
 													"Caller tuples",
 													ALLOCSET_DEFAULT_SIZES);
 	else
-		state->tuplecontext = GenerationContextCreate(state->sortcontext,
+		state->ops.tuplecontext = GenerationContextCreate(state->ops.sortcontext,
 													  "Caller tuples",
 													  ALLOCSET_DEFAULT_SIZES);
 
@@ -1028,10 +1068,11 @@ tuplesort_begin_heap(TupleDesc tupDesc,
 {
 	Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
 												   sortopt);
+	TuplesortOps   *ops = TuplesortstateGetOps(state);
 	MemoryContext oldcontext;
 	int			i;
 
-	oldcontext = MemoryContextSwitchTo(state->maincontext);
+	oldcontext = MemoryContextSwitchTo(state->ops.maincontext);
 
 	AssertArg(nkeys > 0);
 
@@ -1042,30 +1083,28 @@ tuplesort_begin_heap(TupleDesc tupDesc,
 			 nkeys, workMem, sortopt & TUPLESORT_RANDOMACCESS ? 't' : 'f');
 #endif
 
-	state->nKeys = nkeys;
+	ops->nKeys = nkeys;
 
 	TRACE_POSTGRESQL_SORT_START(HEAP_SORT,
 								false,	/* no unique check */
 								nkeys,
 								workMem,
 								sortopt & TUPLESORT_RANDOMACCESS,
-								PARALLEL_SORT(state));
+								PARALLEL_SORT(coordinate));
 
-	state->getdatum1 = getdatum1_heap;
-	state->comparetup = comparetup_heap;
-	state->writetup = writetup_heap;
-	state->readtup = readtup_heap;
-	state->haveDatum1 = true;
-
-	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
-	state->abbrevNext = 10;
+	ops->getdatum1 = getdatum1_heap;
+	ops->comparetup = comparetup_heap;
+	ops->writetup = writetup_heap;
+	ops->readtup = readtup_heap;
+	ops->haveDatum1 = true;
+	ops->arg = tupDesc;	/* assume we need not copy tupDesc */
 
 	/* Prepare SortSupport data for each column */
-	state->sortKeys = (SortSupport) palloc0(nkeys * sizeof(SortSupportData));
+	ops->sortKeys = (SortSupport) palloc0(nkeys * sizeof(SortSupportData));
 
 	for (i = 0; i < nkeys; i++)
 	{
-		SortSupport sortKey = state->sortKeys + i;
+		SortSupport sortKey = ops->sortKeys + i;
 
 		AssertArg(attNums[i] != 0);
 		AssertArg(sortOperators[i] != 0);
@@ -1075,7 +1114,7 @@ tuplesort_begin_heap(TupleDesc tupDesc,
 		sortKey->ssup_nulls_first = nullsFirstFlags[i];
 		sortKey->ssup_attno = attNums[i];
 		/* Convey if abbreviation optimization is applicable in principle */
-		sortKey->abbreviate = (i == 0 && state->haveDatum1);
+		sortKey->abbreviate = (i == 0 && ops->haveDatum1);
 
 		PrepareSortSupportFromOrderingOp(sortOperators[i], sortKey);
 	}
@@ -1086,8 +1125,8 @@ tuplesort_begin_heap(TupleDesc tupDesc,
 	 * is only of value to pass-by-value types anyway, whereas abbreviated
 	 * keys are typically only of value to pass-by-reference types.
 	 */
-	if (nkeys == 1 && !state->sortKeys->abbrev_converter)
-		state->onlyKey = state->sortKeys;
+	if (nkeys == 1 && !ops->sortKeys->abbrev_converter)
+		ops->onlyKey = ops->sortKeys;
 
 	MemoryContextSwitchTo(oldcontext);
 
@@ -1102,13 +1141,16 @@ tuplesort_begin_cluster(TupleDesc tupDesc,
 {
 	Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
 												   sortopt);
+	TuplesortOps   *ops = TuplesortstateGetOps(state);
 	BTScanInsert indexScanKey;
 	MemoryContext oldcontext;
+	TupleSortClusterArg *arg;
 	int			i;
 
 	Assert(indexRel->rd_rel->relam == BTREE_AM_OID);
 
-	oldcontext = MemoryContextSwitchTo(state->maincontext);
+	oldcontext = MemoryContextSwitchTo(state->ops.maincontext);
+	arg = (TupleSortClusterArg *) palloc0(sizeof(TupleSortClusterArg));
 
 #ifdef TRACE_SORT
 	if (trace_sort)
@@ -1118,37 +1160,38 @@ tuplesort_begin_cluster(TupleDesc tupDesc,
 			 workMem, sortopt & TUPLESORT_RANDOMACCESS ? 't' : 'f');
 #endif
 
-	state->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
+	ops->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
 
 	TRACE_POSTGRESQL_SORT_START(CLUSTER_SORT,
 								false,	/* no unique check */
-								state->nKeys,
+								ops->nKeys,
 								workMem,
 								sortopt & TUPLESORT_RANDOMACCESS,
-								PARALLEL_SORT(state));
+								PARALLEL_SORT(coordinate));
 
-	state->getdatum1 = getdatum1_cluster;
-	state->comparetup = comparetup_cluster;
-	state->writetup = writetup_cluster;
-	state->readtup = readtup_cluster;
-	state->abbrevNext = 10;
+	ops->getdatum1 = getdatum1_cluster;
+	ops->comparetup = comparetup_cluster;
+	ops->writetup = writetup_cluster;
+	ops->readtup = readtup_cluster;
+	ops->freestate = freestate_cluster;
+	ops->arg = arg;
 
-	state->indexInfo = BuildIndexInfo(indexRel);
+	arg->indexInfo = BuildIndexInfo(indexRel);
 
 	/*
 	 * If we don't have a simple leading attribute, we don't currently
 	 * initialize datum1, so disable optimizations that require it.
 	 */
-	if (state->indexInfo->ii_IndexAttrNumbers[0] == 0)
-		state->haveDatum1 = false;
+	if (arg->indexInfo->ii_IndexAttrNumbers[0] == 0)
+		ops->haveDatum1 = false;
 	else
-		state->haveDatum1 = true;
+		ops->haveDatum1 = true;
 
-	state->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
+	arg->tupDesc = tupDesc;	/* assume we need not copy tupDesc */
 
 	indexScanKey = _bt_mkscankey(indexRel, NULL);
 
-	if (state->indexInfo->ii_Expressions != NULL)
+	if (arg->indexInfo->ii_Expressions != NULL)
 	{
 		TupleTableSlot *slot;
 		ExprContext *econtext;
@@ -1159,19 +1202,19 @@ tuplesort_begin_cluster(TupleDesc tupDesc,
 		 * TupleTableSlot to put the table tuples into.  The econtext's
 		 * scantuple has to point to that slot, too.
 		 */
-		state->estate = CreateExecutorState();
+		arg->estate = CreateExecutorState();
 		slot = MakeSingleTupleTableSlot(tupDesc, &TTSOpsHeapTuple);
-		econtext = GetPerTupleExprContext(state->estate);
+		econtext = GetPerTupleExprContext(arg->estate);
 		econtext->ecxt_scantuple = slot;
 	}
 
 	/* Prepare SortSupport data for each column */
-	state->sortKeys = (SortSupport) palloc0(state->nKeys *
+	ops->sortKeys = (SortSupport) palloc0(ops->nKeys *
 											sizeof(SortSupportData));
 
-	for (i = 0; i < state->nKeys; i++)
+	for (i = 0; i < ops->nKeys; i++)
 	{
-		SortSupport sortKey = state->sortKeys + i;
+		SortSupport sortKey = ops->sortKeys + i;
 		ScanKey		scanKey = indexScanKey->scankeys + i;
 		int16		strategy;
 
@@ -1181,7 +1224,7 @@ tuplesort_begin_cluster(TupleDesc tupDesc,
 			(scanKey->sk_flags & SK_BT_NULLS_FIRST) != 0;
 		sortKey->ssup_attno = scanKey->sk_attno;
 		/* Convey if abbreviation optimization is applicable in principle */
-		sortKey->abbreviate = (i == 0 && state->haveDatum1);
+		sortKey->abbreviate = (i == 0 && ops->haveDatum1);
 
 		AssertState(sortKey->ssup_attno != 0);
 
@@ -1209,11 +1252,14 @@ tuplesort_begin_index_btree(Relation heapRel,
 {
 	Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
 												   sortopt);
+	TuplesortOps   *ops = TuplesortstateGetOps(state);
 	BTScanInsert indexScanKey;
+	TupleSortIndexBTreeArg *arg;
 	MemoryContext oldcontext;
 	int			i;
 
-	oldcontext = MemoryContextSwitchTo(state->maincontext);
+	oldcontext = MemoryContextSwitchTo(state->ops.maincontext);
+	arg = (TupleSortIndexBTreeArg *) palloc(sizeof(TupleSortIndexBTreeArg));
 
 #ifdef TRACE_SORT
 	if (trace_sort)
@@ -1223,36 +1269,36 @@ tuplesort_begin_index_btree(Relation heapRel,
 			 workMem, sortopt & TUPLESORT_RANDOMACCESS ? 't' : 'f');
 #endif
 
-	state->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
+	ops->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
 
 	TRACE_POSTGRESQL_SORT_START(INDEX_SORT,
 								enforceUnique,
 								state->nKeys,
 								workMem,
 								sortopt & TUPLESORT_RANDOMACCESS,
-								PARALLEL_SORT(state));
+								PARALLEL_SORT(coordinate));
 
-	state->getdatum1 = getdatum1_index;
-	state->comparetup = comparetup_index_btree;
-	state->writetup = writetup_index;
-	state->readtup = readtup_index;
-	state->abbrevNext = 10;
-	state->haveDatum1 = true;
+	ops->getdatum1 = getdatum1_index;
+	ops->comparetup = comparetup_index_btree;
+	ops->writetup = writetup_index;
+	ops->readtup = readtup_index;
+	ops->haveDatum1 = true;
+	ops->arg = arg;
 
-	state->heapRel = heapRel;
-	state->indexRel = indexRel;
-	state->enforceUnique = enforceUnique;
-	state->uniqueNullsNotDistinct = uniqueNullsNotDistinct;
+	arg->index.heapRel = heapRel;
+	arg->index.indexRel = indexRel;
+	arg->enforceUnique = enforceUnique;
+	arg->uniqueNullsNotDistinct = uniqueNullsNotDistinct;
 
 	indexScanKey = _bt_mkscankey(indexRel, NULL);
 
 	/* Prepare SortSupport data for each column */
-	state->sortKeys = (SortSupport) palloc0(state->nKeys *
-											sizeof(SortSupportData));
+	ops->sortKeys = (SortSupport) palloc0(ops->nKeys *
+												sizeof(SortSupportData));
 
-	for (i = 0; i < state->nKeys; i++)
+	for (i = 0; i < ops->nKeys; i++)
 	{
-		SortSupport sortKey = state->sortKeys + i;
+		SortSupport sortKey = ops->sortKeys + i;
 		ScanKey		scanKey = indexScanKey->scankeys + i;
 		int16		strategy;
 
@@ -1262,7 +1308,7 @@ tuplesort_begin_index_btree(Relation heapRel,
 			(scanKey->sk_flags & SK_BT_NULLS_FIRST) != 0;
 		sortKey->ssup_attno = scanKey->sk_attno;
 		/* Convey if abbreviation optimization is applicable in principle */
-		sortKey->abbreviate = (i == 0 && state->haveDatum1);
+		sortKey->abbreviate = (i == 0 && ops->haveDatum1);
 
 		AssertState(sortKey->ssup_attno != 0);
 
@@ -1291,9 +1337,12 @@ tuplesort_begin_index_hash(Relation heapRel,
 {
 	Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
 												   sortopt);
+	TuplesortOps   *ops = TuplesortstateGetOps(state);
 	MemoryContext oldcontext;
+	TupleSortIndexHashArg *arg;
 
-	oldcontext = MemoryContextSwitchTo(state->maincontext);
+	oldcontext = MemoryContextSwitchTo(state->ops.maincontext);
+	arg = (TupleSortIndexHashArg *) palloc(sizeof(TupleSortIndexHashArg));
 
 #ifdef TRACE_SORT
 	if (trace_sort)
@@ -1307,20 +1356,21 @@ tuplesort_begin_index_hash(Relation heapRel,
 			 sortopt & TUPLESORT_RANDOMACCESS ? 't' : 'f');
 #endif
 
-	state->nKeys = 1;			/* Only one sort column, the hash code */
+	ops->nKeys = 1;			/* Only one sort column, the hash code */
 
-	state->getdatum1 = getdatum1_index;
-	state->comparetup = comparetup_index_hash;
-	state->writetup = writetup_index;
-	state->readtup = readtup_index;
-	state->haveDatum1 = true;
+	ops->getdatum1 = getdatum1_index;
+	ops->comparetup = comparetup_index_hash;
+	ops->writetup = writetup_index;
+	ops->readtup = readtup_index;
+	ops->haveDatum1 = true;
+	ops->arg = arg;
 
-	state->heapRel = heapRel;
-	state->indexRel = indexRel;
+	arg->index.heapRel = heapRel;
+	arg->index.indexRel = indexRel;
 
-	state->high_mask = high_mask;
-	state->low_mask = low_mask;
-	state->max_buckets = max_buckets;
+	arg->high_mask = high_mask;
+	arg->low_mask = low_mask;
+	arg->max_buckets = max_buckets;
 
 	MemoryContextSwitchTo(oldcontext);
 
@@ -1336,10 +1386,13 @@ tuplesort_begin_index_gist(Relation heapRel,
 {
 	Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
 												   sortopt);
+	TuplesortOps   *ops = TuplesortstateGetOps(state);
 	MemoryContext oldcontext;
+	TupleSortIndexBTreeArg *arg;
 	int			i;
 
-	oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	oldcontext = MemoryContextSwitchTo(state->ops.maincontext);
+	arg = (TupleSortIndexBTreeArg *) palloc(sizeof(TupleSortIndexBTreeArg));
 
 #ifdef TRACE_SORT
 	if (trace_sort)
@@ -1348,31 +1401,34 @@ tuplesort_begin_index_gist(Relation heapRel,
 			 workMem, sortopt & TUPLESORT_RANDOMACCESS ? 't' : 'f');
 #endif
 
-	state->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
+	ops->nKeys = IndexRelationGetNumberOfKeyAttributes(indexRel);
 
-	state->getdatum1 = getdatum1_index;
-	state->comparetup = comparetup_index_btree;
-	state->writetup = writetup_index;
-	state->readtup = readtup_index;
-	state->haveDatum1 = true;
+	ops->getdatum1 = getdatum1_index;
+	ops->comparetup = comparetup_index_btree;
+	ops->writetup = writetup_index;
+	ops->readtup = readtup_index;
+	ops->haveDatum1 = true;
+	ops->arg = arg;
 
-	state->heapRel = heapRel;
-	state->indexRel = indexRel;
+	arg->index.heapRel = heapRel;
+	arg->index.indexRel = indexRel;
+	arg->enforceUnique = false;
+	arg->uniqueNullsNotDistinct = false;
 
 	/* Prepare SortSupport data for each column */
-	state->sortKeys = (SortSupport) palloc0(state->nKeys *
-											sizeof(SortSupportData));
+	ops->sortKeys = (SortSupport) palloc0(ops->nKeys *
+										  sizeof(SortSupportData));
 
-	for (i = 0; i < state->nKeys; i++)
+	for (i = 0; i < ops->nKeys; i++)
 	{
-		SortSupport sortKey = state->sortKeys + i;
+		SortSupport sortKey = ops->sortKeys + i;
 
 		sortKey->ssup_cxt = CurrentMemoryContext;
 		sortKey->ssup_collation = indexRel->rd_indcollation[i];
 		sortKey->ssup_nulls_first = false;
 		sortKey->ssup_attno = i + 1;
 		/* Convey if abbreviation optimization is applicable in principle */
-		sortKey->abbreviate = (i == 0 && state->haveDatum1);
+		sortKey->abbreviate = (i == 0 && ops->haveDatum1);
 
 		AssertState(sortKey->ssup_attno != 0);
 
@@ -1392,11 +1448,14 @@ tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
 {
 	Tuplesortstate *state = tuplesort_begin_common(workMem, coordinate,
 												   sortopt);
+	TuplesortOps   *ops = TuplesortstateGetOps(state);
+	TupleSortDatumArg *arg;
 	MemoryContext oldcontext;
 	int16		typlen;
 	bool		typbyval;
 
-	oldcontext = MemoryContextSwitchTo(state->maincontext);
+	oldcontext = MemoryContextSwitchTo(state->ops.maincontext);
+	arg = (TupleSortDatumArg *) palloc(sizeof(TupleSortDatumArg));
 
 #ifdef TRACE_SORT
 	if (trace_sort)
@@ -1405,35 +1464,36 @@ tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
 			 workMem, sortopt & TUPLESORT_RANDOMACCESS ? 't' : 'f');
 #endif
 
-	state->nKeys = 1;			/* always a one-column sort */
+	ops->nKeys = 1;			/* always a one-column sort */
 
 	TRACE_POSTGRESQL_SORT_START(DATUM_SORT,
 								false,	/* no unique check */
 								1,
 								workMem,
 								sortopt & TUPLESORT_RANDOMACCESS,
-								PARALLEL_SORT(state));
+								PARALLEL_SORT(coordinate));
 
-	state->getdatum1 = getdatum1_datum;
-	state->comparetup = comparetup_datum;
-	state->writetup = writetup_datum;
-	state->readtup = readtup_datum;
+	ops->getdatum1 = getdatum1_datum;
+	ops->comparetup = comparetup_datum;
+	ops->writetup = writetup_datum;
+	ops->readtup = readtup_datum;
 	state->abbrevNext = 10;
-	state->haveDatum1 = true;
+	ops->haveDatum1 = true;
+	ops->arg = arg;
 
-	state->datumType = datumType;
+	arg->datumType = datumType;
 
 	/* lookup necessary attributes of the datum type */
 	get_typlenbyval(datumType, &typlen, &typbyval);
-	state->datumTypeLen = typlen;
-	state->tuples = !typbyval;
+	arg->datumTypeLen = typlen;
+	ops->tuples = !typbyval;
 
 	/* Prepare SortSupport data */
-	state->sortKeys = (SortSupport) palloc0(sizeof(SortSupportData));
+	ops->sortKeys = (SortSupport) palloc0(sizeof(SortSupportData));
 
-	state->sortKeys->ssup_cxt = CurrentMemoryContext;
-	state->sortKeys->ssup_collation = sortCollation;
-	state->sortKeys->ssup_nulls_first = nullsFirstFlag;
+	ops->sortKeys->ssup_cxt = CurrentMemoryContext;
+	ops->sortKeys->ssup_collation = sortCollation;
+	ops->sortKeys->ssup_nulls_first = nullsFirstFlag;
 
 	/*
 	 * Abbreviation is possible here only for by-reference types.  In theory,
@@ -1443,9 +1503,9 @@ tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
 	 * can't, because a datum sort only stores a single copy of the datum; the
 	 * "tuple" field of each SortTuple is NULL.
 	 */
-	state->sortKeys->abbreviate = !typbyval;
+	ops->sortKeys->abbreviate = !typbyval;
 
-	PrepareSortSupportFromOrderingOp(sortOperator, state->sortKeys);
+	PrepareSortSupportFromOrderingOp(sortOperator, ops->sortKeys);
 
 	/*
 	 * The "onlyKey" optimization cannot be used with abbreviated keys, since
@@ -1453,8 +1513,8 @@ tuplesort_begin_datum(Oid datumType, Oid sortOperator, Oid sortCollation,
 	 * is only of value to pass-by-value types anyway, whereas abbreviated
 	 * keys are typically only of value to pass-by-reference types.
 	 */
-	if (!state->sortKeys->abbrev_converter)
-		state->onlyKey = state->sortKeys;
+	if (!ops->sortKeys->abbrev_converter)
+		ops->onlyKey = ops->sortKeys;
 
 	MemoryContextSwitchTo(oldcontext);
 
@@ -1479,7 +1539,7 @@ tuplesort_set_bound(Tuplesortstate *state, int64 bound)
 	/* Assert we're called before loading any tuples */
 	Assert(state->status == TSS_INITIAL && state->memtupcount == 0);
 	/* Assert we allow bounded sorts */
-	Assert(state->sortopt & TUPLESORT_ALLOWBOUNDED);
+	Assert(state->ops.sortopt & TUPLESORT_ALLOWBOUNDED);
 	/* Can't set the bound twice, either */
 	Assert(!state->bounded);
 	/* Also, this shouldn't be called in a parallel worker */
@@ -1507,13 +1567,13 @@ tuplesort_set_bound(Tuplesortstate *state, int64 bound)
 	 * optimization.  Disable by setting state to be consistent with no
 	 * abbreviation support.
 	 */
-	state->sortKeys->abbrev_converter = NULL;
-	if (state->sortKeys->abbrev_full_comparator)
-		state->sortKeys->comparator = state->sortKeys->abbrev_full_comparator;
+	state->ops.sortKeys->abbrev_converter = NULL;
+	if (state->ops.sortKeys->abbrev_full_comparator)
+		state->ops.sortKeys->comparator = state->ops.sortKeys->abbrev_full_comparator;
 
 	/* Not strictly necessary, but be tidy */
-	state->sortKeys->abbrev_abort = NULL;
-	state->sortKeys->abbrev_full_comparator = NULL;
+	state->ops.sortKeys->abbrev_abort = NULL;
+	state->ops.sortKeys->abbrev_full_comparator = NULL;
 }
 
 /*
@@ -1536,7 +1596,7 @@ static void
 tuplesort_free(Tuplesortstate *state)
 {
 	/* context swap probably not needed, but let's be safe */
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 
 #ifdef TRACE_SORT
 	long		spaceUsed;
@@ -1583,21 +1643,13 @@ tuplesort_free(Tuplesortstate *state)
 	TRACE_POSTGRESQL_SORT_DONE(state->tapeset != NULL, 0L);
 #endif
 
-	/* Free any execution state created for CLUSTER case */
-	if (state->estate != NULL)
-	{
-		ExprContext *econtext = GetPerTupleExprContext(state->estate);
-
-		ExecDropSingleTupleTableSlot(econtext->ecxt_scantuple);
-		FreeExecutorState(state->estate);
-	}
-
+	FREESTATE(state);
 	MemoryContextSwitchTo(oldcontext);
 
 	/*
 	 * Free the per-sort memory context, thereby releasing all working memory.
 	 */
-	MemoryContextReset(state->sortcontext);
+	MemoryContextReset(state->ops.sortcontext);
 }
 
 /*
@@ -1618,7 +1670,7 @@ tuplesort_end(Tuplesortstate *state)
 	 * Free the main memory context, including the Tuplesortstate struct
 	 * itself.
 	 */
-	MemoryContextDelete(state->maincontext);
+	MemoryContextDelete(state->ops.maincontext);
 }
 
 /*
@@ -1832,7 +1884,9 @@ noalloc:
 void
 tuplesort_puttupleslot(Tuplesortstate *state, TupleTableSlot *slot)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->tuplecontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.tuplecontext);
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleDesc	tupDesc = (TupleDesc) ops->arg;
 	SortTuple	stup;
 	MinimalTuple tuple;
 	HeapTupleData htup;
@@ -1844,8 +1898,8 @@ tuplesort_puttupleslot(Tuplesortstate *state, TupleTableSlot *slot)
 	htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
 	htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
 	stup.datum1 = heap_getattr(&htup,
-							   state->sortKeys[0].ssup_attno,
-							   state->tupDesc,
+							   ops->sortKeys[0].ssup_attno,
+							   tupDesc,
 							   &stup.isnull1);
 
 	puttuple_common(state, &stup);
@@ -1862,7 +1916,9 @@ void
 tuplesort_putheaptuple(Tuplesortstate *state, HeapTuple tup)
 {
 	SortTuple	stup;
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->tuplecontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.tuplecontext);
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortClusterArg *arg = (TupleSortClusterArg *) ops->arg;
 
 	/* copy the tuple into sort storage */
 	tup = heap_copytuple(tup);
@@ -1872,11 +1928,11 @@ tuplesort_putheaptuple(Tuplesortstate *state, HeapTuple tup)
 	 * set up first-column key value, and potentially abbreviate, if it's a
 	 * simple column
 	 */
-	if (state->haveDatum1)
+	if (ops->haveDatum1)
 	{
 		stup.datum1 = heap_getattr(tup,
-								   state->indexInfo->ii_IndexAttrNumbers[0],
-								   state->tupDesc,
+								   arg->indexInfo->ii_IndexAttrNumbers[0],
+								   arg->tupDesc,
 								   &stup.isnull1);
 	}
 
@@ -1894,9 +1950,11 @@ tuplesort_putindextuplevalues(Tuplesortstate *state, Relation rel,
 							  ItemPointer self, Datum *values,
 							  bool *isnull)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->tuplecontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.tuplecontext);
 	SortTuple	stup;
 	IndexTuple	tuple;
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortIndexArg *arg = (TupleSortIndexArg *) ops->arg;
 
 	stup.tuple = index_form_tuple(RelationGetDescr(rel), values, isnull);
 	tuple = ((IndexTuple) stup.tuple);
@@ -1904,7 +1962,7 @@ tuplesort_putindextuplevalues(Tuplesortstate *state, Relation rel,
 	/* set up first-column key value */
 	stup.datum1 = index_getattr(tuple,
 								1,
-								RelationGetDescr(state->indexRel),
+								RelationGetDescr(arg->indexRel),
 								&stup.isnull1);
 
 	puttuple_common(state, &stup);
@@ -1920,7 +1978,9 @@ tuplesort_putindextuplevalues(Tuplesortstate *state, Relation rel,
 void
 tuplesort_putdatum(Tuplesortstate *state, Datum val, bool isNull)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->tuplecontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.tuplecontext);
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortDatumArg *arg = (TupleSortDatumArg *) ops->arg;
 	SortTuple	stup;
 
 	/*
@@ -1935,7 +1995,7 @@ tuplesort_putdatum(Tuplesortstate *state, Datum val, bool isNull)
 	 * identical to stup.tuple.
 	 */
 
-	if (isNull || !state->tuples)
+	if (isNull || !state->ops.tuples)
 	{
 		/*
 		 * Set datum1 to zeroed representation for NULLs (to be consistent,
@@ -1948,7 +2008,7 @@ tuplesort_putdatum(Tuplesortstate *state, Datum val, bool isNull)
 	else
 	{
 		stup.isnull1 = false;
-		stup.datum1 = datumCopy(val, false, state->datumTypeLen);
+		stup.datum1 = datumCopy(val, false, arg->datumTypeLen);
 		stup.tuple = DatumGetPointer(stup.datum1);
 	}
 
@@ -1963,15 +2023,15 @@ tuplesort_putdatum(Tuplesortstate *state, Datum val, bool isNull)
 static void
 puttuple_common(Tuplesortstate *state, SortTuple *tuple)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 
 	Assert(!LEADER(state));
 
 	if (tuple->tuple != NULL)
 		USEMEM(state, GetMemoryChunkSpace(tuple->tuple));
 
-	if (!state->sortKeys || !state->haveDatum1 || !state->tuples ||
-		!state->sortKeys->abbrev_converter || tuple->isnull1)
+	if (!state->ops.sortKeys || !state->ops.haveDatum1 || !state->ops.tuples ||
+		!state->ops.sortKeys->abbrev_converter || tuple->isnull1)
 	{
 		/*
 		 * Store ordinary Datum representation, or NULL value.  If there is a
@@ -1985,8 +2045,8 @@ puttuple_common(Tuplesortstate *state, SortTuple *tuple)
 	else if (!consider_abort_common(state))
 	{
 		/* Store abbreviated key representation */
-		tuple->datum1 = state->sortKeys->abbrev_converter(tuple->datum1,
-														  state->sortKeys);
+		tuple->datum1 = state->ops.sortKeys->abbrev_converter(tuple->datum1,
+															  state->ops.sortKeys);
 	}
 	else
 	{
@@ -2130,9 +2190,9 @@ writetuple_common(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 static bool
 consider_abort_common(Tuplesortstate *state)
 {
-	Assert(state->sortKeys[0].abbrev_converter != NULL);
-	Assert(state->sortKeys[0].abbrev_abort != NULL);
-	Assert(state->sortKeys[0].abbrev_full_comparator != NULL);
+	Assert(state->ops.sortKeys[0].abbrev_converter != NULL);
+	Assert(state->ops.sortKeys[0].abbrev_abort != NULL);
+	Assert(state->ops.sortKeys[0].abbrev_full_comparator != NULL);
 
 	/*
 	 * Check effectiveness of abbreviation optimization.  Consider aborting
@@ -2147,19 +2207,19 @@ consider_abort_common(Tuplesortstate *state)
 		 * Check opclass-supplied abbreviation abort routine.  It may indicate
 		 * that abbreviation should not proceed.
 		 */
-		if (!state->sortKeys->abbrev_abort(state->memtupcount,
-										   state->sortKeys))
+		if (!state->ops.sortKeys->abbrev_abort(state->memtupcount,
+											   state->ops.sortKeys))
 			return false;
 
 		/*
 		 * Finally, restore authoritative comparator, and indicate that
 		 * abbreviation is not in play by setting abbrev_converter to NULL
 		 */
-		state->sortKeys[0].comparator = state->sortKeys[0].abbrev_full_comparator;
-		state->sortKeys[0].abbrev_converter = NULL;
+		state->ops.sortKeys[0].comparator = state->ops.sortKeys[0].abbrev_full_comparator;
+		state->ops.sortKeys[0].abbrev_converter = NULL;
 		/* Not strictly necessary, but be tidy */
-		state->sortKeys[0].abbrev_abort = NULL;
-		state->sortKeys[0].abbrev_full_comparator = NULL;
+		state->ops.sortKeys[0].abbrev_abort = NULL;
+		state->ops.sortKeys[0].abbrev_full_comparator = NULL;
 
 		/* Give up - expect original pass-by-value representation */
 		return true;
@@ -2174,7 +2234,7 @@ consider_abort_common(Tuplesortstate *state)
 void
 tuplesort_performsort(Tuplesortstate *state)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 
 #ifdef TRACE_SORT
 	if (trace_sort)
@@ -2294,7 +2354,7 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
 	switch (state->status)
 	{
 		case TSS_SORTEDINMEM:
-			Assert(forward || state->sortopt & TUPLESORT_RANDOMACCESS);
+			Assert(forward || state->ops.sortopt & TUPLESORT_RANDOMACCESS);
 			Assert(!state->slabAllocatorUsed);
 			if (forward)
 			{
@@ -2338,7 +2398,7 @@ tuplesort_gettuple_common(Tuplesortstate *state, bool forward,
 			break;
 
 		case TSS_SORTEDONTAPE:
-			Assert(forward || state->sortopt & TUPLESORT_RANDOMACCESS);
+			Assert(forward || state->ops.sortopt & TUPLESORT_RANDOMACCESS);
 			Assert(state->slabAllocatorUsed);
 
 			/*
@@ -2540,7 +2600,7 @@ bool
 tuplesort_gettupleslot(Tuplesortstate *state, bool forward, bool copy,
 					   TupleTableSlot *slot, Datum *abbrev)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 	SortTuple	stup;
 
 	if (!tuplesort_gettuple_common(state, forward, &stup))
@@ -2551,7 +2611,7 @@ tuplesort_gettupleslot(Tuplesortstate *state, bool forward, bool copy,
 	if (stup.tuple)
 	{
 		/* Record abbreviated key for caller */
-		if (state->sortKeys->abbrev_converter && abbrev)
+		if (state->ops.sortKeys->abbrev_converter && abbrev)
 			*abbrev = stup.datum1;
 
 		if (copy)
@@ -2576,7 +2636,7 @@ tuplesort_gettupleslot(Tuplesortstate *state, bool forward, bool copy,
 HeapTuple
 tuplesort_getheaptuple(Tuplesortstate *state, bool forward)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 	SortTuple	stup;
 
 	if (!tuplesort_gettuple_common(state, forward, &stup))
@@ -2596,7 +2656,7 @@ tuplesort_getheaptuple(Tuplesortstate *state, bool forward)
 IndexTuple
 tuplesort_getindextuple(Tuplesortstate *state, bool forward)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 	SortTuple	stup;
 
 	if (!tuplesort_gettuple_common(state, forward, &stup))
@@ -2626,7 +2686,9 @@ bool
 tuplesort_getdatum(Tuplesortstate *state, bool forward,
 				   Datum *val, bool *isNull, Datum *abbrev)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortDatumArg *arg = (TupleSortDatumArg *) ops->arg;
 	SortTuple	stup;
 
 	if (!tuplesort_gettuple_common(state, forward, &stup))
@@ -2639,10 +2701,10 @@ tuplesort_getdatum(Tuplesortstate *state, bool forward,
 	MemoryContextSwitchTo(oldcontext);
 
 	/* Record abbreviated key for caller */
-	if (state->sortKeys->abbrev_converter && abbrev)
+	if (ops->sortKeys->abbrev_converter && abbrev)
 		*abbrev = stup.datum1;
 
-	if (stup.isnull1 || !state->tuples)
+	if (stup.isnull1 || !state->ops.tuples)
 	{
 		*val = stup.datum1;
 		*isNull = stup.isnull1;
@@ -2650,7 +2712,7 @@ tuplesort_getdatum(Tuplesortstate *state, bool forward,
 	else
 	{
 		/* use stup.tuple because stup.datum1 may be an abbreviation */
-		*val = datumCopy(PointerGetDatum(stup.tuple), false, state->datumTypeLen);
+		*val = datumCopy(PointerGetDatum(stup.tuple), false, arg->datumTypeLen);
 		*isNull = false;
 	}
 
@@ -2703,7 +2765,7 @@ tuplesort_skiptuples(Tuplesortstate *state, int64 ntuples, bool forward)
 			 * We could probably optimize these cases better, but for now it's
 			 * not worth the trouble.
 			 */
-			oldcontext = MemoryContextSwitchTo(state->sortcontext);
+			oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 			while (ntuples-- > 0)
 			{
 				SortTuple	stup;
@@ -2979,7 +3041,7 @@ mergeruns(Tuplesortstate *state)
 	Assert(state->status == TSS_BUILDRUNS);
 	Assert(state->memtupcount == 0);
 
-	if (state->sortKeys != NULL && state->sortKeys->abbrev_converter != NULL)
+	if (state->ops.sortKeys != NULL && state->ops.sortKeys->abbrev_converter != NULL)
 	{
 		/*
 		 * If there are multiple runs to be merged, when we go to read back
@@ -2987,19 +3049,19 @@ mergeruns(Tuplesortstate *state)
 		 * we don't care to regenerate them.  Disable abbreviation from this
 		 * point on.
 		 */
-		state->sortKeys->abbrev_converter = NULL;
-		state->sortKeys->comparator = state->sortKeys->abbrev_full_comparator;
+		state->ops.sortKeys->abbrev_converter = NULL;
+		state->ops.sortKeys->comparator = state->ops.sortKeys->abbrev_full_comparator;
 
 		/* Not strictly necessary, but be tidy */
-		state->sortKeys->abbrev_abort = NULL;
-		state->sortKeys->abbrev_full_comparator = NULL;
+		state->ops.sortKeys->abbrev_abort = NULL;
+		state->ops.sortKeys->abbrev_full_comparator = NULL;
 	}
 
 	/*
 	 * Reset tuple memory.  We've freed all the tuples that we previously
 	 * allocated.  We will use the slab allocator from now on.
 	 */
-	MemoryContextResetOnly(state->tuplecontext);
+	MemoryContextResetOnly(state->ops.tuplecontext);
 
 	/*
 	 * We no longer need a large memtuples array.  (We will allocate a smaller
@@ -3022,7 +3084,7 @@ mergeruns(Tuplesortstate *state)
 	 * From this point on, we no longer use the USEMEM()/LACKMEM() mechanism
 	 * to track memory usage of individual tuples.
 	 */
-	if (state->tuples)
+	if (state->ops.tuples)
 		init_slab_allocator(state, state->nOutputTapes + 1);
 	else
 		init_slab_allocator(state, 0);
@@ -3036,7 +3098,7 @@ mergeruns(Tuplesortstate *state)
 	 * number of input tapes will not increase between passes.)
 	 */
 	state->memtupsize = state->nOutputTapes;
-	state->memtuples = (SortTuple *) MemoryContextAlloc(state->maincontext,
+	state->memtuples = (SortTuple *) MemoryContextAlloc(state->ops.maincontext,
 														state->nOutputTapes * sizeof(SortTuple));
 	USEMEM(state, GetMemoryChunkSpace(state->memtuples));
 
@@ -3113,7 +3175,7 @@ mergeruns(Tuplesortstate *state)
 			 * sorted tape, we can stop at this point and do the final merge
 			 * on-the-fly.
 			 */
-			if ((state->sortopt & TUPLESORT_RANDOMACCESS) == 0
+			if ((state->ops.sortopt & TUPLESORT_RANDOMACCESS) == 0
 				&& state->nInputRuns <= state->nInputTapes
 				&& !WORKER(state))
 			{
@@ -3339,7 +3401,7 @@ dumptuples(Tuplesortstate *state, bool alltuples)
 	 * AllocSetFree's bucketing by size class might be particularly bad if
 	 * this step wasn't taken.
 	 */
-	MemoryContextReset(state->tuplecontext);
+	MemoryContextReset(state->ops.tuplecontext);
 
 	markrunend(state->destTape);
 
@@ -3357,9 +3419,9 @@ dumptuples(Tuplesortstate *state, bool alltuples)
 void
 tuplesort_rescan(Tuplesortstate *state)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 
-	Assert(state->sortopt & TUPLESORT_RANDOMACCESS);
+	Assert(state->ops.sortopt & TUPLESORT_RANDOMACCESS);
 
 	switch (state->status)
 	{
@@ -3390,9 +3452,9 @@ tuplesort_rescan(Tuplesortstate *state)
 void
 tuplesort_markpos(Tuplesortstate *state)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 
-	Assert(state->sortopt & TUPLESORT_RANDOMACCESS);
+	Assert(state->ops.sortopt & TUPLESORT_RANDOMACCESS);
 
 	switch (state->status)
 	{
@@ -3421,9 +3483,9 @@ tuplesort_markpos(Tuplesortstate *state)
 void
 tuplesort_restorepos(Tuplesortstate *state)
 {
-	MemoryContext oldcontext = MemoryContextSwitchTo(state->sortcontext);
+	MemoryContext oldcontext = MemoryContextSwitchTo(state->ops.sortcontext);
 
-	Assert(state->sortopt & TUPLESORT_RANDOMACCESS);
+	Assert(state->ops.sortopt & TUPLESORT_RANDOMACCESS);
 
 	switch (state->status)
 	{
@@ -3639,9 +3701,9 @@ tuplesort_sort_memtuples(Tuplesortstate *state)
 		 * Do we have the leading column's value or abbreviation in datum1,
 		 * and is there a specialization for its comparator?
 		 */
-		if (state->haveDatum1 && state->sortKeys)
+		if (state->ops.haveDatum1 && state->ops.sortKeys)
 		{
-			if (state->sortKeys[0].comparator == ssup_datum_unsigned_cmp)
+			if (state->ops.sortKeys[0].comparator == ssup_datum_unsigned_cmp)
 			{
 				qsort_tuple_unsigned(state->memtuples,
 									 state->memtupcount,
@@ -3649,7 +3711,7 @@ tuplesort_sort_memtuples(Tuplesortstate *state)
 				return;
 			}
 #if SIZEOF_DATUM >= 8
-			else if (state->sortKeys[0].comparator == ssup_datum_signed_cmp)
+			else if (state->ops.sortKeys[0].comparator == ssup_datum_signed_cmp)
 			{
 				qsort_tuple_signed(state->memtuples,
 								   state->memtupcount,
@@ -3657,7 +3719,7 @@ tuplesort_sort_memtuples(Tuplesortstate *state)
 				return;
 			}
 #endif
-			else if (state->sortKeys[0].comparator == ssup_datum_int32_cmp)
+			else if (state->ops.sortKeys[0].comparator == ssup_datum_int32_cmp)
 			{
 				qsort_tuple_int32(state->memtuples,
 								  state->memtupcount,
@@ -3667,16 +3729,16 @@ tuplesort_sort_memtuples(Tuplesortstate *state)
 		}
 
 		/* Can we use the single-key sort function? */
-		if (state->onlyKey != NULL)
+		if (state->ops.onlyKey != NULL)
 		{
 			qsort_ssup(state->memtuples, state->memtupcount,
-					   state->onlyKey);
+					   state->ops.onlyKey);
 		}
 		else
 		{
 			qsort_tuple(state->memtuples,
 						state->memtupcount,
-						state->comparetup,
+						state->ops.comparetup,
 						state);
 		}
 	}
@@ -3793,10 +3855,10 @@ tuplesort_heap_replace_top(Tuplesortstate *state, SortTuple *tuple)
 static void
 reversedirection(Tuplesortstate *state)
 {
-	SortSupport sortKey = state->sortKeys;
+	SortSupport sortKey = state->ops.sortKeys;
 	int			nkey;
 
-	for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
+	for (nkey = 0; nkey < state->ops.nKeys; nkey++, sortKey++)
 	{
 		sortKey->ssup_reverse = !sortKey->ssup_reverse;
 		sortKey->ssup_nulls_first = !sortKey->ssup_nulls_first;
@@ -3847,7 +3909,7 @@ readtup_alloc(Tuplesortstate *state, Size tuplen)
 	Assert(state->slabFreeHead);
 
 	if (tuplen > SLAB_SLOT_SIZE || !state->slabFreeHead)
-		return MemoryContextAlloc(state->sortcontext, tuplen);
+		return MemoryContextAlloc(state->ops.sortcontext, tuplen);
 	else
 	{
 		buf = state->slabFreeHead;
@@ -3866,6 +3928,7 @@ readtup_alloc(Tuplesortstate *state, Size tuplen)
 static void
 getdatum1_heap(Tuplesortstate *state, SortTuple *stup)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
 	HeapTupleData htup;
 
 	htup.t_len = ((MinimalTuple) stup->tuple)->t_len +
@@ -3874,8 +3937,8 @@ getdatum1_heap(Tuplesortstate *state, SortTuple *stup)
 										MINIMAL_TUPLE_OFFSET);
 
 	stup->datum1 = heap_getattr(&htup,
-								state->sortKeys[0].ssup_attno,
-								state->tupDesc,
+								ops->sortKeys[0].ssup_attno,
+								(TupleDesc) ops->arg,
 								&stup->isnull1);
 
 }
@@ -3883,7 +3946,8 @@ getdatum1_heap(Tuplesortstate *state, SortTuple *stup)
 static int
 comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 {
-	SortSupport sortKey = state->sortKeys;
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	SortSupport sortKey = ops->sortKeys;
 	HeapTupleData ltup;
 	HeapTupleData rtup;
 	TupleDesc	tupDesc;
@@ -3908,7 +3972,7 @@ comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 	ltup.t_data = (HeapTupleHeader) ((char *) a->tuple - MINIMAL_TUPLE_OFFSET);
 	rtup.t_len = ((MinimalTuple) b->tuple)->t_len + MINIMAL_TUPLE_OFFSET;
 	rtup.t_data = (HeapTupleHeader) ((char *) b->tuple - MINIMAL_TUPLE_OFFSET);
-	tupDesc = state->tupDesc;
+	tupDesc = (TupleDesc) ops->arg;
 
 	if (sortKey->abbrev_converter)
 	{
@@ -3925,7 +3989,7 @@ comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 	}
 
 	sortKey++;
-	for (nkey = 1; nkey < state->nKeys; nkey++, sortKey++)
+	for (nkey = 1; nkey < ops->nKeys; nkey++, sortKey++)
 	{
 		attno = sortKey->ssup_attno;
 
@@ -3945,6 +4009,7 @@ comparetup_heap(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 static void
 writetup_heap(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
 	MinimalTuple tuple = (MinimalTuple) stup->tuple;
 
 	/* the part of the MinimalTuple we'll write: */
@@ -3956,7 +4021,7 @@ writetup_heap(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 
 	LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
 	LogicalTapeWrite(tape, (void *) tupbody, tupbodylen);
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
 }
@@ -3969,12 +4034,13 @@ readtup_heap(Tuplesortstate *state, SortTuple *stup,
 	unsigned int tuplen = tupbodylen + MINIMAL_TUPLE_DATA_OFFSET;
 	MinimalTuple tuple = (MinimalTuple) readtup_alloc(state, tuplen);
 	char	   *tupbody = (char *) tuple + MINIMAL_TUPLE_DATA_OFFSET;
+	TuplesortOps *ops = TuplesortstateGetOps(state);
 	HeapTupleData htup;
 
 	/* read in the tuple proper */
 	tuple->t_len = tuplen;
 	LogicalTapeReadExact(tape, tupbody, tupbodylen);
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
 	stup->tuple = (void *) tuple;
@@ -3982,8 +4048,8 @@ readtup_heap(Tuplesortstate *state, SortTuple *stup,
 	htup.t_len = tuple->t_len + MINIMAL_TUPLE_OFFSET;
 	htup.t_data = (HeapTupleHeader) ((char *) tuple - MINIMAL_TUPLE_OFFSET);
 	stup->datum1 = heap_getattr(&htup,
-								state->sortKeys[0].ssup_attno,
-								state->tupDesc,
+								ops->sortKeys[0].ssup_attno,
+								(TupleDesc) ops->arg,
 								&stup->isnull1);
 }
 
@@ -3995,12 +4061,14 @@ readtup_heap(Tuplesortstate *state, SortTuple *stup,
 static void
 getdatum1_cluster(Tuplesortstate *state, SortTuple *stup)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortClusterArg *arg = (TupleSortClusterArg *) ops->arg;
 	HeapTuple tup;
 
 	tup = (HeapTuple) stup->tuple;
 	stup->datum1 = heap_getattr(tup,
-								state->indexInfo->ii_IndexAttrNumbers[0],
-								state->tupDesc,
+								arg->indexInfo->ii_IndexAttrNumbers[0],
+								arg->tupDesc,
 								&stup->isnull1);
 }
 
@@ -4008,7 +4076,9 @@ static int
 comparetup_cluster(const SortTuple *a, const SortTuple *b,
 				   Tuplesortstate *state)
 {
-	SortSupport sortKey = state->sortKeys;
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortClusterArg *arg = (TupleSortClusterArg *) ops->arg;
+	SortSupport sortKey = ops->sortKeys;
 	HeapTuple	ltup;
 	HeapTuple	rtup;
 	TupleDesc	tupDesc;
@@ -4022,10 +4092,10 @@ comparetup_cluster(const SortTuple *a, const SortTuple *b,
 	/* Be prepared to compare additional sort keys */
 	ltup = (HeapTuple) a->tuple;
 	rtup = (HeapTuple) b->tuple;
-	tupDesc = state->tupDesc;
+	tupDesc = arg->tupDesc;
 
 	/* Compare the leading sort key, if it's simple */
-	if (state->haveDatum1)
+	if (ops->haveDatum1)
 	{
 		compare = ApplySortComparator(a->datum1, a->isnull1,
 									  b->datum1, b->isnull1,
@@ -4035,7 +4105,7 @@ comparetup_cluster(const SortTuple *a, const SortTuple *b,
 
 		if (sortKey->abbrev_converter)
 		{
-			AttrNumber	leading = state->indexInfo->ii_IndexAttrNumbers[0];
+			AttrNumber	leading = arg->indexInfo->ii_IndexAttrNumbers[0];
 
 			datum1 = heap_getattr(ltup, leading, tupDesc, &isnull1);
 			datum2 = heap_getattr(rtup, leading, tupDesc, &isnull2);
@@ -4044,7 +4114,7 @@ comparetup_cluster(const SortTuple *a, const SortTuple *b,
 													datum2, isnull2,
 													sortKey);
 		}
-		if (compare != 0 || state->nKeys == 1)
+		if (compare != 0 || ops->nKeys == 1)
 			return compare;
 		/* Compare additional columns the hard way */
 		sortKey++;
@@ -4056,13 +4126,13 @@ comparetup_cluster(const SortTuple *a, const SortTuple *b,
 		nkey = 0;
 	}
 
-	if (state->indexInfo->ii_Expressions == NULL)
+	if (arg->indexInfo->ii_Expressions == NULL)
 	{
 		/* If not expression index, just compare the proper heap attrs */
 
-		for (; nkey < state->nKeys; nkey++, sortKey++)
+		for (; nkey < ops->nKeys; nkey++, sortKey++)
 		{
-			AttrNumber	attno = state->indexInfo->ii_IndexAttrNumbers[nkey];
+			AttrNumber	attno = arg->indexInfo->ii_IndexAttrNumbers[nkey];
 
 			datum1 = heap_getattr(ltup, attno, tupDesc, &isnull1);
 			datum2 = heap_getattr(rtup, attno, tupDesc, &isnull2);
@@ -4089,19 +4159,19 @@ comparetup_cluster(const SortTuple *a, const SortTuple *b,
 		TupleTableSlot *ecxt_scantuple;
 
 		/* Reset context each time to prevent memory leakage */
-		ResetPerTupleExprContext(state->estate);
+		ResetPerTupleExprContext(arg->estate);
 
-		ecxt_scantuple = GetPerTupleExprContext(state->estate)->ecxt_scantuple;
+		ecxt_scantuple = GetPerTupleExprContext(arg->estate)->ecxt_scantuple;
 
 		ExecStoreHeapTuple(ltup, ecxt_scantuple, false);
-		FormIndexDatum(state->indexInfo, ecxt_scantuple, state->estate,
+		FormIndexDatum(arg->indexInfo, ecxt_scantuple, arg->estate,
 					   l_index_values, l_index_isnull);
 
 		ExecStoreHeapTuple(rtup, ecxt_scantuple, false);
-		FormIndexDatum(state->indexInfo, ecxt_scantuple, state->estate,
+		FormIndexDatum(arg->indexInfo, ecxt_scantuple, arg->estate,
 					   r_index_values, r_index_isnull);
 
-		for (; nkey < state->nKeys; nkey++, sortKey++)
+		for (; nkey < ops->nKeys; nkey++, sortKey++)
 		{
 			compare = ApplySortComparator(l_index_values[nkey],
 										  l_index_isnull[nkey],
@@ -4119,6 +4189,7 @@ comparetup_cluster(const SortTuple *a, const SortTuple *b,
 static void
 writetup_cluster(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
 	HeapTuple	tuple = (HeapTuple) stup->tuple;
 	unsigned int tuplen = tuple->t_len + sizeof(ItemPointerData) + sizeof(int);
 
@@ -4126,7 +4197,7 @@ writetup_cluster(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 	LogicalTapeWrite(tape, &tuplen, sizeof(tuplen));
 	LogicalTapeWrite(tape, &tuple->t_self, sizeof(ItemPointerData));
 	LogicalTapeWrite(tape, tuple->t_data, tuple->t_len);
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeWrite(tape, &tuplen, sizeof(tuplen));
 }
@@ -4135,6 +4206,8 @@ static void
 readtup_cluster(Tuplesortstate *state, SortTuple *stup,
 				LogicalTape *tape, unsigned int tuplen)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortClusterArg *arg = (TupleSortClusterArg *) ops->arg;
 	unsigned int t_len = tuplen - sizeof(ItemPointerData) - sizeof(int);
 	HeapTuple	tuple = (HeapTuple) readtup_alloc(state,
 												  t_len + HEAPTUPLESIZE);
@@ -4147,18 +4220,34 @@ readtup_cluster(Tuplesortstate *state, SortTuple *stup,
 	tuple->t_tableOid = InvalidOid;
 	/* Read in the tuple body */
 	LogicalTapeReadExact(tape, tuple->t_data, tuple->t_len);
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
 	stup->tuple = (void *) tuple;
 	/* set up first-column key value, if it's a simple column */
-	if (state->haveDatum1)
+	if (ops->haveDatum1)
 		stup->datum1 = heap_getattr(tuple,
-									state->indexInfo->ii_IndexAttrNumbers[0],
-									state->tupDesc,
+									arg->indexInfo->ii_IndexAttrNumbers[0],
+									arg->tupDesc,
 									&stup->isnull1);
 }
 
+static void
+freestate_cluster(Tuplesortstate *state)
+{
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortClusterArg *arg = (TupleSortClusterArg *) ops->arg;
+
+	/* Free any execution state created for CLUSTER case */
+	if (arg->estate != NULL)
+	{
+		ExprContext *econtext = GetPerTupleExprContext(arg->estate);
+
+		ExecDropSingleTupleTableSlot(econtext->ecxt_scantuple);
+		FreeExecutorState(arg->estate);
+	}
+}
+
 /*
  * Routines specialized for IndexTuple case
  *
@@ -4170,12 +4259,14 @@ readtup_cluster(Tuplesortstate *state, SortTuple *stup,
 static void
 getdatum1_index(Tuplesortstate *state, SortTuple *stup)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortIndexArg *arg = (TupleSortIndexArg *) ops->arg;
 	IndexTuple	tuple;
 
 	tuple = stup->tuple;
 	stup->datum1 = index_getattr(tuple,
 								 1,
-								 RelationGetDescr(state->indexRel),
+								 RelationGetDescr(arg->indexRel),
 								 &stup->isnull1);
 }
 
@@ -4188,7 +4279,9 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 	 * is also special handling for enforcing uniqueness, and special
 	 * treatment for equal keys at the end.
 	 */
-	SortSupport sortKey = state->sortKeys;
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortIndexBTreeArg *arg = (TupleSortIndexBTreeArg *) ops->arg;
+	SortSupport sortKey = ops->sortKeys;
 	IndexTuple	tuple1;
 	IndexTuple	tuple2;
 	int			keysz;
@@ -4212,8 +4305,8 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 	/* Compare additional sort keys */
 	tuple1 = (IndexTuple) a->tuple;
 	tuple2 = (IndexTuple) b->tuple;
-	keysz = state->nKeys;
-	tupDes = RelationGetDescr(state->indexRel);
+	keysz = ops->nKeys;
+	tupDes = RelationGetDescr(arg->index.indexRel);
 
 	if (sortKey->abbrev_converter)
 	{
@@ -4258,7 +4351,7 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 	 * sort algorithm wouldn't have checked whether one must appear before the
 	 * other.
 	 */
-	if (state->enforceUnique && !(!state->uniqueNullsNotDistinct && equal_hasnull))
+	if (arg->enforceUnique && !(!arg->uniqueNullsNotDistinct && equal_hasnull))
 	{
 		Datum		values[INDEX_MAX_KEYS];
 		bool		isnull[INDEX_MAX_KEYS];
@@ -4274,16 +4367,16 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
 
 		index_deform_tuple(tuple1, tupDes, values, isnull);
 
-		key_desc = BuildIndexValueDescription(state->indexRel, values, isnull);
+		key_desc = BuildIndexValueDescription(arg->index.indexRel, values, isnull);
 
 		ereport(ERROR,
 				(errcode(ERRCODE_UNIQUE_VIOLATION),
 				 errmsg("could not create unique index \"%s\"",
-						RelationGetRelationName(state->indexRel)),
+						RelationGetRelationName(arg->index.indexRel)),
 				 key_desc ? errdetail("Key %s is duplicated.", key_desc) :
 				 errdetail("Duplicate keys exist."),
-				 errtableconstraint(state->heapRel,
-									RelationGetRelationName(state->indexRel))));
+				 errtableconstraint(arg->index.heapRel,
+									RelationGetRelationName(arg->index.indexRel))));
 	}
 
 	/*
@@ -4321,6 +4414,8 @@ comparetup_index_hash(const SortTuple *a, const SortTuple *b,
 	Bucket		bucket2;
 	IndexTuple	tuple1;
 	IndexTuple	tuple2;
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortIndexHashArg *arg = (TupleSortIndexHashArg *) ops->arg;
 
 	/*
 	 * Fetch hash keys and mask off bits we don't want to sort by. We know
@@ -4328,12 +4423,12 @@ comparetup_index_hash(const SortTuple *a, const SortTuple *b,
 	 */
 	Assert(!a->isnull1);
 	bucket1 = _hash_hashkey2bucket(DatumGetUInt32(a->datum1),
-								   state->max_buckets, state->high_mask,
-								   state->low_mask);
+								   arg->max_buckets, arg->high_mask,
+								   arg->low_mask);
 	Assert(!b->isnull1);
 	bucket2 = _hash_hashkey2bucket(DatumGetUInt32(b->datum1),
-								   state->max_buckets, state->high_mask,
-								   state->low_mask);
+								   arg->max_buckets, arg->high_mask,
+								   arg->low_mask);
 	if (bucket1 > bucket2)
 		return 1;
 	else if (bucket1 < bucket2)
@@ -4371,13 +4466,14 @@ comparetup_index_hash(const SortTuple *a, const SortTuple *b,
 static void
 writetup_index(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
 	IndexTuple	tuple = (IndexTuple) stup->tuple;
 	unsigned int tuplen;
 
 	tuplen = IndexTupleSize(tuple) + sizeof(tuplen);
 	LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
 	LogicalTapeWrite(tape, (void *) tuple, IndexTupleSize(tuple));
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeWrite(tape, (void *) &tuplen, sizeof(tuplen));
 }
@@ -4386,18 +4482,20 @@ static void
 readtup_index(Tuplesortstate *state, SortTuple *stup,
 			  LogicalTape *tape, unsigned int len)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortIndexArg *arg = (TupleSortIndexArg *) ops->arg;
 	unsigned int tuplen = len - sizeof(unsigned int);
 	IndexTuple	tuple = (IndexTuple) readtup_alloc(state, tuplen);
 
 	LogicalTapeReadExact(tape, tuple, tuplen);
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
 	stup->tuple = (void *) tuple;
 	/* set up first-column key value */
 	stup->datum1 = index_getattr(tuple,
 								 1,
-								 RelationGetDescr(state->indexRel),
+								 RelationGetDescr(arg->indexRel),
 								 &stup->isnull1);
 }
 
@@ -4414,20 +4512,21 @@ getdatum1_datum(Tuplesortstate *state, SortTuple *stup)
 static int
 comparetup_datum(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
 	int			compare;
 
 	compare = ApplySortComparator(a->datum1, a->isnull1,
 								  b->datum1, b->isnull1,
-								  state->sortKeys);
+								  ops->sortKeys);
 	if (compare != 0)
 		return compare;
 
 	/* if we have abbreviations, then "tuple" has the original value */
 
-	if (state->sortKeys->abbrev_converter)
+	if (ops->sortKeys->abbrev_converter)
 		compare = ApplySortAbbrevFullComparator(PointerGetDatum(a->tuple), a->isnull1,
 												PointerGetDatum(b->tuple), b->isnull1,
-												state->sortKeys);
+												ops->sortKeys);
 
 	return compare;
 }
@@ -4435,6 +4534,8 @@ comparetup_datum(const SortTuple *a, const SortTuple *b, Tuplesortstate *state)
 static void
 writetup_datum(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
+	TupleSortDatumArg *arg = (TupleSortDatumArg *) ops->arg;
 	void	   *waddr;
 	unsigned int tuplen;
 	unsigned int writtenlen;
@@ -4444,7 +4545,7 @@ writetup_datum(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 		waddr = NULL;
 		tuplen = 0;
 	}
-	else if (!state->tuples)
+	else if (!state->ops.tuples)
 	{
 		waddr = &stup->datum1;
 		tuplen = sizeof(Datum);
@@ -4452,7 +4553,7 @@ writetup_datum(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 	else
 	{
 		waddr = stup->tuple;
-		tuplen = datumGetSize(PointerGetDatum(stup->tuple), false, state->datumTypeLen);
+		tuplen = datumGetSize(PointerGetDatum(stup->tuple), false, arg->datumTypeLen);
 		Assert(tuplen != 0);
 	}
 
@@ -4460,7 +4561,7 @@ writetup_datum(Tuplesortstate *state, LogicalTape *tape, SortTuple *stup)
 
 	LogicalTapeWrite(tape, (void *) &writtenlen, sizeof(writtenlen));
 	LogicalTapeWrite(tape, waddr, tuplen);
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeWrite(tape, (void *) &writtenlen, sizeof(writtenlen));
 }
@@ -4469,6 +4570,7 @@ static void
 readtup_datum(Tuplesortstate *state, SortTuple *stup,
 			  LogicalTape *tape, unsigned int len)
 {
+	TuplesortOps *ops = TuplesortstateGetOps(state);
 	unsigned int tuplen = len - sizeof(unsigned int);
 
 	if (tuplen == 0)
@@ -4478,7 +4580,7 @@ readtup_datum(Tuplesortstate *state, SortTuple *stup,
 		stup->isnull1 = true;
 		stup->tuple = NULL;
 	}
-	else if (!state->tuples)
+	else if (!state->ops.tuples)
 	{
 		Assert(tuplen == sizeof(Datum));
 		LogicalTapeReadExact(tape, &stup->datum1, tuplen);
@@ -4495,7 +4597,7 @@ readtup_datum(Tuplesortstate *state, SortTuple *stup,
 		stup->tuple = raddr;
 	}
 
-	if (state->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
+	if (ops->sortopt & TUPLESORT_RANDOMACCESS)	/* need trailing length
 													 * word? */
 		LogicalTapeReadExact(tape, &tuplen, sizeof(tuplen));
 }
-- 
2.30.2