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