divide metric into manifest and data; cube.c

12 files changed, 1135 insertions, 951 deletions

diff --git a/include/fieldstat/fieldstat.h b/include/fieldstat/fieldstat.h
index be78ab4..a6101a8 100644
--- a/include/fieldstat/fieldstat.h
+++ b/include/fieldstat/fieldstat.h
@@ -215,13 +215,6 @@ struct fieldstat_tag_list *fieldstat_get_shared_tags(const struct fieldstat *ins
 */
 int fieldstat_find_cube(const struct fieldstat *instance, const struct fieldstat_tag *shared_tags, size_t n_shared_tags);
 
-/*
-    return FS_ERR_NULL_HANDLER, FS_ERR_INVALID_CUBE_ID if fail.
-    When return OK, mode == COMPREHENSIVE, primary_metric_id == -1.
-    When return OK, mode == TOPK, primary_metric_id >= 0.
-*/
-int fieldstat_get_cube_mode(const struct fieldstat *instance, int cube_id, enum sampling_mode *mode, int *primary_metric_id);
-
 /* 
     get the cell numbers in a cube. Return FS_ERR_INVALID_CUBE_ID if cube_id is invalid.
 */
diff --git a/src/cube.c b/src/cube.c
new file mode 100644
index 0000000..e5ebd19
--- /dev/null
+++ b/src/cube.c
@@ -0,0 +1,731 @@
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+
+#include "cube.h"
+#include "metric_manifest.h"
+#include "metric.h"
+#include "heavy_keeper.h"
+#include "tag_map.h"
+
+#define DEFAULT_N_METRIC 32
+
+struct exdata_new_args {
+    const struct fieldstat_tag *tags;
+    size_t n_tags;
+};
+
+struct cell {
+    struct metric **metrics;
+    size_t max_n_metric;
+    struct fieldstat_tag_list tags; // cell identifier
+};
+
+struct cube {
+    enum sampling_mode sampling_mode;
+    union {
+        struct heavy_keeper *topk;
+        struct tag_map *comprehensive;
+    };
+    size_t max_n_cell;
+
+    // the key of cube is the combination of shared tags
+    struct fieldstat_tag *cube_identifier;
+    size_t n_shared_tags;
+
+    int primary_metric_id;
+};
+
+static void tag_array_copy(struct fieldstat_tag *tags_dst, const struct fieldstat_tag *tags_src, size_t n_tag)
+{
+    for (size_t i = 0; i < n_tag; i++) {
+        tags_dst[i].key = strdup(tags_src[i].key);
+        tags_dst[i].type = tags_src[i].type;
+        switch (tags_src[i].type)
+        {
+        case TAG_INTEGER:
+            tags_dst[i].value_longlong = tags_src[i].value_longlong;
+            break;
+        case TAG_CSTRING:
+            tags_dst[i].value_str = strdup(tags_src[i].value_str);
+            break;
+        case TAG_DOUBLE:
+            tags_dst[i].value_double = tags_src[i].value_double;
+            break;
+        default:
+            break;
+        }
+    }
+}
+
+static void fieldstat_free_tag_array(struct fieldstat_tag *tags, size_t n_tags)
+{
+    for (size_t i = 0; i < n_tags; i++) {
+        struct fieldstat_tag *tag = &tags[i];
+        free((char *)tag->key);
+        if (tag->type == TAG_CSTRING) {
+            free((char *)tag->value_str);
+        }
+    }
+    free(tags);
+}
+
+
+struct metric *find_metric_in_cell(const struct cell *cell, int metric_id)
+{
+    if (metric_id >= cell->max_n_metric) {
+        return NULL;
+    }
+    return cell->metrics[metric_id];
+}
+
+void add_metric_to_cell(struct cell *cell, struct metric *metric, int metric_id)
+{
+    if (metric_id >= cell->max_n_metric) {
+        cell->metrics = realloc(cell->metrics, sizeof(struct metric *) * cell->max_n_metric * 2);
+        memset(cell->metrics + cell->max_n_metric, 0, sizeof(struct metric *) * cell->max_n_metric);
+        cell->max_n_metric *= 2;
+    }
+
+    cell->metrics[metric_id] = metric;
+}
+
+struct metric *add_or_find_metric_in_cell(const struct metric_manifest *manifest, struct cell *cell)
+{
+    struct metric *metric = find_metric_in_cell(cell, manifest->id);
+    if (metric != NULL) {
+        return metric;
+    }
+   
+    metric = metric_new(manifest);
+    add_metric_to_cell(cell, metric, manifest->id);
+    return metric;
+}
+
+struct cell *cell_new(const struct exdata_new_args *args) {
+    struct cell *pthis = malloc(sizeof(struct cell));
+    pthis->metrics = calloc(DEFAULT_N_METRIC, sizeof(struct metric *));
+    pthis->max_n_metric = DEFAULT_N_METRIC;
+
+    pthis->tags.n_tag = args->n_tags;
+    pthis->tags.tag = malloc(sizeof(struct fieldstat_tag) * args->n_tags);
+    tag_array_copy(pthis->tags.tag, args->tags, args->n_tags);
+    return pthis;
+}
+
+void cell_free(struct cell *pthis) {
+    for (size_t i = 0; i < pthis->max_n_metric; i++) {
+        metric_free(pthis->metrics[i]);
+    }
+    free(pthis->metrics);
+    for (size_t i = 0; i < pthis->tags.n_tag; i++) {
+        free((char *)pthis->tags.tag[i].key);
+        if (pthis->tags.tag[i].type == TAG_CSTRING) {
+            free((char *)pthis->tags.tag[i].value_str);
+        }
+    }
+    free(pthis->tags.tag);
+    free(pthis);
+}
+
+struct cell *cell_copy(const struct cell *src) {
+    struct cell *pthis = malloc(sizeof(struct cell));
+    pthis->metrics = calloc(src->max_n_metric, sizeof(struct metric *));
+    pthis->max_n_metric = src->max_n_metric;
+    for (size_t i = 0; i < src->max_n_metric; i++) {
+        if (src->metrics[i] == NULL) {
+            continue;
+        }
+
+        pthis->metrics[i] = metric_copy(src->metrics[i]);
+    }
+    
+    pthis->tags.n_tag = src->tags.n_tag;
+    pthis->tags.tag = malloc(sizeof(struct fieldstat_tag) * src->tags.n_tag);
+    tag_array_copy(pthis->tags.tag, src->tags.tag, src->tags.n_tag);
+
+    return pthis;
+}
+
+void cell_reset(struct cell *pthis) {
+    for (size_t i = 0; i < pthis->max_n_metric; i++) {
+        if (pthis->metrics[i] == NULL) {
+            continue;
+        }
+        metric_reset(pthis->metrics[i]);
+    }
+}
+
+void cell_merge(struct cell *dest, const struct cell *src) {
+    for (size_t i = 0; i < src->max_n_metric; i++) {
+        const struct metric *metric_src = src->metrics[i];
+        if (metric_src == NULL) {
+            continue;
+        }
+        struct metric *metric_dst = find_metric_in_cell(dest, i);
+
+        if (metric_dst == NULL) {
+            metric_dst = metric_copy(metric_src);
+            add_metric_to_cell(dest, metric_dst, i);
+        } else {
+            metric_merge(metric_dst, metric_src);
+        }
+    }
+}
+
+void *exdata_new_i(void *arg) {
+    return cell_new((struct exdata_new_args *)arg);
+}
+
+void exdata_free_i(void *exdata) {
+    cell_free((struct cell *)exdata);
+}
+
+void exdata_reset_i(void *exdata) {
+    cell_reset((struct cell *)exdata);
+}
+
+void exdata_merge_i(void *dest, void *src) {
+    cell_merge((struct cell *)dest, (struct cell *)src);
+}
+
+void *exdata_copy_i(void *exdata) {
+    return cell_copy((struct cell *)exdata);
+}
+
+struct cube *fieldstat_cube_info_init(const struct fieldstat_tag *shared_tags, size_t n_tag, enum sampling_mode mode, size_t max_n_cell)
+{
+    struct cube *cube = calloc(1, sizeof(struct cube));
+    cube->sampling_mode = mode;
+
+    if (n_tag == 0) {
+        cube->cube_identifier = NULL;
+    } else {
+        cube->cube_identifier = malloc(sizeof(struct fieldstat_tag) * n_tag);
+        tag_array_copy(cube->cube_identifier, shared_tags, n_tag);
+    }
+
+    cube->n_shared_tags = n_tag;
+
+    cube->max_n_cell = max_n_cell;
+
+    return cube;
+}
+
+struct cube *cube_new(const struct fieldstat_tag *shared_tags, size_t n_tag, enum sampling_mode mode, size_t max_n_cell)
+{
+    struct cube *cube = fieldstat_cube_info_init(shared_tags, n_tag, mode, max_n_cell);
+
+    switch (mode)
+    {
+        case SAMPLING_MODE_TOPK:
+            cube->topk = heavy_keeper_new(max_n_cell);
+            heavy_keeper_set_exdata_schema(cube->topk, exdata_new_i, exdata_free_i, exdata_merge_i, exdata_reset_i, exdata_copy_i);
+            break;
+        case SAMPLING_MODE_COMPREHENSIVE:
+            cube->comprehensive = tag_map_new(max_n_cell);
+            tag_map_set_exdata_schema(cube->comprehensive, exdata_new_i, exdata_free_i, exdata_merge_i, exdata_reset_i, exdata_copy_i);
+            break;
+        default:
+            assert(0);
+            break;
+    }
+
+    return cube;
+}
+
+void cube_free(struct cube *cube) {
+    switch (cube->sampling_mode)
+    {
+    case SAMPLING_MODE_TOPK:
+        heavy_keeper_free(cube->topk);
+        break;
+    case SAMPLING_MODE_COMPREHENSIVE:
+        tag_map_free(cube->comprehensive);
+        break;
+    default:
+        assert(0);
+        break;
+    }
+
+    fieldstat_free_tag_array(cube->cube_identifier, cube->n_shared_tags);
+
+    free(cube);
+}
+
+void cube_reset(struct cube *cube) {
+    if (cube->sampling_mode == SAMPLING_MODE_TOPK) {
+        heavy_keeper_reset(cube->topk);
+    } else {
+        tag_map_reset(cube->comprehensive);
+    }
+}
+
+void cube_set_primary_metric(struct cube *cube, int metric_id) {
+    cube->primary_metric_id = metric_id;
+}
+
+struct cell *find_or_add_exdata_comprehensive(struct tag_map *comprehensive, const char *key, size_t key_len, struct exdata_new_args *args)
+{
+    struct cell *cell_data = tag_map_get0_exdata(comprehensive, key, key_len);
+    if (cell_data == NULL) {
+        int tmp_ret = tag_map_add(comprehensive, key, key_len, (void *)args);
+        if (tmp_ret != 1) {
+            return NULL;
+        }
+        cell_data = tag_map_get0_exdata(comprehensive, key, key_len);
+    }
+    return cell_data;
+}
+
+struct cell *find_or_add_exdata_none_primary_topk(struct heavy_keeper *topk, const char *key, size_t key_len, struct exdata_new_args *args)
+{
+    struct cell *cell_data = heavy_keeper_get0_exdata(topk, key, key_len);
+    if (cell_data == NULL) {
+        int tmp_ret = heavy_keeper_add(topk, key, key_len, 0, (void *)args);
+        if (tmp_ret != 1) {
+            return NULL;
+        }
+        cell_data = heavy_keeper_get0_exdata(topk, key, key_len);
+    }
+    return cell_data;
+}
+
+
+int cube_histogram_record(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, long long value) {
+    char *tag_in_string;
+    size_t tag_len;
+    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
+
+    struct exdata_new_args args;
+    args.tags = tags;
+    args.n_tags = n_tag;
+
+    struct cell *cell_data = NULL;
+    switch (cube->sampling_mode) {
+    case SAMPLING_MODE_TOPK:
+        cell_data = find_or_add_exdata_none_primary_topk(cube->topk, tag_in_string, tag_len, &args);
+        break;
+    case SAMPLING_MODE_COMPREHENSIVE:
+        cell_data = find_or_add_exdata_comprehensive(cube->comprehensive, tag_in_string, tag_len, &args);
+        break;
+    }
+    free(tag_in_string);
+
+    if (cell_data == NULL) {
+        return FS_ERR_TOO_MANY_CELLS;
+    }
+    struct metric *metric = add_or_find_metric_in_cell(manifest, cell_data);
+
+    int ret = metric_histogram_record(metric, value);
+    if (ret < 0) {
+        return FS_ERR_INVALID_PARAM;
+    }
+    return FS_OK;
+}
+
+int cube_hll_add(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, const char *key, size_t key_len) {
+    char *tag_in_string;
+    size_t tag_len;
+    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
+
+    struct exdata_new_args args;
+    args.tags = tags;
+    args.n_tags = n_tag;
+
+    struct cell *cell_data = NULL;
+    switch (cube->sampling_mode) {
+    case SAMPLING_MODE_TOPK:
+        cell_data = find_or_add_exdata_none_primary_topk(cube->topk, tag_in_string, tag_len, &args);
+        break;
+    case SAMPLING_MODE_COMPREHENSIVE:
+        cell_data = find_or_add_exdata_comprehensive(cube->comprehensive, tag_in_string, tag_len, &args);
+        break;
+    }
+
+    if (cell_data == NULL) {
+        return FS_ERR_TOO_MANY_CELLS;
+    }
+    struct metric *metric = add_or_find_metric_in_cell(manifest, cell_data);
+
+    metric_hll_add(metric, key, key_len);
+    free(tag_in_string);
+
+    return FS_OK;
+}
+
+int cube_counter_incrby(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, long long increment) {
+    char *tag_in_string;
+    size_t tag_len;
+    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
+
+    struct exdata_new_args args;
+    args.tags = tags;
+    args.n_tags = n_tag;
+
+    struct cell *cell_data = NULL;
+    switch (cube->sampling_mode)
+    {
+    case SAMPLING_MODE_TOPK:
+        if (cube->primary_metric_id != manifest->id) {
+            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
+
+            if (cell_data == NULL) {
+                int tmp_ret = heavy_keeper_add(cube->topk, tag_in_string, tag_len, 0, &args);
+                if (tmp_ret != 1) {
+                    free(tag_in_string);
+                    return FS_ERR_TOO_MANY_CELLS;
+                }
+                cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
+            }
+        } else {
+            if (increment < 0) {
+                free(tag_in_string);
+                return FS_ERR_INVALID_PARAM;
+            } else if (increment == 0) {
+                free(tag_in_string);
+                return FS_OK;
+            }
+
+            // heavy_keeper_add should be called anyway, to let the topk record update.
+            int tmp_ret = heavy_keeper_add(cube->topk, tag_in_string, tag_len, increment, &args);
+            if (tmp_ret != 1) {
+                free(tag_in_string);
+                return FS_ERR_TOO_MANY_CELLS;
+            }
+            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
+        }
+        break;
+    case SAMPLING_MODE_COMPREHENSIVE:
+        cell_data = find_or_add_exdata_comprehensive(cube->comprehensive, tag_in_string, tag_len, &args);
+        if (cell_data == NULL) {
+            free(tag_in_string);
+            return FS_ERR_TOO_MANY_CELLS;
+        }
+        break;
+    default:
+        assert(0);
+        break;
+    }
+
+    struct metric *metric = add_or_find_metric_in_cell(manifest, cell_data);
+
+    metric_counter_incrby(metric, increment);
+    free(tag_in_string);
+    return FS_OK;
+}
+
+int cube_counter_set(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, long long value) {
+    char *tag_in_string;
+    size_t tag_len;
+    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
+
+    struct exdata_new_args args;
+    args.tags = tags;
+    args.n_tags = n_tag;
+    int metric_id = manifest->id;
+
+    struct cell *cell_data = NULL;
+    switch (cube->sampling_mode)
+    {
+    case SAMPLING_MODE_TOPK: { // TODO: 这个地方想办法拿到值以后进counter incrby 流程，重构
+        if (cube->primary_metric_id != metric_id) {
+            cell_data = find_or_add_exdata_none_primary_topk(cube->topk, tag_in_string, tag_len, &args);
+        } else {
+            long long current_count = 0;
+            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
+            if (cell_data != NULL) {
+                const struct metric *tmp_metric = find_metric_in_cell(cell_data, metric_id);
+                if (tmp_metric != NULL) {
+                    current_count = metric_counter_get(tmp_metric);
+                }
+            }
+            long long increment = value - current_count;
+            if (increment < 0) {
+                free(tag_in_string);
+                return FS_ERR_INVALID_PARAM;
+            } else if (increment == 0) {
+                free(tag_in_string);
+                return FS_OK;
+            }
+            
+            int tmp_ret = heavy_keeper_add(cube->topk, tag_in_string, tag_len, increment, &args);
+            if (tmp_ret != 1) {
+                return FS_ERR_TOO_MANY_CELLS;
+            }
+            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
+        }
+        break;}
+    case SAMPLING_MODE_COMPREHENSIVE: {
+        cell_data = find_or_add_exdata_comprehensive(cube->comprehensive, tag_in_string, tag_len, &args);
+        break;}
+    default:
+        assert(0);
+        break;
+    }
+
+    struct metric *metric = add_or_find_metric_in_cell(manifest, cell_data);
+    free(tag_in_string);
+    metric_counter_set(metric, value);
+    return FS_OK;
+}
+
+struct cube *cube_copy(const struct cube *cube)
+{
+    struct cube *cube_dup = fieldstat_cube_info_init(cube->cube_identifier, cube->n_shared_tags, cube->sampling_mode, cube->max_n_cell);
+    cube_dup->primary_metric_id = cube->primary_metric_id;
+
+    switch (cube->sampling_mode)
+    {
+    case SAMPLING_MODE_TOPK:
+        cube_dup->topk = heavy_keeper_copy(cube->topk);
+        break;
+    case SAMPLING_MODE_COMPREHENSIVE:
+        cube_dup->comprehensive = tag_map_copy(cube->comprehensive);
+        break;
+    default:
+        assert(0);
+        break;
+    }
+
+    return cube_dup;
+}
+
+void cube_merge(struct cube *dest, const struct cube *src)
+{
+    assert(dest->sampling_mode == src->sampling_mode);
+
+    switch (dest->sampling_mode)
+    {
+    case SAMPLING_MODE_TOPK:
+        heavy_keeper_merge(dest->topk, src->topk);
+        break;
+    case SAMPLING_MODE_COMPREHENSIVE:
+        tag_map_merge(dest->comprehensive, src->comprehensive);
+        break;
+    default:
+        assert(0);
+        break;
+    }
+}
+
+struct cube *cube_fork(const struct cube *cube) {
+    struct cube *ret = cube_new(cube->cube_identifier, cube->n_shared_tags, cube->sampling_mode, cube->max_n_cell);
+    ret->primary_metric_id = cube->primary_metric_id;
+
+    return ret;
+}
+
+void cube_get_cells(const struct cube *cube, struct fieldstat_tag_list **tag_list, size_t *n_cell)
+{
+    size_t n_cell_tmp = 0;
+    switch (cube->sampling_mode) {
+        case SAMPLING_MODE_COMPREHENSIVE:
+            n_cell_tmp = tag_map_get_count(cube->comprehensive);
+            break;
+        case SAMPLING_MODE_TOPK:
+            n_cell_tmp = heavy_keeper_get_count(cube->topk);
+            break;
+        default:
+            assert(0);
+    }
+
+    if (n_cell_tmp == 0) {
+        *tag_list = NULL;
+        *n_cell = 0;
+        return;
+    }
+
+    struct cell **cell_datas = (struct cell **)malloc(sizeof(struct cell *) * n_cell_tmp);
+    switch (cube->sampling_mode) {
+        case SAMPLING_MODE_COMPREHENSIVE:
+            tag_map_list(cube->comprehensive, (void **)cell_datas, n_cell_tmp);
+            break;
+        case SAMPLING_MODE_TOPK:
+            heavy_keeper_list(cube->topk, (void **)cell_datas, n_cell_tmp);
+            break;
+        default:
+            assert(0);
+    }
+
+    struct fieldstat_tag_list *tag_list_ret = (struct fieldstat_tag_list *)malloc(sizeof(struct fieldstat_tag_list) * n_cell_tmp);
+    *tag_list = tag_list_ret;
+    *n_cell = n_cell_tmp;
+
+    for (int i = 0; i < n_cell_tmp; i++) {
+        struct cell *cell_data = cell_datas[i];
+        struct fieldstat_tag_list *tag_list_tmp = &tag_list_ret[i];
+        tag_list_tmp->n_tag = cell_data->tags.n_tag;
+        if (tag_list_tmp->n_tag == 0) {
+            tag_list_tmp->tag = NULL;
+            continue;
+        }
+        tag_list_tmp->tag = (struct fieldstat_tag *)malloc(sizeof(struct fieldstat_tag) * tag_list_tmp->n_tag);
+        tag_array_copy(tag_list_tmp->tag, cell_data->tags.tag, tag_list_tmp->n_tag);
+    }
+
+    free(cell_datas);
+}
+
+const struct cell *get_cell_by_tag_list(const struct cube *cube, const struct fieldstat_tag_list *tags)
+{
+    const struct cell *ret = NULL;
+    char *tag_in_string;
+    size_t tag_len;
+    build_dynamic_cell_key(tags->tag, tags->n_tag, &tag_in_string, &tag_len);
+
+    switch (cube->sampling_mode)
+    {
+    case SAMPLING_MODE_TOPK:
+        ret = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
+        break;
+    case SAMPLING_MODE_COMPREHENSIVE:
+        ret = tag_map_get0_exdata(cube->comprehensive, tag_in_string, tag_len);
+        break;
+    default:
+        assert(0);
+        return NULL;
+    }
+    free(tag_in_string);
+
+    return ret;
+}
+
+const struct metric *get_metric_by_tag_list(const struct cube *cube, const struct fieldstat_tag_list *tags, int metric_id,int *ret)
+{
+    const struct cell *data = get_cell_by_tag_list(cube, tags);
+    
+    if (data == NULL) {
+        *ret = FS_ERR_INVALID_TAG;
+        return NULL;
+    }
+
+    if (metric_id < 0 || metric_id >= data->max_n_metric) {
+        *ret = FS_ERR_INVALID_METRIC_ID;
+        return NULL;
+    }
+    *ret = FS_OK;
+
+    return data->metrics[metric_id];
+}
+
+int cube_counter_get(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, long long *value)
+{
+    int ret;
+    const struct metric *metric = get_metric_by_tag_list(cube, tags, metric_id, &ret);
+    if (ret != FS_OK) {
+        return ret;
+    }
+    if (metric == NULL) {
+        return FS_ERR_INVALID_METRIC_ID;
+    }
+
+    *value = metric_counter_get(metric);
+    return FS_OK;
+}
+
+int cube_hll_get(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, double *value)
+{
+    int ret;
+    const struct metric *metric = get_metric_by_tag_list(cube, tags, metric_id, &ret);
+    if (ret != FS_OK) {
+        return ret;
+    }
+    if (metric == NULL) {
+        return FS_ERR_INVALID_METRIC_ID;
+    }
+
+    *value = metric_hll_get(metric);
+    return FS_OK;
+}
+
+int cube_histogram_value_at_percentile(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, double percentile, long long *value)
+{
+    int ret;
+    const struct metric *metric = get_metric_by_tag_list(cube, tags, metric_id, &ret);
+    if (ret != FS_OK) {
+        return ret;
+    }
+    if (metric == NULL) {
+        return FS_ERR_INVALID_METRIC_ID;
+    }
+
+    *value = metric_histogram_value_at_percentile(metric, percentile);
+    return FS_OK;
+}
+
+int cube_histogram_count_le_value(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, long long value, long long *count) {
+    int ret;
+    const struct metric *metric = get_metric_by_tag_list(cube, tags, metric_id, &ret);
+    if (ret != FS_OK) {
+        return ret;
+    }
+    if (metric == NULL) {
+        return FS_ERR_INVALID_METRIC_ID;
+    }
+
+    *count = metric_histogram_count_le_value(metric, value);
+    return FS_OK;
+}
+
+int cube_get_serialization(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, char **blob, size_t *blob_size) {
+    int ret;
+    const struct metric *metric = get_metric_by_tag_list(cube, tags, metric_id, &ret);
+    if (ret != FS_OK) {
+        return ret;
+    }
+    if (metric == NULL) {
+        return FS_ERR_INVALID_METRIC_ID;
+    }
+
+    metric_get_plain_blob(metric, blob, blob_size);
+    return FS_OK;
+}
+
+int cube_get_cell_count(const struct cube *cube) {
+    switch (cube->sampling_mode) {
+        case SAMPLING_MODE_COMPREHENSIVE:
+            return tag_map_get_count(cube->comprehensive);
+        case SAMPLING_MODE_TOPK:
+            return heavy_keeper_get_count(cube->topk);
+        default:
+            return FS_ERR_INVALID_PARAM;
+    }
+}
+
+void cube_get_cells_used_by_metric(const struct cube *cube, const struct fieldstat_tag_list *tags, int **metric_id_out, size_t *n_metric_out) {
+    const struct cell *cell_data = get_cell_by_tag_list(cube, tags);
+    if (cell_data == NULL) {
+        *metric_id_out = NULL;
+        *n_metric_out = 0;
+        return;
+    }
+
+    *metric_id_out = (int *)malloc(sizeof(int) * cell_data->max_n_metric);
+    int n_metric = 0;
+    for (int i = 0; i < cell_data->max_n_metric; i++) {
+        if (cell_data->metrics[i] != NULL) {
+            (*metric_id_out)[n_metric] = i;
+            n_metric++;
+        }
+    }
+    *n_metric_out = n_metric;
+}
+
+struct fieldstat_tag_list *cube_get_identifier(const struct cube *cube) {
+    struct fieldstat_tag_list *tag_list = (struct fieldstat_tag_list *)malloc(sizeof(struct fieldstat_tag_list));
+
+    if (cube->n_shared_tags == 0) {
+        tag_list->tag = NULL;
+        tag_list->n_tag = 0;
+        return tag_list;
+    }
+
+    tag_list->tag = (struct fieldstat_tag *)malloc(sizeof(struct fieldstat_tag) * cube->n_shared_tags);
+    tag_list->n_tag = cube->n_shared_tags;
+    tag_array_copy(tag_list->tag, cube->cube_identifier, cube->n_shared_tags);
+
+    return tag_list;
+}
\ No newline at end of file
diff --git a/src/cube.h b/src/cube.h
new file mode 100644
index 0000000..71b83f4
--- /dev/null
+++ b/src/cube.h
@@ -0,0 +1,41 @@
+#pragma once
+#include <stdio.h>
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+#include <stddef.h>
+#include <stdbool.h>
+#include "fieldstat.h" // for tags
+#include "metric_manifest.h"
+
+struct cube;
+
+struct cube *cube_new(const struct fieldstat_tag *shared_tags, size_t n_tag, enum sampling_mode mode, size_t max_n_cell);
+void cube_free(struct cube *cube);
+void cube_reset(struct cube *cube);
+struct cube *cube_copy(const struct cube *cube);
+void cube_merge(struct cube *dest, const struct cube *src);
+struct cube *cube_fork(const struct cube *cube); // only copy the cube configurations, leave the cells empty
+
+int cube_histogram_record(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, long long value);
+int cube_hll_add(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, const char *key, size_t key_len);
+int cube_counter_incrby(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, long long increment);
+int cube_counter_set(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, long long value);
+
+int cube_counter_get(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, long long *value);
+int cube_hll_get(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, double *value);
+int cube_histogram_value_at_percentile(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, double percentile, long long *value);
+int cube_histogram_count_le_value(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, long long value, long long *count);
+int cube_get_serialization(const struct cube *cube, int metric_id, const struct fieldstat_tag_list *tags, char **blob, size_t *blob_size);
+
+int cube_get_cell_count(const struct cube *cube);
+void cube_get_cells(const struct cube *cube, struct fieldstat_tag_list **tag_list, size_t *n_cell);
+void cube_get_cells_used_by_metric(const struct cube *cube, const struct fieldstat_tag_list *tags, int **metric_id_out, size_t *n_metric_out);
+void cube_set_primary_metric(struct cube *cube, int metric_id);
+struct fieldstat_tag_list *cube_get_identifier(const struct cube *cube);
+
+#ifdef __cplusplus
+}
+#endif
\ No newline at end of file
diff --git a/src/fieldstat.c b/src/fieldstat.c
index 5e2d3b1..71a1e0a 100644
--- a/src/fieldstat.c
+++ b/src/fieldstat.c
@@ -3,6 +3,7 @@
 #include <assert.h>
 #include <stdlib.h>
 #include <stdio.h>
