Merging upstream version 1.44.3.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

7 files changed, 1139 insertions, 0 deletions

diff --git a/fluent-bit/tests/internal/include/sp_cb_functions.h b/fluent-bit/tests/internal/include/sp_cb_functions.h
new file mode 100644
index 000000000..aee3fb393
--- /dev/null
+++ b/fluent-bit/tests/internal/include/sp_cb_functions.h
@@ -0,0 +1,655 @@
+#include "sp_helpers.h"
+
+#ifndef FLB_TEST_CP_FUNCTIONS
+#define FLB_TEST_CP_FUNCTIONS
+
+/* Callback functions to perform checks over results */
+static void cb_select_all(int id, struct task_check *check,
+                          char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect all 11 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 11);
+}
+
+/* Callback test: expect one key per record */
+static void cb_select_id(int id, struct task_check *check,
+                         char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect all 11 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 11);
+
+    ret = mp_count_keys(buf, size);
+    TEST_CHECK(ret == 13);
+}
+
+static void cb_select_cond_1(int id, struct task_check *check,
+                             char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+}
+
+static void cb_select_cond_2(int id, struct task_check *check,
+                             char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 2 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+}
+
+static void cb_select_cond_not_null(int id, struct task_check *check,
+                                    char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+}
+
+static void cb_select_cond_null(int id, struct task_check *check,
+                                char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+}
+
+static void cb_select_not_equal_1(int id, struct task_check *check,
+                                  char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+}
+
+static void cb_select_not_equal_2(int id, struct task_check *check,
+                                  char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 2 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+}
+
+static void cb_select_aggr_count(int id, struct task_check *check,
+                                 char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    /* COUNT(*) is 11 */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "COUNT(*)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 11, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_func_time_now(int id, struct task_check *check,
+                             char *buf, size_t size)
+{
+    int ret;
+    char tmp[32];
+    struct tm *local;
+    time_t now = time(NULL);
+
+    local = localtime(&now);
+    strftime(tmp, sizeof(tmp) - 1, "%Y-%m-%d %H:%M:%S", local);
+
+    /* Expect 2 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+
+    /* NOW() */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "NOW()",
+                            MSGPACK_OBJECT_STR,
+                            tmp, 0, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* tnow */
+    ret = mp_record_key_cmp(buf, size,
+                            1, "tnow",
+                            MSGPACK_OBJECT_STR,
+                            tmp, 0, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_func_time_unix_timestamp(int id, struct task_check *check,
+                                        char *buf, size_t size)
+{
+    int ret;
+    time_t now = time(NULL);
+
+    /* Expect 2 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+
+    /* UNIX_TIMESTAMP() */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "UNIX_TIMESTAMP()",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, now, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* tnow */
+    ret = mp_record_key_cmp(buf, size,
+                            1, "ts",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, now, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+/* No records must be selected */
+static void cb_select_tag_error(int id, struct task_check *check,
+                                char *buf, size_t size)
+{
+    int ret;
+
+    TEST_CHECK(buf == NULL && size == 0);
+
+    /* no records expected */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 0);
+}
+
+/* No records must be selected */
+static void cb_select_tag_ok(int id, struct task_check *check,
+                             char *buf, size_t size)
+{
+    int ret;
+
+    TEST_CHECK(buf != NULL && size > 0);
+
+    /* 2 records expected */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+}
+
+static void cb_record_contains(int id, struct task_check *check,
+                               char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 2 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+}
+
+static void cb_record_not_contains(int id, struct task_check *check,
+                                   char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 0 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 0);
+}
+
+/* Callback functions to perform checks over results */
+static void cb_select_sub_blue(int id, struct task_check *check,
+                               char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+}
+
+static void cb_select_sub_num(int id, struct task_check *check,
+                              char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 2 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+}
+
+static void cb_select_sub_colors(int id, struct task_check *check,
+                                 char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 3 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 3);
+}
+
+static void cb_select_sub_record_contains(int id, struct task_check *check,
+                                          char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 5 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 5);
+}
+
+static void cb_select_sub_keys(int id, struct task_check *check,
+                               char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "map['sub1']['sub2']['color']",
+                            MSGPACK_OBJECT_STR,
+                            "blue", 0, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_sum_sub_keys(int id, struct task_check *check,
+                                   char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "SUM(map['sub1']['sub2'])",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 246, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_avg_sub_keys(int id, struct task_check *check,
+                                   char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "AVG(map['sub1']['sub2'])",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 123.0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_count_sub_keys(int id, struct task_check *check,
+                                     char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "COUNT(map['sub1']['sub2'])",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 2, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_min_sub_keys(int id, struct task_check *check,
+                                   char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "MIN(map['sub1']['sub2'])",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 123, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_max_sub_keys(int id, struct task_check *check,
+                                   char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "MAX(map['sub1']['sub3'])",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 100);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_sum_sub_keys_group_by(int id, struct task_check *check,
+                                            char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 3 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 3);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "SUM(map['sub1']['sub3'])",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 105.5);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_sum_sub_keys_group_by_2(int id, struct task_check *check,
+                                              char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 3 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 3);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "SUM(map['sub1']['sub3'])",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 105.5);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_sum_sub_keys_group_by_3(int id, struct task_check *check,
+                                              char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 3 rows */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 3);
+
+    ret = mp_record_key_cmp(buf, size,
+                            0, "SUM(map['sub1']['sub3'])",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 100, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    ret = mp_record_key_cmp(buf, size,
+                            1, "SUM(map['sub1']['sub3'])",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 11);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    ret = mp_record_key_cmp(buf, size,
+                            2, "SUM(map['sub1']['sub3'])",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 5.5);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+
+static void cb_forecast_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, "TIMESERIES_FORECAST(usage)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 310.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);
+}
+
+/* Callback functions to perform checks over results */
+static void cb_snapshot_create(int id, struct task_check *check,
+                               char *buf, size_t size)
+{
+    int ret;
+
+    ret = mp_count_rows(buf, size);
+    /* Snapshot doesn't return anything */
+    TEST_CHECK(ret == 0);
+};
+
+static void cb_snapshot_purge(int id, struct task_check *check,
+                              char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 5 rows, as set in snapshot query */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 5);
+};
+
+static void cb_snapshot_purge_time(int id, struct task_check *check,
+                                   char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 11 rows, as set in snapshot query */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 11);
+};
+
+static void cb_window_5_second(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, "SUM(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 225, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* Check AVG value result */
+    ret = mp_record_key_cmp(buf, size, 0, "AVG(id)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 4.5);
+
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_aggr(int id, struct task_check *check,
+                           char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 1);
+
+    /* MIN(id) is 0 */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "MIN(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 0, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* MAX(id) is 10 */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "MAX(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 10, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* COUNT(*) is 11 */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "COUNT(*)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 11, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* SUM(bytes) is 110.50 */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "SUM(bytes)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 110.50);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* AVG(bytes) is 10.04545 */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "AVG(bytes)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 10.045455);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_select_groupby(int id, struct task_check *check,
+                              char *buf, size_t size)
+{
+    int ret;
+
+    /* Expect 1 row */
+    ret = mp_count_rows(buf, size);
+    TEST_CHECK(ret == 2);
+
+    /* MIN(id) is 0 for record 0 (bool=true) */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "MIN(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 0, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* MIN(id) is 6 for record 1 (bool=false)  */
+    ret = mp_record_key_cmp(buf, size,
+                            1, "MIN(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 6, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* MAX(id) is 8 for record 0 (bool=true)  */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "MAX(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 8, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* MAX(id) is i9 for record 1 (bool=false)  */
+    ret = mp_record_key_cmp(buf, size,
+                            1, "MAX(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 9, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* COUNT(*) is 8 for record 0 (bool=true) */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "COUNT(*)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 8, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* COUNT(*) is 2 for record 1 (bool=false) */
+    ret = mp_record_key_cmp(buf, size,
+                            1, "COUNT(*)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 2, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* SUM(bytes) is 80.0 for record 0 (bool=true) */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "SUM(bytes)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 80.0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* SUM(bytes) is 20.50 for record 1 (bool=false) */
+    ret = mp_record_key_cmp(buf, size,
+                            1, "SUM(bytes)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 20.50);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* AVG(bytes) is 10.0 for record 0 (bool=true) */
+    ret = mp_record_key_cmp(buf, size,
+                            0, "AVG(bytes)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 10.0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* AVG(bytes) is 10.25 for record 1 (bool=false) */
+    ret = mp_record_key_cmp(buf, size,
+                            1, "AVG(bytes)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 10.25);
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_hopping_window_5_second(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, "SUM(id)",
+                            MSGPACK_OBJECT_POSITIVE_INTEGER,
+                            NULL, 266, 0);
+    TEST_CHECK(ret == FLB_TRUE);
+
+    /* Check AVG value result */
+    ret = mp_record_key_cmp(buf, size, 0, "AVG(id)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 16.625);
+
+    TEST_CHECK(ret == FLB_TRUE);
+}
+
+static void cb_forecast_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, "TIMESERIES_FORECAST(usage)",
+                            MSGPACK_OBJECT_FLOAT,
+                            NULL, 0, 460.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);
+}
+
+#endif
diff --git a/fluent-bit/tests/internal/include/sp_helpers.h b/fluent-bit/tests/internal/include/sp_helpers.h
new file mode 100644
index 000000000..2efaaeff2
--- /dev/null
+++ b/fluent-bit/tests/internal/include/sp_helpers.h
@@ -0,0 +1,158 @@
+#ifndef FLB_TEST_SP_HELPERS
+#define FLB_TEST_SP_HELPERS
+
+#define MP_UOK MSGPACK_UNPACK_SUCCESS
+
+struct task_check {
+    int id;                /* Test identifier */
+    int window_type;       /* Type of window in window-based queries */
+    int window_size_sec;   /* Size of the window in window-based queries */
+    int window_hop_sec;    /* Hopping size in hopping window */
+    char *name;            /* Name of the test */
+    char *exec;            /* Query */
+    void (*cb_check)(int, struct task_check *, char *, size_t);
+                           /* Callback function:
+                              id, test_check, output buffer, buffer_size */
+};
+
+/* Helper functions */
+static int mp_count_rows(char *buf, size_t size)
+{
+    int total = 0;
+    size_t off = 0;
+    msgpack_unpacked result;
+
+    msgpack_unpacked_init(&result);
+    while (msgpack_unpack_next(&result, buf, size, &off) == MP_UOK) {
+        total++;
+    }
+
+    msgpack_unpacked_destroy(&result);
+    return total;
+}
+
+/* Count total number of keys considering all rows */
+static int mp_count_keys(char *buf, size_t size)
+{
+    int keys = 0;
+    size_t off = 0;
+    msgpack_unpacked result;
+    msgpack_object root;
+    msgpack_object map;
+
+    msgpack_unpacked_init(&result);
+    while (msgpack_unpack_next(&result, buf, size, &off) == MP_UOK) {
+        root = result.data;
+        map = root.via.array.ptr[1];
+        keys += map.via.map.size;
+    }
+    msgpack_unpacked_destroy(&result);
+
+    return keys;
+}
+
+static inline int float_cmp(double f1, double f2)
+{
+    double precision = 0.00001;
+
+    if (((f1 - precision) < f2) &&
+        ((f1 + precision) > f2)) {
+        return 1;
+    }
+    else {
+        return 0;
+    }
+}
+
+/* Lookup record/row number 'id' and check that 'key' matches 'val' */
+static int mp_record_key_cmp(char *buf, size_t size,
+                             int record_id, char *key,
+                             int val_type, char *val_str, int64_t val_int64,
+                             double val_f64)
+{
+    int i;
+    int ret = FLB_FALSE;
+    int id = 0;
+    int k_len;
+    int v_len;
+    int keys = 0;
+    size_t off = 0;
+    msgpack_unpacked result;
+    msgpack_object root;
+    msgpack_object map;
+    msgpack_object k;
+    msgpack_object v;
+
+    k_len = strlen(key);
+
+    msgpack_unpacked_init(&result);
+    while (msgpack_unpack_next(&result, buf, size, &off) == MP_UOK) {
+        if (id != record_id) {
+            id++;
+            continue;
+        }
+
+        root = result.data;
+        map = root.via.array.ptr[1];
+        keys += map.via.map.size;
+
+        for (i = 0; i < keys; i++) {
+            k = map.via.map.ptr[i].key;
+            v = map.via.map.ptr[i].val;
+
+            if (k.type != MSGPACK_OBJECT_STR) {
+                continue;
+            }
+
+            if (k.via.str.size != k_len) {
+                continue;
+            }
+
+            if (strncmp(k.via.str.ptr, key, k_len) != 0) {
+                continue;
+            }
+
+            /* at this point the key matched, now validate the expected value */
+            if (val_type == MSGPACK_OBJECT_FLOAT) {
+                if (v.type != MSGPACK_OBJECT_FLOAT32 &&
+                    v.type != MSGPACK_OBJECT_FLOAT) {
+                    msgpack_unpacked_destroy(&result);
+                    return FLB_FALSE;
+                }
+            }
+            else if (v.type != val_type) {
+                msgpack_unpacked_destroy(&result);
+                return FLB_FALSE;
+            }
+
+            switch (val_type) {
+            case MSGPACK_OBJECT_STR:
+                v_len = strlen(val_str);
+                if (strncmp(v.via.str.ptr, val_str, v_len) == 0) {
+                    ret = FLB_TRUE;
+                }
+                goto exit;
+            case MSGPACK_OBJECT_POSITIVE_INTEGER:
+                if (v.via.i64 == val_int64) {
+                    ret = FLB_TRUE;
+                }
+                goto exit;
+            case MSGPACK_OBJECT_FLOAT:
+                if (float_cmp(v.via.f64, val_f64)) {
+                    ret = FLB_TRUE;
+                }
+                else {
+                    printf("double mismatch: %f exp %f\n",
+                           v.via.f64, val_f64);
+                }
+                goto exit;
+            };
+        }
+    }
+
+exit:
+    msgpack_unpacked_destroy(&result);
+    return ret;
+}
+
+#endif
diff --git a/fluent-bit/tests/internal/include/sp_invalid_queries.h b/fluent-bit/tests/internal/include/sp_invalid_queries.h
new file mode 100644
index 000000000..e299e0fd6
--- /dev/null
+++ b/fluent-bit/tests/internal/include/sp_invalid_queries.h
@@ -0,0 +1,20 @@
+#ifndef FLB_TEST_INVALID_QUERIES
+#define FLB_TEST_INVALID_QUERIES
+
+/* Tests to check if syntactically invalid queries return error */
+char *invalid_query_checks[] = {
+    "SELECT id, MIN(id) FROM STREAM:FLB;",
+    "SELECT *, COUNT(id) FROM STREAM:FLB;",
+    "SELECT * FROM TAG:FLB WHERE bool = NULL ;",
+    "SELECT * FROM TAG:FLB WHERE @record.some_random_func() ;",
+    "SELECT id, MIN(id) FROM STREAM:FLB WINDOW TUMBLING (1 SECOND)" \
+    " GROUP BY bool;",
+    "SELECT *, COUNT(id) FROM STREAM:FLB WINDOW TUMBLING (1 SECOND)" \
+    " GROUP BY bool;",
+    "SELECT *, COUNT(bool) FROM STREAM:FLB WINDOW TUMBLING (1 SECOND)" \
+    " GROUP BY bool;",
+    "SELECT *, bool, COUNT(bool) FROM STREAM:FLB WINDOW TUMBLING (1 SECOND)" \
+    " GROUP BY bool;"
+};
+
+#endif
diff --git a/fluent-bit/tests/internal/include/sp_select_keys.h b/fluent-bit/tests/internal/include/sp_select_keys.h
new file mode 100644
index 000000000..ba3c809f3
--- /dev/null
+++ b/fluent-bit/tests/internal/include/sp_select_keys.h
@@ -0,0 +1,131 @@
+#include "sp_cb_functions.h"
+
+#ifndef FLB_TEST_SP_SELECT_KEYS
+#define FLB_TEST_SP_SELECT_KEYS
+
+/* Tests for 'test_select_keys' */
+struct task_check select_keys_checks[] = {
+    {
+        0, 0, 0, 0,
+        "select_all",
+        "SELECT * FROM STREAM:FLB;",
+        cb_select_all
+    },
+    {
+        1, 0, 0, 0,
+        "select_id",
+        "SELECT id, word2 FROM STREAM:FLB;",
+        cb_select_id
+    },
+
+    /* Conditionals */
+    {
+        2, 0, 0, 0,
+        "select_cond_1",
+        "SELECT * FROM STREAM:FLB WHERE bytes > 10.290;",
+        cb_select_cond_1
+    },
+    {
+        3, 0, 0, 0,
+        "select_cond_2",
+        "SELECT * FROM STREAM:FLB WHERE word2 = 'rlz' or word3 = 'rlz';",
+        cb_select_cond_2
+    },
+    {
+        4, 0, 0, 0,
+        "select_cond_not_null",
+        "SELECT * FROM STREAM:FLB WHERE word2 = 'rlz' and word3 IS NOT NULL;",
+        cb_select_cond_not_null
+    },
+    {
+        5, 0, 0, 0,
+        "select_cond_null",
+        "SELECT * FROM STREAM:FLB WHERE word3 IS NULL;",
+        cb_select_cond_null
+    },
+    {
+        6, 0, 0, 0,
+        "select_not_equal_1",
+        "SELECT * FROM STREAM:FLB WHERE bool != true;",
+        cb_select_not_equal_1
+    },
+    {
+        7, 0, 0, 0,
+        "select_not_equal_2",
+        "SELECT * FROM STREAM:FLB WHERE bytes <> 10;",
+        cb_select_not_equal_2
+    },
+
+
+    /* Aggregation functions */
+    {
+        8, 0, 0, 0,
+        "select_aggr",
+        "SELECT MIN(id), MAX(id), COUNT(*), SUM(bytes), AVG(bytes) " \
+        "FROM STREAM:FLB;",
+        cb_select_aggr,
+    },
+    {
+        9, 0, 0, 0,
+        "select_aggr_coount",
+        "SELECT COUNT(*) " \
+        "FROM STREAM:FLB;",
+        cb_select_aggr_count,
+    },
+    {
+        10, 0, 0, 0,
+        "select_aggr_window_tumbling",
+        "SELECT MIN(id), MAX(id), COUNT(*), SUM(bytes), AVG(bytes) FROM STREAM:FLB;",
+        cb_select_aggr,
+    },
+    {
+        11, 0, 0, 0,
+        "select_aggr_window_tumbling_groupby",
+        "SELECT bool, MIN(id), MAX(id), COUNT(*), SUM(bytes), AVG(bytes) " \
+        "FROM STREAM:FLB WHERE word3 IS NOT NULL GROUP BY bool;",
+        cb_select_groupby,
+    },
+
+    /* Time functions */
+    {
+        12, 0, 0, 0,
+        "func_time_now",
+        "SELECT NOW(), NOW() as tnow FROM STREAM:FLB WHERE bytes > 10;",
+        cb_func_time_now,
+    },
+    {
+        13, 0, 0, 0,
+        "func_time_unix_timestamp",
+        "SELECT UNIX_TIMESTAMP(), UNIX_TIMESTAMP() as ts " \
+        "FROM STREAM:FLB WHERE bytes > 10;",
+        cb_func_time_unix_timestamp,
+    },
+
+    /* Stream selection using Tag rules */
+    {
+        14, 0, 0, 0,
+        "select_from_tag_error",
+        "SELECT id FROM TAG:'no-matches' WHERE bytes > 10;",
+        cb_select_tag_error,
+    },
+    {
+        15, 0, 0, 0,
+        "select_from_tag",
+        "SELECT id FROM TAG:'samples' WHERE bytes > 10;",
+        cb_select_tag_ok,
+    },
+    {
+        16, 0, 0, 0,
+        "@recond.contains",
+        "SELECT id FROM TAG:'samples' WHERE bytes = 10 AND @record.contains(word2);",
+        cb_record_contains,
+    },
+    {
+        17, 0, 0, 0,
+        "@recond.contains",
+        "SELECT id FROM TAG:'samples' WHERE @record.contains(x);",
+        cb_record_not_contains,
+    },
+};
+
+#endif
diff --git a/fluent-bit/tests/internal/include/sp_select_subkeys.h b/fluent-bit/tests/internal/include/sp_select_subkeys.h
new file mode 100644
index 000000000..a5feebf01
--- /dev/null
+++ b/fluent-bit/tests/internal/include/sp_select_subkeys.h
@@ -0,0 +1,84 @@
+#include "sp_cb_functions.h"
+
+#ifndef FLB_TEST_SP_SELECT_SUBKEYS
+#define FLB_TEST_SP_SELECT_SUBKEYS
+
+/* Tests for 'test_select_subkeys' */
+struct task_check select_subkeys_checks[] = {
+    {
+        0, 0, 0, 0,
+        "select_sub_blue",
+        "SELECT * FROM STREAM:FLB WHERE map['sub1']['sub2']['color'] = 'blue';",
+        cb_select_sub_blue
+    },
+    {
+        1, 0, 0, 0,
+        "select_sub_num",
+        "SELECT * FROM STREAM:FLB WHERE map['sub1']['sub2'] = 123;",
+        cb_select_sub_num
+    },
+    {
+        2, 0, 0, 0,
+        "select_sub_colors",
+        "SELECT * FROM STREAM:FLB WHERE "            \
+        "map['sub1']['sub2']['color'] = 'blue' OR "  \
+        "map['sub1']['sub2']['color'] = 'red'  OR "  \
+        "map['color'] = 'blue'; ",
+        cb_select_sub_colors
+    },
+    {
+        3, 0, 0, 0,
+        "cb_select_sub_record_contains",
+        "SELECT * FROM STREAM:FLB WHERE "            \
+        "@record.contains(map['sub1']['sub3']) OR "  \
+        "@record.contains(map['color']); ",
+        cb_select_sub_record_contains
+    },
+    {   4, 0, 0, 0,
+        "cb_select_sub_keys",
+        "SELECT map['sub1']['sub2']['color'] FROM STREAM:FLB WHERE "    \
+        "map['sub1']['sub2']['color'] = 'blue';",
+        cb_select_sub_keys},
+    {   5, 0, 0, 0,
+        "cb_select_sum_sub_keys",
+        "SELECT SUM(map['sub1']['sub2']) FROM STREAM:FLB WHERE "    \
+        "map['sub1']['sub2'] = 123;",
+        cb_select_sum_sub_keys},
+    {   6, 0, 0, 0,
+        "cb_select_avg_sub_keys",
+        "SELECT AVG(map['sub1']['sub2']) FROM STREAM:FLB WHERE "    \
+        "map['sub1']['sub2'] = 123;",
+        cb_select_avg_sub_keys},
+    {   7, 0, 0, 0,
+        "cb_select_count_sub_keys",
+        "SELECT COUNT(map['sub1']['sub2']) FROM STREAM:FLB WHERE "    \
+        "map['sub1']['sub2'] = 123;",
+        cb_select_count_sub_keys},
+    {   8, 0, 0, 0,
+        "cb_select_min_sub_keys",
+        "SELECT MIN(map['sub1']['sub2']) FROM STREAM:FLB WHERE "  \
+        "map['sub1']['sub2'] > 0;",
+        cb_select_min_sub_keys},
+    {   9, 0, 0, 0,
+        "cb_select_max_sub_keys",
+        "SELECT MAX(map['sub1']['sub3']) FROM STREAM:FLB WHERE "  \
+        "map['sub1']['sub3'] > 0;",
+        cb_select_max_sub_keys},
+    {   10, 0, 0, 0,
+        "cb_select_sum_sub_keys_group_by",
+        "SELECT SUM(map['sub1']['sub3']) FROM STREAM:FLB "  \
+        "GROUP BY map['mtype'];",
+        cb_select_sum_sub_keys_group_by},
+    {   11, 0, 0, 0,
+        "cb_select_sum_sub_keys_group_by_2",
+        "SELECT map['sub1']['stype'], map['mtype'], SUM(map['sub1']['sub3']) " \
+        "FROM STREAM:FLB GROUP BY map['mtype'], map['sub1']['stype'];",
+        cb_select_sum_sub_keys_group_by_2},
+    {   12, 0, 0, 0,
+        "cb_select_sum_sub_keys_group_by_3",
+        "SELECT map['sub1']['stype'], map['sub1']['sub4'], SUM(map['sub1']['sub3']) " \
+        "FROM STREAM:FLB GROUP BY map['sub1']['stype'], map['sub1']['sub4'];",
+        cb_select_sum_sub_keys_group_by_3}
+};
+
+#endif
diff --git a/fluent-bit/tests/internal/include/sp_snapshot.h b/fluent-bit/tests/internal/include/sp_snapshot.h
new file mode 100644
index 000000000..e8ca6fad5
--- /dev/null
+++ b/fluent-bit/tests/internal/include/sp_snapshot.h
@@ -0,0 +1,38 @@
+#include "sp_cb_functions.h"
+
+#ifndef FLB_TEST_SP_SNAPSHOT
+#define FLB_TEST_SP_SNAPSHOT
+
+/* Tests for 'test_snapshot' */
+struct task_check snapshot_checks[][2] = {
+    {
+        {   // Snapshot
+            0, 0, 0, 0,
+            "snapshot_create",
+            "SELECT * FROM STREAM:FLB LIMIT 5;",
+            cb_snapshot_create
+        },
+        {   // Flush
+            1, 0, 0, 0,
+            "snapshot_purge",
+            "SELECT * FROM STREAM:FLB;",
+            cb_snapshot_purge
+        },
+    },
+    {
+        {  // Snapshot
+            2, 0, 5, 0,
+            "snapshot_create",
+            "SELECT * FROM STREAM:FLB;",
+            cb_snapshot_create
+        },
+        {  // Flush
+            3, 0, 0, 0,
+            "snapshot_purge",
+            "SELECT * FROM STREAM:FLB;",
+            cb_snapshot_purge_time
+        },
+    },
+};
+
+#endif
diff --git a/fluent-bit/tests/internal/include/sp_window.h b/fluent-bit/tests/internal/include/sp_window.h
new file mode 100644
index 000000000..117b10144
--- /dev/null
+++ b/fluent-bit/tests/internal/include/sp_window.h
@@ -0,0 +1,53 @@
+#include "sp_cb_functions.h"
+
+#ifndef FLB_TEST_SP_WINDOW
+#define FLB_TEST_SP_WINDOW
+
+/* Tests for test_window */
+struct task_check window_checks[] = {
+    {
+        0, FLB_SP_WINDOW_TUMBLING, 5, 0,
+        "window_5_seconds",
+        "SELECT SUM(id), AVG(id) FROM STREAM:FLB WINDOW TUMBLING (5 SECOND) " \
+        "WHERE word3 IS NOT NULL;",
+        cb_window_5_second
+    },
+    {
+        1, FLB_SP_WINDOW_TUMBLING, 1, 0,
+        "select_aggr_window_tumbling",
+        "SELECT MIN(id), MAX(id), COUNT(*), SUM(bytes), AVG(bytes) " \
+        "FROM STREAM:FLB WINDOW TUMBLING (1 SECOND);",
+        cb_select_aggr,
+    },
+    {
+        2, FLB_SP_WINDOW_TUMBLING, 1, 0,
+        "select_aggr_window_tumbling_groupby",
+        "SELECT bool, MIN(id), MAX(id), COUNT(*), SUM(bytes), AVG(bytes) " \
+        "FROM STREAM:FLB WINDOW TUMBLING (1 SECOND) WHERE word3 IS NOT NULL " \
+        "GROUP BY bool;",
+        cb_select_groupby,
+    },
+    {
+        3, FLB_SP_WINDOW_HOPPING, 5, 2,
+        "hopping_window_5_seconds",
+        "SELECT SUM(id), AVG(id) FROM STREAM:FLB WINDOW HOPPING (5 SECOND, " \
+        "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(usage, 20) FROM " \
+        "STREAM:FLB WINDOW TUMBLING (5 SECOND);",
+        cb_forecast_tumbling_window
+    },
+    {
+        5, FLB_SP_WINDOW_HOPPING, 5, 2,
+        "timeseries_forecast_window_hopping",
+        "SELECT AVG(usage), TIMESERIES_FORECAST(usage, 20) FROM " \
+        "STREAM:FLB WINDOW HOPPING (5 SECOND, ADVANCE BY 2 SECOND);",
+        cb_forecast_hopping_window
+    },
+};
+
+#endif