Repository: dask/dask
Pull request: Revise divisions logic in from_pandas
- [x] Closes #9218
- [x] Tests added / passed
- [x] Passes `pre-commit run --all-files`

This PR was originally meant to be a small bug-fix for #9218 - However, I struggled to work with the existing `sorted_division_locations` function, and eventually decided to rewrite it in a slightly different way.

**Side Notes**:

- Most of the complexity in this code comes from the fact that `sorted_division_locations` is expected to prioritize the uniqueness of `divisions` elements over the output satisfying `npartitions` or `chunksize`. Although I can understand the motivation for having unique divisions, it is not clear to me that uniqueness is always more important than the partition size and/or count. Perhaps it makes sense to add something like a `unique_partitions` argument to `from_pandas`?
- Even when there are no duplicate values in the index column, the `npartitions=` argument is not always satisfied. This result seems a bit suprising to me.

Problem statement:
Using ddf.from_pandas with chunksize=1 results in one partition having two rows
**What happened**: When creating a ddf using `from_pandas(chunksize=1)` not all partitions have 1 row. The last partition has two rows in it.

**What you expected to happen**: The dataframe should be evenly split into 10 partitions

**Minimal Complete Verifiable Example**:

```python
import dask.dataframe as ddf
import pandas as pd
import numpy as np

if __name__ == "__main__":
    df = pd.DataFrame(np.random.randint(0, 10, size=(10, 4)), columns=list('ABCD'))
    dataset: ddf.DataFrame = ddf.from_pandas(df, chunksize=1).persist()
    # get each partition length
    num_rows = dataset.map_partitions(lambda df: len(df)).compute()
    print(list(num_rows))
```

Results in 

```
[1, 1, 1, 1, 1, 1, 1, 1, 2]
```

**Anything else we need to know?**:

**Environment**:

- Dask version: 2022.6.1
- Python version: 3.10
- Operating System: MacOS
- Install method (conda, pip, source): Pip

Hints from issue discussion:
Thanks for the simple example - I can also reproduce. I'll take a look at the `from_pandas` code to see whats going on.

Commit message: trigger linting

Patch to review:
diff --git a/dask/array/core.py b/dask/array/core.py
index b24444058e9..aad26408c72 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -5686,7 +5686,7 @@ def size(self) -> int:
         """
         The total number of blocks in the array.
         """
-        return np.prod(self.shape)
+        return int(np.prod(self.shape))
 
     @property
     def shape(self) -> tuple[int, ...]:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index a02c2d4404a..26e454bc094 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -7277,7 +7277,9 @@ def repartition_npartitions(df, npartitions):
         ]
         return _repartition_from_boundaries(df, new_partitions_boundaries, new_name)
     else:
