tier/libgfdb: Replacing ASCII query file with binary

Earlier, when the database was queried we used to save
all the queried records in an ASCII format in the query file.
This caused issues like filename having ASCII delimiter and used
to take a lot of space. The tier.c file also had a lot of parsing code.

Here we changed the format of the query file to binary.
All the logic of serialization and formating of query record is done
by libgfdb. Libgfdb provides API,
gfdb_write_query_record() and gfdb_read_query_record(),
which the user i.e tier migrator and CTR xlator can use to
write to and read from query file.
With this binary format we save on disk space i.e reduce to 50% atleast
as we are saving GFID's in binary format 16 bytes and not the string format
which takes 36 bytes + We are not saving path of the file + we are also saving on
ASCII delimiters.

The on disk format of query record is as follows,

+---------------------------------------------------------------------------+
| Length of serialized query record |       Serialized Query Record         |
+---------------------------------------------------------------------------+
             4 bytes                     Length of serialized query record
                                                      |
                                                      |
     -------------------------------------------------|
     |
     |
     V
   Serialized Query Record Format:
   +---------------------------------------------------------------------------+
   | GFID |  Link count   |  <LINK INFO>  |.....                      | FOOTER |
   +---------------------------------------------------------------------------+
     16 B        4 B         Link Length                                  4 B
                                |                                          |
                                |                                          |
   -----------------------------|                                          |
   |                                                                       |
   |                                                                       |
   V                                                                       |
   Each <Link Info> will be serialized as                                  |
   +-----------------------------------------------+                       |
   | PGID | BASE_NAME_LENGTH |      BASE_NAME      |                       |
   +-----------------------------------------------+                       |
     16 B       4 B             BASE_NAME_LENGTH                           |
                                                                           |
                                                                           |
   ------------------------------------------------------------------------|
   |
   |
   V
   FOOTER is a magic number 0xBAADF00D indicating the end of the record.
   This also serves as a serialized schema validator.

Change-Id: I9db7416fd421e118dd44eafab8b535caafe50d5a
BUG: 1272207
Signed-off-by: Joseph Fernandes <josferna@redhat.com>
Reviewed-on: http://review.gluster.org/12354
Reviewed-by: N Balachandran <nbalacha@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Dan Lambright <dlambrig@redhat.com>
Tested-by: Dan Lambright <dlambrig@redhat.com>

9 files changed, 1034 insertions, 271 deletions

diff --git a/libglusterfs/src/gfdb/Makefile.am b/libglusterfs/src/gfdb/Makefile.am
index 30d1b7bcdde..50cf3402787 100644
--- a/libglusterfs/src/gfdb/Makefile.am
+++ b/libglusterfs/src/gfdb/Makefile.am
@@ -18,14 +18,14 @@ endif
 
 CONTRIB_BUILDDIR = $(top_builddir)/contrib
 
-libgfdb_la_SOURCES = gfdb_data_store.c gfdb_sqlite3_helper.c\
+libgfdb_la_SOURCES = gfdb_data_store.c gfdb_data_store_helper.c gfdb_sqlite3_helper.c\
    gfdb_sqlite3.c
 
 noinst_HEADERS =  gfdb_data_store.h gfdb_data_store_types.h gfdb_sqlite3_helper.h\
-   gfdb_sqlite3.h gfdb_mem-types.h
+   gfdb_sqlite3.h gfdb_mem-types.h gfdb_data_store_helper.h
 
-libgfdb_HEADERS = gfdb_data_store.h gfdb_data_store_types.h \
-                gfdb_sqlite3.h gfdb_mem-types.h gfdb_sqlite3_helper.c
+libgfdb_HEADERS = gfdb_data_store.h gfdb_data_store_types.h gfdb_data_store_helper.h\
+                gfdb_sqlite3.h gfdb_mem-types.h gfdb_sqlite3_helper.h
 
 CLEANFILES =
 
diff --git a/libglusterfs/src/gfdb/gfdb_data_store.c b/libglusterfs/src/gfdb/gfdb_data_store.c
index af4be2116d0..a3b680f3e7e 100644
--- a/libglusterfs/src/gfdb/gfdb_data_store.c
+++ b/libglusterfs/src/gfdb/gfdb_data_store.c
@@ -796,5 +796,16 @@ void get_gfdb_methods (gfdb_methods_t *methods)
         methods->get_db_version = get_db_version;
         methods->get_db_setting = get_db_setting;
         methods->get_db_path_key = get_db_path_key;
+
+        /* Query Record related functions */
+        methods->gfdb_query_record_new = gfdb_query_record_new;
+        methods->gfdb_query_record_free = gfdb_query_record_free;
+        methods->gfdb_add_link_to_query_record = gfdb_add_link_to_query_record;
+        methods->gfdb_write_query_record = gfdb_write_query_record;
+        methods->gfdb_read_query_record = gfdb_read_query_record;
+
+        /* Link info related functions */
+        methods->gfdb_link_info_new = gfdb_link_info_new;
+        methods->gfdb_link_info_free = gfdb_link_info_free;
 }
 
diff --git a/libglusterfs/src/gfdb/gfdb_data_store.h b/libglusterfs/src/gfdb/gfdb_data_store.h
index 109c72768df..5d25fa44a4a 100644
--- a/libglusterfs/src/gfdb/gfdb_data_store.h
+++ b/libglusterfs/src/gfdb/gfdb_data_store.h
@@ -345,6 +345,18 @@ typedef struct gfdb_methods_s {
         /* access function: get_db_path_key(). */
         char *dbpath;
         get_db_path_key_t get_db_path_key;
+
+        /* Query Record related functions */
+        gfdb_query_record_new_t gfdb_query_record_new;
+        gfdb_query_record_free_t gfdb_query_record_free;
+        gfdb_add_link_to_query_record_t gfdb_add_link_to_query_record;
+        gfdb_write_query_record_t gfdb_write_query_record;
+        gfdb_read_query_record_t gfdb_read_query_record;
+
+        /* Link info related functions */
+        gfdb_link_info_new_t gfdb_link_info_new;
+        gfdb_link_info_free_t gfdb_link_info_free;
+
 } gfdb_methods_t;
 
 void get_gfdb_methods (gfdb_methods_t *methods);
