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