Repository: dask/dask (python)

Issue:
test_with_spec_integer_method fails on s390x
**Describe the issue**:

I'm not sure if this is a big-endian problem, or some other thing; I'm not really sure what the test is about.

```pytb
________________________ test_with_spec_integer_method _________________________
[gw2] linux -- Python 3.12.0 /usr/bin/python3
    def test_with_spec_integer_method():
        from dask.dataframe.io.demo import ColumnSpec, DatasetSpec, with_spec
    
        spec = DatasetSpec(
            npartitions=1,
            nrecords=5,
            column_specs=[
                ColumnSpec(prefix="pois", dtype=int, method="poisson"),
                ColumnSpec(prefix="norm", dtype=int, method="normal"),
                ColumnSpec(prefix="unif", dtype=int, method="uniform"),
                ColumnSpec(prefix="binom", dtype=int, method="binomial", args=(100, 0.4)),
                ColumnSpec(prefix="choice", dtype=int, method="choice", args=(10,)),
                ColumnSpec(prefix="rand", dtype=int, random=True, low=0, high=10),
                ColumnSpec(prefix="rand", dtype=int, random=True),
            ],
        )
        ddf = with_spec(spec, seed=42)
        res = ddf.compute()
>       assert res["pois1"].tolist() == [1002, 985, 947, 1003, 1017]
E       assert [1021, 1006, 1024, 1030, 930] == [1002, 985, 947, 1003, 1017]
E         At index 0 diff: 1021 != 1002
E         Full diff:
E         - [1002, 985, 947, 1003, 1017]
E         + [1021, 1006, 1024, 1030, 930]
../../../BUILDROOT/python-dask-2023.7.1-1.fc39~bootstrap.s390x/usr/lib/python3.12/site-packages/dask/dataframe/io/tests/test_demo.py:338: AssertionError
```

**Minimal Complete Verifiable Example**:

```
pytest
```

**Anything else we need to know?**:

**Environment**:

- Dask version: 2023.7.1
- Python version: 3.12b4
- Operating System: Fedora Rawhide
- Install method (conda, pip, source): source

Pull request description:
Force initializing the random seed with the same byte order interpret…
…ation as on x86

- [x] Closes #10423
- [x] Tests added / passed
- [x] Passes `pre-commit run --all-files`


Patch:
diff --git a/dask/blockwise.py b/dask/blockwise.py
index 51d822418..1be3c6f3f 100644
--- a/dask/blockwise.py
+++ b/dask/blockwise.py
@@ -1138,16 +1138,14 @@ def _optimize_blockwise(full_graph, keys=()):
                 blockwise_layers.add(dep)
 
                 # traverse further to this child's children
+                output_indices = set(layers[dep].output_indices)
+                input_indices = {
+                    i for _, ind in layers[dep].indices if ind for i in ind
+                }
+                is_io_superset = output_indices.issuperset(input_indices)
                 for d in full_graph.dependencies.get(dep, ()):
                     # Don't allow reductions to proceed
-                    output_indices = set(layers[dep].output_indices)
-                    input_indices = {
-                        i for _, ind in layers[dep].indices if ind for i in ind
-                    }
-
-                    if len(dependents[d]) <= 1 and output_indices.issuperset(
-                        input_indices
-                    ):
+                    if is_io_superset and len(dependents[d]) <= 1:
                         deps.add(d)
                     else:
                         stack.append(d)
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index e5b5a40bb..c98a7761a 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -8501,7 +8501,11 @@ def new_dd_object(dsk, name, meta, divisions, parent_meta=None):
             (d,) for d in meta.shape[1:]
         )
         if len(chunks) > 1:
-            layer = dsk.layers[name]
+            if isinstance(dsk, HighLevelGraph):
+                layer = dsk.layers[name]
+            else:
+                # dask-expr provides a dict only
+                layer = dsk
             if isinstance(layer, Blockwise):
                 layer.new_axes["j"] = chunks[1][0]
                 layer.output_indices = layer.output_indices + ("j",)
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
index 5a847789c..5ba11de36 100644
--- a/dask/dataframe/io/csv.py
+++ b/dask/dataframe/io/csv.py
@@ -484,6 +484,15 @@ def read_pandas(
         kwargs["lineterminator"] = lineterminator
     else:
         lineterminator = "\n"
+    if "encoding" in kwargs:
+        b_lineterminator = lineterminator.encode(kwargs["encoding"])
+        empty_blob = "".encode(kwargs["encoding"])
+        if empty_blob:
+            # This encoding starts with a Byte Order Mark (BOM), so strip that from the
+            # start of the line terminator, since this value is not a full file.
+            b_lineterminator = b_lineterminator[len(empty_blob) :]
+    else:
+        b_lineterminator = lineterminator.encode()
     if include_path_column and isinstance(include_path_column, bool):
         include_path_column = "path"
     if "index" in kwargs or (
@@ -559,7 +568,6 @@ def read_pandas(
             "Setting ``sample=blocksize``"
         )
         sample = blocksize
-    b_lineterminator = lineterminator.encode()
     b_out = read_bytes(
         urlpath,
         delimiter=b_lineterminator,
diff --git a/dask/utils.py b/dask/utils.py
index 5090b38f4..d192fe029 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -557,7 +557,7 @@ def random_state_data(n: int, random_state=None) -> list:
         random_state = np.random.RandomState(random_state)
 
     random_data = random_state.bytes(624 * n * 4)  # `n * 624` 32-bit integers
-    l = list(np.frombuffer(random_data, dtype=np.uint32).reshape((n, -1)))
+    l = list(np.frombuffer(random_data, dtype="<u4").reshape((n, -1)))
     assert len(l) == n
     return l
 


Test patch:
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index a0924d2ac..6fd980152 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -1155,17 +1155,8 @@ def test_read_csv_with_datetime_index_partitions_n():
         assert_eq(df, ddf)
 
 
-xfail_pandas_100 = pytest.mark.xfail(reason="https://github.com/dask/dask/issues/5787")
-
-
-@pytest.mark.parametrize(
-    "encoding",
-    [
-        pytest.param("utf-16", marks=xfail_pandas_100),
-        pytest.param("utf-16-le", marks=xfail_pandas_100),
-        "utf-16-be",
-    ],
-)
+# utf-8-sig and utf-16 both start with a BOM.
+@pytest.mark.parametrize("encoding", ["utf-8-sig", "utf-16", "utf-16-le", "utf-16-be"])
 def test_encoding_gh601(encoding):
     ar = pd.Series(range(0, 100))
     br = ar % 7
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 8a854ca41..2d4b35d29 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -728,10 +728,11 @@ def test_categorical(tmpdir, write_engine, read_engine):
     assert (df.x == ddf2.x.compute()).all()
 
 
-@pytest.mark.xfail(DASK_EXPR_ENABLED, reason="will be supported after string option")
 @pytest.mark.parametrize("metadata_file", [False, True])
 def test_append(tmpdir, engine, metadata_file):
     """Test that appended parquet equal to the original one."""
+    if DASK_EXPR_ENABLED and metadata_file:
+        pytest.xfail("doesn't work yet")
     tmp = str(tmpdir)
     df = pd.DataFrame(
         {
@@ -762,7 +763,6 @@ def test_append(tmpdir, engine, metadata_file):
     assert_eq(df, ddf3)
 
 
-@pytest.mark.xfail(DASK_EXPR_ENABLED, reason="will be supported after string option")
 def test_append_create(tmpdir, engine):
     """Test that appended parquet equal to the original one."""
     tmp_path = str(tmpdir)
@@ -916,10 +916,11 @@ def test_partition_on_cats_2(tmpdir, engine):
     assert set(df.cat.categories) == {"x", "y", "z"}
 
 
-@pytest.mark.xfail(DASK_EXPR_ENABLED, reason="will be supported after string option")
 @pytest.mark.parametrize("metadata_file", [False, True])
 def test_append_wo_index(tmpdir, engine, metadata_file):
     """Test append with write_index=False."""
+    if DASK_EXPR_ENABLED and metadata_file:
+        pytest.xfail("doesn't work yet")
     tmp = str(tmpdir.join("tmp1.parquet"))
     df = pd.DataFrame(
         {
@@ -950,7 +951,6 @@ def test_append_wo_index(tmpdir, engine, metadata_file):
     assert_eq(df.set_index("f"), ddf3)
 
 
-@pytest.mark.xfail(DASK_EXPR_ENABLED, reason="will be supported after string option")
 @pytest.mark.parametrize("metadata_file", [False, True])
 @pytest.mark.parametrize(
     ("index", "offset"),
@@ -4934,6 +4934,7 @@ def test_read_parquet_lists_not_converting(tmpdir):
     assert_eq(df, result)
 
 
+@PYARROW_MARK
 @pytest.mark.skipif(not PANDAS_GE_200, reason="Requires pandas>=2.0")
 def test_parquet_string_roundtrip(tmpdir):
     pdf = pd.DataFrame({"a": ["a", "b", "c"]}, dtype="string[pyarrow]")
diff --git a/dask/dataframe/tests/test_accessors.py b/dask/dataframe/tests/test_accessors.py
index 2bf904270..a2b8f133e 100644
--- a/dask/dataframe/tests/test_accessors.py
+++ b/dask/dataframe/tests/test_accessors.py
@@ -164,7 +164,7 @@ def test_str_accessor(df_ddf):
     # with `boolean` dtype, while using object strings returns a `bool`. We cast
     # the pandas DataFrame here to ensure pandas and Dask return the same dtype.
     ctx = contextlib.nullcontext()
-    if pyarrow_strings_enabled() and not dd._dask_expr_enabled():
+    if pyarrow_strings_enabled():
         df.str_col = to_pyarrow_string(df.str_col)
         if not PANDAS_GE_210:
             ctx = pytest.warns(
diff --git a/dask/dataframe/tests/test_categorical.py b/dask/dataframe/tests/test_categorical.py
index 85430febd..772c6dfc9 100644
--- a/dask/dataframe/tests/test_categorical.py
+++ b/dask/dataframe/tests/test_categorical.py
@@ -266,10 +266,7 @@ def test_categorize():
 
         ddf2 = ddf.categorize("y_", index=index)
         assert ddf2.y_.cat.known
-        if not dd._dask_expr_enabled():
-            assert ddf2.v.dtype == get_string_dtype()
-        else:
-            assert ddf.v.dtype == object
+        assert ddf2.v.dtype == get_string_dtype()
         assert ddf2.index.cat.known == known_index
         assert_eq(ddf2, df)
 
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 60568e6e3..3663320bf 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -1496,7 +1496,11 @@ def test_len():
     assert len(dd.from_pandas(pd.DataFrame(), npartitions=1)) == 0
     assert len(dd.from_pandas(pd.DataFrame(columns=[1, 2]), npartitions=1)) == 0
     # Regression test for https://github.com/dask/dask/issues/6110
-    assert len(dd.from_pandas(pd.DataFrame(columns=["foo", "foo"]), npartitions=1)) == 0
+    if not DASK_EXPR_ENABLED:
+        assert (
+            len(dd.from_pandas(pd.DataFrame(columns=["foo", "foo"]), npartitions=1))
+            == 0
+        )
 
 
 def test_size():
@@ -1872,7 +1876,7 @@ def test_assign_dtypes():
     new_col = {"col3": pd.Series(["0", "1"])}
     res = ddf.assign(**new_col)
 
-    string_dtype = get_string_dtype() if not DASK_EXPR_ENABLED else "object"
+    string_dtype = get_string_dtype()
     assert_eq(
         res.dtypes,
         pd.Series(
@@ -5400,7 +5404,6 @@ def test_cumulative_multiple_columns():
     assert_eq(ddf, df)
 
 
-@pytest.mark.xfail(DASK_EXPR_ENABLED, reason="array conversion not yet supported")
 @pytest.mark.parametrize("func", [np.asarray, M.to_records])
 def test_map_partition_array(func):
     from dask.array.utils import assert_eq
@@ -5423,7 +5426,6 @@ def test_map_partition_array(func):
         assert x.chunks[0] == (np.nan, np.nan)
 
 
-@pytest.mark.xfail(DASK_EXPR_ENABLED, reason="array conversion not yet supported")
 def test_map_partition_sparse():
     sparse = pytest.importorskip("sparse")
     # Avoid searchsorted failure.
@@ -5530,7 +5532,6 @@ def test_meta_nonempty_uses_meta_value_if_provided():
         assert_eq(expected, actual)
 
 
-@pytest.mark.xfail(DASK_EXPR_ENABLED, reason="array conversion not yet supported")
 def test_dask_dataframe_holds_scipy_sparse_containers():
     sparse = pytest.importorskip("scipy.sparse")
     da = pytest.importorskip("dask.array")
@@ -6267,7 +6268,6 @@ def test_to_backend():
             df.to_backend("missing")
 
 
-@pytest.mark.skipif(DASK_EXPR_ENABLED, reason="skip temporarily")
 @pytest.mark.parametrize("func", ["max", "sum"])
 def test_transform_getitem_works(func):
     df = pd.DataFrame({"ints": [1, 2, 3], "grouper": [0, 1, 0]})
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 444207686..a731ca622 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -2172,7 +2172,7 @@ def record_numeric_only_warnings():
         pytest.param(
             "sum",
             marks=pytest.mark.xfail(
-                pyarrow_strings_enabled() and not DASK_EXPR_ENABLED,
+                pyarrow_strings_enabled(),
                 reason="works in dask-expr",
             ),
         ),
diff --git a/dask/dataframe/tests/test_numeric.py b/dask/dataframe/tests/test_numeric.py
index 785f8e7ea..c36864188 100644
--- a/dask/dataframe/tests/test_numeric.py
+++ b/dask/dataframe/tests/test_numeric.py
@@ -45,7 +45,7 @@ def test_to_numeric_on_dask_dataframe_series():
     expected = pd.to_numeric(s)
     output = to_numeric(arg)
     expected_dtype = "int64"
-    if pyarrow_strings_enabled() and not DASK_EXPR_ENABLED:
+    if pyarrow_strings_enabled():
         # `to_numeric` output depends on input dtype
         expected_dtype = "Int64"
     assert output.dtype == expected_dtype
diff --git a/dask/dataframe/tests/test_ufunc.py b/dask/dataframe/tests/test_ufunc.py
index 48695dd39..befb5c45b 100644
--- a/dask/dataframe/tests/test_ufunc.py
+++ b/dask/dataframe/tests/test_ufunc.py
@@ -146,10 +146,6 @@ def test_ufunc(pandas_input, ufunc):
         assert_eq(dafunc(pandas_input), npfunc(pandas_input))
 
 
-@pytest.mark.skipif(
-    dd._dask_expr_enabled(),
-    reason="returns an Array on the expression level, which is something we can't represent at the moment",
-)
 @pytest.mark.parametrize(
     "ufunc",
     [
diff --git a/dask/tests/test_sizeof.py b/dask/tests/test_sizeof.py
index ef3d6ba37..40c479b94 100644
--- a/dask/tests/test_sizeof.py
+++ b/dask/tests/test_sizeof.py
@@ -81,7 +81,7 @@ def test_pandas_multiindex():
     index = pd.MultiIndex.from_product([range(50), list("abcdefghilmnopqrstuvwxyz")])
     actual_size = sys.getsizeof(index)
 
-    assert 0.5 * actual_size < sizeof(index) < 2 * actual_size
+    assert 0.5 * actual_size < sizeof(index) < 3 * actual_size
     assert isinstance(sizeof(index), int)
 
 
@@ -104,7 +104,7 @@ def test_sparse_matrix():
     assert sizeof(sp.tocoo()) >= 240
     assert sizeof(sp.tocsc()) >= 232
     assert sizeof(sp.tocsr()) >= 232
-    assert sizeof(sp.todok()) >= 188
+    assert sizeof(sp.todok()) >= 184
     assert sizeof(sp.tolil()) >= 204