diff --git a/libglusterfs/src/gfdb/gfdb_data_store_helper.c b/libglusterfs/src/gfdb/gfdb_data_store_helper.c
new file mode 100644
index 00000000000..ff85e17169d
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_data_store_helper.c
@@ -0,0 +1,605 @@
+#include "gfdb_data_store_helper.h"
+
+
+/*Create a single link info structure*/
+gfdb_link_info_t*
+gfdb_link_info_new ()
+{
+        gfdb_link_info_t *link_info = NULL;
+
+        link_info = GF_CALLOC (1, sizeof(gfdb_link_info_t),
+                                        gf_mt_gfdb_link_info_t);
+        if (!link_info) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, ENOMEM,
+                        LG_MSG_NO_MEMORY, "Memory allocation failed for "
+                        "link_info ");
+                goto out;
+        }
+
+        INIT_LIST_HEAD (&link_info->list);
+
+out:
+
+        return link_info;
+}
+
+/*Destroy a link info structure*/
+void
+gfdb_link_info_free(gfdb_link_info_t *link_info)
+{
+        GF_FREE (link_info);
+}
+
+
+/*Function to create the query_record*/
+gfdb_query_record_t *
+gfdb_query_record_new()
+{
+        int ret = -1;
+        gfdb_query_record_t *query_record = NULL;
+
+        query_record = GF_CALLOC (1, sizeof(gfdb_query_record_t),
+                                        gf_mt_gfdb_query_record_t);
+        if (!query_record) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, ENOMEM,
+                        LG_MSG_NO_MEMORY, "Memory allocation failed for "
+                        "query_record ");
+                goto out;
+        }
+
+        INIT_LIST_HEAD (&query_record->link_list);
+
+        ret = 0;
+out:
+        if (ret == -1) {
+                GF_FREE (query_record);
+        }
+        return query_record;
+}
+
+
+/*Function to delete a single linkinfo from list*/
+static void
+gfdb_delete_linkinfo_from_list (gfdb_link_info_t **link_info)
+{
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, link_info, out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, *link_info, out);
+
+        /*Remove hard link from list*/
+        list_del(&(*link_info)->list);
+        gfdb_link_info_free (*link_info);
+        link_info = NULL;
+out:
+        return;
+}
+
+
+/*Function to destroy link_info list*/
+void
+gfdb_free_link_info_list (gfdb_query_record_t *query_record)
+{
+        gfdb_link_info_t        *link_info = NULL;
+        gfdb_link_info_t        *temp = NULL;
+
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, query_record, out);
+
+        list_for_each_entry_safe(link_info, temp,
+                        &query_record->link_list, list)
+        {
+                gfdb_delete_linkinfo_from_list (&link_info);
+                link_info = NULL;
+        }
+
+out:
+        return;
+}
+
+
+
+/* Function to add linkinfo to the query record */
+int
+gfdb_add_link_to_query_record (gfdb_query_record_t      *query_record,
+                           uuid_t                   pgfid,
+                           char               *base_name)
+{
+        int ret                                 = -1;
+        gfdb_link_info_t *link_info             = NULL;
+        int base_name_len                       = 0;
+
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, query_record, out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, pgfid, out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, base_name, out);
+
+        link_info = gfdb_link_info_new ();
+        if (!link_info) {
+                goto out;
+        }
+
+        gf_uuid_copy (link_info->pargfid, pgfid);
+        base_name_len = strlen (base_name);
+        memcpy (link_info->file_name, base_name, base_name_len);
+        link_info->file_name[base_name_len] = '\0';
+
+        list_add_tail (&link_info->list,
+                        &query_record->link_list);
+
+        query_record->link_count++;
+
+        ret = 0;
+out:
+        if (ret) {
+                gfdb_link_info_free (link_info);
+                link_info = NULL;
+        }
+        return ret;
+}
+
+
+
+/*Function to destroy query record*/
+void
+gfdb_query_record_free(gfdb_query_record_t *query_record)
+{
+        if (query_record) {
+                gfdb_free_link_info_list (query_record);
+                GF_FREE (query_record);
+        }
+}
+
+
+/******************************************************************************
+                SERIALIZATION/DE-SERIALIZATION OF QUERY RECORD
+*******************************************************************************/
+/******************************************************************************
+ The on disk format of query record is as follows,
+
++---------------------------------------------------------------------------+
+| Length of serialized query record |       Serialized Query Record         |
++---------------------------------------------------------------------------+
+             4 bytes                     Length of serialized query record
+                                                      |
+                                                      |
+     -------------------------------------------------|
+     |
+     |
+     V
+   Serialized Query Record Format:
+   +---------------------------------------------------------------------------+
+   | GFID |  Link count   |  <LINK INFO>  |.....                      | FOOTER |
+   +---------------------------------------------------------------------------+
+     16 B        4 B         Link Length                                  4 B
+                                |                                          |
+                                |                                          |
+   -----------------------------|                                          |
+   |                                                                       |
+   |                                                                       |
+   V                                                                       |
+   Each <Link Info> will be serialized as                                  |
+   +-----------------------------------------------+                       |
+   | PGID | BASE_NAME_LENGTH |      BASE_NAME      |                       |
+   +-----------------------------------------------+                       |
+     16 B       4 B             BASE_NAME_LENGTH                           |
+                                                                           |
+                                                                           |
+   ------------------------------------------------------------------------|
+   |
+   |
+   V
+   FOOTER is a magic number 0xBAADF00D indicating the end of the record.
+   This also serves as a serialized schema validator.
+ * ****************************************************************************/
+
+#define GFDB_QUERY_RECORD_FOOTER 0xBAADF00D
+#define UUID_LEN                 16
+
+/*Function to get the potential length of the serialized buffer*/
+static int32_t
+gfdb_query_record_serialized_length (gfdb_query_record_t *query_record)
+{
+        int32_t len                         = -1;
+        gfdb_link_info_t *link_info     = NULL;
+
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, query_record, out);
+
+        /* Length of GFID */
+        len = UUID_LEN;
+
+        /* length of number of links*/
+        len += sizeof (int32_t);
+
+        list_for_each_entry (link_info, &query_record->link_list, list) {
+
+                /* length of PFID */
+                len += UUID_LEN;
+
+                /* Add size of base name length*/
+                len += sizeof (int32_t);
+
+                /* Length of base_name */
+                len += strlen (link_info->file_name);
+
+        }
+
+        /* length of footer */
+        len += sizeof (int32_t);
+out:
+        return len;
+}
+
+/* Function for serializing query record.
+ *
+ * Query Record Serialization Format
+ * +---------------------------------------------------------------------------+
+ * | GFID |  Link count   |  <LINK INFO>  |.....                      | FOOTER |
+ * +---------------------------------------------------------------------------+
+ *   16 B        4 B         Link Length                                  4 B
+ *
+ *
+ * Each <Link Info> will be serialized as
+ * +-----------------------------------------------+
+ * | PGID | BASE_NAME_LENGTH |      BASE_NAME      |
+ * +-----------------------------------------------+
+ *   16 B       4 B             BASE_NAME_LENGTH
+ *
+ *
+ * FOOTER is a magic number 0xBAADF00D indicating the end of the record.
+ * This also serves as a serialized schema validator.
+ *
+ * The function will allocate memory to the serialized buffer,
+ * the caller needs to free it.
+ * Returns the length of the serialized buffer on success
+ * or -1 on failure.
+ *
+ * */
+static int
+gfdb_query_record_serialize (gfdb_query_record_t *query_record,
+                             char **in_buffer)
+{
+        gfdb_link_info_t *link_info     = NULL;
+        int              count          = -1;
+        int              base_name_len  = 0;
+        int              buffer_length  = 0;
+        int              footer         = GFDB_QUERY_RECORD_FOOTER;
+        char             *buffer        = NULL;
+        char             *ret_buffer    = NULL;
+
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, query_record, out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE,
+                             (query_record->link_count > 0), out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, in_buffer, out);
+
+
+        /* Calculate the total length of the serialized buffer */
+        buffer_length = gfdb_query_record_serialized_length (query_record);
+        if (buffer_length <= 0) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                        LG_MSG_DB_ERROR, "Failed to calculate the length of "
+                        "serialized buffer");
+                goto out;
+        }
+
+        /* Allocate memory to the serialized buffer */
+        ret_buffer = GF_CALLOC (1, buffer_length,  gf_common_mt_char);
+        if (!ret_buffer) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                        LG_MSG_DB_ERROR, "Memory allocation failed for "
+                        "serialized buffer.");
+                goto out;
+        }
+
+        buffer = ret_buffer;
+
+        count = 0;
+
+        /* Copying the GFID */
+        memcpy (buffer, query_record->gfid, UUID_LEN);
+        buffer += UUID_LEN;
+        count += UUID_LEN;
+
+        /* Copying the number of links */
+        memcpy (buffer, &query_record->link_count, sizeof (int32_t));
+        buffer += sizeof (int32_t);
+        count += sizeof (int32_t);
+
+        list_for_each_entry (link_info, &query_record->link_list, list) {
+
+                /* Copying the PFID */
+                memcpy(buffer, link_info->pargfid, UUID_LEN);
+                buffer += UUID_LEN;
+                count += UUID_LEN;
+
+                /* Copying base name length*/
+                base_name_len = strlen (link_info->file_name);
+                memcpy (buffer, &base_name_len, sizeof (int32_t));
+                buffer += sizeof (int32_t);
+                count += sizeof (int32_t);
+
+                /* Length of base_name */
+                memcpy(buffer, link_info->file_name, base_name_len);
+                buffer += base_name_len;
+                count += base_name_len;
+
+        }
+
+        /* Copying the Footer of the record */
+        memcpy (buffer, &footer, sizeof (int32_t));
+        buffer += sizeof (int32_t);
+        count += sizeof (int32_t);
+
+out:
+        if (count < 0) {
+                GF_FREE (ret_buffer);
+                ret_buffer = NULL;
+        }
+        *in_buffer = ret_buffer;
+        return count;
+}
+
+static gf_boolean_t
+is_serialized_buffer_valid (char *in_buffer, int buffer_length) {
+        gf_boolean_t    ret        = _gf_false;
+        int             footer     = 0;
+
+        /* Read the footer */
+        in_buffer += (buffer_length - sizeof (int32_t));
+        memcpy (&footer, in_buffer, sizeof (int32_t));
+
+        /*
+         * if the footer is not GFDB_QUERY_RECORD_FOOTER
+         * then the serialized record is invalid
+         *
+         * */
+        if (footer != GFDB_QUERY_RECORD_FOOTER) {
+                goto out;
+        }
+
+        ret = _gf_true;
+out:
+        return ret;
+}
+
+
+static int
+gfdb_query_record_deserialize (char *in_buffer,
+                               int buffer_length,
+                               gfdb_query_record_t **query_record)
+{
+        int ret                                 = -1;
+        char *buffer                            = NULL;
+        int i                                   = 0;
+        gfdb_link_info_t *link_info             = NULL;
+        int count                               = 0;
+        int base_name_len                       = 0;
+        gfdb_query_record_t *ret_qrecord        = NULL;
+
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, in_buffer, out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, query_record, out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, (buffer_length > 0), out);
+
+        if (!is_serialized_buffer_valid (in_buffer, buffer_length)) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                        LG_MSG_DB_ERROR, "Invalid serialized query record");
+                goto out;
+        }
+
+        buffer = in_buffer;
+
+        ret_qrecord = gfdb_query_record_new ();
+        if (!ret_qrecord) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                        LG_MSG_DB_ERROR, "Failed to allocate space to "
+                        "gfdb_query_record_t");
+                goto out;
+        }
+
+        /* READ GFID */
+        memcpy ((ret_qrecord)->gfid, buffer, UUID_LEN);
+        buffer += UUID_LEN;
+        count += UUID_LEN;
+
+        /* Read the number of link */
+        memcpy (&(ret_qrecord->link_count), buffer, sizeof (int32_t));
+        buffer += sizeof (int32_t);
+        count += sizeof (int32_t);
+
+        /* Read all the links */
+        for (i = 0; i < ret_qrecord->link_count; i++) {
+                if (count >= buffer_length) {
+                        gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                                LG_MSG_DB_ERROR, "Invalid serialized "
+                                "query record");
+                        ret = -1;
+                        goto out;
+                }
+
+                link_info = gfdb_link_info_new ();
+                if (!link_info) {
+                        gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                                LG_MSG_DB_ERROR, "Failed to create link_info");
+                        goto out;
+                }
+
+                /* READ PGFID */
+                memcpy (link_info->pargfid, buffer, UUID_LEN);
+                buffer += UUID_LEN;
+                count += UUID_LEN;
+
+                /* Read base name length */
+                memcpy (&base_name_len, buffer, sizeof (int32_t));
+                buffer += sizeof (int32_t);
+                count += sizeof (int32_t);
+
+                /* READ basename */
+                memcpy (link_info->file_name, buffer, base_name_len);
+                buffer += base_name_len;
+                count += base_name_len;
+                link_info->file_name[base_name_len] = '\0';
+
+                /* Add link_info to the list */
+                list_add_tail (&link_info->list,
+                               &(ret_qrecord->link_list));
+
+                /* Reseting link_info */
+                link_info = NULL;
+        }
+
+        ret = 0;
+out:
+        if (ret) {
+                gfdb_query_record_free (ret_qrecord);
+                ret_qrecord = NULL;
+        }
+        *query_record = ret_qrecord;
+        return ret;
+}
+
+
+
+
+
+/* Function to write query record to file
+ *
+ * Disk format
+ * +---------------------------------------------------------------------------+
+ * | Length of serialized query record |       Serialized Query Record         |
+ * +---------------------------------------------------------------------------+
+ *            4 bytes                     Length of serialized query record
+ *
+ * Please refer gfdb_query_record_serialize () for format of
+ * Serialized Query Record
+ *
+ * */
+int
+gfdb_write_query_record (int fd,
+                        gfdb_query_record_t *query_record)
+{
+        int ret                 = -1;
+        int buffer_len          = 0;
+        char *buffer            = NULL;
+        int write_len           = 0;
+        char *write_buffer      = NULL;
+
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, (fd >= 0), out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, query_record, out);
+
+        buffer_len = gfdb_query_record_serialize (query_record, &buffer);
+        if (buffer_len < 0) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                        LG_MSG_DB_ERROR, "Failed to serialize query record");
+                goto out;
+        }
+
+        /* Serialize the buffer length and write to file */
+        ret = write (fd, &buffer_len, sizeof (int32_t));
+        if (ret < 0) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                                LG_MSG_DB_ERROR, "Failed to write buffer length"
+                                " to file");
+                goto out;
+        }
+
+        /* Write the serialized query record to file */
+        write_len = buffer_len;
+        write_buffer = buffer;
+        while ((ret = write (fd, write_buffer, write_len)) <  write_len) {
+                if (ret < 0) {
+                        gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, errno,
+                                LG_MSG_DB_ERROR, "Failed to write serialized "
+                                "query record to file");
+                        goto out;
+                }
+
+                write_buffer += ret;
+                write_len -= ret;
+        }
+
+        ret = 0;
+out:
+        GF_FREE (buffer);
+        return ret;
+}
+
+
+
+/* Function to read query record from file.
+ * Allocates memory to query record and
+ * returns length of serialized query record when successful
+ * Return -1 when failed.
+ * Return 0 when reached EOF.
+ * */
+int
+gfdb_read_query_record (int fd,
+                        gfdb_query_record_t **query_record)
+{
+        int ret                 = -1;
+        int buffer_len          = 0;
+        int read_len            = 0;
+        char *buffer            = NULL;
+        char *read_buffer       = NULL;
+
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, (fd >= 0), out);
+        GF_VALIDATE_OR_GOTO (GFDB_DATA_STORE, query_record, out);
+
+
+        /* Read serialized query record length from the file*/
+        ret = read (fd, &buffer_len, sizeof (int32_t));
+        if (ret < 0) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                                LG_MSG_DB_ERROR, "Failed reading buffer length"
+                                " from file");
+                goto out;
+        }
+        /* EOF */
+        else if (ret == 0) {
+                ret = 0;
+                goto out;
+        }
+
+        /* Allocating memory to the serialization buffer */
+        buffer = GF_CALLOC (1, buffer_len,  gf_common_mt_char);
+        if (!buffer) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                        LG_MSG_DB_ERROR, "Failed to allocate space to "
+                        "serialized buffer");
+                goto out;
+        }
+
+
+        /* Read the serialized query record from file */
+        read_len = buffer_len;
+        read_buffer = buffer;
+        while ((ret = read (fd, read_buffer, read_len)) < read_len) {
+
+                /*Any error */
+                if (ret < 0) {
+                        gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, errno,
+                                LG_MSG_DB_ERROR, "Failed to read serialized "
+                                "query record from file");
+                        goto out;
+                }
+                /* EOF */
+                else if (ret == 0) {
+                        gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                                LG_MSG_DB_ERROR, "Invalid query record or "
+                                "corrupted query file");
+                        ret = -1;
+                        goto out;
+                }
+
+                read_buffer += ret;
+                read_len -= ret;
+        }
+
+        ret = gfdb_query_record_deserialize (buffer, buffer_len,
+                                             query_record);
+        if (ret) {
+                gf_msg (GFDB_DATA_STORE, GF_LOG_ERROR, 0,
+                        LG_MSG_DB_ERROR, "Failed to de-serialize query record");
+                goto out;
+        }
+
+        ret = buffer_len;
+out:
+        GF_FREE (buffer);
+        return ret;
+}
diff --git a/libglusterfs/src/gfdb/gfdb_data_store_helper.h b/libglusterfs/src/gfdb/gfdb_data_store_helper.h
new file mode 100644
index 00000000000..fe9fbba8795
--- /dev/null
+++ b/libglusterfs/src/gfdb/gfdb_data_store_helper.h
@@ -0,0 +1,125 @@
+/*
+   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_HELPER_H
+#define __GFDB_DATA_STORE_HELPER_H
+
+#include <time.h>
+#include <sys/time.h>
+#include <string.h>
+
+#include "common-utils.h"
+#include "compat-uuid.h"
+#include "gfdb_mem-types.h"
+#include "dict.h"
+#include "byte-order.h"
+#include "libglusterfs-messages.h"
+
+
+#define GFDB_DATA_STORE               "gfdbdatastore"
+
+/*******************************************************************************
+ *
+ *                     Query related data structure and functions
+ *
+ * ****************************************************************************/
+
+#ifdef NAME_MAX
+#define GF_NAME_MAX NAME_MAX
+#else
+#define GF_NAME_MAX 255
+#endif
+
+/*Structure to hold the link information*/
+typedef struct gfdb_link_info {
+        uuid_t                          pargfid;
+        char                            file_name[GF_NAME_MAX];
+        struct list_head                list;
+} gfdb_link_info_t;
+
+
+/*Create a single link info structure*/
+gfdb_link_info_t *gfdb_link_info_new ();
+typedef gfdb_link_info_t *(*gfdb_link_info_new_t) ();
+
+/*Destroy a link info structure*/
+void
+gfdb_link_info_free (gfdb_link_info_t *gfdb_link_info);
+typedef void
+(*gfdb_link_info_free_t) (gfdb_link_info_t *gfdb_link_info);
+
+
+
+
+
+/*Structure used for querying purpose*/
+typedef struct gfdb_query_record {
+        uuid_t                          gfid;
+        /*This is the hardlink list*/
+        struct list_head                link_list;
+        int                             link_count;
+} gfdb_query_record_t;
+
+
+
+/* Function to create the query_record */
+gfdb_query_record_t *
+gfdb_query_record_new();
+typedef gfdb_query_record_t *
+(*gfdb_query_record_new_t)();
+
+
+
+
+/* Fuction to add linkinfo to query record */
+int
+gfdb_add_link_to_query_record (gfdb_query_record_t      *gfdb_query_record,
+                           uuid_t                   pgfid,
+                           char               *base_name);
+typedef int
+(*gfdb_add_link_to_query_record_t) (gfdb_query_record_t *, uuid_t, char *);
+
+
+
+
+/*Function to destroy query record*/
+void
+gfdb_query_record_free (gfdb_query_record_t *gfdb_query_record);
+typedef void
+(*gfdb_query_record_free_t) (gfdb_query_record_t *);
+
+
+
+
+
+
+/* Function to write query record to file */
+int
+gfdb_write_query_record (int fd,
+                        gfdb_query_record_t *gfdb_query_record);
+typedef int
+(*gfdb_write_query_record_t) (int, gfdb_query_record_t *);
+
+
+
+
+
+/* Function to read query record from file.
+ * Allocates memory to query record and return 0 when successful
+ * Return -1 when failed.
+ * Return 0 when EOF.
+ * */
+int
+gfdb_read_query_record (int fd,
+                        gfdb_query_record_t **gfdb_query_record);
+typedef int
+(*gfdb_read_query_record_t) (int, gfdb_query_record_t **);
+
+
+#endif
\ No newline at end of file
diff --git a/libglusterfs/src/gfdb/gfdb_data_store_types.h b/libglusterfs/src/gfdb/gfdb_data_store_types.h
index ab8b7945d2c..ce09e731746 100644
--- a/libglusterfs/src/gfdb/gfdb_data_store_types.h
+++ b/libglusterfs/src/gfdb/gfdb_data_store_types.h
@@ -10,16 +10,7 @@
 #ifndef __GFDB_DATA_STORE_TYPE_H
 #define __GFDB_DATA_STORE_TYPE_H
 
