public static Map<List<RowExpression>, Boolean> getExpressionsPartitionedByCSE(Collection<? extends RowExpression> expressions, int expressionGroupSize)
    {
        if (expressions.isEmpty()) {
            return ImmutableMap.of();
        }

        CommonSubExpressionCollector expressionCollector = new CommonSubExpressionCollector();
        expressions.forEach(expression -> expression.accept(expressionCollector, null));
        Set<RowExpression> cse = expressionCollector.cseByLevel.values().stream().flatMap(Set::stream).collect(toImmutableSet());

        if (cse.isEmpty()) {
            return expressions.stream().collect(toImmutableMap(ImmutableList::of, m -> false));
        }

        ImmutableMap.Builder<List<RowExpression>, Boolean> expressionsPartitionedByCse = ImmutableMap.builder();
        SubExpressionChecker subExpressionChecker = new SubExpressionChecker(cse);
        Map<Boolean, List<RowExpression>> expressionsWithCseFlag = expressions.stream().collect(Collectors.partitioningBy(expression -> expression.accept(subExpressionChecker, null)));
        expressionsWithCseFlag.get(false).forEach(expression -> expressionsPartitionedByCse.put(ImmutableList.of(expression), false));

        List<RowExpression> expressionsWithCse = expressionsWithCseFlag.get(true);
        if (expressionsWithCse.size() == 1) {
            RowExpression expression = expressionsWithCse.get(0);
            expressionsPartitionedByCse.put(ImmutableList.of(expression), true);
            return expressionsPartitionedByCse.build();
        }

        List<Set<RowExpression>> cseDependency = expressionsWithCse.stream()
                .map(expression -> subExpressions(expression).stream()
                        .filter(cse::contains)
                        .collect(toImmutableSet()))
                .collect(toImmutableList());

        boolean[] merged = new boolean[expressionsWithCse.size()];

        int i = 0;
        while (i < merged.length) {
            while (i < merged.length && merged[i]) {
                i++;
            }
            if (i >= merged.length) {
                break;
            }
            merged[i] = true;
            List<RowExpression> newList = new ArrayList<>();
            newList.add(expressionsWithCse.get(i));
            Set<RowExpression> dependencies = new HashSet<>();
            Set<RowExpression> first = cseDependency.get(i);
            dependencies.addAll(first);
            int j = i + 1;
            while (j < merged.length && newList.size() < expressionGroupSize) {
                while (j < merged.length && merged[j]) {
                    j++;
                }
                if (j >= merged.length) {
                    break;
                }
                Set<RowExpression> second = cseDependency.get(j);
                if (!Sets.intersection(dependencies, second).isEmpty()) {
                    RowExpression expression = expressionsWithCse.get(j);
                    newList.add(expression);
                    dependencies.addAll(second);
                    merged[j] = true;
                    j = i + 1;
                }
                else {
                    j++;
                }
            }
            expressionsPartitionedByCse.put(ImmutableList.copyOf(newList), true);
        }

        return expressionsPartitionedByCse.build();
    }
@Test
    void testGetExpressionsWithCSE()
    {
        List<RowExpression> expressions = ImmutableList.of(rowExpression("x + y"), rowExpression("(x + y) * 2"), rowExpression("x + 2"), rowExpression("y * (x + 2)"), rowExpression("x * y"));
        Map<List<RowExpression>, Boolean> expressionsWithCSE = getExpressionsPartitionedByCSE(expressions, 3);
        assertEquals(
                expressionsWithCSE,
                ImmutableMap.of(
                        ImmutableList.of(rowExpression("x + y"), rowExpression("(x + y) * 2")), true,
                        ImmutableList.of(rowExpression("x + 2"), rowExpression("y * (x + 2)")), true,
                        ImmutableList.of(rowExpression("x * y")), false));
        expressions = ImmutableList.of(rowExpression("x + y"), rowExpression("x * 2"), rowExpression("x + y + x * 2"), rowExpression("y * 2"), rowExpression("x + y * 2"));
        expressionsWithCSE = getExpressionsPartitionedByCSE(expressions, 3);
        assertEquals(
                expressionsWithCSE,
                ImmutableMap.of(
                        ImmutableList.of(rowExpression("x + y"), rowExpression("x + y + x * 2"), rowExpression("x * 2")), true,
                        ImmutableList.of(rowExpression("y * 2"), rowExpression("x + y * 2")), true));

        expressionsWithCSE = getExpressionsPartitionedByCSE(expressions, 2);
        assertEquals(
                expressionsWithCSE,
                ImmutableMap.of(
                        ImmutableList.of(rowExpression("x + y"), rowExpression("x + y + x * 2")), true,
                        ImmutableList.of(rowExpression("y * 2"), rowExpression("x + y * 2")), true,
                        ImmutableList.of(rowExpression("x * 2")), true));
    }