There is a missing thing. This patch also should handle symmetric repartition join. The way is very similar to your group-by work.
For that, please take a look at the below code. This code chooses a smaller table and gets the proper number of partitions. After this patch is applied, the maximum number of partitions are limited to worker slots. So, you need to choose lager tables as the base table for calculating the number of task.
ExecutionBlock inner = childs.get(1);
long innerVolume = getInputVolume(subQuery.masterPlan, subQuery.context, inner);
LOG.info("Outer volume: " + Math.ceil((double)outerVolume / 1048576));
LOG.info("Inner volume: " + Math.ceil((double)innerVolume / 1048576));
long smaller = Math.min(outerVolume, innerVolume);
int mb = (int) Math.ceil((double)smaller / 1048576);
LOG.info("Smaller Table's volume is approximately " + mb + " MB");
int taskNum = (int) Math.ceil((double)mb /
LOG.info("The determined number of join partitions is " + taskNum);