Allow FDWs and custom scan providers to replace joins with scans.

Foreign data wrappers can use this capability for so-called "join
pushdown"; that is, instead of executing two separate foreign scans
and then joining the results locally, they can generate a path which
performs the join on the remote server and then is scanned locally.
This commit does not extend postgres_fdw to take advantage of this
capability; it just provides the infrastructure.

Custom scan providers can use this in a similar way.  Previously,
it was only possible for a custom scan provider to scan a single
relation.  Now, it can scan an entire join tree, provided of course
that it knows how to produce the same results that the join would
have produced if executed normally.

KaiGai Kohei, reviewed by Shigeru Hanada, Ashutosh Bapat, and me.

Author: robertmhaas

doc/src/sgml/custom-scan.sgml

@@ -81,6 +81,28 @@ typedef struct CustomPath
     detailed below.
   </para>
 
+  <para>
+   A custom scan provider can also add join paths; in this case, the scan
+   must produce the same output as would normally be produced by the join
+   it replaces.  To do this, the join provider should set the following hook.
+   This hook may be invoked repeatedly for the same pair of relations, with
+   different combinations of inner and outer relations; it is the
+   responsibility of the hook to minimize duplicated work.
+<programlisting>
+typedef void (*set_join_pathlist_hook_type) (PlannerInfo *root,
+                                             RelOptInfo *joinrel,
+                                             RelOptInfo *outerrel,
+                                             RelOptInfo *innerrel,
+                                             List *restrictlist,
+                                             JoinType jointype,
+                                             SpecialJoinInfo *sjinfo,
+                                             SemiAntiJoinFactors *semifactors,
+                                             Relids param_source_rels,
+                                             Relids extra_lateral_rels);
+extern PGDLLIMPORT set_join_pathlist_hook_type set_join_pathlist_hook;
+</programlisting>
+  </para>
+
   <sect2 id="custom-scan-path-callbacks">
   <title>Custom Path Callbacks</title>
 
@@ -124,7 +146,9 @@ typedef struct CustomScan
     Scan      scan;
     uint32    flags;
     List     *custom_exprs;
+    List     *custom_ps_tlist;
     List     *custom_private;
+    List     *custom_relids;
     const CustomScanMethods *methods;
 } CustomScan;
 </programlisting>
@@ -141,11 +165,27 @@ typedef struct CustomScan
     is only used by the custom scan provider itself.  Plan trees must be able
     to be duplicated using <function>copyObject</>, so all the data stored
     within these two fields must consist of nodes that function can handle.
+    <literal>custom_relids</> is set by the core code to the set of relations
+    which this scan node must handle; except when this scan is replacing a
+    join, it will have only one member.
     <structfield>methods</> must point to a (usually statically allocated)
     object implementing the required custom scan methods, which are further
     detailed below.
   </para>
 
+  <para>
+   When a <structname>CustomScan</> scans a single relation,
+   <structfield>scan.scanrelid</> should be the range table index of the table
+   to be scanned, and <structfield>custom_ps_tlist</> should be
+   <literal>NULL</>.  When it replaces a join, <structfield>scan.scanrelid</>
+   should be zero, and <structfield>custom_ps_tlist</> should be a list of
+   <structname>TargetEntry</> nodes.  This is necessary because, when a join
+   is replaced, the target list cannot be constructed from the table
+   definition.  At execution time, this list will be used to initialize the
+   tuple descriptor of the <structname>TupleTableSlot</>.  It will also be
+   used by <command>EXPLAIN</>, when deparsing.
+  </para>
+
   <sect2 id="custom-scan-plan-callbacks">
    <title>Custom Scan Callbacks</title>
    <para>

doc/src/sgml/fdwhandler.sgml

@@ -598,6 +598,42 @@ IsForeignRelUpdatable (Relation rel);
 
    </sect2>
 
