# Copyright (c) 2016, 2024, Oracle and/or its affiliates.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License, version 2.0, as
# published by the Free Software Foundation.
#
# This program is designed to work with certain software (including
# but not limited to OpenSSL) that is licensed under separate terms,
# as designated in a particular file or component or in included license
# documentation. The authors of MySQL hereby grant you an
# additional permission to link the program and your derivative works
# with the separately licensed software that they have either included with
# the program or referenced in the documentation.
#
# Without limiting anything contained in the foregoing, this file,
# which is part of MySQL Connector/Python, is also subject to the
# Universal FOSS Exception, version 1.0, a copy of which can be found at
# http://oss.oracle.com/licenses/universal-foss-exception.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
# See the GNU General Public License, version 2.0, for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA

"""Expression Parser."""

from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Union

from .dbdoc import DbDoc
from .helpers import BYTE_TYPES, get_item_or_attr
from .protobuf import Message, mysqlxpb_enum
from .types import (
    BuildExprTypes,
    BuildScalarTypes,
    MessageType,
    ProtobufMessageCextType,
    ProtobufMessageType,
    StrOrBytes,
)


# pylint: disable=missing-function-docstring
class TokenType:
    """Token types class."""

    NOT = 1
    AND = 2
    OR = 3
    XOR = 4
    IS = 5
    LPAREN = 6
    RPAREN = 7
    LSQBRACKET = 8
    RSQBRACKET = 9
    BETWEEN = 10
    TRUE = 11
    NULL = 12
    FALSE = 13
    IN = 14
    LIKE = 15
    INTERVAL = 16
    REGEXP = 17
    ESCAPE = 18
    IDENT = 19
    LSTRING = 20
    LNUM = 21
    DOT = 22
    DOLLAR = 23
    COMMA = 24
    EQ = 25
    NE = 26
    GT = 27
    GE = 28
    LT = 29
    LE = 30
    BITAND = 31
    BITOR = 32
    BITXOR = 33
    LSHIFT = 34
    RSHIFT = 35
    PLUS = 36
    MINUS = 37
    MUL = 38
    DIV = 39
    HEX = 40
    BIN = 41
    NEG = 42
    BANG = 43
    MICROSECOND = 44
    SECOND = 45
    MINUTE = 46
    HOUR = 47
    DAY = 48
    WEEK = 49
    MONTH = 50
    QUARTER = 51
    YEAR = 52
    EROTEME = 53
    DOUBLESTAR = 54
    MOD = 55
    COLON = 56
    OROR = 57
    ANDAND = 58
    LCURLY = 59
    RCURLY = 60
    CAST = 61
    DOTSTAR = 62
    ORDERBY_ASC = 63
    ORDERBY_DESC = 64
    AS = 65
    ARROW = 66
    QUOTE = 67
    BINARY = 68
    DATETIME = 69
    TIME = 70
    CHAR = 71
    DATE = 72
    DECIMAL = 73
    SIGNED = 74
    INTEGER = 75
    UNSIGNED = 76
    JSON = 77
    SECOND_MICROSECOND = 78
    MINUTE_MICROSECOND = 79
    MINUTE_SECOND = 80
    HOUR_MICROSECOND = 81
    HOUR_SECOND = 82
    HOUR_MINUTE = 83
    DAY_MICROSECOND = 84
    DAY_SECOND = 85
    DAY_MINUTE = 86
    DAY_HOUR = 87
    YEAR_MONTH = 88
    OVERLAPS = 89


_INTERVAL_UNITS = set(
    [
        TokenType.MICROSECOND,
        TokenType.SECOND,
        TokenType.MINUTE,
        TokenType.HOUR,
        TokenType.DAY,
        TokenType.WEEK,
        TokenType.MONTH,
        TokenType.QUARTER,
        TokenType.YEAR,
        TokenType.SECOND_MICROSECOND,
        TokenType.MINUTE_MICROSECOND,
        TokenType.MINUTE_SECOND,
        TokenType.HOUR_MICROSECOND,
        TokenType.HOUR_SECOND,
        TokenType.HOUR_MINUTE,
        TokenType.DAY_MICROSECOND,
        TokenType.DAY_SECOND,
        TokenType.DAY_MINUTE,
        TokenType.DAY_HOUR,
        TokenType.YEAR_MONTH,
    ]
)

# map of reserved word to token type
_RESERVED_WORDS = {
    "and": TokenType.AND,
    "or": TokenType.OR,
    "xor": TokenType.XOR,
    "is": TokenType.IS,
    "not": TokenType.NOT,
    "like": TokenType.LIKE,
    "in": TokenType.IN,
    "overlaps": TokenType.OVERLAPS,
    "regexp": TokenType.REGEXP,
    "between": TokenType.BETWEEN,
    "interval": TokenType.INTERVAL,
    "escape": TokenType.ESCAPE,
    "cast": TokenType.CAST,
    "div": TokenType.DIV,
    "hex": TokenType.HEX,
    "bin": TokenType.BIN,
    "true": TokenType.TRUE,
    "false": TokenType.FALSE,
    "null": TokenType.NULL,
    "second": TokenType.SECOND,
    "minute": TokenType.MINUTE,
    "hour": TokenType.HOUR,
    "day": TokenType.DAY,
    "week": TokenType.WEEK,
    "month": TokenType.MONTH,
    "quarter": TokenType.QUARTER,
    "year": TokenType.YEAR,
    "microsecond": TokenType.MICROSECOND,
    "asc": TokenType.ORDERBY_ASC,
    "desc": TokenType.ORDERBY_DESC,
    "as": TokenType.AS,
    "binary": TokenType.BINARY,
    "datetime": TokenType.DATETIME,
    "time": TokenType.TIME,
    "char": TokenType.CHAR,
    "date": TokenType.DATE,
    "decimal": TokenType.DECIMAL,
    "signed": TokenType.SIGNED,
    "unsigned": TokenType.UNSIGNED,
    "integer": TokenType.INTEGER,
    "json": TokenType.JSON,
    "second_microsecond": TokenType.SECOND_MICROSECOND,
    "minute_microsecond": TokenType.MINUTE_MICROSECOND,
    "minute_second": TokenType.MINUTE_SECOND,
    "hour_microsecond": TokenType.HOUR_MICROSECOND,
    "hour_second": TokenType.HOUR_SECOND,
    "hour_minute": TokenType.HOUR_MINUTE,
    "day_microsecond": TokenType.DAY_MICROSECOND,
    "day_second": TokenType.DAY_SECOND,
    "day_minute": TokenType.DAY_MINUTE,
    "day_hour": TokenType.DAY_HOUR,
    "year_month": TokenType.YEAR_MONTH,
}

