Skip to content

Commit

Permalink
fix series tests
Browse files Browse the repository at this point in the history
Signed-off-by: Dmitry Chigarev <[email protected]>
  • Loading branch information
dchigarev committed Apr 2, 2024
1 parent 1d227b5 commit cc0a66b
Show file tree
Hide file tree
Showing 2 changed files with 14 additions and 15 deletions.
13 changes: 3 additions & 10 deletions modin/core/dataframe/pandas/dataframe/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -630,22 +630,15 @@ def split_partitions_using_pivots_for_sort(
**kwargs: dict,
) -> "tuple[pandas.DataFrame, ...]": # noqa
def add_attr(df, timestamp):
if "dummy_timestamp" in df.attrs:
df.attrs["dummy_timestamp"] = (*df.attrs["dummy_timestamp"], timestamp)
if "bin_bounds" in df.attrs:
df.attrs["bin_bounds"] = (*df.attrs["bin_bounds"], timestamp)
else:
df.attrs["dummy_timestamp"] = (timestamp,)
df.attrs["bin_bounds"] = (timestamp,)
return df

result = ShuffleSortFunctions.split_partitions_using_pivots_for_sort(
df, columns_info, ascending, **kwargs
)
lvl = columns_info[0].name
if isinstance(lvl, int):
to_drop = [i for i in range(result[0].index.nlevels) if i != lvl]
else:
to_drop = [i for i in result[0].index.names if i != lvl]
for rs in result:
rs.index = rs.index.droplevel(to_drop)
for i, pivot in enumerate(columns_info[0].pivots):
add_attr(result[i], pivot - pandas.Timedelta(1, unit="ns"))
if i + 1 <= len(result):
Expand Down
16 changes: 11 additions & 5 deletions modin/core/storage_formats/pandas/query_compiler.py
Original file line number Diff line number Diff line change
Expand Up @@ -1066,15 +1066,21 @@ def map_func(df, resample_kwargs=resample_kwargs): # pragma: no cover
if resample_kwargs["on"] is not None:
df = df.set_index(resample_kwargs["on"])
return df
if "dummy_timestamp" in df.attrs:
timestamps = df.attrs["dummy_timestamp"]
if "bin_bounds" in df.attrs:
timestamps = df.attrs["bin_bounds"]
if isinstance(df.index, pandas.MultiIndex):
level_to_keep = resample_kwargs["level"]
if isinstance(level_to_keep, int):
to_drop = [lvl for lvl in range(df.index.nlevels) if lvl != level_to_keep]
else:
to_drop = [lvl for lvl in df.index.names if lvl != level_to_keep]
df.index = df.index.droplevel(to_drop)
resample_kwargs = resample_kwargs.copy()
resample_kwargs["level"] = None
for ts in timestamps:
df.loc[ts] = np.NaN
if df_op is not None:
df = df_op(df)
if use_range_impl:
resample_kwargs = resample_kwargs.copy()
resample_kwargs["level"] = None
resampled_val = df.resample(**resample_kwargs)
op = getattr(pandas.core.resample.Resampler, func_name)
if callable(op):
Expand Down

0 comments on commit cc0a66b

Please sign in to comment.