[DBSP] Fix incorrect hashing in UpsertHandle - #6755
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
Nice bug fix. The default_hash() as u32 truncation was silently disagreeing with the u64 hash used everywhere else, and the disagreement only showed up when workers was not a power of two (which is why the existing 2-worker tests never tripped it).
Two improvements make this durable:
- Widening
HashFuncfromu32tou64removes the hash-space mismatch by construction, not by hoping downstream operators truncate the same way. dyn_shard_pairsbeforeupdate_set(with theis_sologuard aroundmark_sharded) removes the "trust the collection handle's placement" contract; the input is now globally sharded on multihost, which was a latent hole forUpsertHandleon cluster deployments.
The new map_join_test_mt3 / set_join_test_mt3 regressions land the fix in the shape that would have caught the original bug (worker count 3 is the essential detail — the review comment on TestCase.java calls this out explicitly). Bumping the SQL compiler tests to 3 workers by default is exactly the right instinct: broken windows for hash-truncation bugs.
The new inline comments on distinct.rs (by_hash uses h % workers, not h.default_hash() % workers, so must not escape the region) and on both join.rs sites ("valid only because the join function copies the input key unchanged") are the kind of load-bearing invariant documentation this code was missing. Please keep them.
One nit: deleting the commented-out shard_vec block is a small win for readability, but the surrounding block comment ("We may want to uncomment...") ended up as an orphan */ in the diff? A quick re-read of the surrounding text before merge would confirm nothing dangles.
Signed off, no AI attribution.
|
@blp the analysis found a multi-host related bug, so I added you as a reviewer |
blp
left a comment
There was a problem hiding this comment.
I only really looked at the particular part related to multihost.
| // UpsertHandle shards its inputs across workers on the current host | ||
| // only; on a multihost layout the updates must be re-sharded globally. | ||
| // `update_set` requires a sharded input and does not shard it itself. | ||
| if Runtime::runtime().is_none_or(|rt| rt.layout().is_solo()) { | ||
| sorted.mark_sharded(); | ||
| } |
There was a problem hiding this comment.
I'd mark it as sharded across the local workers instead:
| // UpsertHandle shards its inputs across workers on the current host | |
| // only; on a multihost layout the updates must be re-sharded globally. | |
| // `update_set` requires a sharded input and does not shard it itself. | |
| if Runtime::runtime().is_none_or(|rt| rt.layout().is_solo()) { | |
| sorted.mark_sharded(); | |
| } | |
| sorted.mark_sharded_workers(rt.layout().local_workers()); |
There was a problem hiding this comment.
@blp, I assume you mean
if let Some(rt) = Runtime::runtime() {
sorted.mark_sharded_workers(rt.layout().local_workers());
}
There was a problem hiding this comment.
That looks right, I didn't actually try to compile it.
| let factories_clone = factories.clone(); | ||
|
|
||
| let sorted = input_stream | ||
| let sorted = input_stream.apply_owned(|mut upserts| { |
There was a problem hiding this comment.
maybe we should just get rid of add_input_set. Noone's using it.
There was a problem hiding this comment.
I can try to delete it and see if there are any problems
There was a problem hiding this comment.
Filed #6757 for this, will follow-up in a separate PR.
Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
Fixes #6753
Also, changed the default number of workers in the java tests to 3, this will hopefully catch this kind of bug sooner.