Implement MySQL-specific MATCH

Added new construct :class:`_mysql.match`, which provides for the full
range of MySQL's MATCH operator including multiple column support and
modifiers. Pull request courtesy Anton Kovalevich.

Fixes: #6132
Closes: #6133
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/6133
Pull-request-sha: dc6842f13688849a848e2ecbb81600e6edf8b3a9

Change-Id: I66bbfd7947aa2e43a031772e9b5ae238d94e5223

3 files changed, 202 insertions, 4 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/__init__.py b/lib/sqlalchemy/dialects/mysql/__init__.py
index 20dd68d8f..4db05984c 100644
--- a/lib/sqlalchemy/dialects/mysql/__init__.py
+++ b/lib/sqlalchemy/dialects/mysql/__init__.py
@@ -49,6 +49,7 @@ from .base import VARCHAR
 from .base import YEAR
 from .dml import Insert
 from .dml import insert
+from .expression import match
 from ...util import compat
 
 if compat.py3k:
@@ -99,4 +100,5 @@ __all__ = (
     "dialect",
     "insert",
     "Insert",
+    "match",
 )
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 92023b3b2..5ebc83a75 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -443,6 +443,15 @@ available.
 
     select(...).with_hint(some_table, "USE INDEX xyz")
 
+* MATCH operator support::
+
+    from sqlalchemy.dialects.mysql import match
+    select(...).where(match(col1, col2, against="some expr").in_boolean_mode())
+
+    .. seealso::
+
+        :class:`_mysql.match`
+
 .. _mysql_insert_on_duplicate_key_update:
 
 INSERT...ON DUPLICATE KEY UPDATE (Upsert)
@@ -928,6 +937,7 @@ output::
 
 from array import array as _array
 from collections import defaultdict
+from itertools import compress
 import re
 
 from sqlalchemy import literal_column
@@ -1583,11 +1593,67 @@ class MySQLCompiler(compiler.SQLCompiler):
             self.process(binary.right, **kw),
         )
 
