/*
 * 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.cache.store.jdbc;

import java.sql.BatchUpdateException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
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.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.integration.CacheLoaderException;
import javax.cache.integration.CacheWriterException;
import javax.sql.DataSource;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cache.store.CacheStoreSession;
import org.apache.ignite.cache.store.jdbc.dialect.BasicJdbcDialect;
import org.apache.ignite.cache.store.jdbc.dialect.DB2Dialect;
import org.apache.ignite.cache.store.jdbc.dialect.H2Dialect;
import org.apache.ignite.cache.store.jdbc.dialect.JdbcDialect;
import org.apache.ignite.cache.store.jdbc.dialect.MySQLDialect;
import org.apache.ignite.cache.store.jdbc.dialect.OracleDialect;
import org.apache.ignite.cache.store.jdbc.dialect.SQLServerDialect;
import org.apache.ignite.internal.binary.BinaryEnumObjectImpl;
import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lifecycle.LifecycleAware;
import org.apache.ignite.resources.CacheStoreSessionResource;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.resources.LoggerResource;
import org.apache.ignite.thread.IgniteThreadFactory;
import org.apache.ignite.transactions.Transaction;
import org.jetbrains.annotations.Nullable;

import static java.sql.Statement.EXECUTE_FAILED;
import static java.sql.Statement.SUCCESS_NO_INFO;
import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreFactory.DFLT_BATCH_SIZE;
import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreFactory.DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD;
import static org.apache.ignite.cache.store.jdbc.CacheJdbcPojoStoreFactory.DFLT_WRITE_ATTEMPTS;
import static org.apache.ignite.cache.store.jdbc.JdbcTypesTransformer.NUMERIC_TYPES;

/**
 * Implementation of {@link CacheStore} backed by JDBC.
 * <p>
 * Store works with database via SQL dialect. Ignite ships with dialects for most popular databases:
 * <ul>
 *     <li>{@link DB2Dialect} - dialect for IBM DB2 database.</li>
 *     <li>{@link OracleDialect} - dialect for Oracle database.</li>
 *     <li>{@link SQLServerDialect} - dialect for Microsoft SQL Server database.</li>
 *     <li>{@link MySQLDialect} - dialect for Oracle MySQL database.</li>
 *     <li>{@link H2Dialect} - dialect for H2 database.</li>
 *     <li>{@link BasicJdbcDialect} - dialect for any database via plain JDBC.</li>
 * </ul>
 * <p>
 * <h2 class="header">Configuration</h2>
 * <ul>
 *     <li>Data source (see {@link #setDataSource(DataSource)}</li>
 *     <li>Dialect (see {@link #setDialect(JdbcDialect)}</li>
 *     <li>Maximum batch size for writeAll and deleteAll operations. (see {@link #setBatchSize(int)})</li>
 *     <li>Max workers thread count. These threads are responsible for load cache. (see {@link #setMaximumPoolSize(int)})</li>
 *     <li>Parallel load cache minimum threshold. (see {@link #setParallelLoadCacheMinimumThreshold(int)})</li>
 * </ul>
 * <h2 class="header">Java Example</h2>
 * <pre name="code" class="java">
 *    ...
 *    // Create store factory.
 *    CacheJdbcPojoStoreFactory storeFactory = new CacheJdbcPojoStoreFactory();
 *    storeFactory.setDataSourceBean("your_data_source_name");
 *    storeFactory.setDialect(new H2Dialect());
 *    storeFactory.setTypes(array_with_your_types);
 *    ...
 *    ccfg.setCacheStoreFactory(storeFactory);
 *    ccfg.setReadThrough(true);
 *    ccfg.setWriteThrough(true);
 *
 *    cfg.setCacheConfiguration(ccfg);
 *    ...
 * </pre>
 */
public abstract class CacheAbstractJdbcStore<K, V> implements CacheStore<K, V>, LifecycleAware {
    /** Connection attribute property name. */
    protected static final String ATTR_CONN_PROP = "JDBC_STORE_CONNECTION";

    /** Thread name. */
    private static final String CACHE_LOADER_THREAD_NAME = "jdbc-cache-loader";

    /** Built in Java types names. */
    protected static final Collection<String> BUILT_IN_TYPES = new HashSet<>();

    static {
        BUILT_IN_TYPES.add("java.math.BigDecimal");
        BUILT_IN_TYPES.add("java.lang.Boolean");
        BUILT_IN_TYPES.add("java.lang.Byte");
        BUILT_IN_TYPES.add("java.lang.Character");
        BUILT_IN_TYPES.add("java.lang.Double");
        BUILT_IN_TYPES.add("java.util.Date");
        BUILT_IN_TYPES.add("java.sql.Date");
        BUILT_IN_TYPES.add("java.lang.Float");
        BUILT_IN_TYPES.add("java.lang.Integer");
        BUILT_IN_TYPES.add("java.lang.Long");
        BUILT_IN_TYPES.add("java.lang.Short");
        BUILT_IN_TYPES.add("java.lang.String");
        BUILT_IN_TYPES.add("java.sql.Timestamp");
        BUILT_IN_TYPES.add("java.util.UUID");
    }

    /** Auto-injected store session. */
    @CacheStoreSessionResource
    private CacheStoreSession ses;

    /** Auto injected ignite instance. */
    @IgniteInstanceResource
    protected Ignite ignite;

    /** Auto-injected logger instance. */
    @LoggerResource
    protected IgniteLogger log;

    /** Lock for metadata cache. */
    @GridToStringExclude
    private final Lock cacheMappingsLock = new ReentrantLock();

    /** Data source. */
    protected volatile DataSource dataSrc;

    /** Cache with entry mapping description. (cache name, (key id, mapping description)). */
    protected volatile Map<String, Map<Object, EntryMapping>> cacheMappings = Collections.emptyMap();

    /** Maximum batch size for writeAll and deleteAll operations. */
    private int batchSize = DFLT_BATCH_SIZE;

    /** Database dialect. */
    protected JdbcDialect dialect;

    /** Maximum write attempts in case of database error. */
    private int maxWrtAttempts = DFLT_WRITE_ATTEMPTS;

    /** Max workers thread count. These threads are responsible for load cache. */
    private int maxPoolSize = Runtime.getRuntime().availableProcessors();

    /** Parallel load cache minimum threshold. If {@code 0} then load sequentially. */
    private int parallelLoadCacheMinThreshold = DFLT_PARALLEL_LOAD_CACHE_MINIMUM_THRESHOLD;

    /** Types that store could process. */
    private JdbcType[] types;

    /** Hash calculator.  */
    protected JdbcTypeHasher hasher = JdbcTypeDefaultHasher.INSTANCE;

    /** Types transformer. */
    protected JdbcTypesTransformer transformer = JdbcTypesDefaultTransformer.INSTANCE;

    /** Flag indicating that table and field names should be escaped in all SQL queries created by JDBC POJO store. */
    private boolean sqlEscapeAll;

    /**
     * Get field value from object for use as query parameter.
     *
     * @param cacheName Cache name.
     * @param typeName Type name.
     * @param typeKind Type kind.
     * @param fieldName Field name.
     * @param obj Cache object.
     * @return Field value from object.
     * @throws CacheException in case of error.
     */
    @Nullable protected abstract Object extractParameter(@Nullable String cacheName, String typeName, TypeKind typeKind,
        String fieldName, Object obj) throws CacheException;

    /**
     * Construct object from query result.
     *
     * @param <R> Type of result object.
     * @param cacheName Cache name.
     * @param typeName Type name.
     * @param typeKind Type kind.
     * @param flds Fields descriptors.
     * @param loadColIdxs Select query columns index.
     * @param rs ResultSet.
     * @return Constructed object.
     * @throws CacheLoaderException If failed to construct cache object.
     */
    protected abstract <R> R buildObject(@Nullable String cacheName, String typeName, TypeKind typeKind,
        JdbcTypeField[] flds, Map<String, Integer> loadColIdxs, ResultSet rs) throws CacheLoaderException;

