package fr.ifremer.adagio.synchro.dao;

/*
 * #%L
 * SIH-Adagio :: Synchronization
 * $Id:$
 * $HeadURL:$
 * %%
 * Copyright (C) 2012 - 2014 Ifremer
 * %%
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 * #L%
 */

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import fr.ifremer.adagio.synchro.SynchroTechnicalException;
import fr.ifremer.adagio.synchro.config.SynchroConfiguration;
import fr.ifremer.adagio.synchro.intercept.SynchroInterceptor;
import fr.ifremer.adagio.synchro.intercept.SynchroInterceptorBase;
import fr.ifremer.adagio.synchro.intercept.SynchroInterceptorUtils;
import fr.ifremer.adagio.synchro.intercept.SynchroWriteBuffer;
import fr.ifremer.adagio.synchro.meta.SynchroDatabaseMetadata;
import fr.ifremer.adagio.synchro.meta.SynchroTableMetadata;
import fr.ifremer.adagio.synchro.meta.SynchroTableMetadata.TableInsertStrategy;
import fr.ifremer.adagio.synchro.service.SynchroContext;
import fr.ifremer.adagio.synchro.service.SynchroPendingOperationBuffer;

public class SynchroTableDaoImpl implements SynchroTableDao {

    /** Logger. */
    private static final Log log =
            LogFactory.getLog(SynchroTableDaoImpl.class);

    public static final String TABLE_TEMP_QUERY_PARAMETER = "temp_query_parameter";

    protected final Connection connection;

    protected final SynchroDatabaseMetadata dbMeta;

    protected final SynchroTableMetadata table;

    protected final PreparedStatement insertStatement;

    protected final PreparedStatement insertQueryParameterStatement;

    protected final PreparedStatement updateStatement;

    protected final List<PreparedStatement> selectStatements;

    protected final boolean insertStrategyGenerateIdFirst;

    protected final int columnCount;

    protected final int batchSize;

    protected final String tableName;

    protected boolean debug;

    protected int insertCount = 0;

    protected int updateCount = 0;

    protected final int idColumnIndex;

    protected final SynchroInterceptor readInterceptor;

    protected final SynchroInterceptor writeInterceptor;

    protected SynchroPendingOperationBuffer pendingOperationBuffer;

    public SynchroTableDaoImpl(
            Connection connection,
            SynchroTableMetadata table,
            boolean enableWriteStatements) throws SQLException {
        this.connection = connection;
        this.table = table;
        this.dbMeta = table.getDatabaseMetadata();
        this.columnCount = table.getColumnsCount();
        this.selectStatements = Lists.newArrayList();
        this.tableName = table.getName();

        // Batch size
        this.batchSize = SynchroConfiguration.getInstance().getImportJdbcBatchSize();
        Preconditions.checkArgument(this.batchSize > 0);

        debug = log.isTraceEnabled();

        if (table.isWithIdColumn()) {
            this.idColumnIndex = table.getColumnIndex(SynchroTableMetadata.COLUMN_ID) + 1;
        }
        else {
            this.idColumnIndex = -1;
        }

        if (enableWriteStatements) {
            // Prepare insert statement
            String insertSql = table.getInsertQuery();
            Preconditions.checkArgument(insertSql.toUpperCase().startsWith("INSERT"),
                    String.format("%s Insert SQL query should be like 'INSERT ...' but was: %s", this.tableName, insertSql));
            this.insertStatement = connection.prepareStatement(insertSql);
            this.insertStrategyGenerateIdFirst = table.getInsertStrategy() == TableInsertStrategy.GENERATE_ID_FIRST;

            // Prepare update statement
            String updateSql = table.getUpdateQuery();
            Preconditions.checkArgument(updateSql.toUpperCase().startsWith("UPDATE"),
                    String.format("%s Update SQL query should be like 'UPDATE ...' but was: %s", this.tableName, updateSql));
            this.updateStatement = connection.prepareStatement(updateSql);

            // Init write interceptors
            this.writeInterceptor = initWriteInterceptor(table.getInterceptors());
        }
        else {
            this.insertStatement = null;
            this.updateStatement = null;
            this.insertStrategyGenerateIdFirst = false;
            this.writeInterceptor = null;
        }

        // Query temp parameter (for query parameter with many values)
        String insertQueryParameterSql = createTempQueryParameterQuery(table);
        if (insertQueryParameterSql != null) {
            insertQueryParameterStatement = getConnection().prepareStatement(insertQueryParameterSql);
        }
        else {
            insertQueryParameterStatement = null;
        }

        this.readInterceptor = initReadInterceptor(table.getInterceptors());
    }

