diff options
Diffstat (limited to 'src/backend/commands/dbcommands.c')
| -rw-r--r-- | src/backend/commands/dbcommands.c | 110 |
1 files changed, 90 insertions, 20 deletions
diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c index 09f1ab41ad3..307729ab7ef 100644 --- a/src/backend/commands/dbcommands.c +++ b/src/backend/commands/dbcommands.c @@ -719,7 +719,7 @@ createdb(ParseState *pstate, const CreatedbStmt *stmt) int encoding = -1; bool dbistemplate = false; bool dballowconnections = true; - int dbconnlimit = -1; + int dbconnlimit = DATCONNLIMIT_UNLIMITED; char *dbcollversion = NULL; int notherbackends; int npreparedxacts; @@ -926,7 +926,7 @@ createdb(ParseState *pstate, const CreatedbStmt *stmt) if (dconnlimit && dconnlimit->arg) { dbconnlimit = defGetInt32(dconnlimit); - if (dbconnlimit < -1) + if (dbconnlimit < DATCONNLIMIT_UNLIMITED) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid connection limit: %d", dbconnlimit))); @@ -978,6 +978,16 @@ createdb(ParseState *pstate, const CreatedbStmt *stmt) dbtemplate))); /* + * If the source database was in the process of being dropped, we can't + * use it as a template. + */ + if (database_is_invalid_oid(src_dboid)) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot use invalid database \"%s\" as template", dbtemplate), + errhint("Use DROP DATABASE to drop invalid databases.")); + + /* * Permission check: to copy a DB that's not marked datistemplate, you * must be superuser or the owner thereof. */ @@ -1576,6 +1586,7 @@ dropdb(const char *dbname, bool missing_ok, bool force) bool db_istemplate; Relation pgdbrel; HeapTuple tup; + Form_pg_database datform; int notherbackends; int npreparedxacts; int nslots, @@ -1692,17 +1703,6 @@ dropdb(const char *dbname, bool missing_ok, bool force) errdetail_busy_db(notherbackends, npreparedxacts))); /* - * Remove the database's tuple from pg_database. - */ - tup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(db_id)); - if (!HeapTupleIsValid(tup)) - elog(ERROR, "cache lookup failed for database %u", db_id); - - CatalogTupleDelete(pgdbrel, &tup->t_self); - - ReleaseSysCache(tup); - - /* * Delete any comments or security labels associated with the database. */ DeleteSharedComments(db_id, DatabaseRelationId); @@ -1719,6 +1719,37 @@ dropdb(const char *dbname, bool missing_ok, bool force) dropDatabaseDependencies(db_id); /* + * Tell the cumulative stats system to forget it immediately, too. + */ + pgstat_drop_database(db_id); + + tup = SearchSysCacheCopy1(DATABASEOID, ObjectIdGetDatum(db_id)); + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for database %u", db_id); + datform = (Form_pg_database) GETSTRUCT(tup); + + /* + * Except for the deletion of the catalog row, subsequent actions are not + * transactional (consider DropDatabaseBuffers() discarding modified + * buffers). But we might crash or get interrupted below. To prevent + * accesses to a database with invalid contents, mark the database as + * invalid using an in-place update. + * + * We need to flush the WAL before continuing, to guarantee the + * modification is durable before performing irreversible filesystem + * operations. + */ + datform->datconnlimit = DATCONNLIMIT_INVALID_DB; + heap_inplace_update(pgdbrel, tup); + XLogFlush(XactLastRecEnd); + + /* + * Also delete the tuple - transactionally. If this transaction commits, + * the row will be gone, but if we fail, dropdb() can be invoked again. + */ + CatalogTupleDelete(pgdbrel, &tup->t_self); + + /* * Drop db-specific replication slots. */ ReplicationSlotsDropDBSlots(db_id); @@ -1731,11 +1762,6 @@ dropdb(const char *dbname, bool missing_ok, bool force) DropDatabaseBuffers(db_id); /* - * Tell the cumulative stats system to forget it immediately, too. - */ - pgstat_drop_database(db_id); - - /* * Tell checkpointer to forget any pending fsync and unlink requests for * files in the database; else the fsyncs will fail at next checkpoint, or * worse, it will delete files that belong to a newly created database @@ -2248,7 +2274,7 @@ AlterDatabase(ParseState *pstate, AlterDatabaseStmt *stmt, bool isTopLevel) ListCell *option; bool dbistemplate = false; bool dballowconnections = true; - int dbconnlimit = -1; + int dbconnlimit = DATCONNLIMIT_UNLIMITED; DefElem *distemplate = NULL; DefElem *dallowconnections = NULL; DefElem *dconnlimit = NULL; @@ -2319,7 +2345,7 @@ AlterDatabase(ParseState *pstate, AlterDatabaseStmt *stmt, bool isTopLevel) if (dconnlimit && dconnlimit->arg) { dbconnlimit = defGetInt32(dconnlimit); - if (dbconnlimit < -1) + if (dbconnlimit < DATCONNLIMIT_UNLIMITED) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid connection limit: %d", dbconnlimit))); @@ -2346,6 +2372,14 @@ AlterDatabase(ParseState *pstate, AlterDatabaseStmt *stmt, bool isTopLevel) datform = (Form_pg_database) GETSTRUCT(tuple); dboid = datform->oid; + if (database_is_invalid_form(datform)) + { + ereport(FATAL, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot alter invalid database \"%s\"", stmt->dbname), + errhint("Use DROP DATABASE to drop invalid databases.")); + } + if (!object_ownercheck(DatabaseRelationId, dboid, GetUserId())) aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE, stmt->dbname); @@ -3064,6 +3098,42 @@ get_database_name(Oid dbid) return result; } + +/* + * While dropping a database the pg_database row is marked invalid, but the + * catalog contents still exist. Connections to such a database are not + * allowed. + */ +bool +database_is_invalid_form(Form_pg_database datform) +{ + return datform->datconnlimit == DATCONNLIMIT_INVALID_DB; +} + + +/* + * Convenience wrapper around database_is_invalid_form() + */ +bool +database_is_invalid_oid(Oid dboid) +{ + HeapTuple dbtup; + Form_pg_database dbform; + bool invalid; + + dbtup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dboid)); + if (!HeapTupleIsValid(dbtup)) + elog(ERROR, "cache lookup failed for database %u", dboid); + dbform = (Form_pg_database) GETSTRUCT(dbtup); + + invalid = database_is_invalid_form(dbform); + + ReleaseSysCache(dbtup); + + return invalid; +} + + /* * recovery_create_dbdir() * |