[FEAT] [WIP] [Execution] Execution model prototype

[clarkzinzow]

This PR contains a prototype of a new execution model. The new execution model is designed with:

• AQE as a first-class citizen,
• division of operator scheduling into exchange operators and pipelinable task trees,
• pipelining scheduling policies that are executor-agnostic (i.e. are shared across local and distributed executors),
• task execution and fusion that are executor-agnostic,
• resource accounting shared across executors,
• all execution implemented in Rust.

This PR contains a proof-of-concept local executor + a scheduling implementation for a handful of operations (in-memory scan, file scan, project, filter, repartition, sort, collect).

TODOs

Pre-merge TODOs

• Implement limit exchange + limit sink
• Implement hash join exchange
• [Prereq for montonically increasing ID] Support in-scheduler mutable state protocol for partition task ops that need to adjust their task configuration based on the input metadata. E.g., monotonically increasing ID needs to bind the partition number to each partition task.
• Implement monotonically increasing ID op.
• Support in-scheduler mutable state protocol for partition task ops that need to adjust their task configuration based on the output metadata of previous partition tasks for that same op. E.g., row number needs to bind the running row count to each partition task, and limit should bind the remaining limit to each partition task as an optimization.
• Fix redundant task ID creation.
• Fix intermediate queue ordering.
• Consolidate bulk and streaming partition task scheduler logic.
• Have schedulers select over future waiting, wait timeout, and SIGINT/SIGTERM signals.
• [Timeboxed - lifetime + Send + Sync issues] Use tokio runtime + spawn for streaming sink execution instead of background thread
• Port stage planner to visitor pattern
• Port physical plan → partition task tree builder to visitor pattern
• [Timeboxed] Try to create static correspondence between e.g. partition task with scan task input + leaf partition task node + scan task input
• Make exchange/sink-imposed backpressure policy for partition task scheduler pluggable
• Implement partial metadata
• Unit tests

Post-merge TODOs

• [Follow-up - Ray executor] Make PartitionTaskOp serializable with enum + PartitionTaskOp.to_type() that returns the appropriate variant
• [Follow-up - Ray executor] Instead of having static executor and partition ref from planner-down, allow both to be dynamic so exchange operator could dynamically choose whether to execute locally or remote