-
Notifications
You must be signed in to change notification settings - Fork 119
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. Weβll occasionally send you account related emails.
Already on GitHub? Sign in to your account
feat: add DataFrame
and LazyFrame
explode
method
#1542
Changes from 3 commits
3061fe9
2326b08
32af22e
3b52ab5
c3bf009
b427e79
c77dc62
72314a2
7f04579
7be326e
5da1ad6
4a098b8
380a6cb
c7a47c9
864e932
cc72f6b
1156beb
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -11,6 +11,7 @@ | |
- columns | ||
- drop | ||
- drop_nulls | ||
- explode | ||
- filter | ||
- gather_every | ||
- get_column | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -10,6 +10,7 @@ | |
- columns | ||
- drop | ||
- drop_nulls | ||
- explode | ||
- filter | ||
- gather_every | ||
- group_by | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -19,6 +19,7 @@ | |
from narwhals.utils import Implementation | ||
from narwhals.utils import flatten | ||
from narwhals.utils import generate_temporary_column_name | ||
from narwhals.utils import import_dtypes_module | ||
from narwhals.utils import is_sequence_but_not_str | ||
from narwhals.utils import parse_columns_to_drop | ||
|
||
|
@@ -743,3 +744,78 @@ def unpivot( | |
) | ||
# TODO(Unassigned): Even with promote_options="permissive", pyarrow does not | ||
# upcast numeric to non-numeric (e.g. string) datatypes | ||
|
||
def explode(self: Self, columns: str | Sequence[str], *more_columns: str) -> Self: | ||
import pyarrow as pa | ||
import pyarrow.compute as pc | ||
|
||
from narwhals.exceptions import InvalidOperationError | ||
|
||
dtypes = import_dtypes_module(self._version) | ||
|
||
to_explode = ( | ||
[columns, *more_columns] | ||
if isinstance(columns, str) | ||
else [*columns, *more_columns] | ||
) | ||
|
||
schema = self.collect_schema() | ||
for col_to_explode in to_explode: | ||
dtype = schema[col_to_explode] | ||
|
||
if dtype != dtypes.List: | ||
msg = f"`explode` operation not supported for dtype `{dtype}`" | ||
raise InvalidOperationError(msg) | ||
|
||
native_frame = self._native_frame | ||
counts = pc.list_value_length(native_frame[to_explode[0]]) | ||
|
||
if not all( | ||
pc.all(pc.equal(pc.list_value_length(native_frame[col_name]), counts)).as_py() | ||
for col_name in to_explode[1:] | ||
): | ||
from narwhals.exceptions import ShapeError | ||
|
||
msg = "exploded columns must have matching element counts" | ||
raise ShapeError(msg) | ||
|
||
original_columns = self.columns | ||
other_columns = [c for c in original_columns if c not in to_explode] | ||
fast_path = pc.all(pc.greater_equal(counts, 1)).as_py() | ||
|
||
if fast_path: | ||
indices = pc.list_parent_indices(native_frame[to_explode[0]]) | ||
flatten_func = pc.list_flatten | ||
else: | ||
indices = pa.array( | ||
[ | ||
i | ||
for i, count in enumerate(counts.to_pylist()) | ||
for _ in range(max(count or 1, 1)) | ||
] | ||
) | ||
|
||
def explode_null_array(array: pa.ChunkedArray) -> pa.ChunkedArray: | ||
exploded_values = [] # type: ignore[var-annotated] | ||
for lst_element in array.to_pylist(): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. i might be missing something but this looks potentially very expensive There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It definitely is π happy to raise for nullable cases for now and try to figure out a native alternative There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I did some improvements ππΌ now there is only one |
||
if lst_element is None or len(lst_element) == 0: | ||
exploded_values.append(None) | ||
else: # Non-empty list) | ||
exploded_values.extend(lst_element) | ||
return pa.chunked_array([exploded_values]) | ||
|
||
flatten_func = explode_null_array | ||
|
||
arrays = [ | ||
native_frame[col_name].take(indices=indices) | ||
if col_name in other_columns | ||
else flatten_func(native_frame[col_name]) | ||
for col_name in original_columns | ||
] | ||
|
||
return self._from_native_frame( | ||
pa.Table.from_arrays( | ||
arrays=arrays, | ||
names=original_columns, | ||
) | ||
) |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -937,3 +937,52 @@ def unpivot( | |
value_name=value_name if value_name is not None else "value", | ||
) | ||
) | ||
|
||
def explode(self: Self, columns: str | Sequence[str], *more_columns: str) -> Self: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If a single column is to be exploded, then we use the pandas native method. If multiple columns, the strategy is to explode the one column with the rest of the dataframe, and the other series individually and finally concatenating them back, plus sorting by original column names order |
||
from narwhals.exceptions import InvalidOperationError | ||
|
||
dtypes = import_dtypes_module(self._version) | ||
|
||
to_explode = ( | ||
[columns, *more_columns] | ||
if isinstance(columns, str) | ||
else [*columns, *more_columns] | ||
) | ||
schema = self.collect_schema() | ||
for col_to_explode in to_explode: | ||
dtype = schema[col_to_explode] | ||
|
||
if dtype != dtypes.List: | ||
msg = f"`explode` operation not supported for dtype `{dtype}`" | ||
raise InvalidOperationError(msg) | ||
FBruzzesi marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
if len(to_explode) == 1: | ||
return self._from_native_frame(self._native_frame.explode(to_explode[0])) | ||
else: | ||
native_frame = self._native_frame | ||
anchor_series = native_frame[to_explode[0]].list.len() | ||
|
||
if not all( | ||
(native_frame[col_name].list.len() == anchor_series).all() | ||
for col_name in to_explode[1:] | ||
): | ||
from narwhals.exceptions import ShapeError | ||
|
||
msg = "exploded columns must have matching element counts" | ||
raise ShapeError(msg) | ||
|
||
original_columns = self.columns | ||
other_columns = [c for c in original_columns if c not in to_explode] | ||
|
||
exploded_frame = native_frame[[*other_columns, to_explode[0]]].explode( | ||
to_explode[0] | ||
) | ||
exploded_series = [ | ||
native_frame[col_name].explode().to_frame() for col_name in to_explode[1:] | ||
] | ||
|
||
plx = self.__native_namespace__() | ||
|
||
return self._from_native_frame( | ||
plx.concat([exploded_frame, *exploded_series], axis=1)[original_columns] | ||
) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
pyarrow has two paths:
pc.list_parent_indices
andpc.list_flatten
, which is a problem. This implementation falls back to a python list both to flatten the array(s) and to create the corresponding indices .After flattening, a new table is created by
take
-ing the indices of the non-flattened arrays and the flattened arrays.