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 plus additions
  11    that are listed 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 John Walker.
  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
  51 /* -----------------------------------------------------------------------
  52  *
  53  *   PostgreSQL dependent defines and subroutines
  54  *
  55  * -----------------------------------------------------------------------
  56  */
  57
  58 /* List of open databases */
  59 static BQUEUE db_list = {&db_list, &db_list};
  60
  61 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  62
  63 /*
  64  * Retrieve database type
  65  */
  66 const char *
  67 db_get_type(void)
  68 {
  69    return "PostgreSQL";
  70
  71 }
  72
  73 /*
  74  * Initialize database data structure. In principal this should
  75  * never have errors, or it is really fatal.
  76  */
  77 B_DB *
  78 db_init_database(JCR *jcr, const char *db_name, const char *db_user, const char *db_password,
  79                  const char *db_address, int db_port, const char *db_socket,
  80                  int mult_db_connections)
  81 {
  82    B_DB *mdb;
  83
  84    if (!db_user) {
  85       Jmsg(jcr, M_FATAL, 0, _("A user name for PostgreSQL must be supplied.\n"));
  86       return NULL;
  87    }
  88    P(mutex);                          /* lock DB queue */
  89    if (!mult_db_connections) {
  90       /* Look to see if DB already open */
  91       for (mdb=NULL; (mdb=(B_DB *)qnext(&db_list, &mdb->bq)); ) {
  92          if (bstrcmp(mdb->db_name, db_name) &&
  93              bstrcmp(mdb->db_address, db_address) &&
  94              mdb->db_port == db_port) {
  95             Dmsg2(100, "DB REopen %d %s\n", mdb->ref_count, db_name);
  96             mdb->ref_count++;
  97             V(mutex);
  98             return mdb;                  /* already open */
  99          }
 100       }
 101    }
 102    Dmsg0(100, "db_open first time\n");
 103    mdb = (B_DB *)malloc(sizeof(B_DB));
 104    memset(mdb, 0, sizeof(B_DB));
 105    mdb->db_name = bstrdup(db_name);
 106    mdb->db_user = bstrdup(db_user);
 107    if (db_password) {
 108       mdb->db_password = bstrdup(db_password);
 109    }
 110    if (db_address) {
 111       mdb->db_address  = bstrdup(db_address);
 112    }
 113    if (db_socket) {
 114       mdb->db_socket   = bstrdup(db_socket);
 115    }
 116    mdb->db_port        = db_port;
 117    mdb->have_insert_id = TRUE;
 118    mdb->errmsg         = get_pool_memory(PM_EMSG); /* get error message buffer */
 119    *mdb->errmsg        = 0;
 120    mdb->cmd            = get_pool_memory(PM_EMSG); /* get command buffer */
 121    mdb->cached_path    = get_pool_memory(PM_FNAME);
 122    mdb->cached_path_id = 0;
 123    mdb->ref_count      = 1;
 124    mdb->fname          = get_pool_memory(PM_FNAME);
 125    mdb->path           = get_pool_memory(PM_FNAME);
 126    mdb->esc_name       = get_pool_memory(PM_FNAME);
 127    mdb->esc_path      = get_pool_memory(PM_FNAME);
 128    mdb->allow_transactions = mult_db_connections;
 129    qinsert(&db_list, &mdb->bq);            /* put db in list */
 130    V(mutex);
 131    return mdb;
 132 }
 133
 134 /*
 135  * Now actually open the database.  This can generate errors,
 136  *   which are returned in the errmsg
 137  *
 138  * DO NOT close the database or free(mdb) here !!!!
 139  */
 140 int
 141 db_open_database(JCR *jcr, B_DB *mdb)
 142 {
 143    int errstat;
 144    char buf[10], *port;
 145
 146    P(mutex);
 147    if (mdb->connected) {
 148       V(mutex);
 149       return 1;
 150    }
 151    mdb->connected = false;
 152
 153    if ((errstat=rwl_init(&mdb->lock)) != 0) {
 154       Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
 155             strerror(errstat));
 156       V(mutex);
 157       return 0;
 158    }
 159
 160    if (mdb->db_port) {
 161       bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
 162       port = buf;
 163    } else {
 164       port = NULL;
 165    }
 166
 167    /* If connection fails, try at 5 sec intervals for 30 seconds. */
 168    for (int retry=0; retry < 6; retry++) {
 169       /* connect to the database */
 170       mdb->db = PQsetdbLogin(
 171            mdb->db_address,           /* default = localhost */
 172            port,                      /* default port */
 173            NULL,                      /* pg options */
 174            NULL,                      /* tty, ignored */
 175            mdb->db_name,              /* database name */
 176            mdb->db_user,              /* login name */
 177            mdb->db_password);         /* password */
 178
 179       /* If no connect, try once more in case it is a timing problem */
 180       if (PQstatus(mdb->db) == CONNECTION_OK) {
 181          break;
 182       }
 183       bmicrosleep(5, 0);
 184    }
 185
 186    Dmsg0(50, "pg_real_connect done\n");
 187    Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n", mdb->db_user, mdb->db_name,
 188             mdb->db_password==NULL?"(NULL)":mdb->db_password);
 189
 190    if (PQstatus(mdb->db) != CONNECTION_OK) {
 191       Mmsg2(&mdb->errmsg, _("Unable to connect to PostgreSQL server.\n"
 192             "Database=%s User=%s\n"
 193             "It is probably not running or your password is incorrect.\n"),
 194              mdb->db_name, mdb->db_user);
 195       V(mutex);
 196       return 0;
 197    }
 198
 199    mdb->connected = true;
 200
 201    if (!check_tables_version(jcr, mdb)) {
 202       V(mutex);
 203       return 0;
 204    }
 205
 206    sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
 207
 208    V(mutex);
 209    return 1;
 210 }
 211
 212 void
 213 db_close_database(JCR *jcr, B_DB *mdb)
 214 {
 215    if (!mdb) {
 216       return;
 217    }
 218    db_end_transaction(jcr, mdb);
 219    P(mutex);
 220    sql_free_result(mdb);
 221    mdb->ref_count--;
 222    if (mdb->ref_count == 0) {
 223       qdchain(&mdb->bq);
 224       if (mdb->connected && mdb->db) {
 225          sql_close(mdb);
 226       }
 227       rwl_destroy(&mdb->lock);
 228       free_pool_memory(mdb->errmsg);
 229       free_pool_memory(mdb->cmd);
 230       free_pool_memory(mdb->cached_path);
 231       free_pool_memory(mdb->fname);
 232       free_pool_memory(mdb->path);
 233       free_pool_memory(mdb->esc_name);
 234       free_pool_memory(mdb->esc_path);
 235       if (mdb->db_name) {
 236          free(mdb->db_name);
 237       }
 238       if (mdb->db_user) {
 239          free(mdb->db_user);
 240       }
 241       if (mdb->db_password) {
 242          free(mdb->db_password);
 243       }
 244       if (mdb->db_address) {
 245          free(mdb->db_address);
 246       }
 247       if (mdb->db_socket) {
 248          free(mdb->db_socket);
 249       }
 250       my_postgresql_free_result(mdb);
 251       free(mdb);
 252    }
 253    V(mutex);
 254 }
 255
 256 void db_thread_cleanup()
 257 { }
 258
 259 /*
 260  * Return the next unique index (auto-increment) for
 261  * the given table.  Return NULL on error.
 262  *
 263  * For PostgreSQL, NULL causes the auto-increment value
 264  *  to be updated.
 265  */
 266 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
 267 {
 268    strcpy(index, "NULL");
 269    return 1;
 270 }
 271
 272
 273 /*
 274  * Escape strings so that PostgreSQL is happy
 275  *
 276  *   NOTE! len is the length of the old string. Your new
 277  *         string must be long enough (max 2*old+1) to hold
 278  *         the escaped output.
 279  */
 280 void
 281 db_escape_string(char *snew, char *old, int len)
 282 {
 283    PQescapeString(snew, old, len);
 284 }
 285
 286 /*
 287  * Submit a general SQL command (cmd), and for each row returned,
 288  *  the sqlite_handler is called with the ctx.
 289  */
 290 int db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
 291 {
 292    SQL_ROW row;
 293
 294    Dmsg0(500, "db_sql_query started\n");
 295
 296    db_lock(mdb);
 297    if (sql_query(mdb, query) != 0) {
 298       Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
 299       db_unlock(mdb);
 300       Dmsg0(500, "db_sql_query failed\n");
 301       return 0;
 302    }
 303    Dmsg0(500, "db_sql_query succeeded. checking handler\n");
 304
 305    if (result_handler != NULL) {
 306       Dmsg0(500, "db_sql_query invoking handler\n");
 307       if ((mdb->result = sql_store_result(mdb)) != NULL) {
 308          int num_fields = sql_num_fields(mdb);
 309
 310          Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
 311          while ((row = sql_fetch_row(mdb)) != NULL) {
 312
 313             Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
 314             if (result_handler(ctx, num_fields, row))
 315                break;
 316          }
 317
 318         sql_free_result(mdb);
 319       }
 320    }
 321    db_unlock(mdb);
 322
 323    Dmsg0(500, "db_sql_query finished\n");
 324
 325    return 1;
 326 }
 327
 328
 329
 330 POSTGRESQL_ROW my_postgresql_fetch_row(B_DB *mdb)
 331 {
 332    int j;
 333    POSTGRESQL_ROW row = NULL; // by default, return NULL
 334
 335    Dmsg0(500, "my_postgresql_fetch_row start\n");
 336
 337    if (mdb->row_number == -1 || mdb->row == NULL) {
 338       Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
 339
 340       if (mdb->row != NULL) {
 341          Dmsg0(500, "my_postgresql_fetch_row freeing space\n");
 342          free(mdb->row);
 343          mdb->row = NULL;
 344       }
 345
 346       mdb->row = (POSTGRESQL_ROW) malloc(sizeof(char *) * mdb->num_fields);
 347
 348       // now reset the row_number now that we have the space allocated
 349       mdb->row_number = 0;
 350    }
 351
 352    // if still within the result set
 353    if (mdb->row_number < mdb->num_rows) {
 354       Dmsg2(500, "my_postgresql_fetch_row row number '%d' is acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
 355       // get each value from this row
 356       for (j = 0; j < mdb->num_fields; j++) {
 357          mdb->row[j] = PQgetvalue(mdb->result, mdb->row_number, j);
 358          Dmsg2(500, "my_postgresql_fetch_row field '%d' has value '%s'\n", j, mdb->row[j]);
 359       }
 360       // increment the row number for the next call
 361       mdb->row_number++;
 362
 363       row = mdb->row;
 364    } else {
 365       Dmsg2(500, "my_postgresql_fetch_row row number '%d' is NOT acceptable (0..%d)\n", mdb->row_number, mdb->num_rows);
 366    }
 367
 368    Dmsg1(500, "my_postgresql_fetch_row finishes returning %x\n", row);
 369
 370    return row;
 371 }
 372
 373 int my_postgresql_max_length(B_DB *mdb, int field_num) {
 374    //
 375    // for a given column, find the max length
 376    //
 377    int max_length;
 378    int i;
 379    int this_length;
 380
 381    max_length = 0;
 382    for (i = 0; i < mdb->num_rows; i++) {
 383       if (PQgetisnull(mdb->result, i, field_num)) {
 384           this_length = 4;        // "NULL"
 385       } else {
 386           this_length = cstrlen(PQgetvalue(mdb->result, i, field_num));
 387       }
 388
 389       if (max_length < this_length) {
 390           max_length = this_length;
 391       }
 392    }
 393
 394    return max_length;
 395 }
 396
 397 POSTGRESQL_FIELD * my_postgresql_fetch_field(B_DB *mdb)
 398 {
 399    int     i;
 400
 401    Dmsg0(500, "my_postgresql_fetch_field starts\n");
 402    if (mdb->fields == NULL) {
 403       Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
 404       mdb->fields = (POSTGRESQL_FIELD *)malloc(sizeof(POSTGRESQL_FIELD) * mdb->num_fields);
 405
 406       for (i = 0; i < mdb->num_fields; i++) {
 407          Dmsg1(500, "filling field %d\n", i);
 408          mdb->fields[i].name           = PQfname(mdb->result, i);
 409          mdb->fields[i].max_length = my_postgresql_max_length(mdb, i);
 410          mdb->fields[i].type       = PQftype(mdb->result, i);
 411          mdb->fields[i].flags      = 0;
 412
 413          Dmsg4(500, "my_postgresql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
 414             mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
 415             mdb->fields[i].flags);
 416       } // end for
 417    } // end if
 418
 419    // increment field number for the next time around
 420
 421    Dmsg0(500, "my_postgresql_fetch_field finishes\n");
 422    return &mdb->fields[mdb->field_number++];
 423 }
 424
 425 void my_postgresql_data_seek(B_DB *mdb, int row)
 426 {
 427    // set the row number to be returned on the next call
 428    // to my_postgresql_fetch_row
 429    mdb->row_number = row;
 430 }
 431
 432 void my_postgresql_field_seek(B_DB *mdb, int field)
 433 {
 434    mdb->field_number = field;
 435 }
 436
 437 /*
 438  * Note, if this routine returns 1 (failure), Bacula expects
 439  *  that no result has been stored.
 440  * This is where QUERY_DB comes with Postgresql.
 441  *
 442  *  Returns:  0  on success
 443  *            1  on failure
 444  *
 445  */
 446 int my_postgresql_query(B_DB *mdb, const char *query)
 447 {
 448    Dmsg0(500, "my_postgresql_query started\n");
 449    // We are starting a new query.  reset everything.
 450    mdb->num_rows     = -1;
 451    mdb->row_number   = -1;
 452    mdb->field_number = -1;
 453
 454    if (mdb->result) {
 455       PQclear(mdb->result);  /* hmm, someone forgot to free?? */
 456       mdb->result = NULL;
 457    }
 458
 459    Dmsg1(500, "my_postgresql_query starts with '%s'\n", query);
 460
 461    for (int i=0; i < 10; i++) {
 462       mdb->result = PQexec(mdb->db, query);
 463       if (mdb->result) {
 464          break;
 465       }
 466       bmicrosleep(5, 0);
 467    }
 468    if (!mdb->result) {
 469       goto bail_out;
 470    }
 471
 472    mdb->status = PQresultStatus(mdb->result);
 473    if (mdb->status == PGRES_TUPLES_OK || mdb->status == PGRES_COMMAND_OK) {
 474       Dmsg1(500, "we have a result\n", query);
 475
 476       // how many fields in the set?
 477       mdb->num_fields = (int)PQnfields(mdb->result);
 478       Dmsg1(500, "we have %d fields\n", mdb->num_fields);
 479
 480       mdb->num_rows = PQntuples(mdb->result);
 481       Dmsg1(500, "we have %d rows\n", mdb->num_rows);
 482
 483       mdb->status = 0;                  /* succeed */
 484    } else {
 485       goto bail_out;
 486    }
 487
 488    Dmsg0(500, "my_postgresql_query finishing\n");
 489    return mdb->status;
 490
 491 bail_out:
 492    Dmsg1(500, "we failed\n", query);
 493    PQclear(mdb->result);
 494    mdb->result = NULL;
 495    mdb->status = 1;                   /* failed */
 496    return mdb->status;
 497 }
 498
 499 void my_postgresql_free_result(B_DB *mdb)
 500 {
 501    if (mdb->result) {
 502       PQclear(mdb->result);
 503       mdb->result = NULL;
 504    }
 505
 506    if (mdb->row) {
 507       free(mdb->row);
 508       mdb->row = NULL;
 509    }
 510
 511    if (mdb->fields) {
 512       free(mdb->fields);
 513       mdb->fields = NULL;
 514    }
 515 }
 516
 517 int my_postgresql_currval(B_DB *mdb, char *table_name)
 518 {
 519    // Obtain the current value of the sequence that
 520    // provides the serial value for primary key of the table.
 521
 522    // currval is local to our session.  It is not affected by
 523    // other transactions.
 524
 525    // Determine the name of the sequence.
 526    // PostgreSQL automatically creates a sequence using
 527    // <table>_<column>_seq.
 528    // At the time of writing, all tables used this format for
 529    // for their primary key: <table>id
 530    // Except for basefiles which has a primary key on baseid.
 531    // Therefore, we need to special case that one table.
 532
 533    // everything else can use the PostgreSQL formula.
 534
 535    char      sequence[NAMEDATALEN-1];
 536    char      query   [NAMEDATALEN+50];
 537    PGresult *result;
 538    int       id = 0;
 539
 540    if (strcasecmp(table_name, "basefiles") == 0) {
 541       bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
 542    } else {
 543       bstrncpy(sequence, table_name, sizeof(sequence));
 544       bstrncat(sequence, "_",        sizeof(sequence));
 545       bstrncat(sequence, table_name, sizeof(sequence));
 546       bstrncat(sequence, "id",       sizeof(sequence));
 547    }
 548
 549    bstrncat(sequence, "_seq", sizeof(sequence));
 550    bsnprintf(query, sizeof(query), "SELECT currval('%s')", sequence);
 551
 552 // Mmsg(query, "SELECT currval('%s')", sequence);
 553    Dmsg1(500, "my_postgresql_currval invoked with '%s'\n", query);
 554    result = PQexec(mdb->db, query);
 555
 556    Dmsg0(500, "exec done");
 557
 558    if (PQresultStatus(result) == PGRES_TUPLES_OK) {
 559       Dmsg0(500, "getting value");
 560       id = atoi(PQgetvalue(result, 0, 0));
 561       Dmsg2(500, "got value '%s' which became %d\n", PQgetvalue(result, 0, 0), id);
 562    } else {
 563       Mmsg1(&mdb->errmsg, _("error fetching currval: %s\n"), PQerrorMessage(mdb->db));
 564    }
 565
 566    PQclear(result);
 567
 568    return id;
 569 }
 570
 571 int my_postgresql_batch_start(JCR *jcr, B_DB *mdb)
 572 {
 573    Dmsg0(500, "my_postgresql_batch_start started\n");
 574
 575    if (my_postgresql_query(mdb,
 576                            " CREATE TEMPORARY TABLE batch "
 577                            "        (fileindex int,       "
 578                            "        jobid int,            "
 579                            "        path varchar,         "
 580                            "        name varchar,         "
 581                            "        lstat varchar,        "
 582                            "        md5 varchar)") == 1)
 583    {
 584       Dmsg0(500, "my_postgresql_batch_start failed\n");
 585       return 1;
 586    }
 587
 588    // We are starting a new query.  reset everything.
 589    mdb->num_rows     = -1;
 590    mdb->row_number   = -1;
 591    mdb->field_number = -1;
 592
 593    if (mdb->result != NULL) {
 594       my_postgresql_free_result(mdb);
 595    }
 596
 597    mdb->result = PQexec(mdb->db, "COPY batch FROM STDIN");
 598    mdb->status = PQresultStatus(mdb->result);
 599    if (mdb->status == PGRES_COPY_IN) {
 600       // how many fields in the set?
 601       mdb->num_fields = (int) PQnfields(mdb->result);
 602       mdb->num_rows   = 0;
 603       mdb->status = 1;
 604    } else {
 605       Dmsg0(500, "we failed\n");
 606       mdb->status = 0;
 607    }
 608
 609    Dmsg0(500, "my_postgresql_batch_start finishing\n");
 610
 611    return mdb->status;
 612 }
 613
 614 /* set error to something to abort operation */
 615 int my_postgresql_batch_end(JCR *jcr, B_DB *mdb, const char *error)
 616 {
 617    int res;
 618    int count=30;
 619    Dmsg0(500, "my_postgresql_batch_end started\n");
 620
 621    if (!mdb) {                  /* no files ? */
 622       return 0;
 623    }
 624
 625    do {
 626       res = PQputCopyEnd(mdb->db, error);
 627    } while (res == 0 && --count > 0);
 628
 629    if (res == 1) {
 630       Dmsg0(500, "ok\n");
 631       mdb->status = 1;
 632    }
 633
 634    if (res <= 0) {
 635       Dmsg0(500, "we failed\n");
 636       mdb->status = 0;
 637       Mmsg1(&mdb->errmsg, _("error ending batch mode: %s\n"), PQerrorMessage(mdb->db));
 638    }
 639
 640    Dmsg0(500, "my_postgresql_batch_end finishing\n");
 641
 642    return mdb->status;
 643 }
 644
 645 int my_postgresql_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
 646 {
 647    int res;
 648    int count=30;
 649    size_t len;
 650    char *digest;
 651    char ed1[50];
 652
 653    mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
 654    my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
 655
 656    mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
 657    my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
 658
 659    if (ar->Digest == NULL || ar->Digest[0] == 0) {
 660       digest = "0";
 661    } else {
 662       digest = ar->Digest;
 663    }
 664
 665    len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
 666               ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
 667               mdb->esc_name, ar->attr, digest);
 668
 669    do {
 670       res = PQputCopyData(mdb->db,
 671                           mdb->cmd,
 672                           len);
 673    } while (res == 0 && --count > 0);
 674
 675    if (res == 1) {
 676       Dmsg0(500, "ok\n");
 677       mdb->changes++;
 678       mdb->status = 1;
 679    }
 680
 681    if (res <= 0) {
 682       Dmsg0(500, "we failed\n");
 683       mdb->status = 0;
 684       Mmsg1(&mdb->errmsg, _("error ending batch mode: %s\n"), PQerrorMessage(mdb->db));
 685    }
 686
 687    Dmsg0(500, "my_postgresql_batch_insert finishing\n");
 688
 689    return mdb->status;
 690 }
 691
 692 /*
 693  * Escape strings so that PostgreSQL is happy on COPY
 694  *
 695  *   NOTE! len is the length of the old string. Your new
 696  *         string must be long enough (max 2*old+1) to hold
 697  *         the escaped output.
 698  */
 699 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
 700 {
 701    /* we have to escape \t, \n, \r, \ */
 702    char c = '\0' ;
 703
 704    while (len > 0 && *src) {
 705       switch (*src) {
 706       case '\n':
 707          c = 'n';
 708          break;
 709       case '\\':
 710          c = '\\';
 711          break;
 712       case '\t':
 713          c = 't';
 714          break;
 715       case '\r':
 716          c = 'r';
 717          break;
 718       default:
 719          c = '\0' ;
 720       }
 721
 722       if (c) {
 723          *dest = '\\';
 724          dest++;
 725          *dest = c;
 726       } else {
 727          *dest = *src;
 728       }
 729
 730       len--;
 731       src++;
 732       dest++;
 733    }
 734
 735    *dest = '\0';
 736    return dest;
 737 }
 738
 739 char *my_pg_batch_lock_path_query = "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE";
 740
 741
 742 char *my_pg_batch_lock_filename_query = "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE";
 743
 744 char *my_pg_batch_unlock_tables_query = "COMMIT";
 745
 746 char *my_pg_batch_fill_path_query = "INSERT INTO Path (Path)                                    "
 747                                     "  SELECT a.Path FROM                                       "
 748                                     "      (SELECT DISTINCT Path FROM batch) AS a               "
 749                                     "  WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ";
 750
 751
 752 char *my_pg_batch_fill_filename_query = "INSERT INTO Filename (Name)        "
 753                                         "  SELECT a.Name FROM               "
 754                                         "    (SELECT DISTINCT Name FROM batch) as a "
 755                                         "    WHERE NOT EXISTS               "
 756                                         "      (SELECT Name FROM Filename WHERE Name = a.Name)";
 757 #endif /* HAVE_POSTGRESQL */