-        original_divisions = divisions = pd.Series(df.divisions)
+        # Drop duplcates in case last partition has same
+        # value for min and max division
+        original_divisions = divisions = pd.Series(df.divisions).drop_duplicates()
         if df.known_divisions and (
             np.issubdtype(divisions.dtype, np.datetime64)
             or np.issubdtype(divisions.dtype, np.number)
diff --git a/dask/dataframe/io/io.py b/dask/dataframe/io/io.py
index 06f65af0586..d0c39de1237 100644
--- a/dask/dataframe/io/io.py
+++ b/dask/dataframe/io/io.py
@@ -174,11 +174,13 @@ def from_pandas(
     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 depending on
-        the size and index of the dataframe, the output may have fewer
-        partitions than requested.
+        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 number of rows per index partition to use.
+        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
@@ -242,7 +244,7 @@ def from_pandas(
             "Please provide chunksize as an int, or possibly as None if you specify npartitions."
         )
 
-    name = name or ("from_pandas-" + tokenize(data, chunksize))
+    name = name or ("from_pandas-" + tokenize(data, chunksize, npartitions))
 
     if not nrows:
         return new_dd_object({(name, 0): data}, name, data, [None, None])
@@ -257,7 +259,9 @@ def from_pandas(
         if not data.index.is_monotonic_increasing:
             data = data.sort_index(ascending=True)
         divisions, locations = sorted_division_locations(
-            data.index, chunksize=chunksize
+            data.index,
+            npartitions=npartitions,
+            chunksize=None if none_chunksize else chunksize,
         )
     else:
         locations = list(range(0, nrows, chunksize)) + [len(data)]
@@ -743,10 +747,10 @@ def sorted_division_locations(seq, npartitions=None, chunksize=None):
 
     >>> L = ['A', 'A', 'A', 'A', 'B', 'B', 'B', 'C']
     >>> sorted_division_locations(L, chunksize=3)
-    (['A', 'B', 'C'], [0, 4, 8])
+    (['A', 'B', 'C', 'C'], [0, 4, 7, 8])
 
     >>> sorted_division_locations(L, chunksize=2)
-    (['A', 'B', 'C'], [0, 4, 8])
+    (['A', 'B', 'C', 'C'], [0, 4, 7, 8])
 
     >>> sorted_division_locations(['A'], chunksize=2)
     (['A', 'A'], [0, 1])
@@ -754,26 +758,77 @@ def sorted_division_locations(seq, npartitions=None, chunksize=None):
     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)
+    seq_unique = (
+        seq.unique() if hasattr(seq, "unique") else np.unique(seq)
+    )  # Unique elements of seq
+    duplicates = len(seq_unique) < len(seq)
+    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")
+        )
+    else:
+        offsets = seq_unique = None
+
+    # Define chunksize and residual so that
+    # npartitions can be exactly satisfied
+    # when duplicates is False
+    residual = 0
     if npartitions:
-        chunksize = ceil(len(seq) / npartitions)
-
-    positions = [0]
-    values = [seq[0]]
-    for pos in range(0, len(seq), chunksize):
-        if pos <= positions[-1]:
-            continue
-        while pos + 1 < len(seq) and seq[pos - 1] == seq[pos]:
-            pos += 1
-        values.append(seq[pos])
-        if pos == len(seq) - 1:
-            pos += 1
-        positions.append(pos)
-
-    if positions[-1] != len(seq):
-        positions.append(len(seq))
-        values.append(seq[-1])
-
-    return values, positions
+        chunksize = len(seq) // npartitions
+        residual = len(seq) % npartitions
+
+    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)
+    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 occurance 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])
+            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
+            i = pos + chunksizes(len(divisions))
+            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):
diff --git a/dask/dataframe/io/tests/test_io.py b/dask/dataframe/io/tests/test_io.py
index 59d87bb0756..50346ad6a7f 100644
--- a/dask/dataframe/io/tests/test_io.py
+++ b/dask/dataframe/io/tests/test_io.py
@@ -377,6 +377,16 @@ def test_from_pandas_with_wrong_args():
         dd.from_pandas(df, chunksize=18.27)
 
 
