From 7ed90f717523b7e203ee1644d2918a073c4fe09d Mon Sep 17 00:00:00 2001
From: David Rowley <dgrowley@gmail.com>
Date: Wed, 4 Mar 2026 16:55:09 +1300
Subject: [PATCH v1] WIP: Introduce selective tuple deforming

Up until now, we have always deformed every attribute of each tuple up
until the last attribute that we require.  This did once make sense to
do as we often had to walk the tuple in order to determine the byte
offset to any attribute that we deform.  Now, since we proactively
populate the CompactAttribute.attcacheoff, in some cases we might be in
a better position to only deform the attributes that we need to deform
by directly jumping to the cached byte offset and skipping deforming any
attributes which are not required.  In some cases the savings can be
very large as it not only allows tts_values and tts_isnull for unneeded
attributes to be left unpopulated, but it might also mean we can skip
loading entire cachelines when the tuple is wide enough to span multiple
cachelines.

We don't want to exclusively always deform tuples this way as doing this
means paying attention to an additional array which states which attnums
we must deform.  Looking at that array for a SELECT * query, which
requires us to deform all attributes, would add overhead.  To support
this a new expression evaluation operator has been added called
EEOP_SCAN_SELECTSOME and each function which builds an ExprState now
accepts a variant function that allows the caller to specify which attnums
are required from the scan side.  This puts it on the caller to decide
which type of deforming should be done.  When the caller provides
the attnums, the expression will be built with EEOP_SCAN_SELECTSOME
rather than EEOP_SCAN_FETCHSOME.  This currently does not interact well
with the physical tlist optimization.  Currently it's the planner's job
to figure out which attributes are actually required.

TODO: JIT support
---
 src/backend/executor/execExpr.c           | 182 ++++++++++-
 src/backend/executor/execExprInterp.c     |  13 +
 src/backend/executor/execScan.c           |  18 ++
 src/backend/executor/execTuples.c         | 377 +++++++++++++++++++++-
 src/backend/executor/execUtils.c          |  47 ++-
 src/backend/executor/nodeSeqscan.c        |   8 +-
 src/backend/optimizer/plan/createplan.c   |  84 ++++-
 src/backend/utils/misc/guc_parameters.dat |   9 +
 src/backend/utils/misc/guc_tables.c       |   7 +
 src/include/executor/execExpr.h           |  24 +-
 src/include/executor/executor.h           |  19 ++
 src/include/executor/tuptable.h           |  22 ++
 src/include/nodes/plannodes.h             |   8 +
 src/include/optimizer/planmain.h          |   9 +
 14 files changed, 798 insertions(+), 29 deletions(-)

diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c
index 77229141b38..e11aa242368 100644
--- a/src/backend/executor/execExpr.c
+++ b/src/backend/executor/execExpr.c
@@ -66,8 +66,18 @@ typedef struct ExprSetupInfo
 	AttrNumber	last_new;
 	/* MULTIEXPR SubPlan nodes appearing in the expression: */
 	List	   *multiexpr_subplans;
+
+	/*
+	 * Fetch only these attnums from the scan with EEOP_SCAN_SELECTSOME. Empty
+	 * set means use EEOP_SCAN_FETCHSOME (i.e fetch all up until last_scan).
+	 * The first user attribute is based at member 0.  System attributes not
+	 * represented.
+	 */
+	Bitmapset  *scan_attrs;
 } ExprSetupInfo;
 
+static ExprState *ExecInitExprInternal(Expr *node, PlanState *parent,
+									   Node *escontext, Bitmapset *scan_attrs);
 static void ExecReadyExpr(ExprState *state);
 static void ExecInitExprRec(Expr *node, ExprState *state,
 							Datum *resv, bool *resnull);
