Fix Trino/Presto SHA256/SHA512 hex rendering under the pinned sqlglot

Bare SHA256(varchar) is a type error on Trino and binary semantics on
Presto, so the macro now builds LOWER(TO_HEX(SHA256(TO_UTF8(...))))
itself for the Presto family, mirroring those generators' MD5 handling.
A cached probe checks whether the dialect already renders exp.SHA2 in
the hex form, so once the sqlglot pin includes tobymao/sqlglot#7824 the
macro defers to the native rendering and never wraps twice. Adds direct
Trino and Presto SHA256/SHA512 output assertions that hold on both
sides of the pin bump.

Signed-off-by: Pawan Singh Kapkoti <42340841+Pawansingh3889@users.noreply.github.com>

Author: Pawansingh3889

sqlmesh/core/macros.py

@@ -982,9 +982,40 @@ def generate_surrogate_key(
         # not a binary digest, on every dialect.
         func = exp.SHA2(this=func.this, length=func.args.get("length"))
 
+    if isinstance(func, exp.SHA2) and _sha2_renders_binary(evaluator.dialect):
+        # Presto/Trino render a bare SHA256(varchar) for exp.SHA2 on sqlglot
+        # versions without tobymao/sqlglot#7824: a type error on Trino, and
+        # binary rather than string semantics where it runs. Build the
+        # hex-string form explicitly, mirroring what those generators do for
+        # MD5: LOWER(TO_HEX(SHA256(TO_UTF8(...)))). The probe keeps this
+        # branch inert once sqlglot renders the hex form natively, so the
+        # expression is never wrapped twice.
+        return exp.Lower(
+            this=exp.Hex(
+                this=exp.SHA2(
+                    this=exp.Encode(this=func.this, charset=exp.Literal.string("utf-8")),
+                    length=func.args.get("length"),
+                )
+            )
+        )
+
     return func
 
 
+@lru_cache(maxsize=None)
+def _sha2_renders_binary(dialect: DialectType) -> bool:
+    """Whether this dialect renders exp.SHA2 as a bare binary-semantics call.
+
+    Only the Presto family models string and binary hashes separately; other
+    dialects' SHA256(varchar) already returns a hex string.
+    """
+    dialect_name = (str(dialect) if dialect else "").split(",")[0].strip().lower()
+    if dialect_name not in ("presto", "trino", "athena"):
+        return False
+    probe = exp.SHA2(this=exp.column("_sqlmesh_probe"), length=exp.Literal.number(256))
+    return "TO_HEX" not in probe.sql(dialect=dialect)
+
+
 @macro()
 def safe_add(_: MacroEvaluator, *fields: exp.Expr) -> exp.Case:
     """Adds numbers together, substitutes nulls for 0s and only returns null if all fields are null.

tests/core/test_macros.py

@@ -1225,7 +1225,29 @@ def render(dialect: str, hash_function: str) -> str:
         render("trino", "MD5")
         == "SELECT LOWER(TO_HEX(MD5(TO_UTF8(CAST(COALESCE(CAST(a AS VARCHAR), '_sqlmesh_surrogate_key_null_') AS VARCHAR))))) FROM foo"
     )
-    # Trino/Presto render SHA256/SHA512 surrogate keys as
-    # LOWER(TO_HEX(SHA256(TO_UTF8(...)))) once sqlglot ships
-    # tobymao/sqlglot#7824; the string assertions for that belong with the
-    # sqlglot version bump.
+
+    # The reported bug (#5871): Trino/Presto SHA256/SHA512 surrogate keys must
+    # be the hex-string form, not a bare SHA256(varchar). The macro-side
+    # fallback produces it under the current sqlglot pin; once sqlglot renders
+    # exp.SHA2 this way natively (tobymao/sqlglot#7824), the probe disables
+    # the fallback and these assertions hold unchanged.
+    for _dialect in ("trino", "presto"):
+        assert (
+            render(_dialect, "SHA256")
+            == "SELECT LOWER(TO_HEX(SHA256(TO_UTF8(CAST(COALESCE(CAST(a AS VARCHAR), '_sqlmesh_surrogate_key_null_') AS VARCHAR))))) FROM foo"
+        )
+        assert (
+            render(_dialect, "SHA512")
+            == "SELECT LOWER(TO_HEX(SHA512(TO_UTF8(CAST(COALESCE(CAST(a AS VARCHAR), '_sqlmesh_surrogate_key_null_') AS VARCHAR))))) FROM foo"
+        )
+
+    # The fallback is scoped to the Presto family: dialects whose bare
+    # SHA256(varchar) already returns a hex string are left to sqlglot.
+    from sqlmesh.core.macros import _sha2_renders_binary
+
+    assert not _sha2_renders_binary("duckdb")
+    assert not _sha2_renders_binary("bigquery")
+    assert (
+        render("snowflake", "SHA256")
+        == "SELECT SHA256(CONCAT(COALESCE(CAST(a AS VARCHAR), '_sqlmesh_surrogate_key_null_'))) FROM foo"
+    )