MXS-1196: Handle sequence related fields/functions

Both 10.3 and Oracle support sequence pseudo colums and corresponding
functions. Getting the next number in the sequence is in both cases
obtained using nextval/nextval() but the current number is in Oracle
obtained using currval/currval() and in 10.3 using lastval/lastval().

These fields/functions are now ignored, in the sense that they will
not show up in the field/function infos. However, they will cause the
type mask of the statement to contain the bit QUERY_TYPE_WRITE so that
statements accessing the sequence will always be sent to the master.

query_classifier/qc_mysqlembedded/qc_mysqlembedded.cc

@@ -90,6 +90,7 @@ typedef struct parsing_info_st
     size_t function_infos_capacity;
     GWBUF* preparable_stmt;
     qc_parse_result_t result;
+    int32_t type_mask;
 #if defined(SS_DEBUG)
     skygw_chk_t pi_chk_tail;
 #endif
@@ -155,6 +156,8 @@ static TABLE_LIST* skygw_get_affected_tables(void* lexptr);
 static bool ensure_query_is_parsed(GWBUF* query);
 static bool parse_query(GWBUF* querybuf);
 static bool query_is_parsed(GWBUF* buf);
+int32_t qc_mysql_get_field_info(GWBUF* buf, const QC_FIELD_INFO** infos, uint32_t* n_infos);
+
 
 #if MYSQL_VERSION_MAJOR >= 10 && MYSQL_VERSION_MINOR >= 3
 inline void get_string_and_length(const LEX_CSTRING& ls, const char** s, size_t* length)
@@ -224,6 +227,8 @@ int32_t qc_mysql_parse(GWBUF* querybuf, uint32_t collect, int32_t* result)
 
 int32_t qc_mysql_get_type_mask(GWBUF* querybuf, uint32_t* type_mask)
 {
+    int32_t rv = QC_RESULT_OK;
+
     *type_mask = QUERY_TYPE_UNKNOWN;
     MYSQL* mysql;
     bool succp;
@@ -254,12 +259,25 @@ int32_t qc_mysql_get_type_mask(GWBUF* querybuf, uint32_t* type_mask)
             if (mysql != NULL)
             {
                 *type_mask = resolve_query_type(pi, (THD *) mysql->thd);
+#if MYSQL_VERSION_MAJOR >= 10 && MYSQL_VERSION_MINOR >= 3
+                // If in 10.3 mode we need to ensure that sequence related functions
+                // are taken into account. That we can ensure by querying for the fields.
+                const QC_FIELD_INFO* field_infos;
+                uint32_t n_field_infos;
+
+                rv = qc_mysql_get_field_info(querybuf, &field_infos, &n_field_infos);
+
+                if (rv == QC_RESULT_OK)
+                {
+                    *type_mask |= pi->type_mask;
+                }
+#endif
             }
         }
     }
 
 retblock:
-    return QC_RESULT_OK;
+    return rv;
 }
 
 /**
@@ -2342,6 +2360,43 @@ static void remove_surrounding_back_ticks(char* s)
     }
 }
 
+static bool should_function_be_ignored(parsing_info_t* pi, const char* func_name)
+{
+    bool rv = false;
+
+    // We want to ignore functions that do not really appear as such in an
+    // actual SQL statement. E.g. "SELECT @a" appears as a function "get_user_var".
+    if ((strcasecmp(func_name, "decimal_typecast") == 0) ||
+        (strcasecmp(func_name, "cast_as_char") == 0) ||
+        (strcasecmp(func_name, "cast_as_date") == 0) ||
+        (strcasecmp(func_name, "cast_as_datetime") == 0) ||
+        (strcasecmp(func_name, "cast_as_time") == 0) ||
+        (strcasecmp(func_name, "cast_as_signed") == 0) ||
+        (strcasecmp(func_name, "cast_as_unsigned") == 0) ||
+        (strcasecmp(func_name, "get_user_var") == 0) ||
+        (strcasecmp(func_name, "get_system_var") == 0) ||
+        (strcasecmp(func_name, "set_user_var") == 0) ||
+        (strcasecmp(func_name, "set_system_var") == 0))
+    {
+        rv = true;
+    }
+
+    // Any sequence related functions should be ignored as well.
+#if MYSQL_VERSION_MAJOR >= 10 && MYSQL_VERSION_MINOR >= 3
+    if (!rv)
+    {
+        if ((strcasecmp(func_name, "lastval") == 0) ||
+            (strcasecmp(func_name, "nextval") == 0))
+        {
+            pi->type_mask |= QUERY_TYPE_WRITE;
+            rv = true;
+        }
+    }
+#endif
+
+    return rv;
+}
+
 static void update_field_infos(parsing_info_t* pi,
                                collect_source_t source,
                                Item* item,
@@ -2459,17 +2514,7 @@ static void update_field_infos(parsing_info_t* pi,
 
             // We want to ignore functions that do not really appear as such in an
             // actual SQL statement. E.g. "SELECT @a" appears as a function "get_user_var".
-            if ((strcasecmp(func_name, "decimal_typecast") != 0) &&
+            if (!should_function_be_ignored(pi, func_name))
-                (strcasecmp(func_name, "cast_as_char") != 0) &&
-                (strcasecmp(func_name, "cast_as_date") != 0) &&
-                (strcasecmp(func_name, "cast_as_datetime") != 0) &&
-                (strcasecmp(func_name, "cast_as_time") != 0) &&
-                (strcasecmp(func_name, "cast_as_signed") != 0) &&
-                (strcasecmp(func_name, "cast_as_unsigned") != 0) &&
-                (strcasecmp(func_name, "get_user_var") != 0) &&
-                (strcasecmp(func_name, "get_system_var") != 0) &&
-                (strcasecmp(func_name, "set_user_var") != 0) &&
-                (strcasecmp(func_name, "set_system_var") != 0))
             {
                 if (strcmp(func_name, "%") == 0)
                 {

query_classifier/qc_sqlite/qc_sqlite.c

@@ -180,6 +180,10 @@ static QC_SQLITE_INFO* info_alloc(uint32_t collect);
 static void info_finish(QC_SQLITE_INFO* info);
 static void info_free(QC_SQLITE_INFO* info);
 static QC_SQLITE_INFO* info_init(QC_SQLITE_INFO* info, uint32_t collect);
+static bool is_sequence_related_field(const char* database,
+                                      const char* table,
+                                      const char* column);
+static bool is_sequence_related_function(const char* func_name);
 static void log_invalid_data(GWBUF* query, const char* message);
 static const char* map_function_name(const char* name);
 static bool parse_query(GWBUF* query, uint32_t collect);
@@ -658,6 +662,57 @@ static bool query_is_parsed(GWBUF* query, uint32_t collect)
     return rc;
 }
 
+/**
+ * Returns whether a field is sequence related.
+ *
+ * @param database The database/schema or NULL.
+ * @param table    The table or NULL.
+ * @param column   The column.
+ *
+ * @return True, if the field is sequence related, false otherwise.
+ */
+static bool is_sequence_related_field(const char* database,
+                                      const char* table,
+                                      const char* column)
+{
+    return is_sequence_related_function(column);
+}
+
+/**
+ * Returns whether a function is sequence related.
+ *
+ * @param func_name   A function.
+ *
+ * @return True, if the function is sequence related, false otherwise.
+ */
+static bool is_sequence_related_function(const char* func_name)
+{
+    bool rv = false;
+
+    if (this_unit.sql_mode == QC_SQL_MODE_ORACLE)
+    {
+        // In Oracle mode we ignore the pseudocolumns "currval" and "nextval".
+        // We also exclude "lastval", the 10.3 equivalent of "currval".
+        if ((strcasecmp(func_name, "currval") == 0) ||
+            (strcasecmp(func_name, "nextval") == 0) ||
+            (strcasecmp(func_name, "lastval") == 0))
+        {
+            rv = true;
+        }
+    }
+
+    if (!rv && (this_unit.parse_as == QC_PARSE_AS_103))
+    {
+        if ((strcasecmp(func_name, "lastval") == 0) ||
+            (strcasecmp(func_name, "nextval") == 0))
+        {
+            rv = true;
+        }
+    }
+
+    return rv;
+}
+
 /**
  * Logs information about invalid data.
  *
@@ -759,6 +814,12 @@ static void update_field_info(QC_SQLITE_INFO* info,
 {
     ss_dassert(column);
 
+    if (is_sequence_related_field(database, table, column))
+    {
+        info->type_mask |= QUERY_TYPE_WRITE;
+        return;
+    }
+
     if (!(info->collect & QC_COLLECT_FIELDS) || (info->collected & QC_COLLECT_FIELDS))
     {
         // If field information should not be collected, or if field information
@@ -1065,6 +1126,8 @@ static void update_field_infos(QC_SQLITE_INFO* info,
 {
     const char* zToken = pExpr->u.zToken;
 
+    bool ignore_exprlist = false;
+
     switch (pExpr->op)
     {
     case TK_ASTERISK: // select *
@@ -1187,6 +1250,11 @@ static void update_field_infos(QC_SQLITE_INFO* info,
                 {
                     info->type_mask |= (QUERY_TYPE_READ | QUERY_TYPE_MASTER_READ);
                 }
+                else if (is_sequence_related_function(zToken))
+                {
+                    info->type_mask |= QUERY_TYPE_WRITE;
+                    ignore_exprlist = true;
+                }
                 else if (!is_builtin_readonly_function(zToken, this_unit.sql_mode == QC_SQL_MODE_ORACLE))
                 {
                     info->type_mask |= QUERY_TYPE_WRITE;
@@ -1194,7 +1262,7 @@ static void update_field_infos(QC_SQLITE_INFO* info,
 
                 // We exclude "row", because we cannot detect all rows the same
                 // way qc_mysqlembedded does.
-                if (strcasecmp(zToken, "row") != 0)
+                if (!ignore_exprlist && (strcasecmp(zToken, "row") != 0))
                 {
                     update_function_info(info, zToken, usage);
                 }
@@ -1227,7 +1295,10 @@ static void update_field_infos(QC_SQLITE_INFO* info,
             case TK_BETWEEN:
             case TK_CASE:
             case TK_FUNCTION:
-                update_field_infos_from_exprlist(info, pExpr->x.pList, usage, pExclude);
+                if (!ignore_exprlist)
+                {
+                    update_field_infos_from_exprlist(info, pExpr->x.pList, usage, pExclude);
+                }
                 break;
 
             case TK_EXISTS: