Merge branch main into colin/streaming-physical-writes

Cargo.toml

@@ -69,7 +69,9 @@ python = [
   "common-daft-config/python",
   "common-system-info/python",
   "common-display/python",
-  "common-resource-request/python"
+  "common-resource-request/python",
+  "dep:pyo3",
+  "dep:pyo3-log"
 ]
 
 [lib]
@@ -151,7 +153,9 @@ members = [
   "src/daft-sketch",
   "src/daft-sql",
   "src/daft-writers",
-  "src/hyperloglog"
+  "src/daft-table",
+  "src/hyperloglog",
+  "src/parquet2"
 ]
 
 [workspace.dependencies]

src/daft-local-execution/src/pipeline.rs

@@ -12,8 +12,9 @@ use daft_core::{
 use daft_dsl::{col, join::get_common_join_keys, Expr};
 use daft_micropartition::MicroPartition;
 use daft_physical_plan::{
-    Concat, EmptyScan, Explode, Filter, HashAggregate, HashJoin, InMemoryScan, Limit,
-    LocalPhysicalPlan, PhysicalWrite, Pivot, Project, Sample, Sort, UnGroupedAggregate, Unpivot,
+    ActorPoolProject, Concat, EmptyScan, Explode, Filter, HashAggregate, HashJoin, InMemoryScan,
+    Limit, LocalPhysicalPlan, PhysicalWrite, Pivot, Project, Sample, Sort, UnGroupedAggregate,
+    Unpivot,
 };
 use daft_plan::{populate_aggregation_stages, JoinType};
 use daft_table::ProbeState;
@@ -136,7 +137,7 @@ pub fn physical_plan_to_pipeline(
             input, projection, ..
         }) => {
             let proj_op = ActorPoolProjectOperator::new(projection.clone());
-            let child_node = physical_plan_to_pipeline(input, psets)?;
+            let child_node = physical_plan_to_pipeline(input, psets, cfg)?;
             IntermediateNode::new(Arc::new(proj_op), vec![child_node]).boxed()
         }
         LocalPhysicalPlan::Sample(Sample {