Repository: tobymao/sqlglot

Issue:
Trino to SparkSQL transpilation error: DATE_DIFF('seconds', x, y)
I'm seeing an error in transpilation in a complex query: 

`date_diff('second', x, y)` transpiling to `date_diff(TO_DATE(x), TO_DATE(y))`

In a simple example, it transpiles to:
```
>>> sqlglot.transpile("select date_diff('seconds', x, y)", read='trino', write='spark')
['SELECT DATEDIFF(y, x)']
```

... but `date_diff('seconds', x, y)` returns the difference in seconds between x and y, whereas DATEDIFF(x,y) returns the difference in days.  A sparksql equivalent might be `unix_timestamp(y) - unix_timestamp(x)`.

Discussion:

Patch:
diff --git a/sqlglot/dialects/hive.py b/sqlglot/dialects/hive.py
index 0110eee5..6d3379a9 100644
--- a/sqlglot/dialects/hive.py
+++ b/sqlglot/dialects/hive.py
@@ -34,6 +34,13 @@ DATE_DELTA_INTERVAL = {
     "DAY": ("DATE_ADD", 1),
 }
 
+TIME_DIFF_FACTOR = {
+    "MILLISECOND": " * 1000",
+    "SECOND": "",
+    "MINUTE": " / 60",
+    "HOUR": " / 3600",
+}
+
 DIFF_MONTH_SWITCH = ("YEAR", "QUARTER", "MONTH")
 
 
@@ -51,6 +58,14 @@ def _add_date_sql(self: generator.Generator, expression: exp.DateAdd) -> str:
 
 def _date_diff_sql(self: generator.Generator, expression: exp.DateDiff) -> str:
     unit = expression.text("unit").upper()
+
+    factor = TIME_DIFF_FACTOR.get(unit)
+    if factor is not None:
+        left = self.sql(expression, "this")
+        right = self.sql(expression, "expression")
+        sec_diff = f"UNIX_TIMESTAMP({left}) - UNIX_TIMESTAMP({right})"
+        return f"({sec_diff}){factor}" if factor else sec_diff
+
     sql_func = "MONTHS_BETWEEN" if unit in DIFF_MONTH_SWITCH else "DATEDIFF"
     _, multiplier = DATE_DELTA_INTERVAL.get(unit, ("", 1))
     multiplier_sql = f" / {multiplier}" if multiplier > 1 else ""
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index 5a1dd3ec..c6b6181b 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -1784,7 +1784,7 @@ class Subqueryable(Unionable):
         instance = _maybe_copy(self, copy)
         return Subquery(
             this=instance,
-            alias=TableAlias(this=to_identifier(alias)),
+            alias=TableAlias(this=to_identifier(alias)) if alias else None,
         )
 
     def limit(self, expression, dialect=None, copy=True, **opts) -> Select:


Test patch:
diff --git a/tests/dialects/test_hive.py b/tests/dialects/test_hive.py
index 84848059..0161f1ea 100644
--- a/tests/dialects/test_hive.py
+++ b/tests/dialects/test_hive.py
@@ -246,6 +246,30 @@ class TestHive(Validator):
         )
 
     def test_time(self):
+        self.validate_all(
+            "(UNIX_TIMESTAMP(y) - UNIX_TIMESTAMP(x)) * 1000",
+            read={
+                "presto": "DATE_DIFF('millisecond', x, y)",
+            },
+        )
+        self.validate_all(
+            "UNIX_TIMESTAMP(y) - UNIX_TIMESTAMP(x)",
+            read={
+                "presto": "DATE_DIFF('second', x, y)",
+            },
+        )
+        self.validate_all(
+            "(UNIX_TIMESTAMP(y) - UNIX_TIMESTAMP(x)) / 60",
+            read={
+                "presto": "DATE_DIFF('minute', x, y)",
+            },
+        )
+        self.validate_all(
+            "(UNIX_TIMESTAMP(y) - UNIX_TIMESTAMP(x)) / 3600",
+            read={
+                "presto": "DATE_DIFF('hour', x, y)",
+            },
+        )
         self.validate_all(
             "DATEDIFF(a, b)",
             write={
diff --git a/tests/test_executor.py b/tests/test_executor.py
index 013ff344..c455f3a3 100644
--- a/tests/test_executor.py
+++ b/tests/test_executor.py
@@ -401,6 +401,16 @@ class TestExecutor(unittest.TestCase):
             ],
         )
 
+        table1_view = exp.Select().select("id", "sub_type").from_("table1").subquery()
+        select_from_sub_query = exp.Select().select("id AS id_alias", "sub_type").from_(table1_view)
+        expression = exp.Select().select("*").from_("cte1").with_("cte1", as_=select_from_sub_query)
+
+        schema = {"table1": {"id": "str", "sub_type": "str"}}
+        executed = execute(expression, tables={t: [] for t in schema}, schema=schema)
+
+        self.assertEqual(executed.rows, [])
+        self.assertEqual(executed.columns, ("id_alias", "sub_type"))
+
     def test_correlated_count(self):
         tables = {
             "parts": [{"pnum": 0, "qoh": 1}],