ronantakizawa/codereview-bench huggingface dataset:
Repository: dask/dask
File: dask/dataframe/core.py (Python)

Code before:
    dfs = [
        from_pandas(df, 1)
        if (is_series_like(df) or is_dataframe_like(df)) and not is_dask_collection(df)
        else df
        for df in dfs
    ]
    return dfs


def hash_shard(
    df, nparts, split_out_setup=None, split_out_setup_kwargs=None, ignore_index=False
):
    if split_out_setup:
        h = split_out_setup(df, **(split_out_setup_kwargs or {}))
    else:
        h = df

    h = hash_object_dispatch(h, index=False)
    return group_split_dispatch(df, h % nparts, nparts, ignore_index=ignore_index)


def split_evenly(df, k):
    """Split dataframe into k roughly equal parts"""
    divisions = np.linspace(0, len(df), k + 1).astype(int)
    return {i: df.iloc[divisions[i] : divisions[i + 1]] for i in range(k)}


def split_out_on_index(df):
    h = df.index
    if isinstance(h, pd.MultiIndex):
        h = meta_frame_constructor(df)([], index=h).reset_index()
    return h


def split_out_on_cols(df, cols=None):
    return df[cols]


@insert_meta_param_description
def apply_concat_apply(
    args,
    chunk=None,
    aggregate=None,
    combine=None,
    meta=no_default,
    token=None,
    chunk_kwargs=None,
    aggregate_kwargs=None,
    combine_kwargs=None,
    split_every=None,
    split_out=None,

Review comment (suggestion):
Changed from None to align to pandas behaviour.
Besides deprecated reductions, this applies to numpy ufuncs.

Code after:
        else:
            out = None

    # Notice, we use .__class__ as opposed to type() in order to support
    # object proxies see <https://github.com/dask/dask/pull/6981>
    if out is not None and out.__class__ != result.__class__:
        raise TypeError(
            "Mismatched types between result and out parameter. "
            "out=%s, result=%s" % (str(type(out)), str(type(result)))
        )

    if isinstance(out, DataFrame):
        if len(out.columns) != len(result.columns):
            raise ValueError(
                "Mismatched columns count between result and out parameter. "
                "out=%s, result=%s" % (str(len(out.columns)), str(len(result.columns)))
            )

    if isinstance(out, (Series, DataFrame, Scalar)):
        out._meta = result._meta
        out._name = result._name
        out.dask = result.dask

        if not isinstance(out, Scalar):
            out._divisions = result.divisions
        return result
    elif out is not None:
        msg = (
            "The out parameter is not fully supported."
            " Received type %s, expected %s "
            % (
                typename(type(out)),
                typename(type(result)),
            )
        )
        raise NotImplementedError(msg)
    else:
        return result


def _maybe_from_pandas(dfs):
    from dask.dataframe.io import from_pandas

    dfs = [
        from_pandas(df, 1)
        if (is_series_like(df) or is_dataframe_like(df)) and not is_dask_collection(df)
        else df
        for df in dfs
    ]
    return dfs
Repository: dask/dask
File: dask/dataframe/__init__.py (Python)

Code before:
            from dask.dataframe.io import read_parquet, to_parquet
        except ImportError:
            pass
        try:
            from dask.dataframe.io import read_orc, to_orc
        except ImportError:
            pass
        try:
            from dask.dataframe.core import isna
        except ImportError:
            pass
    except ImportError as e:
        msg = (
            "Dask dataframe requirements are not installed.\n\n"
            "Please either conda or pip install as follows:\n\n"
            "  conda install dask                     # either conda install\n"
            '  python -m pip install "dask[dataframe]" --upgrade  # or python -m pip install'
        )
        raise ImportError(msg) from e
    warnings.warn(
        """The current Dask DataFrame implementation is deprecated. 
Dask will switch to a new implementation that contains a logical query planning 
layer and many other improvements in a future version. The new user-facing API is
compatible with the existing API.

The new implementation can be enabled with

pip install dask-expr

and through setting the configuration option

dask.config.set({'dataframe.query-planning': True})

The API-Docs can be found at https://docs.dask.org/en/stable/dask-expr-api.html
""",
        DeprecationWarning,
        stacklevel=2,
    )


from dask.dataframe._testing import test_dataframe


Review comment (suggestion):

```suggestion
The new implementation is already available and you can opt-in with
```

Code after:
            from dask.dataframe.io import read_parquet, to_parquet
        except ImportError:
            pass
        try:
            from dask.dataframe.io import read_orc, to_orc
        except ImportError:
            pass
        try:
            from dask.dataframe.core import isna
        except ImportError:
            pass
    except ImportError as e:
        msg = (
            "Dask dataframe requirements are not installed.\n\n"
            "Please either conda or pip install as follows:\n\n"
            "  conda install dask                     # either conda install\n"
            '  python -m pip install "dask[dataframe]" --upgrade  # or python -m pip install'
        )
        raise ImportError(msg) from e
    warnings.warn(
        """The current Dask DataFrame implementation is deprecated. 
Dask will switch to a new implementation that contains a logical query planning 
layer and many other improvements in a future version. The new user-facing API is
compatible with the existing API.

The new implementation is already available and you can opt-in with

pip install dask-expr

and through setting the configuration option

dask.config.set({'dataframe.query-planning': True})

The API-Docs can be found at https://docs.dask.org/en/stable/dask-expr-api.html
""",
        DeprecationWarning,
        stacklevel=2,
    )


from dask.dataframe._testing import test_dataframe
Repository: dask/dask
File: dask/base.py (Python)

Code before:
    def normalize_categorical_dtype(dtype):
        return [normalize_token(dtype.categories), normalize_token(dtype.ordered)]

    @normalize_token.register(pd.api.extensions.ExtensionDtype)
    def normalize_period_dtype(dtype):
        return normalize_token(dtype.name)

    @normalize_token.register(type(pd.NA))
    def normalize_na(na):
        return pd.NA

    @normalize_token.register(pd.offsets.BaseOffset)
    def normalize_offset(offset):
        return offset.freqstr


@normalize_token.register_lazy("pyarrow")
def register_pyarrow():
    import pyarrow as pa

    @normalize_token.register(pa.DataType)
    def normalize_datatype(dt):
        return pickle.dumps(dt, protocol=4)


@normalize_token.register_lazy("numpy")
def register_numpy():
    import numpy as np

    @normalize_token.register(np.generic)
    @normalize_token.register(np.ndarray)
    def normalize_array(x):
        if not x.shape:
            # Note: this path is used for both scalar and empty arrays
            return (x.item(), x.dtype)

        if x.size == 0:
            return (None, x.dtype, x.shape)

        if x.dtype.hasobject:
            try:
                try:
                    # string fast-path
                    data = hash_buffer_hex(
                        "-".join(x.flat).encode(
                            encoding="utf-8", errors="surrogatepass"
                        )
                    )
                except UnicodeDecodeError:
                    # bytes fast-path
                    data = hash_buffer_hex(b"-".join(x.flat))

Review comment (suggestion):
This feels like an unrelated change. While the functionality is similar, the raised exception is a little different

Code after:
@normalize_token.register_lazy("numpy")
def register_numpy():
    import numpy as np

    @normalize_token.register(np.ufunc)
    def normalize_ufunc(func):
        try:
            return _normalize_pickle(func)
        except Exception:
            _maybe_raise_nondeterministic(
                f"Cannot tokenize numpy ufunc {func!r}. Please use functions "
                "of the dask.array.ufunc module instead. See also "
                "https://docs.dask.org/en/latest/array-numpy-compatibility.html"
            )
            return uuid.uuid4().hex


def _colorize(t):
    """Convert (r, g, b) triple to "#RRGGBB" string

    For use with ``visualize(color=...)``

    Examples
    --------
    >>> _colorize((255, 255, 255))
    '#FFFFFF'
    >>> _colorize((0, 32, 128))
    '#002080'
    """
    t = t[:3]
    i = sum(v * 256 ** (len(t) - i - 1) for i, v in enumerate(t))
    h = hex(int(i))[2:].upper()
    h = "0" * (6 - len(h)) + h
    return "#" + h


named_schedulers: dict[str, SchedulerGetCallable] = {
    "sync": local.get_sync,
    "synchronous": local.get_sync,
    "single-threaded": local.get_sync,
}

if not EMSCRIPTEN:
    from dask import threaded

    named_schedulers.update(
        {
            "threads": threaded.get,
            "threading": threaded.get,
        }
    )
Repository: dask/dask
File: dask/cli.py (Python)

Code before:
            click.style(
                """Config key not specified. Are you looking for "dask config list"?"""
            ),
            err=True,
        )
        exit(1)
    else:
        value = dask.config.interpret_value(value)
        _, path = save_config({key: value})
        click.echo(f"Updated [{key}] to [{value}], config saved to {path}")


