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

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2000-2011 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 three of the GNU Affero 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 Affero 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 SQLite
  30  *
  31  *    Kern Sibbald, January 2002
  32  *
  33  * Major rewrite by Marco van Wieringen, January 2010 for catalog refactoring.
  34  */
  35
  36 #include "bacula.h"
  37
  38 #if HAVE_SQLITE3
  39
  40 #include "cats.h"
  41 #include "bdb_priv.h"
  42 #include <sqlite3.h>
  43 #include "bdb_sqlite.h"
  44
  45 /* -----------------------------------------------------------------------
  46  *
  47  *    SQLite dependent defines and subroutines
  48  *
  49  * -----------------------------------------------------------------------
  50  */
  51
  52 /*
  53  * List of open databases
  54  */
  55 static dlist *db_list = NULL;
  56
  57 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  58
  59 /*
  60  * When using mult_db_connections = true,
  61  * sqlite can be BUSY. We just need sleep a little in this case.
  62  */
  63 static int sqlite_busy_handler(void *arg, int calls)
  64 {
  65    bmicrosleep(0, 500);
  66    return 1;
  67 }
  68
  69 B_DB_SQLITE::B_DB_SQLITE(JCR *jcr,
  70                          const char *db_driver,
  71                          const char *db_name,
  72                          const char *db_user,
  73                          const char *db_password,
  74                          const char *db_address,
  75                          int db_port,
  76                          const char *db_socket,
  77                          bool mult_db_connections,
  78                          bool disable_batch_insert)
  79 {
  80    /*
  81     * Initialize the parent class members.
  82     */
  83    m_db_interface_type = SQL_INTERFACE_TYPE_SQLITE3;
  84    m_db_type = SQL_TYPE_SQLITE3;
  85    m_db_driver = bstrdup("SQLite3");
  86    m_db_name = bstrdup(db_name);
  87    if (disable_batch_insert) {
  88       m_disabled_batch_insert = true;
  89       m_have_batch_insert = false;
  90    } else {
  91       m_disabled_batch_insert = false;
  92 #if defined(USE_BATCH_FILE_INSERT)
  93 #if defined(HAVE_SQLITE3_THREADSAFE)
  94       m_have_batch_insert = sqlite3_threadsafe();
  95 #else
  96       m_have_batch_insert = false;
  97 #endif /* HAVE_SQLITE3_THREADSAFE */
  98 #else
  99       m_have_batch_insert = false;
 100 #endif /* USE_BATCH_FILE_INSERT */
 101    }
 102    errmsg = get_pool_memory(PM_EMSG); /* get error message buffer */
 103    *errmsg = 0;
 104    cmd = get_pool_memory(PM_EMSG);    /* get command buffer */
 105    cached_path = get_pool_memory(PM_FNAME);
 106    cached_path_id = 0;
 107    m_ref_count = 1;
 108    fname = get_pool_memory(PM_FNAME);
 109    path = get_pool_memory(PM_FNAME);
 110    esc_name = get_pool_memory(PM_FNAME);
 111    esc_path = get_pool_memory(PM_FNAME);
 112    esc_obj  = get_pool_memory(PM_FNAME);
 113    m_allow_transactions = mult_db_connections;
 114
 115    /* At this time, when mult_db_connections == true, this is for
 116     * specific console command such as bvfs or batch mode, and we don't
 117     * want to share a batch mode or bvfs. In the future, we can change
 118     * the creation function to add this parameter.
 119     */
 120    m_dedicated = mult_db_connections;
 121
 122    /*
 123     * Initialize the private members.
 124     */
 125    m_db_handle = NULL;
 126    m_result = NULL;
 127    m_sqlite_errmsg = NULL;
 128
 129    /*
 130     * Put the db in the list.
 131     */
 132    if (db_list == NULL) {
 133       db_list = New(dlist(this, &this->m_link));
 134    }
 135    db_list->append(this);
 136 }
 137
 138 B_DB_SQLITE::~B_DB_SQLITE()
 139 {
 140 }
 141
 142 /*
 143  * Now actually open the database.  This can generate errors,
 144  * which are returned in the errmsg
 145  *
 146  * DO NOT close the database or delete mdb here !!!!
 147  */
 148 bool B_DB_SQLITE::db_open_database(JCR *jcr)
 149 {
 150    bool retval = false;
 151    char *db_path;
 152    int len;
 153    struct stat statbuf;
 154    int ret;
 155    int errstat;
 156    int retry = 0;
 157
 158    P(mutex);
 159    if (m_connected) {
 160       retval = true;
 161       goto bail_out;
 162    }
 163
 164    if ((errstat=rwl_init(&m_lock)) != 0) {
 165       berrno be;
 166       Mmsg1(&errmsg, _("Unable to initialize DB lock. ERR=%s\n"),
 167             be.bstrerror(errstat));
 168       goto bail_out;
 169    }
 170
 171    /*
 172     * Open the database
 173     */
 174    len = strlen(working_directory) + strlen(m_db_name) + 5;
 175    db_path = (char *)malloc(len);
 176    strcpy(db_path, working_directory);
 177    strcat(db_path, "/");
 178    strcat(db_path, m_db_name);
 179    strcat(db_path, ".db");
 180    if (stat(db_path, &statbuf) != 0) {
 181       Mmsg1(&errmsg, _("Database %s does not exist, please create it.\n"),
 182          db_path);
 183       free(db_path);
 184       goto bail_out;
 185    }
 186
 187    for (m_db_handle = NULL; !m_db_handle && retry++ < 10; ) {
 188       ret = sqlite3_open(db_path, &m_db_handle);
 189       if (ret != SQLITE_OK) {
 190          m_sqlite_errmsg = (char *)sqlite3_errmsg(m_db_handle);
 191          sqlite3_close(m_db_handle);
 192          m_db_handle = NULL;
 193       } else {
 194          m_sqlite_errmsg = NULL;
 195       }
 196
 197       Dmsg0(300, "sqlite_open\n");
 198       if (!m_db_handle) {
 199          bmicrosleep(1, 0);
 200       }
 201    }
 202    if (m_db_handle == NULL) {
 203       Mmsg2(&errmsg, _("Unable to open Database=%s. ERR=%s\n"),
 204          db_path, m_sqlite_errmsg ? m_sqlite_errmsg : _("unknown"));
 205       free(db_path);
 206       goto bail_out;
 207    }
 208    m_connected = true;
 209    free(db_path);
 210
 211    /*
 212     * Set busy handler to wait when we use mult_db_connections = true
 213     */
 214    sqlite3_busy_handler(m_db_handle, sqlite_busy_handler, NULL);
 215
 216 #if defined(SQLITE3_INIT_QUERY)
 217    sql_query(SQLITE3_INIT_QUERY);
 218 #endif
 219
 220    if (!check_tables_version(jcr, this)) {
 221       goto bail_out;
 222    }
 223
 224    retval = true;
 225
 226 bail_out:
 227    V(mutex);
 228    return retval;
 229 }
 230
 231 void B_DB_SQLITE::db_close_database(JCR *jcr)
 232 {
 233    db_end_transaction(jcr);
 234    P(mutex);
 235    m_ref_count--;
 236    if (m_ref_count == 0) {
 237       sql_free_result();
 238       db_list->remove(this);
 239       if (m_connected && m_db_handle) {
 240          sqlite3_close(m_db_handle);
 241       }
 242       rwl_destroy(&m_lock);
 243       free_pool_memory(errmsg);
 244       free_pool_memory(cmd);
 245       free_pool_memory(cached_path);
 246       free_pool_memory(fname);
 247       free_pool_memory(path);
 248       free_pool_memory(esc_name);
 249       free_pool_memory(esc_path);
 250       free_pool_memory(esc_obj);
 251       if (m_db_driver) {
 252          free(m_db_driver);
 253       }
 254       if (m_db_name) {
 255          free(m_db_name);
 256       }
 257       delete this;
 258       if (db_list->size() == 0) {
 259          delete db_list;
 260          db_list = NULL;
 261       }
 262    }
 263    V(mutex);
 264 }
 265
 266 void B_DB_SQLITE::db_thread_cleanup(void)
 267 {
 268    sqlite3_thread_cleanup();
 269 }
 270
 271 /*
 272  * Escape strings so that SQLite is happy
 273  *
 274  *   NOTE! len is the length of the old string. Your new
 275  *         string must be long enough (max 2*old+1) to hold
 276  *         the escaped output.
 277  */
 278 void B_DB_SQLITE::db_escape_string(JCR *jcr, char *snew, char *old, int len)
 279 {
 280    char *n, *o;
 281
 282    n = snew;
 283    o = old;
 284    while (len--) {
 285       switch (*o) {
 286       case '\'':
 287          *n++ = '\'';
 288          *n++ = '\'';
 289          o++;
 290          break;
 291       case 0:
 292          *n++ = '\\';
 293          *n++ = 0;
 294          o++;
 295          break;
 296       default:
 297          *n++ = *o++;
 298          break;
 299       }
 300    }
 301    *n = 0;
 302 }
 303
 304 /*
 305  * Escape binary object so that SQLite is happy
 306  * Memory is stored in B_DB struct, no need to free it
 307  *
 308  * TODO: this should be implemented  (escape \0)
 309  */
 310 char *B_DB_SQLITE::db_escape_object(JCR *jcr, char *old, int len)
 311 {
 312    int l;
 313    int max = len*2;           /* TODO: too big, should be *4/3 */
 314
 315    esc_obj = check_pool_memory_size(esc_obj, max);
 316    l = bin_to_base64(esc_obj, max, old, len, true);
 317    esc_obj[l] = 0;
 318    ASSERT(l < max);    /* TODO: add check for l */
 319
 320    return esc_obj;
 321 }
 322
 323 /*
 324  * Unescape binary object so that SQLIte is happy
 325  *
 326  * TODO: need to be implemented (escape \0)
 327  */
 328
 329 void B_DB_SQLITE::db_unescape_object(JCR *jcr, char *from, int32_t expected_len,
 330                                      POOLMEM **dest, int32_t *dest_len)
 331 {
 332    if (!from) {
 333       *dest[0] = 0;
 334       *dest_len = 0;
 335       return;
 336    }
 337    *dest = check_pool_memory_size(*dest, expected_len+1);
 338    base64_to_bin(*dest, expected_len+1, from, strlen(from));
 339    *dest_len = expected_len;
 340    (*dest)[expected_len]=0;
 341 }
 342
 343 /*
 344  * Start a transaction. This groups inserts and makes things
 345  * much more efficient. Usually started when inserting
 346  * file attributes.
 347  */
 348 void B_DB_SQLITE::db_start_transaction(JCR *jcr)
 349 {
 350    if (!jcr->attr) {
 351       jcr->attr = get_pool_memory(PM_FNAME);
 352    }
 353    if (!jcr->ar) {
 354       jcr->ar = (ATTR_DBR *)malloc(sizeof(ATTR_DBR));
 355    }
 356
 357    if (!m_allow_transactions) {
 358       return;
 359    }
 360
 361    db_lock(this);
 362    /*
 363     * Allow only 10,000 changes per transaction
 364     */
 365    if (m_transaction && changes > 10000) {
 366       db_end_transaction(jcr);
 367    }
 368    if (!m_transaction) {
 369       sql_query("BEGIN");  /* begin transaction */
 370       Dmsg0(400, "Start SQLite transaction\n");
 371       m_transaction = true;
 372    }
 373    db_unlock(this);
 374 }
 375
 376 void B_DB_SQLITE::db_end_transaction(JCR *jcr)
 377 {
 378    if (jcr && jcr->cached_attribute) {
 379       Dmsg0(400, "Flush last cached attribute.\n");
 380       if (!db_create_attributes_record(jcr, this, jcr->ar)) {
 381          Jmsg1(jcr, M_FATAL, 0, _("Attribute create error. %s"), db_strerror(jcr->db));
 382       }
 383       jcr->cached_attribute = false;
 384    }
 385
 386    if (!m_allow_transactions) {
 387       return;
 388    }
 389
 390    db_lock(this);
 391    if (m_transaction) {
 392       sql_query("COMMIT"); /* end transaction */
 393       m_transaction = false;
 394       Dmsg1(400, "End SQLite transaction changes=%d\n", changes);
 395    }
 396    changes = 0;
 397    db_unlock(this);
 398 }
 399
 400 struct rh_data {
 401    B_DB_SQLITE *mdb;
 402    DB_RESULT_HANDLER *result_handler;
 403    void *ctx;
 404    bool initialized;
 405 };
 406
 407 /*
 408  * Convert SQLite's callback into Bacula DB callback
 409  */
 410 static int sqlite_result_handler(void *arh_data, int num_fields, char **rows, char **col_names)
 411 {
 412    struct rh_data *rh_data = (struct rh_data *)arh_data;
 413
 414    /* The db_sql_query doesn't have access to m_results, so if we wan't to get
 415     * fields information, we need to use col_names
 416     */
 417    if (!rh_data->initialized) {
 418       rh_data->mdb->set_column_names(col_names, num_fields);
 419       rh_data->initialized = true;
 420    }
 421    if (rh_data->result_handler) {
 422       (*(rh_data->result_handler))(rh_data->ctx, num_fields, rows);
 423    }
 424
 425    return 0;
 426 }
 427
 428 /*
 429  * Submit a general SQL command (cmd), and for each row returned,
 430  *  the result_handler is called with the ctx.
 431  */
 432 bool B_DB_SQLITE::db_sql_query(const char *query, DB_RESULT_HANDLER *result_handler, void *ctx)
 433 {
 434    bool retval = false;
 435    int stat;
 436    struct rh_data rh_data;
 437
 438    Dmsg1(500, "db_sql_query starts with '%s'\n", query);
 439
 440    db_lock(this);
 441    if (m_sqlite_errmsg) {
 442       sqlite3_free(m_sqlite_errmsg);
 443       m_sqlite_errmsg = NULL;
 444    }
 445    sql_free_result();
 446
 447    rh_data.ctx = ctx;
 448    rh_data.mdb = this;
 449    rh_data.initialized = false;
 450    rh_data.result_handler = result_handler;
 451
 452    stat = sqlite3_exec(m_db_handle, query, sqlite_result_handler,
 453                        (void *)&rh_data, &m_sqlite_errmsg);
 454
 455    if (stat != SQLITE_OK) {
 456       Mmsg(errmsg, _("Query failed: %s: ERR=%s\n"), query, sql_strerror());
 457       Dmsg0(500, "db_sql_query finished\n");
 458       goto bail_out;
 459    }
 460    Dmsg0(500, "db_sql_query finished\n");
 461    sql_free_result();
 462    retval = true;
 463
 464 bail_out:
 465    db_unlock(this);
 466    return retval;
 467 }
 468
 469 /*
 470  * Submit a sqlite query and retrieve all the data
 471  */
 472 bool B_DB_SQLITE::sql_query(const char *query, int flags)
 473 {
 474    int stat;
 475    bool retval = false;
 476
 477    Dmsg1(500, "sql_query starts with '%s'\n", query);
 478
 479    sql_free_result();
 480    if (m_sqlite_errmsg) {
 481       sqlite3_free(m_sqlite_errmsg);
 482       m_sqlite_errmsg = NULL;
 483    }
 484
 485    stat = sqlite3_get_table(m_db_handle, (char *)query, &m_result,
 486                             &m_num_rows, &m_num_fields, &m_sqlite_errmsg);
 487
 488    m_row_number = 0;               /* no row fetched */
 489    if (stat != 0) {                   /* something went wrong */
 490       m_num_rows = m_num_fields = 0;
 491       Dmsg0(500, "sql_query finished\n");
 492    } else {
 493       Dmsg0(500, "sql_query finished\n");
 494       retval = true;
 495    }
 496    return retval;
 497 }
 498
 499 void B_DB_SQLITE::sql_free_result(void)
 500 {
 501    db_lock(this);
 502    if (m_fields) {
 503       free(m_fields);
 504       m_fields = NULL;
 505    }
 506    if (m_result) {
 507       sqlite3_free_table(m_result);
 508       m_result = NULL;
 509    }
 510    m_col_names = NULL;
 511    m_num_rows = m_num_fields = 0;
 512    db_unlock(this);
 513 }
 514
 515 /*
 516  * Fetch one row at a time
 517  */
 518 SQL_ROW B_DB_SQLITE::sql_fetch_row(void)
 519 {
 520    if (!m_result || (m_row_number >= m_num_rows)) {
 521       return NULL;
 522    }
 523    m_row_number++;
 524    return &m_result[m_num_fields * m_row_number];
 525 }
 526
 527 const char *B_DB_SQLITE::sql_strerror(void)
 528 {
 529    return m_sqlite_errmsg ? m_sqlite_errmsg : "unknown";
 530 }
 531
 532 void B_DB_SQLITE::sql_data_seek(int row)
 533 {
 534    /*
 535     * Set the row number to be returned on the next call to sql_fetch_row
 536     */
 537    m_row_number = row;
 538 }
 539
 540 int B_DB_SQLITE::sql_affected_rows(void)
 541 {
 542    return sqlite3_changes(m_db_handle);
 543 }
 544
 545 uint64_t B_DB_SQLITE::sql_insert_autokey_record(const char *query, const char *table_name)
 546 {
 547    /*
 548     * First execute the insert query and then retrieve the currval.
 549     */
 550    if (!sql_query(query)) {
 551       return 0;
 552    }
 553
 554    m_num_rows = sql_affected_rows();
 555    if (m_num_rows != 1) {
 556       return 0;
 557    }
 558
 559    changes++;
 560
 561    return sqlite3_last_insert_rowid(m_db_handle);
 562 }
 563
 564 SQL_FIELD *B_DB_SQLITE::sql_fetch_field(void)
 565 {
 566    int i, j, len;
 567
 568    /* We are in the middle of a db_sql_query and we want to get fields info */
 569    if (m_col_names != NULL) {
 570       if (m_num_fields > m_field_number) {
 571          m_sql_field.name = m_col_names[m_field_number];
 572          /* We don't have the maximum field length, so we can use 80 as
 573           * estimation.
 574           */
 575          len = MAX(cstrlen(m_sql_field.name), 80/m_num_fields);
 576          m_sql_field.max_length = len;
 577
 578          m_field_number++;
 579          m_sql_field.type = 0;  /* not numeric */
 580          m_sql_field.flags = 1; /* not null */
 581          return &m_sql_field;
 582       } else {                  /* too much fetch_field() */
 583          return NULL;
 584       }
 585    }
 586
 587    /* We are after a sql_query() that stores the result in m_results */
 588    if (!m_fields || m_fields_size < m_num_fields) {
 589       if (m_fields) {
 590          free(m_fields);
 591          m_fields = NULL;
 592       }
 593       Dmsg1(500, "allocating space for %d fields\n", m_num_fields);
 594       m_fields = (SQL_FIELD *)malloc(sizeof(SQL_FIELD) * m_num_fields);
 595       m_fields_size = m_num_fields;
 596
 597       for (i = 0; i < m_num_fields; i++) {
 598          Dmsg1(500, "filling field %d\n", i);
 599          m_fields[i].name = m_result[i];
 600          m_fields[i].max_length = cstrlen(m_fields[i].name);
 601          for (j = 1; j <= m_num_rows; j++) {
 602             if (m_result[i + m_num_fields * j]) {
 603                len = (uint32_t)cstrlen(m_result[i + m_num_fields * j]);
 604             } else {
 605                len = 0;
 606             }
 607             if (len > m_fields[i].max_length) {
 608                m_fields[i].max_length = len;
 609             }
 610          }
 611          m_fields[i].type = 0;
 612          m_fields[i].flags = 1;        /* not null */
 613
 614          Dmsg4(500, "sql_fetch_field finds field '%s' has length='%d' type='%d' and IsNull=%d\n",
 615                m_fields[i].name, m_fields[i].max_length, m_fields[i].type, m_fields[i].flags);
 616       }
 617    }
 618
 619    /*
 620     * Increment field number for the next time around
 621     */
 622    return &m_fields[m_field_number++];
 623 }
 624
 625 bool B_DB_SQLITE::sql_field_is_not_null(int field_type)
 626 {
 627    switch (field_type) {
 628    case 1:
 629       return true;
 630    default:
 631       return false;
 632    }
 633 }
 634
 635 bool B_DB_SQLITE::sql_field_is_numeric(int field_type)
 636 {
 637    switch (field_type) {
 638    case 1:
 639       return true;
 640    default:
 641       return false;
 642    }
 643 }
 644
 645 /*
 646  * Returns true if OK
 647  *         false if failed
 648  */
 649 bool B_DB_SQLITE::sql_batch_start(JCR *jcr)
 650 {
 651    bool retval;
 652
 653    db_lock(this);
 654    retval = sql_query("CREATE TEMPORARY TABLE batch ("
 655                               "FileIndex integer,"
 656                               "JobId integer,"
 657                               "Path blob,"
 658                               "Name blob,"
 659                               "LStat tinyblob,"
 660                               "MD5 tinyblob,"
 661                               "DeltaSeq integer)");
 662    db_unlock(this);
 663
 664    return retval;
 665 }
 666
 667 /* set error to something to abort operation */
 668 /*
 669  * Returns true if OK
 670  *         false if failed
 671  */
 672 bool B_DB_SQLITE::sql_batch_end(JCR *jcr, const char *error)
 673 {
 674    m_status = 0;
 675
 676    return true;
 677 }
 678
 679 /*
 680  * Returns true if OK
 681  *         false if failed
 682  */
 683 bool B_DB_SQLITE::sql_batch_insert(JCR *jcr, ATTR_DBR *ar)
 684 {
 685    const char *digest;
 686    char ed1[50];
 687
 688    esc_name = check_pool_memory_size(esc_name, fnl*2+1);
 689    db_escape_string(jcr, esc_name, fname, fnl);
 690
 691    esc_path = check_pool_memory_size(esc_path, pnl*2+1);
 692    db_escape_string(jcr, esc_path, path, pnl);
 693
 694    if (ar->Digest == NULL || ar->Digest[0] == 0) {
 695       digest = "0";
 696    } else {
 697       digest = ar->Digest;
 698    }
 699
 700    Mmsg(cmd, "INSERT INTO batch VALUES "
 701         "(%u,%s,'%s','%s','%s','%s',%u)",
 702         ar->FileIndex, edit_int64(ar->JobId,ed1), esc_path,
 703         esc_name, ar->attr, digest, ar->DeltaSeq);
 704
 705    return sql_query(cmd);
 706 }
 707
 708 /*
 709  * Initialize database data structure. In principal this should
 710  * never have errors, or it is really fatal.
 711  */
 712 B_DB *db_init_database(JCR *jcr, const char *db_driver, const char *db_name,
 713                        const char *db_user, const char *db_password,
 714                        const char *db_address, int db_port,
 715                        const char *db_socket, bool mult_db_connections,
 716                        bool disable_batch_insert)
 717 {
 718    B_DB *mdb = NULL;
 719
 720    P(mutex);                          /* lock DB queue */
 721    /*
 722     * Look to see if DB already open
 723     */
 724    if (db_list && !mult_db_connections) {
 725       foreach_dlist(mdb, db_list) {
 726          if (mdb->db_match_database(db_driver, db_name, db_address, db_port)) {
 727             Dmsg1(300, "DB REopen %s\n", db_name);
 728             mdb->increment_refcount();
 729             goto bail_out;
 730          }
 731       }
 732    }
 733    Dmsg0(300, "db_init_database first time\n");
 734    mdb = New(B_DB_SQLITE(jcr, db_driver, db_name, db_user, db_password,
 735                          db_address, db_port, db_socket, mult_db_connections,
 736                          disable_batch_insert));
 737
 738 bail_out:
 739    V(mutex);
 740    return mdb;
 741 }
 742
 743 #endif /* HAVE_SQLITE3 */