/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.ignite.yardstick.cache.load;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicLong;
import javax.cache.CacheException;
import javax.cache.configuration.FactoryBuilder;
import javax.cache.event.CacheEntryEvent;
import javax.cache.event.CacheEntryListenerException;
import javax.cache.event.CacheEntryUpdatedListener;
import javax.cache.processor.EntryProcessor;
import javax.cache.processor.MutableEntry;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCompute;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteException;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.cache.query.ContinuousQuery;
import org.apache.ignite.cache.query.Query;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import org.apache.ignite.yardstick.IgniteAbstractBenchmark;
import org.apache.ignite.yardstick.IgniteBenchmarkUtils;
import org.apache.ignite.yardstick.IgniteNode;
import org.apache.ignite.yardstick.cache.load.model.ModelUtil;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.BeansException;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.yardstickframework.BenchmarkConfiguration;
import org.yardstickframework.BenchmarkUtils;

/**
 * Ignite cache random operation benchmark.
 */
public class IgniteCacheRandomOperationBenchmark extends IgniteAbstractBenchmark {
    /** */
    public static final int operations = Operation.values().length;

    /***/
    public static final int CONTINUOUS_QUERY_PER_CACHE = 3;

    /** Scan query predicate. */
    private static BenchmarkIgniteBiPredicate igniteBiPred = new BenchmarkIgniteBiPredicate();

    /** Amount partitions. */
    private static final int SCAN_QUERY_PARTITION_AMOUNT = 10;

    /** List off all available cache. */
    private List<IgniteCache<Object, Object>> availableCaches;

    /** List of available transactional cache. */
    private List<IgniteCache<Object, Object>> txCaches;

    /** List of affinity cache. */
    private List<IgniteCache<Object, Object>> affCaches;

    /** Map cache name on key classes. */
    private Map<String, Class[]> keysCacheClasses;

    /** Map cache name on value classes. */
    private Map<String, Class[]> valuesCacheClasses;

    /** List of query descriptors by cache names. */
    private Map<String, List<SqlCacheDescriptor>> cacheSqlDescriptors;

    /** List of SQL queries. */
    private List<TestQuery> queries;

    /** List of allowed cache operations which will be executed. */
    private List<Operation> allowedLoadTestOps;

    /**
     * Replace value entry processor.
     */
    private BenchmarkReplaceValueEntryProcessor replaceEntryProc;

    /**
     * Remove entry processor.
     */
    private BenchmarkRemoveEntryProcessor rmvEntryProc;

    /**
     * Map of statistic information.
     */
    private Map<String, AtomicLong> operationStatistics;

    /** {@inheritDoc} */
    @Override public void setUp(BenchmarkConfiguration cfg) throws Exception {
        super.setUp(cfg);

        if (args.additionalCachesNumber() > 0)
            createAdditionalCaches();

        searchCache();

        preLoading();
    }

    /**
     * Creates additional caches.
     *
     * @throws Exception If failed.
     */
    private void createAdditionalCaches() throws Exception {
        Map<String, CacheConfiguration> cfgMap;

        try {
            // Loading spring application context and getting all of the caches configurations in the map.
            cfgMap = IgniteNode.loadConfiguration(args.configuration()).get2().getBeansOfType(CacheConfiguration.class);
        }
        catch (BeansException e) {
            throw new Exception("Failed to instantiate bean [type=" + CacheConfiguration.class + ", err=" +
                e.getMessage() + ']', e);
        }

        if (cfgMap == null || cfgMap.isEmpty())
            throw new Exception("Failed to find cache configurations in: " + args.configuration());

        // Getting cache configuration from the map using name specified in property file.
        CacheConfiguration<Object, Object> ccfg = cfgMap.get(args.additionalCachesName());

        if (ccfg == null)
            throw new Exception("Failed to find cache configuration [cache=" + args.additionalCachesName() +
                ", cfg=" + args.configuration() + ']');

        for (int i = 0; i < args.additionalCachesNumber(); i++) {
            CacheConfiguration<Object, Object> newCfg = new CacheConfiguration<>(ccfg);

            newCfg.setName("additional_" + args.additionalCachesName() + "_cache_" + i);

            try {
                ignite().createCache(newCfg);
            }
            catch (CacheException e) {
                BenchmarkUtils.error("Failed to create additional cache [ name = " + args.additionalCachesName() +
                    ", err" + e.getMessage() + ']', e);
            }
        }
    }