    @Override
    public Connection getConnection() {
        return connection;
    }

    @Override
    public SynchroTableMetadata getTable() {
        return table;
    }

    public void setPendingOperationBuffer(SynchroPendingOperationBuffer pendingChangesBuffer) {
        this.pendingOperationBuffer = pendingChangesBuffer;
    }

    public SynchroPendingOperationBuffer getPendingOperationBuffer() {
        return this.pendingOperationBuffer;
    }

    public int getInsertCount() {
        return insertCount;
    }

    public int getUpdateCount() {
        return updateCount;
    }

    public void flush() throws SQLException {
        if (!debug) {
            if (insertCount > 0 && insertCount % batchSize != 0) {
                insertStatement.executeBatch();
                insertStatement.clearBatch();
            }
            if (updateCount > 0 && updateCount % batchSize != 0) {
                updateStatement.executeBatch();
                updateStatement.clearBatch();
            }
        }
    }

    @Override
    public void close() throws IOException {
        DaoUtils.closeSilently(insertStatement);
        DaoUtils.closeSilently(updateStatement);
        DaoUtils.closeSilently(insertQueryParameterStatement);
        closeSelectStatements();
    }

    /**
     * Gets the last updateDate for the given {@code table} using
     * the given datasource.
     * 
     * @return the last update date of the given table, or {@code null} if table does not use a updateDate columns or if
     *         there
     *         is no data in table.
     */
    public Timestamp getLastUpdateDate() throws SQLException {
        if (!table.isWithUpdateDateColumn()) {
            return null;
        }

        String sql = table.getSelectMaxUpdateDateQuery();
        if (sql == null) {
            return null;
        }

        PreparedStatement statement = getConnection().prepareStatement(sql);
        try {
            ResultSet resultSet = statement.executeQuery();
            if (!resultSet.next()) {
                return null;
            }
            Timestamp result = resultSet.getTimestamp(1);
            return result;
        } finally {
            DaoUtils.closeSilently(statement);
        }
    }

    public long count() throws SQLException {

        String sql = table.getCountQuery();

        PreparedStatement statement = getConnection().prepareStatement(sql);

        try {
            ResultSet resultSet = statement.executeQuery();
            resultSet.next();
            long result = resultSet.getLong(1);
            return result;
        } finally {
            DaoUtils.closeSilently(statement);
        }
    }

    public long countDataToUpdate(Date fromDate) throws SQLException {

        String sql = table.getCountDataToUpdateQuery(fromDate);

        PreparedStatement statement = getConnection().prepareStatement(sql);
        log.debug(sql);
        try {
            if (table.isWithUpdateDateColumn() &&
                    fromDate != null) {
                Timestamp updateDateValue = new Timestamp(fromDate.getTime());
                // Set all query parameter (could have more than one,
                // when query has been override by an interceptor)
                for (int i = 1; i <= statement.getParameterMetaData().getParameterCount(); i++) {
                    statement.setTimestamp(i, updateDateValue);
                }
            }

            ResultSet queryResult = statement.executeQuery();
            queryResult.next();
            long result = queryResult.getLong(1);
            return result;
        } finally {
            DaoUtils.closeSilently(statement);
        }
    }

