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       if (!ret) {
 152          Mmsg(mdb->errmsg,
 153               _("Encoding error for database \"%s\". Wanted SQL_ASCII, got %s\n"),
 154               mdb->db_name, row[0]);
 155          Jmsg(jcr, M_WARNING, 0, "%s", mdb->errmsg);
 156          Dmsg1(50, "%s", mdb->errmsg);
 157       }
 158    }
 159    return ret;
 160 }
 161
 162 /*
 163  * Now actually open the database.  This can generate errors,
 164  *   which are returned in the errmsg
 165  *
 166  * DO NOT close the database or free(mdb) here !!!!
 167  */
 168 int
 169 db_open_database(JCR *jcr, B_DB *mdb)
 170 {
 171    int errstat;
 172    char buf[10], *port;
 173
 174 #ifdef xxx                      /* require libpq >= 8.2 */
 175    if (!PQisthreadsafe()) {
 176       Jmsg(jcr, M_ABORT, 0, _("PostgreSQL configuration problem. "
 177            "PostgreSQL library is not thread safe. Cannot continue.\n"));
 178    }
 179 #endif
 180    P(mutex);
 181    if (mdb->connected) {
 182       V(mutex);
 183       return 1;
 184    }
 185    mdb->connected = false;
 186
 187    if ((errstat=rwl_init(&mdb->lock)) != 0) {
 188       berrno be;
 189       Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
 190             be.bstrerror(errstat));
 191       V(mutex);
 192       return 0;
 193    }
 194
 195    if (mdb->db_port) {
 196       bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
 197       port = buf;
 198    } else {
 199       port = NULL;
 200    }
 201
 202    /* If connection fails, try at 5 sec intervals for 30 seconds. */
 203    for (int retry=0; retry < 6; retry++) {
 204       /* connect to the database */
 205       mdb->db = PQsetdbLogin(
 206            mdb->db_address,           /* default = localhost */
 207            port,                      /* default port */
 208            NULL,                      /* pg options */
 209            NULL,                      /* tty, ignored */
 210            mdb->db_name,              /* database name */
 211            mdb->db_user,              /* login name */
 212            mdb->db_password);         /* password */
 213
 214       /* If no connect, try once more in case it is a timing problem */
 215       if (PQstatus(mdb->db) == CONNECTION_OK) {
 216          break;
 217       }
 218       bmicrosleep(5, 0);
 219    }
 220
 221    Dmsg0(50, "pg_real_connect done\n");
 222    Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n", mdb->db_user, mdb->db_name,
 223             mdb->db_password==NULL?"(NULL)":mdb->db_password);
 224
 225    if (PQstatus(mdb->db) != CONNECTION_OK) {
 226       Mmsg2(&mdb->errmsg, _("Unable to connect to PostgreSQL server.\n"
 227             "Database=%s User=%s\n"
 228             "It is probably not running or your password is incorrect.\n"),
 229              mdb->db_name, mdb->db_user);
 230       V(mutex);
 231       return 0;
 232    }
 233
 234    mdb->connected = true;
 235
 236    if (!check_tables_version(jcr, mdb)) {
 237       V(mutex);
 238       return 0;
 239    }
 240
 241    sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
 242
 243    /* tell PostgreSQL we are using standard conforming strings
 244       and avoid warnings such as:
 245        WARNING:  nonstandard use of \\ in a string literal
 246    */
 247    sql_query(mdb, "set standard_conforming_strings=on");
 248
 249    /* check that encoding is SQL_ASCII */
 250    check_database_encoding(jcr, mdb);
 251
 252    V(mutex);
 253    return 1;
 254 }
 255
 256 void
 257 db_close_database(JCR *jcr, B_DB *mdb)
 258 {
 259    if (!mdb) {
 260       return;
 261    }
 262    db_end_transaction(jcr, mdb);
 263    P(mutex);
 264    sql_free_result(mdb);
 265    mdb->ref_count--;
 266    if (mdb->ref_count == 0) {
 267       qdchain(&mdb->bq);
 268       if (mdb->connected && mdb->db) {
 269          sql_close(mdb);
 270       }
 271       rwl_destroy(&mdb->lock);
 272       free_pool_memory(mdb->errmsg);
 273       free_pool_memory(mdb->cmd);
 274       free_pool_memory(mdb->cached_path);
 275       free_pool_memory(mdb->fname);
 276       free_pool_memory(mdb->path);
 277       free_pool_memory(mdb->esc_name);
 278       free_pool_memory(mdb->esc_path);
 279       if (mdb->db_name) {
 280          free(mdb->db_name);
 281       }
 282       if (mdb->db_user) {
 283          free(mdb->db_user);
 284       }
 285       if (mdb->db_password) {
 286          free(mdb->db_password);
 287       }
 288       if (mdb->db_address) {
 289          free(mdb->db_address);
 290       }
 291       if (mdb->db_socket) {
 292          free(mdb->db_socket);
 293       }
 294       free(mdb);
 295    }
 296    V(mutex);
 297 }
 298
 299 void db_thread_cleanup()
 300 { }
 301
 302 /*
 303  * Return the next unique index (auto-increment) for
 304  * the given table.  Return NULL on error.
 305  *
 306  * For PostgreSQL, NULL causes the auto-increment value
 307  *  to be updated.
 308  */
 309 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
 310 {
 311    strcpy(index, "NULL");
 312    return 1;
 313 }
 314
 315
 316 /*
 317  * Escape strings so that PostgreSQL is happy
 318  *
 319  *   NOTE! len is the length of the old string. Your new
 320  *         string must be long enough (max 2*old+1) to hold
 321  *         the escaped output.
 322  */
 323 void
 324 db_escape_string(JCR *jcr, B_DB *mdb, char *snew, char *old, int len)
 325 {
 326    int error;
 327
 328    PQescapeStringConn(mdb->db, snew, old, len, &error);
 329    if (error) {
 330       Jmsg(jcr, M_FATAL, 0, _("PQescapeStringConn returned non-zero.\n"));
 331       /* error on encoding, probably invalid multibyte encoding in the source string
 332         see PQescapeStringConn documentation for details. */
 333       Dmsg0(500, "PQescapeStringConn failed\n");
 334    }
 335 }
 336
 337 /*
 338  * Submit a general SQL command (cmd), and for each row returned,
 339  *  the sqlite_handler is called with the ctx.
 340  */
 341 bool db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
 342 {
 343    SQL_ROW row;
 344
 345    Dmsg0(500, "db_sql_query started\n");
 346
 347    db_lock(mdb);
 348    if (sql_query(mdb, query) != 0) {
 349       Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
 350       db_unlock(mdb);
 351       Dmsg0(500, "db_sql_query failed\n");
 352       return false;
 353    }
 354    Dmsg0(500, "db_sql_query succeeded. checking handler\n");
 355
 356    if (result_handler != NULL) {
 357       Dmsg0(500, "db_sql_query invoking handler\n");
 358       if ((mdb->result = sql_store_result(mdb)) != NULL) {
 359          int num_fields = sql_num_fields(mdb);
 360
 361          Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
 362          while ((row = sql_fetch_row(mdb)) != NULL) {
 363
 364             Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
 365             if (result_handler(ctx, num_fields, row))
 366                break;
 367          }
 368
 369         sql_free_result(mdb);
 370       }
 371    }
 372    db_unlock(mdb);
 373
 374    Dmsg0(500, "db_sql_query finished\n");
 375
 376    return true;
 377 }
 378
 379
 380
 381 POSTGRESQL_ROW my_postgresql_fetch_row(B_DB *mdb)
 382 {
 383    int j;
 384    POSTGRESQL_ROW row = NULL; // by default, return NULL
 385
 386    Dmsg0(500, "my_postgresql_fetch_row start\n");
 387
 388    if (!mdb->row || mdb->row_size < mdb->num_fields) {
 389       int num_fields = mdb->num_fields;
 390       Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
 391
 392       if (mdb->row) {
 393          Dmsg0(500, "my_postgresql_fetch_row freeing space\n");
 394          free(mdb->row);
 395       }
 396       num_fields += 20;                  /* add a bit extra */
 397       mdb->row = (POSTGRESQL_ROW)malloc(sizeof(char *) * num_fields);
 398       mdb->row_size = num_fields;
 399
 400       // now reset the row_number now that we have the space allocated
 401       mdb->row_number = 0;
 402    }
 403
 404    // if still within the result set
 405    if (mdb->row_number < mdb->num_rows) {
 406       Dmsg2(500, "my_postgresql_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
 407       // get each value from this row
 408       for (j = 0; j < mdb->num_fields; j++) {
 409          mdb->row[j] = PQgetvalue(mdb->result, mdb->row_number, j);
 410          Dmsg2(500, "my_postgresql_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
 411       }
 412       // increment the row number for the next call
 413       mdb->row_number++;
 414
 415       row = mdb->row;
 416    } else {
 417       Dmsg2(500, "my_postgresql_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
 418    }
 419
 420    Dmsg1(500, "my_postgresql_fetch_row finishes returning %p\n", row);
 421
 422    return row;
 423 }
 424
 425 int my_postgresql_max_length(B_DB *mdb, int field_num) {
 426    //
 427    // for a given column, find the max length
 428    //
 429    int max_length;
 430    int i;
 431    int this_length;
 432
 433    max_length = 0;
 434    for (i = 0; i < mdb->num_rows; i++) {
 435       if (PQgetisnull(mdb->result, i, field_num)) {
 436           this_length = 4;        // "NULL"
 437       } else {
 438           this_length = cstrlen(PQgetvalue(mdb->result, i, field_num));
 439       }
 440
 441       if (max_length < this_length) {
 442           max_length = this_length;
 443       }
 444    }
 445
 446    return max_length;
 447 }
 448
 449 POSTGRESQL_FIELD * my_postgresql_fetch_field(B_DB *mdb)
 450 {
 451    int     i;
 452
 453    Dmsg0(500, "my_postgresql_fetch_field starts\n");
 454
 455    if (!mdb->fields || mdb->fields_size < mdb->num_fields) {
 456       if (mdb->fields) {
 457          free(mdb->fields);
 458       }
 459       Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
 460       mdb->fields = (POSTGRESQL_FIELD *)malloc(sizeof(POSTGRESQL_FIELD) * mdb->num_fields);
 461       mdb->fields_size = mdb->num_fields;
 462
 463       for (i = 0; i < mdb->num_fields; i++) {
 464          Dmsg1(500, "filling field %d\n", i);
 465          mdb->fields[i].name           = PQfname(mdb->result, i);
 466          mdb->fields[i].max_length = my_postgresql_max_length(mdb, i);
 467          mdb->fields[i].type       = PQftype(mdb->result, i);
 468          mdb->fields[i].flags      = 0;
 469
 470          Dmsg4(500, "my_postgresql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
 471             mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
 472             mdb->fields[i].flags);
 473       } // end for
 474    } // end if
 475
 476    // increment field number for the next time around
 477
 478    Dmsg0(500, "my_postgresql_fetch_field finishes\n");
 479    return &mdb->fields[mdb->field_number++];
 480 }
 481
 482 void my_postgresql_data_seek(B_DB *mdb, int row)
 483 {
 484    // set the row number to be returned on the next call
 485    // to my_postgresql_fetch_row
 486    mdb->row_number = row;
 487 }
 488
 489 void my_postgresql_field_seek(B_DB *mdb, int field)
 490 {
 491    mdb->field_number = field;
 492 }
 493
 494 /*
 495  * Note, if this routine returns 1 (failure), Bacula expects
 496  *  that no result has been stored.
 497  * This is where QUERY_DB comes with Postgresql.
 498  *
 499  *  Returns:  0  on success
 500  *            1  on failure
 501  *
 502  */
 503 int my_postgresql_query(B_DB *mdb, const char *query)
 504 {
 505    Dmsg0(500, "my_postgresql_query started\n");
 506    // We are starting a new query.  reset everything.
 507    mdb->num_rows     = -1;
 508    mdb->row_number   = -1;
 509    mdb->field_number = -1;
 510
 511    if (mdb->result) {
 512       PQclear(mdb->result);  /* hmm, someone forgot to free?? */
 513       mdb->result = NULL;
 514    }
 515
 516    Dmsg1(500, "my_postgresql_query starts with '%s'\n", query);
 517
 518    for (int i=0; i < 10; i++) {
 519       mdb->result = PQexec(mdb->db, query);
 520       if (mdb->result) {
 521          break;
 522       }
 523       bmicrosleep(5, 0);
 524    }
 525    if (!mdb->result) {
 526       Dmsg1(50, "Query failed: %s\n", query);
 527       goto bail_out;
 528    }
 529
 530    mdb->status = PQresultStatus(mdb->result);
 531    if (mdb->status == PGRES_TUPLES_OK || mdb->status == PGRES_COMMAND_OK) {
 532       Dmsg1(500, "we have a result\n", query);
 533
 534       // how many fields in the set?
 535       mdb->num_fields = (int)PQnfields(mdb->result);
 536       Dmsg1(500, "we have %d fields\n", mdb->num_fields);
 537
 538       mdb->num_rows = PQntuples(mdb->result);
 539       Dmsg1(500, "we have %d rows\n", mdb->num_rows);
 540
 541       mdb->status = 0;                  /* succeed */
 542    } else {
 543       Dmsg1(50, "Result status failed: %s\n", query);
 544       goto bail_out;
 545    }
 546
 547    Dmsg0(500, "my_postgresql_query finishing\n");
 548    return mdb->status;
 549
 550 bail_out:
 551    Dmsg1(500, "we failed\n", query);
 552    PQclear(mdb->result);
 553    mdb->result = NULL;
 554    mdb->status = 1;                   /* failed */
 555    return mdb->status;
 556 }
 557
 558 void my_postgresql_free_result(B_DB *mdb)
 559 {
 560
 561    db_lock(mdb);
 562    if (mdb->result) {
 563       PQclear(mdb->result);
 564       mdb->result = NULL;
 565    }
 566
 567    if (mdb->row) {
 568       free(mdb->row);
 569       mdb->row = NULL;
 570    }
 571
 572    if (mdb->fields) {
 573       free(mdb->fields);
 574       mdb->fields = NULL;
 575    }
 576    db_unlock(mdb);
 577 }
 578
 579 int my_postgresql_currval(B_DB *mdb, const char *table_name)
 580 {
 581    // Obtain the current value of the sequence that
 582    // provides the serial value for primary key of the table.
 583
 584    // currval is local to our session.  It is not affected by
 585    // other transactions.
 586
 587    // Determine the name of the sequence.
 588    // PostgreSQL automatically creates a sequence using
 589    // <table>_<column>_seq.
 590    // At the time of writing, all tables used this format for
 591    // for their primary key: <table>id
 592    // Except for basefiles which has a primary key on baseid.
 593    // Therefore, we need to special case that one table.
 594
 595    // everything else can use the PostgreSQL formula.
 596
 597    char      sequence[NAMEDATALEN-1];
 598    char      query   [NAMEDATALEN+50];
 599    PGresult *result;
 600    int       id = 0;
 601
 602    if (strcasecmp(table_name, "basefiles") == 0) {
 603       bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
 604    } else {
 605       bstrncpy(sequence, table_name, sizeof(sequence));
 606       bstrncat(sequence, "_",        sizeof(sequence));
 607       bstrncat(sequence, table_name, sizeof(sequence));
 608       bstrncat(sequence, "id",       sizeof(sequence));
 609    }
 610
 611    bstrncat(sequence, "_seq", sizeof(sequence));
 612    bsnprintf(query, sizeof(query), "SELECT currval('%s')", sequence);
 613
 614    Dmsg1(500, "my_postgresql_currval invoked with '%s'\n", query);
 615    for (int i=0; i < 10; i++) {
 616       result = PQexec(mdb->db, query);
 617       if (result) {
 618          break;
 619       }
 620       bmicrosleep(5, 0);
 621    }
 622    if (!result) {
 623       Dmsg1(50, "Query failed: %s\n", query);
 624       goto bail_out;
 625    }
 626
 627    Dmsg0(500, "exec done");
 628
 629    if (PQresultStatus(result) == PGRES_TUPLES_OK) {
 630       Dmsg0(500, "getting value");
 631       id = atoi(PQgetvalue(result, 0, 0));
 632       Dmsg2(500, "got value '%s' which became %d\n", PQgetvalue(result, 0, 0), id);
 633    } else {
 634       Dmsg1(50, "Result status failed: %s\n", query);
 635       Mmsg1(&mdb->errmsg, _("error fetching currval: %s\n"), PQerrorMessage(mdb->db));
 636    }
 637
 638 bail_out:
 639    PQclear(result);
 640
 641    return id;
 642 }
 643
 644 #ifdef HAVE_BATCH_FILE_INSERT
 645
 646 int my_postgresql_batch_start(JCR *jcr, B_DB *mdb)
 647 {
 648    const char *query = "COPY batch FROM STDIN";
 649
 650    Dmsg0(500, "my_postgresql_batch_start started\n");
 651
 652    if (my_postgresql_query(mdb,
 653                            "CREATE TEMPORARY TABLE batch ("
 654                                "fileindex int,"
 655                                "jobid int,"
 656                                "path varchar,"
 657                                "name varchar,"
 658                                "lstat varchar,"
 659                                "md5 varchar)") == 1)
 660    {
 661       Dmsg0(500, "my_postgresql_batch_start failed\n");
 662       return 1;
 663    }
 664
 665    // We are starting a new query.  reset everything.
 666    mdb->num_rows     = -1;
 667    mdb->row_number   = -1;
 668    mdb->field_number = -1;
 669
 670    my_postgresql_free_result(mdb);
 671
 672    for (int i=0; i < 10; i++) {
 673       mdb->result = PQexec(mdb->db, query);
 674       if (mdb->result) {
 675          break;
 676       }
 677       bmicrosleep(5, 0);
 678    }
 679    if (!mdb->result) {
 680       Dmsg1(50, "Query failed: %s\n", query);
 681       goto bail_out;
 682    }
 683
 684    mdb->status = PQresultStatus(mdb->result);
 685    if (mdb->status == PGRES_COPY_IN) {
 686       // how many fields in the set?
 687       mdb->num_fields = (int) PQnfields(mdb->result);
 688       mdb->num_rows   = 0;
 689       mdb->status = 1;
 690    } else {
 691       Dmsg1(50, "Result status failed: %s\n", query);
 692       goto bail_out;
 693    }
 694
 695    Dmsg0(500, "my_postgresql_batch_start finishing\n");
 696
 697    return mdb->status;
 698
 699 bail_out:
 700    Mmsg1(&mdb->errmsg, _("error starting batch mode: %s"), PQerrorMessage(mdb->db));
 701    mdb->status = 0;
 702    PQclear(mdb->result);
 703    mdb->result = NULL;
 704    return mdb->status;
 705 }
 706
 707 /* set error to something to abort operation */
 708 int my_postgresql_batch_end(JCR *jcr, B_DB *mdb, const char *error)
 709 {
 710    int res;
 711    int count=30;
 712    Dmsg0(500, "my_postgresql_batch_end started\n");
 713
 714    if (!mdb) {                  /* no files ? */
 715       return 0;
 716    }
 717
 718    do {
 719       res = PQputCopyEnd(mdb->db, error);
 720    } while (res == 0 && --count > 0);
 721
 722    if (res == 1) {
 723       Dmsg0(500, "ok\n");
 724       mdb->status = 1;
 725    }
 726
 727    if (res <= 0) {
 728       Dmsg0(500, "we failed\n");
 729       mdb->status = 0;
 730       Mmsg1(&mdb->errmsg, _("error ending batch mode: %s"), PQerrorMessage(mdb->db));
 731    }
 732
 733    Dmsg0(500, "my_postgresql_batch_end finishing\n");
 734
 735    return mdb->status;
 736 }
 737
 738 int my_postgresql_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
 739 {
 740    int res;
 741    int count=30;
 742    size_t len;
 743    const char *digest;
 744    char ed1[50];
 745
 746    mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
 747    my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
 748
 749    mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
 750    my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
 751
 752    if (ar->Digest == NULL || ar->Digest[0] == 0) {
 753       digest = "0";
 754    } else {
 755       digest = ar->Digest;
 756    }
 757
 758    len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
 759               ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
 760               mdb->esc_name, ar->attr, digest);
 761
 762    do {
 763       res = PQputCopyData(mdb->db,
 764                           mdb->cmd,
 765                           len);
 766    } while (res == 0 && --count > 0);
 767
 768    if (res == 1) {
 769       Dmsg0(500, "ok\n");
 770       mdb->changes++;
 771       mdb->status = 1;
 772    }
 773
 774    if (res <= 0) {
 775       Dmsg0(500, "we failed\n");
 776       mdb->status = 0;
 777       Mmsg1(&mdb->errmsg, _("error ending batch mode: %s"), PQerrorMessage(mdb->db));
 778    }
 779
 780    Dmsg0(500, "my_postgresql_batch_insert finishing\n");
 781
 782    return mdb->status;
 783 }
 784
 785 #endif /* HAVE_BATCH_FILE_INSERT */
 786
 787 /*
 788  * Escape strings so that PostgreSQL is happy on COPY
 789  *
 790  *   NOTE! len is the length of the old string. Your new
 791  *         string must be long enough (max 2*old+1) to hold
 792  *         the escaped output.
 793  */
 794 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
 795 {
 796    /* we have to escape \t, \n, \r, \ */
 797    char c = '\0' ;
 798
 799    while (len > 0 && *src) {
 800       switch (*src) {
 801       case '\n':
 802          c = 'n';
 803          break;
 804       case '\\':
 805          c = '\\';
 806          break;
 807       case '\t':
 808          c = 't';
 809          break;
 810       case '\r':
 811          c = 'r';
 812          break;
 813       default:
 814          c = '\0' ;
 815       }
 816
 817       if (c) {
 818          *dest = '\\';
 819          dest++;
 820          *dest = c;
 821       } else {
 822          *dest = *src;
 823       }
 824
 825       len--;
 826       src++;
 827       dest++;
 828    }
 829
 830    *dest = '\0';
 831    return dest;
 832 }
 833
 834 #ifdef HAVE_BATCH_FILE_INSERT
 835 const char *my_pg_batch_lock_path_query =
 836    "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE";
 837
 838
 839 const char *my_pg_batch_lock_filename_query =
 840    "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE";
 841
 842 const char *my_pg_batch_unlock_tables_query = "COMMIT";
 843
 844 const char *my_pg_batch_fill_path_query =
 845    "INSERT INTO Path (Path) "
 846     "SELECT a.Path FROM "
 847      "(SELECT DISTINCT Path FROM batch) AS a "
 848       "WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
 849
 850
 851 const char *my_pg_batch_fill_filename_query =
 852    "INSERT INTO Filename (Name) "
 853     "SELECT a.Name FROM "
 854      "(SELECT DISTINCT Name FROM batch) as a "
 855       "WHERE NOT EXISTS "
 856        "(SELECT Name FROM Filename WHERE Name = a.Name)";
 857 #endif /* HAVE_BATCH_FILE_INSERT */
 858
 859 #endif /* HAVE_POSTGRESQL */