<commit_before>/*
 * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.hazelcast.map.impl.query;

import com.hazelcast.core.HazelcastException;
import com.hazelcast.core.Member;
import com.hazelcast.internal.cluster.ClusterService;
import com.hazelcast.logging.ILogger;
import com.hazelcast.map.QueryResultSizeExceededException;
import com.hazelcast.map.impl.MapService;
import com.hazelcast.map.impl.MapServiceContext;
import com.hazelcast.query.PagingPredicate;
import com.hazelcast.query.Predicate;
import com.hazelcast.query.QueryException;
import com.hazelcast.query.TruePredicate;
import com.hazelcast.spi.NodeEngine;
import com.hazelcast.spi.Operation;
import com.hazelcast.spi.OperationService;
import com.hazelcast.spi.partition.IPartitionService;
import com.hazelcast.util.BitSetUtils;
import com.hazelcast.util.IterationType;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Future;

import static com.hazelcast.cluster.memberselector.MemberSelectors.DATA_MEMBER_SELECTOR;
import static com.hazelcast.util.BitSetUtils.hasAllBitsSet;
import static com.hazelcast.util.ExceptionUtil.rethrow;
import static java.util.Collections.singletonList;

/**
 * Invokes and orchestrates the query logic returning the final result.
 * <p>
 * Knows nothing about query logic.
 * Should never invoke any query operations directly.
 * <p>
 * Should be used from top-level proxy-layer only (e.g. MapProxy, etc.).
 * <p>
 * Top level-query actors:
 * - QueryEngine orchestrates the queries and  merging the result
 * - QueryRunner -> runs the query logic in the calling thread (so like evaluates the predicates and asks the index)
 */
public class QueryEngineImpl implements QueryEngine {

    private final MapServiceContext mapServiceContext;
    private final NodeEngine nodeEngine;
    private final ILogger logger;
    private final QueryResultSizeLimiter queryResultSizeLimiter;
    private final IPartitionService partitionService;
    private final OperationService operationService;
    private final ClusterService clusterService;
    private final ResultProcessorRegistry resultProcessorRegistry;

    public QueryEngineImpl(MapServiceContext mapServiceContext) {
        this.mapServiceContext = mapServiceContext;
        this.nodeEngine = mapServiceContext.getNodeEngine();
        this.partitionService = nodeEngine.getPartitionService();
        this.logger = nodeEngine.getLogger(getClass());
        this.queryResultSizeLimiter = new QueryResultSizeLimiter(mapServiceContext, logger);
        this.operationService = nodeEngine.getOperationService();
        this.clusterService = nodeEngine.getClusterService();
        this.resultProcessorRegistry = mapServiceContext.getResultProcessorRegistry();
    }

    @Override
    public Result execute(Query query, Target target) {
        Query adjustedQuery = adjustQuery(query);
        if (target.isTargetAllNodes()) {
            return runQueryOnAllPartitions(adjustedQuery);
        } else if (target.isTargetLocalNode()) {
            return runQueryOnLocalPartitions(adjustedQuery);
        } else if (target.isTargetPartitionOwner()) {
            return runQueryOnGivenPartition(adjustedQuery, target);
        }
        throw new IllegalArgumentException("Illegal target " + query);
    }

    private Query adjustQuery(Query query) {
        IterationType retrievalIterationType = getRetrievalIterationType(query.getPredicate(), query.getIterationType());
        Query adjustedQuery = Query.of(query).iterationType(retrievalIterationType).build();
        if (adjustedQuery.getPredicate() instanceof PagingPredicate) {
            ((PagingPredicate) adjustedQuery.getPredicate()).setIterationType(query.getIterationType());
        } else {
            if (adjustedQuery.getPredicate() == TruePredicate.INSTANCE) {
                queryResultSizeLimiter.precheckMaxResultLimitOnLocalPartitions(adjustedQuery.getMapName());
            }
        }
        return adjustedQuery;
    }

    // query thread first, fallback to partition thread
    private Result runQueryOnLocalPartitions(Query query) {
        BitSet mutablePartitionIds = getLocalPartitionIds();

        Result result = doRunQueryOnQueryThreads(query, mutablePartitionIds, Target.LOCAL_NODE);
        if (isResultFromAnyPartitionMissing(mutablePartitionIds)) {
            doRunQueryOnPartitionThreads(query, mutablePartitionIds, result);
        }
        assertAllPartitionsQueried(mutablePartitionIds);

        return result;
    }

    // query thread first, fallback to partition thread
    private Result runQueryOnAllPartitions(Query query) {
        BitSet mutablePartitionIds = getAllPartitionIds();

        Result result = doRunQueryOnQueryThreads(query, mutablePartitionIds, Target.ALL_NODES);
        if (isResultFromAnyPartitionMissing(mutablePartitionIds)) {
            doRunQueryOnPartitionThreads(query, mutablePartitionIds, result);
        }
        assertAllPartitionsQueried(mutablePartitionIds);

        return result;
    }

