Add SQLITE_RT_BLOB to try writing to blob directlypoc2

5 files changed, 78 insertions, 1 deletions

diff --git a/database/engine/pagecache.c b/database/engine/pagecache.c
index 265e954f46..452a480376 100644
--- a/database/engine/pagecache.c
+++ b/database/engine/pagecache.c
@@ -89,6 +89,9 @@ struct rrdeng_page_descr *pg_cache_create_descr(void)
     descr->extent = NULL;
     descr->pg_cache_descr_state = 0;
     descr->pg_cache_descr = NULL;
+#ifdef SQLITE_RT_BLOB
+    descr->blob = NULL;
+#endif
 
     return descr;
 }
diff --git a/database/engine/pagecache.h b/database/engine/pagecache.h
index 31e9739da6..7ac869fc78 100644
--- a/database/engine/pagecache.h
+++ b/database/engine/pagecache.h
@@ -3,6 +3,7 @@
 #ifndef NETDATA_PAGECACHE_H
 #define NETDATA_PAGECACHE_H
 
+#include <sqlite3.h>
 #include "rrdengine.h"
 
 /* Forward declarations */
@@ -61,6 +62,9 @@ struct rrdeng_page_descr {
     usec_t start_time;
     usec_t end_time;
     uint32_t page_length;
+#ifdef SQLITE_RT_BLOB
+    sqlite3_blob *blob;
+#endif
 };
 
 #define PAGE_INFO_SCRATCH_SZ (8)
diff --git a/database/engine/rrdengineapi.c b/database/engine/rrdengineapi.c
index fbe960a435..1491dc1730 100755
--- a/database/engine/rrdengineapi.c
+++ b/database/engine/rrdengineapi.c
@@ -187,6 +187,9 @@ void rrdeng_store_metric_flush_current_page(RRDDIM *rd)
         return;
     }
     sql_add_metric_page(rd->state->metric_uuid, descr);
+#ifdef SQLITE_RT_BLOB
+    sqlite3_blob_close(descr->blob);
+#endif
     if (likely(descr->page_length)) {
         int ret, page_is_empty;
 
@@ -277,6 +280,22 @@ void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number n
     }
     page = descr->pg_cache_descr->page;
     page[descr->page_length / sizeof(number)] = number;
+#ifdef SQLITE_RT_BLOB
+    if (descr->blob) {
+        //info("Writing to BLOB at offset %d", descr->page_length);
+        int rc= sqlite3_blob_write(descr->blob, &number, sizeof(number), descr->page_length);
+        if (rc != SQLITE_OK)
+            info("Writing to BLOB at offset %d FAILED", descr->page_length);
+
+        storage_number new_number = 0;
+        rc = sqlite3_blob_read(descr->blob, &new_number, sizeof(new_number), descr->page_length);
+        if (rc != SQLITE_OK)
+            info("Reading from BLOB at offset %d FAILED", descr->page_length);
+
+        if (new_number != number)
+            info("BLOB numbers are NOT OK %d vs %d", number, new_number);
+    }
+#endif
     pg_cache_atomic_set_pg_info(descr, point_in_time, descr->page_length + sizeof(number));
 
     if (perfect_page_alignment)
@@ -704,8 +723,14 @@ void *rrdeng_create_page(struct rrdengine_instance *ctx, uuid_t *id, struct rrde
     void *page;
     /* TODO: check maximum number of pages in page cache limit */
 
+#ifdef SQLITE_RT_BLOB
+    sqlite3_blob *blob = (*ret_descr) ? (*ret_descr)->blob : sql_open_metric_blob(descr->id);
+#endif
     descr = pg_cache_create_descr();
     descr->id = id; /* TODO: add page type: metric, log, something? */
+#ifdef SQLITE_RT_BLOB
+    descr->blob = blob;
+#endif
     page = mallocz(RRDENG_BLOCK_SIZE); /*TODO: add page size */
     rrdeng_page_descr_mutex_lock(ctx, descr);
     pg_cache_descr = descr->pg_cache_descr;
diff --git a/database/sqlite/sqlite_functions.c b/database/sqlite/sqlite_functions.c
index 9676073cf9..f6fdeadec1 100644
--- a/database/sqlite/sqlite_functions.c
+++ b/database/sqlite/sqlite_functions.c
@@ -656,7 +656,11 @@ void sql_add_metric_page(uuid_t *dim_uuid, struct rrdeng_page_descr *descr)
 
 
     if (!res) {
-        rc = sqlite3_prepare_v2(db, "insert or replace into metric_update (dim_uuid, date_created) values (@dim_uuid, @date);", -1, &res, 0);
+#ifdef SQLITE_RT_BLOB
+        rc = sqlite3_prepare_v2(db, "insert into metric_update (dim_uuid, date_created, metric) values (@dim_uuid, @date, zeroblob(4096)) on conflict(dim_uuid) DO update set date_created=excluded.date_created;", -1, &res, 0);
+#else
+        rc = sqlite3_prepare_v2(db, "insert into metric_update (dim_uuid, date_created, metric) values (@dim_uuid, @date) on conflict(dim_uuid) DO update set date_created=excluded.date_created;", -1, &res, 0);
+#endif
         if (rc != SQLITE_OK) {
             info("SQLITE: Failed to prepare statement");
             return;
@@ -779,6 +783,46 @@ void sql_store_datafile_info(char *path, int fileno, size_t file_size)
     return;
 }
 
+sqlite3_blob *sql_open_metric_blob(uuid_t *dim_uuid)
+{
+    sqlite3_blob *blob;
+    static sqlite3_stmt *res = NULL;
+    int rc;
+
+    if (!db)
+        return NULL;
+
+    if (!res) {
+        rc = sqlite3_prepare_v2(db, "select rowid from metric_update where dim_uuid = @dim;", -1, &res, 0);
+        if (rc != SQLITE_OK)
+            return NULL;
+    }
+
+
+    rc = sqlite3_bind_blob(res, 1, dim_uuid, 16, SQLITE_TRANSIENT);
+    if (rc != SQLITE_OK) // Release the RES
+        return NULL;
+
+    sqlite3_int64 row = 0;
+    if ((rc = sqlite3_step(res)) == SQLITE_ROW)
+       row = sqlite3_column_int64(res, 0);
+    else
+        info("BLOB execution find row failed %d", rc);
+
+//    if (row != 2681)
+//        return NULL;
+
+    rc = sqlite3_blob_open(db, "main", "metric_update", "metric", row, 1, &blob);
+    if (rc != SQLITE_OK)
+        info("BLOB open failed");
+
+    char dim_str[37];
+    uuid_unparse_lower(dim_uuid, dim_str);
+    info("BLOB open for %s on line %lld", dim_str, row);
+    sqlite3_reset(res);
+    return blob;
+}
+
 void sql_store_page_info(uuid_t dim_uuid, int valid_page, int page_length, usec_t  start_time, usec_t end_time, int fileno, size_t offset, size_t size)
 {
     char sql[512];
diff --git a/database/sqlite/sqlite_functions.h b/database/sqlite/sqlite_functions.h
index 39f831010a..4ea917220b 100644
--- a/database/sqlite/sqlite_functions.h
+++ b/database/sqlite/sqlite_functions.h
@@ -44,6 +44,7 @@ extern void sql_compact_database();
 extern void sql_store_datafile_info(char *path, int fileno, size_t file_size);
 extern void sql_store_page_info(uuid_t temp_id, int valid_page, int page_length, usec_t  start_time, usec_t end_time, int , size_t offset, size_t size);
 extern void sql_add_metric_page_from_extent(struct rrdeng_page_descr *descr);
+extern struct sqlite3_blob *sql_open_metric_blob(uuid_t *dim_uuid);