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

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2000-2009 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 interface routines
  30  *
  31  *     Almost generic set of SQL database interface routines
  32  *      (with a little more work)
  33  *     SQL engine specific routines are in mysql.c, postgresql.c,
  34  *       sqlite.c, ...
  35  *
  36  *    Kern Sibbald, March 2000
  37  *
  38  *    Version $Id: sql.c 8034 2008-11-11 14:33:46Z ricozz $
  39  */
  40
  41 /* The following is necessary so that we do not include
  42  * the dummy external definition of B_DB.
  43  */
  44 #define __SQL_C                       /* indicate that this is sql.c */
  45
  46 #include "bacula.h"
  47 #include "cats.h"
  48
  49 #if    HAVE_SQLITE3 || HAVE_MYSQL || HAVE_SQLITE || HAVE_POSTGRESQL || HAVE_INGRES || HAVE_DBI
  50
  51 uint32_t bacula_db_version = 0;
  52
  53 int db_type = -1;        /* SQL engine type index */
  54
  55 /* Forward referenced subroutines */
  56 void print_dashes(B_DB *mdb);
  57 void print_result(B_DB *mdb);
  58
  59 B_DB *db_init(JCR *jcr, const char *db_driver, const char *db_name, const char *db_user,
  60               const char *db_password, const char *db_address, int db_port,
  61               const char *db_socket, int mult_db_connections)
  62 {
  63 #ifdef HAVE_DBI
  64    char *p;
  65    if (!db_driver) {
  66       Jmsg0(jcr, M_ABORT, 0, _("Driver type not specified in Catalog resource.\n"));
  67    }
  68    if (strlen(db_driver) < 5 || db_driver[3] != ':' || strncasecmp(db_driver, "dbi", 3) != 0) {
  69       Jmsg0(jcr, M_ABORT, 0, _("Invalid driver type, must be \"dbi:<type>\"\n"));
  70    }
  71    p = (char *)(db_driver + 4);
  72    if (strcasecmp(p, "mysql") == 0) {
  73       db_type = SQL_TYPE_MYSQL;
  74    } else if (strcasecmp(p, "postgresql") == 0) {
  75       db_type = SQL_TYPE_POSTGRESQL;
  76    } else if (strcasecmp(p, "sqlite") == 0) {
  77       db_type = SQL_TYPE_SQLITE;
  78    } else if (strcasecmp(p, "sqlite3") == 0) {
  79       db_type = SQL_TYPE_SQLITE3;
  80    } else if (strcasecmp(p, "ingres") == 0) {
  81       db_type = SQL_TYPE_INGRES;
  82    } else {
  83       Jmsg1(jcr, M_ABORT, 0, _("Unknown database type: %s\n"), p);
  84    }
  85 #elif HAVE_MYSQL
  86    db_type = SQL_TYPE_MYSQL;
  87 #elif HAVE_POSTGRESQL
  88    db_type = SQL_TYPE_POSTGRESQL;
  89 #elif HAVE_INGRES
  90    db_type = SQL_TYPE_INGRES;
  91 #elif HAVE_SQLITE
  92    db_type = SQL_TYPE_SQLITE;
  93 #elif HAVE_SQLITE3
  94    db_type = SQL_TYPE_SQLITE3;
  95 #endif
  96
  97    return db_init_database(jcr, db_name, db_user, db_password, db_address,
  98              db_port, db_socket, mult_db_connections);
  99 }
 100
 101 dbid_list::dbid_list()
 102 {
 103    memset(this, 0, sizeof(dbid_list));
 104    max_ids = 1000;
 105    DBId = (DBId_t *)malloc(max_ids * sizeof(DBId_t));
 106    num_ids = num_seen = tot_ids = 0;
 107    PurgedFiles = NULL;
 108 }
 109
 110 dbid_list::~dbid_list()
 111 {
 112    free(DBId);
 113 }
 114
 115 /*
 116  * Called here to retrieve an integer from the database
 117  */
 118 int db_int_handler(void *ctx, int num_fields, char **row)
 119 {
 120    uint32_t *val = (uint32_t *)ctx;
 121
 122    Dmsg1(800, "int_handler starts with row pointing at %x\n", row);
 123
 124    if (row[0]) {
 125       Dmsg1(800, "int_handler finds '%s'\n", row[0]);
 126       *val = str_to_int64(row[0]);
 127    } else {
 128       Dmsg0(800, "int_handler finds zero\n");
 129       *val = 0;
 130    }
 131    Dmsg0(800, "int_handler finishes\n");
 132    return 0;
 133 }
 134
 135 /*
 136  * Called here to retrieve a 32/64 bit integer from the database.
 137  *   The returned integer will be extended to 64 bit.
 138  */
 139 int db_int64_handler(void *ctx, int num_fields, char **row)
 140 {
 141    db_int64_ctx *lctx = (db_int64_ctx *)ctx;
 142
 143    if (row[0]) {
 144       lctx->value = str_to_int64(row[0]);
 145       lctx->count++;
 146    }
 147    return 0;
 148 }
 149
 150 /*
 151  * Use to build a comma separated list of values from a query. "10,20,30"
 152  */
 153 int db_list_handler(void *ctx, int num_fields, char **row)
 154 {
 155    db_list_ctx *lctx = (db_list_ctx *)ctx;
 156    if (num_fields == 1 && row[0]) {
 157       if (lctx->list[0]) {
 158          pm_strcat(lctx->list, ",");
 159       }
 160       pm_strcat(lctx->list, row[0]);
 161       lctx->count++;
 162    }
 163    return 0;
 164 }
 165
 166
 167 /*
 168  * Called here to retrieve an integer from the database
 169  */
 170 static int db_max_connections_handler(void *ctx, int num_fields, char **row)
 171 {
 172    uint32_t *val = (uint32_t *)ctx;
 173    uint32_t index = sql_get_max_connections_index[db_type];
 174    if (row[index]) {
 175       *val = str_to_int64(row[index]);
 176    } else {
 177       Dmsg0(800, "int_handler finds zero\n");
 178       *val = 0;
 179    }
 180    return 0;
 181 }
 182
 183 /*
 184  * Check catalog max_connections setting
 185  */
 186 bool db_check_max_connections(JCR *jcr, B_DB *mdb, uint32_t max_concurrent_jobs)
 187 {
 188    uint32_t max_conn=0;
 189    int ret=true;
 190
 191    /* Without Batch insert, no need to verify max_connections */
 192 #ifndef HAVE_BATCH_FILE_INSERT
 193    return ret;
 194 #endif
 195
 196    /* Check max_connections setting */
 197    if (!db_sql_query(mdb, sql_get_max_connections[db_type],
 198                      db_max_connections_handler, &max_conn)) {
 199       Jmsg(jcr, M_ERROR, 0, "Can't verify max_connections settings %s", mdb->errmsg);
 200       return ret;
 201    }
 202    if (max_conn && max_concurrent_jobs && max_concurrent_jobs > max_conn) {
 203       Mmsg(mdb->errmsg,
 204            _("On db_name=%s, %s max_connections=%d is lower than Director "
 205              "MaxConcurentJobs=%d\n"),
 206            mdb->db_name, db_get_type(), max_conn, max_concurrent_jobs);
 207       Jmsg(jcr, M_WARNING, 0, "%s", mdb->errmsg);
 208       ret = false;
 209    }
 210
 211    return ret;
 212 }
 213
 214 /* NOTE!!! The following routines expect that the
 215  *  calling subroutine sets and clears the mutex
 216  */
 217
 218 /* Check that the tables correspond to the version we want */
 219 bool check_tables_version(JCR *jcr, B_DB *mdb)
 220 {
 221    const char *query = "SELECT VersionId FROM Version";
 222
 223    bacula_db_version = 0;
 224    if (!db_sql_query(mdb, query, db_int_handler, (void *)&bacula_db_version)) {
 225       Jmsg(jcr, M_FATAL, 0, "%s", mdb->errmsg);
 226       return false;
 227    }
 228    if (bacula_db_version != BDB_VERSION) {
 229       Mmsg(mdb->errmsg, "Version error for database \"%s\". Wanted %d, got %d\n",
 230           mdb->db_name, BDB_VERSION, bacula_db_version);
 231       Jmsg(jcr, M_FATAL, 0, "%s", mdb->errmsg);
 232       return false;
 233    }
 234    return true;
 235 }
 236
 237 /* Utility routine for queries. The database MUST be locked before calling here. */
 238 int
 239 QueryDB(const char *file, int line, JCR *jcr, B_DB *mdb, char *cmd)
 240 {
 241    int status;
 242
 243    sql_free_result(mdb);
 244    if ((status=sql_query(mdb, cmd)) != 0) {
 245       m_msg(file, line, &mdb->errmsg, _("query %s failed:\n%s\n"), cmd, sql_strerror(mdb));
 246       j_msg(file, line, jcr, M_FATAL, 0, "%s", mdb->errmsg);
 247       if (verbose) {
 248          j_msg(file, line, jcr, M_INFO, 0, "%s\n", cmd);
 249       }
 250       return 0;
 251    }
 252
 253    mdb->result = sql_store_result(mdb);
 254
 255    return mdb->result != NULL;
 256 }
 257
 258 /*
 259  * Utility routine to do inserts
 260  * Returns: 0 on failure
 261  *          1 on success
 262  */
 263 int
 264 InsertDB(const char *file, int line, JCR *jcr, B_DB *mdb, char *cmd)
 265 {
 266    if (sql_query(mdb, cmd)) {
 267       m_msg(file, line, &mdb->errmsg,  _("insert %s failed:\n%s\n"), cmd, sql_strerror(mdb));
 268       j_msg(file, line, jcr, M_FATAL, 0, "%s", mdb->errmsg);
 269       if (verbose) {
 270          j_msg(file, line, jcr, M_INFO, 0, "%s\n", cmd);
 271       }
 272       return 0;
 273    }
 274    if (mdb->have_insert_id) {
 275       mdb->num_rows = sql_affected_rows(mdb);
 276    } else {
 277       mdb->num_rows = 1;
 278    }
 279    if (mdb->num_rows != 1) {
 280       char ed1[30];
 281       m_msg(file, line, &mdb->errmsg, _("Insertion problem: affected_rows=%s\n"),
 282          edit_uint64(mdb->num_rows, ed1));
 283       if (verbose) {
 284          j_msg(file, line, jcr, M_INFO, 0, "%s\n", cmd);
 285       }
 286       return 0;
 287    }
 288    mdb->changes++;
 289    return 1;
 290 }
 291
 292 /* Utility routine for updates.
 293  *  Returns: 0 on failure
 294  *           1 on success
 295  */
 296 int
 297 UpdateDB(const char *file, int line, JCR *jcr, B_DB *mdb, char *cmd)
 298 {
 299
 300    if (sql_query(mdb, cmd)) {
 301       m_msg(file, line, &mdb->errmsg, _("update %s failed:\n%s\n"), cmd, sql_strerror(mdb));
 302       j_msg(file, line, jcr, M_ERROR, 0, "%s", mdb->errmsg);
 303       if (verbose) {
 304          j_msg(file, line, jcr, M_INFO, 0, "%s\n", cmd);
 305       }
 306       return 0;
 307    }
 308    mdb->num_rows = sql_affected_rows(mdb);
 309    if (mdb->num_rows < 1) {
 310       char ed1[30];
 311       m_msg(file, line, &mdb->errmsg, _("Update failed: affected_rows=%s for %s\n"),
 312          edit_uint64(mdb->num_rows, ed1), cmd);
 313       if (verbose) {
 314 //       j_msg(file, line, jcr, M_INFO, 0, "%s\n", cmd);
 315       }
 316       return 0;
 317    }
 318    mdb->changes++;
 319    return 1;
 320 }
 321
 322 /* Utility routine for deletes
 323  *
 324  * Returns: -1 on error
 325  *           n number of rows affected
 326  */
 327 int
 328 DeleteDB(const char *file, int line, JCR *jcr, B_DB *mdb, char *cmd)
 329 {
 330
 331    if (sql_query(mdb, cmd)) {
 332       m_msg(file, line, &mdb->errmsg, _("delete %s failed:\n%s\n"), cmd, sql_strerror(mdb));
 333       j_msg(file, line, jcr, M_ERROR, 0, "%s", mdb->errmsg);
 334       if (verbose) {
 335          j_msg(file, line, jcr, M_INFO, 0, "%s\n", cmd);
 336       }
 337       return -1;
 338    }
 339    mdb->changes++;
 340    return sql_affected_rows(mdb);
 341 }
 342
 343
 344 /*
 345  * Get record max. Query is already in mdb->cmd
 346  *  No locking done
 347  *
 348  * Returns: -1 on failure
 349  *          count on success
 350  */
 351 int get_sql_record_max(JCR *jcr, B_DB *mdb)
 352 {
 353    SQL_ROW row;
 354    int stat = 0;
 355
 356    if (QUERY_DB(jcr, mdb, mdb->cmd)) {
 357       if ((row = sql_fetch_row(mdb)) == NULL) {
 358          Mmsg1(&mdb->errmsg, _("error fetching row: %s\n"), sql_strerror(mdb));
 359          stat = -1;
 360       } else {
 361          stat = str_to_int64(row[0]);
 362       }
 363       sql_free_result(mdb);
 364    } else {
 365       Mmsg1(&mdb->errmsg, _("error fetching row: %s\n"), sql_strerror(mdb));
 366       stat = -1;
 367    }
 368    return stat;
 369 }
 370
 371 /*
 372  * Return pre-edited error message
 373  */
 374 char *db_strerror(B_DB *mdb)
 375 {
 376    return mdb->errmsg;
 377 }
 378
 379 /*
 380  * Lock database, this can be called multiple times by the same
 381  *   thread without blocking, but must be unlocked the number of
 382  *   times it was locked.
 383  */
 384 void _db_lock(const char *file, int line, B_DB *mdb)
 385 {
 386    int errstat;
 387    if ((errstat=rwl_writelock(&mdb->lock)) != 0) {
 388       berrno be;
 389       e_msg(file, line, M_FATAL, 0, "rwl_writelock failure. stat=%d: ERR=%s\n",
 390            errstat, be.bstrerror(errstat));
 391    }
 392 }
 393
 394 /*
 395  * Unlock the database. This can be called multiple times by the
 396  *   same thread up to the number of times that thread called
 397  *   db_lock()/
 398  */
 399 void _db_unlock(const char *file, int line, B_DB *mdb)
 400 {
 401    int errstat;
 402    if ((errstat=rwl_writeunlock(&mdb->lock)) != 0) {
 403       berrno be;
 404       e_msg(file, line, M_FATAL, 0, "rwl_writeunlock failure. stat=%d: ERR=%s\n",
 405            errstat, be.bstrerror(errstat));
 406    }
 407 }
 408
 409 /*
 410  * Start a transaction. This groups inserts and makes things
 411  *  much more efficient. Usually started when inserting
 412  *  file attributes.
 413  */
 414 void db_start_transaction(JCR *jcr, B_DB *mdb)
 415 {
 416    if (!jcr->attr) {
 417       jcr->attr = get_pool_memory(PM_FNAME);
 418    }
 419    if (!jcr->ar) {
 420       jcr->ar = (ATTR_DBR *)malloc(sizeof(ATTR_DBR));
 421    }
 422
 423 #ifdef HAVE_SQLITE
 424    if (!mdb->allow_transactions) {
 425       return;
 426    }
 427    db_lock(mdb);
 428    /* Allow only 10,000 changes per transaction */
 429    if (mdb->transaction && mdb->changes > 10000) {
 430       db_end_transaction(jcr, mdb);
 431    }
 432    if (!mdb->transaction) {
 433       my_sqlite_query(mdb, "BEGIN");  /* begin transaction */
 434       Dmsg0(400, "Start SQLite transaction\n");
 435       mdb->transaction = 1;
 436    }
 437    db_unlock(mdb);
 438 #endif
 439
 440 /*
 441  * This is turned off because transactions break
 442  * if multiple simultaneous jobs are run.
 443  */
 444 #ifdef HAVE_POSTGRESQL
 445    if (!mdb->allow_transactions) {
 446       return;
 447    }
 448    db_lock(mdb);
 449    /* Allow only 25,000 changes per transaction */
 450    if (mdb->transaction && mdb->changes > 25000) {
 451       db_end_transaction(jcr, mdb);
 452    }
 453    if (!mdb->transaction) {
 454       db_sql_query(mdb, "BEGIN", NULL, NULL);  /* begin transaction */
 455       Dmsg0(400, "Start PosgreSQL transaction\n");
 456       mdb->transaction = 1;
 457    }
 458    db_unlock(mdb);
 459 #endif
 460
 461 #ifdef HAVE_INGRES
 462    if (!mdb->allow_transactions) {
 463       return;
 464    }
 465    db_lock(mdb);
 466    /* Allow only 25,000 changes per transaction */
 467    if (mdb->transaction && mdb->changes > 25000) {
 468       db_end_transaction(jcr, mdb);
 469    }
 470    if (!mdb->transaction) {
 471       db_sql_query(mdb, "BEGIN", NULL, NULL);  /* begin transaction */
 472       Dmsg0(400, "Start Ingres transaction\n");
 473       mdb->transaction = 1;
 474    }
 475    db_unlock(mdb);
 476 #endif
 477
 478 #ifdef HAVE_DBI
 479    if (db_type == SQL_TYPE_SQLITE) {
 480       if (!mdb->allow_transactions) {
 481          return;
 482       }
 483       db_lock(mdb);
 484       /* Allow only 10,000 changes per transaction */
 485       if (mdb->transaction && mdb->changes > 10000) {
 486          db_end_transaction(jcr, mdb);
 487       }
 488       if (!mdb->transaction) {
 489          //my_sqlite_query(mdb, "BEGIN");  /* begin transaction */
 490          db_sql_query(mdb, "BEGIN", NULL, NULL);  /* begin transaction */
 491          Dmsg0(400, "Start SQLite transaction\n");
 492          mdb->transaction = 1;
 493       }
 494       db_unlock(mdb);
 495    } else if (db_type == SQL_TYPE_POSTGRESQL) {
 496       if (!mdb->allow_transactions) {
 497          return;
 498       }
 499       db_lock(mdb);
 500       /* Allow only 25,000 changes per transaction */
 501       if (mdb->transaction && mdb->changes > 25000) {
 502          db_end_transaction(jcr, mdb);
 503       }
 504       if (!mdb->transaction) {
 505          db_sql_query(mdb, "BEGIN", NULL, NULL);  /* begin transaction */
 506          Dmsg0(400, "Start PosgreSQL transaction\n");
 507          mdb->transaction = 1;
 508       }
 509       db_unlock(mdb);
 510    }
 511 #endif
 512 }
 513
 514 void db_end_transaction(JCR *jcr, B_DB *mdb)
 515 {
 516    /*
 517     * This can be called during thread cleanup and
 518     *   the db may already be closed.  So simply return.
 519     */
 520    if (!mdb) {
 521       return;
 522    }
 523
 524    if (jcr && jcr->cached_attribute) {
 525       Dmsg0(400, "Flush last cached attribute.\n");
 526       if (!db_create_attributes_record(jcr, mdb, jcr->ar)) {
 527          Jmsg1(jcr, M_FATAL, 0, _("Attribute create error. %s"), db_strerror(jcr->db));
 528       }
 529       jcr->cached_attribute = false;
 530    }
 531
 532 #ifdef HAVE_SQLITE
 533    if (!mdb->allow_transactions) {
 534       return;
 535    }
 536    db_lock(mdb);
 537    if (mdb->transaction) {
 538       my_sqlite_query(mdb, "COMMIT"); /* end transaction */
 539       mdb->transaction = 0;
 540       Dmsg1(400, "End SQLite transaction changes=%d\n", mdb->changes);
 541    }
 542    mdb->changes = 0;
 543    db_unlock(mdb);
 544 #endif
 545
 546
 547
 548 #ifdef HAVE_INGRES
 549    if (!mdb->allow_transactions) {
 550       return;
 551    }
 552    db_lock(mdb);
 553    if (mdb->transaction) {
 554       db_sql_query(mdb, "COMMIT", NULL, NULL); /* end transaction */
 555       mdb->transaction = 0;
 556       Dmsg1(400, "End Ingres transaction changes=%d\n", mdb->changes);
 557    }
 558    mdb->changes = 0;
 559    db_unlock(mdb);
 560 #endif
 561
 562
 563 #ifdef HAVE_POSTGRESQL
 564    if (!mdb->allow_transactions) {
 565       return;
 566    }
 567    db_lock(mdb);
 568    if (mdb->transaction) {
 569       db_sql_query(mdb, "COMMIT", NULL, NULL); /* end transaction */
 570       mdb->transaction = 0;
 571       Dmsg1(400, "End PostgreSQL transaction changes=%d\n", mdb->changes);
 572    }
 573    mdb->changes = 0;
 574    db_unlock(mdb);
 575 #endif
 576
 577 #ifdef HAVE_DBI
 578    if (db_type == SQL_TYPE_SQLITE) {
 579       if (!mdb->allow_transactions) {
 580          return;
 581       }
 582       db_lock(mdb);
 583       if (mdb->transaction) {
 584          //my_sqlite_query(mdb, "COMMIT"); /* end transaction */
 585          db_sql_query(mdb, "COMMIT", NULL, NULL); /* end transaction */
 586          mdb->transaction = 0;
 587          Dmsg1(400, "End SQLite transaction changes=%d\n", mdb->changes);
 588       }
 589       mdb->changes = 0;
 590       db_unlock(mdb);
 591    } else if (db_type == SQL_TYPE_POSTGRESQL) {
 592       if (!mdb->allow_transactions) {
 593          return;
 594       }
 595       db_lock(mdb);
 596       if (mdb->transaction) {
 597          db_sql_query(mdb, "COMMIT", NULL, NULL); /* end transaction */
 598          mdb->transaction = 0;
 599          Dmsg1(400, "End PostgreSQL transaction changes=%d\n", mdb->changes);
 600       }
 601       mdb->changes = 0;
 602       db_unlock(mdb);
 603    }
 604 #endif
 605 }
 606
 607 /*
 608  * Given a full filename, split it into its path
 609  *  and filename parts. They are returned in pool memory
 610  *  in the mdb structure.
 611  */
 612 void split_path_and_file(JCR *jcr, B_DB *mdb, const char *fname)
 613 {
 614    const char *p, *f;
 615
 616    /* Find path without the filename.
 617     * I.e. everything after the last / is a "filename".
 618     * OK, maybe it is a directory name, but we treat it like
 619     * a filename. If we don't find a / then the whole name
 620     * must be a path name (e.g. c:).
 621     */
 622    for (p=f=fname; *p; p++) {
 623       if (IsPathSeparator(*p)) {
 624          f = p;                       /* set pos of last slash */
 625       }
 626    }
 627    if (IsPathSeparator(*f)) {                   /* did we find a slash? */
 628       f++;                            /* yes, point to filename */
 629    } else {                           /* no, whole thing must be path name */
 630       f = p;
 631    }
 632
 633    /* If filename doesn't exist (i.e. root directory), we
 634     * simply create a blank name consisting of a single
 635     * space. This makes handling zero length filenames
 636     * easier.
 637     */
 638    mdb->fnl = p - f;
 639    if (mdb->fnl > 0) {
 640       mdb->fname = check_pool_memory_size(mdb->fname, mdb->fnl+1);
 641       memcpy(mdb->fname, f, mdb->fnl);    /* copy filename */
 642       mdb->fname[mdb->fnl] = 0;
 643    } else {
 644       mdb->fname[0] = 0;
 645       mdb->fnl = 0;
 646    }
 647
 648    mdb->pnl = f - fname;
 649    if (mdb->pnl > 0) {
 650       mdb->path = check_pool_memory_size(mdb->path, mdb->pnl+1);
 651       memcpy(mdb->path, fname, mdb->pnl);
 652       mdb->path[mdb->pnl] = 0;
 653    } else {
 654       Mmsg1(&mdb->errmsg, _("Path length is zero. File=%s\n"), fname);
 655       Jmsg(jcr, M_ERROR, 0, "%s", mdb->errmsg);
 656       mdb->path[0] = 0;
 657       mdb->pnl = 0;
 658    }
 659
 660    Dmsg2(500, "split path=%s file=%s\n", mdb->path, mdb->fname);
 661 }
 662
 663 /*
 664  * Set maximum field length to something reasonable
 665  */
 666 static int max_length(int max_length)
 667 {
 668    int max_len = max_length;
 669    /* Sanity check */
 670    if (max_len < 0) {
 671       max_len = 2;
 672    } else if (max_len > 100) {
 673       max_len = 100;
 674    }
 675    return max_len;
 676 }
 677
 678 /*
 679  * List dashes as part of header for listing SQL results in a table
 680  */
 681 void
 682 list_dashes(B_DB *mdb, DB_LIST_HANDLER *send, void *ctx)
 683 {
 684    SQL_FIELD  *field;
 685    int i, j;
 686    int len;
 687
 688    sql_field_seek(mdb, 0);
 689    send(ctx, "+");
 690    for (i = 0; i < sql_num_fields(mdb); i++) {
 691       field = sql_fetch_field(mdb);
 692       if (!field) {
 693          break;
 694       }
 695       len = max_length(field->max_length + 2);
 696       for (j = 0; j < len; j++) {
 697          send(ctx, "-");
 698       }
 699       send(ctx, "+");
 700    }
 701    send(ctx, "\n");
 702 }
 703
 704 /*
 705  * If full_list is set, we list vertically, otherwise, we
 706  * list on one line horizontally.
 707  */
 708 void
 709 list_result(JCR *jcr, B_DB *mdb, DB_LIST_HANDLER *send, void *ctx, e_list_type type)
 710 {
 711    SQL_FIELD *field;
 712    SQL_ROW row;
 713    int i, col_len, max_len = 0;
 714    char buf[2000], ewc[30];
 715
 716    Dmsg0(800, "list_result starts\n");
 717    if (mdb->result == NULL || sql_num_rows(mdb) == 0) {
 718       send(ctx, _("No results to list.\n"));
 719       return;
 720    }
 721
 722    Dmsg1(800, "list_result starts looking at %d fields\n", sql_num_fields(mdb));
 723    /* determine column display widths */
 724    sql_field_seek(mdb, 0);
 725    for (i = 0; i < sql_num_fields(mdb); i++) {
 726       Dmsg1(800, "list_result processing field %d\n", i);
 727       field = sql_fetch_field(mdb);
 728       if (!field) {
 729          break;
 730       }
 731       col_len = cstrlen(field->name);
 732       if (type == VERT_LIST) {
 733          if (col_len > max_len) {
 734             max_len = col_len;
 735          }
 736       } else {
 737          if (IS_NUM(field->type) && (int)field->max_length > 0) { /* fixup for commas */
 738             field->max_length += (field->max_length - 1) / 3;
 739          }
 740          if (col_len < (int)field->max_length) {
 741             col_len = field->max_length;
 742          }
 743          if (col_len < 4 && !IS_NOT_NULL(field->flags)) {
 744             col_len = 4;                 /* 4 = length of the word "NULL" */
 745          }
 746          field->max_length = col_len;    /* reset column info */
 747       }
 748    }
 749
 750    Dmsg0(800, "list_result finished first loop\n");
 751    if (type == VERT_LIST) {
 752       goto vertical_list;
 753    }
 754
 755    Dmsg1(800, "list_result starts second loop looking at %d fields\n", sql_num_fields(mdb));
 756    list_dashes(mdb, send, ctx);
 757    send(ctx, "|");
 758    sql_field_seek(mdb, 0);
 759    for (i = 0; i < sql_num_fields(mdb); i++) {
 760       Dmsg1(800, "list_result looking at field %d\n", i);
 761       field = sql_fetch_field(mdb);
 762       if (!field) {
 763          break;
 764       }
 765       max_len = max_length(field->max_length);
 766       bsnprintf(buf, sizeof(buf), " %-*s |", max_len, field->name);
 767       send(ctx, buf);
 768    }
 769    send(ctx, "\n");
 770    list_dashes(mdb, send, ctx);
 771
 772    Dmsg1(800, "list_result starts third loop looking at %d fields\n", sql_num_fields(mdb));
 773    while ((row = sql_fetch_row(mdb)) != NULL) {
 774       sql_field_seek(mdb, 0);
 775       send(ctx, "|");
 776       for (i = 0; i < sql_num_fields(mdb); i++) {
 777          field = sql_fetch_field(mdb);
 778          if (!field) {
 779             break;
 780          }
 781          max_len = max_length(field->max_length);
 782          if (row[i] == NULL) {
 783             bsnprintf(buf, sizeof(buf), " %-*s |", max_len, "NULL");
 784          } else if (IS_NUM(field->type) && !jcr->gui && is_an_integer(row[i])) {
 785             bsnprintf(buf, sizeof(buf), " %*s |", max_len,
 786                       add_commas(row[i], ewc));
 787          } else {
 788             bsnprintf(buf, sizeof(buf), " %-*s |", max_len, row[i]);
 789          }
 790          send(ctx, buf);
 791       }
 792       send(ctx, "\n");
 793    }
 794    list_dashes(mdb, send, ctx);
 795    return;
 796
 797 vertical_list:
 798
 799    Dmsg1(800, "list_result starts vertical list at %d fields\n", sql_num_fields(mdb));
 800    while ((row = sql_fetch_row(mdb)) != NULL) {
 801       sql_field_seek(mdb, 0);
 802       for (i = 0; i < sql_num_fields(mdb); i++) {
 803          field = sql_fetch_field(mdb);
 804          if (!field) {
 805             break;
 806          }
 807          if (row[i] == NULL) {
 808             bsnprintf(buf, sizeof(buf), " %*s: %s\n", max_len, field->name, "NULL");
 809          } else if (IS_NUM(field->type) && !jcr->gui && is_an_integer(row[i])) {
 810             bsnprintf(buf, sizeof(buf), " %*s: %s\n", max_len, field->name,
 811                 add_commas(row[i], ewc));
 812          } else {
 813             bsnprintf(buf, sizeof(buf), " %*s: %s\n", max_len, field->name, row[i]);
 814          }
 815          send(ctx, buf);
 816       }
 817       send(ctx, "\n");
 818    }
 819    return;
 820 }
 821
 822 /*
 823  * Open a new connexion to mdb catalog. This function is used
 824  * by batch and accurate mode.
 825  */
 826 bool db_open_batch_connexion(JCR *jcr, B_DB *mdb)
 827 {
 828 #ifdef HAVE_BATCH_FILE_INSERT
 829    const int multi_db = true;   /* we force a new connection only if batch insert is enabled */
 830 #else
 831    const int multi_db = false;
 832 #endif
 833
 834    if (!jcr->db_batch) {
 835       jcr->db_batch = db_init_database(jcr,
 836                                       mdb->db_name,
 837                                       mdb->db_user,
 838                                       mdb->db_password,
 839                                       mdb->db_address,
 840                                       mdb->db_port,
 841                                       mdb->db_socket,
 842                                       multi_db /* multi_db = true when using batch mode */);
 843       if (!jcr->db_batch) {
 844          Mmsg0(&mdb->errmsg, _("Could not init database batch connection"));
 845          Jmsg(jcr, M_FATAL, 0, "%s", mdb->errmsg);
 846          return false;
 847       }
 848
 849       if (!db_open_database(jcr, jcr->db_batch)) {
 850          Mmsg2(&mdb->errmsg,  _("Could not open database \"%s\": ERR=%s\n"),
 851               jcr->db_batch->db_name, db_strerror(jcr->db_batch));
 852          Jmsg(jcr, M_FATAL, 0, "%s", mdb->errmsg);
 853          return false;
 854       }
 855       Dmsg3(100, "initdb ref=%d connected=%d db=%p\n", jcr->db_batch->ref_count,
 856             jcr->db_batch->connected, jcr->db_batch->db);
 857
 858    }
 859    return true;
 860 }
 861
 862 /*
 863  * !!! WARNING !!! Use this function only when bacula is stopped.
 864  * ie, after a fatal signal and before exiting the program
 865  * Print information about a B_DB object.
 866  */
 867 void db_debug_print(JCR *jcr, FILE *fp)
 868 {
 869    B_DB *mdb = jcr->db;
 870
 871    if (!mdb) {
 872       return;
 873    }
 874
 875    fprintf(fp, "B_DB=%p db_name=%s db_user=%s connected=%i\n",
 876            mdb, NPRTB(mdb->db_name), NPRTB(mdb->db_user), mdb->connected);
 877    fprintf(fp, "\tcmd=\"%s\" changes=%i\n", NPRTB(mdb->cmd), mdb->changes);
 878    if (mdb->lock.valid == RWLOCK_VALID) {
 879       fprintf(fp, "\tRWLOCK=%p w_active=%i w_wait=%i\n", &mdb->lock, mdb->lock.w_active, mdb->lock.w_wait);
 880    }
 881 }
 882
 883 #endif /* HAVE_SQLITE3 || HAVE_MYSQL || HAVE_SQLITE || HAVE_POSTGRESQL || HAVE_INGRES*/