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