+   <sect2>
+    <title>FDW Routines For Remote Joins</title>
+    <para>
+<programlisting>
+void
+GetForeignJoinPaths(PlannerInfo *root,
+                    RelOptInfo *joinrel,
+                    RelOptInfo *outerrel,
+                    RelOptInfo *innerrel,
+                    List *restrictlist,
+                    JoinType jointype,
+                    SpecialJoinInfo *sjinfo,
+                    SemiAntiJoinFactors *semifactors,
+                    Relids param_source_rels,
+                    Relids extra_lateral_rels);
+</programlisting>
+     Create possible access paths for a join of two foreign tables managed
+     by the same foreign data wrapper.
+     This optional function is called during query planning.
+    </para>
+    <para>
+     This function the FDW to add <structname>ForeignScan</> paths for the
+     supplied <literal>joinrel</>.  Typically, the FDW will send the whole
+     join to the remote server as a single query, as performing the join
+     remotely rather than locally is typically much more efficient.
+    </para>
+    <para>
+     Since we cannot construct the slot descriptor for a remote join from
+     the catalogs, the FDW should set the <structfield>scanrelid</> of the
+     <structname>ForeignScan</> to zero and <structfield>fdw_ps_tlist</>
+     to an appropriate list of <structfield>TargetEntry</> nodes.
+     Junk entries will be ignored, but can be present for the benefit of
+     deparsing performed by <command>EXPLAIN</>.
+    </para>
+   </sect2>
+
    <sect2 id="fdw-callbacks-explain">
     <title>FDW Routines for <command>EXPLAIN</></title>
 

src/backend/commands/explain.c

@@ -730,11 +730,17 @@ ExplainPreScanNode(PlanState *planstate, Bitmapset **rels_used)
 		case T_ValuesScan:
 		case T_CteScan:
 		case T_WorkTableScan:
-		case T_ForeignScan:
-		case T_CustomScan:
 			*rels_used = bms_add_member(*rels_used,
 										((Scan *) plan)->scanrelid);
 			break;
+		case T_ForeignScan:
+			*rels_used = bms_add_members(*rels_used,
+										 ((ForeignScan *) plan)->fdw_relids);
+			break;
+		case T_CustomScan:
+			*rels_used = bms_add_members(*rels_used,
+										 ((CustomScan *) plan)->custom_relids);
+			break;
 		case T_ModifyTable:
 			*rels_used = bms_add_member(*rels_used,
 									((ModifyTable *) plan)->nominalRelation);
