Repository: dask/dask

Issue:
Fix CSV encoding with pandas 1.0
With pandas 1.0.0rc0, the test `dask/dataframe/io/tests/test_csv.py::test_encoding_gh601[utf-16]` is failing

```pytb
pytest dask/dataframe/io/tests/test_csv.py -k test_encoding_gh601[utf-16] -v
================================================================================ test session starts ================================================================================
platform darwin -- Python 3.7.6, pytest-5.2.1, py-1.8.0, pluggy-0.13.0 -- /Users/taugspurger/Envs/pandas-dev/bin/python
cachedir: .pytest_cache
hypothesis profile 'default' -> database=DirectoryBasedExampleDatabase('/Users/taugspurger/sandbox/dask/.hypothesis/examples')
rootdir: /Users/taugspurger/sandbox/dask, inifile: setup.cfg
plugins: xdist-1.29.0, hypothesis-4.36.2, forked-1.0.2, repeat-0.8.0, cov-2.8.1
collected 122 items / 121 deselected / 1 selected

dask/dataframe/io/tests/test_csv.py::test_encoding_gh601[utf-16] FAILED                                                                                                       [100%]
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> traceback >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

encoding = 'utf-16'

    @pytest.mark.parametrize("encoding", ["utf-16", "utf-16-le", "utf-16-be"])
    def test_encoding_gh601(encoding):
        ar = pd.Series(range(0, 100))
        br = ar % 7
        cr = br * 3.3
        dr = br / 1.9836
        test_df = pd.DataFrame({"a": ar, "b": br, "c": cr, "d": dr})

        with tmpfile(".csv") as fn:
            test_df.to_csv(fn, encoding=encoding, index=False)

            a = pd.read_csv(fn, encoding=encoding)
            d = dd.read_csv(fn, encoding=encoding, blocksize=1000)
>           d = d.compute()

dask/dataframe/io/tests/test_csv.py:1055:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
dask/base.py:165: in compute
    (result,) = compute(self, traverse=False, **kwargs)
dask/base.py:436: in compute
    results = schedule(dsk, keys, **kwargs)
dask/threaded.py:81: in get
    **kwargs
dask/local.py:486: in get_async
    raise_exception(exc, tb)
dask/local.py:316: in reraise
    raise exc
dask/local.py:222: in execute_task
    result = _execute_task(task, data)
dask/core.py:118: in _execute_task
    args2 = [_execute_task(a, cache) for a in args]
dask/core.py:118: in <listcomp>
    args2 = [_execute_task(a, cache) for a in args]
dask/core.py:119: in _execute_task
    return func(*args2)
dask/utils.py:29: in apply
    return func(*args, **kwargs)
dask/dataframe/io/csv.py:68: in pandas_read_text
    df = reader(bio, **kwargs)
../pandas/pandas/io/parsers.py:675: in parser_f
    return _read(filepath_or_buffer, kwds)
../pandas/pandas/io/parsers.py:448: in _read
    parser = TextFileReader(fp_or_buf, **kwds)
../pandas/pandas/io/parsers.py:879: in __init__
    self._make_engine(self.engine)
../pandas/pandas/io/parsers.py:1111: in _make_engine
    self._engine = CParserWrapper(self.f, **self.options)
../pandas/pandas/io/parsers.py:1883: in __init__
    self._reader = parsers.TextReader(src, **kwds)
pandas/_libs/parsers.pyx:529: in pandas._libs.parsers.TextReader.__cinit__
    ???
pandas/_libs/parsers.pyx:719: in pandas._libs.parsers.TextReader._get_header
    ???
pandas/_libs/parsers.pyx:915: in pandas._libs.parsers.TextReader._tokenize_rows
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

>   ???
E   UnicodeDecodeError: 'utf-16-le' codec can't decode byte 0x0a in position 0: truncated data

pandas/_libs/parsers.pyx:2062: UnicodeDecodeError

```

I haven't been able to fix this yet. A few tidbits

* Pandas fixed an issue with CSV IO and encoding in https://github.com/pandas-dev/pandas/issues/24130. Maybe related
* https://github.com/dask/dask/blob/5db6dad7947f186a9764cdf409b8f50bf7c38f22/dask/dataframe/io/csv.py#L397 looks fishy. That should presumably be encoded with the user-provided encoding?

Discussion:
github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     15 files  ± 0       15 suites  ±0   3h 20m 40s :stopwatch: + 1m 52s
 13 105 tests + 1   12 178 :white_check_mark: + 3     927 :zzz:  -  2  0 :x: ±0 
162 299 runs  +12  145 790 :white_check_mark: +37  16 509 :zzz:  - 25  0 :x: ±0 

Results for commit cb95e212. ± Comparison against base commit 9c20facd.

[test-results]:data:application/gzip;base64,H4sIAIa84WUC/02MSQ7DIBAEv2JxzmEYFpt8xmKzhOIlwnCK8vcQ4mDfpqpH9SJTmP1O7h0Vt47sOaSTXI46hW39MgKHosqa6s4oiD+Pe7a2PtF+OOUjPItU2Dc16TAX1UKjj3GLh4l5rWWJqNTBrcxFr6DJX5lKAe3xkq58LdttWUIqQKxRwiNFNpTDUcUcQzB8chyt9NryAbnREsj7A8wlAbQZAQAA


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,


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 de619a923..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"),
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