chore(bigframes): update the datetime_ops and isin_join in the sqlglot compiler (#16568)

This update addresses issues identified during performance testing
against Ibis golden SQL.

- `datetime_ops`: Aligned with Ibis behavior by implementing safe
parsing instead of throwing errors.
- `isin_join`: Improved performance; benchmark details are available
here: [screenshot/vmb8ACvJhDNpccp].

Author: chelsea-lin

packages/bigframes/bigframes/core/compile/sqlglot/expressions/datetime_ops.py

@@ -366,10 +366,7 @@ def _(expr: TypedExpr, op: ops.ToDatetimeOp) -> sge.Expression:
         result = expr.expr
         if expr.dtype != dtypes.STRING_DTYPE:
             result = sge.Cast(this=result, to="STRING")
-        result = sge.func(
-            "PARSE_TIMESTAMP", sge.convert(op.format), result, sge.convert("UTC")
-        )
-        return sge.Cast(this=result, to="DATETIME")
+        return sge.TryCast(this=result, to="DATETIME")
 
     if expr.dtype in (
         dtypes.STRING_DTYPE,

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

@@ -364,34 +364,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/expressions/snapshots/test_datetime_ops/test_to_datetime/out.sql

@@ -3,5 +3,5 @@ SELECT
   SAFE_CAST(`string_col` AS DATETIME),
   CAST(TIMESTAMP_MICROS(CAST(TRUNC(`float64_col` * 0.001) AS INT64)) AS DATETIME) AS `float64_col`,
   SAFE_CAST(`timestamp_col` AS DATETIME),
-  CAST(PARSE_TIMESTAMP('%Y-%m-%d', `string_col`, 'UTC') AS DATETIME) AS `string_col_fmt`
+  SAFE_CAST(`string_col` AS DATETIME) AS `string_col_fmt`
 FROM `bigframes-dev`.`sqlglot_test`.`scalar_types` AS `bft_0`

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`
 )