_SQL_FUNTION_RESERVED_WORDS_COLLISION = {
    "binary": TokenType.BINARY,
    "cast": TokenType.CAST,
    "char": TokenType.CHAR,
    "date": TokenType.DATE,
    "decimal": TokenType.DECIMAL,
    "signed": TokenType.SIGNED,
    "time": TokenType.TIME,
    "unsigned": TokenType.UNSIGNED,
}

_OPERATORS = {
    "=": "==",
    "==": "==",
    "and": "&&",
    "div": "div",
    "||": "||",
    "or": "||",
    "not": "not",
    "xor": "xor",
    "^": "^",
    "is": "is",
    "between": "between",
    "in": "in",
    "like": "like",
    "!=": "!=",
    "<>": "!=",
    ">": ">",
    ">=": ">=",
    "<": "<",
    "<=": "<=",
    "&": "&",
    "&&": "&&",
    "|": "|",
    "<<": "<<",
    ">>": ">>",
    "+": "+",
    "-": "-",
    "*": "*",
    "/": "/",
    "~": "~",
    "%": "%",
    "cast": "cast",
    "cont_in": "cont_in",
    "overlaps": "overlaps",
}

_UNARY_OPERATORS = {
    "+": "sign_plus",
    "-": "sign_minus",
    "~": "~",
    "not": "not",
    "!": "!",
}

_NEGATION = {
    "is": "is_not",
    "between": "not_between",
    "regexp": "not_regexp",
    "like": "not_like",
    "in": "not_in",
    "cont_in": "not_cont_in",
    "overlaps": "not_overlaps",
}


class Token:
    """Token representation class."""

    def __init__(self, token_type: int, value: str, length: int = 1) -> None:
        self.token_type: int = token_type
        self.value: str = value
        self.length: int = length

    def __repr__(self) -> str:
        return self.__str__()

    def __str__(self) -> str:
        if self.token_type in (
            TokenType.IDENT,
            TokenType.LNUM,
            TokenType.LSTRING,
        ):
            return f"{self.token_type}({self.value})"
        return f"{self.token_type}"


# static protobuf helper functions


def build_expr(value: BuildExprTypes) -> MessageType:
    msg = Message("Mysqlx.Expr.Expr")
    if isinstance(value, (Message)):
        return value
    if isinstance(value, (ExprParser)):
        return value.expr(reparse=True)
    if isinstance(value, (dict, DbDoc)):
        msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OBJECT")
        msg["object"] = build_object(value).get_message()
    elif isinstance(value, (list, tuple)):
        msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.ARRAY")
        msg["array"] = build_array(value).get_message()
    else:
        msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.LITERAL")
        msg["literal"] = build_scalar(value).get_message()
    return msg


def build_scalar(value: BuildScalarTypes) -> MessageType:
    if isinstance(value, str):
        return build_string_scalar(value)
    if isinstance(value, BYTE_TYPES):
        return build_bytes_scalar(value)
    if isinstance(value, bool):
        return build_bool_scalar(value)
    if isinstance(value, int):
        return build_int_scalar(value)
    if isinstance(value, float):
        return build_double_scalar(value)
    if value is None:
        return build_null_scalar()
    raise ValueError(f"Unsupported data type: {type(value)}")


def build_object(obj: Union[Dict, DbDoc]) -> MessageType:
    if isinstance(obj, DbDoc):
        return build_object(obj.__dict__)

    msg = Message("Mysqlx.Expr.Object")
    for key, value in obj.items():
        pair = Message("Mysqlx.Expr.Object.ObjectField")
        pair["key"] = key.encode() if isinstance(key, str) else key
        pair["value"] = build_expr(value).get_message()
        msg["fld"].extend([pair.get_message()])
    return msg


def build_array(array: Sequence[BuildExprTypes]) -> MessageType:
    msg = Message("Mysqlx.Expr.Array")
    msg["value"].extend([build_expr(value).get_message() for value in array])
    return msg


def build_null_scalar() -> MessageType:
    msg = Message("Mysqlx.Datatypes.Scalar")
    msg["type"] = mysqlxpb_enum("Mysqlx.Datatypes.Scalar.Type.V_NULL")
    return msg


def build_double_scalar(value: float) -> MessageType:
    msg = Message("Mysqlx.Datatypes.Scalar")
    msg["type"] = mysqlxpb_enum("Mysqlx.Datatypes.Scalar.Type.V_DOUBLE")
    msg["v_double"] = value
    return msg


def build_int_scalar(value: int) -> MessageType:
    msg = Message("Mysqlx.Datatypes.Scalar")
    msg["type"] = mysqlxpb_enum("Mysqlx.Datatypes.Scalar.Type.V_SINT")
    msg["v_signed_int"] = value
    return msg


def build_unsigned_int_scalar(value: int) -> MessageType:
    msg = Message("Mysqlx.Datatypes.Scalar")
    msg["type"] = mysqlxpb_enum("Mysqlx.Datatypes.Scalar.Type.V_UINT")
    msg["v_unsigned_int"] = value
    return msg


def build_string_scalar(value: StrOrBytes) -> MessageType:
    if isinstance(value, str):
        value = bytes(bytearray(value, "utf-8"))
    msg = Message("Mysqlx.Datatypes.Scalar")
    msg["type"] = mysqlxpb_enum("Mysqlx.Datatypes.Scalar.Type.V_STRING")
    msg["v_string"] = Message("Mysqlx.Datatypes.Scalar.String", value=value)
    return msg


