Skip to content

Commit 68ef051

Browse files
robertmhaasrobertmhaas
committed
Refactor broken CREATE TABLE IF NOT EXISTS support.
Per bug #5988, reported by Marko Tiikkaja, and further analyzed by Tom Lane, the previous coding was broken in several respects: even if the target table already existed, a subsequent CREATE TABLE IF NOT EXISTS might try to add additional constraints or sequences-for-serial specified in the new CREATE TABLE statement. In passing, this also fixes a minor information leak: it's no longer possible to figure out whether a schema to which you don't have CREATE access contains a sequence named like "x_y_seq" by attempting to create a table in that schema called "x" with a serial column called "y". Some more refactoring of this code in the future might be warranted, but that will need to wait for a later major release.
1 parent be90032 commit 68ef051

File tree

11 files changed

+73
-79
lines changed

11 files changed

+73
-79
lines changed

‎src/backend/bootstrap/bootparse.y

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -247,8 +247,7 @@ Boot_CreateStmt:
247247
ONCOMMIT_NOOP,
248248
(Datum) 0,
249249
false,
250-
true,
251-
false);
250+
true);
252251
elog(DEBUG4, "relation created with oid %u", id);
253252
}
254253
do_end();

‎src/backend/catalog/heap.c

Lines changed: 3 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -973,8 +973,7 @@ heap_create_with_catalog(const char *relname,
973973
OnCommitAction oncommit,
974974
Datum reloptions,
975975
bool use_user_acl,
976-
bool allow_system_table_mods,
977-
bool if_not_exists)
976+
bool allow_system_table_mods)
978977
{
979978
Relation pg_class_desc;
980979
Relation new_rel_desc;
@@ -994,26 +993,14 @@ heap_create_with_catalog(const char *relname,
994993
CheckAttributeNamesTypes(tupdesc, relkind, allow_system_table_mods);
995994

996995
/*
997-
* If the relation already exists, it's an error, unless the user
998-
* specifies "IF NOT EXISTS". In that case, we just print a notice and do
999-
* nothing further.
996+
* This would fail later on anyway, if the relation already exists. But
997+
* by catching it here we can emit a nicer error message.
1000998
*/
1001999
existing_relid = get_relname_relid(relname, relnamespace);
10021000
if (existing_relid != InvalidOid)
1003-
{
1004-
if (if_not_exists)
1005-
{
1006-
ereport(NOTICE,
1007-
(errcode(ERRCODE_DUPLICATE_TABLE),
1008-
errmsg("relation \"%s\" already exists, skipping",
1009-
relname)));
1010-
heap_close(pg_class_desc, RowExclusiveLock);
1011-
return InvalidOid;
1012-
}
10131001
ereport(ERROR,
10141002
(errcode(ERRCODE_DUPLICATE_TABLE),
10151003
errmsg("relation \"%s\" already exists", relname)));
1016-
}
10171004

10181005
/*
10191006
* Since we are going to create a rowtype as well, also check for

‎src/backend/catalog/namespace.c

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -360,6 +360,35 @@ RangeVarGetCreationNamespace(const RangeVar *newRelation)
360360
return namespaceId;
361361
}
362362

363+
/*
364+
* RangeVarGetAndCheckCreationNamespace
365+
* As RangeVarGetCreationNamespace, but with a permissions check.
366+
*/
367+
Oid
368+
RangeVarGetAndCheckCreationNamespace(const RangeVar *newRelation)
369+
{
370+
Oid namespaceId;
371+
372+
namespaceId = RangeVarGetCreationNamespace(newRelation);
373+
374+
/*
375+
* Check we have permission to create there. Skip check if bootstrapping,
376+
* since permissions machinery may not be working yet.
377+
*/
378+
if (!IsBootstrapProcessingMode())
379+
{
380+
AclResult aclresult;
381+
382+
aclresult = pg_namespace_aclcheck(namespaceId, GetUserId(),
383+
ACL_CREATE);
384+
if (aclresult != ACLCHECK_OK)
385+
aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
386+
get_namespace_name(namespaceId));
387+
}
388+
389+
return namespaceId;
390+
}
391+
363392
/*
364393
* RelnameGetRelid
365394
* Try to resolve an unqualified relation name.

‎src/backend/catalog/toasting.c

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -227,8 +227,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid, Datum reloptio
227227
ONCOMMIT_NOOP,
228228
reloptions,
229229
false,
230-
true,
231-
false);
230+
true);
232231
Assert(toast_relid != InvalidOid);
233232

234233
/* make the toast relation visible, else heap_open will fail */

‎src/backend/commands/cluster.c

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -646,8 +646,7 @@ make_new_heap(Oid OIDOldHeap, Oid NewTableSpace)
646646
ONCOMMIT_NOOP,
647647
reloptions,
648648
false,
649-
true,
650-
false);
649+
true);
651650
Assert(OIDNewHeap != InvalidOid);
652651