    /**
     * Calculate type ID for object.
     *
     * @param obj Object to calculate type ID for.
     * @return Type ID.
     * @throws CacheException If failed to calculate type ID for given object.
     */
    protected abstract Object typeIdForObject(Object obj) throws CacheException;

    /**
     * Calculate type ID for given type name.
     *
     * @param kind If {@code true} then calculate type ID for POJO otherwise for binary object .
     * @param typeName String description of type name.
     * @return Type ID.
     * @throws CacheException If failed to get type ID for given type name.
     */
    protected abstract Object typeIdForTypeName(TypeKind kind, String typeName) throws CacheException;

    /**
     * Prepare internal store specific builders for provided types metadata.
     *
     * @param cacheName Cache name to prepare builders for.
     * @param types Collection of types.
     * @throws CacheException If failed to prepare internal builders for types.
     */
    protected abstract void prepareBuilders(@Nullable String cacheName, Collection<JdbcType> types)
        throws CacheException;

    /**
     * Perform dialect resolution.
     *
     * @return The resolved dialect.
     * @throws CacheException Indicates problems accessing the metadata.
     */
    protected JdbcDialect resolveDialect() throws CacheException {
        Connection conn = null;

        String dbProductName = null;

        try {
            conn = openConnection(false);

            dbProductName = conn.getMetaData().getDatabaseProductName();
        }
        catch (SQLException e) {
            throw new CacheException("Failed access to metadata for detect database dialect.", e);
        }
        finally {
            U.closeQuiet(conn);
        }

        if ("H2".equals(dbProductName))
            return new H2Dialect();

        if ("MySQL".equals(dbProductName))
            return new MySQLDialect();

        if (dbProductName.startsWith("Microsoft SQL Server"))
            return new SQLServerDialect();

        if ("Oracle".equals(dbProductName))
            return new OracleDialect();

        if (dbProductName.startsWith("DB2/"))
            return new DB2Dialect();

        U.warn(log, "Failed to resolve dialect (BasicJdbcDialect will be used): " + dbProductName);

        return new BasicJdbcDialect();
    }

    /** {@inheritDoc} */
    @Override public void start() throws IgniteException {
        if (dataSrc == null)
            throw new IgniteException("Failed to initialize cache store (data source is not provided).");

        if (dialect == null) {
            dialect = resolveDialect();

            if (log.isDebugEnabled() && dialect.getClass() != BasicJdbcDialect.class)
                log.debug("Resolved database dialect: " + U.getSimpleName(dialect.getClass()));
        }
    }

    /** {@inheritDoc} */
    @Override public void stop() throws IgniteException {
        // No-op.
    }

    /**
     * Gets connection from a pool.
     *
     * @param autocommit {@code true} If connection should use autocommit mode.
     * @return Pooled connection.
     * @throws SQLException In case of error.
     */
    protected Connection openConnection(boolean autocommit) throws SQLException {
        Connection conn = dataSrc.getConnection();

        conn.setAutoCommit(autocommit);

        return conn;
    }

    /**
     * @return Connection.
     * @throws SQLException In case of error.
     */
    protected Connection connection() throws SQLException {
        CacheStoreSession ses = session();

        if (ses.transaction() != null) {
            Map<String, Connection> prop = ses.properties();

            Connection conn = prop.get(ATTR_CONN_PROP);

            if (conn == null) {
                conn = openConnection(false);

                // Store connection in session to used it for other operations in the same session.
                prop.put(ATTR_CONN_PROP, conn);
            }

            return conn;
        }
        // Transaction can be null in case of simple load operation.
        else
            return openConnection(true);
    }

    /**
     * Closes connection.
     *
     * @param conn Connection to close.
     */
    protected void closeConnection(@Nullable Connection conn) {
        CacheStoreSession ses = session();

        // Close connection right away if there is no transaction.
        if (ses.transaction() == null)
            U.closeQuiet(conn);
    }

    /**
     * Closes allocated resources depending on transaction status.
     *
     * @param conn Allocated connection.
     * @param st Created statement,
     */
    protected void end(@Nullable Connection conn, @Nullable Statement st) {
        U.closeQuiet(st);

        closeConnection(conn);
    }

    /** {@inheritDoc} */
    @Override public void sessionEnd(boolean commit) throws CacheWriterException {
        CacheStoreSession ses = session();

        Transaction tx = ses.transaction();

        if (tx != null) {
            Map<String, Connection> sesProps = ses.properties();

            Connection conn = sesProps.get(ATTR_CONN_PROP);

            if (conn != null) {
                sesProps.remove(ATTR_CONN_PROP);

                try {
                    if (commit)
                        conn.commit();
                    else
                        conn.rollback();
                }
                catch (SQLException e) {
                    throw new CacheWriterException(
                            "Failed to end transaction [xid=" + tx.xid() + ", commit=" + commit + ']', e);
                }
                finally {
                    U.closeQuiet(conn);
                }
            }

            if (log.isDebugEnabled())
                log.debug("Transaction ended [xid=" + tx.xid() + ", commit=" + commit + ']');
        }
    }

    /**
     * Construct load cache from range.
     *
     * @param em Type mapping description.
     * @param clo Closure that will be applied to loaded values.
     * @param lowerBound Lower bound for range.
     * @param upperBound Upper bound for range.
     * @param fetchSize Number of rows to fetch from DB.
     * @return Callable for pool submit.
     */
    private Callable<Void> loadCacheRange(
        final EntryMapping em,
        final IgniteBiInClosure<K, V> clo,
        @Nullable final Object[] lowerBound,
        @Nullable final Object[] upperBound,
        final int fetchSize
    ) {
        return new Callable<Void>() {
            @Override public Void call() throws Exception {
                Connection conn = null;

                PreparedStatement stmt = null;

                try {
                    conn = openConnection(true);

                    stmt = conn.prepareStatement(lowerBound == null && upperBound == null
                        ? em.loadCacheQry
                        : em.loadCacheRangeQuery(lowerBound != null, upperBound != null));

                    stmt.setFetchSize(fetchSize);

                    int idx = 1;

                    if (lowerBound != null)
                        for (int i = lowerBound.length; i > 0; i--)
                            for (int j = 0; j < i; j++)
                                stmt.setObject(idx++, lowerBound[j]);

                    if (upperBound != null)
                        for (int i = upperBound.length; i > 0; i--)
                            for (int j = 0; j < i; j++)
                                stmt.setObject(idx++, upperBound[j]);

                    ResultSet rs = stmt.executeQuery();

                    while (rs.next()) {
                        K key = buildObject(em.cacheName, em.keyType(), em.keyKind(), em.keyColumns(),
                            em.loadColIdxs, rs);

                        V val = buildObject(em.cacheName, em.valueType(), em.valueKind(), em.valueColumns(),
                            em.loadColIdxs, rs);

                        clo.apply(key, val);
                    }
                }
                catch (SQLException e) {
                    throw new IgniteCheckedException("Failed to load cache", e);
                }
                finally {
                    U.closeQuiet(stmt);

                    U.closeQuiet(conn);
                }

                return null;
            }
        };
    }

    /**
     * Construct load cache in one select.
     *
     * @param m Type mapping description.
     * @param clo Closure for loaded values.
     * @return Callable for pool submit.
     */
    private Callable<Void> loadCacheFull(EntryMapping m, IgniteBiInClosure<K, V> clo) {
        return loadCacheRange(m, clo, null, null, dialect.getFetchSize());
    }

