Fix(bigquery): properly consume dashed table parts (#4477)

Author: georgesittas

sqlglot/dialects/bigquery.py

@@ -589,6 +589,8 @@ class Parser(parser.Parser):
 
         NULL_TOKENS = {TokenType.NULL, TokenType.UNKNOWN}
 
+        DASHED_TABLE_PART_FOLLOW_TOKENS = {TokenType.DOT, TokenType.L_PAREN, TokenType.R_PAREN}
+
         STATEMENT_PARSERS = {
             **parser.Parser.STATEMENT_PARSERS,
             TokenType.ELSE: lambda self: self._parse_as_command(self._prev),
@@ -615,11 +617,13 @@ def _parse_table_part(self, schema: bool = False) -> t.Optional[exp.Expression]:
             if isinstance(this, exp.Identifier):
                 table_name = this.name
                 while self._match(TokenType.DASH, advance=False) and self._next:
-                    text = ""
-                    while self._is_connected() and self._curr.token_type != TokenType.DOT:
+                    start = self._curr
+                    while self._is_connected() and not self._match_set(
+                        self.DASHED_TABLE_PART_FOLLOW_TOKENS, advance=False
+                    ):
                         self._advance()
-                        text += self._prev.text
-                    table_name += text
+
+                    table_name += self._find_sql(start, self._prev)
 
                 this = exp.Identifier(this=table_name, quoted=this.args.get("quoted"))
             elif isinstance(this, exp.Literal):

tests/dialects/test_bigquery.py

@@ -200,24 +200,7 @@ def test_bigquery(self):
         self.validate_identity("CAST(x AS NVARCHAR)", "CAST(x AS STRING)")
         self.validate_identity("CAST(x AS TIMESTAMPTZ)", "CAST(x AS TIMESTAMP)")
         self.validate_identity("CAST(x AS RECORD)", "CAST(x AS STRUCT)")
-        self.validate_all(
-            "EDIT_DISTANCE(col1, col2, max_distance => 3)",
-            write={
-                "bigquery": "EDIT_DISTANCE(col1, col2, max_distance => 3)",
-                "clickhouse": UnsupportedError,
-                "databricks": UnsupportedError,
-                "drill": UnsupportedError,
-                "duckdb": UnsupportedError,
-                "hive": UnsupportedError,
-                "postgres": "LEVENSHTEIN_LESS_EQUAL(col1, col2, 3)",
-                "presto": UnsupportedError,
-                "snowflake": "EDITDISTANCE(col1, col2, 3)",
-                "spark": UnsupportedError,
-                "spark2": UnsupportedError,
-                "sqlite": UnsupportedError,
-            },
-        )
-
+        self.validate_identity("SELECT * FROM x WHERE x.y >= (SELECT MAX(a) FROM b-c) - 20")
         self.validate_identity(
             "MERGE INTO dataset.NewArrivals USING (SELECT * FROM UNNEST([('microwave', 10, 'warehouse #1'), ('dryer', 30, 'warehouse #1'), ('oven', 20, 'warehouse #2')])) ON FALSE WHEN NOT MATCHED THEN INSERT ROW WHEN NOT MATCHED BY SOURCE THEN DELETE"
         )
@@ -332,6 +315,23 @@ def test_bigquery(self):
             "SELECT CAST(1 AS INT64)",
         )
 
+        self.validate_all(
+            "EDIT_DISTANCE(col1, col2, max_distance => 3)",
+            write={
+                "bigquery": "EDIT_DISTANCE(col1, col2, max_distance => 3)",
+                "clickhouse": UnsupportedError,
+                "databricks": UnsupportedError,
+                "drill": UnsupportedError,
+                "duckdb": UnsupportedError,
+                "hive": UnsupportedError,
+                "postgres": "LEVENSHTEIN_LESS_EQUAL(col1, col2, 3)",
+                "presto": UnsupportedError,
+                "snowflake": "EDITDISTANCE(col1, col2, 3)",
+                "spark": UnsupportedError,
+                "spark2": UnsupportedError,
+                "sqlite": UnsupportedError,
+            },
+        )
         self.validate_all(
             "EDIT_DISTANCE(a, b)",
             write={