fix: avoid creating aliases for already-known tables (#361)

Closes #353.

sqlalchemy_bigquery/base.py

@@ -50,6 +50,7 @@
 from sqlalchemy.engine.default import DefaultDialect, DefaultExecutionContext
 from sqlalchemy.engine.base import Engine
 from sqlalchemy.sql.schema import Column
+from sqlalchemy.sql.schema import Table
 from sqlalchemy.sql import elements, selectable
 import re
 
@@ -289,12 +290,18 @@ def visit_column(
             if isinstance(tablename, elements._truncated_label):
                 tablename = self._truncated_identifier("alias", tablename)
             elif TABLE_VALUED_ALIAS_ALIASES in kwargs:
-                aliases = kwargs[TABLE_VALUED_ALIAS_ALIASES]
-                if tablename not in aliases:
-                    aliases[tablename] = self.anon_map[
-                        f"{TABLE_VALUED_ALIAS_ALIASES} {tablename}"
-                    ]
-                tablename = aliases[tablename]
+                known_tables = set(
+                    from_.name
+                    for from_ in self.compile_state.froms
+                    if isinstance(from_, Table)
+                )
+                if tablename not in known_tables:
+                    aliases = kwargs[TABLE_VALUED_ALIAS_ALIASES]
+                    if tablename not in aliases:
+                        aliases[tablename] = self.anon_map[
+                            f"{TABLE_VALUED_ALIAS_ALIASES} {tablename}"
+                        ]
+                    tablename = aliases[tablename]
 
             return self.preparer.quote(tablename) + "." + name
 

tests/unit/test_compiler.py

@@ -21,6 +21,7 @@
 import sqlalchemy.exc
 
 from conftest import setup_table
+from conftest import sqlalchemy_1_4_or_higher
 
 
 def test_constraints_are_ignored(faux_conn, metadata):
@@ -53,3 +54,25 @@ def test_cant_compile_unnamed_column(faux_conn, metadata):
         match="Cannot compile Column object until its 'name' is assigned.",
     ):
         sqlalchemy.Column(sqlalchemy.Integer).compile(faux_conn)
+
+
+@sqlalchemy_1_4_or_higher
+def test_no_alias_for_known_tables(faux_conn, metadata):
+    # See: https://github.com/googleapis/python-bigquery-sqlalchemy/issues/353
+    table = setup_table(
+        faux_conn,
+        "table1",
+        metadata,
+        sqlalchemy.Column("foo", sqlalchemy.Integer),
+        sqlalchemy.Column("bar", sqlalchemy.ARRAY(sqlalchemy.Integer)),
+    )
+    F = sqlalchemy.func
+    q = sqlalchemy.select(table.c.foo).where(F.unnest(table.c.bar).column_valued() == 1)
+
+    expected_sql = (
+        "SELECT `table1`.`foo` \n"
+        "FROM `table1`, unnest(`table1`.`bar`) AS `anon_1` \n"
+        "WHERE `anon_1` = %(param_1:INT64)s"
+    )
+    found_sql = q.compile(faux_conn).string
+    assert found_sql == expected_sql