+def test_from_pandas_chunksize_one():
+    # See: https://github.com/dask/dask/issues/9218
+    df = pd.DataFrame(np.random.randint(0, 10, size=(10, 4)), columns=list("ABCD"))
+    ddf = dd.from_pandas(df, chunksize=1)
+    num_rows = list(ddf.map_partitions(lambda df: len(df)).compute())
+    # chunksize=1 with range index should
+    # always have unit-length partitions
+    assert num_rows == [1] * 10
+
+
 def test_DataFrame_from_dask_array():
     x = da.ones((10, 3), chunks=(4, 2))
     pdf = pd.DataFrame(np.ones((10, 3)), columns=["a", "b", "c"])
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index c56c1784dce..ea1d8da68da 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -2388,11 +2388,11 @@ def test_read_dir_nometa(tmpdir, write_engine, read_engine, divisions, remove_co
 
     ddf2 = dd.read_parquet(tmp_path, engine=read_engine, calculate_divisions=divisions)
     assert_eq(ddf, ddf2, check_divisions=False)
-    assert ddf.divisions == tuple(range(0, 420, 30))
+    assert None not in ddf.divisions
     if divisions is False:
-        assert ddf2.divisions == (None,) * 14
+        assert ddf2.divisions == (None,) * (ddf.npartitions + 1)
     else:
-        assert ddf2.divisions == tuple(range(0, 420, 30))
+        assert ddf2.divisions == ddf.divisions
 
 
 @write_read_engines()
@@ -2402,8 +2402,8 @@ def test_statistics_nometa(tmpdir, write_engine, read_engine):
 
     ddf2 = dd.read_parquet(tmp_path, engine=read_engine, calculate_divisions=True)
     assert_eq(ddf, ddf2)
-    assert ddf.divisions == tuple(range(0, 420, 30))
-    assert ddf2.divisions == tuple(range(0, 420, 30))
+    assert None not in ddf.divisions
+    assert ddf2.divisions == ddf.divisions
 
 
 @pytest.mark.parametrize("schema", ["infer", None])
@@ -4054,9 +4054,7 @@ def test_custom_filename_with_partition(tmpdir, engine):
             )
         for file in files:
             assert file in (
-                "0-cool.parquet",
-                "1-cool.parquet",
-                "2-cool.parquet",
+                *[f"{i}-cool.parquet" for i in range(df.npartitions)],
                 "_common_metadata",
                 "_metadata",
             )
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 4af5a5cd8c1..c96f38364ad 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -1137,7 +1137,7 @@ def test_drop_duplicates_subset():
 
 def test_get_partition():
     pdf = pd.DataFrame(np.random.randn(10, 5), columns=list("abcde"))
-    ddf = dd.from_pandas(pdf, 3)
+    ddf = dd.from_pandas(pdf, chunksize=4)
     assert ddf.divisions == (0, 4, 8, 9)
 
     # DataFrame
@@ -1389,7 +1389,8 @@ def test_quantile_missing(method):
     if method == "tdigest":
         pytest.importorskip("crick")
     df = pd.DataFrame({"A": [0, np.nan, 2]})
-    ddf = dd.from_pandas(df, 2)
+    # TODO: Test npartitions=2
+    ddf = dd.from_pandas(df, npartitions=1)
     expected = df.quantile()
     result = ddf.quantile(method=method)
     assert_eq(result, expected)
@@ -2925,7 +2926,7 @@ def test_to_dask_array_raises(as_frame):
 @pytest.mark.parametrize("as_frame", [False, True])
 def test_to_dask_array_unknown(as_frame):
     s = pd.Series([1, 2, 3, 4, 5], name="foo")
-    a = dd.from_pandas(s, chunksize=2)
+    a = dd.from_pandas(s, npartitions=2)
 
     if as_frame:
         a = a.to_frame()