    // partition thread ONLY (for now)
    private Result runQueryOnGivenPartition(Query query, Target target) {
        try {
            return dispatchPartitionScanQueryOnOwnerMemberOnPartitionThread(
                    query, target.getPartitionId()).get();
        } catch (Throwable t) {
            throw rethrow(t);
        }
    }

    private Result doRunQueryOnQueryThreads(Query query, BitSet partitionIds, Target target) {
        Result result = populateResult(query, partitionIds);
        List<Future<Result>> futures = dispatchOnQueryThreads(query, target);
        addResultsOfPredicate(futures, result, partitionIds, false);
        return result;
    }

    private List<Future<Result>> dispatchOnQueryThreads(Query query, Target target) {
        try {
            return dispatchFullQueryOnQueryThread(query, target);
        } catch (Throwable t) {
            if (!(t instanceof HazelcastException)) {
                // these are programmatic errors that needs to be visible
                throw rethrow(t);
            } else if (t.getCause() instanceof QueryResultSizeExceededException) {
                throw rethrow(t);
            } else {
                // log failure to invoke query on member at fine level
                // the missing partition IDs will be queried anyway, so it's not a terminal failure
                if (logger.isFineEnabled()) {
                    logger.fine("Query invocation failed on member ", t);
                }
            }
        }
        return Collections.emptyList();
    }

    private Result populateResult(Query query, BitSet partitionIds) {
        return resultProcessorRegistry.get(query.getResultType()).populateResult(query,
                queryResultSizeLimiter.getNodeResultLimit(partitionIds.cardinality()));
    }

    private void doRunQueryOnPartitionThreads(Query query, BitSet partitionIds, Result result) {
        try {
            List<Future<Result>> futures = dispatchPartitionScanQueryOnOwnerMemberOnPartitionThread(
                    query, partitionIds);
            addResultsOfPredicate(futures, result, partitionIds, true);
        } catch (Throwable t) {
            throw rethrow(t);
        }
    }

