<commit_before>package com.jivesoftware.os.miru.plugin.solution;

import com.google.common.base.Optional;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.jivesoftware.os.filer.io.api.KeyRange;
import com.jivesoftware.os.filer.io.api.StackBuffer;
import com.jivesoftware.os.jive.utils.base.interfaces.CallbackStream;
import com.jivesoftware.os.miru.api.MiruPartitionCoord;
import com.jivesoftware.os.miru.api.activity.schema.MiruFieldDefinition;
import com.jivesoftware.os.miru.api.activity.schema.MiruSchema;
import com.jivesoftware.os.miru.api.base.MiruTermId;
import com.jivesoftware.os.miru.api.field.MiruFieldType;
import com.jivesoftware.os.miru.api.query.filter.MiruFieldFilter;
import com.jivesoftware.os.miru.api.query.filter.MiruFilter;
import com.jivesoftware.os.miru.api.query.filter.MiruFilterOperation;
import com.jivesoftware.os.miru.api.query.filter.MiruValue;
import com.jivesoftware.os.miru.api.wal.MiruSipCursor;
import com.jivesoftware.os.miru.plugin.bitmap.MiruBitmaps;
import com.jivesoftware.os.miru.plugin.bitmap.MiruIntIterator;
import com.jivesoftware.os.miru.plugin.context.MiruRequestContext;
import com.jivesoftware.os.miru.plugin.index.FieldMultiTermTxIndex;
import com.jivesoftware.os.miru.plugin.index.LastIdAndTermIdStream;
import com.jivesoftware.os.miru.plugin.index.MiruActivityIndex;
import com.jivesoftware.os.miru.plugin.index.MiruFieldIndex;
import com.jivesoftware.os.miru.plugin.index.MiruFieldIndexProvider;
import com.jivesoftware.os.miru.plugin.index.MiruTermComposer;
import com.jivesoftware.os.miru.plugin.index.MiruTxIndex;
import com.jivesoftware.os.miru.plugin.partition.TrackError;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang.mutable.MutableInt;
import org.apache.commons.lang.mutable.MutableLong;

/**
 *
 */
public class MiruAggregateUtil {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    public interface StreamBitmaps<BM> {

        boolean stream(int streamIndex, int lastId, MiruTermId termId, int scoredToLastId, BM[] answers, int fromId, Set<Feature>[] features) throws Exception;
    }

    public interface ConsumeBitmaps<BM> {

        boolean consume(StreamBitmaps<BM> streamBitmaps) throws Exception;
    }

    public <BM extends IBM, IBM, S extends MiruSipCursor<S>> void gatherFeatures(String name,
        MiruBitmaps<BM, IBM> bitmaps,
        MiruRequestContext<BM, IBM, S> requestContext,
        ConsumeBitmaps<BM> consumeAnswers,
        int[][] featureFieldIds,
        FeatureStream stream,
        MiruSolutionLog solutionLog,
        StackBuffer stackBuffer) throws Exception {

        Set<Integer> uniqueFieldIds = Sets.newHashSet();
        for (int i = 0; i < featureFieldIds.length; i++) {
            if (featureFieldIds[i] != null) {
                for (int j = 0; j < featureFieldIds[i].length; j++) {
                    uniqueFieldIds.add(featureFieldIds[i][j]);
                }
            }
        }

        MiruSchema schema = requestContext.getSchema();
        MiruActivityIndex activityIndex = requestContext.getActivityIndex();

        MiruTermId[][][] fieldTerms = new MiruTermId[schema.fieldCount()][][];

        /*@SuppressWarnings("unchecked")
        Set<Feature>[] features = new Set[featureFieldIds.length];
        if (dedupe) {
            for (int i = 0; i < features.length; i++) {
                features[i] = Sets.newHashSet();
            }
        }*/

        int[] featureCount = { 0 };
        int[] termCount = { 0 };
        int batchSize = 1_000;
        boolean[][] featuresContained = new boolean[batchSize][featureFieldIds.length];
        int[] ids = new int[batchSize];
        long start = System.currentTimeMillis();
        int[] skippedCount = { 0 };
        int[] consumedCount = { 0 };
        int[] minFromId = { Integer.MAX_VALUE };
        int[] maxFromId = { Integer.MIN_VALUE };
        consumeAnswers.consume((streamIndex, lastId, answerTermId, answerScoredLastId, answerBitmaps, fromId, features) -> {
            minFromId[0] = Math.min(minFromId[0], fromId);
            maxFromId[0] = Math.max(maxFromId[0], fromId);

            termCount[0]++;
            /*for (Set<Feature> featureSet : features) {
                if (featureSet != null) {
                    featureSet.clear();
                }
            }*/
            MiruIntIterator[] intIters = new MiruIntIterator[answerBitmaps.length];
            for (int i = 0; i < intIters.length; i++) {
                if (answerBitmaps[i] != null) {
                    intIters[i] = bitmaps.intIterator(answerBitmaps[i]);
                }
            }

            CollatingIntIterator iter = new CollatingIntIterator(intIters, false);
            int count = 0;
            while (iter.hasNext()) {
                ids[count] = iter.next(featuresContained[count]);
                if (ids[count] < fromId) {
                    skippedCount[0]++;
                    continue;
                }
                consumedCount[0]++;
                count++;
                if (count == batchSize) {
                    if (!gatherAndStreamFeatures(name, streamIndex, featureFieldIds, stream, stackBuffer, uniqueFieldIds, activityIndex, fieldTerms,
                        ids, featuresContained, count, lastId, answerTermId, answerScoredLastId, featureCount, features)) {
                        return false;
                    }
                    count = 0;
                }
            }
            if (count > 0) {
                if (!gatherAndStreamFeatures(name, streamIndex, featureFieldIds, stream, stackBuffer, uniqueFieldIds, activityIndex, fieldTerms,
                    ids, featuresContained, count, lastId, answerTermId, answerScoredLastId, featureCount, features)) {
                    return false;
                }
                count = 0;
            }
            return stream.stream(streamIndex, lastId, answerTermId, answerScoredLastId, -1, null);
        });
        LOG.info("Gathered {} features for {} terms in {} ms - skipped {} consumed {} fromId {}/{}",
            featureCount[0], termCount[0], System.currentTimeMillis() - start, skippedCount[0], consumedCount[0], minFromId[0], maxFromId[0]);
        solutionLog.log(MiruSolutionLogLevel.INFO, "Gathered {} features for {} terms in {} ms - skipped {} consumed {} fromId {}/{}",
            featureCount[0], termCount[0], System.currentTimeMillis() - start, skippedCount[0], consumedCount[0], minFromId[0], maxFromId[0]);
    }