-    def visit_match_op_binary(self, binary, operator, **kw):
-        return "MATCH (%s) AGAINST (%s IN BOOLEAN MODE)" % (
-            self.process(binary.left, **kw),
-            self.process(binary.right, **kw),
+    _match_valid_flag_combinations = frozenset(
+        (
+            # (boolean_mode, natural_language, query_expansion)
+            (False, False, False),
+            (True, False, False),
+            (False, True, False),
+            (False, False, True),
+            (False, True, True),
         )
+    )
+
+    _match_flag_expressions = (
+        "IN BOOLEAN MODE",
+        "IN NATURAL LANGUAGE MODE",
+        "WITH QUERY EXPANSION",
+    )
+
+    def visit_mysql_match(self, element, **kw):
+        return self.visit_match_op_binary(element, element.operator, **kw)
+
+    def visit_match_op_binary(self, binary, operator, **kw):
+        """
+        Note that `mysql_boolean_mode` is enabled by default because of
+        backward compatibility
+        """
+
+        modifiers = binary.modifiers
+
+        boolean_mode = modifiers.get("mysql_boolean_mode", True)
+        natural_language = modifiers.get("mysql_natural_language", False)
+        query_expansion = modifiers.get("mysql_query_expansion", False)
+
+        flag_combination = (boolean_mode, natural_language, query_expansion)
+
+        if flag_combination not in self._match_valid_flag_combinations:
+            flags = (
+                "in_boolean_mode=%s" % boolean_mode,
+                "in_natural_language_mode=%s" % natural_language,
+                "with_query_expansion=%s" % query_expansion,
+            )
+
+            flags = ", ".join(flags)
+
+            raise exc.CompileError("Invalid MySQL match flags: %s" % flags)
+
+        match_clause = binary.left
+        match_clause = self.process(match_clause, **kw)
+        against_clause = self.process(binary.right, **kw)
+
+        if any(flag_combination):
+            flag_expressions = compress(
+                self._match_flag_expressions,
+                flag_combination,
+            )
+
+            against_clause = [against_clause]
+            against_clause.extend(flag_expressions)
+
+            against_clause = " ".join(against_clause)
+
+        return "MATCH (%s) AGAINST (%s)" % (match_clause, against_clause)
 
     def get_from_hint_text(self, table, text):
         return text
diff --git a/lib/sqlalchemy/dialects/mysql/expression.py b/lib/sqlalchemy/dialects/mysql/expression.py
new file mode 100644
index 000000000..d6ef80ef6
--- /dev/null
+++ b/lib/sqlalchemy/dialects/mysql/expression.py
@@ -0,0 +1,130 @@
+from ... import exc
+from ... import util
+from ...sql import coercions
+from ...sql import elements
+from ...sql import operators
+from ...sql import roles
+from ...sql.base import _generative
+from ...sql.base import Generative
+
+
+class match(Generative, elements.BinaryExpression):
+    """Produce a ``MATCH (X, Y) AGAINST ('TEXT')`` clause.
+
+    E.g.::
+
+        from sqlalchemy import desc
+        from sqlalchemy.dialects.mysql import match
+
+        match_expr = match(
+            users_table.c.firstname,
+            users_table.c.lastname,
+            against="Firstname Lastname",
+        )
+
+        stmt = (
+            select(users_table)
+            .where(match_expr.in_boolean_mode())
+            .order_by(desc(match_expr))
+        )
+
+    Would produce SQL resembling::
+
+        SELECT id, firstname, lastname
+        FROM user
+        WHERE MATCH(firstname, lastname) AGAINST (:param_1 IN BOOLEAN MODE)
+        ORDER BY MATCH(firstname, lastname) AGAINST (:param_2) DESC
+
+    The :func:`_mysql.match` function is a standalone version of the
+    :meth:`_sql.ColumnElement.match` method available on all
+    SQL expressions, as when :meth:`_expression.ColumnElement.match` is
+    used, but allows to pass multiple columns
+
+    :param cols: column expressions to match against
+
+    :param against: expression to be compared towards
+
+    :param in_boolean_mode: boolean, set "boolean mode" to true
+
+    :param in_natural_language_mode: boolean , set "natural language" to true
+
+    :param with_query_expansion: boolean, set "query expansion" to true
+
+    .. versionadded:: 1.4.19
+
+    .. seealso::
+
+        :meth:`_expression.ColumnElement.match`
+
+    """
+
+    __visit_name__ = "mysql_match"
+
+    inherit_cache = True
+
+    def __init__(self, *cols, **kw):
+        if not cols:
+            raise exc.ArgumentError("columns are required")
+
+        against = kw.pop("against", None)
+
+        if not against:
+            raise exc.ArgumentError("against is required")
+        against = coercions.expect(
+            roles.ExpressionElementRole,
+            against,
+        )
+
+        left = elements.BooleanClauseList._construct_raw(
+            operators.comma_op,
+            clauses=cols,
+        )
+        left.group = False
+
+        flags = util.immutabledict(
+            {
+                "mysql_boolean_mode": kw.pop("in_boolean_mode", False),
+                "mysql_natural_language": kw.pop(
+                    "in_natural_language_mode", False
+                ),
+                "mysql_query_expansion": kw.pop("with_query_expansion", False),
+            }
+        )
+
+        if kw:
+            raise exc.ArgumentError("unknown arguments: %s" % (", ".join(kw)))
+
+        super(match, self).__init__(
+            left, against, operators.match_op, modifiers=flags
+        )
+
+    @_generative
+    def in_boolean_mode(self):
+        """Apply the "IN BOOLEAN MODE" modifier to the MATCH expression.
+
+        :return: a new :class:`_mysql.match` instance with modifications
+         applied.
+        """
+
+        self.modifiers = self.modifiers.union({"mysql_boolean_mode": True})
+
+    @_generative
+    def in_natural_language_mode(self):
+        """Apply the "IN NATURAL LANGUAGE MODE" modifier to the MATCH
+        expression.
+
+        :return: a new :class:`_mysql.match` instance with modifications
+         applied.
+        """
+
+        self.modifiers = self.modifiers.union({"mysql_natural_language": True})
+
+    @_generative
+    def with_query_expansion(self):
+        """Apply the "WITH QUERY EXPANSION" modifier to the MATCH expression.
+
+        :return: a new :class:`_mysql.match` instance with modifications
+         applied.
+        """
+
+        self.modifiers = self.modifiers.union({"mysql_query_expansion": True})