sp: refactor aggregate functions (#3288)

Signed-off-by: Masoud Koleini <[email protected]>

Author: koleini

include/fluent-bit/stream_processor/flb_sp.h

@@ -29,13 +29,13 @@
 #include <monkey/mk_core.h>
 #include <rbtree.h>
 
-/* Aggr num type */
+/* Aggregate num type */
 #define FLB_SP_NUM_I64       0
 #define FLB_SP_NUM_F64       1
 #define FLB_SP_BOOLEAN       2
 #define FLB_SP_STRING        3
 
-struct aggr_num {
+struct aggregate_num {
     int type;
     int ops;
     int64_t i64;
@@ -44,17 +44,21 @@ struct aggr_num {
     flb_sds_t string;
 };
 
-struct timeseries {
-    struct aggr_num *nums;
+struct aggregate_data {
+    struct aggregate_num *nums;
     struct mk_list _head;
 };
 
 struct timeseries_forecast {
-    struct aggr_num *nums;
+    struct aggregate_num *nums;
     struct mk_list _head;
 
-    double *offset;
-    double *latest_x;
+    // future time to forecast
+    double future_time;
+
+    // time offset (the first time value captured)
+    double offset;
+    double latest_x;
 
     double sigma_x;
     double sigma_y;
@@ -63,15 +67,15 @@ struct timeseries_forecast {
     double sigma_x2;
 };
 
-struct aggr_node {
+struct aggregate_node {
     int groupby_keys;
     int records;
     int nums_size;
-    struct aggr_num *nums;
-    struct aggr_num *groupby_nums;
+    struct aggregate_num *nums;
+    struct aggregate_num *groupby_nums;
 
-    /* Timeseries data */
-    struct timeseries **ts;
+    /* Aggregate data */
+    struct aggregate_data **aggregate_data;
 
     /* To keep track of the aggregation nodes */
     struct rb_tree_node _rb_head;
@@ -85,8 +89,8 @@ struct flb_sp_window_data {
 };
 
 struct flb_sp_hopping_slot {
-    struct rb_tree aggr_tree;
-    struct mk_list aggr_list;
+    struct rb_tree aggregate_tree;
+    struct mk_list aggregate_list;
     int records;
     struct mk_list _head;
 };
@@ -98,8 +102,8 @@ struct flb_sp_task_window {
     struct mk_event event;
     struct mk_event event_hop;
 
-    struct rb_tree aggr_tree;
-    struct mk_list aggr_list;
+    struct rb_tree aggregate_tree;
+    struct mk_list aggregate_list;
 
     /* Hopping window parameters */
     /*
@@ -133,7 +137,7 @@ struct flb_sp_task {
      */
     void *stream;
 
-    int aggr_keys;           /* do commands contains aggregated keys ? */
+    int aggregate_keys;      /* do commands contains aggregate keys? */
     struct flb_sp *sp;       /* parent context */
     struct flb_sp_cmd *cmd;  /* (SQL) commands */
 
@@ -167,8 +171,8 @@ struct flb_sp_task *flb_sp_task_create(struct flb_sp *sp, const char *name,
                                        const char *query);
 int flb_sp_fd_event(int fd, struct flb_sp *sp);
 void flb_sp_task_destroy(struct flb_sp_task *task);
-void groupby_nums_destroy(struct aggr_num *groupby_nums, int size);
-void flb_sp_aggr_node_destroy(struct flb_sp_cmd *cmd,
-                              struct aggr_node *aggr_node);
+void groupby_nums_destroy(struct aggregate_num *groupby_nums, int size);
+void flb_sp_aggregate_node_destroy(struct flb_sp_cmd *cmd,
+                                   struct aggregate_node *aggregate_node);
 
 #endif

include/fluent-bit/stream_processor/flb_sp_aggregate_func.h

@@ -0,0 +1,56 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2020 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_SP_AGGREGATE_FUNC_H
+#define FLB_SP_AGGREGATE_FUNC_H
+
+
+typedef void (*aggregate_function_destroy)(struct aggregate_node *,
+                                           int);
+
+typedef int (*aggregate_function_clone)(struct aggregate_node *,
+                                        struct aggregate_node *,
+                                        struct flb_sp_cmd_key *,
+                                        int);
+
+typedef void (*aggregate_function_add)(struct aggregate_node *,
+                                       struct flb_sp_cmd_key *,
+                                       int,
+                                       struct flb_time *,
+                                       int64_t, double);
+
+typedef void (*aggregate_function_calc)(struct aggregate_node *,
+                                        struct flb_sp_cmd_key *,
+                                        msgpack_packer *,
+                                        int);
+
+typedef void (*aggregate_function_remove)(struct aggregate_node *,
+                                          struct aggregate_node *,
+                                          int);
+
+extern char aggregate_func_string[AGGREGATE_FUNCTIONS][sizeof("TIMESERIES_FORECAST") + 1];
+
+extern aggregate_function_clone aggregate_func_clone[AGGREGATE_FUNCTIONS];
+extern aggregate_function_add aggregate_func_add[AGGREGATE_FUNCTIONS];
+extern aggregate_function_calc aggregate_func_calc[AGGREGATE_FUNCTIONS];
+extern aggregate_function_remove aggregate_func_remove[AGGREGATE_FUNCTIONS];
+extern aggregate_function_destroy aggregate_func_destroy[AGGREGATE_FUNCTIONS];
+
+#endif

include/fluent-bit/stream_processor/flb_sp_parser.h

@@ -34,6 +34,10 @@
 #define FLB_SP_COUNT     3
 #define FLB_SP_MIN       4
 #define FLB_SP_MAX       5
+#define FLB_SP_FORECAST  6
+
+/* Update this whenever a new aggregate function is added */
+#define AGGREGATE_FUNCTIONS    6
 
 /* Date time functions */
 #define FLB_SP_NOW             10
@@ -43,12 +47,6 @@
 #define FLB_SP_RECORD_TAG      20
 #define FLB_SP_RECORD_TIME     21
 
-/* Timeseries functions */
-#define FLB_SP_TIMESERIES_START  30
-#define FLB_SP_FORECAST          30
-#define FLB_SP_FORECAST_R        31
-#define FLB_SP_TIMESERIES_END    39
-
 /* Status */
 #define FLB_SP_OK            0
 #define FLB_SP_ERROR        -1
@@ -78,7 +76,6 @@ enum Expressions {
     FLB_EXP_NULL,
     FLB_EXP_FUNC,
     FLB_EXP_PARAM,
-    FLB_EXP_TIMESERIES,
 };
 
 /* Logical operation */
@@ -124,13 +121,13 @@ struct flb_sp_cmd_key {
     int aggr_func;             /* Aggregation function */
     int time_func;             /* Time function */
     int record_func;           /* Record function */
-    int timeseries_func;       /* Timeseries function */
     flb_sds_t name;            /* Parent Key name */
     flb_sds_t alias;           /* Key output alias (key AS alias) */
-    flb_sds_t name_keys;       /* Key name with sub-keys */
     void *gb_key;              /* Key name reference to gb_key */
+    // TODO: make it a general union type (or array of values)
+    int constant;              /* constant parameter value
+                                  (used specifically for timeseries_forecast) */
     struct mk_list *subkeys;   /* sub-keys selection */
-    struct flb_exp_timeseries *timeseries;  /* Timeseries functions */
     struct mk_list _head;      /* Link to flb_sp_cmd->keys */
 };
 
@@ -158,13 +155,7 @@ struct flb_sp_cmd {
     struct flb_sp_window window;   /* WINDOW window in select statement */
 
     struct mk_list gb_keys;        /* list head of group-by record fields */
-
-    int timeseries_num;            /* Number of timeseries functions */
-    /*
-     * This keeps the temporary list of parameters parameters in timeseries
-     * functions during SQL statement parsing.
-     */
-    struct mk_list *tmp_params;
+    char *alias;
 
     /*
      * When parsing a SQL statement that have references to keys with sub-keys
@@ -244,21 +235,6 @@ struct flb_exp_param {
     struct flb_exp *param;
 };
 
-struct flb_exp_timeseries {
-    int type;
-    struct mk_list _head;
-    struct mk_list params;
-
-    struct timeseries *(*cb_func_alloc) (int);
-    struct timeseries *(*cb_func_clone) (struct timeseries *);
-    void (*cb_func_add) (struct timeseries *, struct flb_time *);
-    void (*cb_func_rem) (struct timeseries *, struct timeseries *,
-                         struct flb_time *);
-    void (*cb_func_calc) (struct timeseries *, struct flb_sp_cmd_key *,
-                          msgpack_packer *, int, struct flb_time *);
-    void (*cb_func_destroy) (struct timeseries *);
-};
-
 struct flb_sp_cmd *flb_sp_cmd_create(const char *sql);
 void flb_sp_cmd_destroy(struct flb_sp_cmd *cmd);
 
@@ -271,9 +247,9 @@ void flb_sp_cmd_stream_prop_del(struct flb_sp_cmd_prop *prop);
 const char *flb_sp_cmd_stream_prop_get(struct flb_sp_cmd *cmd, const char *key);
 
 /* Selection keys */
-int flb_sp_cmd_key_add(struct flb_sp_cmd *cmd, int func,
-                       const char *key_name, const char *key_alias);
+int flb_sp_cmd_key_add(struct flb_sp_cmd *cmd, int func, const char *key_name);
 void flb_sp_cmd_key_del(struct flb_sp_cmd_key *key);
+void flb_sp_cmd_alias_add(struct flb_sp_cmd *cmd, const char *key_alias);
 int flb_sp_cmd_source(struct flb_sp_cmd *cmd, int type, const char *source);
 void flb_sp_cmd_dump(struct flb_sp_cmd *cmd);
 
@@ -307,9 +283,7 @@ void flb_sp_cmd_gb_key_del(struct flb_sp_cmd_gb_key *key);
 
 void flb_sp_cmd_limit_add(struct flb_sp_cmd *cmd, int limit);
 
-/* Timeseries */
-int flb_sp_cmd_param_add(struct flb_sp_cmd *cmd, int func, struct flb_exp *param);
-int flb_sp_cmd_timeseries(struct flb_sp_cmd *cmd, char *func, const char *key_alias);
-void flb_cmd_params_del(struct mk_list *params);
+int flb_sp_cmd_timeseries_forecast(struct flb_sp_cmd *cmd, int func,
+                                   const char *key_name, int seconds);
 
 #endif