1 files changed, 409 insertions, 349 deletions

diff --git a/database/rrddim.c b/database/rrddim.c
index 90165a25..1b3d9952 100644
--- a/database/rrddim.c
+++ b/database/rrddim.c
@@ -2,281 +2,117 @@
 
 #define NETDATA_RRD_INTERNALS
 #include "rrd.h"
-#ifdef ENABLE_DBENGINE
-#include "database/engine/rrdengineapi.h"
-#endif
 #include "storage_engine.h"
 
 // ----------------------------------------------------------------------------
 // RRDDIM index
 
-int rrddim_compare(void* a, void* b) {
-    if(((RRDDIM *)a)->hash < ((RRDDIM *)b)->hash) return -1;
-    else if(((RRDDIM *)a)->hash > ((RRDDIM *)b)->hash) return 1;
-    else return strcmp(((RRDDIM *)a)->id, ((RRDDIM *)b)->id);
-}
-
-#define rrddim_index_add(st, rd) (RRDDIM *)avl_insert_lock(&((st)->dimensions_index), (avl_t *)(rd))
-#define rrddim_index_del(st,rd ) (RRDDIM *)avl_remove_lock(&((st)->dimensions_index), (avl_t *)(rd))
-
-static inline RRDDIM *rrddim_index_find(RRDSET *st, const char *id, uint32_t hash) {
-    RRDDIM tmp = {
-            .id = id,
-            .hash = (hash)?hash:simple_hash(id)
-    };
-    return (RRDDIM *)avl_search_lock(&(st->dimensions_index), (avl_t *) &tmp);
-}
-
-
-// ----------------------------------------------------------------------------
-// RRDDIM - find a dimension
-
-inline RRDDIM *rrddim_find(RRDSET *st, const char *id) {
-    debug(D_RRD_CALLS, "rrddim_find() for chart %s, dimension %s", st->name, id);
-
-    return rrddim_index_find(st, id, 0);
-}
-
-
-// ----------------------------------------------------------------------------
-// RRDDIM rename a dimension
-
-inline int rrddim_set_name(RRDSET *st, RRDDIM *rd, const char *name) {
-    if(unlikely(!name || !*name || (rd->name && !strcmp(rd->name, name))))
-        return 0;
-
-    debug(D_RRD_CALLS, "rrddim_set_name() from %s.%s to %s.%s", st->name, rd->name, st->name, name);
-
-    if (rd->name)
-        freez((void *) rd->name);
-
-    rd->name = strdupz(name);
-    rd->hash_name = simple_hash(rd->name);
-
-    if (!st->state->is_ar_chart)
-        rrddimvar_rename_all(rd);
-
-    rd->exposed = 0;
-    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
-
-    ml_dimension_update_name(st, rd, name);
-
-    return 1;
-}
-
-inline int rrddim_set_algorithm(RRDSET *st, RRDDIM *rd, RRD_ALGORITHM algorithm) {
-    if(unlikely(rd->algorithm == algorithm))
-        return 0;
-
-    debug(D_RRD_CALLS, "Updating algorithm of dimension '%s/%s' from %s to %s", st->id, rd->name, rrd_algorithm_name(rd->algorithm), rrd_algorithm_name(algorithm));
-    rd->algorithm = algorithm;
-    rd->exposed = 0;
-    rrdset_flag_set(st, RRDSET_FLAG_HOMOGENEOUS_CHECK);
-    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
-    rrdcontext_updated_rrddim_algorithm(rd);
-    return 1;
-}
-
-inline int rrddim_set_multiplier(RRDSET *st, RRDDIM *rd, collected_number multiplier) {
-    if(unlikely(rd->multiplier == multiplier))
-        return 0;
+struct rrddim_constructor {
+    RRDSET *st;
+    const char *id;
+    const char *name;
+    collected_number multiplier;
+    collected_number divisor;
+    RRD_ALGORITHM algorithm;
+    RRD_MEMORY_MODE memory_mode;
+
+    enum {
+        RRDDIM_REACT_NONE    = 0,
+        RRDDIM_REACT_NEW     = (1 << 0),
+        RRDDIM_REACT_UPDATED = (1 << 2),
+    } react_action;
 
-    debug(D_RRD_CALLS, "Updating multiplier of dimension '%s/%s' from " COLLECTED_NUMBER_FORMAT " to " COLLECTED_NUMBER_FORMAT, st->id, rd->name, rd->multiplier, multiplier);
-    rd->multiplier = multiplier;
-    rd->exposed = 0;
-    rrdset_flag_set(st, RRDSET_FLAG_HOMOGENEOUS_CHECK);
-    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
-    rrdcontext_updated_rrddim_multiplier(rd);
-    return 1;
-}
-
-inline int rrddim_set_divisor(RRDSET *st, RRDDIM *rd, collected_number divisor) {
-    if(unlikely(rd->divisor == divisor))
-        return 0;
-
-    debug(D_RRD_CALLS, "Updating divisor of dimension '%s/%s' from " COLLECTED_NUMBER_FORMAT " to " COLLECTED_NUMBER_FORMAT, st->id, rd->name, rd->divisor, divisor);
-    rd->divisor = divisor;
-    rd->exposed = 0;
-    rrdset_flag_set(st, RRDSET_FLAG_HOMOGENEOUS_CHECK);
-    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
-    rrdcontext_updated_rrddim_divisor(rd);
-    return 1;
-}
-
-// ----------------------------------------------------------------------------
-// RRDDIM create a dimension
-
-void rrdcalc_link_to_rrddim(RRDDIM *rd, RRDSET *st, RRDHOST *host) {
-    RRDCALC *rrdc;
-    
-    for (rrdc = host->alarms_with_foreach; rrdc ; rrdc = rrdc->next) {
-        if (simple_pattern_matches(rrdc->spdim, rd->id) || simple_pattern_matches(rrdc->spdim, rd->name)) {
-            if (rrdc->hash_chart == st->hash_name || !strcmp(rrdc->chart, st->name) || !strcmp(rrdc->chart, st->id)) {
-                char *name = alarm_name_with_dim(rrdc->name, strlen(rrdc->name), rd->name, strlen(rd->name));
-                if(rrdcalc_exists(host, st->name, name, 0, 0)) {
-                    freez(name);
-                    continue;
-                }
-
-                netdata_rwlock_wrlock(&host->health_log.alarm_log_rwlock);
-                RRDCALC *child = rrdcalc_create_from_rrdcalc(rrdc, host, name, rd->name);
-                netdata_rwlock_unlock(&host->health_log.alarm_log_rwlock);
-
-                if (child) {
-                    rrdcalc_add_to_host(host, child);
-                    RRDCALC *rdcmp  = (RRDCALC *) avl_insert_lock(&(host)->alarms_idx_health_log,(avl_t *)child);
-                    if (rdcmp != child) {
-                        error("Cannot insert the alarm index ID %s",child->name);
-                    }
-                }
-                else {
-                    error("Cannot allocate a new alarm.");
-                    rrdc->foreachcounter--;
-                }
-            }
-        }
-    }
-}
-
-// Return either
-//   0 : Dimension is live
-//   last collected time : Dimension is not live
+};
 