@@ -2948,7 +2949,7 @@ def test_to_dask_array_unknown(as_frame):
 @pytest.mark.parametrize(
     "lengths,as_frame,meta",
     [
-        ([2, 3], False, None),
+        ([2, 2, 1], False, None),
         (True, False, None),
         (True, False, np.array([], dtype="f4")),
     ],
@@ -2963,7 +2964,7 @@ def test_to_dask_array(meta, as_frame, lengths):
     result = a.to_dask_array(lengths=lengths, meta=meta)
     assert isinstance(result, da.Array)
 
-    expected_chunks = ((2, 3),)
+    expected_chunks = ((2, 2, 1),)
 
     if as_frame:
         expected_chunks = expected_chunks + ((1,),)
@@ -3095,7 +3096,7 @@ def test_round():
 def test_cov():
     # DataFrame
     df = _compat.makeMissingDataframe()
-    ddf = dd.from_pandas(df, npartitions=6)
+    ddf = dd.from_pandas(df, chunksize=5)
 
     res = ddf.cov()
     res2 = ddf.cov(split_every=2)
@@ -3115,8 +3116,8 @@ def test_cov():
     # Series
     a = df.A
     b = df.B
-    da = dd.from_pandas(a, npartitions=6)
-    db = dd.from_pandas(b, npartitions=7)
+    da = dd.from_pandas(a, chunksize=5)
+    db = dd.from_pandas(b, chunksize=5)
 
     res = da.cov(db)
     res2 = da.cov(db, split_every=2)
@@ -3137,7 +3138,7 @@ def test_cov():
 def test_corr():
     # DataFrame
     df = _compat.makeMissingDataframe()
-    ddf = dd.from_pandas(df, npartitions=6)
+    ddf = dd.from_pandas(df, chunksize=5)
 
     res = ddf.corr()
     res2 = ddf.corr(split_every=2)
@@ -3159,8 +3160,8 @@ def test_corr():
     # Series
     a = df.A
     b = df.B
-    da = dd.from_pandas(a, npartitions=6)
-    db = dd.from_pandas(b, npartitions=7)
+    da = dd.from_pandas(a, chunksize=5)
+    db = dd.from_pandas(b, chunksize=5)
 
     res = da.corr(db)
     res2 = da.corr(db, split_every=2)
@@ -3199,11 +3200,11 @@ def test_corr_same_name():
 def test_cov_corr_meta():
     df = pd.DataFrame(
         {
-            "a": np.array([1, 2, 3]),
-            "b": np.array([1.0, 2.0, 3.0], dtype="f4"),
-            "c": np.array([1.0, 2.0, 3.0]),
+            "a": np.array([1, 2, 3, 4]),
+            "b": np.array([1.0, 2.0, 3.0, 4.0], dtype="f4"),
+            "c": np.array([1.0, 2.0, 3.0, 4.0]),
         },
-        index=pd.Index([1, 2, 3], name="myindex"),
+        index=pd.Index([1, 2, 3, 4], name="myindex"),
     )
     ddf = dd.from_pandas(df, npartitions=2)
     assert_eq(ddf.corr(), df.corr())
@@ -3648,7 +3649,17 @@ def test_categorize_info():
         {"x": [1, 2, 3, 4], "y": pd.Series(list("aabc")), "z": pd.Series(list("aabc"))},
         index=[0, 1, 2, 3],
     )
-    ddf = dd.from_pandas(df, npartitions=4).categorize(["y"])
+
+    # Use from_map to construct custom partitioning
+    def myfunc(bounds):
+        start, stop = bounds
+        return df.iloc[start:stop]
+
+    ddf = dd.from_map(
+        myfunc,
+        [(0, 1), (1, 2), (2, 4)],
+        divisions=[0, 1, 2, 3],
+    ).categorize(["y"])
 
     # Verbose=False
     buf = StringIO()
@@ -4193,7 +4204,9 @@ def test_del():
 @pytest.mark.parametrize("deep", [True, False])
 def test_memory_usage(index, deep):
     df = pd.DataFrame({"x": [1, 2, 3], "y": [1.0, 2.0, 3.0], "z": ["a", "b", "c"]})
-    ddf = dd.from_pandas(df, npartitions=2)
+    # Multi-partition memory usage will not
+    # match perfectly between pandas and dask
+    ddf = dd.from_pandas(df, npartitions=1)
 
     assert_eq(
         df.memory_usage(index=index, deep=deep),
@@ -5231,7 +5244,7 @@ def test_from_dict(dtype, orient, npartitions):
     )
     if orient == "index":
         # DataFrame only has two rows with this orientation
-        assert result.npartitions == 1
+        assert result.npartitions == min(npartitions, 2)
     else:
         assert result.npartitions == npartitions
     assert_eq(result, expected)
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index eee161c746f..f56572696aa 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -2030,7 +2030,7 @@ def test_df_groupby_idxmin():
         {"idx": list(range(4)), "group": [1, 1, 2, 2], "value": [10, 20, 20, 10]}
     ).set_index("idx")
 
-    ddf = dd.from_pandas(pdf, npartitions=3)
+    ddf = dd.from_pandas(pdf, npartitions=2)
 
     expected = pd.DataFrame({"group": [1, 2], "value": [0, 3]}).set_index("group")
 
@@ -2051,7 +2051,7 @@ def test_df_groupby_idxmin_skipna(skipna):
         }
     ).set_index("idx")
 
