sqlite3_wrapper.c

Go to the documentation of this file.

/*
 * File Name: sqlite3_wrapper.c
 */

/*
 * This file is part of libermetadb.
 *
 * libermetadb is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 2 of the License, or
 * (at your option) any later version.
 *
 * libermetadb is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

/**
 * Copyright (C) 2008 iRex Technologies B.V.
 * All rights reserved.
 */

//----------------------------------------------------------------------------
// Include Files
//----------------------------------------------------------------------------

// system include files, between < >
#include <glib.h>
#include <sqlite3.h>
#include <sys/types.h>
#include <unistd.h>

// ereader include files, between < >
#include <liberutils/er_error.h>

// local include files, between " "
#define LOGGING_ON 0
#include "ermetadb_log.h"
#include "ermetadb.h"
#include "ermetadb_error.h"
#include "ermetadb_private.h"
#include "metadata_table.h"
#include "metadata_table_private.h"
#include "sqlite3_wrapper.h"

#define UNUSED(x) (void)(x)


//----------------------------------------------------------------------------
// Global Constants
//----------------------------------------------------------------------------

static const int BEGIN_MAX_RETRIES  =  20;      // max. retries for a begin statement
static const int BEGIN_RETRY_MS     = 250;      // interval in ms. between retries
                                                // 20 x 250 ms = 5 seconds

static const int COMMIT_MAX_RETRIES =  30;      // max. retries for a begin statement
static const int COMMIT_RETRY_MS    = 100;      // interval in ms. between retries
                                                // 30 x 100 ms = 3 seconds

static const int STEP_MAX_RETRIES   =  10;      // max. retries per execution step
static const int STEP_RETRY_MS      = 100;      // interval in ms. between retries
                                                // 10 x 100 ms = 1 second

//============================================================================
// Functions Implementation
//============================================================================

/**---------------------------------------------------------------------------
 *
 * Name :  sql3_finalize_statements
 *
 * @brief  Release compiled sql statements
 *
 * @param  [in]  statements - GPtrArray with compiled statements
 *
 *--------------------------------------------------------------------------*/
static void sql3_finalize_statements(GPtrArray *statements)
{
    LOGPRINTF("entry");

    if (statements)
    {
        unsigned int i;
        for (i = 0 ; i < statements->len ; i++)
        {
            sqlite3_stmt *stmt = g_ptr_array_index(statements, i);
            sqlite3_finalize(stmt);
        }
        g_ptr_array_free(statements, TRUE);
    }
}


/**---------------------------------------------------------------------------
 *
 * Name :  sql3_prepare_statements
 *
 * @brief  Parse sql string and compile into byte code
 *
 * @param  [in]  db - sqlite database object
 * @param  [in]  sql - SQL statement(s) as text
 * @param  [out] statements - GPtrArray with compiled statements, or NULL
 *
 * @return ER_OK or error code
 *
 *--------------------------------------------------------------------------*/
static int sql3_prepare_statements (sqlite3 *db, const char *sql, GPtrArray **statements)
{
    int          ret = ER_OK;     // return code
    const char   *head = NULL;    // start of current sql statement
    const char   *tail = NULL;    // start of next sql statement, if any
    sqlite3_stmt *stmt = NULL;
    GPtrArray    *stmt_list = NULL;

    LOGPRINTF("entry: sql [%s]", sql);

    g_return_val_if_fail( db                  , ER_INVALID_PARAMETER);
    g_return_val_if_fail( sql                 , ER_INVALID_PARAMETER);
    g_return_val_if_fail((*sql != '\0'       ), ER_INVALID_PARAMETER);
    g_return_val_if_fail( statements          , ER_INVALID_PARAMETER);
    g_return_val_if_fail((*statements == NULL), ER_INVALID_PARAMETER);

    // parse sql string and store compiled statements
    gboolean done = FALSE;
    for ( done = FALSE, head = sql ; !done ; head = tail )
    {
        int rc = sqlite3_prepare_v2(db, head, -1, &stmt, &tail);
        if (rc != ER_OK)
        {
            ERRORPRINTF("sqlite error [%d] [%s] on prepare_v2 [%s]", rc, sqlite3_errmsg(db), head);
            ret  = ERMETADB_FAIL;
            done = TRUE;
        }
        else
        {
            if (stmt)
            {
                // statement compiled: store it
                if (stmt_list == NULL)
                {
                    stmt_list = g_ptr_array_new();
                    if (stmt_list == NULL)
                    {
                        ERRORPRINTF("cannot create GPtrArray");
                        ret  = ER_OUT_OF_MEMORY;
                        done = TRUE;
                    }
                }
                if (stmt_list)
                {
                    g_ptr_array_add(stmt_list, stmt);
                }
                stmt = NULL;
            }

            if (*tail == '\0')
            {
                // no more sql to process
                done = TRUE;
            }
        }
    }

    // discard compiled statements in case of error
    if (ret != ER_OK)
    {
        if (stmt_list)
        {
            sql3_finalize_statements(stmt_list);
            stmt_list = NULL;
        }
    }

    // return list of prepared statements
    *statements = stmt_list;
    return ret;
}