-
-#include <time.h>
-#include <sys/time.h>
-#include <string.h>
-
-#include "common-utils.h"
-#include "compat-uuid.h"
-#include "gfdb_mem-types.h"
-#include "dict.h"
-#include "libglusterfs-messages.h"
+#include "gfdb_data_store_helper.h"
 
 /*
  * Helps in dynamically choosing log level
@@ -149,7 +140,6 @@ typedef enum gfdb_db_type {
 } 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"
@@ -314,168 +304,6 @@ typedef struct gfdb_db_record {
 } 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_msg (GFDB_DATA_STORE, GF_LOG_ERROR, ENOMEM,
-                        LG_MSG_NO_MEMORY, "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_msg (GFDB_DATA_STORE, GF_LOG_ERROR, ENOMEM,
-                        LG_MSG_NO_MEMORY, "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 = gf_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
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3.c b/libglusterfs/src/gfdb/gfdb_sqlite3.c
index c50de4a47bc..84ee9c32f08 100644
--- a/libglusterfs/src/gfdb/gfdb_sqlite3.c
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3.c
@@ -603,6 +603,28 @@ out:
  *
  * ***************************************************************************/
 
+static int
+gf_get_basic_query_stmt (char **out_stmt)
+{
+        int ret = -1;
+        ret = gf_asprintf (out_stmt, "select GF_FILE_TB.GF_ID,"
+                                  "GF_FLINK_TB.GF_PID ,"
+                                  "GF_FLINK_TB.FNAME "
+                                  "from GF_FLINK_TB, GF_FILE_TB "
+                                  "where "
+                                  "GF_FILE_TB.GF_ID = GF_FLINK_TB.GF_ID ");
+        if (ret <= 0) {
+                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0, LG_MSG_QUERY_FAILED,
+                        "Failed to create base query statement");
+                *out_stmt = NULL;
+        }
+        return ret;
+}
+
+
+
+
+
 /*
  * Find All files recorded in the DB
  * Input:
@@ -618,22 +640,19 @@ gf_sqlite3_find_all (void *db_conn, gf_query_callback_t query_callback,
         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 = gf_get_basic_query_stmt (&query_str);
+        if (ret <= 0) {
+                goto out;
+        }
 
         ret = sqlite3_prepare (sql_conn->sqlite3_db_conn, query_str, -1,
                                 &prep_stmt, 0);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_PREPARE_FAILED, "Failed preparing statment %s :"
+                        LG_MSG_PREPARE_FAILED, "Failed to prepare statment %s :"
                         "%s", query_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -650,6 +669,7 @@ gf_sqlite3_find_all (void *db_conn, gf_query_callback_t query_callback,
         ret = 0;
 out:
         sqlite3_finalize (prep_stmt);
+        GF_FREE (query_str);
         return ret;
 }
 
@@ -672,22 +692,31 @@ gf_sqlite3_find_recently_changed_files(void *db_conn,
         gf_sql_connection_t *sql_conn           =       db_conn;
         sqlite3_stmt *prep_stmt                 =       NULL;
         uint64_t  from_time_usec                =       0;
+        char *base_query_str                    =       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 where "
+        ret = gf_get_basic_query_stmt (&base_query_str);
+        if (ret <= 0) {
+                goto out;
+        }
+
+        ret = gf_asprintf (&query_str, "%s AND"
                 /*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 ") >= ?)";
+                GF_COL_TB_RWMSEC ") >= ?)", base_query_str);
+
+        if (ret < 0) {
+                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0, LG_MSG_QUERY_FAILED,
+                        "Failed creating query statement");
+                query_str = NULL;
+                goto out;
+        }
 
         from_time_usec = gfdb_time_2_usec (from_time);
 
@@ -695,7 +724,7 @@ gf_sqlite3_find_recently_changed_files(void *db_conn,
                                &prep_stmt, 0);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_PREPARE_FAILED, "Failed preparing statment %s :"
+                        LG_MSG_PREPARE_FAILED, "Failed to prepare statment %s :"
                         " %s", query_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -706,7 +735,7 @@ gf_sqlite3_find_recently_changed_files(void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 1, from_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding from_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind from_time_usec "
                         "%"PRIu64" : %s", from_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -717,7 +746,7 @@ gf_sqlite3_find_recently_changed_files(void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 2, from_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding from_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind from_time_usec "
                         "%"PRIu64" : %s ", from_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -735,6 +764,8 @@ gf_sqlite3_find_recently_changed_files(void *db_conn,
         ret = 0;
 out:
         sqlite3_finalize (prep_stmt);
+        GF_FREE (base_query_str);
+        GF_FREE (query_str);
         return ret;
 }
 
@@ -757,22 +788,31 @@ gf_sqlite3_find_unchanged_for_time (void *db_conn,
         gf_sql_connection_t *sql_conn           =       db_conn;
         sqlite3_stmt *prep_stmt                 =       NULL;
         uint64_t  for_time_usec                 =       0;
+        char *base_query_str                    =       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 where "
+        ret = gf_get_basic_query_stmt (&base_query_str);
+        if (ret <= 0) {
+                goto out;
+        }
+
+        ret = gf_asprintf (&query_str, "%s AND "
                 /*First condition: For writes*/
                 "((" GF_COL_TB_WSEC " * " TOSTRING(GFDB_MICROSEC) " + "
                 GF_COL_TB_WMSEC ") <= ? )"
                 " AND "
                 /*Second condition: For reads*/
                 "((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
