contrib/dblink/postgresql_fdw.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * postgresql_fdw.c
   4  *                foreign-data wrapper for PostgreSQL
   5  *
   6  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
   7  *
   8  * IDENTIFICATION
   9  *                $PostgreSQL$
  10  *
  11  *-------------------------------------------------------------------------
  12  */
  13 #include "postgres.h"
  14
  15 #include "catalog/pg_operator.h"
  16 #include "catalog/pg_proc.h"
  17 #include "funcapi.h"
  18 #include "libpq-fe.h"
  19 #include "mb/pg_wchar.h"
  20 #include "miscadmin.h"
  21 #include "nodes/nodeFuncs.h"
  22 #include "nodes/makefuncs.h"
  23 #include "optimizer/clauses.h"
  24 #include "parser/scansup.h"
  25 #include "utils/builtins.h"
  26 #include "utils/lsyscache.h"
  27 #include "utils/memutils.h"
  28 #include "utils/syscache.h"
  29
  30 #include "dblink.h"
  31
  32 #define DEBUG_FDW
  33
  34 /*
  35  * WHERE caluse optimization level
  36  */
  37 #define EVAL_QUAL_LOCAL         0       /* evaluate none in foreign, all in local */
  38 #define EVAL_QUAL_BOTH          1       /* evaluate some in foreign, all in local */
  39 #define EVAL_QUAL_FOREIGN       2       /* evaluate some in foreign, rest in local */
  40
  41 #define OPTIMIZE_WHERE_CLAUSE   EVAL_QUAL_FOREIGN
  42
  43 extern Datum postgresql_fdw_handler(PG_FUNCTION_ARGS);
  44
  45 static FSConnection* pgConnectServer(ForeignServer *server, UserMapping *user);
  46 static void pgFreeFSConnection(FSConnection *conn);
  47 static void pgOpen(ForeignScanState *scanstate);
  48 static void pgIterate(ForeignScanState *scanstate);
  49 static void pgClose(ForeignScanState *scanstate);
  50 static void pgReOpen(ForeignScanState *scanstate);
  51
  52 /* deparse SQL from the request */
  53 static bool is_immutable_func(Oid funcid);
  54 static bool is_foreign_qual(ExprState *state);
  55 static bool foreign_qual_walker(Node *node, void *context);
  56 static char *deparseSql(ForeignScanState *scanstate);
  57 static int flatten_deflist(List *options,
  58                                                    const char **keywords, const char **values);
  59 static void check_conn_params(const char **keywords, const char **values);
  60 static remoteConn *createRemoteConn(PGconn *conn);
  61 static void dblink_connstr_check(const char *connstr);
  62
  63 /*
  64  * Concrete cursor for a foreign table on a PostgreSQL
  65  */
  66 typedef struct pgFdwReply
  67 {
  68         char                       *sql;                /* SQL text sent to foreign server */
  69         Tuplestorestate    *tupstore;   /* result set */
  70 } pgFdwReply;
  71
  72 FdwRoutine postgresql_fdw_routine =
  73 {
  74         pgConnectServer,
  75         pgFreeFSConnection,
  76         pgOpen,
  77         pgIterate,
  78         pgClose,
  79         pgReOpen
  80 };
  81
  82 /*
  83  * return foreign-data wrapper handler object to execute foreign-data wrapper
  84  * routines.
  85  */
  86 PG_FUNCTION_INFO_V1(postgresql_fdw_handler);
  87 Datum
  88 postgresql_fdw_handler(PG_FUNCTION_ARGS)
  89 {
  90         PG_RETURN_POINTER(&postgresql_fdw_routine);
  91 }
  92
  93 /*
  94  * Connect to foreign PostgreSQL server with libpq.
  95  */
  96 static FSConnection *
  97 pgConnectServer(ForeignServer *server, UserMapping *user)
  98 {
  99         PGconn             *conn;
 100         const char        **keywords;
 101         const char        **values;
 102         int                             n;
 103
 104 elog(DEBUG2, "%s(%s) called", PG_FUNCNAME_MACRO, server->servername);
 105
 106         /*
 107          * construct connection params from options of ForeignDataWrapper,
 108          * ForeignServer and UserMapping. Assuming all GENERIC OPTIONS are
 109          * conneciton information.
 110          *
 111          * TODO: allow non-connection options and ignore them during constructing
 112          * connection string.
 113          */
 114         n = list_length(server->options) + list_length(user->options) + 1;
 115         keywords = (const char **) palloc(sizeof(char *) * n);
 116         values = (const char **) palloc(sizeof(char *) * n);
 117         n = 0;
 118         n += flatten_deflist(server->options, keywords + n, values + n);
 119         n += flatten_deflist(user->options, keywords + n, values + n);
 120         keywords[n] = values[n] = NULL;
 121
 122         /* verify connection parameters and do connect */
 123         check_conn_params(keywords, values);
 124         conn = PQconnectdbParams(keywords, values, 0);
 125         pfree(keywords);
 126         pfree(values);
 127
 128         return (FSConnection *) createRemoteConn(conn);
 129 }
 130
 131 /*
 132  * Disconnect from the foreign server and free memory.
 133  * The cache entry of the hash table will be remove by executor.
 134  */
 135 static void
 136 pgFreeFSConnection(FSConnection *conn)
 137 {
 138         Assert(conn != NULL);
 139
 140 elog(DEBUG2, "%s() called", PG_FUNCNAME_MACRO);
 141
 142         PQfinish(((remoteConn *) conn)->conn);
 143 }
 144
 145 /*
 146  * Check whether the function is IMMUTABLE.
 147  */
 148 static bool
 149 is_immutable_func(Oid funcid)
 150 {
 151         HeapTuple               tp;
 152         bool                    isnull;
 153         Datum                   datum;
 154
 155         tp = SearchSysCache1(PROCOID, funcid);
 156         if (!HeapTupleIsValid(tp))
 157                 elog(ERROR, "cache lookup failed for function %u", funcid);
 158
 159 #ifdef DEBUG_FDW
 160         /* print function name and its immutability */
 161         {
 162                 char               *proname;
 163                 datum = SysCacheGetAttr(PROCOID, tp, Anum_pg_proc_proname, &isnull);
 164                 proname = pstrdup(DatumGetName(datum)->data);
 165                 elog(DEBUG1, "func %s(%u) is%s immutable", proname, funcid,
 166                         (DatumGetChar(datum) == PROVOLATILE_IMMUTABLE) ? "" : " not");
 167                 pfree(proname);
 168         }
 169 #endif
 170
 171         datum = SysCacheGetAttr(PROCOID, tp, Anum_pg_proc_provolatile, &isnull);
 172         ReleaseSysCache(tp);
 173
 174         return (DatumGetChar(datum) == PROVOLATILE_IMMUTABLE);
 175 }
 176
 177 /*
 178  * Check whether the ExprState node should be evaluated in foreign server.
 179  *
 180  * An expression which consists of expressions below will be evaluated in
 181  * the foreign server.
 182  *  - constant value
 183  *  - variable (foreign table column)
 184  *  - external parameter (parameter of prepared statement)
 185  *  - array
 186  *  - bool expression (AND/OR/NOT)
 187  *  - NULL test (IS [NOT] NULL)
 188  *  - operator
 189  *    - IMMUTABLE only
 190  *    - It is required that the meaning of the operator be the same as the
 191  *      local server in the foreign server.
 192  *  - function
 193  *    - IMMUTABLE only
 194  *    - It is required that the meaning of the operator be the same as the
 195  *      local server in the foreign server.
 196  *  - scalar array operator (ANY/ALL)
 197  */
 198 static bool
 199 is_foreign_qual(ExprState *state)
 200 {
 201         return !foreign_qual_walker((Node *) state->expr, NULL);
 202 }
 203
 204 /*
 205  * return true if node cannot be evaluatated in foreign server.
 206  */
 207 static bool
 208 foreign_qual_walker(Node *node, void *context)
 209 {
 210         if (node == NULL)
 211                 return false;
 212
 213         switch (nodeTag(node))
 214         {
 215                 case T_Param:
 216                         /* TODO: pass internal parameters to the foreign server */
 217                         if (((Param *) node)->paramkind != PARAM_EXTERN)
 218                                 return true;
 219                         break;
 220                 case T_DistinctExpr:
 221                 case T_OpExpr:
 222                         /*
 223                          * An operator which uses IMMUTABLE function can be evaluated in
 224                          * foreign server . It is not necessary to worry about oprrest
 225                          * and oprjoin here because they are invoked by planner but not
 226                          * executor. DistinctExpr is a typedef of OpExpr.
 227                          */
 228                         if (!is_immutable_func(((OpExpr*) node)->opfuncid))
 229                                 return true;
 230                         break;
 231                 case T_ScalarArrayOpExpr:
 232                         if (!is_immutable_func(((ScalarArrayOpExpr*) node)->opfuncid))
 233                                 return true;
 234                         break;
 235                 case T_FuncExpr:
 236                         /* IMMUTABLE function can be evaluated in foreign server */
 237                         if (!is_immutable_func(((FuncExpr*) node)->funcid))
 238                                 return true;
 239                         break;
 240                 case T_TargetEntry:
 241                 case T_PlaceHolderVar:
 242                 case T_AppendRelInfo:
 243                 case T_PlaceHolderInfo:
 244                         /* TODO: research whether those complex nodes are evaluatable. */
 245                         return true;
 246                 default:
 247                         break;
 248         }
 249
 250         return expression_tree_walker(node, foreign_qual_walker, context);
 251 }
 252
 253 /*
 254  * Deparse SQL string from query request.
 255  *
 256  * The expressions in Plan.qual are deparsed when it satisfies is_foreign_qual()
 257  * and removed.
 258  */
 259 static char *
 260 deparseSql(ForeignScanState *scanstate)
 261 {
 262         EState             *estate = scanstate->ss.ps.state;
 263         bool                    prefix;
 264         List               *context;
 265         StringInfoData  sql;
 266         ForeignScan        *scan;
 267         RangeTblEntry  *rte;
 268         ForeignTable   *table;
 269         char               *nspname = NULL;
 270         char               *relname = NULL;
 271         const char         *nspname_q;
 272         const char         *relname_q;
 273         const char         *aliasname_q;
 274         ListCell           *lc;
 275         int                             i;
 276         TupleDesc               tupdesc;
 277         bool                    first;
 278
 279 elog(DEBUG2, "%s(%u) called", __FUNCTION__, __LINE__);
 280
 281         /* extract ForeignScan and RangeTblEntry */
 282         scan = (ForeignScan *)scanstate->ss.ps.plan;
 283         rte = list_nth(estate->es_range_table, scan->scan.scanrelid - 1);
 284
 285         /* prepare to deparse plan */
 286         initStringInfo(&sql);
 287         context = deparse_context_for_planstate((Node *)&scanstate->ss.ps, NULL,
 288                                                                            estate->es_range_table);
 289
 290         /*
 291          * Scanning multiple relations in a ForeignScan node is not supported.
 292          */
 293         prefix = false;
 294 #if 0
 295         prefix = list_length(estate->es_range_table) > 1;
 296 #endif
 297
 298         /* The alias of relation is used in both SELECT clause and FROM clause. */
 299         aliasname_q = quote_identifier(rte->eref->aliasname);
 300
 301         /* deparse SELECT clause */
 302         appendStringInfo(&sql, "SELECT ");
 303
 304         /*
 305          * TODO: omit (deparse to "NULL") columns which are not used in the
 306          * original SQL.
 307          *
 308          * We must parse nodes parents of this ForeignScan node to determine unused
 309          * columns because some columns may be used only in parent Sort/Agg/Limit
 310          * nodes.
 311          */
 312         tupdesc = scanstate->ss.ss_currentRelation->rd_att;
 313         first = true;
 314         for (i = 0; i < tupdesc->natts; i++)
 315         {
 316                 /* skip dropped attributes */
 317                 if (tupdesc->attrs[i]->attisdropped)
 318                         continue;
 319
 320                 if (!first)
 321                         appendStringInfoString(&sql, ", ");
 322
 323                 if (prefix)
 324                         appendStringInfo(&sql, "%s.%s",
 325                                                         aliasname_q, tupdesc->attrs[i]->attname.data);
 326                 else
 327                         appendStringInfo(&sql, "%s", tupdesc->attrs[i]->attname.data);
 328                 first = false;
 329         }
 330
 331         /* if target list is composed only of system attributes, add dummy column */
 332         if (first)
 333                 appendStringInfo(&sql, "NULL");
 334
 335         /* deparse FROM clause */
 336         appendStringInfo(&sql, " FROM ");
 337
 338         /*
 339          * If the foreign table has generic option "nspname" and/or "relname", use
 340          * them in the foreign query.  Otherwise, use local catalog names.
 341          * Each identifier must be quoted if they are case sensitive.
 342          */
 343         table = GetForeignTable(rte->relid);
 344         foreach(lc, table->options)
 345         {
 346                 DefElem *opt = lfirst(lc);
 347                 if (strcmp(opt->defname, "nspname") == 0)
 348                         nspname = pstrdup(strVal(opt->arg));
 349                 else if (strcmp(opt->defname, "relname") == 0)
 350                         relname = pstrdup(strVal(opt->arg));
 351         }
 352         if (nspname == NULL)
 353                 nspname = get_namespace_name(get_rel_namespace(rte->relid));
 354         if (relname == NULL)
 355                 relname = get_rel_name(rte->relid);
 356         nspname_q = quote_identifier(nspname);
 357         relname_q = quote_identifier(relname);
 358         appendStringInfo(&sql, "%s.%s %s", nspname_q, relname_q, aliasname_q);
 359         pfree(nspname);
 360         pfree(relname);
 361         if (nspname_q != nspname)
 362                 pfree((char *) nspname_q);
 363         if (relname_q != relname)
 364                 pfree((char * ) relname_q);
 365         if (aliasname_q != rte->eref->aliasname)
 366                 pfree((char *) aliasname_q);
 367
 368         /*
 369          * deparse WHERE cluase
 370          *
 371          * The expressions which satisfy is_foreign_qual() are deparsed into WHERE
 372          * clause of result SQL string, and they could be removed from qual of
 373          * PlanState to avoid duplicate evaluation at ExecScan().
 374          *
 375          * The Plan.qual is never changed, so multiple use of the Plan with
 376          * PREPARE/EXECUTE work properly.
 377          */
 378 #if OPTIMIZE_WHERE_CLAUSE > EVAL_QUAL_LOCAL
 379         if (scanstate->ss.ps.plan->qual)
 380         {
 381                 List       *local_qual = NIL;
 382                 List       *foreign_qual = NIL;
 383                 List       *foreign_expr = NIL;
 384                 ListCell   *lc;
 385
 386                 /*
 387                  * Divide qual of PlanState into two lists, one for local evaluation
 388                  * and one for foreign evaluation.
 389                  */
 390                 foreach (lc, scanstate->ss.ps.qual)
 391                 {
 392                         ExprState          *state = lfirst(lc);
 393
 394                         if (is_foreign_qual(state))
 395                         {
 396                                 elog(DEBUG1, "foreign qual: %s", nodeToString(state->expr));
 397                                 foreign_qual = lappend(foreign_qual, state);
 398                                 foreign_expr = lappend(foreign_expr, state->expr);
 399                         }
 400                         else
 401                         {
 402                                 elog(DEBUG1, "local qual: %s", nodeToString(state->expr));
 403                                 local_qual = lappend(local_qual, state);
 404                         }
 405                 }
 406 #if OPTIMIZE_WHERE_CLAUSE == EVAL_QUAL_FOREIGN
 407                 /*
 408                  * If the optimization level is EVAL_QUAL_FOREIGN, replace the original
 409                  * qual with the list of ExprStates which should be evaluated in the
 410                  * local server.
 411                  */
 412                 scanstate->ss.ps.qual = local_qual;
 413 #endif
 414
 415                 /*
 416                  * Deparse quals to be evaluated in the foreign server if any.
 417                  * TODO: modify deparse_expression() to deparse conditions which use
 418                  * internal parameters.
 419                  */
 420                 if (foreign_expr != NIL)
 421                 {
 422                         Node   *node;
 423                         node = (Node *) make_ands_explicit(foreign_expr);
 424                         appendStringInfo(&sql, " WHERE ");
 425                         appendStringInfo(&sql,
 426                                 deparse_expression(node, context, prefix, false));
 427                         /*
 428                          * The contents of the list MUST NOT be free-ed because they are
 429                          * referenced from Plan.qual list.
 430                          */
 431                         list_free(foreign_expr);
 432                 }
 433         }
 434 #endif
 435
 436         elog(DEBUG1, "deparsed SQL is \"%s\"", sql.data);
 437
 438         return sql.data;
 439 }
 440
 441 /*
 442  * pgOpen()
 443  *   - deparse SQL statement from ForeignScanState and EState
 444  */
 445 static void
 446 pgOpen(ForeignScanState *scanstate)
 447 {
 448         pgFdwReply         *reply;
 449
 450 elog(DEBUG2, "%s(%u) called", __FUNCTION__, __LINE__);
 451
 452         /* FWD-specific portion */
 453         reply = (pgFdwReply *) palloc0(sizeof(*reply));
 454         reply->sql = deparseSql(scanstate);
 455         scanstate->reply = (FdwReply *) reply;
 456 }
 457
 458 /*
 459  * return tuples one by one.
 460  *   - execute SQL statement which was deparsed in pgOpen()
 461  *
 462  * The all of result are fetched at once when pgIterate() is called first after
 463  * pgOpen() or pgReOpen().
 464  * pgIterate() takes out a tuple from tupstore with tupslot and returns it.
 465  */
 466 static void
 467 pgIterate(ForeignScanState *scanstate)
 468 {
 469         pgFdwReply         *reply = (pgFdwReply *) scanstate->reply;
 470         TupleTableSlot *slot = scanstate->ss.ss_ScanTupleSlot;
 471
 472 elog(DEBUG2, "%s(%u) called", __FUNCTION__, __LINE__);
 473
 474         /*
 475          * Execute query with current parameters.
 476          */
 477         if (reply->tupstore == NULL)
 478         {
 479                 PGconn             *conn = ((remoteConn *) scanstate->conn)->conn;
 480                 PGresult           *res;
 481                 ParamListInfo   info = scanstate->ss.ps.state->es_param_list_info;
 482                 int                             numParams = info ? info->numParams : 0;
 483                 Oid                        *types = NULL;
 484                 const char        **values = NULL;
 485
 486                 /* construct parameter array in text format */
 487                 /* TODO: omit unused parameter */
 488                 if (numParams > 0)
 489                 {
 490                         int                     i;
 491
 492                         types = palloc0(sizeof(Oid) * numParams);
 493                         values = palloc0(sizeof(char *) * numParams);
 494                         for (i = 0; i < numParams; i++)
 495                         {
 496                                 types[i] = info->params[i].ptype;
 497                                 if (info->params[i].isnull)
 498                                         values[i] = NULL;
 499                                 else
 500                                 {
 501                                         Oid                     out_func_oid;
 502                                         bool            isvarlena;
 503                                         FmgrInfo        func;
 504
 505                                         /* TODO: cache FmgrInfo to use it again after pgReOpen() */
 506                                         /* TODO: send parameters in binary format rather than text */
 507                                         getTypeOutputInfo(types[i], &out_func_oid, &isvarlena);
 508                                         fmgr_info(out_func_oid, &func);
 509                                         values[i] =
 510                                                 OutputFunctionCall(&func, info->params[i].value);
 511                                 }
 512                         }
 513                 }
 514
 515                 /*
 516                  * Execute query with the parameters.
 517                  * TODO: support internal parameters(PARAM_EXTERN)
 518                  * TODO: support cursor mode for huge result sets.
 519                  */
 520                 res = PQexecParams(conn, reply->sql,
 521                                                                 numParams, types, values, NULL, NULL, 0);
 522                 if (numParams > 0)
 523                 {
 524                         int i;
 525                         pfree(types);
 526                         for (i = 0; i < numParams; i++)
 527                                 pfree((char *) values[i]);
 528                         pfree(values);
 529                 }
 530                 if (!res || PQresultStatus(res) != PGRES_TUPLES_OK)
 531                 {
 532                         PQclear(res);
 533                         ereport(ERROR, (
 534                                         errmsg("could not execute foreign query"),
 535                                         errdetail("%s", PQerrorMessage(conn)),
 536                                         errhint("%s", reply->sql)));
 537                 }
 538
 539                 /* Note: use PG_TRY to ensure freeing PGresult. */
 540                 PG_TRY();
 541                 {
 542                         TupleDesc       tupdesc = ExecGetScanType((ScanState *) scanstate);
 543
 544                         /* create tuplestore to store results */
 545                         reply->tupstore = tuplestore_begin_heap(true, false, work_mem);
 546
 547                         storeResult(reply->tupstore, false, tupdesc, res);
 548
 549                         PQclear(res);
 550                 }
 551                 PG_CATCH();
 552                 {
 553                         PQclear(res);
 554                         PG_RE_THROW();
 555                 }
 556                 PG_END_TRY();
 557         }
 558
 559         /* store the next tuple into the slot from the tuplestore */
 560         if (tuplestore_gettupleslot(reply->tupstore, true, false, slot))
 561         {
 562                 /*
 563                  * Because the tuples stored in the tupstore are minimal tuples,
 564                  * they have to be materialized to retrieve system attributes.
 565                  */
 566                 ExecMaterializeSlot(slot);
 567         }
 568         else
 569         {
 570                 /* TODO: if cursor mode, reset tuple slot and fetch the next batch. */
 571         }
 572 }
 573
 574 /*
 575  * Finish scanning foreign table and dispose objects used for this scan.
 576  */
 577 static void
 578 pgClose(ForeignScanState *scanstate)
 579 {
 580         pgFdwReply         *reply = (pgFdwReply *) scanstate->reply;
 581
 582 elog(DEBUG2, "%s(%u) called", __FUNCTION__, __LINE__);
 583
 584         if (reply == NULL)
 585                 return;
 586
 587         if (reply->tupstore != NULL)
 588                 tuplestore_end(reply->tupstore);
 589
 590         /*
 591          * reply->conn is not free-ed here because foreign connections are
 592          * managed by executor, not FDW.
 593          */
 594         pfree(reply);
 595         scanstate->reply = NULL;
 596 }
 597
 598 /*
 599  * Execute query with new parameter.
 600  */
 601 static void
 602 pgReOpen(ForeignScanState *scanstate)
 603 {
 604         pgFdwReply         *reply = (pgFdwReply *) scanstate->reply;
 605
 606 elog(DEBUG2, "%s(%u) called", __FUNCTION__, __LINE__);
 607
 608         /* Free Tuplestore to execute query again */
 609         /* TODO: reuse tupstore through the scan to avoid overhead */
 610         if (reply->tupstore)
 611         {
 612                 tuplestore_end(reply->tupstore);
 613                 reply->tupstore = NULL;
 614                 elog(DEBUG1, "tuplestore disposed");
 615         }
 616 }
 617
 618 /*
 619  * Flattern options into keywords and values buffers.
 620  */
 621 static int
 622 flatten_deflist(List *options, const char **keywords, const char **values)
 623 {
 624         ListCell   *cell;
 625         int                     n = 0;
 626
 627         foreach(cell, options)
 628         {
 629                 DefElem    *def = lfirst(cell);
 630
 631                 keywords[n] = def->defname;
 632                 values[n] = strVal(def->arg);
 633                 n++;
 634         }
 635         return n;
 636 }
 637
 638 /*
 639  * For non-superusers, insist that the connstr specify a password.      This
 640  * prevents a password from being picked up from .pgpass, a service file,
 641  * the environment, etc.  We don't want the postgres user's passwords
 642  * to be accessible to non-superusers.
 643  */
 644 static void
 645 check_conn_params(const char **keywords, const char **values)
 646 {
 647         int                     i;
 648
 649         /* no check required if superuser */
 650         if (superuser())
 651                 return;
 652
 653         /* ok if params contain a non-empty password */
 654         for (i = 0; keywords[i] != NULL; i++)
 655         {
 656                 if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0')
 657                         return;
 658         }
 659
 660         ereport(ERROR,
 661                   (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
 662                    errmsg("password is required"),
 663                    errdetail("Non-superusers must provide a password in the connection string.")));
 664 }
 665
 666 /*
 667  * Returns the connection which has name `conname' and is against PostgreSQL
 668  * server from connection cache.
 669  * If the connection which has the name was not found, returns NULL.
 670  * If the connection which has the name was found but the connection is not
 671  * against PostgreSQL, quit with error.
 672  */
 673 remoteConn *
 674 getConnectionByName(const char *conname)
 675 {
 676         FSConnection       *conn;
 677         FdwRoutine                 *routine;
 678         char                       *name;
 679
 680         if (conname)
 681         {
 682                 name = pstrdup(conname);
 683                 truncate_identifier(name, strlen(name), true);
 684         }
 685         else
 686                 name = UNNAMED_CONN_NAME;
 687
 688         conn = GetFSConnectionByName(name, &routine);
 689         if (conn == NULL)
 690                 return NULL;
 691
 692         if (routine != &postgresql_fdw_routine)
 693                 ereport(ERROR,
 694                         (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 695                          errmsg("connection \"%s\" is not a postgresql_fdw", conname)));
 696
 697         return (remoteConn *) conn;
 698 }
 699
 700 /*
 701  * Create a new connection to a PostgreSQL server and register the connection
 702  * to the FDW connection cache.
 703  * If the same name connection found, abort even if an unnamed connection was
 704  * requested.
 705  */
 706 void
 707 createNewConnection(const char *conname, const char *conninfo)
 708 {
 709         ForeignServer      *server;
 710         char                       *name;
 711         char                       *srvname;
 712
 713         AssertArg(conninfo != NULL);
 714
 715         if (conname)
 716         {
 717                 name = pstrdup(conname);
 718                 truncate_identifier(name, strlen(name), true);
 719         }
 720         else
 721                 name = UNNAMED_CONN_NAME;
 722
 723         /* raise an error if the same name connection found. */
 724         if (GetFSConnectionByName(name, NULL) != NULL)
 725                 ereport(ERROR,
 726                         (errcode(ERRCODE_DUPLICATE_OBJECT),
 727                          errmsg("duplicate connection name")));
 728
 729         /* first check for valid foreign data server */
 730         srvname = pstrdup(conninfo);
 731         truncate_identifier(srvname, strlen(srvname), false);
 732         server = GetForeignServerByName(srvname, true);
 733
 734         if (server != NULL)
 735         {
 736                 UserMapping                *user;
 737                 ForeignDataWrapper *wrapper;
 738
 739                 user = GetUserMapping(GetOuterUserId(), server->serverid);
 740                 wrapper = GetForeignDataWrapper(server->fdwid);
 741                 if (OidIsValid(wrapper->fdwhandler))
 742                 {
 743                         FdwRoutine                 *routine;
 744
 745                         routine = GetFdwRoutine(wrapper->fdwhandler);
 746                         (void) ConnectToForeignServer(routine, server, user, name);
 747                 }
 748                 else
 749                 {
 750                         /*
 751                          * For backward compatibility, assume all of the options are
 752                          * for postgresql connections if the fdw doesn't have a handler.
 753                          */
 754                         RegisterFSConnection(
 755                                 pgConnectServer(server, user),
 756                                 &postgresql_fdw_routine, name);
 757                 }
 758         }
 759         else
 760         {
 761                 /* check password in connection string if not superuser */
 762                 dblink_connstr_check(conninfo);
 763
 764                 RegisterFSConnection(
 765                         (FSConnection *) createRemoteConn(PQconnectdb(conninfo)),
 766                         &postgresql_fdw_routine, name);
 767         }
 768 }
 769
 770 /*
 771  * For non-superusers, insist that the connstr specify a password.      This
 772  * prevents a password from being picked up from .pgpass, a service file,
 773  * the environment, etc.  We don't want the postgres user's passwords
 774  * to be accessible to non-superusers.
 775  */
 776 static void
 777 dblink_connstr_check(const char *connstr)
 778 {
 779         if (!superuser())
 780         {
 781                 PQconninfoOption *options;
 782                 PQconninfoOption *option;
 783                 bool            connstr_gives_password = false;
 784
 785                 options = PQconninfoParse(connstr, NULL);
 786                 if (options)
 787                 {
 788                         for (option = options; option->keyword != NULL; option++)
 789                         {
 790                                 if (strcmp(option->keyword, "password") == 0)
 791                                 {
 792                                         if (option->val != NULL && option->val[0] != '\0')
 793                                         {
 794                                                 connstr_gives_password = true;
 795                                                 break;
 796                                         }
 797                                 }
 798                         }
 799                         PQconninfoFree(options);
 800                 }
 801
 802                 if (!connstr_gives_password)
 803                         ereport(ERROR,
 804                                   (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
 805                                    errmsg("password is required"),
 806                                    errdetail("Non-superusers must provide a password in the connection string.")));
 807         }
 808 }
 809
 810 /*
 811  * Create a remoteConn object from a PGconn.
 812  * Also check that password was specified explicitly if the user was a
 813  * non-superuser.
 814  */
 815 static remoteConn *
 816 createRemoteConn(PGconn *conn)
 817 {
 818         remoteConn         *rconn;
 819
 820         if (PQstatus(conn) != CONNECTION_OK)
 821         {
 822                 char *message = pstrdup(PQerrorMessage(conn));
 823                 PQfinish(conn);
 824                 ereport(ERROR,
 825                                 (errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
 826                                  errmsg("could not establish connection"),
 827                                  errdetail("%s", message)));
 828         }
 829
 830         if (!superuser() && !PQconnectionUsedPassword(conn))
 831         {
 832                 PQfinish(conn);
 833                 ereport(ERROR,
 834                           (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
 835                            errmsg("password is required"),
 836                            errdetail("Non-superuser cannot connect if the server does not request a password."),
 837                            errhint("Target server's authentication method must be changed.")));
 838         }
 839
 840         /* attempt to set client encoding to match server encoding */
 841         PQsetClientEncoding(conn, GetDatabaseEncodingName());
 842
 843         /* Cosntruct remoteConn */
 844         rconn = MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn));
 845         rconn->conn = conn;
 846         rconn->openCursorCount = 0;
 847         rconn->newXactForCursor = false;
 848
 849         return rconn;
 850 }