static void bind_single_statement (void *p1, void *p2)
{
    sqlite3_stmt         *stmt       = (sqlite3_stmt*)    p1;
    const metadata_table *parameters = (const metadata_table*) p2;

    LOGPRINTF("entry: stmt [%p]", stmt);

    // get #parameters in statement
    int n_parms = sqlite3_bind_parameter_count(stmt);

    // bind parameters
    GString *parm_name = g_string_new("");
    int parm_idx;
    const metadata_cell *parm = NULL;
    for (parm_idx = 0 ; parm_idx < n_parms ; parm_idx++)
    {
        // get parameter name
        g_string_printf(parm_name, "?%d", parm_idx + 1);

        // find parameter value
        metadata_cell_type parm_type = METADATA_NULL;
        if (parameters)
        {
            parm = metadata_table_get_cell(parameters, parm_idx);
            if (parm)
            {
                parm_type = parm->type;
            }
        }

        int rc;
        // bind parameter
        switch (parm_type)
        {
            case METADATA_INT64:
                rc = sqlite3_bind_int64(stmt, parm_idx + 1, parm->value.v_int64);
                break;
            case METADATA_DOUBLE:
                rc = sqlite3_bind_double(stmt, parm_idx + 1, parm->value.v_double);
                break;
            case METADATA_TEXT:
                rc = sqlite3_bind_text(stmt, parm_idx + 1, parm->value.v_text->str, -1, SQLITE_STATIC);
                break;
            case METADATA_BLOB:
                rc = sqlite3_bind_blob(stmt, parm_idx + 1, parm->value.v_blob.data, parm->value.v_blob.len, SQLITE_STATIC);
                break;
            default:
                rc = sqlite3_bind_null(stmt, parm_idx + 1);
        }
        if (rc != ER_OK)
        {
            ERRORPRINTF("sqlite3_bind_xx returns [%d] on stmt [%p]", rc, stmt);
            sqlite3_bind_null(stmt, parm_idx + 1);
        }
    }

    g_string_free(parm_name, TRUE);
}


/**---------------------------------------------------------------------------
 *
 * Name :  sql3_bind_parameters
 *
 * @brief  Bind positional parameters to a set of compiled sql statements
 *         Assumes no parameters are currently bound to this statement
 *         Assumes TEXT and BLOB parameters unchanged until statements have executed
 *
 * @param  [in/out] statements - GPtrArray with compiled statements
 * @param  [in]     parameters - metadata_table with parameters (value), or NULL
 *                               table must have exactly one row
 *
 * @return ER_OK or error code
 *
 *--------------------------------------------------------------------------*/
int sql3_bind_parameters (GPtrArray *statements, const metadata_table *parameters)
{
    g_return_val_if_fail( statements          , ER_INVALID_PARAMETER);
    g_return_val_if_fail((statements->len > 0), ER_INVALID_PARAMETER);

    metadata_table *parms = (metadata_table*) parameters;  // const_cast
    g_ptr_array_foreach(statements, bind_single_statement, parms);

    return ER_OK;
}


static metadata_table* create_result_table (sqlite3_stmt *stmt)
{
    metadata_table *result = metadata_table_new();
    g_assert(result);

    // add columns to result table
    int n_col = sqlite3_column_count(stmt);
    int col;
    for (col = 0 ; col < n_col ; col++)
    {
        const char* name = sqlite3_column_name(stmt, col);
        if (name == NULL)
        {
            ERRORPRINTF("sqlite3_column returns NULL");
        }
        LOGPRINTF("column [%d] name [%s]", col, name);
        metadata_table_add_column(result, name);
    }
    return result;
}