    public ResultSet getDataToUpdate(Date fromDate) throws SQLException {

        String sql = table.getSelectDataToUpdateQuery(fromDate);

        PreparedStatement statement = getConnection().prepareStatement(sql);
        selectStatements.add(statement);

        if (table.isWithUpdateDateColumn() &&
                fromDate != null) {
            statement.setTimestamp(1, new Timestamp(fromDate.getTime()));
            if (statement.getParameterMetaData().getParameterCount() > 1) {
                statement.setTimestamp(2, new Timestamp(fromDate.getTime()));
            }
        }
        statement.setFetchSize(batchSize);

        ResultSet result = statement.executeQuery();
        return result;
    }

    public void deleteAll() throws SQLException {
        PreparedStatement deleteStatement =
                getConnection().prepareStatement(
                        "DELETE FROM " + table.getName());
        try {
            deleteStatement.execute();
        } finally {
            DaoUtils.closeSilently(deleteStatement);
        }
    }

    public Object[] findByPk(List<Object> pk) throws SQLException {
        String selectDataSql = table.getSelectDataQueryFromPk();

        PreparedStatement selectStatement = getConnection().prepareStatement(selectDataSql);
        int columnCountIndex = 1;

        for (Object pkColumn : pk) {
            selectStatement.setObject(columnCountIndex++, pkColumn);
        }

        int columnsCount = table.getColumnsCount();

        try {
            ResultSet resultSet = selectStatement.executeQuery();
            resultSet.next();

            Object[] result = new Object[columnsCount];

            for (int i = 1; i <= columnsCount; i++) {

                result[i - 1] = resultSet.getObject(i);
            }
            return result;
        } finally {
            DaoUtils.closeSilently(selectStatement);
        }
    }

    public Set<String> getExistingPrimaryKeys() throws SQLException {
        Set<String> result = Sets.newHashSet();

        // If simple SQL (select with only one output String column)
        if (table.isSelectPrimaryKeysAsStringQueryEnable()) {
            String sql = table.getSelectPrimaryKeysAsStringQuery();
            PreparedStatement statement = getConnection().prepareStatement(sql);
            try {
                ResultSet resultSet = statement.executeQuery();

                while (resultSet.next()) {
                    String pk = resultSet.getString(1);
                    result.add(pk);
                }
            } finally {
                DaoUtils.closeSilently(statement);
            }
        }

        // If more than one one column as String, in the select query
        else {
            String sql = table.getSelectPrimaryKeysQuery();
            PreparedStatement statement = getConnection().prepareStatement(sql);

            try {
                ResultSet resultSet = statement.executeQuery();

                int columnCount = table.getPkNames().size();
                List<Object> pks = Lists.newArrayListWithCapacity(columnCount);
                while (resultSet.next()) {
                    for (int i = 1; i <= columnCount; i++) {
                        Object pk = resultSet.getObject(i);
                        pks.add(pk);
                    }
                    result.add(table.toPkStr(pks));
                    pks.clear();
                }
            } finally {
                DaoUtils.closeSilently(statement);
            }
        }

        return result;
    }

    public Integer generateNewId() throws SQLException {
        String sql = table.getSequenceNextValString();
        Preconditions.checkNotNull(sql);

        if (sql == null) {
            return null;
        }

        PreparedStatement statement = getConnection().prepareStatement(sql);
        try {
            ResultSet resultSet = statement.executeQuery();
            if (!resultSet.next()) {
                return null;
            }
            Integer result = resultSet.getInt(1);
            return result;
        } finally {
            DaoUtils.closeSilently(statement);
        }
    }

