Adding Libgfdb to GlusterFS

*************************************************************************
			Libgfdb						|
*************************************************************************
Libgfdb provides abstract mechanism to record extra/rich metadata
required for data maintenance, such as data tiering/classification.
It provides consumer with API for recording and querying, keeping
the consumer abstracted from the data store used beneath for storing data.
It works in a plug-and-play model, where data stores can be plugged-in.
Presently we have plugin for Sqlite3. In the future will provide recording
and querying performance optimizer. In the current implementation the schema
of metadata is fixed.

Schema:
~~~~~~
      GF_FILE_TB Table:
      ~~~~~~~~~~~~~~~~~
      This table has one entry per file inode. It holds the metadata required to
      make decisions in data maintenance.
      GF_ID (Primary key)	: File GFID (Universal Unique IDentifier in the namespace)
      W_SEC, W_MSEC 		: Write wind time in sec & micro-sec
      UW_SEC, UW_MSEC		: Write un-wind time in sec & micro-sec
      W_READ_SEC, W_READ_MSEC 	: Read wind time in sec & micro-sec
      UW_READ_SEC, UW_READ_MSEC : Read un-wind time in sec & micro-sec
      WRITE_FREQ_CNTR INTEGER	: Write Frequency Counter
      READ_FREQ_CNTR INTEGER	: Read Frequency Counter

      GF_FLINK_TABLE:
      ~~~~~~~~~~~~~~
      This table has all the hardlinks to a file inode.
      GF_ID		: File GFID               (Composite Primary Key)``|
      GF_PID		: Parent Directory GFID  (Composite Primary Key)   |-> Primary Key
      FNAME 		: File Base Name          (Composite Primary Key)__|
      FPATH 		: File Full Path (Its redundant for now, this will go)
      W_DEL_FLAG 	: This Flag is used for crash consistancy, when a link is unlinked.
                  	  i.e Set to 1 during unlink wind and during unwind this record
                          is deleted
      LINK_UPDATE 	: This Flag is used when a link is changed i.e rename.
                          Set to 1 when rename wind and set to 0 in rename unwind

Libgfdb API:
~~~~~~~~~~~
Refer libglusterfs/src/gfdb/gfdb_data_store.h

Change-Id: I2e9fbab3878ce630a7f41221ef61017dc43db11f
BUG: 1194753
Signed-off-by: Joseph Fernandes <josferna@redhat.com>
Signed-off-by: Dan Lambright <dlambrig@redhat.com>
Signed-off-by: Joseph Fernandes <josferna@redhat.com>
Reviewed-on: http://review.gluster.org/9683
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

11 files changed, 4259 insertions, 1 deletions

diff --git a/libglusterfs/Makefile.am b/libglusterfs/Makefile.am
index d471a3f9243..8e5a4a0ccbf 100644
--- a/libglusterfs/Makefile.am
+++ b/libglusterfs/Makefile.am
@@ -1,3 +1,3 @@
-SUBDIRS = src
+SUBDIRS = src src/gfdb
 
 CLEANFILES = 