// execute one statement, get all result rows
static int execute_single_statement (sqlite3_stmt *stmt, metadata_table **result_table)
{
    int ret = ER_OK;

    // repeat step while new rows are delivered
    metadata_table *result = NULL;
    int retries_left = STEP_MAX_RETRIES;
    gboolean done = FALSE;
    while (!done)
    {
        int step_rc = sqlite3_step(stmt);
        switch (step_rc)
        {
            case SQLITE_ROW:
            {
                if (result == NULL) result = create_result_table(stmt);

                // read column values into result table
                int n_row = metadata_table_n_rows(result);
                int n_col = metadata_table_n_columns(result);
                int idx = n_row * n_col;
                int col;
                for (col = 0 ; col < n_col && ret == ER_OK ; col++)
                {
                    sqlite3_value *value = sqlite3_column_value(stmt, col);
                    int rc = metadata_table_set_value(result, idx, value);
                    if (rc != ER_OK)
                    {
                        ret = rc;
                        done = TRUE;
                    }
                    idx++;
                }
                retries_left = STEP_MAX_RETRIES;
                break;
            }
            case SQLITE_DONE:
                done = TRUE;
                break;

            case SQLITE_BUSY:
                if (--retries_left >= 0)
                {
                    // wait a while, then try again
                    usleep(STEP_RETRY_MS * 1000);
                    break;
                }
                else
                {
                    // no more retries: error
                    ; // FALLTHROUGH to case default
                }

            default:
                ERRORPRINTF("sqlite3_step reports error [%d]", step_rc);
                ret  = ERMETADB_FAIL;
                done = TRUE;
        }
    }

    if (result)
    {
        // report the result
        if (*result_table)
        {
            ERRORPRINTF( "Not allowed to have SQL statements produce multiple result sets.\n"
                         "Second result set discarded:" );
            metadata_table_dump(result);
            metadata_table_free(result);
            ret = ERMETADB_FAIL;
        }
        else
        {
            *result_table = result;
        }
    }

    return ret;
}


/**---------------------------------------------------------------------------
 *
 * Name :  sql3_execute_statements
 *
 * @brief  Parse sql string and compile into byte code
 *
 * @param  [in]  db - sqlite database object
 * @param  [in]  statements - GPtrArray with compiled statements
 * @param  [in]  result_table - location to store metadata_table holding query results,
 *                              or NULL when caller is not interested in query results
 * @param  [out] result_table - metadata_table holding query results, or NULL when no results
 *                              First row holds column names and cell values,
 *                              other rows hold cell values only.
 *
 * @return ER_OK or error code
 *
 *--------------------------------------------------------------------------*/
static int sql3_execute_statements(sqlite3 *db, const GPtrArray *statements, metadata_table **result_table)
{
    LOGPRINTF("entry");

    g_return_val_if_fail( db                       , ER_INVALID_PARAMETER);
    g_return_val_if_fail( statements               , ER_INVALID_PARAMETER);
    g_return_val_if_fail((statements->len > 0     ), ER_INVALID_PARAMETER);
    g_return_val_if_fail( result_table             , ER_INVALID_PARAMETER);
    g_return_val_if_fail((   result_table == NULL
                          || *result_table == NULL), ER_INVALID_PARAMETER);

    // execute each statement
    int ret = ER_OK;
    unsigned int stmt_idx;
    metadata_table *result = NULL;
    for ( stmt_idx = 0 ; stmt_idx < statements->len  &&  ret == ER_OK ; stmt_idx++ )
    {
        sqlite3_stmt *stmt = g_ptr_array_index(statements, stmt_idx);
        // NOTE: memleak on result if several loops?
        ret = execute_single_statement(stmt, &result);
        if (ret != ER_OK)
        {
            ERRORPRINTF( "execute_single_statement returns [%d] on stmt_idx [%d], sqlite error [%s]",
                         ret, stmt_idx, sqlite3_errmsg(db) );
        }
    }

    // unbind and reset all statements
    for (stmt_idx = 0 ; stmt_idx < statements->len ; stmt_idx++)
    {
        sqlite3_stmt *stmt = g_ptr_array_index(statements, stmt_idx);

        // unbind all parameters on this statement
        int n_parms = sqlite3_bind_parameter_count(stmt);
        int parm_idx;
        for (parm_idx = 0 ; parm_idx < n_parms ; parm_idx++)
        {
            sqlite3_bind_null(stmt, parm_idx + 1);
        }

        // reset statement
        sqlite3_reset(stmt);
    }

    // discard results on error
    if (ret != ER_OK)
    {
        metadata_table_free(result);
    }

    if (result_table)
    {
        *result_table = result;
        result = NULL;
    }
    metadata_table_free(result);
    return ret;
}


