postgres_fdw.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * postgres_fdw.c
 *        Foreign-data wrapper for remote PostgreSQL servers
 *
 * Portions Copyright (c) 2012-2026, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *        contrib/postgres_fdw/postgres_fdw.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"
 
#include <limits.h>
 
#include "access/htup_details.h"
#include "access/sysattr.h"
#include "access/table.h"
#include "catalog/pg_opfamily.h"
#include "commands/defrem.h"
#include "commands/explain_format.h"
#include "commands/explain_state.h"
#include "commands/vacuum.h"
#include "executor/execAsync.h"
#include "executor/instrument.h"
#include "executor/spi.h"
#include "foreign/fdwapi.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "optimizer/appendinfo.h"
#include "optimizer/cost.h"
#include "optimizer/inherit.h"
#include "optimizer/optimizer.h"
#include "optimizer/pathnode.h"
#include "optimizer/paths.h"
#include "optimizer/planmain.h"
#include "optimizer/prep.h"
#include "optimizer/restrictinfo.h"
#include "optimizer/tlist.h"
#include "parser/parsetree.h"
#include "postgres_fdw.h"
#include "statistics/statistics.h"
#include "storage/latch.h"
#include "utils/builtins.h"
#include "utils/float.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/sampling.h"
#include "utils/selfuncs.h"
 
PG_MODULE_MAGIC_EXT(
                    .name = "postgres_fdw",
                    .version = PG_VERSION
);
 
/* Default CPU cost to start up a foreign query. */
#define DEFAULT_FDW_STARTUP_COST    100.0
 
/* Default CPU cost to process 1 row (above and beyond cpu_tuple_cost). */
#define DEFAULT_FDW_TUPLE_COST      0.2
 
/* If no remote estimates, assume a sort costs 20% extra */
#define DEFAULT_FDW_SORT_MULTIPLIER 1.2
 
/*
 * Indexes of FDW-private information stored in fdw_private lists.
 *
 * These items are indexed with the enum FdwScanPrivateIndex, so an item
 * can be fetched with list_nth().  For example, to get the SELECT statement:
 *      sql = strVal(list_nth(fdw_private, FdwScanPrivateSelectSql));
 */
enum FdwScanPrivateIndex
{
    /* SQL statement to execute remotely (as a String node) */
    FdwScanPrivateSelectSql,
    /* Integer list of attribute numbers retrieved by the SELECT */
    FdwScanPrivateRetrievedAttrs,
    /* Integer representing the desired fetch_size */
    FdwScanPrivateFetchSize,
 
    /*
     * String describing join i.e. names of relations being joined and types
     * of join, added when the scan is join
     */
    FdwScanPrivateRelations,
};
 
/*
 * Similarly, this enum describes what's kept in the fdw_private list for
 * a ModifyTable node referencing a postgres_fdw foreign table.  We store:
 *
 * 1) INSERT/UPDATE/DELETE statement text to be sent to the remote server
 * 2) Integer list of target attribute numbers for INSERT/UPDATE
 *    (NIL for a DELETE)
 * 3) Length till the end of VALUES clause for INSERT
 *    (-1 for a DELETE/UPDATE)
 * 4) Boolean flag showing if the remote query has a RETURNING clause
 * 5) Integer list of attribute numbers retrieved by RETURNING, if any
 */
enum FdwModifyPrivateIndex
{
    /* SQL statement to execute remotely (as a String node) */
    FdwModifyPrivateUpdateSql,
    /* Integer list of target attribute numbers for INSERT/UPDATE */
    FdwModifyPrivateTargetAttnums,
    /* Length till the end of VALUES clause (as an Integer node) */
    FdwModifyPrivateLen,
    /* has-returning flag (as a Boolean node) */
    FdwModifyPrivateHasReturning,
    /* Integer list of attribute numbers retrieved by RETURNING */
    FdwModifyPrivateRetrievedAttrs,
};
 
/*
 * Similarly, this enum describes what's kept in the fdw_private list for
 * a ForeignScan node that modifies a foreign table directly.  We store:
 *
 * 1) UPDATE/DELETE statement text to be sent to the remote server
 * 2) Boolean flag showing if the remote query has a RETURNING clause
 * 3) Integer list of attribute numbers retrieved by RETURNING, if any
 * 4) Boolean flag showing if we set the command es_processed
 */
enum FdwDirectModifyPrivateIndex
{
    /* SQL statement to execute remotely (as a String node) */
    FdwDirectModifyPrivateUpdateSql,
    /* has-returning flag (as a Boolean node) */
    FdwDirectModifyPrivateHasReturning,
    /* Integer list of attribute numbers retrieved by RETURNING */
    FdwDirectModifyPrivateRetrievedAttrs,
    /* set-processed flag (as a Boolean node) */
    FdwDirectModifyPrivateSetProcessed,
};
 
/*
 * Execution state of a foreign scan using postgres_fdw.
 */
typedef struct PgFdwScanState
{
    Relation    rel;            /* relcache entry for the foreign table. NULL
                                 * for a foreign join scan. */
    TupleDesc   tupdesc;        /* tuple descriptor of scan */
    AttInMetadata *attinmeta;   /* attribute datatype conversion metadata */
 
    /* extracted fdw_private data */
    char       *query;          /* text of SELECT command */
    List       *retrieved_attrs;    /* list of retrieved attribute numbers */
 
    /* for remote query execution */
    PGconn     *conn;           /* connection for the scan */
    PgFdwConnState *conn_state; /* extra per-connection state */
    unsigned int cursor_number; /* quasi-unique ID for my cursor */
    bool        cursor_exists;  /* have we created the cursor? */
    int         numParams;      /* number of parameters passed to query */
    FmgrInfo   *param_flinfo;   /* output conversion functions for them */
    List       *param_exprs;    /* executable expressions for param values */
    const char **param_values;  /* textual values of query parameters */
 
    /* for storing result tuples */
    HeapTuple  *tuples;         /* array of currently-retrieved tuples */
    int         num_tuples;     /* # of tuples in array */
    int         next_tuple;     /* index of next one to return */
 
    /* batch-level state, for optimizing rewinds and avoiding useless fetch */
    int         fetch_ct_2;     /* Min(# of fetches done, 2) */
    bool        eof_reached;    /* true if last fetch reached EOF */
 
    /* for asynchronous execution */
    bool        async_capable;  /* engage asynchronous-capable logic? */
 
    /* working memory contexts */
    MemoryContext batch_cxt;    /* context holding current batch of tuples */
    MemoryContext temp_cxt;     /* context for per-tuple temporary data */
 
    int         fetch_size;     /* number of tuples per fetch */
} PgFdwScanState;
 
/*
 * Execution state of a foreign insert/update/delete operation.
 */
typedef struct PgFdwModifyState
{
    Relation    rel;            /* relcache entry for the foreign table */
    AttInMetadata *attinmeta;   /* attribute datatype conversion metadata */
 
    /* for remote query execution */
    PGconn     *conn;           /* connection for the scan */
    PgFdwConnState *conn_state; /* extra per-connection state */
    char       *p_name;         /* name of prepared statement, if created */
 
    /* extracted fdw_private data */
    char       *query;          /* text of INSERT/UPDATE/DELETE command */
    char       *orig_query;     /* original text of INSERT command */
    List       *target_attrs;   /* list of target attribute numbers */
    int         values_end;     /* length up to the end of VALUES */
    int         batch_size;     /* value of FDW option "batch_size" */
    bool        has_returning;  /* is there a RETURNING clause? */
    List       *retrieved_attrs;    /* attr numbers retrieved by RETURNING */
 
    /* info about parameters for prepared statement */
    AttrNumber  ctidAttno;      /* attnum of input resjunk ctid column */
    int         p_nums;         /* number of parameters to transmit */
    FmgrInfo   *p_flinfo;       /* output conversion functions for them */
 
    /* batch operation stuff */
    int         num_slots;      /* number of slots to insert */
 
    /* working memory context */
    MemoryContext temp_cxt;     /* context for per-tuple temporary data */
 
    /* for update row movement if subplan result rel */
    struct PgFdwModifyState *aux_fmstate;   /* foreign-insert state, if
                                             * created */
} PgFdwModifyState;
 
/*
 * Execution state of a foreign scan that modifies a foreign table directly.
 */
typedef struct PgFdwDirectModifyState
{
    Relation    rel;            /* relcache entry for the foreign table */
    AttInMetadata *attinmeta;   /* attribute datatype conversion metadata */
 
    /* extracted fdw_private data */
    char       *query;          /* text of UPDATE/DELETE command */
    bool        has_returning;  /* is there a RETURNING clause? */
    List       *retrieved_attrs;    /* attr numbers retrieved by RETURNING */
    bool        set_processed;  /* do we set the command es_processed? */
 
    /* for remote query execution */
    PGconn     *conn;           /* connection for the update */
    PgFdwConnState *conn_state; /* extra per-connection state */
    int         numParams;      /* number of parameters passed to query */
    FmgrInfo   *param_flinfo;   /* output conversion functions for them */
    List       *param_exprs;    /* executable expressions for param values */
    const char **param_values;  /* textual values of query parameters */
 
    /* for storing result tuples */
    PGresult   *result;         /* result for query */
    int         num_tuples;     /* # of result tuples */
    int         next_tuple;     /* index of next one to return */
    Relation    resultRel;      /* relcache entry for the target relation */
    AttrNumber *attnoMap;       /* array of attnums of input user columns */
    AttrNumber  ctidAttno;      /* attnum of input ctid column */
    AttrNumber  oidAttno;       /* attnum of input oid column */
    bool        hasSystemCols;  /* are there system columns of resultRel? */
 
    /* working memory context */
    MemoryContext temp_cxt;     /* context for per-tuple temporary data */
} PgFdwDirectModifyState;
 
/*
 * Workspace for analyzing a foreign table.
 */
typedef struct PgFdwAnalyzeState
{
    Relation    rel;            /* relcache entry for the foreign table */
    AttInMetadata *attinmeta;   /* attribute datatype conversion metadata */
    List       *retrieved_attrs;    /* attr numbers retrieved by query */
 
    /* collected sample rows */
    HeapTuple  *rows;           /* array of size targrows */
    int         targrows;       /* target # of sample rows */
    int         numrows;        /* # of sample rows collected */
 
    /* for random sampling */
    double      samplerows;     /* # of rows fetched */
    double      rowstoskip;     /* # of rows to skip before next sample */
    ReservoirStateData rstate;  /* state for reservoir sampling */
 
    /* working memory contexts */
    MemoryContext anl_cxt;      /* context for per-analyze lifespan data */
    MemoryContext temp_cxt;     /* context for per-tuple temporary data */
} PgFdwAnalyzeState;
 
/*
 * This enum describes what's kept in the fdw_private list for a ForeignPath.
 * We store:
 *
 * 1) Boolean flag showing if the remote query has the final sort
 * 2) Boolean flag showing if the remote query has the LIMIT clause
 */
enum FdwPathPrivateIndex
{
    /* has-final-sort flag (as a Boolean node) */
    FdwPathPrivateHasFinalSort,
    /* has-limit flag (as a Boolean node) */
    FdwPathPrivateHasLimit,
};
 
/* Struct for extra information passed to estimate_path_cost_size() */
typedef struct
{
    PathTarget *target;
    bool        has_final_sort;
    bool        has_limit;
    double      limit_tuples;
    int64       count_est;
    int64       offset_est;
} PgFdwPathExtraData;
 
/*
 * Identify the attribute where data conversion fails.
 */
typedef struct ConversionLocation
{
    AttrNumber  cur_attno;      /* attribute number being processed, or 0 */
    Relation    rel;            /* foreign table being processed, or NULL */
    ForeignScanState *fsstate;  /* plan node being processed, or NULL */
} ConversionLocation;
 
/* Callback argument for ec_member_matches_foreign */
typedef struct
{
    Expr       *current;        /* current expr, or NULL if not yet found */
    List       *already_used;   /* expressions already dealt with */
} ec_member_foreign_arg;
 
/* Pairs of remote columns with local columns */
typedef struct
{
    AttrNumber  local_attnum;
    char       *local_attname;
    char       *remote_attname;
    int         res_index;
} RemoteAttributeMapping;
 
/* Result sets that are returned from a foreign statistics scan */
typedef struct
{
    PGresult   *rel;
    PGresult   *att;
    int         server_version_num;
} RemoteStatsResults;
 
/* Column order in relation stats query */
enum RelStatsColumns
{
    RELSTATS_RELPAGES = 0,
    RELSTATS_RELTUPLES,
    RELSTATS_RELKIND,
    RELSTATS_NUM_FIELDS,
};
 
/* Column order in attribute stats query */
enum AttStatsColumns
{
    ATTSTATS_ATTNAME = 0,
    ATTSTATS_NULL_FRAC,
    ATTSTATS_AVG_WIDTH,
    ATTSTATS_N_DISTINCT,
    ATTSTATS_MOST_COMMON_VALS,
    ATTSTATS_MOST_COMMON_FREQS,
    ATTSTATS_HISTOGRAM_BOUNDS,
    ATTSTATS_CORRELATION,
    ATTSTATS_MOST_COMMON_ELEMS,
    ATTSTATS_MOST_COMMON_ELEM_FREQS,
    ATTSTATS_ELEM_COUNT_HISTOGRAM,
    ATTSTATS_RANGE_LENGTH_HISTOGRAM,
    ATTSTATS_RANGE_EMPTY_FRAC,
    ATTSTATS_RANGE_BOUNDS_HISTOGRAM,
    ATTSTATS_NUM_FIELDS,
};
 
/* Relation stats import query */
static const char *relimport_sql =
"SELECT pg_catalog.pg_restore_relation_stats(\n"
"\t'version', $1,\n"
"\t'schemaname', $2,\n"
"\t'relname', $3,\n"
"\t'relpages', $4::integer,\n"
"\t'reltuples', $5::real)";
 
/* Argument order in relation stats import query */
enum RelImportSqlArgs
{
    RELIMPORT_SQL_VERSION = 0,
    RELIMPORT_SQL_SCHEMANAME,
    RELIMPORT_SQL_RELNAME,
    RELIMPORT_SQL_RELPAGES,
    RELIMPORT_SQL_RELTUPLES,
    RELIMPORT_SQL_NUM_FIELDS
};
 
/* Argument types in relation stats import query */
static const Oid relimport_argtypes[RELIMPORT_SQL_NUM_FIELDS] =
{
    INT4OID, TEXTOID, TEXTOID, TEXTOID,
    TEXTOID,
};
 
/* Attribute stats import query */
static const char *attimport_sql =
"SELECT pg_catalog.pg_restore_attribute_stats(\n"
"\t'version', $1,\n"
"\t'schemaname', $2,\n"
"\t'relname', $3,\n"
"\t'attnum', $4,\n"
"\t'inherited', false::boolean,\n"
"\t'null_frac', $5::real,\n"
"\t'avg_width', $6::integer,\n"
"\t'n_distinct', $7::real,\n"
"\t'most_common_vals', $8,\n"
"\t'most_common_freqs', $9::real[],\n"
"\t'histogram_bounds', $10,\n"
"\t'correlation', $11::real,\n"
"\t'most_common_elems', $12,\n"
"\t'most_common_elem_freqs', $13::real[],\n"
"\t'elem_count_histogram', $14::real[],\n"
"\t'range_length_histogram', $15,\n"
"\t'range_empty_frac', $16::real,\n"
"\t'range_bounds_histogram', $17)";
 
/* Argument order in attribute stats import query */
enum AttImportSqlArgs
{
    ATTIMPORT_SQL_VERSION = 0,
    ATTIMPORT_SQL_SCHEMANAME,
    ATTIMPORT_SQL_RELNAME,
    ATTIMPORT_SQL_ATTNUM,
    ATTIMPORT_SQL_NULL_FRAC,
    ATTIMPORT_SQL_AVG_WIDTH,
    ATTIMPORT_SQL_N_DISTINCT,
    ATTIMPORT_SQL_MOST_COMMON_VALS,
    ATTIMPORT_SQL_MOST_COMMON_FREQS,
    ATTIMPORT_SQL_HISTOGRAM_BOUNDS,
    ATTIMPORT_SQL_CORRELATION,
    ATTIMPORT_SQL_MOST_COMMON_ELEMS,
    ATTIMPORT_SQL_MOST_COMMON_ELEM_FREQS,
    ATTIMPORT_SQL_ELEM_COUNT_HISTOGRAM,
    ATTIMPORT_SQL_RANGE_LENGTH_HISTOGRAM,
    ATTIMPORT_SQL_RANGE_EMPTY_FRAC,
    ATTIMPORT_SQL_RANGE_BOUNDS_HISTOGRAM,
    ATTIMPORT_SQL_NUM_FIELDS
};
 
/* Argument types in attribute stats import query */
static const Oid attimport_argtypes[ATTIMPORT_SQL_NUM_FIELDS] =
{
    INT4OID, TEXTOID, TEXTOID, INT2OID,
    TEXTOID, TEXTOID, TEXTOID, TEXTOID,
    TEXTOID, TEXTOID, TEXTOID, TEXTOID,
    TEXTOID, TEXTOID, TEXTOID, TEXTOID,
    TEXTOID,
};
 
/*
 * The mapping of attribute stats query columns to the positional arguments in
 * the prepared pg_restore_attribute_stats() statement.
 */
typedef struct
{
    enum AttStatsColumns res_field;
    enum AttImportSqlArgs arg_num;
} AttrResultArgMap;
 
#define NUM_MAPPED_ATTIMPORT_ARGS 13
 
static const AttrResultArgMap attr_result_arg_map[NUM_MAPPED_ATTIMPORT_ARGS] =
{
    {ATTSTATS_NULL_FRAC, ATTIMPORT_SQL_NULL_FRAC},
    {ATTSTATS_AVG_WIDTH, ATTIMPORT_SQL_AVG_WIDTH},
    {ATTSTATS_N_DISTINCT, ATTIMPORT_SQL_N_DISTINCT},
    {ATTSTATS_MOST_COMMON_VALS, ATTIMPORT_SQL_MOST_COMMON_VALS},
    {ATTSTATS_MOST_COMMON_FREQS, ATTIMPORT_SQL_MOST_COMMON_FREQS},
    {ATTSTATS_HISTOGRAM_BOUNDS, ATTIMPORT_SQL_HISTOGRAM_BOUNDS},
    {ATTSTATS_CORRELATION, ATTIMPORT_SQL_CORRELATION},
    {ATTSTATS_MOST_COMMON_ELEMS, ATTIMPORT_SQL_MOST_COMMON_ELEMS},
    {ATTSTATS_MOST_COMMON_ELEM_FREQS, ATTIMPORT_SQL_MOST_COMMON_ELEM_FREQS},
    {ATTSTATS_ELEM_COUNT_HISTOGRAM, ATTIMPORT_SQL_ELEM_COUNT_HISTOGRAM},
    {ATTSTATS_RANGE_LENGTH_HISTOGRAM, ATTIMPORT_SQL_RANGE_LENGTH_HISTOGRAM},
    {ATTSTATS_RANGE_EMPTY_FRAC, ATTIMPORT_SQL_RANGE_EMPTY_FRAC},
    {ATTSTATS_RANGE_BOUNDS_HISTOGRAM, ATTIMPORT_SQL_RANGE_BOUNDS_HISTOGRAM},
};
 
/* Attribute stats clear query */
static const char *attclear_sql =
"SELECT pg_catalog.pg_clear_attribute_stats($1, $2, $3, false)";
 
/* Argument order in attribute stats clear query */
enum AttClearSqlArgs
{
    ATTCLEAR_SQL_SCHEMANAME = 0,
    ATTCLEAR_SQL_RELNAME,
    ATTCLEAR_SQL_ATTNAME,
    ATTCLEAR_SQL_NUM_FIELDS
};
 
/* Argument types in attribute stats clear query */
static const Oid attclear_argtypes[ATTCLEAR_SQL_NUM_FIELDS] =
{
    TEXTOID, TEXTOID, TEXTOID,
};
 
/*
 * SQL functions
 */
PG_FUNCTION_INFO_V1(postgres_fdw_handler);
 
/*
 * FDW callback routines
 */
static void postgresGetForeignRelSize(PlannerInfo *root,
                                      RelOptInfo *baserel,
                                      Oid foreigntableid);
static void postgresGetForeignPaths(PlannerInfo *root,
                                    RelOptInfo *baserel,
                                    Oid foreigntableid);
static ForeignScan *postgresGetForeignPlan(PlannerInfo *root,
                                           RelOptInfo *foreignrel,
                                           Oid foreigntableid,
                                           ForeignPath *best_path,
                                           List *tlist,
                                           List *scan_clauses,
                                           Plan *outer_plan);
static void postgresBeginForeignScan(ForeignScanState *node, int eflags);
static TupleTableSlot *postgresIterateForeignScan(ForeignScanState *node);
static void postgresReScanForeignScan(ForeignScanState *node);
static void postgresEndForeignScan(ForeignScanState *node);
static void postgresAddForeignUpdateTargets(PlannerInfo *root,
                                            Index rtindex,
                                            RangeTblEntry *target_rte,
                                            Relation target_relation);
static List *postgresPlanForeignModify(PlannerInfo *root,
                                       ModifyTable *plan,
                                       Index resultRelation,
                                       int subplan_index);
static void postgresBeginForeignModify(ModifyTableState *mtstate,
                                       ResultRelInfo *resultRelInfo,
                                       List *fdw_private,
                                       int subplan_index,
                                       int eflags);
static TupleTableSlot *postgresExecForeignInsert(EState *estate,
                                                 ResultRelInfo *resultRelInfo,
                                                 TupleTableSlot *slot,
                                                 TupleTableSlot *planSlot);
static TupleTableSlot **postgresExecForeignBatchInsert(EState *estate,
                                                       ResultRelInfo *resultRelInfo,
                                                       TupleTableSlot **slots,
                                                       TupleTableSlot **planSlots,
                                                       int *numSlots);
static int  postgresGetForeignModifyBatchSize(ResultRelInfo *resultRelInfo);
static TupleTableSlot *postgresExecForeignUpdate(EState *estate,
                                                 ResultRelInfo *resultRelInfo,
                                                 TupleTableSlot *slot,
                                                 TupleTableSlot *planSlot);
static TupleTableSlot *postgresExecForeignDelete(EState *estate,
                                                 ResultRelInfo *resultRelInfo,
                                                 TupleTableSlot *slot,
                                                 TupleTableSlot *planSlot);
static void postgresEndForeignModify(EState *estate,
                                     ResultRelInfo *resultRelInfo);
static void postgresBeginForeignInsert(ModifyTableState *mtstate,
                                       ResultRelInfo *resultRelInfo);
static void postgresEndForeignInsert(EState *estate,
                                     ResultRelInfo *resultRelInfo);
static int  postgresIsForeignRelUpdatable(Relation rel);
static bool postgresPlanDirectModify(PlannerInfo *root,
                                     ModifyTable *plan,
                                     Index resultRelation,
                                     int subplan_index);
static void postgresBeginDirectModify(ForeignScanState *node, int eflags);
static TupleTableSlot *postgresIterateDirectModify(ForeignScanState *node);
static void postgresEndDirectModify(ForeignScanState *node);
static void postgresExplainForeignScan(ForeignScanState *node,
                                       ExplainState *es);
static void postgresExplainForeignModify(ModifyTableState *mtstate,
                                         ResultRelInfo *rinfo,
                                         List *fdw_private,
                                         int subplan_index,
                                         ExplainState *es);
static void postgresExplainDirectModify(ForeignScanState *node,
                                        ExplainState *es);
static void postgresExecForeignTruncate(List *rels,
                                        DropBehavior behavior,
                                        bool restart_seqs);
static bool postgresAnalyzeForeignTable(Relation relation,
                                        AcquireSampleRowsFunc *func,
                                        BlockNumber *totalpages);
static bool postgresImportForeignStatistics(Relation relation,
                                            List *va_cols,
                                            int elevel);
static List *postgresImportForeignSchema(ImportForeignSchemaStmt *stmt,
                                         Oid serverOid);
static void postgresGetForeignJoinPaths(PlannerInfo *root,
                                        RelOptInfo *joinrel,
                                        RelOptInfo *outerrel,
                                        RelOptInfo *innerrel,
                                        JoinType jointype,
                                        JoinPathExtraData *extra);
static bool postgresRecheckForeignScan(ForeignScanState *node,
                                       TupleTableSlot *slot);
static void postgresGetForeignUpperPaths(PlannerInfo *root,
                                         UpperRelationKind stage,
                                         RelOptInfo *input_rel,
                                         RelOptInfo *output_rel,
                                         void *extra);
static bool postgresIsForeignPathAsyncCapable(ForeignPath *path);
static void postgresForeignAsyncRequest(AsyncRequest *areq);
static void postgresForeignAsyncConfigureWait(AsyncRequest *areq);
static void postgresForeignAsyncNotify(AsyncRequest *areq);
 
/*
 * Helper functions
 */
static void estimate_path_cost_size(PlannerInfo *root,
                                    RelOptInfo *foreignrel,
                                    List *param_join_conds,
                                    List *pathkeys,
                                    PgFdwPathExtraData *fpextra,
                                    double *p_rows, int *p_width,
                                    int *p_disabled_nodes,
                                    Cost *p_startup_cost, Cost *p_total_cost);
static void get_remote_estimate(const char *sql,
                                PGconn *conn,
                                double *rows,
                                int *width,
                                Cost *startup_cost,
                                Cost *total_cost);
static void adjust_foreign_grouping_path_cost(PlannerInfo *root,
                                              List *pathkeys,
                                              double retrieved_rows,
                                              double width,
                                              double limit_tuples,
                                              int *p_disabled_nodes,
                                              Cost *p_startup_cost,
                                              Cost *p_run_cost);
static bool ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel,
                                      EquivalenceClass *ec, EquivalenceMember *em,
                                      void *arg);
static void create_cursor(ForeignScanState *node);
static void fetch_more_data(ForeignScanState *node);
static void close_cursor(PGconn *conn, unsigned int cursor_number,
                         PgFdwConnState *conn_state);
static PgFdwModifyState *create_foreign_modify(EState *estate,
                                               RangeTblEntry *rte,
                                               ResultRelInfo *resultRelInfo,
                                               CmdType operation,
                                               Plan *subplan,
                                               char *query,
                                               List *target_attrs,
                                               int values_end,
                                               bool has_returning,
                                               List *retrieved_attrs);
static TupleTableSlot **execute_foreign_modify(EState *estate,
                                               ResultRelInfo *resultRelInfo,
                                               CmdType operation,
                                               TupleTableSlot **slots,
                                               TupleTableSlot **planSlots,
                                               int *numSlots);
static void prepare_foreign_modify(PgFdwModifyState *fmstate);
static const char **convert_prep_stmt_params(PgFdwModifyState *fmstate,
                                             ItemPointer tupleid,
                                             TupleTableSlot **slots,
                                             int numSlots);
static void store_returning_result(PgFdwModifyState *fmstate,
                                   TupleTableSlot *slot, PGresult *res);
static void finish_foreign_modify(PgFdwModifyState *fmstate);
static void deallocate_query(PgFdwModifyState *fmstate);
static List *build_remote_returning(Index rtindex, Relation rel,
                                    List *returningList);
static void rebuild_fdw_scan_tlist(ForeignScan *fscan, List *tlist);
static void execute_dml_stmt(ForeignScanState *node);
static TupleTableSlot *get_returning_data(ForeignScanState *node);
static void init_returning_filter(PgFdwDirectModifyState *dmstate,
                                  List *fdw_scan_tlist,
                                  Index rtindex);
static TupleTableSlot *apply_returning_filter(PgFdwDirectModifyState *dmstate,
                                              ResultRelInfo *resultRelInfo,
                                              TupleTableSlot *slot,
                                              EState *estate);
static void prepare_query_params(PlanState *node,
                                 List *fdw_exprs,
                                 int numParams,
                                 FmgrInfo **param_flinfo,
                                 List **param_exprs,
                                 const char ***param_values);
static void process_query_params(ExprContext *econtext,
                                 FmgrInfo *param_flinfo,
                                 List *param_exprs,
                                 const char **param_values);
static int  postgresAcquireSampleRowsFunc(Relation relation, int elevel,
                                          HeapTuple *rows, int targrows,
                                          double *totalrows,
                                          double *totaldeadrows);
static void analyze_row_processor(PGresult *res, int row,
                                  PgFdwAnalyzeState *astate);
static bool fetch_remote_statistics(Relation relation,
                                    List *va_cols,
                                    ForeignTable *table,
                                    const char *local_schemaname,
                                    const char *local_relname,
                                    int *p_attrcnt,
                                    RemoteAttributeMapping **p_remattrmap,
                                    RemoteStatsResults *remstats);
static PGresult *fetch_relstats(PGconn *conn, Relation relation);
static PGresult *fetch_attstats(PGconn *conn, int server_version_num,
                                const char *remote_schemaname, const char *remote_relname,
                                const char *column_list);
static RemoteAttributeMapping *build_remattrmap(Relation relation, List *va_cols,
                                                int *p_attrcnt, StringInfo column_list);
static void free_remattrmap(RemoteAttributeMapping *map, int len);
static bool attname_in_list(const char *attname, List *va_cols);
static int  remattrmap_cmp(const void *v1, const void *v2);
static bool match_attrmap(PGresult *res,
                          const char *local_schemaname,
                          const char *local_relname,
                          const char *remote_schemaname,
                          const char *remote_relname,
                          int attrcnt,
                          RemoteAttributeMapping *remattrmap);
static bool import_fetched_statistics(const char *schemaname,
                                      const char *relname,
                                      int attrcnt,
                                      const RemoteAttributeMapping *remattrmap,
                                      RemoteStatsResults *remstats);
static void map_field_to_arg(PGresult *res, int row, int field,
                             int arg, Datum *values, char *nulls);
static bool import_spi_query_ok(void);
static void produce_tuple_asynchronously(AsyncRequest *areq, bool fetch);
static void fetch_more_data_begin(AsyncRequest *areq);
static void complete_pending_request(AsyncRequest *areq);
static HeapTuple make_tuple_from_result_row(PGresult *res,
                                            int row,
                                            Relation rel,
                                            AttInMetadata *attinmeta,
                                            List *retrieved_attrs,
                                            ForeignScanState *fsstate,
                                            MemoryContext temp_context);
static void conversion_error_callback(void *arg);
static bool foreign_join_ok(PlannerInfo *root, RelOptInfo *joinrel,
                            JoinType jointype, RelOptInfo *outerrel, RelOptInfo *innerrel,
                            JoinPathExtraData *extra);
static bool foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
                                Node *havingQual);
static List *get_useful_pathkeys_for_relation(PlannerInfo *root,
                                              RelOptInfo *rel);
static List *get_useful_ecs_for_relation(PlannerInfo *root, RelOptInfo *rel);
static void add_paths_with_pathkeys_for_rel(PlannerInfo *root, RelOptInfo *rel,
                                            Path *epq_path, List *restrictlist);
static void add_foreign_grouping_paths(PlannerInfo *root,
                                       RelOptInfo *input_rel,
                                       RelOptInfo *grouped_rel,
                                       GroupPathExtraData *extra);
static void add_foreign_ordered_paths(PlannerInfo *root,
                                      RelOptInfo *input_rel,
                                      RelOptInfo *ordered_rel);
static void add_foreign_final_paths(PlannerInfo *root,
                                    RelOptInfo *input_rel,
                                    RelOptInfo *final_rel,
                                    FinalPathExtraData *extra);
static void apply_server_options(PgFdwRelationInfo *fpinfo);
static void apply_table_options(PgFdwRelationInfo *fpinfo);
static void merge_fdw_options(PgFdwRelationInfo *fpinfo,
                              const PgFdwRelationInfo *fpinfo_o,
                              const PgFdwRelationInfo *fpinfo_i);
static int  get_batch_size_option(Relation rel);
 
 
/*
 * Foreign-data wrapper handler function: return a struct with pointers
 * to my callback routines.
 */
Datum
postgres_fdw_handler(PG_FUNCTION_ARGS)
{
    FdwRoutine *routine = makeNode(FdwRoutine);
 
    /* Functions for scanning foreign tables */
    routine->GetForeignRelSize = postgresGetForeignRelSize;
    routine->GetForeignPaths = postgresGetForeignPaths;
    routine->GetForeignPlan = postgresGetForeignPlan;
    routine->BeginForeignScan = postgresBeginForeignScan;
    routine->IterateForeignScan = postgresIterateForeignScan;
    routine->ReScanForeignScan = postgresReScanForeignScan;
    routine->EndForeignScan = postgresEndForeignScan;
 
    /* Functions for updating foreign tables */
    routine->AddForeignUpdateTargets = postgresAddForeignUpdateTargets;
    routine->PlanForeignModify = postgresPlanForeignModify;
    routine->BeginForeignModify = postgresBeginForeignModify;
    routine->ExecForeignInsert = postgresExecForeignInsert;
    routine->ExecForeignBatchInsert = postgresExecForeignBatchInsert;
    routine->GetForeignModifyBatchSize = postgresGetForeignModifyBatchSize;
    routine->ExecForeignUpdate = postgresExecForeignUpdate;
    routine->ExecForeignDelete = postgresExecForeignDelete;
    routine->EndForeignModify = postgresEndForeignModify;
    routine->BeginForeignInsert = postgresBeginForeignInsert;
    routine->EndForeignInsert = postgresEndForeignInsert;
    routine->IsForeignRelUpdatable = postgresIsForeignRelUpdatable;
    routine->PlanDirectModify = postgresPlanDirectModify;
    routine->BeginDirectModify = postgresBeginDirectModify;
    routine->IterateDirectModify = postgresIterateDirectModify;
    routine->EndDirectModify = postgresEndDirectModify;
 
    /* Function for EvalPlanQual rechecks */
    routine->RecheckForeignScan = postgresRecheckForeignScan;
    /* Support functions for EXPLAIN */
    routine->ExplainForeignScan = postgresExplainForeignScan;
    routine->ExplainForeignModify = postgresExplainForeignModify;
    routine->ExplainDirectModify = postgresExplainDirectModify;
 
    /* Support function for TRUNCATE */
    routine->ExecForeignTruncate = postgresExecForeignTruncate;
 
    /* Support functions for ANALYZE */
    routine->AnalyzeForeignTable = postgresAnalyzeForeignTable;
    routine->ImportForeignStatistics = postgresImportForeignStatistics;
 
    /* Support functions for IMPORT FOREIGN SCHEMA */
    routine->ImportForeignSchema = postgresImportForeignSchema;
 
    /* Support functions for join push-down */
    routine->GetForeignJoinPaths = postgresGetForeignJoinPaths;
 
    /* Support functions for upper relation push-down */
    routine->GetForeignUpperPaths = postgresGetForeignUpperPaths;
 
    /* Support functions for asynchronous execution */
    routine->IsForeignPathAsyncCapable = postgresIsForeignPathAsyncCapable;
    routine->ForeignAsyncRequest = postgresForeignAsyncRequest;
    routine->ForeignAsyncConfigureWait = postgresForeignAsyncConfigureWait;
    routine->ForeignAsyncNotify = postgresForeignAsyncNotify;
 
    PG_RETURN_POINTER(routine);
}
 
/*
 * postgresGetForeignRelSize
 *      Estimate # of rows and width of the result of the scan
 *
 * We should consider the effect of all baserestrictinfo clauses here, but
 * not any join clauses.
 */
