Local worker allocations

[gabotechs]

Small optimization that allows the dynamic planner to collocate single-tasked stages into the same machine that is acting as coordinator.

During dynamic planning, if all stages are small, they will just use 1 task, and before this PR, there was already some logic for co-locating all stages in a single worker. The challenge:

The single worker was a random worker in the cluster.

This means that if the coordinating context also happens to be in the scope of a worker, there's a missed chance for running everything locally, almost as if it was just single-node DataFusion.

This PRs unlocks this scenario by allowing users to inject their own LocalWorkerContext into the coordinating SessionContext. That way, the coordinator itself can recognize it's also a worker, and can send to itself all these single-tasked stages, avoiding network hops and data serialization.

Benchmarks

TPCH-SF1: 1.12 faster ✔
TPCH-SF10: 1.06 faster ✔
TPCH-SF100: 1.01 faster ✔
TPC-DS: 1.08 faster ✔
ClickBench: 1.01 faster ✔

[jayshrivastava]

stray println

[gabotechs]

Yeah, this one is on purpose, like the ones we have on metrics_collection.rs, for local iteration is nice to see the plan in the output.

[jayshrivastava]

Nit: Do you think CoordinatorContext is more clear than LocalWorkerContext because it only exists on the coordinator?

[gabotechs]

It also exists on workers actually, during a query, in each remote worker, we populate the SessionContext with this same information:

datafusion-distributed/src/worker/impl_coordinator_channel.rs

Lines 62 to 66 in e240854

	.with_extension(Arc::new(LocalWorkerContext {
	task_data_entries: Arc::clone(&self.task_data_entries),
	self_url: request.target_worker_url,
	}))
	.with_distributed_option_extension_from_headers::<DistributedConfig>(&headers)?;

This way, workers can recognize when they need to reach themselves, and fallback to a local connection, rather than a remote one. This is the same, but in the coordinator, in case it also happens to be a worker.