def save_config(new_config: dict) -> tuple[dict, pathlib.Path]:
    """
    Save new config values to dask config file,
    return new config and path to file.
    """
    config_file = pathlib.Path(dask.config.PATH).joinpath("dask.yaml")
    config_file.parent.mkdir(exist_ok=True, parents=True)
    if config_file.exists():
        with open(str(config_file)) as f:
            config = yaml.safe_load(f) or dict()
    else:
        config = dict()

    config = dask.config.set(new_config, config=config).config

    try:
        config_file.write_text(yaml.dump(config))
    except OSError as e:
        raise RuntimeError(
            f"""

For some reason we couldn't write config to {config_file}.
Perhaps you don't have permissions here?

You can change the directory where Dask writes config files
to somewhere else using the DASK_CONFIG environment variable.

For example, you could set the following:

    export DASK_CONFIG=~/.dask
"""
        ) from e

    return config, config_file


@config.command(name="list")
def config_list():
    """Print the whole config."""

Review comment (suggestion):
```suggestion
    dask.config.set(new_config, config=config)
```

Code after:
            click.style(
                """Config key not specified. Are you looking for "dask config list"?"""
            ),
            err=True,
        )
        exit(1)
    else:
        value = dask.config.interpret_value(value)
        _, path = save_config({key: value})
        click.echo(f"Updated [{key}] to [{value}], config saved to {path}")


