<commit_before>package com.airhacks.enhydrator;

import com.airhacks.enhydrator.flexpipe.JDBCPipeline;
import com.airhacks.enhydrator.in.Entry;
import com.airhacks.enhydrator.in.JDBCSource;
import com.airhacks.enhydrator.out.Sink;
import com.airhacks.enhydrator.transform.EntryTransformer;
import com.airhacks.enhydrator.transform.FunctionScriptLoader;
import com.airhacks.enhydrator.transform.ResultSetToEntries;
import com.airhacks.enhydrator.transform.RowTransformer;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 *
 * @author <EMAIL>
 */
public class Pump {

    private final JDBCSource source;
    private final Function<ResultSet, List<Entry>> rowTransformer;
    private final Map<String, Function<Entry, List<Entry>>> namedEntryFunctions;
    private final Map<Integer, Function<Entry, List<Entry>>> indexedEntryFunctions;
    private final Function<List<Entry>, List<Entry>> before;
    private final Function<List<Entry>, List<Entry>> after;

    private final Sink sink;

    private Pump(JDBCSource source, Function<ResultSet, List<Entry>> rowTransformer,
            Function<List<Entry>, List<Entry>> before,
            Map<String, Function<Entry, List<Entry>>> namedFunctions,
            Map<Integer, Function<Entry, List<Entry>>> indexedFunctions,
            Function<List<Entry>, List<Entry>> after,
            Sink sink) {
        this.source = source;
        this.before = before;
        this.rowTransformer = rowTransformer;
        this.namedEntryFunctions = namedFunctions;
        this.indexedEntryFunctions = indexedFunctions;
        this.after = after;
        this.sink = sink;
    }

    void process(String sql, Object... params) {
        Iterable<ResultSet> results = this.source.query(sql, params);
        this.sink.init();
        results.forEach(this::onNewRow);
        this.sink.close();
    }

    void onNewRow(ResultSet columns) {
        List<Entry> convertedColumns = this.rowTransformer.apply(columns);
        List<Entry> entryColumns = this.before.apply(convertedColumns);
        List<Entry> transformed = entryColumns.stream().
                map(e -> applyOrReturnOnIndexed(e)).
                flatMap(l -> l.stream()).
                map(e -> applyOrReturnOnNamed(e)).
                flatMap(l -> l.stream()).
                collect(Collectors.toList());
        List<Entry> afterProcessed = this.after.apply(transformed);
        this.sink.processRow(afterProcessed);

    }

    List<Entry> applyOrReturnOnIndexed(Entry e) {
        final Function<Entry, List<Entry>> function = this.indexedEntryFunctions.get(e.getSlot());
        if (function != null) {
            return function.apply(e);
        } else {
            return e.asList();
        }
    }

    List<Entry> applyOrReturnOnNamed(Entry e) {
        final Function<Entry, List<Entry>> function = this.namedEntryFunctions.get(e.getSlot());
        if (function != null) {
            return function.apply(e);
        } else {
            return e.asList();
        }
    }

    public static class Engine {

        private Sink sink;
        private JDBCSource source;
        private Function<ResultSet, List<Entry>> resultSetToEntries;
        private Map<String, Function<Entry, List<Entry>>> entryFunctions;
        private Map<Integer, Function<Entry, List<Entry>>> indexedFunctions;
        private Function<List<Entry>, List<Entry>> before;<commit_msg>Bug with named slots fixed<commit_after>package com.airhacks.enhydrator;

import com.airhacks.enhydrator.flexpipe.JDBCPipeline;
import com.airhacks.enhydrator.in.Entry;
import com.airhacks.enhydrator.in.JDBCSource;
import com.airhacks.enhydrator.out.Sink;
import com.airhacks.enhydrator.transform.EntryTransformer;
import com.airhacks.enhydrator.transform.FunctionScriptLoader;
import com.airhacks.enhydrator.transform.ResultSetToEntries;
import com.airhacks.enhydrator.transform.RowTransformer;
import java.sql.ResultSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 *
 * @author <EMAIL>hacks.com
 */
public class Pump {

    private final JDBCSource source;
    private final Function<ResultSet, List<Entry>> rowTransformer;
    private final Map<String, Function<Entry, List<Entry>>> namedEntryFunctions;
    private final Map<Integer, Function<Entry, List<Entry>>> indexedEntryFunctions;
    private final Function<List<Entry>, List<Entry>> before;
    private final Function<List<Entry>, List<Entry>> after;

    private final Sink sink;

    private Pump(JDBCSource source, Function<ResultSet, List<Entry>> rowTransformer,
            Function<List<Entry>, List<Entry>> before,
            Map<String, Function<Entry, List<Entry>>> namedFunctions,
            Map<Integer, Function<Entry, List<Entry>>> indexedFunctions,
            Function<List<Entry>, List<Entry>> after,
            Sink sink) {
        this.source = source;
        this.before = before;
        this.rowTransformer = rowTransformer;
        this.namedEntryFunctions = namedFunctions;
        this.indexedEntryFunctions = indexedFunctions;
        this.after = after;
        this.sink = sink;
    }

    void process(String sql, Object... params) {
        Iterable<ResultSet> results = this.source.query(sql, params);
        this.sink.init();
        results.forEach(this::onNewRow);
        this.sink.close();
    }

    void onNewRow(ResultSet columns) {
        List<Entry> convertedColumns = this.rowTransformer.apply(columns);
        List<Entry> entryColumns = this.before.apply(convertedColumns);
        List<Entry> transformed = entryColumns.stream().
                map(e -> applyOrReturnOnIndexed(e)).
                flatMap(l -> l.stream()).
                map(e -> applyOrReturnOnNamed(e)).
                flatMap(l -> l.stream()).
                collect(Collectors.toList());
        List<Entry> afterProcessed = this.after.apply(transformed);
        this.sink.processRow(afterProcessed);

    }

    List<Entry> applyOrReturnOnIndexed(Entry e) {
        final Function<Entry, List<Entry>> function = this.indexedEntryFunctions.get(e.getSlot());
        if (function != null) {
            return function.apply(e);
        } else {
            return e.asList();
        }
    }

    List<Entry> applyOrReturnOnNamed(Entry e) {
        final Function<Entry, List<Entry>> function = this.namedEntryFunctions.get(e.getName());
        if (function != null) {
            return function.apply(e);
        } else {
            return e.asList();
        }
    }

    public static class Engine {

        private Sink sink;
        private JDBCSource source;
        private Function<ResultSet, List<Entry>> resultSetToEntries;
        private Map<String, Function<Entry, List<Entry>>> entryFunctions;
        private Map<Integer, Function<Entry, List<Entry>>> indexedFunctions;
        private Function<List<Entry>, List<Entry>> before;
{'fraction_non_alphanumeric': 0.11377160668744687, 'fraction_numerical': 0.0, 'mean_word_length': 3.0945475638051043, 'pattern_counts': {'":': 0, '<': 105, '<?xml version=': 0, '>': 113, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5009499', 'n_tokens_mistral': 2041, 'n_tokens_neox': 1970, 'n_words': 409}