Repository: tobymao/sqlglot

Issue:
PL/SQL supports pseudo column LEVEL in hierarchical queries
```
script = """
SELECT
	t1.c1

FROM t1
LEFT JOIN t2 ON t1.c2 = t2.c2

WHERE (t1.TOP_PARENT_INDICATOR = 'Y' or LEVEL = 1)
	  start with (t1.id is not null) connect by prior t1.id = t2.id
"""

qualify(parse_one(script, dialect="oracle"))
```

produces error: OptimizeError: Column '"level"' could not be resolved

**Official Documentation**
There is an example with LEVEL here: https://docs.oracle.com/en/database/oracle/oracle-database/18/sqlrf/Hierarchical-Queries.html#GUID-E3D35EF7-33C3-4D88-81B3-00030C47AE56

Discussion:
@georgesittas: Leaving this here for future reference. This is valid Oracle SQL:

```sql
WITH
t1 AS (
  SELECT
    1 AS c1,
    1 AS c2,
    'Y' AS TOP_PARENT_INDICATOR,
    1 AS id
  FROM DUAL
),
t2 AS (
  SELECT
    1 AS c2,
    2 AS id
  FROM DUAL
)
SELECT t1.c1
FROM t1
LEFT JOIN t2 ON t1.c2 = t2.c2
WHERE (t1.TOP_PARENT_INDICATOR = 'Y' OR LEVEL = 1)
START WITH (t1.id IS NOT NULL)
CONNECT BY PRIOR t1.id = t2.id;
```

I think I made an incorrect assumption [here](https://github.com/tobymao/sqlglot/commit/946cd4234a2ca403785b7c6a026a39ef604e8754) that queries with `CONNECT BY` in Snowflake can't include `JOIN`s, hence why I returned `LEVEL` as a "pseudocolumn".

This is problematic for the above query though, because that column will appear in both `t1` and `t2` and so it will end up being ambiguous. Perhaps we need to treat it as a special identifier at the AST level, i.e. get rid of the `Column` node.

Note that the query above doesn't run in Snowflake (not sure why yet):

```
Error: invalid identifier 'LEVEL' (line 34)
```

Solution patch:
diff --git a/sqlglot/dialects/snowflake.py b/sqlglot/dialects/snowflake.py
index b52eb9dc..ac699944 100644
--- a/sqlglot/dialects/snowflake.py
+++ b/sqlglot/dialects/snowflake.py
@@ -357,8 +357,6 @@ class Snowflake(Dialect):
         "mi": "%M",
         "SS": "%S",
         "ss": "%S",
-        "FF": "%f",
-        "ff": "%f",
         "FF6": "%f",
         "ff6": "%f",
     }
diff --git a/sqlglot/dialects/trino.py b/sqlglot/dialects/trino.py
index fca96396..adfaa5a5 100644
--- a/sqlglot/dialects/trino.py
+++ b/sqlglot/dialects/trino.py
@@ -47,6 +47,7 @@ class Trino(Presto):
             **Presto.Generator.TRANSFORMS,
             exp.ArraySum: lambda self,
             e: f"REDUCE({self.sql(e, 'this')}, 0, (acc, x) -> acc + x, acc -> acc)",
+            exp.ArrayUniqueAgg: lambda self, e: f"ARRAY_AGG(DISTINCT {self.sql(e, 'this')})",
             exp.Merge: merge_without_target_sql,
             exp.TimeStrToTime: lambda self, e: timestrtotime_sql(self, e, include_precision=True),
             exp.Trim: trim_sql,