def save_config(new_config: dict) -> tuple[dict, pathlib.Path]:
    """
    Save new config values to dask config file,
    return new config and path to file.
    """
    config_file = pathlib.Path(dask.config.PATH).joinpath("dask.yaml")
    config_file.parent.mkdir(exist_ok=True, parents=True)
    if config_file.exists():
        with open(str(config_file)) as f:
            config = yaml.safe_load(f) or dict()
    else:
        config = dict()

    dask.config.set(new_config, config=config)

    try:
        config_file.write_text(yaml.dump(config))
    except OSError as e:
        raise RuntimeError(
            f"""

For some reason we couldn't write config to {config_file}.
Perhaps you don't have permissions here?

You can change the directory where Dask writes config files
to somewhere else using the DASK_CONFIG environment variable.

For example, you could set the following:

    export DASK_CONFIG=~/.dask
"""
        ) from e

    return config, config_file


@config.command(name="list")
def config_list():
    """Print the whole config."""
Repository: dask/dask
File: dask/config.py (Python)

Code before:
    -  Treats ``__`` (double-underscore) as nested access
    -  Calls ``ast.literal_eval`` on the value

    Any serialized config passed via ``DASK_INTERNAL_INHERIT_CONFIG`` is also set here.

    """

    if env is None:
        env = os.environ

    if "DASK_INTERNAL_INHERIT_CONFIG" in env:
        d = deserialize(env["DASK_INTERNAL_INHERIT_CONFIG"])
    else:
        d = {}

    for name, value in env.items():
        if name.startswith("DASK_"):
            varname = name[5:].lower().replace("__", ".")
            d[varname] = interpret_value(value)

    result: dict = {}
    set(d, config=result)
    return result


def interpret_value(value):
    try:
        return ast.literal_eval(value)
    except (SyntaxError, ValueError):
        if value.lower() in ("none", "null"):
            return None
        else:
            return value


def ensure_file(
    source: str, destination: str | None = None, comment: bool = True
) -> None:
    """
    Copy file to default location if it does not already exist

    This tries to move a default configuration file to a default location if
    if does not already exist.  It also comments out that file by default.

    This is to be used by downstream modules (like dask.distributed) that may
    have default configuration files that they wish to include in the default
    configuration path.

    Parameters
    ----------
    source : string, filename

Review comment (suggestion):
```suggestion
def interpret_value(value: str) -> Any:
```

