Back-patch contrib/vacuumlo's new -l (limit) option into 9.0 and 9.1.

Since 9.0, removing lots of large objects in a single transaction risks
exceeding max_locks_per_transaction, because we merged large object removal
into the generic object-drop mechanism, which takes out an exclusive lock
on each object to be dropped.  This creates a hazard for contrib/vacuumlo,
which has historically tried to drop all unreferenced large objects in one
transaction.  There doesn't seem to be any correctness requirement to do it
that way, though; we only need to drop enough large objects per transaction
to amortize the commit costs.

To prevent a regression from pre-9.0 releases wherein vacuumlo worked just
fine, back-patch commits b69f2e36402aaa222ed03c1769b3de6d5be5f302 and
64c604898e812aa93c124c666e8709fff1b8dd26, which break vacuumlo's deletions
into multiple transactions with a user-controllable upper limit on the
number of objects dropped per transaction.

Tim Lewis, Robert Haas, Tom Lane

diff --git a/contrib/vacuumlo/vacuumlo.c b/contrib/vacuumlo/vacuumlo.c
index f6e2a28cf38cc5645c83a6eb0b73aebe87bdb2cc..641a8c3425d97614a14a547f91651f0bbd83eae8 100644 (file)
--- a/contrib/vacuumlo/vacuumlo.c
+++ b/contrib/vacuumlo/vacuumlo.c
@@ -3,7 +3,7 @@
  * vacuumlo.c
  *   This removes orphaned large objects from a database.
  *
- * Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  *
@@ -22,7 +22,6 @@
 #endif
 
 #include "libpq-fe.h"
-#include "libpq/libpq-fs.h"
 
 #define atooid(x)  ((Oid) strtoul((x), NULL, 10))
 
@@ -30,8 +29,7 @@
 
 extern char *optarg;
 extern int optind,
-           opterr,
-           optopt;
+           opterr;
 
 enum trivalue
 {
@@ -48,29 +46,32 @@ struct _param
    char       *pg_host;
    int         verbose;
    int         dry_run;
+   long        transaction_limit;
 };
 
-int            vacuumlo(char *, struct _param *);
-void       usage(const char *progname);
+static int vacuumlo(const char *database, const struct _param * param);
+static void usage(const char *progname);
 
 
 
 /*
  * This vacuums LOs of one database. It returns 0 on success, -1 on failure.
  */
-int
-vacuumlo(char *database, struct _param * param)
+static int
+vacuumlo(const char *database, const struct _param * param)
 {
    PGconn     *conn;
    PGresult   *res,
               *res2;
    char        buf[BUFSIZE];
-   int         matched;
-   int         deleted;
+   long        matched;
+   long        deleted;
    int         i;
    static char *password = NULL;
    bool        new_pass;
+   bool        success = true;
 
+   /* Note: password can be carried over from a previous call */
    if (param->pg_prompt == TRI_YES && password == NULL)
        password = simple_prompt("Password: ", 100, false);
 
@@ -118,7 +119,7 @@ vacuumlo(char *database, struct _param * param)
 
    if (param->verbose)
    {
-       fprintf(stdout, "Connected to %s\n", database);
+       fprintf(stdout, "Connected to database \"%s\"\n", database);
        if (param->dry_run)
            fprintf(stdout, "Test run: no large objects will be removed!\n");
    }
@@ -219,9 +220,21 @@ vacuumlo(char *database, struct _param * param)
        if (param->verbose)
            fprintf(stdout, "Checking %s in %s.%s\n", field, schema, table);
 
+       schema = PQescapeIdentifier(conn, schema, strlen(schema));
+       table = PQescapeIdentifier(conn, table, strlen(table));
+       field = PQescapeIdentifier(conn, field, strlen(field));
+
+       if (!schema || !table || !field)
+       {
+           fprintf(stderr, "Out of memory\n");
+           PQclear(res);
+           PQfinish(conn);
+           return -1;
+       }
+
        snprintf(buf, BUFSIZE,
                 "DELETE FROM vacuum_l "
-                "WHERE lo IN (SELECT \"%s\" FROM \"%s\".\"%s\")",
+                "WHERE lo IN (SELECT %s FROM %s.%s)",
                 field, schema, table);
        res2 = PQexec(conn, buf);
        if (PQresultStatus(res2) != PGRES_COMMAND_OK)
@@ -235,23 +248,35 @@ vacuumlo(char *database, struct _param * param)
            return -1;
        }
        PQclear(res2);
