Repository: TobikoData/sqlmesh
Pull request #2238 (2024-03-08 20:51:54)

Feat: truncate comments to engine-specific maximum
Some engines error if comments are longer than a maximum length. With this PR they will be automatically truncated to the engine's max.

Some engines, like Spark and Trino, have modular compute, metadata, and storage layers. The max comment length may differ across choices of metadata layer, such as Hive (max 4000 chars) or Iceberg (no max). For modular engines, we handle truncation in the Hive mixin rather than the engine adapter itself. 

crericha: Approach looks good. Nice job, @treysp. I left a couple minor comments. I didn't deeply review where the truncation was added. I'll leave that to @eakmanrq who knows that code better than I.

Patch:
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index c61f5785..482e6f8f 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -85,6 +85,8 @@ class EngineAdapter:
     SUPPORTS_INDEXES = False
     COMMENT_CREATION_TABLE = CommentCreationTable.IN_SCHEMA_DEF_CTAS
     COMMENT_CREATION_VIEW = CommentCreationView.IN_SCHEMA_DEF_AND_COMMANDS
+    MAX_TABLE_COMMENT_LENGTH: t.Optional[int] = None
+    MAX_COLUMN_COMMENT_LENGTH: t.Optional[int] = None
     INSERT_OVERWRITE_STRATEGY = InsertOverwriteStrategy.DELETE_INSERT
     SUPPORTS_MATERIALIZED_VIEWS = False
     SUPPORTS_MATERIALIZED_VIEW_SCHEMA = False
@@ -581,7 +583,9 @@ class EngineAdapter:
         if comment:
             return [
                 exp.ColumnConstraint(
-                    kind=exp.CommentColumnConstraint(this=exp.Literal.string(comment))
+                    kind=exp.CommentColumnConstraint(
+                        this=exp.Literal.string(self._truncate_column_comment(comment))
+                    )
                 )
             ]
         return []
@@ -1883,7 +1887,11 @@ class EngineAdapter:
         properties: t.List[exp.Expression] = []
 
         if table_description:
-            properties.append(exp.SchemaCommentProperty(this=exp.Literal.string(table_description)))
+            properties.append(
+                exp.SchemaCommentProperty(
+                    this=exp.Literal.string(self._truncate_table_comment(table_description))
+                )
+            )
 
         if properties:
             return exp.Properties(expressions=properties)
@@ -1898,12 +1906,25 @@ class EngineAdapter:
         properties: t.List[exp.Expression] = []
 
         if table_description:
-            properties.append(exp.SchemaCommentProperty(this=exp.Literal.string(table_description)))
+            properties.append(
+                exp.SchemaCommentProperty(
+                    this=exp.Literal.string(self._truncate_table_comment(table_description))
+                )
+            )
 
         if properties:
             return exp.Properties(expressions=properties)
         return None
 
