fix dialect passing

Author: TrevorBergeron

packages/bigframes/bigframes/core/compile/lowering.py

@@ -301,7 +301,9 @@ def lower(self, expr: expression.OpExpression) -> expression.Expression:
         if self.dialect == "polars":
             return _lower_cast_to_polars(expr.op, expr.inputs[0])
         else:
-            return _lower_cast_to_substrait(expr.op, expr.inputs[0], dialect=self.dialect)
+            return _lower_cast_to_substrait(
+                expr.op, expr.inputs[0], dialect=self.dialect
+            )
 
 
 def invert_bytes(byte_string):
@@ -476,7 +478,9 @@ def _lower_cast_to_polars(cast_op: ops.AsTypeOp, arg: expression.Expression):
 def _lower_cast_to_substrait(
     cast_op: ops.AsTypeOp,
     arg: expression.Expression,
-    dialect: Literal["substrait-datafusion", "substrait-acero"] = "substrait-datafusion",
+    dialect: Literal[
+        "substrait-datafusion", "substrait-acero"
+    ] = "substrait-datafusion",
 ):
     if arg.output_type == dtypes.BOOL_DTYPE and cast_op.to_type == dtypes.STRING_DTYPE:
         is_true_cond = ops.eq_op.as_expr(arg, expression.const(True))
@@ -504,7 +508,9 @@ def _lower_cast_to_substrait(
         elif arg.output_type == dtypes.TIMESTAMP_DTYPE:
             cast_expr = cast_op.as_expr(arg)
             if dialect == "substrait-datafusion":
-                replaced_t = string_ops.ReplaceStrOp(pat="T", repl=" ").as_expr(cast_expr)
+                replaced_t = string_ops.ReplaceStrOp(pat="T", repl=" ").as_expr(
+                    cast_expr
+                )
                 return string_ops.ReplaceStrOp(pat="Z", repl="+00").as_expr(replaced_t)
             else:
                 # Acero: native cast (excluded in test)
@@ -513,14 +519,16 @@ def _lower_cast_to_substrait(
     return cast_op.as_expr(arg)
 
 
-class LowerEqNullsMatchRule(op_lowering.OpLoweringRule):
+class SubstraitLowerEqNullsMatchRule(op_lowering.OpLoweringRule):
     @property
     def op(self) -> type[ops.ScalarOp]:
         return comparison_ops.EqNullsMatchOp
 
     def lower(self, expr: expression.OpExpression) -> expression.Expression:
         assert isinstance(expr.op, comparison_ops.EqNullsMatchOp)
-        arg1, arg2 = _coerce_comparables(expr.children[0], expr.children[1])
+        arg1, arg2 = _coerce_comparables(
+            expr.children[0], expr.children[1], dialect="substrait"
+        )
 
         # True constant
         true_const = expression.const(True)
@@ -594,16 +602,19 @@ def lower(self, expr: expression.OpExpression) -> expression.Expression:
     LowerFloorOp(),
 )
 
+
 def lower_ops_to_polars(root: bigframe_node.BigFrameNode) -> bigframe_node.BigFrameNode:
     return op_lowering.lower_ops(root, rules=POLARS_LOWERING_RULES)
 
 
 def lower_ops_to_substrait(
     root: bigframe_node.BigFrameNode,
-    dialect: Literal["substrait-datafusion", "substrait-acero"] = "substrait-datafusion",
+    dialect: Literal[
+        "substrait-datafusion", "substrait-acero"
+    ] = "substrait-datafusion",
 ) -> bigframe_node.BigFrameNode:
     rules = (
-        LowerEqNullsMatchRule(),
+        SubstraitLowerEqNullsMatchRule(),
         *SUBSTRAIT_LOWER_COMPARISONS,
         LowerAsTypeRule(dialect=dialect),
     )

packages/bigframes/bigframes/core/compile/polars/compiler.py

@@ -766,7 +766,9 @@ def compile_join(self, node: nodes.JoinNode):
             for left_ex, right_ex in node.conditions:
                 from bigframes.core.compile import lowering
 
-                left_ex, right_ex = lowering._coerce_comparables(left_ex, right_ex)
+                left_ex, right_ex = lowering._coerce_comparables(
+                    left_ex, right_ex, dialect="polars"
+                )
                 left_on.append(self.expr_compiler.compile_expression(left_ex))
                 right_on.append(self.expr_compiler.compile_expression(right_ex))
 
@@ -787,7 +789,9 @@ def compile_isin(self, node: nodes.InNode):
             right_col = ex.ResolvedDerefOp.from_field(node.right_child.fields[0])
             from bigframes.core.compile import lowering
 
-            left_ex, right_ex = lowering._coerce_comparables(node.left_col, right_col)
+            left_ex, right_ex = lowering._coerce_comparables(
+                node.left_col, right_col, dialect="polars"
+            )
 
             left_pl_ex = self.expr_compiler.compile_expression(left_ex)
             right_pl_ex = self.expr_compiler.compile_expression(right_ex)

packages/bigframes/bigframes/core/compile/substrait/compiler.py

@@ -29,8 +29,8 @@
 import bigframes.operations.comparison_ops as comparison_ops
 import bigframes.operations.generic_ops as generic_ops
 import bigframes.operations.numeric_ops as numeric_ops
-import bigframes.operations.struct_ops as struct_ops
 import bigframes.operations.string_ops as string_ops
+import bigframes.operations.struct_ops as struct_ops
 from bigframes.core import agg_expressions, bigframe_node, nodes, rewrite
 from bigframes.core.compile import lowering
 
@@ -44,7 +44,9 @@ def __init__(
         self,
         duration_type: Literal["interval_day", "int"],
         use_precision_types: bool = True,
-        dialect: Literal["substrait-datafusion", "substrait-acero"] = "substrait-datafusion",
+        dialect: Literal[
+            "substrait-datafusion", "substrait-acero"
+        ] = "substrait-datafusion",
     ):
         self._duration_type = duration_type
         self._use_precision_types = use_precision_types
@@ -911,18 +913,20 @@ def _compile_replace(
         replacement: str,
     ) -> algebra_pb2.Expression:
         pb_expr = algebra_pb2.Expression()
-        pb_expr.scalar_function.function_reference = 76  # "replace" or "replace_substring"
-        
+        pb_expr.scalar_function.function_reference = (
+            76  # "replace" or "replace_substring"
+        )
+
         pb_expr.scalar_function.arguments.add().value.CopyFrom(str_expr)
-        
+
         search_expr = algebra_pb2.Expression()
         search_expr.literal.string = search
         pb_expr.scalar_function.arguments.add().value.CopyFrom(search_expr)
-        
+
         replace_expr = algebra_pb2.Expression()
         replace_expr.literal.string = replacement
         pb_expr.scalar_function.arguments.add().value.CopyFrom(replace_expr)
-        
+
         return pb_expr
 
     @_compile_op.register(struct_ops.StructOp)

packages/bigframes/tests/system/small/engines/test_generic_ops.py

@@ -183,7 +183,11 @@ def test_engines_astype_string(scalars_array_value: array_value.ArrayValue, engi
     # Acero's Substrait consumer lacks support for string functions like replace/replace_substring
     # and precision_time, so we cannot format time_col and timestamp_col inside the Substrait plan.
     from bigframes.session.substrait_executor import SubstraitExecutor
-    if isinstance(engine, SubstraitExecutor) and not engine._compiler._use_precision_types:
+
+    if (
+        isinstance(engine, SubstraitExecutor)
+        and not engine._compiler._use_precision_types
+    ):
         excluded_cols.extend(["time_col", "timestamp_col"])
 
     arr = apply_op(