Biflow aggregator - Add UNIQUE_COUNT aggregation function

Author: Damir Zainullin

biflow_aggregator/aggregator.cpp

@@ -184,6 +184,18 @@ int Field_template::assign_append() noexcept
     return 0;
 }
 
+template<typename T>
+int Field_template::assign_unique_count() noexcept
+{
+    typename_size = sizeof(T);
+    ag_fnc = unique_count<T>;
+    post_proc_fnc = Unique_count_data<T>::postprocessing;
+    init_fnc = Unique_count_data<T>::init;
+    deinit_fnc = Unique_count_data<T>::deinit;
+    ag_data_size = sizeof(Unique_count_data<T>);
+    return 0;
+}
+
 template<Field_type ag_type, typename T>
 int Field_template::assign_min_max() noexcept
 {
@@ -431,6 +443,23 @@ int Field_template::set_templates(const Field_type ag_type, const ur_field_type_
             std::cerr << "Only string and int, uint, float, double, mac, time, and IP array can be used to APPEND function." << std::endl;
             return 1;
         }
+    case UNIQUE_COUNT:
+        switch (ur_f_type) {
+        case UR_TYPE_UINT8:  return assign_unique_count<uint8_t>();
+        case UR_TYPE_INT8:   return assign_unique_count<int8_t>();
+        case UR_TYPE_UINT16: return assign_unique_count<uint16_t>();
+        case UR_TYPE_INT16:  return assign_unique_count<int16_t>();
+        case UR_TYPE_UINT32: return assign_unique_count<uint32_t>();
+        case UR_TYPE_INT32:  return assign_unique_count<int32_t>();
+        case UR_TYPE_UINT64: return assign_unique_count<uint64_t>();
+        case UR_TYPE_INT64:  return assign_unique_count<int64_t>();
+        case UR_TYPE_MAC:    return assign_unique_count<Mac_addr>();
+        case UR_TYPE_IP:     return assign_unique_count<uint128_t>();
+        case UR_TYPE_STRING:   return assign_unique_count<char>();
+        default:
+            std::cerr << "Only string and int, uint, float, double, mac, time, and IP array can be used to UNIQUE_COUNT function." << std::endl;
+            return 1;
+        }
     default:
         assert("Invalid case option.\n");
         return 1;
@@ -732,8 +761,10 @@ int Field_template::set_templates_dir(const ur_field_type_t ur_f_type, const ur_
     }
 }
 
-Field::Field(const Field_config cfg, const ur_field_id_t ur_fid, const ur_field_id_t ur_r_fid) :
-    ur_fid(ur_fid), ur_r_fid(ur_r_fid), ur_sort_key_id(0), ur_sort_key_type()
+Field::Field(const Field_config cfg, const ur_field_id_t ur_fid_in, const ur_field_id_t ur_r_fid_in,
+    const ur_field_id_t ur_fid_out, const ur_field_id_t ur_r_fid_out) 
+    : ur_fid(ur_fid_in), ur_r_fid(ur_r_fid_in), ur_fid_out(ur_fid_out), 
+    ur_r_fid_out(ur_r_fid_out), ur_sort_key_id(0), ur_sort_key_type()
 {
     ur_field_type_t ur_field_type = ur_get_type(ur_fid);
 
@@ -813,6 +844,11 @@ void Fields::init(uint8_t *memory)
             data.first.init(memory, &cfg);
             break;
         }