static void
postgresGetForeignRelSize(PlannerInfo *root,
                          RelOptInfo *baserel,
                          Oid foreigntableid)
{
    PgFdwRelationInfo *fpinfo;
    ListCell   *lc;
 
    /*
     * We use PgFdwRelationInfo to pass various information to subsequent
     * functions.
     */
    fpinfo = palloc0_object(PgFdwRelationInfo);
    baserel->fdw_private = fpinfo;
 
    /* Base foreign tables need to be pushed down always. */
    fpinfo->pushdown_safe = true;
 
    /* Look up foreign-table catalog info. */
    fpinfo->table = GetForeignTable(foreigntableid);
    fpinfo->server = GetForeignServer(fpinfo->table->serverid);
 
    /*
     * Extract user-settable option values.  Note that per-table settings of
     * use_remote_estimate, fetch_size and async_capable override per-server
     * settings of them, respectively.
     */
    fpinfo->use_remote_estimate = false;
    fpinfo->fdw_startup_cost = DEFAULT_FDW_STARTUP_COST;
    fpinfo->fdw_tuple_cost = DEFAULT_FDW_TUPLE_COST;
    fpinfo->shippable_extensions = NIL;
    fpinfo->fetch_size = 100;
    fpinfo->async_capable = false;
 
    apply_server_options(fpinfo);
    apply_table_options(fpinfo);
 
    /*
     * If the table or the server is configured to use remote estimates,
     * identify which user to do remote access as during planning.  This
     * should match what ExecCheckPermissions() does.  If we fail due to lack
     * of permissions, the query would have failed at runtime anyway.
     */
    if (fpinfo->use_remote_estimate)
    {
        Oid         userid;
 
        userid = OidIsValid(baserel->userid) ? baserel->userid : GetUserId();
        fpinfo->user = GetUserMapping(userid, fpinfo->server->serverid);
    }
    else
        fpinfo->user = NULL;
 
    /*
     * Identify which baserestrictinfo clauses can be sent to the remote
     * server and which can't.
     */
    classifyConditions(root, baserel, baserel->baserestrictinfo,
                       &fpinfo->remote_conds, &fpinfo->local_conds);
 
    /*
     * Identify which attributes will need to be retrieved from the remote
     * server.  These include all attrs needed for joins or final output, plus
     * all attrs used in the local_conds.  (Note: if we end up using a
     * parameterized scan, it's possible that some of the join clauses will be
     * sent to the remote and thus we wouldn't really need to retrieve the
     * columns used in them.  Doesn't seem worth detecting that case though.)
     */
    fpinfo->attrs_used = NULL;
    pull_varattnos((Node *) baserel->reltarget->exprs, baserel->relid,
                   &fpinfo->attrs_used);
    foreach(lc, fpinfo->local_conds)
    {
        RestrictInfo *rinfo = lfirst_node(RestrictInfo, lc);
 
        pull_varattnos((Node *) rinfo->clause, baserel->relid,
                       &fpinfo->attrs_used);
    }
 
    /*
     * Compute the selectivity and cost of the local_conds, so we don't have
     * to do it over again for each path.  The best we can do for these
     * conditions is to estimate selectivity on the basis of local statistics.
     */
    fpinfo->local_conds_sel = clauselist_selectivity(root,
                                                     fpinfo->local_conds,
                                                     baserel->relid,
                                                     JOIN_INNER,
                                                     NULL);
 
    cost_qual_eval(&fpinfo->local_conds_cost, fpinfo->local_conds, root);
 
    /*
     * Set # of retrieved rows and cached relation costs to some negative
     * value, so that we can detect when they are set to some sensible values,
     * during one (usually the first) of the calls to estimate_path_cost_size.
     */
    fpinfo->retrieved_rows = -1;
    fpinfo->rel_startup_cost = -1;
    fpinfo->rel_total_cost = -1;
 
    /*
     * If the table or the server is configured to use remote estimates,
     * connect to the foreign server and execute EXPLAIN to estimate the
     * number of rows selected by the restriction clauses, as well as the
     * average row width.  Otherwise, estimate using whatever statistics we
     * have locally, in a way similar to ordinary tables.
     */
    if (fpinfo->use_remote_estimate)
    {
        /*
         * Get cost/size estimates with help of remote server.  Save the
         * values in fpinfo so we don't need to do it again to generate the
         * basic foreign path.
         */
        estimate_path_cost_size(root, baserel, NIL, NIL, NULL,
                                &fpinfo->rows, &fpinfo->width,
                                &fpinfo->disabled_nodes,
                                &fpinfo->startup_cost, &fpinfo->total_cost);
 
        /* Report estimated baserel size to planner. */
        baserel->rows = fpinfo->rows;
        baserel->reltarget->width = fpinfo->width;
    }
    else
    {
        /*
         * If the foreign table has never been ANALYZEd, it will have
         * reltuples < 0, meaning "unknown".  We can't do much if we're not
         * allowed to consult the remote server, but we can use a hack similar
         * to plancat.c's treatment of empty relations: use a minimum size
         * estimate of 10 pages, and divide by the column-datatype-based width
         * estimate to get the corresponding number of tuples.
         */
        if (baserel->tuples < 0)
        {
            baserel->pages = 10;
            baserel->tuples =
                (10 * BLCKSZ) / (baserel->reltarget->width +
                                 MAXALIGN(SizeofHeapTupleHeader));
        }
 
        /* Estimate baserel size as best we can with local statistics. */
        set_baserel_size_estimates(root, baserel);
 
        /* Fill in basically-bogus cost estimates for use later. */
        estimate_path_cost_size(root, baserel, NIL, NIL, NULL,
                                &fpinfo->rows, &fpinfo->width,
                                &fpinfo->disabled_nodes,
                                &fpinfo->startup_cost, &fpinfo->total_cost);
    }
 
    /*
     * fpinfo->relation_name gets the numeric rangetable index of the foreign
     * table RTE.  (If this query gets EXPLAIN'd, we'll convert that to a
     * human-readable string at that time.)
     */
    fpinfo->relation_name = psprintf("%u", baserel->relid);
 
    /* No outer and inner relations. */
    fpinfo->make_outerrel_subquery = false;
    fpinfo->make_innerrel_subquery = false;
    fpinfo->lower_subquery_rels = NULL;
    fpinfo->hidden_subquery_rels = NULL;
    /* Set the relation index. */
    fpinfo->relation_index = baserel->relid;
}
 
/*
 * get_useful_ecs_for_relation
 *      Determine which EquivalenceClasses might be involved in useful
 *      orderings of this relation.
 *
 * This function is in some respects a mirror image of the core function
 * pathkeys_useful_for_merging: for a regular table, we know what indexes
 * we have and want to test whether any of them are useful.  For a foreign
 * table, we don't know what indexes are present on the remote side but
 * want to speculate about which ones we'd like to use if they existed.
 *
 * This function returns a list of potentially-useful equivalence classes,
 * but it does not guarantee that an EquivalenceMember exists which contains
 * Vars only from the given relation.  For example, given ft1 JOIN t1 ON
 * ft1.x + t1.x = 0, this function will say that the equivalence class
 * containing ft1.x + t1.x is potentially useful.  Supposing ft1 is remote and
 * t1 is local (or on a different server), it will turn out that no useful
 * ORDER BY clause can be generated.  It's not our job to figure that out
 * here; we're only interested in identifying relevant ECs.
 */
static List *
get_useful_ecs_for_relation(PlannerInfo *root, RelOptInfo *rel)
{
    List       *useful_eclass_list = NIL;
    ListCell   *lc;
    Relids      relids;
 
    /*
     * First, consider whether any active EC is potentially useful for a merge
     * join against this relation.
     */
    if (rel->has_eclass_joins)
    {
        foreach(lc, root->eq_classes)
        {
            EquivalenceClass *cur_ec = (EquivalenceClass *) lfirst(lc);
 
            if (eclass_useful_for_merging(root, cur_ec, rel))
                useful_eclass_list = lappend(useful_eclass_list, cur_ec);
        }
    }
 
    /*
     * Next, consider whether there are any non-EC derivable join clauses that
     * are merge-joinable.  If the joininfo list is empty, we can exit
     * quickly.
     */
    if (rel->joininfo == NIL)
        return useful_eclass_list;
 
    /* If this is a child rel, we must use the topmost parent rel to search. */
    if (IS_OTHER_REL(rel))
    {
        Assert(!bms_is_empty(rel->top_parent_relids));
        relids = rel->top_parent_relids;
    }
    else
        relids = rel->relids;
 
    /* Check each join clause in turn. */
    foreach(lc, rel->joininfo)
    {
        RestrictInfo *restrictinfo = (RestrictInfo *) lfirst(lc);
 
        /* Consider only mergejoinable clauses */
        if (restrictinfo->mergeopfamilies == NIL)
            continue;
 
        /* Make sure we've got canonical ECs. */
        update_mergeclause_eclasses(root, restrictinfo);
 
        /*
         * restrictinfo->mergeopfamilies != NIL is sufficient to guarantee
         * that left_ec and right_ec will be initialized, per comments in
         * distribute_qual_to_rels.
         *
         * We want to identify which side of this merge-joinable clause
         * contains columns from the relation produced by this RelOptInfo. We
         * test for overlap, not containment, because there could be extra
         * relations on either side.  For example, suppose we've got something
         * like ((A JOIN B ON A.x = B.x) JOIN C ON A.y = C.y) LEFT JOIN D ON
         * A.y = D.y.  The input rel might be the joinrel between A and B, and
         * we'll consider the join clause A.y = D.y. relids contains a
         * relation not involved in the join class (B) and the equivalence
         * class for the left-hand side of the clause contains a relation not
         * involved in the input rel (C).  Despite the fact that we have only
         * overlap and not containment in either direction, A.y is potentially
         * useful as a sort column.
         *
         * Note that it's even possible that relids overlaps neither side of
         * the join clause.  For example, consider A LEFT JOIN B ON A.x = B.x
         * AND A.x = 1.  The clause A.x = 1 will appear in B's joininfo list,
         * but overlaps neither side of B.  In that case, we just skip this
         * join clause, since it doesn't suggest a useful sort order for this
         * relation.
         */
        if (bms_overlap(relids, restrictinfo->right_ec->ec_relids))
            useful_eclass_list = list_append_unique_ptr(useful_eclass_list,
                                                        restrictinfo->right_ec);
        else if (bms_overlap(relids, restrictinfo->left_ec->ec_relids))
            useful_eclass_list = list_append_unique_ptr(useful_eclass_list,
                                                        restrictinfo->left_ec);
    }
 
    return useful_eclass_list;
}
 
/*
 * get_useful_pathkeys_for_relation
 *      Determine which orderings of a relation might be useful.
 *
 * Getting data in sorted order can be useful either because the requested
 * order matches the final output ordering for the overall query we're
 * planning, or because it enables an efficient merge join.  Here, we try
 * to figure out which pathkeys to consider.
 */
static List *
get_useful_pathkeys_for_relation(PlannerInfo *root, RelOptInfo *rel)
{
    List       *useful_pathkeys_list = NIL;
    List       *useful_eclass_list;
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) rel->fdw_private;
    EquivalenceClass *query_ec = NULL;
    ListCell   *lc;
 
    /*
     * Pushing the query_pathkeys to the remote server is always worth
     * considering, because it might let us avoid a local sort.
     */
    fpinfo->qp_is_pushdown_safe = false;
    if (root->query_pathkeys)
    {
        bool        query_pathkeys_ok = true;
 
        foreach(lc, root->query_pathkeys)
        {
            PathKey    *pathkey = (PathKey *) lfirst(lc);
 
            /*
             * The planner and executor don't have any clever strategy for
             * taking data sorted by a prefix of the query's pathkeys and
             * getting it to be sorted by all of those pathkeys. We'll just
             * end up resorting the entire data set.  So, unless we can push
             * down all of the query pathkeys, forget it.
             */
            if (!is_foreign_pathkey(root, rel, pathkey))
            {
                query_pathkeys_ok = false;
                break;
            }
        }
 
        if (query_pathkeys_ok)
        {
            useful_pathkeys_list = list_make1(list_copy(root->query_pathkeys));
            fpinfo->qp_is_pushdown_safe = true;
        }
    }
 
    /*
     * Even if we're not using remote estimates, having the remote side do the
     * sort generally won't be any worse than doing it locally, and it might
     * be much better if the remote side can generate data in the right order
     * without needing a sort at all.  However, what we're going to do next is
     * try to generate pathkeys that seem promising for possible merge joins,
     * and that's more speculative.  A wrong choice might hurt quite a bit, so
     * bail out if we can't use remote estimates.
     */
    if (!fpinfo->use_remote_estimate)
        return useful_pathkeys_list;
 
    /* Get the list of interesting EquivalenceClasses. */
    useful_eclass_list = get_useful_ecs_for_relation(root, rel);
 
    /* Extract unique EC for query, if any, so we don't consider it again. */
    if (list_length(root->query_pathkeys) == 1)
    {
        PathKey    *query_pathkey = linitial(root->query_pathkeys);
 
        query_ec = query_pathkey->pk_eclass;
    }
 
    /*
     * As a heuristic, the only pathkeys we consider here are those of length
     * one.  It's surely possible to consider more, but since each one we
     * choose to consider will generate a round-trip to the remote side, we
     * need to be a bit cautious here.  It would sure be nice to have a local
     * cache of information about remote index definitions...
     */
    foreach(lc, useful_eclass_list)
    {
        EquivalenceClass *cur_ec = lfirst(lc);
        PathKey    *pathkey;
 
        /* If redundant with what we did above, skip it. */
        if (cur_ec == query_ec)
            continue;
 
        /* Can't push down the sort if the EC's opfamily is not shippable. */
        if (!is_shippable(linitial_oid(cur_ec->ec_opfamilies),
                          OperatorFamilyRelationId, fpinfo))
            continue;
 
        /* If no pushable expression for this rel, skip it. */
        if (find_em_for_rel(root, cur_ec, rel) == NULL)
            continue;
 
        /* Looks like we can generate a pathkey, so let's do it. */
        pathkey = make_canonical_pathkey(root, cur_ec,
                                         linitial_oid(cur_ec->ec_opfamilies),
                                         COMPARE_LT,
                                         false);
        useful_pathkeys_list = lappend(useful_pathkeys_list,
                                       list_make1(pathkey));
    }
 
    return useful_pathkeys_list;
}
 
/*
 * postgresGetForeignPaths
 *      Create possible scan paths for a scan on the foreign table
 */
static void
postgresGetForeignPaths(PlannerInfo *root,
                        RelOptInfo *baserel,
                        Oid foreigntableid)
{
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) baserel->fdw_private;
    ForeignPath *path;
    List       *ppi_list;
    ListCell   *lc;
 
    /*
     * Create simplest ForeignScan path node and add it to baserel.  This path
     * corresponds to SeqScan path of regular tables (though depending on what
     * baserestrict conditions we were able to send to remote, there might
     * actually be an indexscan happening there).  We already did all the work
     * to estimate cost and size of this path.
     *
     * Although this path uses no join clauses, it could still have required
     * parameterization due to LATERAL refs in its tlist.
     */
    path = create_foreignscan_path(root, baserel,
                                   NULL,    /* default pathtarget */
                                   fpinfo->rows,
                                   fpinfo->disabled_nodes,
                                   fpinfo->startup_cost,
                                   fpinfo->total_cost,
                                   NIL, /* no pathkeys */
                                   baserel->lateral_relids,
                                   NULL,    /* no extra plan */
                                   NIL, /* no fdw_restrictinfo list */
                                   NIL);    /* no fdw_private list */
    add_path(baserel, (Path *) path);
 
    /* Add paths with pathkeys */
    add_paths_with_pathkeys_for_rel(root, baserel, NULL, NIL);
 
    /*
     * If we're not using remote estimates, stop here.  We have no way to
     * estimate whether any join clauses would be worth sending across, so
     * don't bother building parameterized paths.
     */
    if (!fpinfo->use_remote_estimate)
        return;
 
    /*
     * Thumb through all join clauses for the rel to identify which outer
     * relations could supply one or more safe-to-send-to-remote join clauses.
     * We'll build a parameterized path for each such outer relation.
     *
     * It's convenient to manage this by representing each candidate outer
     * relation by the ParamPathInfo node for it.  We can then use the
     * ppi_clauses list in the ParamPathInfo node directly as a list of the
     * interesting join clauses for that rel.  This takes care of the
     * possibility that there are multiple safe join clauses for such a rel,
     * and also ensures that we account for unsafe join clauses that we'll
     * still have to enforce locally (since the parameterized-path machinery
     * insists that we handle all movable clauses).
     */
    ppi_list = NIL;
    foreach(lc, baserel->joininfo)
    {
        RestrictInfo *rinfo = (RestrictInfo *) lfirst(lc);
        Relids      required_outer;
        ParamPathInfo *param_info;
 
        /* Check if clause can be moved to this rel */
        if (!join_clause_is_movable_to(rinfo, baserel))
            continue;
 
        /* See if it is safe to send to remote */
        if (!is_foreign_expr(root, baserel, rinfo->clause))
            continue;
 
        /* Calculate required outer rels for the resulting path */
        required_outer = bms_union(rinfo->clause_relids,
                                   baserel->lateral_relids);
        /* We do not want the foreign rel itself listed in required_outer */
        required_outer = bms_del_member(required_outer, baserel->relid);
 
        /*
         * required_outer probably can't be empty here, but if it were, we
         * couldn't make a parameterized path.
         */
        if (bms_is_empty(required_outer))
            continue;
 
        /* Get the ParamPathInfo */
        param_info = get_baserel_parampathinfo(root, baserel,
                                               required_outer);
        Assert(param_info != NULL);
 
        /*
         * Add it to list unless we already have it.  Testing pointer equality
         * is OK since get_baserel_parampathinfo won't make duplicates.
         */
        ppi_list = list_append_unique_ptr(ppi_list, param_info);
    }
 
    /*
     * The above scan examined only "generic" join clauses, not those that
     * were absorbed into EquivalenceClauses.  See if we can make anything out
     * of EquivalenceClauses.
     */
    if (baserel->has_eclass_joins)
    {
        /*
         * We repeatedly scan the eclass list looking for column references
         * (or expressions) belonging to the foreign rel.  Each time we find
         * one, we generate a list of equivalence joinclauses for it, and then
         * see if any are safe to send to the remote.  Repeat till there are
         * no more candidate EC members.
         */
        ec_member_foreign_arg arg;
 
        arg.already_used = NIL;
        for (;;)
        {
            List       *clauses;
 
            /* Make clauses, skipping any that join to lateral_referencers */
            arg.current = NULL;
            clauses = generate_implied_equalities_for_column(root,
                                                             baserel,
                                                             ec_member_matches_foreign,
                                                             &arg,
                                                             baserel->lateral_referencers);
 
            /* Done if there are no more expressions in the foreign rel */
            if (arg.current == NULL)
            {
                Assert(clauses == NIL);
                break;
            }
 
            /* Scan the extracted join clauses */
            foreach(lc, clauses)
            {
                RestrictInfo *rinfo = (RestrictInfo *) lfirst(lc);
                Relids      required_outer;
                ParamPathInfo *param_info;
 
                /* Check if clause can be moved to this rel */
                if (!join_clause_is_movable_to(rinfo, baserel))
                    continue;
 
                /* See if it is safe to send to remote */
                if (!is_foreign_expr(root, baserel, rinfo->clause))
                    continue;
 
                /* Calculate required outer rels for the resulting path */
                required_outer = bms_union(rinfo->clause_relids,
                                           baserel->lateral_relids);
                required_outer = bms_del_member(required_outer, baserel->relid);
                if (bms_is_empty(required_outer))
                    continue;
 
                /* Get the ParamPathInfo */
                param_info = get_baserel_parampathinfo(root, baserel,
                                                       required_outer);
                Assert(param_info != NULL);
 
                /* Add it to list unless we already have it */
                ppi_list = list_append_unique_ptr(ppi_list, param_info);
            }
 
            /* Try again, now ignoring the expression we found this time */
            arg.already_used = lappend(arg.already_used, arg.current);
        }
    }
 
    /*
     * Now build a path for each useful outer relation.
     */
    foreach(lc, ppi_list)
    {
        ParamPathInfo *param_info = (ParamPathInfo *) lfirst(lc);
        double      rows;
        int         width;
        int         disabled_nodes;
        Cost        startup_cost;
        Cost        total_cost;
 
        /* Get a cost estimate from the remote */
        estimate_path_cost_size(root, baserel,
                                param_info->ppi_clauses, NIL, NULL,
                                &rows, &width, &disabled_nodes,
                                &startup_cost, &total_cost);
 
        /*
         * ppi_rows currently won't get looked at by anything, but still we
         * may as well ensure that it matches our idea of the rowcount.
         */
        param_info->ppi_rows = rows;
 
        /* Make the path */
        path = create_foreignscan_path(root, baserel,
                                       NULL,    /* default pathtarget */
                                       rows,
                                       disabled_nodes,
                                       startup_cost,
                                       total_cost,
                                       NIL, /* no pathkeys */
                                       param_info->ppi_req_outer,
                                       NULL,
                                       NIL, /* no fdw_restrictinfo list */
                                       NIL);    /* no fdw_private list */
        add_path(baserel, (Path *) path);
    }
}
 
/*
 * postgresGetForeignPlan
 *      Create ForeignScan plan node which implements selected best path
 */
static ForeignScan *
postgresGetForeignPlan(PlannerInfo *root,
                       RelOptInfo *foreignrel,
                       Oid foreigntableid,
                       ForeignPath *best_path,
                       List *tlist,
                       List *scan_clauses,
                       Plan *outer_plan)
{
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) foreignrel->fdw_private;
    Index       scan_relid;
    List       *fdw_private;
    List       *remote_exprs = NIL;
    List       *local_exprs = NIL;
    List       *params_list = NIL;
    List       *fdw_scan_tlist = NIL;
    List       *fdw_recheck_quals = NIL;
    List       *retrieved_attrs;
    StringInfoData sql;
    bool        has_final_sort = false;
    bool        has_limit = false;
    ListCell   *lc;
 
    /*
     * Get FDW private data created by postgresGetForeignUpperPaths(), if any.
     */
    if (best_path->fdw_private)
    {
        has_final_sort = boolVal(list_nth(best_path->fdw_private,
                                          FdwPathPrivateHasFinalSort));
        has_limit = boolVal(list_nth(best_path->fdw_private,
                                     FdwPathPrivateHasLimit));
    }
 
    if (IS_SIMPLE_REL(foreignrel))
    {
        /*
         * For base relations, set scan_relid as the relid of the relation.
         */
        scan_relid = foreignrel->relid;
 
        /*
         * In a base-relation scan, we must apply the given scan_clauses.
         *
         * Separate the scan_clauses into those that can be executed remotely
         * and those that can't.  baserestrictinfo clauses that were
         * previously determined to be safe or unsafe by classifyConditions
         * are found in fpinfo->remote_conds and fpinfo->local_conds. Anything
         * else in the scan_clauses list will be a join clause, which we have
         * to check for remote-safety.
         *
         * Note: the join clauses we see here should be the exact same ones
         * previously examined by postgresGetForeignPaths.  Possibly it'd be
         * worth passing forward the classification work done then, rather
         * than repeating it here.
         *
         * This code must match "extract_actual_clauses(scan_clauses, false)"
         * except for the additional decision about remote versus local
         * execution.
         */
        foreach(lc, scan_clauses)
        {
            RestrictInfo *rinfo = lfirst_node(RestrictInfo, lc);
 
            /* Ignore any pseudoconstants, they're dealt with elsewhere */
            if (rinfo->pseudoconstant)
                continue;
 
            if (list_member_ptr(fpinfo->remote_conds, rinfo))
                remote_exprs = lappend(remote_exprs, rinfo->clause);
            else if (list_member_ptr(fpinfo->local_conds, rinfo))
                local_exprs = lappend(local_exprs, rinfo->clause);
            else if (is_foreign_expr(root, foreignrel, rinfo->clause))
                remote_exprs = lappend(remote_exprs, rinfo->clause);
            else
                local_exprs = lappend(local_exprs, rinfo->clause);
        }
 
        /*
         * For a base-relation scan, we have to support EPQ recheck, which
         * should recheck all the remote quals.
         */
        fdw_recheck_quals = remote_exprs;
    }
    else
    {
        /*
         * Join relation or upper relation - set scan_relid to 0.
         */
        scan_relid = 0;
 
        /*
         * For a join rel, baserestrictinfo is NIL and we are not considering
         * parameterization right now, so there should be no scan_clauses for
         * a joinrel or an upper rel either.
         */
        Assert(!scan_clauses);
 
        /*
         * Instead we get the conditions to apply from the fdw_private
         * structure.
         */
        remote_exprs = extract_actual_clauses(fpinfo->remote_conds, false);
        local_exprs = extract_actual_clauses(fpinfo->local_conds, false);
 
        /*
         * We leave fdw_recheck_quals empty in this case, since we never need
         * to apply EPQ recheck clauses.  In the case of a joinrel, EPQ
         * recheck is handled elsewhere --- see postgresGetForeignJoinPaths().
         * If we're planning an upperrel (ie, remote grouping or aggregation)
         * then there's no EPQ to do because SELECT FOR UPDATE wouldn't be
         * allowed, and indeed we *can't* put the remote clauses into
         * fdw_recheck_quals because the unaggregated Vars won't be available
         * locally.
         */
 
        /* Build the list of columns to be fetched from the foreign server. */
        fdw_scan_tlist = build_tlist_to_deparse(foreignrel);
 
        /*
         * Ensure that the outer plan produces a tuple whose descriptor
         * matches our scan tuple slot.  Also, remove the local conditions
         * from outer plan's quals, lest they be evaluated twice, once by the
         * local plan and once by the scan.
         */
        if (outer_plan)
        {
            /*
             * Right now, we only consider grouping and aggregation beyond
             * joins. Queries involving aggregates or grouping do not require
             * EPQ mechanism, hence should not have an outer plan here.
             */
            Assert(!IS_UPPER_REL(foreignrel));
 
            /*
             * First, update the plan's qual list if possible.  In some cases
             * the quals might be enforced below the topmost plan level, in
             * which case we'll fail to remove them; it's not worth working
             * harder than this.
             */
            foreach(lc, local_exprs)
            {
                Node       *qual = lfirst(lc);
 
                outer_plan->qual = list_delete(outer_plan->qual, qual);
 
                /*
                 * For an inner join the local conditions of foreign scan plan
                 * can be part of the joinquals as well.  (They might also be
                 * in the mergequals or hashquals, but we can't touch those
                 * without breaking the plan.)
                 */
                if (IsA(outer_plan, NestLoop) ||
                    IsA(outer_plan, MergeJoin) ||
                    IsA(outer_plan, HashJoin))
                {
                    Join       *join_plan = (Join *) outer_plan;
 
                    if (join_plan->jointype == JOIN_INNER)
                        join_plan->joinqual = list_delete(join_plan->joinqual,
                                                          qual);
                }
            }
 
            /*
             * Now fix the subplan's tlist --- this might result in inserting
             * a Result node atop the plan tree.
             */
            outer_plan = change_plan_targetlist(outer_plan, fdw_scan_tlist,
                                                best_path->path.parallel_safe);
        }
    }
 
    /*
     * Build the query string to be sent for execution, and identify
     * expressions to be sent as parameters.
     */
    initStringInfo(&sql);
    deparseSelectStmtForRel(&sql, root, foreignrel, fdw_scan_tlist,
                            remote_exprs, best_path->path.pathkeys,
                            has_final_sort, has_limit, false,
                            &retrieved_attrs, &params_list);
 
    /* Remember remote_exprs for possible use by postgresPlanDirectModify */
    fpinfo->final_remote_exprs = remote_exprs;
 
    /*
     * Build the fdw_private list that will be available to the executor.
     * Items in the list must match order in enum FdwScanPrivateIndex.
     */
    fdw_private = list_make3(makeString(sql.data),
                             retrieved_attrs,
                             makeInteger(fpinfo->fetch_size));
    if (IS_JOIN_REL(foreignrel) || IS_UPPER_REL(foreignrel))
        fdw_private = lappend(fdw_private,
                              makeString(fpinfo->relation_name));
 
    /*
     * Create the ForeignScan node for the given relation.
     *
     * Note that the remote parameter expressions are stored in the fdw_exprs
     * field of the finished plan node; we can't keep them in private state
     * because then they wouldn't be subject to later planner processing.
     */
    return make_foreignscan(tlist,
                            local_exprs,
                            scan_relid,
                            params_list,
                            fdw_private,
                            fdw_scan_tlist,
                            fdw_recheck_quals,
                            outer_plan);
}
 
/*
 * Construct a tuple descriptor for the scan tuples handled by a foreign join.
 */
static TupleDesc
get_tupdesc_for_join_scan_tuples(ForeignScanState *node)
{
    ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan;
    EState     *estate = node->ss.ps.state;
    TupleDesc   tupdesc;
 
    /*
     * The core code has already set up a scan tuple slot based on
     * fsplan->fdw_scan_tlist, and this slot's tupdesc is mostly good enough,
     * but there's one case where it isn't.  If we have any whole-row row
     * identifier Vars, they may have vartype RECORD, and we need to replace
     * that with the associated table's actual composite type.  This ensures
     * that when we read those ROW() expression values from the remote server,
     * we can convert them to a composite type the local server knows.
     */
    tupdesc = CreateTupleDescCopy(node->ss.ss_ScanTupleSlot->tts_tupleDescriptor);
    for (int i = 0; i < tupdesc->natts; i++)
    {
        Form_pg_attribute att = TupleDescAttr(tupdesc, i);
        Var        *var;
        RangeTblEntry *rte;
        Oid         reltype;
 
        /* Nothing to do if it's not a generic RECORD attribute */
        if (att->atttypid != RECORDOID || att->atttypmod >= 0)
            continue;
 
        /*
         * If we can't identify the referenced table, do nothing.  This'll
         * likely lead to failure later, but perhaps we can muddle through.
         */
        var = (Var *) list_nth_node(TargetEntry, fsplan->fdw_scan_tlist,
                                    i)->expr;
        if (!IsA(var, Var) || var->varattno != 0)
            continue;
        rte = list_nth(estate->es_range_table, var->varno - 1);
        if (rte->rtekind != RTE_RELATION)
            continue;
        reltype = get_rel_type_id(rte->relid);
        if (!OidIsValid(reltype))
            continue;
        att->atttypid = reltype;
        /* shouldn't need to change anything else */
    }
    return tupdesc;
}
 
/*
 * postgresBeginForeignScan
 *      Initiate an executor scan of a foreign PostgreSQL table.
 */
static void
postgresBeginForeignScan(ForeignScanState *node, int eflags)
{
    ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan;
    EState     *estate = node->ss.ps.state;
    PgFdwScanState *fsstate;
    RangeTblEntry *rte;
    Oid         userid;
    ForeignTable *table;
    UserMapping *user;
    int         rtindex;
    int         numParams;
 
    /*
     * Do nothing in EXPLAIN (no ANALYZE) case.  node->fdw_state stays NULL.
     */
    if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
        return;
 
    /*
     * We'll save private state in node->fdw_state.
     */
    fsstate = palloc0_object(PgFdwScanState);
    node->fdw_state = fsstate;
 
    /*
     * Identify which user to do the remote access as.  This should match what
     * ExecCheckPermissions() does.
     */
    userid = OidIsValid(fsplan->checkAsUser) ? fsplan->checkAsUser : GetUserId();
    if (fsplan->scan.scanrelid > 0)
        rtindex = fsplan->scan.scanrelid;
    else
        rtindex = bms_next_member(fsplan->fs_base_relids, -1);
    rte = exec_rt_fetch(rtindex, estate);
 
    /* Get info about foreign table. */
    table = GetForeignTable(rte->relid);
    user = GetUserMapping(userid, table->serverid);
 
    /*
     * Get connection to the foreign server.  Connection manager will
     * establish new connection if necessary.
     */
    fsstate->conn = GetConnection(user, false, &fsstate->conn_state);
 
    /* Assign a unique ID for my cursor */
    fsstate->cursor_number = GetCursorNumber(fsstate->conn);
    fsstate->cursor_exists = false;
 
    /* Get private info created by planner functions. */
    fsstate->query = strVal(list_nth(fsplan->fdw_private,
                                     FdwScanPrivateSelectSql));
    fsstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private,
                                                 FdwScanPrivateRetrievedAttrs);
    fsstate->fetch_size = intVal(list_nth(fsplan->fdw_private,
                                          FdwScanPrivateFetchSize));
 
    /* Create contexts for batches of tuples and per-tuple temp workspace. */
    fsstate->batch_cxt = AllocSetContextCreate(estate->es_query_cxt,
                                               "postgres_fdw tuple data",
                                               ALLOCSET_DEFAULT_SIZES);
    fsstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
                                              "postgres_fdw temporary data",
                                              ALLOCSET_SMALL_SIZES);
 
    /*
     * Get info we'll need for converting data fetched from the foreign server
     * into local representation and error reporting during that process.
     */
    if (fsplan->scan.scanrelid > 0)
    {
        fsstate->rel = node->ss.ss_currentRelation;
        fsstate->tupdesc = RelationGetDescr(fsstate->rel);
    }
    else
    {
        fsstate->rel = NULL;
        fsstate->tupdesc = get_tupdesc_for_join_scan_tuples(node);
    }
 
    fsstate->attinmeta = TupleDescGetAttInMetadata(fsstate->tupdesc);
 
    /*
     * Prepare for processing of parameters used in remote query, if any.
     */
    numParams = list_length(fsplan->fdw_exprs);
    fsstate->numParams = numParams;
    if (numParams > 0)
        prepare_query_params((PlanState *) node,
                             fsplan->fdw_exprs,
                             numParams,
                             &fsstate->param_flinfo,
                             &fsstate->param_exprs,
                             &fsstate->param_values);
 
    /* Set the async-capable flag */
    fsstate->async_capable = node->ss.ps.async_capable;
}
 
/*
 * postgresIterateForeignScan
 *      Retrieve next row from the result set, or clear tuple slot to indicate
 *      EOF.
 */
static TupleTableSlot *
postgresIterateForeignScan(ForeignScanState *node)
{
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
    TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
 
    /*
     * In sync mode, if this is the first call after Begin or ReScan, we need
     * to create the cursor on the remote side.  In async mode, we would have
     * already created the cursor before we get here, even if this is the
     * first call after Begin or ReScan.
     */
    if (!fsstate->cursor_exists)
        create_cursor(node);
 
    /*
     * Get some more tuples, if we've run out.
     */
    if (fsstate->next_tuple >= fsstate->num_tuples)
    {
        /* In async mode, just clear tuple slot. */
        if (fsstate->async_capable)
            return ExecClearTuple(slot);
        /* No point in another fetch if we already detected EOF, though. */
        if (!fsstate->eof_reached)
            fetch_more_data(node);
        /* If we didn't get any tuples, must be end of data. */
        if (fsstate->next_tuple >= fsstate->num_tuples)
            return ExecClearTuple(slot);
    }
 
    /*
     * Return the next tuple.
     */
    ExecStoreHeapTuple(fsstate->tuples[fsstate->next_tuple++],
                       slot,
                       false);
 
    return slot;
}
 
/*
 * postgresReScanForeignScan
 *      Restart the scan.
 */
static void
postgresReScanForeignScan(ForeignScanState *node)
{
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
    char        sql[64];
    PGresult   *res;
 
    /* If we haven't created the cursor yet, nothing to do. */
    if (!fsstate->cursor_exists)
        return;
 
    /*
     * If the node is async-capable, and an asynchronous fetch for it has
     * begun, the asynchronous fetch might not have yet completed.  Check if
     * the node is async-capable, and an asynchronous fetch for it is still in
     * progress; if so, complete the asynchronous fetch before restarting the
     * scan.
     */
    if (fsstate->async_capable &&
        fsstate->conn_state->pendingAreq &&
        fsstate->conn_state->pendingAreq->requestee == (PlanState *) node)
        fetch_more_data(node);
 
    /*
     * If any internal parameters affecting this node have changed, we'd
     * better destroy and recreate the cursor.  Otherwise, if the remote
     * server is v14 or older, rewinding it should be good enough; if not,
     * rewind is only allowed for scrollable cursors, but we don't have a way
     * to check the scrollability of it, so destroy and recreate it in any
     * case.  If we've only fetched zero or one batch, we needn't even rewind
     * the cursor, just rescan what we have.
     */
    if (node->ss.ps.chgParam != NULL)
    {
        fsstate->cursor_exists = false;
        snprintf(sql, sizeof(sql), "CLOSE c%u",
                 fsstate->cursor_number);
    }
    else if (fsstate->fetch_ct_2 > 1)
    {
        if (PQserverVersion(fsstate->conn) < 150000)
            snprintf(sql, sizeof(sql), "MOVE BACKWARD ALL IN c%u",
                     fsstate->cursor_number);
        else
        {
            fsstate->cursor_exists = false;
            snprintf(sql, sizeof(sql), "CLOSE c%u",
                     fsstate->cursor_number);
        }
    }
    else
    {
        /* Easy: just rescan what we already have in memory, if anything */
        fsstate->next_tuple = 0;
        return;
    }
 
    res = pgfdw_exec_query(fsstate->conn, sql, fsstate->conn_state);
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pgfdw_report_error(res, fsstate->conn, sql);
    PQclear(res);
 
    /* Now force a fresh FETCH. */
    fsstate->tuples = NULL;
    fsstate->num_tuples = 0;
    fsstate->next_tuple = 0;
    fsstate->fetch_ct_2 = 0;
    fsstate->eof_reached = false;
}
 
/*
 * postgresEndForeignScan
 *      Finish scanning foreign table and dispose objects used for this scan
 */
static void
postgresEndForeignScan(ForeignScanState *node)
{
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
 
    /* if fsstate is NULL, we are in EXPLAIN; nothing to do */
    if (fsstate == NULL)
        return;
 
    /* Close the cursor if open, to prevent accumulation of cursors */
    if (fsstate->cursor_exists)
        close_cursor(fsstate->conn, fsstate->cursor_number,
                     fsstate->conn_state);
 
    /* Release remote connection */
    ReleaseConnection(fsstate->conn);
    fsstate->conn = NULL;
 
    /* MemoryContexts will be deleted automatically. */
}
 
/*
 * postgresAddForeignUpdateTargets
 *      Add resjunk column(s) needed for update/delete on a foreign table
 */
static void
postgresAddForeignUpdateTargets(PlannerInfo *root,
                                Index rtindex,
                                RangeTblEntry *target_rte,
                                Relation target_relation)
{
    Var        *var;
 
    /*
     * In postgres_fdw, what we need is the ctid, same as for a regular table.
     */
 
    /* Make a Var representing the desired value */
    var = makeVar(rtindex,
                  SelfItemPointerAttributeNumber,
                  TIDOID,
                  -1,
                  InvalidOid,
                  0);
 
    /* Register it as a row-identity column needed by this target rel */
    add_row_identity_var(root, var, rtindex, "ctid");
}
 
/*
 * postgresPlanForeignModify
 *      Plan an insert/update/delete operation on a foreign table
 */
