fix a compatible issue where the sg.union cannot support multiple expressions in the older version of sqlglot

Author: chelsea-lin

bigframes/core/compile/sqlglot/sqlglot_ir.py

@@ -175,7 +175,7 @@ def from_union(
         ), f"At least two select expressions must be provided, but got {selects}."
 
         existing_ctes: list[sge.CTE] = []
-        union_selects: list[sge.Select] = []
+        union_selects: list[sge.Expression] = []
         for select in selects:
             assert isinstance(
                 select, sge.Select
@@ -204,10 +204,14 @@ def from_union(
                 sge.Select().select(*selections).from_(sge.Table(this=new_cte_name))
             )
 
-        union_expr = sg.union(
-            *union_selects,
-            distinct=False,
-            copy=False,
+        union_expr = typing.cast(
+            sge.Select,
+            functools.reduce(
+                lambda x, y: sge.Union(
+                    this=x, expression=y, distinct=False, copy=False
+                ),
+                union_selects,
+            ),
         )
         final_select_expr = sge.Select().select(sge.Star()).from_(union_expr.subquery())
         final_select_expr.set("with", sge.With(expressions=existing_ctes))

tests/unit/core/compile/sqlglot/snapshots/test_compile_concat/test_compile_concat_filter_sorted/out.sql

@@ -134,8 +134,8 @@ WITH `bfcte_3` AS (
   )
 )
 SELECT
-  `bfcol_42` AS `bfuid_col_12`,
-  `bfcol_43` AS `bfuid_col_13`
+  `bfcol_42` AS `float64_col`,
+  `bfcol_43` AS `int64_col`
 FROM `bfcte_18`
 ORDER BY
   `bfcol_44` ASC NULLS LAST,

tests/unit/core/compile/sqlglot/test_compile_concat.py

@@ -28,6 +28,7 @@ def test_compile_concat(scalar_types_df: bpd.DataFrame, snapshot):
 
 
 def test_compile_concat_filter_sorted(scalar_types_df: bpd.DataFrame, snapshot):
+
     scalars_array_value = scalar_types_df._block.expr
     input_1 = scalars_array_value.select_columns(["float64_col", "int64_col"]).order_by(
         [ordering.ascending_over("int64_col")]
@@ -37,5 +38,12 @@ def test_compile_concat_filter_sorted(scalar_types_df: bpd.DataFrame, snapshot):
     )
 
     result = input_1.concat([input_2, input_1, input_2])
+
+    new_names = ["float64_col", "int64_col"]
+    col_ids = {
+        old_name: new_name for old_name, new_name in zip(result.column_ids, new_names)
+    }
+    result = result.rename_columns(col_ids).select_columns(new_names)
+
     sql = result.session._executor.to_sql(result, enable_cache=False)
     snapshot.assert_match(sql, "out.sql")