    public void executeInsert(ResultSet incomingData) throws SQLException {
        Preconditions.checkArgument(!this.insertStrategyGenerateIdFirst);

        List<Object> params = null;
        if (debug) {
            params = Lists.newArrayList();
        }

        if (writeInterceptor != null) {
            transformAndSetData(insertStatement, incomingData, pendingOperationBuffer, writeInterceptor, params);
        }
        else {
            setData(insertStatement, incomingData, params);
        }
        insertCount++;

        if (debug) {
            List<Object> pk = table.getPk(incomingData);
            log.debug(String.format("%s Execute insert query (pk:%s), params: %s", tableName, table.toPkStr(pk), params));
            int nbRowInsert = insertStatement.executeUpdate();
            if (nbRowInsert != 1) {
                throw new SynchroTechnicalException(String.format("%s Could not insert a row into the table (pk:%s), params: %s", tableName,
                        table.toPkStr(pk), params));
            }
        }
        else {
            insertStatement.addBatch();
            if (insertCount > 0 && insertCount % batchSize == 0) {
                insertStatement.executeBatch();
                insertStatement.clearBatch();
            }
        }
    }

    public void executeInsert(Object[] incomingData) throws SQLException {
        Preconditions.checkArgument(!this.insertStrategyGenerateIdFirst);

        List<Object> params = null;
        if (debug) {
            params = Lists.newArrayList();
        }

        if (writeInterceptor != null) {
            transformAndSetData(insertStatement, incomingData, pendingOperationBuffer, writeInterceptor, params);
        }
        else {
            setData(insertStatement, incomingData, params);
        }
        insertCount++;

        if (debug) {
            List<Object> pk = table.getPk(incomingData);
            log.debug(String.format("%s Execute insert query (pk:%s), params: %s", tableName, table.toPkStr(pk), params));
            insertStatement.executeUpdate();
        }
        else {
            insertStatement.addBatch();
            if (insertCount > 0 && insertCount % batchSize == 0) {
                insertStatement.executeBatch();
                insertStatement.clearBatch();
            }
        }
    }

    @Override
    public Integer executeInsertAndReturnId(ResultSet incomingData) throws SQLException {
        Preconditions.checkArgument(this.insertStrategyGenerateIdFirst);

        // Generate Id
        Integer id = generateNewId();
        List<Object> params = null;

        if (debug) {
            params = Lists.newArrayList();
        }

        if (writeInterceptor != null) {
            transformAndSetData(insertStatement, incomingData, pendingOperationBuffer, writeInterceptor, params);
        }
        else {
            setData(insertStatement, incomingData, params);
        }

        insertStatement.setObject(columnCount + 1, id);
        insertCount++;

        if (debug) {
            log.debug(String.format("%s Execute insert query (pk:%s), params: %s", tableName, id, params));
            insertStatement.executeUpdate();
        }
        else {
            insertStatement.addBatch();
            if (insertCount > 0 && insertCount % batchSize == 0) {
                insertStatement.executeBatch();
                insertStatement.clearBatch();
            }
        }

        return id;
    }

    @Override
    public void executeUpdate(List<Object> pk, ResultSet incomingData) throws SQLException {

        List<Object> params = null;
        if (debug) {
            params = Lists.newArrayList();
        }

        if (writeInterceptor != null) {
            transformAndSetData(updateStatement, incomingData, pendingOperationBuffer, writeInterceptor, params);
        }
        else {
            setData(updateStatement, incomingData, params);
        }

        int columnCountIndex = columnCount + 1;
        for (Object pkColumn : pk) {
            updateStatement.setObject(columnCountIndex++, pkColumn);
        }

        updateCount++;

        if (debug) {
            log.debug(String.format("%s Execute update query (pk:%s), params: %s", tableName, pk, params));
            int nbRowUpdated = updateStatement.executeUpdate();
            Preconditions.checkArgument(nbRowUpdated == 1, String.format("%s rows has been updated, but expected 1 row.", nbRowUpdated));
        }
        else {
            updateStatement.addBatch();
            if (updateCount > 0 && updateCount % batchSize == 0) {
                updateStatement.executeBatch();
                updateStatement.clearBatch();
            }
        }
    }