static List *
postgresPlanForeignModify(PlannerInfo *root,
                          ModifyTable *plan,
                          Index resultRelation,
                          int subplan_index)
{
    CmdType     operation = plan->operation;
    RangeTblEntry *rte = planner_rt_fetch(resultRelation, root);
    Relation    rel;
    StringInfoData sql;
    List       *targetAttrs = NIL;
    List       *withCheckOptionList = NIL;
    List       *returningList = NIL;
    List       *retrieved_attrs = NIL;
    bool        doNothing = false;
    int         values_end_len = -1;
 
    initStringInfo(&sql);
 
    /*
     * Core code already has some lock on each rel being planned, so we can
     * use NoLock here.
     */
    rel = table_open(rte->relid, NoLock);
 
    /*
     * In an INSERT, we transmit all columns that are defined in the foreign
     * table.  In an UPDATE, if there are BEFORE ROW UPDATE triggers on the
     * foreign table, we transmit all columns like INSERT; else we transmit
     * only columns that were explicitly targets of the UPDATE, so as to avoid
     * unnecessary data transmission.  (We can't do that for INSERT since we
     * would miss sending default values for columns not listed in the source
     * statement, and for UPDATE if there are BEFORE ROW UPDATE triggers since
     * those triggers might change values for non-target columns, in which
     * case we would miss sending changed values for those columns.)
     */
    if (operation == CMD_INSERT ||
        (operation == CMD_UPDATE &&
         rel->trigdesc &&
         rel->trigdesc->trig_update_before_row))
    {
        TupleDesc   tupdesc = RelationGetDescr(rel);
        int         attnum;
 
        for (attnum = 1; attnum <= tupdesc->natts; attnum++)
        {
            CompactAttribute *attr = TupleDescCompactAttr(tupdesc, attnum - 1);
 
            if (!attr->attisdropped)
                targetAttrs = lappend_int(targetAttrs, attnum);
        }
    }
    else if (operation == CMD_UPDATE)
    {
        int         col;
        RelOptInfo *rel = find_base_rel(root, resultRelation);
        Bitmapset  *allUpdatedCols = get_rel_all_updated_cols(root, rel);
 
        col = -1;
        while ((col = bms_next_member(allUpdatedCols, col)) >= 0)
        {
            /* bit numbers are offset by FirstLowInvalidHeapAttributeNumber */
            AttrNumber  attno = col + FirstLowInvalidHeapAttributeNumber;
 
            if (attno <= InvalidAttrNumber) /* shouldn't happen */
                elog(ERROR, "system-column update is not supported");
            targetAttrs = lappend_int(targetAttrs, attno);
        }
    }
 
    /*
     * Extract the relevant WITH CHECK OPTION list if any.
     */
    if (plan->withCheckOptionLists)
        withCheckOptionList = (List *) list_nth(plan->withCheckOptionLists,
                                                subplan_index);
 
    /*
     * Extract the relevant RETURNING list if any.
     */
    if (plan->returningLists)
        returningList = (List *) list_nth(plan->returningLists, subplan_index);
 
    /*
     * ON CONFLICT DO NOTHING/SELECT/UPDATE with inference specification
     * should have already been rejected in the optimizer, as presently there
     * is no way to recognize an arbiter index on a foreign table.  Only DO
     * NOTHING is supported without an inference specification.
     */
    if (plan->onConflictAction == ONCONFLICT_NOTHING)
        doNothing = true;
    else if (plan->onConflictAction != ONCONFLICT_NONE)
        elog(ERROR, "unexpected ON CONFLICT specification: %d",
             (int) plan->onConflictAction);
 
    /*
     * Construct the SQL command string.
     */
    switch (operation)
    {
        case CMD_INSERT:
            deparseInsertSql(&sql, rte, resultRelation, rel,
                             targetAttrs, doNothing,
                             withCheckOptionList, returningList,
                             &retrieved_attrs, &values_end_len);
            break;
        case CMD_UPDATE:
            deparseUpdateSql(&sql, rte, resultRelation, rel,
                             targetAttrs,
                             withCheckOptionList, returningList,
                             &retrieved_attrs);
            break;
        case CMD_DELETE:
            deparseDeleteSql(&sql, rte, resultRelation, rel,
                             returningList,
                             &retrieved_attrs);
            break;
        default:
            elog(ERROR, "unexpected operation: %d", (int) operation);
            break;
    }
 
    table_close(rel, NoLock);
 
    /*
     * Build the fdw_private list that will be available to the executor.
     * Items in the list must match enum FdwModifyPrivateIndex, above.
     */
    return list_make5(makeString(sql.data),
                      targetAttrs,
                      makeInteger(values_end_len),
                      makeBoolean((retrieved_attrs != NIL)),
                      retrieved_attrs);
}
 
/*
 * postgresBeginForeignModify
 *      Begin an insert/update/delete operation on a foreign table
 */
static void
postgresBeginForeignModify(ModifyTableState *mtstate,
                           ResultRelInfo *resultRelInfo,
                           List *fdw_private,
                           int subplan_index,
                           int eflags)
{
    PgFdwModifyState *fmstate;
    char       *query;
    List       *target_attrs;
    bool        has_returning;
    int         values_end_len;
    List       *retrieved_attrs;
    RangeTblEntry *rte;
 
    /*
     * Do nothing in EXPLAIN (no ANALYZE) case.  resultRelInfo->ri_FdwState
     * stays NULL.
     */
    if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
        return;
 
    /* Deconstruct fdw_private data. */
    query = strVal(list_nth(fdw_private,
                            FdwModifyPrivateUpdateSql));
    target_attrs = (List *) list_nth(fdw_private,
                                     FdwModifyPrivateTargetAttnums);
    values_end_len = intVal(list_nth(fdw_private,
                                     FdwModifyPrivateLen));
    has_returning = boolVal(list_nth(fdw_private,
                                     FdwModifyPrivateHasReturning));
    retrieved_attrs = (List *) list_nth(fdw_private,
                                        FdwModifyPrivateRetrievedAttrs);
 
    /* Find RTE. */
    rte = exec_rt_fetch(resultRelInfo->ri_RangeTableIndex,
                        mtstate->ps.state);
 
    /* Construct an execution state. */
    fmstate = create_foreign_modify(mtstate->ps.state,
                                    rte,
                                    resultRelInfo,
                                    mtstate->operation,
                                    outerPlanState(mtstate)->plan,
                                    query,
                                    target_attrs,
                                    values_end_len,
                                    has_returning,
                                    retrieved_attrs);
 
    resultRelInfo->ri_FdwState = fmstate;
}
 
/*
 * postgresExecForeignInsert
 *      Insert one row into a foreign table
 */
static TupleTableSlot *
postgresExecForeignInsert(EState *estate,
                          ResultRelInfo *resultRelInfo,
                          TupleTableSlot *slot,
                          TupleTableSlot *planSlot)
{
    PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
    TupleTableSlot **rslot;
    int         numSlots = 1;
 
    /*
     * If the fmstate has aux_fmstate set, use the aux_fmstate (see
     * postgresBeginForeignInsert())
     */
    if (fmstate->aux_fmstate)
        resultRelInfo->ri_FdwState = fmstate->aux_fmstate;
    rslot = execute_foreign_modify(estate, resultRelInfo, CMD_INSERT,
                                   &slot, &planSlot, &numSlots);
    /* Revert that change */
    if (fmstate->aux_fmstate)
        resultRelInfo->ri_FdwState = fmstate;
 
    return rslot ? *rslot : NULL;
}
 
/*
 * postgresExecForeignBatchInsert
 *      Insert multiple rows into a foreign table
 */
static TupleTableSlot **
postgresExecForeignBatchInsert(EState *estate,
                               ResultRelInfo *resultRelInfo,
                               TupleTableSlot **slots,
                               TupleTableSlot **planSlots,
                               int *numSlots)
{
    PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
    TupleTableSlot **rslot;
 
    /*
     * If the fmstate has aux_fmstate set, use the aux_fmstate (see
     * postgresBeginForeignInsert())
     */
    if (fmstate->aux_fmstate)
        resultRelInfo->ri_FdwState = fmstate->aux_fmstate;
    rslot = execute_foreign_modify(estate, resultRelInfo, CMD_INSERT,
                                   slots, planSlots, numSlots);
    /* Revert that change */
    if (fmstate->aux_fmstate)
        resultRelInfo->ri_FdwState = fmstate;
 
    return rslot;
}
 
/*
 * postgresGetForeignModifyBatchSize
 *      Determine the maximum number of tuples that can be inserted in bulk
 *
 * Returns the batch size specified for server or table. When batching is not
 * allowed (e.g. for tables with BEFORE/AFTER ROW triggers or with RETURNING
 * clause), returns 1.
 */
static int
postgresGetForeignModifyBatchSize(ResultRelInfo *resultRelInfo)
{
    int         batch_size;
    PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
 
    /* should be called only once */
    Assert(resultRelInfo->ri_BatchSize == 0);
 
    /*
     * Should never get called when the insert is being performed on a table
     * that is also among the target relations of an UPDATE operation, because
     * postgresBeginForeignInsert() currently rejects such insert attempts.
     */
    Assert(fmstate == NULL || fmstate->aux_fmstate == NULL);
 
    /*
     * In EXPLAIN without ANALYZE, ri_FdwState is NULL, so we have to lookup
     * the option directly in server/table options. Otherwise just use the
     * value we determined earlier.
     */
    if (fmstate)
        batch_size = fmstate->batch_size;
    else
        batch_size = get_batch_size_option(resultRelInfo->ri_RelationDesc);
 
    /*
     * Disable batching when we have to use RETURNING, there are any
     * BEFORE/AFTER ROW INSERT triggers on the foreign table, or there are any
     * WITH CHECK OPTION constraints from parent views.
     *
     * When there are any BEFORE ROW INSERT triggers on the table, we can't
     * support it, because such triggers might query the table we're inserting
     * into and act differently if the tuples that have already been processed
     * and prepared for insertion are not there.
     */
    if (resultRelInfo->ri_projectReturning != NULL ||
        resultRelInfo->ri_WithCheckOptions != NIL ||
        (resultRelInfo->ri_TrigDesc &&
         (resultRelInfo->ri_TrigDesc->trig_insert_before_row ||
          resultRelInfo->ri_TrigDesc->trig_insert_after_row)))
        return 1;
 
    /*
     * If the foreign table has no columns, disable batching as the INSERT
     * syntax doesn't allow batching multiple empty rows into a zero-column
     * table in a single statement.  This is needed for COPY FROM, in which
     * case fmstate must be non-NULL.
     */
    if (fmstate && list_length(fmstate->target_attrs) == 0)
        return 1;
 
    /*
     * Otherwise use the batch size specified for server/table. The number of
     * parameters in a batch is limited to 65535 (uint16), so make sure we
     * don't exceed this limit by using the maximum batch_size possible.
     */
    if (fmstate && fmstate->p_nums > 0)
        batch_size = Min(batch_size, PQ_QUERY_PARAM_MAX_LIMIT / fmstate->p_nums);
 
    return batch_size;
}
 
/*
 * postgresExecForeignUpdate
 *      Update one row in a foreign table
 */
static TupleTableSlot *
postgresExecForeignUpdate(EState *estate,
                          ResultRelInfo *resultRelInfo,
                          TupleTableSlot *slot,
                          TupleTableSlot *planSlot)
{
    TupleTableSlot **rslot;
    int         numSlots = 1;
 
    rslot = execute_foreign_modify(estate, resultRelInfo, CMD_UPDATE,
                                   &slot, &planSlot, &numSlots);
 
    return rslot ? rslot[0] : NULL;
}
 
/*
 * postgresExecForeignDelete
 *      Delete one row from a foreign table
 */
static TupleTableSlot *
postgresExecForeignDelete(EState *estate,
                          ResultRelInfo *resultRelInfo,
                          TupleTableSlot *slot,
                          TupleTableSlot *planSlot)
{
    TupleTableSlot **rslot;
    int         numSlots = 1;
 
    rslot = execute_foreign_modify(estate, resultRelInfo, CMD_DELETE,
                                   &slot, &planSlot, &numSlots);
 
    return rslot ? rslot[0] : NULL;
}
 
/*
 * postgresEndForeignModify
 *      Finish an insert/update/delete operation on a foreign table
 */
static void
postgresEndForeignModify(EState *estate,
                         ResultRelInfo *resultRelInfo)
{
    PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
 
    /* If fmstate is NULL, we are in EXPLAIN; nothing to do */
    if (fmstate == NULL)
        return;
 
    /* Destroy the execution state */
    finish_foreign_modify(fmstate);
}
 
/*
 * postgresBeginForeignInsert
 *      Begin an insert operation on a foreign table
 */
static void
postgresBeginForeignInsert(ModifyTableState *mtstate,
                           ResultRelInfo *resultRelInfo)
{
    PgFdwModifyState *fmstate;
    ModifyTable *plan = castNode(ModifyTable, mtstate->ps.plan);
    EState     *estate = mtstate->ps.state;
    Index       resultRelation;
    Relation    rel = resultRelInfo->ri_RelationDesc;
    RangeTblEntry *rte;
    TupleDesc   tupdesc = RelationGetDescr(rel);
    int         attnum;
    int         values_end_len;
    StringInfoData sql;
    List       *targetAttrs = NIL;
    List       *retrieved_attrs = NIL;
    bool        doNothing = false;
 
    /*
     * If the foreign table we are about to insert routed rows into is also an
     * UPDATE subplan result rel that will be updated later, proceeding with
     * the INSERT will result in the later UPDATE incorrectly modifying those
     * routed rows, so prevent the INSERT --- it would be nice if we could
     * handle this case; but for now, throw an error for safety.
     */
    if (plan && plan->operation == CMD_UPDATE &&
        (resultRelInfo->ri_usesFdwDirectModify ||
         resultRelInfo->ri_FdwState))
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot route tuples into foreign table to be updated \"%s\"",
                        RelationGetRelationName(rel))));
 
    initStringInfo(&sql);
 
    /* We transmit all columns that are defined in the foreign table. */
    for (attnum = 1; attnum <= tupdesc->natts; attnum++)
    {
        CompactAttribute *attr = TupleDescCompactAttr(tupdesc, attnum - 1);
 
        if (!attr->attisdropped)
            targetAttrs = lappend_int(targetAttrs, attnum);
    }
 
    /* Check if we add the ON CONFLICT clause to the remote query. */
    if (plan)
    {
        OnConflictAction onConflictAction = plan->onConflictAction;
 
        /* We only support DO NOTHING without an inference specification. */
        if (onConflictAction == ONCONFLICT_NOTHING)
            doNothing = true;
        else if (onConflictAction != ONCONFLICT_NONE)
            elog(ERROR, "unexpected ON CONFLICT specification: %d",
                 (int) onConflictAction);
    }
 
    /*
     * If the foreign table is a partition that doesn't have a corresponding
     * RTE entry, we need to create a new RTE describing the foreign table for
     * use by deparseInsertSql and create_foreign_modify() below, after first
     * copying the parent's RTE and modifying some fields to describe the
     * foreign partition to work on. However, if this is invoked by UPDATE,
     * the existing RTE may already correspond to this partition if it is one
     * of the UPDATE subplan target rels; in that case, we can just use the
     * existing RTE as-is.
     */
    if (resultRelInfo->ri_RangeTableIndex == 0)
    {
        ResultRelInfo *rootResultRelInfo = resultRelInfo->ri_RootResultRelInfo;
 
        rte = exec_rt_fetch(rootResultRelInfo->ri_RangeTableIndex, estate);
        rte = copyObject(rte);
        rte->relid = RelationGetRelid(rel);
        rte->relkind = RELKIND_FOREIGN_TABLE;
 
        /*
         * For UPDATE, we must use the RT index of the first subplan target
         * rel's RTE, because the core code would have built expressions for
         * the partition, such as RETURNING, using that RT index as varno of
         * Vars contained in those expressions.
         */
        if (plan && plan->operation == CMD_UPDATE &&
            rootResultRelInfo->ri_RangeTableIndex == plan->rootRelation)
            resultRelation = mtstate->resultRelInfo[0].ri_RangeTableIndex;
        else
            resultRelation = rootResultRelInfo->ri_RangeTableIndex;
    }
    else
    {
        resultRelation = resultRelInfo->ri_RangeTableIndex;
        rte = exec_rt_fetch(resultRelation, estate);
    }
 
    /* Construct the SQL command string. */
    deparseInsertSql(&sql, rte, resultRelation, rel, targetAttrs, doNothing,
                     resultRelInfo->ri_WithCheckOptions,
                     resultRelInfo->ri_returningList,
                     &retrieved_attrs, &values_end_len);
 
    /* Construct an execution state. */
    fmstate = create_foreign_modify(mtstate->ps.state,
                                    rte,
                                    resultRelInfo,
                                    CMD_INSERT,
                                    NULL,
                                    sql.data,
                                    targetAttrs,
                                    values_end_len,
                                    retrieved_attrs != NIL,
                                    retrieved_attrs);
 
    /*
     * If the given resultRelInfo already has PgFdwModifyState set, it means
     * the foreign table is an UPDATE subplan result rel; in which case, store
     * the resulting state into the aux_fmstate of the PgFdwModifyState.
     */
    if (resultRelInfo->ri_FdwState)
    {
        Assert(plan && plan->operation == CMD_UPDATE);
        Assert(resultRelInfo->ri_usesFdwDirectModify == false);
        ((PgFdwModifyState *) resultRelInfo->ri_FdwState)->aux_fmstate = fmstate;
    }
    else
        resultRelInfo->ri_FdwState = fmstate;
}
 
/*
 * postgresEndForeignInsert
 *      Finish an insert operation on a foreign table
 */
static void
postgresEndForeignInsert(EState *estate,
                         ResultRelInfo *resultRelInfo)
{
    PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
 
    Assert(fmstate != NULL);
 
    /*
     * If the fmstate has aux_fmstate set, get the aux_fmstate (see
     * postgresBeginForeignInsert())
     */
    if (fmstate->aux_fmstate)
        fmstate = fmstate->aux_fmstate;
 
    /* Destroy the execution state */
    finish_foreign_modify(fmstate);
}
 
/*
 * postgresIsForeignRelUpdatable
 *      Determine whether a foreign table supports INSERT, UPDATE and/or
 *      DELETE.
 */
static int
postgresIsForeignRelUpdatable(Relation rel)
{
    bool        updatable;
    ForeignTable *table;
    ForeignServer *server;
    ListCell   *lc;
 
    /*
     * By default, all postgres_fdw foreign tables are assumed updatable. This
     * can be overridden by a per-server setting, which in turn can be
     * overridden by a per-table setting.
     */
    updatable = true;
 
    table = GetForeignTable(RelationGetRelid(rel));
    server = GetForeignServer(table->serverid);
 
    foreach(lc, server->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "updatable") == 0)
            updatable = defGetBoolean(def);
    }
    foreach(lc, table->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "updatable") == 0)
            updatable = defGetBoolean(def);
    }
 
    /*
     * Currently "updatable" means support for INSERT, UPDATE and DELETE.
     */
    return updatable ?
        (1 << CMD_INSERT) | (1 << CMD_UPDATE) | (1 << CMD_DELETE) : 0;
}
 
/*
 * postgresRecheckForeignScan
 *      Execute a local join execution plan for a foreign join
 */
static bool
postgresRecheckForeignScan(ForeignScanState *node, TupleTableSlot *slot)
{
    Index       scanrelid = ((Scan *) node->ss.ps.plan)->scanrelid;
    PlanState  *outerPlan = outerPlanState(node);
    TupleTableSlot *result;
 
    /* For base foreign relations, it suffices to set fdw_recheck_quals */
    if (scanrelid > 0)
        return true;
 
    Assert(outerPlan != NULL);
 
    /* Execute a local join execution plan */
    result = ExecProcNode(outerPlan);
    if (TupIsNull(result))
        return false;
 
    /* Store result in the given slot */
    ExecCopySlot(slot, result);
 
    return true;
}
 
/*
 * find_modifytable_subplan
 *      Helper routine for postgresPlanDirectModify to find the
 *      ModifyTable subplan node that scans the specified RTI.
 *
 * Returns NULL if the subplan couldn't be identified.  That's not a fatal
 * error condition, we just abandon trying to do the update directly.
 */
static ForeignScan *
find_modifytable_subplan(PlannerInfo *root,
                         ModifyTable *plan,
                         Index rtindex,
                         int subplan_index)
{
    Plan       *subplan = outerPlan(plan);
 
    /*
     * The cases we support are (1) the desired ForeignScan is the immediate
     * child of ModifyTable, or (2) it is the subplan_index'th child of an
     * Append node that is the immediate child of ModifyTable.  There is no
     * point in looking further down, as that would mean that local joins are
     * involved, so we can't do the update directly.
     *
     * There could be a Result atop the Append too, acting to compute the
     * UPDATE targetlist values.  We ignore that here; the tlist will be
     * checked by our caller.
     *
     * In principle we could examine all the children of the Append, but it's
     * currently unlikely that the core planner would generate such a plan
     * with the children out-of-order.  Moreover, such a search risks costing
     * O(N^2) time when there are a lot of children.
     */
    if (IsA(subplan, Append))
    {
        Append     *appendplan = (Append *) subplan;
 
        if (subplan_index < list_length(appendplan->appendplans))
            subplan = (Plan *) list_nth(appendplan->appendplans, subplan_index);
    }
    else if (IsA(subplan, Result) &&
             outerPlan(subplan) != NULL &&
             IsA(outerPlan(subplan), Append))
    {
        Append     *appendplan = (Append *) outerPlan(subplan);
 
        if (subplan_index < list_length(appendplan->appendplans))
            subplan = (Plan *) list_nth(appendplan->appendplans, subplan_index);
    }
 
    /* Now, have we got a ForeignScan on the desired rel? */
    if (IsA(subplan, ForeignScan))
    {
        ForeignScan *fscan = (ForeignScan *) subplan;
 
        if (bms_is_member(rtindex, fscan->fs_base_relids))
            return fscan;
    }
 
    return NULL;
}
 
/*
 * postgresPlanDirectModify
 *      Consider a direct foreign table modification
 *
 * Decide whether it is safe to modify a foreign table directly, and if so,
 * rewrite subplan accordingly.
 */
static bool
postgresPlanDirectModify(PlannerInfo *root,
                         ModifyTable *plan,
                         Index resultRelation,
                         int subplan_index)
{
    CmdType     operation = plan->operation;
    RelOptInfo *foreignrel;
    RangeTblEntry *rte;
    PgFdwRelationInfo *fpinfo;
    Relation    rel;
    StringInfoData sql;
    ForeignScan *fscan;
    List       *processed_tlist = NIL;
    List       *targetAttrs = NIL;
    List       *remote_exprs;
    List       *params_list = NIL;
    List       *returningList = NIL;
    List       *retrieved_attrs = NIL;
 
    /*
     * Decide whether it is safe to modify a foreign table directly.
     */
 
    /*
     * The table modification must be an UPDATE or DELETE.
     */
    if (operation != CMD_UPDATE && operation != CMD_DELETE)
        return false;
 
    /*
     * Try to locate the ForeignScan subplan that's scanning resultRelation.
     */
    fscan = find_modifytable_subplan(root, plan, resultRelation, subplan_index);
    if (!fscan)
        return false;
 
    /*
     * It's unsafe to modify a foreign table directly if there are any quals
     * that should be evaluated locally.
     */
    if (fscan->scan.plan.qual != NIL)
        return false;
 
    /* Safe to fetch data about the target foreign rel */
    if (fscan->scan.scanrelid == 0)
    {
        foreignrel = find_join_rel(root, fscan->fs_relids);
        /* We should have a rel for this foreign join. */
        Assert(foreignrel);
    }
    else
        foreignrel = root->simple_rel_array[resultRelation];
    rte = root->simple_rte_array[resultRelation];
    fpinfo = (PgFdwRelationInfo *) foreignrel->fdw_private;
 
    /*
     * It's unsafe to update a foreign table directly, if any expressions to
     * assign to the target columns are unsafe to evaluate remotely.
     */
    if (operation == CMD_UPDATE)
    {
        ListCell   *lc,
                   *lc2;
 
        /*
         * The expressions of concern are the first N columns of the processed
         * targetlist, where N is the length of the rel's update_colnos.
         */
        get_translated_update_targetlist(root, resultRelation,
                                         &processed_tlist, &targetAttrs);
        forboth(lc, processed_tlist, lc2, targetAttrs)
        {
            TargetEntry *tle = lfirst_node(TargetEntry, lc);
            AttrNumber  attno = lfirst_int(lc2);
 
            /* update's new-value expressions shouldn't be resjunk */
            Assert(!tle->resjunk);
 
            if (attno <= InvalidAttrNumber) /* shouldn't happen */
                elog(ERROR, "system-column update is not supported");
 
            if (!is_foreign_expr(root, foreignrel, (Expr *) tle->expr))
                return false;
        }
    }
 
    /*
     * Ok, rewrite subplan so as to modify the foreign table directly.
     */
    initStringInfo(&sql);
 
    /*
     * Core code already has some lock on each rel being planned, so we can
     * use NoLock here.
     */
    rel = table_open(rte->relid, NoLock);
 
    /*
     * Recall the qual clauses that must be evaluated remotely.  (These are
     * bare clauses not RestrictInfos, but deparse.c's appendConditions()
     * doesn't care.)
     */
    remote_exprs = fpinfo->final_remote_exprs;
 
    /*
     * Extract the relevant RETURNING list if any.
     */
    if (plan->returningLists)
    {
        returningList = (List *) list_nth(plan->returningLists, subplan_index);
 
        /*
         * When performing an UPDATE/DELETE .. RETURNING on a join directly,
         * we fetch from the foreign server any Vars specified in RETURNING
         * that refer not only to the target relation but to non-target
         * relations.  So we'll deparse them into the RETURNING clause of the
         * remote query; use a targetlist consisting of them instead, which
         * will be adjusted to be new fdw_scan_tlist of the foreign-scan plan
         * node below.
         */
        if (fscan->scan.scanrelid == 0)
            returningList = build_remote_returning(resultRelation, rel,
                                                   returningList);
    }
 
    /*
     * Construct the SQL command string.
     */
    switch (operation)
    {
        case CMD_UPDATE:
            deparseDirectUpdateSql(&sql, root, resultRelation, rel,
                                   foreignrel,
                                   processed_tlist,
                                   targetAttrs,
                                   remote_exprs, &params_list,
                                   returningList, &retrieved_attrs);
            break;
        case CMD_DELETE:
            deparseDirectDeleteSql(&sql, root, resultRelation, rel,
                                   foreignrel,
                                   remote_exprs, &params_list,
                                   returningList, &retrieved_attrs);
            break;
        default:
            elog(ERROR, "unexpected operation: %d", (int) operation);
            break;
    }
 
    /*
     * Update the operation and target relation info.
     */
    fscan->operation = operation;
    fscan->resultRelation = resultRelation;
 
    /*
     * Update the fdw_exprs list that will be available to the executor.
     */
    fscan->fdw_exprs = params_list;
 
    /*
     * Update the fdw_private list that will be available to the executor.
     * Items in the list must match enum FdwDirectModifyPrivateIndex, above.
     */
    fscan->fdw_private = list_make4(makeString(sql.data),
                                    makeBoolean((retrieved_attrs != NIL)),
                                    retrieved_attrs,
                                    makeBoolean(plan->canSetTag));
 
    /*
     * Update the foreign-join-related fields.
     */
    if (fscan->scan.scanrelid == 0)
    {
        /* No need for the outer subplan. */
        fscan->scan.plan.lefttree = NULL;
 
        /* Build new fdw_scan_tlist if UPDATE/DELETE .. RETURNING. */
        if (returningList)
            rebuild_fdw_scan_tlist(fscan, returningList);
    }
 
    /*
     * Finally, unset the async-capable flag if it is set, as we currently
     * don't support asynchronous execution of direct modifications.
     */
    if (fscan->scan.plan.async_capable)
        fscan->scan.plan.async_capable = false;
 
    table_close(rel, NoLock);
    return true;
}
 
/*
 * postgresBeginDirectModify
 *      Prepare a direct foreign table modification
 */
static void
postgresBeginDirectModify(ForeignScanState *node, int eflags)
{
    ForeignScan *fsplan = (ForeignScan *) node->ss.ps.plan;
    EState     *estate = node->ss.ps.state;
    PgFdwDirectModifyState *dmstate;
    Index       rtindex;
    Oid         userid;
    ForeignTable *table;
    UserMapping *user;
    int         numParams;
 
    /*
     * Do nothing in EXPLAIN (no ANALYZE) case.  node->fdw_state stays NULL.
     */
    if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
        return;
 
    /*
     * We'll save private state in node->fdw_state.
     */
    dmstate = palloc0_object(PgFdwDirectModifyState);
    node->fdw_state = dmstate;
 
    /*
     * Identify which user to do the remote access as.  This should match what
     * ExecCheckPermissions() does.
     */
    userid = OidIsValid(fsplan->checkAsUser) ? fsplan->checkAsUser : GetUserId();
 
    /* Get info about foreign table. */
    rtindex = node->resultRelInfo->ri_RangeTableIndex;
    if (fsplan->scan.scanrelid == 0)
        dmstate->rel = ExecOpenScanRelation(estate, rtindex, eflags);
    else
        dmstate->rel = node->ss.ss_currentRelation;
    table = GetForeignTable(RelationGetRelid(dmstate->rel));
    user = GetUserMapping(userid, table->serverid);
 
    /*
     * Get connection to the foreign server.  Connection manager will
     * establish new connection if necessary.
     */
    dmstate->conn = GetConnection(user, false, &dmstate->conn_state);
 
    /* Update the foreign-join-related fields. */
    if (fsplan->scan.scanrelid == 0)
    {
        /* Save info about foreign table. */
        dmstate->resultRel = dmstate->rel;
 
        /*
         * Set dmstate->rel to NULL to teach get_returning_data() and
         * make_tuple_from_result_row() that columns fetched from the remote
         * server are described by fdw_scan_tlist of the foreign-scan plan
         * node, not the tuple descriptor for the target relation.
         */
        dmstate->rel = NULL;
    }
 
    /* Initialize state variable */
    dmstate->num_tuples = -1;   /* -1 means not set yet */
 
    /* Get private info created by planner functions. */
    dmstate->query = strVal(list_nth(fsplan->fdw_private,
                                     FdwDirectModifyPrivateUpdateSql));
    dmstate->has_returning = boolVal(list_nth(fsplan->fdw_private,
                                              FdwDirectModifyPrivateHasReturning));
    dmstate->retrieved_attrs = (List *) list_nth(fsplan->fdw_private,
                                                 FdwDirectModifyPrivateRetrievedAttrs);
    dmstate->set_processed = boolVal(list_nth(fsplan->fdw_private,
                                              FdwDirectModifyPrivateSetProcessed));
 
    /* Create context for per-tuple temp workspace. */
    dmstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
                                              "postgres_fdw temporary data",
                                              ALLOCSET_SMALL_SIZES);
 
    /* Prepare for input conversion of RETURNING results. */
    if (dmstate->has_returning)
    {
        TupleDesc   tupdesc;
 
        if (fsplan->scan.scanrelid == 0)
            tupdesc = get_tupdesc_for_join_scan_tuples(node);
        else
            tupdesc = RelationGetDescr(dmstate->rel);
 
        dmstate->attinmeta = TupleDescGetAttInMetadata(tupdesc);
 
        /*
         * When performing an UPDATE/DELETE .. RETURNING on a join directly,
         * initialize a filter to extract an updated/deleted tuple from a scan
         * tuple.
         */
        if (fsplan->scan.scanrelid == 0)
            init_returning_filter(dmstate, fsplan->fdw_scan_tlist, rtindex);
    }
 
    /*
     * Prepare for processing of parameters used in remote query, if any.
     */
    numParams = list_length(fsplan->fdw_exprs);
    dmstate->numParams = numParams;
    if (numParams > 0)
        prepare_query_params((PlanState *) node,
                             fsplan->fdw_exprs,
                             numParams,
                             &dmstate->param_flinfo,
                             &dmstate->param_exprs,
                             &dmstate->param_values);
}
 
/*
 * postgresIterateDirectModify
 *      Execute a direct foreign table modification
 */
static TupleTableSlot *
postgresIterateDirectModify(ForeignScanState *node)
{
    PgFdwDirectModifyState *dmstate = (PgFdwDirectModifyState *) node->fdw_state;
    EState     *estate = node->ss.ps.state;
    ResultRelInfo *resultRelInfo = node->resultRelInfo;
 
    /*
     * If this is the first call after Begin, execute the statement.
     */
    if (dmstate->num_tuples == -1)
        execute_dml_stmt(node);
 
    /*
     * If the local query doesn't specify RETURNING, just clear tuple slot.
     */
    if (!resultRelInfo->ri_projectReturning)
    {
        TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
        NodeInstrumentation *instr = node->ss.ps.instrument;
 
        Assert(!dmstate->has_returning);
 
        /* Increment the command es_processed count if necessary. */
        if (dmstate->set_processed)
            estate->es_processed += dmstate->num_tuples;
 
        /* Increment the tuple count for EXPLAIN ANALYZE if necessary. */
        if (instr)
            instr->tuplecount += dmstate->num_tuples;
 
        return ExecClearTuple(slot);
    }
 
    /*
     * Get the next RETURNING tuple.
     */
    return get_returning_data(node);
}
 
/*
 * postgresEndDirectModify
 *      Finish a direct foreign table modification
 */
static void
postgresEndDirectModify(ForeignScanState *node)
{
    PgFdwDirectModifyState *dmstate = (PgFdwDirectModifyState *) node->fdw_state;
 
    /* if dmstate is NULL, we are in EXPLAIN; nothing to do */
    if (dmstate == NULL)
        return;
 
    /* Release PGresult */
    PQclear(dmstate->result);
 
    /* Release remote connection */
    ReleaseConnection(dmstate->conn);
    dmstate->conn = NULL;
 
    /* MemoryContext will be deleted automatically. */
}
 
/*
 * postgresExplainForeignScan
 *      Produce extra output for EXPLAIN of a ForeignScan on a foreign table
 */
static void
postgresExplainForeignScan(ForeignScanState *node, ExplainState *es)
{
    ForeignScan *plan = castNode(ForeignScan, node->ss.ps.plan);
    List       *fdw_private = plan->fdw_private;
 
    /*
     * Identify foreign scans that are really joins or upper relations.  The
     * input looks something like "(1) LEFT JOIN (2)", and we must replace the
     * digit string(s), which are RT indexes, with the correct relation names.
     * We do that here, not when the plan is created, because we can't know
     * what aliases ruleutils.c will assign at plan creation time.
     */
    if (list_length(fdw_private) > FdwScanPrivateRelations)
    {
        StringInfoData relations;
        char       *rawrelations;
        char       *ptr;
        int         minrti,
                    rtoffset;
 
        rawrelations = strVal(list_nth(fdw_private, FdwScanPrivateRelations));
 
        /*
         * A difficulty with using a string representation of RT indexes is
         * that setrefs.c won't update the string when flattening the
         * rangetable.  To find out what rtoffset was applied, identify the
         * minimum RT index appearing in the string and compare it to the
         * minimum member of plan->fs_base_relids.  (We expect all the relids
         * in the join will have been offset by the same amount; the Asserts
         * below should catch it if that ever changes.)
         */
        minrti = INT_MAX;
        ptr = rawrelations;
        while (*ptr)
        {
            if (isdigit((unsigned char) *ptr))
            {
                int         rti = strtol(ptr, &ptr, 10);
 
                if (rti < minrti)
                    minrti = rti;
            }
            else
                ptr++;
        }
        rtoffset = bms_next_member(plan->fs_base_relids, -1) - minrti;
 
        /* Now we can translate the string */
        initStringInfo(&relations);
        ptr = rawrelations;
        while (*ptr)
        {
            if (isdigit((unsigned char) *ptr))
            {
                int         rti = strtol(ptr, &ptr, 10);
                RangeTblEntry *rte;
                char       *relname;
                char       *refname;
 
                rti += rtoffset;
                Assert(bms_is_member(rti, plan->fs_base_relids));
                rte = rt_fetch(rti, es->rtable);
                Assert(rte->rtekind == RTE_RELATION);
                /* This logic should agree with explain.c's ExplainTargetRel */
                relname = get_rel_name(rte->relid);
                if (es->verbose)
                {
                    char       *namespace;
 
                    namespace = get_namespace_name_or_temp(get_rel_namespace(rte->relid));
                    appendStringInfo(&relations, "%s.%s",
                                     quote_identifier(namespace),
                                     quote_identifier(relname));
                }
                else
                    appendStringInfoString(&relations,
                                           quote_identifier(relname));
                refname = (char *) list_nth(es->rtable_names, rti - 1);
                if (refname == NULL)
                    refname = rte->eref->aliasname;
                if (strcmp(refname, relname) != 0)
                    appendStringInfo(&relations, " %s",
                                     quote_identifier(refname));
            }
            else
                appendStringInfoChar(&relations, *ptr++);
        }
        ExplainPropertyText("Relations", relations.data, es);
    }
 
    /*
     * Add remote query, when VERBOSE option is specified.
     */
    if (es->verbose)
    {
        char       *sql;
 
        sql = strVal(list_nth(fdw_private, FdwScanPrivateSelectSql));
        ExplainPropertyText("Remote SQL", sql, es);
    }
}
 
/*
 * postgresExplainForeignModify
 *      Produce extra output for EXPLAIN of a ModifyTable on a foreign table
 */
