Fix DDL table identifier parsing

The parsing was inadequate and did not support all forms of quoted
identifiers.

server/modules/routing/avrorouter/avro_schema.c

@@ -720,46 +720,28 @@ int resolve_table_version(const char* db, const char* table)
 
 /**
  * @brief Handle a query event which contains a CREATE TABLE statement
- * @param sql Query SQL
- * @param db Database where this query was executed
+ *
+ * @param ident Table identifier in database.table format
+ * @param sql   The CREATE TABLE statement
+ * @param len   Length of @c sql
+ *
  * @return New CREATE_TABLE object or NULL if an error occurred
  */
-TABLE_CREATE* table_create_alloc(const char* sql, int len, const char* db)
+TABLE_CREATE* table_create_alloc(const char* ident, const char* sql, int len)
 {
     /** Extract the table definition so we can get the column names from it */
     int stmt_len = 0;
     const char* statement_sql = get_table_definition(sql, len, &stmt_len);
     ss_dassert(statement_sql);
+
+    char* tbl_start = strchr(ident, '.');
+    ss_dassert(tbl_start);
+    *tbl_start++ = '\0';
+
     char table[MYSQL_TABLE_MAXLEN + 1];
     char database[MYSQL_DATABASE_MAXLEN + 1];
-    const char* err = NULL;
-    MXS_INFO("Create table: %.*s", len, sql);
-
-    if (!statement_sql)
-    {
-        err = "table definition";
-    }
-    else if (!get_table_name(sql, table))
-    {
-        err = "table name";
-    }
-
-    if (get_database_name(sql, database))
-    {
-        // The CREATE statement contains the database name
-        db = database;
-    }
-    else if (*db == '\0')
-    {
-        // No explicit or current database
-        err = "database name";
-    }
-
-    if (err)
-    {
-        MXS_ERROR("Malformed CREATE TABLE statement, could not extract %s: %.*s", err, len, sql);
-        return NULL;
-    }
+    strcpy(database, ident);
+    strcpy(table, tbl_start);
 
     int* lengths = NULL;
     char **names = NULL;
@@ -1253,15 +1235,130 @@ static bool tok_eq(const char *a, const char *b, size_t len)
     return true;
 }
 
-void read_alter_identifier(const char *sql, const char *end, char *dest, int size)
+static void skip_whitespace(const char** saved)
 {
-    int len = 0;
-    const char *tok = get_tok(sql, &len, end); // ALTER
-    if (tok && (tok = get_tok(tok + len, &len, end)) // TABLE
-        && (tok = get_tok(tok + len, &len, end))) // Table identifier
+    const char* ptr = *saved;
+
+    while (*ptr && isspace(*ptr))
     {
-        snprintf(dest, size, "%.*s", len, tok);
-        remove_backticks(dest);
+        ptr++;
+    }
+
+    *saved = ptr;
+}
+
+static void skip_token(const char** saved)
+{
+    const char* ptr = *saved;
+
+    while (*ptr && !isspace(*ptr))
+    {
+        ptr++;
+    }
+
+    *saved = ptr;
+}
+
+static void skip_non_backtick(const char** saved)
+{
+    const char* ptr = *saved;
+
+    while (*ptr && *ptr != '`')
+    {
+        ptr++;
+    }
+
+    *saved = ptr;
+}
+
+const char* keywords[] =
+{
+    "CREATE",
+    "DROP",
+    "ALTER",
+    "IF",
+    "EXISTS",
+    "REPLACE",
+    "OR",
+    "TABLE",
+    "NOT",
+    NULL
+};
+
+static bool token_is_keyword(const char* tok, int len)
+{
+    for (int i = 0; keywords[i]; i++)
+    {
+        if (strncasecmp(keywords[i], tok, len) == 0)
+        {
+            return true;
+        }
+    }
+
+    return false;
+}
+
+void read_table_identifier(const char* db, const char *sql, const char *end, char *dest, int size)
+{
+    const char* start;
+    int len = 0;
+    bool is_keyword = true;
+
+    while (is_keyword)
+    {
+        skip_whitespace(&sql); // Leading whitespace
+
+        if (*sql == '`')
+        {
+            // Quoted identifier, not a keyword
+            is_keyword = false;
+            sql++;
+            start = sql;
+            skip_non_backtick(&sql);
+            len = sql - start;
+            sql++;
+        }
+        else
+        {
+            start = sql;
+            skip_token(&sql);
+            len = sql - start;
+            is_keyword = token_is_keyword(start, len);
+        }
+    }
+
+    skip_whitespace(&sql); // Space after first identifier
+
+    if (*sql != '.')
+    {
+        // No explicit database
+        snprintf(dest, size, "%s.%.*s", db, len, start);
+    }
+    else
+    {
+        // Explicit database, skip the period
+        sql++;
+        skip_whitespace(&sql); // Space after first identifier
+
+        const char* id_start;
+        int id_len = 0;
+
+        if (*sql == '`')
+        {
+            sql++;
+            id_start = sql;
+            skip_non_backtick(&sql);
+            id_len = sql - id_start;
+            sql++;
+        }
+        else
+        {
+            id_start = sql;
+            skip_token(&sql);
+            id_len = sql - id_start;
+        }
+
+        snprintf(dest, size, "%.*s.%.*s", len, start, id_len, id_start);
     }
 }