chore(bigframes): update the isin_join sqlglot implementation for better performance

Author: chelsea-lin

packages/bigframes/bigframes/core/compile/sqlglot/sqlglot_ir.py

@@ -357,34 +357,25 @@ def isin_join(
                 or conditions[1].dtype == dtypes.FLOAT_DTYPE
             ):
                 force_float_domain = True
-            part1_id = sql.identifier("bfpart1")
-            part2_id = sql.identifier("bfpart2")
             left_expr1, left_expr2 = _value_to_non_null_identity(
                 conditions[0], force_float_domain
             )
-            left_as_struct = sge.Struct(
-                expressions=[
-                    sge.PropertyEQ(this=part1_id, expression=left_expr1),
-                    sge.PropertyEQ(this=part2_id, expression=left_expr2),
-                ]
-            )
             right_expr1, right_expr2 = _value_to_non_null_identity(
                 conditions[1], force_float_domain
             )
-            right_select = right.expr.select(
-                *[
-                    sge.Struct(
-                        expressions=[
-                            sge.PropertyEQ(this=part1_id, expression=right_expr1),
-                            sge.PropertyEQ(this=part2_id, expression=right_expr2),
-                        ]
-                    )
-                ],
-            )
 
-            new_column = sge.In(
-                this=left_as_struct,
-                expressions=[right_select.subquery()],
+            # Use EXISTS for better performance.
+            # We use COALESCE on both sides in the WHERE clause as requested.
+            new_column = sge.Exists(
+                this=sge.Select()
+                .select(sge.convert(1))
+                .from_(right.expr.as_from_item())
+                .where(
+                    sge.and_(
+                        sge.EQ(this=left_expr1, expression=right_expr1),
+                        sge.EQ(this=left_expr2, expression=right_expr2),
+                    )
+                )
             )
         else:
             new_column = sge.In(

packages/bigframes/tests/unit/core/compile/sqlglot/snapshots/test_compile_isin/test_compile_isin/out.sql

@@ -20,12 +20,13 @@ WITH `bfcte_0` AS (
 ), `bfcte_4` AS (
   SELECT
     *,
-    STRUCT(COALESCE(`bfcol_4`, 0) AS `bfpart1`, COALESCE(`bfcol_4`, 1) AS `bfpart2`) IN (
-      (
-        SELECT
-          STRUCT(COALESCE(`bfcol_0`, 0) AS `bfpart1`, COALESCE(`bfcol_0`, 1) AS `bfpart2`)
-        FROM `bfcte_3`
-      )
+    EXISTS(
+      SELECT
+        1
+      FROM `bfcte_3`
+      WHERE
+        COALESCE(`bfcol_4`, 0) = COALESCE(`bfcol_0`, 0)
+        AND COALESCE(`bfcol_4`, 1) = COALESCE(`bfcol_0`, 1)
     ) AS `bfcol_5`
   FROM `bfcte_1`
 )