    private boolean gatherAndStreamFeatures(String name,
        int streamIndex,
        int[][] featureFieldIds,
        FeatureStream stream,
        StackBuffer stackBuffer,
        Set<Integer> uniqueFieldIds,
        MiruActivityIndex activityIndex,
        MiruTermId[][][] fieldTerms,
        int[] ids,
        boolean[][] contained,
        int count,
        int lastId,
        MiruTermId answerTermId,<commit_msg>include name in feature gather logging<commit_after>package com.jivesoftware.os.miru.plugin.solution;

import com.google.common.base.Optional;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.jivesoftware.os.filer.io.api.KeyRange;
import com.jivesoftware.os.filer.io.api.StackBuffer;
import com.jivesoftware.os.jive.utils.base.interfaces.CallbackStream;
import com.jivesoftware.os.miru.api.MiruPartitionCoord;
import com.jivesoftware.os.miru.api.activity.schema.MiruFieldDefinition;
import com.jivesoftware.os.miru.api.activity.schema.MiruSchema;
import com.jivesoftware.os.miru.api.base.MiruTermId;
import com.jivesoftware.os.miru.api.field.MiruFieldType;
import com.jivesoftware.os.miru.api.query.filter.MiruFieldFilter;
import com.jivesoftware.os.miru.api.query.filter.MiruFilter;
import com.jivesoftware.os.miru.api.query.filter.MiruFilterOperation;
import com.jivesoftware.os.miru.api.query.filter.MiruValue;
import com.jivesoftware.os.miru.api.wal.MiruSipCursor;
import com.jivesoftware.os.miru.plugin.bitmap.MiruBitmaps;
import com.jivesoftware.os.miru.plugin.bitmap.MiruIntIterator;
import com.jivesoftware.os.miru.plugin.context.MiruRequestContext;
import com.jivesoftware.os.miru.plugin.index.FieldMultiTermTxIndex;
import com.jivesoftware.os.miru.plugin.index.LastIdAndTermIdStream;
import com.jivesoftware.os.miru.plugin.index.MiruActivityIndex;
import com.jivesoftware.os.miru.plugin.index.MiruFieldIndex;
import com.jivesoftware.os.miru.plugin.index.MiruFieldIndexProvider;
import com.jivesoftware.os.miru.plugin.index.MiruTermComposer;
import com.jivesoftware.os.miru.plugin.index.MiruTxIndex;
import com.jivesoftware.os.miru.plugin.partition.TrackError;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.lang.mutable.MutableInt;
import org.apache.commons.lang.mutable.MutableLong;

/**
 *
 */
public class MiruAggregateUtil {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    public interface StreamBitmaps<BM> {

        boolean stream(int streamIndex, int lastId, MiruTermId termId, int scoredToLastId, BM[] answers, int fromId, Set<Feature>[] features) throws Exception;
    }

    public interface ConsumeBitmaps<BM> {

        boolean consume(StreamBitmaps<BM> streamBitmaps) throws Exception;
    }

    public <BM extends IBM, IBM, S extends MiruSipCursor<S>> void gatherFeatures(String name,
        MiruBitmaps<BM, IBM> bitmaps,
        MiruRequestContext<BM, IBM, S> requestContext,
        ConsumeBitmaps<BM> consumeAnswers,
        int[][] featureFieldIds,
        FeatureStream stream,
        MiruSolutionLog solutionLog,
        StackBuffer stackBuffer) throws Exception {

        Set<Integer> uniqueFieldIds = Sets.newHashSet();
        for (int i = 0; i < featureFieldIds.length; i++) {
            if (featureFieldIds[i] != null) {
                for (int j = 0; j < featureFieldIds[i].length; j++) {
                    uniqueFieldIds.add(featureFieldIds[i][j]);
                }
            }
        }

        MiruSchema schema = requestContext.getSchema();
        MiruActivityIndex activityIndex = requestContext.getActivityIndex();

        MiruTermId[][][] fieldTerms = new MiruTermId[schema.fieldCount()][][];

        /*@SuppressWarnings("unchecked")
        Set<Feature>[] features = new Set[featureFieldIds.length];
        if (dedupe) {
            for (int i = 0; i < features.length; i++) {
                features[i] = Sets.newHashSet();
            }
        }*/

        int[] featureCount = { 0 };
        int[] termCount = { 0 };
        int batchSize = 1_000;
        boolean[][] featuresContained = new boolean[batchSize][featureFieldIds.length];
        int[] ids = new int[batchSize];
        long start = System.currentTimeMillis();
        int[] skippedCount = { 0 };
        int[] consumedCount = { 0 };
        int[] minFromId = { Integer.MAX_VALUE };
        int[] maxFromId = { Integer.MIN_VALUE };
        consumeAnswers.consume((streamIndex, lastId, answerTermId, answerScoredLastId, answerBitmaps, fromId, features) -> {
            minFromId[0] = Math.min(minFromId[0], fromId);
            maxFromId[0] = Math.max(maxFromId[0], fromId);

            termCount[0]++;
            /*for (Set<Feature> featureSet : features) {
                if (featureSet != null) {
                    featureSet.clear();
                }
            }*/
            MiruIntIterator[] intIters = new MiruIntIterator[answerBitmaps.length];
            for (int i = 0; i < intIters.length; i++) {
                if (answerBitmaps[i] != null) {
                    intIters[i] = bitmaps.intIterator(answerBitmaps[i]);
                }
            }

            CollatingIntIterator iter = new CollatingIntIterator(intIters, false);
            int count = 0;
            while (iter.hasNext()) {
                ids[count] = iter.next(featuresContained[count]);
                if (ids[count] < fromId) {
                    skippedCount[0]++;
                    continue;
                }
                consumedCount[0]++;
                count++;
                if (count == batchSize) {
                    if (!gatherAndStreamFeatures(name, streamIndex, featureFieldIds, stream, stackBuffer, uniqueFieldIds, activityIndex, fieldTerms,
                        ids, featuresContained, count, lastId, answerTermId, answerScoredLastId, featureCount, features)) {
                        return false;
                    }
                    count = 0;
                }
            }
            if (count > 0) {
                if (!gatherAndStreamFeatures(name, streamIndex, featureFieldIds, stream, stackBuffer, uniqueFieldIds, activityIndex, fieldTerms,
                    ids, featuresContained, count, lastId, answerTermId, answerScoredLastId, featureCount, features)) {
                    return false;
                }
                count = 0;
            }
            return stream.stream(streamIndex, lastId, answerTermId, answerScoredLastId, -1, null);
        });
        LOG.info("Gathered name:{} features:{} terms:{} elapsed:{} - skipped:{} consumed:{} fromId:{}/{}",
            name, featureCount[0], termCount[0], System.currentTimeMillis() - start, skippedCount[0], consumedCount[0], minFromId[0], maxFromId[0]);
        solutionLog.log(MiruSolutionLogLevel.INFO, "Gathered name:{} features:{} terms:{} elapsed:{} - skipped:{} consumed:{} fromId:{}/{}",
            name, featureCount[0], termCount[0], System.currentTimeMillis() - start, skippedCount[0], consumedCount[0], minFromId[0], maxFromId[0]);
    }

    private boolean gatherAndStreamFeatures(String name,
        int streamIndex,
        int[][] featureFieldIds,
        FeatureStream stream,
        StackBuffer stackBuffer,
        Set<Integer> uniqueFieldIds,
        MiruActivityIndex activityIndex,
        MiruTermId[][][] fieldTerms,
        int[] ids,
        boolean[][] contained,
        int count,
        int lastId,
        MiruTermId answerTermId,
{'fraction_non_alphanumeric': 0.1154562383612663, 'fraction_numerical': 0.004965859714463066, 'mean_word_length': 2.7448347107438016, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 33, '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': '6004367', 'n_tokens_mistral': 4388, 'n_tokens_neox': 4281, 'n_words': 802}