+        case UNIQUE_COUNT: {
+            struct Config_unique_count cfg = {data.first.limit};
+            data.first.init(memory, &cfg);
+            break;
+        }
         case SORTED_MERGE:
         case SORTED_MERGE_DIR: {
             struct Config_sorted_merge cfg = {data.first.limit, data.first.delimiter, data.first.sort_type};

biflow_aggregator/aggregator.h

@@ -40,6 +40,7 @@ enum Field_type {
     LAST,
     LAST_NON_EMPTY,
     APPEND,
+    UNIQUE_COUNT,
     SORTED_MERGE,
     SORTED_MERGE_DIR,
     INVALID_TYPE,
@@ -95,6 +96,9 @@ class Field_template {
     template<typename T>
     int assign_append() noexcept;
 
+    template<typename T>
+    int assign_unique_count() noexcept;
+
     template<Field_type ag_type, typename T>
     int assign_min_max() noexcept;
 
@@ -275,7 +279,7 @@ struct Field_config {
     char delimiter;
 
     /**
-     * @brief Max size of append and sortd merge data
+     * @brief Max size of data
      */
     std::size_t limit;
 
@@ -301,6 +305,16 @@ class Field : public Field_config, public Field_template {
      */
     ur_field_id_t ur_r_fid;
 
+    /**
+     * @brief ID of output unirec field
+     */
+    ur_field_id_t ur_fid_out;
+
+    /**
+     * @brief Reverse ID of output unirec field
+     */
+    ur_field_id_t ur_r_fid_out;
+
     /**
      * @brief ID of sort key unirec field
      * 
@@ -321,8 +335,11 @@ class Field : public Field_config, public Field_template {
      * @param cfg      Field configuration
      * @param ur_fid   Field ID
      * @param ur_r_fid Reverse field ID
+     * @param ur_fid_out Output field ID
+     * @param ur_r_fid_out Reverse output field ID
      */
-    Field(const Field_config cfg, const ur_field_id_t ur_fid, const ur_field_id_t ur_r_fid);
+    Field(const Field_config cfg, const ur_field_id_t ur_fid, const ur_field_id_t ur_r_fid,
+        const ur_field_id_t ur_fid_out, const ur_field_id_t ur_r_fid_out);
 
     /**
      * @brief Call field init function.

biflow_aggregator/aggregator_functions.h

@@ -15,6 +15,7 @@
 
 #include <iostream>
 #include <limits>
+#include <nemea-common/BloomFilter.hpp>
 
 #include <unirec/unirec.h>
 
@@ -174,6 +175,50 @@ struct Append_data : Config_append {
     }
 };
 
+/**
+ * @brief Configuration for unique count function 
+ */
+struct Config_unique_count {
+    std::size_t filter_size;
+};
+
+/**
+ * @brief Structure used to store data for unique count function.
+ */
+template<typename T>
+struct Unique_count_data : Config_unique_count {
+    bloom_filter filter;
+    std::size_t unique_count;
+
+    Unique_count_data(const bloom_parameters& parameters)
+    : filter(parameters)
+    {}
+
+    static inline void init(void* memory, const void* config)
+    {
+        const auto* config_unique_count = static_cast<const Config_unique_count*>(config);
+        bloom_parameters parameters;
+        parameters.projected_element_count = config_unique_count->filter_size;
+        parameters.false_positive_probability = 0.01;
+        parameters.compute_optimal_parameters();
+        new(memory) Unique_count_data<T>(parameters);
+    }
+
+    static inline void deinit(void* memory)
+    {
+        Unique_count_data<T>* unique_count_data = static_cast<Unique_count_data<T>*>(memory);
+        unique_count_data->unique_count = 0;
+        unique_count_data->~Unique_count_data<T>();
+    }
+
+    static inline const void* postprocessing(void* memory, std::size_t& elem_cnt) noexcept
+    {
+        Unique_count_data<T>* unique_count_data = static_cast<Unique_count_data<T>*>(memory);
+        elem_cnt = 1;
+        return static_cast<void*>(&unique_count_data->unique_count);
+    }
+};
+
 /**
  * @brief Configuration to sorted merge function
  */
@@ -519,6 +564,32 @@ inline void append(const void *src, void *dst) noexcept
             static_cast<const T*>(src_data->ptr_first) + src_data->cnt_elements);
 }
 
+/**
+ * @brief Inserts element from src pointer into bloom filter.
+ * @tparam T template type variable.
+ * @param [in] src pointer to source of new data.
+ * @param [in,out] dst pointer to already stored data which will be updated (modified).
+ */
+template<typename T>
+inline void unique_count(const void* src, void* dst) noexcept
+{
+    Unique_count_data<T>* unique_count_data = static_cast<Unique_count_data<T>*>(dst);
+
+    if (std::is_same<T, char>::value) {
+        const ur_array_data* src_data = (static_cast<const ur_array_data*>(src));
+        if (src_data->cnt_elements != 0 && !unique_count_data->filter.contains(
+            static_cast<const unsigned char*>(src_data->ptr_first), src_data->cnt_elements)) {
+            unique_count_data->filter.insert(static_cast<const unsigned char*>(src_data->ptr_first), src_data->cnt_elements);
+            unique_count_data->unique_count++;
+        }
+        return;
+    }
+    if (!unique_count_data->filter.contains(static_cast<const unsigned char*>(src), sizeof(T))) {
+        unique_count_data->filter.insert(static_cast<const unsigned char*>(src), sizeof(T));
+        unique_count_data->unique_count++;
+    }    
+}
+
 template <typename T, typename K>
 inline void sorted_merge(const void *src, void *dst) noexcept
 {

biflow_aggregator/configuration.cpp

@@ -128,6 +128,7 @@ agg::Field_type Configuration::get_field_type(const char *input)
     if (!std::strcmp(input, "BITAND")) return agg::BIT_AND;
     if (!std::strcmp(input, "BITOR")) return agg::BIT_OR;
     if (!std::strcmp(input, "APPEND")) return agg::APPEND;
+    if (!std::strcmp(input, "UNIQUE_COUNT")) return agg::UNIQUE_COUNT;
     if (!std::strcmp(input, "SORTED_MERGE")) return agg::SORTED_MERGE;
     if (!std::strcmp(input, "SORTED_MERGE_DIR")) return agg::SORTED_MERGE_DIR;
     std::cerr << "Invalid type field. Given: " << input << ", Expected: KEY|SUM|MIN|MAX|AVG|FIRST|FIRST_NON_EMPTY|LAST|LAST_NON_EMPTY|BITAND|BITOR|APPEND|SORTED_MERGE|SORTED_MERGE_DIR." << std::endl;

biflow_aggregator/main.cpp

@@ -198,10 +198,10 @@ proccess_and_send(agg::Aggregator<agg::FlowKey>& agg, const agg::FlowKey& key, c
             agg_data = field->post_processing(&flow_data.ctx->data[agg_field.second], size, elem_cnt);
         }
         if (ur_is_array(field->ur_fid)) {
-            ur_array_allocate(out_tmplt, out_rec, field->ur_fid, elem_cnt);
-            std::memcpy(ur_get_ptr_by_id(out_tmplt, out_rec, field->ur_fid), agg_data, size * elem_cnt);
+            ur_array_allocate(out_tmplt, out_rec, field->ur_fid_out, elem_cnt);
+            std::memcpy(ur_get_ptr_by_id(out_tmplt, out_rec, field->ur_fid_out), agg_data, size * elem_cnt);
         } else {
-            field_id = flow_data.reverse ? field->ur_r_fid : field->ur_fid;
+            field_id = flow_data.reverse ? field->ur_r_fid_out : field->ur_fid_out;
             std::memcpy(ur_get_ptr_by_id(out_tmplt, out_rec, field_id), agg_data, size);
         }
     }
@@ -249,8 +249,19 @@ static int process_format_change(
             if (ur_get_type(ur_fid) == UR_TYPE_STRING)
                 is_string_key = true;
             agg::Key_template::add(ur_fid, ur_r_fid);
+        } else if (field_cfg.type == agg::UNIQUE_COUNT) {
+            const int ur_fid_out = ur_define_field((field_cfg.name + "_UNIQUE_COUNT").c_str(), UR_TYPE_UINT32);
+            out_template.append("," + field_cfg.name + "_UNIQUE_COUNT");
+            int ur_r_fid_out = ur_fid_out;
+            if (ur_fid != ur_r_fid) {
+                ur_define_field((field_cfg.reverse_name + "_UNIQUE_COUNT").c_str(), UR_TYPE_UINT32);
+                out_template.append("," + field_cfg.reverse_name + "_UNIQUE_COUNT");
+            }
+            field_cfg.to_output = false;
+            agg::Field field(field_cfg, ur_fid, ur_r_fid, ur_fid_out, ur_r_fid_out);
+            agg.fields.add_field(field);
         } else {
-            agg::Field field(field_cfg, ur_fid, ur_r_fid);
+            agg::Field field(field_cfg, ur_fid, ur_r_fid, ur_fid, ur_r_fid);
             agg.fields.add_field(field);
         }
         if (field_cfg.to_output)