    /** {@inheritDoc} */
    @Override public void onException(Throwable e) {
        BenchmarkUtils.error("The benchmark of random operation failed.", e);

        super.onException(e);
    }

    /** {@inheritDoc} */
    @Override public boolean test(Map<Object, Object> map) throws Exception {
        if (nextBoolean()) {
            executeInTransaction(map);

            executeOutOfTx(map, true);
        }
        else
            executeOutOfTx(map, false);

        return true;
    }

    /** {@inheritDoc} */
    @Override public void tearDown() throws Exception {
        try {
            BenchmarkUtils.println("Benchmark statistics");

            for (String cacheName : ignite().cacheNames()) {
                BenchmarkUtils.println(String.format("Operations over cache '%s'", cacheName));

                for (Operation op : Operation.values()) {
                    BenchmarkUtils.println(cfg, String.format("%s: %s", op,
                        operationStatistics.get(op + "_" + cacheName)));
                }
            }
        }
        finally {
            super.tearDown();
        }
    }

    /**
     * @throws Exception If failed.
     */
    private void searchCache() throws Exception {
        availableCaches = new ArrayList<>(ignite().cacheNames().size());
        txCaches = new ArrayList<>();
        affCaches = new ArrayList<>();
        keysCacheClasses = new HashMap<>();
        valuesCacheClasses = new HashMap<>();
        replaceEntryProc = new BenchmarkReplaceValueEntryProcessor(null);
        rmvEntryProc = new BenchmarkRemoveEntryProcessor();
        cacheSqlDescriptors = new HashMap<>();
        operationStatistics = new HashMap<>();

        loadQueries();

        loadAllowedOperations();

        for (String cacheName : ignite().cacheNames()) {
            IgniteCache<Object, Object> cache = ignite().cache(cacheName);

            for (Operation op : Operation.values())
                operationStatistics.put(op + "_" + cacheName, new AtomicLong(0));

            CacheConfiguration configuration = cache.getConfiguration(CacheConfiguration.class);

            if (isClassDefinedInConfig(configuration)) {
                ArrayList<Class> keys = new ArrayList<>();
                ArrayList<Class> values = new ArrayList<>();

                if (configuration.getQueryEntities() != null) {
                    Collection<QueryEntity> entries = configuration.getQueryEntities();

                    for (QueryEntity queryEntity : entries) {
                        try {
                            if (queryEntity.getKeyType() != null) {
                                Class keyCls = Class.forName(queryEntity.getKeyType());

                                if (ModelUtil.canCreateInstance(keyCls))
                                    keys.add(keyCls);
                                else
                                    throw new IgniteException("Class is unknown for the load test. Make sure you " +
                                        "specified its full name [cache=" + cacheName + ", clsName=" + queryEntity.getKeyType() + ']');
                            }

                            if (queryEntity.getValueType() != null) {
                                Class valCls = Class.forName(queryEntity.getValueType());

                                if (ModelUtil.canCreateInstance(valCls))
                                    values.add(valCls);
                                else
                                    throw new IgniteException("Class is unknown for the load test. Make sure you " +
                                        "specified its full name [cache=" + cacheName + ", clsName=" + queryEntity.getValueType() + ']');

                                configureCacheSqlDescriptor(cacheName, queryEntity, valCls);
                            }
                        }
                        catch (ClassNotFoundException e) {
                            BenchmarkUtils.println(e.getMessage());
                            BenchmarkUtils.println("This can be a BinaryObject. Ignoring exception.");

                            if (!cacheSqlDescriptors.containsKey(cacheName))
                                cacheSqlDescriptors.put(cacheName, new ArrayList<SqlCacheDescriptor>());
                        }
                    }
                }

                if (configuration.getQueryEntities() != null) {
                    Collection<QueryEntity> entities = configuration.getQueryEntities();

                    for (QueryEntity entity : entities) {
                        try {
                            if (entity.getKeyType() != null) {
                                Class keyCls = Class.forName(entity.getKeyType());

                                if (ModelUtil.canCreateInstance(keyCls))
                                    keys.add(keyCls);
                                else
                                    throw new IgniteException("Class is unknown for the load test. Make sure you " +
                                        "specified its full name [clsName=" + entity.getKeyType() + ']');
                            }

                            if (entity.getValueType() != null) {
                                Class valCls = Class.forName(entity.getValueType());

                                if (ModelUtil.canCreateInstance(valCls))
                                    values.add(valCls);
                                else
                                    throw new IgniteException("Class is unknown for the load test. Make sure you " +
                                        "specified its full name [clsName=" + entity.getKeyType() + ']');
                            }
                        }
                        catch (ClassNotFoundException e) {
                            BenchmarkUtils.println(e.getMessage());
                            BenchmarkUtils.println("This can be a BinaryObject. Ignoring exception.");

                            if (!cacheSqlDescriptors.containsKey(cacheName))
                                cacheSqlDescriptors.put(cacheName, new ArrayList<SqlCacheDescriptor>());
                        }
                    }
                }

                keysCacheClasses.put(cacheName, keys.toArray(new Class[] {}));

                valuesCacheClasses.put(cacheName, values.toArray(new Class[] {}));
            }
            else
                keysCacheClasses.put(cacheName, new Class[] {randomKeyClass(cacheName)});

            valuesCacheClasses.put(cacheName, determineValueClasses(cacheName));

            affCaches.add(cache);

            if (configuration.getAtomicityMode() == CacheAtomicityMode.TRANSACTIONAL)
                txCaches.add(cache);

            availableCaches.add(cache);
        }
    }

