Repository: tobymao/sqlglot

Issue:
TSQL convert function incorrect conversion to sql string
I'm using the TSQL dialect and can confirm that it's an issue with the latest version 21.2.1

```python
from sqlglot import select
from sqlglot.dialects.tsql import TSQL

select_statement = select(
    "CONVERT(varchar, '2023-01-01 23:10:11.123', 121) AS current_datetime", dialect=TSQL
).from_("my_table")
sql = select_statement.sql(dialect=TSQL)
print(sql)
```

Output

```tsql
 SELECT CAST(FORMAT('2023-01-01 23:10:11.123', 'yyyy-MM-dd HH:mm:ss.S') AS VARCHAR(30)) AS current_datetime FROM my_table
```

Which isn't valid TSQL, apparently you can't pass it a varchar in the first position, you have to do something like this apparently with format.
Note .S should be .f repeated to however many digits of precision you need.

```tsql
DECLARE @d DATETIME = '2023-01-01 23:10:11.123'
SELECT CAST(FORMAT(@d, 'yyyy-MM-dd HH:mm:ss.fff') AS VARCHAR(30)) AS current_datetime
```

It would probably be much better that it not try and optimise the original convert statement which can take a varchar.

Discussion:

Patch:
diff --git a/sqlglot/dialects/hive.py b/sqlglot/dialects/hive.py
index 50c41f81..b0d11bab 100644
--- a/sqlglot/dialects/hive.py
+++ b/sqlglot/dialects/hive.py
@@ -609,9 +609,8 @@ class Hive(Dialect):
             return self.properties(properties, prefix=self.seg("TBLPROPERTIES"))
 
         def datatype_sql(self, expression: exp.DataType) -> str:
-            if (
-                expression.this in (exp.DataType.Type.VARCHAR, exp.DataType.Type.NVARCHAR)
-                and not expression.expressions
+            if expression.this in self.PARAMETERIZABLE_TEXT_TYPES and (
+                not expression.expressions or expression.expressions[0].name == "MAX"
             ):
                 expression = exp.DataType.build("text")
             elif expression.is_type(exp.DataType.Type.TEXT) and expression.expressions:
diff --git a/sqlglot/dialects/tsql.py b/sqlglot/dialects/tsql.py
index 2b4c4eee..b6f491f9 100644
--- a/sqlglot/dialects/tsql.py
+++ b/sqlglot/dialects/tsql.py
@@ -18,7 +18,6 @@ from sqlglot.dialects.dialect import (
     timestrtotime_sql,
     trim_sql,
 )
-from sqlglot.expressions import DataType
 from sqlglot.helper import seq_get
 from sqlglot.time import format_time
 from sqlglot.tokens import TokenType
@@ -456,7 +455,6 @@ class TSQL(Dialect):
             "IMAGE": TokenType.IMAGE,
             "MONEY": TokenType.MONEY,
             "NTEXT": TokenType.TEXT,
-            "NVARCHAR(MAX)": TokenType.TEXT,
             "PRINT": TokenType.COMMAND,
             "PROC": TokenType.PROCEDURE,
             "REAL": TokenType.FLOAT,
@@ -467,7 +465,6 @@ class TSQL(Dialect):
             "TOP": TokenType.TOP,
             "UNIQUEIDENTIFIER": TokenType.UNIQUEIDENTIFIER,
             "UPDATE STATISTICS": TokenType.COMMAND,
-            "VARCHAR(MAX)": TokenType.TEXT,
             "XML": TokenType.XML,
             "OUTPUT": TokenType.RETURNING,
             "SYSTEM_USER": TokenType.CURRENT_USER,
@@ -518,19 +515,7 @@ class TSQL(Dialect):
             "TIMEFROMPARTS": _build_timefromparts,
         }
 