static void
postgresExplainForeignModify(ModifyTableState *mtstate,
                             ResultRelInfo *rinfo,
                             List *fdw_private,
                             int subplan_index,
                             ExplainState *es)
{
    if (es->verbose)
    {
        char       *sql = strVal(list_nth(fdw_private,
                                          FdwModifyPrivateUpdateSql));
 
        ExplainPropertyText("Remote SQL", sql, es);
 
        /*
         * For INSERT we should always have batch size >= 1, but UPDATE and
         * DELETE don't support batching so don't show the property.
         */
        if (rinfo->ri_BatchSize > 0)
            ExplainPropertyInteger("Batch Size", NULL, rinfo->ri_BatchSize, es);
    }
}
 
/*
 * postgresExplainDirectModify
 *      Produce extra output for EXPLAIN of a ForeignScan that modifies a
 *      foreign table directly
 */
static void
postgresExplainDirectModify(ForeignScanState *node, ExplainState *es)
{
    List       *fdw_private;
    char       *sql;
 
    if (es->verbose)
    {
        fdw_private = ((ForeignScan *) node->ss.ps.plan)->fdw_private;
        sql = strVal(list_nth(fdw_private, FdwDirectModifyPrivateUpdateSql));
        ExplainPropertyText("Remote SQL", sql, es);
    }
}
 
/*
 * postgresExecForeignTruncate
 *      Truncate one or more foreign tables
 */
static void
postgresExecForeignTruncate(List *rels,
                            DropBehavior behavior,
                            bool restart_seqs)
{
    Oid         serverid = InvalidOid;
    UserMapping *user = NULL;
    PGconn     *conn = NULL;
    StringInfoData sql;
    ListCell   *lc;
    bool        server_truncatable = true;
 
    /*
     * By default, all postgres_fdw foreign tables are assumed truncatable.
     * This can be overridden by a per-server setting, which in turn can be
     * overridden by a per-table setting.
     */
    foreach(lc, rels)
    {
        ForeignServer *server = NULL;
        Relation    rel = lfirst(lc);
        ForeignTable *table = GetForeignTable(RelationGetRelid(rel));
        ListCell   *cell;
        bool        truncatable;
 
        /*
         * First time through, determine whether the foreign server allows
         * truncates. Since all specified foreign tables are assumed to belong
         * to the same foreign server, this result can be used for other
         * foreign tables.
         */
        if (!OidIsValid(serverid))
        {
            serverid = table->serverid;
            server = GetForeignServer(serverid);
 
            foreach(cell, server->options)
            {
                DefElem    *defel = (DefElem *) lfirst(cell);
 
                if (strcmp(defel->defname, "truncatable") == 0)
                {
                    server_truncatable = defGetBoolean(defel);
                    break;
                }
            }
        }
 
        /*
         * Confirm that all specified foreign tables belong to the same
         * foreign server.
         */
        Assert(table->serverid == serverid);
 
        /* Determine whether this foreign table allows truncations */
        truncatable = server_truncatable;
        foreach(cell, table->options)
        {
            DefElem    *defel = (DefElem *) lfirst(cell);
 
            if (strcmp(defel->defname, "truncatable") == 0)
            {
                truncatable = defGetBoolean(defel);
                break;
            }
        }
 
        if (!truncatable)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("foreign table \"%s\" does not allow truncates",
                            RelationGetRelationName(rel))));
    }
    Assert(OidIsValid(serverid));
 
    /*
     * Get connection to the foreign server.  Connection manager will
     * establish new connection if necessary.
     */
    user = GetUserMapping(GetUserId(), serverid);
    conn = GetConnection(user, false, NULL);
 
    /* Construct the TRUNCATE command string */
    initStringInfo(&sql);
    deparseTruncateSql(&sql, rels, behavior, restart_seqs);
 
    /* Issue the TRUNCATE command to remote server */
    do_sql_command(conn, sql.data);
 
    pfree(sql.data);
}
 
/*
 * estimate_path_cost_size
 *      Get cost and size estimates for a foreign scan on given foreign relation
 *      either a base relation or a join between foreign relations or an upper
 *      relation containing foreign relations.
 *
 * param_join_conds are the parameterization clauses with outer relations.
 * pathkeys specify the expected sort order if any for given path being costed.
 * fpextra specifies additional post-scan/join-processing steps such as the
 * final sort and the LIMIT restriction.
 *
 * The function returns the cost and size estimates in p_rows, p_width,
 * p_disabled_nodes, p_startup_cost and p_total_cost variables.
 */
static void
estimate_path_cost_size(PlannerInfo *root,
                        RelOptInfo *foreignrel,
                        List *param_join_conds,
                        List *pathkeys,
                        PgFdwPathExtraData *fpextra,
                        double *p_rows, int *p_width,
                        int *p_disabled_nodes,
                        Cost *p_startup_cost, Cost *p_total_cost)
{
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) foreignrel->fdw_private;
    double      rows;
    double      retrieved_rows;
    int         width;
    int         disabled_nodes = 0;
    Cost        startup_cost;
    Cost        total_cost;
 
    /* Make sure the core code has set up the relation's reltarget */
    Assert(foreignrel->reltarget);
 
    /*
     * If the table or the server is configured to use remote estimates,
     * connect to the foreign server and execute EXPLAIN to estimate the
     * number of rows selected by the restriction+join clauses.  Otherwise,
     * estimate rows using whatever statistics we have locally, in a way
     * similar to ordinary tables.
     */
    if (fpinfo->use_remote_estimate)
    {
        List       *remote_param_join_conds;
        List       *local_param_join_conds;
        StringInfoData sql;
        PGconn     *conn;
        Selectivity local_sel;
        QualCost    local_cost;
        List       *fdw_scan_tlist = NIL;
        List       *remote_conds;
 
        /* Required only to be passed to deparseSelectStmtForRel */
        List       *retrieved_attrs;
 
        /*
         * param_join_conds might contain both clauses that are safe to send
         * across, and clauses that aren't.
         */
        classifyConditions(root, foreignrel, param_join_conds,
                           &remote_param_join_conds, &local_param_join_conds);
 
        /* Build the list of columns to be fetched from the foreign server. */
        if (IS_JOIN_REL(foreignrel) || IS_UPPER_REL(foreignrel))
            fdw_scan_tlist = build_tlist_to_deparse(foreignrel);
        else
            fdw_scan_tlist = NIL;
 
        /*
         * The complete list of remote conditions includes everything from
         * baserestrictinfo plus any extra join_conds relevant to this
         * particular path.
         */
        remote_conds = list_concat(remote_param_join_conds,
                                   fpinfo->remote_conds);
 
        /*
         * Construct EXPLAIN query including the desired SELECT, FROM, and
         * WHERE clauses. Params and other-relation Vars are replaced by dummy
         * values, so don't request params_list.
         */
        initStringInfo(&sql);
        appendStringInfoString(&sql, "EXPLAIN ");
        deparseSelectStmtForRel(&sql, root, foreignrel, fdw_scan_tlist,
                                remote_conds, pathkeys,
                                fpextra ? fpextra->has_final_sort : false,
                                fpextra ? fpextra->has_limit : false,
                                false, &retrieved_attrs, NULL);
 
        /* Get the remote estimate */
        conn = GetConnection(fpinfo->user, false, NULL);
        get_remote_estimate(sql.data, conn, &rows, &width,
                            &startup_cost, &total_cost);
        ReleaseConnection(conn);
 
        retrieved_rows = rows;
 
        /* Factor in the selectivity of the locally-checked quals */
        local_sel = clauselist_selectivity(root,
                                           local_param_join_conds,
                                           foreignrel->relid,
                                           JOIN_INNER,
                                           NULL);
        local_sel *= fpinfo->local_conds_sel;
 
        rows = clamp_row_est(rows * local_sel);
 
        /* Add in the eval cost of the locally-checked quals */
        startup_cost += fpinfo->local_conds_cost.startup;
        total_cost += fpinfo->local_conds_cost.per_tuple * retrieved_rows;
        cost_qual_eval(&local_cost, local_param_join_conds, root);
        startup_cost += local_cost.startup;
        total_cost += local_cost.per_tuple * retrieved_rows;
 
        /*
         * Add in tlist eval cost for each output row.  In case of an
         * aggregate, some of the tlist expressions such as grouping
         * expressions will be evaluated remotely, so adjust the costs.
         */
        startup_cost += foreignrel->reltarget->cost.startup;
        total_cost += foreignrel->reltarget->cost.startup;
        total_cost += foreignrel->reltarget->cost.per_tuple * rows;
        if (IS_UPPER_REL(foreignrel))
        {
            QualCost    tlist_cost;
 
            cost_qual_eval(&tlist_cost, fdw_scan_tlist, root);
            startup_cost -= tlist_cost.startup;
            total_cost -= tlist_cost.startup;
            total_cost -= tlist_cost.per_tuple * rows;
        }
    }
    else
    {
        Cost        run_cost = 0;
 
        /*
         * We don't support join conditions in this mode (hence, no
         * parameterized paths can be made).
         */
        Assert(param_join_conds == NIL);
 
        /*
         * We will come here again and again with different set of pathkeys or
         * additional post-scan/join-processing steps that caller wants to
         * cost.  We don't need to calculate the cost/size estimates for the
         * underlying scan, join, or grouping each time.  Instead, use those
         * estimates if we have cached them already.
         */
        if (fpinfo->rel_startup_cost >= 0 && fpinfo->rel_total_cost >= 0)
        {
            Assert(fpinfo->retrieved_rows >= 0);
 
            rows = fpinfo->rows;
            retrieved_rows = fpinfo->retrieved_rows;
            width = fpinfo->width;
            startup_cost = fpinfo->rel_startup_cost;
            run_cost = fpinfo->rel_total_cost - fpinfo->rel_startup_cost;
 
            /*
             * If we estimate the costs of a foreign scan or a foreign join
             * with additional post-scan/join-processing steps, the scan or
             * join costs obtained from the cache wouldn't yet contain the
             * eval costs for the final scan/join target, which would've been
             * updated by apply_scanjoin_target_to_paths(); add the eval costs
             * now.
             */
            if (fpextra && !IS_UPPER_REL(foreignrel))
            {
                /* Shouldn't get here unless we have LIMIT */
                Assert(fpextra->has_limit);
                Assert(foreignrel->reloptkind == RELOPT_BASEREL ||
                       foreignrel->reloptkind == RELOPT_JOINREL);
                startup_cost += foreignrel->reltarget->cost.startup;
                run_cost += foreignrel->reltarget->cost.per_tuple * rows;
            }
        }
        else if (IS_JOIN_REL(foreignrel))
        {
            PgFdwRelationInfo *fpinfo_i;
            PgFdwRelationInfo *fpinfo_o;
            QualCost    join_cost;
            QualCost    remote_conds_cost;
            double      nrows;
 
            /* Use rows/width estimates made by the core code. */
            rows = foreignrel->rows;
            width = foreignrel->reltarget->width;
 
            /* For join we expect inner and outer relations set */
            Assert(fpinfo->innerrel && fpinfo->outerrel);
 
            fpinfo_i = (PgFdwRelationInfo *) fpinfo->innerrel->fdw_private;
            fpinfo_o = (PgFdwRelationInfo *) fpinfo->outerrel->fdw_private;
 
            /* Estimate of number of rows in cross product */
            nrows = fpinfo_i->rows * fpinfo_o->rows;
 
            /*
             * Back into an estimate of the number of retrieved rows.  Just in
             * case this is nuts, clamp to at most nrows.
             */
            retrieved_rows = clamp_row_est(rows / fpinfo->local_conds_sel);
            retrieved_rows = Min(retrieved_rows, nrows);
 
            /*
             * The cost of foreign join is estimated as cost of generating
             * rows for the joining relations + cost for applying quals on the
             * rows.
             */
 
            /*
             * Calculate the cost of clauses pushed down to the foreign server
             */
            cost_qual_eval(&remote_conds_cost, fpinfo->remote_conds, root);
            /* Calculate the cost of applying join clauses */
            cost_qual_eval(&join_cost, fpinfo->joinclauses, root);
 
            /*
             * Startup cost includes startup cost of joining relations and the
             * startup cost for join and other clauses. We do not include the
             * startup cost specific to join strategy (e.g. setting up hash
             * tables) since we do not know what strategy the foreign server
             * is going to use.
             */
            startup_cost = fpinfo_i->rel_startup_cost + fpinfo_o->rel_startup_cost;
            startup_cost += join_cost.startup;
            startup_cost += remote_conds_cost.startup;
            startup_cost += fpinfo->local_conds_cost.startup;
 
            /*
             * Run time cost includes:
             *
             * 1. Run time cost (total_cost - startup_cost) of relations being
             * joined
             *
             * 2. Run time cost of applying join clauses on the cross product
             * of the joining relations.
             *
             * 3. Run time cost of applying pushed down other clauses on the
             * result of join
             *
             * 4. Run time cost of applying nonpushable other clauses locally
             * on the result fetched from the foreign server.
             */
            run_cost = fpinfo_i->rel_total_cost - fpinfo_i->rel_startup_cost;
            run_cost += fpinfo_o->rel_total_cost - fpinfo_o->rel_startup_cost;
            run_cost += nrows * join_cost.per_tuple;
            nrows = clamp_row_est(nrows * fpinfo->joinclause_sel);
            run_cost += nrows * remote_conds_cost.per_tuple;
            run_cost += fpinfo->local_conds_cost.per_tuple * retrieved_rows;
 
            /* Add in tlist eval cost for each output row */
            startup_cost += foreignrel->reltarget->cost.startup;
            run_cost += foreignrel->reltarget->cost.per_tuple * rows;
        }
        else if (IS_UPPER_REL(foreignrel))
        {
            RelOptInfo *outerrel = fpinfo->outerrel;
            PgFdwRelationInfo *ofpinfo;
            AggClauseCosts aggcosts = {0};
            double      input_rows;
            int         numGroupCols;
            double      numGroups = 1;
 
            /* The upper relation should have its outer relation set */
            Assert(outerrel);
            /* and that outer relation should have its reltarget set */
            Assert(outerrel->reltarget);
 
            /*
             * This cost model is mixture of costing done for sorted and
             * hashed aggregates in cost_agg().  We are not sure which
             * strategy will be considered at remote side, thus for
             * simplicity, we put all startup related costs in startup_cost
             * and all finalization and run cost are added in total_cost.
             */
 
            ofpinfo = (PgFdwRelationInfo *) outerrel->fdw_private;
 
            /* Get rows from input rel */
            input_rows = ofpinfo->rows;
 
            /* Collect statistics about aggregates for estimating costs. */
            if (root->parse->hasAggs)
            {
                get_agg_clause_costs(root, AGGSPLIT_SIMPLE, &aggcosts);
            }
 
            /* Get number of grouping columns and possible number of groups */
            numGroupCols = list_length(root->processed_groupClause);
            numGroups = estimate_num_groups(root,
                                            get_sortgrouplist_exprs(root->processed_groupClause,
                                                                    fpinfo->grouped_tlist),
                                            input_rows, NULL, NULL);
 
            /*
             * Get the retrieved_rows and rows estimates.  If there are HAVING
             * quals, account for their selectivity.
             */
            if (root->hasHavingQual)
            {
                /* Factor in the selectivity of the remotely-checked quals */
                retrieved_rows =
                    clamp_row_est(numGroups *
                                  clauselist_selectivity(root,
                                                         fpinfo->remote_conds,
                                                         0,
                                                         JOIN_INNER,
                                                         NULL));
                /* Factor in the selectivity of the locally-checked quals */
                rows = clamp_row_est(retrieved_rows * fpinfo->local_conds_sel);
            }
            else
            {
                rows = retrieved_rows = numGroups;
            }
 
            /* Use width estimate made by the core code. */
            width = foreignrel->reltarget->width;
 
            /*-----
             * Startup cost includes:
             *    1. Startup cost for underneath input relation, adjusted for
             *       tlist replacement by apply_scanjoin_target_to_paths()
             *    2. Cost of performing aggregation, per cost_agg()
             *-----
             */
            startup_cost = ofpinfo->rel_startup_cost;
            startup_cost += outerrel->reltarget->cost.startup;
            startup_cost += aggcosts.transCost.startup;
            startup_cost += aggcosts.transCost.per_tuple * input_rows;
            startup_cost += aggcosts.finalCost.startup;
            startup_cost += (cpu_operator_cost * numGroupCols) * input_rows;
 
            /*-----
             * Run time cost includes:
             *    1. Run time cost of underneath input relation, adjusted for
             *       tlist replacement by apply_scanjoin_target_to_paths()
             *    2. Run time cost of performing aggregation, per cost_agg()
             *-----
             */
            run_cost = ofpinfo->rel_total_cost - ofpinfo->rel_startup_cost;
            run_cost += outerrel->reltarget->cost.per_tuple * input_rows;
            run_cost += aggcosts.finalCost.per_tuple * numGroups;
            run_cost += cpu_tuple_cost * numGroups;
 
            /* Account for the eval cost of HAVING quals, if any */
            if (root->hasHavingQual)
            {
                QualCost    remote_cost;
 
                /* Add in the eval cost of the remotely-checked quals */
                cost_qual_eval(&remote_cost, fpinfo->remote_conds, root);
                startup_cost += remote_cost.startup;
                run_cost += remote_cost.per_tuple * numGroups;
                /* Add in the eval cost of the locally-checked quals */
                startup_cost += fpinfo->local_conds_cost.startup;
                run_cost += fpinfo->local_conds_cost.per_tuple * retrieved_rows;
            }
 
            /* Add in tlist eval cost for each output row */
            startup_cost += foreignrel->reltarget->cost.startup;
            run_cost += foreignrel->reltarget->cost.per_tuple * rows;
        }
        else
        {
            Cost        cpu_per_tuple;
 
            /* Use rows/width estimates made by set_baserel_size_estimates. */
            rows = foreignrel->rows;
            width = foreignrel->reltarget->width;
 
            /*
             * Back into an estimate of the number of retrieved rows.  Just in
             * case this is nuts, clamp to at most foreignrel->tuples.
             */
            retrieved_rows = clamp_row_est(rows / fpinfo->local_conds_sel);
            retrieved_rows = Min(retrieved_rows, foreignrel->tuples);
 
            /*
             * Cost as though this were a seqscan, which is pessimistic.  We
             * effectively imagine the local_conds are being evaluated
             * remotely, too.
             */
            startup_cost = 0;
            run_cost = 0;
            run_cost += seq_page_cost * foreignrel->pages;
 
            startup_cost += foreignrel->baserestrictcost.startup;
            cpu_per_tuple = cpu_tuple_cost + foreignrel->baserestrictcost.per_tuple;
            run_cost += cpu_per_tuple * foreignrel->tuples;
 
            /* Add in tlist eval cost for each output row */
            startup_cost += foreignrel->reltarget->cost.startup;
            run_cost += foreignrel->reltarget->cost.per_tuple * rows;
        }
 
        /*
         * Without remote estimates, we have no real way to estimate the cost
         * of generating sorted output.  It could be free if the query plan
         * the remote side would have chosen generates properly-sorted output
         * anyway, but in most cases it will cost something.  Estimate a value
         * high enough that we won't pick the sorted path when the ordering
         * isn't locally useful, but low enough that we'll err on the side of
         * pushing down the ORDER BY clause when it's useful to do so.
         */
        if (pathkeys != NIL)
        {
            if (IS_UPPER_REL(foreignrel))
            {
                Assert(foreignrel->reloptkind == RELOPT_UPPER_REL &&
                       fpinfo->stage == UPPERREL_GROUP_AGG);
 
                /*
                 * We can only get here when this function is called from
                 * add_foreign_ordered_paths() or add_foreign_final_paths();
                 * in which cases, the passed-in fpextra should not be NULL.
                 */
                Assert(fpextra);
                adjust_foreign_grouping_path_cost(root, pathkeys,
                                                  retrieved_rows, width,
                                                  fpextra->limit_tuples,
                                                  &disabled_nodes,
                                                  &startup_cost, &run_cost);
            }
            else
            {
                startup_cost *= DEFAULT_FDW_SORT_MULTIPLIER;
                run_cost *= DEFAULT_FDW_SORT_MULTIPLIER;
            }
        }
 
        total_cost = startup_cost + run_cost;
 
        /* Adjust the cost estimates if we have LIMIT */
        if (fpextra && fpextra->has_limit)
        {
            adjust_limit_rows_costs(&rows, &startup_cost, &total_cost,
                                    fpextra->offset_est, fpextra->count_est);
            retrieved_rows = rows;
        }
    }
 
    /*
     * If this includes the final sort step, the given target, which will be
     * applied to the resulting path, might have different expressions from
     * the foreignrel's reltarget (see make_sort_input_target()); adjust tlist
     * eval costs.
     */
    if (fpextra && fpextra->has_final_sort &&
        fpextra->target != foreignrel->reltarget)
    {
        QualCost    oldcost = foreignrel->reltarget->cost;
        QualCost    newcost = fpextra->target->cost;
 
        startup_cost += newcost.startup - oldcost.startup;
        total_cost += newcost.startup - oldcost.startup;
        total_cost += (newcost.per_tuple - oldcost.per_tuple) * rows;
    }
 
    /*
     * Cache the retrieved rows and cost estimates for scans, joins, or
     * groupings without any parameterization, pathkeys, or additional
     * post-scan/join-processing steps, before adding the costs for
     * transferring data from the foreign server.  These estimates are useful
     * for costing remote joins involving this relation or costing other
     * remote operations on this relation such as remote sorts and remote
     * LIMIT restrictions, when the costs can not be obtained from the foreign
     * server.  This function will be called at least once for every foreign
     * relation without any parameterization, pathkeys, or additional
     * post-scan/join-processing steps.
     */
    if (pathkeys == NIL && param_join_conds == NIL && fpextra == NULL)
    {
        fpinfo->retrieved_rows = retrieved_rows;
        fpinfo->rel_startup_cost = startup_cost;
        fpinfo->rel_total_cost = total_cost;
    }
 
    /*
     * Add some additional cost factors to account for connection overhead
     * (fdw_startup_cost), transferring data across the network
     * (fdw_tuple_cost per retrieved row), and local manipulation of the data
     * (cpu_tuple_cost per retrieved row).
     */
    startup_cost += fpinfo->fdw_startup_cost;
    total_cost += fpinfo->fdw_startup_cost;
    total_cost += fpinfo->fdw_tuple_cost * retrieved_rows;
    total_cost += cpu_tuple_cost * retrieved_rows;
 
    /*
     * If we have LIMIT, we should prefer performing the restriction remotely
     * rather than locally, as the former avoids extra row fetches from the
     * remote that the latter might cause.  But since the core code doesn't
     * account for such fetches when estimating the costs of the local
     * restriction (see create_limit_path()), there would be no difference
     * between the costs of the local restriction and the costs of the remote
     * restriction estimated above if we don't use remote estimates (except
     * for the case where the foreignrel is a grouping relation, the given
     * pathkeys is not NIL, and the effects of a bounded sort for that rel is
     * accounted for in costing the remote restriction).  Tweak the costs of
     * the remote restriction to ensure we'll prefer it if LIMIT is a useful
     * one.
     */
    if (!fpinfo->use_remote_estimate &&
        fpextra && fpextra->has_limit &&
        fpextra->limit_tuples > 0 &&
        fpextra->limit_tuples < fpinfo->rows)
    {
        Assert(fpinfo->rows > 0);
        total_cost -= (total_cost - startup_cost) * 0.05 *
            (fpinfo->rows - fpextra->limit_tuples) / fpinfo->rows;
    }
 
    /* Return results. */
    *p_rows = rows;
    *p_width = width;
    *p_disabled_nodes = disabled_nodes;
    *p_startup_cost = startup_cost;
    *p_total_cost = total_cost;
}
 
/*
 * Estimate costs of executing a SQL statement remotely.
 * The given "sql" must be an EXPLAIN command.
 */
static void
get_remote_estimate(const char *sql, PGconn *conn,
                    double *rows, int *width,
                    Cost *startup_cost, Cost *total_cost)
{
    PGresult   *res;
    char       *line;
    char       *p;
    int         n;
 
    /*
     * Execute EXPLAIN remotely.
     */
    res = pgfdw_exec_query(conn, sql, NULL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pgfdw_report_error(res, conn, sql);
 
    /*
     * Extract cost numbers for topmost plan node.  Note we search for a left
     * paren from the end of the line to avoid being confused by other uses of
     * parentheses.
     */
    line = PQgetvalue(res, 0, 0);
    p = strrchr(line, '(');
    if (p == NULL)
        elog(ERROR, "could not interpret EXPLAIN output: \"%s\"", line);
    n = sscanf(p, "(cost=%lf..%lf rows=%lf width=%d)",
               startup_cost, total_cost, rows, width);
    if (n != 4)
        elog(ERROR, "could not interpret EXPLAIN output: \"%s\"", line);
    PQclear(res);
}
 
/*
 * Adjust the cost estimates of a foreign grouping path to include the cost of
 * generating properly-sorted output.
 */
static void
adjust_foreign_grouping_path_cost(PlannerInfo *root,
                                  List *pathkeys,
                                  double retrieved_rows,
                                  double width,
                                  double limit_tuples,
                                  int *p_disabled_nodes,
                                  Cost *p_startup_cost,
                                  Cost *p_run_cost)
{
    /*
     * If the GROUP BY clause isn't sort-able, the plan chosen by the remote
     * side is unlikely to generate properly-sorted output, so it would need
     * an explicit sort; adjust the given costs with cost_sort().  Likewise,
     * if the GROUP BY clause is sort-able but isn't a superset of the given
     * pathkeys, adjust the costs with that function.  Otherwise, adjust the
     * costs by applying the same heuristic as for the scan or join case.
     */
    if (!grouping_is_sortable(root->processed_groupClause) ||
        !pathkeys_contained_in(pathkeys, root->group_pathkeys))
    {
        Path        sort_path;  /* dummy for result of cost_sort */
 
        cost_sort(&sort_path,
                  root,
                  pathkeys,
                  0,
                  *p_startup_cost + *p_run_cost,
                  retrieved_rows,
                  width,
                  0.0,
                  work_mem,
                  limit_tuples);
 
        *p_startup_cost = sort_path.startup_cost;
        *p_run_cost = sort_path.total_cost - sort_path.startup_cost;
    }
    else
    {
        /*
         * The default extra cost seems too large for foreign-grouping cases;
         * add 1/4th of that default.
         */
        double      sort_multiplier = 1.0 + (DEFAULT_FDW_SORT_MULTIPLIER
                                             - 1.0) * 0.25;
 
        *p_startup_cost *= sort_multiplier;
        *p_run_cost *= sort_multiplier;
    }
}
 
/*
 * Detect whether we want to process an EquivalenceClass member.
 *
 * This is a callback for use by generate_implied_equalities_for_column.
 */
static bool
ec_member_matches_foreign(PlannerInfo *root, RelOptInfo *rel,
                          EquivalenceClass *ec, EquivalenceMember *em,
                          void *arg)
{
    ec_member_foreign_arg *state = (ec_member_foreign_arg *) arg;
    Expr       *expr = em->em_expr;
 
    /*
     * If we've identified what we're processing in the current scan, we only
     * want to match that expression.
     */
    if (state->current != NULL)
        return equal(expr, state->current);
 
    /*
     * Otherwise, ignore anything we've already processed.
     */
    if (list_member(state->already_used, expr))
        return false;
 
    /* This is the new target to process. */
    state->current = expr;
    return true;
}
 
/*
 * Create cursor for node's query with current parameter values.
 */
static void
create_cursor(ForeignScanState *node)
{
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
    ExprContext *econtext = node->ss.ps.ps_ExprContext;
    int         numParams = fsstate->numParams;
    const char **values = fsstate->param_values;
    PGconn     *conn = fsstate->conn;
    StringInfoData buf;
    PGresult   *res;
 
    /* First, process a pending asynchronous request, if any. */
    if (fsstate->conn_state->pendingAreq)
        process_pending_request(fsstate->conn_state->pendingAreq);
 
    /*
     * Construct array of query parameter values in text format.  We do the
     * conversions in the short-lived per-tuple context, so as not to cause a
     * memory leak over repeated scans.
     */
    if (numParams > 0)
    {
        MemoryContext oldcontext;
 
        oldcontext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
 
        process_query_params(econtext,
                             fsstate->param_flinfo,
                             fsstate->param_exprs,
                             values);
 
        MemoryContextSwitchTo(oldcontext);
    }
 
    /* Construct the DECLARE CURSOR command */
    initStringInfo(&buf);
    appendStringInfo(&buf, "DECLARE c%u CURSOR FOR\n%s",
                     fsstate->cursor_number, fsstate->query);
 
    /*
     * Notice that we pass NULL for paramTypes, thus forcing the remote server
     * to infer types for all parameters.  Since we explicitly cast every
     * parameter (see deparse.c), the "inference" is trivial and will produce
     * the desired result.  This allows us to avoid assuming that the remote
     * server has the same OIDs we do for the parameters' types.
     */
    if (!PQsendQueryParams(conn, buf.data, numParams,
                           NULL, values, NULL, NULL, 0))
        pgfdw_report_error(NULL, conn, buf.data);
 
    /*
     * Get the result, and check for success.
     */
    res = pgfdw_get_result(conn);
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pgfdw_report_error(res, conn, fsstate->query);
    PQclear(res);
 
    /* Mark the cursor as created, and show no tuples have been retrieved */
    fsstate->cursor_exists = true;
    fsstate->tuples = NULL;
    fsstate->num_tuples = 0;
    fsstate->next_tuple = 0;
    fsstate->fetch_ct_2 = 0;
    fsstate->eof_reached = false;
 
    /* Clean up */
    pfree(buf.data);
}
 
/*
 * Fetch some more rows from the node's cursor.
 */
static void
fetch_more_data(ForeignScanState *node)
{
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
    PGconn     *conn = fsstate->conn;
    PGresult   *res;
    int         numrows;
    int         i;
    MemoryContext oldcontext;
 
    /*
     * We'll store the tuples in the batch_cxt.  First, flush the previous
     * batch.
     */
    fsstate->tuples = NULL;
    MemoryContextReset(fsstate->batch_cxt);
    oldcontext = MemoryContextSwitchTo(fsstate->batch_cxt);
 
    if (fsstate->async_capable)
    {
        Assert(fsstate->conn_state->pendingAreq);
 
        /*
         * The query was already sent by an earlier call to
         * fetch_more_data_begin.  So now we just fetch the result.
         */
        res = pgfdw_get_result(conn);
        /* On error, report the original query, not the FETCH. */
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
            pgfdw_report_error(res, conn, fsstate->query);
 
        /* Reset per-connection state */
        fsstate->conn_state->pendingAreq = NULL;
    }
    else
    {
        char        sql[64];
 
        /* This is a regular synchronous fetch. */
        snprintf(sql, sizeof(sql), "FETCH %d FROM c%u",
                 fsstate->fetch_size, fsstate->cursor_number);
 
        res = pgfdw_exec_query(conn, sql, fsstate->conn_state);
        /* On error, report the original query, not the FETCH. */
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
            pgfdw_report_error(res, conn, fsstate->query);
    }
 
    /* Convert the data into HeapTuples */
    numrows = PQntuples(res);
    fsstate->tuples = (HeapTuple *) palloc0(numrows * sizeof(HeapTuple));
    fsstate->num_tuples = numrows;
    fsstate->next_tuple = 0;
 
    for (i = 0; i < numrows; i++)
    {
        Assert(IsA(node->ss.ps.plan, ForeignScan));
 
        fsstate->tuples[i] =
            make_tuple_from_result_row(res, i,
                                       fsstate->rel,
                                       fsstate->attinmeta,
                                       fsstate->retrieved_attrs,
                                       node,
                                       fsstate->temp_cxt);
    }
 
    /* Update fetch_ct_2 */
    if (fsstate->fetch_ct_2 < 2)
        fsstate->fetch_ct_2++;
 
    /* Must be EOF if we didn't get as many tuples as we asked for. */
    fsstate->eof_reached = (numrows < fsstate->fetch_size);
 
    PQclear(res);
 
    MemoryContextSwitchTo(oldcontext);
}
 
/*
 * Force assorted GUC parameters to settings that ensure that we'll output
 * data values in a form that is unambiguous to the remote server.
 *
 * This is rather expensive and annoying to do once per row, but there's
 * little choice if we want to be sure values are transmitted accurately;
 * we can't leave the settings in place between rows for fear of affecting
 * user-visible computations.
 *
 * We use the equivalent of a function SET option to allow the settings to
 * persist only until the caller calls reset_transmission_modes().  If an
 * error is thrown in between, guc.c will take care of undoing the settings.
 *
 * The return value is the nestlevel that must be passed to
 * reset_transmission_modes() to undo things.
 */
int
set_transmission_modes(void)
{
    int         nestlevel = NewGUCNestLevel();
 
    /*
     * The values set here should match what pg_dump does.  See also
     * configure_remote_session in connection.c.
     */
    if (DateStyle != USE_ISO_DATES)
        (void) set_config_option("datestyle", "ISO",
                                 PGC_USERSET, PGC_S_SESSION,
                                 GUC_ACTION_SAVE, true, 0, false);
    if (IntervalStyle != INTSTYLE_POSTGRES)
        (void) set_config_option("intervalstyle", "postgres",
                                 PGC_USERSET, PGC_S_SESSION,
                                 GUC_ACTION_SAVE, true, 0, false);
    if (extra_float_digits < 3)
        (void) set_config_option("extra_float_digits", "3",
                                 PGC_USERSET, PGC_S_SESSION,
                                 GUC_ACTION_SAVE, true, 0, false);
 
    /*
     * In addition force restrictive search_path, in case there are any
     * regproc or similar constants to be printed.
     */
    (void) set_config_option("search_path", "pg_catalog",
                             PGC_USERSET, PGC_S_SESSION,
                             GUC_ACTION_SAVE, true, 0, false);
 
    return nestlevel;
}
 
/*
 * Undo the effects of set_transmission_modes().
 */
void
reset_transmission_modes(int nestlevel)
{
    AtEOXact_GUC(true, nestlevel);
}
 
/*
 * Utility routine to close a cursor.
 */
static void
close_cursor(PGconn *conn, unsigned int cursor_number,
             PgFdwConnState *conn_state)
{
    char        sql[64];
    PGresult   *res;
 
    snprintf(sql, sizeof(sql), "CLOSE c%u", cursor_number);
    res = pgfdw_exec_query(conn, sql, conn_state);
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pgfdw_report_error(res, conn, sql);
    PQclear(res);
}
 
/*
 * create_foreign_modify
 *      Construct an execution state of a foreign insert/update/delete
 *      operation
 */
static PgFdwModifyState *
create_foreign_modify(EState *estate,
                      RangeTblEntry *rte,
                      ResultRelInfo *resultRelInfo,
                      CmdType operation,
                      Plan *subplan,
                      char *query,
                      List *target_attrs,
                      int values_end,
                      bool has_returning,
                      List *retrieved_attrs)
{
    PgFdwModifyState *fmstate;
    Relation    rel = resultRelInfo->ri_RelationDesc;
    TupleDesc   tupdesc = RelationGetDescr(rel);
    Oid         userid;
    ForeignTable *table;
    UserMapping *user;
    AttrNumber  n_params;
    Oid         typefnoid;
    bool        isvarlena;
    ListCell   *lc;
 
    /* Begin constructing PgFdwModifyState. */
    fmstate = palloc0_object(PgFdwModifyState);
    fmstate->rel = rel;
 
    /* Identify which user to do the remote access as. */
    userid = ExecGetResultRelCheckAsUser(resultRelInfo, estate);
 
    /* Get info about foreign table. */
    table = GetForeignTable(RelationGetRelid(rel));
    user = GetUserMapping(userid, table->serverid);
 
    /* Open connection; report that we'll create a prepared statement. */
    fmstate->conn = GetConnection(user, true, &fmstate->conn_state);
    fmstate->p_name = NULL;     /* prepared statement not made yet */
 
    /* Set up remote query information. */
    fmstate->query = query;
    if (operation == CMD_INSERT)
    {
        fmstate->query = pstrdup(fmstate->query);
        fmstate->orig_query = pstrdup(fmstate->query);
    }
    fmstate->target_attrs = target_attrs;
    fmstate->values_end = values_end;
    fmstate->has_returning = has_returning;
    fmstate->retrieved_attrs = retrieved_attrs;
 
    /* Create context for per-tuple temp workspace. */
    fmstate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
                                              "postgres_fdw temporary data",
                                              ALLOCSET_SMALL_SIZES);
 
    /* Prepare for input conversion of RETURNING results. */
    if (fmstate->has_returning)
        fmstate->attinmeta = TupleDescGetAttInMetadata(tupdesc);
 
    /* Prepare for output conversion of parameters used in prepared stmt. */
    n_params = list_length(fmstate->target_attrs) + 1;
    fmstate->p_flinfo = palloc0_array(FmgrInfo, n_params);
    fmstate->p_nums = 0;
 
    if (operation == CMD_UPDATE || operation == CMD_DELETE)
    {
        Assert(subplan != NULL);
 
        /* Find the ctid resjunk column in the subplan's result */
        fmstate->ctidAttno = ExecFindJunkAttributeInTlist(subplan->targetlist,
                                                          "ctid");
        if (!AttributeNumberIsValid(fmstate->ctidAttno))
            elog(ERROR, "could not find junk ctid column");
 
        /* First transmittable parameter will be ctid */
        getTypeOutputInfo(TIDOID, &typefnoid, &isvarlena);
        fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
        fmstate->p_nums++;
    }
 
    if (operation == CMD_INSERT || operation == CMD_UPDATE)
    {
        /* Set up for remaining transmittable parameters */
        foreach(lc, fmstate->target_attrs)
        {
            int         attnum = lfirst_int(lc);
            Form_pg_attribute attr = TupleDescAttr(tupdesc, attnum - 1);
 
            Assert(!attr->attisdropped);
 
            /* Ignore generated columns; they are set to DEFAULT */
            if (attr->attgenerated)
                continue;
            getTypeOutputInfo(attr->atttypid, &typefnoid, &isvarlena);
            fmgr_info(typefnoid, &fmstate->p_flinfo[fmstate->p_nums]);
            fmstate->p_nums++;
        }
    }
 
    Assert(fmstate->p_nums <= n_params);
 
    /* Set batch_size from foreign server/table options. */
    if (operation == CMD_INSERT)
        fmstate->batch_size = get_batch_size_option(rel);
 
    fmstate->num_slots = 1;
 
    /* Initialize auxiliary state */
    fmstate->aux_fmstate = NULL;
 
    return fmstate;
}
 