    /**
     * Load allowed operation from parameters.
     */
    private void loadAllowedOperations() {
        allowedLoadTestOps = new ArrayList<>();

        if (args.allowedLoadTestOps().isEmpty())
            Collections.addAll(allowedLoadTestOps, Operation.values());
        else {
            for (String opName : args.allowedLoadTestOps())
                allowedLoadTestOps.add(Operation.valueOf(opName.toUpperCase()));
        }
    }

    /**
     * Load query from file.
     *
     * @throws Exception If fail.
     */
    private void loadQueries() throws Exception {
        queries = new ArrayList<>();

        if (args.loadTestQueriesFile() != null) {
            try (FileReader fr = new FileReader(args.loadTestQueriesFile())) {
                try (BufferedReader br = new BufferedReader(fr)) {
                    String line;
                    while ((line = br.readLine()) != null) {
                        if (line.trim().isEmpty())
                            continue;

                        boolean distributedJoins = false;

                        int commentIdx = line.lastIndexOf('#');

                        if (commentIdx >= 0) {
                            if (line.toUpperCase().indexOf("DISTRIBUTEDJOINS", commentIdx) > 0)
                                distributedJoins = true;

                            line = line.substring(0, commentIdx);
                        }

                        line = line.trim();

                        TestQuery qry = new TestQuery(line, distributedJoins);

                        queries.add(qry);

                        BenchmarkUtils.println("Loaded query: " + qry);
                    }
                }
            }
        }
    }

    /**
     * @param cacheName Ignite cache name.
     * @param qryEntity Query entry.
     * @param valCls Class of value.
     * @throws ClassNotFoundException If fail.
     */
    private void configureCacheSqlDescriptor(String cacheName, QueryEntity qryEntity, Class valCls)
        throws ClassNotFoundException {
        List<SqlCacheDescriptor> descs = cacheSqlDescriptors.get(cacheName);

        if (descs == null) {
            descs = new ArrayList<>();

            cacheSqlDescriptors.put(cacheName, descs);
        }

        Map<String, Class> indexedFields = new HashMap<>();

        for (QueryIndex index : qryEntity.getIndexes()) {
            for (String iField : index.getFieldNames()) {
                indexedFields.put(iField,
                    Class.forName(qryEntity.getFields().get(iField)));
            }
        }

        descs.add(new SqlCacheDescriptor(valCls, qryEntity.getFields().keySet(),
            indexedFields));
    }

    /**
     * @param configuration Ignite cache configuration.
     * @return True if defined.
     */
    private boolean isClassDefinedInConfig(CacheConfiguration configuration) {
        return (configuration.getIndexedTypes() != null && configuration.getIndexedTypes().length > 0)
            || !CollectionUtils.isEmpty(configuration.getQueryEntities());
    }

