feat(duckDB): Cast inputs (BLOB → VARCHAR) for duckDB STARTS_WITH

sqlglot/dialects/duckdb.py

@@ -1209,6 +1209,19 @@ def objectinsert_sql(self, expression: exp.ObjectInsert) -> str:
 
             return self.func("STRUCT_INSERT", this, kv_sql)
 
+        def _prepare_startswith_arg(self, arg: exp.Expression) -> None:
+            """Prepare argument for STARTS_WITH by converting to VARCHAR."""
+            # Cast non-VARCHAR types to VARCHAR (includes double-cast for BLOB types)
+            if arg.type and not arg.is_type(exp.DataType.Type.VARCHAR, exp.DataType.Type.UNKNOWN):
+                arg.replace(exp.cast(arg, exp.DataType.Type.VARCHAR))
+
+        def startswith_sql(self, expression: exp.StartsWith) -> str:
+            # Prepare both arguments for STARTS_WITH (converts to VARCHAR)
+            self._prepare_startswith_arg(expression.this)
+            self._prepare_startswith_arg(expression.expression)
+
+            return self.func("STARTS_WITH", expression.this, expression.expression)
+
         def unnest_sql(self, expression: exp.Unnest) -> str:
             explode_array = expression.args.get("explode_array")
             if explode_array:

tests/dialects/test_bigquery.py

@@ -1205,6 +1205,21 @@ def test_bigquery(self):
                 "spark": "CAST(a AS BINARY)",
             },
         )
+        # Test STARTS_WITH with BYTES/BLOB handling from BigQuery to DuckDB
+        # Requires type annotation for proper BLOB -> VARCHAR casting
+        expr = self.parse_one("STARTS_WITH(CAST('foo' AS BYTES), CAST('f' AS BYTES))")
+        annotated = annotate_types(expr, dialect="bigquery")
+        self.assertEqual(
+            annotated.sql("duckdb"),
+            "STARTS_WITH(CAST(CAST('foo' AS BLOB) AS TEXT), CAST(CAST('f' AS BLOB) AS TEXT))",
+        )
+
+        expr = self.parse_one("STARTS_WITH(CAST('foo' AS BYTES), b'f')")
+        annotated = annotate_types(expr, dialect="bigquery")
+        self.assertEqual(
+            annotated.sql("duckdb"),
+            "STARTS_WITH(CAST(CAST('foo' AS BLOB) AS TEXT), CAST(CAST(e'f' AS BLOB) AS TEXT))",
+        )
         self.validate_all(
             "CAST(a AS NUMERIC)",
             write={