Add support for parallel by DAGs

This will not yet work for nested parallel by, because we need to split up the inputs as part of the split, the way we join outputs with the join code and not inside the input connector code.