Code after:
    -  Treats ``__`` (double-underscore) as nested access
    -  Calls ``ast.literal_eval`` on the value

    Any serialized config passed via ``DASK_INTERNAL_INHERIT_CONFIG`` is also set here.

    """

    if env is None:
        env = os.environ

    if "DASK_INTERNAL_INHERIT_CONFIG" in env:
        d = deserialize(env["DASK_INTERNAL_INHERIT_CONFIG"])
    else:
        d = {}

    for name, value in env.items():
        if name.startswith("DASK_"):
            varname = name[5:].lower().replace("__", ".")
            d[varname] = interpret_value(value)

    result: dict = {}
    set(d, config=result)
    return result


def interpret_value(value: str) -> Any:
    try:
        return ast.literal_eval(value)
    except (SyntaxError, ValueError):
        if value.lower() in ("none", "null"):
            return None
        else:
            return value


def ensure_file(
    source: str, destination: str | None = None, comment: bool = True
) -> None:
    """
    Copy file to default location if it does not already exist

    This tries to move a default configuration file to a default location if
    if does not already exist.  It also comments out that file by default.

    This is to be used by downstream modules (like dask.distributed) that may
    have default configuration files that they wish to include in the default
    configuration path.

    Parameters
    ----------
    source : string, filename
Repository: dask/dask
File: dask/dataframe/__init__.py (Python)

Code before:
        warnings.warn(
            """The current Dask DataFrame implementation is deprecated. 
In a future release, Dask DataFrame will use a new implementation that
contains several improvements including a logical query planning.
The user-facing DataFrame API will remain unchanged.

The new implementation is already available and can be enabled by
installing the dask-expr library:

    $ pip install dask-expr

and turning the query planning option on:

    >>> import dask
    >>> dask.config.set({'dataframe.query-planning': True})
    >>> import dask.dataframe as dd

API documentation for the new implementation is available at
https://docs.dask.org/en/stable/dask-expr-api.html

Any feedback can be reported on the Dask issue tracker
https://github.com/dask/dask/issues 

To disable this warning in the future, set dask config:

    dataframe.query-planning-warning=False
""",
            DeprecationWarning,
            stacklevel=2,
        )


from dask.dataframe._testing import test_dataframe


Review comment (suggestion):
```suggestion
    >>> dask.config.set({'dataframe.query-planning-warning': False})
```
Changed to match the paragraph just above

Code after:
In a future release, Dask DataFrame will use a new implementation that
contains several improvements including a logical query planning.
The user-facing DataFrame API will remain unchanged.

The new implementation is already available and can be enabled by
installing the dask-expr library:

    $ pip install dask-expr

and turning the query planning option on:

    >>> import dask
    >>> dask.config.set({'dataframe.query-planning': True})
    >>> import dask.dataframe as dd

API documentation for the new implementation is available at
https://docs.dask.org/en/stable/dask-expr-api.html

Any feedback can be reported on the Dask issue tracker
https://github.com/dask/dask/issues 

To disable this warning in the future, set dask config:

    >>> dask.config.set({'dataframe.query-planning-warning': False})
""",
            DeprecationWarning,
            stacklevel=2,
        )


from dask.dataframe._testing import test_dataframe
Repository: dask/dask
File: dask/cli.py (Python)

Code before:
    else:
        value = dask.config.interpret_value(value)
        _, path = save_config(key, value, file)
        click.echo(f"Updated [{key}] to [{value}], config saved to {path}")


def save_config(
    key: str,
    value: Any,
    config_file: pathlib.Path | None = None,
) -> tuple[dict, pathlib.Path]:
    """
    Save new config values to dask config file,
    return new config and path to file.
    """
    # If explicit config file wasn't supplied, try finding the first existing
    # config file w/ this key, falling back to the default config path
    if config_file is None:
        config_file = next(
            dask.config.paths_containing_key(key),
            pathlib.Path(dask.config.PATH).joinpath("dask.yaml"),
        )
    config_file.parent.mkdir(exist_ok=True, parents=True)

    if config_file.exists():
        config = yaml.safe_load(config_file.read_text()) or dict()
    else:
        config = dict()

    dask.config.set({key: value}, config=config)
    try:
        config_file.write_text(yaml.dump(config))
    except OSError as e:
        raise RuntimeError(
            f"""

For some reason we couldn't write config to {config_file}.
Perhaps you don't have permissions here?