    @Override
    public void executeUpdate(List<Object> pk, Object[] row) throws SQLException {
        List<Object> params = null;
        if (debug) {
            params = Lists.newArrayList();
        }

        if (writeInterceptor != null) {
            transformAndSetData(updateStatement, row, pendingOperationBuffer, writeInterceptor, params);
        }
        else {
            setData(updateStatement, row, params);
        }

        int columnCountIndex = columnCount + 1;

        for (Object pkColumn : pk) {
            updateStatement.setObject(columnCountIndex++, pkColumn);
        }

        updateCount++;

        if (debug) {
            log.debug(String.format("%s Execute update query (pk:%s), params: %s", tableName, pk, params));
            int nbRowUpdated = updateStatement.executeUpdate();
            Preconditions.checkArgument(nbRowUpdated == 1, String.format("%s rows has been updated, but expected 1 row.", nbRowUpdated));
        }
        else {
            updateStatement.addBatch();
            if (updateCount > 0 && updateCount % batchSize == 0) {
                updateStatement.executeBatch();
                updateStatement.clearBatch();
            }
        }
    }

    @Override
    public ResultSet getDataByFk(String fkColumnName, Set<Integer> fkValues) throws SQLException {

        boolean insertUsingTempQueryParameter = isTempQueryParameterEnable() && fkValues.size() > dbMeta.getInExpressionCountLimit();

        if (insertUsingTempQueryParameter) {
            return getDataByFkUsingTempParameterTable(fkColumnName, fkValues);
        }

        return getDataByFkWithInOperator(fkColumnName, fkValues);
    }

    @Override
    public Integer getIdFromRemoteId(String tableName, Integer remoteId) throws SQLException {
        String sql = getTable().getSelectIdFromRemoteIdQuery(tableName);
        if (sql == null) {
            return null;
        }

        PreparedStatement statement = getConnection().prepareStatement(sql);
        statement.setInt(1, remoteId);
        try {
            ResultSet resultSet = statement.executeQuery();
            if (!resultSet.next()) {
                return null;
            }
            Integer result = resultSet.getInt(1);

            return result;
        } finally {
            DaoUtils.closeSilently(statement);
        }
    }

    @Override
    public Map<Integer, Integer> getExistingRemoteIdsMap() throws SQLException {

        String sql = getTable().getSelectRemoteIdsQuery();

        PreparedStatement statement = getConnection().prepareStatement(sql);

        Map<Integer, Integer> result = Maps.newHashMap();

        try {
            ResultSet resultSet = statement.executeQuery();
            while (resultSet.next()) {
                int id = resultSet.getInt(1);
                int remoteId = resultSet.getInt(2);
                result.put(remoteId, id);
            }
            statement.close();
            return result;
        } finally {
            DaoUtils.closeSilently(statement);
        }
    }

    public List<Object> getPk(ResultSet incomingData, SynchroWriteBuffer transformResult) throws SQLException {
        List<Object> result;

        if (readInterceptor != null) {
            Object[] data = transformData(incomingData, readInterceptor, transformResult);
            result = getTable().getPk(data);
        }
        else {
            result = getTable().getPk(incomingData);
        }

        return result;
    }

    public boolean isTempQueryParameterEnable() {
        return insertQueryParameterStatement != null;
    }

    public void clearCounts() {
        insertCount = 0;
        updateCount = 0;
    }

    /* -- Abstract method -- */

    /* -- Protected methods -- */