def build_bool_scalar(value: bool) -> MessageType:
    msg = Message("Mysqlx.Datatypes.Scalar")
    msg["type"] = mysqlxpb_enum("Mysqlx.Datatypes.Scalar.Type.V_BOOL")
    msg["v_bool"] = value
    return msg


def build_bytes_scalar(value: bytes) -> MessageType:
    msg = Message("Mysqlx.Datatypes.Scalar")
    msg["type"] = mysqlxpb_enum("Mysqlx.Datatypes.Scalar.Type.V_OCTETS")
    msg["v_octets"] = Message("Mysqlx.Datatypes.Scalar.Octets", value=value)
    return msg


def build_literal_expr(value: MessageType) -> MessageType:
    msg = Message("Mysqlx.Expr.Expr")
    msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.LITERAL")
    msg["literal"] = value
    return msg


def build_unary_op(name: str, param: MessageType) -> MessageType:
    operator = Message("Mysqlx.Expr.Operator")
    operator["name"] = _UNARY_OPERATORS[name]
    operator["param"] = [param.get_message()]
    msg = Message("Mysqlx.Expr.Expr")
    msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OPERATOR")
    msg["operator"] = operator.get_message()
    return msg


def escape_literal(string: str) -> str:
    return string.replace('"', '""')