-                GF_COL_TB_RWMSEC ") <= ?)";
+                GF_COL_TB_RWMSEC ") <= ?)", base_query_str);
+
+        if (ret < 0) {
+                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0, LG_MSG_QUERY_FAILED,
+                        "Failed to create query statement");
+                query_str = NULL;
+                goto out;
+        }
 
         for_time_usec = gfdb_time_2_usec (for_time);
 
@@ -780,7 +820,7 @@ gf_sqlite3_find_unchanged_for_time (void *db_conn,
                                &prep_stmt, 0);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_PREPARE_FAILED, "Failed preparing statment %s :"
+                        LG_MSG_PREPARE_FAILED, "Failed to prepare statment %s :"
                         " %s", query_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -791,7 +831,7 @@ gf_sqlite3_find_unchanged_for_time (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 1, for_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding for_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind for_time_usec "
                         "%"PRIu64" : %s", for_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -802,7 +842,7 @@ gf_sqlite3_find_unchanged_for_time (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 2, for_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding for_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind for_time_usec "
                         "%"PRIu64" : %s", for_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -820,6 +860,8 @@ gf_sqlite3_find_unchanged_for_time (void *db_conn,
         ret = 0;
 out:
         sqlite3_finalize (prep_stmt);
+        GF_FREE (base_query_str);
+        GF_FREE (query_str);
         return ret;
 }
 
@@ -852,15 +894,16 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
         gf_sql_connection_t *sql_conn           =       db_conn;
         sqlite3_stmt *prep_stmt                 =       NULL;
         uint64_t  from_time_usec                =       0;
+        char *base_query_str                    =       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 where "
+        ret = gf_get_basic_query_stmt (&base_query_str);
+        if (ret <= 0) {
+                goto out;
+        }
+        ret = gf_asprintf (&query_str, "%s AND "
                 /*First condition: For Writes*/
                 "( ((" GF_COL_TB_WSEC " * " TOSTRING(GFDB_MICROSEC) " + "
                 GF_COL_TB_WMSEC ") >= ? )"
@@ -869,7 +912,14 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
                 /*Second condition: For Reads */
                 "( ((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
                 GF_COL_TB_RWMSEC ") >= ?)"
-                " AND "" (" GF_COL_TB_RFC " >= ? ) )";
+                " AND "" (" GF_COL_TB_RFC " >= ? ) )", base_query_str);
+
+        if (ret < 0) {
+                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0, LG_MSG_QUERY_FAILED,
+                        "Failed to create query statement");
+                query_str = NULL;
+                goto out;
+        }
 
         from_time_usec = gfdb_time_2_usec (from_time);
 
@@ -877,7 +927,7 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
                                 &prep_stmt, 0);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_PREPARE_FAILED, "Failed preparing statment %s :"