@@ -1072,9 +1078,12 @@ ExplainNode(PlanState *planstate, List *ancestors,
 		case T_ValuesScan:
 		case T_CteScan:
 		case T_WorkTableScan:
+			ExplainScanTarget((Scan *) plan, es);
+			break;
 		case T_ForeignScan:
 		case T_CustomScan:
-			ExplainScanTarget((Scan *) plan, es);
+			if (((Scan *) plan)->scanrelid > 0)
+				ExplainScanTarget((Scan *) plan, es);
 			break;
 		case T_IndexScan:
 			{

src/backend/executor/execScan.c

@@ -251,6 +251,12 @@ ExecAssignScanProjectionInfo(ScanState *node)
 	/* Vars in an index-only scan's tlist should be INDEX_VAR */
 	if (IsA(scan, IndexOnlyScan))
 		varno = INDEX_VAR;
+	/* Also foreign or custom scan on pseudo relation should be INDEX_VAR */
+	else if (scan->scanrelid == 0)
+	{
+		Assert(IsA(scan, ForeignScan) || IsA(scan, CustomScan));
+		varno = INDEX_VAR;
+	}
 	else
 		varno = scan->scanrelid;
 

src/backend/executor/nodeCustom.c

@@ -23,7 +23,7 @@ CustomScanState *
 ExecInitCustomScan(CustomScan *cscan, EState *estate, int eflags)
 {
 	CustomScanState    *css;
-	Relation			scan_rel;
+	Index				scan_relid = cscan->scan.scanrelid;
 
 	/* populate a CustomScanState according to the CustomScan */
 	css = (CustomScanState *) cscan->methods->CreateCustomScanState(cscan);
@@ -48,12 +48,26 @@ ExecInitCustomScan(CustomScan *cscan, EState *estate, int eflags)
 	ExecInitScanTupleSlot(estate, &css->ss);
 	ExecInitResultTupleSlot(estate, &css->ss.ps);
 
-	/* initialize scan relation */
-	scan_rel = ExecOpenScanRelation(estate, cscan->scan.scanrelid, eflags);
-	css->ss.ss_currentRelation = scan_rel;
-	css->ss.ss_currentScanDesc = NULL;	/* set by provider */
-	ExecAssignScanType(&css->ss, RelationGetDescr(scan_rel));
-
+	/*
+	 * open the base relation and acquire an appropriate lock on it;
+	 * also, get and assign the scan type
+	 */
+	if (scan_relid > 0)
+	{
+		Relation		scan_rel;
+
+		scan_rel = ExecOpenScanRelation(estate, scan_relid, eflags);
+		css->ss.ss_currentRelation = scan_rel;
+		css->ss.ss_currentScanDesc = NULL;	/* set by provider */
+		ExecAssignScanType(&css->ss, RelationGetDescr(scan_rel));
+	}
+	else
+	{
+		TupleDesc	ps_tupdesc;
+
+		ps_tupdesc = ExecCleanTypeFromTL(cscan->custom_ps_tlist, false);
+		ExecAssignScanType(&css->ss, ps_tupdesc);
+	}
 	css->ss.ps.ps_TupFromTlist = false;
 
 	/*
@@ -89,11 +103,11 @@ ExecEndCustomScan(CustomScanState *node)
 
 	/* Clean out the tuple table */
 	ExecClearTuple(node->ss.ps.ps_ResultTupleSlot);
-	if (node->ss.ss_ScanTupleSlot)
-		ExecClearTuple(node->ss.ss_ScanTupleSlot);
+	ExecClearTuple(node->ss.ss_ScanTupleSlot);
 
 	/* Close the heap relation */
-	ExecCloseScanRelation(node->ss.ss_currentRelation);
+	if (node->ss.ss_currentRelation)
+		ExecCloseScanRelation(node->ss.ss_currentRelation);
 }
 
 void

src/backend/executor/nodeForeignscan.c

@@ -102,7 +102,7 @@ ForeignScanState *
 ExecInitForeignScan(ForeignScan *node, EState *estate, int eflags)
 {
 	ForeignScanState *scanstate;
-	Relation	currentRelation;
+	Index		scanrelid = node->scan.scanrelid;
 	FdwRoutine *fdwroutine;
 
 	/* check for unsupported flags */
@@ -141,16 +141,24 @@ ExecInitForeignScan(ForeignScan *node, EState *estate, int eflags)
 	ExecInitScanTupleSlot(estate, &scanstate->ss);
 
 	/*
-	 * open the base relation and acquire appropriate lock on it.
+	 * open the base relation and acquire an appropriate lock on it;
+	 * also, get and assign the scan type
 	 */
-	currentRelation = ExecOpenScanRelation(estate, node->scan.scanrelid, eflags);
-	scanstate->ss.ss_currentRelation = currentRelation;
+	if (scanrelid > 0)
+	{
+		Relation	currentRelation;
 
-	/*
-	 * get the scan type from the relation descriptor.  (XXX at some point we
-	 * might want to let the FDW editorialize on the scan tupdesc.)
-	 */
-	ExecAssignScanType(&scanstate->ss, RelationGetDescr(currentRelation));
+		currentRelation = ExecOpenScanRelation(estate, scanrelid, eflags);
+		scanstate->ss.ss_currentRelation = currentRelation;
+		ExecAssignScanType(&scanstate->ss, RelationGetDescr(currentRelation));
+	}
+	else
+	{
+		TupleDesc	ps_tupdesc;
+
+		ps_tupdesc = ExecCleanTypeFromTL(node->fdw_ps_tlist, false);
+		ExecAssignScanType(&scanstate->ss, ps_tupdesc);
+	}
 
 	/*
 	 * Initialize result tuple type and projection info.
@@ -161,7 +169,7 @@ ExecInitForeignScan(ForeignScan *node, EState *estate, int eflags)
 	/*
 	 * Acquire function pointers from the FDW's handler, and init fdw_state.
 	 */
-	fdwroutine = GetFdwRoutineForRelation(currentRelation, true);
+	fdwroutine = GetFdwRoutine(node->fdw_handler);
 	scanstate->fdwroutine = fdwroutine;
 	scanstate->fdw_state = NULL;
 
@@ -193,7 +201,8 @@ ExecEndForeignScan(ForeignScanState *node)
 	ExecClearTuple(node->ss.ss_ScanTupleSlot);
 
 	/* close the relation. */
-	ExecCloseScanRelation(node->ss.ss_currentRelation);
+	if (node->ss.ss_currentRelation)
+		ExecCloseScanRelation(node->ss.ss_currentRelation);
 }
 
 /* ----------------------------------------------------------------

src/backend/foreign/foreign.c

@@ -304,11 +304,11 @@ GetFdwRoutine(Oid fdwhandler)
 
 
 /*
- * GetFdwRoutineByRelId - look up the handler of the foreign-data wrapper
- * for the given foreign table, and retrieve its FdwRoutine struct.
+ * GetFdwHandlerByRelId - look up the handler of the foreign-data wrapper
+ * for the given foreign table
  */
-FdwRoutine *
-GetFdwRoutineByRelId(Oid relid)
+Oid
+GetFdwHandlerByRelId(Oid relid)
 {
 	HeapTuple	tp;
 	Form_pg_foreign_data_wrapper fdwform;
@@ -350,7 +350,18 @@ GetFdwRoutineByRelId(Oid relid)
 
 	ReleaseSysCache(tp);
 
-	/* And finally, call the handler function. */
+	return fdwhandler;
+}
+
+/*
+ * GetFdwRoutineByRelId - look up the handler of the foreign-data wrapper
+ * for the given foreign table, and retrieve its FdwRoutine struct.
+ */
+FdwRoutine *
+GetFdwRoutineByRelId(Oid relid)
+{
+	Oid			fdwhandler = GetFdwHandlerByRelId(relid);
+
 	return GetFdwRoutine(fdwhandler);
 }
 

src/backend/nodes/copyfuncs.c

@@ -592,8 +592,11 @@ _copyForeignScan(const ForeignScan *from)
 	/*
 	 * copy remainder of node
 	 */
+	COPY_SCALAR_FIELD(fdw_handler);
 	COPY_NODE_FIELD(fdw_exprs);
+	COPY_NODE_FIELD(fdw_ps_tlist);
 	COPY_NODE_FIELD(fdw_private);
+	COPY_BITMAPSET_FIELD(fdw_relids);
 	COPY_SCALAR_FIELD(fsSystemCol);
 
 	return newnode;
@@ -617,7 +620,9 @@ _copyCustomScan(const CustomScan *from)
 	 */
 	COPY_SCALAR_FIELD(flags);
 	COPY_NODE_FIELD(custom_exprs);
+	COPY_NODE_FIELD(custom_ps_tlist);
 	COPY_NODE_FIELD(custom_private);
+	COPY_BITMAPSET_FIELD(custom_relids);
 
 	/*
 	 * NOTE: The method field of CustomScan is required to be a pointer to a

src/backend/nodes/outfuncs.c

@@ -558,8 +558,11 @@ _outForeignScan(StringInfo str, const ForeignScan *node)
 
 	_outScanInfo(str, (const Scan *) node);
 
+	WRITE_OID_FIELD(fdw_handler);
 	WRITE_NODE_FIELD(fdw_exprs);
+	WRITE_NODE_FIELD(fdw_ps_tlist);
 	WRITE_NODE_FIELD(fdw_private);
+	WRITE_BITMAPSET_FIELD(fdw_relids);
 	WRITE_BOOL_FIELD(fsSystemCol);
 }
 
@@ -572,7 +575,9 @@ _outCustomScan(StringInfo str, const CustomScan *node)
 
 	WRITE_UINT_FIELD(flags);
 	WRITE_NODE_FIELD(custom_exprs);
+	WRITE_NODE_FIELD(custom_ps_tlist);
 	WRITE_NODE_FIELD(custom_private);
+	WRITE_BITMAPSET_FIELD(custom_relids);
 	appendStringInfoString(str, " :methods ");
 	_outToken(str, node->methods->CustomName);
 	if (node->methods->TextOutCustomScan)

src/backend/optimizer/path/joinpath.c

@@ -17,10 +17,13 @@
 #include <math.h>
 
 #include "executor/executor.h"
+#include "foreign/fdwapi.h"
 #include "optimizer/cost.h"
 #include "optimizer/pathnode.h"
 #include "optimizer/paths.h"
 
+/* Hook for plugins to get control in add_paths_to_joinrel() */
+set_join_pathlist_hook_type set_join_pathlist_hook = NULL;
 
 #define PATH_PARAM_BY_REL(path, rel)  \
 	((path)->param_info && bms_overlap(PATH_REQ_OUTER(path), (rel)->relids))
@@ -260,6 +263,27 @@ add_paths_to_joinrel(PlannerInfo *root,
 							 restrictlist, jointype,
 							 sjinfo, &semifactors,
 							 param_source_rels, extra_lateral_rels);
+
+	/*
+	 * 5. If both inner and outer relations are managed by the same FDW,
+	 * give it a chance to push down joins.
+	 */
+	if (joinrel->fdwroutine &&
+		joinrel->fdwroutine->GetForeignJoinPaths)
+		joinrel->fdwroutine->GetForeignJoinPaths(root, joinrel,
+												 outerrel, innerrel,
+												 restrictlist, jointype, sjinfo,
+												 &semifactors,
+												 param_source_rels,
+												 extra_lateral_rels);
+	/*
+	 * 6. Finally, give extensions a chance to manipulate the path list.
+	 */
+	if (set_join_pathlist_hook)
+		set_join_pathlist_hook(root, joinrel, outerrel, innerrel,
+							   restrictlist, jointype,
+							   sjinfo, &semifactors,
+							   param_source_rels, extra_lateral_rels);
 }
 
 /*