    /**
     * @throws Exception If fail.
     */
    private void preLoading() throws Exception {
        if (args.preloadAmount() > args.range())
            throw new IllegalArgumentException("Preloading amount (\"-pa\", \"--preloadAmount\") " +
                "must by less then the range (\"-r\", \"--range\").");

        startPreloadLogging(args.preloadLogsInterval());

        ExecutorService executor = Executors.newFixedThreadPool(10);

        try {
            List<Future<?>> futs = new ArrayList<>();

            final Thread thread = Thread.currentThread();

            for (int i = 0; i < availableCaches.size(); i++) {
                final String cacheName = availableCaches.get(i).getName();

                futs.add(executor.submit(new Runnable() {
                    @Override public void run() {
                        try (IgniteDataStreamer<Object, Object> dataLdr = ignite().dataStreamer(cacheName)) {
                            for (int i = 0; i < args.preloadAmount(); i++) {
                                if (i % 100 == 0 && thread.isInterrupted())
                                    break;

                                dataLdr.addData(createRandomKey(i, cacheName), createRandomValue(i, cacheName));
                            }
                        }
                    }
                }));
            }

            for (Future<?> fut : futs)
                fut.get();
        }
        finally {
            executor.shutdown();
        }

        stopPreloadLogging();
    }

    /**
     * Building a map that contains mapping of node ID to a list of partitions stored on the node.
     *
     * @param cacheName Name of Ignite cache.
     * @return Node to partitions map.
     */
    private Map<UUID, List<Integer>> personCachePartitions(String cacheName) {
        // Getting affinity for person cache.
        Affinity<Object> affinity = ignite().affinity(cacheName);

        // Building a list of all partitions numbers.
        List<Integer> rndParts = new ArrayList<>(10);

        if (affinity.partitions() <= SCAN_QUERY_PARTITION_AMOUNT)
            for (int i = 0; i < affinity.partitions(); i++)
                rndParts.add(i);
        else {
            for (int i = 0; i < SCAN_QUERY_PARTITION_AMOUNT; i++) {
                int partNum;

                do
                    partNum = nextRandom(affinity.partitions());
                while (rndParts.contains(partNum));

                rndParts.add(partNum);
            }
        }

        Collections.sort(rndParts);

        // Getting partition to node mapping.
        Map<Integer, ClusterNode> partPerNodes = affinity.mapPartitionsToNodes(rndParts);

        // Building node to partitions mapping.
        Map<UUID, List<Integer>> nodesToPart = new HashMap<>();

        for (Map.Entry<Integer, ClusterNode> entry : partPerNodes.entrySet()) {
            List<Integer> nodeParts = nodesToPart.get(entry.getValue().id());

            if (nodeParts == null) {
                nodeParts = new ArrayList<>();

                nodesToPart.put(entry.getValue().id(), nodeParts);
            }

            nodeParts.add(entry.getKey());
        }

        return nodesToPart;
    }

    /**
     * @param id Object identifier.
     * @param cacheName Name of Ignite cache.
     * @return Random object.
     */
    private Object createRandomKey(int id, String cacheName) {
        Class clazz = randomKeyClass(cacheName);

        return ModelUtil.create(clazz, id);
    }

    /**
     * @param cacheName Ignite cache name.
     * @return Random key class.
     */
    private Class randomKeyClass(String cacheName) {
        Class[] keys = keysCacheClasses.containsKey(cacheName)
            ? keysCacheClasses.get(cacheName) : ModelUtil.keyClasses();

        return keys[nextRandom(keys.length)];
    }

    /**
     * @param cacheName Cache name.
     * @return Set classes for cache.
     */
    private Class[] determineValueClasses(@NotNull String cacheName) {
        return cacheName.toLowerCase().contains("fat-values") ? ModelUtil.fatValueClasses() :
            ModelUtil.simpleValueClasses();
    }

    /**
     * @param id Object identifier.
     * @param cacheName Name of Ignite cache.
     * @return Random object.
     */
    private Object createRandomValue(int id, String cacheName) {
        Class clazz = randomValueClass(cacheName);

        return ModelUtil.create(clazz, id);
    }

    /**
     * @param cacheName Ignite cache name.
     * @return Random value class.
     */
    private Class randomValueClass(String cacheName) {
        Class[] values = valuesCacheClasses.containsKey(cacheName)
            ? valuesCacheClasses.get(cacheName) : ModelUtil.valueClasses();

        return values[nextRandom(values.length)];
    }

    /**
     * @param map Parameters map.
     * @param withoutTransactionCache Without transaction cache.
     * @throws Exception If fail.
     */
    private void executeOutOfTx(Map<Object, Object> map, boolean withoutTransactionCache) throws Exception {
        for (IgniteCache<Object, Object> cache : availableCaches) {
            if (withoutTransactionCache && txCaches.contains(cache))
                continue;

            executeRandomOperation(map, cache);
        }
    }