    /**
     * Checks if type configured properly.
     *
     * @param cacheName Cache name to check mapping for.
     * @param typeName Type name.
     * @param flds Fields descriptors.
     * @throws CacheException If failed to check type configuration.
     */
    private void checkTypeConfiguration(@Nullable String cacheName, TypeKind kind, String typeName,
        JdbcTypeField[] flds) throws CacheException {
        try {
            if (kind == TypeKind.BUILT_IN) {
                if (flds.length != 1)
                    throw new CacheException("More than one field for built in type " +
                        "[cache=" + U.maskName(cacheName) + ", type=" + typeName + " ]");

                JdbcTypeField field = flds[0];

                if (field.getDatabaseFieldName() == null)
                    throw new CacheException("Missing database name in mapping description [cache=" +
                        U.maskName(cacheName) + ", type=" + typeName + " ]");

                field.setJavaFieldType(Class.forName(typeName));
            }
            else
                for (JdbcTypeField field : flds) {
                    if (field.getDatabaseFieldName() == null)
                        throw new CacheException("Missing database name in mapping description " +
                            "[cache=" + U.maskName(cacheName) + ", type=" + typeName + " ]");

                    if (field.getJavaFieldName() == null)
                        throw new CacheException("Missing field name in mapping description " +
                            "[cache=" + U.maskName(cacheName) + ", type=" + typeName + " ]");

                    if (field.getJavaFieldType() == null)
                        throw new CacheException("Missing field type in mapping description " +
                            "[cache=" + U.maskName(cacheName) + ", type=" + typeName + " ]");
                }
        }
        catch (ClassNotFoundException e) {
            throw new CacheException("Failed to find class: " + typeName, e);
        }
    }

    /**
     * @param type Type name to check.
     * @param binarySupported True if binary marshaller enable.
     * @return {@code True} if class not found.
     */
    protected TypeKind kindForName(String type, boolean binarySupported) {
        if (BUILT_IN_TYPES.contains(type))
            return TypeKind.BUILT_IN;

        if (binarySupported)
            return TypeKind.BINARY;

        try {
            Class.forName(type);

            return TypeKind.POJO;
        }
        catch (ClassNotFoundException e) {
            throw new CacheException("Failed to find class " + type +
                " (make sure the class is present in classPath or use BinaryMarshaller)", e);
        }
    }

    /**
     * @param type Type name to check.
     * @return {@code True} if class not found.
     */
    protected TypeKind kindForName(String type) {
        return kindForName(type, ignite.configuration().getMarshaller() instanceof BinaryMarshaller);
    }

    /**
     * @param cacheName Cache name to check mappings for.
     * @return Type mappings for specified cache name.
     * @throws CacheException If failed to initialize cache mappings.
     */
    private Map<Object, EntryMapping> getOrCreateCacheMappings(@Nullable String cacheName) throws CacheException {
        Map<Object, EntryMapping> entryMappings = cacheMappings.get(cacheName);

        if (entryMappings != null)
            return entryMappings;

        cacheMappingsLock.lock();
        try {
            entryMappings = cacheMappings.get(cacheName);

            if (entryMappings != null)
                return entryMappings;

            List<JdbcType> cacheTypes = new ArrayList<>(types.length);

            for (JdbcType type : types)
                if ((cacheName != null && cacheName.equals(type.getCacheName())) ||
                    (cacheName == null && type.getCacheName() == null))
                    cacheTypes.add(type);

            entryMappings = U.newHashMap(cacheTypes.size());

            if (!cacheTypes.isEmpty()) {
                boolean binarySupported = ignite.configuration().getMarshaller() instanceof BinaryMarshaller;

                for (JdbcType type : cacheTypes) {
                    String keyType = type.getKeyType();
                    String valType = type.getValueType();

                    TypeKind keyKind = kindForName(keyType, binarySupported);

                    checkTypeConfiguration(cacheName, keyKind, keyType, type.getKeyFields());

                    Object keyTypeId = typeIdForTypeName(keyKind, keyType);

                    if (entryMappings.containsKey(keyTypeId))
                        throw new CacheException("Key type must be unique in type metadata [cache=" +
                            U.maskName(cacheName) + ", type=" + keyType + "]");

                    TypeKind valKind = kindForName(valType, binarySupported);

                    checkTypeConfiguration(cacheName, valKind, valType, type.getValueFields());

                    entryMappings.put(keyTypeId, new EntryMapping(cacheName, dialect, type, keyKind, valKind, sqlEscapeAll));
                }

                Map<String, Map<Object, EntryMapping>> mappings = new HashMap<>(cacheMappings);

                mappings.put(cacheName, entryMappings);

                prepareBuilders(cacheName, cacheTypes);

                cacheMappings = mappings;
            }

            return entryMappings;
        }
        finally {
            cacheMappingsLock.unlock();
        }
    }

    /**
     * @param cacheName Cache name.
     * @param typeId Type id.
     * @return Entry mapping.
     * @throws CacheException If mapping for key was not found.
     */
    private EntryMapping entryMapping(String cacheName, Object typeId) throws CacheException {
        Map<Object, EntryMapping> mappings = getOrCreateCacheMappings(cacheName);

        EntryMapping em = mappings.get(typeId);

        if (em == null) {
            String maskedCacheName = U.maskName(cacheName);

            throw new CacheException("Failed to find mapping description [cache=" + maskedCacheName +
                ", typeId=" + typeId + "]. Please configure JdbcType to associate cache '" + maskedCacheName +
                "' with JdbcPojoStore.");
        }

        return em;
    }

    /**
     * Find column index by database name.
     *
     * @param loadColIdxs Select query columns indexes.
     * @param dbName Column name in database.
     * @return Column index.
     * @throws IllegalStateException if column not found.
     */
    protected Integer columnIndex(Map<String, Integer> loadColIdxs, String dbName) {
        Integer colIdx = loadColIdxs.get(dbName.toUpperCase());

        if (colIdx == null)
            throw new IllegalStateException("Failed to find column index for database field: " + dbName);

        return colIdx;
    }

