[data] Fix multi-input operator object store memory attribution to upstream operators

[justinvyu]

Description

• Fix resource attribution for multi-input operators (union, zip): the resource manager previously charged each upstream operator the total internal inqueue bytes (across all inputs) of the downstream operator, double-counting memory for multi-input operators. Now, OpRuntimeMetrics tracks per-input inqueues, so each upstream is only charged for the blocks it actually produced.
• This fixes a deadlock with preserve_order=True where a fast input inflates the total inqueue across inputs in the union, incorrectly backpressuring the slow input that the union is waiting on (strict round-robin), so neither side can make progress.

Before:

    op_A (slow) ───┐                                                                                                                           
                    ├──▶ UnionOp [inqueue: A=10B, B=90B, total=100B]                                                                           
    op_B (fast) ───┘                                                                                                                           
                                                                                                                                               
    Resource manager charges each upstream for the AGGREGATE inqueue:                                                                          
      op_A owes: 100B (total inqueue) ← wrong, A only contributed 10B                                                                          
      op_B owes: 100B (total inqueue) ← wrong, B only contributed 90B                                                                          
                                         
    Both get backpressured → op_A can't produce → union waits for A → deadlock

After:

    op_A (slow) ───┐
                    ├──▶ UnionOp [inqueue[0]: 10B, inqueue[1]: 90B]
    op_B (fast) ───┘

    Resource manager charges each upstream for its OWN per-input inqueue:
      op_A owes: 10B  (inqueue[0]) ← correct
      op_B owes: 90B  (inqueue[1]) ← correct

    op_A stays unthrottled → produces blocks → union makes progress ✓

Followup work

This PR only solves the preserve_order case where union input queues build up. However, the default case can still run into problems:

• The default case doesn't even use input queues -- it just automatically routes inputs to the output queue, so the input attribution is immediately thrown away since all the blocks get mixed together in the output queue.

preserve_order=False case illustrated:

Step 1: Blocks arrive — union puts them directly in internal output buffer
                                                       
    op_A ──▶ union._output_buffer: [A₁]                           
    op_B ──▶ union._output_buffer: [A₁, B₁, B₂, B₃, B₄]
                                                                                                                                               
    (on_input_queued never called, on_output_queued instead — no input attribution)             
                                                                                                                                               
  Step 2: Executor eagerly drains union into external outqueue (every step)                                                                    
                                                                                                                                               
    while op.has_next():                 
        op_state.add_output(op.get_next())                                                                                                     
                                         
    union._output_buffer: []
    union external outqueue: [A₁, B₁, B₂, B₃, B₄] = 100B  ← charged to union

  Step 3: Downstream is backpressured, stops consuming — blocks pile up

    op_A ──▶ union external outqueue: 100B  ← all charged to union
    op_B ──▶                                   (no per-input info anywhere)
                      │
                downstream (backpressured, not dispatching)

  Step 4: If union is ineligible, full 100B rolls up to BOTH upstreams

    op_A usage += 100B  (but A only contributed 10B)
    op_B usage += 100B  (but B only contributed 90B)

  Blocks never touch the inqueues at all in preserve_order=False — they go straight to the output buffer, get eagerly drained to the external
  outqueue, and from there there's no information about which input produced them.

[gemini-code-assist]

Code Review

This pull request correctly fixes a memory attribution issue for multi-input operators like Union and Zip. By introducing per-input inqueues in OpRuntimeMetrics and updating the resource manager to use this fine-grained information, the change prevents double-counting of memory and resolves a potential deadlock scenario. The changes are well-structured, backward-compatible, and include a new unit test that validates the fix. I have one suggestion to improve code conciseness.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.