int sql3_execute_query(sqlite3 *db, const char *sql, const metadata_table *parameters, metadata_table **result_table)
{
    QUERYPRINTF("entry: sql [%s]", sql);
    metadata_table_dump(parameters);

    g_return_val_if_fail( db                       , ER_INVALID_PARAMETER);
    g_return_val_if_fail( sql                      , ER_INVALID_PARAMETER);
    g_return_val_if_fail((   result_table == NULL
                          || *result_table == NULL), ER_INVALID_PARAMETER);

    int ret = ER_OK;
    GPtrArray *statements = NULL;
    int rc = sql3_prepare_statements(db, sql, &statements);
    if (rc != ER_OK)
    {
        ERRORPRINTF("sql3_prepare_statements returns [%d] sql [%s]", rc, sql);
        ret = rc;
    }

    if (ret == ER_OK  &&  parameters)
    {
        rc = sql3_bind_parameters(statements, parameters);
        if (rc != ER_OK)
        {
            ERRORPRINTF("sql3_bind_parameters returns [%d] sql [%s]", rc, sql);
            ret = rc;
        }
    }

    metadata_table *result = NULL;
    if (ret == ER_OK)
    {
        rc = sql3_execute_statements(db, statements, &result);
        if (rc != ER_OK)
        {
            ERRORPRINTF("sql3_execute_statements returns [%d] sql [%s]", rc, sql);
            ret = rc;
        }
    }

    // clean up
    if (statements)
    {
        sql3_finalize_statements(statements);
        statements = NULL;
    }

    // return result
    if (result_table)
    {
        *result_table = result;
        result = NULL;
    }
    metadata_table_free(result);
    return ret;
}


static int begin_transaction(erMetadb thiz, const gboolean is_readonly)
{
    int         ret = ER_OK;
    int         rc;
    const char  *sql    = NULL;
    char        *errmsg = NULL;
    int         retries_left = BEGIN_MAX_RETRIES;
    gboolean    done = FALSE;

    g_assert(thiz && thiz->database);
    g_assert(thiz->transaction_nesting < 100);

    sqlite3 *db = thiz->database;

    LOGPRINTF( "entry: db [%p] is_readonly [%d] [%d] transaction_nesting [%d]",
                db,
                is_readonly,
                thiz->is_transaction_readonly,
                thiz->transaction_nesting      );

    g_return_val_if_fail(db, ER_INVALID_PARAMETER);


    // start a transaction, acquire write lock
    if (thiz->transaction_nesting == 0)
    {
        thiz->is_transaction_readonly = is_readonly;
        thiz->do_force_rollback       = FALSE;

        // repeat while database locked
        if ( is_readonly )
        {
            sql = "BEGIN DEFERRED;";
        }
        else
        {
            sql = "BEGIN IMMEDIATE;";
        }
        QUERYPRINTF("sql [%s]", sql);
        while (!done)
        {
            rc = sqlite3_exec(db, sql, NULL, NULL, &errmsg);
            if (rc == SQLITE_OK)
            {
                // database lock obtained: increase transaction nesting level
                thiz->transaction_nesting++;
                done = TRUE;

            }
            else if (   rc == SQLITE_BUSY
                     && --retries_left >= 0 )
            {
                // database locked: wait a while then try again
                WARNPRINTF("pid [%ld] waiting for database lock", (long)getpid());
                usleep(BEGIN_RETRY_MS * 1000);
            }
            else
            {
                ERRORPRINTF("pid [%ld] cannot begin transaction, error [%d] [%s]", (long)getpid(), rc, errmsg);
                switch (rc)
                {
                    case SQLITE_BUSY:
                        ret = ERMETADB_DATABASE_BUSY;
                        break;
                    case SQLITE_READONLY:
                        ret = ER_READONLY;
                        break;
                    default:
                        ret = ERMETADB_FAIL;
                }
                done = TRUE;
            }

            // free error message, if any
            sqlite3_free(errmsg);
            errmsg = NULL;
        }
    }
    else
    {
        // transaction in progress: check read-only status
        if ( is_readonly == thiz->is_transaction_readonly )
        {
            // same is_readonly: ok
            thiz->transaction_nesting++;
        }
        else if ( is_readonly )
        {
            // read-only begin on writable transaction: ok
            thiz->transaction_nesting++;
        }
        else
        {
            // error: read-only transaction cannot be promoted to writable
            ret = ER_READONLY;
        }
    }

    if (ret != ER_OK) WARNPRINTF("cannot begin transaction. ret [%d]", ret);
    return ret;
}