    @SuppressWarnings("unchecked")
    // modifies partitionIds list! Optimization not to allocate an extra collection with collected partitionIds
    private void addResultsOfPredicate(List<Future<Result>> futures, Result result,
                                       BitSet finishedPartitionIds, boolean rethrowAll) {
        for (Future<Result> future : futures) {
            Result queryResult = null;

            try {
                queryResult = future.get();
            } catch (Throwable t) {
                if (t.getCause() instanceof QueryResultSizeExceededException || rethrowAll) {
                    throw rethrow(t);
                }<commit_msg>Minor QueryEngineImpl method renames<commit_after>/*
 * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.hazelcast.map.impl.query;

import com.hazelcast.core.HazelcastException;
import com.hazelcast.core.Member;
import com.hazelcast.internal.cluster.ClusterService;
import com.hazelcast.logging.ILogger;
import com.hazelcast.map.QueryResultSizeExceededException;
import com.hazelcast.map.impl.MapService;
import com.hazelcast.map.impl.MapServiceContext;
import com.hazelcast.query.PagingPredicate;
import com.hazelcast.query.Predicate;
import com.hazelcast.query.QueryException;
import com.hazelcast.query.TruePredicate;
import com.hazelcast.spi.NodeEngine;
import com.hazelcast.spi.Operation;
import com.hazelcast.spi.OperationService;
import com.hazelcast.spi.partition.IPartitionService;
import com.hazelcast.util.BitSetUtils;
import com.hazelcast.util.IterationType;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Future;

import static com.hazelcast.cluster.memberselector.MemberSelectors.DATA_MEMBER_SELECTOR;
import static com.hazelcast.util.BitSetUtils.hasAllBitsSet;
import static com.hazelcast.util.ExceptionUtil.rethrow;
import static java.util.Collections.singletonList;

/**
 * Invokes and orchestrates the query logic returning the final result.
 * <p>
 * Knows nothing about query logic.
 * Should never invoke any query operations directly.
 * <p>
 * Should be used from top-level proxy-layer only (e.g. MapProxy, etc.).
 * <p>
 * Top level-query actors:
 * - QueryEngine orchestrates the queries and  merging the result
 * - QueryRunner -> runs the query logic in the calling thread (so like evaluates the predicates and asks the index)
 */
public class QueryEngineImpl implements QueryEngine {

    private final MapServiceContext mapServiceContext;
    private final NodeEngine nodeEngine;
    private final ILogger logger;
    private final QueryResultSizeLimiter queryResultSizeLimiter;
    private final IPartitionService partitionService;
    private final OperationService operationService;
    private final ClusterService clusterService;
    private final ResultProcessorRegistry resultProcessorRegistry;

    public QueryEngineImpl(MapServiceContext mapServiceContext) {
        this.mapServiceContext = mapServiceContext;
        this.nodeEngine = mapServiceContext.getNodeEngine();
        this.partitionService = nodeEngine.getPartitionService();
        this.logger = nodeEngine.getLogger(getClass());
        this.queryResultSizeLimiter = new QueryResultSizeLimiter(mapServiceContext, logger);
        this.operationService = nodeEngine.getOperationService();
        this.clusterService = nodeEngine.getClusterService();
        this.resultProcessorRegistry = mapServiceContext.getResultProcessorRegistry();
    }

    @Override
    public Result execute(Query query, Target target) {
        Query adjustedQuery = adjustQuery(query);
        if (target.isTargetAllNodes()) {
            return runOnAllPartitions(adjustedQuery);
        } else if (target.isTargetLocalNode()) {
            return runOnLocalPartitions(adjustedQuery);
        } else if (target.isTargetPartitionOwner()) {
            return runOnGivenPartition(adjustedQuery, target);
        }
        throw new IllegalArgumentException("Illegal target " + query);
    }

    private Query adjustQuery(Query query) {
        IterationType retrievalIterationType = getRetrievalIterationType(query.getPredicate(), query.getIterationType());
        Query adjustedQuery = Query.of(query).iterationType(retrievalIterationType).build();
        if (adjustedQuery.getPredicate() instanceof PagingPredicate) {
            ((PagingPredicate) adjustedQuery.getPredicate()).setIterationType(query.getIterationType());
        } else {
            if (adjustedQuery.getPredicate() == TruePredicate.INSTANCE) {
                queryResultSizeLimiter.precheckMaxResultLimitOnLocalPartitions(adjustedQuery.getMapName());
            }
        }
        return adjustedQuery;
    }

    // query thread first, fallback to partition thread
    private Result runOnLocalPartitions(Query query) {
        BitSet mutablePartitionIds = getLocalPartitionIds();

        Result result = doRunOnQueryThreads(query, mutablePartitionIds, Target.LOCAL_NODE);
        if (isResultFromAnyPartitionMissing(mutablePartitionIds)) {
            doRunOnPartitionThreads(query, mutablePartitionIds, result);
        }
        assertAllPartitionsQueried(mutablePartitionIds);

        return result;
    }

    // query thread first, fallback to partition thread
    private Result runOnAllPartitions(Query query) {
        BitSet mutablePartitionIds = getAllPartitionIds();

        Result result = doRunOnQueryThreads(query, mutablePartitionIds, Target.ALL_NODES);
        if (isResultFromAnyPartitionMissing(mutablePartitionIds)) {
            doRunOnPartitionThreads(query, mutablePartitionIds, result);
        }
        assertAllPartitionsQueried(mutablePartitionIds);

        return result;
    }

    // partition thread ONLY (for now)
    private Result runOnGivenPartition(Query query, Target target) {
        try {
            return dispatchPartitionScanQueryOnOwnerMemberOnPartitionThread(
                    query, target.getPartitionId()).get();
        } catch (Throwable t) {
            throw rethrow(t);
        }
    }

    private Result doRunOnQueryThreads(Query query, BitSet partitionIds, Target target) {
        Result result = populateResult(query, partitionIds);
        List<Future<Result>> futures = dispatchOnQueryThreads(query, target);
        addResultsOfPredicate(futures, result, partitionIds, false);
        return result;
    }

    private List<Future<Result>> dispatchOnQueryThreads(Query query, Target target) {
        try {
            return dispatchFullQueryOnQueryThread(query, target);
        } catch (Throwable t) {
            if (!(t instanceof HazelcastException)) {
                // these are programmatic errors that needs to be visible
                throw rethrow(t);
            } else if (t.getCause() instanceof QueryResultSizeExceededException) {
                throw rethrow(t);
            } else {
                // log failure to invoke query on member at fine level
                // the missing partition IDs will be queried anyway, so it's not a terminal failure
                if (logger.isFineEnabled()) {
                    logger.fine("Query invocation failed on member ", t);
                }
            }
        }
        return Collections.emptyList();
    }

    private Result populateResult(Query query, BitSet partitionIds) {
        return resultProcessorRegistry.get(query.getResultType()).populateResult(query,
                queryResultSizeLimiter.getNodeResultLimit(partitionIds.cardinality()));
    }

    private void doRunOnPartitionThreads(Query query, BitSet partitionIds, Result result) {
        try {
            List<Future<Result>> futures = dispatchPartitionScanQueryOnOwnerMemberOnPartitionThread(
                    query, partitionIds);
            addResultsOfPredicate(futures, result, partitionIds, true);
        } catch (Throwable t) {
            throw rethrow(t);
        }
    }

    @SuppressWarnings("unchecked")
    // modifies partitionIds list! Optimization not to allocate an extra collection with collected partitionIds
    private void addResultsOfPredicate(List<Future<Result>> futures, Result result,
                                       BitSet finishedPartitionIds, boolean rethrowAll) {
        for (Future<Result> future : futures) {
            Result queryResult = null;

            try {
                queryResult = future.get();
            } catch (Throwable t) {
                if (t.getCause() instanceof QueryResultSizeExceededException || rethrowAll) {
                    throw rethrow(t);
                }
{'fraction_non_alphanumeric': 0.07842219230994323, 'fraction_numerical': 0.0014045765786855505, 'mean_word_length': 3.2571001494768312, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2372310', 'n_tokens_mistral': 4430, 'n_tokens_neox': 4312, 'n_words': 1135}