fix(instrumented_exec): close spans when streams finish, not on plan

[antonmry]

Which issue does this PR close?

• Closes Excution plan takes a long time to drop for some reason #27.

Rationale for this change

When datafusion-tracing is used with an OpenTelemetry subscriber backed by
SimpleSpanProcessor, dropping an Arc<dyn ExecutionPlan> could block the
calling thread for several seconds.

Each InstrumentedExec node held a strong Arc to its recorders and a clone
of its tracing Span in OnceLock fields. Those fields lived until the plan
itself was dropped. With SimpleSpanProcessor, every span close calls
futures_executor::block_on(exporter.export(...)) synchronously, so:

plan_drop_time ≈ N_nodes × OTLP_round_trip_latency

A query plan with 6 nodes and a 50 ms/span exporter blocked drop() for
~328 ms. At real-world OTLP latencies (200 ms to Jaeger under load) and larger
plans (40 nodes), this matches the 8-second drop reported in the issue.

What changes are included in this PR?

InstrumentedExec now holds Mutex<Option<Weak<*Recorder>>> instead of
OnceLock<Arc<*Recorder>>, and no longer stores a Span clone at all.

• get_or_create_node_recorder() creates the span (inside NodeRecorder) on
  the first execute() call and stores only a Weak — the Arc is owned by
  the execution stream.
• MetricsRecorder and PreviewRecorder follow the same pattern.
• All span clones live in the stream pipeline. When the last stream finishes,
  the recorders drop, span fields are recorded, and the span closes.
• Dropping the plan drops only the Weak pointers — no span I/O, no blocking.

A new span() accessor was added to NodeRecorder so execute() can get the
span from the recorder instead of a separate OnceLock.

Are these changes tested?

A self-contained reproducer is available at:
https://gist.github.com/antonmry/7f2355b83cd1cf2c37ea92bea7cd245b

It simulates a slow OTLP backend (50 ms/span) and measures drop() time for
both SimpleSpanProcessor and BatchSpanProcessor.

Before the fix (published datafusion-tracing = "53.0.0"):

=== Step 2: problem — SimpleSpanProcessor (synchronous export) ===
  drop() blocked for: 328ms  (exported 6 spans × 50ms)

=== Step 3: fix — BatchSpanProcessor (async export) ===
  drop() returned in:  456µs  ← instant, spans queued

After the fix (this PR):

=== Step 2: SimpleSpanProcessor ===
  drop() blocked for: 161µs  (exported 6 spans × 50ms)

=== Step 3: BatchSpanProcessor ===
  drop() returned in:  73µs  ← instant, spans queued

SimpleSpanProcessor drop time drops from 328 ms → 161 µs (~2000×).
Both processors are now equivalent from the caller's perspective: drop() is
instant in both cases because all span work completes during collect().

The existing unit-test suite passes without changes.

Are there any user-facing changes?

Spans now close when the last execution stream is consumed rather than when the
plan is dropped. This is a behavioral improvement: span end times more
accurately reflect when execution actually finished. There are no API changes.

[antonmry]

@geoffreyclaude I was able to reproduce the issue, but I'm not so sure about the fix, in particular with the required changes in the snapshots. I would appreciate your feedback before I continue with it

[geoffreyclaude]

@geoffreyclaude I was able to reproduce the issue, but I'm not so sure about the fix, in particular with the required changes in the snapshots. I would appreciate your feedback before I continue with it

@antonmry I think you're on the right track, but I believe there’s still one tricky edge case here.

The current state is still tied to the plan node itself, and seems to assume each partition is executed once. But ExecutionPlan::execute(partition, context) can legally be called multiple times on the same plan (unfortunately!), including for the same partition, or for only some partitions. In those cases, I don’t think completed_partitions: Vec<bool> can accurately represent what has really finished.

I'm not sure what the proper solution here is though!

[antonmry]

@geoffreyclaude excellent point, thank you. I've committed d554065 addressing it. Would that work?

Numbers are still good:

Case	SimpleSpanProcessor drop(plan_clone)	BatchSpanProcessor drop(plan_clone)	Exported spans
Before fix	330.791375ms	470.541µs	6
After fix	395.667µs	71.709µs	6

[geoffreyclaude]

Thanks a lot for continuing to push this! The active-stream change fixes the weird cases of execute being called multiple times.

But I think there is still one minor issue left. Overlapping/concurrent executions will share the same recorders and so the same span, when we probably want one different span per execution call.

Maybe keying the recorders by TaskContext (to avoid mixing up two concurrent plan executions on different context) and by "has this partition been seen" (to avoid mixing up concurrent calls in the same context and on the same partition) would work?

To be clearer, for both these cases, we want two different spans:

different contexts

let s1 = plan.execute(0, ctx_a.clone())?;
let s2 = plan.execute(1, ctx_b.clone())?;

duplicate executions

let s1 = plan.execute(0, ctx.clone())?;
let s2 = plan.execute(0, ctx.clone())?;

Sorry for being so strict about this! Your initial fix of the long drops is great, but it uncovered a whole class of preexisting concurrency issues.

[antonmry]

Hey @geoffreyclaude, I appreciate the careful review. I’m still getting familiar with this part of the codebase, so I’m very happy to keep iterating until we have a solid fix. Please keep the comments coming.

I added 13a90c0 following your suggested direction. Let me know what you think