+    def _truncate_comment(self, comment: str, length: t.Optional[int]) -> str:
+        return comment[:length] if length else comment
+
+    def _truncate_table_comment(self, comment: str) -> str:
+        return self._truncate_comment(comment, self.MAX_TABLE_COMMENT_LENGTH)
+
+    def _truncate_column_comment(self, comment: str) -> str:
+        return self._truncate_comment(comment, self.MAX_COLUMN_COMMENT_LENGTH)
+
     def _to_sql(self, expression: exp.Expression, quote: bool = True, **kwargs: t.Any) -> str:
         """
         Converts an expression to a SQL string. Has a set of default kwargs to apply, and then default
@@ -1980,7 +2001,7 @@ class EngineAdapter:
         return exp.Comment(
             this=table,
             kind=table_kind,
-            expression=exp.Literal.string(table_comment),
+            expression=exp.Literal.string(self._truncate_table_comment(table_comment)),
         )
 
     def _create_table_comment(
@@ -2002,7 +2023,7 @@ class EngineAdapter:
         return exp.Comment(
             this=exp.column(column_name, *reversed(table.parts)),  # type: ignore
             kind="COLUMN",
-            expression=exp.Literal.string(column_comment),
+            expression=exp.Literal.string(self._truncate_column_comment(column_comment)),
         )
 
     def _create_column_comments(
diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py
index 17e8aebf..c83ba3e0 100644
--- a/sqlmesh/core/engine_adapter/bigquery.py
+++ b/sqlmesh/core/engine_adapter/bigquery.py
@@ -50,6 +50,8 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
     SUPPORTS_MATERIALIZED_VIEWS = True
     SUPPORTS_CLONING = True
     CATALOG_SUPPORT = CatalogSupport.FULL_SUPPORT
+    MAX_TABLE_COMMENT_LENGTH = 1024
+    MAX_COLUMN_COMMENT_LENGTH = 1024
 
     # SQL is not supported for adding columns to structs: https://cloud.google.com/bigquery/docs/managing-table-schemas#api_1
     # Can explore doing this with the API in the future
@@ -442,7 +444,9 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
         for i in range(len(table_def["schema"]["fields"])):
             comment = column_comments.get(table_def["schema"]["fields"][i]["name"], None)
             if comment:
-                table_def["schema"]["fields"][i]["description"] = comment
+                table_def["schema"]["fields"][i]["description"] = self._truncate_column_comment(
+                    comment
+                )
 
         # convert dict back to a Table object
         table = table.from_api_repr(table_def)
@@ -451,10 +455,14 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
         logger.info(f"Registering column comments for table {table_name}")
         self._db_call(self.client.update_table, table=table, fields=["schema"])
 
-    def _build_description_property_exp(self, description: str) -> exp.Property:
+    def _build_description_property_exp(
+        self,
+        description: str,
+        trunc_method: t.Callable,
+    ) -> exp.Property:
         return exp.Property(
             this=exp.to_identifier("description", quoted=True),
-            value=exp.Literal.string(description),
+            value=exp.Literal.string(trunc_method(description)),
         )
 
     def _build_table_properties_exp(
@@ -512,7 +520,9 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
 
         if table_description:
             properties.append(
-                self._build_description_property_exp(table_description),
+                self._build_description_property_exp(
+                    table_description, self._truncate_table_comment
+                ),
             )
 
         properties.extend(self._table_properties_to_expressions(table_properties))
@@ -530,7 +540,9 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
                 exp.ColumnConstraint(
                     kind=exp.Properties(
                         expressions=[
-                            self._build_description_property_exp(comment),
+                            self._build_description_property_exp(
+                                comment, self._truncate_column_comment
+                            ),
                         ]
                     )
                 )
@@ -547,7 +559,9 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
 
         if table_description:
             properties.append(
-                self._build_description_property_exp(table_description),
+                self._build_description_property_exp(
+                    table_description, self._truncate_table_comment
+                ),
             )
 
         if properties:
@@ -559,7 +573,7 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
     ) -> exp.Comment | str:
         table_sql = table.sql(dialect=self.dialect, identify=True)
 
-        return f"ALTER {table_kind} {table_sql} SET OPTIONS(description = '{table_comment}')"
+        return f"ALTER {table_kind} {table_sql} SET OPTIONS(description = '{self._truncate_table_comment(table_comment)}')"
 
     def _build_create_comment_column_exp(
         self, table: exp.Table, column_name: str, column_comment: str, table_kind: str = "TABLE"
@@ -567,7 +581,7 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
         table_sql = table.sql(dialect=self.dialect, identify=True)
         column_sql = exp.column(column_name).sql(dialect=self.dialect, identify=True)
 
-        return f"ALTER {table_kind} {table_sql} ALTER COLUMN {column_sql} SET OPTIONS(description = '{column_comment}')"
+        return f"ALTER {table_kind} {table_sql} ALTER COLUMN {column_sql} SET OPTIONS(description = '{self._truncate_column_comment(column_comment)}')"
 
     def create_state_table(
         self,
diff --git a/sqlmesh/core/engine_adapter/mixins.py b/sqlmesh/core/engine_adapter/mixins.py
index 7b375316..8cd56c28 100644
--- a/sqlmesh/core/engine_adapter/mixins.py
+++ b/sqlmesh/core/engine_adapter/mixins.py
@@ -126,6 +126,9 @@ class InsertOverwriteWithMergeMixin(EngineAdapter):
 
 
 class HiveMetastoreTablePropertiesMixin(EngineAdapter):
+    MAX_TABLE_COMMENT_LENGTH = 4000
+    MAX_COLUMN_COMMENT_LENGTH = 4000
+
     def _build_table_properties_exp(
         self,
         catalog_name: t.Optional[str] = None,
@@ -167,7 +170,11 @@ class HiveMetastoreTablePropertiesMixin(EngineAdapter):
             properties.append(property)
 
         if table_description:
-            properties.append(exp.SchemaCommentProperty(this=exp.Literal.string(table_description)))
+            properties.append(
+                exp.SchemaCommentProperty(
+                    this=exp.Literal.string(self._truncate_table_comment(table_description))
+                )
+            )
 
         properties.extend(self._table_properties_to_expressions(table_properties))
 
@@ -184,7 +191,11 @@ class HiveMetastoreTablePropertiesMixin(EngineAdapter):
         properties: t.List[exp.Expression] = []
 
         if table_description:
-            properties.append(exp.SchemaCommentProperty(this=exp.Literal.string(table_description)))
+            properties.append(
+                exp.SchemaCommentProperty(
+                    this=exp.Literal.string(self._truncate_table_comment(table_description))
+                )
+            )
 
         properties.extend(self._table_properties_to_expressions(table_properties))
 
@@ -192,6 +203,12 @@ class HiveMetastoreTablePropertiesMixin(EngineAdapter):
             return exp.Properties(expressions=properties)
         return None
 
+    def _truncate_comment(self, comment: str, length: t.Optional[int]) -> str:
+        # iceberg does not have a comment length limit
+        if self.current_catalog_type == "iceberg":
+            return comment
+        return super()._truncate_comment(comment, length)
+
 
 class GetCurrentCatalogFromFunctionMixin(EngineAdapter):
     CURRENT_CATALOG_EXPRESSION: exp.Expression = exp.func("current_catalog")
diff --git a/sqlmesh/core/engine_adapter/mysql.py b/sqlmesh/core/engine_adapter/mysql.py
index e3e68095..c072ab9e 100644
--- a/sqlmesh/core/engine_adapter/mysql.py
+++ b/sqlmesh/core/engine_adapter/mysql.py
@@ -37,6 +37,8 @@ class MySQLEngineAdapter(
     SUPPORTS_INDEXES = True
     COMMENT_CREATION_TABLE = CommentCreationTable.IN_SCHEMA_DEF_NO_CTAS
     COMMENT_CREATION_VIEW = CommentCreationView.UNSUPPORTED
+    MAX_TABLE_COMMENT_LENGTH = 2048
+    MAX_COLUMN_COMMENT_LENGTH = 1024
     SUPPORTS_REPLACE_TABLE = False
 
     def get_current_catalog(self) -> t.Optional[str]:
@@ -102,7 +104,7 @@ class MySQLEngineAdapter(
     ) -> exp.Comment | str:
         table_sql = table.sql(dialect=self.dialect, identify=True)
 
-        return f"ALTER TABLE {table_sql} COMMENT = '{table_comment}'"
+        return f"ALTER TABLE {table_sql} COMMENT = '{self._truncate_table_comment(table_comment)}'"
 
     def _create_column_comments(
         self,
diff --git a/sqlmesh/core/engine_adapter/spark.py b/sqlmesh/core/engine_adapter/spark.py
index 5e4232ec..dd65f76b 100644
--- a/sqlmesh/core/engine_adapter/spark.py
+++ b/sqlmesh/core/engine_adapter/spark.py
@@ -529,7 +529,7 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
         table_sql = table.sql(dialect=self.dialect, identify=True)
         column_sql = exp.column(column_name).sql(dialect=self.dialect, identify=True)
 
-        return f"ALTER TABLE {table_sql} ALTER COLUMN {column_sql} COMMENT '{column_comment}'"
+        return f"ALTER TABLE {table_sql} ALTER COLUMN {column_sql} COMMENT '{self._truncate_column_comment(column_comment)}'"
 
     @classmethod
     def _wap_branch_name(cls, wap_id: str) -> str:


Test patch:
diff --git a/tests/core/engine_adapter/test_bigquery.py b/tests/core/engine_adapter/test_bigquery.py
index 1c78fa86..ba96c502 100644
--- a/tests/core/engine_adapter/test_bigquery.py
+++ b/tests/core/engine_adapter/test_bigquery.py
@@ -587,6 +587,14 @@ def test_create_table_table_options(make_mocked_engine_adapter: t.Callable, mock
 def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
     adapter = make_mocked_engine_adapter(BigQueryEngineAdapter)
 
+    allowed_table_comment_length = BigQueryEngineAdapter.MAX_TABLE_COMMENT_LENGTH
+    truncated_table_comment = "a" * allowed_table_comment_length
+    long_table_comment = truncated_table_comment + "b"
+
+    allowed_column_comment_length = BigQueryEngineAdapter.MAX_COLUMN_COMMENT_LENGTH
+    truncated_column_comment = "c" * allowed_column_comment_length
+    long_column_comment = truncated_column_comment + "d"
+
     execute_mock = mocker.patch(
         "sqlmesh.core.engine_adapter.bigquery.BigQueryEngineAdapter.execute"
     )
@@ -594,35 +602,35 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
     adapter.create_table(
         "test_table",
         {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
-        table_description="test description",
-        column_descriptions={"a": "a description"},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
     )
 
     adapter.ctas(
         "test_table",
         parse_one("SELECT a, b FROM source_table"),
         {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
-        table_description="test description",
-        column_descriptions={"a": "a description"},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
     )
 
     adapter.create_view(
         "test_table",
         parse_one("SELECT a, b FROM source_table"),
-        table_description="test description",
+        table_description=long_table_comment,
     )
 
     adapter._create_table_comment(
         "test_table",
-        "test description",
+        long_table_comment,
     )
 
     sql_calls = _to_sql_calls(execute_mock)
     assert sql_calls == [
-        "CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='a description'), `b` INT64) OPTIONS (description='test description')",
-        "CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='a description'), `b` INT64) OPTIONS (description='test description') AS SELECT `a`, `b` FROM `source_table`",
-        "CREATE OR REPLACE VIEW `test_table` OPTIONS (description='test description') AS SELECT `a`, `b` FROM `source_table`",
-        "ALTER TABLE `test_table` SET OPTIONS(description = 'test description')",
+        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='{truncated_column_comment}'), `b` INT64) OPTIONS (description='{truncated_table_comment}')",
+        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='{truncated_column_comment}'), `b` INT64) OPTIONS (description='{truncated_table_comment}') AS SELECT `a`, `b` FROM `source_table`",
+        f"CREATE OR REPLACE VIEW `test_table` OPTIONS (description='{truncated_table_comment}') AS SELECT `a`, `b` FROM `source_table`",
+        f"ALTER TABLE `test_table` SET OPTIONS(description = '{truncated_table_comment}')",
     ]
 
 
diff --git a/tests/core/engine_adapter/test_mysql.py b/tests/core/engine_adapter/test_mysql.py
index 1fb04b34..e0f07de1 100644
--- a/tests/core/engine_adapter/test_mysql.py
+++ b/tests/core/engine_adapter/test_mysql.py
@@ -2,6 +2,7 @@
 import typing as t
 
 from pytest_mock.plugin import MockerFixture
+from sqlglot import exp, parse_one
 
 from sqlmesh.core.engine_adapter import MySQLEngineAdapter
 from tests.core.engine_adapter import to_sql_calls
@@ -10,21 +11,54 @@ from tests.core.engine_adapter import to_sql_calls
 def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
     adapter = make_mocked_engine_adapter(MySQLEngineAdapter)
 
+    allowed_table_comment_length = MySQLEngineAdapter.MAX_TABLE_COMMENT_LENGTH
+    truncated_table_comment = "a" * allowed_table_comment_length
+    long_table_comment = truncated_table_comment + "b"
+
+    allowed_column_comment_length = MySQLEngineAdapter.MAX_COLUMN_COMMENT_LENGTH
+    truncated_column_comment = "c" * allowed_column_comment_length
+    long_column_comment = truncated_column_comment + "d"
+
     fetchone_mock = mocker.patch("sqlmesh.core.engine_adapter.mysql.MySQLEngineAdapter.fetchone")
     fetchone_mock.return_value = ["test_table", "CREATE TABLE test_table (a INT)"]
 
+    adapter.create_table(
+        "test_table",
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.ctas(
+        "test_table",
+        parse_one("SELECT a, b FROM source_table"),
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.create_view(
+        "test_view",
+        parse_one("SELECT a, b FROM source_table"),
+        table_description=long_table_comment,
+    )
+
     adapter._create_table_comment(
         "test_table",
-        "test description",
+        long_table_comment,
     )
 
     adapter._create_column_comments(
         "test_table",
-        {"a": "a description"},
+        {"a": long_column_comment},
     )
 
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
-        "ALTER TABLE `test_table` COMMENT = 'test description'",
-        "ALTER TABLE `test_table` MODIFY `a` INT COMMENT 'a description'",
+        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT COMMENT '{truncated_column_comment}', `b` INT) COMMENT='{truncated_table_comment}'",
+        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT='{truncated_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"ALTER TABLE `test_table` MODIFY `a` INT COMMENT '{truncated_column_comment}'",
+        f"CREATE OR REPLACE VIEW `test_view` AS SELECT `a`, `b` FROM `source_table`",
+        f"ALTER TABLE `test_table` COMMENT = '{truncated_table_comment}'",
+        f"ALTER TABLE `test_table` MODIFY `a` INT COMMENT '{truncated_column_comment}'",
     ]
diff --git a/tests/core/engine_adapter/test_spark.py b/tests/core/engine_adapter/test_spark.py
index 00408c6f..c0e1194d 100644
--- a/tests/core/engine_adapter/test_spark.py
+++ b/tests/core/engine_adapter/test_spark.py
@@ -854,23 +854,119 @@ def test_create_table_iceberg(mocker: MockerFixture, make_mocked_engine_adapter:
     ]
 
 
-def test_comments(make_mocked_engine_adapter: t.Callable):
+def test_comments_hive(mocker: MockerFixture, make_mocked_engine_adapter: t.Callable):
     adapter = make_mocked_engine_adapter(SparkEngineAdapter)
 
+    current_catalog_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.EngineAdapter.get_catalog_type"
+    )
+    current_catalog_mock.return_value = "hive"
+
+    allowed_table_comment_length = SparkEngineAdapter.MAX_TABLE_COMMENT_LENGTH
+    truncated_table_comment = "a" * allowed_table_comment_length
+    long_table_comment = truncated_table_comment + "b"
+
+    allowed_column_comment_length = SparkEngineAdapter.MAX_COLUMN_COMMENT_LENGTH
+    truncated_column_comment = "c" * allowed_column_comment_length
+    long_column_comment = truncated_column_comment + "d"
+
+    adapter.create_table(
+        "test_table",
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.ctas(
+        "test_table",
+        parse_one("SELECT a, b FROM source_table"),
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.create_view(
+        "test_view",
+        parse_one("SELECT a, b FROM source_table"),
+        table_description=long_table_comment,
+    )
+
+    adapter._create_table_comment(
+        "test_table",
+        long_table_comment,
+    )
+
+    adapter._create_column_comments(
+        "test_table",
+        {"a": long_column_comment},
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    assert sql_calls == [
+        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT COMMENT '{truncated_column_comment}', `b` INT) COMMENT '{truncated_table_comment}'",
+        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT '{truncated_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"ALTER TABLE `test_table` ALTER COLUMN `a` COMMENT '{truncated_column_comment}'",
+        f"CREATE OR REPLACE VIEW `test_view` COMMENT '{truncated_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"COMMENT ON TABLE `test_table` IS '{truncated_table_comment}'",
+        f"ALTER TABLE `test_table` ALTER COLUMN `a` COMMENT '{truncated_column_comment}'",
+    ]
+
+
+def test_comments_iceberg(mocker: MockerFixture, make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(SparkEngineAdapter)
+
+    current_catalog_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.EngineAdapter.get_catalog_type"
+    )
+    current_catalog_mock.return_value = "iceberg"
+
+    allowed_table_comment_length = SparkEngineAdapter.MAX_TABLE_COMMENT_LENGTH
+    truncated_table_comment = "a" * allowed_table_comment_length
+    long_table_comment = truncated_table_comment + "b"
+
+    allowed_column_comment_length = SparkEngineAdapter.MAX_COLUMN_COMMENT_LENGTH
+    truncated_column_comment = "c" * allowed_column_comment_length
+    long_column_comment = truncated_column_comment + "d"
+
+    adapter.create_table(
+        "test_table",
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.ctas(
+        "test_table",
+        parse_one("SELECT a, b FROM source_table"),
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.create_view(
+        "test_view",
+        parse_one("SELECT a, b FROM source_table"),
+        table_description=long_table_comment,
+    )
+
     adapter._create_table_comment(
         "test_table",
-        "test description",
+        long_table_comment,
     )
 
     adapter._create_column_comments(
         "test_table",
-        {"a": "a description"},
+        {"a": long_column_comment},
     )
 
     sql_calls = to_sql_calls(adapter)
     assert sql_calls == [
-        "COMMENT ON TABLE `test_table` IS 'test description'",
-        "ALTER TABLE `test_table` ALTER COLUMN `a` COMMENT 'a description'",
+        f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT COMMENT '{long_column_comment}', `b` INT) COMMENT '{long_table_comment}'",
+        f"CREATE TABLE IF NOT EXISTS `test_table` COMMENT '{long_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"ALTER TABLE `test_table` ALTER COLUMN `a` COMMENT '{long_column_comment}'",
+        f"CREATE OR REPLACE VIEW `test_view` COMMENT '{long_table_comment}' AS SELECT `a`, `b` FROM `source_table`",
+        f"COMMENT ON TABLE `test_table` IS '{long_table_comment}'",
+        f"ALTER TABLE `test_table` ALTER COLUMN `a` COMMENT '{long_column_comment}'",
     ]
 
 
diff --git a/tests/core/engine_adapter/test_trino.py b/tests/core/engine_adapter/test_trino.py
index 3a8a5067..a8531f18 100644
--- a/tests/core/engine_adapter/test_trino.py
+++ b/tests/core/engine_adapter/test_trino.py
@@ -178,3 +178,129 @@ def test_partitioned_by_with_multiple_catalogs_same_server(
         """CREATE TABLE IF NOT EXISTS "datalake_iceberg"."test_schema"."test_table" ("cola" INTEGER, "colb" VARCHAR) WITH (PARTITIONING=ARRAY['colb'])""",
         """CREATE TABLE IF NOT EXISTS "datalake_iceberg"."test_schema"."test_table" WITH (PARTITIONING=ARRAY['colb']) AS SELECT 1""",
     ]
+
+
+def test_comments_hive(mocker: MockerFixture, make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(TrinoEngineAdapter)
+
+    current_catalog_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog"
+    )
+    current_catalog_mock.return_value = "hive"
+    catalog_type_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_catalog_type"
+    )
+    catalog_type_mock.return_value = "hive"
+
+    allowed_table_comment_length = TrinoEngineAdapter.MAX_TABLE_COMMENT_LENGTH
+    truncated_table_comment = "a" * allowed_table_comment_length
+    long_table_comment = truncated_table_comment + "b"
+
+    allowed_column_comment_length = TrinoEngineAdapter.MAX_COLUMN_COMMENT_LENGTH
+    truncated_column_comment = "c" * allowed_column_comment_length
+    long_column_comment = truncated_column_comment + "d"
+
+    adapter.create_table(
+        "test_table",
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.ctas(
+        "test_table",
+        parse_one("SELECT a, b FROM source_table"),
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.create_view(
+        "test_view",
+        parse_one("SELECT a, b FROM source_table"),
+        table_description=long_table_comment,
+    )
+
+    adapter._create_table_comment(
+        "test_table",
+        long_table_comment,
+    )
+
+    adapter._create_column_comments(
+        "test_table",
+        {"a": long_column_comment},
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    assert sql_calls == [
+        f"""CREATE TABLE IF NOT EXISTS "test_table" ("a" INTEGER COMMENT '{truncated_column_comment}', "b" INTEGER) COMMENT '{truncated_table_comment}'""",
+        f"""CREATE TABLE IF NOT EXISTS "test_table" COMMENT '{truncated_table_comment}' AS SELECT "a", "b" FROM "source_table\"""",
+        f"""COMMENT ON COLUMN "test_table"."a" IS '{truncated_column_comment}'""",
+        f"""CREATE OR REPLACE VIEW "test_view" AS SELECT "a", "b" FROM "source_table\"""",
+        f"""COMMENT ON VIEW "test_view" IS '{truncated_table_comment}'""",
+        f"""COMMENT ON TABLE "test_table" IS '{truncated_table_comment}'""",
+        f"""COMMENT ON COLUMN "test_table"."a" IS '{truncated_column_comment}'""",
+    ]
+
+
+def test_comments_iceberg(mocker: MockerFixture, make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(TrinoEngineAdapter)
+
+    current_catalog_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_current_catalog"
+    )
+    current_catalog_mock.return_value = "iceberg"
+    catalog_type_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.trino.TrinoEngineAdapter.get_catalog_type"
+    )
+    catalog_type_mock.return_value = "iceberg"
+
+    allowed_table_comment_length = TrinoEngineAdapter.MAX_TABLE_COMMENT_LENGTH
+    truncated_table_comment = "a" * allowed_table_comment_length
+    long_table_comment = truncated_table_comment + "b"
+
+    allowed_column_comment_length = TrinoEngineAdapter.MAX_COLUMN_COMMENT_LENGTH
+    truncated_column_comment = "c" * allowed_column_comment_length
+    long_column_comment = truncated_column_comment + "d"
+
+    adapter.create_table(
+        "test_table",
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.ctas(
+        "test_table",
+        parse_one("SELECT a, b FROM source_table"),
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+    )
+
+    adapter.create_view(
+        "test_view",
+        parse_one("SELECT a, b FROM source_table"),
+        table_description=long_table_comment,
+    )
+
+    adapter._create_table_comment(
+        "test_table",
+        long_table_comment,
+    )
+
+    adapter._create_column_comments(
+        "test_table",
+        {"a": long_column_comment},
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    assert sql_calls == [
+        f"""CREATE TABLE IF NOT EXISTS "test_table" ("a" INTEGER COMMENT '{long_column_comment}', "b" INTEGER) COMMENT '{long_table_comment}'""",
+        f"""CREATE TABLE IF NOT EXISTS "test_table" COMMENT '{long_table_comment}' AS SELECT "a", "b" FROM "source_table\"""",
+        f"""COMMENT ON COLUMN "test_table"."a" IS '{long_column_comment}'""",
+        f"""CREATE OR REPLACE VIEW "test_view" AS SELECT "a", "b" FROM "source_table\"""",
+        f"""COMMENT ON VIEW "test_view" IS '{long_table_comment}'""",
+        f"""COMMENT ON TABLE "test_table" IS '{long_table_comment}'""",
+        f"""COMMENT ON COLUMN "test_table"."a" IS '{long_column_comment}'""",
+    ]