/*
 * execute_foreign_modify
 *      Perform foreign-table modification as required, and fetch RETURNING
 *      result if any.  (This is the shared guts of postgresExecForeignInsert,
 *      postgresExecForeignBatchInsert, postgresExecForeignUpdate, and
 *      postgresExecForeignDelete.)
 */
static TupleTableSlot **
execute_foreign_modify(EState *estate,
                       ResultRelInfo *resultRelInfo,
                       CmdType operation,
                       TupleTableSlot **slots,
                       TupleTableSlot **planSlots,
                       int *numSlots)
{
    PgFdwModifyState *fmstate = (PgFdwModifyState *) resultRelInfo->ri_FdwState;
    ItemPointer ctid = NULL;
    const char **p_values;
    PGresult   *res;
    int         n_rows;
    StringInfoData sql;
 
    /* The operation should be INSERT, UPDATE, or DELETE */
    Assert(operation == CMD_INSERT ||
           operation == CMD_UPDATE ||
           operation == CMD_DELETE);
 
    /* First, process a pending asynchronous request, if any. */
    if (fmstate->conn_state->pendingAreq)
        process_pending_request(fmstate->conn_state->pendingAreq);
 
    /*
     * If the existing query was deparsed and prepared for a different number
     * of rows, rebuild it for the proper number.
     */
    if (operation == CMD_INSERT && fmstate->num_slots != *numSlots)
    {
        /* Destroy the prepared statement created previously */
        if (fmstate->p_name)
            deallocate_query(fmstate);
 
        /* Build INSERT string with numSlots records in its VALUES clause. */
        initStringInfo(&sql);
        rebuildInsertSql(&sql, fmstate->rel,
                         fmstate->orig_query, fmstate->target_attrs,
                         fmstate->values_end, fmstate->p_nums,
                         *numSlots - 1);
        pfree(fmstate->query);
        fmstate->query = sql.data;
        fmstate->num_slots = *numSlots;
    }
 
    /* Set up the prepared statement on the remote server, if we didn't yet */
    if (!fmstate->p_name)
        prepare_foreign_modify(fmstate);
 
    /*
     * For UPDATE/DELETE, get the ctid that was passed up as a resjunk column
     */
    if (operation == CMD_UPDATE || operation == CMD_DELETE)
    {
        Datum       datum;
        bool        isNull;
 
        datum = ExecGetJunkAttribute(planSlots[0],
                                     fmstate->ctidAttno,
                                     &isNull);
        /* shouldn't ever get a null result... */
        if (isNull)
            elog(ERROR, "ctid is NULL");
        ctid = (ItemPointer) DatumGetPointer(datum);
    }
 
    /* Convert parameters needed by prepared statement to text form */
    p_values = convert_prep_stmt_params(fmstate, ctid, slots, *numSlots);
 
    /*
     * Execute the prepared statement.
     */
    if (!PQsendQueryPrepared(fmstate->conn,
                             fmstate->p_name,
                             fmstate->p_nums * (*numSlots),
                             p_values,
                             NULL,
                             NULL,
                             0))
        pgfdw_report_error(NULL, fmstate->conn, fmstate->query);
 
    /*
     * Get the result, and check for success.
     */
    res = pgfdw_get_result(fmstate->conn);
    if (PQresultStatus(res) !=
        (fmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK))
        pgfdw_report_error(res, fmstate->conn, fmstate->query);
 
    /* Check number of rows affected, and fetch RETURNING tuple if any */
    if (fmstate->has_returning)
    {
        Assert(*numSlots == 1);
        n_rows = PQntuples(res);
        if (n_rows > 0)
            store_returning_result(fmstate, slots[0], res);
    }
    else
        n_rows = atoi(PQcmdTuples(res));
 
    /* And clean up */
    PQclear(res);
 
    MemoryContextReset(fmstate->temp_cxt);
 
    *numSlots = n_rows;
 
    /*
     * Return NULL if nothing was inserted/updated/deleted on the remote end
     */
    return (n_rows > 0) ? slots : NULL;
}
 
/*
 * prepare_foreign_modify
 *      Establish a prepared statement for execution of INSERT/UPDATE/DELETE
 */
static void
prepare_foreign_modify(PgFdwModifyState *fmstate)
{
    char        prep_name[NAMEDATALEN];
    char       *p_name;
    PGresult   *res;
 
    /*
     * The caller would already have processed a pending asynchronous request
     * if any, so no need to do it here.
     */
 
    /* Construct name we'll use for the prepared statement. */
    snprintf(prep_name, sizeof(prep_name), "pgsql_fdw_prep_%u",
             GetPrepStmtNumber(fmstate->conn));
    p_name = pstrdup(prep_name);
 
    /*
     * We intentionally do not specify parameter types here, but leave the
     * remote server to derive them by default.  This avoids possible problems
     * with the remote server using different type OIDs than we do.  All of
     * the prepared statements we use in this module are simple enough that
     * the remote server will make the right choices.
     */
    if (!PQsendPrepare(fmstate->conn,
                       p_name,
                       fmstate->query,
                       0,
                       NULL))
        pgfdw_report_error(NULL, fmstate->conn, fmstate->query);
 
    /*
     * Get the result, and check for success.
     */
    res = pgfdw_get_result(fmstate->conn);
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pgfdw_report_error(res, fmstate->conn, fmstate->query);
    PQclear(res);
 
    /* This action shows that the prepare has been done. */
    fmstate->p_name = p_name;
}
 
/*
 * convert_prep_stmt_params
 *      Create array of text strings representing parameter values
 *
 * tupleid is ctid to send, or NULL if none
 * slot is slot to get remaining parameters from, or NULL if none
 *
 * Data is constructed in temp_cxt; caller should reset that after use.
 */
static const char **
convert_prep_stmt_params(PgFdwModifyState *fmstate,
                         ItemPointer tupleid,
                         TupleTableSlot **slots,
                         int numSlots)
{
    const char **p_values;
    int         i;
    int         j;
    int         pindex = 0;
    MemoryContext oldcontext;
 
    oldcontext = MemoryContextSwitchTo(fmstate->temp_cxt);
 
    p_values = (const char **) palloc(sizeof(char *) * fmstate->p_nums * numSlots);
 
    /* ctid is provided only for UPDATE/DELETE, which don't allow batching */
    Assert(!(tupleid != NULL && numSlots > 1));
 
    /* 1st parameter should be ctid, if it's in use */
    if (tupleid != NULL)
    {
        Assert(numSlots == 1);
        /* don't need set_transmission_modes for TID output */
        p_values[pindex] = OutputFunctionCall(&fmstate->p_flinfo[pindex],
                                              PointerGetDatum(tupleid));
        pindex++;
    }
 
    /* get following parameters from slots */
    if (slots != NULL && fmstate->target_attrs != NIL)
    {
        TupleDesc   tupdesc = RelationGetDescr(fmstate->rel);
        int         nestlevel;
        ListCell   *lc;
 
        nestlevel = set_transmission_modes();
 
        for (i = 0; i < numSlots; i++)
        {
            j = (tupleid != NULL) ? 1 : 0;
            foreach(lc, fmstate->target_attrs)
            {
                int         attnum = lfirst_int(lc);
                CompactAttribute *attr = TupleDescCompactAttr(tupdesc, attnum - 1);
                Datum       value;
                bool        isnull;
 
                /* Ignore generated columns; they are set to DEFAULT */
                if (attr->attgenerated)
                    continue;
                value = slot_getattr(slots[i], attnum, &isnull);
                if (isnull)
                    p_values[pindex] = NULL;
                else
                    p_values[pindex] = OutputFunctionCall(&fmstate->p_flinfo[j],
                                                          value);
                pindex++;
                j++;
            }
        }
 
        reset_transmission_modes(nestlevel);
    }
 
    Assert(pindex == fmstate->p_nums * numSlots);
 
    MemoryContextSwitchTo(oldcontext);
 
    return p_values;
}
 
/*
 * store_returning_result
 *      Store the result of a RETURNING clause
 */
static void
store_returning_result(PgFdwModifyState *fmstate,
                       TupleTableSlot *slot, PGresult *res)
{
    HeapTuple   newtup;
 
    newtup = make_tuple_from_result_row(res, 0,
                                        fmstate->rel,
                                        fmstate->attinmeta,
                                        fmstate->retrieved_attrs,
                                        NULL,
                                        fmstate->temp_cxt);
 
    /*
     * The returning slot will not necessarily be suitable to store heaptuples
     * directly, so allow for conversion.
     */
    ExecForceStoreHeapTuple(newtup, slot, true);
}
 
/*
 * finish_foreign_modify
 *      Release resources for a foreign insert/update/delete operation
 */
static void
finish_foreign_modify(PgFdwModifyState *fmstate)
{
    Assert(fmstate != NULL);
 
    /* If we created a prepared statement, destroy it */
    deallocate_query(fmstate);
 
    /* Release remote connection */
    ReleaseConnection(fmstate->conn);
    fmstate->conn = NULL;
}
 
/*
 * deallocate_query
 *      Deallocate a prepared statement for a foreign insert/update/delete
 *      operation
 */
static void
deallocate_query(PgFdwModifyState *fmstate)
{
    char        sql[64];
    PGresult   *res;
 
    /* do nothing if the query is not allocated */
    if (!fmstate->p_name)
        return;
 
    snprintf(sql, sizeof(sql), "DEALLOCATE %s", fmstate->p_name);
    res = pgfdw_exec_query(fmstate->conn, sql, fmstate->conn_state);
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pgfdw_report_error(res, fmstate->conn, sql);
    PQclear(res);
    pfree(fmstate->p_name);
    fmstate->p_name = NULL;
}
 
/*
 * build_remote_returning
 *      Build a RETURNING targetlist of a remote query for performing an
 *      UPDATE/DELETE .. RETURNING on a join directly
 */
static List *
build_remote_returning(Index rtindex, Relation rel, List *returningList)
{
    bool        have_wholerow = false;
    List       *tlist = NIL;
    List       *vars;
    ListCell   *lc;
 
    Assert(returningList);
 
    vars = pull_var_clause((Node *) returningList, PVC_INCLUDE_PLACEHOLDERS);
 
    /*
     * If there's a whole-row reference to the target relation, then we'll
     * need all the columns of the relation.
     */
    foreach(lc, vars)
    {
        Var        *var = (Var *) lfirst(lc);
 
        if (IsA(var, Var) &&
            var->varno == rtindex &&
            var->varattno == InvalidAttrNumber)
        {
            have_wholerow = true;
            break;
        }
    }
 
    if (have_wholerow)
    {
        TupleDesc   tupdesc = RelationGetDescr(rel);
        int         i;
 
        for (i = 1; i <= tupdesc->natts; i++)
        {
            Form_pg_attribute attr = TupleDescAttr(tupdesc, i - 1);
            Var        *var;
 
            /* Ignore dropped attributes. */
            if (attr->attisdropped)
                continue;
 
            var = makeVar(rtindex,
                          i,
                          attr->atttypid,
                          attr->atttypmod,
                          attr->attcollation,
                          0);
 
            tlist = lappend(tlist,
                            makeTargetEntry((Expr *) var,
                                            list_length(tlist) + 1,
                                            NULL,
                                            false));
        }
    }
 
    /* Now add any remaining columns to tlist. */
    foreach(lc, vars)
    {
        Var        *var = (Var *) lfirst(lc);
 
        /*
         * No need for whole-row references to the target relation.  We don't
         * need system columns other than ctid and oid either, since those are
         * set locally.
         */
        if (IsA(var, Var) &&
            var->varno == rtindex &&
            var->varattno <= InvalidAttrNumber &&
            var->varattno != SelfItemPointerAttributeNumber)
            continue;           /* don't need it */
 
        if (tlist_member((Expr *) var, tlist))
            continue;           /* already got it */
 
        tlist = lappend(tlist,
                        makeTargetEntry((Expr *) var,
                                        list_length(tlist) + 1,
                                        NULL,
                                        false));
    }
 
    list_free(vars);
 
    return tlist;
}
 
/*
 * rebuild_fdw_scan_tlist
 *      Build new fdw_scan_tlist of given foreign-scan plan node from given
 *      tlist
 *
 * There might be columns that the fdw_scan_tlist of the given foreign-scan
 * plan node contains that the given tlist doesn't.  The fdw_scan_tlist would
 * have contained resjunk columns such as 'ctid' of the target relation and
 * 'wholerow' of non-target relations, but the tlist might not contain them,
 * for example.  So, adjust the tlist so it contains all the columns specified
 * in the fdw_scan_tlist; else setrefs.c will get confused.
 */
static void
rebuild_fdw_scan_tlist(ForeignScan *fscan, List *tlist)
{
    List       *new_tlist = tlist;
    List       *old_tlist = fscan->fdw_scan_tlist;
    ListCell   *lc;
 
    foreach(lc, old_tlist)
    {
        TargetEntry *tle = (TargetEntry *) lfirst(lc);
 
        if (tlist_member(tle->expr, new_tlist))
            continue;           /* already got it */
 
        new_tlist = lappend(new_tlist,
                            makeTargetEntry(tle->expr,
                                            list_length(new_tlist) + 1,
                                            NULL,
                                            false));
    }
    fscan->fdw_scan_tlist = new_tlist;
}
 
/*
 * Execute a direct UPDATE/DELETE statement.
 */
static void
execute_dml_stmt(ForeignScanState *node)
{
    PgFdwDirectModifyState *dmstate = (PgFdwDirectModifyState *) node->fdw_state;
    ExprContext *econtext = node->ss.ps.ps_ExprContext;
    int         numParams = dmstate->numParams;
    const char **values = dmstate->param_values;
 
    /* First, process a pending asynchronous request, if any. */
    if (dmstate->conn_state->pendingAreq)
        process_pending_request(dmstate->conn_state->pendingAreq);
 
    /*
     * Construct array of query parameter values in text format.
     */
    if (numParams > 0)
        process_query_params(econtext,
                             dmstate->param_flinfo,
                             dmstate->param_exprs,
                             values);
 
    /*
     * Notice that we pass NULL for paramTypes, thus forcing the remote server
     * to infer types for all parameters.  Since we explicitly cast every
     * parameter (see deparse.c), the "inference" is trivial and will produce
     * the desired result.  This allows us to avoid assuming that the remote
     * server has the same OIDs we do for the parameters' types.
     */
    if (!PQsendQueryParams(dmstate->conn, dmstate->query, numParams,
                           NULL, values, NULL, NULL, 0))
        pgfdw_report_error(NULL, dmstate->conn, dmstate->query);
 
    /*
     * Get the result, and check for success.
     */
    dmstate->result = pgfdw_get_result(dmstate->conn);
    if (PQresultStatus(dmstate->result) !=
        (dmstate->has_returning ? PGRES_TUPLES_OK : PGRES_COMMAND_OK))
        pgfdw_report_error(dmstate->result, dmstate->conn,
                           dmstate->query);
 
    /*
     * The result potentially needs to survive across multiple executor row
     * cycles, so move it to the context where the dmstate is.
     */
    dmstate->result = libpqsrv_PGresultSetParent(dmstate->result,
                                                 GetMemoryChunkContext(dmstate));
 
    /* Get the number of rows affected. */
    if (dmstate->has_returning)
        dmstate->num_tuples = PQntuples(dmstate->result);
    else
        dmstate->num_tuples = atoi(PQcmdTuples(dmstate->result));
}
 
/*
 * Get the result of a RETURNING clause.
 */
static TupleTableSlot *
get_returning_data(ForeignScanState *node)
{
    PgFdwDirectModifyState *dmstate = (PgFdwDirectModifyState *) node->fdw_state;
    EState     *estate = node->ss.ps.state;
    ResultRelInfo *resultRelInfo = node->resultRelInfo;
    TupleTableSlot *slot = node->ss.ss_ScanTupleSlot;
    TupleTableSlot *resultSlot;
 
    Assert(resultRelInfo->ri_projectReturning);
 
    /* If we didn't get any tuples, must be end of data. */
    if (dmstate->next_tuple >= dmstate->num_tuples)
        return ExecClearTuple(slot);
 
    /* Increment the command es_processed count if necessary. */
    if (dmstate->set_processed)
        estate->es_processed += 1;
 
    /*
     * Store a RETURNING tuple.  If has_returning is false, just emit a dummy
     * tuple.  (has_returning is false when the local query is of the form
     * "UPDATE/DELETE .. RETURNING 1" for example.)
     */
    if (!dmstate->has_returning)
    {
        ExecStoreAllNullTuple(slot);
        resultSlot = slot;
    }
    else
    {
        HeapTuple   newtup;
 
        newtup = make_tuple_from_result_row(dmstate->result,
                                            dmstate->next_tuple,
                                            dmstate->rel,
                                            dmstate->attinmeta,
                                            dmstate->retrieved_attrs,
                                            node,
                                            dmstate->temp_cxt);
        ExecStoreHeapTuple(newtup, slot, false);
        /* Get the updated/deleted tuple. */
        if (dmstate->rel)
            resultSlot = slot;
        else
            resultSlot = apply_returning_filter(dmstate, resultRelInfo, slot, estate);
    }
    dmstate->next_tuple++;
 
    /* Make slot available for evaluation of the local query RETURNING list. */
    resultRelInfo->ri_projectReturning->pi_exprContext->ecxt_scantuple =
        resultSlot;
 
    return slot;
}
 
/*
 * Initialize a filter to extract an updated/deleted tuple from a scan tuple.
 */
static void
init_returning_filter(PgFdwDirectModifyState *dmstate,
                      List *fdw_scan_tlist,
                      Index rtindex)
{
    TupleDesc   resultTupType = RelationGetDescr(dmstate->resultRel);
    ListCell   *lc;
    int         i;
 
    /*
     * Calculate the mapping between the fdw_scan_tlist's entries and the
     * result tuple's attributes.
     *
     * The "map" is an array of indexes of the result tuple's attributes in
     * fdw_scan_tlist, i.e., one entry for every attribute of the result
     * tuple.  We store zero for any attributes that don't have the
     * corresponding entries in that list, marking that a NULL is needed in
     * the result tuple.
     *
     * Also get the indexes of the entries for ctid and oid if any.
     */
    dmstate->attnoMap = (AttrNumber *)
        palloc0(resultTupType->natts * sizeof(AttrNumber));
 
    dmstate->ctidAttno = dmstate->oidAttno = 0;
 
    i = 1;
    dmstate->hasSystemCols = false;
    foreach(lc, fdw_scan_tlist)
    {
        TargetEntry *tle = (TargetEntry *) lfirst(lc);
        Var        *var = (Var *) tle->expr;
 
        Assert(IsA(var, Var));
 
        /*
         * If the Var is a column of the target relation to be retrieved from
         * the foreign server, get the index of the entry.
         */
        if (var->varno == rtindex &&
            list_member_int(dmstate->retrieved_attrs, i))
        {
            int         attrno = var->varattno;
 
            if (attrno < 0)
            {
                /*
                 * We don't retrieve system columns other than ctid and oid.
                 */
                if (attrno == SelfItemPointerAttributeNumber)
                    dmstate->ctidAttno = i;
                else
                    Assert(false);
                dmstate->hasSystemCols = true;
            }
            else
            {
                /*
                 * We don't retrieve whole-row references to the target
                 * relation either.
                 */
                Assert(attrno > 0);
 
                dmstate->attnoMap[attrno - 1] = i;
            }
        }
        i++;
    }
}
 
/*
 * Extract and return an updated/deleted tuple from a scan tuple.
 */
static TupleTableSlot *
apply_returning_filter(PgFdwDirectModifyState *dmstate,
                       ResultRelInfo *resultRelInfo,
                       TupleTableSlot *slot,
                       EState *estate)
{
    TupleDesc   resultTupType = RelationGetDescr(dmstate->resultRel);
    TupleTableSlot *resultSlot;
    Datum      *values;
    bool       *isnull;
    Datum      *old_values;
    bool       *old_isnull;
    int         i;
 
    /*
     * Use the return tuple slot as a place to store the result tuple.
     */
    resultSlot = ExecGetReturningSlot(estate, resultRelInfo);
 
    /*
     * Extract all the values of the scan tuple.
     */
    slot_getallattrs(slot);
    old_values = slot->tts_values;
    old_isnull = slot->tts_isnull;
 
    /*
     * Prepare to build the result tuple.
     */
    ExecClearTuple(resultSlot);
    values = resultSlot->tts_values;
    isnull = resultSlot->tts_isnull;
 
    /*
     * Transpose data into proper fields of the result tuple.
     */
    for (i = 0; i < resultTupType->natts; i++)
    {
        int         j = dmstate->attnoMap[i];
 
        if (j == 0)
        {
            values[i] = (Datum) 0;
            isnull[i] = true;
        }
        else
        {
            values[i] = old_values[j - 1];
            isnull[i] = old_isnull[j - 1];
        }
    }
 
    /*
     * Build the virtual tuple.
     */
    ExecStoreVirtualTuple(resultSlot);
 
    /*
     * If we have any system columns to return, materialize a heap tuple in
     * the slot from column values set above and install system columns in
     * that tuple.
     */
    if (dmstate->hasSystemCols)
    {
        HeapTuple   resultTup = ExecFetchSlotHeapTuple(resultSlot, true, NULL);
 
        /* ctid */
        if (dmstate->ctidAttno)
        {
            ItemPointer ctid = NULL;
 
            ctid = (ItemPointer) DatumGetPointer(old_values[dmstate->ctidAttno - 1]);
            resultTup->t_self = *ctid;
        }
 
        /*
         * And remaining columns
         *
         * Note: since we currently don't allow the target relation to appear
         * on the nullable side of an outer join, any system columns wouldn't
         * go to NULL.
         *
         * Note: no need to care about tableoid here because it will be
         * initialized in ExecProcessReturning().
         */
        HeapTupleHeaderSetXmin(resultTup->t_data, InvalidTransactionId);
        HeapTupleHeaderSetXmax(resultTup->t_data, InvalidTransactionId);
        HeapTupleHeaderSetCmin(resultTup->t_data, InvalidTransactionId);
    }
 
    /*
     * And return the result tuple.
     */
    return resultSlot;
}
 
/*
 * Prepare for processing of parameters used in remote query.
 */
static void
prepare_query_params(PlanState *node,
                     List *fdw_exprs,
                     int numParams,
                     FmgrInfo **param_flinfo,
                     List **param_exprs,
                     const char ***param_values)
{
    int         i;
    ListCell   *lc;
 
    Assert(numParams > 0);
 
    /* Prepare for output conversion of parameters used in remote query. */
    *param_flinfo = palloc0_array(FmgrInfo, numParams);
 
    i = 0;
    foreach(lc, fdw_exprs)
    {
        Node       *param_expr = (Node *) lfirst(lc);
        Oid         typefnoid;
        bool        isvarlena;
 
        getTypeOutputInfo(exprType(param_expr), &typefnoid, &isvarlena);
        fmgr_info(typefnoid, &(*param_flinfo)[i]);
        i++;
    }
 
    /*
     * Prepare remote-parameter expressions for evaluation.  (Note: in
     * practice, we expect that all these expressions will be just Params, so
     * we could possibly do something more efficient than using the full
     * expression-eval machinery for this.  But probably there would be little
     * benefit, and it'd require postgres_fdw to know more than is desirable
     * about Param evaluation.)
     */
    *param_exprs = ExecInitExprList(fdw_exprs, node);
 
    /* Allocate buffer for text form of query parameters. */
    *param_values = (const char **) palloc0(numParams * sizeof(char *));
}
 
/*
 * Construct array of query parameter values in text format.
 */
static void
process_query_params(ExprContext *econtext,
                     FmgrInfo *param_flinfo,
                     List *param_exprs,
                     const char **param_values)
{
    int         nestlevel;
    int         i;
    ListCell   *lc;
 
    nestlevel = set_transmission_modes();
 
    i = 0;
    foreach(lc, param_exprs)
    {
        ExprState  *expr_state = (ExprState *) lfirst(lc);
        Datum       expr_value;
        bool        isNull;
 
        /* Evaluate the parameter expression */
        expr_value = ExecEvalExpr(expr_state, econtext, &isNull);
 
        /*
         * Get string representation of each parameter value by invoking
         * type-specific output function, unless the value is null.
         */
        if (isNull)
            param_values[i] = NULL;
        else
            param_values[i] = OutputFunctionCall(&param_flinfo[i], expr_value);
 
        i++;
    }
 
    reset_transmission_modes(nestlevel);
}
 
/*
 * postgresAnalyzeForeignTable
 *      Test whether analyzing this foreign table is supported
 */
static bool
postgresAnalyzeForeignTable(Relation relation,
                            AcquireSampleRowsFunc *func,
                            BlockNumber *totalpages)
{
    ForeignTable *table;
    UserMapping *user;
    PGconn     *conn;
    StringInfoData sql;
    PGresult   *res;
 
    /* Return the row-analysis function pointer */
    *func = postgresAcquireSampleRowsFunc;
 
    /*
     * Now we have to get the number of pages.  It's annoying that the ANALYZE
     * API requires us to return that now, because it forces some duplication
     * of effort between this routine and postgresAcquireSampleRowsFunc.  But
     * it's probably not worth redefining that API at this point.
     */
 
    /*
     * Get the connection to use.  We do the remote access as the table's
     * owner, even if the ANALYZE was started by some other user.
     */
    table = GetForeignTable(RelationGetRelid(relation));
    user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
    conn = GetConnection(user, false, NULL);
 
    /*
     * Construct command to get page count for relation.
     */
    initStringInfo(&sql);
    deparseAnalyzeSizeSql(&sql, relation);
 
    res = pgfdw_exec_query(conn, sql.data, NULL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pgfdw_report_error(res, conn, sql.data);
 
    if (PQntuples(res) != 1 || PQnfields(res) != 1)
        elog(ERROR, "unexpected result from deparseAnalyzeSizeSql query");
    *totalpages = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
    PQclear(res);
 
    ReleaseConnection(conn);
 
    return true;
}
 
/*
 * postgresGetAnalyzeInfoForForeignTable
 *      Count tuples in foreign table (just get pg_class.reltuples).
 *
 * can_tablesample determines if the remote relation supports acquiring the
 * sample using TABLESAMPLE.
 */
static double
postgresGetAnalyzeInfoForForeignTable(Relation relation, bool *can_tablesample)
{
    ForeignTable *table;
    UserMapping *user;
    PGconn     *conn;
    StringInfoData sql;
    PGresult   *res;
    double      reltuples;
    char        relkind;
 
    /* assume the remote relation does not support TABLESAMPLE */
    *can_tablesample = false;
 
    /*
     * Get the connection to use.  We do the remote access as the table's
     * owner, even if the ANALYZE was started by some other user.
     */
    table = GetForeignTable(RelationGetRelid(relation));
    user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
    conn = GetConnection(user, false, NULL);
 
    /*
     * Construct command to get page count for relation.
     */
    initStringInfo(&sql);
    deparseAnalyzeInfoSql(&sql, relation);
 
    res = pgfdw_exec_query(conn, sql.data, NULL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pgfdw_report_error(res, conn, sql.data);
 
    if (PQntuples(res) != 1 || PQnfields(res) != RELSTATS_NUM_FIELDS)
        elog(ERROR, "unexpected result from deparseAnalyzeInfoSql query");
    /* We don't use relpages here */
    reltuples = strtod(PQgetvalue(res, 0, RELSTATS_RELTUPLES), NULL);
    relkind = *(PQgetvalue(res, 0, RELSTATS_RELKIND));
    PQclear(res);
 
    ReleaseConnection(conn);
 
    /* TABLESAMPLE is supported only for regular tables and matviews */
    *can_tablesample = (relkind == RELKIND_RELATION ||
                        relkind == RELKIND_MATVIEW ||
                        relkind == RELKIND_PARTITIONED_TABLE);
 
    return reltuples;
}
 
/*
 * Acquire a random sample of rows from foreign table managed by postgres_fdw.
 *
 * Selected rows are returned in the caller-allocated array rows[],
 * which must have at least targrows entries.
 * The actual number of rows selected is returned as the function result.
 * We also count the total number of rows in the table and return it into
 * *totalrows.  Note that *totaldeadrows is always set to 0.
 *
 * Note that the returned list of rows is not always in order by physical
 * position in the table.  Therefore, correlation estimates derived later
 * may be meaningless, but it's OK because we don't use the estimates
 * currently (the planner only pays attention to correlation for indexscans).
 */
static int
postgresAcquireSampleRowsFunc(Relation relation, int elevel,
                              HeapTuple *rows, int targrows,
                              double *totalrows,
                              double *totaldeadrows)
{
    PgFdwAnalyzeState astate;
    ForeignTable *table;
    ForeignServer *server;
    UserMapping *user;
    PGconn     *conn;
    int         server_version_num;
    PgFdwSamplingMethod method = ANALYZE_SAMPLE_AUTO;   /* auto is default */
    double      sample_frac = -1.0;
    double      reltuples = -1.0;
    unsigned int cursor_number;
    StringInfoData sql;
    PGresult   *res;
    char        fetch_sql[64];
    int         fetch_size;
    ListCell   *lc;
 
    /* Initialize workspace state */
    astate.rel = relation;
    astate.attinmeta = TupleDescGetAttInMetadata(RelationGetDescr(relation));
 
    astate.rows = rows;
    astate.targrows = targrows;
    astate.numrows = 0;
    astate.samplerows = 0;
    astate.rowstoskip = -1;     /* -1 means not set yet */
    reservoir_init_selection_state(&astate.rstate, targrows);
 
    /* Remember ANALYZE context, and create a per-tuple temp context */
    astate.anl_cxt = CurrentMemoryContext;
    astate.temp_cxt = AllocSetContextCreate(CurrentMemoryContext,
                                            "postgres_fdw temporary data",
                                            ALLOCSET_SMALL_SIZES);
 
    /*
     * Get the connection to use.  We do the remote access as the table's
     * owner, even if the ANALYZE was started by some other user.
     */
    table = GetForeignTable(RelationGetRelid(relation));
    server = GetForeignServer(table->serverid);
    user = GetUserMapping(relation->rd_rel->relowner, table->serverid);
    conn = GetConnection(user, false, NULL);
 
    /* We'll need server version, so fetch it now. */
    server_version_num = PQserverVersion(conn);
 
    /*
     * What sampling method should we use?
     */
    foreach(lc, server->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "analyze_sampling") == 0)
        {
            char       *value = defGetString(def);
 
            if (strcmp(value, "off") == 0)
                method = ANALYZE_SAMPLE_OFF;
            else if (strcmp(value, "auto") == 0)
                method = ANALYZE_SAMPLE_AUTO;
            else if (strcmp(value, "random") == 0)
                method = ANALYZE_SAMPLE_RANDOM;
            else if (strcmp(value, "system") == 0)
                method = ANALYZE_SAMPLE_SYSTEM;
            else if (strcmp(value, "bernoulli") == 0)
                method = ANALYZE_SAMPLE_BERNOULLI;
 
            break;
        }
    }
 
    foreach(lc, table->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "analyze_sampling") == 0)
        {
            char       *value = defGetString(def);
 
            if (strcmp(value, "off") == 0)
                method = ANALYZE_SAMPLE_OFF;
            else if (strcmp(value, "auto") == 0)
                method = ANALYZE_SAMPLE_AUTO;
            else if (strcmp(value, "random") == 0)
                method = ANALYZE_SAMPLE_RANDOM;
            else if (strcmp(value, "system") == 0)
                method = ANALYZE_SAMPLE_SYSTEM;
            else if (strcmp(value, "bernoulli") == 0)
                method = ANALYZE_SAMPLE_BERNOULLI;
 
            break;
        }
    }
 
    /*
     * Error-out if explicitly required one of the TABLESAMPLE methods, but
     * the server does not support it.
     */
    if ((server_version_num < 95000) &&
        (method == ANALYZE_SAMPLE_SYSTEM ||
         method == ANALYZE_SAMPLE_BERNOULLI))
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("remote server does not support TABLESAMPLE feature")));
 
    /*
     * If we've decided to do remote sampling, calculate the sampling rate. We
     * need to get the number of tuples from the remote server, but skip that
     * network round-trip if not needed.
     */
    if (method != ANALYZE_SAMPLE_OFF)
    {
        bool        can_tablesample;
 
        reltuples = postgresGetAnalyzeInfoForForeignTable(relation,
                                                          &can_tablesample);
 
        /*
         * Make sure we're not choosing TABLESAMPLE when the remote relation
         * does not support that. But only do this for "auto" - if the user
         * explicitly requested BERNOULLI/SYSTEM, it's better to fail.
         */
        if (!can_tablesample && (method == ANALYZE_SAMPLE_AUTO))
            method = ANALYZE_SAMPLE_RANDOM;
 
        /*
         * Remote's reltuples could be 0 or -1 if the table has never been
         * vacuumed/analyzed.  In that case, disable sampling after all.
         */
        if ((reltuples <= 0) || (targrows >= reltuples))
            method = ANALYZE_SAMPLE_OFF;
        else
        {
            /*
             * All supported sampling methods require sampling rate, not
             * target rows directly, so we calculate that using the remote
             * reltuples value. That's imperfect, because it might be off a
             * good deal, but that's not something we can (or should) address
             * here.
             *
             * If reltuples is too low (i.e. when table grew), we'll end up
             * sampling more rows - but then we'll apply the local sampling,
             * so we get the expected sample size. This is the same outcome as
             * without remote sampling.
             *
             * If reltuples is too high (e.g. after bulk DELETE), we will end
             * up sampling too few rows.
             *
             * We can't really do much better here - we could try sampling a
             * bit more rows, but we don't know how off the reltuples value is
             * so how much is "a bit more"?
             *
             * Furthermore, the targrows value for partitions is determined
             * based on table size (relpages), which can be off in different
             * ways too. Adjusting the sampling rate here might make the issue
             * worse.
             */
            sample_frac = targrows / reltuples;
 
            /*
             * We should never get sampling rate outside the valid range
             * (between 0.0 and 1.0), because those cases should be covered by
             * the previous branch that sets ANALYZE_SAMPLE_OFF.
             */
            Assert(sample_frac >= 0.0 && sample_frac <= 1.0);
        }
    }
 
    /*
     * For "auto" method, pick the one we believe is best. For servers with
     * TABLESAMPLE support we pick BERNOULLI, for old servers we fall-back to
     * random() to at least reduce network transfer.
     */
    if (method == ANALYZE_SAMPLE_AUTO)
    {
        if (server_version_num < 95000)
            method = ANALYZE_SAMPLE_RANDOM;
        else
            method = ANALYZE_SAMPLE_BERNOULLI;
    }
 
    /*
     * Construct cursor that retrieves whole rows from remote.
     */
    cursor_number = GetCursorNumber(conn);
    initStringInfo(&sql);
    appendStringInfo(&sql, "DECLARE c%u CURSOR FOR ", cursor_number);
 
    deparseAnalyzeSql(&sql, relation, method, sample_frac, &astate.retrieved_attrs);
 
    res = pgfdw_exec_query(conn, sql.data, NULL);
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        pgfdw_report_error(res, conn, sql.data);
    PQclear(res);
 
    /*
     * Determine the fetch size.  The default is arbitrary, but shouldn't be
     * enormous.
     */
    fetch_size = 100;
    foreach(lc, server->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "fetch_size") == 0)
        {
            (void) parse_int(defGetString(def), &fetch_size, 0, NULL);
            break;
        }
    }
    foreach(lc, table->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "fetch_size") == 0)
        {
            (void) parse_int(defGetString(def), &fetch_size, 0, NULL);
            break;
        }
    }
 
    /* Construct command to fetch rows from remote. */
    snprintf(fetch_sql, sizeof(fetch_sql), "FETCH %d FROM c%u",
             fetch_size, cursor_number);
 
    /* Retrieve and process rows a batch at a time. */
    for (;;)
    {
        int         numrows;
        int         i;
 
        /* Allow users to cancel long query */
        CHECK_FOR_INTERRUPTS();
 
        /*
         * XXX possible future improvement: if rowstoskip is large, we could
         * issue a MOVE rather than physically fetching the rows, then just
         * adjust rowstoskip and samplerows appropriately.
         */
 
        /* Fetch some rows */
        res = pgfdw_exec_query(conn, fetch_sql, NULL);
        /* On error, report the original query, not the FETCH. */
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
            pgfdw_report_error(res, conn, sql.data);
 
        /* Process whatever we got. */
        numrows = PQntuples(res);
        for (i = 0; i < numrows; i++)
            analyze_row_processor(res, i, &astate);
 
        PQclear(res);
 
        /* Must be EOF if we didn't get all the rows requested. */
        if (numrows < fetch_size)
            break;
    }
 
    /* Close the cursor, just to be tidy. */
    close_cursor(conn, cursor_number, NULL);
 
    ReleaseConnection(conn);
 
    /* We assume that we have no dead tuple. */
    *totaldeadrows = 0.0;
 
    /*
     * Without sampling, we've retrieved all living tuples from foreign
     * server, so report that as totalrows.  Otherwise use the reltuples
     * estimate we got from the remote side.
     */
    if (method == ANALYZE_SAMPLE_OFF)
        *totalrows = astate.samplerows;
    else
        *totalrows = reltuples;
 
    /*
     * Emit some interesting relation info
     */
    ereport(elevel,
            (errmsg("\"%s\": table contains %.0f rows, %d rows in sample",
                    RelationGetRelationName(relation),
                    *totalrows, astate.numrows)));
 
    return astate.numrows;
}
 
