Feat: Introduce runtime check for identifier limits per engine

Author: VaggelisD

sqlmesh/core/engine_adapter/base.py

@@ -106,6 +106,7 @@ class EngineAdapter:
     SUPPORTS_REPLACE_TABLE = True
     DEFAULT_CATALOG_TYPE = DIALECT
     QUOTE_IDENTIFIERS_IN_VIEWS = True
+    MAX_IDENTIFIER_LENGTH: t.Optional[int] = None
 
     def __init__(
         self,
@@ -2138,14 +2139,12 @@ def execute(
         )
         with self.transaction():
             for e in ensure_list(expressions):
-                sql = t.cast(
-                    str,
-                    (
-                        self._to_sql(e, quote=quote_identifiers, **to_sql_kwargs)
-                        if isinstance(e, exp.Expression)
-                        else e
-                    ),
-                )
+                if isinstance(e, exp.Expression):
+                    self._check_identifier_length(e, check_only_ddl=True)
+                    sql = self._to_sql(e, quote=quote_identifiers, **to_sql_kwargs)
+                else:
+                    sql = t.cast(str, e)
+
                 self._log_sql(
                     sql,
                     expression=e if isinstance(e, exp.Expression) else None,
@@ -2516,6 +2515,22 @@ def ping(self) -> None:
     def _select_columns(cls, columns: t.Iterable[str]) -> exp.Select:
         return exp.select(*(exp.column(c, quoted=True) for c in columns))
 
+    def _check_identifier_length(
+        self, expression: exp.Expression, check_only_ddl: bool = True
+    ) -> None:
+        if self.MAX_IDENTIFIER_LENGTH is None or (
+            check_only_ddl and not isinstance(expression, exp.DDL)
+        ):
+            return
+
+        for identifier in expression.find_all(exp.Identifier):
+            name = identifier.name
+            name_length = len(name)
+            if name_length > self.MAX_IDENTIFIER_LENGTH:
+                raise SQLMeshError(
+                    f"Identifier name {name} (length {name_length}) exceeds {self.dialect.capitalize()}'s max identifier limit of {self.MAX_IDENTIFIER_LENGTH} characters"
+                )
+
 
 class EngineAdapterWithIndexSupport(EngineAdapter):
     SUPPORTS_INDEXES = True

sqlmesh/core/engine_adapter/mysql.py

@@ -39,6 +39,7 @@ class MySQLEngineAdapter(
     MAX_TABLE_COMMENT_LENGTH = 2048
     MAX_COLUMN_COMMENT_LENGTH = 1024
     SUPPORTS_REPLACE_TABLE = False
+    MAX_IDENTIFIER_LENGTH = 64
     SCHEMA_DIFFER = SchemaDiffer(
         parameterized_type_defaults={
             exp.DataType.build("BIT", dialect=DIALECT).this: [(1,)],

sqlmesh/core/engine_adapter/postgres.py

@@ -34,6 +34,7 @@ class PostgresEngineAdapter(
     HAS_VIEW_BINDING = True
     CURRENT_CATALOG_EXPRESSION = exp.column("current_catalog")
     SUPPORTS_REPLACE_TABLE = False
+    MAX_IDENTIFIER_LENGTH = 63
     SCHEMA_DIFFER = SchemaDiffer(
         parameterized_type_defaults={
             # DECIMAL without precision is "up to 131072 digits before the decimal point; up to 16383 digits after the decimal point"

tests/core/engine_adapter/integration/test_integration.py

@@ -2652,3 +2652,17 @@ def execute(
         {"id": 1, "name": "foo"} if ctx.dialect != "snowflake" else {"ID": 1, "NAME": "foo"}
     )
     assert df.iloc[0].to_dict() == expected_result
+
+
+def test_identifier_length_limit(ctx: TestContext):
+    adapter = ctx.engine_adapter
+    if adapter.MAX_IDENTIFIER_LENGTH is None:
+        pytest.skip(f"Engine {adapter.dialect} does not have identifier length limits set.")
+
+    long_table_name = "a" * (adapter.MAX_IDENTIFIER_LENGTH + 1)
+
+    with pytest.raises(
+        SQLMeshError,
+        match=f"Identifier name {long_table_name} (length {len(long_table_name)}) exceeds {adapter.dialect.capitalize()}'s max identifier limit of {adapter.MAX_IDENTIFIER_LENGTH} characters",
+    ):
+        adapter.create_table(long_table_name, {"col": exp.DataType.build("int")})