PostgreSQL Source Code  git master
postgres_fdw.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * postgres_fdw.c
4  * Foreign-data wrapper for remote PostgreSQL servers
5  *
6  * Portions Copyright (c) 2012-2024, PostgreSQL Global Development Group
7  *
8  * IDENTIFICATION
9  * contrib/postgres_fdw/postgres_fdw.c
10  *
11  *-------------------------------------------------------------------------
12  */
13 #include "postgres.h"
14 
15 #include <limits.h>
16 
17 #include "access/htup_details.h"
18 #include "access/sysattr.h"
19 #include "access/table.h"
20 #include "catalog/pg_class.h"
21 #include "catalog/pg_opfamily.h"
22 #include "commands/defrem.h"
23 #include "commands/explain.h"
24 #include "commands/vacuum.h"
25 #include "executor/execAsync.h"
26 #include "foreign/fdwapi.h"
27 #include "funcapi.h"
28 #include "miscadmin.h"
29 #include "nodes/makefuncs.h"
30 #include "nodes/nodeFuncs.h"
31 #include "optimizer/appendinfo.h"
32 #include "optimizer/clauses.h"
33 #include "optimizer/cost.h"
34 #include "optimizer/inherit.h"
35 #include "optimizer/optimizer.h"
36 #include "optimizer/pathnode.h"
37 #include "optimizer/paths.h"
38 #include "optimizer/planmain.h"
39 #include "optimizer/prep.h"
40 #include "optimizer/restrictinfo.h"
41 #include "optimizer/tlist.h"
42 #include "parser/parsetree.h"
43 #include "postgres_fdw.h"
44 #include "storage/latch.h"
45 #include "utils/builtins.h"
46 #include "utils/float.h"
47 #include "utils/guc.h"
48 #include "utils/lsyscache.h"
49 #include "utils/memutils.h"
50 #include "utils/rel.h"
51 #include "utils/sampling.h"
52 #include "utils/selfuncs.h"
53 
55 
56 /* Default CPU cost to start up a foreign query. */
57 #define DEFAULT_FDW_STARTUP_COST 100.0
58 
59 /* Default CPU cost to process 1 row (above and beyond cpu_tuple_cost). */
60 #define DEFAULT_FDW_TUPLE_COST 0.2
61 
62 /* If no remote estimates, assume a sort costs 20% extra */
63 #define DEFAULT_FDW_SORT_MULTIPLIER 1.2
64 
65 /*
66  * Indexes of FDW-private information stored in fdw_private lists.
67  *
68  * These items are indexed with the enum FdwScanPrivateIndex, so an item
69  * can be fetched with list_nth(). For example, to get the SELECT statement:
70  * sql = strVal(list_nth(fdw_private, FdwScanPrivateSelectSql));
71  */
73 {
74  /* SQL statement to execute remotely (as a String node) */
76  /* Integer list of attribute numbers retrieved by the SELECT */
78  /* Integer representing the desired fetch_size */
80 
81  /*
82  * String describing join i.e. names of relations being joined and types
83  * of join, added when the scan is join
84  */
86 };
87 
88 /*
89  * Similarly, this enum describes what's kept in the fdw_private list for
90  * a ModifyTable node referencing a postgres_fdw foreign table. We store:
91  *
92  * 1) INSERT/UPDATE/DELETE statement text to be sent to the remote server
93  * 2) Integer list of target attribute numbers for INSERT/UPDATE
94  * (NIL for a DELETE)
95  * 3) Length till the end of VALUES clause for INSERT
96  * (-1 for a DELETE/UPDATE)
97  * 4) Boolean flag showing if the remote query has a RETURNING clause
98  * 5) Integer list of attribute numbers retrieved by RETURNING, if any
99  */
101 {
102  /* SQL statement to execute remotely (as a String node) */
104  /* Integer list of target attribute numbers for INSERT/UPDATE */
106  /* Length till the end of VALUES clause (as an Integer node) */
108  /* has-returning flag (as a Boolean node) */
110  /* Integer list of attribute numbers retrieved by RETURNING */
112 };
113 
114 /*
115  * Similarly, this enum describes what's kept in the fdw_private list for
116  * a ForeignScan node that modifies a foreign table directly. We store:
117  *
118  * 1) UPDATE/DELETE statement text to be sent to the remote server
119  * 2) Boolean flag showing if the remote query has a RETURNING clause
120  * 3) Integer list of attribute numbers retrieved by RETURNING, if any
121  * 4) Boolean flag showing if we set the command es_processed
122  */
124 {
125  /* SQL statement to execute remotely (as a String node) */
127  /* has-returning flag (as a Boolean node) */
129  /* Integer list of attribute numbers retrieved by RETURNING */
131  /* set-processed flag (as a Boolean node) */
133 };
134 
135 /*
136  * Execution state of a foreign scan using postgres_fdw.
137  */
138 typedef struct PgFdwScanState
139 {
140  Relation rel; /* relcache entry for the foreign table. NULL
141  * for a foreign join scan. */
142  TupleDesc tupdesc; /* tuple descriptor of scan */
143  AttInMetadata *attinmeta; /* attribute datatype conversion metadata */
144 
145  /* extracted fdw_private data */
146  char *query; /* text of SELECT command */
147  List *retrieved_attrs; /* list of retrieved attribute numbers */
148 
149  /* for remote query execution */
150  PGconn *conn; /* connection for the scan */
151  PgFdwConnState *conn_state; /* extra per-connection state */
152  unsigned int cursor_number; /* quasi-unique ID for my cursor */
153  bool cursor_exists; /* have we created the cursor? */
154  int numParams; /* number of parameters passed to query */
155  FmgrInfo *param_flinfo; /* output conversion functions for them */
156  List *param_exprs; /* executable expressions for param values */
157  const char **param_values; /* textual values of query parameters */
158 
159  /* for storing result tuples */
160  HeapTuple *tuples; /* array of currently-retrieved tuples */
161  int num_tuples; /* # of tuples in array */
162  int next_tuple; /* index of next one to return */
163 
164  /* batch-level state, for optimizing rewinds and avoiding useless fetch */
165  int fetch_ct_2; /* Min(# of fetches done, 2) */
166  bool eof_reached; /* true if last fetch reached EOF */
167 
168  /* for asynchronous execution */
169  bool async_capable; /* engage asynchronous-capable logic? */
170 
171  /* working memory contexts */
172  MemoryContext batch_cxt; /* context holding current batch of tuples */
173  MemoryContext temp_cxt; /* context for per-tuple temporary data */
174 
175  int fetch_size; /* number of tuples per fetch */
177 
178 /*
179  * Execution state of a foreign insert/update/delete operation.
180  */
181 typedef struct PgFdwModifyState
182 {
183  Relation rel; /* relcache entry for the foreign table */
184  AttInMetadata *attinmeta; /* attribute datatype conversion metadata */
185 
186  /* for remote query execution */
187  PGconn *conn; /* connection for the scan */
188  PgFdwConnState *conn_state; /* extra per-connection state */
189  char *p_name; /* name of prepared statement, if created */
190 
191  /* extracted fdw_private data */
192  char *query; /* text of INSERT/UPDATE/DELETE command */
193  char *orig_query; /* original text of INSERT command */
194  List *target_attrs; /* list of target attribute numbers */
195  int values_end; /* length up to the end of VALUES */
196  int batch_size; /* value of FDW option "batch_size" */
197  bool has_returning; /* is there a RETURNING clause? */
198  List *retrieved_attrs; /* attr numbers retrieved by RETURNING */
199 
200  /* info about parameters for prepared statement */
201  AttrNumber ctidAttno; /* attnum of input resjunk ctid column */
202  int p_nums; /* number of parameters to transmit */
203  FmgrInfo *p_flinfo; /* output conversion functions for them */
204 
205  /* batch operation stuff */
206  int num_slots; /* number of slots to insert */
207 
208  /* working memory context */
209  MemoryContext temp_cxt; /* context for per-tuple temporary data */
210 
211  /* for update row movement if subplan result rel */
212  struct PgFdwModifyState *aux_fmstate; /* foreign-insert state, if
213  * created */
215 
216 /*
217  * Execution state of a foreign scan that modifies a foreign table directly.
218  */
220 {
221  Relation rel; /* relcache entry for the foreign table */
222  AttInMetadata *attinmeta; /* attribute datatype conversion metadata */
223 
224  /* extracted fdw_private data */
225  char *query; /* text of UPDATE/DELETE command */
226  bool has_returning; /* is there a RETURNING clause? */
227  List *retrieved_attrs; /* attr numbers retrieved by RETURNING */
228  bool set_processed; /* do we set the command es_processed? */
229 
230  /* for remote query execution */
231  PGconn *conn; /* connection for the update */
232  PgFdwConnState *conn_state; /* extra per-connection state */
233  int numParams; /* number of parameters passed to query */
234  FmgrInfo *param_flinfo; /* output conversion functions for them */
235  List *param_exprs; /* executable expressions for param values */
236  const char **param_values; /* textual values of query parameters */
237 
238  /* for storing result tuples */
239  PGresult *result; /* result for query */
240  int num_tuples; /* # of result tuples */
241  int next_tuple; /* index of next one to return */
242  Relation resultRel; /* relcache entry for the target relation */
243  AttrNumber *attnoMap; /* array of attnums of input user columns */
244  AttrNumber ctidAttno; /* attnum of input ctid column */
245  AttrNumber oidAttno; /* attnum of input oid column */
246  bool hasSystemCols; /* are there system columns of resultRel? */
247 
248  /* working memory context */
249  MemoryContext temp_cxt; /* context for per-tuple temporary data */
251 
252 /*
253  * Workspace for analyzing a foreign table.
254  */
255 typedef struct PgFdwAnalyzeState
256 {
257  Relation rel; /* relcache entry for the foreign table */
258  AttInMetadata *attinmeta; /* attribute datatype conversion metadata */
259  List *retrieved_attrs; /* attr numbers retrieved by query */
260 
261  /* collected sample rows */
262  HeapTuple *rows; /* array of size targrows */
263  int targrows; /* target # of sample rows */
264  int numrows; /* # of sample rows collected */
265 
266  /* for random sampling */
267  double samplerows; /* # of rows fetched */
268  double rowstoskip; /* # of rows to skip before next sample */
269  ReservoirStateData rstate; /* state for reservoir sampling */
270 
271  /* working memory contexts */
272  MemoryContext anl_cxt; /* context for per-analyze lifespan data */
273  MemoryContext temp_cxt; /* context for per-tuple temporary data */
275 
276 /*
277  * This enum describes what's kept in the fdw_private list for a ForeignPath.
278  * We store:
279  *
280  * 1) Boolean flag showing if the remote query has the final sort
281  * 2) Boolean flag showing if the remote query has the LIMIT clause
282  */
284 {
285  /* has-final-sort flag (as a Boolean node) */
287  /* has-limit flag (as a Boolean node) */
289 };
290 
291 /* Struct for extra information passed to estimate_path_cost_size() */
292 typedef struct
293 {
296  bool has_limit;
297  double limit_tuples;
298  int64 count_est;
299  int64 offset_est;
301 
302 /*
303  * Identify the attribute where data conversion fails.
304  */
305 typedef struct ConversionLocation
306 {
307  AttrNumber cur_attno; /* attribute number being processed, or 0 */
308  Relation rel; /* foreign table being processed, or NULL */
309  ForeignScanState *fsstate; /* plan node being processed, or NULL */
311 
312 /* Callback argument for ec_member_matches_foreign */
313 typedef struct
314 {
315  Expr *current; /* current expr, or NULL if not yet found */
316  List *already_used; /* expressions already dealt with */
318 
319 /*
320  * SQL functions
321  */
323 
324 /*
325  * FDW callback routines
326  */
327 static void postgresGetForeignRelSize(PlannerInfo *root,
328  RelOptInfo *baserel,
329  Oid foreigntableid);
330 static void postgresGetForeignPaths(PlannerInfo *root,
331  RelOptInfo *baserel,
332  Oid foreigntableid);
334  RelOptInfo *foreignrel,
335  Oid foreigntableid,
336  ForeignPath *best_path,
337  List *tlist,
338  List *scan_clauses,
339  Plan *outer_plan);
340 static void postgresBeginForeignScan(ForeignScanState *node, int eflags);
343 static void postgresEndForeignScan(ForeignScanState *node);
345  Index rtindex,
346  RangeTblEntry *target_rte,
347  Relation target_relation);
349  ModifyTable *plan,
350  Index resultRelation,
351  int subplan_index);
352 static void postgresBeginForeignModify(ModifyTableState *mtstate,
353  ResultRelInfo *resultRelInfo,
354  List *fdw_private,
355  int subplan_index,
356  int eflags);
358  ResultRelInfo *resultRelInfo,
359  TupleTableSlot *slot,
360  TupleTableSlot *planSlot);
362  ResultRelInfo *resultRelInfo,
363  TupleTableSlot **slots,
364  TupleTableSlot **planSlots,
365  int *numSlots);
366 static int postgresGetForeignModifyBatchSize(ResultRelInfo *resultRelInfo);
368  ResultRelInfo *resultRelInfo,
369  TupleTableSlot *slot,
370  TupleTableSlot *planSlot);
372  ResultRelInfo *resultRelInfo,
373  TupleTableSlot *slot,
374  TupleTableSlot *planSlot);
375 static void postgresEndForeignModify(EState *estate,
376  ResultRelInfo *resultRelInfo);
377 static void postgresBeginForeignInsert(ModifyTableState *mtstate,
378  ResultRelInfo *resultRelInfo);
379 static void postgresEndForeignInsert(EState *estate,
380  ResultRelInfo *resultRelInfo);
382 static bool postgresPlanDirectModify(PlannerInfo *root,
383  ModifyTable *plan,
384  Index resultRelation,
385  int subplan_index);
386 static void postgresBeginDirectModify(ForeignScanState *node, int eflags);
388 static void postgresEndDirectModify(ForeignScanState *node);
390  ExplainState *es);
392  ResultRelInfo *rinfo,
393  List *fdw_private,
394  int subplan_index,
395  ExplainState *es);
397  ExplainState *es);
398 static void postgresExecForeignTruncate(List *rels,
399  DropBehavior behavior,
400  bool restart_seqs);
401 static bool postgresAnalyzeForeignTable(Relation relation,
402  AcquireSampleRowsFunc *func,
403  BlockNumber *totalpages);
405  Oid serverOid);
406 static void postgresGetForeignJoinPaths(PlannerInfo *root,
407  RelOptInfo *joinrel,
408  RelOptInfo *outerrel,
409  RelOptInfo *innerrel,
410  JoinType jointype,
411  JoinPathExtraData *extra);
413  TupleTableSlot *slot);
414 static void postgresGetForeignUpperPaths(PlannerInfo *root,
415  UpperRelationKind stage,
416  RelOptInfo *input_rel,
417  RelOptInfo *output_rel,
418  void *extra);
420 static void postgresForeignAsyncRequest(AsyncRequest *areq);
422 static void postgresForeignAsyncNotify(AsyncRequest *areq);
423 
424 /*
425  * Helper functions
426  */
427 static void estimate_path_cost_size(PlannerInfo *root,
428  RelOptInfo *foreignrel,
429  List *param_join_conds,
430  List *pathkeys,
431  PgFdwPathExtraData *fpextra,
432  double *p_rows, int *p_width,
433  Cost *p_startup_cost, Cost *p_total_cost);
434 static void get_remote_estimate(const char *sql,
435  PGconn *conn,
436  double *rows,
437  int *width,
438  Cost *startup_cost,
439  Cost *total_cost);
441  List *pathkeys,
442  double retrieved_rows,
443  double width,
444  double limit_tuples,
445  Cost *p_startup_cost,
446  Cost *p_run_cost);
447 static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel,
449  void *arg);
450 static void create_cursor(ForeignScanState *node);
451 static void fetch_more_data(ForeignScanState *node);
452 static void close_cursor(PGconn *conn, unsigned int cursor_number,
453  PgFdwConnState *conn_state);
455  RangeTblEntry *rte,
456  ResultRelInfo *resultRelInfo,
457  CmdType operation,
458  Plan *subplan,
459  char *query,
460  List *target_attrs,
461  int values_end,
462  bool has_returning,
463  List *retrieved_attrs);
465  ResultRelInfo *resultRelInfo,
466  CmdType operation,
467  TupleTableSlot **slots,
468  TupleTableSlot **planSlots,
469  int *numSlots);
470 static void prepare_foreign_modify(PgFdwModifyState *fmstate);
471 static const char **convert_prep_stmt_params(PgFdwModifyState *fmstate,
472  ItemPointer tupleid,
473  TupleTableSlot **slots,
474  int numSlots);
475 static void store_returning_result(PgFdwModifyState *fmstate,
476  TupleTableSlot *slot, PGresult *res);
477 static void finish_foreign_modify(PgFdwModifyState *fmstate);
478 static void deallocate_query(PgFdwModifyState *fmstate);
479 static List *build_remote_returning(Index rtindex, Relation rel,
480  List *returningList);
481 static void rebuild_fdw_scan_tlist(ForeignScan *fscan, List *tlist);
482 static void execute_dml_stmt(ForeignScanState *node);
484 static void init_returning_filter(PgFdwDirectModifyState *dmstate,
485  List *fdw_scan_tlist,
486  Index rtindex);
488  ResultRelInfo *resultRelInfo,
489  TupleTableSlot *slot,
490  EState *estate);
491 static void prepare_query_params(PlanState *node,
492  List *fdw_exprs,
493  int numParams,
494  FmgrInfo **param_flinfo,
495  List **param_exprs,
496  const char ***param_values);
497 static void process_query_params(ExprContext *econtext,
498  FmgrInfo *param_flinfo,
499  List *param_exprs,
500  const char **param_values);
501 static int postgresAcquireSampleRowsFunc(Relation relation, int elevel,
502  HeapTuple *rows, int targrows,
503  double *totalrows,
504  double *totaldeadrows);
505 static void analyze_row_processor(PGresult *res, int row,
506  PgFdwAnalyzeState *astate);
507 static void produce_tuple_asynchronously(AsyncRequest *areq, bool fetch);
508 static void fetch_more_data_begin(AsyncRequest *areq);
509 static void complete_pending_request(AsyncRequest *areq);
511  int row,
512  Relation rel,
513  AttInMetadata *attinmeta,
514  List *retrieved_attrs,
515  ForeignScanState *fsstate,
516  MemoryContext temp_context);
517 static void conversion_error_callback(void *arg);
518 static bool foreign_join_ok(PlannerInfo *root, RelOptInfo *joinrel,
519  JoinType jointype, RelOptInfo *outerrel, RelOptInfo *innerrel,
520  JoinPathExtraData *extra);
521 static bool foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
522  Node *havingQual);
524  RelOptInfo *rel);
527  Path *epq_path, List *restrictlist);
528 static void add_foreign_grouping_paths(PlannerInfo *root,
529  RelOptInfo *input_rel,
530  RelOptInfo *grouped_rel,
531  GroupPathExtraData *extra);
532 static void add_foreign_ordered_paths(PlannerInfo *root,
533  RelOptInfo *input_rel,
534  RelOptInfo *ordered_rel);
535 static void add_foreign_final_paths(PlannerInfo *root,
536  RelOptInfo *input_rel,
537  RelOptInfo *final_rel,
538  FinalPathExtraData *extra);
539 static void apply_server_options(PgFdwRelationInfo *fpinfo);
540 static void apply_table_options(PgFdwRelationInfo *fpinfo);
541 static void merge_fdw_options(PgFdwRelationInfo *fpinfo,
542  const PgFdwRelationInfo *fpinfo_o,
543  const PgFdwRelationInfo *fpinfo_i);
544 static int get_batch_size_option(Relation rel);
545 
546 
547 /*
548  * Foreign-data wrapper handler function: return a struct with pointers
549  * to my callback routines.
550  */
551 Datum
553 {
554  FdwRoutine *routine = makeNode(FdwRoutine);
555 
556  /* Functions for scanning foreign tables */
564 
565  /* Functions for updating foreign tables */
582 
583  /* Function for EvalPlanQual rechecks */
585  /* Support functions for EXPLAIN */
589 
590  /* Support function for TRUNCATE */
592 
593  /* Support functions for ANALYZE */
595 
596  /* Support functions for IMPORT FOREIGN SCHEMA */
598 
599  /* Support functions for join push-down */
601 
602  /* Support functions for upper relation push-down */
604 
605  /* Support functions for asynchronous execution */
610 
611  PG_RETURN_POINTER(routine);
612 }
613 
614 /*
615  * postgresGetForeignRelSize
616  * Estimate # of rows and width of the result of the scan
617  *
618  * We should consider the effect of all baserestrictinfo clauses here, but
619  * not any join clauses.
620  */
621 static void
623  RelOptInfo *baserel,
624  Oid foreigntableid)
625 {
626  PgFdwRelationInfo *fpinfo;
627  ListCell *lc;
628 
629  /*
630  * We use PgFdwRelationInfo to pass various information to subsequent
631  * functions.
632  */
633  fpinfo = (PgFdwRelationInfo *) palloc0(sizeof(PgFdwRelationInfo));
634  baserel->fdw_private = (void *) fpinfo;
635 
636  /* Base foreign tables need to be pushed down always. */
637  fpinfo->pushdown_safe = true;
638 
639  /* Look up foreign-table catalog info. */
640  fpinfo->table = GetForeignTable(foreigntableid);
641  fpinfo->server = GetForeignServer(fpinfo->table->serverid);
642 
643  /*
644  * Extract user-settable option values. Note that per-table settings of
645  * use_remote_estimate, fetch_size and async_capable override per-server
646  * settings of them, respectively.
647  */
648  fpinfo->use_remote_estimate = false;
651  fpinfo->shippable_extensions = NIL;
652  fpinfo->fetch_size = 100;
653  fpinfo->async_capable = false;
654 
655  apply_server_options(fpinfo);
656  apply_table_options(fpinfo);
657 
658  /*
659  * If the table or the server is configured to use remote estimates,
660  * identify which user to do remote access as during planning. This
661  * should match what ExecCheckPermissions() does. If we fail due to lack
662  * of permissions, the query would have failed at runtime anyway.
663  */
664  if (fpinfo->use_remote_estimate)
665  {
666  Oid userid;
667 
668  userid = OidIsValid(baserel->userid) ? baserel->userid : GetUserId();
669  fpinfo->user = GetUserMapping(userid, fpinfo->server->serverid);
670  }
671  else
672  fpinfo->user = NULL;
673 
674  /*
675  * Identify which baserestrictinfo clauses can be sent to the remote
676  * server and which can't.
677  */
678  classifyConditions(root, baserel, baserel->baserestrictinfo,
679  &fpinfo->remote_conds, &fpinfo->local_conds);
680 
681  /*
682  * Identify which attributes will need to be retrieved from the remote
683  * server. These include all attrs needed for joins or final output, plus
684  * all attrs used in the local_conds. (Note: if we end up using a
685  * parameterized scan, it's possible that some of the join clauses will be
686  * sent to the remote and thus we wouldn't really need to retrieve the
687  * columns used in them. Doesn't seem worth detecting that case though.)
688  */
689  fpinfo->attrs_used = NULL;
690  pull_varattnos((Node *) baserel->reltarget->exprs, baserel->relid,
691  &fpinfo->attrs_used);
692  foreach(lc, fpinfo->local_conds)
693  {
694  RestrictInfo *rinfo = lfirst_node(RestrictInfo, lc);
695 
696  pull_varattnos((Node *) rinfo->clause, baserel->relid,
697  &fpinfo->attrs_used);
698  }
699 
700  /*
701  * Compute the selectivity and cost of the local_conds, so we don't have
702  * to do it over again for each path. The best we can do for these
703  * conditions is to estimate selectivity on the basis of local statistics.
704  */
706  fpinfo->local_conds,
707  baserel->relid,
708  JOIN_INNER,
709  NULL);
710 
711  cost_qual_eval(&fpinfo->local_conds_cost, fpinfo->local_conds, root);
712 
713  /*
714  * Set # of retrieved rows and cached relation costs to some negative
715  * value, so that we can detect when they are set to some sensible values,
716  * during one (usually the first) of the calls to estimate_path_cost_size.
717  */
718  fpinfo->retrieved_rows = -1;
719  fpinfo->rel_startup_cost = -1;
720  fpinfo->rel_total_cost = -1;
721 
722  /*
723  * If the table or the server is configured to use remote estimates,
724  * connect to the foreign server and execute EXPLAIN to estimate the
725  * number of rows selected by the restriction clauses, as well as the
726  * average row width. Otherwise, estimate using whatever statistics we
727  * have locally, in a way similar to ordinary tables.
728  */
729  if (fpinfo->use_remote_estimate)
730  {
731  /*
732  * Get cost/size estimates with help of remote server. Save the
733  * values in fpinfo so we don't need to do it again to generate the
734  * basic foreign path.
735  */
736  estimate_path_cost_size(root, baserel, NIL, NIL, NULL,
737  &fpinfo->rows, &fpinfo->width,
738  &fpinfo->startup_cost, &fpinfo->total_cost);
739 
740  /* Report estimated baserel size to planner. */
741  baserel->rows = fpinfo->rows;
742  baserel->reltarget->width = fpinfo->width;
743  }
744  else
745  {
746  /*
747  * If the foreign table has never been ANALYZEd, it will have
748  * reltuples < 0, meaning "unknown". We can't do much if we're not
749  * allowed to consult the remote server, but we can use a hack similar
750  * to plancat.c's treatment of empty relations: use a minimum size
751  * estimate of 10 pages, and divide by the column-datatype-based width
752  * estimate to get the corresponding number of tuples.
753  */
754  if (baserel->tuples < 0)
755  {
756  baserel->pages = 10;
757  baserel->tuples =
758  (10 * BLCKSZ) / (baserel->reltarget->width +
760  }
761 
762  /* Estimate baserel size as best we can with local statistics. */
763  set_baserel_size_estimates(root, baserel);
764 
765  /* Fill in basically-bogus cost estimates for use later. */
766  estimate_path_cost_size(root, baserel, NIL, NIL, NULL,
767  &fpinfo->rows, &fpinfo->width,
768  &fpinfo->startup_cost, &fpinfo->total_cost);
769  }
770 
771  /*
772  * fpinfo->relation_name gets the numeric rangetable index of the foreign
773  * table RTE. (If this query gets EXPLAIN'd, we'll convert that to a
774  * human-readable string at that time.)
775  */
776  fpinfo->relation_name = psprintf("%u", baserel->relid);
777 
778  /* No outer and inner relations. */
779  fpinfo->make_outerrel_subquery = false;
780  fpinfo->make_innerrel_subquery = false;
781  fpinfo->lower_subquery_rels = NULL;
782  fpinfo->hidden_subquery_rels = NULL;
783  /* Set the relation index. */
784  fpinfo->relation_index = baserel->relid;
785 }
786 
787 /*
788  * get_useful_ecs_for_relation
789  * Determine which EquivalenceClasses might be involved in useful
790  * orderings of this relation.
791  *
792  * This function is in some respects a mirror image of the core function
793  * pathkeys_useful_for_merging: for a regular table, we know what indexes
794  * we have and want to test whether any of them are useful. For a foreign
795  * table, we don't know what indexes are present on the remote side but
796  * want to speculate about which ones we'd like to use if they existed.
797  *
798  * This function returns a list of potentially-useful equivalence classes,
799  * but it does not guarantee that an EquivalenceMember exists which contains
800  * Vars only from the given relation. For example, given ft1 JOIN t1 ON
801  * ft1.x + t1.x = 0, this function will say that the equivalence class
802  * containing ft1.x + t1.x is potentially useful. Supposing ft1 is remote and
803  * t1 is local (or on a different server), it will turn out that no useful
804  * ORDER BY clause can be generated. It's not our job to figure that out
805  * here; we're only interested in identifying relevant ECs.
806  */
807 static List *
809 {
810  List *useful_eclass_list = NIL;
811  ListCell *lc;
812  Relids relids;
813 
814  /*
815  * First, consider whether any active EC is potentially useful for a merge
816  * join against this relation.
817  */
818  if (rel->has_eclass_joins)
819  {
820  foreach(lc, root->eq_classes)
821  {
822  EquivalenceClass *cur_ec = (EquivalenceClass *) lfirst(lc);
823 
824  if (eclass_useful_for_merging(root, cur_ec, rel))
825  useful_eclass_list = lappend(useful_eclass_list, cur_ec);
826  }
827  }
828 
829  /*
830  * Next, consider whether there are any non-EC derivable join clauses that
831  * are merge-joinable. If the joininfo list is empty, we can exit
832  * quickly.
833  */
834  if (rel->joininfo == NIL)
835  return useful_eclass_list;
836 
837  /* If this is a child rel, we must use the topmost parent rel to search. */
838  if (IS_OTHER_REL(rel))
839  {
841  relids = rel->top_parent_relids;
842  }
843  else
844  relids = rel->relids;
845 
846  /* Check each join clause in turn. */
847  foreach(lc, rel->joininfo)
848  {
849  RestrictInfo *restrictinfo = (RestrictInfo *) lfirst(lc);
850 
851  /* Consider only mergejoinable clauses */
852  if (restrictinfo->mergeopfamilies == NIL)
853  continue;
854 
855  /* Make sure we've got canonical ECs. */
856  update_mergeclause_eclasses(root, restrictinfo);
857 
858  /*
859  * restrictinfo->mergeopfamilies != NIL is sufficient to guarantee
860  * that left_ec and right_ec will be initialized, per comments in
861  * distribute_qual_to_rels.
862  *
863  * We want to identify which side of this merge-joinable clause
864  * contains columns from the relation produced by this RelOptInfo. We
865  * test for overlap, not containment, because there could be extra
866  * relations on either side. For example, suppose we've got something
867  * like ((A JOIN B ON A.x = B.x) JOIN C ON A.y = C.y) LEFT JOIN D ON
868  * A.y = D.y. The input rel might be the joinrel between A and B, and
869  * we'll consider the join clause A.y = D.y. relids contains a
870  * relation not involved in the join class (B) and the equivalence
871  * class for the left-hand side of the clause contains a relation not
872  * involved in the input rel (C). Despite the fact that we have only
873  * overlap and not containment in either direction, A.y is potentially
874  * useful as a sort column.
875  *
876  * Note that it's even possible that relids overlaps neither side of
877  * the join clause. For example, consider A LEFT JOIN B ON A.x = B.x
878  * AND A.x = 1. The clause A.x = 1 will appear in B's joininfo list,
879  * but overlaps neither side of B. In that case, we just skip this
880  * join clause, since it doesn't suggest a useful sort order for this
881  * relation.
882  */
883  if (bms_overlap(relids, restrictinfo->right_ec->ec_relids))
884  useful_eclass_list = list_append_unique_ptr(useful_eclass_list,
885  restrictinfo->right_ec);
886  else if (bms_overlap(relids, restrictinfo->left_ec->ec_relids))
887  useful_eclass_list = list_append_unique_ptr(useful_eclass_list,
888  restrictinfo->left_ec);
889  }
890 
891  return useful_eclass_list;
892 }
893 
894 /*
895  * get_useful_pathkeys_for_relation
896  * Determine which orderings of a relation might be useful.
897  *
898  * Getting data in sorted order can be useful either because the requested
899  * order matches the final output ordering for the overall query we're
900  * planning, or because it enables an efficient merge join. Here, we try
901  * to figure out which pathkeys to consider.
902  */
903 static List *
905 {
906  List *useful_pathkeys_list = NIL;
907  List *useful_eclass_list;
908  PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) rel->fdw_private;
909  EquivalenceClass *query_ec = NULL;
910  ListCell *lc;
911 
912  /*
913  * Pushing the query_pathkeys to the remote server is always worth
914  * considering, because it might let us avoid a local sort.
915  */
916  fpinfo->qp_is_pushdown_safe = false;
917  if (root->query_pathkeys)
918  {
919  bool query_pathkeys_ok = true;
920 
921  foreach(lc, root->query_pathkeys)
922  {
923  PathKey *pathkey = (PathKey *) lfirst(lc);
924 
925  /*
926  * The planner and executor don't have any clever strategy for
927  * taking data sorted by a prefix of the query's pathkeys and
928  * getting it to be sorted by all of those pathkeys. We'll just
929  * end up resorting the entire data set. So, unless we can push
930  * down all of the query pathkeys, forget it.
931  */
932  if (!is_foreign_pathkey(root, rel, pathkey))
933  {
934  query_pathkeys_ok = false;
935  break;
936  }
937  }
938 
939  if (query_pathkeys_ok)
940  {
941  useful_pathkeys_list = list_make1(list_copy(root->query_pathkeys));
942  fpinfo->qp_is_pushdown_safe = true;
943  }
944  }
945 
946  /*
947  * Even if we're not using remote estimates, having the remote side do the
948  * sort generally won't be any worse than doing it locally, and it might
949  * be much better if the remote side can generate data in the right order
950  * without needing a sort at all. However, what we're going to do next is
951  * try to generate pathkeys that seem promising for possible merge joins,
952  * and that's more speculative. A wrong choice might hurt quite a bit, so
953  * bail out if we can't use remote estimates.
954  */
955  if (!fpinfo->use_remote_estimate)
956  return useful_pathkeys_list;
957 
958  /* Get the list of interesting EquivalenceClasses. */
959  useful_eclass_list = get_useful_ecs_for_relation(root, rel);
960 
961  /* Extract unique EC for query, if any, so we don't consider it again. */
962  if (list_length(root->query_pathkeys) == 1)
963  {
964  PathKey *query_pathkey = linitial(root->query_pathkeys);
965 
966  query_ec = query_pathkey->pk_eclass;
967  }
968 
969  /*
970  * As a heuristic, the only pathkeys we consider here are those of length
971  * one. It's surely possible to consider more, but since each one we
972  * choose to consider will generate a round-trip to the remote side, we
973  * need to be a bit cautious here. It would sure be nice to have a local
974  * cache of information about remote index definitions...
975  */
976  foreach(lc, useful_eclass_list)
977  {
978  EquivalenceClass *cur_ec = lfirst(lc);
979  PathKey *pathkey;
980 
981  /* If redundant with what we did above, skip it. */
982  if (cur_ec == query_ec)
983  continue;
984 
985  /* Can't push down the sort if the EC's opfamily is not shippable. */
987  OperatorFamilyRelationId, fpinfo))
988  continue;
989 
990  /* If no pushable expression for this rel, skip it. */
991  if (find_em_for_rel(root, cur_ec, rel) == NULL)
992  continue;
993 
994  /* Looks like we can generate a pathkey, so let's do it. */
995  pathkey = make_canonical_pathkey(root, cur_ec,
996  linitial_oid(cur_ec->ec_opfamilies),
998  false);
999  useful_pathkeys_list = lappend(useful_pathkeys_list,
1000  list_make1(pathkey));
1001  }
1002 
1003  return useful_pathkeys_list;
1004 }
1005 
1006 /*
1007  * postgresGetForeignPaths
1008  * Create possible scan paths for a scan on the foreign table
1009  */
1010 static void
1012  RelOptInfo *baserel,
1013  Oid foreigntableid)
1014 {
1015  PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) baserel->fdw_private;
1016  ForeignPath *path;
1017  List *ppi_list;
1018  ListCell *lc;
1019 
1020  /*
1021  * Create simplest ForeignScan path node and add it to baserel. This path
1022  * corresponds to SeqScan path of regular tables (though depending on what
1023  * baserestrict conditions we were able to send to remote, there might
1024  * actually be an indexscan happening there). We already did all the work
1025  * to estimate cost and size of this path.
1026  *
1027  * Although this path uses no join clauses, it could still have required
1028  * parameterization due to LATERAL refs in its tlist.
1029  */
1030  path = create_foreignscan_path(root, baserel,
1031  NULL, /* default pathtarget */
1032  fpinfo->rows,
1033  fpinfo->startup_cost,
1034  fpinfo->total_cost,
1035  NIL, /* no pathkeys */
1036  baserel->lateral_relids,
1037  NULL, /* no extra plan */
1038  NIL, /* no fdw_restrictinfo list */
1039  NIL); /* no fdw_private list */
1040  add_path(baserel, (Path *) path);
1041 
1042  /* Add paths with pathkeys */
1043  add_paths_with_pathkeys_for_rel(root, baserel, NULL, NIL);
1044 
1045  /*
1046  * If we're not using remote estimates, stop here. We have no way to
1047  * estimate whether any join clauses would be worth sending across, so
1048  * don't bother building parameterized paths.
1049  */
1050  if (!fpinfo->use_remote_estimate)
1051  return;
1052 
1053  /*
1054  * Thumb through all join clauses for the rel to identify which outer
1055  * relations could supply one or more safe-to-send-to-remote join clauses.
1056  * We'll build a parameterized path for each such outer relation.
1057  *
1058  * It's convenient to manage this by representing each candidate outer
1059  * relation by the ParamPathInfo node for it. We can then use the
1060  * ppi_clauses list in the ParamPathInfo node directly as a list of the
1061  * interesting join clauses for that rel. This takes care of the
1062  * possibility that there are multiple safe join clauses for such a rel,
1063  * and also ensures that we account for unsafe join clauses that we'll
1064  * still have to enforce locally (since the parameterized-path machinery
1065  * insists that we handle all movable clauses).
1066  */
1067  ppi_list = NIL;
1068  foreach(lc, baserel->joininfo)
1069  {
1070  RestrictInfo *rinfo = (RestrictInfo *) lfirst(lc);
1071  Relids required_outer;
1072  ParamPathInfo *param_info;
1073 
1074  /* Check if clause can be moved to this rel */
1075  if (!join_clause_is_movable_to(rinfo, baserel))
1076  continue;
1077 
1078  /* See if it is safe to send to remote */
1079  if (!is_foreign_expr(root, baserel, rinfo->clause))
1080  continue;
1081 
1082  /* Calculate required outer rels for the resulting path */
1083  required_outer = bms_union(rinfo->clause_relids,
1084  baserel->lateral_relids);
1085  /* We do not want the foreign rel itself listed in required_outer */
1086  required_outer = bms_del_member(required_outer, baserel->relid);
1087 
1088  /*
1089  * required_outer probably can't be empty here, but if it were, we
1090  * couldn't make a parameterized path.
1091  */
1092  if (bms_is_empty(required_outer))
1093  continue;
1094 
1095  /* Get the ParamPathInfo */
1096  param_info = get_baserel_parampathinfo(root, baserel,
1097  required_outer);
1098  Assert(param_info != NULL);
1099 
1100  /*
1101  * Add it to list unless we already have it. Testing pointer equality
1102  * is OK since get_baserel_parampathinfo won't make duplicates.
1103  */
1104  ppi_list = list_append_unique_ptr(ppi_list, param_info);
1105  }
1106 
1107  /*
1108  * The above scan examined only "generic" join clauses, not those that
1109  * were absorbed into EquivalenceClauses. See if we can make anything out
1110  * of EquivalenceClauses.
1111  */
1112  if (baserel->has_eclass_joins)
1113  {
1114  /*
1115  * We repeatedly scan the eclass list looking for column references
1116  * (or expressions) belonging to the foreign rel. Each time we find
1117  * one, we generate a list of equivalence joinclauses for it, and then
1118  * see if any are safe to send to the remote. Repeat till there are
1119  * no more candidate EC members.
1120  */
1122 
1123  arg.already_used = NIL;
1124  for (;;)
1125  {
1126  List *clauses;
1127 
1128  /* Make clauses, skipping any that join to lateral_referencers */
1129  arg.current = NULL;
1131  baserel,
1133  (void *) &arg,
1134  baserel->lateral_referencers);
1135 
1136  /* Done if there are no more expressions in the foreign rel */
1137  if (arg.current == NULL)
1138  {
1139  Assert(clauses == NIL);
1140  break;
1141  }
1142 
1143  /* Scan the extracted join clauses */
1144  foreach(lc, clauses)
1145  {
1146  RestrictInfo *rinfo = (RestrictInfo *) lfirst(lc);
1147  Relids required_outer;
1148  ParamPathInfo *param_info;
1149 
1150  /* Check if clause can be moved to this rel */
1151  if (!join_clause_is_movable_to(rinfo, baserel))
1152  continue;
1153 
1154  /* See if it is safe to send to remote */
1155  if (!is_foreign_expr(root, baserel, rinfo->clause))
1156  continue;
1157 
1158  /* Calculate required outer rels for the resulting path */
1159  required_outer = bms_union(rinfo->clause_relids,
1160  baserel->lateral_relids);
1161  required_outer = bms_del_member(required_outer, baserel->relid);
1162  if (bms_is_empty(required_outer))
1163  continue;
1164 
1165  /* Get the ParamPathInfo */
1166  param_info = get_baserel_parampathinfo(root, baserel,
1167  required_outer);
1168  Assert(param_info != NULL);
1169 
1170  /* Add it to list unless we already have it */
1171  ppi_list = list_append_unique_ptr(ppi_list, param_info);
1172  }
1173 
1174  /* Try again, now ignoring the expression we found this time */
1175  arg.already_used = lappend(arg.already_used, arg.current);
1176  }
1177  }
1178 
1179  /*
1180  * Now build a path for each useful outer relation.
1181  */
1182  foreach(lc, ppi_list)
1183  {
1184  ParamPathInfo *param_info = (ParamPathInfo *) lfirst(lc);
1185  double rows;
1186  int width;
1187  Cost startup_cost;
1188  Cost total_cost;
1189 
1190  /* Get a cost estimate from the remote */
1191  estimate_path_cost_size(root, baserel,
1192  param_info->ppi_clauses, NIL, NULL,
1193  &rows, &width,
1194  &startup_cost, &total_cost);
1195 
1196  /*
1197  * ppi_rows currently won't get looked at by anything, but still we
1198  * may as well ensure that it matches our idea of the rowcount.
1199  */
1200  param_info->ppi_rows = rows;
1201 
1202  /* Make the path */
1203  path = create_foreignscan_path(root, baserel,
1204  NULL, /* default pathtarget */
1205  rows,
1206  startup_cost,
1207  total_cost,
1208  NIL, /* no pathkeys */
1209  param_info->ppi_req_outer,
1210  NULL,
1211  NIL, /* no fdw_restrictinfo list */
1212  NIL); /* no fdw_private list */
1213  add_path(baserel, (Path *) path);
1214  }
1215 }
1216 
1217 /*
1218  * postgresGetForeignPlan
1219  * Create ForeignScan plan node which implements selected best path
1220  */
1221 static ForeignScan *
1223  RelOptInfo *foreignrel,
1224  Oid foreigntableid,
1225  ForeignPath *best_path,
1226  List *tlist,
1227  List *scan_clauses,
1228  Plan *outer_plan)
1229 {
1230  PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) foreignrel->fdw_private;
1231  Index scan_relid;
1232  List *fdw_private;
1233  List *remote_exprs = NIL;
1234  List *local_exprs = NIL;
1235  List *params_list = NIL;
1236  List *fdw_scan_tlist = NIL;
1237  List *fdw_recheck_quals = NIL;
1238  List *retrieved_attrs;
1239  StringInfoData sql;
1240  bool has_final_sort = false;
1241  bool has_limit = false;
1242  ListCell *lc;
1243 
1244  /*
1245  * Get FDW private data created by postgresGetForeignUpperPaths(), if any.
1246  */
1247  if (best_path->fdw_private)
1248  {
1249  has_final_sort = boolVal(list_nth(best_path->fdw_private,
1251  has_limit = boolVal(list_nth(best_path->fdw_private,
1253  }
1254 
1255  if (IS_SIMPLE_REL(foreignrel))
1256  {
1257  /*
1258  * For base relations, set scan_relid as the relid of the relation.
1259  */
1260  scan_relid = foreignrel->relid;
1261 
1262  /*
1263  * In a base-relation scan, we must apply the given scan_clauses.
1264  *
1265  * Separate the scan_clauses into those that can be executed remotely
1266  * and those that can't. baserestrictinfo clauses that were
1267  * previously determined to be safe or unsafe by classifyConditions
1268  * are found in fpinfo->remote_conds and fpinfo->local_conds. Anything
1269  * else in the scan_clauses list will be a join clause, which we have
1270  * to check for remote-safety.
1271  *
1272  * Note: the join clauses we see here should be the exact same ones
1273  * previously examined by postgresGetForeignPaths. Possibly it'd be
1274  * worth passing forward the classification work done then, rather
1275  * than repeating it here.
1276  *
1277  * This code must match "extract_actual_clauses(scan_clauses, false)"
1278  * except for the additional decision about remote versus local
1279  * execution.
1280  */
1281  foreach(lc, scan_clauses)
1282  {
1283  RestrictInfo *rinfo = lfirst_node(RestrictInfo, lc);
1284 
1285  /* Ignore any pseudoconstants, they're dealt with elsewhere */
1286  if (rinfo->pseudoconstant)
1287  continue;
1288 
1289  if (list_member_ptr(fpinfo->remote_conds, rinfo))
1290  remote_exprs = lappend(remote_exprs, rinfo->clause);
1291  else if (list_member_ptr(fpinfo->local_conds, rinfo))
1292  local_exprs = lappend(local_exprs, rinfo->clause);
1293  else if (is_foreign_expr(root, foreignrel, rinfo->clause))
1294  remote_exprs = lappend(remote_exprs, rinfo->clause);
1295  else
1296  local_exprs = lappend(local_exprs, rinfo->clause);
1297  }
1298 
1299  /*
1300  * For a base-relation scan, we have to support EPQ recheck, which
1301  * should recheck all the remote quals.
1302  */
1303  fdw_recheck_quals = remote_exprs;
1304  }
1305  else
1306  {
1307  /*
1308  * Join relation or upper relation - set scan_relid to 0.
1309  */
1310  scan_relid = 0;
1311 
1312  /*
1313  * For a join rel, baserestrictinfo is NIL and we are not considering
1314  * parameterization right now, so there should be no scan_clauses for
1315  * a joinrel or an upper rel either.
1316  */
1317  Assert(!scan_clauses);
1318 
1319  /*
1320  * Instead we get the conditions to apply from the fdw_private
1321  * structure.
1322  */
1323  remote_exprs = extract_actual_clauses(fpinfo->remote_conds, false);
1324  local_exprs = extract_actual_clauses(fpinfo->local_conds, false);
1325 
1326  /*
1327  * We leave fdw_recheck_quals empty in this case, since we never need
1328  * to apply EPQ recheck clauses. In the case of a joinrel, EPQ
1329  * recheck is handled elsewhere --- see postgresGetForeignJoinPaths().
1330  * If we're planning an upperrel (ie, remote grouping or aggregation)
1331  * then there's no EPQ to do because SELECT FOR UPDATE wouldn't be
1332  * allowed, and indeed we *can't* put the remote clauses into
1333  * fdw_recheck_quals because the unaggregated Vars won't be available
1334  * locally.
1335  */
1336 
1337  /* Build the list of columns to be fetched from the foreign server. */
1338  fdw_scan_tlist = build_tlist_to_deparse(foreignrel);
1339 
1340  /*
1341  * Ensure that the outer plan produces a tuple whose descriptor
1342  * matches our scan tuple slot. Also, remove the local conditions
1343  * from outer plan's quals, lest they be evaluated twice, once by the
1344  * local plan and once by the scan.
1345  */
1346  if (outer_plan)
1347  {
1348  /*
1349  * Right now, we only consider grouping and aggregation beyond
1350  * joins. Queries involving aggregates or grouping do not require
1351  * EPQ mechanism, hence should not have an outer plan here.
1352  */
1353  Assert(!IS_UPPER_REL(foreignrel));
1354 
1355  /*
1356  * First, update the plan's qual list if possible. In some cases
1357  * the quals might be enforced below the topmost plan level, in
1358  * which case we'll fail to remove them; it's not worth working
1359  * harder than this.
1360  */
1361  foreach(lc, local_exprs)
1362  {
1363  Node *qual = lfirst(lc);
1364 
1365  outer_plan->qual = list_delete(outer_plan->qual, qual);
1366 
1367  /*
1368  * For an inner join the local conditions of foreign scan plan
1369  * can be part of the joinquals as well. (They might also be
1370  * in the mergequals or hashquals, but we can't touch those
1371  * without breaking the plan.)
1372  */
1373  if (IsA(outer_plan, NestLoop) ||
1374  IsA(outer_plan, MergeJoin) ||
1375  IsA(outer_plan, HashJoin))
1376  {
1377  Join *join_plan = (Join *) outer_plan;
1378 
1379  if (join_plan->jointype == JOIN_INNER)
1380  join_plan->joinqual = list_delete(join_plan->joinqual,
1381  qual);
1382  }
1383  }
1384 
1385  /*
1386  * Now fix the subplan's tlist --- this might result in inserting
1387  * a Result node atop the plan tree.
1388  */
1389  outer_plan = change_plan_targetlist(outer_plan, fdw_scan_tlist,
1390  best_path->path.parallel_safe);
1391  }
1392  }
1393 
1394  /*
1395  * Build the query string to be sent for execution, and identify
1396  * expressions to be sent as parameters.
1397  */
1398  initStringInfo(&sql);
1399  deparseSelectStmtForRel(&sql, root, foreignrel, fdw_scan_tlist,
1400  remote_exprs, best_path->path.pathkeys,
1401  has_final_sort, has_limit, false,
1402  &retrieved_attrs, &params_list);
1403 
1404  /* Remember remote_exprs for possible use by postgresPlanDirectModify */
1405  fpinfo->final_remote_exprs = remote_exprs;
1406 
1407  /*
1408  * Build the fdw_private list that will be available to the executor.
1409  * Items in the list must match order in enum FdwScanPrivateIndex.
1410  */
1411  fdw_private = list_make3(makeString(sql.data),
1412  retrieved_attrs,
1413  makeInteger(fpinfo->fetch_size));
1414  if (IS_JOIN_REL(foreignrel) || IS_UPPER_REL(foreignrel))
1415  fdw_private = lappend(fdw_private,
1416  makeString(fpinfo->relation_name));
1417 
1418  /*
1419  * Create the ForeignScan node for the given relation.
1420  *
1421  * Note that the remote parameter expressions are stored in the fdw_exprs
1422  * field of the finished plan node; we can't keep them in private state
1423  * because then they wouldn't be subject to later planner processing.
1424  */
1425  return make_foreignscan(tlist,
1426  local_exprs,
1427  scan_relid,
1428  params_list,
1429  fdw_private,
1430  fdw_scan_tlist,
1431  fdw_recheck_quals,
1432  outer_plan);
1433 }
1434 
1435 /*
1436  * Construct a tuple descriptor for the scan tuples handled by a foreign join.
1437  */
1438 static TupleDesc
1440 {
1441  ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan;
1442  EState *estate = node->ss.ps.state;
1443  TupleDesc tupdesc;
1444 
1445  /*
1446  * The core code has already set up a scan tuple slot based on
1447  * fsplan->fdw_scan_tlist, and this slot's tupdesc is mostly good enough,
1448  * but there's one case where it isn't. If we have any whole-row row
1449  * identifier Vars, they may have vartype RECORD, and we need to replace
1450  * that with the associated table's actual composite type. This ensures
1451  * that when we read those ROW() expression values from the remote server,
1452  * we can convert them to a composite type the local server knows.
1453  */
1455  for (int i = 0; i < tupdesc->natts; i++)
1456  {
1457  Form_pg_attribute att = TupleDescAttr(tupdesc, i);
1458  Var *var;
1459  RangeTblEntry *rte;
1460  Oid reltype;
1461 
1462  /* Nothing to do if it's not a generic RECORD attribute */
1463  if (att->atttypid != RECORDOID || att->atttypmod >= 0)
1464  continue;
1465 
1466  /*
1467  * If we can't identify the referenced table, do nothing. This'll
1468  * likely lead to failure later, but perhaps we can muddle through.
1469  */
1470  var = (Var *) list_nth_node(TargetEntry, fsplan->fdw_scan_tlist,
1471  i)->expr;
1472  if (!IsA(var, Var) || var->varattno != 0)
1473  continue;
1474  rte = list_nth(estate->es_range_table, var->varno - 1);
1475  if (rte->rtekind != RTE_RELATION)
1476  continue;
1477  reltype = get_rel_type_id(rte->relid);
1478  if (!OidIsValid(reltype))
1479  continue;
1480  att->atttypid = reltype;
1481  /* shouldn't need to change anything else */
1482  }
1483  return tupdesc;
1484 }
1485 
1486 /*
1487  * postgresBeginForeignScan
1488  * Initiate an executor scan of a foreign PostgreSQL table.
1489  */
1490 static void
1492 {
1493  ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan;
1494  EState *estate = node->ss.ps.state;
1495  PgFdwScanState *fsstate;
1496  RangeTblEntry *rte;
1497  Oid userid;
1498  ForeignTable *table;
1499  UserMapping *user;
1500  int rtindex;
1501  int numParams;
1502 
1503  /*
1504  * Do nothing in EXPLAIN (no ANALYZE) case. node->fdw_state stays NULL.
1505  */
1506  if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
1507  return;
1508 
1509  /*
1510  * We'll save private state in node->fdw_state.
1511  */
1512  fsstate = (PgFdwScanState *) palloc0(sizeof(PgFdwScanState));
1513  node->fdw_state = (void *) fsstate;
1514 
1515  /*
1516  * Identify which user to do the remote access as. This should match what
1517  * ExecCheckPermissions() does.
1518  */
1519  userid = OidIsValid(fsplan->checkAsUser) ? fsplan->checkAsUser : GetUserId();
1520  if (fsplan->scan.scanrelid > 0)
1521  rtindex = fsplan->scan.scanrelid;
1522  else
1523  rtindex = bms_next_member(fsplan->fs_base_relids, -1);
1524  rte = exec_rt_fetch(rtindex, estate);
1525 
1526  /* Get info about foreign table. */
1527  table = GetForeignTable(rte->relid);
1528  user = GetUserMapping(userid, table->serverid);
1529 
1530  /*
1531  * Get connection to the foreign server. Connection manager will
1532  * establish new connection if necessary.
1533  */
1534  fsstate->conn = GetConnection(user, false, &fsstate->conn_state);
1535 
1536  /* Assign a unique ID for my cursor */
1537  fsstate->cursor_number = GetCursorNumber(fsstate->conn);
1538  fsstate->cursor_exists = false;
1539 
1540  /* Get private info created by planner functions. */
1541  fsstate->query = strVal(list_nth(fsplan->fdw_private,
1543  fsstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private,
1545  fsstate->fetch_size = intVal(list_nth(fsplan->fdw_private,
1547 
1548  /* Create contexts for batches of tuples and per-tuple temp workspace. */
1549  fsstate->batch_cxt = AllocSetContextCreate(estate->es_query_cxt,
1550  "postgres_fdw tuple data",
1552  fsstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
1553  "postgres_fdw temporary data",
1555 
1556  /*
1557  * Get info we'll need for converting data fetched from the foreign server
1558  * into local representation and error reporting during that process.
1559  */
1560  if (fsplan->scan.scanrelid > 0)
1561  {
1562  fsstate->rel = node->ss.ss_currentRelation;
1563  fsstate->tupdesc = RelationGetDescr(fsstate->rel);
1564  }
1565  else
1566  {
1567  fsstate->rel = NULL;
1568  fsstate->tupdesc = get_tupdesc_for_join_scan_tuples(node);
1569  }
1570 
1571  fsstate->attinmeta = TupleDescGetAttInMetadata(fsstate->tupdesc);
1572 
1573  /*
1574  * Prepare for processing of parameters used in remote query, if any.
1575  */
1576  numParams = list_length(fsplan->fdw_exprs);
1577  fsstate->numParams = numParams;
1578  if (numParams > 0)
1580  fsplan->fdw_exprs,
1581  numParams,
1582  &fsstate->param_flinfo,
1583  &fsstate->param_exprs,
1584  &fsstate->param_values);
1585 
1586  /* Set the async-capable flag */
1587  fsstate->async_capable = node->ss.ps.async_capable;
1588 }
1589 
1590 /*
1591  * postgresIterateForeignScan
1592  * Retrieve next row from the result set, or clear tuple slot to indicate
1593  * EOF.
1594  */
1595 static TupleTableSlot *
1597 {
1598  PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
1599  TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
1600 
1601  /*
1602  * In sync mode, if this is the first call after Begin or ReScan, we need
1603  * to create the cursor on the remote side. In async mode, we would have
1604  * already created the cursor before we get here, even if this is the
1605  * first call after Begin or ReScan.
1606  */
1607  if (!fsstate->cursor_exists)
1608  create_cursor(node);
1609 
1610  /*
1611  * Get some more tuples, if we've run out.
1612  */
1613  if (fsstate->next_tuple >= fsstate->num_tuples)
1614  {
1615  /* In async mode, just clear tuple slot. */
1616  if (fsstate->async_capable)
1617  return ExecClearTuple(slot);
1618  /* No point in another fetch if we already detected EOF, though. */
1619  if (!fsstate->eof_reached)
1620  fetch_more_data(node);
1621  /* If we didn't get any tuples, must be end of data. */
1622  if (fsstate->next_tuple >= fsstate->num_tuples)
1623  return ExecClearTuple(slot);
1624  }
1625 
1626  /*
1627  * Return the next tuple.
1628  */
1629  ExecStoreHeapTuple(fsstate->tuples[fsstate->next_tuple++],
1630  slot,
1631  false);
1632 
1633  return slot;
1634 }
1635 
1636 /*
1637  * postgresReScanForeignScan
1638  * Restart the scan.
1639  */
1640 static void
1642 {
1643  PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
1644  char sql[64];
1645  PGresult *res;
1646 
1647  /* If we haven't created the cursor yet, nothing to do. */
1648  if (!fsstate->cursor_exists)
1649  return;
1650 
1651  /*
1652  * If the node is async-capable, and an asynchronous fetch for it has
1653  * begun, the asynchronous fetch might not have yet completed. Check if
1654  * the node is async-capable, and an asynchronous fetch for it is still in
1655  * progress; if so, complete the asynchronous fetch before restarting the
1656  * scan.
1657  */
1658  if (fsstate->async_capable &&
1659  fsstate->conn_state->pendingAreq &&
1660  fsstate->conn_state->pendingAreq->requestee == (PlanState *) node)
1661  fetch_more_data(node);
1662 
1663  /*
1664  * If any internal parameters affecting this node have changed, we'd
1665  * better destroy and recreate the cursor. Otherwise, rewinding it should
1666  * be good enough. If we've only fetched zero or one batch, we needn't
1667  * even rewind the cursor, just rescan what we have.
1668  */
1669  if (node->ss.ps.chgParam != NULL)
1670  {
1671  fsstate->cursor_exists = false;
1672  snprintf(sql, sizeof(sql), "CLOSE c%u",
1673  fsstate->cursor_number);
1674  }
1675  else if (fsstate->fetch_ct_2 > 1)
1676  {
1677  snprintf(sql, sizeof(sql), "MOVE BACKWARD ALL IN c%u",
1678  fsstate->cursor_number);
1679  }
1680  else
1681  {
1682  /* Easy: just rescan what we already have in memory, if anything */
1683  fsstate->next_tuple = 0;
1684  return;
1685  }
1686 
1687  /*
1688  * We don't use a PG_TRY block here, so be careful not to throw error
1689  * without releasing the PGresult.
1690  */
1691  res = pgfdw_exec_query(fsstate->conn, sql, fsstate->conn_state);
1693  pgfdw_report_error(ERROR, res, fsstate->conn, true, sql);
1694  PQclear(res);
1695 
1696  /* Now force a fresh FETCH. */
1697  fsstate->tuples = NULL;
1698  fsstate->num_tuples = 0;
1699  fsstate->next_tuple = 0;
1700  fsstate->fetch_ct_2 = 0;
1701  fsstate->eof_reached = false;
1702 }
1703 
1704 /*
1705  * postgresEndForeignScan
1706  * Finish scanning foreign table and dispose objects used for this scan
1707  */
1708 static void
1710 {
1711  PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
1712 
1713  /* if fsstate is NULL, we are in EXPLAIN; nothing to do */
1714  if (fsstate == NULL)
1715  return;
1716 
1717  /* Close the cursor if open, to prevent accumulation of cursors */
1718  if (fsstate->cursor_exists)
1719  close_cursor(fsstate->conn, fsstate->cursor_number,
1720  fsstate->conn_state);
1721 
1722  /* Release remote connection */
1723  ReleaseConnection(fsstate->conn);
1724  fsstate->conn = NULL;
1725 
1726  /* MemoryContexts will be deleted automatically. */
1727 }
1728 
1729 /*
1730  * postgresAddForeignUpdateTargets
1731  * Add resjunk column(s) needed for update/delete on a foreign table
1732  */
1733 static void
1735  Index rtindex,
1736  RangeTblEntry *target_rte,
1737  Relation target_relation)
1738 {
1739  Var *var;
1740 
1741  /*
1742  * In postgres_fdw, what we need is the ctid, same as for a regular table.
1743  */
1744 
1745  /* Make a Var representing the desired value */
1746  var = makeVar(rtindex,
1748  TIDOID,
1749  -1,
1750  InvalidOid,
1751  0);
1752 
1753  /* Register it as a row-identity column needed by this target rel */
1754  add_row_identity_var(root, var, rtindex, "ctid");
1755 }
1756 
1757 /*
1758  * postgresPlanForeignModify
1759  * Plan an insert/update/delete operation on a foreign table
1760  */
1761 static List *
1763  ModifyTable *plan,
1764  Index resultRelation,
1765  int subplan_index)
1766 {
1767  CmdType operation = plan->operation;
1768  RangeTblEntry *rte = planner_rt_fetch(resultRelation, root);
1769  Relation rel;
1770  StringInfoData sql;
1771  List *targetAttrs = NIL;
1772  List *withCheckOptionList = NIL;
1773  List *returningList = NIL;
1774  List *retrieved_attrs = NIL;
1775  bool doNothing = false;
1776  int values_end_len = -1;
1777 
1778  initStringInfo(&sql);
1779 
1780  /*
1781  * Core code already has some lock on each rel being planned, so we can
1782  * use NoLock here.
1783  */
1784  rel = table_open(rte->relid, NoLock);
1785 
1786  /*
1787  * In an INSERT, we transmit all columns that are defined in the foreign
1788  * table. In an UPDATE, if there are BEFORE ROW UPDATE triggers on the
1789  * foreign table, we transmit all columns like INSERT; else we transmit
1790  * only columns that were explicitly targets of the UPDATE, so as to avoid
1791  * unnecessary data transmission. (We can't do that for INSERT since we
1792  * would miss sending default values for columns not listed in the source
1793  * statement, and for UPDATE if there are BEFORE ROW UPDATE triggers since
1794  * those triggers might change values for non-target columns, in which
1795  * case we would miss sending changed values for those columns.)
1796  */
1797  if (operation == CMD_INSERT ||
1798  (operation == CMD_UPDATE &&
1799  rel->trigdesc &&
1801  {
1802  TupleDesc tupdesc = RelationGetDescr(rel);
1803  int attnum;
1804 
1805  for (attnum = 1; attnum <= tupdesc->natts; attnum++)
1806  {
1807  Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
1808 
1809  if (!attr->attisdropped)
1810  targetAttrs = lappend_int(targetAttrs, attnum);
1811  }
1812  }
1813  else if (operation == CMD_UPDATE)
1814  {
1815  int col;
1816  RelOptInfo *rel = find_base_rel(root, resultRelation);
1817  Bitmapset *allUpdatedCols = get_rel_all_updated_cols(root, rel);
1818 
1819  col = -1;
1820  while ((col = bms_next_member(allUpdatedCols, col)) >= 0)
1821  {
1822  /* bit numbers are offset by FirstLowInvalidHeapAttributeNumber */
1824 
1825  if (attno <= InvalidAttrNumber) /* shouldn't happen */
1826  elog(ERROR, "system-column update is not supported");
1827  targetAttrs = lappend_int(targetAttrs, attno);
1828  }
1829  }
1830 
1831  /*
1832  * Extract the relevant WITH CHECK OPTION list if any.
1833  */
1834  if (plan->withCheckOptionLists)
1835  withCheckOptionList = (List *) list_nth(plan->withCheckOptionLists,
1836  subplan_index);
1837 
1838  /*
1839  * Extract the relevant RETURNING list if any.
1840  */
1841  if (plan->returningLists)
1842  returningList = (List *) list_nth(plan->returningLists, subplan_index);
1843 
1844  /*
1845  * ON CONFLICT DO UPDATE and DO NOTHING case with inference specification
1846  * should have already been rejected in the optimizer, as presently there
1847  * is no way to recognize an arbiter index on a foreign table. Only DO
1848  * NOTHING is supported without an inference specification.
1849  */
1850  if (plan->onConflictAction == ONCONFLICT_NOTHING)
1851  doNothing = true;
1852  else if (plan->onConflictAction != ONCONFLICT_NONE)
1853  elog(ERROR, "unexpected ON CONFLICT specification: %d",
1854  (int) plan->onConflictAction);
1855 
1856  /*
1857  * Construct the SQL command string.
1858  */
1859  switch (operation)
1860  {
1861  case CMD_INSERT:
1862  deparseInsertSql(&sql, rte, resultRelation, rel,
1863  targetAttrs, doNothing,
1864  withCheckOptionList, returningList,
1865  &retrieved_attrs, &values_end_len);
1866  break;
1867  case CMD_UPDATE:
1868  deparseUpdateSql(&sql, rte, resultRelation, rel,
1869  targetAttrs,
1870  withCheckOptionList, returningList,
1871  &retrieved_attrs);
1872  break;
1873  case CMD_DELETE:
1874  deparseDeleteSql(&sql, rte, resultRelation, rel,
1875  returningList,
1876  &retrieved_attrs);
1877  break;
1878  default:
1879  elog(ERROR, "unexpected operation: %d", (int) operation);
1880  break;
1881  }
1882 
1883  table_close(rel, NoLock);
1884 
1885  /*
1886  * Build the fdw_private list that will be available to the executor.
1887  * Items in the list must match enum FdwModifyPrivateIndex, above.
1888  */
1889  return list_make5(makeString(sql.data),
1890  targetAttrs,
1891  makeInteger(values_end_len),
1892  makeBoolean((retrieved_attrs != NIL)),
1893  retrieved_attrs);
1894 }
1895 
1896 /*
1897  * postgresBeginForeignModify
1898  * Begin an insert/update/delete operation on a foreign table
1899  */
1900 static void
1902  ResultRelInfo *resultRelInfo,
1903  List *fdw_private,
1904  int subplan_index,
1905  int eflags)
1906 {
1907  PgFdwModifyState *fmstate;
1908  char *query;
1909  List *target_attrs;
1910  bool has_returning;
1911  int values_end_len;
1912  List *retrieved_attrs;
1913  RangeTblEntry *rte;
1914 
1915  /*
1916  * Do nothing in EXPLAIN (no ANALYZE) case. resultRelInfo->ri_FdwState
1917  * stays NULL.
1918  */
1919  if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
1920  return;
1921 
1922  /* Deconstruct fdw_private data. */
1923  query = strVal(list_nth(fdw_private,
1925  target_attrs = (List *) list_nth(fdw_private,
1927  values_end_len = intVal(list_nth(fdw_private,
1929  has_returning = boolVal(list_nth(fdw_private,
1931  retrieved_attrs = (List *) list_nth(fdw_private,
1933 
1934  /* Find RTE. */
1935  rte = exec_rt_fetch(resultRelInfo->ri_RangeTableIndex,
1936  mtstate->ps.state);
1937 
1938  /* Construct an execution state. */
1939  fmstate = create_foreign_modify(mtstate->ps.state,
1940  rte,
1941  resultRelInfo,
1942  mtstate->operation,
1943  outerPlanState(mtstate)->plan,
1944  query,
1945  target_attrs,
1946  values_end_len,
1947  has_returning,
1948  retrieved_attrs);
1949 
1950  resultRelInfo->ri_FdwState = fmstate;
1951 }
1952 
1953 /*
1954  * postgresExecForeignInsert
1955  * Insert one row into a foreign table
1956  */
1957 static TupleTableSlot *
1959  ResultRelInfo *resultRelInfo,
1960  TupleTableSlot *slot,
1961  TupleTableSlot *planSlot)
1962 {
1963  PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
1964  TupleTableSlot **rslot;
1965  int numSlots = 1;
1966 
1967  /*
1968  * If the fmstate has aux_fmstate set, use the aux_fmstate (see
1969  * postgresBeginForeignInsert())
1970  */
1971  if (fmstate->aux_fmstate)
1972  resultRelInfo->ri_FdwState = fmstate->aux_fmstate;
1973  rslot = execute_foreign_modify(estate, resultRelInfo, CMD_INSERT,
1974  &slot, &planSlot, &numSlots);
1975  /* Revert that change */
1976  if (fmstate->aux_fmstate)
1977  resultRelInfo->ri_FdwState = fmstate;
1978 
1979  return rslot ? *rslot : NULL;
1980 }
1981 
1982 /*
1983  * postgresExecForeignBatchInsert
1984  * Insert multiple rows into a foreign table
1985  */
1986 static TupleTableSlot **
1988  ResultRelInfo *resultRelInfo,
1989  TupleTableSlot **slots,
1990  TupleTableSlot **planSlots,
1991  int *numSlots)
1992 {
1993  PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
1994  TupleTableSlot **rslot;
1995 
1996  /*
1997  * If the fmstate has aux_fmstate set, use the aux_fmstate (see
1998  * postgresBeginForeignInsert())
1999  */
2000  if (fmstate->aux_fmstate)
2001  resultRelInfo->ri_FdwState = fmstate->aux_fmstate;
2002  rslot = execute_foreign_modify(estate, resultRelInfo, CMD_INSERT,
2003  slots, planSlots, numSlots);
2004  /* Revert that change */
2005  if (fmstate->aux_fmstate)
2006  resultRelInfo->ri_FdwState = fmstate;
2007 
2008  return rslot;
2009 }
2010 
2011 /*
2012  * postgresGetForeignModifyBatchSize
2013  * Determine the maximum number of tuples that can be inserted in bulk
2014  *
2015  * Returns the batch size specified for server or table. When batching is not
2016  * allowed (e.g. for tables with BEFORE/AFTER ROW triggers or with RETURNING
2017  * clause), returns 1.
2018  */
2019 static int
2021 {
2022  int batch_size;
2023  PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
2024 
2025  /* should be called only once */
2026  Assert(resultRelInfo->ri_BatchSize == 0);
2027 
2028  /*
2029  * Should never get called when the insert is being performed on a table
2030  * that is also among the target relations of an UPDATE operation, because
2031  * postgresBeginForeignInsert() currently rejects such insert attempts.
2032  */
2033  Assert(fmstate == NULL || fmstate->aux_fmstate == NULL);
2034 
2035  /*
2036  * In EXPLAIN without ANALYZE, ri_FdwState is NULL, so we have to lookup
2037  * the option directly in server/table options. Otherwise just use the
2038  * value we determined earlier.
2039  */
2040  if (fmstate)
2041  batch_size = fmstate->batch_size;
2042  else
2043  batch_size = get_batch_size_option(resultRelInfo->ri_RelationDesc);
2044 
2045  /*
2046  * Disable batching when we have to use RETURNING, there are any
2047  * BEFORE/AFTER ROW INSERT triggers on the foreign table, or there are any
2048  * WITH CHECK OPTION constraints from parent views.
2049  *
2050  * When there are any BEFORE ROW INSERT triggers on the table, we can't
2051  * support it, because such triggers might query the table we're inserting
2052  * into and act differently if the tuples that have already been processed
2053  * and prepared for insertion are not there.
2054  */
2055  if (resultRelInfo->ri_projectReturning != NULL ||
2056  resultRelInfo->ri_WithCheckOptions != NIL ||
2057  (resultRelInfo->ri_TrigDesc &&
2058  (resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
2059  resultRelInfo->ri_TrigDesc->trig_insert_after_row)))
2060  return 1;
2061 
2062  /*
2063  * If the foreign table has no columns, disable batching as the INSERT
2064  * syntax doesn't allow batching multiple empty rows into a zero-column
2065  * table in a single statement. This is needed for COPY FROM, in which
2066  * case fmstate must be non-NULL.
2067  */
2068  if (fmstate && list_length(fmstate->target_attrs) == 0)
2069  return 1;
2070 
2071  /*
2072  * Otherwise use the batch size specified for server/table. The number of
2073  * parameters in a batch is limited to 65535 (uint16), so make sure we
2074  * don't exceed this limit by using the maximum batch_size possible.
2075  */
2076  if (fmstate && fmstate->p_nums > 0)
2077  batch_size = Min(batch_size, PQ_QUERY_PARAM_MAX_LIMIT / fmstate->p_nums);
2078 
2079  return batch_size;
2080 }
2081 
2082 /*
2083  * postgresExecForeignUpdate
2084  * Update one row in a foreign table
2085  */
2086 static TupleTableSlot *
2088  ResultRelInfo *resultRelInfo,
2089  TupleTableSlot *slot,
2090  TupleTableSlot *planSlot)
2091 {
2092  TupleTableSlot **rslot;
2093  int numSlots = 1;
2094 
2095  rslot = execute_foreign_modify(estate, resultRelInfo, CMD_UPDATE,
2096  &slot, &planSlot, &numSlots);
2097 
2098  return rslot ? rslot[0] : NULL;
2099 }
2100 
2101 /*
2102  * postgresExecForeignDelete
2103  * Delete one row from a foreign table
2104  */
2105 static TupleTableSlot *
2107  ResultRelInfo *resultRelInfo,
2108  TupleTableSlot *slot,
2109  TupleTableSlot *planSlot)
2110 {
2111  TupleTableSlot **rslot;
2112  int numSlots = 1;
2113 
2114  rslot = execute_foreign_modify(estate, resultRelInfo, CMD_DELETE,
2115  &slot, &planSlot, &numSlots);
2116 
2117  return rslot ? rslot[0] : NULL;
2118 }
2119 
2120 /*
2121  * postgresEndForeignModify
2122  * Finish an insert/update/delete operation on a foreign table
2123  */
2124 static void
2126  ResultRelInfo *resultRelInfo)
2127 {
2128  PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
2129 
2130  /* If fmstate is NULL, we are in EXPLAIN; nothing to do */
2131  if (fmstate == NULL)
2132  return;
2133 
2134  /* Destroy the execution state */
2135  finish_foreign_modify(fmstate);
2136 }
2137 
2138 /*
2139  * postgresBeginForeignInsert
2140  * Begin an insert operation on a foreign table
2141  */
2142 static void
2144  ResultRelInfo *resultRelInfo)
2145 {
2146  PgFdwModifyState *fmstate;
2147  ModifyTable *plan = castNode(ModifyTable, mtstate->ps.plan);
2148  EState *estate = mtstate->ps.state;
2149  Index resultRelation;
2150  Relation rel = resultRelInfo->ri_RelationDesc;
2151  RangeTblEntry *rte;
2152  TupleDesc tupdesc = RelationGetDescr(rel);
2153  int attnum;
2154  int values_end_len;
2155  StringInfoData sql;
2156  List *targetAttrs = NIL;
2157  List *retrieved_attrs = NIL;
2158  bool doNothing = false;
2159 
2160  /*
2161  * If the foreign table we are about to insert routed rows into is also an
2162  * UPDATE subplan result rel that will be updated later, proceeding with
2163  * the INSERT will result in the later UPDATE incorrectly modifying those
2164  * routed rows, so prevent the INSERT --- it would be nice if we could
2165  * handle this case; but for now, throw an error for safety.
2166  */
2167  if (plan && plan->operation == CMD_UPDATE &&
2168  (resultRelInfo->ri_usesFdwDirectModify ||
2169  resultRelInfo->ri_FdwState))
2170  ereport(ERROR,
2171  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
2172  errmsg("cannot route tuples into foreign table to be updated \"%s\"",
2173  RelationGetRelationName(rel))));
2174 
2175  initStringInfo(&sql);
2176 
2177  /* We transmit all columns that are defined in the foreign table. */
2178  for (attnum = 1; attnum <= tupdesc->natts; attnum++)
2179  {
2180  Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
2181 
2182  if (!attr->attisdropped)
2183  targetAttrs = lappend_int(targetAttrs, attnum);
2184  }
2185 
2186  /* Check if we add the ON CONFLICT clause to the remote query. */
2187  if (plan)
2188  {
2189  OnConflictAction onConflictAction = plan->onConflictAction;
2190 
2191  /* We only support DO NOTHING without an inference specification. */
2192  if (onConflictAction == ONCONFLICT_NOTHING)
2193  doNothing = true;
2194  else if (onConflictAction != ONCONFLICT_NONE)
2195  elog(ERROR, "unexpected ON CONFLICT specification: %d",
2196  (int) onConflictAction);
2197  }
2198 
2199  /*
2200  * If the foreign table is a partition that doesn't have a corresponding
2201  * RTE entry, we need to create a new RTE describing the foreign table for
2202  * use by deparseInsertSql and create_foreign_modify() below, after first
2203  * copying the parent's RTE and modifying some fields to describe the
2204  * foreign partition to work on. However, if this is invoked by UPDATE,
2205  * the existing RTE may already correspond to this partition if it is one
2206  * of the UPDATE subplan target rels; in that case, we can just use the
2207  * existing RTE as-is.
2208  */
2209  if (resultRelInfo->ri_RangeTableIndex == 0)
2210  {
2211  ResultRelInfo *rootResultRelInfo = resultRelInfo->ri_RootResultRelInfo;
2212 
2213  rte = exec_rt_fetch(rootResultRelInfo->ri_RangeTableIndex, estate);
2214  rte = copyObject(rte);
2215  rte->relid = RelationGetRelid(rel);
2216  rte->relkind = RELKIND_FOREIGN_TABLE;
2217 
2218  /*
2219  * For UPDATE, we must use the RT index of the first subplan target
2220  * rel's RTE, because the core code would have built expressions for
2221  * the partition, such as RETURNING, using that RT index as varno of
2222  * Vars contained in those expressions.
2223  */
2224  if (plan && plan->operation == CMD_UPDATE &&
2225  rootResultRelInfo->ri_RangeTableIndex == plan->rootRelation)
2226  resultRelation = mtstate->resultRelInfo[0].ri_RangeTableIndex;
2227  else
2228  resultRelation = rootResultRelInfo->ri_RangeTableIndex;
2229  }
2230  else
2231  {
2232  resultRelation = resultRelInfo->ri_RangeTableIndex;
2233  rte = exec_rt_fetch(resultRelation, estate);
2234  }
2235 
2236  /* Construct the SQL command string. */
2237  deparseInsertSql(&sql, rte, resultRelation, rel, targetAttrs, doNothing,
2238  resultRelInfo->ri_WithCheckOptions,
2239  resultRelInfo->ri_returningList,
2240  &retrieved_attrs, &values_end_len);
2241 
2242  /* Construct an execution state. */
2243  fmstate = create_foreign_modify(mtstate->ps.state,
2244  rte,
2245  resultRelInfo,
2246  CMD_INSERT,
2247  NULL,
2248  sql.data,
2249  targetAttrs,
2250  values_end_len,
2251  retrieved_attrs != NIL,
2252  retrieved_attrs);
2253 
2254  /*
2255  * If the given resultRelInfo already has PgFdwModifyState set, it means
2256  * the foreign table is an UPDATE subplan result rel; in which case, store
2257  * the resulting state into the aux_fmstate of the PgFdwModifyState.
2258  */
2259  if (resultRelInfo->ri_FdwState)
2260  {
2261  Assert(plan && plan->operation == CMD_UPDATE);
2262  Assert(resultRelInfo->ri_usesFdwDirectModify == false);
2263  ((PgFdwModifyState *) resultRelInfo->ri_FdwState)->aux_fmstate = fmstate;
2264  }
2265  else
2266  resultRelInfo->ri_FdwState = fmstate;
2267 }
2268 
2269 /*
2270  * postgresEndForeignInsert
2271  * Finish an insert operation on a foreign table
2272  */
2273 static void
2275  ResultRelInfo *resultRelInfo)
2276 {
2277  PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
2278 
2279  Assert(fmstate != NULL);
2280 
2281  /*
2282  * If the fmstate has aux_fmstate set, get the aux_fmstate (see
2283  * postgresBeginForeignInsert())
2284  */
2285  if (fmstate->aux_fmstate)
2286  fmstate = fmstate->aux_fmstate;
2287 
2288  /* Destroy the execution state */
2289  finish_foreign_modify(fmstate);
2290 }
2291 
2292 /*
2293  * postgresIsForeignRelUpdatable
2294  * Determine whether a foreign table supports INSERT, UPDATE and/or
2295  * DELETE.
2296  */
2297 static int
2299 {
2300  bool updatable;
2301  ForeignTable *table;
2302  ForeignServer *server;
2303  ListCell *lc;
2304 
2305  /*
2306  * By default, all postgres_fdw foreign tables are assumed updatable. This
2307  * can be overridden by a per-server setting, which in turn can be
2308  * overridden by a per-table setting.
2309  */
2310  updatable = true;
2311 
2312  table = GetForeignTable(RelationGetRelid(rel));
2313  server = GetForeignServer(table->serverid);
2314 
2315  foreach(lc, server->options)
2316  {
2317  DefElem *def = (DefElem *) lfirst(lc);
2318 
2319  if (strcmp(def->defname, "updatable") == 0)
2320  updatable = defGetBoolean(def);
2321  }
2322  foreach(lc, table->options)
2323  {
2324  DefElem *def = (DefElem *) lfirst(lc);
2325 
2326  if (strcmp(def->defname, "updatable") == 0)
2327  updatable = defGetBoolean(def);
2328  }
2329 
2330  /*
2331  * Currently "updatable" means support for INSERT, UPDATE and DELETE.
2332  */
2333  return updatable ?
2334  (1 << CMD_INSERT) | (1 << CMD_UPDATE) | (1 << CMD_DELETE) : 0;
2335 }
2336 
2337 /*
2338  * postgresRecheckForeignScan
2339  * Execute a local join execution plan for a foreign join
2340  */
2341 static bool
2343 {
2344  Index scanrelid = ((Scan *) node->ss.ps.plan)->scanrelid;
2346  TupleTableSlot *result;
2347 
2348  /* For base foreign relations, it suffices to set fdw_recheck_quals */
2349  if (scanrelid > 0)
2350  return true;
2351 
2352  Assert(outerPlan != NULL);
2353 
2354  /* Execute a local join execution plan */
2355  result = ExecProcNode(outerPlan);
2356  if (TupIsNull(result))
2357  return false;
2358 
2359  /* Store result in the given slot */
2360  ExecCopySlot(slot, result);
2361 
2362  return true;
2363 }
2364 
2365 /*
2366  * find_modifytable_subplan
2367  * Helper routine for postgresPlanDirectModify to find the
2368  * ModifyTable subplan node that scans the specified RTI.
2369  *
2370  * Returns NULL if the subplan couldn't be identified. That's not a fatal
2371  * error condition, we just abandon trying to do the update directly.
2372  */
2373 static ForeignScan *
2375  ModifyTable *plan,
2376  Index rtindex,
2377  int subplan_index)
2378 {
2379  Plan *subplan = outerPlan(plan);
2380 
2381  /*
2382  * The cases we support are (1) the desired ForeignScan is the immediate
2383  * child of ModifyTable, or (2) it is the subplan_index'th child of an
2384  * Append node that is the immediate child of ModifyTable. There is no
2385  * point in looking further down, as that would mean that local joins are
2386  * involved, so we can't do the update directly.
2387  *
2388  * There could be a Result atop the Append too, acting to compute the
2389  * UPDATE targetlist values. We ignore that here; the tlist will be
2390  * checked by our caller.
2391  *
2392  * In principle we could examine all the children of the Append, but it's
2393  * currently unlikely that the core planner would generate such a plan
2394  * with the children out-of-order. Moreover, such a search risks costing
2395  * O(N^2) time when there are a lot of children.
2396  */
2397  if (IsA(subplan, Append))
2398  {
2399  Append *appendplan = (Append *) subplan;
2400 
2401  if (subplan_index < list_length(appendplan->appendplans))
2402  subplan = (Plan *) list_nth(appendplan->appendplans, subplan_index);
2403  }
2404  else if (IsA(subplan, Result) &&
2405  outerPlan(subplan) != NULL &&
2406  IsA(outerPlan(subplan), Append))
2407  {
2408  Append *appendplan = (Append *) outerPlan(subplan);
2409 
2410  if (subplan_index < list_length(appendplan->appendplans))
2411  subplan = (Plan *) list_nth(appendplan->appendplans, subplan_index);
2412  }
2413 
2414  /* Now, have we got a ForeignScan on the desired rel? */
2415  if (IsA(subplan, ForeignScan))
2416  {
2417  ForeignScan *fscan = (ForeignScan *) subplan;
2418 
2419  if (bms_is_member(rtindex, fscan->fs_base_relids))
2420  return fscan;
2421  }
2422 
2423  return NULL;
2424 }
2425 
2426 /*
2427  * postgresPlanDirectModify
2428  * Consider a direct foreign table modification
2429  *
2430  * Decide whether it is safe to modify a foreign table directly, and if so,
2431  * rewrite subplan accordingly.
2432  */
2433 static bool
2435  ModifyTable *plan,
2436  Index resultRelation,
2437  int subplan_index)
2438 {
2439  CmdType operation = plan->operation;
2440  RelOptInfo *foreignrel;
2441  RangeTblEntry *rte;
2442  PgFdwRelationInfo *fpinfo;
2443  Relation rel;
2444  StringInfoData sql;
2445  ForeignScan *fscan;
2446  List *processed_tlist = NIL;
2447  List *targetAttrs = NIL;
2448  List *remote_exprs;
2449  List *params_list = NIL;
2450  List *returningList = NIL;
2451  List *retrieved_attrs = NIL;
2452 
2453  /*
2454  * Decide whether it is safe to modify a foreign table directly.
2455  */
2456 
2457  /*
2458  * The table modification must be an UPDATE or DELETE.
2459  */
2460  if (operation != CMD_UPDATE && operation != CMD_DELETE)
2461  return false;
2462 
2463  /*
2464  * Try to locate the ForeignScan subplan that's scanning resultRelation.
2465  */
2466  fscan = find_modifytable_subplan(root, plan, resultRelation, subplan_index);
2467  if (!fscan)
2468  return false;
2469 
2470  /*
2471  * It's unsafe to modify a foreign table directly if there are any quals
2472  * that should be evaluated locally.
2473  */
2474  if (fscan->scan.plan.qual != NIL)
2475  return false;
2476 
2477  /* Safe to fetch data about the target foreign rel */
2478  if (fscan->scan.scanrelid == 0)
2479  {
2480  foreignrel = find_join_rel(root, fscan->fs_relids);
2481  /* We should have a rel for this foreign join. */
2482  Assert(foreignrel);
2483  }
2484  else
2485  foreignrel = root->simple_rel_array[resultRelation];
2486  rte = root->simple_rte_array[resultRelation];
2487  fpinfo = (PgFdwRelationInfo *) foreignrel->fdw_private;
2488 
2489  /*
2490  * It's unsafe to update a foreign table directly, if any expressions to
2491  * assign to the target columns are unsafe to evaluate remotely.
2492  */
2493  if (operation == CMD_UPDATE)
2494  {
2495  ListCell *lc,
2496  *lc2;
2497 
2498  /*
2499  * The expressions of concern are the first N columns of the processed
2500  * targetlist, where N is the length of the rel's update_colnos.
2501  */
2502  get_translated_update_targetlist(root, resultRelation,
2503  &processed_tlist, &targetAttrs);
2504  forboth(lc, processed_tlist, lc2, targetAttrs)
2505  {
2506  TargetEntry *tle = lfirst_node(TargetEntry, lc);
2507  AttrNumber attno = lfirst_int(lc2);
2508 
2509  /* update's new-value expressions shouldn't be resjunk */
2510  Assert(!tle->resjunk);
2511 
2512  if (attno <= InvalidAttrNumber) /* shouldn't happen */
2513  elog(ERROR, "system-column update is not supported");
2514 
2515  if (!is_foreign_expr(root, foreignrel, (Expr *) tle->expr))
2516  return false;
2517  }
2518  }
2519 
2520  /*
2521  * Ok, rewrite subplan so as to modify the foreign table directly.
2522  */
2523  initStringInfo(&sql);
2524 
2525  /*
2526  * Core code already has some lock on each rel being planned, so we can
2527  * use NoLock here.
2528  */
2529  rel = table_open(rte->relid, NoLock);
2530 
2531  /*
2532  * Recall the qual clauses that must be evaluated remotely. (These are
2533  * bare clauses not RestrictInfos, but deparse.c's appendConditions()
2534  * doesn't care.)
2535  */
2536  remote_exprs = fpinfo->final_remote_exprs;
2537 
2538  /*
2539  * Extract the relevant RETURNING list if any.
2540  */
2541  if (plan->returningLists)
2542  {
2543  returningList = (List *) list_nth(plan->returningLists, subplan_index);
2544 
2545  /*
2546  * When performing an UPDATE/DELETE .. RETURNING on a join directly,
2547  * we fetch from the foreign server any Vars specified in RETURNING
2548  * that refer not only to the target relation but to non-target
2549  * relations. So we'll deparse them into the RETURNING clause of the
2550  * remote query; use a targetlist consisting of them instead, which
2551  * will be adjusted to be new fdw_scan_tlist of the foreign-scan plan
2552  * node below.
2553  */
2554  if (fscan->scan.scanrelid == 0)
2555  returningList = build_remote_returning(resultRelation, rel,
2556  returningList);
2557  }
2558 
2559  /*
2560  * Construct the SQL command string.
2561  */
2562  switch (operation)
2563  {
2564  case CMD_UPDATE:
2565  deparseDirectUpdateSql(&sql, root, resultRelation, rel,
2566  foreignrel,
2567  processed_tlist,
2568  targetAttrs,
2569  remote_exprs, &params_list,
2570  returningList, &retrieved_attrs);
2571  break;
2572  case CMD_DELETE:
2573  deparseDirectDeleteSql(&sql, root, resultRelation, rel,
2574  foreignrel,
2575  remote_exprs, &params_list,
2576  returningList, &retrieved_attrs);
2577  break;
2578  default:
2579  elog(ERROR, "unexpected operation: %d", (int) operation);
2580  break;
2581  }
2582 
2583  /*
2584  * Update the operation and target relation info.
2585  */
2586  fscan->operation = operation;
2587  fscan->resultRelation = resultRelation;
2588 
2589  /*
2590  * Update the fdw_exprs list that will be available to the executor.
2591  */
2592  fscan->fdw_exprs = params_list;
2593 
2594  /*
2595  * Update the fdw_private list that will be available to the executor.
2596  * Items in the list must match enum FdwDirectModifyPrivateIndex, above.
2597  */
2598  fscan->fdw_private = list_make4(makeString(sql.data),
2599  makeBoolean((retrieved_attrs != NIL)),
2600  retrieved_attrs,
2601  makeBoolean(plan->canSetTag));
2602 
2603  /*
2604  * Update the foreign-join-related fields.
2605  */
2606  if (fscan->scan.scanrelid == 0)
2607  {
2608  /* No need for the outer subplan. */
2609  fscan->scan.plan.lefttree = NULL;
2610 
2611  /* Build new fdw_scan_tlist if UPDATE/DELETE .. RETURNING. */
2612  if (returningList)
2613  rebuild_fdw_scan_tlist(fscan, returningList);
2614  }
2615 
2616  /*
2617  * Finally, unset the async-capable flag if it is set, as we currently
2618  * don't support asynchronous execution of direct modifications.
2619  */
2620  if (fscan->scan.plan.async_capable)
2621  fscan->scan.plan.async_capable = false;
2622 
2623  table_close(rel, NoLock);
2624  return true;
2625 }
2626 
2627 /*
2628  * postgresBeginDirectModify
2629  * Prepare a direct foreign table modification
2630  */
2631 static void
2633 {
2634  ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan;
2635  EState *estate = node->ss.ps.state;
2636  PgFdwDirectModifyState *dmstate;
2637  Index rtindex;
2638  Oid userid;
2639  ForeignTable *table;
2640  UserMapping *user;
2641  int numParams;
2642 
2643  /*
2644  * Do nothing in EXPLAIN (no ANALYZE) case. node->fdw_state stays NULL.
2645  */
2646  if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
2647  return;
2648 
2649  /*
2650  * We'll save private state in node->fdw_state.
2651  */
2652  dmstate = (PgFdwDirectModifyState *) palloc0(sizeof(PgFdwDirectModifyState));
2653  node->fdw_state = (void *) dmstate;
2654 
2655  /*
2656  * Identify which user to do the remote access as. This should match what
2657  * ExecCheckPermissions() does.
2658  */
2659  userid = OidIsValid(fsplan->checkAsUser) ? fsplan->checkAsUser : GetUserId();
2660 
2661  /* Get info about foreign table. */
2662  rtindex = node->resultRelInfo->ri_RangeTableIndex;
2663  if (fsplan->scan.scanrelid == 0)
2664  dmstate->rel = ExecOpenScanRelation(estate, rtindex, eflags);
2665  else
2666  dmstate->rel = node->ss.ss_currentRelation;
2667  table = GetForeignTable(RelationGetRelid(dmstate->rel));
2668  user = GetUserMapping(userid, table->serverid);
2669 
2670  /*
2671  * Get connection to the foreign server. Connection manager will
2672  * establish new connection if necessary.
2673  */
2674  dmstate->conn = GetConnection(user, false, &dmstate->conn_state);
2675 
2676  /* Update the foreign-join-related fields. */
2677  if (fsplan->scan.scanrelid == 0)
2678  {
2679  /* Save info about foreign table. */
2680  dmstate->resultRel = dmstate->rel;
2681 
2682  /*
2683  * Set dmstate->rel to NULL to teach get_returning_data() and
2684  * make_tuple_from_result_row() that columns fetched from the remote
2685  * server are described by fdw_scan_tlist of the foreign-scan plan
2686  * node, not the tuple descriptor for the target relation.
2687  */
2688  dmstate->rel = NULL;
2689  }
2690 
2691  /* Initialize state variable */
2692  dmstate->num_tuples = -1; /* -1 means not set yet */
2693 
2694  /* Get private info created by planner functions. */
2695  dmstate->query = strVal(list_nth(fsplan->fdw_private,
2697  dmstate->has_returning = boolVal(list_nth(fsplan->fdw_private,
2699  dmstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private,
2701  dmstate->set_processed = boolVal(list_nth(fsplan->fdw_private,
2703 
2704  /* Create context for per-tuple temp workspace. */
2705  dmstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
2706  "postgres_fdw temporary data",
2708 
2709  /* Prepare for input conversion of RETURNING results. */
2710  if (dmstate->has_returning)
2711  {
2712  TupleDesc tupdesc;
2713 
2714  if (fsplan->scan.scanrelid == 0)
2715  tupdesc = get_tupdesc_for_join_scan_tuples(node);
2716  else
2717  tupdesc = RelationGetDescr(dmstate->rel);
2718 
2719  dmstate->attinmeta = TupleDescGetAttInMetadata(tupdesc);
2720 
2721  /*
2722  * When performing an UPDATE/DELETE .. RETURNING on a join directly,
2723  * initialize a filter to extract an updated/deleted tuple from a scan
2724  * tuple.
2725  */
2726  if (fsplan->scan.scanrelid == 0)
2727  init_returning_filter(dmstate, fsplan->fdw_scan_tlist, rtindex);
2728  }
2729 
2730  /*
2731  * Prepare for processing of parameters used in remote query, if any.
2732  */
2733  numParams = list_length(fsplan->fdw_exprs);
2734  dmstate->numParams = numParams;
2735  if (numParams > 0)
2737  fsplan->fdw_exprs,
2738  numParams,
2739  &dmstate->param_flinfo,
2740  &dmstate->param_exprs,
2741  &dmstate->param_values);
2742 }
2743 
2744 /*
2745  * postgresIterateDirectModify
2746  * Execute a direct foreign table modification
2747  */
2748 static TupleTableSlot *
2750 {
2752  EState *estate = node->ss.ps.state;
2753  ResultRelInfo *resultRelInfo = node->resultRelInfo;
2754 
2755  /*
2756  * If this is the first call after Begin, execute the statement.
2757  */
2758  if (dmstate->num_tuples == -1)
2759  execute_dml_stmt(node);
2760 
2761  /*
2762  * If the local query doesn't specify RETURNING, just clear tuple slot.
2763  */
2764  if (!resultRelInfo->ri_projectReturning)
2765  {
2766  TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
2767  Instrumentation *instr = node->ss.ps.instrument;
2768 
2769  Assert(!dmstate->has_returning);
2770 
2771  /* Increment the command es_processed count if necessary. */
2772  if (dmstate->set_processed)
2773  estate->es_processed += dmstate->num_tuples;
2774 
2775  /* Increment the tuple count for EXPLAIN ANALYZE if necessary. */
2776  if (instr)
2777  instr->tuplecount += dmstate->num_tuples;
2778 
2779  return ExecClearTuple(slot);
2780  }
2781 
2782  /*
2783  * Get the next RETURNING tuple.
2784  */
2785  return get_returning_data(node);
2786 }
2787 
2788 /*
2789  * postgresEndDirectModify
2790  * Finish a direct foreign table modification
2791  */
2792 static void
2794 {
2796 
2797  /* if dmstate is NULL, we are in EXPLAIN; nothing to do */
2798  if (dmstate == NULL)
2799  return;
2800 
2801  /* Release PGresult */
2802  PQclear(dmstate->result);
2803 
2804  /* Release remote connection */
2805  ReleaseConnection(dmstate->conn);
2806  dmstate->conn = NULL;
2807 
2808  /* MemoryContext will be deleted automatically. */
2809 }
2810 
2811 /*
2812  * postgresExplainForeignScan
2813  * Produce extra output for EXPLAIN of a ForeignScan on a foreign table
2814  */
2815 static void
2817 {
2819  List *fdw_private = plan->fdw_private;
2820 
2821  /*
2822  * Identify foreign scans that are really joins or upper relations. The
2823  * input looks something like "(1) LEFT JOIN (2)", and we must replace the
2824  * digit string(s), which are RT indexes, with the correct relation names.
2825  * We do that here, not when the plan is created, because we can't know
2826  * what aliases ruleutils.c will assign at plan creation time.
2827  */
2828  if (list_length(fdw_private) > FdwScanPrivateRelations)
2829  {
2830  StringInfo relations;
2831  char *rawrelations;
2832  char *ptr;
2833  int minrti,
2834  rtoffset;
2835 
2836  rawrelations = strVal(list_nth(fdw_private, FdwScanPrivateRelations));
2837 
2838  /*
2839  * A difficulty with using a string representation of RT indexes is
2840  * that setrefs.c won't update the string when flattening the
2841  * rangetable. To find out what rtoffset was applied, identify the
2842  * minimum RT index appearing in the string and compare it to the
2843  * minimum member of plan->fs_base_relids. (We expect all the relids
2844  * in the join will have been offset by the same amount; the Asserts
2845  * below should catch it if that ever changes.)
2846  */
2847  minrti = INT_MAX;
2848  ptr = rawrelations;
2849  while (*ptr)
2850  {
2851  if (isdigit((unsigned char) *ptr))
2852  {
2853  int rti = strtol(ptr, &ptr, 10);
2854 
2855  if (rti < minrti)
2856  minrti = rti;
2857  }
2858  else
2859  ptr++;
2860  }
2861  rtoffset = bms_next_member(plan->fs_base_relids, -1) - minrti;
2862 
2863  /* Now we can translate the string */
2864  relations = makeStringInfo();
2865  ptr = rawrelations;
2866  while (*ptr)
2867  {
2868  if (isdigit((unsigned char) *ptr))
2869  {
2870  int rti = strtol(ptr, &ptr, 10);
2871  RangeTblEntry *rte;
2872  char *relname;
2873  char *refname;
2874 
2875  rti += rtoffset;
2876  Assert(bms_is_member(rti, plan->fs_base_relids));
2877  rte = rt_fetch(rti, es->rtable);
2878  Assert(rte->rtekind == RTE_RELATION);
2879  /* This logic should agree with explain.c's ExplainTargetRel */
2880  relname = get_rel_name(rte->relid);
2881  if (es->verbose)
2882  {
2883  char *namespace;
2884 
2885  namespace = get_namespace_name_or_temp(get_rel_namespace(rte->relid));
2886  appendStringInfo(relations, "%s.%s",
2887  quote_identifier(namespace),
2889  }
2890  else
2891  appendStringInfoString(relations,
2893  refname = (char *) list_nth(es->rtable_names, rti - 1);
2894  if (refname == NULL)
2895  refname = rte->eref->aliasname;
2896  if (strcmp(refname, relname) != 0)
2897  appendStringInfo(relations, " %s",
2898  quote_identifier(refname));
2899  }
2900  else
2901  appendStringInfoChar(relations, *ptr++);
2902  }
2903  ExplainPropertyText("Relations", relations->data, es);
2904  }
2905 
2906  /*
2907  * Add remote query, when VERBOSE option is specified.
2908  */
2909  if (es->verbose)
2910  {
2911  char *sql;
2912 
2913  sql = strVal(list_nth(fdw_private, FdwScanPrivateSelectSql));
2914  ExplainPropertyText("Remote SQL", sql, es);
2915  }
2916 }
2917 
2918 /*
2919  * postgresExplainForeignModify
2920  * Produce extra output for EXPLAIN of a ModifyTable on a foreign table
2921  */
2922 static void
2924  ResultRelInfo *rinfo,
2925  List *fdw_private,
2926  int subplan_index,
2927  ExplainState *es)
2928 {
2929  if (es->verbose)
2930  {
2931  char *sql = strVal(list_nth(fdw_private,
2933 
2934  ExplainPropertyText("Remote SQL", sql, es);
2935 
2936  /*
2937  * For INSERT we should always have batch size >= 1, but UPDATE and
2938  * DELETE don't support batching so don't show the property.
2939  */
2940  if (rinfo->ri_BatchSize > 0)
2941  ExplainPropertyInteger("Batch Size", NULL, rinfo->ri_BatchSize, es);
2942  }
2943 }
2944 
2945 /*
2946  * postgresExplainDirectModify
2947  * Produce extra output for EXPLAIN of a ForeignScan that modifies a
2948  * foreign table directly
2949  */
2950 static void
2952 {
2953  List *fdw_private;
2954  char *sql;
2955 
2956  if (es->verbose)
2957  {
2958  fdw_private = ((ForeignScan *) node->ss.ps.plan)->fdw_private;
2959  sql = strVal(list_nth(fdw_private, FdwDirectModifyPrivateUpdateSql));
2960  ExplainPropertyText("Remote SQL", sql, es);
2961  }
2962 }
2963 
2964 /*
2965  * postgresExecForeignTruncate
2966  * Truncate one or more foreign tables
2967  */
2968 static void
2970  DropBehavior behavior,
2971  bool restart_seqs)
2972 {
2973  Oid serverid = InvalidOid;
2974  UserMapping *user = NULL;
2975  PGconn *conn = NULL;
2976  StringInfoData sql;
2977  ListCell *lc;
2978  bool server_truncatable = true;
2979 
2980  /*
2981  * By default, all postgres_fdw foreign tables are assumed truncatable.
2982  * This can be overridden by a per-server setting, which in turn can be
2983  * overridden by a per-table setting.
2984  */
2985  foreach(lc, rels)
2986  {
2987  ForeignServer *server = NULL;
2988  Relation rel = lfirst(lc);
2990  ListCell *cell;
2991  bool truncatable;
2992 
2993  /*
2994  * First time through, determine whether the foreign server allows
2995  * truncates. Since all specified foreign tables are assumed to belong
2996  * to the same foreign server, this result can be used for other
2997  * foreign tables.
2998  */
2999  if (!OidIsValid(serverid))
3000  {
3001  serverid = table->serverid;
3002  server = GetForeignServer(serverid);
3003 
3004  foreach(cell, server->options)
3005  {
3006  DefElem *defel = (DefElem *) lfirst(cell);
3007 
3008  if (strcmp(defel->defname, "truncatable") == 0)
3009  {
3010  server_truncatable = defGetBoolean(defel);
3011  break;
3012  }
3013  }
3014  }
3015 
3016  /*
3017  * Confirm that all specified foreign tables belong to the same
3018  * foreign server.
3019  */
3020  Assert(table->serverid == serverid);
3021 
3022  /* Determine whether this foreign table allows truncations */
3023  truncatable = server_truncatable;
3024  foreach(cell, table->options)
3025  {
3026  DefElem *defel = (DefElem *) lfirst(cell);
3027 
3028  if (strcmp(defel->defname, "truncatable") == 0)
3029  {
3030  truncatable = defGetBoolean(defel);
3031  break;
3032  }
3033  }
3034 
3035  if (!truncatable)
3036  ereport(ERROR,
3037  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
3038  errmsg("foreign table \"%s\" does not allow truncates",
3039  RelationGetRelationName(rel))));
3040  }
3041  Assert(OidIsValid(serverid));
3042 
3043  /*
3044  * Get connection to the foreign server. Connection manager will
3045  * establish new connection if necessary.
3046  */
3047  user = GetUserMapping(GetUserId(), serverid);
3048  conn = GetConnection(user, false, NULL);
3049 
3050  /* Construct the TRUNCATE command string */
3051  initStringInfo(&sql);
3052  deparseTruncateSql(&sql, rels, behavior, restart_seqs);
3053 
3054  /* Issue the TRUNCATE command to remote server */
3055  do_sql_command(conn, sql.data);
3056 
3057  pfree(sql.data);
3058 }
3059 
3060 /*
3061  * estimate_path_cost_size
3062  * Get cost and size estimates for a foreign scan on given foreign relation
3063  * either a base relation or a join between foreign relations or an upper
3064  * relation containing foreign relations.
3065  *
3066  * param_join_conds are the parameterization clauses with outer relations.
3067  * pathkeys specify the expected sort order if any for given path being costed.
3068  * fpextra specifies additional post-scan/join-processing steps such as the
3069  * final sort and the LIMIT restriction.
3070  *
3071  * The function returns the cost and size estimates in p_rows, p_width,
3072  * p_startup_cost and p_total_cost variables.
3073  */
3074 static void
3076  RelOptInfo *foreignrel,
3077  List *param_join_conds,
3078  List *pathkeys,
3079  PgFdwPathExtraData *fpextra,
3080  double *p_rows, int *p_width,
3081  Cost *p_startup_cost, Cost *p_total_cost)
3082 {
3083  PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) foreignrel->fdw_private;
3084  double rows;
3085  double retrieved_rows;
3086  int width;
3087  Cost startup_cost;
3088  Cost total_cost;
3089 
3090  /* Make sure the core code has set up the relation's reltarget */
3091  Assert(foreignrel->reltarget);
3092 
3093  /*
3094  * If the table or the server is configured to use remote estimates,
3095  * connect to the foreign server and execute EXPLAIN to estimate the
3096  * number of rows selected by the restriction+join clauses. Otherwise,
3097  * estimate rows using whatever statistics we have locally, in a way
3098  * similar to ordinary tables.
3099  */
3100  if (fpinfo->use_remote_estimate)
3101  {
3102  List *remote_param_join_conds;
3103  List *local_param_join_conds;
3104  StringInfoData sql;
3105  PGconn *conn;
3106  Selectivity local_sel;
3107  QualCost local_cost;
3108  List *fdw_scan_tlist = NIL;
3109  List *remote_conds;
3110 
3111  /* Required only to be passed to deparseSelectStmtForRel */
3112  List *retrieved_attrs;
3113 
3114  /*
3115  * param_join_conds might contain both clauses that are safe to send
3116  * across, and clauses that aren't.
3117  */
3118  classifyConditions(root, foreignrel, param_join_conds,
3119  &remote_param_join_conds, &local_param_join_conds);
3120 
3121  /* Build the list of columns to be fetched from the foreign server. */
3122  if (IS_JOIN_REL(foreignrel) || IS_UPPER_REL(foreignrel))
3123  fdw_scan_tlist = build_tlist_to_deparse(foreignrel);
3124  else
3125  fdw_scan_tlist = NIL;
3126 
3127  /*
3128  * The complete list of remote conditions includes everything from
3129  * baserestrictinfo plus any extra join_conds relevant to this
3130  * particular path.
3131  */
3132  remote_conds = list_concat(remote_param_join_conds,
3133  fpinfo->remote_conds);
3134 
3135  /*
3136  * Construct EXPLAIN query including the desired SELECT, FROM, and
3137  * WHERE clauses. Params and other-relation Vars are replaced by dummy
3138  * values, so don't request params_list.
3139  */
3140  initStringInfo(&sql);
3141  appendStringInfoString(&sql, "EXPLAIN ");
3142  deparseSelectStmtForRel(&sql, root, foreignrel, fdw_scan_tlist,
3143  remote_conds, pathkeys,
3144  fpextra ? fpextra->has_final_sort : false,
3145  fpextra ? fpextra->has_limit : false,
3146  false, &retrieved_attrs, NULL);
3147 
3148  /* Get the remote estimate */
3149  conn = GetConnection(fpinfo->user, false, NULL);
3150  get_remote_estimate(sql.data, conn, &rows, &width,
3151  &startup_cost, &total_cost);
3153 
3154  retrieved_rows = rows;
3155 
3156  /* Factor in the selectivity of the locally-checked quals */
3157  local_sel = clauselist_selectivity(root,
3158  local_param_join_conds,
3159  foreignrel->relid,
3160  JOIN_INNER,
3161  NULL);
3162  local_sel *= fpinfo->local_conds_sel;
3163 
3164  rows = clamp_row_est(rows * local_sel);
3165 
3166  /* Add in the eval cost of the locally-checked quals */
3167  startup_cost += fpinfo->local_conds_cost.startup;
3168  total_cost += fpinfo->local_conds_cost.per_tuple * retrieved_rows;
3169  cost_qual_eval(&local_cost, local_param_join_conds, root);
3170  startup_cost += local_cost.startup;
3171  total_cost += local_cost.per_tuple * retrieved_rows;
3172 
3173  /*
3174  * Add in tlist eval cost for each output row. In case of an
3175  * aggregate, some of the tlist expressions such as grouping
3176  * expressions will be evaluated remotely, so adjust the costs.
3177  */
3178  startup_cost += foreignrel->reltarget->cost.startup;
3179  total_cost += foreignrel->reltarget->cost.startup;
3180  total_cost += foreignrel->reltarget->cost.per_tuple * rows;
3181  if (IS_UPPER_REL(foreignrel))
3182  {
3183  QualCost tlist_cost;
3184 
3185  cost_qual_eval(&tlist_cost, fdw_scan_tlist, root);
3186  startup_cost -= tlist_cost.startup;
3187  total_cost -= tlist_cost.startup;
3188  total_cost -= tlist_cost.per_tuple * rows;
3189  }
3190  }
3191  else
3192  {
3193  Cost run_cost = 0;
3194 
3195  /*
3196  * We don't support join conditions in this mode (hence, no
3197  * parameterized paths can be made).
3198  */
3199  Assert(param_join_conds == NIL);
3200 
3201  /*
3202  * We will come here again and again with different set of pathkeys or
3203  * additional post-scan/join-processing steps that caller wants to
3204  * cost. We don't need to calculate the cost/size estimates for the
3205  * underlying scan, join, or grouping each time. Instead, use those
3206  * estimates if we have cached them already.
3207  */
3208  if (fpinfo->rel_startup_cost >= 0 && fpinfo->rel_total_cost >= 0)
3209  {
3210  Assert(fpinfo->retrieved_rows >= 0);
3211 
3212  rows = fpinfo->rows;
3213  retrieved_rows = fpinfo->retrieved_rows;
3214  width = fpinfo->width;
3215  startup_cost = fpinfo->rel_startup_cost;
3216  run_cost = fpinfo->rel_total_cost - fpinfo->rel_startup_cost;
3217 
3218  /*
3219  * If we estimate the costs of a foreign scan or a foreign join
3220  * with additional post-scan/join-processing steps, the scan or
3221  * join costs obtained from the cache wouldn't yet contain the
3222  * eval costs for the final scan/join target, which would've been
3223  * updated by apply_scanjoin_target_to_paths(); add the eval costs
3224  * now.
3225  */
3226  if (fpextra && !IS_UPPER_REL(foreignrel))
3227  {
3228  /* Shouldn't get here unless we have LIMIT */
3229  Assert(fpextra->has_limit);
3230  Assert(foreignrel->reloptkind == RELOPT_BASEREL ||
3231  foreignrel->reloptkind == RELOPT_JOINREL);
3232  startup_cost += foreignrel->reltarget->cost.startup;
3233  run_cost += foreignrel->reltarget->cost.per_tuple * rows;
3234  }
3235  }
3236  else if (IS_JOIN_REL(foreignrel))
3237  {
3238  PgFdwRelationInfo *fpinfo_i;
3239  PgFdwRelationInfo *fpinfo_o;
3240  QualCost join_cost;
3241  QualCost remote_conds_cost;
3242  double nrows;
3243 
3244  /* Use rows/width estimates made by the core code. */
3245  rows = foreignrel->rows;
3246  width = foreignrel->reltarget->width;
3247 
3248  /* For join we expect inner and outer relations set */
3249  Assert(fpinfo->innerrel && fpinfo->outerrel);
3250 
3251  fpinfo_i = (PgFdwRelationInfo *) fpinfo->innerrel->fdw_private;
3252  fpinfo_o = (PgFdwRelationInfo *) fpinfo->outerrel->fdw_private;
3253 
3254  /* Estimate of number of rows in cross product */
3255  nrows = fpinfo_i->rows * fpinfo_o->rows;
3256 
3257  /*
3258  * Back into an estimate of the number of retrieved rows. Just in
3259  * case this is nuts, clamp to at most nrows.
3260  */
3261  retrieved_rows = clamp_row_est(rows / fpinfo->local_conds_sel);
3262  retrieved_rows = Min(retrieved_rows, nrows);
3263 
3264  /*
3265  * The cost of foreign join is estimated as cost of generating
3266  * rows for the joining relations + cost for applying quals on the
3267  * rows.
3268  */
3269 
3270  /*
3271  * Calculate the cost of clauses pushed down to the foreign server
3272  */
3273  cost_qual_eval(&remote_conds_cost, fpinfo->remote_conds, root);
3274  /* Calculate the cost of applying join clauses */
3275  cost_qual_eval(&join_cost, fpinfo->joinclauses, root);
3276 
3277  /*
3278  * Startup cost includes startup cost of joining relations and the
3279  * startup cost for join and other clauses. We do not include the
3280  * startup cost specific to join strategy (e.g. setting up hash
3281  * tables) since we do not know what strategy the foreign server
3282  * is going to use.
3283  */
3284  startup_cost = fpinfo_i->rel_startup_cost + fpinfo_o->rel_startup_cost;
3285  startup_cost += join_cost.startup;
3286  startup_cost += remote_conds_cost.startup;
3287  startup_cost += fpinfo->local_conds_cost.startup;
3288 
3289  /*
3290  * Run time cost includes:
3291  *
3292  * 1. Run time cost (total_cost - startup_cost) of relations being
3293  * joined
3294  *
3295  * 2. Run time cost of applying join clauses on the cross product
3296  * of the joining relations.
3297  *
3298  * 3. Run time cost of applying pushed down other clauses on the
3299  * result of join
3300  *
3301  * 4. Run time cost of applying nonpushable other clauses locally
3302  * on the result fetched from the foreign server.
3303  */
3304  run_cost = fpinfo_i->rel_total_cost - fpinfo_i->rel_startup_cost;
3305  run_cost += fpinfo_o->rel_total_cost - fpinfo_o->rel_startup_cost;
3306  run_cost += nrows * join_cost.per_tuple;
3307  nrows = clamp_row_est(nrows * fpinfo->joinclause_sel);
3308  run_cost += nrows * remote_conds_cost.per_tuple;
3309  run_cost += fpinfo->local_conds_cost.per_tuple * retrieved_rows;
3310 
3311  /* Add in tlist eval cost for each output row */
3312  startup_cost += foreignrel->reltarget->cost.startup;
3313  run_cost += foreignrel->reltarget->cost.per_tuple * rows;
3314  }
3315  else if (IS_UPPER_REL(foreignrel))
3316  {
3317  RelOptInfo *outerrel = fpinfo->outerrel;
3318  PgFdwRelationInfo *ofpinfo;
3319  AggClauseCosts aggcosts;
3320  double input_rows;
3321  int numGroupCols;
3322  double numGroups = 1;
3323 
3324  /* The upper relation should have its outer relation set */
3325  Assert(outerrel);
3326  /* and that outer relation should have its reltarget set */
3327  Assert(outerrel->reltarget);
3328 
3329  /*
3330  * This cost model is mixture of costing done for sorted and
3331  * hashed aggregates in cost_agg(). We are not sure which
3332  * strategy will be considered at remote side, thus for
3333  * simplicity, we put all startup related costs in startup_cost
3334  * and all finalization and run cost are added in total_cost.
3335  */
3336 
3337  ofpinfo = (PgFdwRelationInfo *) outerrel->fdw_private;
3338 
3339  /* Get rows from input rel */
3340  input_rows = ofpinfo->rows;
3341 
3342  /* Collect statistics about aggregates for estimating costs. */
3343  MemSet(&aggcosts, 0, sizeof(AggClauseCosts));
3344  if (root->parse->hasAggs)
3345  {
3346  get_agg_clause_costs(root, AGGSPLIT_SIMPLE, &aggcosts);
3347  }
3348 
3349  /* Get number of grouping columns and possible number of groups */
3350  numGroupCols = list_length(root->processed_groupClause);
3351  numGroups = estimate_num_groups(root,
3353  fpinfo->grouped_tlist),
3354  input_rows, NULL, NULL);
3355 
3356  /*
3357  * Get the retrieved_rows and rows estimates. If there are HAVING
3358  * quals, account for their selectivity.
3359  */
3360  if (root->hasHavingQual)
3361  {
3362  /* Factor in the selectivity of the remotely-checked quals */
3363  retrieved_rows =
3364  clamp_row_est(numGroups *
3366  fpinfo->remote_conds,
3367  0,
3368  JOIN_INNER,
3369  NULL));
3370  /* Factor in the selectivity of the locally-checked quals */
3371  rows = clamp_row_est(retrieved_rows * fpinfo->local_conds_sel);
3372  }
3373  else
3374  {
3375  rows = retrieved_rows = numGroups;
3376  }
3377 
3378  /* Use width estimate made by the core code. */
3379  width = foreignrel->reltarget->width;
3380 
3381  /*-----
3382  * Startup cost includes:
3383  * 1. Startup cost for underneath input relation, adjusted for
3384  * tlist replacement by apply_scanjoin_target_to_paths()
3385  * 2. Cost of performing aggregation, per cost_agg()
3386  *-----
3387  */
3388  startup_cost = ofpinfo->rel_startup_cost;
3389  startup_cost += outerrel->reltarget->cost.startup;
3390  startup_cost += aggcosts.transCost.startup;
3391  startup_cost += aggcosts.transCost.per_tuple * input_rows;
3392  startup_cost += aggcosts.finalCost.startup;
3393  startup_cost += (cpu_operator_cost * numGroupCols) * input_rows;
3394 
3395  /*-----
3396  * Run time cost includes:
3397  * 1. Run time cost of underneath input relation, adjusted for
3398  * tlist replacement by apply_scanjoin_target_to_paths()
3399  * 2. Run time cost of performing aggregation, per cost_agg()
3400  *-----
3401  */
3402  run_cost = ofpinfo->rel_total_cost - ofpinfo->rel_startup_cost;
3403  run_cost += outerrel->reltarget->cost.per_tuple * input_rows;
3404  run_cost += aggcosts.finalCost.per_tuple * numGroups;
3405  run_cost += cpu_tuple_cost * numGroups;
3406 
3407  /* Account for the eval cost of HAVING quals, if any */
3408  if (root->hasHavingQual)
3409  {
3410  QualCost remote_cost;
3411 
3412  /* Add in the eval cost of the remotely-checked quals */
3413  cost_qual_eval(&remote_cost, fpinfo->remote_conds, root);
3414  startup_cost += remote_cost.startup;
3415  run_cost += remote_cost.per_tuple * numGroups;
3416  /* Add in the eval cost of the locally-checked quals */
3417  startup_cost += fpinfo->local_conds_cost.startup;
3418  run_cost += fpinfo->local_conds_cost.per_tuple * retrieved_rows;
3419  }
3420 
3421  /* Add in tlist eval cost for each output row */
3422  startup_cost += foreignrel->reltarget->cost.startup;
3423  run_cost += foreignrel->reltarget->cost.per_tuple * rows;
3424  }
3425  else
3426  {
3427  Cost cpu_per_tuple;
3428 
3429  /* Use rows/width estimates made by set_baserel_size_estimates. */
3430  rows = foreignrel->rows;
3431  width = foreignrel->reltarget->width;
3432 
3433  /*
3434  * Back into an estimate of the number of retrieved rows. Just in
3435  * case this is nuts, clamp to at most foreignrel->tuples.
3436  */
3437  retrieved_rows = clamp_row_est(rows / fpinfo->local_conds_sel);
3438  retrieved_rows = Min(retrieved_rows, foreignrel->tuples);
3439 
3440  /*
3441  * Cost as though this were a seqscan, which is pessimistic. We
3442  * effectively imagine the local_conds are being evaluated
3443  * remotely, too.
3444  */
3445  startup_cost = 0;
3446  run_cost = 0;
3447  run_cost += seq_page_cost * foreignrel->pages;
3448 
3449  startup_cost += foreignrel->baserestrictcost.startup;
3450  cpu_per_tuple = cpu_tuple_cost + foreignrel->baserestrictcost.per_tuple;
3451  run_cost += cpu_per_tuple * foreignrel->tuples;
3452 
3453  /* Add in tlist eval cost for each output row */
3454  startup_cost += foreignrel->reltarget->cost.startup;
3455  run_cost += foreignrel->reltarget->cost.per_tuple * rows;
3456  }
3457 
3458  /*
3459  * Without remote estimates, we have no real way to estimate the cost
3460  * of generating sorted output. It could be free if the query plan
3461  * the remote side would have chosen generates properly-sorted output
3462  * anyway, but in most cases it will cost something. Estimate a value
3463  * high enough that we won't pick the sorted path when the ordering
3464  * isn't locally useful, but low enough that we'll err on the side of
3465  * pushing down the ORDER BY clause when it's useful to do so.
3466  */
3467  if (pathkeys != NIL)
3468  {
3469  if (IS_UPPER_REL(foreignrel))
3470  {
3471  Assert(foreignrel->reloptkind == RELOPT_UPPER_REL &&
3472  fpinfo->stage == UPPERREL_GROUP_AGG);
3473  adjust_foreign_grouping_path_cost(root, pathkeys,
3474  retrieved_rows, width,
3475  fpextra->limit_tuples,
3476  &startup_cost, &run_cost);
3477  }
3478  else
3479  {
3480  startup_cost *= DEFAULT_FDW_SORT_MULTIPLIER;
3481  run_cost *= DEFAULT_FDW_SORT_MULTIPLIER;
3482  }
3483  }
3484 
3485  total_cost = startup_cost + run_cost;
3486 
3487  /* Adjust the cost estimates if we have LIMIT */
3488  if (fpextra && fpextra->has_limit)
3489  {
3490  adjust_limit_rows_costs(&rows, &startup_cost, &total_cost,
3491  fpextra->offset_est, fpextra->count_est);
3492  retrieved_rows = rows;
3493  }
3494  }
3495 
3496  /*
3497  * If this includes the final sort step, the given target, which will be
3498  * applied to the resulting path, might have different expressions from
3499  * the foreignrel's reltarget (see make_sort_input_target()); adjust tlist
3500  * eval costs.
3501  */
3502  if (fpextra && fpextra->has_final_sort &&
3503  fpextra->target != foreignrel->reltarget)
3504  {
3505  QualCost oldcost = foreignrel->reltarget->cost;
3506  QualCost newcost = fpextra->target->cost;
3507 
3508  startup_cost += newcost.startup - oldcost.startup;
3509  total_cost += newcost.startup - oldcost.startup;
3510  total_cost += (newcost.per_tuple - oldcost.per_tuple) * rows;
3511  }
3512 
3513  /*
3514  * Cache the retrieved rows and cost estimates for scans, joins, or
3515  * groupings without any parameterization, pathkeys, or additional
3516  * post-scan/join-processing steps, before adding the costs for
3517  * transferring data from the foreign server. These estimates are useful
3518  * for costing remote joins involving this relation or costing other
3519  * remote operations on this relation such as remote sorts and remote
3520  * LIMIT restrictions, when the costs can not be obtained from the foreign
3521  * server. This function will be called at least once for every foreign
3522  * relation without any parameterization, pathkeys, or additional
3523  * post-scan/join-processing steps.
3524  */
3525  if (pathkeys == NIL && param_join_conds == NIL && fpextra == NULL)
3526  {
3527  fpinfo->retrieved_rows = retrieved_rows;
3528  fpinfo->rel_startup_cost = startup_cost;
3529  fpinfo->rel_total_cost = total_cost;
3530  }
3531 
3532  /*
3533  * Add some additional cost factors to account for connection overhead
3534  * (fdw_startup_cost), transferring data across the network
3535  * (fdw_tuple_cost per retrieved row), and local manipulation of the data
3536  * (cpu_tuple_cost per retrieved row).
3537  */
3538  startup_cost += fpinfo->fdw_startup_cost;
3539  total_cost += fpinfo->fdw_startup_cost;
3540  total_cost += fpinfo->fdw_tuple_cost * retrieved_rows;
3541  total_cost += cpu_tuple_cost * retrieved_rows;
3542 
3543  /*
3544  * If we have LIMIT, we should prefer performing the restriction remotely
3545  * rather than locally, as the former avoids extra row fetches from the
3546  * remote that the latter might cause. But since the core code doesn't
3547  * account for such fetches when estimating the costs of the local
3548  * restriction (see create_limit_path()), there would be no difference
3549  * between the costs of the local restriction and the costs of the remote
3550  * restriction estimated above if we don't use remote estimates (except
3551  * for the case where the foreignrel is a grouping relation, the given
3552  * pathkeys is not NIL, and the effects of a bounded sort for that rel is
3553  * accounted for in costing the remote restriction). Tweak the costs of
3554  * the remote restriction to ensure we'll prefer it if LIMIT is a useful
3555  * one.
3556  */
3557  if (!fpinfo->use_remote_estimate &&
3558  fpextra && fpextra->has_limit &&
3559  fpextra->limit_tuples > 0 &&
3560  fpextra->limit_tuples < fpinfo->rows)
3561  {
3562  Assert(fpinfo->rows > 0);
3563  total_cost -= (total_cost - startup_cost) * 0.05 *
3564  (fpinfo->rows - fpextra->limit_tuples) / fpinfo->rows;
3565  }
3566 
3567  /* Return results. */
3568  *p_rows = rows;
3569  *p_width = width;
3570  *p_startup_cost = startup_cost;
3571  *p_total_cost = total_cost;
3572 }
3573 
3574 /*
3575  * Estimate costs of executing a SQL statement remotely.
3576  * The given "sql" must be an EXPLAIN command.
3577  */
3578 static void
3579 get_remote_estimate(const char *sql, PGconn *conn,
3580  double *rows, int *width,
3581  Cost *startup_cost, Cost *total_cost)
3582 {
3583  PGresult *volatile res = NULL;
3584 
3585  /* PGresult must be released before leaving this function. */
3586  PG_TRY();
3587  {
3588  char *line;
3589  char *p;
3590  int n;
3591 
3592  /*
3593  * Execute EXPLAIN remotely.
3594  */
3595  res = pgfdw_exec_query(conn, sql, NULL);
3597  pgfdw_report_error(ERROR, res, conn, false, sql);
3598 
3599  /*
3600  * Extract cost numbers for topmost plan node. Note we search for a
3601  * left paren from the end of the line to avoid being confused by
3602  * other uses of parentheses.
3603  */
3604  line = PQgetvalue(res, 0, 0);
3605  p = strrchr(line, '(');
3606  if (p == NULL)
3607  elog(ERROR, "could not interpret EXPLAIN output: \"%s\"", line);
3608  n = sscanf(p, "(cost=%lf..%lf rows=%lf width=%d)",
3609  startup_cost, total_cost, rows, width);
3610  if (n != 4)
3611  elog(ERROR, "could not interpret EXPLAIN output: \"%s\"", line);
3612  }
3613  PG_FINALLY();
3614  {
3615  PQclear(res);
3616  }
3617  PG_END_TRY();
3618 }
3619 
3620 /*
3621  * Adjust the cost estimates of a foreign grouping path to include the cost of
3622  * generating properly-sorted output.
3623  */
3624 static void
3626  List *pathkeys,
3627  double retrieved_rows,
3628  double width,
3629  double limit_tuples,
3630  Cost *p_startup_cost,
3631  Cost *p_run_cost)
3632 {
3633  /*
3634  * If the GROUP BY clause isn't sort-able, the plan chosen by the remote
3635  * side is unlikely to generate properly-sorted output, so it would need
3636  * an explicit sort; adjust the given costs with cost_sort(). Likewise,
3637  * if the GROUP BY clause is sort-able but isn't a superset of the given
3638  * pathkeys, adjust the costs with that function. Otherwise, adjust the
3639  * costs by applying the same heuristic as for the scan or join case.
3640  */
3642  !pathkeys_contained_in(pathkeys, root->group_pathkeys))
3643  {
3644  Path sort_path; /* dummy for result of cost_sort */
3645 
3646  cost_sort(&sort_path,
3647  root,
3648  pathkeys,
3649  *p_startup_cost + *p_run_cost,
3650  retrieved_rows,
3651  width,
3652  0.0,
3653  work_mem,
3654  limit_tuples);
3655 
3656  *p_startup_cost = sort_path.startup_cost;
3657  *p_run_cost = sort_path.total_cost - sort_path.startup_cost;
3658  }
3659  else
3660  {
3661  /*
3662  * The default extra cost seems too large for foreign-grouping cases;
3663  * add 1/4th of that default.
3664  */
3665  double sort_multiplier = 1.0 + (DEFAULT_FDW_SORT_MULTIPLIER
3666  - 1.0) * 0.25;
3667 
3668  *p_startup_cost *= sort_multiplier;
3669  *p_run_cost *= sort_multiplier;
3670  }
3671 }
3672 
3673 /*
3674  * Detect whether we want to process an EquivalenceClass member.
3675  *
3676  * This is a callback for use by generate_implied_equalities_for_column.
3677  */
3678 static bool
3681  void *arg)
3682 {
3684  Expr *expr = em->em_expr;
3685 
3686  /*
3687  * If we've identified what we're processing in the current scan, we only
3688  * want to match that expression.
3689  */
3690  if (state->current != NULL)
3691  return equal(expr, state->current);
3692 
3693  /*
3694  * Otherwise, ignore anything we've already processed.
3695  */
3696  if (list_member(state->already_used, expr))
3697  return false;
3698 
3699  /* This is the new target to process. */
3700  state->current = expr;
3701  return true;
3702 }
3703 
3704 /*
3705  * Create cursor for node's query with current parameter values.
3706  */
3707 static void
3709 {
3710  PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
3711  ExprContext *econtext = node->ss.ps.ps_ExprContext;
3712  int numParams = fsstate->numParams;
3713  const char **values = fsstate->param_values;
3714  PGconn *conn = fsstate->conn;
3716  PGresult *res;
3717 
3718  /* First, process a pending asynchronous request, if any. */
3719  if (fsstate->conn_state->pendingAreq)
3721 
3722  /*
3723  * Construct array of query parameter values in text format. We do the
3724  * conversions in the short-lived per-tuple context, so as not to cause a
3725  * memory leak over repeated scans.
3726  */
3727  if (numParams > 0)
3728  {
3729  MemoryContext oldcontext;
3730 
3731  oldcontext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
3732 
3733  process_query_params(econtext,
3734  fsstate->param_flinfo,
3735  fsstate->param_exprs,
3736  values);
3737 
3738  MemoryContextSwitchTo(oldcontext);
3739  }
3740 
3741  /* Construct the DECLARE CURSOR command */
3742  initStringInfo(&buf);
3743  appendStringInfo(&buf, "DECLARE c%u CURSOR FOR\n%s",
3744  fsstate->cursor_number, fsstate->query);
3745 
3746  /*
3747  * Notice that we pass NULL for paramTypes, thus forcing the remote server
3748  * to infer types for all parameters. Since we explicitly cast every
3749  * parameter (see deparse.c), the "inference" is trivial and will produce
3750  * the desired result. This allows us to avoid assuming that the remote
3751  * server has the same OIDs we do for the parameters' types.
3752  */
3753  if (!PQsendQueryParams(conn, buf.data, numParams,
3754  NULL, values, NULL, NULL, 0))
3755  pgfdw_report_error(ERROR, NULL, conn, false, buf.data);
3756 
3757  /*
3758  * Get the result, and check for success.
3759  *
3760  * We don't use a PG_TRY block here, so be careful not to throw error
3761  * without releasing the PGresult.
3762  */
3765  pgfdw_report_error(ERROR, res, conn, true, fsstate->query);
3766  PQclear(res);
3767 
3768  /* Mark the cursor as created, and show no tuples have been retrieved */
3769  fsstate->cursor_exists = true;
3770  fsstate->tuples = NULL;
3771  fsstate->num_tuples = 0;
3772  fsstate->next_tuple = 0;
3773  fsstate->fetch_ct_2 = 0;
3774  fsstate->eof_reached = false;
3775 
3776  /* Clean up */
3777  pfree(buf.data);
3778 }
3779 
3780 /*
3781  * Fetch some more rows from the node's cursor.
3782  */
3783 static void
3785 {
3786  PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
3787  PGresult *volatile res = NULL;
3788  MemoryContext oldcontext;
3789 
3790  /*
3791  * We'll store the tuples in the batch_cxt. First, flush the previous
3792  * batch.
3793  */
3794  fsstate->tuples = NULL;
3795  MemoryContextReset(fsstate->batch_cxt);
3796  oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt);
3797 
3798  /* PGresult must be released before leaving this function. */
3799  PG_TRY();
3800  {
3801  PGconn *conn = fsstate->conn;
3802  int numrows;
3803  int i;
3804 
3805  if (fsstate->async_capable)
3806  {
3807  Assert(fsstate->conn_state->pendingAreq);
3808 
3809  /*
3810  * The query was already sent by an earlier call to
3811  * fetch_more_data_begin. So now we just fetch the result.
3812  */
3814  /* On error, report the original query, not the FETCH. */
3816  pgfdw_report_error(ERROR, res, conn, false, fsstate->query);
3817 
3818  /* Reset per-connection state */
3819  fsstate->conn_state->pendingAreq = NULL;
3820  }
3821  else
3822  {
3823  char sql[64];
3824 
3825  /* This is a regular synchronous fetch. */
3826  snprintf(sql, sizeof(sql), "FETCH %d FROM c%u",
3827  fsstate->fetch_size, fsstate->cursor_number);
3828 
3829  res = pgfdw_exec_query(conn, sql, fsstate->conn_state);
3830  /* On error, report the original query, not the FETCH. */
3832  pgfdw_report_error(ERROR, res, conn, false, fsstate->query);
3833  }
3834 
3835  /* Convert the data into HeapTuples */
3836  numrows = PQntuples(res);
3837  fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple));
3838  fsstate->num_tuples = numrows;
3839  fsstate->next_tuple = 0;
3840 
3841  for (i = 0; i < numrows; i++)
3842  {
3843  Assert(IsA(node->ss.ps.plan, ForeignScan));
3844 
3845  fsstate->tuples[i] =
3847  fsstate->rel,
3848  fsstate->attinmeta,
3849  fsstate->retrieved_attrs,
3850  node,
3851  fsstate->temp_cxt);
3852  }
3853 
3854  /* Update fetch_ct_2 */
3855  if (fsstate->fetch_ct_2 < 2)
3856  fsstate->fetch_ct_2++;
3857 
3858  /* Must be EOF if we didn't get as many tuples as we asked for. */
3859  fsstate->eof_reached = (numrows < fsstate->fetch_size);
3860  }
3861  PG_FINALLY();
3862  {
3863  PQclear(res);
3864  }
3865  PG_END_TRY();
3866 
3867  MemoryContextSwitchTo(oldcontext);
3868 }
3869 
3870 /*
3871  * Force assorted GUC parameters to settings that ensure that we'll output
3872  * data values in a form that is unambiguous to the remote server.
3873  *
3874  * This is rather expensive and annoying to do once per row, but there's
3875  * little choice if we want to be sure values are transmitted accurately;
3876  * we can't leave the settings in place between rows for fear of affecting
3877  * user-visible computations.
3878  *
3879  * We use the equivalent of a function SET option to allow the settings to
3880  * persist only until the caller calls reset_transmission_modes(). If an
3881  * error is thrown in between, guc.c will take care of undoing the settings.
3882  *
3883  * The return value is the nestlevel that must be passed to
3884  * reset_transmission_modes() to undo things.
3885  */
3886 int
3888 {
3889  int nestlevel = NewGUCNestLevel();
3890 
3891  /*
3892  * The values set here should match what pg_dump does. See also
3893  * configure_remote_session in connection.c.
3894  */
3895  if (DateStyle != USE_ISO_DATES)
3896  (void) set_config_option("datestyle", "ISO",
3898  GUC_ACTION_SAVE, true, 0, false);
3900  (void) set_config_option("intervalstyle", "postgres",
3902  GUC_ACTION_SAVE, true, 0, false);
3903  if (extra_float_digits < 3)
3904  (void) set_config_option("extra_float_digits", "3",
3906  GUC_ACTION_SAVE, true, 0, false);
3907 
3908  /*
3909  * In addition force restrictive search_path, in case there are any
3910  * regproc or similar constants to be printed.
3911  */
3912  (void) set_config_option("search_path", "pg_catalog",
3914  GUC_ACTION_SAVE, true, 0, false);
3915 
3916  return nestlevel;
3917 }
3918 
3919 /*
3920  * Undo the effects of set_transmission_modes().
3921  */
3922 void
3924 {
3925  AtEOXact_GUC(true, nestlevel);
3926 }
3927 
3928 /*
3929  * Utility routine to close a cursor.
3930  */
3931 static void
3933  PgFdwConnState *conn_state)
3934 {
3935  char sql[64];
3936  PGresult *res;
3937 
3938  snprintf(sql, sizeof(sql), "CLOSE c%u", cursor_number);
3939 
3940  /*
3941  * We don't use a PG_TRY block here, so be careful not to throw error
3942  * without releasing the PGresult.
3943  */
3944  res = pgfdw_exec_query(conn, sql, conn_state);
3946  pgfdw_report_error(ERROR, res, conn, true, sql);
3947  PQclear(res);
3948 }
3949 
3950 /*
3951  * create_foreign_modify
3952  * Construct an execution state of a foreign insert/update/delete
3953  * operation
3954  */
3955 static PgFdwModifyState *
3957  RangeTblEntry *rte,
3958  ResultRelInfo *resultRelInfo,
3959  CmdType operation,
3960  Plan *subplan,
3961  char *query,
3962  List *target_attrs,
3963  int values_end,
3964  bool has_returning,
3965  List *retrieved_attrs)
3966 {
3967  PgFdwModifyState *fmstate;
3968  Relation rel = resultRelInfo->ri_RelationDesc;
3969  TupleDesc tupdesc = RelationGetDescr(rel);
3970  Oid userid;
3971  ForeignTable *table;
3972  UserMapping *user;
3973  AttrNumber n_params;
3974  Oid typefnoid;
3975  bool isvarlena;
3976  ListCell *lc;
3977 
3978  /* Begin constructing PgFdwModifyState. */
3979  fmstate = (PgFdwModifyState *) palloc0(sizeof(PgFdwModifyState));
3980  fmstate->rel = rel;
3981 
3982  /* Identify which user to do the remote access as. */
3983  userid = ExecGetResultRelCheckAsUser(resultRelInfo, estate);
3984 
3985  /* Get info about foreign table. */
3986  table = GetForeignTable(RelationGetRelid(rel));
3987  user = GetUserMapping(userid, table->serverid);
3988 
3989  /* Open connection; report that we'll create a prepared statement. */
3990  fmstate->conn = GetConnection(user, true, &fmstate->conn_state);
3991  fmstate->p_name = NULL; /* prepared statement not made yet */
3992 
3993  /* Set up remote query information. */
3994  fmstate->query = query;
3995  if (operation == CMD_INSERT)
3996  {
3997  fmstate->query = pstrdup(fmstate->query);
3998  fmstate->orig_query = pstrdup(fmstate->query);
3999  }
4000  fmstate->target_attrs = target_attrs;
4001  fmstate->values_end = values_end;
4002  fmstate->has_returning = has_returning;
4003  fmstate->retrieved_attrs = retrieved_attrs;
4004 
4005  /* Create context for per-tuple temp workspace. */
4006  fmstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
4007  "postgres_fdw temporary data",
4009 
4010  /* Prepare for input conversion of RETURNING results. */
4011  if (fmstate->has_returning)
4012  fmstate->attinmeta = TupleDescGetAttInMetadata(tupdesc);
4013 
4014  /* Prepare for output conversion of parameters used in prepared stmt. */
4015  n_params = list_length(fmstate->target_attrs) + 1;
4016  fmstate->p_flinfo = (FmgrInfo *) palloc0(sizeof(FmgrInfo) * n_params);
4017  fmstate->p_nums = 0;
4018 
4019  if (operation == CMD_UPDATE || operation == CMD_DELETE)
4020  {
4021  Assert(subplan != NULL);
4022 
4023  /* Find the ctid resjunk column in the subplan's result */
4025  "ctid");
4026  if (!AttributeNumberIsValid(fmstate->ctidAttno))
4027  elog(ERROR, "could not find junk ctid column");
4028 
4029  /* First transmittable parameter will be ctid */
4030  getTypeOutputInfo(TIDOID, &typefnoid, &isvarlena);
4031  fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
4032  fmstate->p_nums++;
4033  }
4034 
4035  if (operation == CMD_INSERT || operation == CMD_UPDATE)
4036  {
4037  /* Set up for remaining transmittable parameters */
4038  foreach(lc, fmstate->target_attrs)
4039  {
4040  int attnum = lfirst_int(lc);
4041  Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
4042 
4043  Assert(!attr->attisdropped);
4044 
4045  /* Ignore generated columns; they are set to DEFAULT */
4046  if (attr->attgenerated)
4047  continue;
4048  getTypeOutputInfo(attr->atttypid, &typefnoid, &isvarlena);
4049  fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
4050  fmstate->p_nums++;
4051  }
4052  }
4053 
4054  Assert(fmstate->p_nums <= n_params);
4055 
4056  /* Set batch_size from foreign server/table options. */
4057  if (operation == CMD_INSERT)
4058  fmstate->batch_size = get_batch_size_option(rel);
4059 
4060  fmstate->num_slots = 1;
4061 
4062  /* Initialize auxiliary state */
4063  fmstate->aux_fmstate = NULL;
4064 
4065  return fmstate;
4066 }
4067 
4068 /*
4069  * execute_foreign_modify
4070  * Perform foreign-table modification as required, and fetch RETURNING
4071  * result if any. (This is the shared guts of postgresExecForeignInsert,
4072  * postgresExecForeignBatchInsert, postgresExecForeignUpdate, and
4073  * postgresExecForeignDelete.)
4074  */
4075 static TupleTableSlot **
4077  ResultRelInfo *resultRelInfo,
4078  CmdType operation,
4079  TupleTableSlot **slots,
4080  TupleTableSlot **planSlots,
4081  int *numSlots)
4082 {
4083  PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
4084  ItemPointer ctid = NULL;
4085  const char **p_values;
4086  PGresult *res;
4087  int n_rows;
4088  StringInfoData sql;
4089 
4090  /* The operation should be INSERT, UPDATE, or DELETE */
4091  Assert(operation == CMD_INSERT ||
4092  operation == CMD_UPDATE ||
4093  operation == CMD_DELETE);
4094 
4095  /* First, process a pending asynchronous request, if any. */
4096  if (fmstate->conn_state->pendingAreq)
4098 
4099  /*
4100  * If the existing query was deparsed and prepared for a different number
4101  * of rows, rebuild it for the proper number.
4102  */
4103  if (operation == CMD_INSERT && fmstate->num_slots != *numSlots)
4104  {
4105  /* Destroy the prepared statement created previously */
4106  if (fmstate->p_name)
4107  deallocate_query(fmstate);
4108 
4109  /* Build INSERT string with numSlots records in its VALUES clause. */
4110  initStringInfo(&sql);
4111  rebuildInsertSql(&sql, fmstate->rel,
4112  fmstate->orig_query, fmstate->target_attrs,
4113  fmstate->values_end, fmstate->p_nums,
4114  *numSlots - 1);
4115  pfree(fmstate->query);
4116  fmstate->query = sql.data;
4117  fmstate->num_slots = *numSlots;
4118  }
4119 
4120  /* Set up the prepared statement on the remote server, if we didn't yet */
4121  if (!fmstate->p_name)
4122  prepare_foreign_modify(fmstate);
4123 
4124  /*
4125  * For UPDATE/DELETE, get the ctid that was passed up as a resjunk column
4126  */
4127  if (operation == CMD_UPDATE || operation == CMD_DELETE)
4128  {
4129  Datum datum;
4130  bool isNull;
4131 
4132  datum = ExecGetJunkAttribute(planSlots[0],
4133  fmstate->ctidAttno,
4134  &isNull);
4135  /* shouldn't ever get a null result... */
4136  if (isNull)
4137  elog(ERROR, "ctid is NULL");
4138  ctid = (ItemPointer) DatumGetPointer(datum);
4139  }
4140 
4141  /* Convert parameters needed by prepared statement to text form */
4142  p_values = convert_prep_stmt_params(fmstate, ctid, slots, *numSlots);
4143 
4144  /*
4145  * Execute the prepared statement.
4146  */
4147  if (!PQsendQueryPrepared(fmstate->conn,
4148  fmstate->p_name,
4149  fmstate->p_nums * (*numSlots),
4150  p_values,
4151  NULL,
4152  NULL,
4153  0))
4154  pgfdw_report_error(ERROR, NULL, fmstate->conn, false, fmstate->query);
4155 
4156  /*
4157  * Get the result, and check for success.
4158  *
4159  * We don't use a PG_TRY block here, so be careful not to throw error
4160  * without releasing the PGresult.
4161  */
4162  res = pgfdw_get_result(fmstate->conn);
4163  if (PQresultStatus(res) !=
4165  pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query);
4166 
4167  /* Check number of rows affected, and fetch RETURNING tuple if any */
4168  if (fmstate->has_returning)
4169  {
4170  Assert(*numSlots == 1);
4171  n_rows = PQntuples(res);
4172  if (n_rows > 0)
4173  store_returning_result(fmstate, slots[0], res);
4174  }
4175  else
4176  n_rows = atoi(PQcmdTuples(res));
4177 
4178  /* And clean up */
4179  PQclear(res);
4180 
4181  MemoryContextReset(fmstate->temp_cxt);
4182 
4183  *numSlots = n_rows;
4184 
4185  /*
4186  * Return NULL if nothing was inserted/updated/deleted on the remote end
4187  */
4188  return (n_rows > 0) ? slots : NULL;
4189 }
4190 
4191 /*
4192  * prepare_foreign_modify
4193  * Establish a prepared statement for execution of INSERT/UPDATE/DELETE
4194  */
4195 static void
4197 {
4198  char prep_name[NAMEDATALEN];
4199  char *p_name;
4200  PGresult *res;
4201 
4202  /*
4203  * The caller would already have processed a pending asynchronous request
4204  * if any, so no need to do it here.
4205  */
4206 
4207  /* Construct name we'll use for the prepared statement. */
4208  snprintf(prep_name, sizeof(prep_name), "pgsql_fdw_prep_%u",
4209  GetPrepStmtNumber(fmstate->conn));
4210  p_name = pstrdup(prep_name);
4211 
4212  /*
4213  * We intentionally do not specify parameter types here, but leave the
4214  * remote server to derive them by default. This avoids possible problems
4215  * with the remote server using different type OIDs than we do. All of
4216  * the prepared statements we use in this module are simple enough that
4217  * the remote server will make the right choices.
4218  */
4219  if (!PQsendPrepare(fmstate->conn,
4220  p_name,
4221  fmstate->query,
4222  0,
4223  NULL))
4224  pgfdw_report_error(ERROR, NULL, fmstate->conn, false, fmstate->query);
4225 
4226  /*
4227  * Get the result, and check for success.
4228  *
4229  * We don't use a PG_TRY block here, so be careful not to throw error
4230  * without releasing the PGresult.
4231  */
4232  res = pgfdw_get_result(fmstate->conn);
4234  pgfdw_report_error(ERROR, res, fmstate->conn, true, fmstate->query);
4235  PQclear(res);
4236 
4237  /* This action shows that the prepare has been done. */
4238  fmstate->p_name = p_name;
4239 }
4240 
4241 /*
4242  * convert_prep_stmt_params
4243  * Create array of text strings representing parameter values
4244  *
4245  * tupleid is ctid to send, or NULL if none
4246  * slot is slot to get remaining parameters from, or NULL if none
4247  *
4248  * Data is constructed in temp_cxt; caller should reset that after use.
4249  */
4250 static const char **
4252  ItemPointer tupleid,
4253  TupleTableSlot **slots,
4254  int numSlots)
4255 {
4256  const char **p_values;
4257  int i;
4258  int j;
4259  int pindex = 0;
4260  MemoryContext oldcontext;
4261 
4262  oldcontext = MemoryContextSwitchTo(fmstate->temp_cxt);
4263 
4264  p_values = (const char **) palloc(sizeof(char *) * fmstate->p_nums * numSlots);
4265 
4266  /* ctid is provided only for UPDATE/DELETE, which don't allow batching */
4267  Assert(!(tupleid != NULL && numSlots > 1));
4268 
4269  /* 1st parameter should be ctid, if it's in use */
4270  if (tupleid != NULL)
4271  {
4272  Assert(numSlots == 1);
4273  /* don't need set_transmission_modes for TID output */
4274  p_values[pindex] = OutputFunctionCall(&fmstate->p_flinfo[pindex],
4275  PointerGetDatum(tupleid));
4276  pindex++;
4277  }
4278 
4279  /* get following parameters from slots */
4280  if (slots != NULL && fmstate->target_attrs != NIL)
4281  {
4282  TupleDesc tupdesc = RelationGetDescr(fmstate->rel);
4283  int nestlevel;
4284  ListCell *lc;
4285 
4286  nestlevel = set_transmission_modes();
4287 
4288  for (i = 0; i < numSlots; i++)
4289  {
4290  j = (tupleid != NULL) ? 1 : 0;
4291  foreach(lc, fmstate->target_attrs)
4292  {
4293  int attnum = lfirst_int(lc);
4294  Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
4295  Datum value;
4296  bool isnull;
4297 
4298  /* Ignore generated columns; they are set to DEFAULT */
4299  if (attr->attgenerated)
4300  continue;
4301  value = slot_getattr(slots[i], attnum, &isnull);
4302  if (isnull)
4303  p_values[pindex] = NULL;
4304  else
4305  p_values[pindex] = OutputFunctionCall(&fmstate->p_flinfo[j],
4306  value);
4307  pindex++;
4308  j++;
4309  }
4310  }
4311 
4312  reset_transmission_modes(nestlevel);
4313  }
4314 
4315  Assert(pindex == fmstate->p_nums * numSlots);
4316 
4317  MemoryContextSwitchTo(oldcontext);
4318 
4319  return p_values;
4320 }
4321 
4322 /*
4323  * store_returning_result
4324  * Store the result of a RETURNING clause
4325  *
4326  * On error, be sure to release the PGresult on the way out. Callers do not
4327  * have PG_TRY blocks to ensure this happens.
4328  */
4329 static void
4331  TupleTableSlot *slot, PGresult *res)
4332 {
4333  PG_TRY();
4334  {
4335  HeapTuple newtup;
4336 
4337  newtup = make_tuple_from_result_row(res, 0,
4338  fmstate->rel,
4339  fmstate->attinmeta,
4340  fmstate->retrieved_attrs,
4341  NULL,
4342  fmstate->temp_cxt);
4343 
4344  /*
4345  * The returning slot will not necessarily be suitable to store
4346  * heaptuples directly, so allow for conversion.
4347  */
4348  ExecForceStoreHeapTuple(newtup, slot, true);
4349  }
4350  PG_CATCH();
4351  {
4352  PQclear(res);
4353  PG_RE_THROW();
4354  }
4355  PG_END_TRY();
4356 }
4357 
4358 /*
4359  * finish_foreign_modify
4360  * Release resources for a foreign insert/update/delete operation
4361  */
4362 static void
4364 {
4365  Assert(fmstate != NULL);
4366 
4367  /* If we created a prepared statement, destroy it */
4368  deallocate_query(fmstate);
4369 
4370  /* Release remote connection */
4371  ReleaseConnection(fmstate->conn);
4372  fmstate->conn = NULL;
4373 }
4374 
4375 /*
4376  * deallocate_query
4377  * Deallocate a prepared statement for a foreign insert/update/delete
4378  * operation
4379  */
4380 static void
4382 {
4383  char sql[64];
4384  PGresult *res;
4385 
4386  /* do nothing if the query is not allocated */
4387  if (!fmstate->p_name)
4388  return;
4389 
4390  snprintf(sql, sizeof(sql), "DEALLOCATE %s", fmstate->p_name);
4391 
4392  /*
4393  * We don't use a PG_TRY block here, so be careful not to throw error
4394  * without releasing the PGresult.
4395  */
4396  res = pgfdw_exec_query(fmstate->conn, sql, fmstate->conn_state);
4398  pgfdw_report_error(ERROR, res, fmstate->conn, true, sql);
4399  PQclear(res);
4400  pfree(fmstate->p_name);
4401  fmstate->p_name = NULL;
4402 }
4403 
4404 /*
4405  * build_remote_returning
4406  * Build a RETURNING targetlist of a remote query for performing an
4407  * UPDATE/DELETE .. RETURNING on a join directly
4408  */
4409 static List *
4410 build_remote_returning(Index rtindex, Relation rel, List *returningList)
4411 {
4412  bool have_wholerow = false;
4413  List *tlist = NIL;
4414  List *vars;
4415  ListCell *lc;
4416 
4417  Assert(returningList);
4418 
4419  vars = pull_var_clause((Node *) returningList, PVC_INCLUDE_PLACEHOLDERS);
4420 
4421  /*
4422  * If there's a whole-row reference to the target relation, then we'll
4423  * need all the columns of the relation.
4424  */
4425  foreach(lc, vars)
4426  {
4427  Var *var = (Var *) lfirst(lc);
4428 
4429  if (IsA(var, Var) &&
4430  var->varno == rtindex &&
4431  var->varattno == InvalidAttrNumber)
4432  {
4433  have_wholerow = true;
4434  break;
4435  }
4436  }
4437 
4438  if (have_wholerow)
4439  {
4440  TupleDesc tupdesc = RelationGetDescr(rel);
4441  int i;
4442 
4443  for (i = 1; i <= tupdesc->natts; i++)
4444  {
4445  Form_pg_attribute attr = TupleDescAttr(tupdesc, i - 1);
4446  Var *var;
4447 
4448  /* Ignore dropped attributes. */
4449  if (attr->attisdropped)
4450  continue;
4451 
4452  var = makeVar(rtindex,
4453  i,
4454  attr->atttypid,
4455  attr->atttypmod,
4456  attr->attcollation,
4457  0);
4458 
4459  tlist = lappend(tlist,
4460  makeTargetEntry((Expr *) var,
4461  list_length(tlist) + 1,
4462  NULL,
4463  false));
4464  }
4465  }
4466 
4467  /* Now add any remaining columns to tlist. */
4468  foreach(lc, vars)
4469  {
4470  Var *var = (Var *) lfirst(lc);
4471 
4472  /*
4473  * No need for whole-row references to the target relation. We don't
4474  * need system columns other than ctid and oid either, since those are
4475  * set locally.
4476  */
4477  if (IsA(var, Var) &&
4478  var->varno == rtindex &&
4479  var->varattno <= InvalidAttrNumber &&
4481  continue; /* don't need it */
4482 
4483  if (tlist_member((Expr *) var, tlist))
4484  continue; /* already got it */
4485 
4486  tlist = lappend(tlist,
4487  makeTargetEntry((Expr *) var,
4488  list_length(tlist) + 1,
4489  NULL,
4490  false));
4491  }
4492 
4493  list_free(vars);
4494 
4495  return tlist;
4496 }
4497 
4498 /*
4499  * rebuild_fdw_scan_tlist
4500  * Build new fdw_scan_tlist of given foreign-scan plan node from given
4501  * tlist
4502  *
4503  * There might be columns that the fdw_scan_tlist of the given foreign-scan
4504  * plan node contains that the given tlist doesn't. The fdw_scan_tlist would
4505  * have contained resjunk columns such as 'ctid' of the target relation and
4506  * 'wholerow' of non-target relations, but the tlist might not contain them,
4507  * for example. So, adjust the tlist so it contains all the columns specified
4508  * in the fdw_scan_tlist; else setrefs.c will get confused.
4509  */
4510 static void
4512 {
4513  List *new_tlist = tlist;
4514  List *old_tlist = fscan->fdw_scan_tlist;
4515  ListCell *lc;
4516 
4517  foreach(lc, old_tlist)
4518  {
4519  TargetEntry *tle = (TargetEntry *) lfirst(lc);
4520 
4521  if (tlist_member(tle->expr, new_tlist))
4522  continue; /* already got it */
4523 
4524  new_tlist = lappend(new_tlist,
4525  makeTargetEntry(tle->expr,
4526  list_length(new_tlist) + 1,
4527  NULL,
4528  false));
4529  }
4530  fscan->fdw_scan_tlist = new_tlist;
4531 }
4532 
4533 /*
4534  * Execute a direct UPDATE/DELETE statement.
4535  */
4536 static void
4538 {
4540  ExprContext *econtext = node->ss.ps.ps_ExprContext;
4541  int numParams = dmstate->numParams;
4542  const char **values = dmstate->param_values;
4543 
4544  /* First, process a pending asynchronous request, if any. */
4545  if (dmstate->conn_state->pendingAreq)
4547 
4548  /*
4549  * Construct array of query parameter values in text format.
4550  */
4551  if (numParams > 0)
4552  process_query_params(econtext,
4553  dmstate->param_flinfo,
4554  dmstate->param_exprs,
4555  values);
4556 
4557  /*
4558  * Notice that we pass NULL for paramTypes, thus forcing the remote server
4559  * to infer types for all parameters. Since we explicitly cast every
4560  * parameter (see deparse.c), the "inference" is trivial and will produce
4561  * the desired result. This allows us to avoid assuming that the remote
4562  * server has the same OIDs we do for the parameters' types.
4563  */
4564  if (!PQsendQueryParams(dmstate->conn, dmstate->query, numParams,
4565  NULL, values, NULL, NULL, 0))
4566  pgfdw_report_error(ERROR, NULL, dmstate->conn, false, dmstate->query);
4567 
4568  /*
4569  * Get the result, and check for success.
4570  *
4571  * We don't use a PG_TRY block here, so be careful not to throw error
4572  * without releasing the PGresult.
4573  */
4574  dmstate->result = pgfdw_get_result(dmstate->conn);
4575  if (PQresultStatus(dmstate->result) !=
4577  pgfdw_report_error(ERROR, dmstate->result, dmstate->conn, true,
4578  dmstate->query);
4579 
4580  /* Get the number of rows affected. */
4581  if (dmstate->has_returning)
4582  dmstate->num_tuples = PQntuples(dmstate->result);
4583  else
4584  dmstate->num_tuples = atoi(PQcmdTuples(dmstate->result));
4585 }
4586 
4587 /*
4588  * Get the result of a RETURNING clause.
4589  */
4590 static TupleTableSlot *
4592 {
4594  EState *estate = node->ss.ps.state;
4595  ResultRelInfo *resultRelInfo = node->resultRelInfo;
4596  TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
4597  TupleTableSlot *resultSlot;
4598 
4599  Assert(resultRelInfo->ri_projectReturning);
4600 
4601  /* If we didn't get any tuples, must be end of data. */
4602  if (dmstate->next_tuple >= dmstate->num_tuples)
4603  return ExecClearTuple(slot);
4604 
4605  /* Increment the command es_processed count if necessary. */
4606  if (dmstate->set_processed)
4607  estate->es_processed += 1;
4608 
4609  /*
4610  * Store a RETURNING tuple. If has_returning is false, just emit a dummy
4611  * tuple. (has_returning is false when the local query is of the form
4612  * "UPDATE/DELETE .. RETURNING 1" for example.)
4613  */
4614  if (!dmstate->has_returning)
4615  {
4616  ExecStoreAllNullTuple(slot);
4617  resultSlot = slot;
4618  }
4619  else
4620  {
4621  /*
4622  * On error, be sure to release the PGresult on the way out. Callers
4623  * do not have PG_TRY blocks to ensure this happens.
4624  */
4625  PG_TRY();
4626  {
4627  HeapTuple newtup;
4628 
4629  newtup = make_tuple_from_result_row(dmstate->result,
4630  dmstate->next_tuple,
4631  dmstate->rel,
4632  dmstate->attinmeta,
4633  dmstate->retrieved_attrs,
4634  node,
4635  dmstate->temp_cxt);
4636  ExecStoreHeapTuple(newtup, slot, false);
4637  }
4638  PG_CATCH();
4639  {
4640  PQclear(dmstate->result);
4641  PG_RE_THROW();
4642  }
4643  PG_END_TRY();
4644 
4645  /* Get the updated/deleted tuple. */
4646  if (dmstate->rel)
4647  resultSlot = slot;
4648  else
4649  resultSlot = apply_returning_filter(dmstate, resultRelInfo, slot, estate);
4650  }
4651  dmstate->next_tuple++;
4652 
4653  /* Make slot available for evaluation of the local query RETURNING list. */
4654  resultRelInfo->ri_projectReturning->pi_exprContext->ecxt_scantuple =
4655  resultSlot;
4656 
4657  return slot;
4658 }
4659 
4660 /*
4661  * Initialize a filter to extract an updated/deleted tuple from a scan tuple.
4662  */
4663 static void
4665  List *fdw_scan_tlist,
4666  Index rtindex)
4667 {
4668  TupleDesc resultTupType = RelationGetDescr(dmstate->resultRel);
4669  ListCell *lc;
4670  int i;
4671 
4672  /*
4673  * Calculate the mapping between the fdw_scan_tlist's entries and the
4674  * result tuple's attributes.
4675  *
4676  * The "map" is an array of indexes of the result tuple's attributes in
4677  * fdw_scan_tlist, i.e., one entry for every attribute of the result
4678  * tuple. We store zero for any attributes that don't have the
4679  * corresponding entries in that list, marking that a NULL is needed in
4680  * the result tuple.
4681  *
4682  * Also get the indexes of the entries for ctid and oid if any.
4683  */
4684  dmstate->attnoMap = (AttrNumber *)
4685  palloc0(resultTupType->natts * sizeof(AttrNumber));
4686 
4687  dmstate->ctidAttno = dmstate->oidAttno = 0;
4688 
4689  i = 1;
4690  dmstate->hasSystemCols = false;
4691  foreach(lc, fdw_scan_tlist)
4692  {
4693  TargetEntry *tle = (TargetEntry *) lfirst(lc);
4694  Var *var = (Var *) tle->expr;
4695 
4696  Assert(IsA(var, Var));
4697 
4698  /*
4699  * If the Var is a column of the target relation to be retrieved from
4700  * the foreign server, get the index of the entry.
4701  */
4702  if (var->varno == rtindex &&
4703  list_member_int(dmstate->retrieved_attrs, i))
4704  {
4705  int attrno = var->varattno;
4706 
4707  if (attrno < 0)
4708  {
4709  /*
4710  * We don't retrieve system columns other than ctid and oid.
4711  */
4712  if (attrno == SelfItemPointerAttributeNumber)
4713  dmstate->ctidAttno = i;
4714  else
4715  Assert(false);
4716  dmstate->hasSystemCols = true;
4717  }
4718  else
4719  {
4720  /*
4721  * We don't retrieve whole-row references to the target
4722  * relation either.
4723  */
4724  Assert(attrno > 0);
4725 
4726  dmstate->attnoMap[attrno - 1] = i;
4727  }
4728  }
4729  i++;
4730  }
4731 }
4732 
4733 /*
4734  * Extract and return an updated/deleted tuple from a scan tuple.
4735  */
4736 static TupleTableSlot *
4738  ResultRelInfo *resultRelInfo,
4739  TupleTableSlot *slot,
4740  EState *estate)
4741 {
4742  TupleDesc resultTupType = RelationGetDescr(dmstate->resultRel);
4743  TupleTableSlot *resultSlot;
4744  Datum *values;
4745  bool *isnull;
4746  Datum *old_values;
4747  bool *old_isnull;
4748  int i;
4749 
4750  /*
4751  * Use the return tuple slot as a place to store the result tuple.
4752  */
4753  resultSlot = ExecGetReturningSlot(estate, resultRelInfo);
4754 
4755  /*
4756  * Extract all the values of the scan tuple.
4757  */
4758  slot_getallattrs(slot);
4759  old_values = slot->tts_values;
4760  old_isnull = slot->tts_isnull;
4761 
4762  /*
4763  * Prepare to build the result tuple.
4764  */
4765  ExecClearTuple(resultSlot);
4766  values = resultSlot->tts_values;
4767  isnull = resultSlot->tts_isnull;
4768 
4769  /*
4770  * Transpose data into proper fields of the result tuple.
4771  */
4772  for (i = 0; i < resultTupType->natts; i++)
4773  {
4774  int j = dmstate->attnoMap[i];
4775 
4776  if (j == 0)
4777  {
4778  values[i] = (Datum) 0;
4779  isnull[i] = true;
4780  }
4781  else
4782  {
4783  values[i] = old_values[j - 1];
4784  isnull[i] = old_isnull[j - 1];
4785  }
4786  }
4787 
4788  /*
4789  * Build the virtual tuple.
4790  */
4791  ExecStoreVirtualTuple(resultSlot);
4792 
4793  /*
4794  * If we have any system columns to return, materialize a heap tuple in
4795  * the slot from column values set above and install system columns in
4796  * that tuple.
4797  */
4798  if (dmstate->hasSystemCols)
4799  {
4800  HeapTuple resultTup = ExecFetchSlotHeapTuple(resultSlot, true, NULL);
4801 
4802  /* ctid */
4803  if (dmstate->ctidAttno)
4804  {
4805  ItemPointer ctid = NULL;
4806 
4807  ctid = (ItemPointer) DatumGetPointer(old_values[dmstate->ctidAttno - 1]);
4808  resultTup->t_self = *ctid;
4809  }
4810 
4811  /*
4812  * And remaining columns
4813  *
4814  * Note: since we currently don't allow the target relation to appear
4815  * on the nullable side of an outer join, any system columns wouldn't
4816  * go to NULL.
4817  *
4818  * Note: no need to care about tableoid here because it will be
4819  * initialized in ExecProcessReturning().
4820  */
4824  }
4825 
4826  /*
4827  * And return the result tuple.
4828  */
4829  return resultSlot;
4830 }
4831 
4832 /*
4833  * Prepare for processing of parameters used in remote query.
4834  */
4835 static void
4837  List *fdw_exprs,
4838  int numParams,
4839  FmgrInfo **param_flinfo,
4840  List **param_exprs,
4841  const char ***param_values)
4842 {
4843  int i;
4844  ListCell *lc;
4845 
4846  Assert(numParams > 0);
4847 
4848  /* Prepare for output conversion of parameters used in remote query. */
4849  *param_flinfo = (FmgrInfo *) palloc0(sizeof(FmgrInfo) * numParams);
4850 
4851  i = 0;
4852  foreach(lc, fdw_exprs)
4853  {
4854  Node *param_expr = (Node *) lfirst(lc);
4855  Oid typefnoid;
4856  bool isvarlena;
4857 
4858  getTypeOutputInfo(exprType(param_expr), &typefnoid, &isvarlena);
4859  fmgr_info(typefnoid, &(*param_flinfo)[i]);
4860  i++;
4861  }
4862 
4863  /*
4864  * Prepare remote-parameter expressions for evaluation. (Note: in
4865  * practice, we expect that all these expressions will be just Params, so
4866  * we could possibly do something more efficient than using the full
4867  * expression-eval machinery for this. But probably there would be little
4868  * benefit, and it'd require postgres_fdw to know more than is desirable
4869  * about Param evaluation.)
4870  */
4871  *param_exprs = ExecInitExprList(fdw_exprs, node);
4872 
4873  /* Allocate buffer for text form of query parameters. */
4874  *param_values = (const char **) palloc0(numParams * sizeof(char *));
4875 }
4876 
4877 /*
4878  * Construct array of query parameter values in text format.
4879  */
4880 static void
4882  FmgrInfo *param_flinfo,
4883  List *param_exprs,
4884  const char **param_values)
4885 {
4886  int nestlevel;
4887  int i;
4888  ListCell *lc;
4889 
4890  nestlevel = set_transmission_modes();
4891 
4892  i = 0;
4893  foreach(lc, param_exprs)
4894  {
4895  ExprState *expr_state = (ExprState *) lfirst(lc);
4896  Datum expr_value;
4897  bool isNull;
4898 
4899  /* Evaluate the parameter expression */
4900  expr_value = ExecEvalExpr(expr_state, econtext, &isNull);
4901 
4902  /*
4903  * Get string representation of each parameter value by invoking
4904  * type-specific output function, unless the value is null.
4905  */
4906  if (isNull)
4907  param_values[i] = NULL;
4908  else
4909  param_values[i] = OutputFunctionCall(&param_flinfo[i], expr_value);
4910 
4911  i++;
4912  }
4913 
4914  reset_transmission_modes(nestlevel);
4915 }
4916 
4917 /*
4918  * postgresAnalyzeForeignTable
4919  * Test whether analyzing this foreign table is supported
4920  */
4921 static bool
4923  AcquireSampleRowsFunc *func,
4924  BlockNumber *totalpages)
4925 {
4926  ForeignTable *table;
4927  UserMapping *user;
4928  PGconn *conn;
4929  StringInfoData sql;
4930  PGresult *volatile res = NULL;
4931 
4932  /* Return the row-analysis function pointer */
4934 
4935  /*
4936  * Now we have to get the number of pages. It's annoying that the ANALYZE
4937  * API requires us to return that now, because it forces some duplication
4938  * of effort between this routine and postgresAcquireSampleRowsFunc. But
4939  * it's probably not worth redefining that API at this point.
4940  */
4941 
4942  /*
4943  * Get the connection to use. We do the remote access as the table's
4944  * owner, even if the ANALYZE was started by some other user.
4945  */
4946  table = GetForeignTable(RelationGetRelid(relation));
4947  user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
4948  conn = GetConnection(user, false, NULL);
4949 
4950  /*
4951  * Construct command to get page count for relation.
4952  */
4953  initStringInfo(&sql);
4954  deparseAnalyzeSizeSql(&sql, relation);
4955 
4956  /* In what follows, do not risk leaking any PGresults. */
4957  PG_TRY();
4958  {
4959  res = pgfdw_exec_query(conn, sql.data, NULL);
4961  pgfdw_report_error(ERROR, res, conn, false, sql.data);
4962 
4963  if (PQntuples(res) != 1 || PQnfields(res) != 1)
4964  elog(ERROR, "unexpected result from deparseAnalyzeSizeSql query");
4965  *totalpages = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
4966  }
4967  PG_FINALLY();
4968  {
4969  PQclear(res);
4970  }
4971  PG_END_TRY();
4972 
4974 
4975  return true;
4976 }
4977 
4978 /*
4979  * postgresGetAnalyzeInfoForForeignTable
4980  * Count tuples in foreign table (just get pg_class.reltuples).
4981  *
4982  * can_tablesample determines if the remote relation supports acquiring the
4983  * sample using TABLESAMPLE.
4984  */
4985 static double
4986 postgresGetAnalyzeInfoForForeignTable(Relation relation, bool *can_tablesample)
4987 {
4988  ForeignTable *table;
4989  UserMapping *user;
4990  PGconn *conn;
4991  StringInfoData sql;
4992  PGresult *volatile res = NULL;
4993  volatile double reltuples = -1;
4994  volatile char relkind = 0;
4995 
4996  /* assume the remote relation does not support TABLESAMPLE */
4997  *can_tablesample = false;
4998 
4999  /*
5000  * Get the connection to use. We do the remote access as the table's
5001  * owner, even if the ANALYZE was started by some other user.
5002  */
5003  table = GetForeignTable(RelationGetRelid(relation));
5004  user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
5005  conn = GetConnection(user, false, NULL);
5006 
5007  /*
5008  * Construct command to get page count for relation.
5009  */
5010  initStringInfo(&sql);
5011  deparseAnalyzeInfoSql(&sql, relation);
5012 
5013  /* In what follows, do not risk leaking any PGresults. */
5014  PG_TRY();
5015  {
5016  res = pgfdw_exec_query(conn, sql.data, NULL);
5018  pgfdw_report_error(ERROR, res, conn, false, sql.data);
5019 
5020  if (PQntuples(res) != 1 || PQnfields(res) != 2)
5021  elog(ERROR, "unexpected result from deparseAnalyzeInfoSql query");
5022  reltuples = strtod(PQgetvalue(res, 0, 0), NULL);
5023  relkind = *(PQgetvalue(res, 0, 1));
5024  }
5025  PG_FINALLY();
5026  {
5027  if (res)
5028  PQclear(res);
5029  }
5030  PG_END_TRY();
5031 
5033 
5034  /* TABLESAMPLE is supported only for regular tables and matviews */
5035  *can_tablesample = (relkind == RELKIND_RELATION ||
5036  relkind == RELKIND_MATVIEW ||
5037  relkind == RELKIND_PARTITIONED_TABLE);
5038 
5039  return reltuples;
5040 }
5041 
5042 /*
5043  * Acquire a random sample of rows from foreign table managed by postgres_fdw.
5044  *
5045  * Selected rows are returned in the caller-allocated array rows[],
5046  * which must have at least targrows entries.
5047  * The actual number of rows selected is returned as the function result.
5048  * We also count the total number of rows in the table and return it into
5049  * *totalrows. Note that *totaldeadrows is always set to 0.
5050  *
5051  * Note that the returned list of rows is not always in order by physical
5052  * position in the table. Therefore, correlation estimates derived later
5053  * may be meaningless, but it's OK because we don't use the estimates
5054  * currently (the planner only pays attention to correlation for indexscans).
5055  */
5056 static int
5058  HeapTuple *rows, int targrows,
5059  double *totalrows,
5060  double *totaldeadrows)
5061 {
5062  PgFdwAnalyzeState astate;
5063  ForeignTable *table;
5064  ForeignServer *server;
5065  UserMapping *user;
5066  PGconn *conn;
5067  int server_version_num;
5068  PgFdwSamplingMethod method = ANALYZE_SAMPLE_AUTO; /* auto is default */
5069  double sample_frac = -1.0;
5070  double reltuples;
5071  unsigned int cursor_number;
5072  StringInfoData sql;
5073  PGresult *volatile res = NULL;
5074  ListCell *lc;
5075 
5076  /* Initialize workspace state */
5077  astate.rel = relation;
5079 
5080  astate.rows = rows;
5081  astate.targrows = targrows;
5082  astate.numrows = 0;
5083  astate.samplerows = 0;
5084  astate.rowstoskip = -1; /* -1 means not set yet */
5085  reservoir_init_selection_state(&astate.rstate, targrows);
5086 
5087  /* Remember ANALYZE context, and create a per-tuple temp context */
5088  astate.anl_cxt = CurrentMemoryContext;
5090  "postgres_fdw temporary data",
5092 
5093  /*
5094  * Get the connection to use. We do the remote access as the table's
5095  * owner, even if the ANALYZE was started by some other user.
5096  */
5097  table = GetForeignTable(RelationGetRelid(relation));
5098  server = GetForeignServer(table->serverid);
5099  user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
5100  conn = GetConnection(user, false, NULL);
5101 
5102  /* We'll need server version, so fetch it now. */
5104 
5105  /*
5106  * What sampling method should we use?
5107  */
5108  foreach(lc, server->options)
5109  {
5110  DefElem *def = (DefElem *) lfirst(lc);
5111 
5112  if (strcmp(def->defname, "analyze_sampling") == 0)
5113  {
5114  char *value = defGetString(def);
5115 
5116  if (strcmp(value, "off") == 0)
5117  method = ANALYZE_SAMPLE_OFF;
5118  else if (strcmp(value, "auto") == 0)
5119  method = ANALYZE_SAMPLE_AUTO;
5120  else if (strcmp(value, "random") == 0)
5121  method = ANALYZE_SAMPLE_RANDOM;
5122  else if (strcmp(value, "system") == 0)
5123  method = ANALYZE_SAMPLE_SYSTEM;
5124  else if (strcmp(value, "bernoulli") == 0)
5125  method = ANALYZE_SAMPLE_BERNOULLI;
5126 
5127  break;
5128  }
5129  }
5130 
5131  foreach(lc, table->options)
5132  {
5133  DefElem *def = (DefElem *) lfirst(lc);
5134 
5135  if (strcmp(def->defname, "analyze_sampling") == 0)
5136  {
5137  char *value = defGetString(def);
5138 
5139  if (strcmp(value, "off") == 0)
5140  method = ANALYZE_SAMPLE_OFF;
5141  else if (strcmp(value, "auto") == 0)
5142  method = ANALYZE_SAMPLE_AUTO;
5143  else if (strcmp(value, "random") == 0)
5144  method = ANALYZE_SAMPLE_RANDOM;
5145  else if (strcmp(value, "system") == 0)
5146  method = ANALYZE_SAMPLE_SYSTEM;
5147  else if (strcmp(value, "bernoulli") == 0)
5148  method = ANALYZE_SAMPLE_BERNOULLI;
5149 
5150  break;
5151  }
5152  }
5153 
5154  /*
5155  * Error-out if explicitly required one of the TABLESAMPLE methods, but
5156  * the server does not support it.
5157  */
5158  if ((server_version_num < 95000) &&
5159  (method == ANALYZE_SAMPLE_SYSTEM ||
5160  method == ANALYZE_SAMPLE_BERNOULLI))
5161  ereport(ERROR,
5162  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
5163  errmsg("remote server does not support TABLESAMPLE feature")));
5164 
5165  /*
5166  * If we've decided to do remote sampling, calculate the sampling rate. We
5167  * need to get the number of tuples from the remote server, but skip that
5168  * network round-trip if not needed.
5169  */
5170  if (method != ANALYZE_SAMPLE_OFF)
5171  {
5172  bool can_tablesample;
5173 
5174  reltuples = postgresGetAnalyzeInfoForForeignTable(relation,
5175  &can_tablesample);
5176 
5177  /*
5178  * Make sure we're not choosing TABLESAMPLE when the remote relation
5179  * does not support that. But only do this for "auto" - if the user
5180  * explicitly requested BERNOULLI/SYSTEM, it's better to fail.
5181  */
5182  if (!can_tablesample && (method == ANALYZE_SAMPLE_AUTO))
5183  method = ANALYZE_SAMPLE_RANDOM;
5184 
5185  /*
5186  * Remote's reltuples could be 0 or -1 if the table has never been
5187  * vacuumed/analyzed. In that case, disable sampling after all.
5188  */
5189  if ((reltuples <= 0) || (targrows >= reltuples))
5190  method = ANALYZE_SAMPLE_OFF;
5191  else
5192  {
5193  /*
5194  * All supported sampling methods require sampling rate, not
5195  * target rows directly, so we calculate that using the remote
5196  * reltuples value. That's imperfect, because it might be off a
5197  * good deal, but that's not something we can (or should) address
5198  * here.
5199  *
5200  * If reltuples is too low (i.e. when table grew), we'll end up
5201  * sampling more rows - but then we'll apply the local sampling,
5202  * so we get the expected sample size. This is the same outcome as
5203  * without remote sampling.
5204  *
5205  * If reltuples is too high (e.g. after bulk DELETE), we will end
5206  * up sampling too few rows.
5207  *
5208  * We can't really do much better here - we could try sampling a
5209  * bit more rows, but we don't know how off the reltuples value is
5210  * so how much is "a bit more"?
5211  *
5212  * Furthermore, the targrows value for partitions is determined
5213  * based on table size (relpages), which can be off in different
5214  * ways too. Adjusting the sampling rate here might make the issue
5215  * worse.
5216  */
5217  sample_frac = targrows / reltuples;
5218 
5219  /*
5220  * We should never get sampling rate outside the valid range
5221  * (between 0.0 and 1.0), because those cases should be covered by
5222  * the previous branch that sets ANALYZE_SAMPLE_OFF.
5223  */
5224  Assert(sample_frac >= 0.0 && sample_frac <= 1.0);
5225  }
5226  }
5227 
5228  /*
5229  * For "auto" method, pick the one we believe is best. For servers with
5230  * TABLESAMPLE support we pick BERNOULLI, for old servers we fall-back to
5231  * random() to at least reduce network transfer.
5232  */
5233  if (method == ANALYZE_SAMPLE_AUTO)
5234  {
5235  if (server_version_num < 95000)
5236  method = ANALYZE_SAMPLE_RANDOM;
5237  else
5238  method = ANALYZE_SAMPLE_BERNOULLI;
5239  }
5240 
5241  /*
5242  * Construct cursor that retrieves whole rows from remote.
5243  */
5245  initStringInfo(&sql);
5246  appendStringInfo(&sql, "DECLARE c%u CURSOR FOR ", cursor_number);
5247 
5248  deparseAnalyzeSql(&sql, relation, method, sample_frac, &astate.retrieved_attrs);
5249 
5250  /* In what follows, do not risk leaking any PGresults. */
5251  PG_TRY();
5252  {
5253  char fetch_sql[64];
5254  int fetch_size;
5255 
5256  res = pgfdw_exec_query(conn, sql.data, NULL);
5258  pgfdw_report_error(ERROR, res, conn, false, sql.data);
5259  PQclear(res);
5260  res = NULL;
5261 
5262  /*
5263  * Determine the fetch size. The default is arbitrary, but shouldn't
5264  * be enormous.
5265  */
5266  fetch_size = 100;
5267  foreach(lc, server->options)
5268  {
5269  DefElem *def = (DefElem *) lfirst(lc);
5270 
5271  if (strcmp(def->defname, "fetch_size") == 0)
5272  {
5273  (void) parse_int(defGetString(def), &fetch_size, 0, NULL);
5274  break;
5275  }
5276  }
5277  foreach(lc, table->options)
5278  {
5279  DefElem *def = (DefElem *) lfirst(lc);
5280 
5281  if (strcmp(def->defname, "fetch_size") == 0)
5282  {
5283  (void) parse_int(defGetString(def), &fetch_size, 0, NULL);
5284  break;
5285  }
5286  }
5287 
5288  /* Construct command to fetch rows from remote. */
5289  snprintf(fetch_sql, sizeof(fetch_sql), "FETCH %d FROM c%u",
5291 
5292  /* Retrieve and process rows a batch at a time. */
5293  for (;;)
5294  {
5295  int numrows;
5296  int i;
5297 
5298  /* Allow users to cancel long query */
5300 
5301  /*
5302  * XXX possible future improvement: if rowstoskip is large, we
5303  * could issue a MOVE rather than physically fetching the rows,
5304  * then just adjust rowstoskip and samplerows appropriately.
5305  */
5306 
5307  /* Fetch some rows */
5308  res = pgfdw_exec_query(conn, fetch_sql, NULL);
5309  /* On error, report the original query, not the FETCH. */
5311  pgfdw_report_error(ERROR, res, conn, false, sql.data);
5312 
5313  /* Process whatever we got. */
5314  numrows = PQntuples(res);
5315  for (i = 0; i < numrows; i++)
5316  analyze_row_processor(res, i, &astate);
5317 
5318  PQclear(res);
5319  res = NULL;
5320 
5321  /* Must be EOF if we didn't get all the rows requested. */
5322  if (numrows < fetch_size)
5323  break;
5324  }
5325 
5326  /* Close the cursor, just to be tidy. */
5328  }
5329  PG_CATCH();
5330  {
5331  PQclear(res);
5332  PG_RE_THROW();
5333  }
5334  PG_END_TRY();
5335 
5337 
5338  /* We assume that we have no dead tuple. */
5339  *totaldeadrows = 0.0;
5340 
5341  /*
5342  * Without sampling, we've retrieved all living tuples from foreign
5343  * server, so report that as totalrows. Otherwise use the reltuples
5344  * estimate we got from the remote side.
5345  */
5346  if (method == ANALYZE_SAMPLE_OFF)
5347  *totalrows = astate.samplerows;
5348  else
5349  *totalrows = reltuples;
5350 
5351  /*
5352  * Emit some interesting relation info
5353  */
5354  ereport(elevel,
5355  (errmsg("\"%s\": table contains %.0f rows, %d rows in sample",
5356  RelationGetRelationName(relation),
5357  *totalrows, astate.numrows)));
5358 
5359  return astate.numrows;
5360 }
5361 
5362 /*
5363  * Collect sample rows from the result of query.
5364  * - Use all tuples in sample until target # of samples are collected.
5365  * - Subsequently, replace already-sampled tuples randomly.
5366  */
5367 static void
5369 {
5370  int targrows = astate->targrows;
5371  int pos; /* array index to store tuple in */
5372  MemoryContext oldcontext;
5373 
5374  /* Always increment sample row counter. */
5375  astate->samplerows += 1;
5376 
5377  /*
5378  * Determine the slot where this sample row should be stored. Set pos to
5379  * negative value to indicate the row should be skipped.
5380  */
5381  if (astate->numrows < targrows)
5382  {
5383  /* First targrows rows are always included into the sample */
5384  pos = astate->numrows++;
5385  }
5386  else
5387  {
5388  /*
5389  * Now we start replacing tuples in the sample until we reach the end
5390  * of the relation. Same algorithm as in acquire_sample_rows in
5391  * analyze.c; see Jeff Vitter's paper.
5392  */
5393  if (astate->rowstoskip < 0)
5394  astate->rowstoskip = reservoir_get_next_S(&astate->rstate, astate->samplerows, targrows);
5395 
5396  if (astate->rowstoskip <= 0)
5397  {
5398  /* Choose a random reservoir element to replace. */
5399  pos = (int) (targrows * sampler_random_fract(&astate->rstate.randstate));
5400  Assert(pos >= 0 && pos < targrows);
5401  heap_freetuple(astate->rows[pos]);
5402  }
5403  else
5404  {
5405  /* Skip this tuple. */
5406  pos = -1;
5407  }
5408 
5409  astate->rowstoskip -= 1;
5410  }
5411 
5412  if (pos >= 0)
5413  {
5414  /*
5415  * Create sample tuple from current result row, and store it in the
5416  * position determined above. The tuple has to be created in anl_cxt.
5417  */
5418  oldcontext = MemoryContextSwitchTo(astate->anl_cxt);
5419 
5420  astate->rows[pos] = make_tuple_from_result_row(res, row,
5421  astate->rel,
5422  astate->attinmeta,
5423  astate->retrieved_attrs,
5424  NULL,
5425  astate->temp_cxt);
5426 
5427  MemoryContextSwitchTo(oldcontext);
5428  }
5429 }
5430 
5431 /*
5432  * Import a foreign schema
5433  */
5434 static List *
5436 {
5437  List *commands = NIL;
5438  bool import_collate = true;
5439  bool import_default = false;
5440  bool import_generated = true;
5441  bool import_not_null = true;
5442  ForeignServer *server;
5443  UserMapping *mapping;
5444  PGconn *conn;
5446  PGresult *volatile res = NULL;
5447  int numrows,
5448  i;
5449  ListCell *lc;
5450 
5451  /* Parse statement options */
5452  foreach(lc, stmt->options)
5453  {
5454  DefElem *def = (DefElem *) lfirst(lc);
5455 
5456  if (strcmp(def->defname, "import_collate") == 0)
5457  import_collate = defGetBoolean(def);
5458  else if (strcmp(def->defname, "import_default") == 0)
5459  import_default = defGetBoolean(def);
5460  else if (strcmp(def->defname, "import_generated") == 0)
5461  import_generated = defGetBoolean(def);
5462  else if (strcmp(def->defname, "import_not_null") == 0)
5463  import_not_null = defGetBoolean(def);
5464  else
5465  ereport(ERROR,
5466  (errcode(ERRCODE_FDW_INVALID_OPTION_NAME),
5467  errmsg("invalid option \"%s\"", def->defname)));
5468  }
5469 
5470  /*
5471  * Get connection to the foreign server. Connection manager will
5472  * establish new connection if necessary.
5473  */
5474  server = GetForeignServer(serverOid);
5475  mapping = GetUserMapping(GetUserId(), server->serverid);
5476  conn = GetConnection(mapping, false, NULL);
5477 
5478  /* Don't attempt to import collation if remote server hasn't got it */
5479  if (PQserverVersion(conn) < 90100)
5480  import_collate = false;
5481 
5482  /* Create workspace for strings */
5483  initStringInfo(&buf);
5484 
5485  /* In what follows, do not risk leaking any PGresults. */
5486  PG_TRY();
5487  {
5488  /* Check that the schema really exists */
5489  appendStringInfoString(&buf, "SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = ");
5490  deparseStringLiteral(&buf, stmt->remote_schema);
5491 
5492  res = pgfdw_exec_query(conn, buf.data, NULL);
5494  pgfdw_report_error(ERROR, res, conn, false, buf.data);
5495 
5496  if (PQntuples(res) != 1)
5497  ereport(ERROR,
5498  (errcode(ERRCODE_FDW_SCHEMA_NOT_FOUND),
5499  errmsg("schema \"%s\" is not present on foreign server \"%s\"",
5500  stmt->remote_schema, server->servername)));
5501 
5502  PQclear(res);
5503  res = NULL;
5504  resetStringInfo(&buf);
5505 
5506  /*
5507  * Fetch all table data from this schema, possibly restricted by
5508  * EXCEPT or LIMIT TO. (We don't actually need to pay any attention
5509  * to EXCEPT/LIMIT TO here, because the core code will filter the
5510  * statements we return according to those lists anyway. But it
5511  * should save a few cycles to not process excluded tables in the
5512  * first place.)
5513  *
5514  * Import table data for partitions only when they are explicitly
5515  * specified in LIMIT TO clause. Otherwise ignore them and only
5516  * include the definitions of the root partitioned tables to allow
5517  * access to the complete remote data set locally in the schema
5518  * imported.
5519  *
5520  * Note: because we run the connection with search_path restricted to
5521  * pg_catalog, the format_type() and pg_get_expr() outputs will always
5522  * include a schema name for types/functions in other schemas, which
5523  * is what we want.
5524  */
5526  "SELECT relname, "
5527  " attname, "
5528  " format_type(atttypid, atttypmod), "
5529  " attnotnull, "
5530  " pg_get_expr(adbin, adrelid), ");
5531 
5532  /* Generated columns are supported since Postgres 12 */
5533  if (PQserverVersion(conn) >= 120000)
5535