Remove legacy DataFrame implementation (#8968)

Author: phofl

.github/workflows/tests.yaml

@@ -57,16 +57,6 @@ jobs:
             label: no_queue
             partition: "not ci1"
 
-          # Set dataframe.query-planning: false
-          - os: ubuntu-latest
-            environment: "3.10"
-            label: no_expr
-            partition: "ci1"
-          - os: ubuntu-latest
-            environment: "3.10"
-            label: no_expr
-            partition: "not ci1"
-
           # dask.array P2P shuffle
           - os: ubuntu-latest
             environment: mindeps
@@ -83,12 +73,12 @@ jobs:
           - os: ubuntu-latest
             environment: mindeps
             label: pandas
-            extra_packages: [numpy=1.24, pandas=2.0, pyarrow=14.0.1]
+            extra_packages: [numpy=1.24, pandas=2.0, pyarrow=14.0.1, dask-expr]
             partition: "ci1"
           - os: ubuntu-latest
             environment: mindeps
             label: pandas
-            extra_packages: [numpy=1.24, pandas=2.0, pyarrow=14.0.1]
+            extra_packages: [numpy=1.24, pandas=2.0, pyarrow=14.0.1, dask-expr]
             partition: "not ci1"
 
           - os: ubuntu-latest
@@ -221,11 +211,6 @@ jobs:
         if: ${{ matrix.label == 'no_queue' }}
         run: echo "DASK_DISTRIBUTED__SCHEDULER__WORKER_SATURATION=inf" >> $GITHUB_ENV
 
-      - name: Set up dask env to disable dask-expr
-        shell: bash -l {0}
-        if: ${{ matrix.label == 'no_expr' }}
-        run: echo "DASK_DATAFRAME__QUERY_PLANNING=False" >> $GITHUB_ENV
-
       - name: Print host info
         # host_info.py imports numpy, which isn't a direct dependency of distributed
         if: matrix.environment != 'mindeps'

distributed/shuffle/__init__.py

@@ -1,18 +1,12 @@
 from __future__ import annotations
 
 from distributed.shuffle._arrow import check_minimal_arrow_version
-from distributed.shuffle._merge import HashJoinP2PLayer, hash_join_p2p
 from distributed.shuffle._rechunk import rechunk_p2p
 from distributed.shuffle._scheduler_plugin import ShuffleSchedulerPlugin
-from distributed.shuffle._shuffle import P2PShuffleLayer, rearrange_by_column_p2p
 from distributed.shuffle._worker_plugin import ShuffleWorkerPlugin
 
 __all__ = [
     "check_minimal_arrow_version",
-    "hash_join_p2p",
-    "HashJoinP2PLayer",
-    "P2PShuffleLayer",
-    "rearrange_by_column_p2p",
     "rechunk_p2p",
     "ShuffleSchedulerPlugin",
     "ShuffleWorkerPlugin",