# Source code for dask.dataframe.io.io

    from __future__ import annotations

    from collections.abc import Iterable
    from functools import partial
    from math import ceil
    from operator import getitem
    from threading import Lock
    from typing import TYPE_CHECKING, Literal, overload

    import numpy as np
    import pandas as pd

    import dask.array as da
    from dask._task_spec import TaskRef
    from dask.base import is_dask_collection, tokenize
    from dask.blockwise import BlockwiseDepDict, blockwise
    from dask.dataframe._compat import is_any_real_numeric_dtype
    from dask.dataframe.backends import dataframe_creation_dispatch
    from dask.dataframe.core import (
        DataFrame,
        Series,
        _concat,
        _emulate,
        _Frame,
        apply_and_enforce,
        has_parallel_type,
        new_dd_object,
    )
    from dask.dataframe.dispatch import meta_lib_from_array
    from dask.dataframe.io.utils import DataFrameIOFunction
    from dask.dataframe.utils import (
        check_meta,
        insert_meta_param_description,
        is_series_like,
        make_meta,
    )
    from dask.delayed import Delayed, delayed
    from dask.highlevelgraph import HighLevelGraph
    from dask.layers import DataFrameIOLayer
    from dask.utils import M, funcname, is_arraylike

    if TYPE_CHECKING:
        import distributed

    lock = Lock()

    def _meta_from_array(x, columns=None, index=None, meta=None):
        """Create empty DataFrame or Series which has correct dtype"""

        if x.ndim > 2:
            raise ValueError(
                "from_array does not input more than 2D array, got"
                " array with shape %r" % (x.shape,)
            )

        if index is not None:
            from dask.dataframe import Index

            if not isinstance(index, Index):
                raise ValueError("'index' must be an instance of dask.dataframe.Index")
            index = index._meta

        if meta is None:
            meta = meta_lib_from_array(x).DataFrame()

        if getattr(x.dtype, "names", None) is not None:
            # record array has named columns
            if columns is None:
                columns = list(x.dtype.names)
            elif np.isscalar(columns):
                raise ValueError("For a struct dtype, columns must be a list.")
            elif not all(i in x.dtype.names for i in columns):
                extra = sorted(set(columns).difference(x.dtype.names))
                raise ValueError(f"dtype {x.dtype} doesn't have fields {extra}")
            fields = x.dtype.fields
            dtypes = [fields[n][0] if n in fields else "f8" for n in columns]
        elif x.ndim == 1:
            if np.isscalar(columns) or columns is None:
                return meta._constructor_sliced(
                    [], name=columns, dtype=x.dtype, index=index
                )
            elif len(columns) == 1:
                return meta._constructor(
                    np.array([], dtype=x.dtype), columns=columns, index=index
                )
            raise ValueError(
                "For a 1d array, columns must be a scalar or single element list"
            )
        else:
            if np.isnan(x.shape[1]):
                raise ValueError("Shape along axis 1 must be known")
            if columns is None:
                columns = list(range(x.shape[1])) if x.ndim == 2 else [0]
            elif len(columns) != x.shape[1]:
                raise ValueError(
                    "Number of column names must match width of the array. "
                    f"Got {len(columns)} names for {x.shape[1]} columns"
                )
            dtypes = [x.dtype] * len(columns)

        data = {c: np.array([], dtype=dt) for (c, dt) in zip(columns, dtypes)}
        return meta._constructor(data, columns=columns, index=index)

    def from_array(x, chunksize=50000, columns=None, meta=None):
        """Read any sliceable array into a Dask Dataframe

        Uses getitem syntax to pull slices out of the array.  The array need not be
        a NumPy array but must support slicing syntax

            x[50000:100000]

        and have 2 dimensions:

            x.ndim == 2

        or have a record dtype:

            x.dtype == [('name', 'O'), ('balance', 'i8')]

        Parameters
        ----------
        x : array_like
        chunksize : int, optional
            The number of rows per partition to use.
        columns : list or string, optional
            list of column names if DataFrame, single string if Series
        meta : object, optional
            An optional `meta` parameter can be passed for dask
            to specify the concrete dataframe type to use for partitions of
            the Dask dataframe. By default, pandas DataFrame is used.

        Returns
        -------
        dask.DataFrame or dask.Series
            A dask DataFrame/Series
        """
        if isinstance(x, da.Array):
            return from_dask_array(x, columns=columns, meta=meta)

        meta = _meta_from_array(x, columns, meta=meta)

        divisions = tuple(range(0, len(x), chunksize))
        divisions = divisions + (len(x) - 1,)
        token = tokenize(x, chunksize, columns)
        name = "from_array-" + token

        dsk = {}
        for i in range(0, int(ceil(len(x) / chunksize))):
            data = (getitem, x, slice(i * chunksize, (i + 1) * chunksize))
            if is_series_like(meta):
                dsk[name, i] = (type(meta), data, None, meta.dtype, meta.name)
            else:
                dsk[name, i] = (type(meta), data, None, meta.columns)
        return new_dd_object(dsk, name, meta, divisions)

    @overload
    def from_pandas(
        data: pd.DataFrame,
        npartitions: int | None = None,
        chunksize: int | None = None,
        sort: bool = True,
        name: str | None = None,
    ) -> DataFrame: ...

    # We ignore this overload for now until pandas-stubs can be added.
    # See https://github.com/dask/dask/issues/9220
    @overload
    def from_pandas(  # type: ignore
        data: pd.Series,
        npartitions: int | None = None,
        chunksize: int | None = None,
        sort: bool = True,
        name: str | None = None,
    ) -> Series: ...

    def from_pandas(
        data: pd.DataFrame | pd.Series,
        npartitions: int | None = None,
        chunksize: int | None = None,
        sort: bool = True,
        name: str | None = None,
    ) -> DataFrame | Series:
        """
        Construct a Dask DataFrame from a Pandas DataFrame

        This splits an in-memory Pandas dataframe into several parts and constructs
        a dask.dataframe from those parts on which Dask.dataframe can operate in
        parallel.  By default, the input dataframe will be sorted by the index to
        produce cleanly-divided partitions (with known divisions).  To preserve the
        input ordering, make sure the input index is monotonically-increasing. The
        ``sort=False`` option will also avoid reordering, but will not result in
        known divisions.

        Note that, despite parallelism, Dask.dataframe may not always be faster
        than Pandas.  We recommend that you stay with Pandas for as long as
        possible before switching to Dask.dataframe.

        Parameters
        ----------
        data : pandas.DataFrame or pandas.Series
            The DataFrame/Series with which to construct a Dask DataFrame/Series
        npartitions : int, optional
            The number of partitions of the index to create. Note that if there
            are duplicate values or insufficient elements in ``data.index``, the
            output may have fewer partitions than requested.
        chunksize : int, optional
            The desired number of rows per index partition to use. Note that
            depending on the size and index of the dataframe, actual partition
            sizes may vary.
        sort: bool
            Sort the input by index first to obtain cleanly divided partitions
            (with known divisions).  If False, the input will not be sorted, and
            all divisions will be set to None. Default is True.
        name: string, optional
            An optional keyname for the dataframe.  Defaults to hashing the input

        Returns
        -------
        dask.DataFrame or dask.Series
            A dask DataFrame/Series partitioned along the index

        Examples
        --------
        >>> from dask.dataframe import from_pandas
        >>> df = pd.DataFrame(dict(a=list('aabbcc'), b=list(range(6))),
        ...                   index=pd.date_range(start='20100101', periods=6))
        >>> ddf = from_pandas(df, npartitions=3)
        >>> ddf.divisions  # doctest: +NORMALIZE_WHITESPACE
        (Timestamp('2010-01-01 00:00:00', freq='D'),
         Timestamp('2010-01-03 00:00:00', freq='D'),
         Timestamp('2010-01-05 00:00:00', freq='D'),
         Timestamp('2010-01-06 00:00:00', freq='D'))
        >>> ddf = from_pandas(df.a, npartitions=3)  # Works with Series too!
        >>> ddf.divisions  # doctest: +NORMALIZE_WHITESPACE
        (Timestamp('2010-01-01 00:00:00', freq='D'),
         Timestamp('2010-01-03 00:00:00', freq='D'),
         Timestamp('2010-01-05 00:00:00', freq='D'),
         Timestamp('2010-01-06 00:00:00', freq='D'))

        Raises
        ------
        TypeError
            If something other than a ``pandas.DataFrame`` or ``pandas.Series`` is
            passed in.

        See Also
        --------
        from_array : Construct a dask.DataFrame from an array that has record dtype
        read_csv : Construct a dask.DataFrame from a CSV file
        """
        if isinstance(getattr(data, "index", None), pd.MultiIndex):
            raise NotImplementedError("Dask does not support MultiIndex Dataframes.")

        if not has_parallel_type(data):
            raise TypeError("Input must be a pandas DataFrame or Series.")

        if (npartitions is None) == (chunksize is None):
            raise ValueError("Exactly one of npartitions and chunksize must be specified.")

        nrows = len(data)

        if chunksize is None:
            if not isinstance(npartitions, int):
                raise TypeError(
                    "Please provide npartitions as an int, or possibly as None if you specify chunksize."
                )
        elif not isinstance(chunksize, int):
            raise TypeError(
                "Please provide chunksize as an int, or possibly as None if you specify npartitions."
            )

        name = name or ("from_pandas-" + tokenize(data, chunksize, npartitions))

        if not nrows:
            return new_dd_object({(name, 0): data}, name, data, [None, None])

        if data.index.isna().any() and not is_any_real_numeric_dtype(data.index):
            raise NotImplementedError(
                "Index in passed data is non-numeric and contains nulls, which Dask does not entirely support.\n"
                "Consider passing `data.loc[~data.isna()]` instead."
            )

        if sort:
            if not data.index.is_monotonic_increasing:
                data = data.sort_index(ascending=True)
            else:
                # sort_index copies as well
                data = data.copy()
            divisions, locations = sorted_division_locations(
                data.index,
                npartitions=npartitions,
                chunksize=chunksize,
            )
        else:
            data = data.copy()
            if chunksize is None:
                assert isinstance(npartitions, int)
                chunksize = int(ceil(nrows / npartitions))
            locations = list(range(0, nrows, chunksize)) + [len(data)]
            divisions = [None] * len(locations)

        dsk = {
            (name, i): data.iloc[start:stop]
            for i, (start, stop) in enumerate(zip(locations[:-1], locations[1:]))
        }
        return new_dd_object(dsk, name, data, divisions)

    @dataframe_creation_dispatch.register_inplace("pandas")
    def from_dict(
        data,
        npartitions,
        orient="columns",
        dtype=None,
        columns=None,
        constructor=pd.DataFrame,
    ):
        """
        Construct a Dask DataFrame from a Python Dictionary

        Parameters
        ----------
        data : dict
            Of the form {field : array-like} or {field : dict}.
        npartitions : int
            The number of partitions of the index to create. Note that depending on
            the size and index of the dataframe, the output may have fewer
            partitions than requested.
        orient : {'columns', 'index', 'tight'}, default 'columns'
            The "orientation" of the data. If the keys of the passed dict
            should be the columns of the resulting DataFrame, pass 'columns'
            (default). Otherwise if the keys should be rows, pass 'index'.
            If 'tight', assume a dict with keys
            ['index', 'columns', 'data', 'index_names', 'column_names'].
        dtype: bool
            Data type to force, otherwise infer.
        columns: string, optional
            Column labels to use when ``orient='index'``. Raises a ValueError
            if used with ``orient='columns'`` or ``orient='tight'``.
        constructor: class, default pd.DataFrame
            Class with which ``from_dict`` should be called with.

        Examples
        --------
        >>> import dask.dataframe as dd
        >>> ddf = dd.from_dict({"num1": [1, 2, 3, 4], "num2": [7, 8, 9, 10]}, npartitions=2)
        """

        collection_types = {type(v) for v in data.values() if is_dask_collection(v)}
        if collection_types:
            raise NotImplementedError(
                "from_dict doesn't currently support Dask collections as inputs. "
                f"Objects of type {collection_types} were given in the input dict."
            )

        return from_pandas(
            constructor.from_dict(data, orient, dtype, columns),
            npartitions,
        )

    def _partition_from_array(data, index=None, initializer=None, **kwargs):
        """Create a Dask partition for either a DataFrame or Series.

        Designed to be used with :func:`dask.blockwise.blockwise`. ``data`` is the array
        from which the partition will be created. ``index`` can be:

        1. ``None``, in which case each partition has an independent RangeIndex
        2. a `tuple` with two elements, the start and stop values for a RangeIndex for
           this partition, which gives a continuously varying RangeIndex over the
           whole Dask DataFrame
        3. an instance of a ``pandas.Index`` or a subclass thereof

        The ``kwargs`` _must_ contain an ``initializer`` key which is set by calling
        ``type(meta)``.
        """
        if isinstance(index, tuple):
            index = pd.RangeIndex(*index)
        return initializer(data, index=index, **kwargs)

    def from_dask_array(x, columns=None, index=None, meta=None):
        """Create a Dask DataFrame from a Dask Array.

        Converts a 2d array into a DataFrame and a 1d array into a Series.

        Parameters
        ----------
        x : da.Array
        columns : list or string
            list of column names if DataFrame, single string if Series
        index : dask.dataframe.Index, optional
            An optional *dask* Index to use for the output Series or DataFrame.

            The default output index depends on whether `x` has any unknown
            chunks. If there are any unknown chunks, the output has ``None``
            for all the divisions (one per chunk). If all the chunks are known,
            a default index with known divisions is created.

            Specifying `index` can be useful if you're conforming a Dask Array
            to an existing dask Series or DataFrame, and you would like the
            indices to match.
        meta : object, optional
            An optional `meta` parameter can be passed for dask
            to specify the concrete dataframe type to be returned.
            By default, pandas DataFrame is used.

        Examples
        --------
        >>> import dask.array as da
        >>> import dask.dataframe as dd
        >>> x = da.ones((4, 2), chunks=(2, 2))
        >>> df = dd.io.from_dask_array(x, columns=['a', 'b'])
        >>> df.compute()
             a    b
        0  1.0  1.0
        1  1.0  1.0
        2  1.0  1.0
        3  1.0  1.0

        See Also
        --------
        dask.bag.to_dataframe: from dask.bag
        dask.dataframe._Frame.values: Reverse conversion
        dask.dataframe._Frame.to_records: Reverse conversion
        """
        meta = _meta_from_array(x, columns, index, meta=meta)

        name = "from-dask-array-" + tokenize(x, columns)
        graph_dependencies = [x]
        arrays_and_indices = [x.name, "ij" if x.ndim == 2 else "i"]
        numblocks = {x.name: x.numblocks}

        if index is not None:
            # An index is explicitly given by the caller, so we can pass it through to the
            # initializer after a few checks.
            if index.npartitions != x.numblocks[0]:
                msg = (
                    "The index and array have different numbers of blocks. "
                    "({} != {})".format(index.npartitions, x.numblocks[0])
                )
                raise ValueError(msg)
            divisions = index.divisions
            graph_dependencies.append(index)
            arrays_and_indices.extend([index._name, "i"])
            numblocks[index._name] = (index.npartitions,)

        elif np.isnan(sum(x.shape)):
            # The shape of the incoming array is not known in at least one dimension. As
            # such, we can't create an index for the entire output DataFrame and we set
            # the divisions to None to represent that.
            divisions = [None] * (len(x.chunks[0]) + 1)
        else:
            # The shape of the incoming array is known and we don't have an explicit index.
            # Create a mapping of chunk number in the incoming array to
            # (start row, stop row) tuples. These tuples will be used to create a sequential
            # RangeIndex later on that is continuous over the whole DataFrame.
            n_elements = sum(x.chunks[0])
            divisions = [0]
            stop = 0
            index_mapping = {}
            for i, increment in enumerate(x.chunks[0]):
                stop += increment
                index_mapping[(i,)] = (divisions[i], stop)

                # last division corrected, even if there are empty chunk(s) at the end
                if stop == n_elements:
                    stop -= 1

                divisions.append(stop)

            arrays_and_indices.extend([BlockwiseDepDict(mapping=index_mapping), "i"])

        if is_series_like(meta):
            kwargs = {"dtype": x.dtype, "name": meta.name, "initializer": type(meta)}
        else:
            kwargs = {"columns": meta.columns, "initializer": type(meta)}

        blk = blockwise(
            _partition_from_array,
            name,
            "i",
            *arrays_and_indices,
            numblocks=numblocks,
            concatenate=True,
            # kwargs passed through to the DataFrame/Series initializer
            **kwargs,
        )

        graph = HighLevelGraph.from_collections(name, blk, dependencies=graph_dependencies)

        import dask.dataframe as dd

        if dd._dask_expr_enabled():
            from dask_expr._collection import from_graph

            from dask.utils import key_split

            return from_graph(
                graph,
                meta,
                divisions,
                [(name, i) for i in range(len(divisions) - 1)],
                key_split(name),
            )

        return new_dd_object(graph, name, meta, divisions)

    def _link(token, result):
        """A dummy function to link results together in a graph

        We use this to enforce an artificial sequential ordering on tasks that
        don't explicitly pass around a shared resource
        """
        return None

    def _df_to_bag(df, index=False, format="tuple"):
        if isinstance(df, pd.DataFrame):
            if format == "tuple":
                return list(map(tuple, df.itertuples(index)))
            elif format == "dict":
                if index:
                    return [
                        {**{"index": idx}, **values}
                        for values, idx in zip(df.to_dict("records"), df.index)
                    ]
                else:
                    return df.to_dict(orient="records")
        elif isinstance(df, pd.Series):
            if format == "tuple":
                return list(df.items()) if index else list(df)
            elif format == "dict":
                return df.to_frame().to_dict(orient="records")

    def to_bag(df, index=False, format="tuple"):
        """Create Dask Bag from a Dask DataFrame

        Parameters
        ----------
        index : bool, optional
            If True, the elements are tuples of ``(index, value)``, otherwise
            they're just the ``value``.  Default is False.
        format : {"tuple", "dict", "frame"}, optional
            Whether to return a bag of tuples, dictionaries, or
            dataframe-like objects. Default is "tuple". If "frame",
            the original partitions of ``df`` will not be transformed
            in any way.

        Examples
        --------
        >>> bag = df.to_bag()  # doctest: +SKIP
        """
        from dask.bag.core import Bag

        if not isinstance(df, (DataFrame, Series)):
            raise TypeError("df must be either DataFrame or Series")
        name = "to_bag-" + tokenize(df, index, format)
        if format == "frame":
            # Use existing graph and name of df, but
            # drop meta to produce a Bag collection
            dsk = df.dask
            name = df._name
        else:
            dsk = {
                (name, i): (_df_to_bag, block, index, format)
                for (i, block) in enumerate(df.__dask_keys__())
            }
            dsk.update(df.__dask_optimize__(df.__dask_graph__(), df.__dask_keys__()))
        return Bag(dsk, name, df.npartitions)

    def to_records(df):
        """Create Dask Array from a Dask Dataframe

        Warning: This creates a dask.array without precise shape information.
        Operations that depend on shape information, like slicing or reshaping,
        will not work.

        Examples
        --------
        >>> df.to_records()  # doctest: +SKIP

        See Also
        --------
        dask.dataframe._Frame.values
        dask.dataframe.from_dask_array
        """
        return df.map_partitions(M.to_records)

    @insert_meta_param_description
    def from_delayed(
        dfs: Delayed | distributed.Future | Iterable[Delayed | distributed.Future],
        meta=None,
        divisions: tuple | Literal["sorted"] | None = None,
        prefix: str = "from-delayed",
        verify_meta: bool = True,
    ) -> DataFrame | Series:
        """Create Dask DataFrame from many Dask Delayed objects.

        .. warning::
            ``from_delayed`` should only be used if the objects that create
            the data are complex and cannot be easily represented as a single
            function in an embarassingly parallel fashion.

            ``from_map`` is recommended if the query can be expressed as a single
            function like:

            def read_xml(path):
                return pd.read_xml(path)

            ddf = dd.from_map(read_xml, paths)

            ``from_delayed`` might be depreacted in the future.

        Parameters
        ----------
        dfs :
            A ``dask.delayed.Delayed``, a ``distributed.Future``, or an iterable of either
            of these objects, e.g. returned by ``client.submit``. These comprise the
            individual partitions of the resulting dataframe.
            If a single object is provided (not an iterable), then the resulting dataframe
            will have only one partition.
        $META
        divisions :
            Partition boundaries along the index.
            For tuple, see https://docs.dask.org/en/latest/dataframe-design.html#partitions
            For string 'sorted' will compute the delayed values to find index
            values.  Assumes that the indexes are mutually sorted.
            If None, then won't use index information
        prefix :
            Prefix to prepend to the keys.
        verify_meta :
            If True check that the partitions have consistent metadata, defaults to True.
        """
        from dask.delayed import Delayed

        if isinstance(dfs, Delayed) or hasattr(dfs, "key"):
            dfs = [dfs]
        dfs = [
            delayed(df) if not isinstance(df, Delayed) and hasattr(df, "key") else df
            for df in dfs
        ]

        for item in dfs:
            if not isinstance(item, Delayed):
                raise TypeError("Expected Delayed object, got %s" % type(item).__name__)

        if meta is None:
            meta = delayed(make_meta)(dfs[0]).compute()
        else:
            meta = make_meta(meta)

        if not dfs:
            dfs = [delayed(make_meta)(meta)]

        if divisions is None or divisions == "sorted":
            divs: list | tuple = [None] * (len(dfs) + 1)
        else:
            divs = list(divisions)
            if len(divs) != len(dfs) + 1:
                raise ValueError("divisions should be a tuple of len(dfs) + 1")

        name = prefix + "-" + tokenize(*dfs)
        layer = DataFrameIOLayer(
            name=name,
            columns=None,
            inputs=BlockwiseDepDict(
                {(i,): TaskRef(inp.key) for i, inp in enumerate(dfs)},
                produces_keys=True,
            ),
            io_func=(
                partial(check_meta, meta=meta, funcname="from_delayed")
                if verify_meta
                else lambda x: x
            ),
        )
        df = new_dd_object(
            HighLevelGraph.from_collections(name, layer, dfs), name, meta, divs
        )

        if divisions == "sorted":
            from dask.dataframe.shuffle import compute_and_set_divisions

            return compute_and_set_divisions(df)

        return df

    def sorted_division_locations(seq, npartitions=None, chunksize=None):
        """Find division locations and values in sorted list

        Examples
        --------

        >>> L = ['A', 'B', 'C', 'D', 'E', 'F']
        >>> sorted_division_locations(L, chunksize=2)
        (['A', 'C', 'E', 'F'], [0, 2, 4, 6])

        >>> sorted_division_locations(L, chunksize=3)
        (['A', 'D', 'F'], [0, 3, 6])

        >>> L = ['A', 'A', 'A', 'A', 'B', 'B', 'B', 'C']
        >>> sorted_division_locations(L, chunksize=3)
        (['A', 'B', 'C', 'C'], [0, 4, 7, 8])

        >>> sorted_division_locations(L, chunksize=2)
        (['A', 'B', 'C', 'C'], [0, 4, 7, 8])

        >>> sorted_division_locations(['A'], chunksize=2)
        (['A', 'A'], [0, 1])
        """
        if (npartitions is None) == (chunksize is None):
            raise ValueError("Exactly one of npartitions and chunksize must be specified.")

        # Find unique-offset array (if duplicates exist).
        # Note that np.unique(seq) should work in all cases
        # for newer versions of numpy/pandas
        seq_unique = seq.unique() if hasattr(seq, "unique") else np.unique(seq)
        duplicates = len(seq_unique) < len(seq)
        enforce_exact = False
        if duplicates:
            offsets = (
                # Avoid numpy conversion (necessary for dask-cudf)
                seq.searchsorted(seq_unique, side="left")
                if hasattr(seq, "searchsorted")
                else np.array(seq).searchsorted(seq_unique, side="left")
            )
            enforce_exact = npartitions and len(offsets) >= npartitions
        else:
            offsets = seq_unique = None

        # Define chunksize and residual so that
        # npartitions can be exactly satisfied
        # when duplicates is False
        residual = 0
        subtract_drift = False
        if npartitions:
            chunksize = len(seq) // npartitions
            residual = len(seq) % npartitions
            subtract_drift = True

        def chunksizes(ind):
            # Helper function to satisfy npartitions
            return chunksize + int(ind < residual)

        # Always start with 0th item in seqarr,
        # and then try to take chunksize steps
        # along the seqarr array
        divisions = [seq[0]]
        locations = [0]
        i = chunksizes(0)
        ind = None  # ind cache (sometimes avoids nonzero call)
        drift = 0  # accumulated drift away from ideal chunksizes
        divs_remain = npartitions - len(divisions) if enforce_exact else None
        while i < len(seq):
            # Map current position selection (i)
            # to the corresponding division value (div)
            div = seq[i]
            # pos is the position of the first occurrence of
            # div (which is i when seq has no duplicates)
            if duplicates:
                # Note: cupy requires casts to `int` below
                if ind is None:
                    ind = int((seq_unique == seq[i]).nonzero()[0][0])
                if enforce_exact:
                    # Avoid "over-stepping" too many unique
                    # values when npartitions is approximately
                    # equal to len(offsets)
                    offs_remain = len(offsets) - ind
                    if divs_remain > offs_remain:
                        ind -= divs_remain - offs_remain
                        i = offsets[ind]
                        div = seq[i]
                pos = int(offsets[ind])
            else:
                pos = i
            if div <= divisions[-1]:
                # pos overlaps with divisions.
                # Try the next element on the following pass
                if duplicates:
                    ind += 1
                    # Note: cupy requires cast to `int`
                    i = int(offsets[ind]) if ind < len(offsets) else len(seq)
                else:
                    i += 1
            else:
                # pos does not overlap with divisions.
                # Append candidate pos/div combination, and
                # take another chunksize step
                if subtract_drift:
                    # Only subtract drift when user specified npartitions
                    drift = drift + ((pos - locations[-1]) - chunksizes(len(divisions) - 1))
                if enforce_exact:
                    divs_remain -= 1
                i = pos + max(1, chunksizes(len(divisions)) - drift)
                divisions.append(div)
                locations.append(pos)
                ind = None

        # The final element of divisions/locations
        # will always be the same
        divisions.append(seq[-1])
        locations.append(len(seq))

        return divisions, locations

    class _PackedArgCallable(DataFrameIOFunction):
        """Packed-argument wrapper for DataFrameIOFunction

        This is a private helper class for ``from_map``. This class
        ensures that packed positional arguments will be expanded
        before the underlying function (``func``) is called. This class
        also handles optional metadata enforcement and column projection
        (when ``func`` satisfies the ``DataFrameIOFunction`` protocol).
        """

        def __init__(
            self,
            func,
            args=None,
            kwargs=None,
            meta=None,
            packed=False,
            enforce_metadata=False,
        ):
            self.func = func
            self.args = args
            self.kwargs = kwargs
            self.meta = meta
            self.enforce_metadata = enforce_metadata
            self.packed = packed
            self.is_dataframe_io_func = isinstance(self.func, DataFrameIOFunction)

        @property
        def columns(self):
            if self.is_dataframe_io_func:
                return self.func.columns
            return None

        def project_columns(self, columns):
            if self.is_dataframe_io_func:
                return _PackedArgCallable(
                    self.func.project_columns(columns),
                    args=self.args,
                    kwargs=self.kwargs,
                    meta=self.meta[columns],
                    packed=self.packed,
                    enforce_metadata=self.enforce_metadata,
                )
            return self

        def __call__(self, packed_arg):
            if not self.packed:
                packed_arg = [packed_arg]
            if self.enforce_metadata:
                return apply_and_enforce(
                    *packed_arg,
                    *(self.args or []),
                    _func=self.func,
                    _meta=self.meta,
                    **(self.kwargs or {}),
                )
            return self.func(
                *packed_arg,
                *(self.args or []),
                **(self.kwargs or {}),
            )

    @insert_meta_param_description
    def from_map(
        func,
        *iterables,
        args=None,
        meta=None,
        divisions=None,
        label=None,
        token=None,
        enforce_metadata=True,
        **kwargs,
    ):
        """Create a DataFrame collection from a custom function map.

        ``from_map`` is the preferred option when reading from data sources
        that are not natively supported by Dask or if the data source
        requires custom handling before handing things of to Dask DataFrames.
        Examples are things like binary files or other unstructured data that
        doesn't have an IO connector.

        Parameters
        ----------
        func : callable
            Function used to create each partition. If ``func`` satisfies the
            ``DataFrameIOFunction`` protocol, column projection will be enabled.
        *iterables : Iterable objects
            Iterable objects to map to each output partition. All iterables must
            be the same length. This length determines the number of partitions
            in the output collection (only one element of each iterable will
            be passed to ``func`` for each partition).
        args : list or tuple, optional
            Positional arguments to broadcast to each output partition. Note
            that these arguments will always be passed to ``func`` after the
            ``iterables`` positional arguments.
        $META
        divisions : tuple, str, optional
            Partition boundaries along the index.
            For tuple, see https://docs.dask.org/en/latest/dataframe-design.html#partitions
            For string 'sorted' will compute the delayed values to find index
            values.  Assumes that the indexes are mutually sorted.
            If None, then won't use index information
        label : str, optional
            String to use as the function-name label in the output
            collection-key names.
        token : str, optional
            String to use as the "token" in the output collection-key names.
        enforce_metadata : bool, default True
            Whether to enforce at runtime that the structure of the DataFrame
            produced by ``func`` actually matches the structure of ``meta``.
            This will rename and reorder columns for each partition,
            and will raise an error if this doesn't work,
            but it won't raise if dtypes don't match.
        **kwargs:
            Key-word arguments to broadcast to each output partition. These
            same arguments will be passed to ``func`` for every output partition.

        Examples
        --------
        >>> import pandas as pd
        >>> import dask.dataframe as dd
        >>> func = lambda x, size=0: pd.Series([x] * size)
        >>> inputs = ["A", "B"]
        >>> dd.from_map(func, inputs, size=2).compute()
        0    A
        1    A
        0    B
        1    B
        dtype: object

        This API can also be used as an alternative to other file-based
        IO functions, like ``read_csv`` (which are already just
        ``from_map`` wrapper functions):

        >>> import pandas as pd
        >>> import dask.dataframe as dd
        >>> paths = ["0.csv", "1.csv", "2.csv"]
        >>> dd.from_map(pd.read_csv, paths).head()  # doctest: +SKIP
                            name
        timestamp
        2000-01-01 00:00:00   Laura
        2000-01-01 00:00:01  Oliver
        2000-01-01 00:00:02   Alice
        2000-01-01 00:00:03  Victor
        2000-01-01 00:00:04     Bob

        Since ``from_map`` allows you to map an arbitrary function
        to any number of iterable objects, it can be a very convenient
        means of implementing functionality that may be missing from
        from other DataFrame-creation methods. For example, if you
        happen to have apriori knowledge about the number of rows
        in each of the files in a dataset, you can generate a
        DataFrame collection with a global RangeIndex:

        >>> import pandas as pd
        >>> import numpy as np
        >>> import dask.dataframe as dd
        >>> paths = ["0.csv", "1.csv", "2.csv"]
        >>> file_sizes = [86400, 86400, 86400]
        >>> def func(path, row_offset):
        ...     # Read parquet file and set RangeIndex offset
        ...     df = pd.read_csv(path)
        ...     return df.set_index(
        ...         pd.RangeIndex(row_offset, row_offset+len(df))
        ...     )
        >>> def get_ddf(paths, file_sizes):
        ...     offsets = [0] + list(np.cumsum(file_sizes))
        ...     return dd.from_map(
        ...         func, paths, offsets[:-1], divisions=offsets
        ...     )
        >>> ddf = get_ddf(paths, file_sizes)  # doctest: +SKIP
        >>> ddf.index  # doctest: +SKIP
        Dask Index Structure:
        npartitions=3
        0         int64
        86400       ...
        172800      ...
        259200      ...
        dtype: int64
        Dask Name: myfunc, 6 tasks

        See Also
        --------
        dask.layers.DataFrameIOLayer
        """

        # Input validation
        if not callable(func):
            raise ValueError("`func` argument must be `callable`")
        lengths = set()
        iterables = list(iterables)
        for i, iterable in enumerate(iterables):
            if not isinstance(iterable, Iterable):
                raise ValueError(
                    f"All elements of `iterables` must be Iterable, got {type(iterable)}"
                )
            try:
                lengths.add(len(iterable))
            except (AttributeError, TypeError):
                iterables[i] = list(iterable)
                lengths.add(len(iterables[i]))
        if len(lengths) == 0:
            raise ValueError("`from_map` requires at least one Iterable input")
        elif len(lengths) > 1:
            raise ValueError("All `iterables` must have the same length")
        if lengths == {0}:
            raise ValueError("All `iterables` must have a non-zero length")

        # Check for `produces_tasks` and `creation_info`.
        # These options are included in the function signature,
        # because they are not intended for "public" use.
        produces_tasks = kwargs.pop("produces_tasks", False)
        creation_info = kwargs.pop("creation_info", None)

        if produces_tasks or len(iterables) == 1:
            if len(iterables) > 1:
                # Tasks are not detected correctly when they are "packed"
                # within an outer list/tuple
                raise ValueError(
                    "Multiple iterables not supported when produces_tasks=True"
                )
            inputs = iterables[0]
            packed = False
        else:
            inputs = list(zip(*iterables))
            packed = True

        # Define collection name
        label = label or funcname(func)
        token = token or tokenize(
            func, meta, inputs, args, divisions, enforce_metadata, **kwargs
        )
        name = f"{label}-{token}"

        # Get "projectable" column selection.
        # Note that this relies on the IO function
        # ducktyping with DataFrameIOFunction
        column_projection = func.columns if isinstance(func, DataFrameIOFunction) else None

        # NOTE: Most of the metadata-handling logic used here
        # is copied directly from `map_partitions`
        if meta is None:
            meta = _emulate(
                func,
                *(inputs[0] if packed else inputs[:1]),
                *(args or []),
                udf=True,
                **kwargs,
            )
            meta_is_emulated = True
        else:
            meta = make_meta(meta)
            meta_is_emulated = False

        if not (has_parallel_type(meta) or is_arraylike(meta) and meta.shape):
            if not meta_is_emulated:
                raise TypeError(
                    "Meta is not valid, `from_map` expects output to be a pandas object. "
                    "Try passing a pandas object as meta or a dict or tuple representing the "
                    "(name, dtype) of the columns."
                )
            # If `meta` is not a pandas object, the concatenated results will be a
            # different type
            meta = make_meta(_concat([meta]))

        # Ensure meta is empty DataFrame
        meta = make_meta(meta)

        # Define io_func
        if packed or args or kwargs or enforce_metadata:
            io_func = _PackedArgCallable(
                func,
                args=args,
                kwargs=kwargs,
                meta=meta if enforce_metadata else None,
                enforce_metadata=enforce_metadata,
                packed=packed,
            )
        else:
            io_func = func

        # Construct DataFrameIOLayer
        layer = DataFrameIOLayer(
            name,
            column_projection,
            inputs,
            io_func,
            label=label,
            produces_tasks=produces_tasks,
            creation_info=creation_info,
        )

        # Return new DataFrame-collection object
        divisions = divisions or [None] * (len(inputs) + 1)
        graph = HighLevelGraph.from_collections(name, layer, dependencies=[])
        return new_dd_object(graph, name, meta, divisions)

    def to_backend(ddf: _Frame, backend: str | None = None, **kwargs):
        """Move a DataFrame collection to a new backend

        Parameters
        ----------
        ddf : DataFrame, Series, or Index
            The input dataframe collection.
        backend : str, Optional
            The name of the new backend to move to. The default
            is the current "dataframe.backend" configuration.

        Returns
        -------
        dask.DataFrame, dask.Series or dask.Index
            A new dataframe collection with the backend
            specified by ``backend``.
        """
        # Get desired backend
        backend = backend or dataframe_creation_dispatch.backend
        # Check that "backend" has a registered entrypoint
        backend_entrypoint = dataframe_creation_dispatch.dispatch(backend)
        # Call `DataFrameBackendEntrypoint.to_backend`
        return backend_entrypoint.to_backend(ddf, **kwargs)

    DataFrame.to_records.__doc__ = to_records.__doc__
    DataFrame.to_bag.__doc__ = to_bag.__doc__