chore: Migrate NaryRemoteFunctionOp operator to SQLGlot

Author: jialuoo

bigframes/core/compile/sqlglot/expressions/generic_ops.py

@@ -184,6 +184,17 @@ def _(
     return sge.func(func_name, left.expr, right.expr)
 
 
+@register_nary_op(ops.NaryRemoteFunctionOp, pass_op=True)
+def _(*operands: TypedExpr, op: ops.NaryRemoteFunctionOp) -> sge.Expression:
+    routine_ref = op.function_def.routine_ref
+    # Quote project, dataset, and routine IDs to avoid keyword clashes.
+    func_name = (
+        f"`{routine_ref.project}`.`{routine_ref.dataset_id}`.`{routine_ref.routine_id}`"
+    )
+
+    return sge.func(func_name, *(operand.expr for operand in operands))
+
+
 @register_nary_op(ops.case_when_op)
 def _(*cases_and_outputs: TypedExpr) -> sge.Expression:
     # Need to upcast BOOL to INT if any output is numeric

tests/unit/core/compile/sqlglot/expressions/snapshots/test_generic_ops/test_nary_remote_function_op/out.sql

@@ -0,0 +1,15 @@
+WITH `bfcte_0` AS (
+  SELECT
+    `float64_col`,
+    `int64_col`,
+    `string_col`
+  FROM `bigframes-dev`.`sqlglot_test`.`scalar_types`
+), `bfcte_1` AS (
+  SELECT
+    *,
+    `my_project`.`my_dataset`.`my_routine`(`int64_col`, `float64_col`, `string_col`) AS `bfcol_3`
+  FROM `bfcte_0`
+)
+SELECT
+  `bfcol_3` AS `int64_col`
+FROM `bfcte_1`

tests/unit/core/compile/sqlglot/expressions/test_generic_ops.py

@@ -242,6 +242,48 @@ def test_binary_remote_function_op(scalar_types_df: bpd.DataFrame, snapshot):
     snapshot.assert_match(sql, "out.sql")
 
 
+def test_nary_remote_function_op(scalar_types_df: bpd.DataFrame, snapshot):
+    from google.cloud import bigquery
+
+    from bigframes.functions import udf_def
+
+    bf_df = scalar_types_df[["int64_col", "float64_col", "string_col"]]
+    op = ops.NaryRemoteFunctionOp(
+        function_def=udf_def.BigqueryUdf(
+            routine_ref=bigquery.RoutineReference.from_string(
+                "my_project.my_dataset.my_routine"
+            ),
+            signature=udf_def.UdfSignature(
+                input_types=(
+                    udf_def.UdfField(
+                        "x",
+                        bigquery.StandardSqlDataType(
+                            type_kind=bigquery.StandardSqlTypeNames.INT64
+                        ),
+                    ),
+                    udf_def.UdfField(
+                        "y",
+                        bigquery.StandardSqlDataType(
+                            type_kind=bigquery.StandardSqlTypeNames.FLOAT64
+                        ),
+                    ),
+                    udf_def.UdfField(
+                        "z",
+                        bigquery.StandardSqlDataType(
+                            type_kind=bigquery.StandardSqlTypeNames.STRING
+                        ),
+                    ),
+                ),
+                output_bq_type=bigquery.StandardSqlDataType(
+                    type_kind=bigquery.StandardSqlTypeNames.FLOAT64
+                ),
+            ),
+        )
+    )
+    sql = utils._apply_nary_op(bf_df, op, "int64_col", "float64_col", "string_col")
+    snapshot.assert_match(sql, "out.sql")
+
+
 def test_case_when_op(scalar_types_df: bpd.DataFrame, snapshot):
     ops_map = {
         "single_case": ops.case_when_op.as_expr(