chore: pyarrow do not combine chunks (#1662)

* fix: pyarrow do not combine chunks

* duplicated no threads?

Author: FBruzzesi

Diff for: narwhals/_arrow/dataframe.py

@@ -597,7 +597,6 @@ def write_csv(self: Self, file: Any) -> Any:
         return pa_csv.write_csv(pa_table, file)
 
     def is_duplicated(self: Self) -> ArrowSeries:
-        import numpy as np  # ignore-banned-import
         import pyarrow as pa
         import pyarrow.compute as pc
 
@@ -608,14 +607,18 @@ def is_duplicated(self: Self) -> ArrowSeries:
         columns = self.columns
         col_token = generate_temporary_column_name(n_bytes=8, columns=columns)
         row_count = (
-            df.append_column(col_token, pa.array(np.arange(len(self))))
+            df.append_column(col_token, pa.repeat(pa.scalar(1), len(self)))
             .group_by(columns)
-            .aggregate([(col_token, "count")])
+            .aggregate([(col_token, "sum")])
         )
         is_duplicated = pc.greater(
             df.join(
-                row_count, keys=columns, right_keys=columns, join_type="inner"
-            ).column(f"{col_token}_count"),
+                row_count,
+                keys=columns,
+                right_keys=columns,
+                join_type="inner",
+                use_threads=False,
+            ).column(f"{col_token}_sum"),
             1,
         )
         return ArrowSeries(

Diff for: narwhals/_arrow/namespace.py

@@ -445,18 +445,18 @@ def __call__(self: Self, df: ArrowDataFrame) -> Sequence[ArrowSeries]:
         except TypeError:
             # `self._otherwise_value` is a scalar and can't be converted to an expression
             value_series = condition.__class__._from_iterable(
-                [self._then_value] * len(condition),
+                pa.repeat(pa.scalar(self._then_value), len(condition)),
                 name="literal",
                 backend_version=self._backend_version,
                 version=self._version,
             )
 
         value_series_native = value_series._native_series
-        condition_native = condition._native_series.combine_chunks()
+        condition_native = condition._native_series
 
         if self._otherwise_value is None:
-            otherwise_native = pa.array(
-                [None] * len(condition_native), type=value_series_native.type
+            otherwise_native = pa.repeat(
+                pa.scalar(None, type=value_series_native.type), len(condition_native)
             )
             return [
                 value_series._from_native_series(

Diff for: narwhals/_arrow/utils.py

@@ -261,7 +261,7 @@ def vertical_concat(dfs: list[pa.Table]) -> pa.Table:
 
     import pyarrow as pa
 
-    return pa.concat_tables(dfs).combine_chunks()
+    return pa.concat_tables(dfs)
 
 
 def diagonal_concat(dfs: list[pa.Table], backend_version: tuple[int, ...]) -> pa.Table:
@@ -276,7 +276,7 @@ def diagonal_concat(dfs: list[pa.Table], backend_version: tuple[int, ...]) -> pa
         if backend_version < (14, 0, 0)
         else {"promote_options": "default"}  # type: ignore[dict-item]
     )
-    return pa.concat_tables(dfs, **kwargs).combine_chunks()
+    return pa.concat_tables(dfs, **kwargs)
 
 
 def floordiv_compat(left: Any, right: Any) -> Any: