fieldstat_hll_add_tag

6 files changed, 69 insertions, 13 deletions

diff --git a/src/cube.c b/src/cube.c
index b1123a0..21f3ef8 100644
--- a/src/cube.c
+++ b/src/cube.c
@@ -4,12 +4,15 @@
 #include <assert.h>
 #include <string.h>
 
+#include "uthash.h"
+#define XXH_INLINE_ALL
+#include "xxhash/xxhash.h"
+
 #include "cube.h"
 #include "metric_manifest.h"
 #include "metric.h"
 #include "heavy_keeper.h"
 #include "tag_map.h"
-#include "uthash.h"
 
 #define DEFAULT_N_METRIC 32
 #define DEFAULT_N_CUBE 64
@@ -131,7 +134,7 @@ struct cube_manager *cube_manager_new() {
     return pthis;
 }
 
-static void tag2key(const struct fieldstat_tag tags[], size_t n_tags, char **out_key, size_t *out_key_size)
+static void tags2key(const struct fieldstat_tag tags[], size_t n_tags, char **out_key, size_t *out_key_size)
 {
     if (n_tags == 0) {
         // use a default dummy key
@@ -241,7 +244,7 @@ int cube_manager_find(const struct cube_manager *pthis, const struct fieldstat_t
 {
     char *key;
     size_t key_len;
-    tag2key(identifier, n_tag, &key, &key_len);
+    tags2key(identifier, n_tag, &key, &key_len);
 
     struct cube *node = NULL;
     HASH_FIND(hh, pthis->hash_table, key, key_len, node);
@@ -479,7 +482,7 @@ struct cube *cube_info_new(const struct fieldstat_tag *shared_tags, size_t n_tag
 
     cube->n_shared_tags = n_tag;
     cube->max_n_cell = max_n_cell;
-    tag2key(shared_tags, n_tag, &cube->key, &cube->key_len);
+    tags2key(shared_tags, n_tag, &cube->key, &cube->key_len);
 
     cube->id = -1;
 
@@ -543,7 +546,7 @@ void cube_set_primary_metric(struct cube *cube, int metric_id) {
 struct cell *get_cell(struct cube *cube, const struct fieldstat_tag *tags, size_t n_tag,long long increment, int metric_id) {
     char *tag_in_string;
     size_t tag_len;
-    tag2key(tags, n_tag, &tag_in_string, &tag_len);
+    tags2key(tags, n_tag, &tag_in_string, &tag_len);
     struct exdata_new_args args;
     args.tags = tags;
     args.n_tags = n_tag;
@@ -614,6 +617,38 @@ int cube_hll_add(struct cube *cube, const struct metric_manifest *manifest, cons
     return FS_OK;
 }
 
+uint64_t tags2hash(const struct fieldstat_tag *tag, size_t n_tag) {
+    XXH3_state_t state = {0};
+    XXH3_64bits_reset(&state);
+
+    for (int i = 0; i < n_tag; i++) {
+        XXH3_64bits_update(&state, tag[i].key, strlen(tag[i].key));
+        if (tag[i].type != TAG_CSTRING) {
+            XXH3_64bits_update(&state, &tag[i].value_longlong, sizeof(long long));
+        } else {
+            XXH3_64bits_update(&state, tag[i].value_str, strlen(tag[i].value_str));
+        }
+    }
+
+    return XXH3_64bits_digest(&state);
+}
+
+int cube_hll_add_tag(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, const struct fieldstat_tag *tags_key, size_t n_tag_key) 
+{
+    assert(manifest->type == METRIC_TYPE_HLL);
+    assert(cube->sampling_mode != SAMPLING_MODE_TOPK || (cube->primary_metric_id != manifest->id));
+
+    struct cell *cell_data = get_cell(cube, tags, n_tag, 0, manifest->id);
+    if (cell_data == NULL) {
+        return FS_ERR_TOO_MANY_CELLS;
+    }
+    struct metric *metric = add_or_find_metric_in_cell(manifest, cell_data);
+
+    uint64_t hash = tags2hash(tags_key, n_tag_key);
+    metric_hll_add_hash(metric, hash);
+    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) {
     assert(manifest->type == METRIC_TYPE_COUNTER);
     assert(cube->sampling_mode == SAMPLING_MODE_COMPREHENSIVE || (cube->primary_metric_id != manifest->id || increment >= 0));
@@ -744,7 +779,7 @@ const struct cell *get_cell_by_tag_list(const struct cube *cube, const struct fi
     const struct cell *ret = NULL;
     char *tag_in_string;
     size_t tag_len;
-    tag2key(tags->tag, tags->n_tag, &tag_in_string, &tag_len);
+    tags2key(tags->tag, tags->n_tag, &tag_in_string, &tag_len);
 
     switch (cube->sampling_mode)
     {
diff --git a/src/cube.h b/src/cube.h
index 60e41d4..3a2d20b 100644
--- a/src/cube.h
+++ b/src/cube.h
@@ -22,6 +22,7 @@ struct cube *cube_fork(const struct cube *cube); // only copy the cube configura
 
 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_hll_add_tag(struct cube *cube, const struct metric_manifest *manifest, const struct fieldstat_tag *tags, size_t n_tag, const struct fieldstat_tag *tags_key, size_t n_tag_key);
 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);
 
diff --git a/src/fieldstat.c b/src/fieldstat.c
index e5a5945..cd43128 100644
--- a/src/fieldstat.c
+++ b/src/fieldstat.c
@@ -340,6 +340,21 @@ int fieldstat_hll_add(struct fieldstat *instance, int cube_id, int metric_id, co
 }
 
 // cppcheck-suppress [constParameterPointer, unmatchedSuppression]
+int fieldstat_hll_add_tag(struct fieldstat *instance, int cube_id, int metric_id, const struct fieldstat_tag *tags, size_t n_tag, const struct fieldstat_tag *tags_key, size_t n_tag_key)
+{
+    struct cube *cube = cube_manager_get_cube_by_id(instance->cube_manager, cube_id);
+    if (cube == NULL) {
+        return FS_ERR_INVALID_CUBE_ID;
+    }
+    const struct metric_manifest *manifest = metric_manifest_manager_get_by_id(instance->manifest_manager, metric_id);
+    if (manifest == NULL || manifest->type != METRIC_TYPE_HLL) {
+        return FS_ERR_INVALID_METRIC_ID;
+    }
+
+    return cube_hll_add_tag(cube, manifest, tags, n_tag, tags_key, n_tag_key);
+}
+
+// cppcheck-suppress [constParameterPointer, unmatchedSuppression]
 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)
 {
     struct cube *cube = cube_manager_get_cube_by_id(instance->cube_manager, cube_id);
diff --git a/src/metrics/metric.c b/src/metrics/metric.c
index 91ef47f..343b102 100644
--- a/src/metrics/metric.c
+++ b/src/metrics/metric.c
@@ -329,6 +329,10 @@ void metric_hll_add(struct metric *pthis, const char *key, size_t key_len) {
     hyperloglog_add(pthis->data->hll, key, key_len);
 }
 
+void metric_hll_add_hash(struct metric *pthis, unsigned long long hash) {
+    hyperloglog_add_hash(pthis->data->hll, hash);
+}
+
 double metric_hll_get(const struct metric *pthis) {
     return hyperloglog_count(pthis->data->hll);
 }
diff --git a/src/metrics/metric.h b/src/metrics/metric.h
index 81629a3..30d5f1a 100644
--- a/src/metrics/metric.h
+++ b/src/metrics/metric.h
@@ -19,6 +19,7 @@ void metric_counter_set(struct metric *pthis, long long value);
 long long metric_counter_get(const struct metric *pthis);
 
 void metric_hll_add(struct metric *pthis, const char *key, size_t key_len);
+void metric_hll_add_hash(struct metric *pthis, unsigned long long hash);
 double metric_hll_get(const struct metric *pthis);
 
 int metric_histogram_record(struct metric *pthis, long long value);
diff --git a/src/tags/heavy_keeper.c b/src/tags/heavy_keeper.c
index e48eb8d..5e9332f 100644
--- a/src/tags/heavy_keeper.c
+++ b/src/tags/heavy_keeper.c
@@ -559,7 +559,7 @@ bool if_need_to_decay(struct heavy_keeper *hk, const struct Bucket *bucket, unsi
     return false;
 }
 
-static inline unsigned int cal_hash_val_with_seed(const char *key, size_t key_len, unsigned int seed) {
+static inline unsigned long long cal_hash_val_with_seed(const char *key, size_t key_len, unsigned int seed) {
     return XXH3_64bits_withSeed(key, key_len, seed);
 }
 
@@ -586,12 +586,12 @@ int heavy_keeper_add(struct heavy_keeper *heavy_keeper, const char *key, size_t
     unsigned long long int old_cnt = sorted_set_get_count(summary, key, key_len);
     bool not_in_sorted_set = (old_cnt == INVALID_COUNT);
     unsigned long long maxv = 0;
-    unsigned int fp = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY);
+    unsigned long long fp = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY);
     unsigned int h1 = fp;
-    unsigned int h2 = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY+1);
+    unsigned long long h2 = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY+1);
 
     for (int j = 0; j < heavy_keeper->params.r; j++) {
-        unsigned int hashv = h1 + j * h2; // use `double hashing` strategy
+        unsigned long long hashv = h1 + j * h2; // use `double hashing` strategy
         struct Bucket *bucket = &(heavy_keeper->sketch[j * heavy_keeper->params.w + (hashv % heavy_keeper->params.w)]);
 
         if (bucket->finger_print == fp) {
@@ -712,9 +712,9 @@ static void heavy_keeper_merge_sketch(struct heavy_keeper *dest, const struct he
 
 unsigned long long find_count_in_sketch(struct heavy_keeper *hk, const char *key, size_t key_len) {
     struct Bucket *bucket;
-    unsigned fp = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY);
-    unsigned h1 = fp;
-    unsigned h2 = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY+1);
+    unsigned long long fp = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY);
+    unsigned long long h1 = fp;
+    unsigned long long h2 = cal_hash_val_with_seed(key, key_len, FP_HASH_KEY+1);
 
     unsigned long long maxv = 0;
     for (int array_id = 0; array_id < hk->params.r; array_id++) {