+#include <stdarg.h>
 
 #include "cjson/cJSON.h"
 #include "uthash.h"
@@ -10,69 +11,73 @@
 #include "fieldstat.h"
 #include "metrics/metric.h"
 #include "cell_manager.h"
-#include "heavy_keeper.h"
-#include "tag_map.h"
 #include "my_ut_hash.h"
+#include "cube.h"
+#include "metric_manifest.h"
 
 #define DEFAULT_N_METRIC 32
 #define DEFAULT_N_CUBE 64
 
 
-struct cell {
-    struct metric **metrics;
-    size_t max_n_metric;
-    struct fieldstat_tag_list tags; // cell identifier
-};
-
-struct cube {
-    enum sampling_mode sampling_mode;
-    union {
-        struct heavy_keeper *topk;
-        struct tag_map *comprehensive;
-    };
-    size_t max_n_cell;
-
-    // the key of cube is the combination of shared tags
-    struct fieldstat_tag *cube_identifier;
-    size_t n_shared_tags;
-    struct tag_hash_key *key_tag; // TODO: 删了它
-    // const char *key_tag;
-
-    int primary_metric_id;
-};
-
-struct metric_name_id_item {
-    char *name;
-    int id;
-    UT_hash_handle hh;
-};
-
-struct metric_name_id_map {
-    struct metric_name_id_item *map;
-};
-
-struct exdata_new_args {
-    const struct fieldstat_tag *tags;
-    size_t n_tags;
-};
-
 struct fieldstat {
     struct cube **cube;
     unsigned long *cube_version; // increase from 0 every time the cube is deleted
     unsigned long cell_version; // increase from 0 every time fieldstat_reset is called
     size_t max_n_cube;
 
-    struct metric **metric_masters;
+    struct metric_manifest **metric_masters;
     size_t max_n_metric_master;
-    struct metric_name_id_map *metric_name_id_map;
 
     struct cube_manager *shared_tag_cube_manager;
 };
 
 
