granicus.if.org Git - postgresql/blob - src/backend/commands/dbcommands.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * dbcommands.c
   4  *              Database management commands (create/drop database).
   5  *
   6  *
   7  * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
   8  * Portions Copyright (c) 1994, Regents of the University of California
   9  *
  10  *
  11  * IDENTIFICATION
  12  *        $PostgreSQL: pgsql/src/backend/commands/dbcommands.c,v 1.150 2005/02/20 02:21:34 tgl Exp $
  13  *
  14  *-------------------------------------------------------------------------
  15  */
  16 #include "postgres.h"
  17
  18 #include <fcntl.h>
  19 #include <unistd.h>
  20 #include <sys/stat.h>
  21
  22 #include "access/genam.h"
  23 #include "access/heapam.h"
  24 #include "catalog/catname.h"
  25 #include "catalog/catalog.h"
  26 #include "catalog/pg_database.h"
  27 #include "catalog/pg_shadow.h"
  28 #include "catalog/pg_tablespace.h"
  29 #include "catalog/indexing.h"
  30 #include "commands/comment.h"
  31 #include "commands/dbcommands.h"
  32 #include "commands/tablespace.h"
  33 #include "mb/pg_wchar.h"
  34 #include "miscadmin.h"
  35 #include "postmaster/bgwriter.h"
  36 #include "storage/fd.h"
  37 #include "storage/freespace.h"
  38 #include "storage/sinval.h"
  39 #include "utils/acl.h"
  40 #include "utils/array.h"
  41 #include "utils/builtins.h"
  42 #include "utils/flatfiles.h"
  43 #include "utils/fmgroids.h"
  44 #include "utils/guc.h"
  45 #include "utils/lsyscache.h"
  46 #include "utils/syscache.h"
  47
  48
  49 /* non-export function prototypes */
  50 static bool get_db_info(const char *name, Oid *dbIdP, int4 *ownerIdP,
  51                         int *encodingP, bool *dbIsTemplateP, Oid *dbLastSysOidP,
  52                         TransactionId *dbVacuumXidP, TransactionId *dbFrozenXidP,
  53                         Oid *dbTablespace);
  54 static bool have_createdb_privilege(void);
  55 static void remove_dbtablespaces(Oid db_id);
  56
  57
  58 /*
  59  * CREATE DATABASE
  60  */
  61 void
  62 createdb(const CreatedbStmt *stmt)
  63 {
  64         HeapScanDesc scan;
  65         Relation        rel;
  66         Oid                     src_dboid;
  67         AclId           src_owner;
  68         int                     src_encoding;
  69         bool            src_istemplate;
  70         Oid                     src_lastsysoid;
  71         TransactionId src_vacuumxid;
  72         TransactionId src_frozenxid;
  73         Oid                     src_deftablespace;
  74         Oid                     dst_deftablespace;
  75         Relation        pg_database_rel;
  76         HeapTuple       tuple;
  77         TupleDesc       pg_database_dsc;
  78         Datum           new_record[Natts_pg_database];
  79         char            new_record_nulls[Natts_pg_database];
  80         Oid                     dboid;
  81         AclId           datdba;
  82         ListCell   *option;
  83         DefElem    *dtablespacename = NULL;
  84         DefElem    *downer = NULL;
  85         DefElem    *dtemplate = NULL;
  86         DefElem    *dencoding = NULL;
  87         char       *dbname = stmt->dbname;
  88         char       *dbowner = NULL;
  89         char       *dbtemplate = NULL;
  90         int                     encoding = -1;
  91
  92 #ifndef WIN32
  93         char            buf[2 * MAXPGPATH + 100];
  94 #endif
  95
  96         /* don't call this in a transaction block */
  97         PreventTransactionChain((void *) stmt, "CREATE DATABASE");
  98
  99         /* Extract options from the statement node tree */
 100         foreach(option, stmt->options)
 101         {
 102                 DefElem    *defel = (DefElem *) lfirst(option);
 103
 104                 if (strcmp(defel->defname, "tablespace") == 0)
 105                 {
 106                         if (dtablespacename)
 107                                 ereport(ERROR,
 108                                                 (errcode(ERRCODE_SYNTAX_ERROR),
 109                                                  errmsg("conflicting or redundant options")));
 110                         dtablespacename = defel;
 111                 }
 112                 else if (strcmp(defel->defname, "owner") == 0)
 113                 {
 114                         if (downer)
 115                                 ereport(ERROR,
 116                                                 (errcode(ERRCODE_SYNTAX_ERROR),
 117                                                  errmsg("conflicting or redundant options")));
 118                         downer = defel;
 119                 }
 120                 else if (strcmp(defel->defname, "template") == 0)
 121                 {
 122                         if (dtemplate)
 123                                 ereport(ERROR,
 124                                                 (errcode(ERRCODE_SYNTAX_ERROR),
 125                                                  errmsg("conflicting or redundant options")));
 126                         dtemplate = defel;
 127                 }
 128                 else if (strcmp(defel->defname, "encoding") == 0)
 129                 {
 130                         if (dencoding)
 131                                 ereport(ERROR,
 132                                                 (errcode(ERRCODE_SYNTAX_ERROR),
 133                                                  errmsg("conflicting or redundant options")));
 134                         dencoding = defel;
 135                 }
 136                 else if (strcmp(defel->defname, "location") == 0)
 137                 {
 138                         ereport(WARNING,
 139                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 140                                          errmsg("LOCATION is not supported anymore"),
 141                                          errhint("Consider using tablespaces instead.")));
 142                 }
 143                 else
 144                         elog(ERROR, "option \"%s\" not recognized",
 145                                  defel->defname);
 146         }
 147
 148         if (downer && downer->arg)
 149                 dbowner = strVal(downer->arg);
 150         if (dtemplate && dtemplate->arg)
 151                 dbtemplate = strVal(dtemplate->arg);
 152         if (dencoding && dencoding->arg)
 153         {
 154                 const char *encoding_name;
 155
 156                 if (IsA(dencoding->arg, Integer))
 157                 {
 158                         encoding = intVal(dencoding->arg);
 159                         encoding_name = pg_encoding_to_char(encoding);
 160                         if (strcmp(encoding_name, "") == 0 ||
 161                                 pg_valid_server_encoding(encoding_name) < 0)
 162                                 ereport(ERROR,
 163                                                 (errcode(ERRCODE_UNDEFINED_OBJECT),
 164                                                  errmsg("%d is not a valid encoding code",
 165                                                                 encoding)));
 166                 }
 167                 else if (IsA(dencoding->arg, String))
 168                 {
 169                         encoding_name = strVal(dencoding->arg);
 170                         if (pg_valid_server_encoding(encoding_name) < 0)
 171                                 ereport(ERROR,
 172                                                 (errcode(ERRCODE_UNDEFINED_OBJECT),
 173                                                  errmsg("%s is not a valid encoding name",
 174                                                                 encoding_name)));
 175                         encoding = pg_char_to_encoding(encoding_name);
 176                 }
 177                 else
 178                         elog(ERROR, "unrecognized node type: %d",
 179                                  nodeTag(dencoding->arg));
 180         }
 181
 182         /* obtain sysid of proposed owner */
 183         if (dbowner)
 184                 datdba = get_usesysid(dbowner); /* will ereport if no such user */
 185         else
 186                 datdba = GetUserId();
 187
 188         if (datdba == GetUserId())
 189         {
 190                 /* creating database for self: can be superuser or createdb */
 191                 if (!superuser() && !have_createdb_privilege())
 192                         ereport(ERROR,
 193                                         (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 194                                          errmsg("permission denied to create database")));
 195         }
 196         else
 197         {
 198                 /* creating database for someone else: must be superuser */
 199                 /* note that the someone else need not have any permissions */
 200                 if (!superuser())
 201                         ereport(ERROR,
 202                                         (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 203                                          errmsg("must be superuser to create database for another user")));
 204         }
 205
 206         /*
 207          * Check for db name conflict.  There is a race condition here, since
 208          * another backend could create the same DB name before we commit.
 209          * However, holding an exclusive lock on pg_database for the whole
 210          * time we are copying the source database doesn't seem like a good
 211          * idea, so accept possibility of race to create.  We will check again
 212          * after we grab the exclusive lock.
 213          */
 214         if (get_db_info(dbname, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
 215                 ereport(ERROR,
 216                                 (errcode(ERRCODE_DUPLICATE_DATABASE),
 217                                  errmsg("database \"%s\" already exists", dbname)));
 218
 219         /*
 220          * Lookup database (template) to be cloned.
 221          */
 222         if (!dbtemplate)
 223                 dbtemplate = "template1";               /* Default template database name */
 224
 225         if (!get_db_info(dbtemplate, &src_dboid, &src_owner, &src_encoding,
 226                                          &src_istemplate, &src_lastsysoid,
 227                                          &src_vacuumxid, &src_frozenxid, &src_deftablespace))
 228                 ereport(ERROR,
 229                                 (errcode(ERRCODE_UNDEFINED_DATABASE),
 230                  errmsg("template database \"%s\" does not exist", dbtemplate)));
 231
 232         /*
 233          * Permission check: to copy a DB that's not marked datistemplate, you
 234          * must be superuser or the owner thereof.
 235          */
 236         if (!src_istemplate)
 237         {
 238                 if (!superuser() && GetUserId() != src_owner)
 239                         ereport(ERROR,
 240                                         (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 241                                          errmsg("permission denied to copy database \"%s\"",
 242                                                         dbtemplate)));
 243         }
 244
 245         /*
 246          * The source DB can't have any active backends, except this one
 247          * (exception is to allow CREATE DB while connected to template1).
 248          * Otherwise we might copy inconsistent data.  This check is not
 249          * bulletproof, since someone might connect while we are copying...
 250          */
 251         if (DatabaseHasActiveBackends(src_dboid, true))
 252                 ereport(ERROR,
 253                                 (errcode(ERRCODE_OBJECT_IN_USE),
 254                 errmsg("source database \"%s\" is being accessed by other users",
 255                            dbtemplate)));
 256
 257         /* If encoding is defaulted, use source's encoding */
 258         if (encoding < 0)
 259                 encoding = src_encoding;
 260
 261         /* Some encodings are client only */
 262         if (!PG_VALID_BE_ENCODING(encoding))
 263                 ereport(ERROR,
 264                                 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
 265                                  errmsg("invalid server encoding %d", encoding)));
 266
 267         /* Resolve default tablespace for new database */
 268         if (dtablespacename && dtablespacename->arg)
 269         {
 270                 char       *tablespacename;
 271                 AclResult       aclresult;
 272
 273                 tablespacename = strVal(dtablespacename->arg);
 274                 dst_deftablespace = get_tablespace_oid(tablespacename);
 275                 if (!OidIsValid(dst_deftablespace))
 276                         ereport(ERROR,
 277                                         (errcode(ERRCODE_UNDEFINED_OBJECT),
 278                                          errmsg("tablespace \"%s\" does not exist",
 279                                                         tablespacename)));
 280                 /* check permissions */
 281                 aclresult = pg_tablespace_aclcheck(dst_deftablespace, GetUserId(),
 282                                                                                    ACL_CREATE);
 283                 if (aclresult != ACLCHECK_OK)
 284                         aclcheck_error(aclresult, ACL_KIND_TABLESPACE,
 285                                                    tablespacename);
 286
 287                 /*
 288                  * If we are trying to change the default tablespace of the template,
 289                  * we require that the template not have any files in the new default
 290                  * tablespace.  This is necessary because otherwise the copied
 291                  * database would contain pg_class rows that refer to its default
 292                  * tablespace both explicitly (by OID) and implicitly (as zero), which
 293                  * would cause problems.  For example another CREATE DATABASE using
 294                  * the copied database as template, and trying to change its default
 295                  * tablespace again, would yield outright incorrect results (it would
 296                  * improperly move tables to the new default tablespace that should
 297                  * stay in the same tablespace).
 298                  */
 299                 if (dst_deftablespace != src_deftablespace)
 300                 {
 301                         char       *srcpath;
 302                         struct stat st;
 303
 304                         srcpath = GetDatabasePath(src_dboid, dst_deftablespace);
 305
 306                         if (stat(srcpath, &st) == 0 &&
 307                                 S_ISDIR(st.st_mode) &&
 308                                 !directory_is_empty(srcpath))
 309                                 ereport(ERROR,
 310                                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 311                                                  errmsg("cannot assign new default tablespace \"%s\"",
 312                                                                 tablespacename),
 313                                                  errdetail("There is a conflict because database \"%s\" already has some tables in this tablespace.",
 314                                                                    dbtemplate)));
 315                         pfree(srcpath);
 316                 }
 317         }
 318         else
 319         {
 320                 /* Use template database's default tablespace */
 321                 dst_deftablespace = src_deftablespace;
 322                 /* Note there is no additional permission check in this path */
 323         }
 324
 325         /*
 326          * Preassign OID for pg_database tuple, so that we can compute db
 327          * path.
 328          */
 329         dboid = newoid();
 330
 331         /*
 332          * Force dirty buffers out to disk, to ensure source database is
 333          * up-to-date for the copy.  (We really only need to flush buffers for
 334          * the source database, but bufmgr.c provides no API for that.)
 335          */
 336         BufferSync(-1, -1);
 337
 338         /*
 339          * Close virtual file descriptors so the kernel has more available for
 340          * the system() calls below.
 341          */
 342         closeAllVfds();
 343
 344         /*
 345          * Iterate through all tablespaces of the template database, and copy
 346          * each one to the new database.
 347          */
 348         rel = heap_openr(TableSpaceRelationName, AccessShareLock);
 349         scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
 350         while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
 351         {
 352                 Oid                     srctablespace = HeapTupleGetOid(tuple);
 353                 Oid                     dsttablespace;
 354                 char       *srcpath;
 355                 char       *dstpath;
 356                 struct stat st;
 357
 358                 /* No need to copy global tablespace */
 359                 if (srctablespace == GLOBALTABLESPACE_OID)
 360                         continue;
 361
 362                 srcpath = GetDatabasePath(src_dboid, srctablespace);
 363
 364                 if (stat(srcpath, &st) < 0 || !S_ISDIR(st.st_mode) ||
 365                         directory_is_empty(srcpath))
 366                 {
 367                         /* Assume we can ignore it */
 368                         pfree(srcpath);
 369                         continue;
 370                 }
 371
 372                 if (srctablespace == src_deftablespace)
 373                         dsttablespace = dst_deftablespace;
 374                 else
 375                         dsttablespace = srctablespace;
 376
 377                 dstpath = GetDatabasePath(dboid, dsttablespace);
 378
 379                 if (stat(dstpath, &st) == 0 || errno != ENOENT)
 380                 {
 381                         remove_dbtablespaces(dboid);
 382                         ereport(ERROR,
 383                                         (errmsg("could not initialize database directory"),
 384                                          errdetail("Directory \"%s\" already exists.",
 385                                                            dstpath)));
 386                 }
 387
 388 #ifndef WIN32
 389
 390                 /*
 391                  * Copy this subdirectory to the new location
 392                  *
 393                  * XXX use of cp really makes this code pretty grotty, particularly
 394                  * with respect to lack of ability to report errors well.  Someday
 395                  * rewrite to do it for ourselves.
 396                  */
 397
 398                 /* We might need to use cp -R one day for portability */
 399                 snprintf(buf, sizeof(buf), "cp -r '%s' '%s'",
 400                                  srcpath, dstpath);
 401                 if (system(buf) != 0)
 402                 {
 403                         remove_dbtablespaces(dboid);
 404                         ereport(ERROR,
 405                                         (errmsg("could not initialize database directory"),
 406                                          errdetail("Failing system command was: %s", buf),
 407                                          errhint("Look in the postmaster's stderr log for more information.")));
 408                 }
 409 #else                                                   /* WIN32 */
 410                 if (copydir(srcpath, dstpath) != 0)
 411                 {
 412                         /* copydir should already have given details of its troubles */
 413                         remove_dbtablespaces(dboid);
 414                         ereport(ERROR,
 415                                         (errmsg("could not initialize database directory")));
 416                 }
 417 #endif   /* WIN32 */
 418
 419                 /* Record the filesystem change in XLOG */
 420                 {
 421                         xl_dbase_create_rec xlrec;
 422                         XLogRecData rdata[3];
 423
 424                         xlrec.db_id = dboid;
 425                         rdata[0].buffer = InvalidBuffer;
 426                         rdata[0].data = (char *) &xlrec;
 427                         rdata[0].len = offsetof(xl_dbase_create_rec, src_path);
 428                         rdata[0].next = &(rdata[1]);
 429
 430                         rdata[1].buffer = InvalidBuffer;
 431                         rdata[1].data = (char *) srcpath;
 432                         rdata[1].len = strlen(srcpath) + 1;
 433                         rdata[1].next = &(rdata[2]);
 434
 435                         rdata[2].buffer = InvalidBuffer;
 436                         rdata[2].data = (char *) dstpath;
 437                         rdata[2].len = strlen(dstpath) + 1;
 438                         rdata[2].next = NULL;
 439
 440                         (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE, rdata);
 441                 }
 442         }
 443         heap_endscan(scan);
 444         heap_close(rel, AccessShareLock);
 445
 446         /*
 447          * Now OK to grab exclusive lock on pg_database.
 448          */
 449         pg_database_rel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 450
 451         /* Check to see if someone else created same DB name meanwhile. */
 452         if (get_db_info(dbname, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
 453         {
 454                 /* Don't hold lock while doing recursive remove */
 455                 heap_close(pg_database_rel, AccessExclusiveLock);
 456                 remove_dbtablespaces(dboid);
 457                 ereport(ERROR,
 458                                 (errcode(ERRCODE_DUPLICATE_DATABASE),
 459                                  errmsg("database \"%s\" already exists", dbname)));
 460         }
 461
 462         /*
 463          * Insert a new tuple into pg_database
 464          */
 465         pg_database_dsc = RelationGetDescr(pg_database_rel);
 466
 467         /* Form tuple */
 468         MemSet(new_record, 0, sizeof(new_record));
 469         MemSet(new_record_nulls, ' ', sizeof(new_record_nulls));
 470
 471         new_record[Anum_pg_database_datname - 1] =
 472                 DirectFunctionCall1(namein, CStringGetDatum(dbname));
 473         new_record[Anum_pg_database_datdba - 1] = Int32GetDatum(datdba);
 474         new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(encoding);
 475         new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(false);
 476         new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(true);
 477         new_record[Anum_pg_database_datlastsysoid - 1] = ObjectIdGetDatum(src_lastsysoid);
 478         new_record[Anum_pg_database_datvacuumxid - 1] = TransactionIdGetDatum(src_vacuumxid);
 479         new_record[Anum_pg_database_datfrozenxid - 1] = TransactionIdGetDatum(src_frozenxid);
 480         new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(dst_deftablespace);
 481
 482         /*
 483          * We deliberately set datconfig and datacl to defaults (NULL), rather
 484          * than copying them from the template database.  Copying datacl would
 485          * be a bad idea when the owner is not the same as the template's
 486          * owner. It's more debatable whether datconfig should be copied.
 487          */
 488         new_record_nulls[Anum_pg_database_datconfig - 1] = 'n';
 489         new_record_nulls[Anum_pg_database_datacl - 1] = 'n';
 490
 491         tuple = heap_formtuple(pg_database_dsc, new_record, new_record_nulls);
 492
 493         HeapTupleSetOid(tuple, dboid);          /* override heap_insert's OID
 494                                                                                  * selection */
 495
 496         simple_heap_insert(pg_database_rel, tuple);
 497
 498         /* Update indexes */
 499         CatalogUpdateIndexes(pg_database_rel, tuple);
 500
 501         /*
 502          * Force dirty buffers out to disk, so that newly-connecting backends
 503          * will see the new database in pg_database right away.  (They'll see
 504          * an uncommitted tuple, but they don't care; see GetRawDatabaseInfo.)
 505          */
 506         FlushRelationBuffers(pg_database_rel, MaxBlockNumber);
 507
 508         /* Close pg_database, but keep exclusive lock till commit */
 509         heap_close(pg_database_rel, NoLock);
 510
 511         /*
 512          * Set flag to update flat database file at commit.
 513          */
 514         database_file_update_needed();
 515 }
 516
 517
 518 /*
 519  * DROP DATABASE
 520  */
 521 void
 522 dropdb(const char *dbname)
 523 {
 524         int4            db_owner;
 525         bool            db_istemplate;
 526         Oid                     db_id;
 527         Relation        pgdbrel;
 528         SysScanDesc pgdbscan;
 529         ScanKeyData key;
 530         HeapTuple       tup;
 531
 532         PreventTransactionChain((void *) dbname, "DROP DATABASE");
 533
 534         AssertArg(dbname);
 535
 536         if (strcmp(dbname, get_database_name(MyDatabaseId)) == 0)
 537                 ereport(ERROR,
 538                                 (errcode(ERRCODE_OBJECT_IN_USE),
 539                                  errmsg("cannot drop the currently open database")));
 540
 541         /*
 542          * Obtain exclusive lock on pg_database.  We need this to ensure that
 543          * no new backend starts up in the target database while we are
 544          * deleting it.  (Actually, a new backend might still manage to start
 545          * up, because it will read pg_database without any locking to
 546          * discover the database's OID.  But it will detect its error in
 547          * ReverifyMyDatabase and shut down before any serious damage is done.
 548          * See postinit.c.)
 549          */
 550         pgdbrel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 551
 552         if (!get_db_info(dbname, &db_id, &db_owner, NULL,
 553                                          &db_istemplate, NULL, NULL, NULL, NULL))
 554                 ereport(ERROR,
 555                                 (errcode(ERRCODE_UNDEFINED_DATABASE),
 556                                  errmsg("database \"%s\" does not exist", dbname)));
 557
 558         if (GetUserId() != db_owner && !superuser())
 559                 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
 560                                            dbname);
 561
 562         /*
 563          * Disallow dropping a DB that is marked istemplate.  This is just to
 564          * prevent people from accidentally dropping template0 or template1;
 565          * they can do so if they're really determined ...
 566          */
 567         if (db_istemplate)
 568                 ereport(ERROR,
 569                                 (errcode(ERRCODE_WRONG_OBJECT_TYPE),
 570                                  errmsg("cannot drop a template database")));
 571
 572         /*
 573          * Check for active backends in the target database.
 574          */
 575         if (DatabaseHasActiveBackends(db_id, false))
 576                 ereport(ERROR,
 577                                 (errcode(ERRCODE_OBJECT_IN_USE),
 578                            errmsg("database \"%s\" is being accessed by other users",
 579                                           dbname)));
 580
 581         /*
 582          * Find the database's tuple by OID (should be unique).
 583          */
 584         ScanKeyInit(&key,
 585                                 ObjectIdAttributeNumber,
 586                                 BTEqualStrategyNumber, F_OIDEQ,
 587                                 ObjectIdGetDatum(db_id));
 588
 589         pgdbscan = systable_beginscan(pgdbrel, DatabaseOidIndex, true,
 590                                                                   SnapshotNow, 1, &key);
 591
 592         tup = systable_getnext(pgdbscan);
 593         if (!HeapTupleIsValid(tup))
 594         {
 595                 /*
 596                  * This error should never come up since the existence of the
 597                  * database is checked earlier
 598                  */
 599                 elog(ERROR, "database \"%s\" doesn't exist despite earlier reports to the contrary",
 600                          dbname);
 601         }
 602
 603         /* Remove the database's tuple from pg_database */
 604         simple_heap_delete(pgdbrel, &tup->t_self);
 605
 606         systable_endscan(pgdbscan);
 607
 608         /*
 609          * Delete any comments associated with the database
 610          *
 611          * NOTE: this is probably dead code since any such comments should have
 612          * been in that database, not mine.
 613          */
 614         DeleteComments(db_id, RelationGetRelid(pgdbrel), 0);
 615
 616         /*
 617          * Drop pages for this database that are in the shared buffer cache.
 618          * This is important to ensure that no remaining backend tries to
 619          * write out a dirty buffer to the dead database later...
 620          */
 621         DropBuffers(db_id);
 622
 623         /*
 624          * Also, clean out any entries in the shared free space map.
 625          */
 626         FreeSpaceMapForgetDatabase(db_id);
 627
 628         /*
 629          * On Windows, force a checkpoint so that the bgwriter doesn't hold any
 630          * open files, which would cause rmdir() to fail.
 631          */
 632 #ifdef WIN32
 633         RequestCheckpoint(true);
 634 #endif
 635
 636         /*
 637          * Remove all tablespace subdirs belonging to the database.
 638          */
 639         remove_dbtablespaces(db_id);
 640
 641         /*
 642          * Force dirty buffers out to disk, so that newly-connecting backends
 643          * will see the database tuple marked dead in pg_database right away.
 644          * (They'll see an uncommitted deletion, but they don't care; see
 645          * GetRawDatabaseInfo.)
 646          */
 647         FlushRelationBuffers(pgdbrel, MaxBlockNumber);
 648
 649         /* Close pg_database, but keep exclusive lock till commit */
 650         heap_close(pgdbrel, NoLock);
 651
 652         /*
 653          * Set flag to update flat database file at commit.
 654          */
 655         database_file_update_needed();
 656 }
 657
 658
 659 /*
 660  * Rename database
 661  */
 662 void
 663 RenameDatabase(const char *oldname, const char *newname)
 664 {
 665         HeapTuple       tup,
 666                                 newtup;
 667         Relation        rel;
 668         SysScanDesc scan,
 669                                 scan2;
 670         ScanKeyData key,
 671                                 key2;
 672
 673         /*
 674          * Obtain AccessExclusiveLock so that no new session gets started
 675          * while the rename is in progress.
 676          */
 677         rel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 678
 679         ScanKeyInit(&key,
 680                                 Anum_pg_database_datname,
 681                                 BTEqualStrategyNumber, F_NAMEEQ,
 682                                 NameGetDatum(oldname));
 683         scan = systable_beginscan(rel, DatabaseNameIndex, true,
 684                                                           SnapshotNow, 1, &key);
 685
 686         tup = systable_getnext(scan);
 687         if (!HeapTupleIsValid(tup))
 688                 ereport(ERROR,
 689                                 (errcode(ERRCODE_UNDEFINED_DATABASE),
 690                                  errmsg("database \"%s\" does not exist", oldname)));
 691
 692         /*
 693          * XXX Client applications probably store the current database
 694          * somewhere, so renaming it could cause confusion.  On the other
 695          * hand, there may not be an actual problem besides a little
 696          * confusion, so think about this and decide.
 697          */
 698         if (HeapTupleGetOid(tup) == MyDatabaseId)
 699                 ereport(ERROR,
 700                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 701                                  errmsg("current database may not be renamed")));
 702
 703         /*
 704          * Make sure the database does not have active sessions.  Might not be
 705          * necessary, but it's consistent with other database operations.
 706          */
 707         if (DatabaseHasActiveBackends(HeapTupleGetOid(tup), false))
 708                 ereport(ERROR,
 709                                 (errcode(ERRCODE_OBJECT_IN_USE),
 710                            errmsg("database \"%s\" is being accessed by other users",
 711                                           oldname)));
 712
 713         /* make sure the new name doesn't exist */
 714         ScanKeyInit(&key2,
 715                                 Anum_pg_database_datname,
 716                                 BTEqualStrategyNumber, F_NAMEEQ,
 717                                 NameGetDatum(newname));
 718         scan2 = systable_beginscan(rel, DatabaseNameIndex, true,
 719                                                            SnapshotNow, 1, &key2);
 720         if (HeapTupleIsValid(systable_getnext(scan2)))
 721                 ereport(ERROR,
 722                                 (errcode(ERRCODE_DUPLICATE_DATABASE),
 723                                  errmsg("database \"%s\" already exists", newname)));
 724         systable_endscan(scan2);
 725
 726         /* must be owner */
 727         if (!pg_database_ownercheck(HeapTupleGetOid(tup), GetUserId()))
 728                 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
 729                                            oldname);
 730
 731         /* must have createdb */
 732         if (!have_createdb_privilege())
 733                 ereport(ERROR,
 734                                 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 735                                  errmsg("permission denied to rename database")));
 736
 737         /* rename */
 738         newtup = heap_copytuple(tup);
 739         namestrcpy(&(((Form_pg_database) GETSTRUCT(newtup))->datname), newname);
 740         simple_heap_update(rel, &newtup->t_self, newtup);
 741         CatalogUpdateIndexes(rel, newtup);
 742
 743         systable_endscan(scan);
 744
 745         /*
 746          * Force dirty buffers out to disk, so that newly-connecting backends
 747          * will see the renamed database in pg_database right away.  (They'll
 748          * see an uncommitted tuple, but they don't care; see
 749          * GetRawDatabaseInfo.)
 750          */
 751         FlushRelationBuffers(rel, MaxBlockNumber);
 752
 753         /* Close pg_database, but keep exclusive lock till commit */
 754         heap_close(rel, NoLock);
 755
 756         /*
 757          * Set flag to update flat database file at commit.
 758          */
 759         database_file_update_needed();
 760 }
 761
 762
 763 /*
 764  * ALTER DATABASE name SET ...
 765  */
 766 void
 767 AlterDatabaseSet(AlterDatabaseSetStmt *stmt)
 768 {
 769         char       *valuestr;
 770         HeapTuple       tuple,
 771                                 newtuple;
 772         Relation        rel;
 773         ScanKeyData scankey;
 774         SysScanDesc scan;
 775         Datum           repl_val[Natts_pg_database];
 776         char            repl_null[Natts_pg_database];
 777         char            repl_repl[Natts_pg_database];
 778
 779         valuestr = flatten_set_variable_args(stmt->variable, stmt->value);
 780
 781         /*
 782          * We need AccessExclusiveLock so we can safely do FlushRelationBuffers.
 783          */
 784         rel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 785         ScanKeyInit(&scankey,
 786                                 Anum_pg_database_datname,
 787                                 BTEqualStrategyNumber, F_NAMEEQ,
 788                                 NameGetDatum(stmt->dbname));
 789         scan = systable_beginscan(rel, DatabaseNameIndex, true,
 790                                                           SnapshotNow, 1, &scankey);
 791         tuple = systable_getnext(scan);
 792         if (!HeapTupleIsValid(tuple))
 793                 ereport(ERROR,
 794                                 (errcode(ERRCODE_UNDEFINED_DATABASE),
 795                                  errmsg("database \"%s\" does not exist", stmt->dbname)));
 796
 797         if (!(superuser()
 798                 || ((Form_pg_database) GETSTRUCT(tuple))->datdba == GetUserId()))
 799                 aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
 800                                            stmt->dbname);
 801
 802         MemSet(repl_repl, ' ', sizeof(repl_repl));
 803         repl_repl[Anum_pg_database_datconfig - 1] = 'r';
 804
 805         if (strcmp(stmt->variable, "all") == 0 && valuestr == NULL)
 806         {
 807                 /* RESET ALL */
 808                 repl_null[Anum_pg_database_datconfig - 1] = 'n';
 809                 repl_val[Anum_pg_database_datconfig - 1] = (Datum) 0;
 810         }
 811         else
 812         {
 813                 Datum           datum;
 814                 bool            isnull;
 815                 ArrayType  *a;
 816
 817                 repl_null[Anum_pg_database_datconfig - 1] = ' ';
 818
 819                 datum = heap_getattr(tuple, Anum_pg_database_datconfig,
 820                                                          RelationGetDescr(rel), &isnull);
 821
 822                 a = isnull ? NULL : DatumGetArrayTypeP(datum);
 823
 824                 if (valuestr)
 825                         a = GUCArrayAdd(a, stmt->variable, valuestr);
 826                 else
 827                         a = GUCArrayDelete(a, stmt->variable);
 828
 829                 if (a)
 830                         repl_val[Anum_pg_database_datconfig - 1] = PointerGetDatum(a);
 831                 else
 832                         repl_null[Anum_pg_database_datconfig - 1] = 'n';
 833         }
 834
 835         newtuple = heap_modifytuple(tuple, RelationGetDescr(rel), repl_val, repl_null, repl_repl);
 836         simple_heap_update(rel, &tuple->t_self, newtuple);
 837
 838         /* Update indexes */
 839         CatalogUpdateIndexes(rel, newtuple);
 840
 841         systable_endscan(scan);
 842
 843         /*
 844          * Force dirty buffers out to disk, so that newly-connecting backends
 845          * will see the altered row in pg_database right away.  (They'll
 846          * see an uncommitted tuple, but they don't care; see
 847          * GetRawDatabaseInfo.)
 848          */
 849         FlushRelationBuffers(rel, MaxBlockNumber);
 850
 851         /* Close pg_database, but keep exclusive lock till commit */
 852         heap_close(rel, NoLock);
 853
 854         /*
 855          * We don't bother updating the flat file since ALTER DATABASE SET
 856          * doesn't affect it.
 857          */
 858 }
 859
 860
 861 /*
 862  * ALTER DATABASE name OWNER TO newowner
 863  */
 864 void
 865 AlterDatabaseOwner(const char *dbname, AclId newOwnerSysId)
 866 {
 867         HeapTuple       tuple;
 868         Relation        rel;
 869         ScanKeyData scankey;
 870         SysScanDesc scan;
 871         Form_pg_database datForm;
 872
 873         /*
 874          * We need AccessExclusiveLock so we can safely do FlushRelationBuffers.
 875          */
 876         rel = heap_openr(DatabaseRelationName, AccessExclusiveLock);
 877         ScanKeyInit(&scankey,
 878                                 Anum_pg_database_datname,
 879                                 BTEqualStrategyNumber, F_NAMEEQ,
 880                                 NameGetDatum(dbname));
 881         scan = systable_beginscan(rel, DatabaseNameIndex, true,
 882                                                           SnapshotNow, 1, &scankey);
 883         tuple = systable_getnext(scan);
 884         if (!HeapTupleIsValid(tuple))
 885                 ereport(ERROR,
 886                                 (errcode(ERRCODE_UNDEFINED_DATABASE),
 887                                  errmsg("database \"%s\" does not exist", dbname)));
 888
 889         datForm = (Form_pg_database) GETSTRUCT(tuple);
 890
 891         /*
 892          * If the new owner is the same as the existing owner, consider the
 893          * command to have succeeded.  This is to be consistent with other
 894          * objects.
 895          */
 896         if (datForm->datdba != newOwnerSysId)
 897         {
 898                 Datum           repl_val[Natts_pg_database];
 899                 char            repl_null[Natts_pg_database];
 900                 char            repl_repl[Natts_pg_database];
 901                 Acl                *newAcl;
 902                 Datum           aclDatum;
 903                 bool            isNull;
 904                 HeapTuple       newtuple;
 905
 906                 /* changing owner's database for someone else: must be superuser */
 907                 /* note that the someone else need not have any permissions */
 908                 if (!superuser())
 909                         ereport(ERROR,
 910                                         (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 911                                          errmsg("must be superuser to change owner")));
 912
 913                 memset(repl_null, ' ', sizeof(repl_null));
 914                 memset(repl_repl, ' ', sizeof(repl_repl));
 915
 916                 repl_repl[Anum_pg_database_datdba - 1] = 'r';
 917                 repl_val[Anum_pg_database_datdba - 1] = Int32GetDatum(newOwnerSysId);
 918
 919                 /*
 920                  * Determine the modified ACL for the new owner.  This is only
 921                  * necessary when the ACL is non-null.
 922                  */
 923                 aclDatum = heap_getattr(tuple,
 924                                                                 Anum_pg_database_datacl,
 925                                                                 RelationGetDescr(rel),
 926                                                                 &isNull);
 927                 if (!isNull)
 928                 {
 929                         newAcl = aclnewowner(DatumGetAclP(aclDatum),
 930                                                                  datForm->datdba, newOwnerSysId);
 931                         repl_repl[Anum_pg_database_datacl - 1] = 'r';
 932                         repl_val[Anum_pg_database_datacl - 1] = PointerGetDatum(newAcl);
 933                 }
 934
 935                 newtuple = heap_modifytuple(tuple, RelationGetDescr(rel), repl_val, repl_null, repl_repl);
 936                 simple_heap_update(rel, &newtuple->t_self, newtuple);
 937                 CatalogUpdateIndexes(rel, newtuple);
 938
 939                 heap_freetuple(newtuple);
 940
 941                 /* must release buffer pins before FlushRelationBuffers */
 942                 systable_endscan(scan);
 943
 944                 /*
 945                  * Force dirty buffers out to disk, so that newly-connecting backends
 946                  * will see the altered row in pg_database right away.  (They'll
 947                  * see an uncommitted tuple, but they don't care; see
 948                  * GetRawDatabaseInfo.)
 949                  */
 950                 FlushRelationBuffers(rel, MaxBlockNumber);
 951         }
 952         else
 953                 systable_endscan(scan);
 954
 955         /* Close pg_database, but keep exclusive lock till commit */
 956         heap_close(rel, NoLock);
 957
 958         /*
 959          * We don't bother updating the flat file since ALTER DATABASE OWNER
 960          * doesn't affect it.
 961          */
 962 }
 963
 964
 965 /*
 966  * Helper functions
 967  */
 968
 969 static bool
 970 get_db_info(const char *name, Oid *dbIdP, int4 *ownerIdP,
 971                         int *encodingP, bool *dbIsTemplateP, Oid *dbLastSysOidP,
 972                         TransactionId *dbVacuumXidP, TransactionId *dbFrozenXidP,
 973                         Oid *dbTablespace)
 974 {
 975         Relation        relation;
 976         ScanKeyData scanKey;
 977         SysScanDesc scan;
 978         HeapTuple       tuple;
 979         bool            gottuple;
 980
 981         AssertArg(name);
 982
 983         /* Caller may wish to grab a better lock on pg_database beforehand... */
 984         relation = heap_openr(DatabaseRelationName, AccessShareLock);
 985
 986         ScanKeyInit(&scanKey,
 987                                 Anum_pg_database_datname,
 988                                 BTEqualStrategyNumber, F_NAMEEQ,
 989                                 NameGetDatum(name));
 990
 991         scan = systable_beginscan(relation, DatabaseNameIndex, true,
 992                                                           SnapshotNow, 1, &scanKey);
 993
 994         tuple = systable_getnext(scan);
 995
 996         gottuple = HeapTupleIsValid(tuple);
 997         if (gottuple)
 998         {
 999                 Form_pg_database dbform = (Form_pg_database) GETSTRUCT(tuple);
1000
1001                 /* oid of the database */
1002                 if (dbIdP)
1003                         *dbIdP = HeapTupleGetOid(tuple);
1004                 /* sysid of the owner */
1005                 if (ownerIdP)
1006                         *ownerIdP = dbform->datdba;
1007                 /* character encoding */
1008                 if (encodingP)
1009                         *encodingP = dbform->encoding;
1010                 /* allowed as template? */
1011                 if (dbIsTemplateP)
1012                         *dbIsTemplateP = dbform->datistemplate;
1013                 /* last system OID used in database */
1014                 if (dbLastSysOidP)
1015                         *dbLastSysOidP = dbform->datlastsysoid;
1016                 /* limit of vacuumed XIDs */
1017                 if (dbVacuumXidP)
1018                         *dbVacuumXidP = dbform->datvacuumxid;
1019                 /* limit of frozen XIDs */
1020                 if (dbFrozenXidP)
1021                         *dbFrozenXidP = dbform->datfrozenxid;
1022                 /* default tablespace for this database */
1023                 if (dbTablespace)
1024                         *dbTablespace = dbform->dattablespace;
1025         }
1026
1027         systable_endscan(scan);
1028         heap_close(relation, AccessShareLock);
1029
1030         return gottuple;
1031 }
1032
1033 static bool
1034 have_createdb_privilege(void)
1035 {
1036         HeapTuple       utup;
1037         bool            retval;
1038
1039         utup = SearchSysCache(SHADOWSYSID,
1040                                                   Int32GetDatum(GetUserId()),
1041                                                   0, 0, 0);
1042
1043         if (!HeapTupleIsValid(utup))
1044                 retval = false;
1045         else
1046                 retval = ((Form_pg_shadow) GETSTRUCT(utup))->usecreatedb;
1047
1048         ReleaseSysCache(utup);
1049
1050         return retval;
1051 }
1052
1053 /*
1054  * Remove tablespace directories
1055  *
1056  * We don't know what tablespaces db_id is using, so iterate through all
1057  * tablespaces removing <tablespace>/db_id
1058  */
1059 static void
1060 remove_dbtablespaces(Oid db_id)
1061 {
1062         Relation        rel;
1063         HeapScanDesc scan;
1064         HeapTuple       tuple;
1065
1066         rel = heap_openr(TableSpaceRelationName, AccessShareLock);
1067         scan = heap_beginscan(rel, SnapshotNow, 0, NULL);
1068         while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
1069         {
1070                 Oid                     dsttablespace = HeapTupleGetOid(tuple);
1071                 char       *dstpath;
1072                 struct stat st;
1073
1074                 /* Don't mess with the global tablespace */
1075                 if (dsttablespace == GLOBALTABLESPACE_OID)
1076                         continue;
1077
1078                 dstpath = GetDatabasePath(db_id, dsttablespace);
1079
1080                 if (stat(dstpath, &st) < 0 || !S_ISDIR(st.st_mode))
1081                 {
1082                         /* Assume we can ignore it */
1083                         pfree(dstpath);
1084                         continue;
1085                 }
1086
1087                 if (!rmtree(dstpath, true))
1088                         ereport(WARNING,
1089                                         (errmsg("could not remove database directory \"%s\"",
1090                                                         dstpath)));
1091
1092                 /* Record the filesystem change in XLOG */
1093                 {
1094                         xl_dbase_drop_rec xlrec;
1095                         XLogRecData rdata[2];
1096
1097                         xlrec.db_id = db_id;
1098                         rdata[0].buffer = InvalidBuffer;
1099                         rdata[0].data = (char *) &xlrec;
1100                         rdata[0].len = offsetof(xl_dbase_drop_rec, dir_path);
1101                         rdata[0].next = &(rdata[1]);
1102
1103                         rdata[1].buffer = InvalidBuffer;
1104                         rdata[1].data = (char *) dstpath;
1105                         rdata[1].len = strlen(dstpath) + 1;
1106                         rdata[1].next = NULL;
1107
1108                         (void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_DROP, rdata);
1109                 }
1110
1111                 pfree(dstpath);
1112         }
1113
1114         heap_endscan(scan);
1115         heap_close(rel, AccessShareLock);
1116 }
1117
1118
1119 /*
1120  * get_database_oid - given a database name, look up the OID
1121  *
1122  * Returns InvalidOid if database name not found.
1123  *
1124  * This is not actually used in this file, but is exported for use elsewhere.
1125  */
1126 Oid
1127 get_database_oid(const char *dbname)
1128 {
1129         Relation        pg_database;
1130         ScanKeyData entry[1];
1131         SysScanDesc scan;
1132         HeapTuple       dbtuple;
1133         Oid                     oid;
1134
1135         /* There's no syscache for pg_database, so must look the hard way */
1136         pg_database = heap_openr(DatabaseRelationName, AccessShareLock);
1137         ScanKeyInit(&entry[0],
1138                                 Anum_pg_database_datname,
1139                                 BTEqualStrategyNumber, F_NAMEEQ,
1140                                 CStringGetDatum(dbname));
1141         scan = systable_beginscan(pg_database, DatabaseNameIndex, true,
1142                                                           SnapshotNow, 1, entry);
1143
1144         dbtuple = systable_getnext(scan);
1145
1146         /* We assume that there can be at most one matching tuple */
1147         if (HeapTupleIsValid(dbtuple))
1148                 oid = HeapTupleGetOid(dbtuple);
1149         else
1150                 oid = InvalidOid;
1151
1152         systable_endscan(scan);
1153         heap_close(pg_database, AccessShareLock);
1154
1155         return oid;
1156 }
1157
1158
1159 /*
1160  * get_database_name - given a database OID, look up the name
1161  *
1162  * Returns a palloc'd string, or NULL if no such database.
1163  *
1164  * This is not actually used in this file, but is exported for use elsewhere.
1165  */
1166 char *
1167 get_database_name(Oid dbid)
1168 {
1169         Relation        pg_database;
1170         ScanKeyData entry[1];
1171         SysScanDesc scan;
1172         HeapTuple       dbtuple;
1173         char       *result;
1174
1175         /* There's no syscache for pg_database, so must look the hard way */
1176         pg_database = heap_openr(DatabaseRelationName, AccessShareLock);
1177         ScanKeyInit(&entry[0],
1178                                 ObjectIdAttributeNumber,
1179                                 BTEqualStrategyNumber, F_OIDEQ,
1180                                 ObjectIdGetDatum(dbid));
1181         scan = systable_beginscan(pg_database, DatabaseOidIndex, true,
1182                                                           SnapshotNow, 1, entry);
1183
1184         dbtuple = systable_getnext(scan);
1185
1186         /* We assume that there can be at most one matching tuple */
1187         if (HeapTupleIsValid(dbtuple))
1188                 result = pstrdup(NameStr(((Form_pg_database) GETSTRUCT(dbtuple))->datname));
1189         else
1190                 result = NULL;
1191
1192         systable_endscan(scan);
1193         heap_close(pg_database, AccessShareLock);
1194
1195         return result;
1196 }
1197
1198 /*
1199  * DATABASE resource manager's routines
1200  */
1201 void
1202 dbase_redo(XLogRecPtr lsn, XLogRecord *record)
1203 {
1204         uint8           info = record->xl_info & ~XLR_INFO_MASK;
1205
1206         if (info == XLOG_DBASE_CREATE)
1207         {
1208                 xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) XLogRecGetData(record);
1209                 char       *dst_path = xlrec->src_path + strlen(xlrec->src_path) + 1;
1210                 struct stat st;
1211
1212 #ifndef WIN32
1213                 char            buf[2 * MAXPGPATH + 100];
1214 #endif
1215
1216                 /*
1217                  * Our theory for replaying a CREATE is to forcibly drop the
1218                  * target subdirectory if present, then re-copy the source data.
1219                  * This may be more work than needed, but it is simple to
1220                  * implement.
1221                  */
1222                 if (stat(dst_path, &st) == 0 && S_ISDIR(st.st_mode))
1223                 {
1224                         if (!rmtree(dst_path, true))
1225                                 ereport(WARNING,
1226                                         (errmsg("could not remove database directory \"%s\"",
1227                                                         dst_path)));
1228                 }
1229
1230                 /*
1231                  * Force dirty buffers out to disk, to ensure source database is
1232                  * up-to-date for the copy.  (We really only need to flush buffers for
1233                  * the source database, but bufmgr.c provides no API for that.)
1234                  */
1235                 BufferSync(-1, -1);
1236
1237 #ifndef WIN32
1238
1239                 /*
1240                  * Copy this subdirectory to the new location
1241                  *
1242                  * XXX use of cp really makes this code pretty grotty, particularly
1243                  * with respect to lack of ability to report errors well.  Someday
1244                  * rewrite to do it for ourselves.
1245                  */
1246
1247                 /* We might need to use cp -R one day for portability */
1248                 snprintf(buf, sizeof(buf), "cp -r '%s' '%s'",
1249                                  xlrec->src_path, dst_path);
1250                 if (system(buf) != 0)
1251                         ereport(ERROR,
1252                                         (errmsg("could not initialize database directory"),
1253                                          errdetail("Failing system command was: %s", buf),
1254                                          errhint("Look in the postmaster's stderr log for more information.")));
1255 #else                                                   /* WIN32 */
1256                 if (copydir(xlrec->src_path, dst_path) != 0)
1257                 {
1258                         /* copydir should already have given details of its troubles */
1259                         ereport(ERROR,
1260                                         (errmsg("could not initialize database directory")));
1261                 }
1262 #endif   /* WIN32 */
1263         }
1264         else if (info == XLOG_DBASE_DROP)
1265         {
1266                 xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) XLogRecGetData(record);
1267
1268                 /*
1269                  * Drop pages for this database that are in the shared buffer
1270                  * cache
1271                  */
1272                 DropBuffers(xlrec->db_id);
1273
1274                 if (!rmtree(xlrec->dir_path, true))
1275                         ereport(WARNING,
1276                                         (errmsg("could not remove database directory \"%s\"",
1277                                                         xlrec->dir_path)));
1278         }
1279         else
1280                 elog(PANIC, "dbase_redo: unknown op code %u", info);
1281 }
1282
1283 void
1284 dbase_undo(XLogRecPtr lsn, XLogRecord *record)
1285 {
1286         elog(PANIC, "dbase_undo: unimplemented");
1287 }
1288
1289 void
1290 dbase_desc(char *buf, uint8 xl_info, char *rec)
1291 {
1292         uint8           info = xl_info & ~XLR_INFO_MASK;
1293
1294         if (info == XLOG_DBASE_CREATE)
1295         {
1296                 xl_dbase_create_rec *xlrec = (xl_dbase_create_rec *) rec;
1297                 char       *dst_path = xlrec->src_path + strlen(xlrec->src_path) + 1;
1298
1299                 sprintf(buf + strlen(buf), "create db: %u copy \"%s\" to \"%s\"",
1300                                 xlrec->db_id, xlrec->src_path, dst_path);
1301         }
1302         else if (info == XLOG_DBASE_DROP)
1303         {
1304                 xl_dbase_drop_rec *xlrec = (xl_dbase_drop_rec *) rec;
1305
1306                 sprintf(buf + strlen(buf), "drop db: %u directory: \"%s\"",
1307                                 xlrec->db_id, xlrec->dir_path);
1308         }
1309         else
1310                 strcat(buf, "UNKNOWN");
1311 }