    /**
     * @param map Parameters map.
     * @param cache Ignite cache.
     * @throws Exception If fail.
     */
    private void executeRandomOperation(Map<Object, Object> map, IgniteCache<Object, Object> cache) throws Exception {
        Operation op = nextRandomOperation();

        try {
            switch (op) {
                case PUT:
                    doPut(cache);
                    break;

                case PUT_ALL:
                    doPutAll(cache);
                    break;

                case GET:
                    doGet(cache);
                    break;

                case GET_ALL:
                    doGetAll(cache);
                    break;

                case INVOKE:
                    doInvoke(cache);
                    break;

                case INVOKE_ALL:
                    doInvokeAll(cache);
                    break;

                case REMOVE:
                    doRemove(cache);
                    break;

                case REMOVE_ALL:
                    doRemoveAll(cache);
                    break;

                case PUT_IF_ABSENT:
                    doPutIfAbsent(cache);
                    break;

                case REPLACE:
                    doReplace(cache);
                    break;

                case SCAN_QUERY:
                    doScanQuery(cache);
                    break;

                case SQL_QUERY:
                    doSqlQuery(cache);
                    break;

                case CONTINUOUS_QUERY:
                    doContinuousQuery(cache, map);
            }

            storeStatistics(cache.getName(), map, op);
        }
        catch (Exception e) {
            BenchmarkUtils.error(String.format("Failed to perform operation %s.", op), e);
        }
    }

    /**
     * @return Operation.
     */
    @NotNull private Operation nextRandomOperation() {
        return allowedLoadTestOps.get(nextRandom(allowedLoadTestOps.size()));
    }

    /**
     * @param cacheName Ignite cache name.
     * @param map Parameters map.
     * @param op Operation.
     */
    private void storeStatistics(String cacheName, Map<Object, Object> map, Operation op) {
        String opCacheKey = op + "_" + cacheName;

        Long opAmount = (Long)map.get("amount");

        opAmount = opAmount == null ? 1L : opAmount + 1;

        Integer opCacheCnt = (Integer)map.get(opCacheKey);

        opCacheCnt = opCacheCnt == null ? 1 : opCacheCnt + 1;

        if (opAmount % 100 == 0)
            updateStat(map);
        else
            map.put(opCacheKey, opCacheCnt);

        map.put("amount", opAmount);
    }

    /**
     * @param map Parameters map.
     */
    private void updateStat(Map<Object, Object> map) {
        for (Operation op : Operation.values())
            for (String cacheName : ignite().cacheNames()) {
                String opCacheKey = op + "_" + cacheName;

                Integer val = (Integer)map.get(opCacheKey);

                if (val != null) {
                    operationStatistics.get(opCacheKey).addAndGet(val.longValue());

                    map.put(opCacheKey, 0);
                }

            }
    }

