git.sur5r.net Git - bacula/bacula/blob - bacula/src/cats/postgresql.c

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2003-2007 Free Software Foundation Europe e.V.
   5
   6    The main author of Bacula is Kern Sibbald, with contributions from
   7    many others, a complete list can be found in the file AUTHORS.
   8    This program is Free Software; you can redistribute it and/or
   9    modify it under the terms of version two of the GNU General Public
  10    License as published by the Free Software Foundation and included
  11    in the file LICENSE.
  12
  13    This program is distributed in the hope that it will be useful, but
  14    WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16    General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  21    02110-1301, USA.
  22
  23    Bacula® is a registered trademark of Kern Sibbald.
  24    The licensor of Bacula is the Free Software Foundation Europe
  25    (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
  26    Switzerland, email:ftf@fsfeurope.org.
  27 */
  28 /*
  29  * Bacula Catalog Database routines specific to PostgreSQL
  30  *   These are PostgreSQL specific routines
  31  *
  32  *    Dan Langille, December 2003
  33  *    based upon work done by Kern Sibbald, March 2000
  34  *
  35  *    Version $Id$
  36  */
  37
  38
  39 /* The following is necessary so that we do not include
  40  * the dummy external definition of DB.
  41  */
  42 #define __SQL_C                       /* indicate that this is sql.c */
  43
  44 #include "bacula.h"
  45 #include "cats.h"
  46
  47 #ifdef HAVE_POSTGRESQL
  48
  49 #include "postgres_ext.h"       /* needed for NAMEDATALEN */
  50 #include "pg_config_manual.h"   /* get NAMEDATALEN on version 8.3 or later */
  51
  52 /* -----------------------------------------------------------------------
  53  *
  54  *   PostgreSQL dependent defines and subroutines
  55  *
  56  * -----------------------------------------------------------------------
  57  */
  58
  59 /* List of open databases */
  60 static BQUEUE db_list = {&db_list, &db_list};
  61
  62 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  63
  64 /*
  65  * Retrieve database type
  66  */
  67 const char *
  68 db_get_type(void)
  69 {
  70    return "PostgreSQL";
  71
  72 }
  73
  74 /*
  75  * Initialize database data structure. In principal this should
  76  * never have errors, or it is really fatal.
  77  */
  78 B_DB *
  79 db_init_database(JCR *jcr, const char *db_name, const char *db_user, const char *db_password,
  80                  const char *db_address, int db_port, const char *db_socket,
  81                  int mult_db_connections)
  82 {
  83    B_DB *mdb;
  84
  85    if (!db_user) {
  86       Jmsg(jcr, M_FATAL, 0, _("A user name for PostgreSQL must be supplied.\n"));
  87       return NULL;
  88    }
  89    P(mutex);                          /* lock DB queue */
  90    if (!mult_db_connections) {
  91       /* Look to see if DB already open */
  92       for (mdb=NULL; (mdb=(B_DB *)qnext(&db_list, &mdb->bq)); ) {
  93          if (bstrcmp(mdb->db_name, db_name) &&
  94              bstrcmp(mdb->db_address, db_address) &&
  95              mdb->db_port == db_port) {
  96             Dmsg2(100, "DB REopen %d %s\n", mdb->ref_count, db_name);
  97             mdb->ref_count++;
  98             V(mutex);
  99             return mdb;                  /* already open */
 100          }
 101       }
 102    }
 103    Dmsg0(100, "db_open first time\n");
 104    mdb = (B_DB *)malloc(sizeof(B_DB));
 105    memset(mdb, 0, sizeof(B_DB));
 106    mdb->db_name = bstrdup(db_name);
 107    mdb->db_user = bstrdup(db_user);
 108    if (db_password) {
 109       mdb->db_password = bstrdup(db_password);
 110    }
 111    if (db_address) {
 112       mdb->db_address  = bstrdup(db_address);
 113    }
 114    if (db_socket) {
 115       mdb->db_socket   = bstrdup(db_socket);
 116    }
 117    mdb->db_port        = db_port;
 118    mdb->have_insert_id = TRUE;
 119    mdb->errmsg         = get_pool_memory(PM_EMSG); /* get error message buffer */
 120    *mdb->errmsg        = 0;
 121    mdb->cmd            = get_pool_memory(PM_EMSG); /* get command buffer */
 122    mdb->cached_path    = get_pool_memory(PM_FNAME);
 123    mdb->cached_path_id = 0;
 124    mdb->ref_count      = 1;
 125    mdb->fname          = get_pool_memory(PM_FNAME);
 126    mdb->path           = get_pool_memory(PM_FNAME);
 127    mdb->esc_name       = get_pool_memory(PM_FNAME);
 128    mdb->esc_path      = get_pool_memory(PM_FNAME);
 129    mdb->allow_transactions = mult_db_connections;
 130    qinsert(&db_list, &mdb->bq);            /* put db in list */
 131    V(mutex);
 132    return mdb;
 133 }
 134
 135 /* Check that the database correspond to the encoding we want */
 136 static bool check_database_encoding(JCR *jcr, B_DB *mdb)
 137 {
 138    SQL_ROW row;
 139    int ret=false;
 140
 141    if (!db_sql_query(mdb, "SELECT getdatabaseencoding()", NULL, NULL)) {
 142       Jmsg(jcr, M_ERROR, 0, "%s", mdb->errmsg);
 143       return false;
 144    }
 145
 146    if ((row = sql_fetch_row(mdb)) == NULL) {
 147       Mmsg1(mdb->errmsg, _("error fetching row: %s\n"), sql_strerror(mdb));
 148       Jmsg(jcr, M_ERROR, 0, "Can't check database encoding %s", mdb->errmsg);
 149    } else {
 150       ret = bstrcmp(row[0], "SQL_ASCII");
 151
 152       if (ret) {
 153          /* if we are in SQL_ASCII, we can force the client_encoding to SQL_ASCII too */
 154          db_sql_query(mdb, "SET client_encoding TO 'SQL_ASCII'", NULL, NULL);
 155
 156       } else {                  /* something is wrong with database encoding */
 157          Mmsg(mdb->errmsg,
 158               _("Encoding error for database \"%s\". Wanted SQL_ASCII, got %s\n"),
 159               mdb->db_name, row[0]);
 160          Jmsg(jcr, M_WARNING, 0, "%s", mdb->errmsg);
 161          Dmsg1(50, "%s", mdb->errmsg);
 162       }
 163    }
 164    return ret;
 165 }
 166
 167 /*
 168  * Now actually open the database.  This can generate errors,
 169  *   which are returned in the errmsg
 170  *
 171  * DO NOT close the database or free(mdb) here !!!!
 172  */
 173 int
 174 db_open_database(JCR *jcr, B_DB *mdb)
 175 {
 176    int errstat;
 177    char buf[10], *port;
 178
 179 #ifdef xxx                      /* require libpq >= 8.2 */
 180    if (!PQisthreadsafe()) {
 181       Jmsg(jcr, M_ABORT, 0, _("PostgreSQL configuration problem. "
 182            "PostgreSQL library is not thread safe. Cannot continue.\n"));
 183    }
 184 #endif
 185    P(mutex);
 186    if (mdb->connected) {
 187       V(mutex);
 188       return 1;
 189    }
 190    mdb->connected = false;
 191
 192    if ((errstat=rwl_init(&mdb->lock)) != 0) {
 193       berrno be;
 194       Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
 195             be.bstrerror(errstat));
 196       V(mutex);
 197       return 0;
 198    }
 199
 200    if (mdb->db_port) {
 201       bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
 202       port = buf;
 203    } else {
 204       port = NULL;
 205    }
 206
 207    /* If connection fails, try at 5 sec intervals for 30 seconds. */
 208    for (int retry=0; retry < 6; retry++) {
 209       /* connect to the database */
 210       mdb->db = PQsetdbLogin(
 211            mdb->db_address,           /* default = localhost */
 212            port,                      /* default port */
 213            NULL,                      /* pg options */
 214            NULL,                      /* tty, ignored */
 215            mdb->db_name,              /* database name */
 216            mdb->db_user,              /* login name */
 217            mdb->db_password);         /* password */
 218
 219       /* If no connect, try once more in case it is a timing problem */
 220       if (PQstatus(mdb->db) == CONNECTION_OK) {
 221          break;
 222       }
 223       bmicrosleep(5, 0);
 224    }
 225
 226    Dmsg0(50, "pg_real_connect done\n");
 227    Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n", mdb->db_user, mdb->db_name,
 228             mdb->db_password==NULL?"(NULL)":mdb->db_password);
 229
 230    if (PQstatus(mdb->db) != CONNECTION_OK) {
 231       Mmsg2(&mdb->errmsg, _("Unable to connect to PostgreSQL server.\n"
 232             "Database=%s User=%s\n"
 233             "It is probably not running or your password is incorrect.\n"),
 234              mdb->db_name, mdb->db_user);
 235       V(mutex);
 236       return 0;
 237    }
 238
 239    mdb->connected = true;
 240
 241    if (!check_tables_version(jcr, mdb)) {
 242       V(mutex);
 243       return 0;
 244    }
 245
 246    sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
 247
 248    /* tell PostgreSQL we are using standard conforming strings
 249       and avoid warnings such as:
 250        WARNING:  nonstandard use of \\ in a string literal
 251    */
 252    sql_query(mdb, "set standard_conforming_strings=on");
 253
 254    /* check that encoding is SQL_ASCII */
 255    check_database_encoding(jcr, mdb);
 256
 257    V(mutex);
 258    return 1;
 259 }
 260
 261 void
 262 db_close_database(JCR *jcr, B_DB *mdb)
 263 {
 264    if (!mdb) {
 265       return;
 266    }
 267    db_end_transaction(jcr, mdb);
 268    P(mutex);
 269    sql_free_result(mdb);
 270    mdb->ref_count--;
 271    if (mdb->ref_count == 0) {
 272       qdchain(&mdb->bq);
 273       if (mdb->connected && mdb->db) {
 274          sql_close(mdb);
 275       }
 276       rwl_destroy(&mdb->lock);
 277       free_pool_memory(mdb->errmsg);
 278       free_pool_memory(mdb->cmd);
 279       free_pool_memory(mdb->cached_path);
 280       free_pool_memory(mdb->fname);
 281       free_pool_memory(mdb->path);
 282       free_pool_memory(mdb->esc_name);
 283       free_pool_memory(mdb->esc_path);
 284       if (mdb->db_name) {
 285          free(mdb->db_name);
 286       }
 287       if (mdb->db_user) {
 288          free(mdb->db_user);
 289       }
 290       if (mdb->db_password) {
 291          free(mdb->db_password);
 292       }
 293       if (mdb->db_address) {
 294          free(mdb->db_address);
 295       }
 296       if (mdb->db_socket) {
 297          free(mdb->db_socket);
 298       }
 299       free(mdb);
 300    }
 301    V(mutex);
 302 }
 303
 304 void db_thread_cleanup()
 305 { }
 306
 307 /*
 308  * Return the next unique index (auto-increment) for
 309  * the given table.  Return NULL on error.
 310  *
 311  * For PostgreSQL, NULL causes the auto-increment value
 312  *  to be updated.
 313  */
 314 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
 315 {
 316    strcpy(index, "NULL");
 317    return 1;
 318 }
 319
 320
 321 /*
 322  * Escape strings so that PostgreSQL is happy
 323  *
 324  *   NOTE! len is the length of the old string. Your new
 325  *         string must be long enough (max 2*old+1) to hold
 326  *         the escaped output.
 327  */
 328 void
 329 db_escape_string(JCR *jcr, B_DB *mdb, char *snew, char *old, int len)
 330 {
 331    int error;
 332
 333    PQescapeStringConn(mdb->db, snew, old, len, &error);
 334    if (error) {
 335       Jmsg(jcr, M_FATAL, 0, _("PQescapeStringConn returned non-zero.\n"));
 336       /* error on encoding, probably invalid multibyte encoding in the source string
 337         see PQescapeStringConn documentation for details. */
 338       Dmsg0(500, "PQescapeStringConn failed\n");
 339    }
 340 }
 341
 342 /*
 343  * Submit a general SQL command (cmd), and for each row returned,
 344  *  the sqlite_handler is called with the ctx.
 345  */
 346 bool db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
 347 {
 348    SQL_ROW row;
 349
 350    Dmsg0(500, "db_sql_query started\n");
 351
 352    db_lock(mdb);
 353    if (sql_query(mdb, query) != 0) {
 354       Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
 355       db_unlock(mdb);
 356       Dmsg0(500, "db_sql_query failed\n");
 357       return false;
 358    }
 359    Dmsg0(500, "db_sql_query succeeded. checking handler\n");
 360
 361    if (result_handler != NULL) {
 362       Dmsg0(500, "db_sql_query invoking handler\n");
 363       if ((mdb->result = sql_store_result(mdb)) != NULL) {
 364          int num_fields = sql_num_fields(mdb);
 365
 366          Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
 367          while ((row = sql_fetch_row(mdb)) != NULL) {
 368
 369             Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
 370             if (result_handler(ctx, num_fields, row))
 371                break;
 372          }
 373
 374         sql_free_result(mdb);
 375       }
 376    }
 377    db_unlock(mdb);
 378
 379    Dmsg0(500, "db_sql_query finished\n");
 380
 381    return true;
 382 }
 383
 384
 385
 386 POSTGRESQL_ROW my_postgresql_fetch_row(B_DB *mdb)
 387 {
 388    int j;
 389    POSTGRESQL_ROW row = NULL; // by default, return NULL
 390
 391    Dmsg0(500, "my_postgresql_fetch_row start\n");
 392
 393    if (!mdb->row || mdb->row_size < mdb->num_fields) {
 394       int num_fields = mdb->num_fields;
 395       Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
 396
 397       if (mdb->row) {
 398          Dmsg0(500, "my_postgresql_fetch_row freeing space\n");
 399          free(mdb->row);
 400       }
 401       num_fields += 20;                  /* add a bit extra */
 402       mdb->row = (POSTGRESQL_ROW)malloc(sizeof(char *) * num_fields);
 403       mdb->row_size = num_fields;
 404
 405       // now reset the row_number now that we have the space allocated
 406       mdb->row_number = 0;
 407    }
 408
 409    // if still within the result set
 410    if (mdb->row_number >= 0 && mdb->row_number < mdb->num_rows) {
 411       Dmsg2(500, "my_postgresql_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
 412       // get each value from this row
 413       for (j = 0; j < mdb->num_fields; j++) {
 414          mdb->row[j] = PQgetvalue(mdb->result, mdb->row_number, j);
 415          Dmsg2(500, "my_postgresql_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
 416       }
 417       // increment the row number for the next call
 418       mdb->row_number++;
 419
 420       row = mdb->row;
 421    } else {
 422       Dmsg2(500, "my_postgresql_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
 423    }
 424
 425    Dmsg1(500, "my_postgresql_fetch_row finishes returning %p\n", row);
 426
 427    return row;
 428 }
 429
 430 int my_postgresql_max_length(B_DB *mdb, int field_num) {
 431    //
 432    // for a given column, find the max length
 433    //
 434    int max_length;
 435    int i;
 436    int this_length;
 437
 438    max_length = 0;
 439    for (i = 0; i < mdb->num_rows; i++) {
 440       if (PQgetisnull(mdb->result, i, field_num)) {
 441           this_length = 4;        // "NULL"
 442       } else {
 443           this_length = cstrlen(PQgetvalue(mdb->result, i, field_num));
 444       }
 445
 446       if (max_length < this_length) {
 447           max_length = this_length;
 448       }
 449    }
 450
 451    return max_length;
 452 }
 453
 454 POSTGRESQL_FIELD * my_postgresql_fetch_field(B_DB *mdb)
 455 {
 456    int     i;
 457
 458    Dmsg0(500, "my_postgresql_fetch_field starts\n");
 459
 460    if (!mdb->fields || mdb->fields_size < mdb->num_fields) {
 461       if (mdb->fields) {
 462          free(mdb->fields);
 463       }
 464       Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
 465       mdb->fields = (POSTGRESQL_FIELD *)malloc(sizeof(POSTGRESQL_FIELD) * mdb->num_fields);
 466       mdb->fields_size = mdb->num_fields;
 467
 468       for (i = 0; i < mdb->num_fields; i++) {
 469          Dmsg1(500, "filling field %d\n", i);
 470          mdb->fields[i].name           = PQfname(mdb->result, i);
 471          mdb->fields[i].max_length = my_postgresql_max_length(mdb, i);
 472          mdb->fields[i].type       = PQftype(mdb->result, i);
 473          mdb->fields[i].flags      = 0;
 474
 475          Dmsg4(500, "my_postgresql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
 476             mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
 477             mdb->fields[i].flags);
 478       } // end for
 479    } // end if
 480
 481    // increment field number for the next time around
 482
 483    Dmsg0(500, "my_postgresql_fetch_field finishes\n");
 484    return &mdb->fields[mdb->field_number++];
 485 }
 486
 487 void my_postgresql_data_seek(B_DB *mdb, int row)
 488 {
 489    // set the row number to be returned on the next call
 490    // to my_postgresql_fetch_row
 491    mdb->row_number = row;
 492 }
 493
 494 void my_postgresql_field_seek(B_DB *mdb, int field)
 495 {
 496    mdb->field_number = field;
 497 }
 498
 499 /*
 500  * Note, if this routine returns 1 (failure), Bacula expects
 501  *  that no result has been stored.
 502  * This is where QUERY_DB comes with Postgresql.
 503  *
 504  *  Returns:  0  on success
 505  *            1  on failure
 506  *
 507  */
 508 int my_postgresql_query(B_DB *mdb, const char *query)
 509 {
 510    Dmsg0(500, "my_postgresql_query started\n");
 511    // We are starting a new query.  reset everything.
 512    mdb->num_rows     = -1;
 513    mdb->row_number   = -1;
 514    mdb->field_number = -1;
 515
 516    if (mdb->result) {
 517       PQclear(mdb->result);  /* hmm, someone forgot to free?? */
 518       mdb->result = NULL;
 519    }
 520
 521    Dmsg1(500, "my_postgresql_query starts with '%s'\n", query);
 522
 523    for (int i=0; i < 10; i++) {
 524       mdb->result = PQexec(mdb->db, query);
 525       if (mdb->result) {
 526          break;
 527       }
 528       bmicrosleep(5, 0);
 529    }
 530    if (!mdb->result) {
 531       Dmsg1(50, "Query failed: %s\n", query);
 532       goto bail_out;
 533    }
 534
 535    mdb->status = PQresultStatus(mdb->result);
 536    if (mdb->status == PGRES_TUPLES_OK || mdb->status == PGRES_COMMAND_OK) {
 537       Dmsg1(500, "we have a result\n", query);
 538
 539       // how many fields in the set?
 540       mdb->num_fields = (int)PQnfields(mdb->result);
 541       Dmsg1(500, "we have %d fields\n", mdb->num_fields);
 542
 543       mdb->num_rows = PQntuples(mdb->result);
 544       Dmsg1(500, "we have %d rows\n", mdb->num_rows);
 545
 546       mdb->row_number = 0;      /* we can start to fetch something */
 547       mdb->status = 0;          /* succeed */
 548    } else {
 549       Dmsg1(50, "Result status failed: %s\n", query);
 550       goto bail_out;
 551    }
 552
 553    Dmsg0(500, "my_postgresql_query finishing\n");
 554    return mdb->status;
 555
 556 bail_out:
 557    Dmsg1(500, "we failed\n", query);
 558    PQclear(mdb->result);
 559    mdb->result = NULL;
 560    mdb->status = 1;                   /* failed */
 561    return mdb->status;
 562 }
 563
 564 void my_postgresql_free_result(B_DB *mdb)
 565 {
 566
 567    db_lock(mdb);
 568    if (mdb->result) {
 569       PQclear(mdb->result);
 570       mdb->result = NULL;
 571    }
 572
 573    if (mdb->row) {
 574       free(mdb->row);
 575       mdb->row = NULL;
 576    }
 577
 578    if (mdb->fields) {
 579       free(mdb->fields);
 580       mdb->fields = NULL;
 581    }
 582    db_unlock(mdb);
 583 }
 584
 585 int my_postgresql_currval(B_DB *mdb, const char *table_name)
 586 {
 587    // Obtain the current value of the sequence that
 588    // provides the serial value for primary key of the table.
 589
 590    // currval is local to our session.  It is not affected by
 591    // other transactions.
 592
 593    // Determine the name of the sequence.
 594    // PostgreSQL automatically creates a sequence using
 595    // <table>_<column>_seq.
 596    // At the time of writing, all tables used this format for
 597    // for their primary key: <table>id
 598    // Except for basefiles which has a primary key on baseid.
 599    // Therefore, we need to special case that one table.
 600
 601    // everything else can use the PostgreSQL formula.
 602
 603    char      sequence[NAMEDATALEN-1];
 604    char      query   [NAMEDATALEN+50];
 605    PGresult *result;
 606    int       id = 0;
 607
 608    if (strcasecmp(table_name, "basefiles") == 0) {
 609       bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
 610    } else {
 611       bstrncpy(sequence, table_name, sizeof(sequence));
 612       bstrncat(sequence, "_",        sizeof(sequence));
 613       bstrncat(sequence, table_name, sizeof(sequence));
 614       bstrncat(sequence, "id",       sizeof(sequence));
 615    }
 616
 617    bstrncat(sequence, "_seq", sizeof(sequence));
 618    bsnprintf(query, sizeof(query), "SELECT currval('%s')", sequence);
 619
 620    Dmsg1(500, "my_postgresql_currval invoked with '%s'\n", query);
 621    for (int i=0; i < 10; i++) {
 622       result = PQexec(mdb->db, query);
 623       if (result) {
 624          break;
 625       }
 626       bmicrosleep(5, 0);
 627    }
 628    if (!result) {
 629       Dmsg1(50, "Query failed: %s\n", query);
 630       goto bail_out;
 631    }
 632
 633    Dmsg0(500, "exec done");
 634
 635    if (PQresultStatus(result) == PGRES_TUPLES_OK) {
 636       Dmsg0(500, "getting value");
 637       id = atoi(PQgetvalue(result, 0, 0));
 638       Dmsg2(500, "got value '%s' which became %d\n", PQgetvalue(result, 0, 0), id);
 639    } else {
 640       Dmsg1(50, "Result status failed: %s\n", query);
 641       Mmsg1(&mdb->errmsg, _("error fetching currval: %s\n"), PQerrorMessage(mdb->db));
 642    }
 643
 644 bail_out:
 645    PQclear(result);
 646
 647    return id;
 648 }
 649
 650 #ifdef HAVE_BATCH_FILE_INSERT
 651
 652 int my_postgresql_batch_start(JCR *jcr, B_DB *mdb)
 653 {
 654    const char *query = "COPY batch FROM STDIN";
 655
 656    Dmsg0(500, "my_postgresql_batch_start started\n");
 657
 658    if (my_postgresql_query(mdb,
 659                            "CREATE TEMPORARY TABLE batch ("
 660                                "fileindex int,"
 661                                "jobid int,"
 662                                "path varchar,"
 663                                "name varchar,"
 664                                "lstat varchar,"
 665                                "md5 varchar)") == 1)
 666    {
 667       Dmsg0(500, "my_postgresql_batch_start failed\n");
 668       return 1;
 669    }
 670
 671    // We are starting a new query.  reset everything.
 672    mdb->num_rows     = -1;
 673    mdb->row_number   = -1;
 674    mdb->field_number = -1;
 675
 676    my_postgresql_free_result(mdb);
 677
 678    for (int i=0; i < 10; i++) {
 679       mdb->result = PQexec(mdb->db, query);
 680       if (mdb->result) {
 681          break;
 682       }
 683       bmicrosleep(5, 0);
 684    }
 685    if (!mdb->result) {
 686       Dmsg1(50, "Query failed: %s\n", query);
 687       goto bail_out;
 688    }
 689
 690    mdb->status = PQresultStatus(mdb->result);
 691    if (mdb->status == PGRES_COPY_IN) {
 692       // how many fields in the set?
 693       mdb->num_fields = (int) PQnfields(mdb->result);
 694       mdb->num_rows   = 0;
 695       mdb->status = 1;
 696    } else {
 697       Dmsg1(50, "Result status failed: %s\n", query);
 698       goto bail_out;
 699    }
 700
 701    Dmsg0(500, "my_postgresql_batch_start finishing\n");
 702
 703    return mdb->status;
 704
 705 bail_out:
 706    Mmsg1(&mdb->errmsg, _("error starting batch mode: %s"), PQerrorMessage(mdb->db));
 707    mdb->status = 0;
 708    PQclear(mdb->result);
 709    mdb->result = NULL;
 710    return mdb->status;
 711 }
 712
 713 /* set error to something to abort operation */
 714 int my_postgresql_batch_end(JCR *jcr, B_DB *mdb, const char *error)
 715 {
 716    int res;
 717    int count=30;
 718    PGresult *result;
 719    Dmsg0(500, "my_postgresql_batch_end started\n");
 720
 721    if (!mdb) {                  /* no files ? */
 722       return 0;
 723    }
 724
 725    do {
 726       res = PQputCopyEnd(mdb->db, error);
 727    } while (res == 0 && --count > 0);
 728
 729    if (res == 1) {
 730       Dmsg0(500, "ok\n");
 731       mdb->status = 1;
 732    }
 733
 734    if (res <= 0) {
 735       Dmsg0(500, "we failed\n");
 736       mdb->status = 0;
 737       Mmsg1(&mdb->errmsg, _("error ending batch mode: %s"), PQerrorMessage(mdb->db));
 738    }
 739
 740    /* Check command status and return to normal libpq state */
 741    result = PQgetResult(mdb->db);
 742    if (PQresultStatus(result) != PGRES_COMMAND_OK) {
 743       Mmsg1(&mdb->errmsg, _("error ending batch mode: %s"), PQerrorMessage(mdb->db));
 744       mdb->status = 0;
 745    }
 746    PQclear(result);
 747
 748    Dmsg0(500, "my_postgresql_batch_end finishing\n");
 749
 750    return mdb->status;
 751 }
 752
 753 int my_postgresql_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
 754 {
 755    int res;
 756    int count=30;
 757    size_t len;
 758    const char *digest;
 759    char ed1[50];
 760
 761    mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
 762    my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
 763
 764    mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
 765    my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
 766
 767    if (ar->Digest == NULL || ar->Digest[0] == 0) {
 768       digest = "0";
 769    } else {
 770       digest = ar->Digest;
 771    }
 772
 773    len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
 774               ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
 775               mdb->esc_name, ar->attr, digest);
 776
 777    do {
 778       res = PQputCopyData(mdb->db,
 779                           mdb->cmd,
 780                           len);
 781    } while (res == 0 && --count > 0);
 782
 783    if (res == 1) {
 784       Dmsg0(500, "ok\n");
 785       mdb->changes++;
 786       mdb->status = 1;
 787    }
 788
 789    if (res <= 0) {
 790       Dmsg0(500, "we failed\n");
 791       mdb->status = 0;
 792       Mmsg1(&mdb->errmsg, _("error copying in batch mode: %s"), PQerrorMessage(mdb->db));
 793    }
 794
 795    Dmsg0(500, "my_postgresql_batch_insert finishing\n");
 796
 797    return mdb->status;
 798 }
 799
 800 #endif /* HAVE_BATCH_FILE_INSERT */
 801
 802 /*
 803  * Escape strings so that PostgreSQL is happy on COPY
 804  *
 805  *   NOTE! len is the length of the old string. Your new
 806  *         string must be long enough (max 2*old+1) to hold
 807  *         the escaped output.
 808  */
 809 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
 810 {
 811    /* we have to escape \t, \n, \r, \ */
 812    char c = '\0' ;
 813
 814    while (len > 0 && *src) {
 815       switch (*src) {
 816       case '\n':
 817          c = 'n';
 818          break;
 819       case '\\':
 820          c = '\\';
 821          break;
 822       case '\t':
 823          c = 't';
 824          break;
 825       case '\r':
 826          c = 'r';
 827          break;
 828       default:
 829          c = '\0' ;
 830       }
 831
 832       if (c) {
 833          *dest = '\\';
 834          dest++;
 835          *dest = c;
 836       } else {
 837          *dest = *src;
 838       }
 839
 840       len--;
 841       src++;
 842       dest++;
 843    }
 844
 845    *dest = '\0';
 846    return dest;
 847 }
 848
 849 #ifdef HAVE_BATCH_FILE_INSERT
 850 const char *my_pg_batch_lock_path_query =
 851    "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE";
 852
 853
 854 const char *my_pg_batch_lock_filename_query =
 855    "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE";
 856
 857 const char *my_pg_batch_unlock_tables_query = "COMMIT";
 858
 859 const char *my_pg_batch_fill_path_query =
 860    "INSERT INTO Path (Path) "
 861     "SELECT a.Path FROM "
 862      "(SELECT DISTINCT Path FROM batch) AS a "
 863       "WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
 864
 865
 866 const char *my_pg_batch_fill_filename_query =
 867    "INSERT INTO Filename (Name) "
 868     "SELECT a.Name FROM "
 869      "(SELECT DISTINCT Name FROM batch) as a "
 870       "WHERE NOT EXISTS "
 871        "(SELECT Name FROM Filename WHERE Name = a.Name)";
 872 #endif /* HAVE_BATCH_FILE_INSERT */
 873
 874 #endif /* HAVE_POSTGRESQL */