    protected void insertValuesIntoTempQueryParameter(Set<Integer> parameterValues, String queryParameterName, int queryPersonId) throws SQLException {
        Preconditions.checkNotNull(insertQueryParameterStatement);
        if (debug) {
            log.debug(String.format("%s Setting query parameters into %s", tableName, TABLE_TEMP_QUERY_PARAMETER.toUpperCase()));
        }

        Statement stm = getConnection().createStatement();
        try {
            stm.executeUpdate(String.format("DELETE FROM %", TABLE_TEMP_QUERY_PARAMETER));
        } finally {
            DaoUtils.closeSilently(stm);
        }

        int i = 1;
        for (Integer parameterValue : parameterValues) {
            insertQueryParameterStatement.setInt(1, i++);
            insertQueryParameterStatement.setString(2, queryParameterName);
            insertQueryParameterStatement.setInt(3, parameterValue);
            insertQueryParameterStatement.setNull(4, Types.VARCHAR);
            insertQueryParameterStatement.setInt(5, queryPersonId);
            insertQueryParameterStatement.addBatch();
        }

        insertQueryParameterStatement.executeBatch();
        insertQueryParameterStatement.clearBatch();
    }

    protected String createTempQueryParameterQuery(SynchroTableMetadata table) {
        SynchroDatabaseMetadata dbMeta = table.getDatabaseMetadata();
        if (dbMeta.getTable(TABLE_TEMP_QUERY_PARAMETER) == null) {
            return null;
        }

        return String.format("INSERT INTO %s (ID, PARAMETER_NAME, NUMERICAL_VALUE, ALPHANUMERICAL_VALUE, PERSON_FK)"
                + " VALUES (?, ?, ?, ?, ?)",
                TABLE_TEMP_QUERY_PARAMETER);

    }

    protected void closeSelectStatements() {
        if (CollectionUtils.isNotEmpty(selectStatements)) {
            for (PreparedStatement statement : selectStatements) {
                DaoUtils.closeSilently(statement);
            }
        }
        selectStatements.clear();
    }

    protected void closePreviousSelectStatements(PreparedStatement newStatement) {
        closeSelectStatements();
        selectStatements.add(newStatement);
    }

    protected void setData(PreparedStatement statement, Object[] values, List<Object> debugParams) throws SQLException {
        for (int c = 1; c <= columnCount; c++) {
            Object object = values[c - 1];
            statement.setObject(c, object);
            if (debug) {
                debugParams.add(object);
            }
        }
    }

    protected void setData(PreparedStatement statement, ResultSet incomingData, List<Object> debugParams) throws SQLException {
        for (int c = 1; c <= columnCount; c++) {
            Object object = incomingData.getObject(c);
            statement.setObject(c, object);
            if (debug) {
                debugParams.add(object);
            }
        }
    }

    protected void transformAndSetData(PreparedStatement statement, ResultSet incomingData,
            SynchroWriteBuffer transformBuffer, SynchroInterceptor interceptor, List<Object> debugParams)
            throws SQLException {
        Preconditions.checkNotNull(transformBuffer);

        // Transform data
        Object[] row = transformData(incomingData, interceptor, transformBuffer);

        // Set the data into the statement
        setData(statement, row, debugParams);
    }

    protected void transformAndSetData(PreparedStatement statement, Object[] incomingData,
            SynchroWriteBuffer transformBuffer, SynchroInterceptor interceptor, List<Object> debugParams)
            throws SQLException {
        Preconditions.checkNotNull(transformBuffer);

        // Transform data
        Object[] row = transformData(incomingData, interceptor, transformBuffer);

        // Set the data into the statement
        setData(statement, row, debugParams);
    }

    protected Object[] transformData(ResultSet incomingData, SynchroInterceptor interceptor, SynchroWriteBuffer buffer)
            throws SQLException {
        Preconditions.checkNotNull(interceptor);

        // Get data
        Object[] result = table.getData(incomingData);

        // Transform values
        try {
            interceptor.onWrite(result, this, buffer);
        } catch (Exception e) {
            throw new SynchroTechnicalException(String.format("Error while transform row, for table %s", table.getName()), e);
        }

        return result;
    }