diff --git a/libglusterfs/src/gfdb/Makefile.am b/libglusterfs/src/gfdb/Makefile.am
new file mode 100644
index 00000000000..8c48463aed3
--- /dev/null
+++ b/libglusterfs/src/gfdb/Makefile.am
@@ -0,0 +1,42 @@
+libgfdb_la_CFLAGS = -Wall $(GF_CFLAGS) $(GF_DARWIN_LIBGLUSTERFS_CFLAGS) \
+  -DDATADIR=\"$(localstatedir)\"
+
+libgfdb_la_CPPFLAGS = $(GF_CPPFLAGS) -D__USE_FILE_OFFSET64 -fpic \
+  -I$(top_srcdir)/libglusterfs/src \
+  -DDATADIR=\"$(localstatedir)\"
+
+libgfdb_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la -lsqlite3
+
+libgfdb_la_LDFLAGS = $(GF_LDFLAGS) -version-info $(LIBGLUSTERFS_LT_VERSION)
+
+libgfdbdir = $(includedir)/glusterfs/gfdb
+
+if BUILD_GFDB
+ lib_LTLIBRARIES = libgfdb.la
+endif
+
+CONTRIB_BUILDDIR = $(top_builddir)/contrib
+
+libgfdb_la_SOURCES = gfdb_data_store.c gfdb_sqlite3_helper.c\
+   gfdb_sqlite3.c \
+   $(CONTRIBDIR)/uuid/clear.c \
+   $(CONTRIBDIR)/uuid/copy.c $(CONTRIBDIR)/uuid/gen_uuid.c \
+   $(CONTRIBDIR)/uuid/pack.c $(CONTRIBDIR)/uuid/parse.c \
+   $(CONTRIBDIR)/uuid/unparse.c $(CONTRIBDIR)/uuid/uuid_time.c \
+   $(CONTRIBDIR)/uuid/compare.c $(CONTRIBDIR)/uuid/isnull.c \
+   $(CONTRIBDIR)/uuid/unpack.c
+
+noinst_HEADERS =  gfdb_data_store.h gfdb_data_store_types.h gfdb_sqlite3_helper.h\
+   gfdb_sqlite3.h gfdb_mem-types.h \
+   $(CONTRIBDIR)/uuid/uuidd.h \
+   $(CONTRIBDIR)/uuid/uuid.h $(CONTRIBDIR)/uuid/uuid.h \
+   $(CONTRIB_BUILDDIR)/uuid/uuid_types.h
+
+libgfdb_HEADERS = gfdb_data_store.h gfdb_data_store_types.h \
+                gfdb_sqlite3.h gfdb_mem-types.h gfdb_sqlite3_helper.c
+
+CLEANFILES =
+CONFIG_CLEAN_FILES = $(CONTRIB_BUILDDIR)/uuid/uuid_types.h
+
+$(top_builddir)/libglusterfs/src/libglusterfs.la:
+	$(MAKE) -C $(top_builddir)/libglusterfs/src/ all
diff --git a/libglusterfs/src/gfdb/gfdb_data_store.c b/libglusterfs/src/gfdb/gfdb_data_store.c
new file mode 100644
index 00000000000..b250ece91e4
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_data_store.c
@@ -0,0 +1,649 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include "gfdb_data_store.h"
+#include "list.h"
+
+/******************************************************************************
+ *
+ *                     Database Connection utils/internals
+ *
+ * ****************************************************************************/
+
+/* GFDB Connection Node:
+ * ~~~~~~~~~~~~~~~~~~~~
+ * Represents the connection to the database while using libgfdb
+ * The connection node is not thread safe as far as fini_db is concerned.
+ * You can use a single connection node
+ * to do multithreaded db operations like insert/delete/find of records.
+ * But you need to wait for all the operating threads to complete i.e
+ * pthread_join() and then do fini_db() to kill the connection node.
+ * gfdb_conn_node_t is an opaque structure.
+ * */
+struct gfdb_conn_node_t {
+        gfdb_connection_t       gfdb_connection;
+        struct list_head        conn_list;
+};
+
+
+/*
+ * db_conn_list is the circular linked list which
+ * will have all the database connections for the process
+ *
+ * */
+static gfdb_conn_node_t *db_conn_list;
+
+/*
+ * db_conn_mutex is the mutex for db_conn_list
+ *
+ * */
+static pthread_mutex_t db_conn_mutex = PTHREAD_MUTEX_INITIALIZER;
+
+
+/*Checks the sanity of the connection node*/
+#define CHECK_CONN_NODE(_conn_node)\
+do {\
+        GF_ASSERT (_conn_node);\
+        GF_ASSERT (_conn_node->gfdb_connection.gf_db_connection);\
+} while (0)
+
+
+/*Check if the conn node is first in the list*/
+#define IS_FIRST_NODE(db_conn_list, _conn_node)\
+        ((_conn_node == db_conn_list) ? _gf_true : _gf_false)
+
+
+/*Check if the conn node is the only node in the list*/
+#define IS_THE_ONLY_NODE(_conn_node)\
+((_conn_node->conn_list.next == _conn_node->conn_list.prev)\
+        ? _gf_true : _gf_false)
+
+
+
+/*Internal Function: Adds connection node to the end of
+ * the db connection list.*/
+static int
+add_connection_node (gfdb_conn_node_t *_conn_node) {
+        int ret = -1;
+
+        GF_ASSERT (_conn_node);
+
+        /*Lock the list*/
+        ret = pthread_mutex_lock (&db_conn_mutex);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                "Failed lock db connection list %s", strerror(ret));
+                ret = -1;
+                goto out;
+        }
+
+        if (db_conn_list == NULL) {
+                db_conn_list = _conn_node;
+        } else {
+                list_add_tail (&_conn_node->conn_list,
+                                &db_conn_list->conn_list);
+        }
+
+        /*unlock the list*/
+        ret = pthread_mutex_unlock (&db_conn_mutex);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                "Failed unlock db connection list %s", strerror(ret));
+                ret = -1;
+                /*TODO What if the unlock fails.
+                * Will it lead to deadlock?
+                * Most of the gluster code
+                * no check for unlock or destory of mutex!*/
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+
+/*Internal Function:
+ * Delete connection node from the list*/
+static inline int
+delete_conn_node (gfdb_conn_node_t *_conn_node)
+{
+        int ret = -1;
+
+        GF_ASSERT (_conn_node);
+
+        /*Lock of the list*/
+        ret = pthread_mutex_lock (&db_conn_mutex);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                        "Failed lock on db connection list %s", strerror(ret));
+                goto out;
+        }
+
+        /*Remove the connection object from list*/
+        if (IS_THE_ONLY_NODE(_conn_node)) {
+                db_conn_list = NULL;
+                GF_FREE (_conn_node);
+        } else {
+                if (IS_FIRST_NODE(db_conn_list, _conn_node)) {
+                        db_conn_list = list_entry (db_conn_list->conn_list.next,
+                                                gfdb_conn_node_t,
+                                                conn_list);
+                }
+                list_del(&_conn_node->conn_list);
+                GF_FREE (_conn_node);
+        }
+
+        /*Release the list lock*/
+        ret =  pthread_mutex_unlock (&db_conn_mutex);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_WARNING,
+                                "Failed unlock on db connection"
+                                " list %s", strerror(ret));
+                /*TODO What if the unlock fails.
+                * Will it lead to deadlock?
+                * Most of the gluster code
+                * no check for unlock or destory of mutex!*/
+                ret = -1;
+                goto out;
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+
+/*Internal function: Used initialize/map db operation of
+ * specified type of db plugin*/
+static int
+init_db_operations (gfdb_db_type_t gfdb_db_type,
+                        gfdb_db_operations_t *gfdb_db_operations) {
+
+        int ret = -1;
+
+        GF_ASSERT (gfdb_db_operations);
+
+        /*Clear the gfdb_db_operations*/
+        gfdb_db_operations = memset(gfdb_db_operations, 0,
+                                        sizeof(*gfdb_db_operations));
+        switch (gfdb_db_type) {
+        case GFDB_SQLITE3:
+                gf_sqlite3_fill_db_operations (gfdb_db_operations);
+                ret = 0;
+                break;
+        case GFDB_HYPERDEX:
+        case GFDB_HASH_FILE_STORE:
+        case GFDB_ROCKS_DB:
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR, "Plugin not supported");
+                break;
+        case GFDB_INVALID_DB:
+        case GFDB_DB_END:
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR, "Invalid DB Type");
+                break;
+        }
+        return ret;
+}
+
+
+/******************************************************************************
+ *
+ *                      LIBGFDB API Functions
+ *
+ * ****************************************************************************/
+
+
+/*Libgfdb API Function: Used to initialize a db connection
+ *                      (Constructor function for db connection object)
+ * Arguments:
+ *      args         :  Dictionary containing database specific parameters
+ *                      eg: For sqlite3, pagesize, cachesize, db name, db path
+                        etc
+ *      gfdb_db_type :  Type of data base used i.e sqlite or hyperdex etc
+ * Returns : if successful return the GFDB Connection node to the caller or
+ *          NULL in case of failure*/
+gfdb_conn_node_t *
+init_db (dict_t *args, gfdb_db_type_t gfdb_db_type)
+{
+        int ret                                 = -1;
+        gfdb_conn_node_t *_conn_node            = NULL;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+
+        /*Create data base connection object*/
+        _conn_node = GF_CALLOC (1, sizeof(gfdb_conn_node_t),
+                                        gf_mt_db_conn_node_t);
+        if (!_conn_node) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Failed mem alloc for gfdb_conn_node_t!");
+                goto alloc_failed;
+        }
+
+        /*Init the list component of db conneciton object*/
+        INIT_LIST_HEAD (&_conn_node->conn_list);
+
+
+        /*Add created connection node to the list*/
+        ret = add_connection_node (_conn_node);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                       "Failed to add connection node to list");
+                goto _conn_failed;
+        }
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+
+        /*init the db ops object of db connection object*/
+        ret = init_db_operations(gfdb_db_type, db_operations_t);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                       "Failed initializing database operation failed.");
+                ret = -1;
+                goto init_db_failed;
+        }
+
+        /*Calling the init_db_op of the respected db type*/
+        GF_ASSERT (db_operations_t->init_db_op);
+        ret = db_operations_t->init_db_op (args, &_conn_node->gfdb_connection.
+                                                gf_db_connection);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                       "Failed initializing database");
+                ret = -1;
+                goto init_db_failed;
+        }
+        _conn_node->gfdb_connection.gfdb_db_type = gfdb_db_type;
+        ret = 0;
+
+        return _conn_node;
+
+        /*****Error Handling********/
+        /* If init_db_operations or init_db of plugin failed delete
+        * conn node from the list.
+        * connection node will be free by delete_conn_node*/
+init_db_failed:
+        ret = delete_conn_node (_conn_node);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Failed deleting connection node from list");
+        }
+        return NULL;
+        /*if adding to the list failed free connection node*/
+_conn_failed:
+        GF_FREE (_conn_node);
+        /*if allocation failed*/
+alloc_failed:
+        return NULL;
+        /*****Error Handling********/
+}
+
+
+
+
+
+/*Libgfdb API Function: Used to terminate/de-initialize db connection
+ *                      (Destructor function for db connection object)
+ * Arguments:
+ *      _conn_node  :  GFDB Connection node
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+fini_db (gfdb_conn_node_t *_conn_node)
+{
+        int ret = -1;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+
+        GF_ASSERT (db_operations_t->fini_db_op);
+
+        ret = db_operations_t->fini_db_op(&_conn_node->gfdb_connection.
+                                                        gf_db_connection);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                       "Failed close the db connection");
+                goto out;
+        }
+
+        ret = delete_conn_node (_conn_node);
+        if (ret) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Failed deleting connection node from list");
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+
+
+
+
+
+/*Libgfdb API Function: Used to insert/update records in the database
+ *                      NOTE: In current gfdb_sqlite plugin we use that
+ *                      same function to delete the record. Set the
+ *                      gfdb_fop_path to GFDB_FOP_UNDEL to delete the
+ *                      link of inode from GF_FLINK_TB and
+ *                      GFDB_FOP_UNDEL_ALL to delete all the records from
+ *                      GF_FLINK_TB and GF_FILE_TB.
+ *                      TODO: Should seperate this function into the
+ *                      delete_record function
+ *                      Refer CTR Xlator features/changetimerecorder for usage
+ * Arguments:
+ *      _conn_node     :  GFDB Connection node
+ *      gfdb_db_record :  Record to be inserted/updated
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+insert_record (gfdb_conn_node_t *_conn_node,
+                        gfdb_db_record_t *gfdb_db_record)
+{
+        int ret = 0;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+        void *gf_db_connection                  = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+        gf_db_connection = _conn_node->gfdb_connection.gf_db_connection;
+
+        if (db_operations_t->insert_record_op) {
+
+                ret = db_operations_t->insert_record_op (gf_db_connection,
+                                                        gfdb_db_record);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Insert/Update operation failed!");
+                }
+        }
+
+        return ret;
+}
+
+
+
+
+/*Libgfdb API Function: Used to delete record from the database
+ *                      NOTE: In the current gfdb_sqlite3 plugin
+ *                      implementation this function is dummy.
+ *                      Use the insert_record function.
+ *                      Refer CTR Xlator features/changetimerecorder for usage
+ * Arguments:
+ *      _conn_node     :  GFDB Connection node
+ *      gfdb_db_record :  Record to be deleted
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+delete_record (gfdb_conn_node_t *_conn_node,
+                        gfdb_db_record_t *gfdb_db_record)
+{
+        int ret = 0;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+        void *gf_db_connection                  = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+        gf_db_connection = _conn_node->gfdb_connection.gf_db_connection;
+
+        if (db_operations_t->delete_record_op) {
+
+                ret = db_operations_t->delete_record_op (gf_db_connection,
+                                                        gfdb_db_record);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Delete operation failed!");
+                }
+
+        }
+
+        return ret;
+}
+
+
+
+
+
+/*Libgfdb API Function: Query all the records from the database
+ * Arguments:
+ *      _conn_node      : GFDB Connection node
+ *      query_callback  : Call back function that will be called
+ *                        for every record found
+ *      _query_cbk_args : Custom argument passed for the call back
+ *                        function query_callback
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+find_all(gfdb_conn_node_t *_conn_node, gf_query_callback_t query_callback,
+                void *_query_cbk_args) {
+        int ret = 0;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+        void *gf_db_connection                  = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+        gf_db_connection = _conn_node->gfdb_connection.gf_db_connection;
+
+        if (db_operations_t->find_all_op) {
+                ret = db_operations_t->find_all_op (gf_db_connection,
+                                                query_callback,
+                                                _query_cbk_args);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Find all operation failed!");
+                }
+
+        }
+
+        return ret;
+}
+
+
+
+/*Libgfdb API Function: Query records/files that have not changed/accessed
+ *                      from a time in past to current time
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are not
+ *                                changed/accessed
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+find_unchanged_for_time(gfdb_conn_node_t *_conn_node,
+                                        gf_query_callback_t query_callback,
+                                        void *_query_cbk_args,
+                                        gfdb_time_t *for_time) {
+
+        int ret = 0;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+        void *gf_db_connection                  = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+        gf_db_connection = _conn_node->gfdb_connection.gf_db_connection;
+
+        if (db_operations_t->find_unchanged_for_time_op) {
+
+                ret = db_operations_t->find_unchanged_for_time_op
+                                                        (gf_db_connection,
+                                                        query_callback,
+                                                        _query_cbk_args,
+                                                        for_time);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Find unchanged operation failed!");
+                }
+
+        }
+
+        return ret;
+}
+
+/*Libgfdb API Function: Query records/files that have changed/accessed from a
+ *                      time in past to current time
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are
+ *                                changed/accessed
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+find_recently_changed_files(gfdb_conn_node_t *_conn_node,
+                                gf_query_callback_t query_callback,
+                                void *_query_cbk_args,
+                                gfdb_time_t *from_time) {
+
+        int ret = 0;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+        void *gf_db_connection                  = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+        gf_db_connection = _conn_node->gfdb_connection.gf_db_connection;
+
+        if (db_operations_t->find_recently_changed_files_op) {
+
+                ret =  db_operations_t->find_recently_changed_files_op (
+                                                        gf_db_connection,
+                                                        query_callback,
+                                                        _query_cbk_args,
+                                                        from_time);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Find changed operation failed!");
+                }
+
+        }
+
+        return ret;
+
+}
+
+/*Libgfdb API Function: Query records/files that have not changed/accessed
+ *                      from a time in past to current time, with
+ *                      a desired frequency
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are not
+ *                                changed/accessed
+ *      write_freq_thresold     : Desired Write Frequency lower limit
+ *      read_freq_thresold      : Desired Read Frequency lower limit
+ *      _clear_counters         : If true, Clears all the frequency counters of
+ *                                all files.
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+find_unchanged_for_time_freq(gfdb_conn_node_t *_conn_node,
+                                        gf_query_callback_t query_callback,
+                                        void *_query_cbk_args,
+                                        gfdb_time_t *for_time,
+                                        int write_freq_thresold,
+                                        int read_freq_thresold,
+                                        gf_boolean_t _clear_counters) {
+
+        int ret = 0;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+        void *gf_db_connection                  = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+        gf_db_connection = _conn_node->gfdb_connection.gf_db_connection;
+
+        if (db_operations_t->find_unchanged_for_time_freq_op) {
+
+                ret = db_operations_t->find_unchanged_for_time_freq_op(
+                                                        gf_db_connection,
+                                                        query_callback,
+                                                        _query_cbk_args,
+                                                        for_time,
+                                                        write_freq_thresold,
+                                                        read_freq_thresold,
+                                                        _clear_counters);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Find unchanged with freq operation failed!");
+                }
+
+        }
+
+        return ret;
+}
+
+/*Libgfdb API Function: Query records/files that have changed/accessed from a
+ *                      time in past to current time, with
+ *                      a desired frequency
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are
+ *                                changed/accessed
+ *      write_freq_thresold     : Desired Write Frequency lower limit
+ *      read_freq_thresold      : Desired Read Frequency lower limit
+ *      _clear_counters         : If true, Clears all the frequency counters of
+ *                                all files.
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+find_recently_changed_files_freq(gfdb_conn_node_t *_conn_node,
+                                gf_query_callback_t query_callback,
+                                void *_query_cbk_args,
+                                gfdb_time_t *from_time,
+                                int write_freq_thresold,
+                                int read_freq_thresold,
+                                gf_boolean_t _clear_counters) {
+
+        int ret = 0;
+        gfdb_db_operations_t *db_operations_t   = NULL;
+        void *gf_db_connection                  = NULL;
+
+        CHECK_CONN_NODE(_conn_node);
+
+        db_operations_t = &_conn_node->gfdb_connection.gfdb_db_operations;
+        gf_db_connection = _conn_node->gfdb_connection.gf_db_connection;
+
+        if (db_operations_t->find_recently_changed_files_freq_op) {
+
+                ret =  db_operations_t->find_recently_changed_files_freq_op(
+                                                        gf_db_connection,
+                                                        query_callback,
+                                                        _query_cbk_args,
+                                                        from_time,
+                                                        write_freq_thresold,
+                                                        read_freq_thresold,
+                                                        _clear_counters);
+                if (ret) {
+                        gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                                "Find changed with freq operation failed!");
+                }
+
+        }
+
+        return ret;
+
+}
diff --git a/libglusterfs/src/gfdb/gfdb_data_store.h b/libglusterfs/src/gfdb/gfdb_data_store.h
new file mode 100644
index 00000000000..1212c2b3fe1
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_data_store.h
@@ -0,0 +1,219 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#ifndef __GFDB_DATA_STORE_H
+#define __GFDB_DATA_STORE_H
+
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "glusterfs.h"
+#include "xlator.h"
+#include "logging.h"
+#include "common-utils.h"
+#include <time.h>
+#include <sys/time.h>
+
+#include "gfdb_data_store_types.h"
+#include "gfdb_sqlite3.h"
+
+
+/* GFDB Connection Node:
+ * ~~~~~~~~~~~~~~~~~~~~
+ * Represents the connection to the database while using libgfdb
+ * The connection node is not thread safe as far as fini_db is concerned.
+ * You can use a single connection node
+ * to do multithreaded db operations like insert/delete/find of records.
+ * But you need to wait for all the operating threads to complete i.e
+ * pthread_join() and then do fini_db() to kill the connection node.
+ * gfdb_conn_node_t is an opaque structure.
+ * */
+typedef struct gfdb_conn_node_t gfdb_conn_node_t;
+
+
+
+
+/*Libgfdb API Function: Used to initialize db connection
+ * Arguments:
+ *      args         :  Dictionary containing database specific parameters
+ *                      eg: For sqlite3, pagesize, cachesize, db name, db path
+                        etc
+ *      gfdb_db_type :  Type of data base used i.e sqlite or hyperdex etc
+ * Returns : if successful return the GFDB Connection Node to the caller or
+ *          NULL value in case of failure*/
+gfdb_conn_node_t *
+init_db(dict_t *arg, gfdb_db_type_t db_type);
+
+
+
+
+
+/*Libgfdb API Function: Used to terminate/de-initialize db connection
+ *                      (Destructor function for db connection object)
+ * Arguments:
+ *      _conn_node  :  DB Connection Index of the DB Connection
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+fini_db(gfdb_conn_node_t *);
+
+
+
+
+/*Libgfdb API Function: Used to insert/updated records in the database
+ *                      NOTE: In current gfdb_sqlite plugin we use that
+ *                      same function to delete the record. Set the
+ *                      gfdb_fop_path to GFDB_FOP_UNDEL to delete the
+ *                      link of inode from GF_FLINK_TB and
+ *                      GFDB_FOP_UNDEL_ALL to delete all the records from
+ *                      GF_FLINK_TB and GF_FILE_TB.
+ *                      TODO: Should seperate this function into the
+ *                      delete_record function
+ *                      Refer CTR Xlator features/changetimerecorder for usage
+ * Arguments:
+ *      _conn_node     :  GFDB Connection node
+ *      gfdb_db_record :  Record to be inserted/updated
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+insert_record(gfdb_conn_node_t *, gfdb_db_record_t *gfdb_db_record);
+
+
+
+
+/*Libgfdb API Function: Used to delete record from the database
+ *                      NOTE: In the current gfdb_sqlite3 plugin
+ *                      implementation this function is dummy.
+ *                      Use the insert_record function.
+ *                      Refer CTR Xlator features/changetimerecorder for usage
+ * Arguments:
+ *      _conn_node     :  GFDB Connection node
+ *      gfdb_db_record :  Record to be deleted
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int
+delete_record(gfdb_conn_node_t *, gfdb_db_record_t *gfdb_db_record);
+
+
+
+
+
+/*Libgfdb API Function: Query all the records from the database
+ * Arguments:
+ *      _conn_node      : GFDB Connection node
+ *      query_callback  : Call back function that will be called
+ *                        for every record found
+ *      _query_cbk_args : Custom argument passed for the call back
+ *                        function query_callback
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int find_all(gfdb_conn_node_t *, gf_query_callback_t query_callback,
+                void *_query_cbk_args);
+
+
+
+
+
+/*Libgfdb API Function: Query records/files that have not changed/accessed
+ *                      from a time in past to current time
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are not
+ *                                changed/accessed
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int find_unchanged_for_time(gfdb_conn_node_t *,
+                        gf_query_callback_t query_callback,
+                        void *_query_cbk_args, gfdb_time_t *for_time);
+
+
+
+
+/*Libgfdb API Function: Query records/files that have changed/accessed from a
+ *                      time in past to current time
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are
+ *                                changed/accessed
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int find_recently_changed_files(gfdb_conn_node_t *_conn,
+                gf_query_callback_t query_callback, void *_query_cbk_args,
+                gfdb_time_t *from_time);
+
+
+
+
+
+/*Libgfdb API Function: Query records/files that have not changed/accessed
+ *                      from a time in past to current time, with
+ *                      a desired frequency
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are not
+ *                                changed/accessed
+ *      write_freq_thresold     : Desired Write Frequency lower limit
+ *      read_freq_thresold      : Desired Read Frequency lower limit
+ *      _clear_counters         : If true, Clears all the frequency counters of
+ *                                all files.
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int find_unchanged_for_time_freq(gfdb_conn_node_t *_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_query_cbk_args,
+                                        gfdb_time_t *for_time,
+                                        int write_freq_thresold,
+                                        int read_freq_thresold,
+                                        gf_boolean_t _clear_counters);
+
+
+
+
+
+/*Libgfdb API Function: Query records/files that have changed/accessed from a
+ *                      time in past to current time, with
+ *                      a desired frequency
+ * Arguments:
+ *      _conn_node              : GFDB Connection node
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _query_cbk_args         : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are
+ *                                changed/accessed
+ *      write_freq_thresold     : Desired Write Frequency lower limit
+ *      read_freq_thresold      : Desired Read Frequency lower limit
+ *      _clear_counters         : If true, Clears all the frequency counters of
+ *                                all files.
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+int find_recently_changed_files_freq(gfdb_conn_node_t *_conn,
+                                gf_query_callback_t query_callback,
+                                void *_query_cbk_args,
+                                gfdb_time_t *from_time,
+                                int write_freq_thresold,
+                                int read_freq_thresold,
+                                gf_boolean_t _clear_counters);
+
+#endif
diff --git a/libglusterfs/src/gfdb/gfdb_data_store_types.h b/libglusterfs/src/gfdb/gfdb_data_store_types.h
new file mode 100644
index 00000000000..e3a2c76435b
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_data_store_types.h
@@ -0,0 +1,733 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#ifndef __GFDB_DATA_STORE_TYPE_H
+#define __GFDB_DATA_STORE_TYPE_H
+
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include <time.h>
+#include <sys/time.h>
+#include <string.h>
+
+#include "common-utils.h"
+#include "gfdb_mem-types.h"
+#include "dict.h"
+
+typedef enum gf_db_operation {
+        GFDB_INVALID_DB_OP = -1,
+        /* Query DB OPS : All the Query DB_OP should be added */
+        /* in between START and END */
+        GFDB_QUERY_DB_OP_START, /* Start of Query DB_OP  */
+        GFDB_QUERY_DB_OP,
+        GF_FTABLE_EXISTS_DB_OP,
+        GFDB_QUERY_DB_OP_END, /* End of Query DB_OP */
+        /* Non-Query DB OPS */
+        GFDB_DB_CREATE_DB_OP,
+        GFDB_GFID_EXIST_DB_OP,
+        GFDB_W_INSERT_DB_OP,
+        GFDB_WU_INSERT_DB_OP,
+        GFDB_W_UPDATE_DB_OP,
+        GFDB_WU_UPDATE_DB_OP,
+        GFDB_W_DELETE_DB_OP,
+        GFDB_UW_DELETE_DB_OP,
+        GFDB_WFC_UPDATE_DB_OP,
+        GFDB_RFC_UPDATE_DB_OP
+} gf_db_operation_t;
+
+
+#define GF_COL_MAX_NUM                  2
+#define GF_COL_ALL                      " * "
+
+/* Column/fields names used in the DB.
+ * If any new field is added should be updated here*/
+#define GF_COL_GF_ID                    "GF_ID"
+#define GF_COL_GF_PID                   "GF_PID"
+#define GF_COL_FILE_NAME                "FNAME"
+#define GF_COL_FPATH                    "FPATH"
+#define GF_COL_WSEC                     "W_SEC"
+#define GF_COL_WMSEC                    "W_MSEC"
+#define GF_COL_UWSEC                    "UW_SEC"
+#define GF_COL_UWMSEC                   "UW_MSEC"
+#define GF_COL_WSEC_READ                "W_READ_SEC"
+#define GF_COL_WMSEC_READ               "W_READ_MSEC"
+#define GF_COL_UWSEC_READ               "UW_READ_SEC"
+#define GF_COL_UWMSEC_READ              "UW_READ_MSEC"
+#define GF_COL_WDEL_FLAG                "W_DEL_FLAG"
+#define GF_COL_WRITE_FREQ_CNTR          "WRITE_FREQ_CNTR"
+#define GF_COL_READ_FREQ_CNTR           "READ_FREQ_CNTR"
+#define GF_COL_LINK_UPDATE              "LINK_UPDATE"
+
+
+
+
+/***********************Time related********************************/
+/*1 sec = 1000000 microsec*/
+#define GFDB_MICROSEC         1000000
+
+/*All the gfdb times are represented using this structure*/
+typedef  struct timeval gfdb_time_t;
+
+/*Convert time into  seconds*/
+static inline long int
+gfdb_time_2_usec(gfdb_time_t *gfdb_time)
+{
+        GF_ASSERT(gfdb_time);
+        return gfdb_time->tv_sec * GFDB_MICROSEC + gfdb_time->tv_usec;
+}
+
+
+
+
+
+/******************************************************************************
+ *
+ *              Insert/Update Record related data structures/functions
+ *
+ * ****************************************************************************/
+
+
+
+
+/*Indicated a generic synchronous write to the db
+ * This may or may not be implemented*/
+typedef enum gfdb_sync_type {
+        GFDB_INVALID_SYNC = -1,
+        GFDB_DB_ASYNC,
+        GFDB_DB_SYNC
+} gfdb_sync_type_t;
+
+/*Strings related to the abvove sync type*/
+#define GFDB_STR_DB_ASYNC        "async"
+#define GFDB_STR_DB_SYNC         "sync"
+
+/*To convert sync type from string to gfdb_sync_type_t*/
+static inline int
+gf_string2gfdbdbsync (char *sync_option)
+{
+        int ret = -1;
+
+        if (!sync_option)
+                goto out;
+        if (strcmp(sync_option, GFDB_STR_DB_ASYNC) == 0) {
+                ret = GFDB_DB_ASYNC;
+        } else if (strcmp(sync_option, GFDB_STR_DB_SYNC) == 0) {
+                ret = GFDB_DB_SYNC;
+        }
+out:
+        return ret;
+}
+
+
+
+
+
+
+/*Indicated different types of db*/
+typedef enum gfdb_db_type {
+        GFDB_INVALID_DB = -1,
+        GFDB_HASH_FILE_STORE,
+        GFDB_ROCKS_DB,
+        GFDB_SQLITE3,
+        GFDB_HYPERDEX,
+        GFDB_DB_END /*Add DB type Entries above this only*/
+} gfdb_db_type_t;
+
+/*String related to the db types*/
+#define GFDB_DATA_STORE               "gfdbdatastore"
+#define GFDB_STR_HASH_FILE_STORE      "hashfile"
+#define GFDB_STR_ROCKS_DB             "rocksdb"
+#define GFDB_STR_SQLITE3              "sqlite3"
+#define GFDB_STR_HYPERDEX             "hyperdex"
+
+/*Convert db type in string to gfdb_db_type_t*/
+static inline int
+gf_string2gfdbdbtype (char *db_option)
+{
+        int ret = -1;
+
+        if (!db_option)
+                goto out;
+        if (strcmp(db_option, GFDB_STR_HASH_FILE_STORE) == 0) {
+                ret = GFDB_HASH_FILE_STORE;
+        } else if (strcmp(db_option, GFDB_STR_ROCKS_DB) == 0) {
+                ret = GFDB_ROCKS_DB;
+        } else if (strcmp(db_option, GFDB_STR_SQLITE3) == 0) {
+                ret = GFDB_SQLITE3;
+        } else if (strcmp(db_option, GFDB_STR_HYPERDEX) == 0) {
+                ret = GFDB_HYPERDEX;
+        }
+out:
+        return ret;
+}
+
+
+
+
+
+
+
+/*Tells the path of the fop*/
+typedef enum gfdb_fop_path {
+        GFDB_FOP_INVALID = -1,
+        /*Filler value for zero*/
+        GFDB_FOP_PATH_ZERO = 0,
+        /*have wind path below this*/
+        GFDB_FOP_WIND = 1,
+        GFDB_FOP_WDEL = 2,
+        /*have unwind path below this*/
+        GFDB_FOP_UNWIND = 4,
+        /*Delete unwind path*/
+        GFDB_FOP_UNDEL = 8,
+        GFDB_FOP_UNDEL_ALL = 16
+} gfdb_fop_path_t;
+/*Strings related to the above fop path*/
+#define GFDB_STR_FOP_INVALID "INVALID"
+#define GFDB_STR_FOP_WIND "ENTRY"
+#define GFDB_STR_FOP_UNWIND "EXIT"
+#define GFDB_STR_FOP_WDEL "WDEL"
+#define GFDB_STR_FOP_UNDEL "UNDEL"
+
+static inline gf_boolean_t
+iswindpath(gfdb_fop_path_t gfdb_fop_path)
+{
+        return ((gfdb_fop_path == GFDB_FOP_WIND) ||
+                        (gfdb_fop_path == GFDB_FOP_WDEL)) ?
+                                                _gf_true : _gf_false;
+}
+
+static inline gf_boolean_t
+isunwindpath(gfdb_fop_path_t gfdb_fop_path)
+{
+        return (gfdb_fop_path >= GFDB_FOP_UNWIND) ? _gf_true : _gf_false;
+}
+
+
+
+
+
+
+
+/*Tell what type of fop it was
+ * Like whether a dentry fop or a inode fop
+ * Read fop or a write fop etc*/
+typedef enum gfdb_fop_type {
+        GFDB_FOP_INVALID_OP = -1,
+        /*Filler value for zero*/
+        GFDB_FOP_TYPE_ZERO = 0,
+        GFDB_FOP_DENTRY_OP = 1,
+        GFDB_FOP_DENTRY_CREATE_OP = 2,
+        GFDB_FOP_INODE_OP = 4,
+        GFDB_FOP_WRITE_OP = 8,
+        GFDB_FOP_READ_OP = 16
+} gfdb_fop_type_t;
+
+#define GFDB_FOP_INODE_WRITE\
+        (GFDB_FOP_INODE_OP | GFDB_FOP_WRITE_OP)
+
+#define GFDB_FOP_DENTRY_WRITE\
+        (GFDB_FOP_DENTRY_OP | GFDB_FOP_WRITE_OP)
+
+#define GFDB_FOP_CREATE_WRITE\
+        (GFDB_FOP_DENTRY_CREATE_OP | GFDB_FOP_WRITE_OP)
+
+#define GFDB_FOP_INODE_READ\
+        (GFDB_FOP_INODE_OP | GFDB_FOP_READ_OP)
+
+static inline gf_boolean_t
+isreadfop(gfdb_fop_type_t fop_type)
+{
+        return (fop_type & GFDB_FOP_READ_OP) ? _gf_true : _gf_false;
+}
+
+static inline gf_boolean_t
+isdentryfop(gfdb_fop_type_t fop_type)
+{
+        return ((fop_type & GFDB_FOP_DENTRY_OP) ||
+                (fop_type & GFDB_FOP_DENTRY_CREATE_OP)) ? _gf_true : _gf_false;
+}
+
+static inline gf_boolean_t
+isdentrycreatefop(gfdb_fop_type_t fop_type)
+{
+        return (fop_type & GFDB_FOP_DENTRY_CREATE_OP) ?
+                        _gf_true : _gf_false;
+}
+
+
+
+
+
+
+
+/*The structure that is used to send insert/update the databases
+ * using insert_db api*/
+typedef struct gfdb_db_record {
+        uuid_t                          gfid;
+        uuid_t                          pargfid;
+        uuid_t                          old_pargfid;
+        char                            file_name[PATH_MAX];
+        char                            file_path[PATH_MAX];
+        char                            old_file_name[PATH_MAX];
+        char                            old_path[PATH_MAX];
+        gfdb_fop_type_t                 gfdb_fop_type;
+        gfdb_fop_path_t                 gfdb_fop_path;
+        /*Time of change or access*/
+        gfdb_time_t                     gfdb_wind_change_time;
+        gfdb_time_t                     gfdb_unwind_change_time;
+        /* For crash consistancy while inserting/updating hard links */
+        gf_boolean_t                    islinkupdate;
+        /* For dentry fops we can choose to ignore recording of unwind time */
+        /* For inode fops "record_exit" volume option does the trick,       */
+        /* but for dentry fops we update the LINK_UPDATE, so an extra       */
+        /* flag is provided to ignore the recording of the unwind time.     */
+        gf_boolean_t                    do_record_uwind_time;
+        gf_boolean_t                    do_record_counters;
+} gfdb_db_record_t;
+
+
+
+/*******************************************************************************
+ *
+ *                     Query related data structure and functions
+ *
+ * ****************************************************************************/
+
+
+
+/*Structure used for querying purpose*/
+typedef struct gfdb_query_record {
+        /*Inode info*/
+        uuid_t                          gfid;
+        /*All the hard link of the inode
+         * All the hard links will be queried as
+         * "GF_PID,FNAME,FPATH,W_DEL_FLAG,LINK_UPDATE"
+         * and multiple hardlinks will be seperated by "::"*/
+        /*Do only shallow copy. The gf_query_callback_t */
+        /* function should do the deep copy.*/
+        char                            *_link_info_str;
+        ssize_t                         link_info_size;
+} gfdb_query_record_t;
+
+/*Function to create the query_record*/
+static inline gfdb_query_record_t *
+gfdb_query_record_init()
+{
+        int ret = -1;
+        gfdb_query_record_t *gfdb_query_record = NULL;
+
+        gfdb_query_record = GF_CALLOC (1, sizeof(gfdb_query_record_t),
+                                        gf_mt_gfdb_query_record_t);
+        if (!gfdb_query_record) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                        "Error allocating memory to gfdb_query_record ");
+                goto out;
+        }
+        ret = 0;
+out:
+        if (ret == -1) {
+                GF_FREE (gfdb_query_record);
+        }
+        return gfdb_query_record;
+}
+
+/*Function to destroy query record*/
+static inline void
+gfdb_query_record_fini(gfdb_query_record_t
+                                                **gfdb_query_record) {
+        GF_FREE (*gfdb_query_record);
+}
+
+
+
+
+
+
+
+
+/*Structure to hold the link information*/
+typedef struct gfdb_link_info {
+        uuid_t                          pargfid;
+        char                            file_name[PATH_MAX];
+        char                            file_path[PATH_MAX];
+        gf_boolean_t                    is_link_updated;
+        gf_boolean_t                    is_del_flag_set;
+} gfdb_link_info_t;
+
+/*Create a single link info structure*/
+static inline gfdb_link_info_t *
+gfdb_link_info_init ()
+{
+        gfdb_link_info_t *gfdb_link_info = NULL;
+
+        gfdb_link_info = GF_CALLOC (1, sizeof(gfdb_link_info_t),
+                                        gf_mt_gfdb_link_info_t);
+        if (!gfdb_link_info) {
+                gf_log (GFDB_DATA_STORE, GF_LOG_ERROR,
+                        "Error allocating memory to gfdb_link_info ");
+        }
+
+        return gfdb_link_info;
+}
+
+/*Destroy a link info structure*/
+static inline void
+gfdb_link_info_fini(gfdb_link_info_t **gfdb_link_info)
+{
+        if (gfdb_link_info)
+                GF_FREE (*gfdb_link_info);
+}
+
+
+/*Length of each hard link string */
+#define DEFAULT_LINK_INFO_STR_LEN       1024
+
+/* Parse a single link string into link_info structure
+ * Input format of str_link
+ *      "GF_PID,FNAME,FPATH,W_DEL_FLAG,LINK_UPDATE"
+ *
+ * */
+static inline int
+str_to_link_info (char *str_link,
+                             gfdb_link_info_t *link_info)
+{
+        int     ret = -1;
+        const char *delimiter = ",";
+        char *token_str = NULL;
+        char *saveptr = NULL;
+        char gfid[200] = "";
+
+        GF_ASSERT (str_link);
+        GF_ASSERT (link_info);
+
+        /*Parent GFID*/
+        token_str = strtok_r(str_link, delimiter, &saveptr);
+        if (token_str != NULL) {
+                strcpy (gfid, token_str);
+                ret = uuid_parse (gfid, link_info->pargfid);
+                if (ret == -1)
+                        goto out;
+        }
+
+        /*Filename*/
+        token_str = strtok_r(NULL, delimiter, &saveptr);
+        if (token_str != NULL) {
+                strcpy (link_info->file_name, token_str);
+        }
+
+        /*Filepath*/
+        token_str = strtok_r(NULL, delimiter, &saveptr);
+        if (token_str != NULL) {
+                strcpy (link_info->file_path, token_str);
+        }
+
+        /*is_link_updated*/
+        token_str = strtok_r(NULL, delimiter, &saveptr);
+        if (token_str != NULL) {
+                link_info->is_link_updated = atoi(token_str);
+                if (link_info->is_link_updated != 0 &&
+                                link_info->is_link_updated != 1) {
+                        goto out;
+                }
+        }
+
+        /*is_del_flag_set*/
+        token_str = strtok_r(NULL, delimiter, &saveptr);
+        if (token_str != NULL) {
+                link_info->is_del_flag_set = atoi (token_str);
+                if (link_info->is_del_flag_set != 0 &&
+                                link_info->is_del_flag_set != 1) {
+                        goto out;
+                }
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+
+/*******************************************************************************
+ *
+ *                           Signatures for the plugin functions
+ *                           i.e Any plugin should implementment
+ *                           these functions to integrate with
+ *                           libgfdb.
+ *
+ * ****************************************************************************/
+
+/*Call back function for querying the database*/
+typedef int
+(*gf_query_callback_t)(gfdb_query_record_t *, void *);
+
+/* Used to initialize db connection
+ * Arguments:
+ *      args         : Dictionary containing database specific parameters
+ *      db_conn      : pointer to plugin specific data base connection
+ *                     that will be created. If the call is successful
+ *                     db_conn will contain the plugin specific connection
+ *                     If call is unsuccessful will have NULL.
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_init_db_t)(dict_t *args, void **db_conn);
+
+
+
+
+/* Used to terminate/de-initialize db connection
+ *                      (Destructor function for db connection object)
+ * Arguments:
+ *      db_conn        : plugin specific data base connection
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_fini_db_t)(void **db_conn);
+
+
+
+
+/*Used to insert/updated records in the database
+ * Arguments:
+ *      db_conn        : plugin specific data base connection
+ *      gfdb_db_record :  Record to be inserted/updated
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_insert_record_t)(void *db_conn,
+                        gfdb_db_record_t *db_record);
+
+
+
+
+/*Used to delete record from the database
+ * Arguments:
+ *      db_conn        : plugin specific data base connection
+ *      gfdb_db_record :  Record to be deleted
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_delete_record_t)(void *db_conn,
+                        gfdb_db_record_t *db_record);
+
+
+
+
+/* Query all the records from the database
+ * Arguments:
+ *      db_conn        : plugin specific data base connection
+ *      query_callback  : Call back function that will be called
+ *                        for every record found
+ *      _query_cbk_args : Custom argument passed for the call back
+ *                        function query_callback
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_find_all_t)(void *db_conn,
+                               gf_query_callback_t query_callback,
+                               void *_cbk_args);
+
+
+
+
+/* Query records/files that have not changed/accessed
+ *                      from a time in past to current time
+ * Arguments:
+ *      db_conn                 : plugin specific data base connection
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _cbk_args               : Custom argument passed for the call back
+ *                                function query_callback
+ *      for_time                : Time from where the file/s are not
+ *                                changed/accessed
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_find_unchanged_for_time_t)(void *db_conn,
+                                  gf_query_callback_t query_callback,
+                                  void *_cbk_args,
+                                  gfdb_time_t *_time);
+
+
+
+/* Query records/files that have changed/accessed from a
+ *                      time in past to current time
+ * Arguments:
+ *      db_conn                 : plugin specific data base connection
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _cbk_args               : Custom argument passed for the call back
+ *                                function query_callback
+ *      _time                   : Time from where the file/s are
+ *                                changed/accessed
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_find_recently_changed_files_t)(void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_cbk_args, gfdb_time_t *_time);
+
+/* Query records/files that have not changed/accessed
+ * from a time in past to current time, with
+ * a desired frequency
+ *
+ * Arguments:
+ *      db_conn                 : plugin specific data base connection
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _cbk_args               : Custom argument passed for the call back
+ *                                function query_callback
+ *      _time                   : Time from where the file/s are not
+ *                                changed/accessed
+ *      _write_freq             : Desired Write Frequency lower limit
+ *      _read_freq              : Desired Read Frequency lower limit
+ *      _clear_counters         : If true, Clears all the frequency counters of
+ *                                all files.
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_find_unchanged_for_time_freq_t)
+                                        (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_cbk_args, gfdb_time_t *_time,
+                                        int _write_freq, int _read_freq,
+                                        gf_boolean_t _clear_counters);
+
+
+
+
+/* Query records/files that have changed/accessed from a
+ * time in past to current time, with a desired frequency
+ * Arguments:
+ *      db_conn                 : plugin specific data base connection
+ *      query_callback          : Call back function that will be called
+ *                                for every record found
+ *      _cbk_args               : Custom argument passed for the call back
+ *                                function query_callback
+ *      _time                   : Time from where the file/s are
+ *                                changed/accessed
+ *      _write_freq             : Desired Write Frequency lower limit
+ *      _read_freq              : Desired Read Frequency lower limit
+ *      _clear_counters         : If true, Clears all the frequency counters of
+ *                                all files.
+ * Returns : if successful return 0 or
+ *          -ve value in case of failure*/
+typedef int
+(*gfdb_find_recently_changed_files_freq_t)(void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_cbk_args, gfdb_time_t *_time,
+                                        int _write_freq, int _read_freq,
+                                        gf_boolean_t _clear_counters);
+
+
+
+
+/*Data structure holding all the above plugin function pointers*/
+typedef struct gfdb_db_operations {
+        gfdb_init_db_t                        init_db_op;
+        gfdb_fini_db_t                        fini_db_op;
+        gfdb_insert_record_t                  insert_record_op;
+        gfdb_delete_record_t                  delete_record_op;
+        gfdb_find_all_t                       find_all_op;
+        gfdb_find_unchanged_for_time_t        find_unchanged_for_time_op;
+        gfdb_find_recently_changed_files_t    find_recently_changed_files_op;
+        gfdb_find_unchanged_for_time_freq_t
+                                        find_unchanged_for_time_freq_op;
+        gfdb_find_recently_changed_files_freq_t
+                                        find_recently_changed_files_freq_op;
+} gfdb_db_operations_t;
+
+
+
+/*******************************************************************************
+ *
+ * Database connection object: This objected is maitained by libgfdb for each
+ * database connection created.
+ * gf_db_connection     : DB connection specific to the plugin
+ * gfdb_db_operations   : Contains all the libgfdb API implementation
+ *                        from the plugin.
+ * gfdb_db_type         : Type of database
+ *
+ * ****************************************************************************/
+
+
+typedef struct gfdb_connection {
+        void *gf_db_connection;
+        gfdb_db_operations_t gfdb_db_operations;
+        gfdb_db_type_t gfdb_db_type;
+} gfdb_connection_t;
+
+
+
+
+/*******************************************************************************
+ *
+ *                    Macros for get and set db options
+ *
+ * ****************************************************************************/
+
+
+/*Set param_key : str_value into param_dict*/
+#define SET_DB_PARAM_TO_DICT(comp_name, params_dict, param_key,\
+                                                        str_value, ret, error)\
+        do {\
+                data_t *data     = NULL;\
+                data            = str_to_data (str_value);\
+                if (!data)\
+                        goto error;\
+                ret = dict_add (params_dict, param_key, data);\
+                if (ret) {\
+                        gf_log (comp_name, GF_LOG_ERROR,\
+                                "Failed setting %s to params dictionary",\
+                                param_key);\
+                        goto error;\
+                };\
+        } while (0)
+
+/*get str_value of param_key from param_dict*/
+#define GET_DB_PARAM_FROM_DICT(comp_name, params_dict, param_key, str_value,\
+                                error)\
+        do {\
+                data_t *data    = NULL;\
+                data = dict_get (params_dict, param_key);\
+                if (!data) {\
+                        gf_log (comp_name, GF_LOG_ERROR,\
+                                "Failed retriving %s from params", param_key);\
+                        goto error;\
+                } else {\
+                        str_value = data->data;\
+                };\
+        } while (0)
+
+
+/*get str_value of param_key from param_dict. if param_key is not present
+ * set _default_v to str_value */
+#define GET_DB_PARAM_FROM_DICT_DEFAULT(comp_name, params_dict, param_key,\
+                                str_value, _default_v)\
+        do {\
+                data_t *data    = NULL;\
+                data = dict_get (params_dict, param_key);\
+                if (!data) {\
+                        str_value = _default_v;\
+                        gf_log (comp_name, GF_LOG_WARNING,\
+                                "Failed retriving %s from params"\
+                                "Assigning default value: %s",\
+                                param_key, _default_v);\
+                } else {\
+                        str_value = data->data;\
+                };\
+        } while (0)
+
+
+#endif
+
+
diff --git a/libglusterfs/src/gfdb/gfdb_mem-types.h b/libglusterfs/src/gfdb/gfdb_mem-types.h
new file mode 100644
index 00000000000..2a84b47fcf5
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_mem-types.h
@@ -0,0 +1,22 @@
+/*
+  Copyright (c) 2008-2015 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
+*/
+
+
+#ifndef __GFDB_MEM_TYPES_H__
+#define __GFDB_MEM_TYPES_H__
+
+#include "mem-types.h"
+
+enum gfdb_mem_types_ {
+        gfdb_mtstart = gf_common_mt_end + 1,
+        gfdb_mt_end
+};
+#endif
+
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3.c b/libglusterfs/src/gfdb/gfdb_sqlite3.c
new file mode 100644
index 00000000000..818691d7cea
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3.c
@@ -0,0 +1,1101 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include "gfdb_sqlite3.h"
+#include "gfdb_sqlite3_helper.h"
+
+/******************************************************************************
+ *
+ *                      Util functions
+ *
+ * ***************************************************************************/
+inline gf_sql_connection_t *
+gf_sql_connection_init ()
+{
+        gf_sql_connection_t *gf_sql_conn = NULL;
+
+        gf_sql_conn = GF_CALLOC (1, sizeof(gf_sql_connection_t),
+                        gf_mt_sql_connection_t);
+        if (gf_sql_conn == NULL) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Error allocating memory to gf_sql_connection_t ");
+        }
+
+        return gf_sql_conn;
+}
+
+inline void
+gf_sql_connection_fini (gf_sql_connection_t **sql_connection)
+{
+        if (!sql_connection)
+                return;
+        GF_FREE (*sql_connection);
+        *sql_connection = NULL;
+}
+
+inline const char *
+gf_sql_jm2str (gf_sql_journal_mode_t jm)
+{
+        switch (jm) {
+        case gf_sql_jm_delete:
+                return GF_SQL_JM_DELETE;
+        case gf_sql_jm_truncate:
+                return GF_SQL_JM_TRUNCATE;
+        case gf_sql_jm_persist:
+                return GF_SQL_JM_PERSIST;
+        case gf_sql_jm_memory:
+                return GF_SQL_JM_MEMORY;
+        case gf_sql_jm_wal:
+                return GF_SQL_JM_WAL;
+        case gf_sql_jm_off:
+                return GF_SQL_JM_OFF;
+        case gf_sql_jm_invalid:
+                break;
+        }
+        return NULL;
+}
+
+inline gf_sql_journal_mode_t
+gf_sql_str2jm (const char *jm_str)
+{
+        if (!jm_str) {
+                return gf_sql_jm_invalid;
+        } else if (strcmp(jm_str, GF_SQL_JM_DELETE) == 0) {
+                return gf_sql_jm_delete;
+        } else if (strcmp(jm_str, GF_SQL_JM_TRUNCATE) == 0) {
+                return gf_sql_jm_truncate;
+        } else if (strcmp(jm_str, GF_SQL_JM_PERSIST) == 0) {
+                return gf_sql_jm_persist;
+        } else if (strcmp(jm_str, GF_SQL_JM_MEMORY) == 0) {
+                return gf_sql_jm_memory;
+        } else if (strcmp(jm_str, GF_SQL_JM_WAL) == 0) {
+                return gf_sql_jm_wal;
+        } else if (strcmp(jm_str, GF_SQL_JM_OFF) == 0) {
+                return gf_sql_jm_off;
+        }
+        return gf_sql_jm_invalid;
+}
+
+inline const char *
+gf_sql_av_t2str (gf_sql_auto_vacuum_t sql_av)
+{
+        switch (sql_av) {
+        case gf_sql_av_none:
+                return GF_SQL_AV_NONE;
+        case gf_sql_av_full:
+                return GF_SQL_AV_FULL;
+        case gf_sql_av_incr:
+                return GF_SQL_AV_INCR;
+        case gf_sql_av_invalid:
+                break;
+        }
+        return NULL;
+}
+
+inline gf_sql_auto_vacuum_t
+gf_sql_str2av_t (const char *av_str)
+{
+        if (!av_str) {
+                return gf_sql_sync_invalid;
+        } else if (strcmp(av_str, GF_SQL_AV_NONE) == 0) {
+                return gf_sql_av_none;
+        } else if (strcmp(av_str, GF_SQL_AV_FULL) == 0) {
+                return gf_sql_av_full;
+        } else if (strcmp(av_str, GF_SQL_AV_INCR) == 0) {
+                return gf_sql_av_incr;
+        }
+        return gf_sql_sync_invalid;
+}
+
+inline const char *
+gf_sync_t2str (gf_sql_sync_t sql_sync)
+{
+        switch (sql_sync) {
+        case gf_sql_sync_off:
+                return GF_SQL_SYNC_OFF;
+        case gf_sql_sync_normal:
+                return GF_SQL_SYNC_NORMAL;
+        case gf_sql_sync_full:
+                return GF_SQL_SYNC_FULL;
+        case gf_sql_sync_invalid:
+                break;
+        }
+        return NULL;
+}
+
+inline gf_sql_sync_t
+gf_sql_str2sync_t (const char *sync_str)
+{
+        if (!sync_str) {
+                return gf_sql_sync_invalid;
+        } else if (strcmp(sync_str, GF_SQL_SYNC_OFF) == 0) {
+                return gf_sql_sync_off;
+        } else if (strcmp(sync_str, GF_SQL_SYNC_NORMAL) == 0) {
+                return gf_sql_sync_normal;
+        } else if (strcmp(sync_str, GF_SQL_SYNC_FULL) == 0) {
+                return gf_sql_sync_full;
+        }
+        return gf_sql_sync_invalid;
+}
+
+
+/*TODO replace GF_CALLOC by mem_pool or iobuff if required for performace */
+static inline char *
+sql_stmt_init ()
+{
+        char *sql_stmt = NULL;
+
+        sql_stmt = GF_CALLOC (GF_STMT_SIZE_MAX, sizeof(char),
+                        gf_common_mt_char);
+
+        if (!sql_stmt) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Error allocating memory to SQL Statement ");
+                goto out;
+        }
+out:
+        return sql_stmt;
+}
+
+/*TODO replace GF_FREE by mem_pool or iobuff if required for performace */
+static inline void
+sql_stmt_fini (char **sql_stmt)
+{
+        GF_FREE (*sql_stmt);
+}
+
+/******************************************************************************
+ *                      DB Essential functions used by
+ *                      create/insert/delete/update/query functions
+ *                      > execute_sqlstmt ()
+ *                      > gf_open_sqlite3_conn ()
+ *                      > gf_close_sqlite3_conn ()
+ * ***************************************************************************/
+static sqlite3 *
+gf_open_sqlite3_conn(char *sqlite3_db_path, int flags)
+{
+        sqlite3 *sqlite3_db_conn = NULL;
+        int ret = -1;
+
+        GF_ASSERT (sqlite3_db_path);
+
+        /*Creates DB if not created*/
+        ret = sqlite3_open_v2(sqlite3_db_path, &sqlite3_db_conn, flags, NULL);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "FATAL: Could open %s : %s",
+                        sqlite3_db_path, sqlite3_errmsg(sqlite3_db_conn));
+        }
+        return sqlite3_db_conn;
+}
+
+static int
+gf_close_sqlite3_conn(sqlite3 *sqlite3_db_conn)
+{
+        int ret = 0;
+
+        GF_ASSERT (sqlite3_db_conn);
+
+        if (sqlite3_db_conn) {
+                ret = sqlite3_close (sqlite3_db_conn);
+                if (ret != SQLITE_OK) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "FATAL: sqlite3 close connection failed %s",
+                                 sqlite3_errmsg(sqlite3_db_conn));
+                        ret = -1;
+                        goto out;
+                }
+        }
+        ret = 0;
+out:
+        return ret;
+}
+
+/******************************************************************************
+*
+*                      Database init / fini / create table
+*
+* ***************************************************************************/
+
+
+/*Function to fill db operations*/
+void
+gf_sqlite3_fill_db_operations(gfdb_db_operations_t  *gfdb_db_ops)
+{
+        GF_ASSERT (gfdb_db_ops);
+
+        gfdb_db_ops->init_db_op = gf_sqlite3_init;
+        gfdb_db_ops->fini_db_op = gf_sqlite3_fini;
+
+        gfdb_db_ops->insert_record_op = gf_sqlite3_insert;
+        gfdb_db_ops->delete_record_op = gf_sqlite3_delete;
+
+        gfdb_db_ops->find_all_op = gf_sqlite3_find_all;
+        gfdb_db_ops->find_unchanged_for_time_op =
+                        gf_sqlite3_find_unchanged_for_time;
+        gfdb_db_ops->find_recently_changed_files_op =
+                        gf_sqlite3_find_recently_changed_files;
+        gfdb_db_ops->find_unchanged_for_time_freq_op =
+                        gf_sqlite3_find_unchanged_for_time_freq;
+        gfdb_db_ops->find_recently_changed_files_freq_op =
+                        gf_sqlite3_find_recently_changed_files_freq;
+}
+
+
+static int
+create_filetable (sqlite3 *sqlite3_db_conn)
+{
+        int ret                         =       -1;
+        char *sql_stmt                  =       NULL;
+        char *sql_strerror              =       NULL;
+
+        GF_ASSERT(sqlite3_db_conn);
+
+        sql_stmt = sql_stmt_init();
+        if (!sql_stmt) {
+                ret = ENOMEM;
+                goto out;
+        }
+
+        GF_CREATE_STMT(sql_stmt);
+
+        ret = sqlite3_exec (sqlite3_db_conn, sql_stmt, NULL, NULL,
+                                &sql_strerror);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing: %s : %s",
+                        sql_stmt, sql_strerror);
+                sqlite3_free (sql_strerror);
+                        ret = -1;
+                        goto out;
+        }
+
+
+        ret = 0;
+out:
+        sql_stmt_fini (&sql_stmt);
+        return ret;
+}
+
+
+
+
+static inline int
+apply_sql_params_db(gf_sql_connection_t *sql_conn, dict_t *param_dict)
+{
+        int ret = -1;
+        char *temp_str = NULL;
+        char sqlite3_config_str[PATH_MAX] = "";
+
+        GF_ASSERT(sql_conn);
+        GF_ASSERT(param_dict);
+
+        /*Extract sql page_size from param_dict,
+         * if not specified default value will be GF_SQL_DEFAULT_PAGE_SIZE*/
+        temp_str = NULL;
+        GET_DB_PARAM_FROM_DICT_DEFAULT(GFDB_STR_SQLITE3, param_dict,
+                        GFDB_SQL_PARAM_PAGE_SIZE, temp_str,
+                        GF_SQL_DEFAULT_PAGE_SIZE);
+        sql_conn->page_size = atoi(temp_str);
+        /*Apply page_size on the sqlite db*/
+        GF_SQLITE3_SET_PRAGMA(sqlite3_config_str, "page_size", "%ld",
+                        sql_conn->page_size, ret, out);
+
+
+
+        /*Extract sql cache size from param_dict,
+         * if not specified default value will be
+         * GF_SQL_DEFAULT_CACHE_SIZE pages*/
+        temp_str = NULL;
+        GET_DB_PARAM_FROM_DICT_DEFAULT(GFDB_STR_SQLITE3, param_dict,
+                        GFDB_SQL_PARAM_CACHE_SIZE, temp_str,
+                        GF_SQL_DEFAULT_CACHE_SIZE);
+        sql_conn->cache_size = atoi(temp_str);
+        /*Apply cache size on the sqlite db*/
+        GF_SQLITE3_SET_PRAGMA(sqlite3_config_str, "cache_size", "%ld",
+                        sql_conn->cache_size, ret, out);
+
+
+
+
+        /*Extract sql journal mode from param_dict,
+         * if not specified default value will be
+         * GF_SQL_DEFAULT_JOURNAL_MODE i.e "wal"*/
+        temp_str = NULL;
+        GET_DB_PARAM_FROM_DICT_DEFAULT(GFDB_STR_SQLITE3, param_dict,
+                        GFDB_SQL_PARAM_JOURNAL_MODE, temp_str,
+                        GF_SQL_DEFAULT_JOURNAL_MODE);
+        sql_conn->journal_mode = gf_sql_str2jm (temp_str);
+        /*Apply journal mode to the sqlite db*/
+        GF_SQLITE3_SET_PRAGMA(sqlite3_config_str, "journal_mode", "%s",
+                        temp_str, ret, out);
+
+
+
+        /*Only when the journal mode is WAL, wal_autocheckpoint makes sense*/
+        if (sql_conn->journal_mode == gf_sql_jm_wal) {
+                /*Extract sql wal auto check point from param_dict
+                * if not specified default value will be
+                * GF_SQL_DEFAULT_WAL_AUTOCHECKPOINT pages*/
+                temp_str = NULL;
+                GET_DB_PARAM_FROM_DICT_DEFAULT(GFDB_STR_SQLITE3, param_dict,
+                        GFDB_SQL_PARAM_WAL_AUTOCHECK, temp_str,
+                        GF_SQL_DEFAULT_WAL_AUTOCHECKPOINT);
+                sql_conn->wal_autocheckpoint = atoi(temp_str);
+                /*Apply wal auto check point to the sqlite db*/
+                GF_SQLITE3_SET_PRAGMA(sqlite3_config_str, "wal_autocheckpoint",
+                        "%ld", sql_conn->wal_autocheckpoint, ret, out);
+        }
+
+
+
+        /*Extract sql synchronous from param_dict
+         * if not specified default value will be GF_SQL_DEFAULT_SYNC*/
+         temp_str = NULL;
+        GET_DB_PARAM_FROM_DICT_DEFAULT(GFDB_STR_SQLITE3, param_dict,
+                        GFDB_SQL_PARAM_SYNC, temp_str, GF_SQL_DEFAULT_SYNC);
+        sql_conn->synchronous = gf_sql_str2sync_t (temp_str);
+        /*Apply synchronous to the sqlite db*/
+        GF_SQLITE3_SET_PRAGMA(sqlite3_config_str, "synchronous", "%d",
+                        sql_conn->synchronous, ret, out);
+
+
+
+        /*Extract sql auto_vacuum from param_dict
+         * if not specified default value will be GF_SQL_DEFAULT_AUTO_VACUUM*/
+         temp_str = NULL;
+        GET_DB_PARAM_FROM_DICT_DEFAULT(GFDB_STR_SQLITE3, param_dict,
+                        GFDB_SQL_PARAM_AUTO_VACUUM, temp_str,
+                        GF_SQL_DEFAULT_AUTO_VACUUM);
+        sql_conn->auto_vacuum = gf_sql_str2av_t (temp_str);
+        /*Apply auto_vacuum to the sqlite db*/
+        GF_SQLITE3_SET_PRAGMA(sqlite3_config_str, "auto_vacuum", "%d",
+                        sql_conn->auto_vacuum, ret, out);
+
+        ret = 0;
+out:
+        return ret;
+}
+
+
+
+int
+gf_sqlite3_init (dict_t *args, void **db_conn) {
+        int ret                         = -1;
+        gf_sql_connection_t *sql_conn   = NULL;
+        struct stat stbuf               = {0,};
+        gf_boolean_t    is_dbfile_exist = _gf_false;
+        char *temp_str                  = NULL;
+
+        GF_ASSERT (args);
+        GF_ASSERT (db_conn);
+
+        if (*db_conn != NULL) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "DB Connection is not empty!");
+                return 0;
+        }
+
+        if (!sqlite3_threadsafe()) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "sqlite3 is not in multithreaded mode");
+                goto out;
+        }
+
+        sql_conn = gf_sql_connection_init ();
+        if (!sql_conn) {
+                goto out;
+        }
+
+        /*Extract sql db path from args*/
+        temp_str = NULL;
+        GET_DB_PARAM_FROM_DICT(GFDB_STR_SQLITE3, args,
+                        GFDB_SQL_PARAM_DBPATH, temp_str, out);
+        strncpy(sql_conn->sqlite3_db_path, temp_str, PATH_MAX);
+
+        is_dbfile_exist = (stat (sql_conn->sqlite3_db_path, &stbuf) == 0) ?
+                                                _gf_true : _gf_false;
+
+        /*Creates DB if not created*/
+        sql_conn->sqlite3_db_conn = gf_open_sqlite3_conn (
+                                        sql_conn->sqlite3_db_path,
+                                        SQLITE_OPEN_READWRITE |
+                                        SQLITE_OPEN_CREATE);
+        if (!sql_conn->sqlite3_db_conn) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed creating db connection");
+                goto out;
+        }
+
+        /* If the file exist we skip the config part
+         * and create the schema if NOT Present*/
+        if (is_dbfile_exist)
+                goto create_table;
+
+
+        /*Apply sqlite3 params to database*/
+        ret = apply_sql_params_db(sql_conn, args);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed applying sql params to %s",
+                        sql_conn->sqlite3_db_path);
+                goto out;
+        }
+
+/*Create the schema if NOT present*/
+create_table:
+
+        ret = create_filetable (sql_conn->sqlite3_db_conn);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed Creating %s Table", GF_FILE_TABLE);
+               goto out;
+        }
+        ret = 0;
+
+out:
+        if (ret) {
+                gf_sqlite3_fini ((void **)&sql_conn);
+        }
+
+        *db_conn = sql_conn;
+
+        return ret;
+}
+
+
+int
+gf_sqlite3_fini (void **db_conn)
+{
+        int ret = -1;
+        gf_sql_connection_t *sql_conn = NULL;
+
+        GF_ASSERT (db_conn);
+        sql_conn = *db_conn;
+
+        if (sql_conn) {
+                if (sql_conn->sqlite3_db_conn) {
+                        ret = gf_close_sqlite3_conn(sql_conn->sqlite3_db_conn);
+                        if (ret) {
+                                /*Logging of error done in
+                                 * gf_close_sqlite3_conn()*/
+                                goto out;
+                        }
+                        sql_conn->sqlite3_db_conn = NULL;
+                }
+                gf_sql_connection_fini (&sql_conn);
+        }
+        *db_conn = sql_conn;
+        ret = 0;
+out:
+        return ret;
+}
+
+/******************************************************************************
+ *
+ *                      INSERT/UPDATE/DELETE Operations
+ *
+ *
+ * ***************************************************************************/
+
+int gf_sqlite3_insert(void *db_conn, gfdb_db_record_t *gfdb_db_record)
+{
+        int ret                         =       -1;
+        gf_sql_connection_t *sql_conn   =       db_conn;
+
+        CHECK_SQL_CONN(sql_conn, out);
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, gfdb_db_record, out);
+
+
+        /*This is for debugging bug. Will be removed with a bug fix*/
+        if ((GFDB_FOP_WIND == gfdb_db_record->gfdb_fop_path) &&
+            (strncmp(gfdb_db_record->file_path, "<gfid", 5) == 0)) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Skip path <gfid fop=%d",
+                        gfdb_db_record->gfdb_fop_type);
+                goto out;
+        }
+
+        switch (gfdb_db_record->gfdb_fop_path) {
+        case GFDB_FOP_WIND:
+                ret = gf_sql_insert_wind(sql_conn, gfdb_db_record);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed wind insert");
+                        goto out;
+                }
+                break;
+        case GFDB_FOP_UNWIND:
+                ret = gf_sql_insert_unwind (sql_conn, gfdb_db_record);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed unwind insert");
+                        goto out;
+                }
+                break;
+
+        case GFDB_FOP_WDEL:
+                ret = gf_sql_update_delete_wind(sql_conn, gfdb_db_record);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed updating delete during wind");
+                        goto out;
+                }
+                break;
+        case GFDB_FOP_UNDEL:
+        case GFDB_FOP_UNDEL_ALL:
+                ret = gf_sql_delete_unwind(sql_conn, gfdb_db_record);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed deleting");
+                        goto out;
+                }
+                break;
+        case GFDB_FOP_INVALID:
+        default:
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Cannot record to DB: Invalid FOP");
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+int
+gf_sqlite3_delete(void *db_conn, gfdb_db_record_t *gfdb_db_record)
+{
+        int ret = -1;
+        gf_sql_connection_t *sql_conn = db_conn;
+
+        CHECK_SQL_CONN(sql_conn, out);
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, gfdb_db_record, out);
+
+        ret = 0;
+out:
+        return ret;
+}
+
+/******************************************************************************
+ *
+ *                      SELECT QUERY FUNCTIONS
+ *
+ *
+ * ***************************************************************************/
+
+/*
+ * Find All files recorded in the DB
+ * Input:
+ *      query_callback  :       query callback fuction to handle
+ *                              result records from the query
+ * */
+int
+gf_sqlite3_find_all (void *db_conn, gf_query_callback_t query_callback,
+                        void *query_cbk_args)
+{
+        int ret                                 =       -1;
+        char *query_str                         =       NULL;
+        gf_sql_connection_t *sql_conn           =       db_conn;
+        sqlite3_stmt *prep_stmt                 =       NULL;
+
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, query_callback, out);
+
+        query_str = "select GF_FILE_TB.GF_ID,"
+                   " (select group_concat( GF_PID || ',' || FNAME || ','"
+                   " || FPATH || ',' || W_DEL_FLAG ||',' || LINK_UPDATE , '::')"
+                   " from GF_FLINK_TB where "
+                   "GF_FILE_TB.GF_ID = GF_FLINK_TB.GF_ID) from GF_FILE_TB ;";
+
+
+        ret = sqlite3_prepare(sql_conn->sqlite3_db_conn, query_str, -1,
+                                &prep_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing statment %s : %s", query_str,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        ret = gf_sql_query_function(prep_stmt, query_callback, query_cbk_args);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed Query %s", query_str);
+                goto out;
+        }
+
+        ret = 0;
+out:
+        sqlite3_finalize(prep_stmt);
+        return ret;
+}
+
+
+/*
+ * Find recently changed files from the DB
+ * Input:
+ *      query_callback  :       query callback fuction to handle
+ *                              result records from the query
+ *      from_time       :       Time to define what is recent
+ * */
+int
+gf_sqlite3_find_recently_changed_files(void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *query_cbk_args,
+                                        gfdb_time_t *from_time)
+{
+        int ret                                 =       -1;
+        char *query_str                         =       NULL;
+        gf_sql_connection_t *sql_conn           =       db_conn;
+        sqlite3_stmt *prep_stmt                 =       NULL;
+        long int  from_time_usec                =       0;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, query_callback, out);
+
+        query_str = "select GF_FILE_TB.GF_ID,"
+                " (select group_concat( GF_PID || ',' || FNAME || ','"
+                " || FPATH || ',' || W_DEL_FLAG ||',' || LINK_UPDATE , '::')"
+                " from GF_FLINK_TB where GF_FILE_TB.GF_ID = GF_FLINK_TB.GF_ID)"
+                "  from GF_FILE_TB where "
+                /*First condition: For writes*/
+                "((" GF_COL_TB_WSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_WMSEC ") >= ? )"
+                " OR "
+                /*Second condition: For reads*/
+                "((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_RWMSEC ") >= ?)";
+
+        from_time_usec = gfdb_time_2_usec(from_time);
+
+        ret = sqlite3_prepare(sql_conn->sqlite3_db_conn, query_str, -1,
+                                &prep_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing statment %s : %s", query_str,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind write wind time*/
+        ret = sqlite3_bind_int64(prep_stmt, 1, from_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding from_time_usec %ld : %s",
+                        from_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind read wind time*/
+        ret = sqlite3_bind_int64(prep_stmt, 2, from_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding from_time_usec %ld : %s ",
+                        from_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the query*/
+        ret = gf_sql_query_function(prep_stmt, query_callback, query_cbk_args);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed Query %s", query_str);
+                goto out;
+        }
+
+        ret = 0;
+out:
+        sqlite3_finalize(prep_stmt);
+        return ret;
+}
+
+
+/*
+ * Find unchanged files from a specified time from the DB
+ * Input:
+ *      query_callback  :       query callback fuction to handle
+ *                              result records from the query
+ *      for_time        :        Time from where the file/s are not changed
+ * */
+int
+gf_sqlite3_find_unchanged_for_time (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *query_cbk_args,
+                                        gfdb_time_t *for_time)
+{
+        int ret                                 =       -1;
+        char *query_str                         =       NULL;
+        gf_sql_connection_t *sql_conn           =       db_conn;
+        sqlite3_stmt *prep_stmt                 =       NULL;
+        long int  for_time_usec                =       0;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, query_callback, out);
+
+        query_str = "select GF_FILE_TB.GF_ID,"
+                " (select group_concat( GF_PID || ',' || FNAME || ','"
+                " || FPATH || ',' || W_DEL_FLAG ||',' || LINK_UPDATE , '::')"
+                " from GF_FLINK_TB where GF_FILE_TB.GF_ID = GF_FLINK_TB.GF_ID)"
+                "  from GF_FILE_TB where "
+                /*First condition: For writes*/
+                "((" GF_COL_TB_WSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_WMSEC ") <= ? )"
+                " OR "
+                /*Second condition: For reads*/
+                "((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_RWMSEC ") <= ?)";
+
+        for_time_usec = gfdb_time_2_usec(for_time);
+
+        ret = sqlite3_prepare(sql_conn->sqlite3_db_conn, query_str, -1,
+                                &prep_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing statment %s : %s", query_str,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind write wind time*/
+        ret = sqlite3_bind_int64(prep_stmt, 1, for_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding for_time_usec %ld : %s", for_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind read wind time*/
+        ret = sqlite3_bind_int64(prep_stmt, 2, for_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding for_time_usec %ld : %s", for_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the query*/
+        ret = gf_sql_query_function(prep_stmt, query_callback, query_cbk_args);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed Query %s", query_str);
+                goto out;
+        }
+
+        ret = 0;
+out:
+        sqlite3_finalize(prep_stmt);
+        return ret;
+}
+
+
+
+
+
+/*
+ * Find recently changed files with a specific frequency from the DB
+ * Input:
+ *      db_conn         :       db connection object
+ *      query_callback  :       query callback fuction to handle
+ *                              result records from the query
+ *      from_time       :       Time to define what is recent
+ *      freq_write_cnt  :       Frequency thresold for write
+ *      freq_read_cnt   :       Frequency thresold for read
+ *      clear_counters  :       Clear counters (r/w) for all inodes in DB
+ * */
+int
+gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *query_cbk_args,
+                                        gfdb_time_t *from_time,
+                                        int freq_write_cnt,
+                                        int freq_read_cnt,
+                                        gf_boolean_t clear_counters)
+{
+        int ret                                 =       -1;
+        char *query_str                         =       NULL;
+        gf_sql_connection_t *sql_conn           =       db_conn;
+        sqlite3_stmt *prep_stmt                 =       NULL;
+        long int  from_time_usec                =       0;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, query_callback, out);
+
+        query_str = "select GF_FILE_TB.GF_ID,"
+                " (select group_concat( GF_PID || ',' || FNAME || ','"
+                " || FPATH || ',' || W_DEL_FLAG ||',' || LINK_UPDATE , '::')"
+                " from GF_FLINK_TB where GF_FILE_TB.GF_ID = GF_FLINK_TB.GF_ID)"
+                "  from GF_FILE_TB where "
+                /*First condition: For Writes*/
+                "( ((" GF_COL_TB_WSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_WMSEC ") >= ? )"
+                " AND "" (" GF_COL_TB_WFC " > = ? ) )"
+                " OR "
+                /*Second condition: For Reads */
+                "( ((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_RWMSEC ") <= ?)"
+                " AND "" (" GF_COL_TB_RFC " > = ? ) )";
+
+        from_time_usec = gfdb_time_2_usec(from_time);
+
+        ret = sqlite3_prepare(sql_conn->sqlite3_db_conn, query_str, -1,
+                                &prep_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing statment %s : %s", query_str,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind write wind time*/
+        ret = sqlite3_bind_int64(prep_stmt, 1, from_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding from_time_usec %ld : %s",
+                        from_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind write frequency thresold*/
+        ret = sqlite3_bind_int(prep_stmt, 2, freq_write_cnt);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding freq_write_cnt %d : %s", freq_write_cnt,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+        /*Bind read wind time*/
+        ret = sqlite3_bind_int64(prep_stmt, 3, from_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding from_time_usec %ld : %s",
+                        from_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind read frequency thresold*/
+        ret = sqlite3_bind_int(prep_stmt, 4, freq_read_cnt);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding freq_read_cnt %d : %s", freq_read_cnt,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the query*/
+        ret = gf_sql_query_function(prep_stmt, query_callback, query_cbk_args);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed Query %s", query_str);
+                goto out;
+        }
+
+
+
+        /*Clear counters*/
+        if (clear_counters) {
+                ret = gf_sql_clear_counters(sql_conn);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed clearing counters!");
+                        goto out;
+                }
+        }
+        ret = 0;
+out:
+        sqlite3_finalize (prep_stmt);
+        return ret;
+}
+
+
+
+
+/*
+ * Find unchanged files from a specified time, w.r.t to frequency, from the DB
+ * Input:
+ *      query_callback  :       query callback fuction to handle
+ *                              result records from the query
+ *      for_time       :        Time from where the file/s are not changed
+ *      freq_write_cnt  :       Frequency thresold for write
+ *      freq_read_cnt   :       Frequency thresold for read
+ *      clear_counters  :       Clear counters (r/w) for all inodes in DB
+ * */
+int
+gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *query_cbk_args,
+                                        gfdb_time_t *for_time,
+                                        int freq_write_cnt,
+                                        int freq_read_cnt,
+                                        gf_boolean_t clear_counters)
+{
+        int ret                                 =       -1;
+        char *query_str                         =       NULL;
+        gf_sql_connection_t *sql_conn           =       db_conn;
+        sqlite3_stmt *prep_stmt                 =       NULL;
+        long int  for_time_usec                 =       0;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, query_callback, out);
+
+        query_str = "select GF_FILE_TB.GF_ID,"
+                " (select group_concat( GF_PID || ',' || FNAME || ','"
+                " || FPATH || ',' || W_DEL_FLAG ||',' || LINK_UPDATE , '::')"
+                " from GF_FLINK_TB where GF_FILE_TB.GF_ID = GF_FLINK_TB.GF_ID)"
+                "  from GF_FILE_TB where "
+                /*First condition: For Writes
+                 * Files that have write wind time smaller than for_time
+                 * OR
+                 * File that have write wind time greater than for_time,
+                 * but write_frequency less than freq_write_cnt*/
+                "( ((" GF_COL_TB_WSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_WMSEC ") < ? )"
+                " OR "
+                "( (" GF_COL_TB_WFC " < ? ) AND"
+                "((" GF_COL_TB_WSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_WMSEC ") >= ? ) ) )"
+                " OR "
+                /*Second condition: For Reads
+                 * Files that have read wind time smaller than for_time
+                 * OR
+                 * File that have read wind time greater than for_time,
+                 * but write_frequency less than freq_write_cnt*/
+                "( ((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_RWMSEC ") < ? )"
+                " OR "
+                "( (" GF_COL_TB_RFC " < ? ) AND"
+                "((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
+                GF_COL_TB_RWMSEC ") >= ? ) ) )";
+
+
+        for_time_usec = gfdb_time_2_usec(for_time);
+
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, query_str, -1,
+                                &prep_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing delete statment %s : %s", query_str,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind write wind time*/
+        ret = sqlite3_bind_int64 (prep_stmt, 1, for_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding for_time_usec %ld : %s",
+                        for_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind write frequency thresold*/
+        ret = sqlite3_bind_int (prep_stmt, 2, freq_write_cnt);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding freq_write_cnt %d : %s",
+                        freq_write_cnt,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind write wind time*/
+        ret = sqlite3_bind_int64 (prep_stmt, 3, for_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding for_time_usec %ld : %s",
+                        for_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+
+        /*Bind read wind time*/
+        ret = sqlite3_bind_int64 (prep_stmt, 4, for_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding for_time_usec %ld : %s",
+                        for_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind read frequency thresold*/
+        ret = sqlite3_bind_int (prep_stmt, 5, freq_read_cnt);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding freq_read_cnt %d : %s",
+                        freq_read_cnt,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind read wind time*/
+        ret = sqlite3_bind_int64 (prep_stmt, 6, for_time_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding for_time_usec %ld : %s",
+                        for_time_usec,
+                        sqlite3_errmsg(sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the query*/
+        ret = gf_sql_query_function (prep_stmt, query_callback, query_cbk_args);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed Query %s", query_str);
+                goto out;
+        }
+
+
+        /*Clear counters*/
+        if (clear_counters) {
+                ret = gf_sql_clear_counters (sql_conn);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed clearing counters!");
+                        goto out;
+                }
+        }
+
+        ret = 0;
+out:
+        sqlite3_finalize(prep_stmt);
+        return ret;
+}
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3.h b/libglusterfs/src/gfdb/gfdb_sqlite3.h
new file mode 100644
index 00000000000..b0c4b2b5fdd
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3.h
@@ -0,0 +1,288 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#ifndef __GFDB_SQLITE3_H
+#define __GFDB_SQLITE3_H
+
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+/*Sqlite3 header file*/
+#include <sqlite3.h>
+
+#include "logging.h"
+#include "gfdb_data_store_types.h"
+#include "gfdb_mem-types.h"
+
+#define GF_STMT_SIZE_MAX        2048
+
+#define GF_DB_NAME              "gfdb.db"
+#define GF_FILE_TABLE           "GF_FILE_TB"
+#define GF_FILE_LINK_TABLE      "GF_FLINK_TB"
+#define GF_MASTER_TABLE         "sqlite_master"
+
+/*Since we have multiple tables to be created we put it in a transaction*/
+#define GF_CREATE_STMT(out_str)\
+do {\
+        sprintf (out_str , "BEGIN; CREATE TABLE IF NOT EXISTS "\
+                GF_FILE_TABLE\
+                "(GF_ID TEXT PRIMARY KEY NOT NULL, "\
+                "W_SEC INTEGER NOT NULL DEFAULT 0, "\
+                "W_MSEC INTEGER NOT NULL DEFAULT 0, "\
+                "UW_SEC INTEGER NOT NULL DEFAULT 0, "\
+                "UW_MSEC INTEGER NOT NULL DEFAULT 0, "\
+                "W_READ_SEC INTEGER NOT NULL DEFAULT 0, "\
+                "W_READ_MSEC INTEGER NOT NULL DEFAULT 0, "\
+                "UW_READ_SEC INTEGER NOT NULL DEFAULT 0, "\
+                "UW_READ_MSEC INTEGER NOT NULL DEFAULT 0, "\
+                "WRITE_FREQ_CNTR INTEGER NOT NULL DEFAULT 1, "\
+                "READ_FREQ_CNTR INTEGER NOT NULL DEFAULT 1); "\
+                "CREATE TABLE IF NOT EXISTS "\
+                GF_FILE_LINK_TABLE\
+                "(GF_ID TEXT NOT NULL, "\
+                "GF_PID TEXT NOT NULL, "\
+                "FNAME TEXT NOT NULL, "\
+                "FPATH TEXT NOT NULL, "\
+                "W_DEL_FLAG INTEGER NOT NULL DEFAULT 0, "\
+                "LINK_UPDATE INTEGER NOT NULL DEFAULT 0, "\
+                "PRIMARY KEY ( GF_ID, GF_PID, FNAME) "\
+                ");"\
+                "COMMIT;"\
+                );;\
+} while (0)
+
+
+#define GF_COL_TB_WSEC          GF_FILE_TABLE "." GF_COL_WSEC
+#define GF_COL_TB_WMSEC         GF_FILE_TABLE "." GF_COL_WMSEC
+#define GF_COL_TB_UWSEC         GF_FILE_TABLE "." GF_COL_UWSEC
+#define GF_COL_TB_UWMSEC        GF_FILE_TABLE "." GF_COL_UWMSEC
+#define GF_COL_TB_RWSEC         GF_FILE_TABLE "." GF_COL_WSEC_READ
+#define GF_COL_TB_RWMSEC        GF_FILE_TABLE "." GF_COL_WMSEC_READ
+#define GF_COL_TB_RUWSEC        GF_FILE_TABLE "." GF_COL_UWSEC_READ
+#define GF_COL_TB_RUWMSEC       GF_FILE_TABLE "." GF_COL_UWMSEC_READ
+#define GF_COL_TB_WFC           GF_FILE_TABLE "." GF_COL_WRITE_FREQ_CNTR
+#define GF_COL_TB_RFC           GF_FILE_TABLE "." GF_COL_READ_FREQ_CNTR
+
+
+/*******************************************************************************
+*                      SQLITE3 Connection details and PRAGMA
+* ****************************************************************************/
+
+#define GF_SQL_AV_NONE  "none"
+#define GF_SQL_AV_FULL  "full"
+#define GF_SQL_AV_INCR  "incr"
+
+
+#define GF_SQL_SYNC_OFF         "off"
+#define GF_SQL_SYNC_NORMAL      "normal"
+#define GF_SQL_SYNC_FULL        "full"
+
+#define GF_SQL_JM_DELETE        "delete"
+#define GF_SQL_JM_TRUNCATE      "truncate"
+#define GF_SQL_JM_PERSIST       "persist"
+#define GF_SQL_JM_MEMORY        "memory"
+#define GF_SQL_JM_WAL           "wal"
+#define GF_SQL_JM_OFF           "off"
+
+
+typedef enum gf_sql_auto_vacuum {
+        gf_sql_av_none = 0,
+        gf_sql_av_full,
+        gf_sql_av_incr,
+        gf_sql_av_invalid
+} gf_sql_auto_vacuum_t;
+
+typedef enum gf_sql_sync {
+        gf_sql_sync_off = 0,
+        gf_sql_sync_normal,
+        gf_sql_sync_full,
+        gf_sql_sync_invalid
+} gf_sql_sync_t;
+
+
+typedef enum gf_sql_journal_mode {
+        gf_sql_jm_wal = 0,
+        gf_sql_jm_delete,
+        gf_sql_jm_truncate,
+        gf_sql_jm_persist,
+        gf_sql_jm_memory,
+        gf_sql_jm_off,
+        gf_sql_jm_invalid
+} gf_sql_journal_mode_t;
+
+
+typedef struct gf_sql_connection {
+        char                    sqlite3_db_path[PATH_MAX];
+        sqlite3                 *sqlite3_db_conn;
+        ssize_t                 cache_size;
+        ssize_t                 page_size;
+        ssize_t                 wal_autocheckpoint;
+        gf_sql_journal_mode_t   journal_mode;
+        gf_sql_sync_t           synchronous;
+        gf_sql_auto_vacuum_t    auto_vacuum;
+} gf_sql_connection_t;
+
+
+
+#define CHECK_SQL_CONN(sql_conn, out)\
+do {\
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, sql_conn, out);\
+        if (!sql_conn->sqlite3_db_conn) {\
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,\
+                        "sqlite3 connection not initialized");\
+                goto out;\
+        };\
+} while (0)
+
+
+#define GF_SQLITE3_SET_PRAGMA(sqlite3_config_str, param_key, format, value,\
+                        ret, error)\
+do {\
+        sprintf(sqlite3_config_str, "PRAGMA " param_key " = " format , value);\
+        ret = sqlite3_exec (sql_conn->sqlite3_db_conn, sqlite3_config_str,\
+                NULL, NULL, NULL);\
+        if (ret != SQLITE_OK) {\
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,\
+                        "Failed executing: %s : %s",\
+                        sqlite3_config_str, sqlite3_errmsg\
+                                               (sql_conn->sqlite3_db_conn));\
+                        ret = -1;\
+                        goto error;\
+        };\
+} while (0)
+
+/************************SQLITE3 PARAMS KEYS***********************************/
+#define GFDB_SQL_PARAM_DBPATH           "sql-db-path"
+#define GFDB_SQL_PARAM_CACHE_SIZE       "sql-db-cachesize"
+#define GFDB_SQL_PARAM_PAGE_SIZE        "sql-db-pagesize"
+#define GFDB_SQL_PARAM_JOURNAL_MODE     "sql-db-journalmode"
+#define GFDB_SQL_PARAM_WAL_AUTOCHECK    "sql-db-wal-autocheckpoint"
+#define GFDB_SQL_PARAM_SYNC             "sql-db-sync"
+#define GFDB_SQL_PARAM_AUTO_VACUUM      "sql-db-autovacuum"
+
+#define GF_SQL_DEFAULT_DBPATH                   ""
+#define GF_SQL_DEFAULT_PAGE_SIZE                "4096"
+#define GF_SQL_DEFAULT_CACHE_SIZE               "1000"
+#define GF_SQL_DEFAULT_WAL_AUTOCHECKPOINT       "1000"
+#define GF_SQL_DEFAULT_JOURNAL_MODE             GF_SQL_JM_WAL
+#define GF_SQL_DEFAULT_SYNC                     GF_SQL_SYNC_NORMAL
+#define GF_SQL_DEFAULT_AUTO_VACUUM              GF_SQL_AV_NONE
+
+
+/* Defines the indexs for sqlite params
+ * The order should be maintained*/
+typedef enum sqlite_param_index {
+        sql_dbpath_ix = 0,
+        sql_pagesize_ix,
+        sql_cachesize_ix,
+        sql_journalmode_ix,
+        sql_walautocheck_ix,
+        sql_dbsync_ix,
+        sql_autovacuum_ix,
+        /*This should be in the end*/
+        sql_index_max
+} sqlite_param_index_t;
+
+/* Array to hold the sqlite param keys
+ * The order should be maintained as sqlite_param_index_t*/
+static char *sqlite_params_keys[] = {
+        GFDB_SQL_PARAM_DBPATH,
+        GFDB_SQL_PARAM_PAGE_SIZE,
+        GFDB_SQL_PARAM_CACHE_SIZE,
+        GFDB_SQL_PARAM_JOURNAL_MODE,
+        GFDB_SQL_PARAM_WAL_AUTOCHECK,
+        GFDB_SQL_PARAM_SYNC,
+        GFDB_SQL_PARAM_AUTO_VACUUM
+};
+
+
+/* Array of default values for sqlite params
+ * The order should be maintained as sqlite_param_index_t*/
+static char *sqlite_params_default_value[] = {
+        GF_SQL_DEFAULT_DBPATH,
+        GF_SQL_DEFAULT_PAGE_SIZE,
+        GF_SQL_DEFAULT_CACHE_SIZE,
+        GF_SQL_DEFAULT_JOURNAL_MODE,
+        GF_SQL_DEFAULT_WAL_AUTOCHECKPOINT,
+        GF_SQL_DEFAULT_SYNC,
+        GF_SQL_DEFAULT_AUTO_VACUUM
+};
+
+/*Extract sql params from page_size to auto_vacumm
+ * The dbpath is extracted in a different way*/
+static inline int
+gfdb_set_sql_params(char *comp_name, dict_t *from_dict, dict_t *to_dict)
+{
+        sqlite_param_index_t sql_index  = sql_pagesize_ix;
+        char *_val_str                  = NULL;
+        int ret                      = -1;
+
+        GF_ASSERT (comp_name);
+        GF_ASSERT (from_dict);
+        GF_ASSERT (to_dict);
+
+        /*Extact and Set of the sql params from page_size*/
+        for (sql_index = sql_pagesize_ix; sql_index < sql_index_max;
+                sql_index++) {
+                _val_str = NULL;
+                GET_DB_PARAM_FROM_DICT_DEFAULT (comp_name, from_dict,
+                        sqlite_params_keys[sql_index], _val_str,
+                        sqlite_params_default_value[sql_index]);
+                SET_DB_PARAM_TO_DICT (comp_name, to_dict,
+                        sqlite_params_keys[sql_index], _val_str, ret, out);
+        }
+out:
+        return ret;
+}
+
+
+
+
+/*************************SQLITE3 GFDB PLUGINS*********************************/
+
+/*Db init and fini modules*/
+int gf_sqlite3_fini (void **db_conn);
+int gf_sqlite3_init (dict_t *args, void **db_conn);
+
+/*insert/update/delete modules*/
+int gf_sqlite3_insert (void *db_conn, gfdb_db_record_t *);
+int gf_sqlite3_delete (void *db_conn, gfdb_db_record_t *);
+
+/*querying modules*/
+int gf_sqlite3_find_all (void *db_conn, gf_query_callback_t,
+                        void *_query_cbk_args);
+int gf_sqlite3_find_unchanged_for_time (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_query_cbk_args,
+                                        gfdb_time_t *for_time);
+int gf_sqlite3_find_recently_changed_files (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_query_cbk_args,
+                                        gfdb_time_t *from_time);
+int gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_query_cbk_args,
+                                        gfdb_time_t *for_time,
+                                        int write_freq_cnt,
+                                        int read_freq_cnt,
+                                        gf_boolean_t clear_counters);
+int gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
+                                        gf_query_callback_t query_callback,
+                                        void *_query_cbk_args,
+                                        gfdb_time_t *from_time,
+                                        int write_freq_cnt,
+                                        int read_freq_cnt,
+                                        gf_boolean_t clear_counters);
+
+void gf_sqlite3_fill_db_operations (gfdb_db_operations_t  *gfdb_db_ops);
+
+#endif
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
new file mode 100644
index 00000000000..50504181d26
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
@@ -0,0 +1,1128 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include "gfdb_sqlite3_helper.h"
+
+/*****************************************************************************
+ *
+ *                 Helper function to execute actual sql queries
+ *
+ *
+ * ****************************************************************************/
+
+static inline int
+gf_sql_delete_all (gf_sql_connection_t *sql_conn,
+                  char              *gfid)
+{
+        int ret = -1;
+        sqlite3_stmt *delete_file_stmt = NULL;
+        sqlite3_stmt *delete_link_stmt = NULL;
+        char *delete_link_str = "DELETE FROM "
+                           GF_FILE_LINK_TABLE
+                           " WHERE GF_ID = ? ;";
+        char *delete_file_str = "DELETE FROM "
+                           GF_FILE_TABLE
+                           " WHERE GF_ID = ? ;";
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfid, out);
+
+        /*
+         * Delete all links associated with this GFID
+         *
+         * */
+        /*Prepare statement for delete all links*/
+        ret = sqlite3_prepare(sql_conn->sqlite3_db_conn, delete_link_str, -1,
+                                &delete_link_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing delete statment %s : %s",
+                        delete_link_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (delete_link_stmt, 1, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step (delete_link_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        delete_link_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+        /*
+         * Delete entry from file table associated with this GFID
+         *
+         * */
+        /*Prepare statement for delete all links*/
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, delete_file_str, -1,
+                                &delete_file_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing delete statment %s : %s",
+                        delete_file_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (delete_file_stmt, 1, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step (delete_file_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        delete_file_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+out:
+        /*Free prepared statement*/
+        sqlite3_finalize (delete_file_stmt);
+        sqlite3_finalize (delete_link_stmt);
+        return ret;
+}
+
+static inline int
+gf_sql_delete_link (gf_sql_connection_t  *sql_conn,
+                   char                 *gfid,
+                   char                 *pargfid,
+                   char                 *basename)
+{
+        int ret = -1;
+        sqlite3_stmt *delete_stmt = NULL;
+        char *delete_str = "DELETE FROM "
+                           GF_FILE_LINK_TABLE
+                           " WHERE GF_ID = ? AND GF_PID = ?"
+                           " AND FNAME = ?;";
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, pargfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, basename, out);
+
+        /*Prepare statement*/
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, delete_str, -1,
+                                &delete_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing delete statment %s : %s", delete_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (delete_stmt, 1, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind pargfid*/
+        ret = sqlite3_bind_text (delete_stmt, 2, pargfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent gfid %s : %s", pargfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind basename*/
+        ret = sqlite3_bind_text (delete_stmt, 3, basename, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding basename %s : %s", basename,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step(delete_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        delete_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+        ret = 0;
+out:
+        /*Free prepared statement*/
+        sqlite3_finalize (delete_stmt);
+        return ret;
+}
+
+
+
+static inline int
+gf_sql_update_link_flags (gf_sql_connection_t  *sql_conn,
+                   char                 *gfid,
+                   char                 *pargfid,
+                   char                 *basename,
+                   int                  update_flag,
+                   gf_boolean_t         is_update_or_delete)
+{
+        int ret = -1;
+        sqlite3_stmt *update_stmt = NULL;
+        char *update_column = NULL;
+        char update_str[1024] = "";
+
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, pargfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, basename, out);
+
+        update_column = (is_update_or_delete) ? "LINK_UPDATE" : "W_DEL_FLAG";
+
+        sprintf (update_str, "UPDATE "
+                            GF_FILE_LINK_TABLE
+                            " SET %s = ?"
+                            " WHERE GF_ID = ? AND GF_PID = ? AND FNAME = ?;",
+                            update_column);
+
+        /*Prepare statement*/
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, update_str, -1,
+                                &update_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing update statment %s : %s", update_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+        /*Bind link_update*/
+        ret = sqlite3_bind_int (update_stmt, 1, update_flag);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding update_flag %d : %s", update_flag,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (update_stmt, 2, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind pargfid*/
+        ret = sqlite3_bind_text (update_stmt, 3, pargfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent gfid %s : %s", pargfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind basename*/
+        ret = sqlite3_bind_text (update_stmt, 4, basename, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding basename %s : %s", basename,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step(update_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        update_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        /*Free prepared statement*/
+        sqlite3_finalize (update_stmt);
+        return ret;
+}
+
+
+static inline int
+gf_sql_insert_link (gf_sql_connection_t  *sql_conn,
+                   char                 *gfid,
+                   char                 *pargfid,
+                   char                 *basename,
+                   char                 *basepath)
+{
+        int ret = -1;
+        sqlite3_stmt *insert_stmt = NULL;
+        char *insert_str = "INSERT INTO "
+                           GF_FILE_LINK_TABLE
+                           " (GF_ID, GF_PID, FNAME, FPATH,"
+                           " W_DEL_FLAG, LINK_UPDATE) "
+                           " VALUES (?, ?, ?, ?, 0, 1);";
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, pargfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, basename, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, basepath, out);
+
+        /*Prepare statement*/
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, insert_str, -1,
+                                &insert_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing insert statment %s : %s", insert_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (insert_stmt, 1, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind pargfid*/
+        ret = sqlite3_bind_text (insert_stmt, 2, pargfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent gfid %s : %s", pargfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind basename*/
+        ret = sqlite3_bind_text (insert_stmt, 3, basename, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding basename %s : %s", basename,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind basepath*/
+        ret = sqlite3_bind_text (insert_stmt, 4, basepath, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding basepath %s : %s", basepath,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step (insert_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        insert_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        /*Free prepared statement*/
+        sqlite3_finalize (insert_stmt);
+        return ret;
+}
+
+
+static inline int
+gf_sql_update_link (gf_sql_connection_t  *sql_conn,
+                   char                 *gfid,
+                   char                 *pargfid,
+                   char                 *basename,
+                   char                 *basepath,
+                   char                 *old_pargfid,
+                   char                 *old_basename)
+{
+        int ret = -1;
+        sqlite3_stmt *insert_stmt = NULL;
+        char *insert_str =  "INSERT INTO "
+                            GF_FILE_LINK_TABLE
+                            " (GF_ID, GF_PID, FNAME, FPATH,"
+                            " W_DEL_FLAG, LINK_UPDATE) "
+                            " VALUES (? , ?, ?, ?, 0, 1);";
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, pargfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, basename, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, basepath, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, old_pargfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, old_basename, out);
+
+        /*
+         *
+         * Delete the old link
+         *
+         * */
+         ret = gf_sql_delete_link (sql_conn, gfid, old_pargfid,
+                                old_basename);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed deleting old link");
+                goto out;
+        }
+
+        /*
+         *
+         * insert new link
+         *
+         * */
+        /*Prepare statement*/
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, insert_str, -1,
+                                &insert_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing insert statment %s : %s", insert_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (insert_stmt, 1, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind new pargfid*/
+        ret = sqlite3_bind_text (insert_stmt, 2, pargfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent gfid %s : %s", pargfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind new basename*/
+        ret = sqlite3_bind_text (insert_stmt, 3, basename, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding basename %s : %s", basename,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind new basepath*/
+        ret = sqlite3_bind_text (insert_stmt, 4, basepath, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding basename %s : %s", basepath,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step (insert_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        insert_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+
+
+        ret = 0;
+out:
+        /*Free prepared statement*/
+        sqlite3_finalize (insert_stmt);
+        return ret;
+}
+
+static inline int
+gf_sql_insert_write_wind_time (gf_sql_connection_t  *sql_conn,
+                                char                 *gfid,
+                                gfdb_time_t          *wind_time)
+{
+        int ret = -1;
+        sqlite3_stmt *insert_stmt = NULL;
+        char *insert_str = "INSERT INTO "
+                           GF_FILE_TABLE
+                           "(GF_ID, W_SEC, W_MSEC, UW_SEC, UW_MSEC)"
+                           " VALUES (?, ?, ?, 0, 0);";
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, wind_time, out);
+
+
+        /*Prepare statement*/
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, insert_str, -1,
+                        &insert_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing insert statment %s : %s", insert_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (insert_stmt, 1, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind wind secs*/
+        ret = sqlite3_bind_int (insert_stmt, 2, wind_time->tv_sec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent wind secs %ld : %s",
+                        wind_time->tv_sec,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind wind msecs*/
+        ret = sqlite3_bind_int (insert_stmt, 3, wind_time->tv_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent wind msecs %ld : %s",
+                        wind_time->tv_usec,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step (insert_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        insert_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        /*Free prepared statement*/
+        sqlite3_finalize (insert_stmt);
+        return ret;
+}
+
+
+
+/*Update write/read times for both wind and unwind*/
+static inline int
+gf_update_time (gf_sql_connection_t    *sql_conn,
+                char                    *gfid,
+                gfdb_time_t             *update_time,
+                gf_boolean_t            record_counter,
+                gf_boolean_t            is_wind,
+                gf_boolean_t            is_read)
+{
+        int ret = -1;
+        sqlite3_stmt *update_stmt = NULL;
+        char update_str[1024] = "";
+        char *freq_cntr_str = NULL;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, update_time, out);
+
+        /*
+         * Constructing the prepare statment string.
+         *
+         * */
+        /*For write time*/
+        if (!is_read) {
+                if (is_wind) {
+                        /*if record counter is on*/
+                        freq_cntr_str = (record_counter) ?
+                        ", WRITE_FREQ_CNTR = WRITE_FREQ_CNTR + 1" : "";
+
+                        /*Prefectly safe as we will not go array of bound*/
+                        sprintf (update_str, "UPDATE "
+                                GF_FILE_TABLE
+                                " SET W_SEC = ?, W_MSEC = ? "
+                                " %s"/*place for read freq counters*/
+                                " WHERE GF_ID = ? ;", freq_cntr_str);
+                } else {
+                        /*Prefectly safe as we will not go array of bound*/
+                        sprintf (update_str, "UPDATE "
+                                GF_FILE_TABLE
+                                " SET UW_SEC = ?, UW_MSEC = ? ;");
+                }
+        }
+        /*For Read Time update*/
+        else {
+                if (is_wind) {
+                        /*if record counter is on*/
+                        freq_cntr_str = (record_counter) ?
+                        ", READ_FREQ_CNTR = READ_FREQ_CNTR + 1" : "";
+
+                        /*Prefectly safe as we will not go array of bound*/
+                        sprintf (update_str, "UPDATE "
+                                GF_FILE_TABLE
+                                " SET W_READ_SEC = ?, W_READ_MSEC = ? "
+                                " %s"/*place for read freq counters*/
+                                " WHERE GF_ID = ? ;", freq_cntr_str);
+                } else {
+                        /*Prefectly safe as we will not go array of bound*/
+                        sprintf (update_str, "UPDATE "
+                                GF_FILE_TABLE
+                                " SET UW_READ_SEC = ?, UW_READ_MSEC = ? ;");
+                }
+        }
+
+        /*Prepare statement*/
+        ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, update_str, -1,
+                                &update_stmt, 0);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed preparing insert statment %s : %s", update_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind time secs*/
+        ret = sqlite3_bind_int (update_stmt, 1, update_time->tv_sec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent wind secs %ld : %s",
+                        update_time->tv_sec,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind time msecs*/
+        ret = sqlite3_bind_int (update_stmt, 2, update_time->tv_usec);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding parent wind msecs %ld : %s",
+                        update_time->tv_usec,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Bind gfid*/
+        ret = sqlite3_bind_text (update_stmt, 3, gfid, -1, NULL);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed binding gfid %s : %s", gfid,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        /*Execute the prepare statement*/
+        if (sqlite3_step (update_stmt) != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing the prepared stmt %s : %s",
+                        update_str,
+                        sqlite3_errmsg (sql_conn->sqlite3_db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        /*Free prepared statement*/
+        sqlite3_finalize (update_stmt);
+        return ret;
+}
+
+/******************************************************************************
+ *
+ *                      Helper functions for gf_sqlite3_insert()
+ *
+ *
+ * ****************************************************************************/
+
+int
+gf_sql_insert_wind (gf_sql_connection_t  *sql_conn,
+                   gfdb_db_record_t     *gfdb_db_record)
+{
+        int ret                 = -1;
+        gfdb_time_t *modtime    = NULL;
+        char *pargfid_str       = NULL;
+        char *gfid_str          = NULL;
+        char *old_pargfid_str   = NULL;
+        gf_boolean_t its_wind   = _gf_true;/*remains true for this function*/
+
+
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfdb_db_record, out);
+
+
+        gfid_str = gf_strdup (uuid_utoa (gfdb_db_record->gfid));
+        if (!gfid_str) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Creating gfid string failed.");
+                goto out;
+        }
+
+        modtime = &gfdb_db_record->gfdb_wind_change_time;
+
+        /* handle all dentry based operations */
+        if (isdentryfop (gfdb_db_record->gfdb_fop_type)) {
+                /*Parent GFID is always set*/
+                pargfid_str = gf_strdup (uuid_utoa (gfdb_db_record->pargfid));
+                if (!pargfid_str) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Creating gfid string failed.");
+                        goto out;
+                }
+
+                /* handle create, mknod */
+                if (isdentrycreatefop (gfdb_db_record->gfdb_fop_type)) {
+                        /*insert link*/
+                        ret = gf_sql_insert_link(sql_conn,
+                                        gfid_str, pargfid_str,
+                                        gfdb_db_record->file_name,
+                                        gfdb_db_record->file_path);
+                        if (ret) {
+                                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed inserting link in DB");
+                                goto out;
+                        }
+                        gfdb_db_record->islinkupdate = _gf_true;
+
+                        /*
+                         * Only for create/mknod insert wind time
+                         * for the first time
+                         * */
+                        ret = gf_sql_insert_write_wind_time (sql_conn, gfid_str,
+                                                                modtime);
+                        if (ret) {
+                                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed inserting wind time in DB");
+                                goto out;
+                        }
+                        goto out;
+                }
+                /*handle rename, link */
+                else {
+                        /*rename*/
+                        if (strlen (gfdb_db_record->old_file_name) != 0) {
+                                old_pargfid_str = gf_strdup (uuid_utoa (
+                                                gfdb_db_record->old_pargfid));
+                                if (!old_pargfid_str) {
+                                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                                "Creating gfid string failed.");
+                                        goto out;
+                                }
+                                gf_sql_update_link (sql_conn, gfid_str,
+                                                pargfid_str,
+                                                gfdb_db_record->file_name,
+                                                gfdb_db_record->file_path,
+                                                old_pargfid_str,
+                                                gfdb_db_record->old_file_name);
+                                if (ret) {
+                                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                                "Failed updating link");
+                                        goto out;
+                                }
+                                gfdb_db_record->islinkupdate = _gf_true;
+                        }
+                        /*link*/
+                        else {
+                                ret = gf_sql_insert_link (sql_conn,
+                                        gfid_str, pargfid_str,
+                                        gfdb_db_record->file_name,
+                                        gfdb_db_record->file_path);
+                                if (ret) {
+                                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                        "Failed inserting link in DB");
+                                        goto out;
+                                }
+                                gfdb_db_record->islinkupdate = _gf_true;
+                        }
+                }
+        }
+
+        /*All fops update times read or write*/
+        ret = gf_update_time (sql_conn, gfid_str, modtime,
+                gfdb_db_record->do_record_counters,
+                its_wind,
+                isreadfop (gfdb_db_record->gfdb_fop_type));
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed update wind time in DB");
+                goto out;
+        }
+
+        ret = 0;
+out:
+        GF_FREE (gfid_str);
+        GF_FREE (pargfid_str);
+        GF_FREE (old_pargfid_str);
+        return ret;
+}
+
+
+
+
+int
+gf_sql_insert_unwind (gf_sql_connection_t  *sql_conn,
+                     gfdb_db_record_t     *gfdb_db_record)
+{
+
+        int ret = -1;
+        gfdb_time_t *modtime    = NULL;
+        gf_boolean_t its_wind   = _gf_true;/*remains true for this function*/
+        char *gfid_str = NULL;
+        char *pargfid_str = NULL;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfdb_db_record, out);
+
+        gfid_str = gf_strdup (uuid_utoa(gfdb_db_record->gfid));
+        if (!gfid_str) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Creating gfid string failed.");
+                goto out;
+        }
+
+        /*Only update if recording unwind is set*/
+        if (gfdb_db_record->do_record_uwind_time) {
+                modtime = &gfdb_db_record->gfdb_unwind_change_time;
+                ret = gf_update_time (sql_conn, gfid_str, modtime,
+                        gfdb_db_record->do_record_counters,
+                        (!its_wind),
+                        isreadfop (gfdb_db_record->gfdb_fop_type));
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed update unwind time in DB");
+                        goto out;
+                }
+        }
+
+        /*For link creation and changes we use link updated*/
+        if (gfdb_db_record->islinkupdate &&
+                isdentryfop(gfdb_db_record->gfdb_fop_type)) {
+
+                pargfid_str = gf_strdup(uuid_utoa(gfdb_db_record->pargfid));
+                if (!pargfid_str) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Creating pargfid_str string failed.");
+                        goto out;
+                }
+
+                ret = gf_sql_update_link_flags (sql_conn, gfid_str, pargfid_str,
+                                        gfdb_db_record->file_name, 0, _gf_true);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed updating link flags in unwind");
+                        goto out;
+                }
+        }
+
+        ret = 0;
+out:
+        GF_FREE (gfid_str);
+        GF_FREE (pargfid_str);
+        return ret;
+}
+
+
+int
+gf_sql_update_delete_wind (gf_sql_connection_t  *sql_conn,
+                          gfdb_db_record_t     *gfdb_db_record)
+{
+        int ret = -1;
+        gfdb_time_t *modtime    = NULL;
+        char *gfid_str          = NULL;
+        char *pargfid_str       = NULL;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfdb_db_record, out);
+
+        gfid_str = gf_strdup (uuid_utoa(gfdb_db_record->gfid));
+        if (!gfid_str) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Creating gfid string failed.");
+                goto out;
+        }
+
+        pargfid_str = gf_strdup (uuid_utoa(gfdb_db_record->pargfid));
+        if (!pargfid_str) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Creating pargfid_str string failed.");
+                        goto out;
+        }
+
+        /*Update the wind write times*/
+        modtime = &gfdb_db_record->gfdb_unwind_change_time;
+        ret = gf_update_time (sql_conn, gfid_str, modtime,
+                        gfdb_db_record->do_record_counters,
+                        _gf_true,
+                        isreadfop (gfdb_db_record->gfdb_fop_type));
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed update wind time in DB");
+                goto out;
+        }
+
+        ret = gf_sql_update_link_flags (sql_conn, gfid_str, pargfid_str,
+                                        gfdb_db_record->file_name, 1,
+                                        _gf_false);
+        if (ret) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed updating link flags in wind");
+                goto out;
+        }
+
+        ret = 0;
+out:
+        GF_FREE (gfid_str);
+        GF_FREE (pargfid_str);
+        return ret;
+}
+
+int
+gf_sql_delete_unwind (gf_sql_connection_t  *sql_conn,
+                          gfdb_db_record_t     *gfdb_db_record)
+{
+        int ret = -1;
+        char *gfid_str = NULL;
+        char *pargfid_str = NULL;
+        gfdb_time_t *modtime    = NULL;
+
+        CHECK_SQL_CONN (sql_conn, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, gfdb_db_record, out);
+
+        gfid_str = gf_strdup (uuid_utoa(gfdb_db_record->gfid));
+        if (!gfid_str) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Creating gfid string failed.");
+                goto out;
+        }
+
+        /*Nuke all the entries for this GFID from DB*/
+        if (gfdb_db_record->gfdb_fop_path == GFDB_FOP_UNDEL_ALL) {
+                gf_sql_delete_all(sql_conn, gfid_str);
+        }
+        /*Remove link entries only*/
+        else if (gfdb_db_record->gfdb_fop_path == GFDB_FOP_UNDEL) {
+
+                pargfid_str = gf_strdup(uuid_utoa(gfdb_db_record->pargfid));
+                if (!pargfid_str) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Creating pargfid_str string failed.");
+                        goto out;
+                }
+
+                modtime = &gfdb_db_record->gfdb_unwind_change_time;
+
+                ret = gf_sql_delete_link(sql_conn, gfid_str, pargfid_str,
+                                        gfdb_db_record->file_name);
+                if (ret) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed deleting link");
+                        goto out;
+                }
+
+                if (gfdb_db_record->do_record_uwind_time) {
+                        ret = gf_update_time (sql_conn, gfid_str, modtime,
+                                gfdb_db_record->do_record_counters,
+                                _gf_false,
+                                isreadfop(gfdb_db_record->gfdb_fop_type));
+                        if (ret) {
+                                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                "Failed update unwind time in DB");
+                                goto out;
+                        }
+                }
+        }
+        ret = 0;
+out:
+        GF_FREE (gfid_str);
+        GF_FREE (pargfid_str);
+        return ret;
+}
+
+/******************************************************************************
+ *
+ *                      Find/Query helper functions
+ *
+ * ****************************************************************************/
+int
+gf_sql_query_function (sqlite3_stmt              *prep_stmt,
+                      gf_query_callback_t       query_callback,
+                      void                      *_query_cbk_args)
+{
+        int ret = -1;
+        gfdb_query_record_t *gfdb_query_record = NULL;
+        char *text_column = NULL;
+        sqlite3 *db_conn = NULL;
+
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, prep_stmt, out);
+        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, query_callback, out);
+
+        db_conn = sqlite3_db_handle(prep_stmt);
+
+        gfdb_query_record = gfdb_query_record_init ();
+        if (!gfdb_query_record) {
+                        gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                        "Failed to create gfdb_query_record");
+                                goto out;
+        }
+
+        /*Loop to access queried rows*/
+        while ((ret = sqlite3_step (prep_stmt)) == SQLITE_ROW) {
+
+                /*Clear the query record*/
+                memset (gfdb_query_record, 0, sizeof(*gfdb_query_record));
+
+                if (sqlite3_column_count(prep_stmt) > 0) {
+
+                        /*Retriving GFID - column index is 0*/
+                        text_column = (char *)sqlite3_column_text
+                                                        (prep_stmt, 0);
+                        if (!text_column) {
+                                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                        "Failed retriving GF_ID");
+                                goto out;
+                        }
+                        ret = uuid_parse (text_column, gfdb_query_record->gfid);
+                        if (ret) {
+                                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                        "Failed parsing GF_ID");
+                                goto out;
+                        }
+
+                        /*Retrive Link Buffer - column index 1*/
+                        text_column = (char *)sqlite3_column_text
+                                                (prep_stmt, 1);
+                        /* Get link string. Do shallow copy here
+                         * query_callback function should do a
+                         * deep copy and then do operations on this field*/
+                        gfdb_query_record->_link_info_str = text_column;
+                        gfdb_query_record->link_info_size = strlen
+                                                                (text_column);
+
+                        /* Call the call back function provided*/
+                        ret = query_callback (gfdb_query_record,
+                                                        _query_cbk_args);
+                        if (ret) {
+                                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                        "Query Call back failed!");
+                                goto out;
+                        }
+
+                }
+
+        }
+
+        if (ret != SQLITE_DONE) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed retriving records from db : %s",
+                        sqlite3_errmsg (db_conn));
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        gfdb_query_record_fini (&gfdb_query_record);
+        return ret;
+}
+
+
+
+int
+gf_sql_clear_counters (gf_sql_connection_t *sql_conn)
+{
+        int ret                 = -1;
+        char *sql_strerror      = NULL;
+        char *query_str         = NULL;
+
+        CHECK_SQL_CONN (sql_conn, out);
+
+        query_str = "BEGIN;UPDATE "
+                    GF_FILE_TABLE
+                    " SET " GF_COL_READ_FREQ_CNTR " = 0 , "
+                    GF_COL_WRITE_FREQ_CNTR " = 0 ;COMMIT;";
+
+        ret = sqlite3_exec (sql_conn->sqlite3_db_conn, query_str, NULL, NULL,
+                                &sql_strerror);
+        if (ret != SQLITE_OK) {
+                gf_log (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                        "Failed executing: %s : %s",
+                        query_str, sql_strerror);
+                sqlite3_free (sql_strerror);
+                        ret = -1;
+                        goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.h b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.h
new file mode 100644
index 00000000000..8b62be1d4d0
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.h
@@ -0,0 +1,64 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#ifndef __GFDB_SQLITE3_HELPER_H
+#define __GFDB_SQLITE3_HELPER_H
+
+
+#ifndef _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "gfdb_sqlite3.h"
+
+/******************************************************************************
+ *
+ *                      Helper functions for gf_sqlite3_insert()
+ *
+ * ****************************************************************************/
+
+
+int
+gf_sql_insert_wind (gf_sql_connection_t  *sql_conn,
+                   gfdb_db_record_t     *gfdb_db_record);
+
+int
+gf_sql_insert_unwind (gf_sql_connection_t  *sql_conn,
+                     gfdb_db_record_t     *gfdb_db_record);
+
+
+int
+gf_sql_update_delete_wind (gf_sql_connection_t  *sql_conn,
+                          gfdb_db_record_t     *gfdb_db_record);
+
+int
+gf_sql_delete_unwind (gf_sql_connection_t  *sql_conn,
+                          gfdb_db_record_t     *gfdb_db_record);
+
+
+
+
+
+/******************************************************************************
+ *
+ *                      Find/Query helper functions
+ *
+ * ****************************************************************************/
+
+
+int
+gf_sql_query_function (sqlite3_stmt              *prep_stmt,
+                      gf_query_callback_t       query_callback,
+                      void                      *_query_cbk_args);
+
+int
+gf_sql_clear_counters (gf_sql_connection_t *sql_conn);
+
+#endif
diff --git a/libglusterfs/src/mem-types.h b/libglusterfs/src/mem-types.h
index 93aea8502e3..9ff78dd19a4 100644
--- a/libglusterfs/src/mem-types.h
+++ b/libglusterfs/src/mem-types.h
@@ -129,6 +129,18 @@ enum gf_common_mem_types_ {
         gf_common_mt_wr                   = 113,
         gf_common_mt_rdma_arena_mr        = 114,
         gf_common_mt_parser_t             = 115,
+        /*related to gfdb library*/
+        gfdb_mt_time_t,
+        gf_mt_sql_cbk_args_t,
+        gf_mt_gfdb_query_record_t,
+        gf_mt_gfdb_link_info_t,
+        gf_mt_gfdb_db_operations_t,
+        gf_mt_sql_connection_t,
+        gf_mt_sql_conn_node_t,
+        gf_mt_db_conn_node_t,
+        gf_mt_db_connection_t,
+        gfdb_mt_db_record_t,
+        /*related to gfdb library*/
         gf_common_mt_end
 };
 #endif