diff --git a/sqlglot/generator.py b/sqlglot/generator.py
index 773d93e5..3ceee215 100644
--- a/sqlglot/generator.py
+++ b/sqlglot/generator.py
@@ -2491,11 +2491,11 @@ class Generator(metaclass=_Generator):
         return csv(
             *sqls,
             *[self.sql(join) for join in expression.args.get("joins") or []],
-            self.sql(expression, "connect"),
             self.sql(expression, "match"),
             *[self.sql(lateral) for lateral in expression.args.get("laterals") or []],
             self.sql(expression, "prewhere"),
             self.sql(expression, "where"),
+            self.sql(expression, "connect"),
             self.sql(expression, "group"),
             self.sql(expression, "having"),
             *[gen(self, expression) for gen in self.AFTER_HAVING_MODIFIER_TRANSFORMS.values()],
diff --git a/sqlglot/optimizer/qualify_columns.py b/sqlglot/optimizer/qualify_columns.py
index 871bb365..73323290 100644
--- a/sqlglot/optimizer/qualify_columns.py
+++ b/sqlglot/optimizer/qualify_columns.py
@@ -56,7 +56,21 @@ def qualify_columns(
     dialect = Dialect.get_or_raise(schema.dialect)
     pseudocolumns = dialect.PSEUDOCOLUMNS
 
+    snowflake_or_oracle = dialect in ("oracle", "snowflake")
+
     for scope in traverse_scope(expression):
+        scope_expression = scope.expression
+        is_select = isinstance(scope_expression, exp.Select)
+
+        if is_select and snowflake_or_oracle and scope_expression.args.get("connect"):
+            # In Snowflake / Oracle queries that have a CONNECT BY clause, one can use the LEVEL
+            # pseudocolumn, which doesn't belong to a table, so we change it into an identifier
+            scope_expression.transform(
+                lambda n: n.this if isinstance(n, exp.Column) and n.name == "LEVEL" else n,
+                copy=False,
+            )
+            scope.clear_cache()
+
         resolver = Resolver(scope, schema, infer_schema=infer_schema)
         _pop_table_column_aliases(scope.ctes)
         _pop_table_column_aliases(scope.derived_tables)
@@ -76,7 +90,7 @@ def qualify_columns(
         if not schema.empty and expand_alias_refs:
             _expand_alias_refs(scope, resolver, dialect)
 
-        if isinstance(scope.expression, exp.Select):
+        if is_select:
             if expand_stars:
                 _expand_stars(
                     scope,
@@ -898,22 +912,10 @@ class Resolver:
                     for (name, alias) in itertools.zip_longest(columns, column_aliases)
                 ]
 
-            pseudocolumns = self._get_source_pseudocolumns(name)
-            if pseudocolumns:
-                columns = list(columns)
-                columns.extend(c for c in pseudocolumns if c not in columns)
-
             self._get_source_columns_cache[cache_key] = columns
 
         return self._get_source_columns_cache[cache_key]
 
-    def _get_source_pseudocolumns(self, name: str) -> t.Sequence[str]:
-        if self.schema.dialect == "snowflake" and self.scope.expression.args.get("connect"):
-            # When there is a CONNECT BY clause, there is only one table being scanned
-            # See: https://docs.snowflake.com/en/sql-reference/constructs/connect-by
-            return ["LEVEL"]
-        return []
-
     def _get_all_source_columns(self) -> t.Dict[str, t.Sequence[str]]:
         if self._source_columns is None:
             self._source_columns = {


Test patch:
diff --git a/tests/dialects/test_hive.py b/tests/dialects/test_hive.py
index e52ee72d..d6f96259 100644
--- a/tests/dialects/test_hive.py
+++ b/tests/dialects/test_hive.py
@@ -740,6 +740,7 @@ class TestHive(Validator):
                 "presto": "SET_AGG(x)",
                 "snowflake": "ARRAY_UNIQUE_AGG(x)",
                 "spark": "COLLECT_SET(x)",
+                "trino": "ARRAY_AGG(DISTINCT x)",
             },
         )
         self.validate_all(
diff --git a/tests/dialects/test_oracle.py b/tests/dialects/test_oracle.py
index 0abef504..cb968ee0 100644
--- a/tests/dialects/test_oracle.py
+++ b/tests/dialects/test_oracle.py
@@ -513,10 +513,10 @@ FROM JSON_TABLE(res, '$.info[*]' COLUMNS(
   LEVEL,
   SYS_CONNECT_BY_PATH(last_name, '/') AS "Path"
 FROM employees
-START WITH last_name = 'King'
-CONNECT BY PRIOR employee_id = manager_id AND LEVEL <= 4
 WHERE
-  level <= 3 AND department_id = 80"""
+  level <= 3 AND department_id = 80
+START WITH last_name = 'King'
+CONNECT BY PRIOR employee_id = manager_id AND LEVEL <= 4"""
 
         for query in (f"{body}{start}{connect}", f"{body}{connect}{start}"):
             self.validate_identity(query, pretty, pretty=True)
diff --git a/tests/dialects/test_snowflake.py b/tests/dialects/test_snowflake.py
index 5c0cc010..e1f43882 100644
--- a/tests/dialects/test_snowflake.py
+++ b/tests/dialects/test_snowflake.py
@@ -105,6 +105,9 @@ class TestSnowflake(Validator):
         self.validate_identity(
             "SELECT * FROM DATA AS DATA_L ASOF JOIN DATA AS DATA_R MATCH_CONDITION (DATA_L.VAL > DATA_R.VAL) ON DATA_L.ID = DATA_R.ID"
         )
+        self.validate_identity(
+            """SELECT TO_TIMESTAMP('2025-01-16T14:45:30.123+0500', 'yyyy-mm-DD"T"hh24:mi:ss.ff3TZHTZM')"""
+        )
         self.validate_identity(
             "WITH t (SELECT 1 AS c) SELECT c FROM t",
             "WITH t AS (SELECT 1 AS c) SELECT c FROM t",
@@ -294,6 +297,13 @@ class TestSnowflake(Validator):
             "SELECT * RENAME (a AS b), c AS d FROM xxx",
         )
 
+        self.validate_all(
+            "SELECT TO_TIMESTAMP('2025-01-16 14:45:30.123', 'yyyy-mm-DD hh24:mi:ss.ff6')",
+            write={
+                "": "SELECT STR_TO_TIME('2025-01-16 14:45:30.123', '%Y-%m-%d %H:%M:%S.%f')",
+                "snowflake": "SELECT TO_TIMESTAMP('2025-01-16 14:45:30.123', 'yyyy-mm-DD hh24:mi:ss.ff6')",
+            },
+        )
         self.validate_all(
             "ARRAY_CONSTRUCT_COMPACT(1, null, 2)",
             write={
diff --git a/tests/fixtures/optimizer/qualify_columns.sql b/tests/fixtures/optimizer/qualify_columns.sql
index 2be88388..96c84aa7 100644
--- a/tests/fixtures/optimizer/qualify_columns.sql
+++ b/tests/fixtures/optimizer/qualify_columns.sql
@@ -227,7 +227,32 @@ START WITH title = 'President'
 CONNECT BY manager_ID = PRIOR employee_id
 ORDER BY
   employee_ID NULLS LAST;
-WITH EMPLOYEES AS (SELECT T.TITLE AS TITLE, T.EMPLOYEE_ID AS EMPLOYEE_ID, T.MANAGER_ID AS MANAGER_ID FROM (VALUES ('President', 1, NULL), ('Vice President Engineering', 10, 1), ('Programmer', 100, 10), ('QA Engineer', 101, 10), ('Vice President HR', 20, 1), ('Health Insurance Analyst', 200, 20)) AS T(TITLE, EMPLOYEE_ID, MANAGER_ID)) SELECT EMPLOYEES.EMPLOYEE_ID AS EMPLOYEE_ID, EMPLOYEES.MANAGER_ID AS MANAGER_ID, EMPLOYEES.TITLE AS TITLE, EMPLOYEES.LEVEL AS LEVEL FROM EMPLOYEES AS EMPLOYEES START WITH EMPLOYEES.TITLE = 'President' CONNECT BY EMPLOYEES.MANAGER_ID = PRIOR EMPLOYEES.EMPLOYEE_ID ORDER BY EMPLOYEE_ID;
+WITH EMPLOYEES AS (SELECT T.TITLE AS TITLE, T.EMPLOYEE_ID AS EMPLOYEE_ID, T.MANAGER_ID AS MANAGER_ID FROM (VALUES ('President', 1, NULL), ('Vice President Engineering', 10, 1), ('Programmer', 100, 10), ('QA Engineer', 101, 10), ('Vice President HR', 20, 1), ('Health Insurance Analyst', 200, 20)) AS T(TITLE, EMPLOYEE_ID, MANAGER_ID)) SELECT EMPLOYEES.EMPLOYEE_ID AS EMPLOYEE_ID, EMPLOYEES.MANAGER_ID AS MANAGER_ID, EMPLOYEES.TITLE AS TITLE, LEVEL AS LEVEL FROM EMPLOYEES AS EMPLOYEES START WITH EMPLOYEES.TITLE = 'President' CONNECT BY EMPLOYEES.MANAGER_ID = PRIOR EMPLOYEES.EMPLOYEE_ID ORDER BY EMPLOYEE_ID;
+
+# execute: false
+# dialect: oracle
+WITH
+t1 AS (
+  SELECT
+    1 AS c1,
+    1 AS c2,
+    'Y' AS TOP_PARENT_INDICATOR,
+    1 AS id
+  FROM DUAL
+),
+t2 AS (
+  SELECT
+    1 AS c2,
+    2 AS id
+  FROM DUAL
+)
+SELECT t1.c1
+FROM t1
+LEFT JOIN t2 ON t1.c2 = t2.c2
+WHERE (t1.TOP_PARENT_INDICATOR = 'Y' OR LEVEL = 1)
+START WITH (t1.id IS NOT NULL)
+CONNECT BY PRIOR t1.id = t2.id;
+WITH T1 AS (SELECT 1 AS C1, 1 AS C2, 'Y' AS TOP_PARENT_INDICATOR, 1 AS ID FROM DUAL DUAL), T2 AS (SELECT 1 AS C2, 2 AS ID FROM DUAL DUAL) SELECT T1.C1 AS C1 FROM T1 T1 LEFT JOIN T2 T2 ON T1.C2 = T2.C2 WHERE (T1.TOP_PARENT_INDICATOR = 'Y' OR LEVEL = 1) START WITH (NOT T1.ID IS NULL) CONNECT BY PRIOR T1.ID = T2.ID;
 
 --------------------------------------
 -- Derived tables