WIP more fixes in EPQ handling

Author: funbringer

src/hooks.c

@@ -987,8 +987,9 @@ pathman_executor_hook(QueryDesc *queryDesc,
 		{
 			CustomScanState *pr_state = (CustomScanState *) mt_state->mt_plans[i];
 
-			/* Check if this is a PartitionRouter node */
-			if (IsPartitionRouterState(pr_state))
+			/* Check if this is a PartitionFilter + PartitionRouter combo */
+			if (IsPartitionFilterState(pr_state) &&
+				IsPartitionRouterState(linitial(pr_state->custom_ps)))
 			{
 				ResultRelInfo *rri = &mt_state->resultRelInfo[i];
 

src/include/partition_filter.h

@@ -108,9 +108,6 @@ typedef struct
 	ResultPartsStorage	result_parts;			/* partition ResultRelInfo cache */
 	CmdType				command_type;
 
-	bool				warning_triggered;		/* warning message counter */
-
-	TupleTableSlot     *subplan_slot;			/* slot that was returned from subplan */
 	TupleTableSlot	   *tup_convert_slot;		/* slot for rebuilt tuples */
 } PartitionFilterState;
 
@@ -170,6 +167,8 @@ PartRelationInfo * refresh_result_parts_storage(ResultPartsStorage *parts_storag
 
 TupleConversionMap * build_part_tuple_map(Relation parent_rel, Relation child_rel);
 
+List * pfilter_build_tlist(Plan *subplan);
+
 
 /* Find suitable partition using 'value' */
 Oid * find_partitions_for_value(Datum value, Oid value_type,
@@ -183,8 +182,8 @@ Plan * make_partition_filter(Plan *subplan,
 							 Oid parent_relid,
 							 Index parent_rti,
 							 OnConflictAction conflict_action,
-							 List *returning_list,
-							 CmdType command_type);
+							 CmdType command_type,
+							 List *returning_list);
 
 
 Node * partition_filter_create_scan_state(CustomScan *node);

src/include/partition_router.h

@@ -31,10 +31,13 @@ typedef struct PartitionRouterState
 	CustomScanState		css;
 
 	Oid					partitioned_table;
-	JunkFilter		   *junkfilter;
+	Plan			   *subplan;	/* proxy variable to store subplan */
+	JunkFilter		   *junkfilter;	/* 'ctid' extraction facility */
+
 	EPQState			epqstate;
 	int					epqparam;
-	Plan			   *subplan; /* proxy variable to store subplan */
+
+	ResultRelInfo	   *current_rri;
 } PartitionRouterState;
 
 

src/partition_filter.c

@@ -16,6 +16,7 @@
 #include "partition_filter.h"
 #include "utils.h"
 
+#include "access/htup_details.h"
 #include "catalog/pg_class.h"
 #include "catalog/pg_type.h"
 #include "foreign/fdwapi.h"
@@ -87,8 +88,6 @@ static Node *fix_returning_list_mutator(Node *node, void *state);
 static Index append_rte_to_estate(EState *estate, RangeTblEntry *rte);
 static int append_rri_to_estate(EState *estate, ResultRelInfo *rri);
 
-static List *pfilter_build_tlist(Plan *subplan);
-
 static void pf_memcxt_callback(void *arg);
 static estate_mod_data * fetch_estate_mod_data(EState *estate);
 
@@ -633,8 +632,8 @@ make_partition_filter(Plan *subplan,
 					  Oid parent_relid,
 					  Index parent_rti,
 					  OnConflictAction conflict_action,
-					  List *returning_list,
-					  CmdType command_type)
+					  CmdType command_type,
+					  List *returning_list)
 {
 	CustomScan *cscan = makeNode(CustomScan);
 
@@ -723,9 +722,6 @@ partition_filter_begin(CustomScanState *node, EState *estate, int eflags)
 							  state->on_conflict_action != ONCONFLICT_NONE,
 							  RPS_RRI_CB(prepare_rri_for_insert, state),
 							  RPS_RRI_CB(NULL, NULL));
-
-	/* No warnings yet */
-	state->warning_triggered = false;
 }
 
 TupleTableSlot *
