MXS-935: Add support for column based rule matching

In addition, test program extended to be able to test the
rule matching. Tests to be extended in a subsequent commit.

server/modules/filter/cache/rules.c

@@ -782,9 +782,119 @@ static bool cache_rule_compare_n(CACHE_RULE *self, const char *value, size_t len
 static bool cache_rule_matches_column(CACHE_RULE *self, const char *default_db, const GWBUF *query)
 {
     ss_dassert(self->attribute == CACHE_ATTRIBUTE_COLUMN);
-    ss_info_dassert(!true, "Column matching not implemented yet.");
 
-    return false;
+    // TODO: Do this "parsing" when the rule item is created.
+    char buffer[strlen(self->value) + 1];
+    strcpy(buffer, self->value);
+
+    const char* rule_column = NULL;
+    const char* rule_table = NULL;
+    const char* rule_database = NULL;
+    char* dot1 = strchr(buffer, '.');
+    char* dot2 = dot1 ? strchr(buffer, '.') : NULL;
+
+    if (dot1 && dot2)
+    {
+        rule_database = buffer;
+        *dot1 = 0;
+        rule_table = dot1 + 1;
+        *dot2 = 0;
+        rule_column = dot2 + 1;
+    }
+    else if (dot1)
+    {
+        rule_table = buffer;
+        *dot1 = 0;
+        rule_column = dot1 + 1;
+    }
+    else
+    {
+        rule_column = buffer;
+    }
+
+    const QC_FIELD_INFO *infos;
+    size_t n_infos;
+
+    int n_tables;
+    char** tables = qc_get_table_names((GWBUF*)query, &n_tables, false);
+
+    const char* default_table = NULL;
+
+    if (n_tables == 1)
+    {
+        // Only if we have exactly one table can we assume anything
+        // about a table that has not been mentioned explicitly.
+        default_table = tables[0];
+    }
+
+    qc_get_field_info((GWBUF*)query, &infos, &n_infos);
+
+    bool matches = false;
+
+    size_t i = 0;
+    while (!matches && (i < n_infos))
+    {
+        const QC_FIELD_INFO *info = (infos + i);
+
+        if ((strcmp(info->column, rule_column) == 0) || (strcmp(info->column, "*") == 0))
+        {
+            if (rule_table)
+            {
+                const char* check_table = info->table ? info->table : default_table;
+
+                if (check_table && (strcmp(check_table, rule_table) == 0))
+                {
+                    if (rule_database)
+                    {
+                        const char *check_database = info->database ? info->database : default_db;
+
+                        if (check_database && (strcmp(check_database, rule_database) == 0))
+                        {
+                            matches = true;
+                        }
+                        else
+                        {
+                            // If the rules specifies a database and either the database
+                            // does not match or we do not know the database, the rule
+                            // does *not* match.
+                            matches = false;
+                        }
+                    }
+                    else
+                    {
+                        // If the rule specifies no table, then if the table and column matches,
+                        // the rule matches.
+                        matches = true;
+                    }
+                }
+                else
+                {
+                    // The rules specifies a table and either the table does not match
+                    // or we do not know the table, the rule does *not* match.
+                    matches = false;
+                }
+            }
+            else
+            {
+                // If the rule specifies no table, then if the column matches, the
+                // rule matches.
+                matches = true;
+            }
+        }
+
+        ++i;
+    }
+
+    if (tables)
+    {
+        for (i = 0; i < (size_t)n_tables; ++i)
+        {
+            MXS_FREE(tables[i]);
+        }
+        MXS_FREE(tables);
+    }
+
+    return matches;
 }
 
 /**

server/modules/filter/cache/test/testrules.c

@@ -14,12 +14,37 @@
 #include <stdlib.h>
 #include "rules.h"
 #include <maxscale/log_manager.h>
+#include <maxscale/query_classifier.h>
+#include <maxscale/protocol/mysql.h>
 #if !defined(SS_DEBUG)
 #define SS_DEBUG
 #endif
 #include <maxscale/debug.h>
 
-struct test_case
+GWBUF* create_gwbuf(const char* s)
+{
+    size_t query_len = strlen(s);
+    size_t payload_len = query_len + 1;
+    size_t gwbuf_len = MYSQL_HEADER_LEN + payload_len;
+
+    GWBUF* gwbuf = gwbuf_alloc(gwbuf_len);
+    ss_dassert(gwbuf);
+
+    *((unsigned char*)((char*)GWBUF_DATA(gwbuf))) = payload_len;
+    *((unsigned char*)((char*)GWBUF_DATA(gwbuf) + 1)) = (payload_len >> 8);
+    *((unsigned char*)((char*)GWBUF_DATA(gwbuf) + 2)) = (payload_len >> 16);
+    *((unsigned char*)((char*)GWBUF_DATA(gwbuf) + 3)) = 0x00;
+    *((unsigned char*)((char*)GWBUF_DATA(gwbuf) + 4)) = 0x03;
+    memcpy((char*)GWBUF_DATA(gwbuf) + MYSQL_HEADER_LEN + 1, s, query_len);
+
+    return gwbuf;
+}
+
+//
+// Test user rules. Basically tests that a user specification is translated
+// into the correct pcre2 regex.
+//
+struct user_test_case
 {
     const char* json;
     struct
@@ -29,31 +54,33 @@ struct test_case
     } expect;
 };
 
-#define TEST_CASE(op_from, from, op_to, to)                                 \
+#define USER_TEST_CASE(op_from, from, op_to, to)                                 \
 { "{ \"use\": [ { \"attribute\": \"user\", \"op\": \"" #op_from "\", \"value\": \"" #from "\" } ] }",\
     { op_to, #to } }
 
-const struct test_case test_cases[] =
+#define COLUMN_
+
+const struct user_test_case user_test_cases[] =
 {
-    TEST_CASE(=, bob,           CACHE_OP_LIKE, bob@.*),
+    USER_TEST_CASE(=, bob,           CACHE_OP_LIKE, bob@.*),
-    TEST_CASE(=, 'bob',         CACHE_OP_LIKE, bob@.*),
+    USER_TEST_CASE(=, 'bob',         CACHE_OP_LIKE, bob@.*),
-    TEST_CASE(=, bob@%,         CACHE_OP_LIKE, bob@.*),
+    USER_TEST_CASE(=, bob@%,         CACHE_OP_LIKE, bob@.*),
-    TEST_CASE(=, 'bob'@'%.52',  CACHE_OP_LIKE, bob@.*\\.52),
+    USER_TEST_CASE(=, 'bob'@'%.52',  CACHE_OP_LIKE, bob@.*\\.52),
-    TEST_CASE(=, bob@127.0.0.1, CACHE_OP_EQ,   bob@127.0.0.1),
+    USER_TEST_CASE(=, bob@127.0.0.1, CACHE_OP_EQ,   bob@127.0.0.1),
-    TEST_CASE(=, b*b@127.0.0.1, CACHE_OP_EQ,   b*b@127.0.0.1),
+    USER_TEST_CASE(=, b*b@127.0.0.1, CACHE_OP_EQ,   b*b@127.0.0.1),
-    TEST_CASE(=, b*b@%.0.0.1,   CACHE_OP_LIKE, b\\*b@.*\\.0\\.0\\.1),
+    USER_TEST_CASE(=, b*b@%.0.0.1,   CACHE_OP_LIKE, b\\*b@.*\\.0\\.0\\.1),
-    TEST_CASE(=, b*b@%.0.%.1,   CACHE_OP_LIKE, b\\*b@.*\\.0\\..*\\.1),
+    USER_TEST_CASE(=, b*b@%.0.%.1,   CACHE_OP_LIKE, b\\*b@.*\\.0\\..*\\.1),
 };
 
-const size_t n_test_cases = sizeof(test_cases) / sizeof(test_cases[0]);
+const size_t n_user_test_cases = sizeof(user_test_cases) / sizeof(user_test_cases[0]);
 
-int test()
+int test_user()
 {
     int errors = 0;
 
-    for (int i = 0; i < n_test_cases; ++i)
+    for (int i = 0; i < n_user_test_cases; ++i)
     {
-        const struct test_case *test_case = &test_cases[i];
+        const struct user_test_case *test_case = &user_test_cases[i];
 
         CACHE_RULES *rules = cache_rules_parse(test_case->json, 0);
         ss_dassert(rules);
@@ -78,9 +105,86 @@ int test()
                    rule->value);
             ++errors;
         }
+
+        cache_rules_free(rules);
     }
 
-    return errors == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
+    return errors;
+}
+
+//
+//
+//
+struct store_test_case
+{
+    const char *rule;       // The rule in JSON format.
+    bool matches;           // Whether or not the rule should match the query.
+    const char *default_db; // The current default db.
+    const char *query;      // The query to be matched against the rule.
+};
+
+#define STORE_TEST_CASE(attribute, op, value, matches, default_db, query) \
+{ "{ \"store\": [ { \"attribute\": \"" attribute "\", \"op\": \"" op "\", \"value\": \"" value "\" } ] }",\
+        matches, default_db, query }
+
+// In the following,
+//   true:  The query SHOULD match the rule,
+//   false: The query should NOT match the rule.
+const struct store_test_case store_test_cases[] =
+{
+    STORE_TEST_CASE("column", "=", "a", true,  NULL, "SELECT a FROM tbl"),
+    STORE_TEST_CASE("column", "=", "b", false, NULL, "SELECT a FROM tbl")
+};
+
+const size_t n_store_test_cases = sizeof(store_test_cases) / sizeof(store_test_cases[0]);
+
+int test_store()
+{
+    int errors = 0;
+
+    for (int i = 0; i < n_store_test_cases; ++i)
+    {
+        const struct store_test_case *test_case = &store_test_cases[i];
+
+        CACHE_RULES *rules = cache_rules_parse(test_case->rule, 0);
+        ss_dassert(rules);
+
+        CACHE_RULE *rule = rules->store_rules;
+        ss_dassert(rule);
+
+        GWBUF *packet = create_gwbuf(test_case->query);
+
+        bool matches = cache_rules_should_store(rules, test_case->default_db, packet);
+
+        if  (matches != test_case->matches)
+        {
+            printf("Query   : %s\n"
+                   "Rule    : %s\n"
+                   "Expected: %s\n"
+                   "Result  : %s\n\n",
+                   test_case->query,
+                   test_case->rule,
+                   test_case->matches ? "A match" : "Not a match",
+                   matches ? "A match" : "Not a match");
+        }
+
+        gwbuf_free(packet);
+
+        cache_rules_free(rules);
+    }
+
+    return errors;
+}
+
+
+int test()
+{
+    int errors = 0;
+
+    errors += test_user();
+    errors += test_store();
+
+    return errors ? EXIT_FAILURE : EXIT_SUCCESS;
 }
 
 int main()
@@ -89,7 +193,14 @@ int main()
 
     if (mxs_log_init(NULL, ".", MXS_LOG_TARGET_DEFAULT))
     {
-        rc = test();
+        if (qc_init("qc_sqlite", ""))
+        {
+            rc = test();
+        }
+        else
+        {
+            MXS_ERROR("Could not initialize query classifier.");
+        }
 
         mxs_log_finish();
     }