    /**
     * Execute operations in transaction.
     *
     * @param map Parameters map.
     * @throws Exception if fail.
     */
    private void executeInTransaction(final Map<Object, Object> map) throws Exception {
        IgniteBenchmarkUtils.doInTransaction(ignite().transactions(),
            TransactionConcurrency.fromOrdinal(nextRandom(TransactionConcurrency.values().length)),
            TransactionIsolation.fromOrdinal(nextRandom(TransactionIsolation.values().length)),

            new Callable<Object>() {
                @Override public Object call() throws Exception {
                    for (IgniteCache<Object, Object> cache : txCaches)
                        if (nextBoolean())
                            executeRandomOperation(map, cache);

                    return null;
                }
            });
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doPut(IgniteCache<Object, Object> cache) throws Exception {
        int i = nextRandom(args.range());

        cache.put(createRandomKey(i, cache.getName()), createRandomValue(i, cache.getName()));
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doPutAll(IgniteCache<Object, Object> cache) throws Exception {
        Map<Object, Object> putMap = new TreeMap<>();

        Class keyCass = randomKeyClass(cache.getName());

        for (int cnt = 0; cnt < args.batch(); cnt++) {
            int i = nextRandom(args.range());

            putMap.put(ModelUtil.create(keyCass, i), createRandomValue(i, cache.getName()));
        }

        cache.putAll(putMap);
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doGet(IgniteCache<Object, Object> cache) throws Exception {
        int i = nextRandom(args.range());

        cache.get(createRandomKey(i, cache.getName()));
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doGetAll(IgniteCache<Object, Object> cache) throws Exception {
        Set<Object> keys = new TreeSet<>();

        Class keyCls = randomKeyClass(cache.getName());

        for (int cnt = 0; cnt < args.batch(); cnt++) {
            int i = nextRandom(args.range());

            keys.add(ModelUtil.create(keyCls, i));
        }

        cache.getAll(keys);
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doInvoke(final IgniteCache<Object, Object> cache) throws Exception {
        final int i = nextRandom(args.range());

        if (nextBoolean())
            cache.invoke(createRandomKey(i, cache.getName()), replaceEntryProc,
                createRandomValue(i + 1, cache.getName()));
        else
            cache.invoke(createRandomKey(i, cache.getName()), rmvEntryProc);

    }

    /**
     * Entry processor for local benchmark replace value task.
     */
    private static class BenchmarkReplaceValueEntryProcessor implements EntryProcessor<Object, Object, Object>, Serializable {
        /**
         * New value for updateCache during process by default.
         */
        private Object newVal;

        /**
         * @param newVal default new value
         */
        private BenchmarkReplaceValueEntryProcessor(Object newVal) {
            this.newVal = newVal;
        }

        /** {@inheritDoc} */
        @Override public Object process(MutableEntry<Object, Object> entry, Object... arguments) {
            Object newVal = arguments == null || arguments[0] == null ? this.newVal : arguments[0];

            Object oldVal = entry.getValue();

            entry.setValue(newVal);

            return oldVal;
        }
    }

    /**
     * Entry processor for local benchmark remove entry task.
     */
    private static class BenchmarkRemoveEntryProcessor implements EntryProcessor<Object, Object, Object>, Serializable {
        /** {@inheritDoc} */
        @Override public Object process(MutableEntry<Object, Object> entry, Object... arguments) {
            Object oldVal = entry.getValue();

            entry.remove();

            return oldVal;
        }
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doInvokeAll(IgniteCache<Object, Object> cache) throws Exception {
        Map<Object, EntryProcessor<Object, Object, Object>> map = new TreeMap<>();

        Class keyCls = randomKeyClass(cache.getName());

        for (int cnt = 0; cnt < args.batch(); cnt++) {
            int i = nextRandom(args.range());

            Object key = ModelUtil.create(keyCls, i);

            if (nextBoolean())
                map.put(key,
                    new BenchmarkReplaceValueEntryProcessor(createRandomValue(i + 1, cache.getName())));
            else
                map.put(key, rmvEntryProc);
        }

        cache.invokeAll(map);
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doRemove(IgniteCache<Object, Object> cache) throws Exception {
        int i = nextRandom(args.range());

        cache.remove(createRandomKey(i, cache.getName()));
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doRemoveAll(IgniteCache<Object, Object> cache) throws Exception {
        Set<Object> keys = new TreeSet<>();

        Class keyCls = randomKeyClass(cache.getName());

        for (int cnt = 0; cnt < args.batch(); cnt++) {
            int i = nextRandom(args.range());

            keys.add(ModelUtil.create(keyCls, i));
        }

        cache.removeAll(keys);
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doPutIfAbsent(IgniteCache<Object, Object> cache) throws Exception {
        int i = nextRandom(args.range());

        cache.putIfAbsent(createRandomKey(i, cache.getName()), createRandomValue(i, cache.getName()));
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doReplace(IgniteCache<Object, Object> cache) throws Exception {
        int i = nextRandom(args.range());

        cache.replace(createRandomKey(i, cache.getName()),
            createRandomValue(i, cache.getName()),
            createRandomValue(i + 1, cache.getName()));
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doScanQuery(IgniteCache<Object, Object> cache) throws Exception {
        if (!affCaches.contains(cache))
            return;

        Map<UUID, List<Integer>> partitionsMap = personCachePartitions(cache.getName());

        ScanQueryBroadcastClosure c = new ScanQueryBroadcastClosure(cache.getName(), partitionsMap);

        ClusterGroup clusterGrp = ignite().cluster().forNodeIds(partitionsMap.keySet());

        IgniteCompute compute = ignite().compute(clusterGrp);

        compute.broadcast(c);
    }

    /**
     * @param cache Ignite cache.
     * @throws Exception If failed.
     */
    private void doSqlQuery(IgniteCache<Object, Object> cache) {
        List<SqlCacheDescriptor> descriptors = cacheSqlDescriptors.get(cache.getName());

        if (descriptors != null) {
            Query sq = null;

            if (queries.isEmpty()) {
                if (!descriptors.isEmpty()) {
                    SqlCacheDescriptor randomDesc = descriptors.get(nextRandom(descriptors.size()));

                    int id = nextRandom(args.range());

                    sq = nextBoolean() ? randomDesc.getSqlQuery(id) : randomDesc.getSqlFieldsQuery(id);
                }
            }
            else {
                TestQuery qry = queries.get(nextRandom(queries.size()));

                String sql = randomizeSql(qry.sql);

                sq = new SqlFieldsQuery(sql);

                ((SqlFieldsQuery)sq).setDistributedJoins(qry.distributedJoin);
            }

            if (sq != null)
                try (QueryCursor cursor = cache.query(sq)) {
                    for (Object obj : cursor) {
                        // No-op.
                    }
                }
        }
    }

    /**
     * @param sql Base SQL.
     * @return Randomized SQL string.
     */
    private String randomizeSql(String sql) {
        int cnt = StringUtils.countOccurrencesOf(sql, "%s");

        Integer[] sub = new Integer[cnt];

        for (int i = 0; i < cnt; i++)
            sub[i] = nextRandom(args.range());

        sql = String.format(sql, sub);

        return sql;
    }

    /**
     * @param cache Ignite cache.
     * @param map Parameters map.
     * @throws Exception If failed.
     */
    private void doContinuousQuery(IgniteCache<Object, Object> cache, Map<Object, Object> map) throws Exception {
        List<QueryCursor> cursors = (ArrayList<QueryCursor>)map.get(cache.getName());

        if (cursors == null) {
            cursors = new ArrayList<>(CONTINUOUS_QUERY_PER_CACHE);
            map.put(cache.getName(), cursors);
        }

        if (cursors.size() == CONTINUOUS_QUERY_PER_CACHE) {
            QueryCursor cursor = cursors.get(nextRandom(cursors.size()));
            cursor.close();
            cursors.remove(cursor);
        }

        ContinuousQuery qry = new ContinuousQuery();

        qry.setLocalListener(new ContinuousQueryUpdater());

        qry.setRemoteFilterFactory(FactoryBuilder.factoryOf(new ContinuousQueryFilter()));

        cursors.add(cache.query(qry));
    }

    /**
     * @return Nex random boolean value.
     */
    private boolean nextBoolean() {
        return ThreadLocalRandom.current().nextBoolean();
    }

    /**
     * Continuous query updater class.
     */
    private static class ContinuousQueryUpdater implements CacheEntryUpdatedListener, Serializable {

        /** {@inheritDoc} */
        @Override public void onUpdated(Iterable iterable) throws CacheEntryListenerException {
            for (Object o : iterable) {
                // No-op.
            }
        }
    }

    /**
     * Continuous query filter class.
     */
    private static class ContinuousQueryFilter implements CacheEntryEventSerializableFilter, Serializable {
        /** */
        private boolean flag = true;

        /** {@inheritDoc} */
        @Override public boolean evaluate(CacheEntryEvent evt) throws CacheEntryListenerException {
            return flag = !flag;
        }
    }

    /**
     * Closure for scan query executing.
     */
    private static class ScanQueryBroadcastClosure implements IgniteRunnable {
        /**
         * Ignite node.
         */
        @IgniteInstanceResource
        private Ignite node;

        /**
         * Information about partition.
         */
        private Map<UUID, List<Integer>> cachePart;

        /**
         * Name of Ignite cache.
         */
        private String cacheName;

        /**
         * @param cacheName Name of Ignite cache.
         * @param cachePart Partition by node for Ignite cache.
         */
        private ScanQueryBroadcastClosure(String cacheName, Map<UUID, List<Integer>> cachePart) {
            this.cachePart = cachePart;
            this.cacheName = cacheName;
        }

        /** {@inheritDoc} */
        @Override public void run() {
            IgniteCache cache = node.cache(cacheName);

            // Getting a list of the partitions owned by this node.
            List<Integer> myPartitions = cachePart.get(node.cluster().localNode().id());

            for (Integer part : myPartitions) {

                ScanQuery scanQry = new ScanQuery();

                scanQry.setPartition(part);

                scanQry.setFilter(igniteBiPred);

                try (QueryCursor cursor = cache.query(scanQry)) {
                    for (Object obj : cursor) {
                        // No-op.
                    }
                }

            }
        }
    }

    /**
     * Scan query predicate class.
     */
    private static class BenchmarkIgniteBiPredicate implements IgniteBiPredicate {
        /**
         * @param key Cache key.
         * @param val Cache value.
         * @return true If is hit.
         */
        @Override public boolean apply(Object key, Object val) {
            return true;
        }
    }

    /**
     * Query descriptor.
     */
    private static class SqlCacheDescriptor {
        /**
         * Class of value.
         */
        private Class valCls;

        /**
         * Select fields.
         */
        private Set<String> fields;

        /**
         * Indexed fields.
         */
        private Map<String, Class> indexedFieldsByCls;

        /**
         * @param valCls Class of value.
         * @param fields All select fields.
         * @param indexedFieldsByCls Indexed fields.
         */
        SqlCacheDescriptor(Class valCls, Set<String> fields,
            Map<String, Class> indexedFieldsByCls) {
            this.valCls = valCls;
            this.fields = fields;
            this.indexedFieldsByCls = indexedFieldsByCls;
        }

        /**
         * @param id Query id.
         * @return Condition string.
         */
        private String makeQuerySelect(int id) {
            return StringUtils.collectionToDelimitedString(fields, ", ");
        }

        /**
         * @param id Query id.
         * @return Condition string.
         */
        private String makeQueryCondition(int id) {
            StringBuilder sb = new StringBuilder();

            for (String iField : indexedFieldsByCls.keySet()) {
                Class cl = indexedFieldsByCls.get(iField);

                if (!Number.class.isAssignableFrom(cl) && !String.class.equals(cl))
                    continue;

                if (sb.length() != 0) {
                    switch (id % 3 % 2) {
                        case 0:
                            sb.append(" OR ");
                            break;
                        case 1:
                            sb.append(" AND ");
                            break;
                    }
                }

                if (Number.class.isAssignableFrom(cl)) {
                    sb.append(iField);
                    switch (id % 2) {
                        case 0:
                            sb.append(" > ");
                            break;
                        case 1:
                            sb.append(" < ");
                            break;
                    }
                    sb.append(id);
                }
                else if (String.class.equals(cl))
                    sb.append("lower(").append(iField).append(") LIKE lower('%").append(id).append("%')");

            }
            return sb.toString();
        }

        /**
         * @param id Query id.
         * @return SQL query object.
         */
        SqlQuery getSqlQuery(int id) {
            return new SqlQuery(valCls, makeQueryCondition(id));
        }

        /**
         * @param id Query id.
         * @return SQL filed query object.
         */
        SqlFieldsQuery getSqlFieldsQuery(int id) {
            return new SqlFieldsQuery(String.format("SELECT %s FROM %s WHERE %s",
                makeQuerySelect(id), valCls.getSimpleName(), makeQueryCondition(id)));
        }

    }

    /**
     * Cache operation enum.
     */
    private enum Operation {
        /** Put operation. */
        PUT,

        /** Put all operation. */
        PUT_ALL,

        /** Get operation. */
        GET,

        /** Get all operation. */
        GET_ALL,

        /** Invoke operation. */
        INVOKE,

        /** Invoke all operation. */
        INVOKE_ALL,

        /** Remove operation. */
        REMOVE,

        /** Remove all operation. */
        REMOVE_ALL,

        /** Put if absent operation. */
        PUT_IF_ABSENT,

        /** Replace operation. */
        REPLACE,

        /** Scan query operation. */
        SCAN_QUERY,

        /** SQL query operation. */
        SQL_QUERY,

        /** Continuous Query. */
        CONTINUOUS_QUERY;

        /**
         * @param num Number of operation.
         * @return Operation.
         */
        public static Operation valueOf(int num) {
            return values()[num];
        }
    }

    /**
     *
     */
    private static class TestQuery {
        /** */
        private final String sql;

        /** */
        private final boolean distributedJoin;

        /**
         * @param sql SQL.
         * @param distributedJoin Distributed join flag.
         */
        public TestQuery(String sql, boolean distributedJoin) {
            this.sql = sql;
            this.distributedJoin = distributedJoin;
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(TestQuery.class, this);
        }
    }
}