/*
 * Collect sample rows from the result of query.
 *   - Use all tuples in sample until target # of samples are collected.
 *   - Subsequently, replace already-sampled tuples randomly.
 */
static void
analyze_row_processor(PGresult *res, int row, PgFdwAnalyzeState *astate)
{
    int         targrows = astate->targrows;
    int         pos;            /* array index to store tuple in */
    MemoryContext oldcontext;
 
    /* Always increment sample row counter. */
    astate->samplerows += 1;
 
    /*
     * Determine the slot where this sample row should be stored.  Set pos to
     * negative value to indicate the row should be skipped.
     */
    if (astate->numrows < targrows)
    {
        /* First targrows rows are always included into the sample */
        pos = astate->numrows++;
    }
    else
    {
        /*
         * Now we start replacing tuples in the sample until we reach the end
         * of the relation.  Same algorithm as in acquire_sample_rows in
         * analyze.c; see Jeff Vitter's paper.
         */
        if (astate->rowstoskip < 0)
            astate->rowstoskip = reservoir_get_next_S(&astate->rstate, astate->samplerows, targrows);
 
        if (astate->rowstoskip <= 0)
        {
            /* Choose a random reservoir element to replace. */
            pos = (int) (targrows * sampler_random_fract(&astate->rstate.randstate));
            Assert(pos >= 0 && pos < targrows);
            heap_freetuple(astate->rows[pos]);
        }
        else
        {
            /* Skip this tuple. */
            pos = -1;
        }
 
        astate->rowstoskip -= 1;
    }
 
    if (pos >= 0)
    {
        /*
         * Create sample tuple from current result row, and store it in the
         * position determined above.  The tuple has to be created in anl_cxt.
         */
        oldcontext = MemoryContextSwitchTo(astate->anl_cxt);
 
        astate->rows[pos] = make_tuple_from_result_row(res, row,
                                                       astate->rel,
                                                       astate->attinmeta,
                                                       astate->retrieved_attrs,
                                                       NULL,
                                                       astate->temp_cxt);
 
        MemoryContextSwitchTo(oldcontext);
    }
}
 
/*
 * postgresImportForeignStatistics
 *      Attempt to fetch/restore remote statistics instead of sampling.
 */
static bool
postgresImportForeignStatistics(Relation relation, List *va_cols, int elevel)
{
    const char *schemaname = NULL;
    const char *relname = NULL;
    ForeignTable *table;
    ForeignServer *server;
    RemoteStatsResults remstats = {.rel = NULL, .att = NULL};
    RemoteAttributeMapping *remattrmap = NULL;
    int         attrcnt = 0;
    bool        restore_stats = false;
    bool        ok = false;
    ListCell   *lc;
 
    schemaname = get_namespace_name(RelationGetNamespace(relation));
    relname = RelationGetRelationName(relation);
    table = GetForeignTable(RelationGetRelid(relation));
    server = GetForeignServer(table->serverid);
 
    /*
     * Check whether the restore_stats option is enabled on the foreign table.
     * If not, silently ignore the foreign table.
     *
     * Server-level options can be overridden by table-level options, so check
     * server-level first.
     */
    foreach(lc, server->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "restore_stats") == 0)
        {
            restore_stats = defGetBoolean(def);
            break;
        }
    }
    foreach(lc, table->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "restore_stats") == 0)
        {
            restore_stats = defGetBoolean(def);
            break;
        }
    }
    if (!restore_stats)
        return false;
 
    /*
     * We don't currently support statistics import for foreign tables with
     * extended statistics objects.
     */
    if (HasRelationExtStatistics(relation))
    {
        ereport(WARNING,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("cannot import statistics for foreign table \"%s.%s\" --- this foreign table has extended statistics objects",
                       schemaname, relname));
        return false;
    }
 
    /*
     * OK, let's do it.
     */
    ereport(elevel,
            (errmsg("importing statistics for foreign table \"%s.%s\"",
                    schemaname, relname)));
 
    ok = fetch_remote_statistics(relation, va_cols,
                                 table, schemaname, relname,
                                 &attrcnt, &remattrmap, &remstats);
 
    if (ok)
        ok = import_fetched_statistics(schemaname, relname,
                                       attrcnt, remattrmap, &remstats);
 
    if (ok)
        ereport(elevel,
                (errmsg("finished importing statistics for foreign table \"%s.%s\"",
                        schemaname, relname)));
 
    PQclear(remstats.rel);
    PQclear(remstats.att);
    free_remattrmap(remattrmap, attrcnt);
 
    return ok;
}
 
/*
 * Attempt to fetch statistics from a remote server.
 */
static bool
fetch_remote_statistics(Relation relation,
                        List *va_cols,
                        ForeignTable *table,
                        const char *local_schemaname,
                        const char *local_relname,
                        int *p_attrcnt,
                        RemoteAttributeMapping **p_remattrmap,
                        RemoteStatsResults *remstats)
{
    const char *remote_schemaname = NULL;
    const char *remote_relname = NULL;
    UserMapping *user;
    PGconn     *conn;
    PGresult   *relstats = NULL;
    PGresult   *attstats = NULL;
    int         server_version_num;
    RemoteAttributeMapping *remattrmap = NULL;
    int         attrcnt = 0;
    char        relkind;
    double      reltuples;
    bool        ok = false;
    ListCell   *lc;
 
    /*
     * Assume the remote schema/relation names are the same as the local name
     * unless the foreign table's options tell us otherwise.
     */
    remote_schemaname = local_schemaname;
    remote_relname = local_relname;
    foreach(lc, table->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "schema_name") == 0)
            remote_schemaname = defGetString(def);
        else if (strcmp(def->defname, "table_name") == 0)
            remote_relname = defGetString(def);
    }
 
    /*
     * Get connection to the foreign server.  Connection manager will
     * establish new connection if necessary.
     */
    user = GetUserMapping(GetUserId(), table->serverid);
    conn = GetConnection(user, false, NULL);
    remstats->server_version_num = server_version_num = PQserverVersion(conn);
 
    /* Fetch relation stats. */
    remstats->rel = relstats = fetch_relstats(conn, relation);
 
    /*
     * Verify that the remote table is the sort that can have meaningful stats
     * in pg_stats.
     *
     * Note that while relations of kinds RELKIND_INDEX and
     * RELKIND_PARTITIONED_INDEX can have rows in pg_stats, they obviously
     * can't support a foreign table.
     */
    relkind = *PQgetvalue(relstats, 0, RELSTATS_RELKIND);
    switch (relkind)
    {
        case RELKIND_RELATION:
        case RELKIND_FOREIGN_TABLE:
        case RELKIND_MATVIEW:
        case RELKIND_PARTITIONED_TABLE:
            break;
        default:
            ereport(WARNING,
                    errmsg("could not import statistics for foreign table \"%s.%s\" --- remote table \"%s.%s\" is of relkind \"%c\" which cannot have statistics",
                           local_schemaname, local_relname,
                           remote_schemaname, remote_relname, relkind));
            goto fetch_cleanup;
    }
 
    /*
     * If the reltuples value > 0, then then we can expect to find attribute
     * stats for the remote table.
     *
     * In v14 or latter, if a reltuples value is -1, it means the table has
     * never been analyzed, so we wouldn't expect to find the stats for the
     * table; fallback to sampling in that case.  If the value is 0, it means
     * it was empty; in which case skip the stats and import relation stats
     * only.
     *
     * In versions prior to v14, a value of 0 was ambiguous; it could mean
     * that the table had never been analyzed, or that it was empty.  Either
     * way, we wouldn't expect to find the stats for the table, so we fallback
     * to sampling.
     */
    reltuples = strtod(PQgetvalue(relstats, 0, RELSTATS_RELTUPLES), NULL);
    if (((server_version_num < 140000) && (reltuples == 0)) ||
        ((server_version_num >= 140000) && (reltuples == -1)))
    {
        ereport(WARNING,
                errmsg("could not import statistics for foreign table \"%s.%s\" --- remote table \"%s.%s\" has no relation statistics to import",
                       local_schemaname, local_relname,
                       remote_schemaname, remote_relname));
        goto fetch_cleanup;
    }
 
 
    if (reltuples > 0)
    {
        StringInfoData column_list;
 
        *p_remattrmap = remattrmap = build_remattrmap(relation, va_cols,
                                                      &attrcnt, &column_list);
        *p_attrcnt = attrcnt;
 
        if (attrcnt > 0)
        {
            /* Fetch attribute stats. */
            remstats->att = attstats = fetch_attstats(conn,
                                                      server_version_num,
                                                      remote_schemaname,
                                                      remote_relname,
                                                      column_list.data);
 
            /* If any attribute stats are missing, fallback to sampling. */
            if (!match_attrmap(attstats,
                               local_schemaname, local_relname,
                               remote_schemaname, remote_relname,
                               attrcnt, remattrmap))
                goto fetch_cleanup;
        }
    }
 
    ok = true;
 
fetch_cleanup:
    ReleaseConnection(conn);
    return ok;
}
 
/*
 * Attempt to fetch remote relation stats.
 */
static PGresult *
fetch_relstats(PGconn *conn, Relation relation)
{
    StringInfoData sql;
    PGresult   *res;
 
    initStringInfo(&sql);
    deparseAnalyzeInfoSql(&sql, relation);
 
    res = pgfdw_exec_query(conn, sql.data, NULL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pgfdw_report_error(res, conn, sql.data);
 
    if (PQntuples(res) != 1 || PQnfields(res) != RELSTATS_NUM_FIELDS)
        elog(ERROR, "unexpected result from deparseAnalyzeInfoSql query");
 
    return res;
}
 
/*
 * Attempt to fetch remote attribute stats.
 */
static PGresult *
fetch_attstats(PGconn *conn, int server_version_num,
               const char *remote_schemaname, const char *remote_relname,
               const char *column_list)
{
    StringInfoData sql;
    PGresult   *res;
 
    initStringInfo(&sql);
    appendStringInfoString(&sql,
                           "SELECT DISTINCT ON (attname COLLATE \"C\") attname,"
                           " null_frac,"
                           " avg_width,"
                           " n_distinct,"
                           " most_common_vals,"
                           " most_common_freqs,"
                           " histogram_bounds,"
                           " correlation,");
 
    /* Elements stats are supported since Postgres 9.2 */
    if (server_version_num >= 92000)
        appendStringInfoString(&sql,
                               " most_common_elems,"
                               " most_common_elem_freqs,"
                               " elem_count_histogram,");
    else
        appendStringInfoString(&sql,
                               " NULL, NULL, NULL,");
 
    /* Range stats are supported since Postgres 17 */
    if (server_version_num >= 170000)
        appendStringInfoString(&sql,
                               " range_length_histogram,"
                               " range_empty_frac,"
                               " range_bounds_histogram");
    else
        appendStringInfoString(&sql,
                               " NULL, NULL, NULL");
 
    appendStringInfoString(&sql,
                           " FROM pg_catalog.pg_stats"
                           " WHERE schemaname = ");
    deparseStringLiteral(&sql, remote_schemaname);
    appendStringInfoString(&sql,
                           " AND tablename = ");
    deparseStringLiteral(&sql, remote_relname);
    appendStringInfo(&sql,
                     " AND attname = ANY(%s)",
                     column_list);
 
    /* inherited is supported since Postgres 9.0 */
    if (server_version_num >= 90000)
        appendStringInfoString(&sql,
                               " ORDER BY attname COLLATE \"C\", inherited DESC");
    else
        appendStringInfoString(&sql,
                               " ORDER BY attname COLLATE \"C\"");
 
    res = pgfdw_exec_query(conn, sql.data, NULL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pgfdw_report_error(res, conn, sql.data);
 
    if (PQnfields(res) != ATTSTATS_NUM_FIELDS)
        elog(ERROR, "unexpected result from fetch_attstats query");
 
    return res;
}
 
/*
 * Build the mapping of local columns to remote columns and create a column
 * list used for constructing the fetch_attstats query.
 */
static RemoteAttributeMapping *
build_remattrmap(Relation relation, List *va_cols,
                 int *p_attrcnt, StringInfo column_list)
{
    TupleDesc   tupdesc = RelationGetDescr(relation);
    RemoteAttributeMapping *remattrmap = NULL;
    int         attrcnt = 0;
 
    remattrmap = palloc_array(RemoteAttributeMapping, tupdesc->natts);
    initStringInfo(column_list);
    appendStringInfoString(column_list, "ARRAY[");
    for (int i = 0; i < tupdesc->natts; i++)
    {
        Form_pg_attribute attr = TupleDescAttr(tupdesc, i);
        char       *attname = NameStr(attr->attname);
        AttrNumber  attnum = attr->attnum;
        char       *remote_attname;
        List       *fc_options;
        ListCell   *lc;
 
        /* If a list is specified, exclude any attnames not in it. */
        if (!attname_in_list(attname, va_cols))
            continue;
 
        if (!attribute_is_analyzable(relation, attnum, attr, NULL))
            continue;
 
        /* If the column_name option is not specified, go with attname. */
        remote_attname = attname;
        fc_options = GetForeignColumnOptions(RelationGetRelid(relation), attnum);
        foreach(lc, fc_options)
        {
            DefElem    *def = (DefElem *) lfirst(lc);
 
            if (strcmp(def->defname, "column_name") == 0)
            {
                remote_attname = defGetString(def);
                break;
            }
        }
 
        if (attrcnt > 0)
            appendStringInfoString(column_list, ", ");
        deparseStringLiteral(column_list, remote_attname);
 
        remattrmap[attrcnt].local_attnum = attnum;
        remattrmap[attrcnt].local_attname = pstrdup(attname);
        remattrmap[attrcnt].remote_attname = pstrdup(remote_attname);
        remattrmap[attrcnt].res_index = -1;
        attrcnt++;
    }
    appendStringInfoChar(column_list, ']');
 
    /* Sort mapping by remote attribute name if needed. */
    if (attrcnt > 1)
        qsort(remattrmap, attrcnt, sizeof(RemoteAttributeMapping), remattrmap_cmp);
 
    *p_attrcnt = attrcnt;
    return remattrmap;
}
 
/*
 * Free the structure created by build_remattrmap().
 */
static void
free_remattrmap(RemoteAttributeMapping *map, int len)
{
    if (!map)
        return;
 
    for (int i = 0; i < len; i++)
    {
        Assert(map[i].local_attname);
        pfree(map[i].local_attname);
        Assert(map[i].remote_attname);
        pfree(map[i].remote_attname);
    }
 
    pfree(map);
}
 
/*
 * Test if an attribute name is in the list.
 *
 * An empty list means that all attribute names are in the list.
 */
static bool
attname_in_list(const char *attname, List *va_cols)
{
    ListCell   *lc;
 
    if (va_cols == NIL)
        return true;
 
    foreach(lc, va_cols)
    {
        char       *col = strVal(lfirst(lc));
 
        if (strcmp(attname, col) == 0)
            return true;
    }
    return false;
}
 
/*
 * Compare two RemoteAttributeMappings for sorting.
 */
static int
remattrmap_cmp(const void *v1, const void *v2)
{
    const RemoteAttributeMapping *r1 = v1;
    const RemoteAttributeMapping *r2 = v2;
 
    return strncmp(r1->remote_attname, r2->remote_attname, NAMEDATALEN);
}
 
/*
 * Match local columns to result set rows.
 *
 * As the result set consists of the attribute stats for some/all of distinct
 * mapped remote columns in the RemoteAttributeMapping, every entry in it
 * should have at most one match in the result set; which is also ordered by
 * attname, so we find such pairs by doing a merge join.
 *
 * Returns true if every entry in it has a match, and false if not.
 */
static bool
match_attrmap(PGresult *res,
              const char *local_schemaname,
              const char *local_relname,
              const char *remote_schemaname,
              const char *remote_relname,
              int attrcnt,
              RemoteAttributeMapping *remattrmap)
{
    int         numrows = PQntuples(res);
    int         row = -1;
 
    /* No work if there are no stats rows. */
    if (numrows == 0)
    {
        ereport(WARNING,
                errmsg("could not import statistics for foreign table \"%s.%s\" --- remote table \"%s.%s\" has no attribute statistics to import",
                       local_schemaname, local_relname,
                       remote_schemaname, remote_relname));
        return false;
    }
 
    /* Scan all entries in the RemoteAttributeMapping. */
    for (int mapidx = 0; mapidx < attrcnt; mapidx++)
    {
        /*
         * First, check whether the entry matches the current stats row, if it
         * is set.
         */
        if (row >= 0 &&
            strcmp(remattrmap[mapidx].remote_attname,
                   PQgetvalue(res, row, ATTSTATS_ATTNAME)) == 0)
        {
            remattrmap[mapidx].res_index = row;
            continue;
        }
 
        /*
         * If we've exhausted all stats rows, it means the stats for the entry
         * are missing.
         */
        if (row >= numrows - 1)
        {
            ereport(WARNING,
                    errmsg("could not import statistics for foreign table \"%s.%s\" --- no attribute statistics found for column \"%s\" of remote table \"%s.%s\"",
                           local_schemaname, local_relname,
                           remattrmap[mapidx].remote_attname,
                           remote_schemaname, remote_relname));
            return false;
        }
 
        /* Advance to the next stats row. */
        row += 1;
 
        /*
         * If the attname in the entry is less than that in the next stats
         * row, it means the stats for the entry are missing.
         */
        if (strcmp(remattrmap[mapidx].remote_attname,
                   PQgetvalue(res, row, ATTSTATS_ATTNAME)) < 0)
        {
            ereport(WARNING,
                    errmsg("could not import statistics for foreign table \"%s.%s\" --- no attribute statistics found for column \"%s\" of remote table \"%s.%s\"",
                           local_schemaname, local_relname,
                           remattrmap[mapidx].remote_attname,
                           remote_schemaname, remote_relname));
            return false;
        }
 
        /* We should not have got a stats row we didn't expect. */
        if (strcmp(remattrmap[mapidx].remote_attname,
                   PQgetvalue(res, row, ATTSTATS_ATTNAME)) > 0)
            elog(ERROR, "unexpected result from fetch_attstats query");
 
        /* We found a match. */
        Assert(strcmp(remattrmap[mapidx].remote_attname,
                      PQgetvalue(res, row, ATTSTATS_ATTNAME)) == 0);
        remattrmap[mapidx].res_index = row;
    }
 
    /* We should have exhausted all stats rows. */
    if (row < numrows - 1)
        elog(ERROR, "unexpected result from fetch_attstats query");
 
    return true;
}
 
/*
 * Import fetched statistics into the local statistics tables.
 */
static bool
import_fetched_statistics(const char *schemaname,
                          const char *relname,
                          int attrcnt,
                          const RemoteAttributeMapping *remattrmap,
                          RemoteStatsResults *remstats)
{
    SPIPlanPtr  attimport_plan = NULL;
    SPIPlanPtr  attclear_plan = NULL;
    Datum       values[ATTIMPORT_SQL_NUM_FIELDS];
    char        nulls[ATTIMPORT_SQL_NUM_FIELDS];
    int         spirc;
    bool        ok = false;
 
    /* Assign all the invariant parameters common to relation/attribute stats */
    values[ATTIMPORT_SQL_VERSION] = Int32GetDatum(remstats->server_version_num);
    nulls[ATTIMPORT_SQL_VERSION] = ' ';
 
    values[ATTIMPORT_SQL_SCHEMANAME] = CStringGetTextDatum(schemaname);
    nulls[ATTIMPORT_SQL_SCHEMANAME] = ' ';
 
    values[ATTIMPORT_SQL_RELNAME] = CStringGetTextDatum(relname);
    nulls[ATTIMPORT_SQL_RELNAME] = ' ';
 
    SPI_connect();
 
    /*
     * We import attribute statistics first, if any, because those are more
     * prone to errors.  This avoids making a modification of pg_class that
     * will just get rolled back by a failed attribute import.
     */
    if (remstats->att != NULL)
    {
        Assert(PQnfields(remstats->att) == ATTSTATS_NUM_FIELDS);
        Assert(PQntuples(remstats->att) >= 1);
 
        attimport_plan = SPI_prepare(attimport_sql, ATTIMPORT_SQL_NUM_FIELDS,
                                     (Oid *) attimport_argtypes);
        if (attimport_plan == NULL)
            elog(ERROR, "failed to prepare attimport_sql query");
 
        attclear_plan = SPI_prepare(attclear_sql, ATTCLEAR_SQL_NUM_FIELDS,
                                    (Oid *) attclear_argtypes);
        if (attclear_plan == NULL)
            elog(ERROR, "failed to prepare attclear_sql query");
 
        nulls[ATTIMPORT_SQL_ATTNUM] = ' ';
 
        for (int mapidx = 0; mapidx < attrcnt; mapidx++)
        {
            int         row = remattrmap[mapidx].res_index;
            Datum      *values2 = values + 1;
            char       *nulls2 = nulls + 1;
 
            /* All mappings should have been assigned a result set row. */
            Assert(row >= 0);
 
            /*
             * Check for user-requested abort.
             */
            CHECK_FOR_INTERRUPTS();
 
            /*
             * First, clear existing attribute stats.
             *
             * We can re-use the values/nulls because the number of parameters
             * is less and the first two params are the same as the second and
             * third ones in attimport_sql.
             */
            values2[ATTCLEAR_SQL_ATTNAME] =
                CStringGetTextDatum(remattrmap[mapidx].local_attname);
 
            spirc = SPI_execute_plan(attclear_plan, values2, nulls2, false, 1);
            if (spirc != SPI_OK_SELECT)
                elog(ERROR, "failed to execute attclear_sql query for column \"%s\" of foreign table \"%s.%s\"",
                     remattrmap[mapidx].local_attname, schemaname, relname);
 
            values[ATTIMPORT_SQL_ATTNUM] =
                Int16GetDatum(remattrmap[mapidx].local_attnum);
 
            /* Loop through all mappable columns to set remaining arguments */
            for (int i = 0; i < NUM_MAPPED_ATTIMPORT_ARGS; i++)
                map_field_to_arg(remstats->att, row,
                                 attr_result_arg_map[i].res_field,
                                 attr_result_arg_map[i].arg_num,
                                 values, nulls);
 
            spirc = SPI_execute_plan(attimport_plan, values, nulls, false, 1);
            if (spirc != SPI_OK_SELECT)
                elog(ERROR, "failed to execute attimport_sql query for column \"%s\" of foreign table \"%s.%s\"",
                     remattrmap[mapidx].local_attname, schemaname, relname);
 
            if (!import_spi_query_ok())
            {
                ereport(WARNING,
                        errmsg("could not import statistics for foreign table \"%s.%s\" --- attribute statistics import failed for column \"%s\" of this foreign table",
                               schemaname, relname,
                               remattrmap[mapidx].local_attname));
                goto import_cleanup;
            }
        }
    }
 
    /*
     * Import relation stats.  We only perform this once, so there is no point
     * in preparing the statement.
     *
     * We can re-use the values/nulls because the number of parameters is less
     * and the first three params are the same as attimport_sql.
     */
    Assert(remstats->rel != NULL);
    Assert(PQnfields(remstats->rel) == RELSTATS_NUM_FIELDS);
    Assert(PQntuples(remstats->rel) == 1);
    map_field_to_arg(remstats->rel, 0, RELSTATS_RELPAGES,
                     RELIMPORT_SQL_RELPAGES, values, nulls);
    map_field_to_arg(remstats->rel, 0, RELSTATS_RELTUPLES,
                     RELIMPORT_SQL_RELTUPLES, values, nulls);
 
    spirc = SPI_execute_with_args(relimport_sql,
                                  RELIMPORT_SQL_NUM_FIELDS,
                                  (Oid *) relimport_argtypes,
                                  values, nulls, false, 1);
    if (spirc != SPI_OK_SELECT)
        elog(ERROR, "failed to execute relimport_sql query for foreign table \"%s.%s\"",
             schemaname, relname);
 
    if (!import_spi_query_ok())
    {
        ereport(WARNING,
                errmsg("could not import statistics for foreign table \"%s.%s\" --- relation statistics import failed for this foreign table",
                       schemaname, relname));
        goto import_cleanup;
    }
 
    ok = true;
 
import_cleanup:
    if (attimport_plan)
        SPI_freeplan(attimport_plan);
    if (attclear_plan)
        SPI_freeplan(attclear_plan);
    SPI_finish();
    return ok;
}
 
/*
 * Move a string value from a result set to a Text value of a Datum array.
 */
static void
map_field_to_arg(PGresult *res, int row, int field,
                 int arg, Datum *values, char *nulls)
{
    if (PQgetisnull(res, row, field))
    {
        values[arg] = (Datum) 0;
        nulls[arg] = 'n';
    }
    else
    {
        const char *s = PQgetvalue(res, row, field);
 
        values[arg] = CStringGetTextDatum(s);
        nulls[arg] = ' ';
    }
}
 
/*
 * Check the 1x1 result set of a pg_restore_*_stats() command for success.
 */
static bool
import_spi_query_ok(void)
{
    TupleDesc   tupdesc;
    Datum       dat;
    bool        isnull;
 
    Assert(SPI_tuptable != NULL);
    Assert(SPI_processed == 1);
 
    tupdesc = SPI_tuptable->tupdesc;
    Assert(tupdesc->natts == 1);
    Assert(TupleDescAttr(tupdesc, 0)->atttypid == BOOLOID);
    dat = SPI_getbinval(SPI_tuptable->vals[0], tupdesc, 1, &isnull);
    Assert(!isnull);
 
    return DatumGetBool(dat);
}
 
/*
 * Import a foreign schema
 */
static List *
postgresImportForeignSchema(ImportForeignSchemaStmt *stmt, Oid serverOid)
{
    List       *commands = NIL;
    bool        import_collate = true;
    bool        import_default = false;
    bool        import_generated = true;
    bool        import_not_null = true;
    ForeignServer *server;
    UserMapping *mapping;
    PGconn     *conn;
    StringInfoData buf;
    PGresult   *res;
    int         numrows,
                i;
    ListCell   *lc;
 
    /* Parse statement options */
    foreach(lc, stmt->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "import_collate") == 0)
            import_collate = defGetBoolean(def);
        else if (strcmp(def->defname, "import_default") == 0)
            import_default = defGetBoolean(def);
        else if (strcmp(def->defname, "import_generated") == 0)
            import_generated = defGetBoolean(def);
        else if (strcmp(def->defname, "import_not_null") == 0)
            import_not_null = defGetBoolean(def);
        else
            ereport(ERROR,
                    (errcode(ERRCODE_FDW_INVALID_OPTION_NAME),
                     errmsg("invalid option \"%s\"", def->defname)));
    }
 
    /*
     * Get connection to the foreign server.  Connection manager will
     * establish new connection if necessary.
     */
    server = GetForeignServer(serverOid);
    mapping = GetUserMapping(GetUserId(), server->serverid);
    conn = GetConnection(mapping, false, NULL);
 
    /* Don't attempt to import collation if remote server hasn't got it */
    if (PQserverVersion(conn) < 90100)
        import_collate = false;
 
    /* Create workspace for strings */
    initStringInfo(&buf);
 
    /* Check that the schema really exists */
    appendStringInfoString(&buf, "SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = ");
    deparseStringLiteral(&buf, stmt->remote_schema);
 
    res = pgfdw_exec_query(conn, buf.data, NULL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pgfdw_report_error(res, conn, buf.data);
 
    if (PQntuples(res) != 1)
        ereport(ERROR,
                (errcode(ERRCODE_FDW_SCHEMA_NOT_FOUND),
                 errmsg("schema \"%s\" is not present on foreign server \"%s\"",
                        stmt->remote_schema, server->servername)));
 
    PQclear(res);
    resetStringInfo(&buf);
 
    /*
     * Fetch all table data from this schema, possibly restricted by EXCEPT or
     * LIMIT TO.  (We don't actually need to pay any attention to EXCEPT/LIMIT
     * TO here, because the core code will filter the statements we return
     * according to those lists anyway.  But it should save a few cycles to
     * not process excluded tables in the first place.)
     *
     * Import table data for partitions only when they are explicitly
     * specified in LIMIT TO clause. Otherwise ignore them and only include
     * the definitions of the root partitioned tables to allow access to the
     * complete remote data set locally in the schema imported.
     *
     * Note: because we run the connection with search_path restricted to
     * pg_catalog, the format_type() and pg_get_expr() outputs will always
     * include a schema name for types/functions in other schemas, which is
     * what we want.
     */
    appendStringInfoString(&buf,
                           "SELECT relname, "
                           "  attname, "
                           "  format_type(atttypid, atttypmod), "
                           "  attnotnull, "
                           "  pg_get_expr(adbin, adrelid), ");
 
    /* Generated columns are supported since Postgres 12 */
    if (PQserverVersion(conn) >= 120000)
        appendStringInfoString(&buf,
                               "  attgenerated, ");
    else
        appendStringInfoString(&buf,
                               "  NULL, ");
 
    if (import_collate)
        appendStringInfoString(&buf,
                               "  collname, "
                               "  collnsp.nspname ");
    else
        appendStringInfoString(&buf,
                               "  NULL, NULL ");
 
    appendStringInfoString(&buf,
                           "FROM pg_class c "
                           "  JOIN pg_namespace n ON "
                           "    relnamespace = n.oid "
                           "  LEFT JOIN pg_attribute a ON "
                           "    attrelid = c.oid AND attnum > 0 "
                           "      AND NOT attisdropped "
                           "  LEFT JOIN pg_attrdef ad ON "
                           "    adrelid = c.oid AND adnum = attnum ");
 
    if (import_collate)
        appendStringInfoString(&buf,
                               "  LEFT JOIN pg_collation coll ON "
                               "    coll.oid = attcollation "
                               "  LEFT JOIN pg_namespace collnsp ON "
                               "    collnsp.oid = collnamespace ");
 
    appendStringInfoString(&buf,
                           "WHERE c.relkind IN ("
                           CppAsString2(RELKIND_RELATION) ","
                           CppAsString2(RELKIND_VIEW) ","
                           CppAsString2(RELKIND_FOREIGN_TABLE) ","
                           CppAsString2(RELKIND_MATVIEW) ","
                           CppAsString2(RELKIND_PARTITIONED_TABLE) ") "
                           "  AND n.nspname = ");
    deparseStringLiteral(&buf, stmt->remote_schema);
 
    /* Partitions are supported since Postgres 10 */
    if (PQserverVersion(conn) >= 100000 &&
        stmt->list_type != FDW_IMPORT_SCHEMA_LIMIT_TO)
        appendStringInfoString(&buf, " AND NOT c.relispartition ");
 
    /* Apply restrictions for LIMIT TO and EXCEPT */
    if (stmt->list_type == FDW_IMPORT_SCHEMA_LIMIT_TO ||
        stmt->list_type == FDW_IMPORT_SCHEMA_EXCEPT)
    {
        bool        first_item = true;
 
        appendStringInfoString(&buf, " AND c.relname ");
        if (stmt->list_type == FDW_IMPORT_SCHEMA_EXCEPT)
            appendStringInfoString(&buf, "NOT ");
        appendStringInfoString(&buf, "IN (");
 
        /* Append list of table names within IN clause */
        foreach(lc, stmt->table_list)
        {
            RangeVar   *rv = (RangeVar *) lfirst(lc);
 
            if (first_item)
                first_item = false;
            else
                appendStringInfoString(&buf, ", ");
            deparseStringLiteral(&buf, rv->relname);
        }
        appendStringInfoChar(&buf, ')');
    }
 
    /* Append ORDER BY at the end of query to ensure output ordering */
    appendStringInfoString(&buf, " ORDER BY c.relname, a.attnum");
 
    /* Fetch the data */
    res = pgfdw_exec_query(conn, buf.data, NULL);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        pgfdw_report_error(res, conn, buf.data);
 
    /* Process results */
    numrows = PQntuples(res);
    /* note: incrementation of i happens in inner loop's while() test */
    for (i = 0; i < numrows;)
    {
        char       *tablename = PQgetvalue(res, i, 0);
        bool        first_item = true;
 
        resetStringInfo(&buf);
        appendStringInfo(&buf, "CREATE FOREIGN TABLE %s (\n",
                         quote_identifier(tablename));
 
        /* Scan all rows for this table */
        do
        {
            char       *attname;
            char       *typename;
            char       *attnotnull;
            char       *attgenerated;
            char       *attdefault;
            char       *collname;
            char       *collnamespace;
 
            /* If table has no columns, we'll see nulls here */
            if (PQgetisnull(res, i, 1))
                continue;
 
            attname = PQgetvalue(res, i, 1);
            typename = PQgetvalue(res, i, 2);
            attnotnull = PQgetvalue(res, i, 3);
            attdefault = PQgetisnull(res, i, 4) ? NULL :
                PQgetvalue(res, i, 4);
            attgenerated = PQgetisnull(res, i, 5) ? NULL :
                PQgetvalue(res, i, 5);
            collname = PQgetisnull(res, i, 6) ? NULL :
                PQgetvalue(res, i, 6);
            collnamespace = PQgetisnull(res, i, 7) ? NULL :
                PQgetvalue(res, i, 7);
 
            if (first_item)
                first_item = false;
            else
                appendStringInfoString(&buf, ",\n");
 
            /* Print column name and type */
            appendStringInfo(&buf, "  %s %s",
                             quote_identifier(attname),
                             typename);
 
            /*
             * Add column_name option so that renaming the foreign table's
             * column doesn't break the association to the underlying column.
             */
            appendStringInfoString(&buf, " OPTIONS (column_name ");
            deparseStringLiteral(&buf, attname);
            appendStringInfoChar(&buf, ')');
 
            /* Add COLLATE if needed */
            if (import_collate && collname != NULL && collnamespace != NULL)
                appendStringInfo(&buf, " COLLATE %s.%s",
                                 quote_identifier(collnamespace),
                                 quote_identifier(collname));
 
            /* Add DEFAULT if needed */
            if (import_default && attdefault != NULL &&
                (!attgenerated || !attgenerated[0]))
                appendStringInfo(&buf, " DEFAULT %s", attdefault);
 
            /* Add GENERATED if needed */
            if (import_generated && attgenerated != NULL &&
                attgenerated[0] == ATTRIBUTE_GENERATED_STORED)
            {
                Assert(attdefault != NULL);
                appendStringInfo(&buf,
                                 " GENERATED ALWAYS AS (%s) STORED",
                                 attdefault);
            }
 
            /* Add NOT NULL if needed */
            if (import_not_null && attnotnull[0] == 't')
                appendStringInfoString(&buf, " NOT NULL");
        }
        while (++i < numrows &&
               strcmp(PQgetvalue(res, i, 0), tablename) == 0);
 
        /*
         * Add server name and table-level options.  We specify remote schema
         * and table name as options (the latter to ensure that renaming the
         * foreign table doesn't break the association).
         */
        appendStringInfo(&buf, "\n) SERVER %s\nOPTIONS (",
                         quote_identifier(server->servername));
 
        appendStringInfoString(&buf, "schema_name ");
        deparseStringLiteral(&buf, stmt->remote_schema);
        appendStringInfoString(&buf, ", table_name ");
        deparseStringLiteral(&buf, tablename);
 
        appendStringInfoString(&buf, ");");
 
        commands = lappend(commands, pstrdup(buf.data));
    }
    PQclear(res);
 
    ReleaseConnection(conn);
 
    return commands;
}
 
/*
 * Check if reltarget is safe enough to push down semi-join.  Reltarget is not
 * safe, if it contains references to inner rel relids, which do not belong to
 * outer rel.
 */
static bool
semijoin_target_ok(PlannerInfo *root, RelOptInfo *joinrel, RelOptInfo *outerrel, RelOptInfo *innerrel)
{
    List       *vars;
    ListCell   *lc;
    bool        ok = true;
 
    Assert(joinrel->reltarget);
 
    vars = pull_var_clause((Node *) joinrel->reltarget->exprs, PVC_INCLUDE_PLACEHOLDERS);
 
    foreach(lc, vars)
    {
        Var        *var = (Var *) lfirst(lc);
 
        if (!IsA(var, Var))
            continue;
 
        if (bms_is_member(var->varno, innerrel->relids))
        {
            /*
             * The planner can create semi-join, which refers to inner rel
             * vars in its target list. However, we deparse semi-join as an
             * exists() subquery, so can't handle references to inner rel in
             * the target list.
             */
            Assert(!bms_is_member(var->varno, outerrel->relids));
            ok = false;
            break;
        }
    }
    return ok;
}
 
/*
 * Assess whether the join between inner and outer relations can be pushed down
 * to the foreign server. As a side effect, save information we obtain in this
 * function to PgFdwRelationInfo passed in.
 */
static bool
foreign_join_ok(PlannerInfo *root, RelOptInfo *joinrel, JoinType jointype,
                RelOptInfo *outerrel, RelOptInfo *innerrel,
                JoinPathExtraData *extra)
{
    PgFdwRelationInfo *fpinfo;
    PgFdwRelationInfo *fpinfo_o;
    PgFdwRelationInfo *fpinfo_i;
    ListCell   *lc;
    List       *joinclauses;
 
    /*
     * We support pushing down INNER, LEFT, RIGHT, FULL OUTER and SEMI joins.
     * Constructing queries representing ANTI joins is hard, hence not
     * considered right now.
     */
    if (jointype != JOIN_INNER && jointype != JOIN_LEFT &&
        jointype != JOIN_RIGHT && jointype != JOIN_FULL &&
        jointype != JOIN_SEMI)
        return false;
 
    /*
     * We can't push down semi-join if its reltarget is not safe
     */
    if ((jointype == JOIN_SEMI) && !semijoin_target_ok(root, joinrel, outerrel, innerrel))
        return false;
 
    /*
     * If either of the joining relations is marked as unsafe to pushdown, the
     * join can not be pushed down.
     */
    fpinfo = (PgFdwRelationInfo *) joinrel->fdw_private;
    fpinfo_o = (PgFdwRelationInfo *) outerrel->fdw_private;
    fpinfo_i = (PgFdwRelationInfo *) innerrel->fdw_private;
    if (!fpinfo_o || !fpinfo_o->pushdown_safe ||
        !fpinfo_i || !fpinfo_i->pushdown_safe)
        return false;
 
    /*
     * If joining relations have local conditions, those conditions are
     * required to be applied before joining the relations. Hence the join can
     * not be pushed down.
     */
    if (fpinfo_o->local_conds || fpinfo_i->local_conds)
        return false;
 
    /*
     * Merge FDW options.  We might be tempted to do this after we have deemed
     * the foreign join to be OK.  But we must do this beforehand so that we
     * know which quals can be evaluated on the foreign server, which might
     * depend on shippable_extensions.
     */
    fpinfo->server = fpinfo_o->server;
    merge_fdw_options(fpinfo, fpinfo_o, fpinfo_i);
 
    /*
     * Separate restrict list into join quals and pushed-down (other) quals.
     *
     * Join quals belonging to an outer join must all be shippable, else we
     * cannot execute the join remotely.  Add such quals to 'joinclauses'.
     *
     * Add other quals to fpinfo->remote_conds if they are shippable, else to
     * fpinfo->local_conds.  In an inner join it's okay to execute conditions
     * either locally or remotely; the same is true for pushed-down conditions
     * at an outer join.
     *
     * Note we might return failure after having already scribbled on
     * fpinfo->remote_conds and fpinfo->local_conds.  That's okay because we
     * won't consult those lists again if we deem the join unshippable.
     */
    joinclauses = NIL;
    foreach(lc, extra->restrictlist)
    {
        RestrictInfo *rinfo = lfirst_node(RestrictInfo, lc);
        bool        is_remote_clause = is_foreign_expr(root, joinrel,
                                                       rinfo->clause);
 
        if (IS_OUTER_JOIN(jointype) &&
            !RINFO_IS_PUSHED_DOWN(rinfo, joinrel->relids))
        {
            if (!is_remote_clause)
                return false;
            joinclauses = lappend(joinclauses, rinfo);
        }
        else
        {
            if (is_remote_clause)
                fpinfo->remote_conds = lappend(fpinfo->remote_conds, rinfo);
            else
                fpinfo->local_conds = lappend(fpinfo->local_conds, rinfo);
        }
    }
 
    /*
     * deparseExplicitTargetList() isn't smart enough to handle anything other
     * than a Var.  In particular, if there's some PlaceHolderVar that would
     * need to be evaluated within this join tree (because there's an upper
     * reference to a quantity that may go to NULL as a result of an outer
     * join), then we can't try to push the join down because we'll fail when
     * we get to deparseExplicitTargetList().  However, a PlaceHolderVar that
     * needs to be evaluated *at the top* of this join tree is OK, because we
     * can do that locally after fetching the results from the remote side.
     */
    foreach(lc, root->placeholder_list)
    {
        PlaceHolderInfo *phinfo = lfirst(lc);
        Relids      relids;
 
        /* PlaceHolderInfo refers to parent relids, not child relids. */
        relids = IS_OTHER_REL(joinrel) ?
            joinrel->top_parent_relids : joinrel->relids;
 
        if (bms_is_subset(phinfo->ph_eval_at, relids) &&
            bms_nonempty_difference(relids, phinfo->ph_eval_at))
            return false;
    }
 
    /* Save the join clauses, for later use. */
    fpinfo->joinclauses = joinclauses;
 
    fpinfo->outerrel = outerrel;
    fpinfo->innerrel = innerrel;
    fpinfo->jointype = jointype;
 
    /*
     * By default, both the input relations are not required to be deparsed as
     * subqueries, but there might be some relations covered by the input
     * relations that are required to be deparsed as subqueries, so save the
     * relids of those relations for later use by the deparser.
     */
    fpinfo->make_outerrel_subquery = false;
    fpinfo->make_innerrel_subquery = false;
    Assert(bms_is_subset(fpinfo_o->lower_subquery_rels, outerrel->relids));
    Assert(bms_is_subset(fpinfo_i->lower_subquery_rels, innerrel->relids));
    fpinfo->lower_subquery_rels = bms_union(fpinfo_o->lower_subquery_rels,
                                            fpinfo_i->lower_subquery_rels);
    fpinfo->hidden_subquery_rels = bms_union(fpinfo_o->hidden_subquery_rels,
                                             fpinfo_i->hidden_subquery_rels);
 
    /*
     * Pull the other remote conditions from the joining relations into join
     * clauses or other remote clauses (remote_conds) of this relation
     * wherever possible. This avoids building subqueries at every join step.
     *
     * For an inner join, clauses from both the relations are added to the
     * other remote clauses. For LEFT and RIGHT OUTER join, the clauses from
     * the outer side are added to remote_conds since those can be evaluated
     * after the join is evaluated. The clauses from inner side are added to
     * the joinclauses, since they need to be evaluated while constructing the
     * join.
     *
     * For SEMI-JOIN clauses from inner relation can not be added to
     * remote_conds, but should be treated as join clauses (as they are
     * deparsed to EXISTS subquery, where inner relation can be referred). A
     * list of relation ids, which can't be referred to from higher levels, is
     * preserved as a hidden_subquery_rels list.
     *
     * For a FULL OUTER JOIN, the other clauses from either relation can not
     * be added to the joinclauses or remote_conds, since each relation acts
     * as an outer relation for the other.
     *
     * The joining sides can not have local conditions, thus no need to test
     * shippability of the clauses being pulled up.
     */
    switch (jointype)
    {
        case JOIN_INNER:
            fpinfo->remote_conds = list_concat(fpinfo->remote_conds,
                                               fpinfo_i->remote_conds);
            fpinfo->remote_conds = list_concat(fpinfo->remote_conds,
                                               fpinfo_o->remote_conds);
            break;
 
        case JOIN_LEFT:
 
            /*
             * When semi-join is involved in the inner or outer part of the
             * left join, it's deparsed as a subquery, and we can't refer to
             * its vars on the upper level.
             */
            if (bms_is_empty(fpinfo_i->hidden_subquery_rels))
                fpinfo->joinclauses = list_concat(fpinfo->joinclauses,
                                                  fpinfo_i->remote_conds);
            if (bms_is_empty(fpinfo_o->hidden_subquery_rels))
                fpinfo->remote_conds = list_concat(fpinfo->remote_conds,
                                                   fpinfo_o->remote_conds);
            break;
 
        case JOIN_RIGHT:
 
            /*
             * When semi-join is involved in the inner or outer part of the
             * right join, it's deparsed as a subquery, and we can't refer to
             * its vars on the upper level.
             */
            if (bms_is_empty(fpinfo_o->hidden_subquery_rels))
                fpinfo->joinclauses = list_concat(fpinfo->joinclauses,
                                                  fpinfo_o->remote_conds);
            if (bms_is_empty(fpinfo_i->hidden_subquery_rels))
                fpinfo->remote_conds = list_concat(fpinfo->remote_conds,
                                                   fpinfo_i->remote_conds);
            break;
 
        case JOIN_SEMI:
            fpinfo->joinclauses = list_concat(fpinfo->joinclauses,
                                              fpinfo_i->remote_conds);
            fpinfo->joinclauses = list_concat(fpinfo->joinclauses,
                                              fpinfo->remote_conds);
            fpinfo->remote_conds = list_copy(fpinfo_o->remote_conds);
            fpinfo->hidden_subquery_rels = bms_union(fpinfo->hidden_subquery_rels,
                                                     innerrel->relids);
            break;
 
        case JOIN_FULL:
 
            /*
             * In this case, if any of the input relations has conditions, we
             * need to deparse that relation as a subquery so that the
             * conditions can be evaluated before the join.  Remember it in
             * the fpinfo of this relation so that the deparser can take
             * appropriate action.  Also, save the relids of base relations
             * covered by that relation for later use by the deparser.
             */
            if (fpinfo_o->remote_conds)
            {
                fpinfo->make_outerrel_subquery = true;
                fpinfo->lower_subquery_rels =
                    bms_add_members(fpinfo->lower_subquery_rels,
                                    outerrel->relids);
            }
            if (fpinfo_i->remote_conds)
            {
                fpinfo->make_innerrel_subquery = true;
                fpinfo->lower_subquery_rels =
                    bms_add_members(fpinfo->lower_subquery_rels,
                                    innerrel->relids);
            }
            break;
 
        default:
            /* Should not happen, we have just checked this above */
            elog(ERROR, "unsupported join type %d", jointype);
    }
 
    /*
     * For an inner join, all restrictions can be treated alike. Treating the
     * pushed down conditions as join conditions allows a top level full outer
     * join to be deparsed without requiring subqueries.
     */
    if (jointype == JOIN_INNER)
    {
        Assert(!fpinfo->joinclauses);
        fpinfo->joinclauses = fpinfo->remote_conds;
        fpinfo->remote_conds = NIL;
    }
    else if (jointype == JOIN_LEFT || jointype == JOIN_RIGHT || jointype == JOIN_FULL)
    {
        /*
         * Conditions, generated from semi-joins, should be evaluated before
         * LEFT/RIGHT/FULL join.
         */
        if (!bms_is_empty(fpinfo_o->hidden_subquery_rels))
        {
            fpinfo->make_outerrel_subquery = true;
            fpinfo->lower_subquery_rels = bms_add_members(fpinfo->lower_subquery_rels, outerrel->relids);
        }
 
        if (!bms_is_empty(fpinfo_i->hidden_subquery_rels))
        {
            fpinfo->make_innerrel_subquery = true;
            fpinfo->lower_subquery_rels = bms_add_members(fpinfo->lower_subquery_rels, innerrel->relids);
        }
    }
 
    /* Mark that this join can be pushed down safely */
    fpinfo->pushdown_safe = true;
 
    /* Get user mapping */
    if (fpinfo->use_remote_estimate)
    {
        if (fpinfo_o->use_remote_estimate)
            fpinfo->user = fpinfo_o->user;
        else
            fpinfo->user = fpinfo_i->user;
    }
    else
        fpinfo->user = NULL;
 
    /*
     * Set # of retrieved rows and cached relation costs to some negative
     * value, so that we can detect when they are set to some sensible values,
     * during one (usually the first) of the calls to estimate_path_cost_size.
     */
    fpinfo->retrieved_rows = -1;
    fpinfo->rel_startup_cost = -1;
    fpinfo->rel_total_cost = -1;
 
    /*
     * Set the string describing this join relation to be used in EXPLAIN
     * output of corresponding ForeignScan.  Note that the decoration we add
     * to the base relation names mustn't include any digits, or it'll confuse
     * postgresExplainForeignScan.
     */
    fpinfo->relation_name = psprintf("(%s) %s JOIN (%s)",
                                     fpinfo_o->relation_name,
                                     get_jointype_name(fpinfo->jointype),
                                     fpinfo_i->relation_name);
 
    /*
     * Set the relation index.  This is defined as the position of this
     * joinrel in the join_rel_list list plus the length of the rtable list.
     * Note that since this joinrel is at the end of the join_rel_list list
     * when we are called, we can get the position by list_length.
     */
    Assert(fpinfo->relation_index == 0);    /* shouldn't be set yet */
    fpinfo->relation_index =
        list_length(root->parse->rtable) + list_length(root->join_rel_list);
 
    return true;
}
 
static void
add_paths_with_pathkeys_for_rel(PlannerInfo *root, RelOptInfo *rel,
                                Path *epq_path, List *restrictlist)
{
    List       *useful_pathkeys_list = NIL; /* List of all pathkeys */
    ListCell   *lc;
 
    useful_pathkeys_list = get_useful_pathkeys_for_relation(root, rel);
 
    /*
     * Before creating sorted paths, arrange for the passed-in EPQ path, if
     * any, to return columns needed by the parent ForeignScan node so that
     * they will propagate up through Sort nodes injected below, if necessary.
     */
    if (epq_path != NULL && useful_pathkeys_list != NIL)
    {
        PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) rel->fdw_private;
        PathTarget *target = copy_pathtarget(epq_path->pathtarget);
 
        /* Include columns required for evaluating PHVs in the tlist. */
        add_new_columns_to_pathtarget(target,
                                      pull_var_clause((Node *) target->exprs,
                                                      PVC_RECURSE_PLACEHOLDERS));
 
        /* Include columns required for evaluating the local conditions. */
        foreach(lc, fpinfo->local_conds)
        {
            RestrictInfo *rinfo = lfirst_node(RestrictInfo, lc);
 
            add_new_columns_to_pathtarget(target,
                                          pull_var_clause((Node *) rinfo->clause,
                                                          PVC_RECURSE_PLACEHOLDERS));
        }
 
        /*
         * If we have added any new columns, adjust the tlist of the EPQ path.
         *
         * Note: the plan created using this path will only be used to execute
         * EPQ checks, where accuracy of the plan cost and width estimates
         * would not be important, so we do not do set_pathtarget_cost_width()
         * for the new pathtarget here.  See also postgresGetForeignPlan().
         */
        if (list_length(target->exprs) > list_length(epq_path->pathtarget->exprs))
        {
            /* The EPQ path is a join path, so it is projection-capable. */
            Assert(is_projection_capable_path(epq_path));
 
            /*
             * Use create_projection_path() here, so as to avoid modifying it
             * in place.
             */
            epq_path = (Path *) create_projection_path(root,
                                                       rel,
                                                       epq_path,
                                                       target);
        }
    }
 
    /* Create one path for each set of pathkeys we found above. */
    foreach(lc, useful_pathkeys_list)
    {
        double      rows;
        int         width;
        int         disabled_nodes;
        Cost        startup_cost;
        Cost        total_cost;
        List       *useful_pathkeys = lfirst(lc);
        Path       *sorted_epq_path;
 
        estimate_path_cost_size(root, rel, NIL, useful_pathkeys, NULL,
                                &rows, &width, &disabled_nodes,
                                &startup_cost, &total_cost);
 
        /*
         * The EPQ path must be at least as well sorted as the path itself, in
         * case it gets used as input to a mergejoin.
         */
        sorted_epq_path = epq_path;
        if (sorted_epq_path != NULL &&
            !pathkeys_contained_in(useful_pathkeys,
                                   sorted_epq_path->pathkeys))
            sorted_epq_path = (Path *)
                create_sort_path(root,
                                 rel,
                                 sorted_epq_path,
                                 useful_pathkeys,
                                 -1.0);
 
        if (IS_SIMPLE_REL(rel))
            add_path(rel, (Path *)
                     create_foreignscan_path(root, rel,
                                             NULL,
                                             rows,
                                             disabled_nodes,
                                             startup_cost,
                                             total_cost,
                                             useful_pathkeys,
                                             rel->lateral_relids,
                                             sorted_epq_path,
                                             NIL,   /* no fdw_restrictinfo
                                                     * list */
                                             NIL));
        else
            add_path(rel, (Path *)
                     create_foreign_join_path(root, rel,
                                              NULL,
                                              rows,
                                              disabled_nodes,
                                              startup_cost,
                                              total_cost,
                                              useful_pathkeys,
                                              rel->lateral_relids,
                                              sorted_epq_path,
                                              restrictlist,
                                              NIL));
    }
}
 