int sql3_begin_transaction(erMetadb thiz)
{
    return begin_transaction(thiz, FALSE);
}


int sql3_begin_transaction_readonly(erMetadb thiz)
{
    return begin_transaction(thiz, TRUE);
}


int sql3_commit_or_rollback(erMetadb thiz, int err)
{
    g_assert(thiz && thiz->database);

    LOGPRINTF( "entry: err [%d] is_readonly [%d] transaction_nesting [%d]",
                err,
                thiz->is_transaction_readonly,
                thiz->transaction_nesting      );

    // decide whether to commit or rollback
    if (err != ER_OK) thiz->do_force_rollback = TRUE;

    // never turn an error code into a good one
    int ret = err;
    if (thiz->transaction_nesting > 0)
    {
        // decrease transaction nesting level
        thiz->transaction_nesting--;

        if (thiz->transaction_nesting == 0)
        {
            const char *sql = NULL;
            // decide commit or rollback ...
            if ( err == ER_OK && thiz->do_force_rollback == FALSE )
            {
                // ok: commit changes to database
                sql = "COMMIT;";
            }
            else
            {
                // error: rollback and return original error code
                sql = "ROLLBACK;";
            }
            QUERYPRINTF("sql [%s]", sql);

            // ... and execute it
            gboolean done = FALSE;
            int retries_left = COMMIT_MAX_RETRIES;
            while (!done)
            {
                char *errmsg = NULL;
                int rc = sqlite3_exec(thiz->database, sql, NULL, NULL, &errmsg);
                if (rc == SQLITE_OK)
                {
                    // transaction ended
                    done = TRUE;
                }
                else if ( rc == SQLITE_BUSY && --retries_left >= 0 )
                {
                    // database locked: wait a while then try again
                    WARNPRINTF("pid [%ld] waiting for database lock, sql [%s]", (long)getpid(), sql);
                    usleep(BEGIN_RETRY_MS * 1000);
                }
                else
                {
                    ERRORPRINTF("cannot end transaction, sql [%s] error [%d] [%s]", sql, rc, errmsg);
                    if (rc == SQLITE_BUSY)
                    {
                        ret = ERMETADB_DATABASE_BUSY;
                    }
                    else
                    {
                        ret = ERMETADB_FAIL;
                    }
                    done = TRUE;
                }

                // free error message, if any
                sqlite3_free(errmsg);
                errmsg = NULL;
            }
        }
    }

    LOGPRINTF("leave: ret [%d]", ret);
    return ret;
}


// custom collation sequence for case-insensitive text ordering
int sql3_strcase_compare( void        *user_data,
                          int         left_len,
                          const void  *left_void,
                          int         right_len,
                          const void  *right_void )
{
    UNUSED(user_data);
    int         ret = 0;
    const gchar *left  = (gchar*)left_void;
    const gchar *right = (gchar*)right_void;

    int len = MIN(left_len, right_len);
    if (len == 0)
    {
        // empty string sorts at end of list
        if (left_len == right_len)
        {
            ret = 0;
        }
        else if (left_len == 0)
        {
            ret = 1;
        }
        else
        {
            ret = -1;
        }
    }
    else
    {
        // non-empty string sorts case-insensitive
        ret = g_ascii_strncasecmp(left, right, len);
        if (ret == 0)
        {
            if (left_len < right_len)
            {
                ret = -1;
            }
            else
            {
                ret = 1;
            }
        }
    }

    LOGPRINTF("return [%d] left [%.*s] right [%.*s]", ret, left_len, left, right_len, right);
    return ret;
}