653652
ReleaseSysCache(tuple);

‎src/backend/commands/tablecmds.c

Lines changed: 4 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -439,22 +439,10 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId)
439439
errmsg("cannot create temporary table within security-restricted operation")));
440440

441441
/*
442-
* Look up the namespace in which we are supposed to create the relation.
443-
* Check we have permission to create there. Skip check if bootstrapping,
444-
* since permissions machinery may not be working yet.
442+
* Look up the namespace in which we are supposed to create the relation,
443+
* and check we have permission to create there.
445444
*/
446-
namespaceId = RangeVarGetCreationNamespace(stmt->relation);
447-
448-
if (!IsBootstrapProcessingMode())
449-
{
450-
AclResult aclresult;
451-
452-
aclresult = pg_namespace_aclcheck(namespaceId, GetUserId(),
453-
ACL_CREATE);
454-
if (aclresult != ACLCHECK_OK)
455-
aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
456-
get_namespace_name(namespaceId));
457-
}
445+
namespaceId = RangeVarGetAndCheckCreationNamespace(stmt->relation);
458446

459447
/*
460448
* Select tablespace to use. If not specified, use default tablespace
@@ -602,16 +590,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId)
602590
stmt->oncommit,
603591
reloptions,
604592
true,
605-
allowSystemTableMods,
606-
stmt->if_not_exists);
607-
608-
/*
609-
* If heap_create_with_catalog returns InvalidOid, it means that the user
610-
* specified "IF NOT EXISTS" and the relation already exists. In that
611-
* case we do nothing further.
612-
*/
613-
if (relationId == InvalidOid)
614-
return InvalidOid;
593+
allowSystemTableMods);
615594

616595
/* Store inheritance information for new rel. */
617596
StoreCatalogInheritance(relationId, inheritOids);

‎src/backend/executor/execMain.c

Lines changed: 2 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -2341,7 +2341,6 @@ OpenIntoRel(QueryDesc *queryDesc)
23412341
Oid namespaceId;
23422342
Oid tablespaceId;
23432343
Datum reloptions;
2344-
AclResult aclresult;
23452344
Oid intoRelationId;
23462345
TupleDesc tupdesc;
23472346
DR_intorel *myState;
@@ -2378,13 +2377,7 @@ OpenIntoRel(QueryDesc *queryDesc)
23782377
* Find namespace to create in, check its permissions
23792378
*/
23802379
intoName = into->rel->relname;
2381-
namespaceId = RangeVarGetCreationNamespace(into->rel);
2382-
2383-
aclresult = pg_namespace_aclcheck(namespaceId, GetUserId(),
2384-
ACL_CREATE);
2385-
if (aclresult != ACLCHECK_OK)
2386-
aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
2387-
get_namespace_name(namespaceId));
2380+
namespaceId = RangeVarGetAndCheckCreationNamespace(into->rel);
23882381

23892382
/*
23902383
* Select tablespace to use. If not specified, use default tablespace
@@ -2444,8 +2437,7 @@ OpenIntoRel(QueryDesc *queryDesc)
24442437
into->onCommit,
24452438
reloptions,
24462439
true,
2447-
allowSystemTableMods,
2448-
false);
2440+
allowSystemTableMods);
24492441
Assert(intoRelationId != InvalidOid);
24502442

24512443
FreeTupleDesc(tupdesc);

‎src/backend/parser/parse_utilcmd.c

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -148,13 +148,41 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
148148
List *result;
149149
List *save_alist;
150150
ListCell *elements;
151+
Oid namespaceid;
151152

152153
/*
153154
* We must not scribble on the passed-in CreateStmt, so copy it. (This is
154155
* overkill, but easy.)
155156
*/
156157
stmt = (CreateStmt *) copyObject(stmt);
157158

