Feat(mysql): improve support for DDL index column constraints (#1961)

* Feat(mysql): improve support for DDL index column constraints

* Fixup

* Cleanup

* Address comment

* Fix a couple of bugs, add more tests

sqlglot/dialects/clickhouse.py

@@ -380,7 +380,7 @@ def after_limit_modifiers(self, expression: exp.Expression) -> t.List[str]:
             ]
 
         def parameterizedagg_sql(self, expression: exp.Anonymous) -> str:
-            params = self.expressions(expression, "params", flat=True)
+            params = self.expressions(expression, key="params", flat=True)
             return self.func(expression.name, *expression.expressions) + f"({params})"
 
         def placeholder_sql(self, expression: exp.Placeholder) -> str:

sqlglot/dialects/mysql.py

@@ -303,6 +303,22 @@ class Parser(parser.Parser):
             "NAMES": lambda self: self._parse_set_item_names(),
         }
 
+        CONSTRAINT_PARSERS = {
+            **parser.Parser.CONSTRAINT_PARSERS,
+            "FULLTEXT": lambda self: self._parse_index_constraint(kind="FULLTEXT"),
+            "INDEX": lambda self: self._parse_index_constraint(),
+            "KEY": lambda self: self._parse_index_constraint(),
+            "SPATIAL": lambda self: self._parse_index_constraint(kind="SPATIAL"),
+        }
+
+        SCHEMA_UNNAMED_CONSTRAINTS = {
+            *parser.Parser.SCHEMA_UNNAMED_CONSTRAINTS,
+            "FULLTEXT",
+            "INDEX",
+            "KEY",
+            "SPATIAL",
+        }
+
         PROFILE_TYPES = {
             "ALL",
             "BLOCK IO",
@@ -327,6 +343,57 @@ class Parser(parser.Parser):
 
         LOG_DEFAULTS_TO_LN = True
 
+        def _parse_index_constraint(
+            self, kind: t.Optional[str] = None
+        ) -> exp.IndexColumnConstraint:
+            if kind:
+                self._match_texts({"INDEX", "KEY"})
+
+            this = self._parse_id_var(any_token=False)
+            type_ = self._match(TokenType.USING) and self._advance_any() and self._prev.text
+            schema = self._parse_schema()
+
+            options = []
+            while True:
+                if self._match_text_seq("KEY_BLOCK_SIZE"):
+                    self._match(TokenType.EQ)
+                    opt = exp.IndexConstraintOption(key_block_size=self._parse_number())
+                elif self._match(TokenType.USING):
+                    opt = exp.IndexConstraintOption(using=self._advance_any() and self._prev.text)
+                elif self._match_text_seq("WITH", "PARSER"):
+                    opt = exp.IndexConstraintOption(parser=self._parse_var(any_token=True))
+                elif self._match(TokenType.COMMENT):
+                    opt = exp.IndexConstraintOption(comment=self._parse_string())
+                elif self._match_text_seq("VISIBLE"):
+                    opt = exp.IndexConstraintOption(visible=True)
+                elif self._match_text_seq("INVISIBLE"):
+                    opt = exp.IndexConstraintOption(visible=False)
+                elif self._match_text_seq("ENGINE_ATTRIBUTE"):
+                    self._match(TokenType.EQ)
+                    opt = exp.IndexConstraintOption(engine_attr=self._parse_string())
+                elif self._match_text_seq("ENGINE_ATTRIBUTE"):
+                    self._match(TokenType.EQ)
+                    opt = exp.IndexConstraintOption(engine_attr=self._parse_string())
+                elif self._match_text_seq("SECONDARY_ENGINE_ATTRIBUTE"):
+                    self._match(TokenType.EQ)
+                    opt = exp.IndexConstraintOption(secondary_engine_attr=self._parse_string())
+                else:
+                    opt = None
+
+                if not opt:
+                    break
+
+                options.append(opt)
+
+            return self.expression(
+                exp.IndexColumnConstraint,
+                this=this,
+                schema=schema,
+                kind=kind,
+                type=type_,
+                options=options,
+            )
+
         def _parse_show_mysql(
             self,
             this: str,

sqlglot/expressions.py

@@ -1206,6 +1206,19 @@ class MergeTreeTTL(Expression):
     }
 
 
+# https://dev.mysql.com/doc/refman/8.0/en/create-table.html
+class IndexConstraintOption(Expression):
+    arg_types = {
+        "key_block_size": False,
+        "using": False,
+        "parser": False,
+        "comment": False,
+        "visible": False,
+        "engine_attr": False,
+        "secondary_engine_attr": False,
+    }
+
+
 class ColumnConstraint(Expression):
     arg_types = {"this": False, "kind": True}
 
@@ -1272,6 +1285,11 @@ class GeneratedAsIdentityColumnConstraint(ColumnConstraintKind):
     }
 
 
