In trying to implement #8306 — specifically converting single_partition_join to blockwise — I realized that map_partitions repartitions single-element inputs to align with multi-partition inputs. I'm not sure if this is intended or desired.

The obvious thing to do for single_partition_join is something like:

joined = multi_partition.map_partitions(merge_chunk, single_partition, **kwargs)

hoping that single_partition would be broadcast to every partition in multi_partition. This would match with the behavior of Array.map_blocks (single-chunk inputs are broadcast to every chunk).

The code in question is _maybe_align_partitions, which first does this is_broadcastable check against the inputs:

But is_broadcastable isn't as generic as it sounds—it's specifically a check for single-row Series, to support df - df.mean():

This was added 5 years ago in #2085 and hasn't changed since.

I'm wondering what the right behavior should be. On one hand, splitting single-partition inputs to align them as we do now means behavior is consistent with multi-partition inputs. OTOH, this is inconsistent with the corollary function from Array (map_blocks), and inconsistent with the Blockwise concept in general.

Personally, I think broadcasting single-partition DataFrames is both less surprising, and a more powerful API. It's just particularly annoying that there's no way around this right now. But, I'm not sure how much existing map_partitions code is relying on the current behavior.

If we don't want to change behavior, I think we should at least add align_inputs=True as a map_partitions kwarg (just like align_arrays in map_blocks) so you can work around this when necessary.

cc @jcrist @ian-r-rose @rjzamora @jrbourbeau