Factoring of columns in USING clause over SQL adapter

edb/pgsql/resolver/context.py

@@ -72,6 +72,13 @@ class Scope:
     # Common Table Expressions
     ctes: List[CTE] = field(default_factory=lambda: [])
 
+    # Pairs of columns of the same name that have been compared in a USING
+    # clause. This makes unqualified references to their name them un-ambagious,
+    # it resolves to the column in the left (first) table.
+    factored_columns: List[Tuple[str, Table, Table]] = field(
+        default_factory=lambda: []
+    )
+
 
 @dataclass(kw_only=True)
 class Table:

edb/pgsql/resolver/expr.py

@@ -268,6 +268,25 @@ def _lookup_column(
             (t, c) for t, c in matched_columns if t.precedence == max_precedence
         ]
 
+    # when ambiguous references have been used in USING clause,
+    # pick the column from the left table
+    if (
+        len(matched_columns) == 2
+        and matched_columns[0][1].name == matched_columns[1][1].name
+    ):
+        matched_name = matched_columns[0][1].name
+        matched_tables = [t for t, _c in matched_columns]
+
+        for c_name, t_left, t_right in ctx.scope.factored_columns:
+            if matched_name != c_name:
+                continue
+            if t_left not in matched_tables or t_right not in matched_tables:
+                continue
+
+            c_left = next(c for c in t_left.columns if c.name == matched_name)
+            matched_columns = [(t_left, c_left)]
+            break
+
     if len(matched_columns) > 1:
         potential_tables = ', '.join([t.name or '' for t, _ in matched_columns])
         raise errors.QueryError(

edb/pgsql/resolver/range_var.py

@@ -180,12 +180,19 @@ def _resolve_JoinExpr(
 
     if join.using_clause:
         for c in join.using_clause:
+            assert len(c.name) == 1
+            assert isinstance(c.name[-1], str)
+            c_name = c.name[-1]
+
             with ctx.child() as subctx:
                 subctx.scope.tables = [ltable]
                 l_expr = dispatch.resolve(c, ctx=subctx)
             with ctx.child() as subctx:
                 subctx.scope.tables = [rtable]
                 r_expr = dispatch.resolve(c, ctx=subctx)
+
+            ctx.scope.factored_columns.append((c_name, ltable, rtable))
+
             quals = pgastutils.extend_binop(
                 quals,
                 pgast.Expr(

tests/test_sql_query.py

@@ -791,6 +791,31 @@ async def test_sql_query_42(self):
         self.assertEqual(res, 'UPDATE 1')
         await tran.rollback()
 
+    async def test_sql_query_43(self):
+        # USING factoring
+
+        res = await self.squery_values(
+            '''
+            WITH
+                a(id) AS (SELECT 1 UNION SELECT 2),
+                b(id) AS (SELECT 1 UNION SELECT 3)
+            SELECT a.id, b.id, id
+            FROM a LEFT JOIN b USING (id);
+            '''
+        )
+        self.assertEqual(res, [[1, 1, 1], [2, None, 2]])
+
+        res = await self.squery_values(
+            '''
+            WITH
+                a(id, sub_id) AS (SELECT 1, 'a' UNION SELECT 2, 'b'),
+                b(id, sub_id) AS (SELECT 1, 'a' UNION SELECT 3, 'c')
+            SELECT a.id, a.sub_id, b.id, b.sub_id, id, sub_id
+            FROM a JOIN b USING (id, sub_id);
+            '''
+        )
+        self.assertEqual(res, [[1, 'a', 1, 'a', 1, 'a']])
+
     async def test_sql_query_introspection_00(self):
         dbname = self.con.dbname
         res = await self.squery_values(