-        JOIN_HINTS = {
-            "LOOP",
-            "HASH",
-            "MERGE",
-            "REMOTE",
-        }
-
-        VAR_LENGTH_DATATYPES = {
-            DataType.Type.NVARCHAR,
-            DataType.Type.VARCHAR,
-            DataType.Type.CHAR,
-            DataType.Type.NCHAR,
-        }
+        JOIN_HINTS = {"LOOP", "HASH", "MERGE", "REMOTE"}
 
         RETURNS_TABLE_TOKENS = parser.Parser.ID_VAR_TOKENS - {
             TokenType.TABLE,
@@ -632,48 +617,13 @@ class TSQL(Dialect):
         def _parse_convert(
             self, strict: bool, safe: t.Optional[bool] = None
         ) -> t.Optional[exp.Expression]:
-            to = self._parse_types()
+            this = self._parse_types()
             self._match(TokenType.COMMA)
-            this = self._parse_conjunction()
-
-            if not to or not this:
-                return None
-
-            # Retrieve length of datatype and override to default if not specified
-            if seq_get(to.expressions, 0) is None and to.this in self.VAR_LENGTH_DATATYPES:
-                to = exp.DataType.build(to.this, expressions=[exp.Literal.number(30)], nested=False)
-
-            # Check whether a conversion with format is applicable
-            if self._match(TokenType.COMMA):
-                format_val = self._parse_number()
-                format_val_name = format_val.name if format_val else ""
-
-                if format_val_name not in TSQL.CONVERT_FORMAT_MAPPING:
-                    raise ValueError(
-                        f"CONVERT function at T-SQL does not support format style {format_val_name}"
-                    )
-
-                format_norm = exp.Literal.string(TSQL.CONVERT_FORMAT_MAPPING[format_val_name])
-
-                # Check whether the convert entails a string to date format
-                if to.this == DataType.Type.DATE:
-                    return self.expression(exp.StrToDate, this=this, format=format_norm)
-                # Check whether the convert entails a string to datetime format
-                elif to.this == DataType.Type.DATETIME:
-                    return self.expression(exp.StrToTime, this=this, format=format_norm)
-                # Check whether the convert entails a date to string format
-                elif to.this in self.VAR_LENGTH_DATATYPES:
-                    return self.expression(
-                        exp.Cast if strict else exp.TryCast,
-                        to=to,
-                        this=self.expression(exp.TimeToStr, this=this, format=format_norm),
-                        safe=safe,
-                    )
-                elif to.this == DataType.Type.TEXT:
-                    return self.expression(exp.TimeToStr, this=this, format=format_norm)
-
-            # Entails a simple cast without any format requirement
-            return self.expression(exp.Cast if strict else exp.TryCast, this=this, to=to, safe=safe)
+            args = [this, *self._parse_csv(self._parse_conjunction)]
+            convert = exp.Convert.from_arg_list(args)
+            convert.set("safe", safe)
+            convert.set("strict", strict)
+            return convert
 
         def _parse_user_defined_function(
             self, kind: t.Optional[TokenType] = None
@@ -804,6 +754,9 @@ class TSQL(Dialect):
             exp.DataType.Type.VARIANT: "SQL_VARIANT",
         }
 
+        TYPE_MAPPING.pop(exp.DataType.Type.NCHAR)
+        TYPE_MAPPING.pop(exp.DataType.Type.NVARCHAR)
+
         TRANSFORMS = {
             **generator.Generator.TRANSFORMS,
             exp.AnyValue: any_value_to_max_sql,
@@ -855,6 +808,12 @@ class TSQL(Dialect):
             exp.VolatileProperty: exp.Properties.Location.UNSUPPORTED,
         }
 
+        def convert_sql(self, expression: exp.Convert) -> str:
+            name = "TRY_CONVERT" if expression.args.get("safe") else "CONVERT"
+            return self.func(
+                name, expression.this, expression.expression, expression.args.get("style")
+            )
+
         def queryoption_sql(self, expression: exp.QueryOption) -> str:
             option = self.sql(expression, "this")
             value = self.sql(expression, "expression")
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index bfffe31d..109d5db8 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -3648,6 +3648,10 @@ class Boolean(Condition):
 class DataTypeParam(Expression):
     arg_types = {"this": True, "expression": False}
 
+    @property
+    def name(self) -> str:
+        return self.this.name
+
 
 class DataType(Expression):
     arg_types = {
@@ -4441,6 +4445,11 @@ class ToChar(Func):
     arg_types = {"this": True, "format": False, "nlsparam": False}
 
 
+# https://learn.microsoft.com/en-us/sql/t-sql/functions/cast-and-convert-transact-sql?view=sql-server-ver16#syntax
+class Convert(Func):
+    arg_types = {"this": True, "expression": True, "style": False}
+
+
 class GenerateSeries(Func):
     arg_types = {"start": True, "end": True, "step": False, "is_end_exclusive": False}
 
diff --git a/sqlglot/generator.py b/sqlglot/generator.py
index e8307dfa..569fc7a3 100644
--- a/sqlglot/generator.py
+++ b/sqlglot/generator.py
@@ -460,6 +460,13 @@ class Generator(metaclass=_Generator):
         exp.Paren,
     )
 
+    PARAMETERIZABLE_TEXT_TYPES = {
+        exp.DataType.Type.NVARCHAR,
+        exp.DataType.Type.VARCHAR,
+        exp.DataType.Type.CHAR,
+        exp.DataType.Type.NCHAR,
+    }
+
     # Expressions that need to have all CTEs under them bubbled up to them
     EXPRESSIONS_WITHOUT_NESTED_CTES: t.Set[t.Type[exp.Expression]] = set()
 
@@ -3462,13 +3469,6 @@ class Generator(metaclass=_Generator):
 
         return expression
 
-    def _ensure_string_if_null(self, values: t.List[exp.Expression]) -> t.List[exp.Expression]:
-        return [
-            exp.func("COALESCE", exp.cast(value, "text"), exp.Literal.string(""))
-            for value in values
-            if value
-        ]
-
     def generateseries_sql(self, expression: exp.GenerateSeries) -> str:
         expression.set("is_end_exclusive", None)
         return self.function_fallback_sql(expression)
@@ -3509,3 +3509,47 @@ class Generator(metaclass=_Generator):
         partition = f" {partition}" if partition else ""
 
         return f"TRUNCATE {target}{exists}{tables}{on_cluster}{identity}{option}{partition}"
+
+    # This transpiles T-SQL's CONVERT function
+    # https://learn.microsoft.com/en-us/sql/t-sql/functions/cast-and-convert-transact-sql?view=sql-server-ver16
+    def convert_sql(self, expression: exp.Convert) -> str:
+        to = expression.this
+        value = expression.expression
+        style = expression.args.get("style")
+        safe = expression.args.get("safe")
+        strict = expression.args.get("strict")
+
+        if not to or not value:
+            return ""
+
+        # Retrieve length of datatype and override to default if not specified
+        if not seq_get(to.expressions, 0) and to.this in self.PARAMETERIZABLE_TEXT_TYPES:
+            to = exp.DataType.build(to.this, expressions=[exp.Literal.number(30)], nested=False)
+
+        transformed: t.Optional[exp.Expression] = None
+        cast = exp.Cast if strict else exp.TryCast
+
+        # Check whether a conversion with format (T-SQL calls this 'style') is applicable
+        if isinstance(style, exp.Literal) and style.is_int:
+            from sqlglot.dialects.tsql import TSQL
+
+            style_value = style.name
+            converted_style = TSQL.CONVERT_FORMAT_MAPPING.get(style_value)
+            if not converted_style:
+                self.unsupported(f"Unsupported T-SQL 'style' value: {style_value}")
+
+            fmt = exp.Literal.string(converted_style)
+
+            if to.this == exp.DataType.Type.DATE:
+                transformed = exp.StrToDate(this=value, format=fmt)
+            elif to.this == exp.DataType.Type.DATETIME:
+                transformed = exp.StrToTime(this=value, format=fmt)
+            elif to.this in self.PARAMETERIZABLE_TEXT_TYPES:
+                transformed = cast(this=exp.TimeToStr(this=value, format=fmt), to=to, safe=safe)
+            elif to.this == exp.DataType.Type.TEXT:
+                transformed = exp.TimeToStr(this=value, format=fmt)
+
+        if not transformed:
+            transformed = cast(this=value, to=to, safe=safe)
+
+        return self.sql(transformed)
diff --git a/sqlglot/lineage.py b/sqlglot/lineage.py
index 0fcf1f58..eb428dc7 100644
--- a/sqlglot/lineage.py
+++ b/sqlglot/lineage.py
@@ -7,7 +7,7 @@ from dataclasses import dataclass, field
 
 from sqlglot import Schema, exp, maybe_parse
 from sqlglot.errors import SqlglotError
-from sqlglot.optimizer import Scope, build_scope, find_all_in_scope, qualify
+from sqlglot.optimizer import Scope, build_scope, find_all_in_scope, normalize_identifiers, qualify
 
 if t.TYPE_CHECKING:
     from sqlglot.dialects.dialect import DialectType
@@ -90,6 +90,7 @@ def lineage(
     """
 
     expression = maybe_parse(sql, dialect=dialect)
+    column = normalize_identifiers.normalize_identifiers(column, dialect=dialect).name
 
     if sources:
         expression = exp.expand(
@@ -110,135 +111,141 @@ def lineage(
     if not scope:
         raise SqlglotError("Cannot build lineage, sql must be SELECT")
 
-    def to_node(
-        column: str | int,
-        scope: Scope,
-        scope_name: t.Optional[str] = None,
-        upstream: t.Optional[Node] = None,
-        source_name: t.Optional[str] = None,
-        reference_node_name: t.Optional[str] = None,
-    ) -> Node:
-        source_names = {
-            dt.alias: dt.comments[0].split()[1]
-            for dt in scope.derived_tables
-            if dt.comments and dt.comments[0].startswith("source: ")
-        }
+    if not any(select.alias_or_name == column for select in scope.expression.selects):
+        raise SqlglotError(f"Cannot find column '{column}' in query.")
 
-        # Find the specific select clause that is the source of the column we want.
-        # This can either be a specific, named select or a generic `*` clause.
-        select = (
-            scope.expression.selects[column]
-            if isinstance(column, int)
-            else next(
-                (select for select in scope.expression.selects if select.alias_or_name == column),
-                exp.Star() if scope.expression.is_star else scope.expression,
-            )
-        )
+    return to_node(column, scope, dialect)
 
-        if isinstance(scope.expression, exp.Union):
-            upstream = upstream or Node(name="UNION", source=scope.expression, expression=select)
-
-            index = (
-                column
-                if isinstance(column, int)
-                else next(
-                    (
-                        i
-                        for i, select in enumerate(scope.expression.selects)
-                        if select.alias_or_name == column or select.is_star
-                    ),
-                    -1,  # mypy will not allow a None here, but a negative index should never be returned
-                )
-            )
 
-            if index == -1:
-                raise ValueError(f"Could not find {column} in {scope.expression}")
-
-            for s in scope.union_scopes:
-                to_node(
-                    index,
-                    scope=s,
-                    upstream=upstream,
-                    source_name=source_name,
-                    reference_node_name=reference_node_name,
-                )
-
-            return upstream
-
-        if isinstance(scope.expression, exp.Select):
-            # For better ergonomics in our node labels, replace the full select with
-            # a version that has only the column we care about.
-            #   "x", SELECT x, y FROM foo
-            #     => "x", SELECT x FROM foo
-            source = t.cast(exp.Expression, scope.expression.select(select, append=False))
-        else:
-            source = scope.expression
-
-        # Create the node for this step in the lineage chain, and attach it to the previous one.
-        node = Node(
-            name=f"{scope_name}.{column}" if scope_name else str(column),
-            source=source,
-            expression=select,
-            source_name=source_name or "",
-            reference_node_name=reference_node_name or "",
+def to_node(
+    column: str | int,
+    scope: Scope,
+    dialect: DialectType,
+    scope_name: t.Optional[str] = None,
+    upstream: t.Optional[Node] = None,
+    source_name: t.Optional[str] = None,
+    reference_node_name: t.Optional[str] = None,
+) -> Node:
+    source_names = {
+        dt.alias: dt.comments[0].split()[1]
+        for dt in scope.derived_tables
+        if dt.comments and dt.comments[0].startswith("source: ")
+    }
+
+    # Find the specific select clause that is the source of the column we want.
+    # This can either be a specific, named select or a generic `*` clause.
+    select = (
+        scope.expression.selects[column]
+        if isinstance(column, int)
+        else next(
+            (select for select in scope.expression.selects if select.alias_or_name == column),
+            exp.Star() if scope.expression.is_star else scope.expression,
         )
+    )
 
-        if upstream:
-            upstream.downstream.append(node)
+    if isinstance(scope.expression, exp.Union):
+        upstream = upstream or Node(name="UNION", source=scope.expression, expression=select)
 
-        subquery_scopes = {
-            id(subquery_scope.expression): subquery_scope
-            for subquery_scope in scope.subquery_scopes
-        }
+        index = (
+            column
+            if isinstance(column, int)
+            else next(
+                (
+                    i
+                    for i, select in enumerate(scope.expression.selects)
+                    if select.alias_or_name == column or select.is_star
+                ),
+                -1,  # mypy will not allow a None here, but a negative index should never be returned
+            )
+        )
 
-        for subquery in find_all_in_scope(select, exp.UNWRAPPED_QUERIES):
-            subquery_scope = subquery_scopes.get(id(subquery))
-            if not subquery_scope:
-                logger.warning(f"Unknown subquery scope: {subquery.sql(dialect=dialect)}")
-                continue
+        if index == -1:
+            raise ValueError(f"Could not find {column} in {scope.expression}")
+
+        for s in scope.union_scopes:
+            to_node(
+                index,
+                scope=s,
+                dialect=dialect,
+                upstream=upstream,
+                source_name=source_name,
+                reference_node_name=reference_node_name,
+            )
 
-            for name in subquery.named_selects:
-                to_node(name, scope=subquery_scope, upstream=node)
+        return upstream
+
+    if isinstance(scope.expression, exp.Select):
+        # For better ergonomics in our node labels, replace the full select with
+        # a version that has only the column we care about.
+        #   "x", SELECT x, y FROM foo
+        #     => "x", SELECT x FROM foo
+        source = t.cast(exp.Expression, scope.expression.select(select, append=False))
+    else:
+        source = scope.expression
+
+    # Create the node for this step in the lineage chain, and attach it to the previous one.
+    node = Node(
+        name=f"{scope_name}.{column}" if scope_name else str(column),
+        source=source,
+        expression=select,
+        source_name=source_name or "",
+        reference_node_name=reference_node_name or "",
+    )
 
-        # if the select is a star add all scope sources as downstreams
-        if select.is_star:
-            for source in scope.sources.values():
-                if isinstance(source, Scope):
-                    source = source.expression
-                node.downstream.append(Node(name=select.sql(), source=source, expression=source))
+    if upstream:
+        upstream.downstream.append(node)
 
-        # Find all columns that went into creating this one to list their lineage nodes.
-        source_columns = set(find_all_in_scope(select, exp.Column))
+    subquery_scopes = {
+        id(subquery_scope.expression): subquery_scope for subquery_scope in scope.subquery_scopes
+    }
 
-        # If the source is a UDTF find columns used in the UTDF to generate the table
-        if isinstance(source, exp.UDTF):
-            source_columns |= set(source.find_all(exp.Column))
+    for subquery in find_all_in_scope(select, exp.UNWRAPPED_QUERIES):
+        subquery_scope = subquery_scopes.get(id(subquery))
+        if not subquery_scope:
+            logger.warning(f"Unknown subquery scope: {subquery.sql(dialect=dialect)}")
+            continue
 
-        for c in source_columns:
-            table = c.table
-            source = scope.sources.get(table)
+        for name in subquery.named_selects:
+            to_node(name, scope=subquery_scope, dialect=dialect, upstream=node)
 
+    # if the select is a star add all scope sources as downstreams
+    if select.is_star:
+        for source in scope.sources.values():
             if isinstance(source, Scope):
-                selected_node, _ = scope.selected_sources.get(table, (None, None))
-                # The table itself came from a more specific scope. Recurse into that one using the unaliased column name.
-                to_node(
-                    c.name,
-                    scope=source,
-                    scope_name=table,
-                    upstream=node,
-                    source_name=source_names.get(table) or source_name,
-                    reference_node_name=selected_node.name if selected_node else None,
-                )
-            else:
-                # The source is not a scope - we've reached the end of the line. At this point, if a source is not found
-                # it means this column's lineage is unknown. This can happen if the definition of a source used in a query
-                # is not passed into the `sources` map.
-                source = source or exp.Placeholder()
-                node.downstream.append(Node(name=c.sql(), source=source, expression=source))
-
-        return node
+                source = source.expression
+            node.downstream.append(Node(name=select.sql(), source=source, expression=source))
+
+    # Find all columns that went into creating this one to list their lineage nodes.
+    source_columns = set(find_all_in_scope(select, exp.Column))
+
+    # If the source is a UDTF find columns used in the UTDF to generate the table
+    if isinstance(source, exp.UDTF):
+        source_columns |= set(source.find_all(exp.Column))
+
+    for c in source_columns:
+        table = c.table
+        source = scope.sources.get(table)
+
+        if isinstance(source, Scope):
+            selected_node, _ = scope.selected_sources.get(table, (None, None))
+            # The table itself came from a more specific scope. Recurse into that one using the unaliased column name.
+            to_node(
+                c.name,
+                scope=source,
+                dialect=dialect,
+                scope_name=table,
+                upstream=node,
+                source_name=source_names.get(table) or source_name,
+                reference_node_name=selected_node.name if selected_node else None,
+            )
+        else:
+            # The source is not a scope - we've reached the end of the line. At this point, if a source is not found
+            # it means this column's lineage is unknown. This can happen if the definition of a source used in a query
+            # is not passed into the `sources` map.
+            source = source or exp.Placeholder()
+            node.downstream.append(Node(name=c.sql(), source=source, expression=source))
 
-    return to_node(column if isinstance(column, str) else column.name, scope)
+    return node
 
 
 class GraphHTML:
diff --git a/sqlglot/tokens.py b/sqlglot/tokens.py
index 20f43b93..da9df7d2 100644
--- a/sqlglot/tokens.py
+++ b/sqlglot/tokens.py
@@ -1041,12 +1041,6 @@ class Tokenizer(metaclass=_Tokenizer):
     def _text(self) -> str:
         return self.sql[self._start : self._current]
 
-    def peek(self, i: int = 0) -> str:
-        i = self._current + i
-        if i < self.size:
-            return self.sql[i]
-        return ""
-
     def _add(self, token_type: TokenType, text: t.Optional[str] = None) -> None:
         self._prev_token_line = self._line
 
diff --git a/sqlglotrs/Cargo.lock b/sqlglotrs/Cargo.lock
index 920ef8be..78e1f64d 100644
--- a/sqlglotrs/Cargo.lock
+++ b/sqlglotrs/Cargo.lock
@@ -188,7 +188,7 @@ checksum = "4dccd0940a2dcdf68d092b8cbab7dc0ad8fa938bf95787e1b916b0e3d0e8e970"
 
 [[package]]
 name = "sqlglotrs"
-version = "0.1.1"
+version = "0.1.2"
 dependencies = [
  "pyo3",
 ]
diff --git a/sqlglotrs/src/tokenizer.rs b/sqlglotrs/src/tokenizer.rs
index 84b107af..2c90a650 100644
--- a/sqlglotrs/src/tokenizer.rs
+++ b/sqlglotrs/src/tokenizer.rs
@@ -178,15 +178,6 @@ impl<'a> TokenizerState<'a> {
         Ok(())
     }
 
-    fn peek(&self, i: usize) -> Result<char, TokenizerError> {
-        let index = self.current + i;
-        if index < self.size {
-            self.char_at(index)
-        } else {
-            Ok('\0')
-        }
-    }
-
     fn chars(&self, size: usize) -> String {
         let start = self.current - 1;
         let end = start + size;


Test patch:
diff --git a/tests/dialects/test_dialect.py b/tests/dialects/test_dialect.py
index f81eae1c..5faed51a 100644
--- a/tests/dialects/test_dialect.py
+++ b/tests/dialects/test_dialect.py
@@ -1782,7 +1782,7 @@ class TestDialect(Validator):
             "CREATE TABLE t (c CHAR, nc NCHAR, v1 VARCHAR, v2 VARCHAR2, nv NVARCHAR, nv2 NVARCHAR2)",
             write={
                 "duckdb": "CREATE TABLE t (c TEXT, nc TEXT, v1 TEXT, v2 TEXT, nv TEXT, nv2 TEXT)",
-                "hive": "CREATE TABLE t (c CHAR, nc CHAR, v1 STRING, v2 STRING, nv STRING, nv2 STRING)",
+                "hive": "CREATE TABLE t (c STRING, nc STRING, v1 STRING, v2 STRING, nv STRING, nv2 STRING)",
                 "oracle": "CREATE TABLE t (c CHAR, nc NCHAR, v1 VARCHAR2, v2 VARCHAR2, nv NVARCHAR2, nv2 NVARCHAR2)",
                 "postgres": "CREATE TABLE t (c CHAR, nc CHAR, v1 VARCHAR, v2 VARCHAR, nv VARCHAR, nv2 VARCHAR)",
                 "sqlite": "CREATE TABLE t (c TEXT, nc TEXT, v1 TEXT, v2 TEXT, nv TEXT, nv2 TEXT)",
diff --git a/tests/dialects/test_duckdb.py b/tests/dialects/test_duckdb.py
index 443b1685..58d1f064 100644
--- a/tests/dialects/test_duckdb.py
+++ b/tests/dialects/test_duckdb.py
@@ -7,10 +7,6 @@ class TestDuckDB(Validator):
     dialect = "duckdb"
 
     def test_duckdb(self):
-        self.validate_identity("CREATE TABLE color (name ENUM('RED', 'GREEN', 'BLUE'))")
-        self.validate_identity(
-            "SELECT * FROM x LEFT JOIN UNNEST(y)", "SELECT * FROM x LEFT JOIN UNNEST(y) ON TRUE"
-        )
         self.validate_all(
             'STRUCT_PACK("a b" := 1)',
             write={
@@ -230,6 +226,10 @@ class TestDuckDB(Validator):
         self.validate_identity("FROM (FROM tbl)", "SELECT * FROM (SELECT * FROM tbl)")
         self.validate_identity("FROM tbl", "SELECT * FROM tbl")
         self.validate_identity("x -> '$.family'")
+        self.validate_identity("CREATE TABLE color (name ENUM('RED', 'GREEN', 'BLUE'))")
+        self.validate_identity(
+            "SELECT * FROM x LEFT JOIN UNNEST(y)", "SELECT * FROM x LEFT JOIN UNNEST(y) ON TRUE"
+        )
         self.validate_identity(
             """SELECT '{"foo": [1, 2, 3]}' -> 'foo' -> 0""",
             """SELECT '{"foo": [1, 2, 3]}' -> '$.foo' -> '$[0]'""",
diff --git a/tests/dialects/test_tsql.py b/tests/dialects/test_tsql.py
index a0df87f0..ed474fd1 100644
--- a/tests/dialects/test_tsql.py
+++ b/tests/dialects/test_tsql.py
@@ -283,7 +283,7 @@ class TestTSQL(Validator):
             "CONVERT(INT, CONVERT(NUMERIC, '444.75'))",
             write={
                 "mysql": "CAST(CAST('444.75' AS DECIMAL) AS SIGNED)",
-                "tsql": "CAST(CAST('444.75' AS NUMERIC) AS INTEGER)",
+                "tsql": "CONVERT(INTEGER, CONVERT(NUMERIC, '444.75'))",
             },
         )
         self.validate_all(
@@ -597,7 +597,7 @@ class TestTSQL(Validator):
             "CAST(x as NCHAR(1))",
             write={
                 "spark": "CAST(x AS CHAR(1))",
-                "tsql": "CAST(x AS CHAR(1))",
+                "tsql": "CAST(x AS NCHAR(1))",
             },
         )
 
@@ -605,7 +605,7 @@ class TestTSQL(Validator):
             "CAST(x as NVARCHAR(2))",
             write={
                 "spark": "CAST(x AS VARCHAR(2))",
-                "tsql": "CAST(x AS VARCHAR(2))",
+                "tsql": "CAST(x AS NVARCHAR(2))",
             },
         )
 
@@ -826,11 +826,6 @@ class TestTSQL(Validator):
         )
 
     def test_transaction(self):
-        # BEGIN { TRAN | TRANSACTION }
-        #    [ { transaction_name | @tran_name_variable }
-        #    [ WITH MARK [ 'description' ] ]
-        #    ]
-        # [ ; ]
         self.validate_identity("BEGIN TRANSACTION")
         self.validate_all("BEGIN TRAN", write={"tsql": "BEGIN TRANSACTION"})
         self.validate_identity("BEGIN TRANSACTION transaction_name")
@@ -838,8 +833,6 @@ class TestTSQL(Validator):
         self.validate_identity("BEGIN TRANSACTION transaction_name WITH MARK 'description'")
 
     def test_commit(self):
-        # COMMIT [ { TRAN | TRANSACTION }  [ transaction_name | @tran_name_variable ] ] [ WITH ( DELAYED_DURABILITY = { OFF | ON } ) ] [ ; ]
-
         self.validate_all("COMMIT", write={"tsql": "COMMIT TRANSACTION"})
         self.validate_all("COMMIT TRAN", write={"tsql": "COMMIT TRANSACTION"})
         self.validate_identity("COMMIT TRANSACTION")
@@ -854,11 +847,6 @@ class TestTSQL(Validator):
         )
 
     def test_rollback(self):
-        # Applies to SQL Server and Azure SQL Database
-        # ROLLBACK { TRAN | TRANSACTION }
-        #     [ transaction_name | @tran_name_variable
-        #     | savepoint_name | @savepoint_variable ]
-        # [ ; ]
         self.validate_all("ROLLBACK", write={"tsql": "ROLLBACK TRANSACTION"})
         self.validate_all("ROLLBACK TRAN", write={"tsql": "ROLLBACK TRANSACTION"})
         self.validate_identity("ROLLBACK TRANSACTION")
@@ -978,7 +966,7 @@ WHERE
 
         expected_sqls = [
             "CREATE PROC [dbo].[transform_proc] AS DECLARE @CurrentDate VARCHAR(20)",
-            "SET @CurrentDate = CAST(FORMAT(GETDATE(), 'yyyy-MM-dd HH:mm:ss') AS VARCHAR(20))",
+            "SET @CurrentDate = CONVERT(VARCHAR(20), GETDATE(), 120)",
             "CREATE TABLE [target_schema].[target_table] (a INTEGER) WITH (DISTRIBUTION=REPLICATE, HEAP)",
         ]
 
@@ -1157,155 +1145,173 @@ WHERE
             },
         )
 
-    def test_convert_date_format(self):
+    def test_convert(self):
         self.validate_all(
             "CONVERT(NVARCHAR(200), x)",
             write={
                 "spark": "CAST(x AS VARCHAR(200))",
+                "tsql": "CONVERT(NVARCHAR(200), x)",
             },
         )
         self.validate_all(
             "CONVERT(NVARCHAR, x)",
             write={
                 "spark": "CAST(x AS VARCHAR(30))",
+                "tsql": "CONVERT(NVARCHAR, x)",
             },
         )
         self.validate_all(
             "CONVERT(NVARCHAR(MAX), x)",
             write={
                 "spark": "CAST(x AS STRING)",
+                "tsql": "CONVERT(NVARCHAR(MAX), x)",
             },
         )
         self.validate_all(
             "CONVERT(VARCHAR(200), x)",
             write={
                 "spark": "CAST(x AS VARCHAR(200))",
+                "tsql": "CONVERT(VARCHAR(200), x)",
             },
         )
         self.validate_all(
             "CONVERT(VARCHAR, x)",
             write={
                 "spark": "CAST(x AS VARCHAR(30))",
+                "tsql": "CONVERT(VARCHAR, x)",
             },
         )
         self.validate_all(
             "CONVERT(VARCHAR(MAX), x)",
             write={
                 "spark": "CAST(x AS STRING)",
+                "tsql": "CONVERT(VARCHAR(MAX), x)",
             },
         )
         self.validate_all(
             "CONVERT(CHAR(40), x)",
             write={
                 "spark": "CAST(x AS CHAR(40))",
+                "tsql": "CONVERT(CHAR(40), x)",
             },
         )
         self.validate_all(
             "CONVERT(CHAR, x)",
             write={
                 "spark": "CAST(x AS CHAR(30))",
+                "tsql": "CONVERT(CHAR, x)",
             },
         )
         self.validate_all(
             "CONVERT(NCHAR(40), x)",
             write={
                 "spark": "CAST(x AS CHAR(40))",
+                "tsql": "CONVERT(NCHAR(40), x)",
             },
         )
         self.validate_all(
             "CONVERT(NCHAR, x)",
             write={
                 "spark": "CAST(x AS CHAR(30))",
+                "tsql": "CONVERT(NCHAR, x)",
             },
         )
         self.validate_all(
             "CONVERT(VARCHAR, x, 121)",
             write={
                 "spark": "CAST(DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS VARCHAR(30))",
+                "tsql": "CONVERT(VARCHAR, x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(VARCHAR(40), x, 121)",
             write={
                 "spark": "CAST(DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS VARCHAR(40))",
+                "tsql": "CONVERT(VARCHAR(40), x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(VARCHAR(MAX), x, 121)",
             write={
-                "spark": "DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS')",
+                "spark": "CAST(DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS STRING)",
+                "tsql": "CONVERT(VARCHAR(MAX), x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(NVARCHAR, x, 121)",
             write={
                 "spark": "CAST(DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS VARCHAR(30))",
+                "tsql": "CONVERT(NVARCHAR, x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(NVARCHAR(40), x, 121)",
             write={
                 "spark": "CAST(DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS VARCHAR(40))",
+                "tsql": "CONVERT(NVARCHAR(40), x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(NVARCHAR(MAX), x, 121)",
             write={
-                "spark": "DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS')",
+                "spark": "CAST(DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS STRING)",
+                "tsql": "CONVERT(NVARCHAR(MAX), x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(DATE, x, 121)",
             write={
                 "spark": "TO_DATE(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS')",
+                "tsql": "CONVERT(DATE, x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(DATETIME, x, 121)",
             write={
                 "spark": "TO_TIMESTAMP(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS')",
+                "tsql": "CONVERT(DATETIME2, x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(DATETIME2, x, 121)",
             write={
                 "spark": "TO_TIMESTAMP(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS')",
+                "tsql": "CONVERT(DATETIME2, x, 121)",
             },
         )
         self.validate_all(
             "CONVERT(INT, x)",
             write={
                 "spark": "CAST(x AS INT)",
+                "tsql": "CONVERT(INTEGER, x)",
             },
         )
         self.validate_all(
             "CONVERT(INT, x, 121)",
             write={
                 "spark": "CAST(x AS INT)",
+                "tsql": "CONVERT(INTEGER, x, 121)",
             },
         )
         self.validate_all(
             "TRY_CONVERT(NVARCHAR, x, 121)",
             write={
                 "spark": "TRY_CAST(DATE_FORMAT(x, 'yyyy-MM-dd HH:mm:ss.SSSSSS') AS VARCHAR(30))",
+                "tsql": "TRY_CONVERT(NVARCHAR, x, 121)",
             },
         )
         self.validate_all(
             "TRY_CONVERT(INT, x)",
             write={
                 "spark": "TRY_CAST(x AS INT)",
+                "tsql": "TRY_CONVERT(INTEGER, x)",
             },
         )
         self.validate_all(
             "TRY_CAST(x AS INT)",
             write={
                 "spark": "TRY_CAST(x AS INT)",
-            },
-        )
-        self.validate_all(
-            "CAST(x AS INT)",
-            write={
-                "spark": "CAST(x AS INT)",
+                "tsql": "TRY_CAST(x AS INTEGER)",
             },
         )
         self.validate_all(
@@ -1313,6 +1319,7 @@ WHERE
             write={
                 "mysql": "SELECT CAST(DATE_FORMAT(testdb.dbo.test.x, '%Y-%m-%d %T') AS CHAR(10)) AS y FROM testdb.dbo.test",
                 "spark": "SELECT CAST(DATE_FORMAT(testdb.dbo.test.x, 'yyyy-MM-dd HH:mm:ss') AS VARCHAR(10)) AS y FROM testdb.dbo.test",
+                "tsql": "SELECT CONVERT(VARCHAR(10), testdb.dbo.test.x, 120) AS y FROM testdb.dbo.test",
             },
         )
         self.validate_all(
@@ -1320,12 +1327,14 @@ WHERE
             write={
                 "mysql": "SELECT CAST(y.x AS CHAR(10)) AS z FROM testdb.dbo.test AS y",
                 "spark": "SELECT CAST(y.x AS VARCHAR(10)) AS z FROM testdb.dbo.test AS y",
+                "tsql": "SELECT CONVERT(VARCHAR(10), y.x) AS z FROM testdb.dbo.test AS y",
             },
         )
         self.validate_all(
             "SELECT CAST((SELECT x FROM y) AS VARCHAR) AS test",
             write={
                 "spark": "SELECT CAST((SELECT x FROM y) AS STRING) AS test",
+                "tsql": "SELECT CAST((SELECT x FROM y) AS VARCHAR) AS test",
             },
         )
 
@@ -1721,7 +1730,7 @@ FROM OPENJSON(@json) WITH (
     Date DATETIME2 '$.Order.Date',
     Customer VARCHAR(200) '$.AccountNumber',
     Quantity INTEGER '$.Item.Quantity',
-    [Order] VARCHAR(MAX) AS JSON
+    [Order] NVARCHAR(MAX) AS JSON
 )"""
             },
             pretty=True,
diff --git a/tests/test_lineage.py b/tests/test_lineage.py
index 1329c385..ed1a4483 100644
--- a/tests/test_lineage.py
+++ b/tests/test_lineage.py
@@ -346,3 +346,10 @@ class TestLineage(unittest.TestCase):
             "with _data as (select [struct(1 as a, 2 as b)] as col) select b from _data cross join unnest(col)",
         )
         self.assertEqual(node.name, "b")
+
+    def test_lineage_normalize(self) -> None:
+        node = lineage("a", "WITH x AS (SELECT 1 a) SELECT a FROM x", dialect="snowflake")
+        self.assertEqual(node.name, "A")
+
+        with self.assertRaises(sqlglot.errors.SqlglotError):
+            lineage('"a"', "WITH x AS (SELECT 1 a) SELECT a FROM x", dialect="snowflake")