comphead commented on PR #2670:
URL:
https://github.com/apache/datafusion-ballista/pull/2670#issuecomment-6452684383
Thanks @andygrove, splitting this out of #2657 makes it much easier to
review. Notes below, most important first. Items marked (unverified) are
expectations I did not test.
### Correctness and design
1. **`remove_job` does not release a producer blocked on an unclaimed queue
in the real writer.** `registry.rs` documents that `remove_job` closes
unclaimed queues, and
`removing_the_job_releases_a_producer_blocked_on_an_unclaimed_queue` passes
because the test drops its `Arc<ProducerOutputs>`. In `execute_pipelined`
(`writer.rs:232-272`) the writer keeps `outputs` alive until the task ends, so
the receivers stay open and the producer waits out the claim timeout instead.
Closing the receivers explicitly in `remove_job` and `remove_stage`
(`rx.close()`) would fix it. The test should hold the `Arc` the way the writer
does.
2. **`fetch_pipelined_partition` may retry after the server has answered.**
The `stream.message().await` error arm goes through `is_transport_failure`, so
a reset after the response headers arrive becomes `GrpcConnectionError` and is
retried. If the executor claims the queue before it answers (the docs imply it
does, but the server side is not in this PR), the retry finds a claimed or
released queue. It then waits the full registration timeout and reports "timed
out waiting for pipelined producer to start", and the batches the first stream
took are lost. Consider mapping only errors from `do_get().await` to
`GrpcConnectionError`. A test that resets the connection after the headers
would show which case applies.
3. **`is_transport_failure` matches `status.message() == "transport
error"`.** String matching is fragile. `Unavailable` is also retried even when
the server sent it, for example from a draining executor. Source-chain
detection alone may be enough.
4. **Memory is outside the DataFusion `MemoryPool`.** Queues
(`queue_capacity` batches per output partition per task), the per-output
`BatchCoalescer`s and the broadcast `ReplayBuffer` (kept until `remove_job`)
are untracked. Is accounting planned for a later PR? Also, where is the
broadcast size bound enforced? The comment points at the planner threshold,
which is not in this PR.
5. **The claim timeout is an absolute deadline from `ProducerOutputs::new`,
not an idle timeout.** A consumer plan that calls `execute` on a child lazily,
after a long first phase, would hit "no consumer opened output partition" while
the producer's queue is full. The hash join probe side is opened eagerly, so it
is fine. Other plan shapes may not be (unverified). A doc note on the plan
shapes this relies on would help.
6. **A second `open` of a released queue gives a misleading error.** Once
all queues are claimed, `release` drops the slot. A repeated open then creates
a new `Waiting` slot and fails after the timeout with "timed out ... to start".
An explicit "already read" error would be clearer, and a test for it is missing.
7. **`create_grpc_server`: `.max_concurrent_streams(None)` appears to be a
no-op.** tonic 0.14.6 defaults it to `None` (`transport/server/mod.rs:128`) and
passes it through to hyper unchanged. It is fine as documentation, but the
comment reads as if it changes behavior for the scheduler and executor servers.
8. **`SharedMap` in `executor/src/client_pool.rs` is unbounded.** Slots are
never removed, and `evict_idle` ignores them. With executor churn, slots for
dead `(host, port, config)` keys accumulate for the life of the process.
Removing the slot on discard, and evicting slots with no outside `Arc` after
the idle timeout, would bound it.
9. **No task or executor loss recovery.** Queues are one-shot, so any loss
fails the job. This is documented in the config text and looks like a
deliberate first step. It would be worth stating in #2652 as a non-goal. For
comparison, Flink restarts the pipelined region and Spark retries a stage from
shuffle files.
### Reuse and simplification
- `open_remote` copies `with_retry` and `is_retriable_fetch_error`
(`shuffle_reader.rs:943-980`). Making those `pub(crate)` removes a second retry
policy.
- `fetch_pipelined_partition` repeats the `do_get` and schema-read sequence
in `execute_do_get`. A shared helper that takes the error mapper would do.
- `ProducerKey` duplicates `TaskKey` (`serde/scheduler/mod.rs`). Using
`TaskKey` avoids a `String` allocation per lookup.
- `output_partition_ids` (writer and `Routing`) re-derives what
`compute_global_output_partition_ids` computes, and nothing outside `writer.rs`
calls it. Keep one.
- The hash-partitioning proto encode is copied from `SortShuffleWriterExec`
(`serde/mod.rs` ~1222 vs ~1176). A small helper would serve both.
- `with_partitions` repeats the slicing in `task_builder.rs`, and
`filter_map` silently drops out-of-range indexes where
`slice_by_global_partition` errors. Only tests call it today, so it could move
to the scheduler PR.
- Writer plumbing: `Arc<HashMap<usize, Output>>` and `Arc<AtomicUsize>` are
threaded through five signatures, so one `Outputs` struct would do.
`Option<Mutex<BatchCoalescer>>` plus `.expect("hash outputs always coalesce")`
carries an invariant at runtime. `finish_outputs` does not use `&self`.
`Routing::Hash` rebuilds `Partitioning::Hash` for every input.
- Small items:
- `try_new_broadcast` returns `Result` but cannot fail.
- `ProducerOutputs::schema()` is `pub` and used once.
- `producer_count` is only used by tests.
- The producer failure is stored twice, in `ProducerOutputs.failure` and
in `ReplayBuffer.outcome`.
- The `io-util` tokio feature is only needed by `copy_bidirectional` in
`tests.rs`, so it can go in dev-dependencies.
- Metric names differ from the siblings (`ShuffleWriteMetrics`,
`local_partitions` and `remote_partitions`).
- Question on design: DataFusion's `RepartitionExec` already hash-partitions
and coalesces. Was reimplementing it in the writer (about 120 lines plus the
proto field and serde) a deliberate choice to avoid a channel hop?
### Efficiency
- Each output's `Mutex<BatchCoalescer>` is shared by all input tasks and
held during the copy. Rows are also copied twice, once by `partition_iter` and
once by the coalescer. Per-input coalescers would avoid the lock at a memory
cost (unverified, needs a benchmark).
- `drive_hashed` awaits sends in order, so one full queue (default capacity
2) stalls the other partitions of that input.
- `shared()` allocates the key (host `String` and config clone) and takes
the DashMap shard lock and the `connecting` mutex on every open, even when a
connection exists. Checking the existing connection first would keep the hot
path read-only.
- `execute()` builds `GrpcClientConfig` and `shared_connection_config` even
when every producer is local.
- `RegistryState::closed` clones `job_id` for each lookup under the registry
mutex.
- Each `PipelinedPartitionLocation` carries a full `ExecutorMetadata`, and
`execute` deep-clones the `Vec` per partition. This follows
`PartitionLocation`, but plan size grows with P x T on wide jobs.
- Small items: a `HashMap` lookup per piece where partition ids are dense,
and `get_array_memory_size()` per batch for the bytes metric.
- Per-stream throughput is roughly stream window / RTT, so about 400 Mbit/s
at 10 ms RTT with the 512 KiB default (my arithmetic, not measured). A note on
how the default was chosen would help.
### Tests
- Overlap: `tests::producer_fails_when_no_consumer_opens_its_full_queue` vs
`registry::producer_gives_up_on_a_full_queue_nobody_opens`, and
`tests::broadcast_outputs_replay_to_every_consumer_task` vs
`registry::broadcast_output_replays_to_every_consumer`.
- `queue_can_be_opened_once_and_is_released_once_claimed` never opens a
second time, so the "once" half is untested.
- `pipelined_queue_capacity_is_never_zero` only checks `.max(1)`.
- `unused_port()` binds, drops and reuses the port, which is racy. Keeping
the listener alive avoids that.
- `ProducerOutputs::new(schema(), [0], false, 1, None, WAIT)` is repeated
about ten times. A helper would shorten the tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