+                        LG_MSG_PREPARE_FAILED, "Failed to prepare statment %s :"
                         " %s", query_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -888,7 +938,7 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 1, from_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding from_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind from_time_usec "
                         "%"PRIu64" : %s", from_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -899,7 +949,7 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
         ret = sqlite3_bind_int (prep_stmt, 2, freq_write_cnt);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding freq_write_cnt "
+                        LG_MSG_BINDING_FAILED, "Failed to bind freq_write_cnt "
                         "%d : %s", freq_write_cnt,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -911,7 +961,7 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 3, from_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding from_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind from_time_usec "
                         "%"PRIu64" : %s", from_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -922,7 +972,7 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
         ret = sqlite3_bind_int (prep_stmt, 4, freq_read_cnt);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding freq_read_cnt "
+                        LG_MSG_BINDING_FAILED, "Failed to bind freq_read_cnt "
                         "%d : %s", freq_read_cnt,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -944,7 +994,7 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
                 ret = gf_sql_clear_counters (sql_conn);
                 if (ret) {
                         gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                                LG_MSG_CLEAR_COUNTER_FAILED, "Failed clearing"
+                                LG_MSG_CLEAR_COUNTER_FAILED, "Failed to clear"
                                 " counters!");
                         goto out;
                 }
