sp: add timeseries function TIMESERIES_FORECAST_R (#1539)

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

Author: koleini

include/fluent-bit/stream_processor/flb_sp_parser.h

@@ -44,7 +44,10 @@
 #define FLB_SP_RECORD_TIME     21
 
 /* Timeseries functions */
-#define FLB_SP_FORECAST        30
+#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

include/fluent-bit/stream_processor/flb_sp_timeseries.h

@@ -22,9 +22,10 @@
 #define FLB_SP_TIMESERIES_H
 
 #include <math.h>
+#include <float.h>
 #include <fluent-bit/stream_processor/flb_sp_parser.h>
 
-#define TIMESERIES_FUNCTIONS_SIZE 1
+#define TIMESERIES_FUNCTIONS_SIZE 2
 
 typedef struct timeseries *(*timeseries_function_alloc_typ)(int);
 typedef struct timeseries *(*timeseries_function_clone_typ)(struct timeseries *);
@@ -209,6 +210,87 @@ void cb_forecast_calc(struct timeseries *ts, struct flb_sp_cmd_key *cmd_key,
     msgpack_pack_float(mp_pck, result);
 }
 
+void cb_forecast_r_calc(struct timeseries *ts, struct flb_sp_cmd_key *cmd_key,
+                        msgpack_packer *mp_pck, int records, struct flb_time *tm)
+{
+    double mean_x;
+    double mean_y;
+    double var_x;
+    double cov_xy;
+    double result;
+    /* y = b0 + b1 * x */
+    double b0;
+    double b1;
+
+    double maximum_x;
+    struct aggr_num *val;
+    struct timeseries_forecast *forecast;
+
+    forecast = (struct timeseries_forecast *) ts;
+
+    mean_x = forecast->sigma_x / records;
+    mean_y = forecast->sigma_y / records;
+    cov_xy = (forecast->sigma_xy / (double) records) - mean_x * mean_y;
+    var_x = (forecast->sigma_x2 / records) - mean_x * mean_x;
+
+    b1 = cov_xy / var_x;
+    b0 = mean_y - b1 * mean_x;
+
+
+    /* Get the cap (4th arguement) */
+    val = ts->nums + 3;
+    switch (val->type) {
+    case FLB_SP_NUM_I64:
+        maximum_x = (double) val->i64;
+        break;
+    case FLB_SP_NUM_F64:
+        maximum_x = val->f64;
+        break;
+    default:
+        return;
+        break;
+    }
+
+    /*
+     * calculate forecast for value (3rd argument).
+     */
+    val = ts->nums + 2;
+
+    if (b1 == 0) {
+        result = maximum_x;
+    }
+    else {
+        switch (val->type) {
+        case FLB_SP_NUM_I64:
+            result = (((double) val->i64 - b0) / b1) - *forecast->latest_x;
+            break;
+        case FLB_SP_NUM_F64:
+            result = ((val->i64 - b0) / b1) - *forecast->latest_x;
+            break;
+        default:
+            result = nan("");
+            break;
+        }
+
+        if (result < 0) {
+            result = maximum_x;
+        }
+    }
+
+    /* pack the result */
+    if (cmd_key->alias) {
+        msgpack_pack_str(mp_pck, flb_sds_len(cmd_key->alias));
+        msgpack_pack_str_body(mp_pck,
+                              cmd_key->alias,
+                              flb_sds_len(cmd_key->alias));
+    }
+    else {
+        msgpack_pack_str(mp_pck, 10);
+        msgpack_pack_str_body(mp_pck, "FORECAST_R", 10);
+    }
+    msgpack_pack_float(mp_pck, result);
+}
+
 void cb_forecast_destroy(struct timeseries *ts)
 {
     struct timeseries_forecast *forecast;
@@ -225,36 +307,43 @@ void cb_forecast_destroy(struct timeseries *ts)
 
 char *timeseries_functions[TIMESERIES_FUNCTIONS_SIZE] = {
     "forecast",
+    "forecast_r",
 };
 
 /* Timeseries function memory allocation */
 timeseries_function_alloc_typ timeseries_functions_alloc_ptr[TIMESERIES_FUNCTIONS_SIZE] = {
     cb_forecast_alloc,
+    cb_forecast_alloc,
 };
 
 /* Timeseries function clone */
 timeseries_function_clone_typ timeseries_functions_clone_ptr[TIMESERIES_FUNCTIONS_SIZE] = {
     cb_forecast_clone,
+    cb_forecast_clone,
 };
 
 /* Timeseries function record addition */
 timeseries_function_add_typ timeseries_functions_add_ptr[TIMESERIES_FUNCTIONS_SIZE] = {
     cb_forecast_add,
+    cb_forecast_add,
 };
 
 /* Timeseries function record removal */
 timeseries_function_rem_typ timeseries_functions_rem_ptr[TIMESERIES_FUNCTIONS_SIZE] = {
     cb_forecast_rem,
+    cb_forecast_rem,
 };
 
 /* Timeseries function calculation */
 timeseries_function_calc_typ timeseries_functions_calc_ptr[TIMESERIES_FUNCTIONS_SIZE] = {
     cb_forecast_calc,
+    cb_forecast_r_calc,
 };
 
 /* Timeseries function calculation */
 timeseries_function_destroy_typ timeseries_functions_destroy_ptr[TIMESERIES_FUNCTIONS_SIZE] = {
     cb_forecast_destroy,
+    cb_forecast_destroy,
 };
 
 #endif

src/stream_processor/parser/flb_sp_parser.c

@@ -184,7 +184,7 @@ struct flb_sp_cmd_key *flb_sp_key_create(struct flb_sp_cmd *cmd, int func,
         /* Record function */
         record_func = func;
     }
-    else if (func >= FLB_SP_FORECAST && func <= FLB_SP_FORECAST) {
+    else if (func >= FLB_SP_TIMESERIES_START && func <= FLB_SP_TIMESERIES_END) {
         /* Timeseries function */
         timeseries_func = func;
     }
@@ -834,9 +834,9 @@ int flb_sp_cmd_timeseries(struct flb_sp_cmd *cmd, char *func, const char *key_al
     for (i = 0; i < TIMESERIES_FUNCTIONS_SIZE; i++)
     {
         ts_name = timeseries_functions[i];
-        if (strncmp(ts_name, func, strlen(ts_name)) == 0)
+        if (strcmp(ts_name, func) == 0)
         {
-            key = flb_sp_key_create(cmd, i + FLB_SP_FORECAST, NULL, key_alias);
+            key = flb_sp_key_create(cmd, i + FLB_SP_TIMESERIES_START, NULL, key_alias);
 
             if (!key) {
                 return -1;

src/stream_processor/parser/sql.l

@@ -81,6 +81,7 @@ TIME                    return TIME;
 
  /* Timeseries Functions */
 TIMESERIES_FORECAST     return TIMESERIES_FORECAST;
+TIMESERIES_FORECAST_R   return TIMESERIES_FORECAST_R;
 
  /* Window Types */
 TUMBLING                return TUMBLING;

src/stream_processor/parser/sql.y

@@ -46,7 +46,7 @@ void yyerror(struct flb_sp_cmd *cmd, const char *query, void *scanner,
 %token RECORD CONTAINS TIME
 
 /* Timeseries functions */
-%token TIMESERIES_FORECAST
+%token TIMESERIES_FORECAST TIMESERIES_FORECAST_R
 
 /* Time functions */
 %token NOW UNIX_TIMESTAMP
@@ -310,15 +310,26 @@ select: SELECT keys FROM source window where groupby ';'
                     flb_free($7);
                   }
                   |
-                  TIMESERIES_FORECAST '(' params ')'
+                  TIMESERIES_FORECAST '(' param ',' param ',' param ')'
                   {
                     flb_sp_cmd_timeseries(cmd, "forecast", NULL);
                   }
                   |
-                  TIMESERIES_FORECAST '(' params ')' AS alias
+                  TIMESERIES_FORECAST '(' param ',' param ',' param ')' AS alias
                   {
-                    flb_sp_cmd_timeseries(cmd, "forecast", $6);
-                    flb_free($6);
+                    flb_sp_cmd_timeseries(cmd, "forecast", $10);
+                    flb_free($10);
+                  }
+                  |
+                  TIMESERIES_FORECAST_R '(' param ',' param ',' param ',' param ')'
+                  {
+                    flb_sp_cmd_timeseries(cmd, "forecast_r", NULL);
+                  }
+                  |
+                  TIMESERIES_FORECAST_R '(' param ',' param ',' param ',' param ')' AS alias
+                  {
+                    flb_sp_cmd_timeseries(cmd, "forecast_r", $12);
+                    flb_free($12);
                   }
                   |
                   NOW '(' ')'
@@ -531,7 +542,6 @@ select: SELECT keys FROM source window where groupby ';'
                {
                  flb_sp_cmd_param_add(cmd, -1, $1);
                }
-        params: param | param ',' params
         value: INTEGER
                {
                  $$ = flb_sp_cmd_condition_integer(cmd, $1);

tests/internal/stream_processor.c

@@ -1179,6 +1179,52 @@ static void cb_forecast_hopping_window(int id, struct task_check *check,
     TEST_CHECK(ret == FLB_TRUE);
 }
 
+static void cb_forecast_r_tumbling_window(int id, struct task_check *check,
+                                          char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect one record only */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    /* Check SUM value result */
+    ret = mp_record_key_cmp(buf, size, 0, "FORECAST_R",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 39.0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* Check AVG value result */
+    ret = mp_record_key_cmp(buf, size, 0, "AVG(usage)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 60.0);
+
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_forecast_r_hopping_window(int id, struct task_check *check,
+                                         char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect one record only */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    /* Check SUM value result */
+    ret = mp_record_key_cmp(buf, size, 0, "FORECAST_R",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 24.0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* Check AVG value result */
+    ret = mp_record_key_cmp(buf, size, 0, "AVG(usage)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 175.0);
+
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
 /* Tests for 'test_window' */
 struct task_check window_checks[] = {
     {
@@ -1210,7 +1256,7 @@ struct task_check window_checks[] = {
         "ADVANCE BY 2 SECOND) WHERE word3 IS NOT NULL;",
         cb_hopping_window_5_second
     },
-    {
+    {    /* FORECAST */
         4, FLB_SP_WINDOW_TUMBLING, 1, 0,
         "timeseries_forecast_window_tumbling",
         "SELECT AVG(usage), TIMESERIES_FORECAST(id, usage, 20) FROM " \
@@ -1224,6 +1270,20 @@ struct task_check window_checks[] = {
         "STREAM:FLB WINDOW HOPPING (5 SECOND, ADVANCE BY 2 SECOND);",
         cb_forecast_hopping_window
     },
+    { /* FORECAST_R */
+        6, FLB_SP_WINDOW_TUMBLING, 1, 0,
+        "timeseries_forecast_r_window_tumbling",
+        "SELECT AVG(usage), TIMESERIES_FORECAST_R(id, usage, 500, 10000) FROM " \
+        "STREAM:FLB WINDOW TUMBLING (5 SECOND);",
+        cb_forecast_r_tumbling_window
+    },
+    {
+        7, FLB_SP_WINDOW_HOPPING, 5, 2,
+        "timeseries_forecast_r_window_hopping",
+        "SELECT AVG(usage), TIMESERIES_FORECAST_R(id, usage, 500, 10000) FROM " \
+        "STREAM:FLB WINDOW HOPPING (5 SECOND, ADVANCE BY 2 SECOND);",
+        cb_forecast_r_hopping_window
+    },
 };
 
 static void test_window()