*
- * $PostgreSQL: pgsql/contrib/dblink/dblink.c,v 1.82.2.5 2010/06/09 03:40:16 itagaki Exp $
+ * $PostgreSQL: pgsql/contrib/dblink/dblink.c,v 1.82.2.6 2010/06/14 20:49:39 tgl Exp $
* Copyright (c) 2001-2009, PostgreSQL Global Development Group
* ALL RIGHTS RESERVED;
*
static HTAB *createConnHash(void);
static void createNewConnection(const char *name, remoteConn *rconn);
static void deleteConnection(const char *name);
-static char **get_pkey_attnames(Oid relid, int16 *numatts);
+static char **get_pkey_attnames(Relation rel, int16 *numatts);
static char **get_text_array_contents(ArrayType *array, int *numitems);
-static char *get_sql_insert(Oid relid, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals);
-static char *get_sql_delete(Oid relid, int2vector *pkattnums, int16 pknumatts, char **tgt_pkattvals);
-static char *get_sql_update(Oid relid, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals);
+static char *get_sql_insert(Relation rel, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals);
+static char *get_sql_delete(Relation rel, int2vector *pkattnums, int16 pknumatts, char **tgt_pkattvals);
+static char *get_sql_update(Relation rel, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals);
static char *quote_literal_cstr(char *rawstr);
static char *quote_ident_cstr(char *rawstr);
static int16 get_attnum_pk_pos(int2vector *pkattnums, int16 pknumatts, int16 key);
-static HeapTuple get_tuple_of_interest(Oid relid, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals);
-static Oid get_relid_from_relname(text *relname_text);
-static char *generate_relation_name(Oid relid);
+static HeapTuple get_tuple_of_interest(Relation rel, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals);
+static Relation get_rel_from_relname(text *relname_text, LOCKMODE lockmode, AclMode aclmode);
+static char *generate_relation_name(Relation rel);
static void dblink_connstr_check(const char *connstr);
static void dblink_security_check(PGconn *conn, remoteConn *rconn);
static void dblink_res_error(const char *conname, PGresult *res, const char *dblink_context_msg, bool fail);
static char *get_connect_string(const char *servername);
static char *escape_param_str(const char *from);
-static int get_nondropped_natts(Oid relid);
+static int get_nondropped_natts(Relation rel);
/* Global */
static remoteConn *pconn = NULL;
dblink_get_pkey(PG_FUNCTION_ARGS)
{
int16 numatts;
- Oid relid;
char **results;
FuncCallContext *funcctx;
int32 call_cntr;
/* stuff done only on the first call of the function */
if (SRF_IS_FIRSTCALL())
{
- TupleDesc tupdesc = NULL;
+ Relation rel;
+ TupleDesc tupdesc;
/* create a function context for cross-call persistence */
funcctx = SRF_FIRSTCALL_INIT();
*/
oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
- /* convert relname to rel Oid */
- relid = get_relid_from_relname(PG_GETARG_TEXT_P(0));
- if (!OidIsValid(relid))
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_TABLE),
- errmsg("relation \"%s\" does not exist",
- text_to_cstring(PG_GETARG_TEXT_PP(0)))));
+ /* open target relation */
+ rel = get_rel_from_relname(PG_GETARG_TEXT_P(0), AccessShareLock, ACL_SELECT);
+
+ /* get the array of attnums */
+ results = get_pkey_attnames(rel, &numatts);
+
+ relation_close(rel, AccessShareLock);
/*
* need a tuple descriptor representing one INT and one TEXT column
attinmeta = TupleDescGetAttInMetadata(tupdesc);
funcctx->attinmeta = attinmeta;
- /* get an array of attnums */
- results = get_pkey_attnames(relid, &numatts);
-
if ((results != NULL) && (numatts > 0))
{
funcctx->max_calls = numatts;
int32 pknumatts_tmp = PG_GETARG_INT32(2);
ArrayType *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
ArrayType *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
- Oid relid;
+ Relation rel;
int16 pknumatts = 0;
char **src_pkattvals;
char **tgt_pkattvals;
int nondropped_natts;
/*
- * Convert relname to rel OID.
+ * Open target relation.
*/
- relid = get_relid_from_relname(relname_text);
- if (!OidIsValid(relid))
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_TABLE),
- errmsg("relation \"%s\" does not exist",
- text_to_cstring(relname_text))));
+ rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
/*
* There should be at least one key attribute
* ensure we don't ask for more pk attributes than we have
* non-dropped columns
*/
- nondropped_natts = get_nondropped_natts(relid);
+ nondropped_natts = get_nondropped_natts(rel);
if (pknumatts > nondropped_natts)
ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
errmsg("number of primary key fields exceeds number of specified relation attributes")));
/*
* Prep work is finally done. Go get the SQL string.
*/
- sql = get_sql_insert(relid, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
+ sql = get_sql_insert(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
+
+ /*
+ * Now we can close the relation.
+ */
+ relation_close(rel, AccessShareLock);
/*
* And send it
int32 pknumatts_tmp = PG_GETARG_INT32(2);
ArrayType *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
int nondropped_natts;
- Oid relid;
+ Relation rel;
int16 pknumatts = 0;
char **tgt_pkattvals;
int tgt_nitems;
char *sql;
/*
- * Convert relname to rel OID.
+ * Open target relation.
*/
- relid = get_relid_from_relname(relname_text);
- if (!OidIsValid(relid))
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_TABLE),
- errmsg("relation \"%s\" does not exist",
- text_to_cstring(relname_text))));
+ rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
/*
* There should be at least one key attribute
* ensure we don't ask for more pk attributes than we have
* non-dropped columns
*/
- nondropped_natts = get_nondropped_natts(relid);
+ nondropped_natts = get_nondropped_natts(rel);
if (pknumatts > nondropped_natts)
ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
errmsg("number of primary key fields exceeds number of specified relation attributes")));
/*
* Prep work is finally done. Go get the SQL string.
*/
- sql = get_sql_delete(relid, pkattnums, pknumatts, tgt_pkattvals);
+ sql = get_sql_delete(rel, pkattnums, pknumatts, tgt_pkattvals);
+
+ /*
+ * Now we can close the relation.
+ */
+ relation_close(rel, AccessShareLock);
/*
* And send it
ArrayType *src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
ArrayType *tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
int nondropped_natts;
- Oid relid;
+ Relation rel;
int16 pknumatts = 0;
char **src_pkattvals;
char **tgt_pkattvals;
char *sql;
/*
- * Convert relname to rel OID.
+ * Open target relation.
*/
- relid = get_relid_from_relname(relname_text);
- if (!OidIsValid(relid))
- ereport(ERROR,
- (errcode(ERRCODE_UNDEFINED_TABLE),
- errmsg("relation \"%s\" does not exist",
- text_to_cstring(relname_text))));
+ rel = get_rel_from_relname(relname_text, AccessShareLock, ACL_SELECT);
/*
* There should be one source array key values for each key attnum
* ensure we don't ask for more pk attributes than we have
* non-dropped columns
*/
- nondropped_natts = get_nondropped_natts(relid);
+ nondropped_natts = get_nondropped_natts(rel);
if (pknumatts > nondropped_natts)
ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR),
errmsg("number of primary key fields exceeds number of specified relation attributes")));
/*
* Prep work is finally done. Go get the SQL string.
*/
- sql = get_sql_update(relid, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
+ sql = get_sql_update(rel, pkattnums, pknumatts, src_pkattvals, tgt_pkattvals);
+
+ /*
+ * Now we can close the relation.
+ */
+ relation_close(rel, AccessShareLock);
/*
* And send it
* Return NULL, and set numatts = 0, if no primary key exists.
*/
static char **
-get_pkey_attnames(Oid relid, int16 *numatts)
+get_pkey_attnames(Relation rel, int16 *numatts)
{
Relation indexRelation;
ScanKeyData skey;
HeapTuple indexTuple;
int i;
char **result = NULL;
- Relation rel;
TupleDesc tupdesc;
- AclResult aclresult;
/* initialize numatts to 0 in case no primary key exists */
*numatts = 0;
- /* open relation using relid, check permissions, get tupdesc */
- rel = relation_open(relid, AccessShareLock);
-
- aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
- ACL_SELECT);
- if (aclresult != ACLCHECK_OK)
- aclcheck_error(aclresult, ACL_KIND_CLASS,
- RelationGetRelationName(rel));
-
tupdesc = rel->rd_att;
/* Prepare to scan pg_index for entries having indrelid = this rel. */
ScanKeyInit(&skey,
Anum_pg_index_indrelid,
BTEqualStrategyNumber, F_OIDEQ,
- ObjectIdGetDatum(relid));
+ ObjectIdGetDatum(RelationGetRelid(rel)));
scan = systable_beginscan(indexRelation, IndexIndrelidIndexId, true,
SnapshotNow, 1, &skey);
systable_endscan(scan);
heap_close(indexRelation, AccessShareLock);
- relation_close(rel, AccessShareLock);
return result;
}
}
static char *
-get_sql_insert(Oid relid, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals)
+get_sql_insert(Relation rel, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals)
{
- Relation rel;
char *relname;
HeapTuple tuple;
TupleDesc tupdesc;
initStringInfo(&buf);
/* get relation name including any needed schema prefix and quoting */
- relname = generate_relation_name(relid);
+ relname = generate_relation_name(rel);
- /*
- * Open relation using relid
- */
- rel = relation_open(relid, AccessShareLock);
tupdesc = rel->rd_att;
natts = tupdesc->natts;
- tuple = get_tuple_of_interest(relid, pkattnums, pknumatts, src_pkattvals);
+ tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
if (!tuple)
ereport(ERROR,
(errcode(ERRCODE_CARDINALITY_VIOLATION),
}
appendStringInfo(&buf, ")");
- relation_close(rel, AccessShareLock);
return (buf.data);
}
static char *
-get_sql_delete(Oid relid, int2vector *pkattnums, int16 pknumatts, char **tgt_pkattvals)
+get_sql_delete(Relation rel, int2vector *pkattnums, int16 pknumatts, char **tgt_pkattvals)
{
- Relation rel;
char *relname;
TupleDesc tupdesc;
int natts;
initStringInfo(&buf);
/* get relation name including any needed schema prefix and quoting */
- relname = generate_relation_name(relid);
+ relname = generate_relation_name(rel);
- /*
- * Open relation using relid
- */
- rel = relation_open(relid, AccessShareLock);
tupdesc = rel->rd_att;
natts = tupdesc->natts;
appendStringInfo(&buf, " IS NULL");
}
- relation_close(rel, AccessShareLock);
return (buf.data);
}
static char *
-get_sql_update(Oid relid, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals)
+get_sql_update(Relation rel, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals, char **tgt_pkattvals)
{
- Relation rel;
char *relname;
HeapTuple tuple;
TupleDesc tupdesc;
initStringInfo(&buf);
/* get relation name including any needed schema prefix and quoting */
- relname = generate_relation_name(relid);
+ relname = generate_relation_name(rel);
- /*
- * Open relation using relid
- */
- rel = relation_open(relid, AccessShareLock);
tupdesc = rel->rd_att;
natts = tupdesc->natts;
- tuple = get_tuple_of_interest(relid, pkattnums, pknumatts, src_pkattvals);
+ tuple = get_tuple_of_interest(rel, pkattnums, pknumatts, src_pkattvals);
if (!tuple)
ereport(ERROR,
(errcode(ERRCODE_CARDINALITY_VIOLATION),
appendStringInfo(&buf, " IS NULL");
}
- relation_close(rel, AccessShareLock);
return (buf.data);
}
}
static HeapTuple
-get_tuple_of_interest(Oid relid, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals)
+get_tuple_of_interest(Relation rel, int2vector *pkattnums, int16 pknumatts, char **src_pkattvals)
{
- Relation rel;
char *relname;
TupleDesc tupdesc;
StringInfoData buf;
initStringInfo(&buf);
/* get relation name including any needed schema prefix and quoting */
- relname = generate_relation_name(relid);
+ relname = generate_relation_name(rel);
- /*
- * Open relation using relid
- */
- rel = relation_open(relid, AccessShareLock);
- tupdesc = CreateTupleDescCopy(rel->rd_att);
- relation_close(rel, AccessShareLock);
+ tupdesc = rel->rd_att;
/*
* Connect to SPI manager
return NULL;
}
-static Oid
-get_relid_from_relname(text *relname_text)
+/*
+ * Open the relation named by relname_text, acquire specified type of lock,
+ * verify we have specified permissions.
+ * Caller must close rel when done with it.
+ */
+static Relation
+get_rel_from_relname(text *relname_text, LOCKMODE lockmode, AclMode aclmode)
{
RangeVar *relvar;
Relation rel;
- Oid relid;
+ AclResult aclresult;
relvar = makeRangeVarFromNameList(textToQualifiedNameList(relname_text));
- rel = heap_openrv(relvar, AccessShareLock);
- relid = RelationGetRelid(rel);
- relation_close(rel, AccessShareLock);
+ rel = heap_openrv(relvar, lockmode);
- return relid;
+ aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
+ aclmode);
+ if (aclresult != ACLCHECK_OK)
+ aclcheck_error(aclresult, ACL_KIND_CLASS,
+ RelationGetRelationName(rel));
+
+ return rel;
}
/*
* generate_relation_name - copied from ruleutils.c
- * Compute the name to display for a relation specified by OID
+ * Compute the name to display for a relation
*
* The result includes all necessary quoting and schema-prefixing.
*/
static char *
-generate_relation_name(Oid relid)
+generate_relation_name(Relation rel)
{
- HeapTuple tp;
- Form_pg_class reltup;
char *nspname;
char *result;
- tp = SearchSysCache(RELOID,
- ObjectIdGetDatum(relid),
- 0, 0, 0);
- if (!HeapTupleIsValid(tp))
- elog(ERROR, "cache lookup failed for relation %u", relid);
-
- reltup = (Form_pg_class) GETSTRUCT(tp);
-
/* Qualify the name if not visible in search path */
- if (RelationIsVisible(relid))
+ if (RelationIsVisible(RelationGetRelid(rel)))
nspname = NULL;
else
- nspname = get_namespace_name(reltup->relnamespace);
+ nspname = get_namespace_name(rel->rd_rel->relnamespace);
- result = quote_qualified_identifier(nspname, NameStr(reltup->relname));
-
- ReleaseSysCache(tp);
+ result = quote_qualified_identifier(nspname, RelationGetRelationName(rel));
return result;
}
}
static int
-get_nondropped_natts(Oid relid)
+get_nondropped_natts(Relation rel)
{
int nondropped_natts = 0;
TupleDesc tupdesc;
- Relation rel;
int natts;
int i;
- rel = relation_open(relid, AccessShareLock);
tupdesc = rel->rd_att;
natts = tupdesc->natts;
nondropped_natts++;
}
- relation_close(rel, AccessShareLock);
return nondropped_natts;
}
projects / users / gsingh / postgres.git / commitdiff