@@ -952,6 +1002,8 @@ gf_sqlite3_find_recently_changed_files_freq (void *db_conn,
         ret = 0;
 out:
         sqlite3_finalize (prep_stmt);
+        GF_FREE (base_query_str);
+        GF_FREE (query_str);
         return ret;
 }
 
@@ -982,15 +1034,17 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         gf_sql_connection_t *sql_conn           =       db_conn;
         sqlite3_stmt *prep_stmt                 =       NULL;
         uint64_t  for_time_usec                 =       0;
+        char *base_query_str                    =       NULL;
 
         CHECK_SQL_CONN (sql_conn, out);
-        GF_VALIDATE_OR_GOTO (GFDB_STR_SQLITE3, query_callback, out);
+        GF_VALIDATE_OR_GOTO(GFDB_STR_SQLITE3, query_callback, out);
+
+        ret = gf_get_basic_query_stmt (&base_query_str);
+        if (ret <= 0) {
+                goto 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 "
+        ret = gf_asprintf (&query_str, "%s AND "
                 /*First condition: For Writes
                  * Files that have write wind time smaller than for_time
                  * OR
@@ -1013,8 +1067,14 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
                 " OR "
                 "( (" GF_COL_TB_RFC " < ? ) AND"
                 "((" GF_COL_TB_RWSEC " * " TOSTRING(GFDB_MICROSEC) " + "
-                GF_COL_TB_RWMSEC ") >= ? ) ) )";
+                GF_COL_TB_RWMSEC ") >= ? ) ) )", base_query_str);
 
+        if (ret < 0) {
+                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0, LG_MSG_QUERY_FAILED,
+                        "Failed to create query statement");
+                query_str = NULL;
+                goto out;
+        }
 
         for_time_usec = gfdb_time_2_usec (for_time);
 
@@ -1022,7 +1082,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
                                 &prep_stmt, 0);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_PREPARE_FAILED, "Failed preparing delete "
+                        LG_MSG_PREPARE_FAILED, "Failed to prepare delete "
                         "statment %s : %s", query_str,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -1033,7 +1093,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 1, for_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding for_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind for_time_usec "
                         "%"PRIu64" : %s", for_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -1044,7 +1104,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         ret = sqlite3_bind_int (prep_stmt, 2, freq_write_cnt);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding freq_write_cnt"
+                        LG_MSG_BINDING_FAILED, "Failed to bind freq_write_cnt"
                         " %d : %s", freq_write_cnt,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -1055,7 +1115,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 3, for_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding for_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind for_time_usec "
                         "%"PRIu64" : %s", for_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -1068,7 +1128,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 4, for_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding for_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind for_time_usec "
                         "%"PRIu64" : %s", for_time_usec,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -1079,7 +1139,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         ret = sqlite3_bind_int (prep_stmt, 5, freq_read_cnt);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding freq_read_cnt "
+                        LG_MSG_BINDING_FAILED, "Failed to bind freq_read_cnt "
                         "%d : %s", freq_read_cnt,
                         sqlite3_errmsg (sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -1090,7 +1150,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         ret = sqlite3_bind_int64 (prep_stmt, 6, for_time_usec);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_BINDING_FAILED, "Failed binding for_time_usec "
+                        LG_MSG_BINDING_FAILED, "Failed to bind for_time_usec "
                         "%"PRIu64" : %s", for_time_usec,
                         sqlite3_errmsg(sql_conn->sqlite3_db_conn));
                 ret = -1;
@@ -1111,7 +1171,7 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
                 ret = gf_sql_clear_counters (sql_conn);
                 if (ret) {
                         gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                                LG_MSG_CLEAR_COUNTER_FAILED, "Failed clearing "
+                                LG_MSG_CLEAR_COUNTER_FAILED, "Failed to clear "
                                 "counters!");
                         goto out;
                 }
@@ -1120,6 +1180,8 @@ gf_sqlite3_find_unchanged_for_time_freq (void *db_conn,
         ret = 0;
 out:
         sqlite3_finalize (prep_stmt);
+        GF_FREE (base_query_str);
+        GF_FREE (query_str);
         return ret;
 }
 
@@ -1135,8 +1197,8 @@ gf_sqlite3_clear_files_heat (void *db_conn)
         ret = gf_sql_clear_counters (sql_conn);
         if (ret) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_CLEAR_COUNTER_FAILED, "Failed clearing "
-                        "files heat!");
+                        LG_MSG_CLEAR_COUNTER_FAILED, "Failed to clear "
+                        "files heat");
                 goto out;
         }
 
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3.h b/libglusterfs/src/gfdb/gfdb_sqlite3.h
index 2051d75fd8e..f2b252faa05 100644
--- a/libglusterfs/src/gfdb/gfdb_sqlite3.h
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3.h
@@ -56,7 +56,6 @@ do {\
                 );;\
 } 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
diff --git a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
index 0cc294a5410..0e9ccdbb380 100644
--- a/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
+++ b/libglusterfs/src/gfdb/gfdb_sqlite3_helper.c
@@ -1103,84 +1103,205 @@ 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;
+        int ret                                         = -1;
+        gfdb_query_record_t *query_record               = NULL;
+        char *text_column                               = NULL;
+        sqlite3 *db_conn                                = NULL;
+        uuid_t  prev_gfid                               = {0};
+        uuid_t  curr_gfid                               = {0};
+        uuid_t  pgfid                                   = {0};
+        char *base_name                                 = NULL;
+        gf_boolean_t is_first_record                    = _gf_true;
+        gf_boolean_t is_query_empty                     = _gf_true;
 
         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_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                                LG_MSG_CREATE_FAILED, "Failed to create "
-                                "gfdb_query_record");
-                                goto out;
-        }
-
-        /*Loop to access queried rows*/
+        /*
+         * Loop to access queried rows
+         * Each db record will have 3 columns
+         * GFID, PGFID, FILE_NAME
+         *
+         * For file with multiple hard links we will get multiple query rows
+         * with the same GFID, but different PGID and FILE_NAME Combination
+         * For Example if a file with
+         *         GFID = 00000000-0000-0000-0000-000000000006
+         * has 3 hardlinks file1, file2 and file3 in 3 different folder
+         * with GFID's
+         * 00000000-0000-0000-0000-0000EFC00001,
+         * 00000000-0000-0000-0000-00000ABC0001 and
+         * 00000000-0000-0000-0000-00000ABC00CD
+         * Then there will be 3 records
+         *         GFID         : 00000000-0000-0000-0000-000000000006
+         *         PGFID        : 00000000-0000-0000-0000-0000EFC00001
+         *         FILE_NAME    : file1
+         *
+         *         GFID         : 00000000-0000-0000-0000-000000000006
+         *         PGFID        : 00000000-0000-0000-0000-00000ABC0001
+         *         FILE_NAME    : file2
+         *
+         *         GFID         : 00000000-0000-0000-0000-000000000006
+         *         PGFID        : 00000000-0000-0000-0000-00000ABC00CD
+         *         FILE_NAME    : file3
+         *
+         * This is retrieved and added to a single query_record
+         *
+         * query_record->gfid = 00000000-0000-0000-0000-000000000006
+         *                  ->link_info = {00000000-0000-0000-0000-0000EFC00001,
+         *                                 "file1"}
+         *                                  |
+         *                                  V
+         *             link_info = {00000000-0000-0000-0000-00000ABC0001,
+         *                                 "file2"}
+         *                                  |
+         *                                  V
+         *             link_info = {00000000-0000-0000-0000-00000ABC0001,
+         *                                 "file3",
+         *                                 list}
+         *
+         * This query record is sent to the registered query_callback()
+         *
+         * */
         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*/
+                        is_query_empty = _gf_false;
+
+                        /*Retrieving GFID - column index is 0*/
                         text_column = (char *)sqlite3_column_text
                                                         (prep_stmt, 0);
                         if (!text_column) {
                                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                                        LG_MSG_GET_ID_FAILED, "Failed "
-                                        "retriving GF_ID");
+                                        LG_MSG_GET_ID_FAILED, "Failed to"
+                                        "retrieve GFID");
+                                goto out;
+                        }
+                        ret = gf_uuid_parse (text_column, curr_gfid);
+                        if (ret) {
+                                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
+                                        LG_MSG_PARSE_FAILED, "Failed to parse "
+                                        "GFID");
+                                goto out;
+                        }
+
+                        /*
+                         * if the previous record was not of the current gfid
+                         * call the call_back function and send the
+                         * query record, which will have all the link_info
+                         * objects associated with this gfid
+                         *
+                         * */
+                        if (gf_uuid_compare (curr_gfid, prev_gfid) != 0) {
+
+                                /* If this is not the first record */
+                                if (!is_first_record) {
+                                        /*Call the call_back function provided*/
+                                        ret = query_callback (query_record,
+                                                        _query_cbk_args);
+                                        if (ret) {
+                                                gf_msg (GFDB_STR_SQLITE3,
+                                                        GF_LOG_ERROR, 0,
+                                                LG_MSG_QUERY_CALL_BACK_FAILED,
+                                                        "Query call back "
+                                                        "failed");
+                                                goto out;
+                                        }
+
+                                }
+
+                                /*Clear the query record*/
+                                gfdb_query_record_free (query_record);
+                                query_record = NULL;
+                                query_record = gfdb_query_record_new ();
+                                if (!query_record) {
+                                        gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR,
+                                                0, LG_MSG_CREATE_FAILED,
+                                                "Failed to create "
+                                                "query_record");
+                                        goto out;
+                                }
+
+                                gf_uuid_copy(query_record->gfid,
+                                                                curr_gfid);
+                                gf_uuid_copy(prev_gfid, curr_gfid);
+
+                        }
+
+                        /* Get PGFID */
+                        text_column = (char *)sqlite3_column_text
+                                                        (prep_stmt, 1);
+                        if (!text_column) {
+                                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
+                                        LG_MSG_GET_ID_FAILED, "Failed to"
+                                        " retrieve GF_ID");
                                 goto out;
                         }