+
+       PQfreemem(schema);
+       PQfreemem(table);
+       PQfreemem(field);
    }
    PQclear(res);
 
    /*
-    * Run the actual deletes in a single transaction.  Note that this would
-    * be a bad idea in pre-7.1 Postgres releases (since rolling back a table
-    * delete used to cause problems), but it should be safe now.
+    * Now, those entries remaining in vacuum_l are orphans.  Delete 'em.
+    *
+    * We don't want to run each delete as an individual transaction, because
+    * the commit overhead would be high.  However, since 9.0 the backend will
+    * acquire a lock per deleted LO, so deleting too many LOs per transaction
+    * risks running out of room in the shared-memory lock table.
+    * Accordingly, we delete up to transaction_limit LOs per transaction.
     */
    res = PQexec(conn, "begin");
+   if (PQresultStatus(res) != PGRES_COMMAND_OK)
+   {
+       fprintf(stderr, "Failed to start transaction:\n");
+       fprintf(stderr, "%s", PQerrorMessage(conn));
+       PQclear(res);
+       PQfinish(conn);
+       return -1;
+   }
    PQclear(res);
 
-   /*
-    * Finally, those entries remaining in vacuum_l are orphans.
-    */
    buf[0] = '\0';
-   strcat(buf, "SELECT lo ");
-   strcat(buf, "FROM vacuum_l");
+   strcat(buf, "SELECT lo FROM vacuum_l");
    res = PQexec(conn, buf);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
    {
@@ -280,37 +305,87 @@ vacuumlo(char *database, struct _param * param)
            {
                fprintf(stderr, "\nFailed to remove lo %u: ", lo);
                fprintf(stderr, "%s", PQerrorMessage(conn));
+               if (PQtransactionStatus(conn) == PQTRANS_INERROR)
+               {
+                   success = false;
+                   break;
+               }
            }
            else
                deleted++;
        }
        else
            deleted++;
+       if (param->transaction_limit > 0 &&
+           (deleted % param->transaction_limit) == 0)
+       {
+           res2 = PQexec(conn, "commit");
+           if (PQresultStatus(res2) != PGRES_COMMAND_OK)
+           {
+               fprintf(stderr, "Failed to commit transaction:\n");
+               fprintf(stderr, "%s", PQerrorMessage(conn));
+               PQclear(res2);
+               PQclear(res);
+               PQfinish(conn);
+               return -1;
+           }
+           PQclear(res2);
+           res2 = PQexec(conn, "begin");
+           if (PQresultStatus(res2) != PGRES_COMMAND_OK)
+           {
+               fprintf(stderr, "Failed to start transaction:\n");
+               fprintf(stderr, "%s", PQerrorMessage(conn));
+               PQclear(res2);
+               PQclear(res);
+               PQfinish(conn);
+               return -1;
+           }
+           PQclear(res2);
+       }
    }
    PQclear(res);
 
    /*
     * That's all folks!
     */
-   res = PQexec(conn, "end");
+   res = PQexec(conn, "commit");
+   if (PQresultStatus(res) != PGRES_COMMAND_OK)
+   {
+       fprintf(stderr, "Failed to commit transaction:\n");
+       fprintf(stderr, "%s", PQerrorMessage(conn));
+       PQclear(res);
+       PQfinish(conn);
+       return -1;
+   }
    PQclear(res);
 
    PQfinish(conn);
 
    if (param->verbose)
-       fprintf(stdout, "\r%s %d large objects from %s.\n",
-          (param->dry_run ? "Would remove" : "Removed"), deleted, database);
+   {
+       if (param->dry_run)
+           fprintf(stdout, "\rWould remove %ld large objects from database \"%s\".\n",
+                   deleted, database);
+       else if (success)
+           fprintf(stdout,
+                   "\rSuccessfully removed %ld large objects from database \"%s\".\n",
+                   deleted, database);
+       else
+           fprintf(stdout, "\rRemoval from database \"%s\" failed at object %ld of %ld.\n",
+                   database, deleted, matched);
+   }
 