/*
 * Parse options from foreign server and apply them to fpinfo.
 *
 * New options might also require tweaking merge_fdw_options().
 */
static void
apply_server_options(PgFdwRelationInfo *fpinfo)
{
    ListCell   *lc;
 
    foreach(lc, fpinfo->server->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "use_remote_estimate") == 0)
            fpinfo->use_remote_estimate = defGetBoolean(def);
        else if (strcmp(def->defname, "fdw_startup_cost") == 0)
            (void) parse_real(defGetString(def), &fpinfo->fdw_startup_cost, 0,
                              NULL);
        else if (strcmp(def->defname, "fdw_tuple_cost") == 0)
            (void) parse_real(defGetString(def), &fpinfo->fdw_tuple_cost, 0,
                              NULL);
        else if (strcmp(def->defname, "extensions") == 0)
            fpinfo->shippable_extensions =
                ExtractExtensionList(defGetString(def), false);
        else if (strcmp(def->defname, "fetch_size") == 0)
            (void) parse_int(defGetString(def), &fpinfo->fetch_size, 0, NULL);
        else if (strcmp(def->defname, "async_capable") == 0)
            fpinfo->async_capable = defGetBoolean(def);
    }
}
 
/*
 * Parse options from foreign table and apply them to fpinfo.
 *
 * New options might also require tweaking merge_fdw_options().
 */
static void
apply_table_options(PgFdwRelationInfo *fpinfo)
{
    ListCell   *lc;
 
    foreach(lc, fpinfo->table->options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "use_remote_estimate") == 0)
            fpinfo->use_remote_estimate = defGetBoolean(def);
        else if (strcmp(def->defname, "fetch_size") == 0)
            (void) parse_int(defGetString(def), &fpinfo->fetch_size, 0, NULL);
        else if (strcmp(def->defname, "async_capable") == 0)
            fpinfo->async_capable = defGetBoolean(def);
    }
}
 
/*
 * Merge FDW options from input relations into a new set of options for a join
 * or an upper rel.
 *
 * For a join relation, FDW-specific information about the inner and outer
 * relations is provided using fpinfo_i and fpinfo_o.  For an upper relation,
 * fpinfo_o provides the information for the input relation; fpinfo_i is
 * expected to NULL.
 */
static void
merge_fdw_options(PgFdwRelationInfo *fpinfo,
                  const PgFdwRelationInfo *fpinfo_o,
                  const PgFdwRelationInfo *fpinfo_i)
{
    /* We must always have fpinfo_o. */
    Assert(fpinfo_o);
 
    /* fpinfo_i may be NULL, but if present the servers must both match. */
    Assert(!fpinfo_i ||
           fpinfo_i->server->serverid == fpinfo_o->server->serverid);
 
    /*
     * Copy the server specific FDW options.  (For a join, both relations come
     * from the same server, so the server options should have the same value
     * for both relations.)
     */
    fpinfo->fdw_startup_cost = fpinfo_o->fdw_startup_cost;
    fpinfo->fdw_tuple_cost = fpinfo_o->fdw_tuple_cost;
    fpinfo->shippable_extensions = fpinfo_o->shippable_extensions;
    fpinfo->use_remote_estimate = fpinfo_o->use_remote_estimate;
    fpinfo->fetch_size = fpinfo_o->fetch_size;
    fpinfo->async_capable = fpinfo_o->async_capable;
 
    /* Merge the table level options from either side of the join. */
    if (fpinfo_i)
    {
        /*
         * We'll prefer to use remote estimates for this join if any table
         * from either side of the join is using remote estimates.  This is
         * most likely going to be preferred since they're already willing to
         * pay the price of a round trip to get the remote EXPLAIN.  In any
         * case it's not entirely clear how we might otherwise handle this
         * best.
         */
        fpinfo->use_remote_estimate = fpinfo_o->use_remote_estimate ||
            fpinfo_i->use_remote_estimate;
 
        /*
         * Set fetch size to maximum of the joining sides, since we are
         * expecting the rows returned by the join to be proportional to the
         * relation sizes.
         */
        fpinfo->fetch_size = Max(fpinfo_o->fetch_size, fpinfo_i->fetch_size);
 
        /*
         * We'll prefer to consider this join async-capable if any table from
         * either side of the join is considered async-capable.  This would be
         * reasonable because in that case the foreign server would have its
         * own resources to scan that table asynchronously, and the join could
         * also be computed asynchronously using the resources.
         */
        fpinfo->async_capable = fpinfo_o->async_capable ||
            fpinfo_i->async_capable;
    }
}
 
/*
 * postgresGetForeignJoinPaths
 *      Add possible ForeignPath to joinrel, if join is safe to push down.
 */
static void
postgresGetForeignJoinPaths(PlannerInfo *root,
                            RelOptInfo *joinrel,
                            RelOptInfo *outerrel,
                            RelOptInfo *innerrel,
                            JoinType jointype,
                            JoinPathExtraData *extra)
{
    PgFdwRelationInfo *fpinfo;
    ForeignPath *joinpath;
    double      rows;
    int         width;
    int         disabled_nodes;
    Cost        startup_cost;
    Cost        total_cost;
    Path       *epq_path;       /* Path to create plan to be executed when
                                 * EvalPlanQual gets triggered. */
 
    /*
     * Skip if this join combination has been considered already.
     */
    if (joinrel->fdw_private)
        return;
 
    /*
     * This code does not work for joins with lateral references, since those
     * must have parameterized paths, which we don't generate yet.
     */
    if (!bms_is_empty(joinrel->lateral_relids))
        return;
 
    /*
     * Create unfinished PgFdwRelationInfo entry which is used to indicate
     * that the join relation is already considered, so that we won't waste
     * time in judging safety of join pushdown and adding the same paths again
     * if found safe. Once we know that this join can be pushed down, we fill
     * the entry.
     */
    fpinfo = palloc0_object(PgFdwRelationInfo);
    fpinfo->pushdown_safe = false;
    joinrel->fdw_private = fpinfo;
    /* attrs_used is only for base relations. */
    fpinfo->attrs_used = NULL;
 
    /*
     * If there is a possibility that EvalPlanQual will be executed, we need
     * to be able to reconstruct the row using scans of the base relations.
     * GetExistingLocalJoinPath will find a suitable path for this purpose in
     * the path list of the joinrel, if one exists.  We must be careful to
     * call it before adding any ForeignPath, since the ForeignPath might
     * dominate the only suitable local path available.  We also do it before
     * calling foreign_join_ok(), since that function updates fpinfo and marks
     * it as pushable if the join is found to be pushable.
     */
    if (root->parse->commandType == CMD_DELETE ||
        root->parse->commandType == CMD_UPDATE ||
        root->rowMarks)
    {
        epq_path = GetExistingLocalJoinPath(joinrel);
        if (!epq_path)
        {
            elog(DEBUG3, "could not push down foreign join because a local path suitable for EPQ checks was not found");
            return;
        }
    }
    else
        epq_path = NULL;
 
    if (!foreign_join_ok(root, joinrel, jointype, outerrel, innerrel, extra))
    {
        /* Free path required for EPQ if we copied one; we don't need it now */
        if (epq_path)
            pfree(epq_path);
        return;
    }
 
    /*
     * Compute the selectivity and cost of the local_conds, so we don't have
     * to do it over again for each path. The best we can do for these
     * conditions is to estimate selectivity on the basis of local statistics.
     * The local conditions are applied after the join has been computed on
     * the remote side like quals in WHERE clause, so pass jointype as
     * JOIN_INNER.
     */
    fpinfo->local_conds_sel = clauselist_selectivity(root,
                                                     fpinfo->local_conds,
                                                     0,
                                                     JOIN_INNER,
                                                     NULL);
    cost_qual_eval(&fpinfo->local_conds_cost, fpinfo->local_conds, root);
 
    /*
     * If we are going to estimate costs locally, estimate the join clause
     * selectivity here while we have special join info.
     */
    if (!fpinfo->use_remote_estimate)
        fpinfo->joinclause_sel = clauselist_selectivity(root, fpinfo->joinclauses,
                                                        0, fpinfo->jointype,
                                                        extra->sjinfo);
 
    /* Estimate costs for bare join relation */
    estimate_path_cost_size(root, joinrel, NIL, NIL, NULL,
                            &rows, &width, &disabled_nodes,
                            &startup_cost, &total_cost);
    /* Now update this information in the joinrel */
    joinrel->rows = rows;
    joinrel->reltarget->width = width;
    fpinfo->rows = rows;
    fpinfo->width = width;
    fpinfo->disabled_nodes = disabled_nodes;
    fpinfo->startup_cost = startup_cost;
    fpinfo->total_cost = total_cost;
 
    /*
     * Create a new join path and add it to the joinrel which represents a
     * join between foreign tables.
     */
    joinpath = create_foreign_join_path(root,
                                        joinrel,
                                        NULL,   /* default pathtarget */
                                        rows,
                                        disabled_nodes,
                                        startup_cost,
                                        total_cost,
                                        NIL,    /* no pathkeys */
                                        joinrel->lateral_relids,
                                        epq_path,
                                        extra->restrictlist,
                                        NIL);   /* no fdw_private */
 
    /* Add generated path into joinrel by add_path(). */
    add_path(joinrel, (Path *) joinpath);
 
    /* Consider pathkeys for the join relation */
    add_paths_with_pathkeys_for_rel(root, joinrel, epq_path,
                                    extra->restrictlist);
 
    /* XXX Consider parameterized paths for the join relation */
}
 
/*
 * Assess whether the aggregation, grouping and having operations can be pushed
 * down to the foreign server.  As a side effect, save information we obtain in
 * this function to PgFdwRelationInfo of the input relation.
 */
static bool
foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel,
                    Node *havingQual)
{
    Query      *query = root->parse;
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) grouped_rel->fdw_private;
    PathTarget *grouping_target = grouped_rel->reltarget;
    PgFdwRelationInfo *ofpinfo;
    ListCell   *lc;
    int         i;
    List       *tlist = NIL;
 
    /* We currently don't support pushing Grouping Sets. */
    if (query->groupingSets)
        return false;
 
    /* Get the fpinfo of the underlying scan relation. */
    ofpinfo = (PgFdwRelationInfo *) fpinfo->outerrel->fdw_private;
 
    /*
     * If underlying scan relation has any local conditions, those conditions
     * are required to be applied before performing aggregation.  Hence the
     * aggregate cannot be pushed down.
     */
    if (ofpinfo->local_conds)
        return false;
 
    /*
     * Examine grouping expressions, as well as other expressions we'd need to
     * compute, and check whether they are safe to push down to the foreign
     * server.  All GROUP BY expressions will be part of the grouping target
     * and thus there is no need to search for them separately.  Add grouping
     * expressions into target list which will be passed to foreign server.
     *
     * A tricky fine point is that we must not put any expression into the
     * target list that is just a foreign param (that is, something that
     * deparse.c would conclude has to be sent to the foreign server).  If we
     * do, the expression will also appear in the fdw_exprs list of the plan
     * node, and setrefs.c will get confused and decide that the fdw_exprs
     * entry is actually a reference to the fdw_scan_tlist entry, resulting in
     * a broken plan.  Somewhat oddly, it's OK if the expression contains such
     * a node, as long as it's not at top level; then no match is possible.
     */
    i = 0;
    foreach(lc, grouping_target->exprs)
    {
        Expr       *expr = (Expr *) lfirst(lc);
        Index       sgref = get_pathtarget_sortgroupref(grouping_target, i);
        ListCell   *l;
 
        /*
         * Check whether this expression is part of GROUP BY clause.  Note we
         * check the whole GROUP BY clause not just processed_groupClause,
         * because we will ship all of it, cf. appendGroupByClause.
         */
        if (sgref && get_sortgroupref_clause_noerr(sgref, query->groupClause))
        {
            TargetEntry *tle;
 
            /*
             * If any GROUP BY expression is not shippable, then we cannot
             * push down aggregation to the foreign server.
             */
            if (!is_foreign_expr(root, grouped_rel, expr))
                return false;
 
            /*
             * If it would be a foreign param, we can't put it into the tlist,
             * so we have to fail.
             */
            if (is_foreign_param(root, grouped_rel, expr))
                return false;
 
            /*
             * Pushable, so add to tlist.  We need to create a TLE for this
             * expression and apply the sortgroupref to it.  We cannot use
             * add_to_flat_tlist() here because that avoids making duplicate
             * entries in the tlist.  If there are duplicate entries with
             * distinct sortgrouprefs, we have to duplicate that situation in
             * the output tlist.
             */
            tle = makeTargetEntry(expr, list_length(tlist) + 1, NULL, false);
            tle->ressortgroupref = sgref;
            tlist = lappend(tlist, tle);
        }
        else
        {
            /*
             * Non-grouping expression we need to compute.  Can we ship it
             * as-is to the foreign server?
             */
            if (is_foreign_expr(root, grouped_rel, expr) &&
                !is_foreign_param(root, grouped_rel, expr))
            {
                /* Yes, so add to tlist as-is; OK to suppress duplicates */
                tlist = add_to_flat_tlist(tlist, list_make1(expr));
            }
            else
            {
                /* Not pushable as a whole; extract its Vars and aggregates */
                List       *aggvars;
 
                aggvars = pull_var_clause((Node *) expr,
                                          PVC_INCLUDE_AGGREGATES);
 
                /*
                 * If any aggregate expression is not shippable, then we
                 * cannot push down aggregation to the foreign server.  (We
                 * don't have to check is_foreign_param, since that certainly
                 * won't return true for any such expression.)
                 */
                if (!is_foreign_expr(root, grouped_rel, (Expr *) aggvars))
                    return false;
 
                /*
                 * Add aggregates, if any, into the targetlist.  Plain Vars
                 * outside an aggregate can be ignored, because they should be
                 * either same as some GROUP BY column or part of some GROUP
                 * BY expression.  In either case, they are already part of
                 * the targetlist and thus no need to add them again.  In fact
                 * including plain Vars in the tlist when they do not match a
                 * GROUP BY column would cause the foreign server to complain
                 * that the shipped query is invalid.
                 */
                foreach(l, aggvars)
                {
                    Expr       *aggref = (Expr *) lfirst(l);
 
                    if (IsA(aggref, Aggref))
                        tlist = add_to_flat_tlist(tlist, list_make1(aggref));
                }
            }
        }
 
        i++;
    }
 
    /*
     * Classify the pushable and non-pushable HAVING clauses and save them in
     * remote_conds and local_conds of the grouped rel's fpinfo.
     */
    if (havingQual)
    {
        foreach(lc, (List *) havingQual)
        {
            Expr       *expr = (Expr *) lfirst(lc);
            RestrictInfo *rinfo;
 
            /*
             * Currently, the core code doesn't wrap havingQuals in
             * RestrictInfos, so we must make our own.
             */
            Assert(!IsA(expr, RestrictInfo));
            rinfo = make_restrictinfo(root,
                                      expr,
                                      true,
                                      false,
                                      false,
                                      false,
                                      root->qual_security_level,
                                      grouped_rel->relids,
                                      NULL,
                                      NULL);
            if (is_foreign_expr(root, grouped_rel, expr))
                fpinfo->remote_conds = lappend(fpinfo->remote_conds, rinfo);
            else
                fpinfo->local_conds = lappend(fpinfo->local_conds, rinfo);
        }
    }
 
    /*
     * If there are any local conditions, pull Vars and aggregates from it and
     * check whether they are safe to pushdown or not.
     */
    if (fpinfo->local_conds)
    {
        List       *aggvars = NIL;
 
        foreach(lc, fpinfo->local_conds)
        {
            RestrictInfo *rinfo = lfirst_node(RestrictInfo, lc);
 
            aggvars = list_concat(aggvars,
                                  pull_var_clause((Node *) rinfo->clause,
                                                  PVC_INCLUDE_AGGREGATES));
        }
 
        foreach(lc, aggvars)
        {
            Expr       *expr = (Expr *) lfirst(lc);
 
            /*
             * If aggregates within local conditions are not safe to push
             * down, then we cannot push down the query.  Vars are already
             * part of GROUP BY clause which are checked above, so no need to
             * access them again here.  Again, we need not check
             * is_foreign_param for a foreign aggregate.
             */
            if (IsA(expr, Aggref))
            {
                if (!is_foreign_expr(root, grouped_rel, expr))
                    return false;
 
                tlist = add_to_flat_tlist(tlist, list_make1(expr));
            }
        }
    }
 
    /* Store generated targetlist */
    fpinfo->grouped_tlist = tlist;
 
    /* Safe to pushdown */
    fpinfo->pushdown_safe = true;
 
    /*
     * Set # of retrieved rows and cached relation costs to some negative
     * value, so that we can detect when they are set to some sensible values,
     * during one (usually the first) of the calls to estimate_path_cost_size.
     */
    fpinfo->retrieved_rows = -1;
    fpinfo->rel_startup_cost = -1;
    fpinfo->rel_total_cost = -1;
 
    /*
     * Set the string describing this grouped relation to be used in EXPLAIN
     * output of corresponding ForeignScan.  Note that the decoration we add
     * to the base relation name mustn't include any digits, or it'll confuse
     * postgresExplainForeignScan.
     */
    fpinfo->relation_name = psprintf("Aggregate on (%s)",
                                     ofpinfo->relation_name);
 
    return true;
}
 
/*
 * postgresGetForeignUpperPaths
 *      Add paths for post-join operations like aggregation, grouping etc. if
 *      corresponding operations are safe to push down.
 */
static void
postgresGetForeignUpperPaths(PlannerInfo *root, UpperRelationKind stage,
                             RelOptInfo *input_rel, RelOptInfo *output_rel,
                             void *extra)
{
    PgFdwRelationInfo *fpinfo;
 
    /*
     * If input rel is not safe to pushdown, then simply return as we cannot
     * perform any post-join operations on the foreign server.
     */
    if (!input_rel->fdw_private ||
        !((PgFdwRelationInfo *) input_rel->fdw_private)->pushdown_safe)
        return;
 
    /* Ignore stages we don't support; and skip any duplicate calls. */
    if ((stage != UPPERREL_GROUP_AGG &&
         stage != UPPERREL_ORDERED &&
         stage != UPPERREL_FINAL) ||
        output_rel->fdw_private)
        return;
 
    fpinfo = palloc0_object(PgFdwRelationInfo);
    fpinfo->pushdown_safe = false;
    fpinfo->stage = stage;
    output_rel->fdw_private = fpinfo;
 
    switch (stage)
    {
        case UPPERREL_GROUP_AGG:
            add_foreign_grouping_paths(root, input_rel, output_rel,
                                       (GroupPathExtraData *) extra);
            break;
        case UPPERREL_ORDERED:
            add_foreign_ordered_paths(root, input_rel, output_rel);
            break;
        case UPPERREL_FINAL:
            add_foreign_final_paths(root, input_rel, output_rel,
                                    (FinalPathExtraData *) extra);
            break;
        default:
            elog(ERROR, "unexpected upper relation: %d", (int) stage);
            break;
    }
}
 
/*
 * add_foreign_grouping_paths
 *      Add foreign path for grouping and/or aggregation.
 *
 * Given input_rel represents the underlying scan.  The paths are added to the
 * given grouped_rel.
 */
