/*
 * Copyright © 2013 by Data Virtuality GmbH
 */
package org.teiid.translator.jdbc;

import com.datavirtuality.dv.core.interfaces.DVCaller;
import org.jboss.logging.Logger;
import org.teiid.language.ColumnReference;
import org.teiid.language.Insert;
import org.teiid.translator.ExecutionContext;
import org.teiid.translator.TranslatorException;
import org.teiid.translator.jdbc.plugin.JDBCPlugin;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;


/**
 * @author Roman Pedchenko
 * @date 2016.08.17
 */
public class MultilInsertExecutionImpl extends JDBCUpdateExecution {
	private static final Logger logger = Logger.getLogger(MultilInsertExecutionImpl.class);

	private final Integer maxChunkSize;
	private final Integer maxRowCount;
	protected Integer numberOfThreads;
	private final boolean insertMultiThread;
	private final MultilInsertExecutionFactoryIntf ef;

	private static void logDebug(String msg, Object... params) {
		logger.debugf(msg, params);
	}

	private static void logTrace(String msg, Object... params) {
		logger.tracef(msg, params);
	}

	MultilInsertExecutionImpl(Insert command, Connection connection, ExecutionContext context, MultilInsertExecutionFactoryIntf ef) {
		super(command, connection, context, (JDBCExecutionFactory) ef);
		this.ef = ef;
		this.insertMultiThread = MultilInsertExecutionFactoryIntf.UploadMode.multiThreaded(ef.uploadMode());
		this.maxChunkSize = ef.maxChunkSize();
		this.maxRowCount = ef.maxRowCount();
		this.numberOfThreads = ef.numberOfThreads();
	}

	@Override
	public void execute() throws TranslatorException {
		TranslatedCommand translatedComm = translateCommand(command);
		executeTranslatedCommand(translatedComm);
	}

    protected void executeTranslatedCommand(TranslatedCommand translatedComm) throws TranslatorException {
        // create statement or PreparedStatement and execute
        String sql = translatedComm.getSql();
        boolean commitType = false;
        boolean succeeded = false;
        JDBCExecutionException exc = null;
        try {
        	int updateCount = 0;
            if (!translatedComm.isPrepared()) {
				// noinspection SqlSourceToSinkFlow
				updateCount = getStatement().executeUpdate(sql);
                addStatementWarnings();
            } else {
            	DVCaller.callWithExc(76, command, connection);
            	PreparedStatement pstatement = getPreparedStatement(sql);

            	Insert insert = (Insert) command;
            	List<ColumnReference> columns = insert.getColumns();
            	Iterator<? extends List<?>> vi = insert.getParameterValues();

                if (vi != null) {
                    // temporarily turn the auto commit off, and set it back to what it was
                    // before at the end of the command execution, so that multiple commands
                	// are executed in the same transaction
                    commitType = getAutoCommit(translatedComm);
                    if (commitType) {
                        connection.setAutoCommit(false);
                    }

					sql = sql.substring(0, sql.lastIndexOf("VALUES"));
                    if (insertMultiThread) {
                    	if (vi.hasNext()) {
							updateCount = executeMultiThread(columns, vi, sql);
                    	}
                    }
                    else
                    	updateCount = executeSingleThread(columns, vi, sql);

                } else {
                	bind(pstatement, translatedComm.getPreparedValues(), null);
        			updateCount = pstatement.executeUpdate();
        			addStatementWarnings();
                }
                succeeded = true;
            }
            result = new int[] {updateCount};
        } catch (SQLException err) {
        	DVCaller.call(91, command);
        	exc = new JDBCExecutionException(JDBCPlugin.Event.TEIID11013, err, translatedComm);
        	throw exc;
        } catch (InterruptedException | ExecutionException e) {
        	exc = new JDBCExecutionException(JDBCPlugin.Event.TEIID11013, new SQLException(e), translatedComm);
        	throw exc;
        } catch (Throwable err) {
        	DVCaller.call(91, command);
        	throw new TranslatorException(err);
		} finally {
        	if (commitType) {
                restoreAutoCommit(!succeeded, translatedComm, exc);
            }
        }
    }

    private int generateQueryForChunk(List<ColumnReference> columns, Iterator<? extends List<?>> vi, StringBuilder insertQuery) throws TranslatorException {
		int currentBytes = 0;
		int currentRows = 0;
		while (batchCondition(currentBytes, currentRows)) {
			List<?> values;
			synchronized (this) {
				if (!vi.hasNext())
					break;
				values = vi.next();
			}
			if (currentRows == 0)
				insertQuery.append(" VALUES ");
			else
				insertQuery.append(",");
			logTrace("MultiInsert - Generating value string");
			String valueString = generateValueString(columns, values);
			logTrace("MultiInsert - Value string generated");
			insertQuery.append(valueString);
			currentBytes += valueString.getBytes().length;
			currentRows++;
		}
		return currentRows;
	}

	private boolean batchCondition(int currentBytes, int currentRows) {
		if (maxChunkSize != null && currentBytes >= maxChunkSize)
			return false;
		return maxRowCount == null || currentRows < maxRowCount;
	}

    private String generateValueString(List<ColumnReference> columns, List<?> values) throws TranslatorException {
    	StringBuilder result = new StringBuilder();
    	result.append("(");
		int colIndex = 0;
    	for (Iterator<?> it = values.iterator(); it.hasNext();) {
			Object value = it.next();
			ColumnReference col = columns.get(colIndex++);
			if (value == null) {
				result.append("null");
			} else {
				result.append(ef.valueToString(col, value));
			}
			if (it.hasNext())
				result.append(",");
		}
    	result.append(")");
    	return result.toString();
    }

	private int executeSingleThread(List<ColumnReference> columns, Iterator<? extends List<?>> vi, String sql) throws SQLException, TranslatorException {
		logDebug("MultiInsert - executing insert statement in a single thread");
    	int updateCount = 0;
		while (true) {
			StringBuilder insertQuery = new StringBuilder();
			logTrace("MultiInsert - Generating query for chunk");
			int counter = generateQueryForChunk(columns, vi, insertQuery);
			logTrace("MultiInsert - Query for chunk generated");
			if (counter == 0)
				break;

			String sql1 = sql + insertQuery;
			logDebug("Size of the query (bytes): %d", sql1.getBytes().length);
			try (Statement stm = connection.createStatement()) {
				//noinspection SqlSourceToSinkFlow
				updateCount += stm.executeUpdate(sql1);
					}
	        	}
		return updateCount;
    }

	private int executeMultiThread(List<ColumnReference> columns, Iterator<? extends List<?>> vi, String sql)
			throws InterruptedException, ExecutionException {
		logDebug("MultiInsert - executing insert statement in multithreading");
		logDebug("Number of threads: %d", numberOfThreads);
		logDebug("Max chunk size: %d", maxChunkSize);
		logDebug("Max row count: %d", maxRowCount);
		int updateCount = 0;
    	ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
		final List<Callable<Integer>> tasks = createTasks(columns, vi, sql);

		List<Future<Integer>> futures = executorService.invokeAll(tasks);

		for (Future<Integer> future : futures) {
			Integer update = future.get();
			if (update != null)
				updateCount += update;
		}
		return updateCount;
    }

	private List<Callable<Integer>> createTasks(List<ColumnReference> columns, Iterator<? extends List<?>> vi, String sql) {
		List<Callable<Integer>> tasks = new ArrayList<>();

		for (int i = 0; i < numberOfThreads; i++) {
			tasks.add(() -> {
				int totalCounter = 0;
				try (Statement stm = connection.createStatement()) {
					while (true) {
						StringBuilder insertQuery = new StringBuilder();
						int counter = generateQueryForChunk(columns, vi, insertQuery);
						if (counter == 0)
							break;
						String sql1 = sql + insertQuery;
						logDebug("MultiInsert - Running query in thread with size of the query (bytes): %d", sql1.getBytes().length);
						//noinspection SqlSourceToSinkFlow
						totalCounter += stm.executeUpdate(sql1);
					}
				}
				return totalCounter;
			});
		}
		return tasks;
	}
}