+# https://dev.mysql.com/doc/refman/8.0/en/create-table.html
+class IndexColumnConstraint(ColumnConstraintKind):
+    arg_types = {"this": False, "schema": True, "kind": False, "type": False, "options": False}
+
+
 class InlineLengthColumnConstraint(ColumnConstraintKind):
     pass
 

sqlglot/generator.py

@@ -2569,6 +2569,51 @@ def querytransform_sql(self, expression: exp.QueryTransform) -> str:
         record_reader = f" RECORDREADER {record_reader}" if record_reader else ""
         return f"{transform}{row_format_before}{record_writer}{using}{schema}{row_format_after}{record_reader}"
 
+    def indexconstraintoption_sql(self, expression: exp.IndexConstraintOption) -> str:
+        key_block_size = self.sql(expression, "key_block_size")
+        if key_block_size:
+            return f"KEY_BLOCK_SIZE = {key_block_size}"
+
+        using = self.sql(expression, "using")
+        if using:
+            return f"USING {using}"
+
+        parser = self.sql(expression, "parser")
+        if parser:
+            return f"WITH PARSER {parser}"
+
+        comment = self.sql(expression, "comment")
+        if comment:
+            return f"COMMENT {comment}"
+
+        visible = expression.args.get("visible")
+        if visible is not None:
+            return "VISIBLE" if visible else "INVISIBLE"
+
+        engine_attr = self.sql(expression, "engine_attr")
+        if engine_attr:
+            return f"ENGINE_ATTRIBUTE = {engine_attr}"
+
+        secondary_engine_attr = self.sql(expression, "secondary_engine_attr")
+        if secondary_engine_attr:
+            return f"SECONDARY_ENGINE_ATTRIBUTE = {secondary_engine_attr}"
+
+        self.unsupported("Unsupported index constraint option.")
+        return ""
+
+    def indexcolumnconstraint_sql(self, expression: exp.IndexColumnConstraint) -> str:
+        kind = self.sql(expression, "kind")
+        kind = f"{kind} INDEX" if kind else "INDEX"
+        this = self.sql(expression, "this")
+        this = f" {this}" if this else ""
+        type_ = self.sql(expression, "type")
+        type_ = f" USING {type_}" if type_ else ""
+        schema = self.sql(expression, "schema")
+        schema = f" {schema}" if schema else ""
+        options = self.expressions(expression, key="options", sep=" ")
+        options = f" {options}" if options else ""
+        return f"{kind}{this}{type_}{schema}{options}"
+
 
 def cached_generator(
     cache: t.Optional[t.Dict[int, str]] = None

tests/dialects/test_mysql.py

@@ -9,6 +9,12 @@ def test_ddl(self):
         self.validate_identity("CREATE TABLE foo (id BIGINT)")
         self.validate_identity("UPDATE items SET items.price = 0 WHERE items.id >= 5 LIMIT 10")
         self.validate_identity("DELETE FROM t WHERE a <= 10 LIMIT 10")
+        self.validate_identity("CREATE TABLE foo (a BIGINT, INDEX USING BTREE (b))")
+        self.validate_identity("CREATE TABLE foo (a BIGINT, FULLTEXT INDEX (b))")
+        self.validate_identity("CREATE TABLE foo (a BIGINT, SPATIAL INDEX (b))")
+        self.validate_identity(
+            "CREATE TABLE foo (a BIGINT, INDEX b USING HASH (c) COMMENT 'd' VISIBLE ENGINE_ATTRIBUTE = 'e' WITH PARSER foo)"
+        )
         self.validate_identity(
             "DELETE t1 FROM t1 LEFT JOIN t2 ON t1.id = t2.id WHERE t2.id IS NULL"
         )
@@ -67,6 +73,12 @@ def test_ddl(self):
                 "mysql": "CREATE TABLE `foo` (`id` CHAR(36) NOT NULL DEFAULT (UUID()), PRIMARY KEY (`id`), UNIQUE `id` (`id`))",
             },
         )
+        self.validate_all(
+            "CREATE TABLE IF NOT EXISTS industry_info (a BIGINT(20) NOT NULL AUTO_INCREMENT, b BIGINT(20) NOT NULL, c VARCHAR(1000), PRIMARY KEY (a), UNIQUE KEY d (b), KEY e (b))",
+            write={
+                "mysql": "CREATE TABLE IF NOT EXISTS industry_info (a BIGINT(20) NOT NULL AUTO_INCREMENT, b BIGINT(20) NOT NULL, c VARCHAR(1000), PRIMARY KEY (a), UNIQUE d (b), INDEX e (b))",
+            },
+        )
 
     def test_identity(self):
         self.validate_identity("SELECT 1 XOR 0")