class ExprParser:
    """Expression parser class."""

    def __init__(self, string: str, allow_relational: bool = True) -> None:
        self.string: str = string
        self.tokens: List[Token] = []
        self.path_name_queue: List[str] = []
        self.pos: int = 0
        self._allow_relational_columns: bool = allow_relational
        self.placeholder_name_to_position: Dict[str, int] = {}
        self.positional_placeholder_count: int = 0
        self.clean_expression()
        self.lex()

    def __str__(self) -> str:
        return f"<mysqlx.ExprParser '{self.string}'>"

    def clean_expression(self) -> None:
        """Removes the keywords that does not form part of the expression.

        Removes the keywords "SELECT" and "WHERE" that does not form part of
        the expression itself.
        """
        if not isinstance(self.string, str):
            self.string = repr(self.string)
        self.string = self.string.strip(" ")
        if len(self.string) > 1 and self.string[-1] == ";":
            self.string = self.string[:-1]
        if "SELECT" in self.string[:6].upper():
            self.string = self.string[6:]
        if "WHERE" in self.string[:5].upper():
            self.string = self.string[5:]

    # convencience checker for lexer
    def next_char_is(self, key: int, char: str) -> bool:
        return key + 1 < len(self.string) and self.string[key + 1] == char

    def lex_number(self, pos: int) -> Token:
        # numeric literal
        start = pos
        found_dot = False
        while pos < len(self.string) and (
            self.string[pos].isdigit() or self.string[pos] == "."
        ):
            if self.string[pos] == ".":
                if found_dot is True:
                    raise ValueError("Invalid number. Found multiple '.'")
                found_dot = True
            # technically we allow more than one "." and let float()'s parsing
            # complain later
            pos += 1
        val = self.string[start:pos]
        return Token(TokenType.LNUM, val, len(val))

    def lex_alpha(self, i: int, allow_space: bool = False) -> Token:
        start = i
        while i < len(self.string) and (
            self.string[i].isalnum()
            or self.string[i] == "_"
            or (self.string[i].isspace() and allow_space)
        ):
            i += 1

        val = self.string[start:i]
        try:
            if (
                i < len(self.string)
                and self.string[i] == "("
                and val.lower() not in _SQL_FUNTION_RESERVED_WORDS_COLLISION
            ):
                token = Token(TokenType.IDENT, val, len(val))
            else:
                token = Token(_RESERVED_WORDS[val.lower()], val.lower(), len(val))
        except KeyError:
            token = Token(TokenType.IDENT, val, len(val))
        return token

    def lex_quoted_token(self, key: int) -> Token:
        quote_char = self.string[key]
        val = ""
        key += 1
        start = key
        while key < len(self.string):
            char = self.string[key]
            if (
                char == quote_char
                and key + 1 < len(self.string)
                and self.string[key + 1] != quote_char
            ):
                # break if we have a quote char that's not double
                break
            if char in (quote_char, "\\"):
                # this quote char has to be doubled
                if key + 1 >= len(self.string):
                    break
                key += 1
                val += self.string[key]
            else:
                val += char
            key += 1
        if key >= len(self.string) or self.string[key] != quote_char:
            raise ValueError(f"Unterminated quoted string starting at {start}")
        if quote_char == "`":
            return Token(TokenType.IDENT, val, len(val) + 2)
        return Token(TokenType.LSTRING, val, len(val) + 2)

    def lex(self) -> None:
        i = 0
        arrow_last = False
        inside_arrow = False
        while i < len(self.string):
            char = self.string[i]
            if char.isspace():
                i += 1
                continue
            if char.isdigit():
                token = self.lex_number(i)
            elif char.isalpha() or char == "_":
                token = self.lex_alpha(i, inside_arrow)
            elif char == "?":
                token = Token(TokenType.EROTEME, char)
            elif char == ":":
                token = Token(TokenType.COLON, char)
            elif char == "{":
                token = Token(TokenType.LCURLY, char)
            elif char == "}":
                token = Token(TokenType.RCURLY, char)
            elif char == "+":
                token = Token(TokenType.PLUS, char)
            elif char == "-":
                if self.next_char_is(i, ">") and not arrow_last:
                    token = Token(TokenType.ARROW, "->", 2)
                    arrow_last = True
                else:
                    token = Token(TokenType.MINUS, char)
            elif char == "*":
                if self.next_char_is(i, "*"):
                    token = Token(TokenType.DOUBLESTAR, "**", 2)
                else:
                    token = Token(TokenType.MUL, char)
            elif char == "/":
                token = Token(TokenType.DIV, char)
            elif char == "$":
                token = Token(TokenType.DOLLAR, char)
            elif char == "%":
                token = Token(TokenType.MOD, char)
            elif char == "=":
                if self.next_char_is(i, "="):
                    token = Token(TokenType.EQ, "==", 2)
                else:
                    token = Token(TokenType.EQ, "==", 1)
            elif char == "&":
                if self.next_char_is(i, "&"):
                    token = Token(TokenType.ANDAND, "&&", 2)
                else:
                    token = Token(TokenType.BITAND, char)
            elif char == "^":
                token = Token(TokenType.BITXOR, char)
            elif char == "|":
                if self.next_char_is(i, "|"):
                    token = Token(TokenType.OROR, "||", 2)
                else:
                    token = Token(TokenType.BITOR, char)
            elif char == "(":
                token = Token(TokenType.LPAREN, char)
            elif char == ")":
                token = Token(TokenType.RPAREN, char)
            elif char == "[":
                token = Token(TokenType.LSQBRACKET, char)
            elif char == "]":
                token = Token(TokenType.RSQBRACKET, char)
            elif char == "~":
                token = Token(TokenType.NEG, char)
            elif char == ",":
                token = Token(TokenType.COMMA, char)
            elif char == "!":
                if self.next_char_is(i, "="):
                    token = Token(TokenType.NE, "!=", 2)
                else:
                    token = Token(TokenType.BANG, char)
            elif char == "<":
                if self.next_char_is(i, ">"):
                    token = Token(TokenType.NE, "<>", 2)
                elif self.next_char_is(i, "<"):
                    token = Token(TokenType.LSHIFT, "<<", 2)
                elif self.next_char_is(i, "="):
                    token = Token(TokenType.LE, "<=", 2)
                else:
                    token = Token(TokenType.LT, char)
            elif char == ">":
                if self.next_char_is(i, ">"):
                    token = Token(TokenType.RSHIFT, ">>", 2)
                elif self.next_char_is(i, "="):
                    token = Token(TokenType.GE, ">=", 2)
                else:
                    token = Token(TokenType.GT, char)
            elif char == ".":
                if self.next_char_is(i, "*"):
                    token = Token(TokenType.DOTSTAR, ".*", 2)
                elif i + 1 < len(self.string) and self.string[i + 1].isdigit():
                    token = self.lex_number(i)
                else:
                    token = Token(TokenType.DOT, char)
            elif char in ("'", '"') and arrow_last:
                token = Token(TokenType.QUOTE, char)
                if not inside_arrow:
                    inside_arrow = True
                else:
                    arrow_last = False
                    inside_arrow = False
            elif char in ('"', "'", "`"):
                token = self.lex_quoted_token(i)
            else:
                raise ValueError(f"Unknown character at {i}")
            self.tokens.append(token)
            i += token.length

    def assert_cur_token(self, token_type: int) -> None:
        if self.pos >= len(self.tokens):
            raise ValueError(
                f"Expected token type {token_type} at pos {self.pos} but no "
                "tokens left"
            )
        if self.tokens[self.pos].token_type != token_type:
            raise ValueError(
                f"Expected token type {token_type} at pos {self.pos} but found "
                f"type {self.tokens[self.pos]}, on tokens {self.tokens}"
            )

    def cur_token_type_is(self, token_type: int) -> bool:
        return self.pos_token_type_is(self.pos, token_type)

    def cur_token_type_in(self, *types: int) -> bool:
        return self.pos < len(self.tokens) and self.tokens[self.pos].token_type in types

    def next_token_type_is(self, token_type: int) -> bool:
        return self.pos_token_type_is(self.pos + 1, token_type)

    def next_token_type_in(self, *types: int) -> bool:
        return (
            self.pos < len(self.tokens)
            and self.tokens[self.pos + 1].token_type in types
        )

    def pos_token_type_is(self, pos: int, token_type: int) -> bool:
        return pos < len(self.tokens) and self.tokens[pos].token_type == token_type

    def consume_token(self, token_type: int) -> str:
        self.assert_cur_token(token_type)
        value = self.tokens[self.pos].value
        self.pos += 1
        return value

    def paren_expr_list(
        self,
    ) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
        """Parse a paren-bounded expression list for function arguments or IN
        list and return a list of Expr objects.
        """
        exprs = []
        path_name_added = False
        self.consume_token(TokenType.LPAREN)
        if not self.cur_token_type_is(TokenType.RPAREN):
            msg_expr = self._expr().get_message()
            if hasattr(msg_expr, "identifier") and msg_expr.identifier.name:
                self.path_name_queue.insert(0, msg_expr.identifier.name)
                path_name_added = True
            elif (
                not hasattr(msg_expr, "identifier")
                and "identifier" in msg_expr
                and "name" in msg_expr["identifier"]
            ):
                self.path_name_queue.insert(0, msg_expr["identifier"]["name"])
                path_name_added = True
            exprs.append(msg_expr)
            while self.cur_token_type_is(TokenType.COMMA):
                self.pos += 1
                exprs.append(self._expr().get_message())
        self.consume_token(TokenType.RPAREN)
        if path_name_added:
            self.path_name_queue.pop()
        return exprs

    def identifier(self) -> MessageType:
        self.assert_cur_token(TokenType.IDENT)
        ident = Message("Mysqlx.Expr.Identifier")
        if self.next_token_type_is(TokenType.DOT):
            ident["schema_name"] = self.consume_token(TokenType.IDENT)
            self.consume_token(TokenType.DOT)
        ident["name"] = self.tokens[self.pos].value
        self.pos += 1
        return ident

    def function_call(self) -> MessageType:
        function_call = Message("Mysqlx.Expr.FunctionCall")
        function_call["name"] = self.identifier()
        function_call["param"] = self.paren_expr_list()
        msg_expr = Message("Mysqlx.Expr.Expr")
        msg_expr["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.FUNC_CALL")
        msg_expr["function_call"] = function_call.get_message()
        return msg_expr

    def docpath_member(self) -> MessageType:
        self.consume_token(TokenType.DOT)
        token = self.tokens[self.pos]

        if token.token_type == TokenType.IDENT:
            if token.value.startswith("`") and token.value.endswith("`"):
                raise ValueError(
                    f"{token.value} is not a valid JSON/ECMAScript identifier"
                )
            self.consume_token(TokenType.IDENT)
            member_name = token.value
        elif token.token_type == TokenType.LSTRING:
            self.consume_token(TokenType.LSTRING)
            member_name = token.value
        else:
            raise ValueError(
                "Expected token type IDENT or LSTRING in JSON path at token "
                f"pos {self.pos}"
            )
        doc_path_item = Message("Mysqlx.Expr.DocumentPathItem")
        doc_path_item["type"] = mysqlxpb_enum(
            "Mysqlx.Expr.DocumentPathItem.Type.MEMBER"
        )
        doc_path_item["value"] = member_name
        return doc_path_item

    def docpath_array_loc(self) -> MessageType:
        self.consume_token(TokenType.LSQBRACKET)
        if self.cur_token_type_is(TokenType.MUL):
            self.consume_token(TokenType.MUL)
            self.consume_token(TokenType.RSQBRACKET)
            doc_path_item = Message("Mysqlx.Expr.DocumentPathItem")
            doc_path_item["type"] = mysqlxpb_enum(
                "Mysqlx.Expr.DocumentPathItem.Type.ARRAY_INDEX_ASTERISK"
            )
            return doc_path_item
        if self.cur_token_type_is(TokenType.LNUM):
            value = int(self.consume_token(TokenType.LNUM))
            if value < 0:
                raise IndexError(f"Array index cannot be negative at {self.pos}")
            self.consume_token(TokenType.RSQBRACKET)
            doc_path_item = Message("Mysqlx.Expr.DocumentPathItem")
            doc_path_item["type"] = mysqlxpb_enum(
                "Mysqlx.Expr.DocumentPathItem.Type.ARRAY_INDEX"
            )
            doc_path_item["index"] = value
            return doc_path_item
        raise ValueError(
            "Exception token type MUL or LNUM in JSON path array index at "
            f"token pos {self.pos}"
        )

    def document_field(self) -> MessageType:
        if not self.tokens:
            raise ValueError("Empty string cannot be used as document field")
        if self.cur_token_type_is(TokenType.DOLLAR):
            self.consume_token(TokenType.DOLLAR)
        col_id = Message("Mysqlx.Expr.ColumnIdentifier")
        if self.cur_token_type_is(TokenType.IDENT):
            doc_path_item = Message("Mysqlx.Expr.DocumentPathItem")
            doc_path_item["type"] = mysqlxpb_enum(
                "Mysqlx.Expr.DocumentPathItem.Type.MEMBER"
            )
            doc_path_item["value"] = self.consume_token(TokenType.IDENT)
            col_id["document_path"].extend([doc_path_item.get_message()])
        col_id["document_path"].extend(self.document_path())
        if self.path_name_queue:
            col_id["name"] = self.path_name_queue[0]
        expr = Message("Mysqlx.Expr.Expr")
        expr["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.IDENT")
        expr["identifier"] = col_id
        return expr

    def document_path(
        self,
    ) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
        """Parse a JSON-style document path, like WL#7909, but prefix by @.
        instead of $. We parse this as a string because the protocol doesn't
        support it. (yet)
        """
        doc_path = []
        while True:
            if self.cur_token_type_is(TokenType.DOT):
                doc_path.append(self.docpath_member().get_message())
            elif self.cur_token_type_is(TokenType.DOTSTAR):
                self.consume_token(TokenType.DOTSTAR)
                doc_path_item = Message("Mysqlx.Expr.DocumentPathItem")
                doc_path_item["type"] = mysqlxpb_enum(
                    "Mysqlx.Expr.DocumentPathItem.Type.MEMBER_ASTERISK"
                )
                doc_path.append(doc_path_item.get_message())
            elif self.cur_token_type_is(TokenType.LSQBRACKET):
                doc_path.append(self.docpath_array_loc().get_message())
            elif self.cur_token_type_is(TokenType.DOUBLESTAR):
                self.consume_token(TokenType.DOUBLESTAR)
                doc_path_item = Message("Mysqlx.Expr.DocumentPathItem")
                doc_path_item["type"] = mysqlxpb_enum(
                    "Mysqlx.Expr.DocumentPathItem.Type.DOUBLE_ASTERISK"
                )
                doc_path.append(doc_path_item.get_message())
            else:
                break
        items = len(doc_path)
        if items > 0 and get_item_or_attr(doc_path[items - 1], "type") == mysqlxpb_enum(
            "Mysqlx.Expr.DocumentPathItem.Type.DOUBLE_ASTERISK"
        ):
            raise ValueError(f"JSON path may not end in '**' at {self.pos}")
        return doc_path

    def column_identifier(self) -> MessageType:
        parts = []
        parts.append(self.consume_token(TokenType.IDENT))
        while self.cur_token_type_is(TokenType.DOT):
            self.consume_token(TokenType.DOT)
            parts.append(self.consume_token(TokenType.IDENT))
        if len(parts) > 3:
            raise ValueError(f"Too many parts to identifier at {self.pos}")
        parts.reverse()
        col_id = Message("Mysqlx.Expr.ColumnIdentifier")
        # clever way to apply them to the struct
        for i in range(0, len(parts)):
            if i == 0:
                col_id["name"] = parts[0]
            elif i == 1:
                col_id["table_name"] = parts[1]
            elif i == 2:
                col_id["schema_name"] = parts[2]

        is_doc = False
        if self.cur_token_type_is(TokenType.DOLLAR):
            is_doc = True
            self.consume_token(TokenType.DOLLAR)
            col_id["document_path"] = self.document_path()
        elif self.cur_token_type_is(TokenType.ARROW):
            is_doc = True
            self.consume_token(TokenType.ARROW)
            is_quoted = False
            if self.cur_token_type_is(TokenType.QUOTE):
                is_quoted = True
                self.consume_token(TokenType.QUOTE)
            self.consume_token(TokenType.DOLLAR)
            col_id["document_path"] = self.document_path()
            if is_quoted:
                self.consume_token(TokenType.QUOTE)

        if is_doc and len(col_id["document_path"]) == 0:
            doc_path_item = Message("Mysqlx.Expr.DocumentPathItem")
            doc_path_item["type"] = mysqlxpb_enum(
                "Mysqlx.Expr.DocumentPathItem.Type.MEMBER"
            )
            doc_path_item["value"] = ""
            col_id["document_path"].extend([doc_path_item.get_message()])

        msg_expr = Message("Mysqlx.Expr.Expr")
        msg_expr["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.IDENT")
        msg_expr["identifier"] = col_id
        return msg_expr

    def next_token(self) -> Token:
        if self.pos >= len(self.tokens):
            raise ValueError("Unexpected end of token stream")
        token = self.tokens[self.pos]
        self.pos += 1
        return token

    def expect_token(self, token_type: int) -> None:
        token = self.next_token()
        if token.token_type != token_type:
            raise ValueError(f"Expected token type {token_type}")

    def peek_token(self) -> Token:
        return self.tokens[self.pos]

    def consume_any_token(self) -> str:
        value = self.tokens[self.pos].value
        self.pos += 1
        return value

    def parse_json_array(self) -> MessageType:
        """
        jsonArray            ::=  "[" [ expr ("," expr)* ] "]"
        """
        msg = Message("Mysqlx.Expr.Array")
        while self.pos < len(self.tokens) and not self.cur_token_type_is(
            TokenType.RSQBRACKET
        ):
            msg["value"].extend([self._expr().get_message()])
            if not self.cur_token_type_is(TokenType.COMMA):
                break
            self.consume_token(TokenType.COMMA)
        self.consume_token(TokenType.RSQBRACKET)

        expr = Message("Mysqlx.Expr.Expr")
        expr["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.ARRAY")
        expr["array"] = msg.get_message()
        return expr

    def parse_json_doc(self) -> MessageType:
        """
        jsonDoc              ::=  "{" [jsonKeyValue ("," jsonKeyValue)*] "}"
        jsonKeyValue         ::=  STRING_DQ ":" expr
        """
        msg = Message("Mysqlx.Expr.Object")
        while self.pos < len(self.tokens) and not self.cur_token_type_is(
            TokenType.RCURLY
        ):
            item = Message("Mysqlx.Expr.Object.ObjectField")
            item["key"] = self.consume_token(TokenType.LSTRING)
            self.consume_token(TokenType.COLON)
            item["value"] = self._expr().get_message()
            msg["fld"].extend([item.get_message()])
            if not self.cur_token_type_is(TokenType.COMMA):
                break
            self.consume_token(TokenType.COMMA)
        self.consume_token(TokenType.RCURLY)

        expr = Message("Mysqlx.Expr.Expr")
        expr["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OBJECT")
        expr["object"] = msg.get_message()
        return expr

    def parse_place_holder(self, token: Token) -> MessageType:
        place_holder_name = ""
        if self.cur_token_type_is(TokenType.LNUM):
            place_holder_name = self.consume_token(TokenType.LNUM)
        elif self.cur_token_type_is(TokenType.IDENT):
            place_holder_name = self.consume_token(TokenType.IDENT)
        elif token.token_type == TokenType.EROTEME:
            place_holder_name = str(self.positional_placeholder_count)
        else:
            raise ValueError(f"Invalid placeholder name at token pos {self.pos}")

        msg_expr = Message("Mysqlx.Expr.Expr")
        msg_expr["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.PLACEHOLDER")
        if place_holder_name in self.placeholder_name_to_position:
            msg_expr["position"] = self.placeholder_name_to_position[place_holder_name]
        else:
            msg_expr["position"] = self.positional_placeholder_count
            self.placeholder_name_to_position[place_holder_name] = (
                self.positional_placeholder_count
            )
            self.positional_placeholder_count += 1
        return msg_expr

    def cast(self) -> MessageType:
        """cast ::= CAST LPAREN expr AS cast_data_type RPAREN"""
        operator = Message("Mysqlx.Expr.Operator", name="cast")
        self.consume_token(TokenType.LPAREN)
        operator["param"].extend([self._expr().get_message()])
        self.consume_token(TokenType.AS)

        type_scalar = build_bytes_scalar(str.encode(self.cast_data_type()))
        operator["param"].extend([build_literal_expr(type_scalar).get_message()])
        self.consume_token(TokenType.RPAREN)
        msg = Message("Mysqlx.Expr.Expr", operator=operator.get_message())
        msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OPERATOR")
        return msg

    def cast_data_type(self) -> str:
        """cast_data_type ::= ( BINARY dimension? ) |
        ( CHAR dimension? ) |
        ( DATE ) |
        ( DATETIME dimension? ) |
        ( TIME dimension? ) |
        ( DECIMAL dimension? ) |
        ( SIGNED INTEGER? ) |
        ( UNSIGNED INTEGER? ) |
        JSON
        """
        token = self.next_token()
        if token.token_type in (
            TokenType.BINARY,
            TokenType.CHAR,
            TokenType.DATETIME,
            TokenType.TIME,
        ):
            dimension = self.cast_data_type_dimension()
            return f"{token.value}{dimension}" if dimension else token.value
        if token.token_type is TokenType.DECIMAL:
            dimension = self.cast_data_type_dimension(True)
            return f"{token.value}{dimension}" if dimension else token.value
        if token.token_type in (TokenType.SIGNED, TokenType.UNSIGNED):
            if self.cur_token_type_is(TokenType.INTEGER):
                self.consume_token(TokenType.INTEGER)
            return token.value
        if token.token_type in (
            TokenType.INTEGER,
            TokenType.JSON,
            TokenType.DATE,
        ):
            return token.value

        raise ValueError(
            f"Unknown token type {token.token_type} at position {self.pos} "
            f"({token.value})"
        )

    def cast_data_type_dimension(self, decimal: bool = False) -> Optional[str]:
        """dimension ::= LPAREN LNUM (, LNUM)? RPAREN"""
        if not self.cur_token_type_is(TokenType.LPAREN):
            return None

        dimension = []
        self.consume_token(TokenType.LPAREN)
        dimension.append(self.consume_token(TokenType.LNUM))
        if decimal and self.cur_token_type_is(TokenType.COMMA):
            self.consume_token(TokenType.COMMA)
            dimension.append(self.consume_token(TokenType.LNUM))
        self.consume_token(TokenType.RPAREN)

        return (
            f"({dimension[0]})"
            if len(dimension) == 1
            else f"({dimension[0]},{dimension[1]})"
        )

    @staticmethod
    def star_operator() -> MessageType:
        msg = Message("Mysqlx.Expr.Expr")
        msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OPERATOR")
        msg["operator"] = Message("Mysqlx.Expr.Operator", name="*")
        return msg

    def atomic_expr(self) -> MessageType:
        """Parse an atomic expression and return a protobuf Expr object"""
        token = self.next_token()

        if token.token_type in [TokenType.EROTEME, TokenType.COLON]:
            return self.parse_place_holder(token)
        if token.token_type == TokenType.LCURLY:
            return self.parse_json_doc()
        if token.token_type == TokenType.LSQBRACKET:
            return self.parse_json_array()
        if token.token_type == TokenType.CAST:
            return self.cast()
        if token.token_type == TokenType.LPAREN:
            expr = self._expr()
            self.expect_token(TokenType.RPAREN)
            return expr
        if token.token_type in [TokenType.PLUS, TokenType.MINUS]:
            peek = self.peek_token()
            if peek.token_type == TokenType.LNUM:
                self.tokens[self.pos].value = token.value + peek.value
                return self.atomic_expr()
            return build_unary_op(token.value, self.atomic_expr())
        if token.token_type in (TokenType.NOT, TokenType.NEG, TokenType.BANG):
            return build_unary_op(token.value, self.atomic_expr())
        if token.token_type == TokenType.LSTRING:
            return build_literal_expr(build_string_scalar(token.value))
        if token.token_type == TokenType.NULL:
            return build_literal_expr(build_null_scalar())
        if token.token_type == TokenType.LNUM:
            if "." in token.value:
                return build_literal_expr(build_double_scalar(float(token.value)))
            return build_literal_expr(build_int_scalar(int(token.value)))
        if token.token_type in [TokenType.TRUE, TokenType.FALSE]:
            return build_literal_expr(
                build_bool_scalar(token.token_type == TokenType.TRUE)
            )
        if token.token_type == TokenType.DOLLAR:
            return self.document_field()
        if token.token_type == TokenType.MUL:
            return self.star_operator()
        if token.token_type == TokenType.IDENT:
            self.pos = self.pos - 1  # stay on the identifier
            if self.next_token_type_is(TokenType.LPAREN) or (
                self.next_token_type_is(TokenType.DOT)
                and self.pos_token_type_is(self.pos + 2, TokenType.IDENT)
                and self.pos_token_type_is(self.pos + 3, TokenType.LPAREN)
            ):
                # Function call
                return self.function_call()
            return (
                self.document_field()
                if not self._allow_relational_columns
                else self.column_identifier()
            )

        raise ValueError(
            f"Unknown token type = {token.token_type}  when expecting atomic "
            f"expression at {self.pos}"
        )

    def parse_left_assoc_binary_op_expr(
        self, types: Iterable[int], inner_parser: Callable[[], MessageType]
    ) -> MessageType:
        """Given a `set' of types and an Expr-returning inner parser function,
        parse a left associate binary operator expression"""
        lhs = inner_parser()
        while self.pos < len(self.tokens) and self.tokens[self.pos].token_type in types:
            msg = Message("Mysqlx.Expr.Expr")
            msg["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OPERATOR")
            operator = Message("Mysqlx.Expr.Operator")
            operator["name"] = _OPERATORS[self.tokens[self.pos].value]
            operator["param"] = [lhs.get_message()]
            self.pos += 1
            operator["param"].extend([inner_parser().get_message()])
            msg["operator"] = operator
            lhs = msg
        return lhs

    # operator precedence is implemented here
    def add_sub_interval(self) -> MessageType:
        lhs = self.atomic_expr()
        if self.cur_token_type_in(
            TokenType.PLUS, TokenType.MINUS
        ) and self.next_token_type_is(TokenType.INTERVAL):
            token = self.next_token()

            operator = Message("Mysqlx.Expr.Operator")
            operator["param"].extend([lhs.get_message()])
            operator["name"] = (
                "date_add" if token.token_type is TokenType.PLUS else "date_sub"
            )

            self.consume_token(TokenType.INTERVAL)
            operator["param"].extend([self.bit_expr().get_message()])

            if not self.cur_token_type_in(*_INTERVAL_UNITS):
                raise ValueError(f"Expected interval type at position {self.pos}")

            token = str.encode(self.consume_any_token().upper())
            operator["param"].extend(
                [build_literal_expr(build_bytes_scalar(token)).get_message()]
            )

            lhs = Message("Mysqlx.Expr.Expr", operator=operator)
            lhs["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OPERATOR")

        return lhs

    def mul_div_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(
            set([TokenType.MUL, TokenType.DIV, TokenType.MOD]),
            self.add_sub_interval,
        )

    def add_sub_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(
            set([TokenType.PLUS, TokenType.MINUS]), self.mul_div_expr
        )

    def shift_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(
            set([TokenType.LSHIFT, TokenType.RSHIFT]), self.add_sub_expr
        )

    def bit_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(
            set([TokenType.BITAND, TokenType.BITOR, TokenType.BITXOR]),
            self.shift_expr,
        )

    def comp_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(
            set(
                [
                    TokenType.GE,
                    TokenType.GT,
                    TokenType.LE,
                    TokenType.LT,
                    TokenType.EQ,
                    TokenType.NE,
                ]
            ),
            self.bit_expr,
        )

    def ilri_expr(self) -> MessageType:
        params = []
        lhs = self.comp_expr()
        is_not = False
        if self.cur_token_type_is(TokenType.NOT):
            is_not = True
            self.consume_token(TokenType.NOT)
        if self.pos < len(self.tokens):
            params.append(lhs.get_message())
            op_name = self.tokens[self.pos].value
            if self.cur_token_type_is(TokenType.IS):
                self.consume_token(TokenType.IS)
                # for IS, NOT comes AFTER
                if self.cur_token_type_is(TokenType.NOT):
                    is_not = True
                    self.consume_token(TokenType.NOT)
                params.append(self.comp_expr().get_message())
            elif self.cur_token_type_is(TokenType.IN):
                self.consume_token(TokenType.IN)
                if self.cur_token_type_is(TokenType.LPAREN):
                    params.extend(self.paren_expr_list())
                else:
                    op_name = "cont_in"
                    params.append(self.comp_expr().get_message())
            elif self.cur_token_type_is(TokenType.OVERLAPS):
                self.consume_token(TokenType.OVERLAPS)
                params.append(self.comp_expr().get_message())

            elif self.cur_token_type_is(TokenType.LIKE):
                self.consume_token(TokenType.LIKE)
                params.append(self.comp_expr().get_message())
                if self.cur_token_type_is(TokenType.ESCAPE):
                    self.consume_token(TokenType.ESCAPE)
                    params.append(self.comp_expr().get_message())
            elif self.cur_token_type_is(TokenType.BETWEEN):
                self.consume_token(TokenType.BETWEEN)
                params.append(self.comp_expr().get_message())
                self.consume_token(TokenType.AND)
                params.append(self.comp_expr().get_message())
            elif self.cur_token_type_is(TokenType.REGEXP):
                self.consume_token(TokenType.REGEXP)
                params.append(self.comp_expr().get_message())
            else:
                if is_not:
                    raise ValueError(f"Unknown token after NOT as pos {self.pos}")
                op_name = None  # not an operator we're interested in
            if op_name:
                operator = Message("Mysqlx.Expr.Operator")
                operator["name"] = _NEGATION[op_name] if is_not else op_name
                operator["param"] = params
                msg_expr = Message("Mysqlx.Expr.Expr")
                msg_expr["type"] = mysqlxpb_enum("Mysqlx.Expr.Expr.Type.OPERATOR")
                msg_expr["operator"] = operator.get_message()
                lhs = msg_expr
        return lhs

    def and_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(
            set([TokenType.AND, TokenType.ANDAND]), self.ilri_expr
        )

    def xor_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(set([TokenType.XOR]), self.and_expr)

    def or_expr(self) -> MessageType:
        return self.parse_left_assoc_binary_op_expr(
            set([TokenType.OR, TokenType.OROR]), self.xor_expr
        )

    def _expr(self, reparse: bool = False) -> MessageType:
        if reparse:
            self.tokens = []
            self.pos = 0
            self.placeholder_name_to_position = {}
            self.positional_placeholder_count = 0
            self.lex()
        return self.or_expr()

    def expr(self, reparse: bool = False) -> MessageType:
        expression = self._expr(reparse)
        used_tokens = self.pos
        if self.pos_token_type_is(len(self.tokens) - 2, TokenType.AS):
            used_tokens += 2
        if used_tokens < len(self.tokens):
            raise ValueError(
                f"Unused token types {self.tokens[self.pos :]} found in "
                f"expression at position: {self.pos}"
            )
        return expression

    def parse_table_insert_field(self) -> MessageType:
        return Message("Mysqlx.Crud.Column", name=self.consume_token(TokenType.IDENT))

    def parse_table_update_field(self) -> Any:
        return self.column_identifier().identifier

    def _table_fields(self) -> List[str]:
        fields = []
        temp = self.string.split(",")
        temp.reverse()
        while temp:
            field = temp.pop()
            while (
                field.count("(") != field.count(")")
                or field.count("[") != field.count("]")
                or field.count("{") != field.count("}")
            ):
                field = f"{temp.pop()},{field}"
            fields.append(field.strip())
        return fields

    def parse_table_select_projection(
        self,
    ) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
        project_expr: List = []
        first = True
        fields = self._table_fields()
        while self.pos < len(self.tokens):
            if not first:
                self.consume_token(TokenType.COMMA)
            first = False
            projection = Message("Mysqlx.Crud.Projection", source=self._expr())
            if self.cur_token_type_is(TokenType.AS):
                self.consume_token(TokenType.AS)
                projection["alias"] = self.consume_token(TokenType.IDENT)
            else:
                projection["alias"] = fields[len(project_expr)]
            project_expr.append(projection.get_message())

        return project_expr

    def parse_order_spec(
        self,
    ) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
        order_specs = []
        first = True
        while self.pos < len(self.tokens):
            if not first:
                self.consume_token(TokenType.COMMA)
            first = False
            order = Message("Mysqlx.Crud.Order", expr=self._expr())
            if self.cur_token_type_is(TokenType.ORDERBY_ASC):
                order["direction"] = mysqlxpb_enum("Mysqlx.Crud.Order.Direction.ASC")
                self.consume_token(TokenType.ORDERBY_ASC)
            elif self.cur_token_type_is(TokenType.ORDERBY_DESC):
                order["direction"] = mysqlxpb_enum("Mysqlx.Crud.Order.Direction.DESC")
                self.consume_token(TokenType.ORDERBY_DESC)
            order_specs.append(order.get_message())
        return order_specs

    def parse_expr_list(
        self,
    ) -> List[Union[ProtobufMessageType, ProtobufMessageCextType]]:
        expr_list = []
        first = True
        while self.pos < len(self.tokens):
            if not first:
                self.consume_token(TokenType.COMMA)
            first = False
            expr_list.append(self._expr().get_message())
        return expr_list