-#ifdef ENABLE_ACLK
-time_t calc_dimension_liveness(RRDDIM *rd, time_t now)
-{
-    time_t last_updated = rd->last_collected_time.tv_sec;
-    int live;
-    if (rd->aclk_live_status == 1)
-        live =
-            ((now - last_updated) <
-             MIN(rrdset_free_obsolete_time, RRDSET_MINIMUM_DIM_OFFLINE_MULTIPLIER * rd->update_every));
-    else
-        live = ((now - last_updated) < RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * rd->update_every);
-    return live ? 0 : last_updated;
+// isolated call to appear
+// separate in statistics
+static void *rrddim_alloc_db(size_t entries) {
+    return callocz(entries, sizeof(storage_number));
 }
-#endif
 
-RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collected_number multiplier,
-                          collected_number divisor, RRD_ALGORITHM algorithm, RRD_MEMORY_MODE memory_mode)
-{
+static void rrddim_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *rrddim, void *constructor_data) {
+    struct rrddim_constructor *ctr = constructor_data;
+    RRDDIM *rd = rrddim;
+    RRDSET *st = ctr->st;
     RRDHOST *host = st->rrdhost;
-    rrdset_wrlock(st);
 
-    RRDDIM *rd = rrddim_find(st, id);
-    if(unlikely(rd)) {
-        debug(D_RRD_CALLS, "Cannot create rrd dimension '%s/%s', it already exists.", st->id, name?name:"<NONAME>");
-
-        int rc = rrddim_set_name(st, rd, name);
-        rc += rrddim_set_algorithm(st, rd, algorithm);
-        rc += rrddim_set_multiplier(st, rd, multiplier);
-        rc += rrddim_set_divisor(st, rd, divisor);
-
-        if (rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
-            store_active_dimension(&rd->metric_uuid);
-
-            for(int tier = 0; tier < storage_tiers ;tier++) {
-                if (rd->tiers[tier])
-                    rd->tiers[tier]->db_collection_handle =
-                        rd->tiers[tier]->collect_ops.init(rd->tiers[tier]->db_metric_handle);
-            }
-
-            rrddim_flag_clear(rd, RRDDIM_FLAG_ARCHIVED);
-            rrddimvar_create(rd, RRDVAR_TYPE_CALCULATED, NULL, NULL, &rd->last_stored_value, RRDVAR_OPTION_DEFAULT);
-            rrddimvar_create(rd, RRDVAR_TYPE_COLLECTED, NULL, "_raw", &rd->last_collected_value, RRDVAR_OPTION_DEFAULT);
-            rrddimvar_create(rd, RRDVAR_TYPE_TIME_T, NULL, "_last_collected_t", &rd->last_collected_time.tv_sec, RRDVAR_OPTION_DEFAULT);
+    rd->flags = RRDDIM_FLAG_NONE;
 
-            rrddim_flag_set(rd, RRDDIM_FLAG_PENDING_FOREACH_ALARM);
-            rrdset_flag_set(st, RRDSET_FLAG_PENDING_FOREACH_ALARMS);
-            rrdhost_flag_set(host, RRDHOST_FLAG_PENDING_FOREACH_ALARMS);
-        }
+    rd->id = string_strdupz(ctr->id);
+    rd->name = (ctr->name && *ctr->name)?rrd_string_strdupz(ctr->name):string_dup(rd->id);
 
-        if (unlikely(rc)) {
-            debug(D_METADATALOG, "DIMENSION [%s] metadata updated", rd->id);
-            (void)sql_store_dimension(&rd->metric_uuid, rd->rrdset->chart_uuid, rd->id, rd->name, rd->multiplier, rd->divisor,
-                                      rd->algorithm);
-#ifdef ENABLE_ACLK
-            queue_dimension_to_aclk(rd, calc_dimension_liveness(rd, now_realtime_sec()));
-#endif
-            rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
-            rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
-        }
-        rrdset_unlock(st);
-        rrdcontext_updated_rrddim(rd);
-        return rd;
-    }
-
-    rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
-    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
-
-    rd = callocz(1, sizeof(RRDDIM));
-    rd->id = strdupz(id);
-    rd->hash = simple_hash(rd->id);
-
-    rd->name = (name && *name)?strdupz(name):strdupz(rd->id);
-    rd->hash_name = simple_hash(rd->name);
-
-    rd->algorithm = algorithm;
-    rd->multiplier = multiplier;
-    rd->divisor = divisor;
+    rd->algorithm = ctr->algorithm;
+    rd->multiplier = ctr->multiplier;
+    rd->divisor = ctr->divisor;
     if(!rd->divisor) rd->divisor = 1;
 
-    rd->entries = st->entries;
     rd->update_every = st->update_every;
 
+    rd->rrdset = st;
+
     if(rrdset_flag_check(st, RRDSET_FLAG_STORE_FIRST))
         rd->collections_counter = 1;
 
-    rd->rrdset = st;
-
-    if(memory_mode == RRD_MEMORY_MODE_MAP || memory_mode == RRD_MEMORY_MODE_SAVE) {
-        if(!rrddim_memory_load_or_create_map_save(st, rd, memory_mode)) {
-            info("Failed to use memory mode %s for chart '%s', dimension '%s', falling back to ram", (memory_mode == RRD_MEMORY_MODE_MAP)?"map":"save", st->name, rd->name);
-            memory_mode = RRD_MEMORY_MODE_RAM;
+    if(ctr->memory_mode == RRD_MEMORY_MODE_MAP || ctr->memory_mode == RRD_MEMORY_MODE_SAVE) {
+        if(!rrddim_memory_load_or_create_map_save(st, rd, ctr->memory_mode)) {
+            info("Failed to use memory mode %s for chart '%s', dimension '%s', falling back to ram", (ctr->memory_mode == RRD_MEMORY_MODE_MAP)?"map":"save", rrdset_name(st), rrddim_name(rd));
+            ctr->memory_mode = RRD_MEMORY_MODE_RAM;
         }
     }
 
-    if(memory_mode == RRD_MEMORY_MODE_RAM) {
+    if(ctr->memory_mode == RRD_MEMORY_MODE_RAM) {
         size_t entries = st->entries;
         if(!entries) entries = 5;
 
         rd->db = netdata_mmap(NULL, entries * sizeof(storage_number), MAP_PRIVATE, 1);
         if(!rd->db) {
-            info("Failed to use memory mode ram for chart '%s', dimension '%s', falling back to alloc", st->name, rd->name);
-            memory_mode = RRD_MEMORY_MODE_ALLOC;
+            info("Failed to use memory mode ram for chart '%s', dimension '%s', falling back to alloc", rrdset_name(st), rrddim_name(rd));
+            ctr->memory_mode = RRD_MEMORY_MODE_ALLOC;
         }
         else rd->memsize = entries * sizeof(storage_number);
     }
 
-    if(memory_mode == RRD_MEMORY_MODE_ALLOC || memory_mode == RRD_MEMORY_MODE_NONE) {
+    if(ctr->memory_mode == RRD_MEMORY_MODE_ALLOC || ctr->memory_mode == RRD_MEMORY_MODE_NONE) {
         size_t entries = st->entries;
         if(entries < 5) entries = 5;
 
-        rd->db = callocz(entries, sizeof(storage_number));
+        rd->db = rrddim_alloc_db(entries);
         rd->memsize = entries * sizeof(storage_number);
     }
 
-    rd->rrd_memory_mode = memory_mode;
-
-#ifdef ENABLE_ACLK
-    rd->aclk_live_status = -1;
+    rd->rrd_memory_mode = ctr->memory_mode;
+
+    if (unlikely(rrdcontext_find_dimension_uuid(st, rrddim_id(rd), &(rd->metric_uuid)))) {
+        uuid_generate(rd->metric_uuid);
+        bool found_in_sql = false; (void)found_in_sql;
+
+//        bool found_in_sql = true;
+//        if(unlikely(sql_find_dimension_uuid(st, rd, &rd->metric_uuid))) {
+//            found_in_sql = false;
+//            uuid_generate(rd->metric_uuid);
+//        }
+
+#ifdef NETDATA_INTERNAL_CHECKS
+        char uuid_str[UUID_STR_LEN];
+        uuid_unparse_lower(rd->metric_uuid, uuid_str);
+        error_report("Dimension UUID for host %s chart [%s] dimension [%s] not found in context. It is now set to %s (%s)",
+                     string2str(host->hostname),
+                     string2str(st->name),
+                     string2str(rd->name),
+                     uuid_str, found_in_sql ? "found in sqlite" : "newly generated");
 #endif
-    (void) find_dimension_uuid(st, rd, &(rd->metric_uuid));
+    }
 
     // initialize the db tiers
     {
         size_t initialized = 0;
-        RRD_MEMORY_MODE wanted_mode = memory_mode;
-        for(int tier = 0; tier < storage_tiers ; tier++, wanted_mode = RRD_MEMORY_MODE_DBENGINE) {
-            STORAGE_ENGINE *eng = storage_engine_get(wanted_mode);
-            if(!eng) continue;
-
+        for(size_t tier = 0; tier < storage_tiers ; tier++) {
+            STORAGE_ENGINE *eng = host->db[tier].eng;
             rd->tiers[tier] = callocz(1, sizeof(struct rrddim_tier));
-            rd->tiers[tier]->tier_grouping = get_tier_grouping(tier);
-            rd->tiers[tier]->mode = eng->id;
-            rd->tiers[tier]->collect_ops = eng->api.collect_ops;
-            rd->tiers[tier]->query_ops = eng->api.query_ops;
-            rd->tiers[tier]->db_metric_handle = eng->api.init(rd, host->storage_instance[tier]);
+            rd->tiers[tier]->tier_grouping = host->db[tier].tier_grouping;
+            rd->tiers[tier]->collect_ops = &eng->api.collect_ops;
+            rd->tiers[tier]->query_ops = &eng->api.query_ops;
+            rd->tiers[tier]->db_metric_handle = eng->api.metric_get_or_create(rd, host->db[tier].instance, rd->rrdset->storage_metrics_groups[tier]);
             storage_point_unset(rd->tiers[tier]->virtual_point);
             initialized++;
 
@@ -284,138 +120,114 @@ RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collecte
         }
 
         if(!initialized)
-            error("Failed to initialize all db tiers for chart '%s', dimension '%s", st->name, rd->name);
+            error("Failed to initialize all db tiers for chart '%s', dimension '%s", rrdset_name(st), rrddim_name(rd));
 
         if(!rd->tiers[0])
-            error("Failed to initialize the first db tier for chart '%s', dimension '%s", st->name, rd->name);
+            error("Failed to initialize the first db tier for chart '%s', dimension '%s", rrdset_name(st), rrddim_name(rd));
     }
 
-    store_active_dimension(&rd->metric_uuid);
-
     // initialize data collection for all tiers
     {
         size_t initialized = 0;
-        for (int tier = 0; tier < storage_tiers; tier++) {
+        for (size_t tier = 0; tier < storage_tiers; tier++) {
             if (rd->tiers[tier]) {
-                rd->tiers[tier]->db_collection_handle = rd->tiers[tier]->collect_ops.init(rd->tiers[tier]->db_metric_handle);
+                rd->tiers[tier]->db_collection_handle = rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every);
                 initialized++;
             }
         }
 
         if(!initialized)
-            error("Failed to initialize data collection for all db tiers for chart '%s', dimension '%s", st->name, rd->name);
+            error("Failed to initialize data collection for all db tiers for chart '%s', dimension '%s", rrdset_name(st), rrddim_name(rd));
     }
 
-    // append this dimension
-    if(!st->dimensions)
-        st->dimensions = rd;
-    else {
-        RRDDIM *td = st->dimensions;
+    if(rrdset_number_of_dimensions(st) != 0) {
+        RRDDIM *td;
+        dfe_start_write(st->rrddim_root_index, td) {
+            if(!td) break;
+        }
+        dfe_done(td);
 
-        if(td->algorithm != rd->algorithm || ABS(td->multiplier) != ABS(rd->multiplier) || ABS(td->divisor) != ABS(rd->divisor)) {
+        if(td && (td->algorithm != rd->algorithm || ABS(td->multiplier) != ABS(rd->multiplier) || ABS(td->divisor) != ABS(rd->divisor))) {
             if(!rrdset_flag_check(st, RRDSET_FLAG_HETEROGENEOUS)) {
-                #ifdef NETDATA_INTERNAL_CHECKS
+#ifdef NETDATA_INTERNAL_CHECKS
                 info("Dimension '%s' added on chart '%s' of host '%s' is not homogeneous to other dimensions already present (algorithm is '%s' vs '%s', multiplier is " COLLECTED_NUMBER_FORMAT " vs " COLLECTED_NUMBER_FORMAT ", divisor is " COLLECTED_NUMBER_FORMAT " vs " COLLECTED_NUMBER_FORMAT ").",
-                        rd->name,
-                        st->name,
-                        host->hostname,
-                        rrd_algorithm_name(rd->algorithm), rrd_algorithm_name(td->algorithm),
-                        rd->multiplier, td->multiplier,
-                        rd->divisor, td->divisor
+                     rrddim_name(rd),
+                     rrdset_name(st),
+                     rrdhost_hostname(host),
+                     rrd_algorithm_name(rd->algorithm), rrd_algorithm_name(td->algorithm),
+                     rd->multiplier, td->multiplier,
+                     rd->divisor, td->divisor
                 );
-                #endif
+#endif
                 rrdset_flag_set(st, RRDSET_FLAG_HETEROGENEOUS);
             }
         }
-
-        for(; td->next; td = td->next) ;
-        td->next = rd;
-    }
-
-    if(host->health_enabled && !st->state->is_ar_chart) {
-        rrddimvar_create(rd, RRDVAR_TYPE_CALCULATED, NULL, NULL, &rd->last_stored_value, RRDVAR_OPTION_DEFAULT);
-        rrddimvar_create(rd, RRDVAR_TYPE_COLLECTED, NULL, "_raw", &rd->last_collected_value, RRDVAR_OPTION_DEFAULT);
-        rrddimvar_create(rd, RRDVAR_TYPE_TIME_T, NULL, "_last_collected_t", &rd->last_collected_time.tv_sec, RRDVAR_OPTION_DEFAULT);
     }
 
-    if(unlikely(rrddim_index_add(st, rd) != rd))
-        error("RRDDIM: INTERNAL ERROR: attempt to index duplicate dimension '%s' on chart '%s'", rd->id, st->id);
+    rrddim_flag_set(rd, RRDDIM_FLAG_PENDING_HEALTH_INITIALIZATION);
+    rrdset_flag_set(rd->rrdset, RRDSET_FLAG_PENDING_HEALTH_INITIALIZATION);
+    rrdhost_flag_set(rd->rrdset->rrdhost, RRDHOST_FLAG_PENDING_HEALTH_INITIALIZATION);
 
-    rrddim_flag_set(rd, RRDDIM_FLAG_PENDING_FOREACH_ALARM);
-    rrdset_flag_set(st, RRDSET_FLAG_PENDING_FOREACH_ALARMS);
-    rrdhost_flag_set(host, RRDHOST_FLAG_PENDING_FOREACH_ALARMS);
+    // let the chart resync
+    rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
 
     ml_new_dimension(rd);
 
-    rrdset_unlock(st);
-    rrdcontext_updated_rrddim(rd);
-    return(rd);
+    ctr->react_action = RRDDIM_REACT_NEW;
+
+    internal_error(false, "RRDDIM: inserted dimension '%s' of chart '%s' of host '%s'",
+                   rrddim_name(rd), rrdset_name(st), rrdhost_hostname(st->rrdhost));
+
 }
 
-// ----------------------------------------------------------------------------
-// RRDDIM remove / free a dimension
+static void rrddim_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *rrddim, void *rrdset) {
+    RRDDIM *rd = rrddim;
+    RRDSET *st = rrdset;
+    RRDHOST *host = st->rrdhost;
+
+    internal_error(false, "RRDDIM: deleting dimension '%s' of chart '%s' of host '%s'",
+                   rrddim_name(rd), rrdset_name(st), rrdhost_hostname(host));
 
-void rrddim_free(RRDSET *st, RRDDIM *rd)
-{
     rrdcontext_removed_rrddim(rd);
-    ml_delete_dimension(rd);
-    
-    debug(D_RRD_CALLS, "rrddim_free() %s.%s", st->name, rd->name);
 
-    if (!rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
+    ml_delete_dimension(rd);
 
-        size_t tiers_available = 0, tiers_said_yes = 0;
-        for(int tier = 0; tier < storage_tiers ;tier++) {
-            if(rd->tiers[tier]) {
-                tiers_available++;
+    debug(D_RRD_CALLS, "rrddim_free() %s.%s", rrdset_name(st), rrddim_name(rd));
 
-                if(rd->tiers[tier]->collect_ops.finalize(rd->tiers[tier]->db_collection_handle))
-                    tiers_said_yes++;
+    size_t tiers_available = 0, tiers_said_yes = 0;
+    for(size_t tier = 0; tier < storage_tiers ;tier++) {
+        if(rd->tiers[tier] && rd->tiers[tier]->db_collection_handle) {
+            tiers_available++;
 
-                rd->tiers[tier]->db_collection_handle = NULL;
-            }
-        }
+            if(rd->tiers[tier]->collect_ops->finalize(rd->tiers[tier]->db_collection_handle))
+                tiers_said_yes++;
 
-        if (tiers_available == tiers_said_yes && tiers_said_yes && rd->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
-            /* This metric has no data and no references */
-            delete_dimension_uuid(&rd->metric_uuid);
+            rd->tiers[tier]->db_collection_handle = NULL;
         }
     }
 
-    if(rd == st->dimensions)
-        st->dimensions = rd->next;
-    else {
-        RRDDIM *i;
-        for (i = st->dimensions; i && i->next != rd; i = i->next) ;
-
-        if (i && i->next == rd)
-            i->next = rd->next;
-        else
-            error("Request to free dimension '%s.%s' but it is not linked.", st->id, rd->name);
+    if (tiers_available == tiers_said_yes && tiers_said_yes && rd->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
+        /* This metric has no data and no references */
+        metaqueue_delete_dimension_uuid(&rd->metric_uuid);
     }
-    rd->next = NULL;
 
-    while(rd->variables)
-        rrddimvar_free(rd->variables);
-
-    if(unlikely(rrddim_index_del(st, rd) != rd))
-        error("RRDDIM: INTERNAL ERROR: attempt to remove from index dimension '%s' on chart '%s', removed a different dimension.", rd->id, st->id);
+    rrddimvar_delete_all(rd);
 
     // free(rd->annotations);
-//#ifdef ENABLE_ACLK
-//    if (!netdata_exit)
-//        aclk_send_dimension_update(rd);
-//#endif
+    //#ifdef ENABLE_ACLK
+    //    if (!netdata_exit)
+    //        aclk_send_dimension_update(rd);
+    //#endif
 
     // this will free MEMORY_MODE_SAVE and MEMORY_MODE_MAP structures
     rrddim_memory_file_free(rd);
 
-    for(int tier = 0; tier < storage_tiers ;tier++) {
+    for(size_t tier = 0; tier < storage_tiers ;tier++) {
         if(!rd->tiers[tier]) continue;
 
-        STORAGE_ENGINE* eng = storage_engine_get(rd->tiers[tier]->mode);
-        if(eng)
-            eng->api.free(rd->tiers[tier]->db_metric_handle);
+        STORAGE_ENGINE* eng = host->db[tier].eng;
+        eng->api.metric_release(rd->tiers[tier]->db_metric_handle);
 
         freez(rd->tiers[tier]);
         rd->tiers[tier] = NULL;
@@ -423,14 +235,252 @@ void rrddim_free(RRDSET *st, RRDDIM *rd)
 
     if(rd->db) {
         if(rd->rrd_memory_mode == RRD_MEMORY_MODE_RAM)
-            munmap(rd->db, rd->memsize);
+            netdata_munmap(rd->db, rd->memsize);
         else
             freez(rd->db);
     }
 
-    freez((void *)rd->id);
-    freez((void *)rd->name);
-    freez(rd);
+    string_freez(rd->id);
+    string_freez(rd->name);
+}
+
+static bool rrddim_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *rrddim, void *new_rrddim, void *constructor_data) {
+    (void)new_rrddim; // it is NULL
+
+    struct rrddim_constructor *ctr = constructor_data;
+    RRDDIM *rd = rrddim;
+    RRDSET *st = ctr->st;
+
+    ctr->react_action = RRDDIM_REACT_NONE;
+
+    int rc = rrddim_reset_name(st, rd, ctr->name);
+    rc += rrddim_set_algorithm(st, rd, ctr->algorithm);
+    rc += rrddim_set_multiplier(st, rd, ctr->multiplier);
+    rc += rrddim_set_divisor(st, rd, ctr->divisor);
+
+    for(size_t tier = 0; tier < storage_tiers ;tier++) {
+        if (rd->tiers[tier] && !rd->tiers[tier]->db_collection_handle)
+            rd->tiers[tier]->db_collection_handle =
+                rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every);
+    }
+
+    if(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
+        rrddim_flag_clear(rd, RRDDIM_FLAG_ARCHIVED);
+
+        rrddim_flag_set(rd, RRDDIM_FLAG_PENDING_HEALTH_INITIALIZATION);
+        rrdset_flag_set(rd->rrdset, RRDSET_FLAG_PENDING_HEALTH_INITIALIZATION);
+        rrdhost_flag_set(rd->rrdset->rrdhost, RRDHOST_FLAG_PENDING_HEALTH_INITIALIZATION);
+    }
+
+    if(unlikely(rc))
+        ctr->react_action = RRDDIM_REACT_UPDATED;
+
+    return ctr->react_action == RRDDIM_REACT_UPDATED;
+}
+
+static void rrddim_react_callback(const DICTIONARY_ITEM *item __maybe_unused, void *rrddim, void *constructor_data) {
+    struct rrddim_constructor *ctr = constructor_data;
+    RRDDIM *rd = rrddim;
+    RRDSET *st = ctr->st;
+
+    if(ctr->react_action & (RRDDIM_REACT_UPDATED | RRDDIM_REACT_NEW)) {
+        rrddim_flag_set(rd, RRDDIM_FLAG_METADATA_UPDATE);
+        rrdset_flag_set(rd->rrdset, RRDSET_FLAG_METADATA_UPDATE);
+        rrdhost_flag_set(rd->rrdset->rrdhost, RRDHOST_FLAG_METADATA_UPDATE);
+    }
+
+    if(ctr->react_action == RRDDIM_REACT_UPDATED) {
+        // the chart needs to be updated to the parent
+        rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
+        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
+    }
+
+    rrdcontext_updated_rrddim(rd);
+}
+
+void rrddim_index_init(RRDSET *st) {
+    if(!st->rrddim_root_index) {
+        st->rrddim_root_index = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);
+
+        dictionary_register_insert_callback(st->rrddim_root_index, rrddim_insert_callback, NULL);
+        dictionary_register_conflict_callback(st->rrddim_root_index, rrddim_conflict_callback, NULL);
+        dictionary_register_delete_callback(st->rrddim_root_index, rrddim_delete_callback, st);
+        dictionary_register_react_callback(st->rrddim_root_index, rrddim_react_callback, st);
+    }
+}
+
+void rrddim_index_destroy(RRDSET *st) {
+    dictionary_destroy(st->rrddim_root_index);
+    st->rrddim_root_index = NULL;
+}
+
+static inline RRDDIM *rrddim_index_find(RRDSET *st, const char *id) {
+    return dictionary_get(st->rrddim_root_index, id);
+}
+
+// ----------------------------------------------------------------------------
+// RRDDIM - find a dimension
+
+inline RRDDIM *rrddim_find(RRDSET *st, const char *id) {
+    debug(D_RRD_CALLS, "rrddim_find() for chart %s, dimension %s", rrdset_name(st), id);
+
+    return rrddim_index_find(st, id);
+}
+
+inline RRDDIM_ACQUIRED *rrddim_find_and_acquire(RRDSET *st, const char *id) {
+    debug(D_RRD_CALLS, "rrddim_find() for chart %s, dimension %s", rrdset_name(st), id);
+
+    return (RRDDIM_ACQUIRED *)dictionary_get_and_acquire_item(st->rrddim_root_index, id);
+}
+
+RRDDIM *rrddim_acquired_to_rrddim(RRDDIM_ACQUIRED *rda) {
+    if(unlikely(!rda))
+        return NULL;
+
+    return (RRDDIM *) dictionary_acquired_item_value((const DICTIONARY_ITEM *)rda);
+}
+
+void rrddim_acquired_release(RRDDIM_ACQUIRED *rda) {
+    if(unlikely(!rda))
+        return;
+
+    RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
+    dictionary_acquired_item_release(rd->rrdset->rrddim_root_index, (const DICTIONARY_ITEM *)rda);
+}
+
+// This will not return dimensions that are archived
+RRDDIM *rrddim_find_active(RRDSET *st, const char *id) {
+    RRDDIM *rd = rrddim_find(st, id);
+
+    if (unlikely(rd && rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)))
+        return NULL;
+
+    return rd;
+}
+
+// ----------------------------------------------------------------------------
+// RRDDIM rename a dimension
+
+inline int rrddim_reset_name(RRDSET *st, RRDDIM *rd, const char *name) {
+    if(unlikely(!name || !*name || !strcmp(rrddim_name(rd), name)))
+        return 0;
+
+    debug(D_RRD_CALLS, "rrddim_reset_name() from %s.%s to %s.%s", rrdset_name(st), rrddim_name(rd), rrdset_name(st), name);
+
+    STRING *old = rd->name;
+    rd->name = rrd_string_strdupz(name);
+    string_freez(old);
+
+    rrddimvar_rename_all(rd);
+
+    rd->exposed = 0;
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
+
+    return 1;
+}
+
+inline int rrddim_set_algorithm(RRDSET *st, RRDDIM *rd, RRD_ALGORITHM algorithm) {
+    if(unlikely(rd->algorithm == algorithm))
+        return 0;
+
+    debug(D_RRD_CALLS, "Updating algorithm of dimension '%s/%s' from %s to %s", rrdset_id(st), rrddim_name(rd), rrd_algorithm_name(rd->algorithm), rrd_algorithm_name(algorithm));
+    rd->algorithm = algorithm;
+    rd->exposed = 0;
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
+    rrdset_flag_set(st, RRDSET_FLAG_HOMOGENEOUS_CHECK);
+    rrdcontext_updated_rrddim_algorithm(rd);
+    return 1;
+}
+
+inline int rrddim_set_multiplier(RRDSET *st, RRDDIM *rd, collected_number multiplier) {
+    if(unlikely(rd->multiplier == multiplier))
+        return 0;
+
+    debug(D_RRD_CALLS, "Updating multiplier of dimension '%s/%s' from " COLLECTED_NUMBER_FORMAT " to " COLLECTED_NUMBER_FORMAT, rrdset_id(st), rrddim_name(rd), rd->multiplier, multiplier);
+    rd->multiplier = multiplier;
+    rd->exposed = 0;
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
+    rrdset_flag_set(st, RRDSET_FLAG_HOMOGENEOUS_CHECK);
+    rrdcontext_updated_rrddim_multiplier(rd);
+    return 1;
+}
+
+inline int rrddim_set_divisor(RRDSET *st, RRDDIM *rd, collected_number divisor) {
+    if(unlikely(rd->divisor == divisor))
+        return 0;
+
+    debug(D_RRD_CALLS, "Updating divisor of dimension '%s/%s' from " COLLECTED_NUMBER_FORMAT " to " COLLECTED_NUMBER_FORMAT, rrdset_id(st), rrddim_name(rd), rd->divisor, divisor);
+    rd->divisor = divisor;
+    rd->exposed = 0;
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
+    rrdset_flag_set(st, RRDSET_FLAG_HOMOGENEOUS_CHECK);
+    rrdcontext_updated_rrddim_divisor(rd);
+    return 1;
+}
+
+// ----------------------------------------------------------------------------
+
+// get the timestamp of the last entry in the round-robin database
+time_t rrddim_last_entry_t(RRDDIM *rd) {
+    time_t latest = rd->tiers[0]->query_ops->latest_time(rd->tiers[0]->db_metric_handle);
+
+    for(size_t tier = 1; tier < storage_tiers ;tier++) {
+        if(unlikely(!rd->tiers[tier])) continue;
+
+        time_t t = rd->tiers[tier]->query_ops->latest_time(rd->tiers[tier]->db_metric_handle);
+        if(t > latest)
+            latest = t;
+    }
+
+    return latest;
+}
+
+time_t rrddim_first_entry_t_of_tier(RRDDIM *rd, size_t tier) {
+    if(unlikely(tier > storage_tiers || !rd->tiers[tier]))
+        return 0;
+
+    return rd->tiers[tier]->query_ops->oldest_time(rd->tiers[tier]->db_metric_handle);
+}
+
+time_t rrddim_first_entry_t(RRDDIM *rd) {
+    time_t oldest = 0;
+
+    for(size_t tier = 0; tier < storage_tiers ;tier++) {
+        time_t t = rrddim_first_entry_t_of_tier(rd, tier);
+        if(t != 0 && (oldest == 0 || t < oldest))
+            oldest = t;
+    }
+
+    return oldest;
+}
+
+RRDDIM *rrddim_add_custom(RRDSET *st
+                          , const char *id
+                          , const char *name
+                          , collected_number multiplier
+                          , collected_number divisor
+                          , RRD_ALGORITHM algorithm
+                          , RRD_MEMORY_MODE memory_mode
+                          ) {
+    struct rrddim_constructor tmp = {
+        .st = st,
+        .id = id,
+        .name = name,
+        .multiplier = multiplier,
+        .divisor = divisor,
+        .algorithm = algorithm,
+        .memory_mode = memory_mode,
+    };
+
+    RRDDIM *rd = dictionary_set_advanced(st->rrddim_root_index, tmp.id, -1, NULL, sizeof(RRDDIM), &tmp);
+    return(rd);
+}
+
+// ----------------------------------------------------------------------------
+// RRDDIM remove / free a dimension
+
+void rrddim_free(RRDSET *st, RRDDIM *rd) {
+    dictionary_del(st->rrddim_root_index, string2str(rd->id));
 }
 
 
@@ -438,56 +488,59 @@ void rrddim_free(RRDSET *st, RRDDIM *rd)
 // RRDDIM - set dimension options
 
 int rrddim_hide(RRDSET *st, const char *id) {
-    debug(D_RRD_CALLS, "rrddim_hide() for chart %s, dimension %s", st->name, id);
+    debug(D_RRD_CALLS, "rrddim_hide() for chart %s, dimension %s", rrdset_name(st), id);
 
     RRDHOST *host = st->rrdhost;
 
     RRDDIM *rd = rrddim_find(st, id);
     if(unlikely(!rd)) {
-        error("Cannot find dimension with id '%s' on stats '%s' (%s) on host '%s'.", id, st->name, st->id, host->hostname);
+        error("Cannot find dimension with id '%s' on stats '%s' (%s) on host '%s'.", id, rrdset_name(st), rrdset_id(st), rrdhost_hostname(host));
         return 1;
     }
-    if (!rrddim_flag_check(rd, RRDDIM_FLAG_META_HIDDEN))
-        (void)sql_set_dimension_option(&rd->metric_uuid, "hidden");
+    if (!rrddim_flag_check(rd, RRDDIM_FLAG_META_HIDDEN)) {
+        rrddim_flag_set(rd, RRDDIM_FLAG_META_HIDDEN);
+        metaqueue_dimension_update_flags(rd);
+    }
 
-    rrddim_flag_set(rd, RRDDIM_FLAG_HIDDEN);
-    rrddim_flag_set(rd, RRDDIM_FLAG_META_HIDDEN);
+    rrddim_option_set(rd, RRDDIM_OPTION_HIDDEN);
     rrdcontext_updated_rrddim_flags(rd);
     return 0;
 }
 
 int rrddim_unhide(RRDSET *st, const char *id) {
-    debug(D_RRD_CALLS, "rrddim_unhide() for chart %s, dimension %s", st->name, id);
+    debug(D_RRD_CALLS, "rrddim_unhide() for chart %s, dimension %s", rrdset_name(st), id);
 
     RRDHOST *host = st->rrdhost;
     RRDDIM *rd = rrddim_find(st, id);
     if(unlikely(!rd)) {
-        error("Cannot find dimension with id '%s' on stats '%s' (%s) on host '%s'.", id, st->name, st->id, host->hostname);
+        error("Cannot find dimension with id '%s' on stats '%s' (%s) on host '%s'.", id, rrdset_name(st), rrdset_id(st), rrdhost_hostname(host));
         return 1;
     }
-    if (rrddim_flag_check(rd, RRDDIM_FLAG_META_HIDDEN))
-        (void)sql_set_dimension_option(&rd->metric_uuid, NULL);
+    if (rrddim_flag_check(rd, RRDDIM_FLAG_META_HIDDEN)) {
+        rrddim_flag_clear(rd, RRDDIM_FLAG_META_HIDDEN);
+        metaqueue_dimension_update_flags(rd);
+    }
 
-    rrddim_flag_clear(rd, RRDDIM_FLAG_HIDDEN);
-    rrddim_flag_clear(rd, RRDDIM_FLAG_META_HIDDEN);
+    rrddim_option_clear(rd, RRDDIM_OPTION_HIDDEN);
     rrdcontext_updated_rrddim_flags(rd);
     return 0;
 }
 
 inline void rrddim_is_obsolete(RRDSET *st, RRDDIM *rd) {
-    debug(D_RRD_CALLS, "rrddim_is_obsolete() for chart %s, dimension %s", st->name, rd->name);
+    debug(D_RRD_CALLS, "rrddim_is_obsolete() for chart %s, dimension %s", rrdset_name(st), rrddim_name(rd));
 
     if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED))) {
-        info("Cannot obsolete already archived dimension %s from chart %s", rd->name, st->name);
+        info("Cannot obsolete already archived dimension %s from chart %s", rrddim_name(rd), rrdset_name(st));
         return;
     }
     rrddim_flag_set(rd, RRDDIM_FLAG_OBSOLETE);
     rrdset_flag_set(st, RRDSET_FLAG_OBSOLETE_DIMENSIONS);
+    rrdhost_flag_set(st->rrdhost, RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS);
     rrdcontext_updated_rrddim_flags(rd);
 }
 
 inline void rrddim_isnot_obsolete(RRDSET *st __maybe_unused, RRDDIM *rd) {
-    debug(D_RRD_CALLS, "rrddim_isnot_obsolete() for chart %s, dimension %s", st->name, rd->name);
+    debug(D_RRD_CALLS, "rrddim_isnot_obsolete() for chart %s, dimension %s", rrdset_name(st), rrddim_name(rd));
 
     rrddim_flag_clear(rd, RRDDIM_FLAG_OBSOLETE);
     rrdcontext_updated_rrddim_flags(rd);
@@ -496,30 +549,34 @@ inline void rrddim_isnot_obsolete(RRDSET *st __maybe_unused, RRDDIM *rd) {
 // ----------------------------------------------------------------------------
 // RRDDIM - collect values for a dimension
 
-inline collected_number rrddim_set_by_pointer(RRDSET *st __maybe_unused, RRDDIM *rd, collected_number value) {
-    debug(D_RRD_CALLS, "rrddim_set_by_pointer() for chart %s, dimension %s, value " COLLECTED_NUMBER_FORMAT, st->name, rd->name, value);
+inline collected_number rrddim_set_by_pointer(RRDSET *st, RRDDIM *rd, collected_number value) {
+    struct timeval now;
+    now_realtime_timeval(&now);
+
+    return rrddim_timed_set_by_pointer(st, rd, now, value);
+}
 
-    rrdcontext_collected_rrddim(rd);
+collected_number rrddim_timed_set_by_pointer(RRDSET *st __maybe_unused, RRDDIM *rd, struct timeval collected_time, collected_number value) {
+    debug(D_RRD_CALLS, "rrddim_set_by_pointer() for chart %s, dimension %s, value " COLLECTED_NUMBER_FORMAT, rrdset_name(st), rrddim_name(rd), value);
 
-    now_realtime_timeval(&rd->last_collected_time);
+    rd->last_collected_time = collected_time;
     rd->collected_value = value;
     rd->updated = 1;
-
     rd->collections_counter++;
 
     collected_number v = (value >= 0) ? value : -value;
-    if(unlikely(v > rd->collected_value_max)) rd->collected_value_max = v;
-
-    // fprintf(stderr, "%s.%s %llu " COLLECTED_NUMBER_FORMAT " dt %0.6f" " rate " NETDATA_DOUBLE_FORMAT "\n", st->name, rd->name, st->usec_since_last_update, value, (float)((double)st->usec_since_last_update / (double)1000000), (NETDATA_DOUBLE)((value - rd->last_collected_value) * (NETDATA_DOUBLE)rd->multiplier / (NETDATA_DOUBLE)rd->divisor * 1000000.0 / (NETDATA_DOUBLE)st->usec_since_last_update));
+    if (unlikely(v > rd->collected_value_max))
+        rd->collected_value_max = v;
 
     return rd->last_collected_value;
 }
 
+
 collected_number rrddim_set(RRDSET *st, const char *id, collected_number value) {
     RRDHOST *host = st->rrdhost;
     RRDDIM *rd = rrddim_find(st, id);
     if(unlikely(!rd)) {
-        error("Cannot find dimension with id '%s' on stats '%s' (%s) on host '%s'.", id, st->name, st->id, host->hostname);
+        error("Cannot find dimension with id '%s' on stats '%s' (%s) on host '%s'.", id, rrdset_name(st), rrdset_id(st), rrdhost_hostname(host));
         return 0;
     }
 
@@ -560,7 +617,7 @@ struct rrddim_map_save_v019 {
     long double last_calculated_value;              // ignored
     long double last_stored_value;                  // ignored
     long long collected_value;                      // ignored
-    long long last_collected_value;                 // ignored
+    long long last_collected_value;                 // load and save
     long double collected_volume;                   // ignored
     long double stored_volume;                      // ignored
     void *next;                                     // ignored
@@ -578,22 +635,23 @@ size_t rrddim_memory_file_header_size(void) {
 }
 
 void rrddim_memory_file_update(RRDDIM *rd) {
-    if(!rd->rd_on_file) return;
+    if(!rd || !rd->rd_on_file) return;
     struct rrddim_map_save_v019 *rd_on_file = rd->rd_on_file;
 
     rd_on_file->last_collected_time.tv_sec = rd->last_collected_time.tv_sec;
     rd_on_file->last_collected_time.tv_usec = rd->last_collected_time.tv_usec;
+    rd_on_file->last_collected_value = rd->last_collected_value;
 }
 
 void rrddim_memory_file_free(RRDDIM *rd) {
-    if(!rd->rd_on_file) return;
+    if(!rd || !rd->rd_on_file) return;
 
     // needed for memory mode map, to save the latest state
     rrddim_memory_file_update(rd);
 
     struct rrddim_map_save_v019 *rd_on_file = rd->rd_on_file;
     freez(rd_on_file->cache_filename);
-    munmap(rd_on_file, rd_on_file->memsize);
+    netdata_munmap(rd_on_file, rd_on_file->memsize);
 
     // remove the pointers from the RRDDIM
     rd->rd_on_file = NULL;
@@ -601,13 +659,13 @@ void rrddim_memory_file_free(RRDDIM *rd) {
 }
 
 const char *rrddim_cache_filename(RRDDIM *rd) {
-    if(!rd->rd_on_file) return NULL;
+    if(!rd || !rd->rd_on_file) return NULL;
     struct rrddim_map_save_v019 *rd_on_file = rd->rd_on_file;
     return rd_on_file->cache_filename;
 }
 
 void rrddim_memory_file_save(RRDDIM *rd) {
-    if(!rd->rd_on_file) return;
+    if(!rd || !rd->rd_on_file) return;
 
     rrddim_memory_file_update(rd);
 
@@ -627,7 +685,7 @@ bool rrddim_memory_load_or_create_map_save(RRDSET *st, RRDDIM *rd, RRD_MEMORY_MO
 
     char filename[FILENAME_MAX + 1];
     char fullfilename[FILENAME_MAX + 1];
-    rrdset_strncpyz_name(filename, rd->id, FILENAME_MAX);
+    rrdset_strncpyz_name(filename, rrddim_id(rd), FILENAME_MAX);
     snprintfz(fullfilename, FILENAME_MAX, "%s/%s.db", st->cache_dir, filename);
 
     rd_on_file = (struct rrddim_map_save_v019 *)netdata_mmap(fullfilename, size,
@@ -646,7 +704,7 @@ bool rrddim_memory_load_or_create_map_save(RRDSET *st, RRDDIM *rd, RRD_MEMORY_MO
         reset = 1;
     }
     else if(rd_on_file->memsize != size) {
-        error("File %s does not have the desired size, expected %lu but found %lu. Clearing it.", fullfilename, size, rd_on_file->memsize);
+        error("File %s does not have the desired size, expected %lu but found %lu. Clearing it.", fullfilename, size, (unsigned long int) rd_on_file->memsize);
         memset(rd_on_file, 0, size);
         reset = 1;
     }
@@ -662,6 +720,8 @@ bool rrddim_memory_load_or_create_map_save(RRDSET *st, RRDDIM *rd, RRD_MEMORY_MO
     }
 
     if(!reset) {
+        rd->last_collected_value = rd_on_file->last_collected_value;
+
         if(rd_on_file->algorithm != rd->algorithm)
             info("File %s does not have the expected algorithm (expected %u '%s', found %u '%s'). Previous values may be wrong.",
                  fullfilename, rd->algorithm, rrd_algorithm_name(rd->algorithm), rd_on_file->algorithm, rrd_algorithm_name(rd_on_file->algorithm));