git.sur5r.net Git - bacula/bacula/blob - bacula/src/cats/dbi.c

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2003-2010 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 Kern Sibbald.
  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 DBI
  30  *   These are DBI specific routines
  31  *
  32  *    João Henrique Freitas, December 2007
  33  *    based upon work done by Dan Langille, December 2003 and
  34  *    by Kern Sibbald, March 2000
  35  *
  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_DBI
  48
  49 /* -----------------------------------------------------------------------
  50  *
  51  *   DBI dependent defines and subroutines
  52  *
  53  * -----------------------------------------------------------------------
  54  */
  55
  56 /* List of open databases */
  57 static dlist *db_list = NULL;
  58
  59 /* Control allocated fields by my_dbi_getvalue */
  60 static dlist *dbi_getvalue_list = NULL;
  61
  62 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  63
  64 /*
  65  * Retrieve database type
  66  */
  67 const char *
  68 db_get_type(void)
  69 {
  70    return "DBI";
  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 = NULL;
  83    DBI_FIELD_GET *field;
  84    char db_driver[10];
  85    char db_driverdir[256];
  86
  87    /* Constraint the db_driver */
  88    if(db_type  == -1) {
  89       Jmsg(jcr, M_FATAL, 0, _("A dbi driver for DBI must be supplied.\n"));
  90       return NULL;
  91    }
  92
  93    /* Do the correct selection of driver.
  94     * Can be one of the varius supported by libdbi
  95     */
  96    switch (db_type) {
  97    case SQL_TYPE_MYSQL:
  98       bstrncpy(db_driver,"mysql", sizeof(db_driver));
  99       break;
 100    case SQL_TYPE_POSTGRESQL:
 101       bstrncpy(db_driver,"pgsql", sizeof(db_driver));
 102       break;
 103    case SQL_TYPE_SQLITE:
 104       bstrncpy(db_driver,"sqlite", sizeof(db_driver));
 105       break;
 106    case SQL_TYPE_SQLITE3:
 107       bstrncpy(db_driver,"sqlite3", sizeof(db_driver));
 108       break;
 109    }
 110
 111    /* Set db_driverdir whereis is the libdbi drivers */
 112    bstrncpy(db_driverdir, DBI_DRIVER_DIR, 255);
 113
 114    if (!db_user) {
 115       Jmsg(jcr, M_FATAL, 0, _("A user name for DBI must be supplied.\n"));
 116       return NULL;
 117    }
 118    P(mutex);                          /* lock DB queue */
 119    if (db_list == NULL) {
 120       db_list = New(dlist(mdb, &mdb->link));
 121       db_getvalue_list = New(dlist(field, field->link));
 122    }
 123    if (!mult_db_connections) {
 124       /* Look to see if DB already open */
 125       foreach_dlist(mdb, db_list) {
 126          if (bstrcmp(mdb->db_name, db_name) &&
 127              bstrcmp(mdb->db_address, db_address) &&
 128              bstrcmp(mdb->db_driver, db_driver) &&
 129              mdb->db_port == db_port) {
 130             Dmsg4(100, "DB REopen %d %s %s erro: %d\n", mdb->ref_count, db_driver, db_name,
 131                   dbi_conn_error(mdb->db, NULL));
 132             mdb->ref_count++;
 133             V(mutex);
 134             return mdb;                  /* already open */
 135          }
 136       }
 137    }
 138    Dmsg0(100, "db_open first time\n");
 139    mdb = (B_DB *)malloc(sizeof(B_DB));
 140    memset(mdb, 0, sizeof(B_DB));
 141    mdb->db_name = bstrdup(db_name);
 142    mdb->db_user = bstrdup(db_user);
 143    if (db_password) {
 144       mdb->db_password = bstrdup(db_password);
 145    }
 146    if (db_address) {
 147       mdb->db_address  = bstrdup(db_address);
 148    }
 149    if (db_socket) {
 150       mdb->db_socket   = bstrdup(db_socket);
 151    }
 152    if (db_driverdir) {
 153       mdb->db_driverdir = bstrdup(db_driverdir);
 154    }
 155    if (db_driver) {
 156       mdb->db_driver    = bstrdup(db_driver);
 157    }
 158    mdb->db_type        = db_type;
 159    mdb->db_port        = db_port;
 160    mdb->have_insert_id = TRUE;
 161    mdb->errmsg         = get_pool_memory(PM_EMSG); /* get error message buffer */
 162    *mdb->errmsg        = 0;
 163    mdb->cmd            = get_pool_memory(PM_EMSG); /* get command buffer */
 164    mdb->cached_path    = get_pool_memory(PM_FNAME);
 165    mdb->cached_path_id = 0;
 166    mdb->ref_count      = 1;
 167    mdb->fname          = get_pool_memory(PM_FNAME);
 168    mdb->path           = get_pool_memory(PM_FNAME);
 169    mdb->esc_name       = get_pool_memory(PM_FNAME);
 170    mdb->esc_path      = get_pool_memory(PM_FNAME);
 171    mdb->allow_transactions = mult_db_connections;
 172    db_list->append(mdb);                   /* put db in list */
 173    V(mutex);
 174    return mdb;
 175 }
 176
 177 /*
 178  * Now actually open the database.  This can generate errors,
 179  *   which are returned in the errmsg
 180  *
 181  * DO NOT close the database or free(mdb) here  !!!!
 182  */
 183 int
 184 db_open_database(JCR *jcr, B_DB *mdb)
 185 {
 186    int errstat;
 187    int dbstat;
 188    uint8_t len;
 189    const char *errmsg;
 190    char buf[10], *port;
 191    int numdrivers;
 192    char *db_name = NULL;
 193    char *db_dir = NULL;
 194
 195    P(mutex);
 196    if (mdb->connected) {
 197       V(mutex);
 198       return 1;
 199    }
 200    mdb->connected = false;
 201
 202    if ((errstat=rwl_init(&mdb->lock)) != 0) {
 203       berrno be;
 204       Mmsg1(&mdb->errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
 205             be.bstrerror(errstat));
 206       V(mutex);
 207       return 0;
 208    }
 209
 210    if (mdb->db_port) {
 211       bsnprintf(buf, sizeof(buf), "%d", mdb->db_port);
 212       port = buf;
 213    } else {
 214       port = NULL;
 215    }
 216
 217    numdrivers = dbi_initialize_r(mdb->db_driverdir, &(mdb->instance));
 218    if (numdrivers < 0) {
 219       Mmsg2(&mdb->errmsg, _("Unable to locate the DBD drivers to DBI interface in: \n"
 220                                "db_driverdir=%s. It is probaly not found any drivers\n"),
 221                                mdb->db_driverdir,numdrivers);
 222       V(mutex);
 223       return 0;
 224    }
 225    mdb->db = (void **)dbi_conn_new_r(mdb->db_driver, mdb->instance);
 226    /* Can be many types of databases */
 227    switch (mdb->db_type) {
 228    case SQL_TYPE_MYSQL:
 229       dbi_conn_set_option(mdb->db, "host", mdb->db_address); /* default = localhost */
 230       dbi_conn_set_option(mdb->db, "port", port);            /* default port */
 231       dbi_conn_set_option(mdb->db, "username", mdb->db_user);     /* login name */
 232       dbi_conn_set_option(mdb->db, "password", mdb->db_password); /* password */
 233       dbi_conn_set_option(mdb->db, "dbname", mdb->db_name);       /* database name */
 234       break;
 235    case SQL_TYPE_POSTGRESQL:
 236       dbi_conn_set_option(mdb->db, "host", mdb->db_address);
 237       dbi_conn_set_option(mdb->db, "port", port);
 238       dbi_conn_set_option(mdb->db, "username", mdb->db_user);
 239       dbi_conn_set_option(mdb->db, "password", mdb->db_password);
 240       dbi_conn_set_option(mdb->db, "dbname", mdb->db_name);
 241       break;
 242    case SQL_TYPE_SQLITE:
 243       len = strlen(working_directory) + 5;
 244       db_dir = (char *)malloc(len);
 245       strcpy(db_dir, working_directory);
 246       strcat(db_dir, "/");
 247       len = strlen(mdb->db_name) + 5;
 248       db_name = (char *)malloc(len);
 249       strcpy(db_name, mdb->db_name);
 250       strcat(db_name, ".db");
 251       dbi_conn_set_option(mdb->db, "sqlite_dbdir", db_dir);
 252       dbi_conn_set_option(mdb->db, "dbname", db_name);
 253       break;
 254    case SQL_TYPE_SQLITE3:
 255       len = strlen(working_directory) + 5;
 256       db_dir = (char *)malloc(len);
 257       strcpy(db_dir, working_directory);
 258       strcat(db_dir, "/");
 259       len = strlen(mdb->db_name) + 5;
 260       db_name = (char *)malloc(len);
 261       strcpy(db_name, mdb->db_name);
 262       strcat(db_name, ".db");
 263       dbi_conn_set_option(mdb->db, "sqlite3_dbdir", db_dir);
 264       dbi_conn_set_option(mdb->db, "dbname", db_name);
 265       Dmsg2(500, "SQLITE: %s %s\n", db_dir, db_name);
 266       break;
 267    }
 268
 269    /* If connection fails, try at 5 sec intervals for 30 seconds. */
 270    for (int retry=0; retry < 6; retry++) {
 271
 272       dbstat = dbi_conn_connect(mdb->db);
 273       if ( dbstat == 0) {
 274          break;
 275       }
 276
 277       dbi_conn_error(mdb->db, &errmsg);
 278       Dmsg1(50, "dbi error: %s\n", errmsg);
 279
 280       bmicrosleep(5, 0);
 281
 282    }
 283
 284    if ( dbstat != 0 ) {
 285       Mmsg3(&mdb->errmsg, _("Unable to connect to DBI interface. Type=%s Database=%s User=%s\n"
 286          "Possible causes: SQL server not running; password incorrect; max_connections exceeded.\n"),
 287          mdb->db_driver, mdb->db_name, mdb->db_user);
 288       V(mutex);
 289       return 0;
 290    }
 291
 292    Dmsg0(50, "dbi_real_connect done\n");
 293    Dmsg3(50, "db_user=%s db_name=%s db_password=%s\n",
 294                     mdb->db_user, mdb->db_name,
 295                     mdb->db_password==NULL?"(NULL)":mdb->db_password);
 296
 297    mdb->connected = true;
 298
 299    if (!check_tables_version(jcr, mdb)) {
 300       V(mutex);
 301       return 0;
 302    }
 303
 304    switch (mdb->db_type) {
 305    case SQL_TYPE_MYSQL:
 306       /* Set connection timeout to 8 days specialy for batch mode */
 307       sql_query(mdb, "SET wait_timeout=691200");
 308       sql_query(mdb, "SET interactive_timeout=691200");
 309       break;
 310    case SQL_TYPE_POSTGRESQL:
 311       /* tell PostgreSQL we are using standard conforming strings
 312          and avoid warnings such as:
 313          WARNING:  nonstandard use of \\ in a string literal
 314       */
 315       sql_query(mdb, "SET datestyle TO 'ISO, YMD'");
 316       sql_query(mdb, "set standard_conforming_strings=on");
 317       break;
 318    }
 319
 320    if(db_dir) {
 321       free(db_dir);
 322    }
 323    if(db_name) {
 324       free(db_name);
 325    }
 326
 327    V(mutex);
 328    return 1;
 329 }
 330
 331 void
 332 db_close_database(JCR *jcr, B_DB *mdb)
 333 {
 334    if (!mdb) {
 335       return;
 336    }
 337    db_end_transaction(jcr, mdb);
 338    P(mutex);
 339    sql_free_result(mdb);
 340    mdb->ref_count--;
 341    if (mdb->ref_count == 0) {
 342       db_list->remove(mdb);
 343       if (mdb->connected && mdb->db) {
 344          //sql_close(mdb);
 345          dbi_shutdown_r(mdb->instance);
 346          mdb->db = NULL;
 347          mdb->instance = NULL;
 348       }
 349       rwl_destroy(&mdb->lock);
 350       free_pool_memory(mdb->errmsg);
 351       free_pool_memory(mdb->cmd);
 352       free_pool_memory(mdb->cached_path);
 353       free_pool_memory(mdb->fname);
 354       free_pool_memory(mdb->path);
 355       free_pool_memory(mdb->esc_name);
 356       free_pool_memory(mdb->esc_path);
 357       if (mdb->db_name) {
 358          free(mdb->db_name);
 359       }
 360       if (mdb->db_user) {
 361          free(mdb->db_user);
 362       }
 363       if (mdb->db_password) {
 364          free(mdb->db_password);
 365       }
 366       if (mdb->db_address) {
 367          free(mdb->db_address);
 368       }
 369       if (mdb->db_socket) {
 370          free(mdb->db_socket);
 371       }
 372       if (mdb->db_driverdir) {
 373          free(mdb->db_driverdir);
 374       }
 375       if (mdb->db_driver) {
 376           free(mdb->db_driver);
 377       }
 378       free(mdb);
 379    }
 380    V(mutex);
 381 }
 382
 383 void db_thread_cleanup()
 384 { }
 385
 386 /*
 387  * Return the next unique index (auto-increment) for
 388  * the given table.  Return NULL on error.
 389  *
 390  */
 391 int db_next_index(JCR *jcr, B_DB *mdb, char *table, char *index)
 392 {
 393    strcpy(index, "NULL");
 394    return 1;
 395 }
 396
 397
 398 /*
 399  * Escape strings so that DBI is happy
 400  *
 401  *   NOTE! len is the length of the old string. Your new
 402  *         string must be long enough (max 2*old+1) to hold
 403  *         the escaped output.
 404  *
 405  * dbi_conn_quote_string_copy receives a pointer to pointer.
 406  * We need copy the value of pointer to snew because libdbi change the
 407  * pointer
 408  */
 409 void
 410 db_escape_string(JCR *jcr, B_DB *mdb, char *snew, char *old, int len)
 411 {
 412    char *inew;
 413    char *pnew;
 414
 415    if (len == 0) {
 416       snew[0] = 0;
 417    } else {
 418       /* correct the size of old basead in len
 419        * and copy new string to inew
 420        */
 421       inew = (char *)malloc(sizeof(char) * len + 1);
 422       bstrncpy(inew,old,len + 1);
 423       /* escape the correct size of old */
 424       dbi_conn_escape_string_copy(mdb->db, inew, &pnew);
 425       free(inew);
 426       /* copy the escaped string to snew */
 427       bstrncpy(snew, pnew, 2 * len + 1);
 428    }
 429
 430    Dmsg2(500, "dbi_conn_escape_string_copy %p %s\n",snew,snew);
 431
 432 }
 433
 434 /*
 435  * Submit a general SQL command (cmd), and for each row returned,
 436  *  the sqlite_handler is called with the ctx.
 437  */
 438 bool db_sql_query(B_DB *mdb, const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
 439 {
 440    SQL_ROW row;
 441
 442    Dmsg0(500, "db_sql_query started\n");
 443
 444    db_lock(mdb);
 445    if (sql_query(mdb, query) != 0) {
 446       Mmsg(mdb->errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror(mdb));
 447       db_unlock(mdb);
 448       Dmsg0(500, "db_sql_query failed\n");
 449       return false;
 450    }
 451    Dmsg0(500, "db_sql_query succeeded. checking handler\n");
 452
 453    if (result_handler != NULL) {
 454       Dmsg0(500, "db_sql_query invoking handler\n");
 455       if ((mdb->result = sql_store_result(mdb)) != NULL) {
 456          int num_fields = sql_num_fields(mdb);
 457
 458          Dmsg0(500, "db_sql_query sql_store_result suceeded\n");
 459          while ((row = sql_fetch_row(mdb)) != NULL) {
 460
 461             Dmsg0(500, "db_sql_query sql_fetch_row worked\n");
 462             if (result_handler(ctx, num_fields, row))
 463                break;
 464          }
 465
 466         sql_free_result(mdb);
 467       }
 468    }
 469    db_unlock(mdb);
 470
 471    Dmsg0(500, "db_sql_query finished\n");
 472
 473    return true;
 474 }
 475
 476
 477
 478 DBI_ROW my_dbi_fetch_row(B_DB *mdb)
 479 {
 480    int j;
 481    DBI_ROW row = NULL; // by default, return NULL
 482
 483    Dmsg0(500, "my_dbi_fetch_row start\n");
 484    if ((!mdb->row || mdb->row_size < mdb->num_fields) && mdb->num_rows > 0) {
 485       int num_fields = mdb->num_fields;
 486       Dmsg1(500, "we have need space of %d bytes\n", sizeof(char *) * mdb->num_fields);
 487
 488       if (mdb->row) {
 489          Dmsg0(500, "my_dbi_fetch_row freeing space\n");
 490          Dmsg2(500, "my_dbi_free_row row: '%p' num_fields: '%d'\n", mdb->row, mdb->num_fields);
 491          if (mdb->num_rows != 0) {
 492             for(j = 0; j < mdb->num_fields; j++) {
 493                Dmsg2(500, "my_dbi_free_row row '%p' '%d'\n", mdb->row[j], j);
 494                   if(mdb->row[j]) {
 495                      free(mdb->row[j]);
 496                   }
 497             }
 498          }
 499          free(mdb->row);
 500       }
 501       //num_fields += 20;                  /* add a bit extra */
 502       mdb->row = (DBI_ROW)malloc(sizeof(char *) * num_fields);
 503       mdb->row_size = num_fields;
 504
 505       // now reset the row_number now that we have the space allocated
 506       mdb->row_number = 1;
 507    }
 508
 509    // if still within the result set
 510    if (mdb->row_number <= mdb->num_rows && mdb->row_number != DBI_ERROR_BADPTR) {
 511       Dmsg2(500, "my_dbi_fetch_row row number '%d' is acceptable (1..%d)\n", mdb->row_number, mdb->num_rows);
 512       // get each value from this row
 513       for (j = 0; j < mdb->num_fields; j++) {
 514          mdb->row[j] = my_dbi_getvalue(mdb->result, mdb->row_number, j);
 515          // allocate space to queue row
 516          mdb->field_get = (DBI_FIELD_GET *)malloc(sizeof(DBI_FIELD_GET));
 517          // store the pointer in queue
 518          mdb->field_get->value = mdb->row[j];
 519          Dmsg4(500, "my_dbi_fetch_row row[%d] field: '%p' in queue: '%p' has value: '%s'\n",
 520                j, mdb->row[j], mdb->field_get->value, mdb->row[j]);
 521          // insert in queue to future free
 522          dbi_getvalue_list->append(mdb->field_get);
 523       }
 524       // increment the row number for the next call
 525       mdb->row_number++;
 526
 527       row = mdb->row;
 528    } else {
 529       Dmsg2(500, "my_dbi_fetch_row row number '%d' is NOT acceptable (1..%d)\n", mdb->row_number, mdb->num_rows);
 530    }
 531
 532    Dmsg1(500, "my_dbi_fetch_row finishes returning %p\n", row);
 533
 534    return row;
 535 }
 536
 537 int my_dbi_max_length(B_DB *mdb, int field_num) {
 538    //
 539    // for a given column, find the max length
 540    //
 541    int max_length;
 542    int i;
 543    int this_length;
 544    char *cbuf = NULL;
 545
 546    max_length = 0;
 547    for (i = 0; i < mdb->num_rows; i++) {
 548       if (my_dbi_getisnull(mdb->result, i, field_num)) {
 549           this_length = 4;        // "NULL"
 550       } else {
 551          cbuf = my_dbi_getvalue(mdb->result, i, field_num);
 552          this_length = cstrlen(cbuf);
 553          // cbuf is always free
 554          free(cbuf);
 555       }
 556
 557       if (max_length < this_length) {
 558           max_length = this_length;
 559       }
 560    }
 561
 562    return max_length;
 563 }
 564
 565 DBI_FIELD * my_dbi_fetch_field(B_DB *mdb)
 566 {
 567    int     i;
 568    int     dbi_index;
 569
 570    Dmsg0(500, "my_dbi_fetch_field starts\n");
 571
 572    if (!mdb->fields || mdb->fields_size < mdb->num_fields) {
 573       if (mdb->fields) {
 574          free(mdb->fields);
 575       }
 576       Dmsg1(500, "allocating space for %d fields\n", mdb->num_fields);
 577       mdb->fields = (DBI_FIELD *)malloc(sizeof(DBI_FIELD) * mdb->num_fields);
 578       mdb->fields_size = mdb->num_fields;
 579
 580       for (i = 0; i < mdb->num_fields; i++) {
 581          // num_fileds is starting at 1, increment i by 1
 582          dbi_index = i + 1;
 583          Dmsg1(500, "filling field %d\n", i);
 584          mdb->fields[i].name       = (char *)dbi_result_get_field_name(mdb->result, dbi_index);
 585          mdb->fields[i].max_length = my_dbi_max_length(mdb, i);
 586          mdb->fields[i].type       = dbi_result_get_field_type_idx(mdb->result, dbi_index);
 587          mdb->fields[i].flags      = dbi_result_get_field_attribs_idx(mdb->result, dbi_index);
 588
 589          Dmsg4(500, "my_dbi_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
 590             mdb->fields[i].name, mdb->fields[i].max_length, mdb->fields[i].type,
 591             mdb->fields[i].flags);
 592       } // end for
 593    } // end if
 594
 595    // increment field number for the next time around
 596
 597    Dmsg0(500, "my_dbi_fetch_field finishes\n");
 598    return &mdb->fields[mdb->field_number++];
 599 }
 600
 601 void my_dbi_data_seek(B_DB *mdb, int row)
 602 {
 603    // set the row number to be returned on the next call
 604    // to my_dbi_fetch_row
 605    mdb->row_number = row;
 606 }
 607
 608 void my_dbi_field_seek(B_DB *mdb, int field)
 609 {
 610    mdb->field_number = field;
 611 }
 612
 613 /*
 614  * Note, if this routine returns 1 (failure), Bacula expects
 615  *  that no result has been stored.
 616  *
 617  *  Returns:  0  on success
 618  *            1  on failure
 619  *
 620  */
 621 int my_dbi_query(B_DB *mdb, const char *query)
 622 {
 623    const char *errmsg;
 624    Dmsg1(500, "my_dbi_query started %s\n", query);
 625    // We are starting a new query.  reset everything.
 626    mdb->num_rows     = -1;
 627    mdb->row_number   = -1;
 628    mdb->field_number = -1;
 629
 630    if (mdb->result) {
 631       dbi_result_free(mdb->result);  /* hmm, someone forgot to free?? */
 632       mdb->result = NULL;
 633    }
 634
 635    mdb->result = (void **)dbi_conn_query(mdb->db, query);
 636
 637    if (!mdb->result) {
 638       Dmsg2(50, "Query failed: %s %p\n", query, mdb->result);
 639       goto bail_out;
 640    }
 641
 642    mdb->status = (dbi_error_flag) dbi_conn_error(mdb->db, &errmsg);
 643
 644    if (mdb->status == DBI_ERROR_NONE) {
 645       Dmsg1(500, "we have a result\n", query);
 646
 647       // how many fields in the set?
 648       // num_fields starting at 1
 649       mdb->num_fields = dbi_result_get_numfields(mdb->result);
 650       Dmsg1(500, "we have %d fields\n", mdb->num_fields);
 651       // if no result num_rows is 0
 652       mdb->num_rows = dbi_result_get_numrows(mdb->result);
 653       Dmsg1(500, "we have %d rows\n", mdb->num_rows);
 654
 655       mdb->status = (dbi_error_flag) 0;                  /* succeed */
 656    } else {
 657       Dmsg1(50, "Result status failed: %s\n", query);
 658       goto bail_out;
 659    }
 660
 661    Dmsg0(500, "my_dbi_query finishing\n");
 662    return mdb->status;
 663
 664 bail_out:
 665    mdb->status = (dbi_error_flag) dbi_conn_error(mdb->db,&errmsg);
 666    //dbi_conn_error(mdb->db, &errmsg);
 667    Dmsg4(500, "my_dbi_query we failed dbi error: "
 668                    "'%s' '%p' '%d' flag '%d''\n", errmsg, mdb->result, mdb->result, mdb->status);
 669    dbi_result_free(mdb->result);
 670    mdb->result = NULL;
 671    mdb->status = (dbi_error_flag) 1;                   /* failed */
 672    return mdb->status;
 673 }
 674
 675 void my_dbi_free_result(B_DB *mdb)
 676 {
 677
 678    DBI_FIELD_GET *f;
 679    db_lock(mdb);
 680    if (mdb->result) {
 681       Dmsg1(500, "my_dbi_free_result result '%p'\n", mdb->result);
 682       dbi_result_free(mdb->result);
 683    }
 684
 685    mdb->result = NULL;
 686
 687    if (mdb->row) {
 688       free(mdb->row);
 689    }
 690
 691    /* now is time to free all value return by my_dbi_get_value
 692     * this is necessary because libdbi don't free memory return by yours results
 693     * and Bacula has some routine wich call more than once time my_dbi_fetch_row
 694     *
 695     * Using a queue to store all pointer allocate is a good way to free all things
 696     * when necessary
 697     */
 698    foreach_dlist(f, dbi_getvalue_list) {
 699       Dmsg2(500, "my_dbi_free_result field value: '%p' in queue: '%p'\n", f->value, f);
 700       free(f->value);
 701       free(f);
 702    }
 703
 704    mdb->row = NULL;
 705
 706    if (mdb->fields) {
 707       free(mdb->fields);
 708       mdb->fields = NULL;
 709    }
 710    db_unlock(mdb);
 711    Dmsg0(500, "my_dbi_free_result finish\n");
 712
 713 }
 714
 715 const char *my_dbi_strerror(B_DB *mdb)
 716 {
 717    const char *errmsg;
 718
 719    dbi_conn_error(mdb->db, &errmsg);
 720
 721    return errmsg;
 722 }
 723
 724 #ifdef HAVE_BATCH_FILE_INSERT
 725
 726 /*
 727  * This can be a bit strang but is the one way to do
 728  *
 729  * Returns 1 if OK
 730  *         0 if failed
 731  */
 732 int my_dbi_batch_start(JCR *jcr, B_DB *mdb)
 733 {
 734    char *query = "COPY batch FROM STDIN";
 735
 736    Dmsg0(500, "my_dbi_batch_start started\n");
 737
 738    switch (mdb->db_type) {
 739    case SQL_TYPE_MYSQL:
 740       db_lock(mdb);
 741       if (my_dbi_query(mdb,
 742                               "CREATE TEMPORARY TABLE batch ("
 743                                   "FileIndex integer,"
 744                                   "JobId integer,"
 745                                   "Path blob,"
 746                                   "Name blob,"
 747                                   "LStat tinyblob,"
 748                                   "MD5 tinyblob)") == 1)
 749       {
 750          Dmsg0(500, "my_dbi_batch_start failed\n");
 751          return 1;
 752       }
 753       db_unlock(mdb);
 754       Dmsg0(500, "my_dbi_batch_start finishing\n");
 755       return 1;
 756       break;
 757    case SQL_TYPE_POSTGRESQL:
 758
 759       if (my_dbi_query(mdb, "CREATE TEMPORARY TABLE batch ("
 760                                   "fileindex int,"
 761                                   "jobid int,"
 762                                   "path varchar,"
 763                                   "name varchar,"
 764                                   "lstat varchar,"
 765                                   "md5 varchar)") == 1)
 766       {
 767          Dmsg0(500, "my_dbi_batch_start failed\n");
 768          return 1;
 769       }
 770
 771       // We are starting a new query.  reset everything.
 772       mdb->num_rows     = -1;
 773       mdb->row_number   = -1;
 774       mdb->field_number = -1;
 775
 776       my_dbi_free_result(mdb);
 777
 778       for (int i=0; i < 10; i++) {
 779          my_dbi_query(mdb, query);
 780          if (mdb->result) {
 781             break;
 782          }
 783          bmicrosleep(5, 0);
 784       }
 785       if (!mdb->result) {
 786          Dmsg1(50, "Query failed: %s\n", query);
 787          goto bail_out;
 788       }
 789
 790       mdb->status = (dbi_error_flag)dbi_conn_error(mdb->db, NULL);
 791       //mdb->status = DBI_ERROR_NONE;
 792
 793       if (mdb->status == DBI_ERROR_NONE) {
 794          // how many fields in the set?
 795          mdb->num_fields = dbi_result_get_numfields(mdb->result);
 796          mdb->num_rows   = dbi_result_get_numrows(mdb->result);
 797          mdb->status = (dbi_error_flag) 1;
 798       } else {
 799          Dmsg1(50, "Result status failed: %s\n", query);
 800          goto bail_out;
 801       }
 802
 803       Dmsg0(500, "my_postgresql_batch_start finishing\n");
 804
 805       return mdb->status;
 806       break;
 807    case SQL_TYPE_SQLITE:
 808       db_lock(mdb);
 809       if (my_dbi_query(mdb, "CREATE TEMPORARY TABLE batch ("
 810                                   "FileIndex integer,"
 811                                   "JobId integer,"
 812                                   "Path blob,"
 813                                   "Name blob,"
 814                                   "LStat tinyblob,"
 815                                   "MD5 tinyblob)") == 1)
 816       {
 817          Dmsg0(500, "my_dbi_batch_start failed\n");
 818          goto bail_out;
 819       }
 820       db_unlock(mdb);
 821       Dmsg0(500, "my_dbi_batch_start finishing\n");
 822       return 1;
 823       break;
 824    case SQL_TYPE_SQLITE3:
 825       db_lock(mdb);
 826       if (my_dbi_query(mdb, "CREATE TEMPORARY TABLE batch ("
 827                                   "FileIndex integer,"
 828                                   "JobId integer,"
 829                                   "Path blob,"
 830                                   "Name blob,"
 831                                   "LStat tinyblob,"
 832                                   "MD5 tinyblob)") == 1)
 833       {
 834          Dmsg0(500, "my_dbi_batch_start failed\n");
 835          goto bail_out;
 836       }
 837       db_unlock(mdb);
 838       Dmsg0(500, "my_dbi_batch_start finishing\n");
 839       return 1;
 840       break;
 841    }
 842
 843 bail_out:
 844    Mmsg1(&mdb->errmsg, _("error starting batch mode: %s"), my_dbi_strerror(mdb));
 845    mdb->status = (dbi_error_flag) 0;
 846    my_dbi_free_result(mdb);
 847    mdb->result = NULL;
 848    return mdb->status;
 849 }
 850
 851 /* set error to something to abort operation */
 852 int my_dbi_batch_end(JCR *jcr, B_DB *mdb, const char *error)
 853 {
 854    int res = 0;
 855    int count = 30;
 856    int (*custom_function)(void*, const char*) = NULL;
 857    dbi_conn_t *myconn = (dbi_conn_t *)(mdb->db);
 858
 859    Dmsg0(500, "my_dbi_batch_end started\n");
 860
 861    if (!mdb) {                  /* no files ? */
 862       return 0;
 863    }
 864
 865    switch (mdb->db_type) {
 866    case SQL_TYPE_MYSQL:
 867       if(mdb) {
 868          mdb->status = (dbi_error_flag) 0;
 869       }
 870       break;
 871    case SQL_TYPE_POSTGRESQL:
 872       custom_function = (custom_function_end_t)dbi_driver_specific_function(dbi_conn_get_driver(mdb->db), "PQputCopyEnd");
 873
 874
 875       do {
 876          res = (*custom_function)(myconn->connection, error);
 877       } while (res == 0 && --count > 0);
 878
 879       if (res == 1) {
 880          Dmsg0(500, "ok\n");
 881          mdb->status = (dbi_error_flag) 1;
 882       }
 883
 884       if (res <= 0) {
 885          Dmsg0(500, "we failed\n");
 886          mdb->status = (dbi_error_flag) 0;
 887          //Mmsg1(&mdb->errmsg, _("error ending batch mode: %s"), PQerrorMessage(mdb->db));
 888        }
 889       break;
 890    case SQL_TYPE_SQLITE:
 891       if(mdb) {
 892          mdb->status = (dbi_error_flag) 0;
 893       }
 894       break;
 895    case SQL_TYPE_SQLITE3:
 896       if(mdb) {
 897          mdb->status = (dbi_error_flag) 0;
 898       }
 899       break;
 900    }
 901
 902    Dmsg0(500, "my_dbi_batch_end finishing\n");
 903
 904    return true;
 905 }
 906
 907 /*
 908  * This function is big and use a big switch.
 909  * In near future is better split in small functions
 910  * and refactory.
 911  *
 912  */
 913 int my_dbi_batch_insert(JCR *jcr, B_DB *mdb, ATTR_DBR *ar)
 914 {
 915    int res;
 916    int count=30;
 917    dbi_conn_t *myconn = (dbi_conn_t *)(mdb->db);
 918    int (*custom_function)(void*, const char*, int) = NULL;
 919    char* (*custom_function_error)(void*) = NULL;
 920    size_t len;
 921    char *digest;
 922    char ed1[50];
 923
 924    Dmsg0(500, "my_dbi_batch_insert started \n");
 925
 926    mdb->esc_name = check_pool_memory_size(mdb->esc_name, mdb->fnl*2+1);
 927    mdb->esc_path = check_pool_memory_size(mdb->esc_path, mdb->pnl*2+1);
 928
 929    if (ar->Digest == NULL || ar->Digest[0] == 0) {
 930       digest = "0";
 931    } else {
 932       digest = ar->Digest;
 933    }
 934
 935    switch (mdb->db_type) {
 936    case SQL_TYPE_MYSQL:
 937       db_escape_string(jcr, mdb, mdb->esc_name, mdb->fname, mdb->fnl);
 938       db_escape_string(jcr, mdb, mdb->esc_path, mdb->path, mdb->pnl);
 939       len = Mmsg(mdb->cmd, "INSERT INTO batch VALUES (%u,%s,'%s','%s','%s','%s')",
 940                       ar->FileIndex, edit_int64(ar->JobId,ed1), mdb->esc_path,
 941                       mdb->esc_name, ar->attr, digest);
 942
 943       if (my_dbi_query(mdb,mdb->cmd) == 1)
 944       {
 945          Dmsg0(500, "my_dbi_batch_insert failed\n");
 946          goto bail_out;
 947       }
 948
 949       Dmsg0(500, "my_dbi_batch_insert finishing\n");
 950
 951       return 1;
 952       break;
 953    case SQL_TYPE_POSTGRESQL:
 954       my_postgresql_copy_escape(mdb->esc_name, mdb->fname, mdb->fnl);
 955       my_postgresql_copy_escape(mdb->esc_path, mdb->path, mdb->pnl);
 956       len = Mmsg(mdb->cmd, "%u\t%s\t%s\t%s\t%s\t%s\n",
 957                      ar->FileIndex, edit_int64(ar->JobId, ed1), mdb->esc_path,
 958                      mdb->esc_name, ar->attr, digest);
 959
 960       /* libdbi don't support CopyData and we need call a postgresql
 961        * specific function to do this work
 962        */
 963       Dmsg2(500, "my_dbi_batch_insert :\n %s \ncmd_size: %d",mdb->cmd, len);
 964       if ((custom_function = (custom_function_insert_t)dbi_driver_specific_function(dbi_conn_get_driver(mdb->db),
 965             "PQputCopyData")) != NULL) {
 966          do {
 967             res = (*custom_function)(myconn->connection, mdb->cmd, len);
 968          } while (res == 0 && --count > 0);
 969
 970          if (res == 1) {
 971             Dmsg0(500, "ok\n");
 972             mdb->changes++;
 973             mdb->status = (dbi_error_flag) 1;
 974          }
 975
 976          if (res <= 0) {
 977             Dmsg0(500, "my_dbi_batch_insert failed\n");
 978             goto bail_out;
 979          }
 980
 981          Dmsg0(500, "my_dbi_batch_insert finishing\n");
 982          return mdb->status;
 983       } else {
 984          // ensure to detect a PQerror
 985          custom_function_error = (custom_function_error_t)dbi_driver_specific_function(dbi_conn_get_driver(mdb->db), "PQerrorMessage");
 986          Dmsg1(500, "my_dbi_batch_insert failed\n PQerrorMessage: %s", (*custom_function_error)(myconn->connection));
 987          goto bail_out;
 988       }
 989       break;
 990    case SQL_TYPE_SQLITE:
 991       db_escape_string(jcr, mdb, mdb->esc_name, mdb->fname, mdb->fnl);
 992       db_escape_string(jcr, mdb, mdb->esc_path, mdb->path, mdb->pnl);
 993       len = Mmsg(mdb->cmd, "INSERT INTO batch VALUES (%u,%s,'%s','%s','%s','%s')",
 994                       ar->FileIndex, edit_int64(ar->JobId,ed1), mdb->esc_path,
 995                       mdb->esc_name, ar->attr, digest);
 996       if (my_dbi_query(mdb,mdb->cmd) == 1)
 997       {
 998          Dmsg0(500, "my_dbi_batch_insert failed\n");
 999          goto bail_out;
1000       }
1001
1002       Dmsg0(500, "my_dbi_batch_insert finishing\n");
1003
1004       return 1;
1005       break;
1006    case SQL_TYPE_SQLITE3:
1007       db_escape_string(jcr, mdb, mdb->esc_name, mdb->fname, mdb->fnl);
1008       db_escape_string(jcr, mdb, mdb->esc_path, mdb->path, mdb->pnl);
1009       len = Mmsg(mdb->cmd, "INSERT INTO batch VALUES (%u,%s,'%s','%s','%s','%s')",
1010                       ar->FileIndex, edit_int64(ar->JobId,ed1), mdb->esc_path,
1011                       mdb->esc_name, ar->attr, digest);
1012       if (my_dbi_query(mdb,mdb->cmd) == 1)
1013       {
1014          Dmsg0(500, "my_dbi_batch_insert failed\n");
1015          goto bail_out;
1016       }
1017
1018       Dmsg0(500, "my_dbi_batch_insert finishing\n");
1019
1020       return 1;
1021       break;
1022    }
1023
1024 bail_out:
1025   Mmsg1(&mdb->errmsg, _("error inserting batch mode: %s"), my_dbi_strerror(mdb));
1026   mdb->status = (dbi_error_flag) 0;
1027   my_dbi_free_result(mdb);
1028   return mdb->status;
1029 }
1030
1031 /*
1032  * Escape strings so that PostgreSQL is happy on COPY
1033  *
1034  *   NOTE! len is the length of the old string. Your new
1035  *         string must be long enough (max 2*old+1) to hold
1036  *         the escaped output.
1037  */
1038 char *my_postgresql_copy_escape(char *dest, char *src, size_t len)
1039 {
1040    /* we have to escape \t, \n, \r, \ */
1041    char c = '\0' ;
1042
1043    while (len > 0 && *src) {
1044       switch (*src) {
1045       case '\n':
1046          c = 'n';
1047          break;
1048       case '\\':
1049          c = '\\';
1050          break;
1051       case '\t':
1052          c = 't';
1053          break;
1054       case '\r':
1055          c = 'r';
1056          break;
1057       default:
1058          c = '\0' ;
1059       }
1060
1061       if (c) {
1062          *dest = '\\';
1063          dest++;
1064          *dest = c;
1065       } else {
1066          *dest = *src;
1067       }
1068
1069       len--;
1070       src++;
1071       dest++;
1072    }
1073
1074    *dest = '\0';
1075    return dest;
1076 }
1077
1078 #endif /* HAVE_BATCH_FILE_INSERT */
1079
1080 /* my_dbi_getisnull
1081  * like PQgetisnull
1082  * int PQgetisnull(const PGresult *res,
1083  *              int row_number,
1084  *               int column_number);
1085  *
1086  *  use dbi_result_seek_row to search in result set
1087  */
1088 int my_dbi_getisnull(dbi_result *result, int row_number, int column_number) {
1089    int i;
1090
1091    if(row_number == 0) {
1092       row_number++;
1093    }
1094
1095    column_number++;
1096
1097    if(dbi_result_seek_row(result, row_number)) {
1098
1099       i = dbi_result_field_is_null_idx(result,column_number);
1100
1101       return i;
1102    } else {
1103
1104       return 0;
1105    }
1106
1107 }
1108 /* my_dbi_getvalue
1109  * like PQgetvalue;
1110  * char *PQgetvalue(const PGresult *res,
1111  *                int row_number,
1112  *                int column_number);
1113  *
1114  * use dbi_result_seek_row to search in result set
1115  * use example to return only strings
1116  */
1117 char *my_dbi_getvalue(dbi_result *result, int row_number, unsigned int column_number) {
1118
1119    char *buf = NULL;
1120    const char *errmsg;
1121    const char *field_name;
1122    unsigned short dbitype;
1123    size_t field_length;
1124    int64_t num;
1125
1126    /* correct the index for dbi interface
1127     * dbi index begins 1
1128     * I prefer do not change others functions
1129     */
1130    Dmsg3(600, "my_dbi_getvalue pre-starting result '%p' row number '%d' column number '%d'\n",
1131                                 result, row_number, column_number);
1132
1133    column_number++;
1134
1135    if(row_number == 0) {
1136      row_number++;
1137    }
1138
1139    Dmsg3(600, "my_dbi_getvalue starting result '%p' row number '%d' column number '%d'\n",
1140                         result, row_number, column_number);
1141
1142    if(dbi_result_seek_row(result, row_number)) {
1143
1144       field_name = dbi_result_get_field_name(result, column_number);
1145       field_length = dbi_result_get_field_length(result, field_name);
1146       dbitype = dbi_result_get_field_type_idx(result,column_number);
1147
1148       Dmsg3(500, "my_dbi_getvalue start: type: '%d' "
1149             "field_length bytes: '%d' fieldname: '%s'\n",
1150             dbitype, field_length, field_name);
1151
1152       if(field_length) {
1153          //buf = (char *)malloc(sizeof(char *) * field_length + 1);
1154          buf = (char *)malloc(field_length + 1);
1155       } else {
1156          /* if numbers */
1157          buf = (char *)malloc(sizeof(char *) * 50);
1158       }
1159
1160       switch (dbitype) {
1161       case DBI_TYPE_INTEGER:
1162          num = dbi_result_get_longlong(result, field_name);
1163          edit_int64(num, buf);
1164          field_length = strlen(buf);
1165          break;
1166       case DBI_TYPE_STRING:
1167          if(field_length) {
1168             field_length = bsnprintf(buf, field_length + 1, "%s",
1169             dbi_result_get_string(result, field_name));
1170          } else {
1171             buf[0] = 0;
1172          }
1173          break;
1174       case DBI_TYPE_BINARY:
1175          /* dbi_result_get_binary return a NULL pointer if value is empty
1176          * following, change this to what Bacula espected
1177          */
1178          if(field_length) {
1179             field_length = bsnprintf(buf, field_length + 1, "%s",
1180                   dbi_result_get_binary(result, field_name));
1181          } else {
1182             buf[0] = 0;
1183          }
1184          break;
1185       case DBI_TYPE_DATETIME:
1186          time_t last;
1187          struct tm tm;
1188
1189          last = dbi_result_get_datetime(result, field_name);
1190
1191          if(last == -1) {
1192                 field_length = bsnprintf(buf, 20, "0000-00-00 00:00:00");
1193          } else {
1194             (void)localtime_r(&last, &tm);
1195             field_length = bsnprintf(buf, 20, "%04d-%02d-%02d %02d:%02d:%02d",
1196                   (tm.tm_year + 1900), (tm.tm_mon + 1), tm.tm_mday,
1197                   tm.tm_hour, tm.tm_min, tm.tm_sec);
1198          }
1199          break;
1200       }
1201
1202    } else {
1203       dbi_conn_error(dbi_result_get_conn(result), &errmsg);
1204       Dmsg1(500, "my_dbi_getvalue error: %s\n", errmsg);
1205    }
1206
1207    Dmsg3(500, "my_dbi_getvalue finish buffer: '%p' num bytes: '%d' data: '%s'\n",
1208       buf, field_length, buf);
1209
1210    // don't worry about this buf
1211    return buf;
1212 }
1213
1214 int my_dbi_sql_insert_id(B_DB *mdb, char *table_name)
1215 {
1216    /*
1217     Obtain the current value of the sequence that
1218     provides the serial value for primary key of the table.
1219
1220     currval is local to our session.  It is not affected by
1221     other transactions.
1222
1223     Determine the name of the sequence.
1224     PostgreSQL automatically creates a sequence using
1225     <table>_<column>_seq.
1226     At the time of writing, all tables used this format for
1227     for their primary key: <table>id
1228     Except for basefiles which has a primary key on baseid.
1229     Therefore, we need to special case that one table.
1230
1231     everything else can use the PostgreSQL formula.
1232    */
1233
1234    char      sequence[30];
1235    uint64_t    id = 0;
1236
1237    if (mdb->db_type == SQL_TYPE_POSTGRESQL) {
1238
1239       if (strcasecmp(table_name, "basefiles") == 0) {
1240          bstrncpy(sequence, "basefiles_baseid", sizeof(sequence));
1241       } else {
1242          bstrncpy(sequence, table_name, sizeof(sequence));
1243          bstrncat(sequence, "_",        sizeof(sequence));
1244          bstrncat(sequence, table_name, sizeof(sequence));
1245          bstrncat(sequence, "id",       sizeof(sequence));
1246       }
1247
1248       bstrncat(sequence, "_seq", sizeof(sequence));
1249       id = dbi_conn_sequence_last(mdb->db, NT_(sequence));
1250    } else {
1251       id = dbi_conn_sequence_last(mdb->db, NT_(table_name));
1252    }
1253
1254    return id;
1255 }
1256
1257 #ifdef HAVE_BATCH_FILE_INSERT
1258 const char *my_dbi_batch_lock_path_query[5] = {
1259    /* Mysql */
1260    "LOCK TABLES Path write, batch write, Path as p write",
1261    /* Postgresql */
1262    "BEGIN; LOCK TABLE Path IN SHARE ROW EXCLUSIVE MODE",
1263    /* SQLite */
1264    "BEGIN",
1265    /* SQLite3 */
1266    "BEGIN",
1267    /* Ingres (TODO) */
1268    "BEGIN"
1269 };
1270
1271 const char *my_dbi_batch_lock_filename_query[5] = {
1272    /* Mysql */
1273    "LOCK TABLES Filename write, batch write, Filename as f write",
1274    /* Postgresql */
1275    "BEGIN; LOCK TABLE Filename IN SHARE ROW EXCLUSIVE MODE",
1276    /* SQLite */
1277    "BEGIN",
1278    /* SQLite3 */
1279    "BEGIN",
1280    /* Ingres (TODO) */
1281    "BEGIN"
1282 };
1283
1284 const char *my_dbi_batch_unlock_tables_query[5] = {
1285    /* Mysql */
1286    "UNLOCK TABLES",
1287    /* Postgresql */
1288    "COMMIT",
1289    /* SQLite */
1290    "COMMIT",
1291    /* SQLite3 */
1292    "COMMIT",
1293    /* Ingres */
1294    "COMMIT"
1295 };
1296
1297 const char *my_dbi_batch_fill_path_query[5] = {
1298    /* Mysql */
1299    "INSERT INTO Path (Path) "
1300    "SELECT a.Path FROM "
1301    "(SELECT DISTINCT Path FROM batch) AS a WHERE NOT EXISTS "
1302    "(SELECT Path FROM Path AS p WHERE p.Path = a.Path)",
1303    /* Postgresql */
1304    "INSERT INTO Path (Path) "
1305    "SELECT a.Path FROM "
1306    "(SELECT DISTINCT Path FROM batch) AS a "
1307    "WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) ",
1308    /* SQLite */
1309    "INSERT INTO Path (Path)"
1310    " SELECT DISTINCT Path FROM batch"
1311    " EXCEPT SELECT Path FROM Path",
1312    /* SQLite3 */
1313    "INSERT INTO Path (Path)"
1314    " SELECT DISTINCT Path FROM batch"
1315    " EXCEPT SELECT Path FROM Path",
1316    /* Ingres (TODO) */
1317    "INSERT INTO Path (Path) "
1318    "SELECT a.Path FROM "
1319    "(SELECT DISTINCT Path FROM batch) AS a "
1320    "WHERE NOT EXISTS (SELECT Path FROM Path WHERE Path = a.Path) "
1321 };
1322
1323 const char *my_dbi_batch_fill_filename_query[5] = {
1324    /* Mysql */
1325    "INSERT INTO Filename (Name) "
1326    "SELECT a.Name FROM "
1327    "(SELECT DISTINCT Name FROM batch) AS a WHERE NOT EXISTS "
1328    "(SELECT Name FROM Filename AS f WHERE f.Name = a.Name)",
1329    /* Postgresql */
1330    "INSERT INTO Filename (Name) "
1331    "SELECT a.Name FROM "
1332    "(SELECT DISTINCT Name FROM batch) as a "
1333    "WHERE NOT EXISTS "
1334    "(SELECT Name FROM Filename WHERE Name = a.Name)",
1335    /* SQLite */
1336    "INSERT INTO Filename (Name)"
1337    " SELECT DISTINCT Name FROM batch "
1338    " EXCEPT SELECT Name FROM Filename",
1339    /* SQLite3 */
1340    "INSERT INTO Filename (Name)"
1341    " SELECT DISTINCT Name FROM batch "
1342    " EXCEPT SELECT Name FROM Filename",
1343    /* Ingres (TODO) */
1344    "INSERT INTO Filename (Name) "
1345    "SELECT a.Name FROM "
1346    "(SELECT DISTINCT Name FROM batch) as a "
1347    "WHERE NOT EXISTS "
1348    "(SELECT Name FROM Filename WHERE Name = a.Name)"
1349 };
1350
1351 #endif /* HAVE_BATCH_FILE_INSERT */
1352
1353 const char *my_dbi_match[4] = {
1354    /* Mysql */
1355    "MATCH",
1356    /* Postgresql */
1357    "~",
1358    /* SQLite */
1359    "MATCH",
1360    /* SQLite3 */
1361    "MATCH"
1362 };
1363
1364 #endif /* HAVE_DBI */