Repository: tobymao/sqlglot

Issue:
Spark functions months_bewteen & to_timestamp not transpiling to DuckDB
I'm attempting to transpile from Spark to DuckDB. Sqlglot doesn't seem to be recognising the spark functions "months_between" and "to_timestamp". 

See example below (apologies for the image, I can't paste the code directly), neither of those functions gets transpiled:

![image](https://github.com/tobymao/sqlglot/assets/136445689/b81a5764-c5dc-482e-a741-9dee1a3c2efa)

"months_bewteen" should become "date_diff", and
"to_timestamp" should become "strptime" in duckdb, see below:

![image](https://github.com/tobymao/sqlglot/assets/136445689/866f299c-50b9-41ca-9ba1-6a8fffebffa1)


**Official Documentation**
[spark: months_bewteen](https://spark.apache.org/docs/3.1.3/api/python/reference/api/pyspark.sql.functions.months_between.html)
[spark: to_timestamp](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.to_timestamp.html)
[duckdb: time functions](https://duckdb.org/docs/sql/functions/timestamp.html)



Discussion:

Patch:
diff --git a/sqlglot/dataframe/sql/column.py b/sqlglot/dataframe/sql/column.py
index f4cfebaf..fcfd71e9 100644
--- a/sqlglot/dataframe/sql/column.py
+++ b/sqlglot/dataframe/sql/column.py
@@ -114,7 +114,7 @@ class Column:
         return self.inverse_binary_op(exp.Or, other)
 
     @classmethod
-    def ensure_col(cls, value: t.Optional[t.Union[ColumnOrLiteral, exp.Expression]]):
+    def ensure_col(cls, value: t.Optional[t.Union[ColumnOrLiteral, exp.Expression]]) -> Column:
         return cls(value)
 
     @classmethod
@@ -259,7 +259,7 @@ class Column:
         new_expression = exp.Not(this=exp.Is(this=self.column_expression, expression=exp.Null()))
         return Column(new_expression)
 
-    def cast(self, dataType: t.Union[str, DataType]):
+    def cast(self, dataType: t.Union[str, DataType]) -> Column:
         """
         Functionality Difference: PySpark cast accepts a datatype instance of the datatype class
         Sqlglot doesn't currently replicate this class so it only accepts a string
diff --git a/sqlglot/dataframe/sql/functions.py b/sqlglot/dataframe/sql/functions.py
index 353bdc0f..4d4f6f11 100644
--- a/sqlglot/dataframe/sql/functions.py
+++ b/sqlglot/dataframe/sql/functions.py
@@ -600,8 +600,13 @@ def months_between(
     date1: ColumnOrName, date2: ColumnOrName, roundOff: t.Optional[bool] = None
 ) -> Column:
     if roundOff is None:
-        return Column.invoke_anonymous_function(date1, "MONTHS_BETWEEN", date2)
-    return Column.invoke_anonymous_function(date1, "MONTHS_BETWEEN", date2, roundOff)
+        return Column.invoke_expression_over_column(
+            date1, expression.MonthsBetween, expression=date2
+        )
+
+    return Column.invoke_expression_over_column(
+        date1, expression.MonthsBetween, expression=date2, roundoff=roundOff
+    )
 
 
 def to_date(col: ColumnOrName, format: t.Optional[str] = None) -> Column:
@@ -614,8 +619,9 @@ def to_date(col: ColumnOrName, format: t.Optional[str] = None) -> Column:
 
 def to_timestamp(col: ColumnOrName, format: t.Optional[str] = None) -> Column:
     if format is not None:
-        return Column.invoke_anonymous_function(col, "TO_TIMESTAMP", lit(format))
-    return Column.invoke_anonymous_function(col, "TO_TIMESTAMP")
+        return Column.invoke_expression_over_column(col, expression.StrToTime, format=lit(format))
+
+    return Column.ensure_col(col).cast("timestamp")
 
 
 def trunc(col: ColumnOrName, format: str) -> Column:
diff --git a/sqlglot/dialects/bigquery.py b/sqlglot/dialects/bigquery.py
index ae1eaf0d..1e544de2 100644
--- a/sqlglot/dialects/bigquery.py
+++ b/sqlglot/dialects/bigquery.py
@@ -40,7 +40,7 @@ def _date_add_sql(
 
 
 def _derived_table_values_to_unnest(self: generator.Generator, expression: exp.Values) -> str:
-    if not isinstance(expression.unnest().parent, exp.From):
+    if not expression.find_ancestor(exp.From, exp.Join):
         return self.values_sql(expression)
 
     alias = expression.args.get("alias")
diff --git a/sqlglot/dialects/duckdb.py b/sqlglot/dialects/duckdb.py
index ae48474a..3d75d29b 100644
--- a/sqlglot/dialects/duckdb.py
+++ b/sqlglot/dialects/duckdb.py
@@ -218,6 +218,12 @@ class DuckDB(Dialect):
             exp.JSONBExtractScalar: arrow_json_extract_scalar_sql,
             exp.LogicalOr: rename_func("BOOL_OR"),
             exp.LogicalAnd: rename_func("BOOL_AND"),
+            exp.MonthsBetween: lambda self, e: self.func(
+                "DATEDIFF",
+                "'month'",
+                exp.cast(e.expression, "timestamp"),
+                exp.cast(e.this, "timestamp"),
+            ),
             exp.Properties: no_properties_sql,
             exp.RegexpExtract: regexp_extract_sql,
             exp.RegexpReplace: regexp_replace_sql,
diff --git a/sqlglot/dialects/hive.py b/sqlglot/dialects/hive.py
index 42a29ffc..6bb4b0b1 100644
--- a/sqlglot/dialects/hive.py
+++ b/sqlglot/dialects/hive.py
@@ -365,6 +365,7 @@ class Hive(Dialect):
             exp.Max: max_or_greatest,
             exp.MD5Digest: lambda self, e: self.func("UNHEX", self.func("MD5", e.this)),
             exp.Min: min_or_least,
+            exp.MonthsBetween: lambda self, e: self.func("MONTHS_BETWEEN", e.this, e.expression),
             exp.VarMap: var_map_sql,
             exp.Create: create_with_partitions_sql,
             exp.Quantile: rename_func("PERCENTILE"),
diff --git a/sqlglot/dialects/spark2.py b/sqlglot/dialects/spark2.py
index 2d0137e4..e1a113ca 100644
--- a/sqlglot/dialects/spark2.py
+++ b/sqlglot/dialects/spark2.py
@@ -5,6 +5,7 @@ import typing as t
 from sqlglot import exp, parser, transforms
 from sqlglot.dialects.dialect import (
     create_with_partitions_sql,
+    format_time_lambda,
     pivot_column_names,
     rename_func,
     trim_sql,
@@ -132,6 +133,9 @@ class Spark2(Hive):
             ),
             "STRING": _parse_as_cast("string"),
             "TIMESTAMP": _parse_as_cast("timestamp"),
+            "TO_TIMESTAMP": lambda args: _parse_as_cast("timestamp")(args)
+            if len(args) == 1
+            else format_time_lambda(exp.StrToTime, "spark")(args),
             "TO_UNIX_TIMESTAMP": exp.StrToUnix.from_arg_list,
             "TRUNC": lambda args: exp.DateTrunc(unit=seq_get(args, 1), this=seq_get(args, 0)),
             "WEEKOFYEAR": lambda args: exp.WeekOfYear(this=exp.TsOrDsToDate(this=seq_get(args, 0))),
@@ -221,6 +225,7 @@ class Spark2(Hive):
         TRANSFORMS.pop(exp.ArraySort)
         TRANSFORMS.pop(exp.ILike)
         TRANSFORMS.pop(exp.Left)
+        TRANSFORMS.pop(exp.MonthsBetween)
         TRANSFORMS.pop(exp.Right)
 
         WRAP_DERIVED_VALUES = False
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index 29c0a0bf..4c65614f 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -4098,6 +4098,10 @@ class WeekOfYear(Func):
     _sql_names = ["WEEK_OF_YEAR", "WEEKOFYEAR"]
 
 
+class MonthsBetween(Func):
+    arg_types = {"this": True, "expression": True, "roundoff": False}
+
+
 class LastDateOfMonth(Func):
     pass
 
diff --git a/sqlglot/parser.py b/sqlglot/parser.py
index f46f088e..ef696349 100644
--- a/sqlglot/parser.py
+++ b/sqlglot/parser.py
@@ -2766,6 +2766,7 @@ class Parser(metaclass=_Parser):
         self, this: t.Optional[exp.Expression] = None, top: bool = False
     ) -> t.Optional[exp.Expression]:
         if self._match(TokenType.TOP if top else TokenType.LIMIT):
+            comments = self._prev_comments
             limit_paren = self._match(TokenType.L_PAREN)
             expression = self._parse_number() if top else self._parse_term()
 
@@ -2775,7 +2776,9 @@ class Parser(metaclass=_Parser):
             else:
                 offset = None
 
-            limit_exp = self.expression(exp.Limit, this=this, expression=expression, offset=offset)
+            limit_exp = self.expression(
+                exp.Limit, this=this, expression=expression, offset=offset, comments=comments
+            )
 
             if limit_paren:
                 self._match_r_paren()


Test patch:
diff --git a/tests/dataframe/unit/test_functions.py b/tests/dataframe/unit/test_functions.py
index 556001c6..2fb56508 100644
--- a/tests/dataframe/unit/test_functions.py
+++ b/tests/dataframe/unit/test_functions.py
@@ -851,9 +851,9 @@ class TestFunctions(unittest.TestCase):
 
     def test_to_timestamp(self):
         col_str = SF.to_timestamp("cola")
-        self.assertEqual("TO_TIMESTAMP(cola)", col_str.sql())
+        self.assertEqual("CAST(cola AS TIMESTAMP)", col_str.sql())
         col = SF.to_timestamp(SF.col("cola"))
-        self.assertEqual("TO_TIMESTAMP(cola)", col.sql())
+        self.assertEqual("CAST(cola AS TIMESTAMP)", col.sql())
         col_with_format = SF.to_timestamp("cola", "yyyy-MM-dd")
         self.assertEqual("TO_TIMESTAMP(cola, 'yyyy-MM-dd')", col_with_format.sql())
 
diff --git a/tests/dialects/test_bigquery.py b/tests/dialects/test_bigquery.py
index 3939ba04..d42b81e5 100644
--- a/tests/dialects/test_bigquery.py
+++ b/tests/dialects/test_bigquery.py
@@ -520,6 +520,12 @@ class TestBigQuery(Validator):
                 "bigquery": "SELECT cola, colb FROM UNNEST([STRUCT(1 AS _c0, 'test' AS _c1)])",
             },
         )
+        self.validate_all(
+            "SELECT * FROM UNNEST([STRUCT(1 AS id)]) CROSS JOIN UNNEST([STRUCT(1 AS id)])",
+            read={
+                "postgres": "SELECT * FROM (VALUES (1)) AS t1(id) CROSS JOIN (VALUES (1)) AS t2(id)",
+            },
+        )
         self.validate_all(
             "SELECT cola, colb, colc FROM (VALUES (1, 'test', NULL)) AS tab(cola, colb, colc)",
             write={
diff --git a/tests/dialects/test_spark.py b/tests/dialects/test_spark.py
index b7412459..494cc154 100644
--- a/tests/dialects/test_spark.py
+++ b/tests/dialects/test_spark.py
@@ -233,6 +233,51 @@ TBLPROPERTIES (
         self.validate_identity("TRIM(TRAILING 'SL' FROM 'SSparkSQLS')")
         self.validate_identity("SPLIT(str, pattern, lim)")
 
+        self.validate_all(
+            "SELECT DATEDIFF(month, CAST('1996-10-30' AS TIMESTAMP), CAST('1997-02-28 10:30:00' AS TIMESTAMP))",
+            read={
+                "duckdb": "SELECT DATEDIFF('month', CAST('1996-10-30' AS TIMESTAMP), CAST('1997-02-28 10:30:00' AS TIMESTAMP))",
+            },
+            write={
+                "spark": "SELECT DATEDIFF(month, TO_DATE(CAST('1996-10-30' AS TIMESTAMP)), TO_DATE(CAST('1997-02-28 10:30:00' AS TIMESTAMP)))",
+                "spark2": "SELECT MONTHS_BETWEEN(TO_DATE(CAST('1997-02-28 10:30:00' AS TIMESTAMP)), TO_DATE(CAST('1996-10-30' AS TIMESTAMP)))",
+            },
+        )
+        self.validate_all(
+            "SELECT MONTHS_BETWEEN('1997-02-28 10:30:00', '1996-10-30')",
+            write={
+                "duckdb": "SELECT DATEDIFF('month', CAST('1996-10-30' AS TIMESTAMP), CAST('1997-02-28 10:30:00' AS TIMESTAMP))",
+                "hive": "SELECT MONTHS_BETWEEN('1997-02-28 10:30:00', '1996-10-30')",
+                "spark": "SELECT MONTHS_BETWEEN('1997-02-28 10:30:00', '1996-10-30')",
+            },
+        )
+        self.validate_all(
+            "SELECT MONTHS_BETWEEN('1997-02-28 10:30:00', '1996-10-30', FALSE)",
+            write={
+                "duckdb": "SELECT DATEDIFF('month', CAST('1996-10-30' AS TIMESTAMP), CAST('1997-02-28 10:30:00' AS TIMESTAMP))",
+                "hive": "SELECT MONTHS_BETWEEN('1997-02-28 10:30:00', '1996-10-30')",
+                "spark": "SELECT MONTHS_BETWEEN('1997-02-28 10:30:00', '1996-10-30', FALSE)",
+            },
+        )
+        self.validate_all(
+            "SELECT TO_TIMESTAMP('2016-12-31 00:12:00')",
+            write={
+                "": "SELECT CAST('2016-12-31 00:12:00' AS TIMESTAMP)",
+                "duckdb": "SELECT CAST('2016-12-31 00:12:00' AS TIMESTAMP)",
+                "spark": "SELECT CAST('2016-12-31 00:12:00' AS TIMESTAMP)",
+            },
+        )
+        self.validate_all(
+            "SELECT TO_TIMESTAMP('2016-12-31', 'yyyy-MM-dd')",
+            read={
+                "duckdb": "SELECT STRPTIME('2016-12-31', '%Y-%m-%d')",
+            },
+            write={
+                "": "SELECT STR_TO_TIME('2016-12-31', '%Y-%m-%d')",
+                "duckdb": "SELECT STRPTIME('2016-12-31', '%Y-%m-%d')",
+                "spark": "SELECT TO_TIMESTAMP('2016-12-31', 'yyyy-MM-dd')",
+            },
+        )
         self.validate_all(
             "SELECT RLIKE('John Doe', 'John.*')",
             write={
diff --git a/tests/test_transpile.py b/tests/test_transpile.py
index 7a5de31c..61d8d7a5 100644
--- a/tests/test_transpile.py
+++ b/tests/test_transpile.py
@@ -211,9 +211,12 @@ FROM bar /* comment 5 */, tbl /*          comment 6 */""",
             -- comment 1
             AND bar
             -- comment 2
-            AND bla;
+            AND bla
+            -- comment 3
+            LIMIT 10
+            ;
             """,
-            "SELECT a FROM b WHERE foo AND /* comment 1 */ bar AND /* comment 2 */ bla",
+            "SELECT a FROM b WHERE foo AND /* comment 1 */ bar AND /* comment 2 */ bla LIMIT 10 /* comment 3 */",
         )
         self.validate(
             """