class SqlStreamOptimizerComponentImpl implements SqlStreamOptimizerComponent {

    private static final Logger LOGGER_STREAM_OPTIMIZER = LoggerManager.getLogger(ApplicationBuilder.LogType.STREAM_OPTIMIZER.getLoggerName());

    private static final SqlStreamOptimizer<?> FALL_BACK = new FallbackStreamOptimizer<>();

    private final List<SqlStreamOptimizer<?>> optimizers;

    public SqlStreamOptimizerComponentImpl() {
        this.optimizers = new CopyOnWriteArrayList<>();
        install(new InitialFilterOptimizer<>());
        install(new FilterSortedSkipOptimizer<>());
    }

    @Override
    public <ENTITY> SqlStreamOptimizer<ENTITY> get(Pipeline initialPipeline, DbmsType dbmsType) {<FILL_FUNCTION_BODY>}

    private static final Comparator<Metrics> METRICS_COMPARATOR
        = comparingInt(Metrics::getPipelineReductions)
            .thenComparing(comparingInt(Metrics::getSqlCount).reversed());

    private <ENTITY> SqlStreamOptimizer<ENTITY> getHelper(Pipeline initialPipeline, DbmsType dbmsType) {
        @SuppressWarnings("unchecked")
        SqlStreamOptimizer<ENTITY> result = (SqlStreamOptimizer<ENTITY>) FALL_BACK;
        if (initialPipeline.isEmpty()) {
            return result;
        }

        Metrics metric = Metrics.empty();
        for (int i = optimizers.size() - 1; i >= 0; i--) {
            @SuppressWarnings("unchecked")
            final SqlStreamOptimizer<ENTITY> candidate = (SqlStreamOptimizer<ENTITY>) optimizers.get(i);
            final Metrics candidateMetric = candidate.metrics(initialPipeline, dbmsType);
            if (DEBUG.isEqualOrHigherThan(LOGGER_STREAM_OPTIMIZER.getLevel())) {
                LOGGER_STREAM_OPTIMIZER.debug("Candidate: %-30s : %s ", candidate.getClass().getSimpleName(), candidateMetric);
            }
            if (METRICS_COMPARATOR.compare(candidateMetric, metric) > 0) {
                metric = candidateMetric;
                result = candidate;
                if (metric.getPipelineReductions() == Integer.MAX_VALUE) {
                    return result;
                }
            }
        }
        return result;
    }

    @Override
    public <ENTITY> void install(SqlStreamOptimizer<ENTITY> sqlStreamOptimizer) {
        requireNonNull(sqlStreamOptimizer);
        optimizers.add(sqlStreamOptimizer);
    }

    @Override
    public Stream<SqlStreamOptimizer<?>> stream() {
        return optimizers.stream();
    }

    private static class FallbackStreamOptimizer<ENTITY> implements SqlStreamOptimizer<ENTITY> {

        @Override
        public <P extends Pipeline> Metrics metrics(P initialPipeline, DbmsType dbmsType) {
            return Metrics.empty();
        }

        @Override
        public <P extends Pipeline> P optimize(P initialPipeline, SqlStreamOptimizerInfo<ENTITY> info, AsynchronousQueryResult<ENTITY> query) {
            return initialPipeline;
        }

    }

}

        if (DEBUG.isEqualOrHigherThan(LOGGER_STREAM_OPTIMIZER.getLevel())) {
            LOGGER_STREAM_OPTIMIZER.debug("Evaluating %s pipeline: %s", initialPipeline.isParallel() ? "parallel" : "sequential", initialPipeline.toString());
        }
        final SqlStreamOptimizer<ENTITY> result = getHelper(initialPipeline, dbmsType);
        if (DEBUG.isEqualOrHigherThan(LOGGER_STREAM_OPTIMIZER.getLevel())) {
            LOGGER_STREAM_OPTIMIZER.debug("Selected: %s", result.getClass().getSimpleName());
        }
        return result;
    
<no_super_class>