Fix: port sqlglot pipe syntax fix in _parse_select (#5373)

Author: georgesittas

sqlmesh/core/dialect.py

@@ -350,13 +350,15 @@ def _parse_select(
     parse_subquery_alias: bool = True,
     parse_set_operation: bool = True,
     consume_pipe: bool = True,
+    from_: t.Optional[exp.From] = None,
 ) -> t.Optional[exp.Expression]:
     select = self.__parse_select(  # type: ignore
         nested=nested,
         table=table,
         parse_subquery_alias=parse_subquery_alias,
         parse_set_operation=parse_set_operation,
         consume_pipe=consume_pipe,
+        from_=from_,
     )
 
     if (

tests/core/test_dialect.py

@@ -722,3 +722,11 @@ def test_sqlglot_extended_correctly(dialect: str) -> None:
 def test_connected_identifier():
     ast = d.parse_one("""SELECT ("x"at time zone 'utc')::timestamp as x""", "redshift")
     assert ast.sql("redshift") == """SELECT CAST(("x" AT TIME ZONE 'utc') AS TIMESTAMP) AS x"""
+
+
+def test_pipe_syntax():
+    ast = d.parse_one("SELECT * FROM (FROM t2 |> SELECT id)", "bigquery")
+    assert (
+        ast.sql("bigquery")
+        == "SELECT * FROM (WITH __tmp1 AS (SELECT id FROM t2) SELECT * FROM __tmp1)"
+    )