@@ -77,7 +87,8 @@ static void ExecInitFunc(ExprEvalStep *scratch, Expr *node, List *args,
 static void ExecInitSubPlanExpr(SubPlan *subplan,
 								ExprState *state,
 								Datum *resv, bool *resnull);
-static void ExecCreateExprSetupSteps(ExprState *state, Node *node);
+static void ExecCreateExprSetupSteps(ExprState *state, Node *node,
+									 Bitmapset *scan_attrs);
 static void ExecPushExprSetupSteps(ExprState *state, ExprSetupInfo *info);
 static bool expr_setup_walker(Node *node, ExprSetupInfo *info);
 static bool ExecComputeSlotInfo(ExprState *state, ExprEvalStep *op);
@@ -142,7 +153,7 @@ static void ExecInitJsonCoercion(ExprState *state, JsonReturning *returning,
 ExprState *
 ExecInitExpr(Expr *node, PlanState *parent)
 {
-	return ExecInitExprWithContext(node, parent, NULL);
+	return ExecInitExprInternal(node, parent, NULL, NULL);
 }
 
 /*
@@ -161,6 +172,31 @@ ExecInitExpr(Expr *node, PlanState *parent)
  */
 ExprState *
 ExecInitExprWithContext(Expr *node, PlanState *parent, Node *escontext)
+{
+	return ExecInitExprInternal(node, parent, escontext, NULL);
+}
+
+/*
+ * ExecInitExprWithScanAttrs
+ *		As ExecInitExpr but when 'scan_attrs' is set, use
+ *		EEOP_SCAN_SELECTSOME instead of EEOP_SCAN_FETCHSOME to deform only
+ *		the mentioned 'scan_attrs' from the scan tuple.
+ */
+ExprState *
+ExecInitExprWithScanAttrs(Expr *node, PlanState *parent,
+						  Bitmapset *scan_attrs)
+{
+	return ExecInitExprInternal(node, parent, NULL, scan_attrs);
+}
+
+/*
+ * ExecInitExprInteral
+ *		Internal version to implement ExecInitExpr, ExecInitExprWithContext
+ *		and ExecInitExprWithScanAttrs.
+ */
+static ExprState *
+ExecInitExprInternal(Expr *node, PlanState *parent, Node *escontext,
+					 Bitmapset *scan_attrs)
 {
 	ExprState  *state;
 	ExprEvalStep scratch = {0};
@@ -177,7 +213,7 @@ ExecInitExprWithContext(Expr *node, PlanState *parent, Node *escontext)
 	state->escontext = (ErrorSaveContext *) escontext;
 
 	/* Insert setup steps as needed */
-	ExecCreateExprSetupSteps(state, (Node *) node);
+	ExecCreateExprSetupSteps(state, (Node *) node, scan_attrs);
 
 	/* Compile the expression proper */
 	ExecInitExprRec(node, state, &state->resvalue, &state->resnull);
@@ -214,7 +250,7 @@ ExecInitExprWithParams(Expr *node, ParamListInfo ext_params)
 	state->ext_params = ext_params;
 
 	/* Insert setup steps as needed */
-	ExecCreateExprSetupSteps(state, (Node *) node);
+	ExecCreateExprSetupSteps(state, (Node *) node, NULL);
 
 	/* Compile the expression proper */
 	ExecInitExprRec(node, state, &state->resvalue, &state->resnull);
@@ -248,6 +284,19 @@ ExecInitExprWithParams(Expr *node, ParamListInfo ext_params)
  */
 ExprState *
 ExecInitQual(List *qual, PlanState *parent)
+{
+	return ExecInitQualWithScanAttrs(qual, parent, NULL);
+}
+
+/*
+ * ExecInitQualWithScanAttrs
+ *		As ExecInitQual but when 'scan_attrs' is set, use
+ *		EEOP_SCAN_SELECTSOME instead of EEOP_SCAN_FETCHSOME to deform only
+ *		the mentioned 'scan_attrs' from the scan tuple.
+ */
+ExprState *
+ExecInitQualWithScanAttrs(List *qual, PlanState *parent,
+						  Bitmapset *scan_attrs)
 {
 	ExprState  *state;
 	ExprEvalStep scratch = {0};
@@ -268,7 +317,7 @@ ExecInitQual(List *qual, PlanState *parent)
 	state->flags = EEO_FLAG_IS_QUAL;
 
 	/* Insert setup steps as needed */
-	ExecCreateExprSetupSteps(state, (Node *) qual);
+	ExecCreateExprSetupSteps(state, (Node *) qual, scan_attrs);
 
 	/*
 	 * ExecQual() needs to return false for an expression returning NULL. That
@@ -393,6 +442,28 @@ ExecBuildProjectionInfo(List *targetList,
 						TupleTableSlot *slot,
 						PlanState *parent,
 						TupleDesc inputDesc)
+{
+	return ExecBuildProjectionInfoWithScanAttrs(targetList,
+												econtext,
+												slot,
+												parent,
+												inputDesc,
+												NULL);
+}
+
+/*
+ * ExecBuildProjectionInfoWithScanAttrs
+ *		As ExecBuildProjectionInfo but when 'scan_attrs' is set, use
+ *		EEOP_SCAN_SELECTSOME instead of EEOP_SCAN_FETCHSOME to deform only
+ *		the mentioned 'scan_attrs' from the scan tuple.
+ */
+ProjectionInfo *
+ExecBuildProjectionInfoWithScanAttrs(List *targetList,
+									 ExprContext *econtext,
+									 TupleTableSlot *slot,
+									 PlanState *parent,
+									 TupleDesc inputDesc,
+									 Bitmapset *scan_attrs)
 {
 	ProjectionInfo *projInfo = makeNode(ProjectionInfo);
 	ExprState  *state;
@@ -410,7 +481,7 @@ ExecBuildProjectionInfo(List *targetList,
 	state->resultslot = slot;
 
 	/* Insert setup steps as needed */
-	ExecCreateExprSetupSteps(state, (Node *) targetList);
+	ExecCreateExprSetupSteps(state, (Node *) targetList, scan_attrs);
 
 	/* Now compile each tlist column */
 	foreach(lc, targetList)
@@ -2904,11 +2975,19 @@ ExecInitSubPlanExpr(SubPlan *subplan,
 /*
  * Add expression steps performing setup that's needed before any of the
  * main execution of the expression.
+ *
+ * 'scan_attrs' may be given an empty set, in which case deforming the scan
+ * tuple is done via EEOP_SCAN_FETCHSOME, which fetches every attribute from
+ * the scan tuple up until the maximum attribute used by this expression.
+ * When 'scan_attrs' is set, EEOP_SCAN_SELECTSOME is used to only fetch the
+ * attributes mentioned.  Callers must create a unioned set of the attributes
+ * needed from the scan for all expressions using the given slot so that we
+ * incrementally fetch the attributes required by all ExprStates.
  */
 static void
-ExecCreateExprSetupSteps(ExprState *state, Node *node)
+ExecCreateExprSetupSteps(ExprState *state, Node *node, Bitmapset *scan_attrs)
 {
-	ExprSetupInfo info = {0, 0, 0, 0, 0, NIL};
+	ExprSetupInfo info = {0, 0, 0, 0, 0, NIL, scan_attrs};
 
 	/* Prescan to find out what we need. */
 	expr_setup_walker(node, &info);
@@ -2956,11 +3035,75 @@ ExecPushExprSetupSteps(ExprState *state, ExprSetupInfo *info)
 	}
 	if (info->last_scan > 0)
 	{
-		scratch.opcode = EEOP_SCAN_FETCHSOME;
-		scratch.d.fetch.last_var = info->last_scan;
-		scratch.d.fetch.fixed = false;
-		scratch.d.fetch.kind = NULL;
-		scratch.d.fetch.known_desc = NULL;
+		/*
+		 * We have two operators for fetching attributes out of a tuple during
+		 * scans.  EEOP_SCAN_FETCHSOME deforms all attributes in the tuple up
+		 * to the 'last_scan' attnum.  This isn't ideal in some cases, as we
+		 * may only need a few attributes, and those might be deep into the
+		 * tuple.  EEOP_SCAN_SELECTSOME is an operator that fetches only the
+		 * required attributes from the tuple.  When the attcacheoff for these
+		 * attributes is known and no NULLs exist in the tuple prior to the
+		 * required attributes, then this can be a very fast operation.
+		 * EEOP_SCAN_FETCHSOME is still supported as many cases require all
+		 * attributes, and EEOP_SCAN_FETCHSOME can do this more efficiently.
+		 */
+		if (bms_is_empty(info->scan_attrs))
+		{
+			scratch.opcode = EEOP_SCAN_FETCHSOME;
+			scratch.d.fetch.last_var = info->last_scan;
+			scratch.d.fetch.fixed = false;
+			scratch.d.fetch.kind = NULL;
+			scratch.d.fetch.known_desc = NULL;
+		}
+		else
+		{
+			int			nattrs = bms_num_members(info->scan_attrs);
+			AttrNumber *atts;
+			int			a;
+			int			i;
+
+			scratch.opcode = EEOP_SCAN_SELECTSOME;
+			scratch.d.fetch.last_var = info->last_scan;
+			scratch.d.fetch.fixed = false;
+			scratch.d.fetch.natts = nattrs;
+			scratch.d.fetch.kind = NULL;
+			scratch.d.fetch.known_desc = NULL;
+
+			/*
+			 * Allocate these two arrays as a single allocation.  The
+			 * req_attnums array needs 1 element for each attnum that's being
+			 * selected, plus a sentinel attnum which we set to the
+			 * 'last_scan' attnum so that we correctly terminate each of the
+			 * loops during selective deformation before walking off the end
+			 * of the array.
+			 */
+			atts = palloc_array(AttrNumber, nattrs + 1 + info->last_scan + 1);
+
+			scratch.d.fetch.req_attnums = atts;
+			scratch.d.fetch.next_req_attnums_index = &atts[nattrs + 1];
+
+			/* Store each attnum in the Bitmapset into the req_attnum array */
+			a = -1;
+			i = 0;
+			while ((a = bms_next_member(info->scan_attrs, a)) >= 0)
+				scratch.d.fetch.req_attnums[i++] = a;
+
+			/* install sentinel */
+			scratch.d.fetch.req_attnums[nattrs] = info->last_scan;
+
+			/*
+			 * Populate the next_req_attnums_index array.  This allows the
+			 * deforming function to refind the position in the
+			 * next_req_attnums_index array from tts_nvalid.
+			 */
+			a = 0;
+			for (i = 0; i <= info->last_scan; i++)
+			{
+				scratch.d.fetch.next_req_attnums_index[i] = a;
+				if (bms_is_member(i, info->scan_attrs))
+					a++;
+			}
+		}
 		if (ExecComputeSlotInfo(state, &scratch))
 			ExprEvalPushStep(state, &scratch);
 	}
@@ -3033,6 +3176,13 @@ expr_setup_walker(Node *node, ExprSetupInfo *info)
 				switch (variable->varreturningtype)
 				{
 					case VAR_RETURNING_DEFAULT:
+
+						/*
+						 * scan_attrs must contain a member for this attnum or
+						 * be completely empty
+						 */
+						Assert(attnum < 0 || bms_is_empty(info->scan_attrs) ||
+							   bms_is_member(attnum - 1, info->scan_attrs));
 						info->last_scan = Max(info->last_scan, attnum);
 						break;
 					case VAR_RETURNING_OLD:
@@ -3099,7 +3249,8 @@ ExecComputeSlotInfo(ExprState *state, ExprEvalStep *op)
 		   opcode == EEOP_OUTER_FETCHSOME ||
 		   opcode == EEOP_SCAN_FETCHSOME ||
 		   opcode == EEOP_OLD_FETCHSOME ||
-		   opcode == EEOP_NEW_FETCHSOME);
+		   opcode == EEOP_NEW_FETCHSOME ||
+		   opcode == EEOP_SCAN_SELECTSOME);
 
 	if (op->d.fetch.known_desc != NULL)
 	{
@@ -3152,6 +3303,7 @@ ExecComputeSlotInfo(ExprState *state, ExprEvalStep *op)
 		}
 	}
 	else if (opcode == EEOP_SCAN_FETCHSOME ||
+			 opcode == EEOP_SCAN_SELECTSOME ||
 			 opcode == EEOP_OLD_FETCHSOME ||
 			 opcode == EEOP_NEW_FETCHSOME)
 	{
@@ -4346,7 +4498,7 @@ ExecBuildHash32Expr(TupleDesc desc, const TupleTableSlotOps *ops,
 	state->parent = parent;
 
 	/* Insert setup steps as needed. */
-	ExecCreateExprSetupSteps(state, (Node *) hash_exprs);
+	ExecCreateExprSetupSteps(state, (Node *) hash_exprs, NULL);
 
 	/*
 	 * Make a place to store intermediate hash values between subsequent
diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c
index 0634af964a9..263a3f7bb67 100644
--- a/src/backend/executor/execExprInterp.c
+++ b/src/backend/executor/execExprInterp.c
@@ -489,6 +489,7 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
 		&&CASE_EEOP_SCAN_FETCHSOME,
 		&&CASE_EEOP_OLD_FETCHSOME,
 		&&CASE_EEOP_NEW_FETCHSOME,
+		&&CASE_EEOP_SCAN_SELECTSOME,
 		&&CASE_EEOP_INNER_VAR,
 		&&CASE_EEOP_OUTER_VAR,
 		&&CASE_EEOP_SCAN_VAR,
@@ -686,6 +687,18 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
 			EEO_NEXT();
 		}
 
+		EEO_CASE(EEOP_SCAN_SELECTSOME)
+		{
+			CheckOpSlotCompatibility(op, scanslot);
+
+			slot_selectattrs(scanslot,
+							 op->d.fetch.last_var,
+							 op->d.fetch.req_attnums,
+							 op->d.fetch.next_req_attnums_index);
+
+			EEO_NEXT();
+		}
+
 		EEO_CASE(EEOP_INNER_VAR)
 		{
 			int			attnum = op->d.var.attnum;
diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c
index 9f68be17b99..525af11aa08 100644
--- a/src/backend/executor/execScan.c
+++ b/src/backend/executor/execScan.c
@@ -86,6 +86,24 @@ ExecAssignScanProjectionInfo(ScanState *node)
 	ExecConditionalAssignProjectionInfo(&node->ps, tupdesc, scan->scanrelid);
 }
 
+/*
+ * ExecAssignScanProjectionInfoWithScanAttrs
+ *		As ExecAssignScanProjectionInfo but when 'scan_attrs' is set, use
+ *		EEOP_SCAN_SELECTSOME instead of EEOP_SCAN_FETCHSOME to deform only
+ *		the mentioned 'scan_attrs' from the scan tuple.
+ */
+void
+ExecAssignScanProjectionInfoWithScanAttrs(ScanState *node,
+										  Bitmapset *scan_attrs)
+{
+	Scan	   *scan = (Scan *) node->ps.plan;
+	TupleDesc	tupdesc = node->ss_ScanTupleSlot->tts_tupleDescriptor;
+
+	ExecConditionalAssignProjectionInfoWithScanAttrs(&node->ps, tupdesc,
+													 scan->scanrelid,
+													 scan_attrs);
+}
+
 /*
  * ExecAssignScanProjectionInfoWithVarno
  *		As above, but caller can specify varno expected in Vars in the tlist.
diff --git a/src/backend/executor/execTuples.c b/src/backend/executor/execTuples.c
index b0a0028b165..9424d87bc37 100644
--- a/src/backend/executor/execTuples.c
+++ b/src/backend/executor/execTuples.c
@@ -74,6 +74,13 @@ static TupleDesc ExecTypeFromTLInternal(List *targetList,
 										bool skipjunk);
 static pg_attribute_always_inline void slot_deform_heap_tuple(TupleTableSlot *slot, HeapTuple tuple, uint32 *offp,
 															  int reqnatts, bool support_cstring);
+static pg_attribute_always_inline void slot_selectively_deform_heap_tuple(TupleTableSlot *slot,
+																		  HeapTuple tuple,
+																		  uint32 *offp,
+																		  int last_attnum,
+																		  AttrNumber *attnums,
+																		  AttrNumber *attnum_map,
+																		  bool support_cstring);
 static inline void tts_buffer_heap_store_tuple(TupleTableSlot *slot,
 											   HeapTuple tuple,
 											   Buffer buffer,
@@ -129,7 +136,22 @@ tts_virtual_clear(TupleTableSlot *slot)
 static void
 tts_virtual_getsomeattrs(TupleTableSlot *slot, int natts)
 {
-	elog(ERROR, "getsomeattrs is not required to be called on a virtual tuple table slot");
+	elog(ERROR,
+		 "%s is not required to be called on a virtual tuple table slot",
+		 "getsomeattrs");
+}
+
+/*
+ * VirtualTupleTableSlots always have fully populated tts_values and
+ * tts_isnull arrays.  So this function should never be called.
+ */
+static void
+tts_virtual_selectattrs(TupleTableSlot *slot, int last_attnum,
+						AttrNumber *attnums, AttrNumber *attnum_map)
+{
+	elog(ERROR,
+		 "%s is not required to be called on a virtual tuple table slot",
+		 "selectattrs");
 }
 
 /*
@@ -352,6 +374,23 @@ tts_heap_getsomeattrs(TupleTableSlot *slot, int natts)
 	slot_deform_heap_tuple(slot, hslot->tuple, &hslot->off, natts, false);
 }
 
+static void
+tts_heap_selectattrs(TupleTableSlot *slot, int last_attnum,
+					 AttrNumber *attnums, AttrNumber *attnum_map)
+{
+	HeapTupleTableSlot *hslot = (HeapTupleTableSlot *) slot;
+
+	Assert(!TTS_EMPTY(slot));
+
+	slot_selectively_deform_heap_tuple(slot,
+									   hslot->tuple,
+									   &hslot->off,
+									   last_attnum,
+									   attnums,
+									   attnum_map,
+									   false);
+}
+
 static Datum
 tts_heap_getsysattr(TupleTableSlot *slot, int attnum, bool *isnull)
 {
@@ -550,6 +589,23 @@ tts_minimal_getsomeattrs(TupleTableSlot *slot, int natts)
 	slot_deform_heap_tuple(slot, mslot->tuple, &mslot->off, natts, true);
 }
 
+static void
+tts_minimal_selectattrs(TupleTableSlot *slot, int last_attnum,
+						AttrNumber *attnums, AttrNumber *attnum_map)
+{
+	MinimalTupleTableSlot *mslot = (MinimalTupleTableSlot *) slot;
+
+	Assert(!TTS_EMPTY(slot));
+
+	slot_selectively_deform_heap_tuple(slot,
+									   mslot->tuple,
+									   &mslot->off,
+									   last_attnum,
+									   attnums,
+									   attnum_map,
+									   true);
+}
+
 /*
  * MinimalTupleTableSlots never provide system attributes. We generally
  * shouldn't get here, but provide a user-friendly message if we do.
@@ -757,6 +813,23 @@ tts_buffer_heap_getsomeattrs(TupleTableSlot *slot, int natts)
 	slot_deform_heap_tuple(slot, bslot->base.tuple, &bslot->base.off, natts, false);
 }
 
+static void
+tts_buffer_heap_selectattrs(TupleTableSlot *slot, int last_attnum,
+							AttrNumber *attnums, AttrNumber *attnum_map)
+{
+	BufferHeapTupleTableSlot *bslot = (BufferHeapTupleTableSlot *) slot;
+
+	Assert(!TTS_EMPTY(slot));
+
+	slot_selectively_deform_heap_tuple(slot,
+									   bslot->base.tuple,
+									   &bslot->base.off,
+									   last_attnum,
+									   attnums,
+									   attnum_map,
+									   false);
+}
+
 static Datum
 tts_buffer_heap_getsysattr(TupleTableSlot *slot, int attnum, bool *isnull)
 {
@@ -1256,12 +1329,311 @@ done:
 	*offp = off;
 }
 
+/*
+ * slot_selectively_deform_heap_tuple
+ *		Deform attributes of 'tuple' into the Datum/isnull arrays in 'slot'.
+ *		Unlike slot_deform_heap_tuple, which deforms every attribute up to the
+ *		given attribute number, here we deform only the attribute numbers
+ *		mentioned in the 'attnums' array.  When only a few attributes are
+ *		required, this can be more efficient.  When the attributes have a
+ *		known attcacheoff and it's valid to use that, then this version can be
+ *		much more efficient than slot_deform_heap_tuple when only a small
+ *		number of the total attributes are required.
+ */
+static pg_attribute_always_inline void
+slot_selectively_deform_heap_tuple(TupleTableSlot *slot, HeapTuple tuple,
+								   uint32 *offp, int last_attnum,
+								   AttrNumber *attnums,
+								   AttrNumber *attnum_map,
+								   bool support_cstring)
+{
+	CompactAttribute *cattrs;
+	CompactAttribute *cattr;
+	TupleDesc	tupleDesc = slot->tts_tupleDescriptor;
+	HeapTupleHeader tup = tuple->t_data;
+	size_t		attnum;
+	int			attnums_idx;
+	int			firstNonCacheOffsetAttr;
+	int			firstNonGuaranteedAttr;
+	int			firstNullAttr;
+	int			natts;
+	Datum	   *values;
+	bool	   *isnull;
+	char	   *tp;				/* ptr to tuple data */
+	uint32		off;			/* offset in tuple data */
+	int			off_attnum;		/* the attnum that 'off' points to */
+
+	/* Did someone forget to call TupleDescFinalize()? */
+	Assert(tupleDesc->firstNonCachedOffsetAttr >= 0);
+
+	isnull = slot->tts_isnull;
+
+	/*
+	 * Some callers may form and deform tuples prior to NOT NULL constraints
+	 * being checked.  Here we'd like to optimize the case where we only need
+	 * to fetch attributes before or up to the point where the attribute is
+	 * guaranteed to exist in the tuple.  We rely on the slot flag being set
+	 * correctly to only enable this optimization when it's valid to do so.
+	 * This optimization allows us to save fetching the number of attributes
+	 * from the tuple and saves the additional cost of handling non-byval
+	 * attrs.
+	 */
+	firstNonGuaranteedAttr = Min(last_attnum, slot->tts_first_nonguaranteed);
+	firstNonCacheOffsetAttr = tupleDesc->firstNonCachedOffsetAttr;
+
+	if (HeapTupleHasNulls(tuple))
+	{
+		natts = HeapTupleHeaderGetNatts(tup);
+		tp = (char *) tup + MAXALIGN(offsetof(HeapTupleHeaderData, t_bits) +
+									 BITMAPLEN(natts));
+
+		natts = Min(natts, last_attnum);
+		if (natts > firstNonGuaranteedAttr)
+		{
+			uint8	   *bp = tup->t_bits;
+
+			/* Find the first NULL attr */
+			firstNullAttr = first_null_attr(bp, natts);
+
+			/*
+			 * And populate the isnull array for all attributes up to the last
+			 * attribute we're deforming.  When not using attcacheoff, we need
+			 * to know if an attribute is NULL even when we're not deforming
+			 * it, so that we can skip over it when calculating the offset to
+			 * attributes that we are deforming.  Note that we purpusefully
+			 * don't try and only set the isnull element for attnums mentioned
+			 * in the attnums[] array.  Checking that would likely be more
+			 * expensive than it's worth.
+			 */
+			populate_isnull_array(bp, natts, isnull);
+		}
+		else
+		{
+			/* Otherwise all required columns are guaranteed to exist */
+			firstNullAttr = natts;
+		}
+	}
+	else
+	{
+		tp = (char *) tup + MAXALIGN(offsetof(HeapTupleHeaderData, t_bits));
+
+		/*
+		 * We only need to look at the tuple's natts if we need more than the
+		 * guaranteed number of columns
+		 */
+		if (last_attnum > firstNonGuaranteedAttr)
+			natts = Min(HeapTupleHeaderGetNatts(tup), last_attnum);
+		else
+		{
+			/* No need to access the number of attributes in the tuple */
+			natts = last_attnum;
+		}
+
+		/* All attrs can be fetched without checking for NULLs */
+		firstNullAttr = natts;
+	}
+
+	attnums_idx = attnum_map[slot->tts_nvalid];
+	attnum = attnums[attnums_idx];
+	values = slot->tts_values;
+
+	/*
+	 * We store the tupleDesc's CompactAttribute array in 'cattrs' as gcc
+	 * seems to be unwilling to optimize accessing the CompactAttribute
+	 * element efficiently when accessing it via TupleDescCompactAttr().
+	 */
+	cattrs = tupleDesc->compact_attrs;
+
+	/* Ensure we calculated tp correctly */
+	Assert(tp == (char *) tup + tup->t_hoff);
+
+	if (attnum < firstNonGuaranteedAttr)
+	{
+		int			attlen;
+
+		/*
+		 * We use a do/while loop as the if condition above guarantees at
+		 * least one loop and confirms to the compiler that 'attlen' and 'off'
+		 * get initialized, which some compilers are not clever enough to
+		 * figure out if we were to use a for loop.
+		 */
+		do
+		{
+			isnull[attnum] = false;
+			cattr = &cattrs[attnum];
+			attlen = cattr->attlen;
+
+			/* We don't expect any non-byval types */
+			pg_assume(attlen > 0);
+			Assert(cattr->attbyval == true);
+
+			off = cattr->attcacheoff;
+			values[attnum] = fetch_att_noerr(tp + off, true, attlen);
+			attnum = attnums[++attnums_idx];
+		} while (attnum < firstNonGuaranteedAttr);
+
+		off += attlen;
+
+		if (attnum == last_attnum)
+			goto done;
+	}
+
+	/* We can use attcacheoff up until the first NULL */
+	firstNonCacheOffsetAttr = Min(firstNonCacheOffsetAttr, firstNullAttr);
+
+	/*
+	 * Handle the portion of the tuple that we have cached the offset for up
+	 * to the first NULL attribute.  The offset is effectively fixed for
+	 * these, so we can use the CompactAttribute's attcacheoff.
+	 */
+	if (attnum < firstNonCacheOffsetAttr)
+	{
+		int			attlen;
+
+		do
+		{
+			isnull[attnum] = false;
+			cattr = &cattrs[attnum];
+			attlen = cattr->attlen;
+			off = cattr->attcacheoff;
+			values[attnum] = fetch_att_noerr(tp + off, cattr->attbyval,
+											 attlen);
+
+			/* fast-forward to the next required attnum */
+			attnum = attnums[++attnums_idx];
+		} while (attnum < firstNonCacheOffsetAttr);
+
+		off += attlen;
+		Assert(attlen > 0);
+
+		if (attnum == last_attnum)
+			goto done;
+	}
+
+	if (slot->tts_nvalid >= firstNonCacheOffsetAttr)
+	{
+		/* Restore state from previous execution */
+		off_attnum = slot->tts_nvalid;
+		off = *offp;
+	}
+	else
+	{
+		off_attnum = firstNonCacheOffsetAttr - 1;
+		off = cattrs[off_attnum].attcacheoff;
+	}
+
+	/*
+	 * We no longer have the ability to use attcacheoff, so we must look
+	 * through all attributes from this point on.  For attributes that we are
+	 * not selecting, we only calculate the offset to skip them, and don't do
+	 * the actual fetch.  Here we loop up to the first NULL attribute.
+	 */
+	for (; off_attnum < firstNullAttr; off_attnum++)
+	{
+		int			attlen;
+
+		cattr = &cattrs[off_attnum];
+		attlen = cattr->attlen;
+
+		/*
+		 * Only emit the cstring-related code in align_fetch_then_add() when
+		 * cstring support is needed.  We assume support_cstring will be
+		 * passed as a const to allow the compiler to eliminate this branch.
+		 */
+		if (!support_cstring)
+			pg_assume(attlen > 0 || attlen == -1);
+
+		off = att_pointer_alignby(off, cattr->attalignby, attlen, tp + off);
+
+		/*
+		 * If this is an attribute we want, do the fetch and then fast-forward
+		 * attnum to the next attribute we want.
+		 */
+		if (off_attnum == attnum)
+		{
+			isnull[off_attnum] = false;
+			values[off_attnum] = fetch_att_noerr(tp + off, cattr->attbyval,
+												 attlen);
+			attnum = attnums[++attnums_idx];
+
+		}
+		/* Move offset beyond this attribute */
+		off = att_addlength_pointer(off, attlen, tp + off);
+	}
+
+	/*
+	 * Now handle any remaining attributes in the tuple up to the requested
+	 * attnum.  This time, include NULL checks as we're now at the first NULL
+	 * attribute.
+	 */
+	for (; off_attnum < natts; off_attnum++)
+	{
+		int			attlen;
+
+		cattr = &cattrs[off_attnum];
+		attlen = cattr->attlen;
+
+		/* As above, only emit cstring code when needed. */
+		if (!support_cstring)
+			pg_assume(attlen > 0 || attlen == -1);
+
+		/* Is this an attribute we're selecting? */
+		if (off_attnum == attnum)
+		{
+			attnum = attnums[++attnums_idx];
+
+			if (isnull[off_attnum])
+			{
+				values[off_attnum] = (Datum) 0;
+				continue;
+			}
+
+			/*
+			 * align 'off', fetch the datum, and increment off beyond the
+			 * datum
+			 */
+			values[off_attnum] = align_fetch_then_add(tp,
+													  &off,
+													  cattr->attbyval,
+													  attlen,
+													  cattr->attalignby);
+		}
+		else if (!isnull[off_attnum])
+		{
+			/* We don't want this attribute, move beyond it */
+			off = att_pointer_alignby(off, cattr->attalignby, attlen, tp + off);
+			off = att_addlength_pointer(off, attlen, tp + off);
+		}
+		else
+		{
+			/* It's a NULL attribute that we're not selecting.  Do nothing. */
+		}
+	}
+
+	/* Fetch any missing attrs and raise an error if reqnatts is invalid */
+	if (unlikely(attnum < last_attnum))
+	{
+		*offp = off;
+		slot->tts_nvalid = last_attnum;
+
+		/* XXX worth doing this selectively too? */
+		slot_getmissingattrs(slot, attnum, last_attnum);
+		return;
+	}
+done:
+
+	slot->tts_nvalid = last_attnum;
+	/* Save current offset for next execution */
+	*offp = off;
+}
+
 const TupleTableSlotOps TTSOpsVirtual = {
 	.base_slot_size = sizeof(VirtualTupleTableSlot),
 	.init = tts_virtual_init,
 	.release = tts_virtual_release,
 	.clear = tts_virtual_clear,
 	.getsomeattrs = tts_virtual_getsomeattrs,
+	.selectattrs = tts_virtual_selectattrs,
 	.getsysattr = tts_virtual_getsysattr,
 	.materialize = tts_virtual_materialize,
 	.is_current_xact_tuple = tts_virtual_is_current_xact_tuple,
@@ -1283,6 +1655,7 @@ const TupleTableSlotOps TTSOpsHeapTuple = {
 	.release = tts_heap_release,
 	.clear = tts_heap_clear,
 	.getsomeattrs = tts_heap_getsomeattrs,
+	.selectattrs = tts_heap_selectattrs,
 	.getsysattr = tts_heap_getsysattr,
 	.is_current_xact_tuple = tts_heap_is_current_xact_tuple,
 	.materialize = tts_heap_materialize,
@@ -1301,6 +1674,7 @@ const TupleTableSlotOps TTSOpsMinimalTuple = {
 	.release = tts_minimal_release,
 	.clear = tts_minimal_clear,
 	.getsomeattrs = tts_minimal_getsomeattrs,
+	.selectattrs = tts_minimal_selectattrs,
 	.getsysattr = tts_minimal_getsysattr,
 	.is_current_xact_tuple = tts_minimal_is_current_xact_tuple,
 	.materialize = tts_minimal_materialize,
@@ -1319,6 +1693,7 @@ const TupleTableSlotOps TTSOpsBufferHeapTuple = {
 	.release = tts_buffer_heap_release,
 	.clear = tts_buffer_heap_clear,
 	.getsomeattrs = tts_buffer_heap_getsomeattrs,
+	.selectattrs = tts_buffer_heap_selectattrs,
 	.getsysattr = tts_buffer_heap_getsysattr,
 	.is_current_xact_tuple = tts_buffer_is_current_xact_tuple,
 	.materialize = tts_buffer_heap_materialize,
diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
index 1eb6b9f1f40..b56ae8841ee 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -585,8 +585,7 @@ ExecGetCommonChildSlotOps(PlanState *ps)
  * ----------------
  */
 void
-ExecAssignProjectionInfo(PlanState *planstate,
-						 TupleDesc inputDesc)
+ExecAssignProjectionInfo(PlanState *planstate, TupleDesc inputDesc)
 {
 	planstate->ps_ProjInfo =
 		ExecBuildProjectionInfo(planstate->plan->targetlist,
@@ -596,6 +595,28 @@ ExecAssignProjectionInfo(PlanState *planstate,
 								inputDesc);
 }
 
+/* ----------------
+ *		ExecAssignProjectionInfoWithScanAttrs
+ *
+ * As ExecAssignScanProjectionInfo but when 'scan_attrs' is set, use
+ * EEOP_SCAN_SELECTSOME instead of EEOP_SCAN_FETCHSOME to deform only the
+ * mentioned 'scan_attrs' from the scan tuple.
+ * ----------------
+*/
+void
+ExecAssignProjectionInfoWithScanAttrs(PlanState *planstate,
+									  TupleDesc inputDesc,
+									  Bitmapset *scan_attrs)
+{
+	planstate->ps_ProjInfo =
+		ExecBuildProjectionInfoWithScanAttrs(planstate->plan->targetlist,
+											 planstate->ps_ExprContext,
+											 planstate->ps_ResultTupleSlot,
+											 planstate,
+											 inputDesc,
+											 scan_attrs);
+}
+
 
 /* ----------------
  *		ExecConditionalAssignProjectionInfo
@@ -607,6 +628,26 @@ ExecAssignProjectionInfo(PlanState *planstate,
 void
 ExecConditionalAssignProjectionInfo(PlanState *planstate, TupleDesc inputDesc,
 									int varno)
+{
+	ExecConditionalAssignProjectionInfoWithScanAttrs(planstate,
+													 inputDesc,
+													 varno,
+													 NULL);
+}
+
+/* ----------------
+ *		ExecConditionalAssignProjectionInfoWithScanAttrs
+ *
+ * As ExecConditionalAssignProjectionInfo but when 'scan_attrs' is set, use
+ * EEOP_SCAN_SELECTSOME instead of EEOP_SCAN_FETCHSOME to deform only the
+ * mentioned 'scan_attrs' from the scan tuple.
+ * ----------------
+*/
+void
+ExecConditionalAssignProjectionInfoWithScanAttrs(PlanState *planstate,
+												 TupleDesc inputDesc,
+												 int varno,
+												 Bitmapset *scan_attrs)
 {
 	if (tlist_matches_tupdesc(planstate,
 							  planstate->plan->targetlist,
@@ -627,7 +668,7 @@ ExecConditionalAssignProjectionInfo(PlanState *planstate, TupleDesc inputDesc,
 			planstate->resultopsfixed = true;
 			planstate->resultopsset = true;
 		}
-		ExecAssignProjectionInfo(planstate, inputDesc);
+		ExecAssignProjectionInfoWithScanAttrs(planstate, inputDesc, scan_attrs);
 	}
 }
 
diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index 5bcb0a861d7..9687052bb47 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -260,13 +260,15 @@ ExecInitSeqScan(SeqScan *node, EState *estate, int eflags)
 	 * Initialize result type and projection.
 	 */
 	ExecInitResultTypeTL(&scanstate->ss.ps);
-	ExecAssignScanProjectionInfo(&scanstate->ss);
+	ExecAssignScanProjectionInfoWithScanAttrs(&scanstate->ss,
+											  node->scan.scan_varattnos);
 
 	/*
 	 * initialize child expressions
 	 */
-	scanstate->ss.ps.qual =
-		ExecInitQual(node->scan.plan.qual, (PlanState *) scanstate);
+	scanstate->ss.ps.qual = ExecInitQualWithScanAttrs(node->scan.plan.qual,
+													  (PlanState *) scanstate,
+													  node->scan.scan_varattnos);
 
 	/*
 	 * When EvalPlanQual() is not in use, assign ExecProcNode for this node
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index de6a183da79..e89411e5136 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -77,6 +77,7 @@ static Plan *create_plan_recurse(PlannerInfo *root, Path *best_path,
 static Plan *create_scan_plan(PlannerInfo *root, Path *best_path,
 							  int flags);
 static List *build_path_tlist(PlannerInfo *root, Path *path);
+static Bitmapset *get_selective_deform_attrs(Bitmapset *attrs);
 static bool use_physical_tlist(PlannerInfo *root, Path *path, int flags);
 static List *get_gating_quals(PlannerInfo *root, List *quals);
 static Plan *create_gating_plan(PlannerInfo *root, Path *path, Plan *plan,
@@ -118,7 +119,8 @@ static ModifyTable *create_modifytable_plan(PlannerInfo *root, ModifyTablePath *
 static Limit *create_limit_plan(PlannerInfo *root, LimitPath *best_path,
 								int flags);
 static SeqScan *create_seqscan_plan(PlannerInfo *root, Path *best_path,
-									List *tlist, List *scan_clauses);
+									List *tlist, List *scan_clauses,
+									Bitmapset *tlist_varattnos);
 static SampleScan *create_samplescan_plan(PlannerInfo *root, Path *best_path,
 										  List *tlist, List *scan_clauses);
 static Scan *create_indexscan_plan(PlannerInfo *root, IndexPath *best_path,
@@ -178,7 +180,8 @@ static void label_sort_with_costsize(PlannerInfo *root, Sort *plan,
 									 double limit_tuples);
 static void label_incrementalsort_with_costsize(PlannerInfo *root, IncrementalSort *plan,
 												List *pathkeys, double limit_tuples);
-static SeqScan *make_seqscan(List *qptlist, List *qpqual, Index scanrelid);
+static SeqScan *make_seqscan(List *qptlist, List *qpqual, Index scanrelid,
+							 Bitmapset *scan_varattnos);
 static SampleScan *make_samplescan(List *qptlist, List *qpqual, Index scanrelid,
 								   TableSampleClause *tsc);
 static IndexScan *make_indexscan(List *qptlist, List *qpqual, Index scanrelid,
@@ -319,6 +322,8 @@ static ModifyTable *make_modifytable(PlannerInfo *root, Plan *subplan,
 static GatherMerge *create_gather_merge_plan(PlannerInfo *root,
 											 GatherMergePath *best_path);
 
+/* GUC parameter */
+int			debug_tuple_deform = DEBUG_TUPLE_DEFORM_AUTO;
 
 /*
  * create_plan
@@ -550,6 +555,7 @@ create_plan_recurse(PlannerInfo *root, Path *best_path, int flags)
 static Plan *
 create_scan_plan(PlannerInfo *root, Path *best_path, int flags)
 {
+	Bitmapset  *tlist_varattnos = NULL;
 	RelOptInfo *rel = best_path->parent;
 	List	   *scan_clauses;
 	List	   *gating_clauses;
@@ -579,6 +585,14 @@ create_scan_plan(PlannerInfo *root, Path *best_path, int flags)
 			break;
 	}
 
+	/*
+	 * Figure out which attributes we need from the scan before applying the
+	 * physical tlist optimization.
+	 */
+	pull_varattnos((Node *) best_path->pathtarget->exprs,
+				   rel->relid,
+				   &tlist_varattnos);
+
 	/*
 	 * If this is a parameterized scan, we also need to enforce all the join
 	 * clauses available from the outer relation(s).
@@ -672,7 +686,8 @@ create_scan_plan(PlannerInfo *root, Path *best_path, int flags)
 			plan = (Plan *) create_seqscan_plan(root,
 												best_path,
 												tlist,
-												scan_clauses);
+												scan_clauses,
+												tlist_varattnos);
 			break;
 
 		case T_SampleScan:
@@ -848,6 +863,39 @@ build_path_tlist(PlannerInfo *root, Path *path)
 	return tlist;
 }
 
+/*
+ * get_selective_deform_attrs
+ *		Given a set of attributes to deform, look at the debug_tuple_deform
+ *		GUC and determine which deforming method the executor should use for
+ *		these attributes.  DEBUG_TUPLE_DEFORM_SELECTIVE will simply return the
+ *		'attrs' value so that selective deforming is done on those attributes.
+ *		(If those attrs are NULL, then incremental deformation is performed).
+ *		When debug_tuple_deform is "incremental", the attrs are returned as
+ *		NULL to force the executor to deform incrementally.  When
+ *		debug_tuple_deform is "auto" we try to decide which deforming method
+ *		is the most efficient for these attrs.
+ */
+static Bitmapset *
+get_selective_deform_attrs(Bitmapset *attrs)
+{
+
+	if (debug_tuple_deform == DEBUG_TUPLE_DEFORM_SELECTIVE)
+		return attrs;
+	else if (debug_tuple_deform == DEBUG_TUPLE_DEFORM_INCREMENTAL || attrs == NULL)
+		return NULL;
+	else
+	{
+		/*
+		 * Keep this simple for now.  If we need less than half of the
+		 * attributes between the min and max attribute, then perform
+		 * selective deformation, otherwise deform incrementally.
+		 */
+		if (bms_num_members(attrs) < (bms_prev_member(attrs, -1) + 1) / 2)
+			return attrs;
+		else
+			return NULL;
+	}
+}
 /*
  * use_physical_tlist
  *		Decide whether to use a tlist matching relation structure,
@@ -2753,10 +2801,13 @@ create_limit_plan(PlannerInfo *root, LimitPath *best_path, int flags)
  */
 static SeqScan *
 create_seqscan_plan(PlannerInfo *root, Path *best_path,
-					List *tlist, List *scan_clauses)
+					List *tlist, List *scan_clauses, Bitmapset *tlist_varattnos)
 {
 	SeqScan    *scan_plan;
 	Index		scan_relid = best_path->parent->relid;
+	Bitmapset  *scan_varattnos = tlist_varattnos;
+	Bitmapset  *non_sys_attrs = NULL;
+	int			i;
 
 	/* it should be a base rel... */
 	Assert(scan_relid > 0);
@@ -2768,6 +2819,23 @@ create_seqscan_plan(PlannerInfo *root, Path *best_path,
 	/* Reduce RestrictInfo list to bare expressions; ignore pseudoconstants */
 	scan_clauses = extract_actual_clauses(scan_clauses, false);
 
+	/* Pull varattnos from WHERE clause Vars */
+	pull_varattnos((Node *) scan_clauses, scan_relid, &scan_varattnos);
+
+	/* Don't set these when whole-row var is present */
+	if (!bms_is_member(0 - FirstLowInvalidHeapAttributeNumber, scan_varattnos))
+	{
+		/*
+		 * Adjust the attnums to they're zero offset rather than offset by
+		 * FirstLowInvalidHeapAttributeNumber.
+		 */
+		/* XXX invent bms_offset_members()? */
+		i = 0 - FirstLowInvalidHeapAttributeNumber;
+		while ((i = bms_next_member(scan_varattnos, i)) >= 0)
+			non_sys_attrs = bms_add_member(non_sys_attrs,
+										   i - 1 + FirstLowInvalidHeapAttributeNumber);
+	}
+
 	/* Replace any outer-relation variables with nestloop params */
 	if (best_path->param_info)
 	{
@@ -2775,9 +2843,11 @@ create_seqscan_plan(PlannerInfo *root, Path *best_path,
 			replace_nestloop_params(root, (Node *) scan_clauses);
 	}
 
+	non_sys_attrs = get_selective_deform_attrs(non_sys_attrs);
 	scan_plan = make_seqscan(tlist,
 							 scan_clauses,
-							 scan_relid);
+							 scan_relid,
+							 non_sys_attrs);
 
 	copy_generic_path_info(&scan_plan->scan.plan, best_path);
 
@@ -5488,7 +5558,8 @@ bitmap_subplan_mark_shared(Plan *plan)
 static SeqScan *
 make_seqscan(List *qptlist,
 			 List *qpqual,
-			 Index scanrelid)
+			 Index scanrelid,
+			 Bitmapset *scan_varattnos)
 {
 	SeqScan    *node = makeNode(SeqScan);
 	Plan	   *plan = &node->scan.plan;
@@ -5498,6 +5569,7 @@ make_seqscan(List *qptlist,
 	plan->lefttree = NULL;
 	plan->righttree = NULL;
 	node->scan.scanrelid = scanrelid;
+	node->scan.scan_varattnos = scan_varattnos;
 
 	return node;
 }
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 83af594d4af..3a1846b65e1 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -739,6 +739,15 @@
   ifdef => 'DEBUG_NODE_TESTS_ENABLED',
 },
 
+{ name => 'debug_tuple_deform', type => 'enum', context => 'PGC_USERSET', group => 'DEVELOPER_OPTIONS',
+  short_desc => 'Allows control over the tuple deform method used during execution.',
+  long_desc => 'This can be useful to test which tuple deform method is most efficient for a given set of attributes.',
+  flags => 'GUC_NOT_IN_SAMPLE | GUC_EXPLAIN',
+  variable => 'debug_tuple_deform',
+  boot_val => 'DEBUG_TUPLE_DEFORM_AUTO',
+  options => 'debug_tuple_deform_options',
+},
+
 { name => 'debug_write_read_parse_plan_trees', type => 'bool', context => 'PGC_SUSET', group => 'DEVELOPER_OPTIONS',
   short_desc => 'Set this to force all parse and plan trees to be passed through outfuncs.c/readfuncs.c, to facilitate catching errors and omissions in those modules.',
   flags => 'GUC_NOT_IN_SAMPLE',
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 290ccbc543e..45033409087 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -416,6 +416,13 @@ static const struct config_enum_entry debug_parallel_query_options[] = {
 	{NULL, 0, false}
 };
 
+static const struct config_enum_entry debug_tuple_deform_options[] = {
+	{"incremental", DEBUG_TUPLE_DEFORM_INCREMENTAL, false},
+	{"selective", DEBUG_TUPLE_DEFORM_SELECTIVE, false},
+	{"auto", DEBUG_TUPLE_DEFORM_AUTO, false},
+	{NULL, 0, false}
+};
+
 static const struct config_enum_entry plan_cache_mode_options[] = {
 	{"auto", PLAN_CACHE_MODE_AUTO, false},
 	{"force_generic_plan", PLAN_CACHE_MODE_FORCE_GENERIC_PLAN, false},
diff --git a/src/include/executor/execExpr.h b/src/include/executor/execExpr.h
index c61b3d624d5..ea385da3550 100644
--- a/src/include/executor/execExpr.h
+++ b/src/include/executor/execExpr.h
@@ -78,6 +78,9 @@ typedef enum ExprEvalOp
 	EEOP_OLD_FETCHSOME,
 	EEOP_NEW_FETCHSOME,
 
+	/* apply slot_selectattrs on the corresponding tuple slot */
+	EEOP_SCAN_SELECTSOME,
+
 	/* compute non-system Var value */
 	EEOP_INNER_VAR,
 	EEOP_OUTER_VAR,
@@ -318,15 +321,34 @@ typedef struct ExprEvalStep
 	 */
 	union
 	{
-		/* for EEOP_INNER/OUTER/SCAN/OLD/NEW_FETCHSOME */
+		/*
+		 * for EEOP_INNER/OUTER/SCAN/OLD/NEW_FETCHSOME and
+		 * EEOP_SCAN_SELECTSOME
+		 */
 		struct
 		{
 			/* attribute number up to which to fetch (inclusive) */
 			int			last_var;
 			/* will the type of slot be the same for every invocation */
 			bool		fixed;
+			/* Number of elements in req_attnums array. XXX needed? */
+			AttrNumber	natts;
+
+			/* One element for each attnum to select, ordered by attnum */
+			AttrNumber *req_attnums;
+
+			/*
+			 * Provides mapping of 0-based attnums back to the index of the
+			 * req_attnums array that tuple deforming should continue from.
+			 * This allows us to re-find the element of req_attnums using the
+			 * slot's tts_nvalid so that we can continue deforming from the
+			 * last deformed attribute.
+			 */
+			AttrNumber *next_req_attnums_index;
+
 			/* tuple descriptor, if known */
 			TupleDesc	known_desc;
+
 			/* type of slot, can only be relied upon if fixed is set */
 			const TupleTableSlotOps *kind;
 		}			fetch;
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 33bbdbfeffb..37ae89b0cea 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -333,8 +333,12 @@ ExecProcNode(PlanState *node)
  */
 extern ExprState *ExecInitExpr(Expr *node, PlanState *parent);
 extern ExprState *ExecInitExprWithContext(Expr *node, PlanState *parent, Node *escontext);
+extern ExprState *ExecInitExprWithScanAttrs(Expr *node, PlanState *parent,
+											Bitmapset *scan_attrs);
 extern ExprState *ExecInitExprWithParams(Expr *node, ParamListInfo ext_params);
 extern ExprState *ExecInitQual(List *qual, PlanState *parent);
+extern ExprState *ExecInitQualWithScanAttrs(List *qual, PlanState *parent,
+											Bitmapset *scan_attrs);
 extern ExprState *ExecInitCheck(List *qual, PlanState *parent);
 extern List *ExecInitExprList(List *nodes, PlanState *parent);
 extern ExprState *ExecBuildAggTrans(AggState *aggstate, struct AggStatePerPhaseData *phase,
@@ -373,6 +377,12 @@ extern ProjectionInfo *ExecBuildProjectionInfo(List *targetList,
 											   TupleTableSlot *slot,
 											   PlanState *parent,
 											   TupleDesc inputDesc);
+extern ProjectionInfo *ExecBuildProjectionInfoWithScanAttrs(List *targetList,
+															ExprContext *econtext,
+															TupleTableSlot *slot,
+															PlanState *parent,
+															TupleDesc inputDesc,
+															Bitmapset *scan_attrs);
 extern ProjectionInfo *ExecBuildUpdateProjection(List *targetList,
 												 bool evalTargetList,
 												 List *targetColnos,
@@ -592,6 +602,8 @@ typedef bool (*ExecScanRecheckMtd) (ScanState *node, TupleTableSlot *slot);
 extern TupleTableSlot *ExecScan(ScanState *node, ExecScanAccessMtd accessMtd,
 								ExecScanRecheckMtd recheckMtd);
 extern void ExecAssignScanProjectionInfo(ScanState *node);
+extern void ExecAssignScanProjectionInfoWithScanAttrs(ScanState *node,
+													  Bitmapset *scan_attrs);
 extern void ExecAssignScanProjectionInfoWithVarno(ScanState *node, int varno);
 extern void ExecScanReScan(ScanState *node);
 
@@ -688,8 +700,15 @@ extern const TupleTableSlotOps *ExecGetCommonSlotOps(PlanState **planstates,
 extern const TupleTableSlotOps *ExecGetCommonChildSlotOps(PlanState *ps);
 extern void ExecAssignProjectionInfo(PlanState *planstate,
 									 TupleDesc inputDesc);
+extern void ExecAssignProjectionInfoWithScanAttrs(PlanState *planstate,
+												  TupleDesc inputDesc,
+												  Bitmapset *scan_attrs);
 extern void ExecConditionalAssignProjectionInfo(PlanState *planstate,
 												TupleDesc inputDesc, int varno);
+extern void ExecConditionalAssignProjectionInfoWithScanAttrs(PlanState *planstate,
+															 TupleDesc inputDesc,
+															 int varno,
+															 Bitmapset *scan_attrs);
 extern void ExecAssignScanType(ScanState *scanstate, TupleDesc tupDesc);
 extern void ExecCreateScanSlotFromOuterPlan(EState *estate,
 											ScanState *scanstate,
diff --git a/src/include/executor/tuptable.h b/src/include/executor/tuptable.h
index d7e8e72aeae..8689d43e76a 100644
--- a/src/include/executor/tuptable.h
+++ b/src/include/executor/tuptable.h
@@ -174,6 +174,16 @@ struct TupleTableSlotOps
 	 */
 	void		(*getsomeattrs) (TupleTableSlot *slot, int natts);
 
+	/*
+	 * Populate the tts_values and tts_isnull elements of the given slot with
+	 * the values of the corresponding attribute from the tuple stored in the
+	 * slot.  Populate up as far as last_attnum and store each attribute
+	 * mentioned in the attnums array.  Use attnum_map to determine the
+	 * starting element in the attnums array from the slot's tts_nvalid.
+	 */
+	void		(*selectattrs) (TupleTableSlot *slot, int last_attnum,
+								AttrNumber *attnums, AttrNumber *attnum_map);
+
 	/*
 	 * Returns value of the given system attribute as a datum and sets isnull
 	 * to false, if it's not NULL. Throws an error if the slot type does not
@@ -380,6 +390,18 @@ slot_getsomeattrs(TupleTableSlot *slot, int attnum)
 		slot->tts_ops->getsomeattrs(slot, attnum);
 }
 
+static inline void
+slot_selectattrs(TupleTableSlot *slot, int last_attnum, AttrNumber *attnums,
+				 AttrNumber *attnum_map)
+{
+	/*
+	 * Populate slot only attributes mentioned in the attnums array, up to
+	 * 'last_attnum', if it's not already
+	 */
+	if (slot->tts_nvalid < last_attnum)
+		slot->tts_ops->selectattrs(slot, last_attnum, attnums, attnum_map);
+}
+
 /*
  * slot_getallattrs
  *		This function forces all the entries of the slot's Datum/isnull
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 14a1dfed2b9..5776434d259 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -542,6 +542,14 @@ typedef struct Scan
 	Plan		plan;
 	/* relid is index into the range table */
 	Index		scanrelid;
+
+	/*
+	 * All varattnos that are required from the scanrelid.  Does not include
+	 * any added due to the physical tlist optimization or system attributes
+	 * or whole-row attributes.  User attributes are 0 based, i.e attnum==1 is
+	 * member 0.
+	 */
+	Bitmapset  *scan_varattnos;
 } Scan;
 
 /* ----------------
diff --git a/src/include/optimizer/planmain.h b/src/include/optimizer/planmain.h
index d0dc3761b13..d67800ed489 100644
--- a/src/include/optimizer/planmain.h
+++ b/src/include/optimizer/planmain.h
@@ -21,6 +21,15 @@
 #define DEFAULT_CURSOR_TUPLE_FRACTION 0.1
 extern PGDLLIMPORT double cursor_tuple_fraction;
 extern PGDLLIMPORT bool enable_self_join_elimination;
+extern PGDLLIMPORT int debug_tuple_deform;
+
+/* possible values for debug_tuple_deform */
+typedef enum
+{
+	DEBUG_TUPLE_DEFORM_INCREMENTAL,
+	DEBUG_TUPLE_DEFORM_SELECTIVE,
+	DEBUG_TUPLE_DEFORM_AUTO,
+}			DebugTupleDeform;
 
 /* query_planner callback to compute query_pathkeys */
 typedef void (*query_pathkeys_callback) (PlannerInfo *root, void *extra);
-- 
2.51.0