-    ddf = dd.from_pandas(pdf, npartitions=3)
+    ddf = dd.from_pandas(pdf, npartitions=2)
 
     result_pd = pdf.groupby("group").idxmin(skipna=skipna)
     result_dd = ddf.groupby("group").idxmin(skipna=skipna)
@@ -2085,7 +2085,7 @@ def test_df_groupby_idxmax_skipna(skipna):
         }
     ).set_index("idx")
 
-    ddf = dd.from_pandas(pdf, npartitions=3)
+    ddf = dd.from_pandas(pdf, npartitions=2)
 
     result_pd = pdf.groupby("group").idxmax(skipna=skipna)
     result_dd = ddf.groupby("group").idxmax(skipna=skipna)
@@ -2098,7 +2098,7 @@ def test_series_groupby_idxmin():
         {"idx": list(range(4)), "group": [1, 1, 2, 2], "value": [10, 20, 20, 10]}
     ).set_index("idx")
 
-    ddf = dd.from_pandas(pdf, npartitions=3)
+    ddf = dd.from_pandas(pdf, npartitions=2)
 
     expected = (
         pd.DataFrame({"group": [1, 2], "value": [0, 3]}).set_index("group").squeeze()
@@ -2121,7 +2121,7 @@ def test_series_groupby_idxmin_skipna(skipna):
         }
     ).set_index("idx")
 
-    ddf = dd.from_pandas(pdf, npartitions=3)
+    ddf = dd.from_pandas(pdf, npartitions=2)
 
     result_pd = pdf.groupby("group")["value"].idxmin(skipna=skipna)
     result_dd = ddf.groupby("group")["value"].idxmin(skipna=skipna)
@@ -2157,7 +2157,7 @@ def test_series_groupby_idxmax_skipna(skipna):
         }
     ).set_index("idx")
 
-    ddf = dd.from_pandas(pdf, npartitions=3)
+    ddf = dd.from_pandas(pdf, npartitions=2)
 
     result_pd = pdf.groupby("group")["value"].idxmax(skipna=skipna)
     result_dd = ddf.groupby("group")["value"].idxmax(skipna=skipna)
diff --git a/dask/dataframe/tests/test_multi.py b/dask/dataframe/tests/test_multi.py
index 09876e50851..7fdeca15861 100644
--- a/dask/dataframe/tests/test_multi.py
+++ b/dask/dataframe/tests/test_multi.py
@@ -305,7 +305,8 @@ def test_merge_asof_on_basic():
 
     C = pd.merge_asof(A, B, on="a")
     c = dd.merge_asof(a, b, on="a")
-    assert_eq(c, C)
+    # merge_asof does not preserve index
+    assert_eq(c, C, check_index=False)
 
 
 def test_merge_asof_on_lefton_righton_error():
@@ -352,7 +353,8 @@ def test_merge_asof_on(allow_exact_matches, direction):
     c = dd.merge_asof(
         a, b, on="a", allow_exact_matches=allow_exact_matches, direction=direction
     )
-    assert_eq(c, C)
+    # merge_asof does not preserve index
+    assert_eq(c, C, check_index=False)
 
 
 @pytest.mark.parametrize("allow_exact_matches", [True, False])
@@ -477,7 +479,12 @@ def test_merge_asof_on_by():
         },
         columns=["time", "ticker", "price", "quantity"],
     )
-    b = dd.from_pandas(B, npartitions=3)
+    # TODO: Use from_pandas(B, npartitions=3)
+    b = dd.from_map(
+        lambda x: x,
+        [B.iloc[0:2], B.iloc[2:5]],
+        divisions=[0, 2, 4],
+    )
 
     C = pd.merge_asof(B, A, on="time", by="ticker")
     c = dd.merge_asof(b, a, on="time", by="ticker")
@@ -529,7 +536,12 @@ def test_merge_asof_on_by_tolerance():
         },
         columns=["time", "ticker", "price", "quantity"],
     )
-    b = dd.from_pandas(B, npartitions=3)
+    # TODO: Use from_pandas(B, npartitions=3)
+    b = dd.from_map(
+        lambda x: x,
+        [B.iloc[0:2], B.iloc[2:5]],
+        divisions=[0, 2, 4],
+    )
 
     C = pd.merge_asof(B, A, on="time", by="ticker", tolerance=pd.Timedelta("2ms"))
     c = dd.merge_asof(b, a, on="time", by="ticker", tolerance=pd.Timedelta("2ms"))