You can change the directory where Dask writes config files
to somewhere else using the DASK_CONFIG environment variable.

For example, you could set the following:

    export DASK_CONFIG=~/.dask
"""
        ) from e

    return config, config_file



Review comment (suggestion):
```suggestion
        config = yaml.safe_load(config_file.read_text()) or {}
```

Code after:
    # If explicit config file wasn't supplied, try finding the first existing
    # config file w/ this key, falling back to the default config path
    if config_file is None:
        config_file = next(
            dask.config.paths_containing_key(key),
            pathlib.Path(dask.config.PATH) / "dask.yaml",
        )
    config_file.parent.mkdir(exist_ok=True, parents=True)

    if config_file.exists():
        config = yaml.safe_load(config_file.read_text()) or {}
    else:
        config = {}

    dask.config.set({key: value}, config=config)
    try:
        config_file.write_text(yaml.dump(config))
    except OSError as e:
        raise RuntimeError(
            f"""

For some reason we couldn't write config to {config_file}.
Perhaps you don't have permissions here?

You can change the directory where Dask writes config files
to somewhere else using the DASK_CONFIG environment variable.

For example, you could set the following:

    export DASK_CONFIG=~/.dask
"""
        ) from e

    return config, config_file


@config.command(name="list")
def config_list():
    """Print the whole config."""
    click.echo_via_pager(yaml.dump(dask.config.config))


def _register_command_ep(interface, entry_point):
    """Add `entry_point` command to `interface`.

    Parameters
    ----------
    interface : click.Command or click.Group
        The click interface to augment with `entry_point`.
    entry_point : importlib.metadata.EntryPoint
        The entry point which loads to a ``click.Command`` or
Repository: dask/dask
File: dask/config.py (Python)

Code before:
    Collect configuration from paths and environment variables

    Parameters
    ----------
    paths : list[str]
        A list of paths to search for yaml config files

    env : Mapping[str, str]
        The system environment variables

    Returns
    -------
    config: dict

    See Also
    --------
    dask.config.refresh: collect configuration and update into primary config
    """
    if env is None:
        env = os.environ

    configs: list[dict] = collect_yaml(paths=paths)  # type: ignore
    configs.append(collect_env(env=env))

    return merge(*configs)


def refresh(
    config: dict = config, defaults: list[Mapping] = defaults, **kwargs
) -> None:
    """
    Update configuration by re-reading yaml files and env variables

    This mutates the global dask.config.config, or the config parameter if
    passed in.

    This goes through the following stages:

    1.  Clearing out all old configuration
    2.  Updating from the stored defaults from downstream libraries
        (see update_defaults)
    3.  Updating from yaml files and environment variables
    4.  Automatically renaming deprecated keys (with a warning)

    Note that some functionality only checks configuration once at startup and
    may not change behavior, even if configuration changes.  It is recommended
    to restart your python process if convenient to ensure that new
    configuration changes take place.

    See Also
    --------

Review comment (suggestion):
To prevent this you could have used `@overload` above

Code after:
                if record:
                    self._record.append(("insert", path, None))
                d[key] = {}
                # No need to record subsequent operations after an insert
                record = False
            self._assign(keys[1:], value, d[key], path, record=record)


def collect(paths: list[str] = paths, env: Mapping[str, str] | None = None) -> dict:
    """
    Collect configuration from paths and environment variables

    Parameters
    ----------
    paths : list[str]
        A list of paths to search for yaml config files

    env : Mapping[str, str]
        The system environment variables

    Returns
    -------
    config: dict

    See Also
    --------
    dask.config.refresh: collect configuration and update into primary config
    """
    if env is None:
        env = os.environ

    configs = [*collect_yaml(paths=paths), collect_env(env=env)]
    return merge(*configs)


def refresh(
    config: dict = config, defaults: list[Mapping] = defaults, **kwargs
) -> None:
    """
    Update configuration by re-reading yaml files and env variables

    This mutates the global dask.config.config, or the config parameter if
    passed in.

    This goes through the following stages:

    1.  Clearing out all old configuration
    2.  Updating from the stored defaults from downstream libraries
        (see update_defaults)
    3.  Updating from yaml files and environment variables
    4.  Automatically renaming deprecated keys (with a warning)