    /** {@inheritDoc} */
    @Override public void loadCache(final IgniteBiInClosure<K, V> clo, @Nullable Object... args)
        throws CacheLoaderException {
        ExecutorService pool = null;

        int fetchSz = dialect.getFetchSize();

        String cacheName = session().cacheName();

        try {
            pool = Executors.newFixedThreadPool(maxPoolSize, new IgniteThreadFactory(ignite.name(), CACHE_LOADER_THREAD_NAME));

            Collection<Future<?>> futs = new ArrayList<>();

            Map<Object, EntryMapping> mappings = getOrCreateCacheMappings(cacheName);

            if (args != null && args.length > 0) {
                if (args.length % 2 != 0)
                    throw new CacheLoaderException("Expected even number of arguments, but found: " + args.length);

                if (log.isDebugEnabled())
                    log.debug("Start loading entries from db using user queries from arguments...");

                for (int i = 0; i < args.length; i += 2) {
                    final String keyType = args[i].toString();

                    if (!F.exist(mappings.values(), new IgnitePredicate<EntryMapping>() {
                        @Override public boolean apply(EntryMapping em) {
                            return em.keyType().equals(keyType);
                        }
                    }))
                        throw new CacheLoaderException("Provided key type is not found in store or cache configuration " +
                            "[cache=" + U.maskName(cacheName) + ", key=" + keyType + ']');

                    EntryMapping em = entryMapping(cacheName, typeIdForTypeName(kindForName(keyType), keyType));

                    Object arg = args[i + 1];

                    LoadCacheCustomQueryWorker<K, V> task;

                    if (arg instanceof PreparedStatement) {
                        PreparedStatement stmt = (PreparedStatement)arg;

                        if (log.isInfoEnabled())
                            log.info("Started load cache using custom statement [cache=" + U.maskName(cacheName) +
                                      ", keyType=" + keyType + ", stmt=" + stmt + ']');

                        task = new LoadCacheCustomQueryWorker<>(em, stmt, clo);
                    }
                    else {
                        String qry = arg.toString();

                        if (log.isInfoEnabled())
                              log.info("Started load cache using custom query [cache=" + U.maskName(cacheName) +
                                  ", keyType=" + keyType + ", query=" + qry + ']');

                        task = new LoadCacheCustomQueryWorker<>(em, qry, clo);
                    }

                    futs.add(pool.submit(task));
                }
            }
            else {
                Collection<String> processedKeyTypes = new HashSet<>();

                for (EntryMapping em : mappings.values()) {
                    String keyType = em.keyType();

                    if (processedKeyTypes.contains(keyType))
                        continue;

                    processedKeyTypes.add(keyType);

                    if (log.isInfoEnabled())
                        log.info("Started load cache [cache=" + U.maskName(cacheName) + ", keyType=" + keyType + ']');

                    if (parallelLoadCacheMinThreshold > 0) {
                        Connection conn = null;

                        try {
                            conn = connection();

                            PreparedStatement stmt = conn.prepareStatement(em.loadCacheSelRangeQry);

                            stmt.setInt(1, parallelLoadCacheMinThreshold);

                            ResultSet rs = stmt.executeQuery();

                            if (rs.next()) {
                                if (log.isDebugEnabled())
                                    log.debug("Multithread loading entries from db [cache=" + U.maskName(cacheName) +
                                        ", keyType=" + keyType + ']');

                                int keyCnt = em.keyCols.size();

                                Object[] upperBound = new Object[keyCnt];

                                for (int i = 0; i < keyCnt; i++)
                                    upperBound[i] = rs.getObject(i + 1);

                                futs.add(pool.submit(loadCacheRange(em, clo, null, upperBound, fetchSz)));

                                while (rs.next()) {
                                    Object[] lowerBound = upperBound;

                                    upperBound = new Object[keyCnt];

                                    for (int i = 0; i < keyCnt; i++)
                                        upperBound[i] = rs.getObject(i + 1);

                                    futs.add(pool.submit(loadCacheRange(em, clo, lowerBound, upperBound, fetchSz)));
                                }

                                futs.add(pool.submit(loadCacheRange(em, clo, upperBound, null, fetchSz)));

                                continue;
                            }
                        }
                        catch (SQLException e) {
                            log.warning("Failed to load entries from db in multithreaded mode, will try in single thread " +
                                "[cache=" + U.maskName(cacheName) + ", keyType=" + keyType + ']', e);
                        }
                        finally {
                            U.closeQuiet(conn);
                        }
                    }

                    if (log.isDebugEnabled())
                        log.debug("Single thread loading entries from db [cache=" + U.maskName(cacheName) +
                            ", keyType=" + keyType + ']');

                    futs.add(pool.submit(loadCacheFull(em, clo)));
                }
            }

            for (Future<?> fut : futs)
                U.get(fut);

            if (log.isInfoEnabled())
                log.info("Finished load cache: " + U.maskName(cacheName));
        }
        catch (IgniteCheckedException e) {
            throw new CacheLoaderException("Failed to load cache: " + U.maskName(cacheName), e.getCause());
        }
        finally {
            U.shutdownNow(getClass(), pool, log);
        }
    }