159+
/*
160+
* Look up the creation namespace. This also checks permissions on the
161+
* target namespace, so that we throw any permissions error as early as
162+
* possible.
163+
*/
164+
namespaceid = RangeVarGetAndCheckCreationNamespace(stmt->relation);
165+
166+
/*
167+
* If the relation already exists and the user specified "IF NOT EXISTS",
168+
* bail out with a NOTICE.
169+
*/
170+
if (stmt->if_not_exists)
171+
{
172+
Oid existing_relid;
173+
174+
existing_relid = get_relname_relid(stmt->relation->relname,
175+
namespaceid);
176+
if (existing_relid != InvalidOid)
177+
{
178+
ereport(NOTICE,
179+
(errcode(ERRCODE_DUPLICATE_TABLE),
180+
errmsg("relation \"%s\" already exists, skipping",
181+
stmt->relation->relname)));
182+
return NIL;
183+
}
184+
}
185+
158186
/*
159187
* If the target relation name isn't schema-qualified, make it so. This
160188
* prevents some corner cases in which added-on rewritten commands might
@@ -164,11 +192,7 @@ transformCreateStmt(CreateStmt *stmt, const char *queryString)
164192
*/
165193
if (stmt->relation->schemaname == NULL
166194
&& stmt->relation->relpersistence != RELPERSISTENCE_TEMP)
167-
{
168-
Oid namespaceid = RangeVarGetCreationNamespace(stmt->relation);
169-
170195
stmt->relation->schemaname = get_namespace_name(namespaceid);
171-
}
172196

173197
/* Set up pstate and CreateStmtContext */
174198
pstate = make_parsestate(NULL);

‎src/backend/tcop/utility.c

Lines changed: 2 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -529,13 +529,6 @@ standard_ProcessUtility(Node *parsetree,
529529
RELKIND_RELATION,
530530
InvalidOid);
531531

532-
/*
533-
* If "IF NOT EXISTS" was specified and the relation
534-
* already exists, do nothing further.
535-
*/
536-
if (relOid == InvalidOid)
537-
continue;
538-
539532
/*
540533
* Let AlterTableCreateToastTable decide if this one
541534
* needs a secondary relation too.
@@ -559,15 +552,8 @@ standard_ProcessUtility(Node *parsetree,
559552
relOid = DefineRelation((CreateStmt *) stmt,
560553
RELKIND_FOREIGN_TABLE,
561554
InvalidOid);
562-
563-
/*
564-
* Unless "IF NOT EXISTS" was specified and the
565-
* relation already exists, create the
566-
* pg_foreign_table entry.
567-
*/
568-
if (relOid != InvalidOid)
569-
CreateForeignTable((CreateForeignTableStmt *) stmt,
570-
relOid);
555+
CreateForeignTable((CreateForeignTableStmt *) stmt,
556+
relOid);
571557
}
572558
else
573559
{

‎src/include/catalog/heap.h

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,8 +63,7 @@ extern Oid heap_create_with_catalog(const char *relname,
6363
OnCommitAction oncommit,
6464
Datum reloptions,
6565
bool use_user_acl,
66-
bool allow_system_table_mods,
67-
bool if_not_exists);
66+
bool allow_system_table_mods);
6867

6968
extern void heap_drop_with_catalog(Oid relid);
7069

‎src/include/catalog/namespace.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ typedef struct OverrideSearchPath
4949

5050
extern Oid RangeVarGetRelid(const RangeVar *relation, bool failOK);
5151
extern Oid RangeVarGetCreationNamespace(const RangeVar *newRelation);
52+
extern Oid RangeVarGetAndCheckCreationNamespace(const RangeVar *newRelation);
5253
extern Oid RelnameGetRelid(const char *relname);
5354
extern bool RelationIsVisible(Oid relid);
5455

0 commit comments

Comments
 (0)