@@ -581,7 +593,12 @@ def test_merge_asof_on_by_tolerance_no_exact_matches():
         },
         columns=["time", "ticker", "price", "quantity"],
     )
-    b = dd.from_pandas(B, npartitions=3)
+    # TODO: Use from_pandas(B, npartitions=3)
+    b = dd.from_map(
+        lambda x: x,
+        [B.iloc[0:2], B.iloc[2:5]],
+        divisions=[0, 2, 4],
+    )
 
     C = pd.merge_asof(
         B,
diff --git a/dask/dataframe/tests/test_shuffle.py b/dask/dataframe/tests/test_shuffle.py
index 5ceea9778ba..4f2abf3530d 100644
--- a/dask/dataframe/tests/test_shuffle.py
+++ b/dask/dataframe/tests/test_shuffle.py
@@ -714,7 +714,7 @@ def test_set_index_interpolate_large_uint(engine):
 
     if engine == "cudf":
         gdf = cudf.from_pandas(df)
-        d = dask_cudf.from_cudf(gdf, npartitions=2)
+        d = dask_cudf.from_cudf(gdf, npartitions=1)
     else:
         d = dd.from_pandas(df, 1)
 
@@ -727,7 +727,7 @@ def test_set_index_timezone():
     s_naive = pd.Series(pd.date_range("20130101", periods=3))
     s_aware = pd.Series(pd.date_range("20130101", periods=3, tz="US/Eastern"))
     df = pd.DataFrame({"tz": s_aware, "notz": s_naive})
-    d = dd.from_pandas(df, 2)
+    d = dd.from_pandas(df, npartitions=1)  # TODO: Use npartitions=2
 
     d1 = d.set_index("notz", npartitions=1)
     s1 = pd.DatetimeIndex(s_naive.values, dtype=s_naive.dtype)


Code review comments:
diff_hunk: @@ -743,37 +747,88 @@ def sorted_division_locations(seq, npartitions=None, chunksize=None):
 
     >>> L = ['A', 'A', 'A', 'A', 'B', 'B', 'B', 'C']
     >>> sorted_division_locations(L, chunksize=3)
-    (['A', 'B', 'C'], [0, 4, 8])
+    (['A', 'B', 'C', 'C'], [0, 4, 7, 8])
 
     >>> sorted_division_locations(L, chunksize=2)
-    (['A', 'B', 'C'], [0, 4, 8])
+    (['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)
+    seq_unique = (
+        seq.unique() if hasattr(seq, "unique") else np.unique(seq)
path: dask/dataframe/io/io.py
text: @user1:
Is this also cudf-related? With pandas series we can seemingly just call `np.unique` on the sequence.

@author:
It looks like I added this in response to a mindeps failure - Apparently older numpy/pandas versions return an ndarray from `np.unique(seq)`, while newer version return an Index

diff_hunk: @@ -257,7 +259,9 @@ def from_pandas(
         if not data.index.is_monotonic_increasing:
             data = data.sort_index(ascending=True)
         divisions, locations = sorted_division_locations(
-            data.index, chunksize=chunksize
+            data.index,
+            npartitions=npartitions,
+            chunksize=None if none_chunksize else chunksize,
path: dask/dataframe/io/io.py
text: @user1:
Style nit that I think of as optional: this is sort of rubbing me the wrong way, we're defining a new varaible `none_chunksize`, then mostly not using it except for error handling and this awkward statement. What if we didn't do any `chunksize` normalization here, and instead moved `chunksize = int(ceil(nrows / npartitions))` into `sorted_division_locations`?

In general, I find the frequent mutation/normalization of user inputs in dask to be kind of hard to read/reason about, but perhaps that's just me :)

@author:
Yeah, I tried to go with the flow there, but I agree that we should just remove `none_chunksize`. Thanks for the nudge