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