public static Set<Set<LogicalVertex>> computePipelinedRegions(
            final Iterable<? extends LogicalVertex> topologicallySortedVertices) {

        final Map<LogicalVertex, Set<LogicalVertex>> vertexToRegion =
                PipelinedRegionComputeUtil.buildRawRegions(
                        topologicallySortedVertices,
                        LogicalPipelinedRegionComputeUtil::getMustBePipelinedConsumedResults);

        // Since LogicalTopology is a DAG, there is no need to do cycle detection nor to merge
        // regions on cycles.
        return uniqueVertexGroups(vertexToRegion);
    }
@Test
    void testTwoInputsMergesIntoOne() {
        JobVertex v1 = new JobVertex("v1");
        JobVertex v2 = new JobVertex("v2");
        JobVertex v3 = new JobVertex("v3");
        JobVertex v4 = new JobVertex("v4");

        v3.connectNewDataSetAsInput(
                v1, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);
        v3.connectNewDataSetAsInput(
                v2, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);
        v4.connectNewDataSetAsInput(
                v3, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);

        Set<Set<LogicalVertex>> regions = computePipelinedRegions(v1, v2, v3, v4);

        checkRegionSize(regions, 2, 3, 1);
    }