-/* -------------------------------------------------------------------------- */
-/*                              metric and exdata                             */
-/* -------------------------------------------------------------------------- */
+union metric_parameter *construct_parameters(enum metric_type type, ...)
+{
+    union metric_parameter *paras = (union metric_parameter *)malloc(sizeof(union metric_parameter));
+    va_list ap;
+    va_start(ap, type);
+    switch (type) {
+        case METRIC_TYPE_COUNTER:
+            break;
+        case METRIC_TYPE_HLL:
+            paras->hll.precision = (char)va_arg(ap, int);
+            break;
+        case METRIC_TYPE_HISTOGRAM:
+            paras->hdr.lowest_trackable_value = va_arg(ap, long long);
+            paras->hdr.highest_trackable_value = va_arg(ap, long long);
+            paras->hdr.significant_figures = va_arg(ap, int);
+            break;
+        default:
+            assert(0);
+    }
+    va_end(ap);
+    return paras;
+}
+
+struct metric_manifest *metric_manifest_copy(const struct metric_manifest *src)
+{
+    struct metric_manifest *dest = (struct metric_manifest *)malloc(sizeof(struct metric_manifest));
+    dest->name = strdup(src->name);
+    if (src->paras == NULL) {
+        dest->paras = NULL;
+    } else {
+        dest->paras = (union metric_parameter *)malloc(sizeof(union metric_parameter));
+        memcpy(dest->paras, src->paras, sizeof(union metric_parameter));
+    }
+
+    dest->type = src->type;
+    dest->id = src->id;
+
+    return dest;
+}
+
+void metric_manifest_free(struct metric_manifest *manifest) {
+    free(manifest->name);
+    free(manifest->paras);
+    free(manifest);
+}
+
 void tag_array_copy(struct fieldstat_tag *tags_dst, const struct fieldstat_tag *tags_src, size_t n_tag)
 {
     for (size_t i = 0; i < n_tag; i++) {
@@ -95,169 +100,51 @@ void tag_array_copy(struct fieldstat_tag *tags_dst, const struct fieldstat_tag *
     }
 }
 
-int name_id_map_get_id_by_name(const struct metric_name_id_map *map, const char *metric_name)
-{
-    struct metric_name_id_item *entry = NULL;
-    HASH_FIND_STR(map->map, metric_name, entry);
-    if (entry == NULL) {
-        return -1;
-    }
-    return entry->id;
-}
-
-void name_id_map_add(struct metric_name_id_map *map, const char *name, int id) {
-    struct metric_name_id_item *entry = malloc(sizeof(struct metric_name_id_item));
-    entry->id = id;
-    entry->name = strdup(name);
-    HASH_ADD_KEYPTR(hh, map->map, entry->name, strlen(entry->name), entry);
-}
-
-void name_id_map_free(struct metric_name_id_map *map)
-{
-    struct metric_name_id_item *entry, *tmp;
-    HASH_ITER(hh, map->map, entry, tmp) {
-        HASH_DEL(map->map, entry);
-        free(entry->name);
-        free(entry);
-    }
-
-    free(map);
-}
-
-struct metric_name_id_map *name_id_map_copy(struct metric_name_id_map *map)
+bool is_tag_array_same(const struct fieldstat_tag *tags1, const struct fieldstat_tag *tags2, size_t n_tag)
 {
-    struct metric_name_id_map *map_dup = malloc(sizeof(struct metric_name_id_map));
-    map_dup->map = NULL;
-    struct metric_name_id_item *entry, *tmp;
-    HASH_ITER(hh, map->map, entry, tmp) {
-        name_id_map_add(map_dup, entry->name, entry->id);
-    }
-    return map_dup;
-}
-
-void add_metric_to_cell(struct cell *cell, struct metric *metric, int metric_id)
-{
-    if (metric_id >= cell->max_n_metric) {
-        cell->metrics = realloc(cell->metrics, sizeof(struct metric *) * cell->max_n_metric * 2);
-        memset(cell->metrics + cell->max_n_metric, 0, sizeof(struct metric *) * cell->max_n_metric);
-        cell->max_n_metric *= 2;
-    }
-
-    cell->metrics[metric_id] = metric;
-}
-
-struct metric *find_metric_in_cell(const struct cell *cell, int metric_id)
-{
-    if (metric_id >= cell->max_n_metric) {
-        return NULL;
-    }
-    return cell->metrics[metric_id];
-}
-
-struct metric *add_or_find_metric_in_cell(struct fieldstat *instance, struct cell *cell, int metric_id)
-{
-    struct metric *metric = find_metric_in_cell(cell, metric_id);
-    if (metric != NULL) {
-        return metric;
-    }
-   
-    metric = metric_fork(instance->metric_masters[metric_id]);
-    add_metric_to_cell(cell, metric, metric_id);
-    return metric;
-}
-
-struct cell *cell_new(const struct exdata_new_args *args) {
-    struct cell *pthis = malloc(sizeof(struct cell));
-    pthis->metrics = calloc(DEFAULT_N_METRIC, sizeof(struct metric *));
-    pthis->max_n_metric = DEFAULT_N_METRIC;
-
-    pthis->tags.n_tag = args->n_tags;
-    pthis->tags.tag = malloc(sizeof(struct fieldstat_tag) * args->n_tags);
-    tag_array_copy(pthis->tags.tag, args->tags, args->n_tags);
-    return pthis;
-}
-
-void cell_free(struct cell *pthis) {
-    for (size_t i = 0; i < pthis->max_n_metric; i++) {
-        metric_free(pthis->metrics[i]);
-    }
-    free(pthis->metrics);
-    for (size_t i = 0; i < pthis->tags.n_tag; i++) {
-        free((char *)pthis->tags.tag[i].key);
-        if (pthis->tags.tag[i].type == TAG_CSTRING) {
-            free((char *)pthis->tags.tag[i].value_str);
+    for (size_t i = 0; i < n_tag; i++) {
+        if (tags1[i].type != tags2[i].type) {
+            return false;
         }
-    }
-    free(pthis->tags.tag);
-    free(pthis);
-}
-
-struct cell *cell_copy(const struct cell *src) {
-    struct cell *pthis = malloc(sizeof(struct cell));
-    pthis->metrics = calloc(src->max_n_metric, sizeof(struct metric *));
-    pthis->max_n_metric = src->max_n_metric;
-    for (size_t i = 0; i < src->max_n_metric; i++) {
-        if (src->metrics[i] == NULL) {
-            continue;
+        if (strcmp(tags1[i].key, tags2[i].key) != 0) {
+            return false;
         }
-
-        pthis->metrics[i] = metric_copy(src->metrics[i]);
-    }
-    
-    pthis->tags.n_tag = src->tags.n_tag;
-    pthis->tags.tag = malloc(sizeof(struct fieldstat_tag) * src->tags.n_tag);
-    tag_array_copy(pthis->tags.tag, src->tags.tag, src->tags.n_tag);
-
-    return pthis;
-}
-
-void cell_reset(struct cell *pthis) {
-    for (size_t i = 0; i < pthis->max_n_metric; i++) {
-        if (pthis->metrics[i] == NULL) {
-            continue;
+        switch (tags1[i].type)
+        {
+        case TAG_INTEGER:
+            if (tags1[i].value_longlong != tags2[i].value_longlong) {
+                return false;
+            }
+            break;
+        case TAG_CSTRING:
+            if (strcmp(tags1[i].value_str, tags2[i].value_str) != 0) {
+                return false;
+            }
+            break;
+        case TAG_DOUBLE:
+            if (tags1[i].value_double != tags2[i].value_double) {
+                return false;
+            }
+            break;
+        default:
+            break;
         }
-        metric_reset(pthis->metrics[i]);
     }
+    return 0;
 }
 
-void cell_merge(struct cell *dest, const struct cell *src) {
-    for (size_t i = 0; i < src->max_n_metric; i++) {
-        const struct metric *metric_src = src->metrics[i];
-        if (metric_src == NULL) {
+bool is_metric_name_duplicate(const struct fieldstat *instance, const char *name)
+{
+    for (size_t i = 0; i < instance->max_n_metric_master; i++) {
+        if (instance->metric_masters[i] == NULL) {
             continue;
         }
-        struct metric *metric_dst = find_metric_in_cell(dest, i);
-        assert(strcmp(metric_get_name(metric_src), metric_get_name(metric_dst)) == 0);
-
-        if (metric_dst == NULL) {
-            metric_dst = metric_copy(metric_src);
-            add_metric_to_cell(dest, metric_dst, i);
-        } else {
-            metric_merge(metric_dst, metric_src);
+        if (strcmp(instance->metric_masters[i]->name, name) == 0) {
+            return true;
         }
     }
+    return false;
 }
-
-void *exdata_new_i(void *arg) {
-    return cell_new((struct exdata_new_args *)arg);
-}
-
-void exdata_free_i(void *exdata) {
-    cell_free((struct cell *)exdata);
-}
-
-void exdata_reset_i(void *exdata) {
-    cell_reset((struct cell *)exdata);
-}
-
-void exdata_merge_i(void *dest, void *src) {
-    cell_merge((struct cell *)dest, (struct cell *)src);
-}
-
-void *exdata_copy_i(void *exdata) {
-    return cell_copy((struct cell *)exdata);
-}
-
 /* -------------------------------------------------------------------------- */
 /*                                  fieldstat                                 */
 /* -------------------------------------------------------------------------- */
@@ -271,8 +158,7 @@ struct fieldstat *fieldstat_new()
     instance->cube_version = calloc(instance->max_n_cube, sizeof(unsigned long));
 
     instance->max_n_metric_master = DEFAULT_N_METRIC;
-    instance->metric_masters = calloc(instance->max_n_metric_master, sizeof(struct metric *));
-    instance->metric_name_id_map = calloc(1, sizeof(struct metric_name_id_map));
+    instance->metric_masters = calloc(instance->max_n_metric_master, sizeof(struct metric_manifest *));
 
     instance->shared_tag_cube_manager = cube_manager_new();
 
@@ -293,10 +179,11 @@ void fieldstat_free(struct fieldstat *instance)
     cube_manager_free(instance->shared_tag_cube_manager);
 
     for (size_t i = 0; i < instance->max_n_metric_master; i++) {
-        metric_free(instance->metric_masters[i]);
+        if (instance->metric_masters[i] != NULL) {
+            metric_manifest_free(instance->metric_masters[i]);
+        }
     }
     free(instance->metric_masters);
-    name_id_map_free(instance->metric_name_id_map);
 
     free(instance);
 }
@@ -312,11 +199,7 @@ void fieldstat_reset(struct fieldstat *instance)
             continue;
         }
 
-        if (cube->sampling_mode == SAMPLING_MODE_TOPK) {
-            heavy_keeper_reset(cube->topk);
-        } else {
-            tag_map_reset(cube->comprehensive);
-        }
+        cube_reset(cube);
     }
     instance->cell_version++;
 }
@@ -342,12 +225,17 @@ int fieldstat_destroy_cube(struct fieldstat *instance, int cube_id)
     }
 
     const struct cube *cube = instance->cube[cube_id];
-    cube_manager_delete(instance->shared_tag_cube_manager, cube->key_tag);
+    struct fieldstat_tag_list *shared_tag = cube_get_identifier(cube);
+    struct tag_hash_key shared_tag_key;
+    tag_hash_key_init_with_fieldstat_tag(&shared_tag_key, shared_tag->tag, shared_tag->n_tag, false);
+    cube_manager_delete(instance->shared_tag_cube_manager, &shared_tag_key);
 
     fieldstat_cube_free(instance, cube_id);
 
     instance->cube[cube_id] = NULL;
     instance->cube_version[cube_id]++;
+
+    fieldstat_tag_list_arr_free(shared_tag, 1);
     return 0;
 }
 
@@ -405,62 +293,16 @@ int fieldstat_append_cube_to_instance(struct fieldstat *instance, struct cube *c
     for (int i = 0; i < instance->max_n_cube; i++) {
         if (instance->cube[i] == NULL) {
             instance->cube[i] = cube;
-            cube_manager_add(instance->shared_tag_cube_manager, cube->key_tag, i);
             return i;
         }
     }
 
     int cube_id = instance->max_n_cube;
     add_cube_to_position(instance, cube, cube_id);
-    cube_manager_add(instance->shared_tag_cube_manager, cube->key_tag, cube_id);
 
     return cube_id;
 }
 
-struct cube *fieldstat_cube_info_init(const struct fieldstat_tag *shared_tags, size_t n_tag, enum sampling_mode mode, size_t max_n_cell)
-{
-    struct cube *cube = calloc(1, sizeof(struct cube));
-    cube->sampling_mode = mode;
-
-    if (n_tag == 0) {
-        cube->cube_identifier = NULL;
-    } else {
-        cube->cube_identifier = malloc(sizeof(struct fieldstat_tag) * n_tag);
-        tag_array_copy(cube->cube_identifier, shared_tags, n_tag);
-    }
-
-    cube->n_shared_tags = n_tag;
-
-    struct tag_hash_key *shared_tag_key = malloc(sizeof(struct tag_hash_key));
-    tag_hash_key_init_with_fieldstat_tag(shared_tag_key, shared_tags, n_tag, true);
-    cube->key_tag = shared_tag_key;
-
-    cube->max_n_cell = max_n_cell;
-
-    return cube;
-}
-
-struct cube *fieldstat_cube_new(const struct fieldstat_tag *shared_tags, size_t n_tag, enum sampling_mode mode, size_t max_n_cell)
-{
-    struct cube *cube = fieldstat_cube_info_init(shared_tags, n_tag, mode, max_n_cell);
-
-    switch (mode)
-    {
-        case SAMPLING_MODE_TOPK:
-            cube->topk = heavy_keeper_new(max_n_cell);
-            heavy_keeper_set_exdata_schema(cube->topk, exdata_new_i, exdata_free_i, exdata_merge_i, exdata_reset_i, exdata_copy_i);
-            break;
-        case SAMPLING_MODE_COMPREHENSIVE:
-            cube->comprehensive = tag_map_new(max_n_cell);
-            tag_map_set_exdata_schema(cube->comprehensive, exdata_new_i, exdata_free_i, exdata_merge_i, exdata_reset_i, exdata_copy_i);
-            break;
-        default:
-            assert(0);
-            break;
-    }
-
-    return cube;
-}
 
 int fieldstat_create_cube(struct fieldstat *instance, const struct fieldstat_tag *shared_tags, size_t n_tag, enum sampling_mode mode, size_t max_n_cell)
 {
@@ -485,46 +327,30 @@ int fieldstat_create_cube(struct fieldstat *instance, const struct fieldstat_tag
         return FS_ERR_INVALID_KEY;
     }
 
-    struct cube *cube = fieldstat_cube_new(shared_tags, n_tag, mode, max_n_cell);
+    struct cube *cube = cube_new(shared_tags, n_tag, mode, max_n_cell);
 
     int cube_id = fieldstat_append_cube_to_instance(instance, cube);
+    cube_manager_add(instance->shared_tag_cube_manager, &shared_tag_key, cube_id);
 
     return cube_id;
 }
 
-void fieldstat_cube_free_contents(struct fieldstat *instance, int cube_id)
-{
-    struct cube *cube = instance->cube[cube_id];
-
-    switch (cube->sampling_mode)
-    {
-    case SAMPLING_MODE_TOPK:
-        heavy_keeper_free(cube->topk);
-        break;
-    case SAMPLING_MODE_COMPREHENSIVE:
-        tag_map_free(cube->comprehensive);
-        break;
-    default:
-        assert(0);
-        break;
-    }
-
-    fieldstat_free_tag_array(cube->cube_identifier, cube->n_shared_tags);
-    tag_hash_key_free(cube->key_tag);
-
-    free(cube);
-    instance->cube[cube_id] = NULL;
-}
-
 void fieldstat_cube_free(struct fieldstat *instance, int cube_id)
 {
-    if (instance->cube[cube_id] == NULL) {
+    struct cube *cube = instance->cube[cube_id];
+    if (cube == NULL) {
         return;
     }
     
-    cube_manager_delete(instance->shared_tag_cube_manager, instance->cube[cube_id]->key_tag);
+    struct fieldstat_tag_list *shared_tag = cube_get_identifier(cube);
+    struct tag_hash_key shared_tag_key;
+    tag_hash_key_init_with_fieldstat_tag(&shared_tag_key, shared_tag->tag, shared_tag->n_tag, false);
+    cube_manager_delete(instance->shared_tag_cube_manager, &shared_tag_key);
+
+    cube_free(cube);
+    instance->cube[cube_id] = NULL;
 
-    fieldstat_cube_free_contents(instance, cube_id);
+    fieldstat_tag_list_arr_free(shared_tag, 1);
 }
 
 int fieldstat_cube_set_primary_metric(struct fieldstat *instance, int cube_id, int metric_id)
@@ -539,20 +365,17 @@ int fieldstat_cube_set_primary_metric(struct fieldstat *instance, int cube_id, i
     if (cube == NULL) {
         return FS_ERR_INVALID_CUBE_ID;
     }
-    if (cube->sampling_mode != SAMPLING_MODE_TOPK) {
-        return FS_ERR_INVALID_PARAM;
-    }
     if (metric_id < 0 || metric_id >= instance->max_n_metric_master) {
         return FS_ERR_INVALID_METRIC_ID;
     }
     if (instance->metric_masters[metric_id] == NULL) {
         return FS_ERR_INVALID_METRIC_ID;
     }
-    if (metric_get_type(instance->metric_masters[metric_id]) != METRIC_TYPE_COUNTER) {
+    if (instance->metric_masters[metric_id]->type != METRIC_TYPE_COUNTER) {
         return FS_ERR_INVALID_PARAM;
     }
 
-    cube->primary_metric_id = metric_id;
+    cube_set_primary_metric(cube, metric_id);
 
     return FS_OK;
 }
@@ -561,28 +384,27 @@ int fieldstat_cube_set_primary_metric(struct fieldstat *instance, int cube_id, i
 /*                               metric register                              */
 /* -------------------------------------------------------------------------- */
 
-void add_metric_to_instance(struct fieldstat *instance, const struct metric *metric, int metric_id)
+void add_manifest_to_instance(struct fieldstat *instance, const struct metric_manifest *manifest, int metric_id)
 {
     if (metric_id >= instance->max_n_metric_master) {
-        instance->metric_masters = realloc(instance->metric_masters, sizeof(struct metric *) * instance->max_n_metric_master * 2);
-        memset(instance->metric_masters + instance->max_n_metric_master, 0, sizeof(struct metric *) * (instance->max_n_metric_master));
+        instance->metric_masters = realloc(instance->metric_masters, sizeof(struct metric_manifest *) * instance->max_n_metric_master * 2);
+        memset(instance->metric_masters + instance->max_n_metric_master, 0, sizeof(struct metric_manifest *) * (instance->max_n_metric_master));
         instance->max_n_metric_master *= 2;
     }
 
-    instance->metric_masters[metric_id] = (struct metric *)metric;
-    name_id_map_add(instance->metric_name_id_map, metric_get_name(metric), metric_id);
+    instance->metric_masters[metric_id] = (struct metric_manifest *)manifest;
 }
 
-static int append_metric_to_instance(struct fieldstat *instance, const struct metric *metric)
+static int append_manifest_to_instance(struct fieldstat *instance, const struct metric_manifest *metric)
 {
     int metric_id = 0;
-    for (;metric_id < instance->max_n_metric_master; metric_id++) {
+    for ( ;metric_id < instance->max_n_metric_master; metric_id++) {
         if (instance->metric_masters[metric_id] == NULL) {
             break;
         }
     }
 
-    add_metric_to_instance(instance, metric, metric_id);
+    add_manifest_to_instance(instance, metric, metric_id);
 
     return metric_id;
 }
@@ -593,7 +415,7 @@ int check_before_register_metric(const struct fieldstat *instance, const char *m
         return FS_ERR_NULL_HANDLER;
     }
 
-    if (name_id_map_get_id_by_name(instance->metric_name_id_map, metric_name) != -1) {
+    if (is_metric_name_duplicate(instance, metric_name)) {
         return FS_ERR_INVALID_KEY;
     }
 
@@ -607,9 +429,15 @@ int fieldstat_register_counter(struct fieldstat *instance, const char *metric_na
         return ret;
     }
 
-    const struct metric *metric = metric_counter_new(metric_name);
+    struct metric_manifest *metric = malloc(sizeof(struct metric_manifest));
+    metric->name = strdup(metric_name);
+    metric->paras = construct_parameters(METRIC_TYPE_COUNTER);
+    metric->type = METRIC_TYPE_COUNTER;
 
-    return append_metric_to_instance(instance, metric);
+    int id = append_manifest_to_instance(instance, metric);
+    metric->id = id;
+
+    return id;
 }
 
 int fieldstat_register_hll(struct fieldstat *instance, const char *metric_name, unsigned char precision)
@@ -622,9 +450,15 @@ int fieldstat_register_hll(struct fieldstat *instance, const char *metric_name,
         return FS_ERR_INVALID_PARAM;
     }
 
-    const struct metric *metric = metric_hll_new(metric_name, precision);
+    struct metric_manifest *metric = malloc(sizeof(struct metric_manifest));
+    metric->name = strdup(metric_name);
+    metric->paras = construct_parameters(METRIC_TYPE_HLL, precision);
+    metric->type = METRIC_TYPE_HLL;
+
+    int id = append_manifest_to_instance(instance, metric);
+    metric->id = id;
 
-    return append_metric_to_instance(instance, metric);
+    return id;
 }
 
 int fieldstat_register_hist(struct fieldstat *instance, const char *metric_name, long long lowest_trackable_value, long long highest_trackable_value, int significant_figures)
@@ -645,9 +479,15 @@ int fieldstat_register_hist(struct fieldstat *instance, const char *metric_name,
 		return FS_ERR_INVALID_PARAM;
 	}
 
-    const struct metric *metric = metric_histogram_new(metric_name, lowest_trackable_value, highest_trackable_value, significant_figures);
+    struct metric_manifest *metric = malloc(sizeof(struct metric_manifest));
+    metric->name = strdup(metric_name);
+    metric->paras = construct_parameters(METRIC_TYPE_HISTOGRAM, lowest_trackable_value, highest_trackable_value, significant_figures);
+    metric->type = METRIC_TYPE_HISTOGRAM;
 
-    return append_metric_to_instance(instance, metric);
+    int id = append_manifest_to_instance(instance, metric);
+    metric->id = id;
+
+    return id;
 }
 
 /* -------------------------------------------------------------------------- */
@@ -669,27 +509,14 @@ int check_before_add(const struct fieldstat *instance, int cube_id, int metric_i
     if (metric_id < 0 || metric_id >= instance->max_n_metric_master) {
         return FS_ERR_INVALID_METRIC_ID;
     }
-    const struct metric *metric = instance->metric_masters[metric_id];
-    if (metric == NULL || metric_get_type(metric) != type) {
+    const struct metric_manifest *metric = instance->metric_masters[metric_id];
+    if (metric == NULL || metric->type != type) {
         return FS_ERR_INVALID_METRIC_ID;
     }
 
     return FS_OK;
 }
 
-struct cell *find_or_add_exdata_comprehensive(struct tag_map *comprehensive, const char *key, size_t key_len, struct exdata_new_args *args)
-{
-    struct cell *cell_data = tag_map_get0_exdata(comprehensive, key, key_len);
-    if (cell_data == NULL) {
-        int tmp_ret = tag_map_add(comprehensive, key, key_len, args);
-        if (tmp_ret != 1) {
-            return NULL;
-        }
-        cell_data = tag_map_get0_exdata(comprehensive, key, key_len);
-    }
-    return cell_data;
-}
-
 int fieldstat_counter_incrby(struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag *tags, size_t n_tag, long long increment)
 {
     int ret = check_before_add(instance, cube_id, metric_id, METRIC_TYPE_COUNTER);
@@ -697,66 +524,11 @@ int fieldstat_counter_incrby(struct fieldstat *instance, int cube_id, int metric
         return ret;
     }
 
-    const struct cube *cube = instance->cube[cube_id];
+    struct cube *cube = instance->cube[cube_id];
+    const struct metric_manifest *manifest = instance->metric_masters[metric_id];
     
-    char *tag_in_string;
-    size_t tag_len;
-    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
-
-    struct exdata_new_args args;
-    args.tags = tags;
-    args.n_tags = n_tag;
-
-    struct cell *cell_data = NULL;
-    switch (cube->sampling_mode)
-    {
-    case SAMPLING_MODE_TOPK:
-        if (cube->primary_metric_id != metric_id) {
-            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-
-            if (cell_data == NULL) {
-                int tmp_ret = heavy_keeper_add(cube->topk, tag_in_string, tag_len, 0, &args);
-                if (tmp_ret != 1) {
-                    free(tag_in_string);
-                    return FS_ERR_TOO_MANY_CELLS;
-                }
-                cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-            }
-        } else {
-            if (increment < 0) {
-                free(tag_in_string);
-                return FS_ERR_INVALID_PARAM;
-            } else if (increment == 0) {
-                free(tag_in_string);
-                return FS_OK;
-            }
+    return cube_counter_incrby(cube, manifest, tags, n_tag, increment);
 
-            // heavy_keeper_add should be called anyway, to let the topk record update.
-            int tmp_ret = heavy_keeper_add(cube->topk, tag_in_string, tag_len, increment, &args);
-            if (tmp_ret != 1) {
-                free(tag_in_string);
-                return FS_ERR_TOO_MANY_CELLS;
-            }
-            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-        }
-        break;
-    case SAMPLING_MODE_COMPREHENSIVE:
-        cell_data = find_or_add_exdata_comprehensive(cube->comprehensive, tag_in_string, tag_len, &args);
-        if (cell_data == NULL) {
-            free(tag_in_string);
-            return FS_ERR_TOO_MANY_CELLS;
-        }
-        break;
-    default:
-        assert(0);
-        break;
-    }
-
-    struct metric *metric = add_or_find_metric_in_cell(instance, cell_data, metric_id);
-
-    metric_counter_incrby(metric, increment);
-    free(tag_in_string);
-    return FS_OK;
 }
 
 int fieldstat_counter_set(struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag *tags, size_t n_tag, long long value)
@@ -765,66 +537,10 @@ int fieldstat_counter_set(struct fieldstat *instance, int cube_id, int metric_id
     if (ret != FS_OK) {
         return ret;
     }
-    const struct cube *cube = instance->cube[cube_id];
-    char *tag_in_string;
-    size_t tag_len;
-    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
-
-    struct exdata_new_args args;
-    args.tags = tags;
-    args.n_tags = n_tag;
-
-    struct cell *cell_data = NULL;
-    switch (cube->sampling_mode)
-    {
-    case SAMPLING_MODE_TOPK: {
-        if (cube->primary_metric_id != metric_id) {
-            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-            if (cell_data == NULL) {
-                int tmp_ret = heavy_keeper_add(cube->topk, tag_in_string, tag_len, 0, &args);
-                if (tmp_ret != 1) {
-                    free(tag_in_string);
-                    return FS_ERR_TOO_MANY_CELLS;
-                }
-                cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-            }
-        } else {
-            long long current_count = 0;
-            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-            if (cell_data != NULL) {
-                const struct metric *tmp_metric = find_metric_in_cell(cell_data, metric_id);
-                if (tmp_metric != NULL) {
-                    current_count = metric_counter_get(tmp_metric);
-                }
-            }
-            long long increment = value - current_count;
-            if (increment < 0) {
-                free(tag_in_string);
-                return FS_ERR_INVALID_PARAM;
-            } else if (increment == 0) {
-                free(tag_in_string);
-                return FS_OK;
-            }
-            
-            int tmp_ret = heavy_keeper_add(cube->topk, tag_in_string, tag_len, increment, &args);
-            if (tmp_ret != 1) {
-                return FS_ERR_TOO_MANY_CELLS;
-            }
-            cell_data = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-        }
-        break;}
-    case SAMPLING_MODE_COMPREHENSIVE: {
-        cell_data = find_or_add_exdata_comprehensive(cube->comprehensive, tag_in_string, tag_len, &args);
-        break;}
-    default:
-        assert(0);
-        break;
-    }
-    assert(cell_data != NULL); // to mute the warning
-    struct metric *metric = add_or_find_metric_in_cell(instance, cell_data, metric_id);
-    free(tag_in_string);
-    metric_counter_set(metric, value);
-    return FS_OK;
+    struct cube *cube = instance->cube[cube_id];
+    const struct metric_manifest *manifest = instance->metric_masters[metric_id];
+
+    return cube_counter_set(cube, manifest, tags, n_tag, value);
 }
 
 int fieldstat_hll_add(struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag *tags, size_t n_tag, const char *key, size_t key_len)
@@ -833,25 +549,11 @@ int fieldstat_hll_add(struct fieldstat *instance, int cube_id, int metric_id, co
     if (ret != FS_OK) {
         return ret;
     }
-    if (instance->cube[cube_id]->sampling_mode == SAMPLING_MODE_TOPK) {
-        return FS_ERR_INVALID_PARAM;
-    }
     
-    char *tag_in_string;
-    size_t tag_len;
-    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
-
-    struct exdata_new_args args;
-    args.tags = tags;
-    args.n_tags = n_tag;
-
-    struct cell *cell_data = find_or_add_exdata_comprehensive(instance->cube[cube_id]->comprehensive, tag_in_string, tag_len, &args);
-    struct metric *metric = add_or_find_metric_in_cell(instance, cell_data, metric_id);
-
-    metric_hll_add(metric, key, key_len);
-    free(tag_in_string);
+    struct cube *cube = instance->cube[cube_id];
+    const struct metric_manifest *manifest = instance->metric_masters[metric_id];
 
-    return 0;
+    return cube_hll_add(cube, manifest, tags, n_tag, key, key_len);
 }
 
 int fieldstat_hist_record(struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag *tags, size_t n_tag, long long value)
@@ -860,84 +562,18 @@ int fieldstat_hist_record(struct fieldstat *instance, int cube_id, int metric_id
     if (ret != FS_OK) {
         return ret;
     }
-    if (instance->cube[cube_id]->sampling_mode == SAMPLING_MODE_TOPK) {
-        return FS_ERR_INVALID_PARAM;
-    }
-
-    char *tag_in_string;
-    size_t tag_len;
-    build_dynamic_cell_key(tags, n_tag, &tag_in_string, &tag_len);
-
-    struct exdata_new_args args;
-    args.tags = tags;
-    args.n_tags = n_tag;
-
-    struct cell *cell_data = find_or_add_exdata_comprehensive(instance->cube[cube_id]->comprehensive, tag_in_string, tag_len, &args);
 
-    free(tag_in_string);
-
-    // // metric_histogram_record may fail, unlike the other add functions.
-    struct metric *metric = find_metric_in_cell(cell_data, metric_id);
-    if (metric != NULL) {
-        ret = metric_histogram_record(metric, value);
-        if (ret < 0) {
-            return FS_ERR_INVALID_PARAM;
-        }
-    } else {
-        metric = metric_fork(instance->metric_masters[metric_id]);
-        ret = metric_histogram_record(metric, value);
-        if (ret < 0) {
-            metric_free(metric);
-            return FS_ERR_INVALID_PARAM;
-        }
-        add_metric_to_cell(cell_data, metric, metric_id);
-    }
+    struct cube *cube = instance->cube[cube_id];
+    const struct metric_manifest *manifest = instance->metric_masters[metric_id];
 
-    return FS_OK;
+    return cube_histogram_record(cube, manifest, tags, n_tag, value);
 }
 
 /* -------------------------------------------------------------------------- */
 /*                                    merge                                   */
 /* -------------------------------------------------------------------------- */
 
-struct cube *fieldstat_cube_copy(const struct cube *cube)
-{
-    struct cube *cube_dup = fieldstat_cube_info_init(cube->cube_identifier, cube->n_shared_tags, cube->sampling_mode, cube->max_n_cell);
-    cube_dup->primary_metric_id = cube->primary_metric_id;
-
-    switch (cube->sampling_mode)
-    {
-    case SAMPLING_MODE_TOPK:
-        cube_dup->topk = heavy_keeper_copy(cube->topk);
-        break;
-    case SAMPLING_MODE_COMPREHENSIVE:
-        cube_dup->comprehensive = tag_map_copy(cube->comprehensive);
-        break;
-    default:
-        assert(0);
-        break;
-    }
-
-    return cube_dup;
-}
 
-void fieldstat_cube_merge(struct cube *dest, const struct cube *src)
-{
-    assert(dest->sampling_mode == src->sampling_mode);
-
-    switch (dest->sampling_mode)
-    {
-    case SAMPLING_MODE_TOPK:
-        heavy_keeper_merge(dest->topk, src->topk);
-        break;
-    case SAMPLING_MODE_COMPREHENSIVE:
-        tag_map_merge(dest->comprehensive, src->comprehensive);
-        break;
-    default:
-        assert(0);
-        break;
-    }
-}
 
 int fieldstat_merge(struct fieldstat *instance, struct fieldstat *src)
 {
@@ -947,13 +583,13 @@ int fieldstat_merge(struct fieldstat *instance, struct fieldstat *src)
 
     int metric_len_src = src->max_n_metric_master;
     for (int i = 0; i < metric_len_src; i++) {
-        const struct metric *metric_src = src->metric_masters[i];
-        const struct metric *metric_dst = instance->metric_masters[i];
+        const struct metric_manifest *metric_src = src->metric_masters[i];
+        const struct metric_manifest *metric_dst = instance->metric_masters[i];
         if (metric_src == NULL || metric_dst == NULL) {
             break;
         }
-        if (metric_get_type(metric_src) != metric_get_type(metric_dst) || 
-                strcmp(metric_get_name(metric_src), metric_get_name(metric_dst)) != 0
+        if (metric_src->type != metric_dst->type ||
+                strcmp(metric_src->name, metric_dst->name) != 0
             ) {
             assert(0);
             return FS_ERR_INVALID_PARAM;
@@ -963,12 +599,11 @@ int fieldstat_merge(struct fieldstat *instance, struct fieldstat *src)
         if (instance->metric_masters[i] != NULL || src->metric_masters[i] == NULL) {
             continue;
         }
-        const struct metric *metric_src = src->metric_masters[i];
-        int id_tmp = append_metric_to_instance(instance, metric_fork(metric_src));
-        name_id_map_add(instance->metric_name_id_map, metric_get_name(metric_src), id_tmp);
+        const struct metric_manifest *metric_src = src->metric_masters[i];
+        append_manifest_to_instance(instance, metric_manifest_copy(metric_src));
     }
 
-    const struct cube_manager *tag_cube_id_map = instance->shared_tag_cube_manager;
+    struct cube_manager *tag_cube_id_map = instance->shared_tag_cube_manager;
     int ret = 0;
     for (int i = 0; i < src->max_n_cube; i++) {
         const struct cube *cube_src = src->cube[i];
@@ -976,36 +611,25 @@ int fieldstat_merge(struct fieldstat *instance, struct fieldstat *src)
             continue;
         }
 
-        const struct tag_hash_key *shared_tag_key_src = cube_src->key_tag;
-        int cube_id_tmp = cube_manager_find(tag_cube_id_map, shared_tag_key_src);
-        if (cube_id_tmp == -1) {
-            struct cube *copied_cube = fieldstat_cube_copy(cube_src);
-            fieldstat_append_cube_to_instance(instance, copied_cube);
+        // const struct tag_hash_key *shared_tag_key_src = cube_src->key_tag;
+        struct tag_hash_key shared_tag_key_src;
+        struct fieldstat_tag_list *shared_tag_list = cube_get_identifier(cube_src);
+        tag_hash_key_init_with_fieldstat_tag(&shared_tag_key_src, shared_tag_list->tag, shared_tag_list->n_tag, false);
+        int cube_id_dest = cube_manager_find(tag_cube_id_map, &shared_tag_key_src);
+        if (cube_id_dest == -1) {
+            struct cube *copied_cube = cube_copy(cube_src);
+            cube_id_dest = fieldstat_append_cube_to_instance(instance, copied_cube);
+            cube_manager_add(tag_cube_id_map, &shared_tag_key_src, cube_id_dest);
         } else {
-            struct cube *cube_dst = instance->cube[cube_id_tmp];
-            if (cube_dst->sampling_mode != cube_src->sampling_mode) {
-                ret = FS_ERR_INVALID_PARAM;
-                continue;
-            }
-            if (cube_dst->sampling_mode == SAMPLING_MODE_TOPK && cube_dst->primary_metric_id != cube_src->primary_metric_id) {
-                ret = FS_ERR_INVALID_PARAM;
-                continue;
-            }
-            fieldstat_cube_merge(cube_dst, cube_src);
+            struct cube *cube_dst = instance->cube[cube_id_dest];
+            cube_merge(cube_dst, cube_src);
         }
+        fieldstat_tag_list_arr_free(shared_tag_list, 1);
     }
 
     return ret;
 }
 
-// only copy the cube configurations, leave the cells empty
-struct cube *fieldstat_cube_fork(const struct cube *cube) {
-    struct cube *ret = fieldstat_cube_new(cube->cube_identifier, cube->n_shared_tags, cube->sampling_mode, cube->max_n_cell);
-    ret->primary_metric_id = cube->primary_metric_id;
-
-    return ret;
-}
-
 struct fieldstat *fieldstat_fork(const struct fieldstat *instance)
 {
     if (instance == NULL) {
@@ -1021,20 +645,23 @@ struct fieldstat *fieldstat_fork(const struct fieldstat *instance)
         if (cube == NULL) {
             continue;
         }
-        new_instance->cube[i] = fieldstat_cube_fork(cube);
-        cube_manager_add(new_instance->shared_tag_cube_manager, cube->key_tag, i);
+        new_instance->cube[i] = cube_fork(cube);
+        struct fieldstat_tag_list *shared_tag = cube_get_identifier(cube);
+        struct tag_hash_key shared_tag_key;
+        tag_hash_key_init_with_fieldstat_tag(&shared_tag_key, shared_tag->tag, shared_tag->n_tag, false);
+        cube_manager_add(new_instance->shared_tag_cube_manager, &shared_tag_key, i);
+        fieldstat_tag_list_arr_free(shared_tag, 1);
     }
     new_instance->cube_version = calloc(new_instance->max_n_cube, sizeof(unsigned long));
     memcpy(new_instance->cube_version, instance->cube_version, sizeof(unsigned long) * new_instance->max_n_cube);
 
-    new_instance->metric_masters = calloc(instance->max_n_metric_master, sizeof(struct metric *));
+    new_instance->metric_masters = calloc(instance->max_n_metric_master, sizeof(struct metric_manifest *));
     new_instance->max_n_metric_master = instance->max_n_metric_master;
     for (size_t i = 0; i < instance->max_n_metric_master; i++) {
         if (instance->metric_masters[i] != NULL) {
-            new_instance->metric_masters[i] = metric_fork(instance->metric_masters[i]);
+            new_instance->metric_masters[i] = metric_manifest_copy(instance->metric_masters[i]);
         }
     }
-    new_instance->metric_name_id_map = name_id_map_copy(instance->metric_name_id_map);
 
     return new_instance;
 }
@@ -1042,34 +669,33 @@ struct fieldstat *fieldstat_fork(const struct fieldstat *instance)
 void calibrate_metrics_in_instance(const struct fieldstat *master, struct fieldstat *replica)
 {
     if (replica->max_n_metric_master < master->max_n_metric_master) {
-        replica->metric_masters = (struct metric **)realloc(replica->metric_masters, sizeof(struct metric *) * master->max_n_metric_master);
-        memset(replica->metric_masters + replica->max_n_metric_master, 0, sizeof(struct metric *) * (master->max_n_metric_master - replica->max_n_metric_master));
+        replica->metric_masters = (struct metric_manifest **)realloc(replica->metric_masters, sizeof(struct metric_manifest *) * master->max_n_metric_master);
+        memset(replica->metric_masters + replica->max_n_metric_master, 0, sizeof(struct metric_manifest *) * (master->max_n_metric_master - replica->max_n_metric_master));
         replica->max_n_metric_master = master->max_n_metric_master;
     }
 
     size_t longer_arr_len = master->max_n_metric_master > replica->max_n_metric_master ? master->max_n_metric_master : replica->max_n_metric_master;
     for (size_t i = 0; i < longer_arr_len; i++) {
-        const struct metric *metric_master = i >= master->max_n_metric_master ? NULL : master->metric_masters[i];
-        struct metric *metric_target = i >= replica->max_n_metric_master ? NULL : replica->metric_masters[i];
+        const struct metric_manifest *metric_master = i >= master->max_n_metric_master ? NULL : master->metric_masters[i];
+        struct metric_manifest *metric_target = i >= replica->max_n_metric_master ? NULL : replica->metric_masters[i];
         if (metric_master == NULL && metric_target == NULL) {
             continue;
         }
         if (metric_master == NULL && metric_target != NULL) {
-            metric_free(metric_target);
+            metric_manifest_free(metric_target);
             replica->metric_masters[i] = NULL;
             continue;
         }
         if (metric_master != NULL && metric_target == NULL) {
-            const struct metric *metric_dup = metric_fork(metric_master);
-            add_metric_to_instance(replica, metric_dup, i);
+            const struct metric_manifest *metric_dup = metric_manifest_copy(metric_master);
+            add_manifest_to_instance(replica, metric_dup, i);
             continue;
         }
-        if (metric_get_type(metric_master) != metric_get_type(metric_target) || 
-                strcmp(metric_get_name(metric_master), metric_get_name(metric_target)) != 0
-            ) {
-            metric_free(metric_target);
-            const struct metric *metric_dup = metric_fork(metric_master);
-            add_metric_to_instance(replica, metric_dup, i);
+
+        if (metric_master->type != metric_target->type || strcmp(metric_master->name, metric_target->name) != 0) {
+            metric_manifest_free(metric_target);
+            const struct metric_manifest *metric_dup = metric_manifest_copy(metric_master);
+            add_manifest_to_instance(replica, metric_dup, i);
             continue;
         }
 
@@ -1096,27 +722,33 @@ int fieldstat_calibrate(const struct fieldstat *master, struct fieldstat *replic
     size_t longer_arr_len = len_master > len_replica ? len_master : len_replica;
     for (size_t i = 0; i < longer_arr_len; i++) {
         const struct cube *cube_master = i >= len_master ? NULL : master->cube[i];
-        const struct cube *cube_target = i >= len_replica ? NULL : replica->cube[i];
+        struct cube *cube_target = i >= len_replica ? NULL : replica->cube[i];
 
         if (cube_master == NULL && cube_target == NULL) {
             continue;
         }
         if (cube_master == NULL && cube_target != NULL) {
-            fieldstat_cube_free_contents(replica, i);
+            cube_free(cube_target);
+            replica->cube[i] = NULL;
             continue;
         }
         if (cube_master != NULL && cube_target == NULL) {
-            struct cube *cube_dup = fieldstat_cube_fork(cube_master);
+            struct cube *cube_dup = cube_fork(cube_master);
             add_cube_to_position(replica, cube_dup, i);
             continue;
         }
 
-        if (master->cube_version[i] == replica->cube_version[i] && tag_hash_key_cmp(cube_master->key_tag, cube_target->key_tag) == 0) {
+        struct fieldstat_tag_list *cube_master_identifier = cube_get_identifier(cube_master);
+        struct fieldstat_tag_list *cube_target_identifier = cube_get_identifier(cube_target);
+        bool is_same = cube_master_identifier->n_tag == cube_target_identifier->n_tag && is_tag_array_same(cube_master_identifier->tag, cube_target_identifier->tag, cube_master_identifier->n_tag);
+        fieldstat_tag_list_arr_free(cube_master_identifier, 1);
+        fieldstat_tag_list_arr_free(cube_target_identifier, 1);
+        if (master->cube_version[i] == replica->cube_version[i] && is_same) {
             continue;
         }
 
-        fieldstat_cube_free_contents(replica, i);
-        struct cube *cube_dup = fieldstat_cube_fork(cube_master);
+        cube_free(cube_target);
+        struct cube *cube_dup = cube_fork(cube_master);
         add_cube_to_position(replica, cube_dup, i);
     }
 
@@ -1185,146 +817,86 @@ struct fieldstat_tag_list *fieldstat_get_shared_tags(const struct fieldstat *ins
     if (cube == NULL) {
         return NULL;
     }
-   
-    struct fieldstat_tag_list *tag_list = (struct fieldstat_tag_list *)malloc(sizeof(struct fieldstat_tag_list));
-
-    if (cube->n_shared_tags == 0) {
-        tag_list->tag = NULL;
-        tag_list->n_tag = 0;
-        return tag_list;
-    }
-
-    tag_list->tag = (struct fieldstat_tag *)malloc(sizeof(struct fieldstat_tag) * cube->n_shared_tags);
-    tag_list->n_tag = cube->n_shared_tags;
-    tag_array_copy(tag_list->tag, cube->cube_identifier, cube->n_shared_tags);
-
-    return tag_list;
-}
-
-const struct cell *get_exdata_by_tag_list(const struct cube *cube, const struct fieldstat_tag_list *tags)
-{
-    const struct cell *ret = NULL;
-    char *tag_in_string;
-    size_t tag_len;
-    build_dynamic_cell_key(tags->tag, tags->n_tag, &tag_in_string, &tag_len);
-
-    switch (cube->sampling_mode)
-    {
-    case SAMPLING_MODE_TOPK:
-        ret = heavy_keeper_get0_exdata(cube->topk, tag_in_string, tag_len);
-        break;
-    case SAMPLING_MODE_COMPREHENSIVE:
-        ret = tag_map_get0_exdata(cube->comprehensive, tag_in_string, tag_len);
-        break;
-    default:
-        assert(0);
-        return NULL;
-    }
-    free(tag_in_string);
-
-    return ret;
+    
+    return cube_get_identifier(cube);
 }
 
-const struct metric *get_metric_by_tag_list(const struct fieldstat *instance, int cube_id, const struct fieldstat_tag_list *tags, int metric_id,int *ret)
+int fieldstat_counter_get(const struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag_list *tags, long long *value)
 {
     if (cube_id < 0 || cube_id >= instance->max_n_cube) {
-        *ret = FS_ERR_INVALID_CUBE_ID;
-        return NULL;
+        return FS_ERR_INVALID_CUBE_ID;
     }
     const struct cube *cube = instance->cube[cube_id];
     if (cube == NULL) {
-        *ret = FS_ERR_INVALID_CUBE_ID;
-        return NULL;
+        return FS_ERR_INVALID_CUBE_ID;
     }
-
-    const struct cell *data = get_exdata_by_tag_list(cube, tags);
     
-    if (data == NULL) {
-        *ret = FS_ERR_INVALID_TAG;
-        return NULL;
-    }
-
-    if (metric_id < 0 || metric_id >= data->max_n_metric) {
-        *ret = FS_ERR_INVALID_METRIC_ID;
-        return NULL;
-    }
-    *ret = FS_OK;
-
-    return data->metrics[metric_id];
-}
-
-int fieldstat_counter_get(const struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag_list *tags, long long *value)
-{
-    int ret;
-    const struct metric *metric = get_metric_by_tag_list(instance, cube_id, tags, metric_id, &ret);
-    if (ret!=FS_OK) {
-        return ret;
-    }
-    if (metric == NULL || metric_get_type(metric) != METRIC_TYPE_COUNTER) {
-        return FS_ERR_INVALID_METRIC_ID;
-    }
-
-    *value = metric_counter_get(metric);
-
-    return FS_OK;
+    return cube_counter_get(cube, metric_id, tags, value);
 }
 
 int fieldstat_hll_get(const struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag_list *tags, double *value)
 {
-    int ret;
-    const struct metric *metric = get_metric_by_tag_list(instance, cube_id, tags, metric_id, &ret);
-    if (ret!=FS_OK) {
-        return ret;
+    if (cube_id < 0 || cube_id >= instance->max_n_cube) {
+        return FS_ERR_INVALID_CUBE_ID;
     }
-    if (metric == NULL || metric_get_type(metric) != METRIC_TYPE_HLL) {
-        return FS_ERR_INVALID_METRIC_ID;
+    const struct cube *cube = instance->cube[cube_id];
+    if (cube == NULL) {
+        return FS_ERR_INVALID_CUBE_ID;
     }
+    
+    int ret = cube_hll_get(cube, metric_id, tags, value);
 
-    *value = metric_hll_get(metric);
-
-    return FS_OK;
+    return ret;
 }
 
 long long fieldstat_hist_value_at_percentile(const struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag_list *tags, double percentile)
 {
-    int ret;
-    const struct metric *metric = get_metric_by_tag_list(instance, cube_id, tags, metric_id, &ret);
-    if (ret!=FS_OK) {
-        return ret;
+    if (cube_id < 0 || cube_id >= instance->max_n_cube) {
+        return FS_ERR_INVALID_CUBE_ID;
     }
-    if (metric == NULL || metric_get_type(metric) != METRIC_TYPE_HISTOGRAM) {
-        return FS_ERR_INVALID_METRIC_ID;
+    const struct cube *cube = instance->cube[cube_id];
+    if (cube == NULL) {
+        return FS_ERR_INVALID_CUBE_ID;
+    }
+    long long value;
+    int ret = cube_histogram_value_at_percentile(cube, metric_id, tags, percentile, &value);
+    if (ret < 0) {
+        return ret;
     }
 
-    return metric_histogram_value_at_percentile(metric, percentile);
+    return value;
 }
 
 long long fieldstat_hist_count_le_value(const struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag_list *tags, long long value)
 {
-    int ret;
-    const struct metric *metric = get_metric_by_tag_list(instance, cube_id, tags, metric_id, &ret);
-    if (ret!=FS_OK) {
-        return ret;
+    if (cube_id < 0 || cube_id >= instance->max_n_cube) {
+        return FS_ERR_INVALID_CUBE_ID;
     }
-    if (metric == NULL || metric_get_type(metric) != METRIC_TYPE_HISTOGRAM) {
-        return FS_ERR_INVALID_METRIC_ID;
+    const struct cube *cube = instance->cube[cube_id];
+    if (cube == NULL) {
+        return FS_ERR_INVALID_CUBE_ID;
     }
-
-    return metric_histogram_count_le_value(metric, value);
+    long long count;
+    int ret = cube_histogram_count_le_value(cube, metric_id, tags, value, &count);
+    if (ret < 0) {
+        return ret;
+    }
+    return count;
 }
 
-// metric_get_plain_blob
 void fieldstat_get_serialized_blob(const struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag_list *tags, char **blob, size_t *blob_size)
 {
-    int ret;
-    const struct metric *metric = get_metric_by_tag_list(instance, cube_id, tags, metric_id, &ret);
-    if (metric == NULL) {
-        *blob = NULL;
-        *blob_size = 0;
+    *blob = NULL;
+    *blob_size = 0;
+    if (cube_id < 0 || cube_id >= instance->max_n_cube) {
+        return;
+    }
+    const struct cube *cube = instance->cube[cube_id];
+    if (cube == NULL) {
         return;
     }
 
-    metric_get_plain_blob(metric, blob, blob_size);
+    cube_get_serialization(cube, metric_id, tags, blob, blob_size);
 }
 
 void fieldstat_tag_list_arr_free(struct fieldstat_tag_list *tag_list, size_t n_cell)
@@ -1343,12 +915,12 @@ const char *fieldstat_get_metric_name(const struct fieldstat *instance, int metr
     if (metric_id < 0 || metric_id >= instance->max_n_metric_master) {
         return NULL;
     }
-    const struct metric *metric = instance->metric_masters[metric_id];
+    const struct metric_manifest *metric = instance->metric_masters[metric_id];
     if (metric == NULL) {
         return NULL;
     }
 
-    return metric_get_name(metric);
+    return metric->name;
 }
 
 enum metric_type fieldstat_get_metric_type(const struct fieldstat *instance, int metric_id)
@@ -1356,12 +928,12 @@ enum metric_type fieldstat_get_metric_type(const struct fieldstat *instance, int
     if (instance == NULL || metric_id < 0 || metric_id >= instance->max_n_metric_master) {
         return (enum metric_type)(-1);
     }
-    const struct metric *metric = instance->metric_masters[metric_id];
+    const struct metric_manifest *metric = instance->metric_masters[metric_id];
     if (metric == NULL) {
         return (enum metric_type)(-1);
     }
 
-    return metric_get_type(metric);
+    return metric->type;
 }
 
 void fieldstat_get_cells_used_by_cube(const struct fieldstat *instance, int cube_id, struct fieldstat_tag_list **tag_list, size_t *n_cell)
@@ -1374,46 +946,7 @@ void fieldstat_get_cells_used_by_cube(const struct fieldstat *instance, int cube
         return;
     }
     
-    struct cell **cell_datas = NULL;
-    size_t n_cell_tmp = 0;
-    switch (cube->sampling_mode) {
-        case SAMPLING_MODE_COMPREHENSIVE:
-            n_cell_tmp = tag_map_get_count(cube->comprehensive);
-            cell_datas = (struct cell **)malloc(sizeof(struct cell *) * n_cell_tmp);
-            tag_map_list(cube->comprehensive, (void **)cell_datas, n_cell_tmp);
-            break;
-        case SAMPLING_MODE_TOPK:
-            n_cell_tmp = heavy_keeper_get_count(cube->topk);
-            cell_datas = (struct cell **)malloc(sizeof(struct cell *) * n_cell_tmp);
-            heavy_keeper_list(cube->topk, (void **)cell_datas, n_cell_tmp);
-            break;
-        default:
-            assert(0);
-    }
-
-    if (n_cell_tmp == 0) {
-        *tag_list = NULL;
-        *n_cell = 0;
-        return;
-    }
-
-    struct fieldstat_tag_list *tag_list_ret = (struct fieldstat_tag_list *)malloc(sizeof(struct fieldstat_tag_list) * n_cell_tmp);
-    *tag_list = tag_list_ret;
-    *n_cell = n_cell_tmp;
-
-    for (int i = 0; i < n_cell_tmp; i++) {
-        struct cell *cell_data = cell_datas[i];
-        struct fieldstat_tag_list *tag_list_tmp = &tag_list_ret[i];
-        tag_list_tmp->n_tag = cell_data->tags.n_tag;
-        if (tag_list_tmp->n_tag == 0) {
-            tag_list_tmp->tag = NULL;
-            continue;
-        }
-        tag_list_tmp->tag = (struct fieldstat_tag *)malloc(sizeof(struct fieldstat_tag) * tag_list_tmp->n_tag);
-        tag_array_copy(tag_list_tmp->tag, cell_data->tags.tag, tag_list_tmp->n_tag);
-    }
-
-    free(cell_datas);
+    cube_get_cells(cube, tag_list, n_cell);
 }
 
 int fieldstat_get_used_sampling(const struct fieldstat *instance, int cube_id)
@@ -1429,14 +962,7 @@ int fieldstat_get_used_sampling(const struct fieldstat *instance, int cube_id)
         return FS_ERR_INVALID_CUBE_ID;
     }
 
-    switch (cube->sampling_mode) {
-        case SAMPLING_MODE_COMPREHENSIVE:
-            return tag_map_get_count(cube->comprehensive);
-        case SAMPLING_MODE_TOPK:
-            return heavy_keeper_get_count(cube->topk);
-        default:
-            return FS_ERR_INVALID_PARAM;
-    }
+    return cube_get_cell_count(cube);
 }
 
 int fieldstat_find_cube(const struct fieldstat *instance, const struct fieldstat_tag *shared_tags, size_t n_shared_tags)
@@ -1455,45 +981,7 @@ int fieldstat_find_cube(const struct fieldstat *instance, const struct fieldstat
     return cube_id;
 }
 
-int fieldstat_get_cube_mode(const struct fieldstat *instance, int cube_id, enum sampling_mode *mode, int *primary_metric_id)
-{
-    if (instance == NULL) {
-        return FS_ERR_NULL_HANDLER;
-    }
-    if (cube_id < 0 || cube_id >= instance->max_n_cube) {
-        return FS_ERR_INVALID_CUBE_ID;
-    }
-    const struct cube *cube = instance->cube[cube_id];
-    if (cube == NULL) {
-        return FS_ERR_INVALID_CUBE_ID;
-    }
-
-    *mode = cube->sampling_mode;
-    if (cube->sampling_mode == SAMPLING_MODE_TOPK) {
-        *primary_metric_id = cube->primary_metric_id;
-    } else {
-        *primary_metric_id = -1;
-    }
-
-    return FS_OK;
-}
-
 void fieldstat_get_metric_in_cell(const struct fieldstat *instance, int cube_id, const struct fieldstat_tag_list *tags, int **metric_id_out, size_t *n_metric_out)
 {
-    const struct cell *cell_data = get_exdata_by_tag_list(instance->cube[cube_id], tags);
-    if (cell_data == NULL) {
-        *metric_id_out = NULL;
-        *n_metric_out = 0;
-        return;
-    }
-
-    *metric_id_out = (int *)malloc(sizeof(int) * cell_data->max_n_metric);
-    int n_metric = 0;
-    for (int i = 0; i < cell_data->max_n_metric; i++) {
-        if (cell_data->metrics[i] != NULL) {
-            (*metric_id_out)[n_metric] = i;
-            n_metric++;
-        }
-    }
-    *n_metric_out = n_metric;
+    return cube_get_cells_used_by_metric(instance->cube[cube_id], tags, metric_id_out, n_metric_out);
 }
\ No newline at end of file
diff --git a/src/metrics/metric.c b/src/metrics/metric.c
index 3b74ae2..aea96a1 100644
--- a/src/metrics/metric.c
+++ b/src/metrics/metric.c
@@ -3,7 +3,6 @@
 #include <stdlib.h>
 #include <assert.h>
 #include <string.h>
-#include <stdarg.h>
 #include <limits.h>
 
 #include "histogram_encoder.h"
@@ -12,38 +11,16 @@
 #include "hdr/hdr_histogram.h"
 #include "hdr/hdr_histogram_log.h"
 #include "st_hyperloglog.h"
+#include "metric_manifest.h"
 
 struct metric_counter_or_gauge {
     long long value;
 };
 
-struct para_hdr{
-    long long lowest_trackable_value;
-    long long highest_trackable_value;
-    int significant_figures;
-};
-
-struct para_hll {
-    char precision;
-};
-
-struct metric_parameter {
-    union {
-        struct para_hdr hdr;
-        struct para_hll hll;
-    };
-};
-
-struct metric_manifest {
-    char *name;
-    struct metric_parameter *paras;
-    enum metric_type type;
-};
-
-typedef struct metric_data *(*metric_func_new)(const struct metric_parameter *);
+typedef struct metric_data *(*metric_func_new)(const union metric_parameter *);
 typedef void (*metric_func_free)(struct metric_data *);
 typedef int (*metric_func_merge)(struct metric_data *, const struct metric_data *);
-typedef int (*metric_func_copy)(struct metric_data *, const struct metric_data *);
+typedef struct metric_data * (*metric_func_copy)(const struct metric_data *);
 typedef void (*metric_func_serialize)(const struct metric_data *, char **/*blob*/, size_t */*blob_size*/);
 // typedef struct metric_data * (*metric_func_deserialize)(const char *, size_t );
 typedef void (*metric_func_reset)(struct metric_data *);
@@ -66,7 +43,7 @@ struct metric_data {
 };
 
 struct metric {
-    struct metric_manifest *info;
+    enum metric_type type;
     struct metric_data *data;
 };
 
@@ -76,7 +53,7 @@ struct metric {
 
 /* --------------------------------- counter -------------------------------- */
 
-struct metric_data *metric_scheme_counter_new(const struct metric_parameter *_para)
+struct metric_data *metric_scheme_counter_new(const union metric_parameter *_para)
 {
     struct metric_data *data = (struct metric_data *)malloc(sizeof(struct metric_data));
     struct metric_counter_or_gauge *counter = (struct metric_counter_or_gauge *)malloc(sizeof(struct metric_counter_or_gauge));
@@ -110,12 +87,14 @@ static int metric_scheme_counter_merge(struct metric_data *pthis, const struct m
     return 0;
 }
 
-static int metric_scheme_counter_copy(struct metric_data *pthis, const struct metric_data *from)
+static struct metric_data *metric_scheme_counter_copy(const struct metric_data *from)
 {
-    struct metric_counter_or_gauge *counter = pthis->counter;
-    const struct metric_counter_or_gauge *from_counter = from->counter;
-    counter->value = from_counter->value;
-    return 0;
+    struct metric_data *pthis = (struct metric_data *)malloc(sizeof(struct metric_data));
+    struct metric_counter_or_gauge *counter = (struct metric_counter_or_gauge *)malloc(sizeof(struct metric_counter_or_gauge));
+    pthis->counter = counter;
+
+    counter->value = from->counter->value;
+    return pthis;
 }
 
 struct metric_data *metric_scheme_counter_deserialize(const char *blob, size_t blob_size) 
@@ -138,7 +117,7 @@ void metric_scheme_counter_reset(struct metric_data *data)
 
 /* --------------------------------- hll -------------------------------- */
 
-struct metric_data *metric_scheme_hll_new(const struct metric_parameter *para)
+struct metric_data *metric_scheme_hll_new(const union metric_parameter *para)
 {
     struct metric_data *data = (struct metric_data *)malloc(sizeof(struct metric_data));
     struct ST_hyperloglog *hll = ST_hyperloglog_new(para->hll.precision);
@@ -162,9 +141,13 @@ static int metric_scheme_hll_merge(struct metric_data *pthis, const struct metri
     return ST_hyperloglog_merge(pthis->hll, from->hll);
 }
 
-static int metric_scheme_hll_copy(struct metric_data *pthis, const struct metric_data *from)
+struct metric_data *metric_scheme_hll_copy(const struct metric_data *from)
 {
-    return ST_hyperloglog_merge(pthis->hll, from->hll);
+    struct metric_data *pthis = (struct metric_data *)malloc(sizeof(struct metric_data));
+    struct ST_hyperloglog *hll = ST_hyperloglog_new(from->hll->cfg.precision);
+    pthis->hll = hll;
+    ST_hyperloglog_merge(hll, from->hll);
+    return pthis;
 }
 
 struct metric_data *metric_scheme_hll_deserialize(const char *blob, size_t blob_size) 
@@ -183,7 +166,7 @@ void metric_scheme_hll_reset(struct metric_data *data)
 
 /* --------------------------------- histogram -------------------------------- */
 
-struct metric_data *metric_scheme_histogram_new(const struct metric_parameter *para)
+struct metric_data *metric_scheme_histogram_new(const union metric_parameter *para)
 {
     struct hdr_histogram* tmp_p_hdr = NULL;
     int ret = hdr_init((int64_t)para->hdr.lowest_trackable_value, (int64_t)para->hdr.highest_trackable_value, para->hdr.significant_figures, &tmp_p_hdr);
@@ -213,10 +196,16 @@ static int metric_scheme_histogram_merge(struct metric_data *pthis, const struct
     return 0;
 }
 
-static int metric_scheme_histogram_copy(struct metric_data *pthis, const struct metric_data *from)
+static struct metric_data *metric_scheme_histogram_copy(const struct metric_data *from)
 {
-    (void)hdr_add(pthis->hdr, from->hdr);
-    return 0;
+    struct hdr_histogram* tmp_p_hdr = NULL;
+    hdr_init(from->hdr->lowest_discernible_value, from->hdr->highest_trackable_value, from->hdr->significant_figures, &tmp_p_hdr);
+
+    struct metric_data *data = (struct metric_data *)malloc(sizeof(struct metric_data));
+    data->hdr = tmp_p_hdr;
+
+    hdr_add(data->hdr, from->hdr);
+    return data;
 }
 
 struct metric_data *metric_scheme_histogram_deserialize(const char *blob, size_t blob_size) 
@@ -266,50 +255,13 @@ static const struct metric_scheme g_metric_scheme_table[] = {
 /* -------------------------------------------------------------------------- */
 /*                                   metric                                   */
 /* -------------------------------------------------------------------------- */
-struct metric_parameter *construct_parameters(enum metric_type type, ...)
-{
-    struct metric_parameter *paras = (struct metric_parameter *)malloc(sizeof(struct metric_parameter));
-    va_list ap;
-    va_start(ap, type);
-    switch (type) {
-        case METRIC_TYPE_COUNTER:
-            break;
-        case METRIC_TYPE_HLL:
-            paras->hll.precision = (char)va_arg(ap, int);
-            break;
-        case METRIC_TYPE_HISTOGRAM:
-            paras->hdr.lowest_trackable_value = va_arg(ap, long long);
-            paras->hdr.highest_trackable_value = va_arg(ap, long long);
-            paras->hdr.significant_figures = va_arg(ap, int);
-            break;
-        default:
-            assert(0);
-    }
-    va_end(ap);
-    return paras;
-}
 
-struct metric_manifest *metric_info_new(const char *name, enum metric_type type, struct metric_parameter *para)
-{
-    struct metric_manifest *info = (struct metric_manifest *)malloc(sizeof(struct metric_manifest));
-    info->name = strdup(name);
-    info->paras = para;
-    info->type = type;
-    return info;
-}
-
-void metric_info_free(struct metric_manifest *info)
-{
-    free(info->name);
-    free(info->paras);
-    free(info);
-}
 
-struct metric *metric_new(const char *name, enum metric_type type, struct metric_parameter *para)
+struct metric *metric_new(const struct metric_manifest *manifest)
 {
     struct metric *pthis = (struct metric *)calloc(1, sizeof(struct metric));
-    pthis->info = metric_info_new(name, type, para);
-    pthis->data = NULL;
+    pthis->data = g_metric_scheme_table[manifest->type].new(manifest->paras);
+    pthis->type = manifest->type;
 
     return pthis;
 }
@@ -320,55 +272,30 @@ void metric_free(struct metric *pthis)
         return;
     }
 
-    if (pthis->data) {
-        g_metric_scheme_table[pthis->info->type].free(pthis->data);
-    }
-    metric_info_free(pthis->info);
+    g_metric_scheme_table[pthis->type].free(pthis->data);
     free(pthis);
 }
 
 void metric_reset(struct metric *pthis) {
-    g_metric_scheme_table[pthis->info->type].reset(pthis->data);
-}
-
-void metric_info_copy(struct metric_manifest *dest, const struct metric_manifest *src)
-{
-    dest->name = strdup(src->name);
-    if (src->paras == NULL) {
-        dest->paras = NULL;
-    } else {
-        dest->paras = (struct metric_parameter *)malloc(sizeof(struct metric_parameter));
-        memcpy(dest->paras, src->paras, sizeof(struct metric_parameter));
-    }
-
-    dest->type = src->type;
-}
-
-struct metric *metric_fork(const struct metric *src) {
-    struct metric *dest = (struct metric *)malloc(sizeof(struct metric));
-    dest->info = (struct metric_manifest *)malloc(sizeof(struct metric_manifest));
-    metric_info_copy(dest->info, src->info);
-
-    dest->data = g_metric_scheme_table[dest->info->type].new(dest->info->paras);
-
-    return dest;
+    g_metric_scheme_table[pthis->type].reset(pthis->data);
 }
 
 struct metric *metric_copy(const struct metric *src) {
-    struct metric *dest = metric_fork(src);
-    g_metric_scheme_table[dest->info->type].copy(dest->data, src->data);
+    struct metric *dest = (struct metric *)malloc(sizeof(struct metric));
+    dest->type = src->type;
+    dest->data = g_metric_scheme_table[dest->type].copy(src->data);
 
     return dest;
 }
 
 //return -1 when merge error. 0 when success. 1 when src has no cell.
 int metric_merge(struct metric *dest, const struct metric *src) {
-    return g_metric_scheme_table[dest->info->type].merge(dest->data, src->data);
+    return g_metric_scheme_table[dest->type].merge(dest->data, src->data);
 }
 
 void metric_get_plain_blob(const struct metric *pthis, char **blob, size_t *blob_size) {
     struct metric_data *data = pthis->data;
-    enum metric_type type = pthis->info->type;
+    enum metric_type type = pthis->type;
     if (type == METRIC_TYPE_HLL) {
         ST_hyperloglog_serialize_for_networking(data->hll, blob, blob_size);
         return;
@@ -384,10 +311,6 @@ void metric_get_plain_blob(const struct metric *pthis, char **blob, size_t *blob
 /*                          specific metric operations                                */ 
 /* -------------------------------------------------------------------------- */
 
-struct metric *metric_counter_new(const char *name) {
-    return metric_new(name, METRIC_TYPE_COUNTER, NULL);
-}
-
 void metric_counter_incrby(struct metric *pthis, long long value) {
     struct metric_counter_or_gauge *counter = pthis->data->counter;
     counter->value += value;
@@ -402,11 +325,6 @@ long long metric_counter_get(const struct metric *pthis) {
     return pthis->data->counter->value;
 }
 
-struct metric *metric_hll_new(const char *name, unsigned char precision) {
-    struct metric_parameter *para = construct_parameters(METRIC_TYPE_HLL, precision);
-    return metric_new(name, METRIC_TYPE_HLL, para);
-}
-
 void metric_hll_add(struct metric *pthis, const char *key, size_t key_len) {
     ST_hyperloglog_add(pthis->data->hll, key, key_len);
 }
@@ -415,14 +333,9 @@ double metric_hll_get(const struct metric *pthis) {
     return ST_hyperloglog_count(pthis->data->hll);
 }
 
-struct metric *metric_histogram_new(const char *name, long long lowest_trackable_value, long long highest_trackable_value, int significant_figures) {
-    struct metric_parameter *para = construct_parameters(METRIC_TYPE_HISTOGRAM, lowest_trackable_value, highest_trackable_value, significant_figures);
-    return metric_new(name, METRIC_TYPE_HISTOGRAM, para);
-}
-
 int metric_histogram_record(struct metric *pthis, long long value) {
-    if (value > pthis->info->paras->hdr.highest_trackable_value) {
-        value = pthis->info->paras->hdr.highest_trackable_value;
+    if (value > pthis->data->hdr->highest_trackable_value) {
+        value = pthis->data->hdr->highest_trackable_value;
     }
     bool ret = hdr_record_value(pthis->data->hdr, value);
     if (!ret) {
@@ -438,13 +351,3 @@ long long metric_histogram_value_at_percentile(const struct metric *pthis, doubl
 long long metric_histogram_count_le_value(const struct metric *pthis, long long value) {
     return hdr_count_le_value(pthis->data->hdr, value);
 }
-
-const char *metric_get_name(const struct metric *metrics)
-{
-    return metrics->info->name;
-}
-
-enum metric_type metric_get_type(const struct metric *metrics)
-{
-    return metrics->info->type;
-}
\ No newline at end of file
diff --git a/src/metrics/metric.h b/src/metrics/metric.h
index 3e04bd2..822735b 100644
--- a/src/metrics/metric.h
+++ b/src/metrics/metric.h
@@ -3,31 +3,24 @@
 #include <stddef.h>
 #include <stdbool.h>
 #include "fieldstat.h" // for enum metric_type
+#include "metric_manifest.h"
 
 struct metric;
 
-// TODO: 删掉metric，单纯提供一个metric_manifest 类，直接把定义写进来，由fieldstat 和 cube 共同依赖。
-
-const char *metric_get_name(const struct metric *metrics); // don't free the return value
-enum metric_type metric_get_type(const struct metric *metrics);
 void metric_free(struct metric *pthis);
 void metric_reset(struct metric *pthis);
 struct metric *metric_copy(const struct metric *src);
-struct metric *metric_fork(const struct metric *src);
-
+struct metric *metric_new(const struct metric_manifest *manifest);
 int metric_merge(struct metric *dest, const struct metric *src);
 void metric_get_plain_blob(const struct metric *pthis, char **blob, size_t *blob_size);
 
-struct metric *metric_counter_new(const char *name);
 void metric_counter_incrby(struct metric *pthis, long long value);
 void metric_counter_set(struct metric *pthis, long long value);
 long long metric_counter_get(const struct metric *pthis);
 
-struct metric *metric_hll_new(const char *name, unsigned char precision);
 void metric_hll_add(struct metric *pthis, const char *key, size_t key_len);
 double metric_hll_get(const struct metric *pthis);
 
-struct metric *metric_histogram_new(const char *name, long long lowest_trackable_value, long long highest_trackable_value, int significant_figures);
 int metric_histogram_record(struct metric *pthis, long long value);
 // for the meaning of these two query method, refer to https://prometheus.io/docs/concepts/metric_types/, and search for "histogram" and "summary"
 long long metric_histogram_value_at_percentile(const struct metric *pthis, double percentile);
diff --git a/src/metrics/metric_manifest.h b/src/metrics/metric_manifest.h
new file mode 100644
index 0000000..83d9ad7
--- /dev/null
+++ b/src/metrics/metric_manifest.h
@@ -0,0 +1,23 @@
+#pragma once
+
+struct para_hdr{
+    long long lowest_trackable_value;
+    long long highest_trackable_value;
+    int significant_figures;
+};
+
+struct para_hll {
+    char precision;
+};
+
+union metric_parameter {
+    struct para_hdr hdr;
+    struct para_hll hll;
+};
+
+struct metric_manifest {
+    int id;
+    char *name;
+    union metric_parameter *paras;
+    enum metric_type type;
+};
\ No newline at end of file
diff --git a/src/tags/heavy_keeper.c b/src/tags/heavy_keeper.c
index 03cf7f5..e44b893 100644
--- a/src/tags/heavy_keeper.c
+++ b/src/tags/heavy_keeper.c
@@ -593,6 +593,11 @@ unsigned my_max(unsigned a, unsigned b) {
 */
 int heavy_keeper_add(struct heavy_keeper *heavy_keeper, const char *key, size_t key_len, int count, void *arg) {
     if (count == 0) {
+        const struct heap_entry *entry = sorted_set_find_entry(heavy_keeper->top_K_heap, key, key_len);
+        if (entry != NULL && !sorted_set_entry_dying(entry)) {
+            return 0;
+        }
+
         if (sorted_set_cardinality(heavy_keeper->top_K_heap) < heavy_keeper->K) {
             sorted_set_insert(heavy_keeper->top_K_heap, key, key_len, count, arg);
             return 1;
diff --git a/src/tags/my_ut_hash.c b/src/tags/my_ut_hash.c
index af27356..0da2a66 100644
--- a/src/tags/my_ut_hash.c
+++ b/src/tags/my_ut_hash.c
@@ -473,4 +473,38 @@ char *tag_hash_key_get_compound_key(const struct tag_hash_key *tag)
     size_t key_size;
     build_dynamic_cell_key(tag->tags, tag->n_my_tag, &key, &key_size);
     return key;
+}
+
+void tag_hash_key_print(const struct tag_hash_key *tag)
+{
+    if (tag == NULL) {
+        printf("tag is NULL\n");
+        return;
+    }
+    if (tag->n_my_tag == 0) {
+        printf("no tags\n");
+        return;
+    }
+
+    printf("n_my_tag: %zu\n", tag->n_my_tag);
+
+    for (int i = 0; i < (int)tag->n_my_tag; i++) {
+        struct fieldstat_tag *tmp_tag = &tag->tags[i];
+        switch (tmp_tag->type)
+        {
+            case TAG_INTEGER:
+                printf("%s: %lld\n", tmp_tag->key, tmp_tag->value_longlong);
+                break;
+            case TAG_DOUBLE:
+                printf("%s: %f\n", tmp_tag->key, tmp_tag->value_double);
+                break;
+            case TAG_CSTRING:
+                printf("%s: %s\n", tmp_tag->key, tmp_tag->value_str);
+                break;
+            default:
+                assert(0);
+                break;
+        }
+    }
+    printf("============================\n");
 }
\ No newline at end of file
diff --git a/src/tags/my_ut_hash.h b/src/tags/my_ut_hash.h
index 4830804..c3ecd78 100644
--- a/src/tags/my_ut_hash.h
+++ b/src/tags/my_ut_hash.h
@@ -28,6 +28,7 @@ char *tag_hash_key_get_compound_key(const struct tag_hash_key *tag);
 int tag_hash_key_cmp(const struct tag_hash_key *a, const struct tag_hash_key *b);
 
 void build_dynamic_cell_key(const struct fieldstat_tag tags[], size_t n_tags, char **out_key, size_t *out_key_size);
+void tag_hash_key_print(const struct tag_hash_key *tag);
 
 #ifdef __cplusplus
 }
diff --git a/test/profiling/main.c b/test/profiling/main.c
index caf6542..3e498af 100644
--- a/test/profiling/main.c
+++ b/test/profiling/main.c
@@ -14,8 +14,6 @@
 const struct fieldstat_tag TEST_TAG_INT = {"INT key_", TAG_INTEGER, {.value_longlong = 100}};
 
 
-// todo: 问一下贺岚风，tag是什么样子的
-
 int main () {
     printf("Start profiling...\n");
     clock_t start, end;
diff --git a/test/test_register_and_reset.cpp b/test/test_register_and_reset.cpp
index 3e3f1dd..8a32efa 100644
--- a/test/test_register_and_reset.cpp
+++ b/test/test_register_and_reset.cpp
@@ -517,32 +517,6 @@ TEST(calibrate, master_many_cube)
     fieldstat_free(target);
 }
 
-TEST(test_register, get_cube_mode)
-{
-    // int fieldstat_get_cube_mode(const struct fieldstat *instance, int cube_id, enum sampling_mode *mode, int *primary_metric_id)
-
-    struct fieldstat *instance = fieldstat_new();
-    int cube_id = fieldstat_create_cube(instance, &TEST_SHARED_TAG, 1, SAMPLING_MODE_TOPK, 10);
-    fieldstat_register_counter(instance, "counter");
-    fieldstat_register_counter(instance, "counter2");
-    fieldstat_cube_set_primary_metric(instance, cube_id, 1);
-    int cube_id2 = fieldstat_create_cube(instance, &TEST_TAG_INT, 1, SAMPLING_MODE_COMPREHENSIVE, 10);
-
-    enum sampling_mode mode;
-    int primary_metric_id;
-    EXPECT_EQ(fieldstat_get_cube_mode(instance, cube_id, &mode, &primary_metric_id), 0);
-    EXPECT_EQ(mode, SAMPLING_MODE_TOPK);
-    EXPECT_EQ(primary_metric_id, 1);
-
-    EXPECT_EQ(fieldstat_get_cube_mode(instance, cube_id2, &mode, &primary_metric_id), 0);
-    EXPECT_EQ(mode, SAMPLING_MODE_COMPREHENSIVE);
-    EXPECT_EQ(primary_metric_id, -1);
-
-    EXPECT_EQ(fieldstat_get_cube_mode(instance, 100, &mode, &primary_metric_id), FS_ERR_INVALID_CUBE_ID);
-
-    fieldstat_free(instance);
-}
-
 // issue： https://jira.geedge.net/browse/TSG-20140
 // 流程：
 // calibrate调用前：