ermetadb_local.c

Go to the documentation of this file.

/*
 * File Name  : ermetadb_local.c
 * 
 * Description: Functions to access local metadb's
 */

/*
 * 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 <string.h>

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

// local include files, between " "
#include "ermetadb_log.h"
#include "ermetadb.h"
#include "ermetadb_error.h"
#include "ermetadb_private.h"
#include "sqlite3_wrapper.h"

#define LOCAL_MAX_COLUMNS  12
static const struct
             {
                 const char     *table_name;
                 const char     *column_names[LOCAL_MAX_COLUMNS+1];
             } local_database_tables[]
             =
             {
                {
                    "application_data",
                    {
                        "file_id",
                        "key",
                        "value",
                         NULL
                    }
                },
                { 
                    "annotations",
                    {
                        "annotation_id",
                        "file_id",
                        "annotation_type",
                        "layer",
                        "file_position",
                        "title",
                        "start_anchor",
                        "end_anchor",
                        "data",
                        NULL
                    }
                },
                { 
                    "file_metadata",
                    {
                        "file_id",
                        "filename",
                        NULL
                    }
                },
                {
                    NULL,
                    {}
                }
             };

#define CHECK_LOCAL \
    g_assert(thiz); \
    g_assert(thiz->database); \
    g_assert(!thiz->is_global);


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

static const char* const * get_local_column_names(const char* table)
{
    TRACE("entry: table [%s]", table);
    g_assert(table);

    int i;
    for (i = 0 ; local_database_tables[i].table_name; i++)
    {
        if ( strcmp(local_database_tables[i].table_name, table) == 0 )
        {
            return local_database_tables[i].column_names;
        }
    }
    return NULL;
}


static gint64 get_local_id(erMetadb thiz, const char* filename)
{
    // build query
    char *query = sqlite3_mprintf("SELECT file_id FROM file_metadata WHERE filename = %Q LIMIT 1;",
                filename);

    metadata_table *result = NULL;
    int rc = sql3_execute_query(thiz->database, query, NULL, &result);
    gint64 file_id = -1;
    if (rc == ER_OK && result) {
        const metadata_cell *cell = metadata_table_get_cell(result, 0);
        g_assert(cell->type == METADATA_INT64);
        file_id = cell->value.v_int64;
    }

    sqlite3_free(query);
    metadata_table_free(result);
    return file_id;
}


int ermetadb_local_remove_annotation (erMetadb thiz, const gint64 annotation_id)
{
    LOGPRINTF("annotation_id [%lld]", annotation_id);

    CHECK_LOCAL

    char query[255];
    sprintf(query, "DELETE FROM annotations WHERE annotation_id = %lld", annotation_id );
    int rc = sql3_execute_query(thiz->database, query, NULL, NULL);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot delete annotation, sql [%s]", query);
        rc = ERMETADB_FAIL;
    }
    return rc;
}


// set annotation details, without begin/commit transaction
static int set_annotation_impl (erMetadb        thiz,
                                const gint64    annotation_id,
                                const gchar     *annotation_type,
                                const gchar     *layer,
                                const gint64    file_position,
                                const gchar     *title,
                                const gchar     *start_anchor,
                                const gchar     *end_anchor,
                                const gchar     *data_blob,
                                const guint     data_len )
{
    TRACE("annotation_id [%lld]", annotation_id);

    g_return_val_if_fail( annotation_type, ER_INVALID_PARAMETER );
    g_return_val_if_fail( layer, ER_INVALID_PARAMETER );
    g_return_val_if_fail( file_position >= 0, ER_INVALID_PARAMETER );
    g_return_val_if_fail( (start_anchor == NULL || start_anchor[0] != 0), ER_INVALID_PARAMETER );
    g_return_val_if_fail( (end_anchor == NULL || end_anchor[0] != 0), ER_INVALID_PARAMETER );
    g_return_val_if_fail( (data_blob == NULL || data_len > 0), ER_INVALID_PARAMETER );

    // build sql query
    GString *sql = g_string_new("");
    char *sql3_annotype = sqlite3_mprintf("%Q", annotation_type);
    char *sql3_layer    = sqlite3_mprintf("%Q", layer);
    g_string_printf(        sql, "UPDATE annotations"
                                 " SET annotation_type = %s,"
                                     " layer = %s,"
                                     " file_position = %lld",
                            sql3_annotype,
                            sql3_layer,
                            file_position );
    sqlite3_free(sql3_layer   );
    sqlite3_free(sql3_annotype);

    g_string_append(        sql, ", title = " );
    if (title)
    {
        char *sql3_string = sqlite3_mprintf("%Q", title);
        g_string_append( sql, sql3_string );
        sqlite3_free(sql3_string);
    }
    else
    {
        // clear old title
        g_string_append(    sql, "NULL" );
    }

    g_string_append(        sql, ", start_anchor = " );
    if (start_anchor)
    {
        char *sql3_string = sqlite3_mprintf("%Q", start_anchor);
        g_string_append( sql, sql3_string );
        sqlite3_free(sql3_string);
    }
    else
    {
        // clear old anchor
        g_string_append(    sql, "NULL" );
    }

    g_string_append(        sql, ", end_anchor = " );
    if (end_anchor)
    {
        char *sql3_string = sqlite3_mprintf("%Q", end_anchor);
        g_string_append( sql, sql3_string);
        sqlite3_free(sql3_string);
    }
    else
    {
        // clear old anchor
        g_string_append(    sql, "NULL" );
    }

    g_string_append(        sql, ", data = " );
    metadata_table *parameters = NULL;
    if (data_blob)
    {
        parameters = metadata_table_new();
        metadata_table_add_column( parameters, "data" );
        metadata_table_set_blob_static( parameters, 0, data_blob, data_len );
        g_string_append(        sql, "?1" );
    }
    else
    {
        // clear old data
        g_string_append(        sql, "NULL" );
    }

    g_string_append_printf( sql, " WHERE annotation_id = %lld;",
                            annotation_id );

    // execute query
    int rc = sql3_execute_query(thiz->database, sql->str, parameters, NULL);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot set annotation details, sql [%s]", sql->str);
        rc = ERMETADB_FAIL;
    }

    metadata_table_free(parameters);
    return rc;
}


static gint64 find_or_create_filename(erMetadb thiz, const char* filename)
{
    gint64 file_id = get_local_id(thiz, filename);
    if (file_id < 0) {
        char *sql = sqlite3_mprintf("INSERT INTO file_metadata (filename) VALUES (%Q);", filename);
        int rc = sql3_execute_query(thiz->database, sql, NULL, NULL);
        sqlite3_free(sql);
        if (rc != ER_OK)
        {
            ERRORPRINTF("cannot add filename [%s]", filename);
            return -1;
        }
        file_id = sqlite3_last_insert_rowid(thiz->database);
    }
    return file_id;
}


int ermetadb_local_create_annotation (erMetadb     thiz,
                                      const gchar  *filename,
                                      gint64       *annotation_id,
                                      const gchar  *annotation_type,
                                      const gchar  *layer,
                                      const gint64 file_position,
                                      const gchar  *title,
                                      const gchar  *start_anchor,
                                      const gchar  *end_anchor,
                                      const gchar  *data_blob,
                                      const guint  data_len )
{
    LOGPRINTF("file [%s]", filename);

    CHECK_LOCAL
    g_assert(filename && filename[0]);
    g_assert(annotation_id);

    gint64 anno_id = -1;

    int ret = sql3_begin_transaction(thiz);
    if (ret != ER_OK) return ret;

    gint64 file_id = find_or_create_filename(thiz, filename);
    if (file_id < 0) {
        ret = ER_FAIL;
        goto out;
    }

    // create an empty annotation
    // build query
    GString *sql = g_string_new("");  // SQL statements
    g_string_printf( sql, "INSERT INTO annotations"
                          " (file_id, annotation_type, layer, file_position)"
                          " VALUES (%lld, 'dummy', 'dummy', 0 )",
                     file_id );

    // execute query
    int rc = sql3_execute_query(thiz->database, sql->str, NULL, NULL);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot create annotation, sql [%s]", sql->str);
        ret = ERMETADB_FAIL;
    }
    else
    {
        // retrieve the new annotation_id
        anno_id = sqlite3_last_insert_rowid(thiz->database);
    }
    g_string_free(sql, TRUE);

    // set annotation fields
    if (ret == ER_OK)
    {
        ret = set_annotation_impl( thiz,
                                   anno_id,
                                   annotation_type,
                                   layer,
                                   file_position,
                                   title,
                                   start_anchor,
                                   end_anchor,
                                   data_blob,
                                   data_len        );
    }
out:
    ret = sql3_commit_or_rollback(thiz, ret);

    *annotation_id = anno_id;
    return ret;
}


int ermetadb_local_set_annotation (erMetadb      thiz,
                                   const gint64  annotation_id,
                                   const gchar   *annotation_type,
                                   const gchar   *layer,
                                   const gint64  file_position,
                                   const gchar   *title,
                                   const gchar   *start_anchor,
                                   const gchar   *end_anchor,
                                   const gchar   *data_blob,
                                   const guint   data_len )
{
    LOGPRINTF("annotation_id [%lld]", annotation_id);

    CHECK_LOCAL
    g_assert(annotation_id);

    int ret = sql3_begin_transaction(thiz);
    if (ret != ER_OK) return ret;

    ret = set_annotation_impl( thiz,
                               annotation_id,
                               annotation_type,
                               layer,
                               file_position,
                               title,
                               start_anchor,
                               end_anchor,
                               data_blob,
                               data_len        );

    ret = sql3_commit_or_rollback(thiz, ret);
    return ret;
}


// get annotation details for a list of annotations
int ermetadb_local_get_annotations (erMetadb       thiz,
                                    const gint64   *annotation_id_list,
                                    long           annotation_id_num,
                                    metadata_table **values_tbl        )
{
    LOGPRINTF("entry");

    CHECK_LOCAL
    g_return_val_if_fail(  annotation_id_list                , ER_INVALID_PARAMETER );
    g_return_val_if_fail(  annotation_id_num >= 0            , ER_INVALID_PARAMETER );
    g_return_val_if_fail( (values_tbl && *values_tbl == NULL), ER_INVALID_PARAMETER );

    // get known column names
    GString *columns = g_string_new("");
    const char* const *name;
    for ( name = get_local_column_names( "annotations" ) ; *name ; name++ )
    {
        if ( strcmp(*name, "file_id" ) != 0 )
        {
            if (columns->len != 0) g_string_append( columns, ", "  );
            g_string_append( columns, *name );
        }
    }

    metadata_table *result = NULL;

    int ret = sql3_begin_transaction_readonly(thiz);
    if (ret != ER_OK) goto out;

    // build sql
    const gint64 *p_anno = annotation_id_list;
    GString *sql = g_string_new("");  // SQL query
    g_string_printf( sql, "SELECT %s"
                          " FROM annotations"
                          " WHERE annotation_id IN (%lld",
                     columns->str,
                     *p_anno      );
    p_anno++;
    int i;
    for ( i = 1 ; i < annotation_id_num ; i++ )
    {
        g_string_append_printf( sql, ", %lld", *p_anno );
        p_anno++;
    }
    g_string_append( sql, ");" );

    // execute query
    int rc = sql3_execute_query(thiz->database, sql->str, NULL, &result);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot get metadata, sql [%s]", sql->str);
        ret = ER_NOT_FOUND;
    }
    g_string_free(sql, TRUE);

    // NOTE can turn ok into fail!
    ret = sql3_commit_or_rollback(thiz, ret);
out:
    g_string_free(columns, TRUE);

    *values_tbl = result;
    return ret;
}


// get a range of annotations's for a file/folder
int ermetadb_local_select_annotations (erMetadb        thiz,
                                       const gchar     *filename,
                                       const gchar     *annotation_type,
                                       const gchar     *layer,
                                       const gint64    file_position_min,
                                       const gint64    file_position_max,
                                       metadata_table  **values_tbl )
{
    LOGPRINTF("file [%s]", filename);

    CHECK_LOCAL
    g_assert(filename && filename[0]);
    g_return_val_if_fail( (annotation_type == NULL || annotation_type[0] != 0), ER_INVALID_PARAMETER );
    g_return_val_if_fail( values_tbl && *values_tbl == NULL, ER_INVALID_PARAMETER );

    // start transaction
    int ret = sql3_begin_transaction_readonly(thiz);
    if (ret != ER_OK) return ret;

    metadata_table *result = NULL;
    
    gint64 file_id = get_local_id(thiz, filename);
    if (file_id < 0) goto out;

    // build sql
    GString *sql = g_string_new("");
    g_string_printf( sql, "SELECT annotation_id"
                               ", annotation_type"
                               ", layer"
                               ", file_position"
                               ", start_anchor"
                               ", end_anchor"
                          " FROM annotations"
                          " WHERE file_id = %lld",
                     file_id );
    if (annotation_type)
    {
        char *str = sqlite3_mprintf("%Q", annotation_type);
        g_string_append_printf(sql, " AND annotation_type = %s", str);
        sqlite3_free(str);
    }
    if (layer)
    {
        char *str = sqlite3_mprintf("%Q", layer);
        g_string_append_printf(sql, " AND layer = %s", str);
        sqlite3_free(str);
    }
    if (file_position_min >= 0)
    {
        g_string_append_printf(sql, " AND file_position >= %lld", file_position_min);
    }
    if (file_position_max >= 0)
    {
        g_string_append_printf(sql, " AND file_position <= %lld", file_position_max);
    }
    g_string_append(sql, "ORDER BY file_position, annotation_type, layer;");

    // execute query
    int rc = sql3_execute_query(thiz->database, sql->str, NULL, &result);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot get annotations, sql [%s]", sql->str);
        ret = ER_NOT_FOUND;
    }
    g_string_free(sql, TRUE);

out:
    ret = sql3_commit_or_rollback(thiz, ret);

    *values_tbl = result;
    return ret;
}


static int local_delete_all_annotations (erMetadb thiz, gint64 file_id)
{
    char query[255];
    sprintf(query, "DELETE FROM annotations WHERE file_id=%lld;", file_id );
    int rc = sql3_execute_query(thiz->database, query, NULL, NULL);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot delete annotations, sql [%s]", query);
        rc = ERMETADB_FAIL;
    }
    return rc;
}


static GString *get_keys(const metadata_table *names_tbl)
{
    GString *sql_keys = g_string_new("");
    int n_cols = metadata_table_n_columns(names_tbl);
    int col;
    for (col = 0 ; col < n_cols ; col++)
    {
        const metadata_cell *cell = metadata_table_get_cell(names_tbl, col);
        if (cell && cell->name && cell->name->str)
        {
            char *sql3_cellname = sqlite3_mprintf("%Q", cell->name->str);
            if (sql_keys->len > 0)
            {
                g_string_append(sql_keys, ", ");
            }
            g_string_append(sql_keys, sql3_cellname);

            sqlite3_free( sql3_cellname );
        }
    }
    return sql_keys;
}


int ermetadb_local_remove_application_data (erMetadb             thiz,
                                            const gchar          *filename,
                                            const metadata_table *names_tbl )
{
    LOGPRINTF("file [%s]", filename);

    CHECK_LOCAL
    g_assert(filename && filename[0]);
    g_return_val_if_fail( names_tbl, ER_INVALID_PARAMETER );
    g_return_val_if_fail( (metadata_table_n_rows(names_tbl) == 0), ER_INVALID_PARAMETER );

    gint64 file_id = get_local_id(thiz, filename);
    if (file_id < 0) return ER_NOT_FOUND;

    // start transaction
    int ret = sql3_begin_transaction(thiz);
    if (ret != ER_OK) return ret;

    // build sql query
    GString *keys = get_keys(names_tbl);
    GString *sql = g_string_new("");
    g_string_printf( sql, "DELETE FROM application_data"
                          " WHERE file_id=%lld"
                          "   AND key IN (%s);",
                     file_id, keys->str );
    g_string_free(keys, TRUE);

    int rc = sql3_execute_query(thiz->database, sql->str, NULL, NULL);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot delete application_data, sql [%s]", sql->str);
        ret = ERMETADB_FAIL;
    }
    g_string_free(sql, TRUE);

    ret = sql3_commit_or_rollback(thiz, ret);
    return ret;
}


int ermetadb_local_set_application_data (erMetadb             thiz,
                                         const gchar          *filename,
                                         const metadata_table *values_tbl )
{
    LOGPRINTF("file [%s]", filename);

    CHECK_LOCAL
    g_assert(filename && filename[0]);
    g_return_val_if_fail( values_tbl, ER_INVALID_PARAMETER );
    g_return_val_if_fail( (metadata_table_n_rows(values_tbl) == 1), ER_INVALID_PARAMETER );

    gint64 file_id = find_or_create_filename(thiz, filename);
    if (file_id < 0) return ER_NOT_FOUND;

    int ret = sql3_begin_transaction(thiz);
    if (ret != ER_OK) return ret;

    // build delete and insert statements for sql query
    GString *sql_del = NULL;     // SQL delete statement
    GString *sql_ins   = NULL;     // SQL insert statements
    int n_cols = metadata_table_n_columns(values_tbl);
    int col;
    for (col = 0 ; col < n_cols ; col++)
    {
        const metadata_cell*cell = metadata_table_get_cell(values_tbl, col);
        if (cell && cell->name && cell->name->str)
        {
            char* sql3_cellname = sqlite3_mprintf("%Q", cell->name->str);

            if (cell->type == METADATA_NULL)
            {
                // add this key to sql delete statement
                if (sql_del == NULL)
                {
                    sql_del = g_string_new("");
                    g_string_printf( sql_del, "DELETE FROM application_data"
                                              " WHERE file_id=%lld"
                                              "   AND key IN ( %s",
                                     file_id,
                                     sql3_cellname );
                }
                else
                {
                    g_string_append_printf( sql_del, ", %s", sql3_cellname );
                }
            }
            else
            {
                // add this key to sql_insert statement
                if (sql_ins == NULL)
                {
                    sql_ins = g_string_new("");
                }
                g_string_append_printf( sql_ins, "INSERT OR REPLACE INTO application_data"
                                                 " (file_id, key, value)"
                                                 " VALUES (%lld, %s, ?%d);",
                                        file_id,
                                        sql3_cellname,
                                        col + 1   );
            }
            sqlite3_free( sql3_cellname );
        }
    }
    if (sql_del) g_string_append( sql_del, " );" );

    // execute delete query
    if (sql_del)
    {
        int rc = sql3_execute_query(thiz->database, sql_del->str, NULL, NULL);
        if (rc != ER_OK)
        {
            ERRORPRINTF("cannot delete application_data, sql [%s]", sql_del->str);
            ret = ERMETADB_FAIL;
        }
    }

    // execute insert query
    if (ret == ER_OK && sql_ins)
    {
        int rc = sql3_execute_query(thiz->database, sql_ins->str, values_tbl, NULL);
        if (rc != ER_OK)
        {
            ERRORPRINTF("cannot insert application_data, sql [%s]", sql_ins->str);
            ret = ERMETADB_FAIL;
        }
    }

    ret = sql3_commit_or_rollback(thiz, ret);

    if (sql_ins) { g_string_free(sql_ins, TRUE); }
    if (sql_del) { g_string_free(sql_del, TRUE); }
    return ret;
}


// rotate table with key-value pairs:
//   in:  table = multiple rows, each row represents a key-value pair
//   out: rotated = single row, each column represents a key-value pair
static metadata_table* rotate_keyval_table(const metadata_table *table)
{
    // find 'key' and 'value' columns
    int n_keys  = metadata_table_n_rows(table);
    int col_key = metadata_table_find_column(table, "key"  );
    int col_val = metadata_table_find_column(table, "value");
    g_assert(col_key >= 0);
    g_assert(col_val >= 0);

    // create new table
    metadata_table *rotated = metadata_table_new();

    // convert table.rows into rotated.columns
    int key_index;
    gboolean ok = TRUE;
    for (key_index = 0 ; key_index < n_keys ; key_index++)
    {
        // set column name = key
        GString *key = metadata_table_get_string( table, 
                                         metadata_table_cell_index(table, key_index, col_key) );
        if (key == NULL || key->len == 0)
        {
            ERRORPRINTF("cannot read key from row [%d]", key_index);
            ok = FALSE;
        }
        else
        {
            int rc = metadata_table_add_column(rotated, key->str);
            if (rc != ER_OK)
            {
                ERRORPRINTF("cannot add column [%s] in rotated, error [%d]", key->str, rc);
                ok = FALSE;
            }
        }
        if (key)
        {
            g_string_free(key, TRUE);
        }

        // set column value = value
        if (ok)
        {
            const metadata_cell *cell = metadata_table_get_cell( table,
                                            metadata_table_cell_index(table, key_index, col_val) );
            int rc = metadata_table_set_cell( rotated, key_index, cell );
            if (rc != ER_OK)
            {
                ERRORPRINTF("cannot set value type [%d] in cell, error [%d]", cell->type, rc);
                ok = FALSE;
            }
        }
    }

    if (!ok)
    {
        metadata_table_free(rotated);
    }
    return rotated;
}


int ermetadb_local_get_application_data (erMetadb             thiz,
                                         const gchar          *filename,
                                         const metadata_table *names_tbl,
                                         metadata_table       **values_tbl )
{
    LOGPRINTF("file [%s]", filename);

    CHECK_LOCAL
    g_assert(filename && filename[0]);
    g_return_val_if_fail( names_tbl, ER_INVALID_PARAMETER );
    g_return_val_if_fail( (values_tbl && *values_tbl == NULL ), ER_INVALID_PARAMETER );

    gint64 file_id = get_local_id(thiz, filename);
    if (file_id < 0) return ER_NOT_FOUND;

    int ret = sql3_begin_transaction_readonly(thiz);
    if (ret != ER_OK) return ret;

    // build sql query
    GString *keys = get_keys(names_tbl);
    GString *sql = g_string_new("");
    g_string_printf( sql, "SELECT key, value"
                          " FROM application_data"
                          " WHERE file_id=%lld"
                          "   AND key IN (%s);",
                     file_id,
                     keys->str );
    g_string_free(keys, TRUE);

    // execute query
    metadata_table *result = NULL;
    int rc = sql3_execute_query(thiz->database, sql->str, NULL, &result);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot get application_data, sql [%s]", sql->str);
        ret = ERMETADB_FAIL;
    }
    g_string_free(sql, TRUE);

    // NOTE: ret can become OK here after rollback!
    ret = sql3_commit_or_rollback(thiz, ret);

    // rotate result: one row per key-val --> one column per key-val
    metadata_table *rotated = NULL;
    if (rc == ER_OK && result)
    {
        rotated = rotate_keyval_table(result);
    }
    *values_tbl = rotated;

    metadata_table_free(result);
    return ret;
}


static int local_delete_all_application_data (erMetadb thiz, gint64 file_id)
{
    char query[255];
    sprintf(query, "DELETE FROM application_data WHERE file_id=%lld;", file_id );

    int rc = sql3_execute_query(thiz->database, query, NULL, NULL);
    if (rc != ER_OK)
    {
        ERRORPRINTF("cannot delete application_data, sql [%s]", query);
        rc = ERMETADB_FAIL;
    }
    return rc;
}


static int local_delete_entry(erMetadb thiz, gint64 file_id)
{
    char query[255];
    snprintf(query, sizeof(query), "DELETE FROM file_metadata WHERE file_id = %lld;", file_id);
    return sql3_execute_query(thiz->database, query, NULL, NULL);
}


int local_delete_all_data_for_file(erMetadb thiz, const char* filename)
{
    g_assert(!thiz->is_global);

    gint64 file_id = get_local_id(thiz, filename);
    if (file_id < 0) return ER_NOT_FOUND;

    return local_delete_all_data_for_id(thiz, file_id);
}


int local_delete_all_data_for_id(erMetadb thiz, gint64 file_id)
{
    g_assert(!thiz->is_global);

    int ret = sql3_begin_transaction(thiz);
    if (ret != ER_OK) return ret;

    // delete annotations, application data and file_metadata entry
    local_delete_all_annotations(thiz, file_id);
    local_delete_all_application_data(thiz, file_id);
    local_delete_entry(thiz, file_id);

    ret = sql3_commit_or_rollback(thiz, ER_OK);
    return ret;

}