Kevin comments

daft/execution/physical_plan.py

@@ -209,6 +209,9 @@ def actor_pool_project(
     actor_pool_name = f"ActorPool_stage{stage_id}"
 
     # Keep track of materializations of the children tasks
+    #
+    # Our goal here is to saturate the actors, and so we need a sufficient number of completed child tasks to do so. However
+    # we do not want too many child tasks to be running (potentially starving our actors) and hence place an upper bound of `num_actors * 2`
     child_materializations_buffer_len = num_actors * 2
     child_materializations: deque[SingleOutputPartitionTask[PartitionT]] = deque()
 

daft/runners/pyrunner.py

@@ -128,6 +128,7 @@ def initialize_actor_global_state(uninitialized_projection: ExpressionsProjectio
 
             logger.info("Initializing stateful UDFs: %s", ", ".join(partial_stateful_udfs.keys()))
 
+            # TODO: Account for Stateful Actor initialization arguments as well as user-provided batch_size
             PyActorPool.initialized_stateful_udfs_process_singleton = {
                 name: partial_udf.func_cls() for name, partial_udf in partial_stateful_udfs.items()
             }

daft/runners/runner.py

@@ -79,6 +79,6 @@ def actor_pool_context(
             name: Name of the actor pool for debugging/observability
             resource_request: Requested amount of resources for each actor
             num_actors: Number of actors to spin up
-            partial_stateful_udf: A stateful UDF that has been "bound" to its arguments, so each actor can run it
+            projection: Projection to be run on the incoming data (contains Stateful UDFs as well as other stateless expressions such as aliases)
         """
         ...

tests/actor_pool/test_pyactor_pool.py

@@ -30,8 +30,7 @@ def test_pyactor_pool():
     ppm = PartialPartitionMetadata(num_rows=None, size_bytes=None)
     instr = StatefulUDFProject(projection=projection)
 
-    pool_id = pool.setup()
-    assert pool_id == "my-pool"
+    pool.setup()
 
     result = pool.submit(
         instruction_stack=[instr],