WIP various changes due to EPQ

Author: funbringer

src/include/partition_router.h

@@ -32,6 +32,8 @@ typedef struct PartitionRouterState
 
 	Oid					partitioned_table;
 	JunkFilter		   *junkfilter;
+	EPQState			epqstate;
+	int					epqparam;
 	Plan			   *subplan; /* proxy variable to store subplan */
 } PartitionRouterState;
 
@@ -64,6 +66,7 @@ void init_partition_router_static_data(void);
 Plan *make_partition_router(Plan *subplan,
 							Oid parent_relid,
 							Index parent_rti,
+							int epq_param,
 							List *returning_list);
 
 

src/partition_filter.c

@@ -87,7 +87,7 @@ 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(Relation parent_rel, Plan *subplan);
+static List *pfilter_build_tlist(Plan *subplan);
 
 static void pf_memcxt_callback(void *arg);
 static estate_mod_data * fetch_estate_mod_data(EState *estate);
@@ -637,7 +637,6 @@ make_partition_filter(Plan *subplan,
 					  CmdType command_type)
 {
 	CustomScan *cscan = makeNode(CustomScan);
-	Relation	parent_rel;
 
 	/* Currently we don't support ON CONFLICT clauses */
 	if (conflict_action != ONCONFLICT_NONE)
@@ -655,14 +654,12 @@ make_partition_filter(Plan *subplan,
 	cscan->methods = &partition_filter_plan_methods;
 	cscan->custom_plans = list_make1(subplan);
 
-	/* Build an appropriate target list using a cached Relation entry */
-	parent_rel = RelationIdGetRelation(parent_relid);
-	cscan->scan.plan.targetlist = pfilter_build_tlist(parent_rel, subplan);
-	RelationClose(parent_rel);
-
 	/* No physical relation will be scanned */
 	cscan->scan.scanrelid = 0;
 
+	/* Build an appropriate target list */
+	cscan->scan.plan.targetlist = pfilter_build_tlist(subplan);
+
 	/* Prepare 'custom_scan_tlist' for EXPLAIN (VERBOSE) */
 	cscan->custom_scan_tlist = copyObject(cscan->scan.plan.targetlist);
 	ChangeVarNodes((Node *) cscan->custom_scan_tlist, INDEX_VAR, parent_rti, 0);
@@ -830,44 +827,37 @@ partition_filter_explain(CustomScanState *node, List *ancestors, ExplainState *e
  * Build partition filter's target list pointing to subplan tuple's elements.
  */
 static List *
-pfilter_build_tlist(Relation parent_rel, Plan *subplan)
+pfilter_build_tlist(Plan *subplan)
 {
 	List	   *result_tlist = NIL;
 	ListCell   *lc;
 
 	foreach (lc, subplan->targetlist)
 	{
-		TargetEntry		   *tle = (TargetEntry *) lfirst(lc),
-						   *newtle = NULL;
+		TargetEntry	   *tle = (TargetEntry *) lfirst(lc),
+					   *newtle = NULL;
 
 		if (IsA(tle->expr, Const))
-			newtle = makeTargetEntry(copyObject(tle->expr), tle->resno, tle->resname,
-											tle->resjunk);
-
+		{
+			/* TODO: maybe we should use copyObject(tle->expr)? */
+			newtle = makeTargetEntry(tle->expr,
+									 tle->resno,
+									 tle->resname,
+									 tle->resjunk);
+		}
 		else
 		{
-			if (tle->expr != NULL && IsA(tle->expr, Var))
-			{
-				Var *var = (Var *) palloc(sizeof(Var));
-				*var = *((Var *)(tle->expr));
-				var->varno = INDEX_VAR;
-				var->varattno = tle->resno;
-
-				newtle = makeTargetEntry((Expr *) var, tle->resno, tle->resname,
-											tle->resjunk);
-			}
-			else
-			{
-				Var *var = makeVar(INDEX_VAR,	/* point to subplan's elements */
-								   tle->resno,
-								   exprType((Node *) tle->expr),
-								   exprTypmod((Node *) tle->expr),
-								   exprCollation((Node *) tle->expr),
-								   0);
-
-				newtle = makeTargetEntry((Expr *) var, tle->resno, tle->resname,
-											tle->resjunk);
-			}
+			Var *var = makeVar(INDEX_VAR,	/* point to subplan's elements */
+							   tle->resno,
+							   exprType((Node *) tle->expr),
+							   exprTypmod((Node *) tle->expr),
+							   exprCollation((Node *) tle->expr),
+							   0);
+
+			newtle = makeTargetEntry((Expr *) var,
+									 tle->resno,
+									 tle->resname,
+									 tle->resjunk);
 		}
 
 		result_tlist = lappend(result_tlist, newtle);

src/partition_router.c

@@ -28,9 +28,9 @@ bool				pg_pathman_enable_partition_router = true;
 CustomScanMethods	partition_router_plan_methods;
 CustomExecMethods	partition_router_exec_methods;
 
-static TupleTableSlot *ExecDeleteInternal(ItemPointer tupleid,
-										  EPQState *epqstate,
-										  EState *estate);
+static bool ExecDeleteInternal(ItemPointer tupleid,
+							   EPQState *epqstate,
+							   EState *estate);
 
 void
 init_partition_router_static_data(void)
@@ -65,6 +65,7 @@ Plan *
 make_partition_router(Plan *subplan,
 					  Oid parent_relid,
 					  Index parent_rti,
+					  int epq_param,
 					  List *returning_list)
 
 {
@@ -85,16 +86,17 @@ make_partition_router(Plan *subplan,
 	cscan->scan.plan.plan_rows		= subplan->plan_rows;
 	cscan->scan.plan.plan_width		= subplan->plan_width;
 
-	/* Setup methods and child plan */
+	/* Setup methods, child plan and param number for EPQ */
 	cscan->methods = &partition_router_plan_methods;
 	cscan->custom_plans = list_make1(pfilter);
-
-	/* Build an appropriate target list */
-	cscan->scan.plan.targetlist = pfilter->targetlist;
+	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;
+
 	/* FIXME: should we use the same tlist? */
 	cscan->custom_scan_tlist = subplan->targetlist;
 
@@ -113,7 +115,9 @@ partition_router_create_scan_state(CustomScan *node)
 	state->css.methods = &partition_router_exec_methods;
 
 	/* Extract necessary variables */
+	state->epqparam = intVal(linitial(node->custom_private));
 	state->subplan = (Plan *) linitial(node->custom_plans);
+
 	return (Node *) state;
 }
 
@@ -122,6 +126,10 @@ partition_router_begin(CustomScanState *node, EState *estate, int eflags)
 {
 	PartitionRouterState   *state = (PartitionRouterState *) node;
 
+	EvalPlanQualInit(&state->epqstate, estate,
+					 state->subplan, NIL,
+					 state->epqparam);
+
 	/* It's convenient to store PlanState in 'custom_ps' */
 	node->custom_ps = list_make1(ExecInitNode(state->subplan, estate, eflags));
 }
@@ -134,14 +142,14 @@ partition_router_exec(CustomScanState *node)
 	TupleTableSlot			*slot;
 	PartitionRouterState	*state = (PartitionRouterState *) node;
 
+take_next_tuple:
 	/* execute PartitionFilter child node */
 	slot = ExecProcNode(child_ps);
 
 	if (!TupIsNull(slot))
 	{
 		ResultRelInfo		   *new_rri,	/* new tuple owner */
 							   *old_rri;	/* previous tuple owner */
-		EPQState				epqstate;
 		PartitionFilterState   *child_state;
 		char					relkind;
 		ItemPointerData			ctid;
@@ -203,8 +211,12 @@ partition_router_exec(CustomScanState *node)
 
 		/* Delete tuple from old partition */
 		Assert(ItemPointerIsValid(&ctid));
-		EvalPlanQualSetSlot(&epqstate, child_state->subplan_slot);
-		ExecDeleteInternal(&ctid, &epqstate, estate);
+		EvalPlanQualSetSlot(&state->epqstate, child_state->subplan_slot);
+		if (!ExecDeleteInternal(&ctid, &state->epqstate, estate))
+		{
+			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;
@@ -244,40 +256,42 @@ partition_router_explain(CustomScanState *node, List *ancestors, ExplainState *e
  * ----------------------------------------------------------------
  */
 
-static TupleTableSlot *
+static bool
 ExecDeleteInternal(ItemPointer tupleid,
 				   EPQState *epqstate,
 				   EState *estate)
 {
-	ResultRelInfo			*resultRelInfo;
-	Relation				 resultRelationDesc;
+	ResultRelInfo			*rri;
+	Relation				 rel;
 	HTSU_Result				 result;
 	HeapUpdateFailureData	 hufd;
 
 	/*
 	 * get information on the (current) result relation
 	 */
-	resultRelInfo = estate->es_result_relation_info;
-	resultRelationDesc = resultRelInfo->ri_RelationDesc;
+	rri = estate->es_result_relation_info;
+	rel = rri->ri_RelationDesc;
 
-	/* BEFORE ROW DELETE Triggers */
-	if (resultRelInfo->ri_TrigDesc &&
-		resultRelInfo->ri_TrigDesc->trig_delete_before_row)
+	/* BEFORE ROW UPDATE triggers */
+	if (rri->ri_TrigDesc &&
+		rri->ri_TrigDesc->trig_update_before_row)
 	{
-		bool		dodelete;
-
-		dodelete = ExecBRDeleteTriggers(estate, epqstate, resultRelInfo,
-										tupleid, NULL);
+		elog(INFO, "kek!");
+	}
 
-		if (!dodelete)
-			elog(ERROR, "the old row always should be deleted from child table");
+	/* BEFORE ROW DELETE triggers */
+	if (rri->ri_TrigDesc &&
+		rri->ri_TrigDesc->trig_delete_before_row)
+	{
+		if (!ExecBRDeleteTriggers(estate, epqstate, rri, tupleid, NULL))
+			return false;
 	}
 
 	if (tupleid != NULL)
 	{
 		/* delete the tuple */
 ldelete:
-		result = heap_delete_compat(resultRelationDesc, tupleid,
+		result = heap_delete_compat(rel, tupleid,
 									estate->es_output_cid,
 									estate->es_crosscheck_snapshot,
 									true /* wait for commit */ ,
@@ -292,7 +306,7 @@ ExecDeleteInternal(ItemPointer tupleid,
 							 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 NULL;
+				return false;
 
 			case HeapTupleMayBeUpdated:
 				break;
@@ -302,37 +316,39 @@ ExecDeleteInternal(ItemPointer tupleid,
 					ereport(ERROR,
 							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
 							 errmsg("could not serialize access due to concurrent update")));
+
 				if (!ItemPointerEquals(tupleid, &hufd.ctid))
 				{
 					TupleTableSlot *epqslot;
 
 					epqslot = EvalPlanQual(estate,
 										   epqstate,
-										   resultRelationDesc,
-										   resultRelInfo->ri_RangeTableIndex,
+										   rel,
+										   rri->ri_RangeTableIndex,
 										   LockTupleExclusive,
 										   &hufd.ctid,
 										   hufd.xmax);
+
 					if (!TupIsNull(epqslot))
 					{
 						Assert(tupleid != NULL);
 						*tupleid = hufd.ctid;
 						goto ldelete;
 					}
 				}
+
 				/* tuple already deleted; nothing to do */
-				return NULL;
+				return false;
 
 			default:
 				elog(ERROR, "unrecognized heap_delete status: %u", result);
-				return NULL;
 		}
 	}
 	else
 		elog(ERROR, "tupleid should be specified for deletion");
 
-	/* AFTER ROW DELETE Triggers */
-	ExecARDeleteTriggersCompat(estate, resultRelInfo, tupleid, NULL, NULL);
+	/* AFTER ROW DELETE triggers */
+	ExecARDeleteTriggersCompat(estate, rri, tupleid, NULL, NULL);
 
-	return NULL;
+	return true;
 }

src/planner_tree_modification.c

@@ -666,15 +666,14 @@ partition_router_visitor(Plan *plan, void *context)
 
 	if (modifytable_contains_fdw(rtable, modify_table))
 	{
-		ereport(NOTICE,
-				(errcode(ERRCODE_STATEMENT_TOO_COMPLEX),
-				 errmsg("discovered mix of local and foreign tables, "
-						UPDATE_NODE_NAME " will be disabled")));
-		return;
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg(UPDATE_NODE_NAME " does not support foreign data wrappers")));
 	}
 
 	lc3 = list_head(modify_table->returningLists);
-	forboth (lc1, modify_table->plans, lc2, modify_table->resultRelations)
+	forboth (lc1, modify_table->plans,
+			 lc2, modify_table->resultRelations)
 	{
 		Index	rindex = lfirst_int(lc2);
 		Oid		relid = getrelid(rindex, rtable),
@@ -698,6 +697,7 @@ partition_router_visitor(Plan *plan, void *context)
 
 			lfirst(lc1) = make_partition_router((Plan *) lfirst(lc1), relid,
 												modify_table->nominalRelation,
+												modify_table->epqParam,
 												returning_list);
 		}
 	}