-                        ret = gf_uuid_parse (text_column, gfdb_query_record->gfid);
+                        ret = gf_uuid_parse (text_column, pgfid);
                         if (ret) {
                                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                                        LG_MSG_PARSE_FAILED, "Failed parsing "
+                                        LG_MSG_PARSE_FAILED, "Failed to parse "
                                         "GF_ID");
                                 goto out;
                         }
 
-                        /*Retrive Link Buffer - column index 1*/
+                        /* Get Base name */
                         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);
+                                                        (prep_stmt, 2);
+                        if (!text_column) {
+                                gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
+                                        LG_MSG_GET_ID_FAILED, "Failed to"
+                                        " retrieve GF_ID");
+                                goto out;
+                        }
+                        base_name = text_column;
+
+
+                        /* Add link info to the list */
+                        ret = gfdb_add_link_to_query_record (query_record,
+                                                         pgfid, base_name);
                         if (ret) {
                                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                                        LG_MSG_QUERY_CALL_BACK_FAILED,
-                                        "Query Call back failed!");
+                                        LG_MSG_GET_ID_FAILED, "Failed to"
+                                        " add link info to query record");
                                 goto out;
                         }
 
+                        is_first_record = _gf_false;
+
                 }
 
         }
 
         if (ret != SQLITE_DONE) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