    protected Object[] transformData(Object[] incomingData, SynchroInterceptor interceptor, SynchroWriteBuffer buffer)
            throws SQLException {
        Preconditions.checkNotNull(interceptor);

        // Transform values
        try {
            interceptor.onWrite(incomingData, this, buffer);
        } catch (Exception e) {
            throw new SynchroTechnicalException(String.format("Error while transform row, for table %s", table.getName()), e);
        }

        return incomingData;
    }

    protected ResultSet getDataByFkWithInOperator(String fkColumnName, Set<Integer> fkValues) throws SQLException {
        String sql = table.getSelectDataFromFkQuery(fkColumnName, fkValues.size());

        Preconditions.checkNotNull(sql, String.format("Columns %s is not referenced for table %s", fkColumnName, table.getName()));

        PreparedStatement statement = getConnection().prepareStatement(sql);
        closePreviousSelectStatements(statement);

        int paramIndex = 1;
        for (Object value : fkValues) {
            statement.setObject(paramIndex, value);
            paramIndex++;
        }

        statement.setFetchSize(batchSize);

        ResultSet result = statement.executeQuery();
        return result;
    }

    protected ResultSet getDataByFkUsingTempParameterTable(String fkColumnName, Set<Integer> fkValues) throws SQLException {

        String sql = table.getSelectDataFromFkQueryBigParams(fkColumnName);
        Preconditions.checkNotNull(sql);

        PreparedStatement statement = getConnection().prepareStatement(sql);
        closePreviousSelectStatements(statement);

        String queryParameterName = "remoteIds";
        int personFk = -1;
        insertValuesIntoTempQueryParameter(fkValues, queryParameterName, personFk);
        statement.setString(1, queryParameterName);
        statement.setInt(2, personFk);

        statement.setFetchSize(batchSize);

        ResultSet result = statement.executeQuery();
        return result;
    }

    @Override
    public List<Object> getPk(ResultSet incomingData) throws SQLException {
        return table.getPk(incomingData);
    }

    private SynchroInterceptor initReadInterceptor(List<SynchroInterceptor> interceptors) {
        if (CollectionUtils.isEmpty(interceptors)) {
            return null;
        }
        SynchroContext context = table.getDatabaseMetadata().getContext();
        List<SynchroInterceptor> readInterceptors = Lists.newArrayList();

        try {
            for (SynchroInterceptor interceptor : interceptors) {
                if (interceptor.enableOnRead()) {
                    SynchroInterceptor newInterceptor;
                    newInterceptor = interceptor.getClass().newInstance();

                    newInterceptor.setContext(context);
                    readInterceptors.add(interceptor);
                }
            }
        } catch (Exception e) {
            throw new SynchroTechnicalException("Could not initialize DAO read interceptors.", e);
        }

        if (CollectionUtils.isEmpty(readInterceptors)) {
            return null;
        }

        return SynchroInterceptorUtils.chain(readInterceptors, SynchroInterceptorBase.class);
    }

    private SynchroInterceptor initWriteInterceptor(List<SynchroInterceptor> interceptors) {
        if (CollectionUtils.isEmpty(interceptors)) {
            return null;
        }
        SynchroContext context = table.getDatabaseMetadata().getContext();
        List<SynchroInterceptor> writeInterceptors = Lists.newArrayList();

        try {
            for (SynchroInterceptor interceptor : interceptors) {
                if (interceptor.enableOnWrite()) {
                    SynchroInterceptor newInterceptor;
                    newInterceptor = interceptor.getClass().newInstance();

                    newInterceptor.setContext(context);
                    writeInterceptors.add(interceptor);
                }
            }
        } catch (Exception e) {
            throw new SynchroTechnicalException("Could not initialize DAO read interceptors.", e);
        }

        if (CollectionUtils.isEmpty(writeInterceptors)) {
            return null;
        }

        return SynchroInterceptorUtils.chain(writeInterceptors, SynchroInterceptorBase.class);
    }

}