Use dialect explicitly while parsing and transforming in Reconciliation (#1365)

Use dialect explicitly while parsing and transforming in Reconciliation.
Currently, it doesn't.

---------

Co-authored-by: SundarShankar89 <[email protected]>

Author: bishwajit-db

src/databricks/labs/remorph/reconcile/query_builder/base.py

@@ -4,13 +4,13 @@
 import sqlglot.expressions as exp
 from sqlglot import Dialect, parse_one
 
-from databricks.labs.remorph.transpiler.sqlglot.dialect_utils import SQLGLOT_DIALECTS
 from databricks.labs.remorph.reconcile.exception import InvalidInputException
 from databricks.labs.remorph.reconcile.query_builder.expression_generator import (
     DataType_transform_mapping,
     transform_expression,
 )
 from databricks.labs.remorph.reconcile.recon_config import Schema, Table, Aggregate
+from databricks.labs.remorph.transpiler.sqlglot.dialect_utils import get_dialect, SQLGLOT_DIALECTS
 
 logger = logging.getLogger(__name__)
 
@@ -91,12 +91,12 @@ def _apply_user_transformation(self, aliases: list[exp.Expression]) -> list[exp.
             with_transform.append(alias.transform(self._user_transformer, self.user_transformations))
         return with_transform
 
-    @staticmethod
-    def _user_transformer(node: exp.Expression, user_transformations: dict[str, str]) -> exp.Expression:
+    def _user_transformer(self, node: exp.Expression, user_transformations: dict[str, str]) -> exp.Expression:
         if isinstance(node, exp.Column) and user_transformations:
+            dialect = self.engine if self.layer == "source" else get_dialect("databricks")
             column_name = node.name
             if column_name in user_transformations.keys():
-                return parse_one(user_transformations.get(column_name, column_name))
+                return parse_one(user_transformations.get(column_name, column_name), read=dialect)
         return node
 
     def _apply_default_transformation(

src/databricks/labs/remorph/reconcile/query_builder/expression_generator.py

@@ -85,7 +85,7 @@ def array_sort(expr: exp.Expression, asc=True) -> exp.Expression:
     return _apply_func_expr(expr, exp.ArraySort, expression=exp.Boolean(this=asc))
 
 
-def anonymous(expr: exp.Column, func: str, is_expr: bool = False) -> exp.Expression:
+def anonymous(expr: exp.Column, func: str, is_expr: bool = False, dialect=None) -> exp.Expression:
     """
 
     This function used in cases where the sql functions are not available in sqlGlot expressions
@@ -104,6 +104,8 @@ def anonymous(expr: exp.Column, func: str, is_expr: bool = False) -> exp.Express
 
     """
     if is_expr:
+        if dialect:
+            return exp.Column(this=func.format(expr.sql(dialect=dialect)))
         return exp.Column(this=func.format(expr))
     is_terminal = isinstance(expr, exp.Column)
     new_expr = expr.copy()
@@ -235,11 +237,17 @@ def _get_is_string(column_types_dict: dict[str, DataType], column_name: str) ->
     "universal": {"default": [partial(coalesce, default='_null_recon_', is_string=True), partial(trim)]},
     "snowflake": {exp.DataType.Type.ARRAY.value: [partial(array_to_string), partial(array_sort)]},
     "oracle": {
-        exp.DataType.Type.NCHAR.value: [partial(anonymous, func="NVL(TRIM(TO_CHAR({})),'_null_recon_')")],
-        exp.DataType.Type.NVARCHAR.value: [partial(anonymous, func="NVL(TRIM(TO_CHAR({})),'_null_recon_')")],
+        exp.DataType.Type.NCHAR.value: [
+            partial(anonymous, func="NVL(TRIM(TO_CHAR({})),'_null_recon_')", dialect=get_dialect("oracle"))
+        ],
+        exp.DataType.Type.NVARCHAR.value: [
+            partial(anonymous, func="NVL(TRIM(TO_CHAR({})),'_null_recon_')", dialect=get_dialect("oracle"))
+        ],
     },
     "databricks": {
-        exp.DataType.Type.ARRAY.value: [partial(anonymous, func="CONCAT_WS(',', SORT_ARRAY({}))")],
+        exp.DataType.Type.ARRAY.value: [
+            partial(anonymous, func="CONCAT_WS(',', SORT_ARRAY({}))", dialect=get_dialect("databricks"))
+        ],
     },
 }
 
@@ -250,7 +258,9 @@ def _get_is_string(column_types_dict: dict[str, DataType], column_name: str) ->
         target=sha256_partial,
     ),
     get_dialect("oracle"): HashAlgoMapping(
-        source=partial(anonymous, func="RAWTOHEX(STANDARD_HASH({}, 'SHA256'))", is_expr=True),
+        source=partial(
+            anonymous, func="RAWTOHEX(STANDARD_HASH({}, 'SHA256'))", is_expr=True, dialect=get_dialect("oracle")
+        ),
         target=sha256_partial,
     ),
     get_dialect("databricks"): HashAlgoMapping(