@@ -739,16 +735,12 @@ partition_filter_exec(CustomScanState *node)
 	TupleTableSlot		   *slot;
 
 	slot = ExecProcNode(child_ps);
-	state->subplan_slot = slot;
-
-	if (state->tup_convert_slot)
-		ExecClearTuple(state->tup_convert_slot);
 
 	if (!TupIsNull(slot))
 	{
 		MemoryContext			old_mcxt;
 		ResultRelInfoHolder	   *rri_holder;
-		ResultRelInfo		   *resultRelInfo;
+		ResultRelInfo		   *rri;
 
 		/* Switch to per-tuple context */
 		old_mcxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
@@ -760,30 +752,28 @@ partition_filter_exec(CustomScanState *node)
 		MemoryContextSwitchTo(old_mcxt);
 		ResetExprContext(econtext);
 
-		resultRelInfo = rri_holder->result_rel_info;
+		rri = rri_holder->result_rel_info;
 
 		/* Magic: replace parent's ResultRelInfo with ours */
-		estate->es_result_relation_info = resultRelInfo;
+		estate->es_result_relation_info = rri;
 
 		/* If there's a transform map, rebuild the tuple */
 		if (rri_holder->tuple_map)
 		{
 			HeapTuple	htup_old,
 						htup_new;
-			Relation	child_rel = resultRelInfo->ri_RelationDesc;
+			Relation	child_rel = rri->ri_RelationDesc;
 
 			htup_old = ExecMaterializeSlot(slot);
 			htup_new = do_convert_tuple(htup_old, rri_holder->tuple_map);
+			ExecClearTuple(slot);
 
 			/* Allocate new slot if needed */
 			if (!state->tup_convert_slot)
 				state->tup_convert_slot = MakeTupleTableSlotCompat();
 
 			ExecSetSlotDescriptor(state->tup_convert_slot, RelationGetDescr(child_rel));
-			ExecStoreTuple(htup_new, state->tup_convert_slot, InvalidBuffer, true);
-
-			/* Now replace the original slot */
-			slot = state->tup_convert_slot;
+			slot = ExecStoreTuple(htup_new, state->tup_convert_slot, InvalidBuffer, true);
 		}
 
 		return slot;
@@ -826,7 +816,7 @@ partition_filter_explain(CustomScanState *node, List *ancestors, ExplainState *e
 /*
  * Build partition filter's target list pointing to subplan tuple's elements.
  */
-static List *
+List *
 pfilter_build_tlist(Plan *subplan)
 {
 	List	   *result_tlist = NIL;

src/partition_router.c

@@ -28,9 +28,10 @@ bool				pg_pathman_enable_partition_router = true;
 CustomScanMethods	partition_router_plan_methods;
 CustomExecMethods	partition_router_exec_methods;
 
-static bool ExecDeleteInternal(ItemPointer tupleid,
-							   EPQState *epqstate,
-							   EState *estate);
+static TupleTableSlot *ExecDeleteInternal(TupleTableSlot *slot,
+										  ItemPointer tupleid,
+										  EPQState *epqstate,
+										  EState *estate);
 
 void
 init_partition_router_static_data(void)
@@ -70,15 +71,6 @@ make_partition_router(Plan *subplan,
 
 {
 	CustomScan *cscan = makeNode(CustomScan);
-	Plan	   *pfilter;
-
-	/* Create child PartitionFilter node */
-	pfilter = make_partition_filter(subplan,
-									parent_relid,
-									parent_rti,
-									ONCONFLICT_NONE,
-									returning_list,
-									CMD_UPDATE);
 
 	/* Copy costs etc */
 	cscan->scan.plan.startup_cost	= subplan->startup_cost;
@@ -88,14 +80,14 @@ make_partition_router(Plan *subplan,
 
 	/* Setup methods, child plan and param number for EPQ */
 	cscan->methods = &partition_router_plan_methods;
-	cscan->custom_plans = list_make1(pfilter);
+	cscan->custom_plans = list_make1(subplan);
 	cscan->custom_private = list_make1(makeInteger(epq_param));
 
 	/* No physical relation will be scanned */
 	cscan->scan.scanrelid = 0;
 
 	/* Build an appropriate target list */
-	cscan->scan.plan.targetlist = pfilter->targetlist;
+	cscan->scan.plan.targetlist = pfilter_build_tlist(subplan);
 
 	/* FIXME: should we use the same tlist? */
 	cscan->custom_scan_tlist = subplan->targetlist;
@@ -126,6 +118,9 @@ partition_router_begin(CustomScanState *node, EState *estate, int eflags)
 {
 	PartitionRouterState   *state = (PartitionRouterState *) node;
 
+	/* Remember current relation we're going to delete from */
+	state->current_rri = estate->es_result_relation_info;
+
 	EvalPlanQualInit(&state->epqstate, estate,
 					 state->subplan, NIL,
 					 state->epqparam);
@@ -148,26 +143,18 @@ partition_router_exec(CustomScanState *node)
 
 	if (!TupIsNull(slot))
 	{
-		ResultRelInfo		   *new_rri,	/* new tuple owner */
-							   *old_rri;	/* previous tuple owner */
-		PartitionFilterState   *child_state;
-		char					relkind;
-		ItemPointerData			ctid;
+		ResultRelInfo	   *current_rri = state->current_rri;
+		char				relkind;
+		ItemPointerData		ctid;
 
 		ItemPointerSetInvalid(&ctid);
 
-		child_state = (PartitionFilterState *) child_ps;
-		Assert(child_state->command_type == CMD_UPDATE);
-
-		old_rri = child_state->result_parts.base_rri;
-		new_rri = estate->es_result_relation_info;
-
 		/* Build new junkfilter if we have to */
 		if (state->junkfilter == NULL)
 		{
 			state->junkfilter =
 				ExecInitJunkFilter(state->subplan->targetlist,
-								   old_rri->ri_RelationDesc->rd_att->tdhasoid,
+								   current_rri->ri_RelationDesc->rd_att->tdhasoid,
 								   ExecInitExtraTupleSlotCompat(estate));
 
 			state->junkfilter->jf_junkAttNo =
@@ -177,13 +164,14 @@ partition_router_exec(CustomScanState *node)
 				elog(ERROR, "could not find junk ctid column");
 		}
 
-		relkind = old_rri->ri_RelationDesc->rd_rel->relkind;
+		/* Additional checks based on 'relkind' */
+		relkind = current_rri->ri_RelationDesc->rd_rel->relkind;
 		if (relkind == RELKIND_RELATION)
 		{
 			Datum	ctid_datum;
 			bool	ctid_isnull;
 
-			ctid_datum = ExecGetJunkAttribute(child_state->subplan_slot,
+			ctid_datum = ExecGetJunkAttribute(slot,
 											  state->junkfilter->jf_junkAttNo,
 											  &ctid_isnull);
 
@@ -199,30 +187,26 @@ partition_router_exec(CustomScanState *node)
 		else
 			elog(ERROR, UPDATE_NODE_NAME " cannot handle relkind %u", relkind);
 
-		/*
-		 * Clean from junk attributes before INSERT,
-		 * but only if slot wasn't transformed in PartitionFilter.
-		 */
-		if (TupIsNull(child_state->tup_convert_slot))
-			slot = ExecFilterJunk(state->junkfilter, slot);
+		elog(INFO, "deleting (%d, %d) from table: %s",
+			 ItemPointerGetBlockNumber(&ctid),
+			 ItemPointerGetOffsetNumber(&ctid),
+			 get_rel_name(RelationGetRelid(current_rri->ri_RelationDesc)));
 
-		/* Magic: replace current ResultRelInfo with parent's one (DELETE) */
-		estate->es_result_relation_info = old_rri;
+		/* Magic: replace parent's ResultRelInfo with ours */
+		estate->es_result_relation_info = current_rri;
 
 		/* Delete tuple from old partition */
 		Assert(ItemPointerIsValid(&ctid));
-		EvalPlanQualSetSlot(&state->epqstate, child_state->subplan_slot);
-		if (!ExecDeleteInternal(&ctid, &state->epqstate, estate))
+		slot = ExecDeleteInternal(slot, &ctid, &state->epqstate, estate);
+
+		if (TupIsNull(slot))
 		{
 			elog(INFO, "oops, deleted, taking next tuple!");
 			goto take_next_tuple;
 		}
 
-		/* Magic: replace parent's ResultRelInfo with child's one (INSERT) */
-		estate->es_result_relation_info = new_rri;
-
 		/* Tuple will be inserted by ModifyTable */
-		return slot;
+		return ExecFilterJunk(state->junkfilter, slot);
 	}
 
 	return NULL;
@@ -231,7 +215,10 @@ partition_router_exec(CustomScanState *node)
 void
 partition_router_end(CustomScanState *node)
 {
+	PartitionRouterState *state = (PartitionRouterState *) node;
+
 	Assert(list_length(node->custom_ps) == 1);
+	EvalPlanQualEnd(&state->epqstate);
 	ExecEndNode((PlanState *) linitial(node->custom_ps));
 }
 
@@ -256,8 +243,9 @@ partition_router_explain(CustomScanState *node, List *ancestors, ExplainState *e
  * ----------------------------------------------------------------
  */
 
-static bool
-ExecDeleteInternal(ItemPointer tupleid,
+static TupleTableSlot *
+ExecDeleteInternal(TupleTableSlot *slot,
+				   ItemPointer tupleid,
 				   EPQState *epqstate,
 				   EState *estate)
 {
@@ -284,13 +272,15 @@ ExecDeleteInternal(ItemPointer tupleid,
 		rri->ri_TrigDesc->trig_delete_before_row)
 	{
 		if (!ExecBRDeleteTriggers(estate, epqstate, rri, tupleid, NULL))
-			return false;
+			return NULL;
 	}
 
 	if (tupleid != NULL)
 	{
-		/* delete the tuple */
+		EvalPlanQualSetSlot(epqstate, slot);
+
 ldelete:
+		/* delete the tuple */
 		result = heap_delete_compat(rel, tupleid,
 									estate->es_output_cid,
 									estate->es_crosscheck_snapshot,
@@ -305,8 +295,8 @@ ExecDeleteInternal(ItemPointer tupleid,
 							 errmsg("tuple to be updated was already modified by an operation triggered by the current command"),
 							 errhint("Consider using an AFTER trigger instead of a BEFORE trigger to propagate changes to other rows.")));
 
-				/* Else, already deleted by self; nothing to do */
-				return false;
+				/* Already deleted by self; nothing to do */
+				return NULL;
 
 			case HeapTupleMayBeUpdated:
 				break;
@@ -333,12 +323,13 @@ ExecDeleteInternal(ItemPointer tupleid,
 					{
 						Assert(tupleid != NULL);
 						*tupleid = hufd.ctid;
+						slot = epqslot;
 						goto ldelete;
 					}
 				}
 
-				/* tuple already deleted; nothing to do */
-				return false;
+				/* Tuple already deleted; nothing to do */
+				return NULL;
 
 			default:
 				elog(ERROR, "unrecognized heap_delete status: %u", result);
@@ -350,5 +341,5 @@ ExecDeleteInternal(ItemPointer tupleid,
 	/* AFTER ROW DELETE triggers */
 	ExecARDeleteTriggersCompat(estate, rri, tupleid, NULL, NULL);
 
-	return true;
+	return slot;
 }