static void
add_foreign_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel,
                           RelOptInfo *grouped_rel,
                           GroupPathExtraData *extra)
{
    Query      *parse = root->parse;
    PgFdwRelationInfo *ifpinfo = input_rel->fdw_private;
    PgFdwRelationInfo *fpinfo = grouped_rel->fdw_private;
    ForeignPath *grouppath;
    double      rows;
    int         width;
    int         disabled_nodes;
    Cost        startup_cost;
    Cost        total_cost;
 
    /* Nothing to be done, if there is no grouping or aggregation required. */
    if (!parse->groupClause && !parse->groupingSets && !parse->hasAggs &&
        !root->hasHavingQual)
        return;
 
    Assert(extra->patype == PARTITIONWISE_AGGREGATE_NONE ||
           extra->patype == PARTITIONWISE_AGGREGATE_FULL);
 
    /* save the input_rel as outerrel in fpinfo */
    fpinfo->outerrel = input_rel;
 
    /*
     * Copy foreign table, foreign server, user mapping, FDW options etc.
     * details from the input relation's fpinfo.
     */
    fpinfo->table = ifpinfo->table;
    fpinfo->server = ifpinfo->server;
    fpinfo->user = ifpinfo->user;
    merge_fdw_options(fpinfo, ifpinfo, NULL);
 
    /*
     * Assess if it is safe to push down aggregation and grouping.
     *
     * Use HAVING qual from extra. In case of child partition, it will have
     * translated Vars.
     */
    if (!foreign_grouping_ok(root, grouped_rel, extra->havingQual))
        return;
 
    /*
     * Compute the selectivity and cost of the local_conds, so we don't have
     * to do it over again for each path.  (Currently we create just a single
     * path here, but in future it would be possible that we build more paths
     * such as pre-sorted paths as in postgresGetForeignPaths and
     * postgresGetForeignJoinPaths.)  The best we can do for these conditions
     * is to estimate selectivity on the basis of local statistics.
     */
    fpinfo->local_conds_sel = clauselist_selectivity(root,
                                                     fpinfo->local_conds,
                                                     0,
                                                     JOIN_INNER,
                                                     NULL);
 
    cost_qual_eval(&fpinfo->local_conds_cost, fpinfo->local_conds, root);
 
    /* Estimate the cost of push down */
    estimate_path_cost_size(root, grouped_rel, NIL, NIL, NULL,
                            &rows, &width, &disabled_nodes,
                            &startup_cost, &total_cost);
 
    /* Now update this information in the fpinfo */
    fpinfo->rows = rows;
    fpinfo->width = width;
    fpinfo->disabled_nodes = disabled_nodes;
    fpinfo->startup_cost = startup_cost;
    fpinfo->total_cost = total_cost;
 
    /* Create and add foreign path to the grouping relation. */
    grouppath = create_foreign_upper_path(root,
                                          grouped_rel,
                                          grouped_rel->reltarget,
                                          rows,
                                          disabled_nodes,
                                          startup_cost,
                                          total_cost,
                                          NIL,  /* no pathkeys */
                                          NULL,
                                          NIL,  /* no fdw_restrictinfo list */
                                          NIL); /* no fdw_private */
 
    /* Add generated path into grouped_rel by add_path(). */
    add_path(grouped_rel, (Path *) grouppath);
}
 
/*
 * add_foreign_ordered_paths
 *      Add foreign paths for performing the final sort remotely.
 *
 * Given input_rel contains the source-data Paths.  The paths are added to the
 * given ordered_rel.
 */
static void
add_foreign_ordered_paths(PlannerInfo *root, RelOptInfo *input_rel,
                          RelOptInfo *ordered_rel)
{
    Query      *parse = root->parse;
    PgFdwRelationInfo *ifpinfo = input_rel->fdw_private;
    PgFdwRelationInfo *fpinfo = ordered_rel->fdw_private;
    PgFdwPathExtraData *fpextra;
    double      rows;
    int         width;
    int         disabled_nodes;
    Cost        startup_cost;
    Cost        total_cost;
    List       *fdw_private;
    ForeignPath *ordered_path;
    ListCell   *lc;
 
    /* Shouldn't get here unless the query has ORDER BY */
    Assert(parse->sortClause);
 
    /* We don't support cases where there are any SRFs in the targetlist */
    if (parse->hasTargetSRFs)
        return;
 
    /* Save the input_rel as outerrel in fpinfo */
    fpinfo->outerrel = input_rel;
 
    /*
     * Copy foreign table, foreign server, user mapping, FDW options etc.
     * details from the input relation's fpinfo.
     */
    fpinfo->table = ifpinfo->table;
    fpinfo->server = ifpinfo->server;
    fpinfo->user = ifpinfo->user;
    merge_fdw_options(fpinfo, ifpinfo, NULL);
 
    /*
     * If the input_rel is a base or join relation, we would already have
     * considered pushing down the final sort to the remote server when
     * creating pre-sorted foreign paths for that relation, because the
     * query_pathkeys is set to the root->sort_pathkeys in that case (see
     * standard_qp_callback()).
     */
    if (input_rel->reloptkind == RELOPT_BASEREL ||
        input_rel->reloptkind == RELOPT_JOINREL)
    {
        Assert(root->query_pathkeys == root->sort_pathkeys);
 
        /* Safe to push down if the query_pathkeys is safe to push down */
        fpinfo->pushdown_safe = ifpinfo->qp_is_pushdown_safe;
 
        return;
    }
 
    /* The input_rel should be a grouping relation */
    Assert(input_rel->reloptkind == RELOPT_UPPER_REL &&
           ifpinfo->stage == UPPERREL_GROUP_AGG);
 
    /*
     * We try to create a path below by extending a simple foreign path for
     * the underlying grouping relation to perform the final sort remotely,
     * which is stored into the fdw_private list of the resulting path.
     */
 
    /* Assess if it is safe to push down the final sort */
    foreach(lc, root->sort_pathkeys)
    {
        PathKey    *pathkey = (PathKey *) lfirst(lc);
        EquivalenceClass *pathkey_ec = pathkey->pk_eclass;
 
        /*
         * is_foreign_expr would detect volatile expressions as well, but
         * checking ec_has_volatile here saves some cycles.
         */
        if (pathkey_ec->ec_has_volatile)
            return;
 
        /*
         * Can't push down the sort if pathkey's opfamily is not shippable.
         */
        if (!is_shippable(pathkey->pk_opfamily, OperatorFamilyRelationId,
                          fpinfo))
            return;
 
        /*
         * The EC must contain a shippable EM that is computed in input_rel's
         * reltarget, else we can't push down the sort.
         */
        if (find_em_for_rel_target(root,
                                   pathkey_ec,
                                   input_rel) == NULL)
            return;
    }
 
    /* Safe to push down */
    fpinfo->pushdown_safe = true;
 
    /* Construct PgFdwPathExtraData */
    fpextra = palloc0_object(PgFdwPathExtraData);
    fpextra->target = root->upper_targets[UPPERREL_ORDERED];
    fpextra->has_final_sort = true;
 
    /* Estimate the costs of performing the final sort remotely */
    estimate_path_cost_size(root, input_rel, NIL, root->sort_pathkeys, fpextra,
                            &rows, &width, &disabled_nodes,
                            &startup_cost, &total_cost);
 
    /*
     * Build the fdw_private list that will be used by postgresGetForeignPlan.
     * Items in the list must match order in enum FdwPathPrivateIndex.
     */
    fdw_private = list_make2(makeBoolean(true), makeBoolean(false));
 
    /* Create foreign ordering path */
    ordered_path = create_foreign_upper_path(root,
                                             input_rel,
                                             root->upper_targets[UPPERREL_ORDERED],
                                             rows,
                                             disabled_nodes,
                                             startup_cost,
                                             total_cost,
                                             root->sort_pathkeys,
                                             NULL,  /* no extra plan */
                                             NIL,   /* no fdw_restrictinfo
                                                     * list */
                                             fdw_private);
 
    /* and add it to the ordered_rel */
    add_path(ordered_rel, (Path *) ordered_path);
}
 
/*
 * add_foreign_final_paths
 *      Add foreign paths for performing the final processing remotely.
 *
 * Given input_rel contains the source-data Paths.  The paths are added to the
 * given final_rel.
 */
static void
add_foreign_final_paths(PlannerInfo *root, RelOptInfo *input_rel,
                        RelOptInfo *final_rel,
                        FinalPathExtraData *extra)
{
    Query      *parse = root->parse;
    PgFdwRelationInfo *ifpinfo = (PgFdwRelationInfo *) input_rel->fdw_private;
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) final_rel->fdw_private;
    bool        has_final_sort = false;
    List       *pathkeys = NIL;
    PgFdwPathExtraData *fpextra;
    bool        save_use_remote_estimate = false;
    double      rows;
    int         width;
    int         disabled_nodes;
    Cost        startup_cost;
    Cost        total_cost;
    List       *fdw_private;
    ForeignPath *final_path;
 
    /*
     * Currently, we only support this for SELECT commands
     */
    if (parse->commandType != CMD_SELECT)
        return;
 
    /*
     * No work if there is no FOR UPDATE/SHARE clause and if there is no need
     * to add a LIMIT node
     */
    if (!parse->rowMarks && !extra->limit_needed)
        return;
 
    /* We don't support cases where there are any SRFs in the targetlist */
    if (parse->hasTargetSRFs)
        return;
 
    /* Save the input_rel as outerrel in fpinfo */
    fpinfo->outerrel = input_rel;
 
    /*
     * Copy foreign table, foreign server, user mapping, FDW options etc.
     * details from the input relation's fpinfo.
     */
    fpinfo->table = ifpinfo->table;
    fpinfo->server = ifpinfo->server;
    fpinfo->user = ifpinfo->user;
    merge_fdw_options(fpinfo, ifpinfo, NULL);
 
    /*
     * If there is no need to add a LIMIT node, there might be a ForeignPath
     * in the input_rel's pathlist that implements all behavior of the query.
     * Note: we would already have accounted for the query's FOR UPDATE/SHARE
     * (if any) before we get here.
     */
    if (!extra->limit_needed)
    {
        ListCell   *lc;
 
        Assert(parse->rowMarks);
 
        /*
         * Grouping and aggregation are not supported with FOR UPDATE/SHARE,
         * so the input_rel should be a base, join, or ordered relation; and
         * if it's an ordered relation, its input relation should be a base or
         * join relation.
         */
        Assert(input_rel->reloptkind == RELOPT_BASEREL ||
               input_rel->reloptkind == RELOPT_JOINREL ||
               (input_rel->reloptkind == RELOPT_UPPER_REL &&
                ifpinfo->stage == UPPERREL_ORDERED &&
                (ifpinfo->outerrel->reloptkind == RELOPT_BASEREL ||
                 ifpinfo->outerrel->reloptkind == RELOPT_JOINREL)));
 
        foreach(lc, input_rel->pathlist)
        {
            Path       *path = (Path *) lfirst(lc);
 
            /*
             * apply_scanjoin_target_to_paths() uses create_projection_path()
             * to adjust each of its input paths if needed, whereas
             * create_ordered_paths() uses apply_projection_to_path() to do
             * that.  So the former might have put a ProjectionPath on top of
             * the ForeignPath; look through ProjectionPath and see if the
             * path underneath it is ForeignPath.
             */
            if (IsA(path, ForeignPath) ||
                (IsA(path, ProjectionPath) &&
                 IsA(((ProjectionPath *) path)->subpath, ForeignPath)))
            {
                /*
                 * Create foreign final path; this gets rid of a
                 * no-longer-needed outer plan (if any), which makes the
                 * EXPLAIN output look cleaner
                 */
                final_path = create_foreign_upper_path(root,
                                                       path->parent,
                                                       path->pathtarget,
                                                       path->rows,
                                                       path->disabled_nodes,
                                                       path->startup_cost,
                                                       path->total_cost,
                                                       path->pathkeys,
                                                       NULL,    /* no extra plan */
                                                       NIL, /* no fdw_restrictinfo
                                                             * list */
                                                       NIL);    /* no fdw_private */
 
                /* and add it to the final_rel */
                add_path(final_rel, (Path *) final_path);
 
                /* Safe to push down */
                fpinfo->pushdown_safe = true;
 
                return;
            }
        }
 
        /*
         * If we get here it means no ForeignPaths; since we would already
         * have considered pushing down all operations for the query to the
         * remote server, give up on it.
         */
        return;
    }
 
    Assert(extra->limit_needed);
 
    /*
     * If the input_rel is an ordered relation, replace the input_rel with its
     * input relation
     */
    if (input_rel->reloptkind == RELOPT_UPPER_REL &&
        ifpinfo->stage == UPPERREL_ORDERED)
    {
        input_rel = ifpinfo->outerrel;
        ifpinfo = (PgFdwRelationInfo *) input_rel->fdw_private;
        has_final_sort = true;
        pathkeys = root->sort_pathkeys;
    }
 
    /* The input_rel should be a base, join, or grouping relation */
    Assert(input_rel->reloptkind == RELOPT_BASEREL ||
           input_rel->reloptkind == RELOPT_JOINREL ||
           (input_rel->reloptkind == RELOPT_UPPER_REL &&
            ifpinfo->stage == UPPERREL_GROUP_AGG));
 
    /*
     * We try to create a path below by extending a simple foreign path for
     * the underlying base, join, or grouping relation to perform the final
     * sort (if has_final_sort) and the LIMIT restriction remotely, which is
     * stored into the fdw_private list of the resulting path.  (We
     * re-estimate the costs of sorting the underlying relation, if
     * has_final_sort.)
     */
 
    /*
     * Assess if it is safe to push down the LIMIT and OFFSET to the remote
     * server
     */
 
    /*
     * If the underlying relation has any local conditions, the LIMIT/OFFSET
     * cannot be pushed down.
     */
    if (ifpinfo->local_conds)
        return;
 
    /*
     * If the query has FETCH FIRST .. WITH TIES, 1) it must have ORDER BY as
     * well, which is used to determine which additional rows tie for the last
     * place in the result set, and 2) ORDER BY must already have been
     * determined to be safe to push down before we get here.  So in that case
     * the FETCH clause is safe to push down with ORDER BY if the remote
     * server is v13 or later, but if not, the remote query will fail entirely
     * for lack of support for it.  Since we do not currently have a way to do
     * a remote-version check (without accessing the remote server), disable
     * pushing the FETCH clause for now.
     */
    if (parse->limitOption == LIMIT_OPTION_WITH_TIES)
        return;
 
    /*
     * Also, the LIMIT/OFFSET cannot be pushed down, if their expressions are
     * not safe to remote.
     */
    if (!is_foreign_expr(root, input_rel, (Expr *) parse->limitOffset) ||
        !is_foreign_expr(root, input_rel, (Expr *) parse->limitCount))
        return;
 
    /* Safe to push down */
    fpinfo->pushdown_safe = true;
 
    /* Construct PgFdwPathExtraData */
    fpextra = palloc0_object(PgFdwPathExtraData);
    fpextra->target = root->upper_targets[UPPERREL_FINAL];
    fpextra->has_final_sort = has_final_sort;
    fpextra->has_limit = extra->limit_needed;
    fpextra->limit_tuples = extra->limit_tuples;
    fpextra->count_est = extra->count_est;
    fpextra->offset_est = extra->offset_est;
 
    /*
     * Estimate the costs of performing the final sort and the LIMIT
     * restriction remotely.  If has_final_sort is false, we wouldn't need to
     * execute EXPLAIN anymore if use_remote_estimate, since the costs can be
     * roughly estimated using the costs we already have for the underlying
     * relation, in the same way as when use_remote_estimate is false.  Since
     * it's pretty expensive to execute EXPLAIN, force use_remote_estimate to
     * false in that case.
     */
    if (!fpextra->has_final_sort)
    {
        save_use_remote_estimate = ifpinfo->use_remote_estimate;
        ifpinfo->use_remote_estimate = false;
    }
    estimate_path_cost_size(root, input_rel, NIL, pathkeys, fpextra,
                            &rows, &width, &disabled_nodes,
                            &startup_cost, &total_cost);
    if (!fpextra->has_final_sort)
        ifpinfo->use_remote_estimate = save_use_remote_estimate;
 
    /*
     * Build the fdw_private list that will be used by postgresGetForeignPlan.
     * Items in the list must match order in enum FdwPathPrivateIndex.
     */
    fdw_private = list_make2(makeBoolean(has_final_sort),
                             makeBoolean(extra->limit_needed));
 
    /*
     * Create foreign final path; this gets rid of a no-longer-needed outer
     * plan (if any), which makes the EXPLAIN output look cleaner
     */
    final_path = create_foreign_upper_path(root,
                                           input_rel,
                                           root->upper_targets[UPPERREL_FINAL],
                                           rows,
                                           disabled_nodes,
                                           startup_cost,
                                           total_cost,
                                           pathkeys,
                                           NULL,    /* no extra plan */
                                           NIL, /* no fdw_restrictinfo list */
                                           fdw_private);
 
    /* and add it to the final_rel */
    add_path(final_rel, (Path *) final_path);
}
 
/*
 * postgresIsForeignPathAsyncCapable
 *      Check whether a given ForeignPath node is async-capable.
 */
static bool
postgresIsForeignPathAsyncCapable(ForeignPath *path)
{
    RelOptInfo *rel = ((Path *) path)->parent;
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) rel->fdw_private;
 
    return fpinfo->async_capable;
}
 
/*
 * postgresForeignAsyncRequest
 *      Asynchronously request next tuple from a foreign PostgreSQL table.
 */
static void
postgresForeignAsyncRequest(AsyncRequest *areq)
{
    produce_tuple_asynchronously(areq, true);
}
 
/*
 * postgresForeignAsyncConfigureWait
 *      Configure a file descriptor event for which we wish to wait.
 */
static void
postgresForeignAsyncConfigureWait(AsyncRequest *areq)
{
    ForeignScanState *node = (ForeignScanState *) areq->requestee;
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
    AsyncRequest *pendingAreq = fsstate->conn_state->pendingAreq;
    AppendState *requestor = (AppendState *) areq->requestor;
    WaitEventSet *set = requestor->as_eventset;
 
    /* This should not be called unless callback_pending */
    Assert(areq->callback_pending);
 
    /*
     * If process_pending_request() has been invoked on the given request
     * before we get here, we might have some tuples already; in which case
     * complete the request
     */
    if (fsstate->next_tuple < fsstate->num_tuples)
    {
        complete_pending_request(areq);
        if (areq->request_complete)
            return;
        Assert(areq->callback_pending);
    }
 
    /* We must have run out of tuples */
    Assert(fsstate->next_tuple >= fsstate->num_tuples);
 
    /* The core code would have registered postmaster death event */
    Assert(GetNumRegisteredWaitEvents(set) >= 1);
 
    /* Begin an asynchronous data fetch if not already done */
    if (!pendingAreq)
        fetch_more_data_begin(areq);
    else if (pendingAreq->requestor != areq->requestor)
    {
        /*
         * This is the case when the in-process request was made by another
         * Append.  Note that it might be useless to process the request made
         * by that Append, because the query might not need tuples from that
         * Append anymore; so we avoid processing it to begin a fetch for the
         * given request if possible.  If there are any child subplans of the
         * same parent that are ready for new requests, skip the given
         * request.  Likewise, if there are any configured events other than
         * the postmaster death event, skip it.  Otherwise, process the
         * in-process request, then begin a fetch to configure the event
         * below, because we might otherwise end up with no configured events
         * other than the postmaster death event.
         */
        if (!bms_is_empty(requestor->as_needrequest))
            return;
        if (GetNumRegisteredWaitEvents(set) > 1)
            return;
        process_pending_request(pendingAreq);
        fetch_more_data_begin(areq);
    }
    else if (pendingAreq->requestee != areq->requestee)
    {
        /*
         * This is the case when the in-process request was made by the same
         * parent but for a different child.  Since we configure only the
         * event for the request made for that child, skip the given request.
         */
        return;
    }
    else
        Assert(pendingAreq == areq);
 
    AddWaitEventToSet(set, WL_SOCKET_READABLE, PQsocket(fsstate->conn),
                      NULL, areq);
}
 
/*
 * postgresForeignAsyncNotify
 *      Fetch some more tuples from a file descriptor that becomes ready,
 *      requesting next tuple.
 */
static void
postgresForeignAsyncNotify(AsyncRequest *areq)
{
    ForeignScanState *node = (ForeignScanState *) areq->requestee;
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
 
    /* The core code would have initialized the callback_pending flag */
    Assert(!areq->callback_pending);
 
    /*
     * If process_pending_request() has been invoked on the given request
     * before we get here, we might have some tuples already; in which case
     * produce the next tuple
     */
    if (fsstate->next_tuple < fsstate->num_tuples)
    {
        produce_tuple_asynchronously(areq, true);
        return;
    }
 
    /* We must have run out of tuples */
    Assert(fsstate->next_tuple >= fsstate->num_tuples);
 
    /* The request should be currently in-process */
    Assert(fsstate->conn_state->pendingAreq == areq);
 
    /* On error, report the original query, not the FETCH. */
    if (!PQconsumeInput(fsstate->conn))
        pgfdw_report_error(NULL, fsstate->conn, fsstate->query);
 
    fetch_more_data(node);
 
    produce_tuple_asynchronously(areq, true);
}
 
/*
 * Asynchronously produce next tuple from a foreign PostgreSQL table.
 */
static void
produce_tuple_asynchronously(AsyncRequest *areq, bool fetch)
{
    ForeignScanState *node = (ForeignScanState *) areq->requestee;
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
    AsyncRequest *pendingAreq = fsstate->conn_state->pendingAreq;
    TupleTableSlot *result;
 
    /* This should not be called if the request is currently in-process */
    Assert(areq != pendingAreq);
 
    /* Fetch some more tuples, if we've run out */
    if (fsstate->next_tuple >= fsstate->num_tuples)
    {
        /* No point in another fetch if we already detected EOF, though */
        if (!fsstate->eof_reached)
        {
            /* Mark the request as pending for a callback */
            ExecAsyncRequestPending(areq);
            /* Begin another fetch if requested and if no pending request */
            if (fetch && !pendingAreq)
                fetch_more_data_begin(areq);
        }
        else
        {
            /* There's nothing more to do; just return a NULL pointer */
            result = NULL;
            /* Mark the request as complete */
            ExecAsyncRequestDone(areq, result);
        }
        return;
    }
 
    /* Get a tuple from the ForeignScan node */
    result = areq->requestee->ExecProcNodeReal(areq->requestee);
    if (!TupIsNull(result))
    {
        /* Mark the request as complete */
        ExecAsyncRequestDone(areq, result);
        return;
    }
 
    /* We must have run out of tuples */
    Assert(fsstate->next_tuple >= fsstate->num_tuples);
 
    /* Fetch some more tuples, if we've not detected EOF yet */
    if (!fsstate->eof_reached)
    {
        /* Mark the request as pending for a callback */
        ExecAsyncRequestPending(areq);
        /* Begin another fetch if requested and if no pending request */
        if (fetch && !pendingAreq)
            fetch_more_data_begin(areq);
    }
    else
    {
        /* There's nothing more to do; just return a NULL pointer */
        result = NULL;
        /* Mark the request as complete */
        ExecAsyncRequestDone(areq, result);
    }
}
 
/*
 * Begin an asynchronous data fetch.
 *
 * Note: this function assumes there is no currently-in-progress asynchronous
 * data fetch.
 *
 * Note: fetch_more_data must be called to fetch the result.
 */
static void
fetch_more_data_begin(AsyncRequest *areq)
{
    ForeignScanState *node = (ForeignScanState *) areq->requestee;
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
    char        sql[64];
 
    Assert(!fsstate->conn_state->pendingAreq);
 
    /* Create the cursor synchronously. */
    if (!fsstate->cursor_exists)
        create_cursor(node);
 
    /* We will send this query, but not wait for the response. */
    snprintf(sql, sizeof(sql), "FETCH %d FROM c%u",
             fsstate->fetch_size, fsstate->cursor_number);
 
    if (!PQsendQuery(fsstate->conn, sql))
        pgfdw_report_error(NULL, fsstate->conn, fsstate->query);
 
    /* Remember that the request is in process */
    fsstate->conn_state->pendingAreq = areq;
}
 
/*
 * Process a pending asynchronous request.
 */
void
process_pending_request(AsyncRequest *areq)
{
    ForeignScanState *node = (ForeignScanState *) areq->requestee;
    PgFdwScanState *fsstate = (PgFdwScanState *) node->fdw_state;
 
    /* The request would have been pending for a callback */
    Assert(areq->callback_pending);
 
    /* The request should be currently in-process */
    Assert(fsstate->conn_state->pendingAreq == areq);
 
    fetch_more_data(node);
 
    /*
     * If we didn't get any tuples, must be end of data; complete the request
     * now.  Otherwise, we postpone completing the request until we are called
     * from postgresForeignAsyncConfigureWait()/postgresForeignAsyncNotify().
     */
    if (fsstate->next_tuple >= fsstate->num_tuples)
    {
        /* Unlike AsyncNotify, we unset callback_pending ourselves */
        areq->callback_pending = false;
        /* Mark the request as complete */
        ExecAsyncRequestDone(areq, NULL);
        /* Unlike AsyncNotify, we call ExecAsyncResponse ourselves */
        ExecAsyncResponse(areq);
    }
}
 
/*
 * Complete a pending asynchronous request.
 */
static void
complete_pending_request(AsyncRequest *areq)
{
    /* The request would have been pending for a callback */
    Assert(areq->callback_pending);
 
    /* Unlike AsyncNotify, we unset callback_pending ourselves */
    areq->callback_pending = false;
 
    /* We begin a fetch afterwards if necessary; don't fetch */
    produce_tuple_asynchronously(areq, false);
 
    /* Unlike AsyncNotify, we call ExecAsyncResponse ourselves */
    ExecAsyncResponse(areq);
 
    /* Also, we do instrumentation ourselves, if required */
    if (areq->requestee->instrument)
        InstrUpdateTupleCount(areq->requestee->instrument,
                              TupIsNull(areq->result) ? 0.0 : 1.0);
}
 
/*
 * Create a tuple from the specified row of the PGresult.
 *
 * rel is the local representation of the foreign table, attinmeta is
 * conversion data for the rel's tupdesc, and retrieved_attrs is an
 * integer list of the table column numbers present in the PGresult.
 * fsstate is the ForeignScan plan node's execution state.
 * temp_context is a working context that can be reset after each tuple.
 *
 * Note: either rel or fsstate, but not both, can be NULL.  rel is NULL
 * if we're processing a remote join, while fsstate is NULL in a non-query
 * context such as ANALYZE, or if we're processing a non-scan query node.
 */
static HeapTuple
make_tuple_from_result_row(PGresult *res,
                           int row,
                           Relation rel,
                           AttInMetadata *attinmeta,
                           List *retrieved_attrs,
                           ForeignScanState *fsstate,
                           MemoryContext temp_context)
{
    HeapTuple   tuple;
    TupleDesc   tupdesc;
    Datum      *values;
    bool       *nulls;
    ItemPointer ctid = NULL;
    ConversionLocation errpos;
    ErrorContextCallback errcallback;
    MemoryContext oldcontext;
    ListCell   *lc;
    int         j;
 
    Assert(row < PQntuples(res));
 
    /*
     * Do the following work in a temp context that we reset after each tuple.
     * This cleans up not only the data we have direct access to, but any
     * cruft the I/O functions might leak.
     */
    oldcontext = MemoryContextSwitchTo(temp_context);
 
    /*
     * Get the tuple descriptor for the row.  Use the rel's tupdesc if rel is
     * provided, otherwise look to the scan node's ScanTupleSlot.
     */
    if (rel)
        tupdesc = RelationGetDescr(rel);
    else
    {
        Assert(fsstate);
        tupdesc = fsstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
    }
 
    values = (Datum *) palloc0(tupdesc->natts * sizeof(Datum));
    nulls = (bool *) palloc(tupdesc->natts * sizeof(bool));
    /* Initialize to nulls for any columns not present in result */
    memset(nulls, true, tupdesc->natts * sizeof(bool));
 
    /*
     * Set up and install callback to report where conversion error occurs.
     */
    errpos.cur_attno = 0;
    errpos.rel = rel;
    errpos.fsstate = fsstate;
    errcallback.callback = conversion_error_callback;
    errcallback.arg = &errpos;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;
 
    /*
     * i indexes columns in the relation, j indexes columns in the PGresult.
     */
    j = 0;
    foreach(lc, retrieved_attrs)
    {
        int         i = lfirst_int(lc);
        char       *valstr;
 
        /* fetch next column's textual value */
        if (PQgetisnull(res, row, j))
            valstr = NULL;
        else
            valstr = PQgetvalue(res, row, j);
 
        /*
         * convert value to internal representation
         *
         * Note: we ignore system columns other than ctid and oid in result
         */
        errpos.cur_attno = i;
        if (i > 0)
        {
            /* ordinary column */
            Assert(i <= tupdesc->natts);
            nulls[i - 1] = (valstr == NULL);
            /* Apply the input function even to nulls, to support domains */
            values[i - 1] = InputFunctionCall(&attinmeta->attinfuncs[i - 1],
                                              valstr,
                                              attinmeta->attioparams[i - 1],
                                              attinmeta->atttypmods[i - 1]);
        }
        else if (i == SelfItemPointerAttributeNumber)
        {
            /* ctid */
            if (valstr != NULL)
            {
                Datum       datum;
 
                datum = DirectFunctionCall1(tidin, CStringGetDatum(valstr));
                ctid = (ItemPointer) DatumGetPointer(datum);
            }
        }
        errpos.cur_attno = 0;
 
        j++;
    }
 
    /* Uninstall error context callback. */
    error_context_stack = errcallback.previous;
 
    /*
     * Check we got the expected number of columns.  Note: j == 0 and
     * PQnfields == 1 is expected, since deparse emits a NULL if no columns.
     */
    if (j > 0 && j != PQnfields(res))
        elog(ERROR, "remote query result does not match the foreign table");
 
    /*
     * Build the result tuple in caller's memory context.
     */
    MemoryContextSwitchTo(oldcontext);
 
    tuple = heap_form_tuple(tupdesc, values, nulls);
 
    /*
     * If we have a CTID to return, install it in both t_self and t_ctid.
     * t_self is the normal place, but if the tuple is converted to a
     * composite Datum, t_self will be lost; setting t_ctid allows CTID to be
     * preserved during EvalPlanQual re-evaluations (see ROW_MARK_COPY code).
     */
    if (ctid)
        tuple->t_self = tuple->t_data->t_ctid = *ctid;
 
    /*
     * Stomp on the xmin, xmax, and cmin fields from the tuple created by
     * heap_form_tuple.  heap_form_tuple actually creates the tuple with
     * DatumTupleFields, not HeapTupleFields, but the executor expects
     * HeapTupleFields and will happily extract system columns on that
     * assumption.  If we don't do this then, for example, the tuple length
     * ends up in the xmin field, which isn't what we want.
     */
    HeapTupleHeaderSetXmax(tuple->t_data, InvalidTransactionId);
    HeapTupleHeaderSetXmin(tuple->t_data, InvalidTransactionId);
    HeapTupleHeaderSetCmin(tuple->t_data, InvalidTransactionId);
 
    /* Clean up */
    MemoryContextReset(temp_context);
 
    return tuple;
}
 
/*
 * Callback function which is called when error occurs during column value
 * conversion.  Print names of column and relation.
 *
 * Note that this function mustn't do any catalog lookups, since we are in
 * an already-failed transaction.  Fortunately, we can get the needed info
 * from the relation or the query's rangetable instead.
 */
static void
conversion_error_callback(void *arg)
{
    ConversionLocation *errpos = (ConversionLocation *) arg;
    Relation    rel = errpos->rel;
    ForeignScanState *fsstate = errpos->fsstate;
    const char *attname = NULL;
    const char *relname = NULL;
    bool        is_wholerow = false;
 
    /*
     * If we're in a scan node, always use aliases from the rangetable, for
     * consistency between the simple-relation and remote-join cases.  Look at
     * the relation's tupdesc only if we're not in a scan node.
     */
    if (fsstate)
    {
        /* ForeignScan case */
        ForeignScan *fsplan = castNode(ForeignScan, fsstate->ss.ps.plan);
        int         varno = 0;
        AttrNumber  colno = 0;
 
        if (fsplan->scan.scanrelid > 0)
        {
            /* error occurred in a scan against a foreign table */
            varno = fsplan->scan.scanrelid;
            colno = errpos->cur_attno;
        }
        else
        {
            /* error occurred in a scan against a foreign join */
            TargetEntry *tle;
 
            tle = list_nth_node(TargetEntry, fsplan->fdw_scan_tlist,
                                errpos->cur_attno - 1);
 
            /*
             * Target list can have Vars and expressions.  For Vars, we can
             * get some information, however for expressions we can't.  Thus
             * for expressions, just show generic context message.
             */
            if (IsA(tle->expr, Var))
            {
                Var        *var = (Var *) tle->expr;
 
                varno = var->varno;
                colno = var->varattno;
            }
        }
 
        if (varno > 0)
        {
            EState     *estate = fsstate->ss.ps.state;
            RangeTblEntry *rte = exec_rt_fetch(varno, estate);
 
            relname = rte->eref->aliasname;
 
            if (colno == 0)
                is_wholerow = true;
            else if (colno > 0 && colno <= list_length(rte->eref->colnames))
                attname = strVal(list_nth(rte->eref->colnames, colno - 1));
            else if (colno == SelfItemPointerAttributeNumber)
                attname = "ctid";
        }
    }
    else if (rel)
    {
        /* Non-ForeignScan case (we should always have a rel here) */
        TupleDesc   tupdesc = RelationGetDescr(rel);
 
        relname = RelationGetRelationName(rel);
        if (errpos->cur_attno > 0 && errpos->cur_attno <= tupdesc->natts)
        {
            Form_pg_attribute attr = TupleDescAttr(tupdesc,
                                                   errpos->cur_attno - 1);
 
            attname = NameStr(attr->attname);
        }
        else if (errpos->cur_attno == SelfItemPointerAttributeNumber)
            attname = "ctid";
    }
 
    if (relname && is_wholerow)
        errcontext("whole-row reference to foreign table \"%s\"", relname);
    else if (relname && attname)
        errcontext("column \"%s\" of foreign table \"%s\"", attname, relname);
    else
        errcontext("processing expression at position %d in select list",
                   errpos->cur_attno);
}
 
/*
 * Given an EquivalenceClass and a foreign relation, find an EC member
 * that can be used to sort the relation remotely according to a pathkey
 * using this EC.
 *
 * If there is more than one suitable candidate, return an arbitrary
 * one of them.  If there is none, return NULL.
 *
 * This checks that the EC member expression uses only Vars from the given
 * rel and is shippable.  Caller must separately verify that the pathkey's
 * ordering operator is shippable.
 */
EquivalenceMember *
find_em_for_rel(PlannerInfo *root, EquivalenceClass *ec, RelOptInfo *rel)
{
    PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) rel->fdw_private;
    EquivalenceMemberIterator it;
    EquivalenceMember *em;
 
    setup_eclass_member_iterator(&it, ec, rel->relids);
    while ((em = eclass_member_iterator_next(&it)) != NULL)
    {
        /*
         * Note we require !bms_is_empty, else we'd accept constant
         * expressions which are not suitable for the purpose.
         */
        if (bms_is_subset(em->em_relids, rel->relids) &&
            !bms_is_empty(em->em_relids) &&
            bms_is_empty(bms_intersect(em->em_relids, fpinfo->hidden_subquery_rels)) &&
            is_foreign_expr(root, rel, em->em_expr))
            return em;
    }
 
    return NULL;
}
 
/*
 * Find an EquivalenceClass member that is to be computed as a sort column
 * in the given rel's reltarget, and is shippable.
 *
 * If there is more than one suitable candidate, return an arbitrary
 * one of them.  If there is none, return NULL.
 *
 * This checks that the EC member expression uses only Vars from the given
 * rel and is shippable.  Caller must separately verify that the pathkey's
 * ordering operator is shippable.
 */
EquivalenceMember *
find_em_for_rel_target(PlannerInfo *root, EquivalenceClass *ec,
                       RelOptInfo *rel)
{
    PathTarget *target = rel->reltarget;
    ListCell   *lc1;
    int         i;
 
    i = 0;
    foreach(lc1, target->exprs)
    {
        Expr       *expr = (Expr *) lfirst(lc1);
        Index       sgref = get_pathtarget_sortgroupref(target, i);
        ListCell   *lc2;
 
        /* Ignore non-sort expressions */
        if (sgref == 0 ||
            get_sortgroupref_clause_noerr(sgref,
                                          root->parse->sortClause) == NULL)
        {
            i++;
            continue;
        }
 
        /* We ignore binary-compatible relabeling on both ends */
        while (expr && IsA(expr, RelabelType))
            expr = ((RelabelType *) expr)->arg;
 
        /*
         * Locate an EquivalenceClass member matching this expr, if any.
         * Ignore child members.
         */
        foreach(lc2, ec->ec_members)
        {
            EquivalenceMember *em = (EquivalenceMember *) lfirst(lc2);
            Expr       *em_expr;
 
            /* Don't match constants */
            if (em->em_is_const)
                continue;
 
            /* Child members should not exist in ec_members */
            Assert(!em->em_is_child);
 
            /* Match if same expression (after stripping relabel) */
            em_expr = em->em_expr;
            while (em_expr && IsA(em_expr, RelabelType))
                em_expr = ((RelabelType *) em_expr)->arg;
 
            if (!equal(em_expr, expr))
                continue;
 
            /* Check that expression (including relabels!) is shippable */
            if (is_foreign_expr(root, rel, em->em_expr))
                return em;
        }
 
        i++;
    }
 
    return NULL;
}
 
/*
 * Determine batch size for a given foreign table. The option specified for
 * a table has precedence.
 */
static int
get_batch_size_option(Relation rel)
{
    Oid         foreigntableid = RelationGetRelid(rel);
    ForeignTable *table;
    ForeignServer *server;
    List       *options;
    ListCell   *lc;
 
    /* we use 1 by default, which means "no batching" */
    int         batch_size = 1;
 
    /*
     * Load options for table and server. We append server options after table
     * options, because table options take precedence.
     */
    table = GetForeignTable(foreigntableid);
    server = GetForeignServer(table->serverid);
 
    options = NIL;
    options = list_concat(options, table->options);
    options = list_concat(options, server->options);
 
    /* See if either table or server specifies batch_size. */
    foreach(lc, options)
    {
        DefElem    *def = (DefElem *) lfirst(lc);
 
        if (strcmp(def->defname, "batch_size") == 0)
        {
            (void) parse_int(defGetString(def), &batch_size, 0, NULL);
            break;
        }
    }
 
    return batch_size;
}