public List<Supplier<PageProjectionWithOutputs>> compileProjections(
            SqlFunctionProperties sqlFunctionProperties,
            Map<SqlFunctionId, SqlInvokedFunction> sessionFunctions,
            List<? extends RowExpression> projections,
            boolean isOptimizeCommonSubExpression,
            Optional<String> classNameSuffix)
    {
        if (isOptimizeCommonSubExpression) {
            ImmutableList.Builder<Supplier<PageProjectionWithOutputs>> pageProjections = ImmutableList.builder();
            ImmutableMap.Builder<RowExpression, Integer> expressionsWithPositionBuilder = ImmutableMap.builder();
            Set<RowExpression> expressionCandidates = new HashSet<>();
            for (int i = 0; i < projections.size(); i++) {
                RowExpression projection = projections.get(i);
                // Duplicate expressions are not expected here in general due to duplicate assignments pruning in query optimization, hence we skip CSE for them to allow for a
                // simpler implementation (and duplicate projections in expressionsWithPositionBuilder will throw exception when calling expressionsWithPositionBuilder.build())
                if (projection instanceof ConstantExpression || projection instanceof InputReferenceExpression || expressionCandidates.contains(projection)) {
                    pageProjections.add(toPageProjectionWithOutputs(compileProjection(sqlFunctionProperties, sessionFunctions, projection, classNameSuffix), new int[] {i}));
                }
                else {
                    expressionsWithPositionBuilder.put(projection, i);
                    expressionCandidates.add(projection);
                }
            }
            Map<RowExpression, Integer> expressionsWithPosition = expressionsWithPositionBuilder.build();

            Map<List<RowExpression>, Boolean> projectionsPartitionedByCSE = getExpressionsPartitionedByCSE(expressionsWithPosition.keySet(), MAX_PROJECTION_GROUP_SIZE);

            for (Map.Entry<List<RowExpression>, Boolean> entry : projectionsPartitionedByCSE.entrySet()) {
                if (entry.getValue()) {
                    pageProjections.add(toPageProjectionWithOutputs(
                            compileProjectionCached(sqlFunctionProperties, sessionFunctions, entry.getKey(), true, classNameSuffix),
                            toIntArray(entry.getKey().stream().map(expressionsWithPosition::get).collect(toImmutableList()))));
                }
                else {
                    verify(entry.getKey().size() == 1, "Expect non-cse expression list to only have one element");
                    RowExpression projection = entry.getKey().get(0);
                    pageProjections.add(toPageProjectionWithOutputs(
                            compileProjection(sqlFunctionProperties, sessionFunctions, projection, classNameSuffix),
                            new int[] {expressionsWithPosition.get(projection)}));
                }
            }
            return pageProjections.build();
        }
        return IntStream.range(0, projections.size())
                .mapToObj(outputChannel -> toPageProjectionWithOutputs(
                        compileProjection(sqlFunctionProperties, sessionFunctions, projections.get(outputChannel), classNameSuffix),
                        new int[] {outputChannel}))
                .collect(toImmutableList());
    }
@Test
    public void testCommonSubExpressionInProjection()
    {
        PageFunctionCompiler functionCompiler = new PageFunctionCompiler(createTestMetadataManager(), 0);

        List<Supplier<PageProjectionWithOutputs>> pageProjectionsCSE = functionCompiler.compileProjections(SESSION.getSqlFunctionProperties(), ImmutableList.of(ADD_X_Y, ADD_X_Y_Z), true, Optional.empty());
        assertEquals(pageProjectionsCSE.size(), 1);
        List<Supplier<PageProjectionWithOutputs>> pageProjectionsNoCSE = functionCompiler.compileProjections(SESSION.getSqlFunctionProperties(), ImmutableList.of(ADD_X_Y, ADD_X_Y_Z), false, Optional.empty());
        assertEquals(pageProjectionsNoCSE.size(), 2);

        Page input = createLongBlockPage(3, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
        List<Block> cseResult = project(pageProjectionsCSE.get(0).get().getPageProjection(), input, SelectedPositions.positionsRange(0, input.getPositionCount()));
        assertEquals(cseResult.size(), 2);
        List<Block> noCseResult1 = project(pageProjectionsNoCSE.get(0).get().getPageProjection(), input, SelectedPositions.positionsRange(0, input.getPositionCount()));
        assertEquals(noCseResult1.size(), 1);
        List<Block> noCseResult2 = project(pageProjectionsNoCSE.get(1).get().getPageProjection(), input, SelectedPositions.positionsRange(0, input.getPositionCount()));
        assertEquals(noCseResult2.size(), 1);
        checkBlockEqual(cseResult.get(0), noCseResult1.get(0));
        checkBlockEqual(cseResult.get(1), noCseResult2.get(0));
    }