    /** {@inheritDoc} */
    @Nullable @Override public V load(K key) throws CacheLoaderException {
        assert key != null;

        EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key));

        if (log.isDebugEnabled())
            log.debug("Load value from db [table= " + em.fullTableName() + ", key=" + key + ']');

        Connection conn = null;

        PreparedStatement stmt = null;

        try {
            conn = connection();

            stmt = conn.prepareStatement(em.loadQrySingle);

            fillKeyParameters(stmt, em, key);

            ResultSet rs = stmt.executeQuery();

            if (rs.next())
                return buildObject(em.cacheName, em.valueType(), em.valueKind(), em.valueColumns(), em.loadColIdxs, rs);
        }
        catch (SQLException e) {
            throw new CacheLoaderException("Failed to load object [table=" + em.fullTableName() +
                ", key=" + key + "]", e);
        }
        finally {
            end(conn, stmt);
        }

        return null;
    }

    /** {@inheritDoc} */
    @Override public Map<K, V> loadAll(Iterable<? extends K> keys) throws CacheLoaderException {
        assert keys != null;

        Connection conn = null;

        try {
            conn = connection();

            String cacheName = session().cacheName();

            Map<Object, LoadWorker<K, V>> workers = U.newHashMap(getOrCreateCacheMappings(cacheName).size());

            Map<K, V> res = new HashMap<>();

            for (K key : keys) {
                Object keyTypeId = typeIdForObject(key);

                EntryMapping em = entryMapping(cacheName, keyTypeId);

                LoadWorker<K, V> worker = workers.get(keyTypeId);

                if (worker == null)
                    workers.put(keyTypeId, worker = new LoadWorker<>(conn, em));

                worker.keys.add(key);

                if (worker.keys.size() == em.maxKeysPerStmt)
                    res.putAll(workers.remove(keyTypeId).call());
            }

            for (LoadWorker<K, V> worker : workers.values())
                res.putAll(worker.call());

            return res;
        }
        catch (Exception e) {
            throw new CacheWriterException("Failed to load entries from database", e);
        }
        finally {
            closeConnection(conn);
        }
    }

    /**
     * @param insStmt Insert statement.
     * @param updStmt Update statement.
     * @param em Entry mapping.
     * @param entry Cache entry.
     * @throws CacheWriterException If failed to update record in database.
     */
    private void writeUpsert(PreparedStatement insStmt, PreparedStatement updStmt,
        EntryMapping em, Cache.Entry<? extends K, ? extends V> entry) throws CacheWriterException {
        try {
            CacheWriterException we = null;

            for (int attempt = 0; attempt < maxWrtAttempts; attempt++) {
                int paramIdx = fillValueParameters(updStmt, 1, em, entry.getValue());

                fillKeyParameters(updStmt, paramIdx, em, entry.getKey());

                if (updStmt.executeUpdate() == 0) {
                    paramIdx = fillKeyParameters(insStmt, em, entry.getKey());

                    fillValueParameters(insStmt, paramIdx, em, entry.getValue());

                    try {
                        insStmt.executeUpdate();

                        if (attempt > 0)
                            U.warn(log, "Entry was inserted in database on second try [table=" + em.fullTableName() +
                                ", entry=" + entry + "]");
                    }
                    catch (SQLException e) {
                        String sqlState = e.getSQLState();

                        SQLException nested = e.getNextException();

                        while (sqlState == null && nested != null) {
                            sqlState = nested.getSQLState();

                            nested = nested.getNextException();
                        }

                        // The error with code 23505 or 23000 is thrown when trying to insert a row that
                        // would violate a unique index or primary key.
                        if ("23505".equals(sqlState) || "23000".equals(sqlState)) {
                            if (we == null)
                                we = new CacheWriterException("Failed insert entry in database, violate a unique" +
                                    " index or primary key [table=" + em.fullTableName() + ", entry=" + entry + "]");

                            we.addSuppressed(e);

                            U.warn(log, "Failed insert entry in database, violate a unique index or primary key" +
                                " [table=" + em.fullTableName() + ", entry=" + entry + "]");

                            continue;
                        }

                        throw new CacheWriterException("Failed insert entry in database [table=" + em.fullTableName() +
                            ", entry=" + entry, e);
                    }
                }

                if (attempt > 0)
                    U.warn(log, "Entry was updated in database on second try [table=" + em.fullTableName() +
                        ", entry=" + entry + "]");

                return;
            }

            throw we;
        }
        catch (SQLException e) {
            throw new CacheWriterException("Failed update entry in database [table=" + em.fullTableName() +
                ", entry=" + entry + "]", e);
        }
    }

    /** {@inheritDoc} */
    @Override public void write(Cache.Entry<? extends K, ? extends V> entry) throws CacheWriterException {
        assert entry != null;

        K key = entry.getKey();

        EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key));

        if (log.isDebugEnabled())
            log.debug("Start write entry to database [table=" + em.fullTableName() + ", entry=" + entry + "]");

        Connection conn = null;

        try {
            conn = connection();

            if (dialect.hasMerge()) {
                PreparedStatement stmt = null;

                try {
                    stmt = conn.prepareStatement(em.mergeQry);

                    int idx = fillKeyParameters(stmt, em, key);

                    fillValueParameters(stmt, idx, em, entry.getValue());

                    int updCnt = stmt.executeUpdate();

                    if (updCnt != 1)
                        U.warn(log, "Unexpected number of updated entries [table=" + em.fullTableName() +
                            ", entry=" + entry + "expected=1, actual=" + updCnt + "]");
                }
                finally {
                    U.closeQuiet(stmt);
                }
            }
            else {
                PreparedStatement insStmt = null;

                PreparedStatement updStmt = null;

                try {
                    insStmt = conn.prepareStatement(em.insQry);

                    updStmt = conn.prepareStatement(em.updQry);

                    writeUpsert(insStmt, updStmt, em, entry);
                }
                finally {
                    U.closeQuiet(insStmt);

                    U.closeQuiet(updStmt);
                }
            }
        }
        catch (SQLException e) {
            throw new CacheWriterException("Failed to write entry to database [table=" + em.fullTableName() +
                ", entry=" + entry + "]", e);
        }
        finally {
            closeConnection(conn);
        }
    }

    /** {@inheritDoc} */
    @Override public void writeAll(final Collection<Cache.Entry<? extends K, ? extends V>> entries)
        throws CacheWriterException {
        assert entries != null;

        Connection conn = null;

        try {
            conn = connection();

            String cacheName = session().cacheName();

            Object currKeyTypeId = null;

            if (dialect.hasMerge()) {
                PreparedStatement mergeStmt = null;

                try {
                    EntryMapping em = null;

                    LazyValue<Object[]> lazyEntries = new LazyValue<Object[]>() {
                        @Override public Object[] create() {
                            return entries.toArray();
                        }
                    };

                    int fromIdx = 0, prepared = 0;

                    for (Cache.Entry<? extends K, ? extends V> entry : entries) {
                        K key = entry.getKey();

                        Object keyTypeId = typeIdForObject(key);

                        em = entryMapping(cacheName, keyTypeId);

                        if (currKeyTypeId == null || !currKeyTypeId.equals(keyTypeId)) {
                            if (mergeStmt != null) {
                                if (log.isDebugEnabled())
                                    log.debug("Write entries to db [cache=" + U.maskName(cacheName) +
                                        ", keyType=" + em.keyType() + ", cnt=" + prepared + "]");

                                executeBatch(em, mergeStmt, "writeAll", fromIdx, prepared, lazyEntries);

                                U.closeQuiet(mergeStmt);
                            }

                            mergeStmt = conn.prepareStatement(em.mergeQry);

                            currKeyTypeId = keyTypeId;

                            fromIdx += prepared;

                            prepared = 0;
                        }

                        int idx = fillKeyParameters(mergeStmt, em, key);

                        fillValueParameters(mergeStmt, idx, em, entry.getValue());

                        mergeStmt.addBatch();

                        if (++prepared % batchSize == 0) {
                            if (log.isDebugEnabled())
                                log.debug("Write entries to db [cache=" + U.maskName(cacheName) +
                                    ", keyType=" + em.keyType() + ", cnt=" + prepared + "]");

                            executeBatch(em, mergeStmt, "writeAll", fromIdx, prepared, lazyEntries);

                            fromIdx += prepared;

                            prepared = 0;
                        }
                    }

                    if (mergeStmt != null && prepared % batchSize != 0) {
                        if (log.isDebugEnabled())
                            log.debug("Write entries to db [cache=" + U.maskName(cacheName) +
                                ", keyType=" + em.keyType() + ", cnt=" + prepared + "]");

                        executeBatch(em, mergeStmt, "writeAll", fromIdx, prepared, lazyEntries);

                    }
                }
                finally {
                    U.closeQuiet(mergeStmt);
                }
            }
            else {
                if (log.isDebugEnabled())
                    log.debug("Write entries to db one by one using update and insert statements " +
                        "[cache=" + U.maskName(cacheName) + ", cnt=" + entries.size() + "]");

                PreparedStatement insStmt = null;

                PreparedStatement updStmt = null;

                try {
                    for (Cache.Entry<? extends K, ? extends V> entry : entries) {
                        K key = entry.getKey();

                        Object keyTypeId = typeIdForObject(key);

                        EntryMapping em = entryMapping(cacheName, keyTypeId);

                        if (currKeyTypeId == null || !currKeyTypeId.equals(keyTypeId)) {
                            U.closeQuiet(insStmt);

                            insStmt = conn.prepareStatement(em.insQry);

                            U.closeQuiet(updStmt);

                            updStmt = conn.prepareStatement(em.updQry);

                            currKeyTypeId = keyTypeId;
                        }

                        writeUpsert(insStmt, updStmt, em, entry);
                    }
                }
                finally {
                    U.closeQuiet(insStmt);

                    U.closeQuiet(updStmt);
                }
            }
        }
        catch (SQLException e) {
            throw new CacheWriterException("Failed to write entries in database", e);
        }
        finally {
            closeConnection(conn);
        }
    }

    /** {@inheritDoc} */
    @Override public void delete(Object key) throws CacheWriterException {
        assert key != null;

        EntryMapping em = entryMapping(session().cacheName(), typeIdForObject(key));

        if (log.isDebugEnabled())
            log.debug("Remove value from db [table=" + em.fullTableName() + ", key=" + key + "]");

        Connection conn = null;

        PreparedStatement stmt = null;

        try {
            conn = connection();

            stmt = conn.prepareStatement(em.remQry);

            fillKeyParameters(stmt, em, key);

            int delCnt = stmt.executeUpdate();

            if (delCnt != 1)
                U.warn(log, "Unexpected number of deleted entries [table=" + em.fullTableName() + ", key=" + key +
                    ", expected=1, actual=" + delCnt + "]");
        }
        catch (SQLException e) {
            throw new CacheWriterException("Failed to remove value from database [table=" + em.fullTableName() +
                ", key=" + key + "]", e);
        }
        finally {
            end(conn, stmt);
        }
    }

    /**
     * @param em Entry mapping.
     * @param stmt Statement.
     * @param desc Statement description for error message.
     * @param fromIdx Objects in batch start from index.
     * @param prepared Expected objects in batch.
     * @param lazyObjs All objects used in batch statement as array.
     * @throws SQLException If failed to execute batch statement.
     */
    private void executeBatch(EntryMapping em, Statement stmt, String desc, int fromIdx, int prepared,
        LazyValue<Object[]> lazyObjs) throws SQLException {
        try {
            int[] rowCounts = stmt.executeBatch();

            int numOfRowCnt = rowCounts.length;

            if (numOfRowCnt != prepared)
                U.warn(log, "Unexpected number of updated rows [table=" + em.fullTableName() + ", expected=" + prepared +
                    ", actual=" + numOfRowCnt + "]");

            for (int i = 0; i < numOfRowCnt; i++) {
                int cnt = rowCounts[i];

                if (cnt != 1 && cnt != SUCCESS_NO_INFO) {
                    Object[] objs = lazyObjs.value();

                    U.warn(log, "Batch " + desc + " returned unexpected updated row count [table=" + em.fullTableName() +
                        ", entry=" + objs[fromIdx + i] + ", expected=1, actual=" + cnt + "]");
                }
            }
        }
        catch (BatchUpdateException be) {
            int[] rowCounts = be.getUpdateCounts();

            for (int i = 0; i < rowCounts.length; i++) {
                if (rowCounts[i] == EXECUTE_FAILED) {
                    Object[] objs = lazyObjs.value();

                    U.warn(log, "Batch " + desc + " failed on execution [table=" + em.fullTableName() +
                        ", entry=" + objs[fromIdx + i] + "]");
                }
            }

            throw be;
        }
    }

    /** {@inheritDoc} */
    @Override public void deleteAll(final Collection<?> keys) throws CacheWriterException {
        assert keys != null;

        Connection conn = null;

        try {
            conn = connection();

            LazyValue<Object[]> lazyKeys = new LazyValue<Object[]>() {
                @Override public Object[] create() {
                    return keys.toArray();
                }
            };

            String cacheName = session().cacheName();

            Object currKeyTypeId = null;

            EntryMapping em = null;

            PreparedStatement delStmt = null;

            int fromIdx = 0, prepared = 0;

            for (Object key : keys) {
                Object keyTypeId = typeIdForObject(key);

                em = entryMapping(cacheName, keyTypeId);

                if (delStmt == null) {
                    delStmt = conn.prepareStatement(em.remQry);

                    currKeyTypeId = keyTypeId;
                }

                if (!currKeyTypeId.equals(keyTypeId)) {
                    if (log.isDebugEnabled())
                        log.debug("Delete entries from db [cache=" + U.maskName(cacheName) +
                            ", keyType=" + em.keyType() + ", cnt=" + prepared + "]");

                    executeBatch(em, delStmt, "deleteAll", fromIdx, prepared, lazyKeys);

                    fromIdx += prepared;

                    prepared = 0;

                    currKeyTypeId = keyTypeId;
                }

                fillKeyParameters(delStmt, em, key);

                delStmt.addBatch();

                if (++prepared % batchSize == 0) {
                    if (log.isDebugEnabled())
                        log.debug("Delete entries from db [cache=" + U.maskName(cacheName) +
                            ", keyType=" + em.keyType() + ", cnt=" + prepared + "]");

                    executeBatch(em, delStmt, "deleteAll", fromIdx, prepared, lazyKeys);

                    fromIdx += prepared;

                    prepared = 0;
                }
            }

            if (delStmt != null && prepared % batchSize != 0) {
                if (log.isDebugEnabled())
                    log.debug("Delete entries from db [cache=" + U.maskName(cacheName) +
                        ", keyType=" + em.keyType() + ", cnt=" + prepared + "]");

                executeBatch(em, delStmt, "deleteAll", fromIdx, prepared, lazyKeys);
            }
        }
        catch (SQLException e) {
            throw new CacheWriterException("Failed to remove values from database", e);
        }
        finally {
            closeConnection(conn);
        }
    }

    /**
     * Sets the value of the designated parameter using the given object.
     *
     * @param stmt Prepare statement.
     * @param idx Index for parameters.
     * @param field Field descriptor.
     * @param fieldVal Field value.
     * @throws CacheException If failed to set statement parameter.
     */
    protected void fillParameter(PreparedStatement stmt, int idx, JdbcTypeField field, @Nullable Object fieldVal)
        throws CacheException {
        try {
            if (fieldVal != null) {
                if (field.getJavaFieldType() == UUID.class) {
                    switch (field.getDatabaseFieldType()) {
                        case Types.BINARY:
                            fieldVal = U.uuidToBytes((UUID)fieldVal);

                            break;
                        case Types.CHAR:
                        case Types.VARCHAR:
                            fieldVal = fieldVal.toString();

                            break;
                        default:
                            // No-op.
                    }
                }
                else if (field.getJavaFieldType().isEnum()) {
                    if (fieldVal instanceof Enum) {
                        Enum val = (Enum)fieldVal;

                        fieldVal = NUMERIC_TYPES.contains(field.getDatabaseFieldType()) ? val.ordinal() : val.name();
                    }
                    else if (fieldVal instanceof BinaryEnumObjectImpl) {
                        BinaryEnumObjectImpl val = (BinaryEnumObjectImpl)fieldVal;

                        fieldVal = val.enumOrdinal();
                    }
                }

                stmt.setObject(idx, fieldVal);
            }
            else
                stmt.setNull(idx, field.getDatabaseFieldType());
        }
        catch (SQLException e) {
            throw new CacheException("Failed to set statement parameter name: " + field.getDatabaseFieldName(), e);
        }
    }

    /**
     * @param stmt Prepare statement.
     * @param idx Start index for parameters.
     * @param em Entry mapping.
     * @param key Key object.
     * @return Next index for parameters.
     * @throws CacheException If failed to set statement parameters.
     */
    protected int fillKeyParameters(PreparedStatement stmt, int idx, EntryMapping em, Object key) throws CacheException {
        for (JdbcTypeField field : em.keyColumns()) {
            Object fieldVal = extractParameter(em.cacheName, em.keyType(), em.keyKind(), field.getJavaFieldName(), key);

            fillParameter(stmt, idx++, field, fieldVal);
        }

        return idx;
    }

    /**
     * @param stmt Prepare statement.
     * @param m Type mapping description.
     * @param key Key object.
     * @return Next index for parameters.
     * @throws CacheException If failed to set statement parameters.
     */
    protected int fillKeyParameters(PreparedStatement stmt, EntryMapping m, Object key) throws CacheException {
        return fillKeyParameters(stmt, 1, m, key);
    }

    /**
     * @param stmt Prepare statement.
     * @param idx Start index for parameters.
     * @param em Type mapping description.
     * @param val Value object.
     * @return Next index for parameters.
     * @throws CacheException If failed to set statement parameters.
     */
    protected int fillValueParameters(PreparedStatement stmt, int idx, EntryMapping em, Object val)
        throws CacheWriterException {
        for (JdbcTypeField field : em.uniqValFlds) {
            Object fieldVal = extractParameter(
                em.cacheName,
                em.valueType(),
                em.valueKind(),
                field.getJavaFieldName(),
                val
            );

            fillParameter(stmt, idx++, field, fieldVal);
        }

        return idx;
    }

    /**
     * @return Data source.
     */
    public DataSource getDataSource() {
        return dataSrc;
    }

    /**
     * @param dataSrc Data source.
     */
    public void setDataSource(DataSource dataSrc) {
        this.dataSrc = dataSrc;
    }

    /**
     * Get database dialect.
     *
     * @return Database dialect.
     */
    public JdbcDialect getDialect() {
        return dialect;
    }

    /**
     * Set database dialect.
     *
     * @param dialect Database dialect.
     */
    public void setDialect(JdbcDialect dialect) {
        this.dialect = dialect;
    }

    /**
     * Get Max workers thread count. These threads are responsible for execute query.
     *
     * @return Max workers thread count.
     */
    public int getMaximumPoolSize() {
        return maxPoolSize;
    }

    /**
     * Set Max workers thread count. These threads are responsible for execute query.
     *
     * @param maxPoolSize Max workers thread count.
     */
    public void setMaximumPoolSize(int maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
    }

    /**
     * Gets maximum number of write attempts in case of database error.
     *
     * @return Maximum number of write attempts.
     */
    public int getMaximumWriteAttempts() {
        return maxWrtAttempts;
    }

    /**
     * Sets maximum number of write attempts in case of database error.
     *
     * @param maxWrtAttempts Number of write attempts.
     */
    public void setMaximumWriteAttempts(int maxWrtAttempts) {
        this.maxWrtAttempts = maxWrtAttempts;
    }

    /**
     * Gets types known by store.
     *
     * @return Types known by store.
     */
    public JdbcType[] getTypes() {
        return types;
    }

    /**
     * Sets store configurations.
     *
     * @param types Store should process.
     */
    public void setTypes(JdbcType... types) {
        this.types = types;
    }

    /**
     * Gets hash code calculator.
     *
     * @return Hash code calculator.
     */
    public JdbcTypeHasher getHasher() {
        return hasher;
    }

    /**
     * Sets hash code calculator.
     *
     * @param hasher Hash code calculator.
     */
    public void setHasher(JdbcTypeHasher hasher) {
        this.hasher = hasher;
    }

    /**
     * Gets types transformer.
     *
     * @return Types transformer.
     */
    public JdbcTypesTransformer getTransformer() {
        return transformer;
    }

    /**
     * Sets types transformer.
     *
     * @param transformer Types transformer.
     */
    public void setTransformer(JdbcTypesTransformer transformer) {
        this.transformer = transformer;
    }

    /**
     * Get maximum batch size for delete and delete operations.
     *
     * @return Maximum batch size.
     */
    public int getBatchSize() {
        return batchSize;
    }

    /**
     * Set maximum batch size for write and delete operations.
     *
     * @param batchSize Maximum batch size.
     */
    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    /**
     * Parallel load cache minimum row count threshold.
     *
     * @return If {@code 0} then load sequentially.
     */
    public int getParallelLoadCacheMinimumThreshold() {
        return parallelLoadCacheMinThreshold;
    }

    /**
     * Parallel load cache minimum row count threshold.
     *
     * @param parallelLoadCacheMinThreshold Minimum row count threshold. If {@code 0} then load sequentially.
     */
    public void setParallelLoadCacheMinimumThreshold(int parallelLoadCacheMinThreshold) {
        this.parallelLoadCacheMinThreshold = parallelLoadCacheMinThreshold;
    }

    /**
     * If {@code true} all the SQL table and field names will be escaped with double quotes like
     * ({@code "tableName"."fieldsName"}). This enforces case sensitivity for field names and
     * also allows having special characters in table and field names.
     *
     * @return Flag value.
     */
    public boolean isSqlEscapeAll() {
        return sqlEscapeAll;
    }

    /**
     * If {@code true} all the SQL table and field names will be escaped with double quotes like
     * ({@code "tableName"."fieldsName"}). This enforces case sensitivity for field names and
     * also allows having special characters in table and field names.
     *
     * @param sqlEscapeAll Flag value.
     */
    public void setSqlEscapeAll(boolean sqlEscapeAll) {
        this.sqlEscapeAll = sqlEscapeAll;
    }

    /**
     * @return Ignite instance.
     */
    protected Ignite ignite() {
        return ignite;
    }

    /**
     * @return Store session.
     */
    protected CacheStoreSession session() {
        return ses;
    }

    /**
     * Type kind.
     */
    protected enum TypeKind {
        /** Type is known as Java built in type, like {@link String} */
        BUILT_IN,
        /** Class for this type is available. */
        POJO,
        /** Class for this type is not available. */
        BINARY
    }

    /**
     * Entry mapping description.
     */
    protected static class EntryMapping {
        /** Cache name. */
        private final String cacheName;

        /** Database dialect. */
        private final JdbcDialect dialect;

        /** Select border for range queries. */
        private final String loadCacheSelRangeQry;

        /** Select all items query. */
        private final String loadCacheQry;

        /** Select item query. */
        private final String loadQrySingle;

        /** Select items query. */
        private final String loadQry;

        /** Merge item(s) query. */
        private final String mergeQry;

        /** Update item query. */
        private final String insQry;

        /** Update item query. */
        private final String updQry;

        /** Remove item(s) query. */
        private final String remQry;

        /** Max key count for load query per statement. */
        private final int maxKeysPerStmt;

        /** Database key columns. */
        private final Collection<String> keyCols;

        /** Database key columns prepared for building SQL queries.. */
        private final Collection<String> sqlKeyCols;

        /** Database unique value columns. */
        private final Collection<String> cols;

        /** Database unique value columns prepared for building SQL queries. */
        private final Collection<String> sqlCols;

        /** Select query columns index. */
        private final Map<String, Integer> loadColIdxs;

        /** Unique value fields. */
        private final Collection<JdbcTypeField> uniqValFlds;

        /** Type metadata. */
        private final JdbcType typeMeta;

        /** Key type kind. */
        private final TypeKind keyKind;

        /** Value type kind. */
        private final TypeKind valKind;

        /** Full table name. */
        private final String fullTblName;

        /** Full table name prepared for building SQL queries. */
        private final String sqlFullTblName;

        /**
         * Escape collection of column names.
         * @param dialect Database dialect.
         * @param cols Columns.
         * @return Collection of escaped names.
         */
        private static Collection<String> escape(JdbcDialect dialect, Collection<String> cols) {
            Collection<String> res = new ArrayList<>(cols.size());

            for (String col : cols)
                res.add(dialect.escape(col));

            return res;
        }

        /**
         * @param cacheName Cache name.
         * @param dialect JDBC dialect.
         * @param typeMeta Type metadata.
         * @param keyKind Type kind.
         * @param valKind Value kind.
         * @param escape Escape SQL identifiers flag.
         */
        public EntryMapping(@Nullable String cacheName, JdbcDialect dialect, JdbcType typeMeta,
            TypeKind keyKind, TypeKind valKind, boolean escape) {
            this.cacheName = cacheName;

            this.dialect = dialect;

            this.typeMeta = typeMeta;

            this.keyKind = keyKind;

            this.valKind = valKind;

            JdbcTypeField[] keyFields = typeMeta.getKeyFields();

            JdbcTypeField[] valFields = typeMeta.getValueFields();

            keyCols = databaseColumns(F.asList(keyFields));

            uniqValFlds = F.view(F.asList(valFields), new IgnitePredicate<JdbcTypeField>() {
                @Override public boolean apply(JdbcTypeField col) {
                    return !keyCols.contains(col.getDatabaseFieldName());
                }
            });

            String schema = typeMeta.getDatabaseSchema();

            String tblName = typeMeta.getDatabaseTable();

            Collection<String> uniqueValCols = databaseColumns(uniqValFlds);

            cols = F.concat(false, keyCols, uniqueValCols);

            loadColIdxs = U.newHashMap(cols.size());

            int idx = 1;

            for (String col : cols)
                loadColIdxs.put(col.toUpperCase(), idx++);

            fullTblName = F.isEmpty(schema) ? tblName : schema + "." + tblName;

            Collection<String> sqlUniqueValCols;

            if (escape) {
                sqlFullTblName = F.isEmpty(schema) ? dialect.escape(tblName) : dialect.escape(schema) + "." + dialect.escape(tblName);

                sqlCols = escape(dialect, cols);
                sqlKeyCols = escape(dialect, keyCols);
                sqlUniqueValCols = escape(dialect, uniqueValCols);
            }
            else {
                sqlFullTblName = fullTblName;
                sqlCols = cols;
                sqlKeyCols = keyCols;
                sqlUniqueValCols = uniqueValCols;
            }

            loadCacheQry = dialect.loadCacheQuery(sqlFullTblName, sqlCols);

            loadCacheSelRangeQry = dialect.loadCacheSelectRangeQuery(sqlFullTblName, sqlKeyCols);

            loadQrySingle = dialect.loadQuery(sqlFullTblName, sqlKeyCols, sqlCols, 1);

            maxKeysPerStmt = dialect.getMaxParameterCount() / sqlKeyCols.size();

            loadQry = dialect.loadQuery(sqlFullTblName, sqlKeyCols, sqlCols, maxKeysPerStmt);

            insQry = dialect.insertQuery(sqlFullTblName, sqlKeyCols, sqlUniqueValCols);

            updQry = dialect.updateQuery(sqlFullTblName, sqlKeyCols, sqlUniqueValCols);

            mergeQry = dialect.mergeQuery(sqlFullTblName, sqlKeyCols, sqlUniqueValCols);

            remQry = dialect.removeQuery(sqlFullTblName, sqlKeyCols);
        }

        /**
         * Extract database column names from {@link JdbcTypeField}.
         *
         * @param dsc collection of {@link JdbcTypeField}.
         * @return Collection with database column names.
         */
        private static Collection<String> databaseColumns(Collection<JdbcTypeField> dsc) {
            return F.transform(dsc, new C1<JdbcTypeField, String>() {
                /** {@inheritDoc} */
                @Override public String apply(JdbcTypeField col) {
                    return col.getDatabaseFieldName();
                }
            });
        }

        /**
         * @return Key type.
         */
        protected String keyType() {
            return typeMeta.getKeyType();
        }

        /**
         * @return Key type kind.
         */
        protected TypeKind keyKind() {
            return keyKind;
        }

        /**
         * @return Value type.
         */
        protected String valueType() {
            return typeMeta.getValueType();
        }

        /**
         * @return Value type kind.
         */
        protected TypeKind valueKind() {
            return valKind;
        }

        /**
         * Construct query for select values with key count less or equal {@code maxKeysPerStmt}
         *
         * @param keyCnt Key count.
         * @return Load query statement text.
         */
        protected String loadQuery(int keyCnt) {
            assert keyCnt <= maxKeysPerStmt;

            if (keyCnt == maxKeysPerStmt)
                return loadQry;

            if (keyCnt == 1)
                return loadQrySingle;

            return dialect.loadQuery(sqlFullTblName, sqlKeyCols, sqlCols, keyCnt);
        }

        /**
         * Construct query for select values in range.
         *
         * @param appendLowerBound Need add lower bound for range.
         * @param appendUpperBound Need add upper bound for range.
         * @return Query with range.
         */
        protected String loadCacheRangeQuery(boolean appendLowerBound, boolean appendUpperBound) {
            return dialect.loadCacheRangeQuery(sqlFullTblName, sqlKeyCols, sqlCols, appendLowerBound, appendUpperBound);
        }

        /**
         * Gets key columns.
         *
         * @return Key columns.
         */
        protected JdbcTypeField[] keyColumns() {
            return typeMeta.getKeyFields();
        }

        /**
         * Gets value columns.
         *
         * @return Value columns.
         */
        protected JdbcTypeField[] valueColumns() {
            return typeMeta.getValueFields();
        }

        /**
         * Get full table name.
         *
         * @return &lt;schema&gt;.&lt;table name&gt
         */
        protected String fullTableName() {
            return fullTblName;
        }
    }

    /**
     * Worker for load cache using custom user query.
     *
     * @param <K1> Key type.
     * @param <V1> Value type.
     */
    private class LoadCacheCustomQueryWorker<K1, V1> implements Callable<Void> {
        /** Entry mapping description. */
        private final EntryMapping em;

        /** User statement. */
        private PreparedStatement stmt;

        /** User query. */
        private String qry;

        /** Closure for loaded values. */
        private final IgniteBiInClosure<K1, V1> clo;

        /**
         * @param em Entry mapping description.
         * @param stmt User statement.
         * @param clo Closure for loaded values.
         */
        private LoadCacheCustomQueryWorker(EntryMapping em, PreparedStatement stmt, IgniteBiInClosure<K1, V1> clo) {
            this.em = em;
            this.stmt = stmt;
            this.clo = clo;
        }

        /**
         * @param em Entry mapping description.
         * @param qry User query.
         * @param clo Closure for loaded values.
         */
        private LoadCacheCustomQueryWorker(EntryMapping em, String qry, IgniteBiInClosure<K1, V1> clo) {
            this.em = em;
            this.qry = qry;
            this.clo = clo;
        }

        /** {@inheritDoc} */
        @Override public Void call() throws Exception {
            Connection conn = null;

            try {
                if (stmt == null) {
                    conn = openConnection(true);

                    stmt = conn.prepareStatement(qry);
                }

                stmt.setFetchSize(dialect.getFetchSize());

                ResultSet rs = stmt.executeQuery();

                ResultSetMetaData meta = rs.getMetaData();

                Map<String, Integer> colIdxs = U.newHashMap(meta.getColumnCount());

                for (int i = 1; i <= meta.getColumnCount(); i++)
                    colIdxs.put(meta.getColumnLabel(i).toUpperCase(), i);

                while (rs.next()) {
                    K1 key = buildObject(em.cacheName, em.keyType(), em.keyKind(), em.keyColumns(), colIdxs, rs);
                    V1 val = buildObject(em.cacheName, em.valueType(), em.valueKind(), em.valueColumns(), colIdxs, rs);

                    clo.apply(key, val);
                }

                return null;
            }
            catch (SQLException e) {
                throw new CacheLoaderException("Failed to execute custom query for load cache", e);
            }
            finally {
                if (conn != null) {
                    U.closeQuiet(stmt);

                    U.closeQuiet(conn);
                }
            }
        }
    }

    /**
     * Lazy initialization of value.
     *
     * @param <T> Cached object type
     */
    private abstract static class LazyValue<T> {
        /** Cached value. */
        private T val;

        /**
         * @return Construct value.
         */
        protected abstract T create();

        /**
         * @return Value.
         */
        public T value() {
            if (val == null)
                val = create();

            return val;
        }
    }

    /**
     * Worker for load by keys.
     *
     * @param <K1> Key type.
     * @param <V1> Value type.
     */
    private class LoadWorker<K1, V1> implements Callable<Map<K1, V1>> {
        /** Connection. */
        private final Connection conn;

        /** Keys for load. */
        private final Collection<K1> keys;

        /** Entry mapping description. */
        private final EntryMapping em;

        /**
         * @param conn Connection.
         * @param em Entry mapping description.
         */
        private LoadWorker(Connection conn, EntryMapping em) {
            this.conn = conn;
            this.em = em;

            keys = new ArrayList<>(em.maxKeysPerStmt);
        }

        /** {@inheritDoc} */
        @Override public Map<K1, V1> call() throws Exception {
            if (log.isDebugEnabled())
                log.debug("Load values from db [table= " + em.fullTableName() + ", keysCnt=" + keys.size() + "]");

            PreparedStatement stmt = null;

            try {
                stmt = conn.prepareStatement(em.loadQuery(keys.size()));

                int idx = 1;

                for (Object key : keys) {
                    for (JdbcTypeField field : em.keyColumns()) {
                        Object fieldVal = extractParameter(em.cacheName, em.keyType(), em.keyKind(), field.getJavaFieldName(), key);

                        fillParameter(stmt, idx++, field, fieldVal);
                    }
                }

                ResultSet rs = stmt.executeQuery();

                Map<K1, V1> entries = U.newHashMap(keys.size());

                while (rs.next()) {
                    K1 key = buildObject(em.cacheName, em.keyType(), em.keyKind(), em.keyColumns(),
                        em.loadColIdxs, rs);

                    V1 val = buildObject(em.cacheName, em.valueType(), em.valueKind(), em.valueColumns(),
                        em.loadColIdxs, rs);

                    entries.put(key, val);
                }

                return entries;
            }
            finally {
                U.closeQuiet(stmt);
            }
        }
    }
}