[CHORE] Remove daft-execution (#2553)

Co-authored-by: Colin Ho <[email protected]>

Cargo.toml

@@ -6,7 +6,6 @@ daft-compression = {path = "src/daft-compression", default-features = false}
 daft-core = {path = "src/daft-core", default-features = false}
 daft-csv = {path = "src/daft-csv", default-features = false}
 daft-dsl = {path = "src/daft-dsl", default-features = false}
-daft-execution = {path = "src/daft-execution", default-features = false}
 daft-functions = {path = "src/daft-functions", default-features = false}
 daft-io = {path = "src/daft-io", default-features = false}
 daft-json = {path = "src/daft-json", default-features = false}
@@ -35,7 +34,6 @@ python = [
   "daft-core/python",
   "daft-csv/python",
   "daft-dsl/python",
-  "daft-execution/python",
   "daft-local-execution/python",
   "daft-io/python",
   "daft-json/python",
@@ -106,7 +104,6 @@ members = [
   "src/common/daft-config",
   "src/common/system-info",
   "src/daft-core",
-  "src/daft-execution",
   "src/daft-local-execution",
   "src/daft-io",
   "src/daft-parquet",

daft/plan_scheduler/physical_plan_scheduler.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from typing import Iterator
-
 from daft.daft import (
     AdaptivePhysicalPlanScheduler as _AdaptivePhysicalPlanScheduler,
 )
@@ -12,12 +10,9 @@
 from daft.execution import physical_plan
 from daft.logical.builder import LogicalPlanBuilder
 from daft.runners.partitioning import (
-    MaterializedResult,
     PartitionCacheEntry,
     PartitionT,
 )
-from daft.runners.pyrunner import PyMaterializedResult
-from daft.table.micropartition import MicroPartition
 
 
 class PhysicalPlanScheduler:
@@ -53,12 +48,6 @@ def __repr__(self) -> str:
     def to_partition_tasks(self, psets: dict[str, list[PartitionT]]) -> physical_plan.MaterializedPhysicalPlan:
         return physical_plan.materialize(self._scheduler.to_partition_tasks(psets))
 
-    def run(self, psets: dict[str, list[MaterializedResult[PartitionT]]]) -> Iterator[PyMaterializedResult]:
-        psets_mp = {part_id: [part.vpartition()._micropartition for part in parts] for part_id, parts in psets.items()}
-        return (
-            PyMaterializedResult(MicroPartition._from_pymicropartition(part)) for part in self._scheduler.run(psets_mp)
-        )
-
 
 class AdaptivePhysicalPlanScheduler:
     def __init__(self, scheduler: _AdaptivePhysicalPlanScheduler) -> None:

daft/runners/pyrunner.py

@@ -165,18 +165,12 @@ def run_iter(
             adaptive_planner = builder.to_adaptive_physical_plan_scheduler(daft_execution_config)
             while not adaptive_planner.is_done():
                 source_id, plan_scheduler = adaptive_planner.next()
-                if daft_execution_config.enable_native_executor:
-                    logger.info("Using new executor")
-                    results_gen = plan_scheduler.run(
-                        {k: v.values() for k, v in self._part_set_cache.get_all_partition_sets().items()}
-                    )
-                else:
-                    # don't store partition sets in variable to avoid reference
-                    tasks = plan_scheduler.to_partition_tasks(
-                        {k: v.values() for k, v in self._part_set_cache.get_all_partition_sets().items()}
-                    )
-                    del plan_scheduler
-                    results_gen = self._physical_plan_to_partitions(tasks)
+                # don't store partition sets in variable to avoid reference
+                tasks = plan_scheduler.to_partition_tasks(
+                    {k: v.values() for k, v in self._part_set_cache.get_all_partition_sets().items()}
+                )
+                del plan_scheduler
+                results_gen = self._physical_plan_to_partitions(tasks)
                 # if source_id is none that means this is the final stage
                 if source_id is None:
                     yield from results_gen