-                        LG_MSG_GET_RECORD_FAILED, "Failed retriving records "
+                        LG_MSG_GET_RECORD_FAILED, "Failed to retrieve records "
                         "from db : %s", sqlite3_errmsg (db_conn));
                 ret = -1;
                 goto out;
         }
 
+
+        if (!is_query_empty) {
+                /*
+                 * Call the call_back function for the last record from the
+                 * Database
+                 * */
+                ret = query_callback (query_record, _query_cbk_args);
+                if (ret) {
+                        gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0,
+                                LG_MSG_QUERY_CALL_BACK_FAILED,
+                                "Query call back failed");
+                        goto out;
+                }
+        }
+
         ret = 0;
 out:
-        gfdb_query_record_fini (&gfdb_query_record);
+        gfdb_query_record_free (query_record);
+        query_record = NULL;
         return ret;
 }
 
@@ -1204,7 +1325,7 @@ gf_sql_clear_counters (gf_sql_connection_t *sql_conn)
                                 &sql_strerror);
         if (ret != SQLITE_OK) {
                 gf_msg (GFDB_STR_SQLITE3, GF_LOG_ERROR, 0, LG_MSG_EXEC_FAILED,
-                        "Failed executing: %s : %s",
+                        "Failed to execute: %s : %s",
                         query_str, sql_strerror);
                 sqlite3_free (sql_strerror);
                         ret = -1;