-   return 0;
+   return ((param->dry_run || success) ? 0 : -1);
 }
 
-void
+static void
 usage(const char *progname)
 {
    printf("%s removes unreferenced large objects from databases.\n\n", progname);
    printf("Usage:\n  %s [OPTION]... DBNAME...\n\n", progname);
    printf("Options:\n");
    printf("  -h HOSTNAME  database server host or socket directory\n");
+   printf("  -l LIMIT     commit after removing each LIMIT large objects\n");
    printf("  -n           don't remove large objects, just show what would be done\n");
    printf("  -p PORT      database server port\n");
    printf("  -U USERNAME  user name to connect as\n");
@@ -335,14 +410,16 @@ main(int argc, char **argv)
 
    progname = get_progname(argv[0]);
 
-   /* Parameter handling */
+   /* Set default parameter values */
    param.pg_user = NULL;
    param.pg_prompt = TRI_DEFAULT;
    param.pg_host = NULL;
    param.pg_port = NULL;
    param.verbose = 0;
    param.dry_run = 0;
+   param.transaction_limit = 1000;
 
+   /* Process command-line arguments */
    if (argc > 1)
    {
        if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
@@ -359,7 +436,7 @@ main(int argc, char **argv)
 
    while (1)
    {
-       c = getopt(argc, argv, "h:U:p:vnwW");
+       c = getopt(argc, argv, "h:l:U:p:vnwW");
        if (c == -1)
            break;
 
@@ -377,6 +454,16 @@ main(int argc, char **argv)
                param.dry_run = 1;
                param.verbose = 1;
                break;
+           case 'l':
+               param.transaction_limit = strtol(optarg, NULL, 10);
+               if (param.transaction_limit < 0)
+               {
+                   fprintf(stderr,
+               "%s: transaction limit must not be negative (0 disables)\n",
+                       progname);
+                   exit(1);
+               }
+               break;
            case 'U':
                param.pg_user = strdup(optarg);
                break;
@@ -405,7 +492,7 @@ main(int argc, char **argv)
    if (optind >= argc)
    {
        fprintf(stderr, "vacuumlo: missing required argument: database name\n");
-       fprintf(stderr, "Try 'vacuumlo -?' for help.\n");
+       fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname);
        exit(1);
    }
 

diff --git a/doc/src/sgml/vacuumlo.sgml b/doc/src/sgml/vacuumlo.sgml
index 471a6ca32be9e75e7a38b3fe51891373ce02e357..97753de6c0c3ab0b9a045cbd36b89df25bc66a77 100644 (file)
--- a/doc/src/sgml/vacuumlo.sgml
+++ b/doc/src/sgml/vacuumlo.sgml
@@ -49,6 +49,19 @@ vacuumlo [options] database [database2 ... databaseN]
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><option>-l</option> <replaceable>limit</></term>
+    <listitem>
+     <para>
+      Remove no more than <replaceable>limit</> large objects per
+      transaction (default 1000).  Since the server acquires a lock per LO
+      removed, removing too many LOs in one transaction risks exceeding
+      <xref linkend="guc-max-locks-per-transaction">.  Set the limit to
+      zero if you want all removals done in a single transaction.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><option>-U</option> <replaceable>username</></term>
     <listitem>
@@ -110,18 +123,19 @@ vacuumlo [options] database [database2 ... databaseN]
   <title>Method</title>
 
   <para>
-   First, it builds a temporary table which contains all of the OIDs of the
-   large objects in that database.
+   First, <application>vacuumlo</> builds a temporary table which contains all
+   of the OIDs of the large objects in the selected database.
   </para>
 
   <para>
    It then scans through all columns in the database that are of type
    <type>oid</> or <type>lo</>, and removes matching entries from the
-  temporary table.
+   temporary table.  (Note: only types with these names are considered;
+   in particular, domains over them are not considered.)
   </para>
 
   <para>
-   The remaining entries in the temp table identify orphaned LOs.
+   The remaining entries in the temporary table identify orphaned LOs.
    These are removed.
   </para>
  </sect2>