/*
 *  Copyright 2010 Anyware Services
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package org.ametys.plugins.forms.content.table;

import java.io.IOException;
import java.net.URISyntaxException;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.avalon.framework.component.Component;
import org.apache.avalon.framework.service.ServiceException;
import org.apache.avalon.framework.service.ServiceManager;
import org.apache.avalon.framework.service.Serviceable;
import org.apache.cocoon.ProcessingException;
import org.apache.commons.lang.StringUtils;
import org.slf4j.LoggerFactory;

import org.ametys.cms.repository.Content;
import org.ametys.core.datasource.ConnectionHelper;
import org.ametys.core.datasource.dbtype.SQLDatabaseTypeExtensionPoint;
import org.ametys.core.ui.Callable;
import org.ametys.core.user.UserIdentity;
import org.ametys.core.util.I18nUtils;
import org.ametys.core.util.URIUtils;
import org.ametys.plugins.forms.FormsException;
import org.ametys.plugins.forms.content.Field;
import org.ametys.plugins.forms.content.Field.FieldType;
import org.ametys.plugins.forms.content.Form;
import org.ametys.plugins.forms.content.data.FieldValue;
import org.ametys.plugins.forms.content.data.UserEntry;
import org.ametys.plugins.forms.content.jcr.FormPropertiesManager;
import org.ametys.plugins.forms.data.Answer;
import org.ametys.plugins.repository.AmetysObjectResolver;
import org.ametys.plugins.workflow.store.JdbcWorkflowStore;
import org.ametys.runtime.config.Config;
import org.ametys.runtime.i18n.I18nizableText;

/**
 * Class that handles creation and deletion of a table used by a form.
 */
public class FormTableManager implements Component, Serviceable
{
    /** Avalon Role */
    public static final String ROLE = FormTableManager.class.getName();
    
    /** The id of the configuration parameter used for the forms' datasource */
    public static final String FORMS_POOL_CONFIG_PARAM = "plugins.forms.datasource";
    
    /** Prefix for database tables */
    public static final String TABLE_PREFIX = "Forms_";
    
    /** ID field name. */
    public static final String ID_FIELD = "id";
    
    /** Creation date field name. */
    public static final String CREATION_DATE_FIELD = "creationDate";
    
    /** Current user login field name. */
    public static final String LOGIN_FIELD = "login";
    
    /** Current user population id field name. */
    public static final String POPULATION_ID_FIELD = "populationId";
    
    /** Workflow id column name. */
    public static final String WORKFLOW_ID_FIELD = "workflowId"; 
    
    /** Suffix for filename column. */
    public static final String FILE_NAME_COLUMN_SUFFIX = "-filename";
    
    /** Constant for table state */
    public static final int TABLE_CREATED_AND_UP_TO_DATE = 2;
    
    /** Constant for table state */
    public static final int TABLE_CREATED_BUT_NEED_UPDATE = 3;
    
    /** Constant for table state */
    public static final int TABLE_NOT_CREATED = 1;
    
    /** Constant for table state */
    public static final int TABLE_UNKOWN_STATUS = 0;

    private static org.slf4j.Logger __logger = LoggerFactory.getLogger(FormTableManager.class);

    private ServiceManager _manager;
    
    private FormPropertiesManager _formPropertiesManager;
    private AmetysObjectResolver _resolver;
    private I18nUtils _i18nUtils;
    private JdbcWorkflowStore _jdbcWorkflowStore;
    private SQLDatabaseTypeExtensionPoint _sqlDatabaseTypeExtensionPoint;
    
    @Override
    public void service(ServiceManager smanager) throws ServiceException
    {
        _manager = smanager;
    }
    
    private SQLDatabaseTypeExtensionPoint getSQLDatabaseTypeExtensionPoint()
    {
        if (_sqlDatabaseTypeExtensionPoint == null)
        {
            try
            {
                _sqlDatabaseTypeExtensionPoint = (SQLDatabaseTypeExtensionPoint) _manager.lookup(SQLDatabaseTypeExtensionPoint.ROLE);
            }
            catch (ServiceException e)
            {
                throw new RuntimeException(e);
            }
        }
        return _sqlDatabaseTypeExtensionPoint;
    }
    
    private I18nUtils getI18nUtils()
    {
        if (_i18nUtils == null)
        {
            try
            {
                _i18nUtils = (I18nUtils) _manager.lookup(I18nUtils.ROLE);
            }
            catch (ServiceException e)
            {
                throw new RuntimeException(e);
            }
        }
        return _i18nUtils;
    }
    
    private AmetysObjectResolver getAmetysObjectResolver()
    {
        if (_resolver == null)
        {
            try
            {
                _resolver = (AmetysObjectResolver) _manager.lookup(AmetysObjectResolver.ROLE);
            }
            catch (ServiceException e)
            {
                throw new RuntimeException(e);
            }
        }
        return _resolver;
    }
    
    private FormPropertiesManager getFormPropertiesManager()
    {
        if (_formPropertiesManager == null)
        {
            try
            {
                _formPropertiesManager = (FormPropertiesManager) _manager.lookup(FormPropertiesManager.ROLE);
            }
            catch (ServiceException e)
            {
                throw new RuntimeException(e);
            }
        }
        return _formPropertiesManager;
    }  
    
    private JdbcWorkflowStore getJdbcWorkflowStore()
    {
        if (_jdbcWorkflowStore == null)
        {
            try
            {
                _jdbcWorkflowStore = (JdbcWorkflowStore) _manager.lookup(JdbcWorkflowStore.ROLE);
            }
            catch (ServiceException e)
            {
                throw new RuntimeException(e);
            }
        }
        return _jdbcWorkflowStore;
    }
    
    /**
     * Create database table to store results of content forms
     * @param form informations used to create table
     * @return true if the table was created false else
     */
    public boolean createTable(Form form)
    {
        if (form == null)
        {
            throw new IllegalArgumentException("Form object can not be null");
        }
        
        Map<String, FieldValue> columns = getColumns(form);
        if (!_createOrUpdateTable(form.getId(), columns))
        {
            return false;
        }
        return true;
    }
    
    /**
     * Drop the table by its name
     * @param table the name of the table to drop
     * @return true if the table was dropped, false otherwise
     */
    public boolean dropTable(String table)
    {
        PreparedStatement stmt = null;
        Connection connection = null;
        
        try
        {
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            
            String request = "DROP TABLE " + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(ConnectionHelper.getDatabaseType(connection), table);
            
            stmt = connection.prepareStatement(request); 
            stmt.executeUpdate();
        }
        catch (SQLException e)
        {
            __logger.error("Error while deleting table " + table, e);
            return false;
        }
        finally
        {
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(connection);
        }

        return true;
    }
    
    /**
     * Get all the table names
     * @return the list of all the forms' table names
     */
    public List<String> getTableNames()
    {
        List<String> formsTableNames = new ArrayList<> ();
        Connection con = null;
        PreparedStatement stmt = null;
        DatabaseMetaData metadata = null;
        ResultSet tables = null;

        try
        {
            // Use a connection pool to establish a connection to the data base
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            con = ConnectionHelper.getConnection(dataSourceId);
            metadata = con.getMetaData();
            tables = metadata.getTables(con.getCatalog(), con.getSchema(), "%", null);
            
            while (tables.next()) 
            {
                String tableName = tables.getString(3);
                if (StringUtils.startsWithIgnoreCase(tableName, TABLE_PREFIX))
                {
                    formsTableNames.add(tableName);
                }
            }
        }
        catch (SQLException e)
        {
            __logger.error("Error while retrieving the forms table names", e);
        }
        finally
        {
            // Clean up connection resources
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(con);
        }
        
        return formsTableNames;
    }
    
    /**
     * Get the user answers from a list of forms
     * @param forms the list of forms
     * @param user the user
     * @return the list of common answer
     * @throws FormsException if an forms error occurred
     */
    public List<Answer> getUserAnwsers(List<Form> forms, UserIdentity user) throws FormsException
    {
        List<Answer> anwsers = new ArrayList<>();
        
        // Get table names to avoid to make a request on non existant table
        List<String> tableNames = getTableNames();
        
        Connection connection = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        
        try
        {
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(connection);

            List<String> queries = new ArrayList<>();
            for (Form form : forms)
            {
                String tableName = TABLE_PREFIX + form.getId();
                if (tableNames.stream().anyMatch(tableName::equalsIgnoreCase))
                {
                    StringBuilder sql = new StringBuilder();
                    if (ConnectionHelper.DATABASE_DERBY.equals(dbType))
                    {
                        sql.append("(SELECT CAST('");
                        sql.append(form.getId());
                        sql.append("' AS VARCHAR(128)) as formId, id, ");
                    }
                    else
                    {
                        sql.append("(SELECT '");
                        sql.append(form.getId());
                        sql.append("' as formId, id, ");
                    }
                    sql.append(LOGIN_FIELD);
                    sql.append(", ");
                    sql.append(POPULATION_ID_FIELD);
                    sql.append(", ");
                    sql.append(CREATION_DATE_FIELD);
                    sql.append(", ");
                    if (StringUtils.isNotBlank(form.getWorkflowName()))
                    {
                        sql.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, WORKFLOW_ID_FIELD));
                    }
                    else
                    {
                        sql.append("-1 AS ");
                        sql.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, WORKFLOW_ID_FIELD));
                    }
                    sql.append(" FROM ");
                    sql.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
                    sql.append(" WHERE ");
                    sql.append(LOGIN_FIELD);
                    sql.append(" = ? AND ");
                    sql.append(POPULATION_ID_FIELD);
                    sql.append(" = ?)");
                    queries.add(sql.toString());
                }
                else
                {
                    __logger.warn("Table name '" + tableName + "' for content id '" + form.getContentId() + "' doesn't exist on the SQL database.");
                }
            }
            
            stmt = connection.prepareStatement(queries.stream().collect(Collectors.joining(" UNION ALL ")));

            int nextParam = 1;
            for (Form form : forms)
            {
                String tableName = TABLE_PREFIX + form.getId();
                if (tableNames.stream().anyMatch(tableName::equalsIgnoreCase))
                {
                    stmt.setString(nextParam++, user.getLogin());
                    stmt.setString(nextParam++, user.getPopulationId());
                }
            }
            
            // Execute the query.
            rs = stmt.executeQuery();
            
            // Extract the result.
            Long number = 0L;
            while (rs.next())
            {
                String id = rs.getString("id");
                String formId = rs.getString("formId");
                Form form = getFormPropertiesManager().getForm(formId);
                
                Timestamp creationDate = rs.getTimestamp(CREATION_DATE_FIELD);
                Integer workflowId = rs.getInt(WORKFLOW_ID_FIELD);
                
                Answer answer = new Answer(
                    id,
                    number++,
                    formId,
                    form.getLabel(),
                    creationDate,
                    form.getWorkflowName(),
                    workflowId,
                    user
                );
                
                anwsers.add(answer);
            }
        }
        catch (Exception e)
        {
            __logger.error("Error while getting current user answers.", e);
            throw new FormsException("Error while getting current user answers.", e);
        }
        finally
        {
            ConnectionHelper.cleanup(rs);
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(connection);
        }
        
        return anwsers;
    }
    
    /**
     * Get all the submissions for a form.
     * @param form the form object.
     * @param columns the columns
     * @param offset The number of results to ignore.
     * @param length The maximum number of results to return
     * @param entryIds the ids of the submissions to retrieve
     * @return the list of submissions.
     * @throws FormsException if an error occurs.
     */
    public List<UserEntry> getSubmissions(Form form, Map<String, FieldValue> columns, int offset, int length, List<Integer> entryIds) throws FormsException
    {
        List<UserEntry> entries = new ArrayList<>();
        
        final String tableName = TABLE_PREFIX + form.getId();
        
        Connection connection = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        
        try
        {

            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(connection);
            
            String request = _getSubmissionsQuery(offset, length, tableName, dbType, form, columns);
            
            stmt = connection.prepareStatement(request);
            
            // Execute the query.
            rs = stmt.executeQuery();
            
            // Extract the results.
            while (rs.next())
            {
                // First get the ID and submission date.
                int id = rs.getInt(ID_FIELD);
                if (entryIds == null || entryIds.contains(id))
                {
                    UserEntry entry = _getUserEntryFromResultSet(rs, form, columns, dbType);
                    entries.add(entry);
                }
            }
        }
        catch (SQLException | IOException e)
        {
            __logger.error("Error while getting entries for table " + tableName, e);
            throw new FormsException("Error while getting entries for table " + tableName, e);
        }
        finally
        {
            ConnectionHelper.cleanup(rs);
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(connection);
        }
        
        return entries;
    }

    /**
     * Get the submissions query
     * @param offset The number of results to ignore.
     * @param length The maximum number of results to return
     * @param tableName the name of the table
     * @param dbType the type of the database used
     * @param form the form
     * @param columns the columns to query. Can be empty to query only common attributes
     * @return the string query for getting the submissions of a form
     */
    private String _getSubmissionsQuery(int offset, int length, final String tableName, String dbType, Form form, Map<String, FieldValue> columns)
    {
        StringBuilder query = _getFieldsForSelectQuery(dbType, form, columns);
        
        query.append(" FROM ");
        query.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
        
        return getSQLDatabaseTypeExtensionPoint().languageLimitQuery(dbType, query.toString(), Integer.toString(length), Integer.toString(offset));
    }
    
    /**
     * Get the submission for a form.
     * @param form the form object.
     * @param columns the columns
     * @param entryId the id of the submission to retrieve
     * @return the submission.
     * @throws FormsException if an error occurs.
     */
    public UserEntry getSubmission(Form form, Map<String, FieldValue> columns, Integer entryId) throws FormsException
    {
        final String tableName = TABLE_PREFIX + form.getId();
        
        Connection connection = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        
        try
        {

            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(connection);
            
            String request = _getSubmissionQuery(tableName, dbType, form, columns);
            
            stmt = connection.prepareStatement(request);
            stmt.setInt(1, entryId);
            
            // Execute the query.
            rs = stmt.executeQuery();
            
            // Extract the results.
            while (rs.next())
            {
                return _getUserEntryFromResultSet(rs, form, columns, dbType);
            }
        }
        catch (SQLException | IOException e)
        {
            __logger.error("Error while getting entry with id '{}' for table '{}'", entryId, tableName, e);
            throw new FormsException("Error while getting entry with id '" + entryId + "' for table " + tableName, e);
        }
        finally
        {
            ConnectionHelper.cleanup(rs);
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(connection);
        }
        
        return null;
    }
    
    /**
     * Get the submission query with entry id
     * @param tableName the name of the table
     * @param dbType the type of the database used
     * @param form the form object.
     * @param columns the columns to query. Can be empty to query only common attributes
     * @return the string query for getting the submissions of a form
     */
    private String _getSubmissionQuery(final String tableName, String dbType, Form form, Map<String, FieldValue> columns)
    {
        StringBuilder query = _getFieldsForSelectQuery(dbType, form, columns);
        
        query.append(" FROM ");
        query.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
        query.append(" WHERE id = ?");
        
        return getSQLDatabaseTypeExtensionPoint().languageLimitQuery(dbType, query.toString(), "1", "0");
    }
    
    private StringBuilder _getFieldsForSelectQuery(String dbType, Form form, Map<String, FieldValue> columns)
    {
        StringBuilder query = new StringBuilder("SELECT ");
        query.append(ID_FIELD);
        query.append(", ");
        query.append(CREATION_DATE_FIELD);
        query.append(", ");
        query.append(LOGIN_FIELD);
        query.append(", ");
        query.append(POPULATION_ID_FIELD);
        if (StringUtils.isNotBlank(form.getWorkflowName()))
        {
            query.append(", ");
            query.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, WORKFLOW_ID_FIELD));
        }
        
        for (String name : columns.keySet())
        {
            FieldValue fieldValue = columns.get(name);
            if (fieldValue.getType() == Types.BLOB)
            {
                // Only file name is needed. Blob will be required by an other way
                String fileNameColumn = DbTypeHelper.normalizeName(dbType, name + FILE_NAME_COLUMN_SUFFIX);
                query.append(", ");
                query.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, fileNameColumn));
            }
            else
            {
                query.append(", ");
                query.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, name));
            }
        }
        
        return query;
    }
    
    /**
     * Get a user entry from a result set
     * @param rs the result set
     * @param form the form
     * @param columns the columns
     * @param dbType the db type
     * @return the user entry
     * @throws SQLException if a sql error occurred
     * @throws IOException is a io error occurred
     */
    protected UserEntry _getUserEntryFromResultSet(ResultSet rs, Form form, Map<String, FieldValue> columns, String dbType) throws SQLException, IOException
    {
     // First get the ID and submission date.
        int id = rs.getInt(ID_FIELD);
        Timestamp creationDate = rs.getTimestamp(CREATION_DATE_FIELD);
        
        UserIdentity userIdentity = null;
        String login = rs.getString(LOGIN_FIELD);
        String populationId = rs.getString(POPULATION_ID_FIELD);
        if (StringUtils.isNotBlank(login) && StringUtils.isNotBlank(populationId))
        {
            userIdentity = new UserIdentity(login, populationId);
        }
        
        List<FieldValue> entryValues = new ArrayList<>();
        
        for (Map.Entry<String, FieldValue> column : columns.entrySet())
        {
            String columnLabel = column.getKey();
            FieldValue columnValue = column.getValue();
            
            // Extract the value.
            FieldValue value = null;
            
            switch (columnValue.getType())
            {
                case Types.VARCHAR:
                case Types.LONGVARCHAR:
                    String sValue = rs.getString(columnLabel);
                    value = new FieldValue(columnValue);
                    value.setValue(sValue);
                    entryValues.add(value);
                    break;
                case Types.BOOLEAN:
                    Boolean bValue = rs.getBoolean(columnLabel);
                    value = new FieldValue(columnValue);
                    value.setValue(bValue);
                    entryValues.add(value);
                    break;
                case Types.BLOB:
                    // Just get filename, blob will be requested by an other way if needed
                    String fileNameColumn = columnLabel + FILE_NAME_COLUMN_SUFFIX;
                    String normalizedName = DbTypeHelper.normalizeName(dbType, fileNameColumn);
                    String fileName = rs.getString(normalizedName);
                    value = new FieldValue(columnValue);
                    value.setValue(fileName);
                    
                    entryValues.add(value);
                    break;
                case Types.INTEGER:
                    Integer iValue = rs.getInt(columnLabel);
                    value = new FieldValue(columnValue);
                    value.setValue(iValue);
                    entryValues.add(value);
                    break;
                default:
                    break;
            }
        }
        
        Integer workflowId = null;
        if (hasWorkflowIdColumn(form.getId()))
        {
            workflowId = rs.getInt(WORKFLOW_ID_FIELD);
        }
        
        return new UserEntry(id, creationDate, entryValues, workflowId, userIdentity);
    }
    
    /**
     * Get the total count of submissions for a form.
     * @param formId the id of the form
     * @return the total count of submissions, or -1 if an error occurs.
     * @throws FormsException if an error occurs.
     */
    public int getTotalSubmissions(String formId) throws FormsException
    {
        final String tableName = TABLE_PREFIX + formId;
        
        Connection connection = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        
        try
        {

            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            
            String dbType = ConnectionHelper.getDatabaseType(connection);
            
            String request = "SELECT COUNT(*) FROM " + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName);
            
            stmt = connection.prepareStatement(request);
            
            // Execute the query.
            rs = stmt.executeQuery();
            
            // Extract the result.
            if (rs.next())
            {
                int count = rs.getInt(1);
                return count;
            }
        }
        catch (SQLException e)
        {
            __logger.error("Error while getting entries for table " + tableName, e);
            throw new FormsException("Error while getting entries for table " + tableName, e);
        }
        finally
        {
            ConnectionHelper.cleanup(rs);
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(connection);
        }
        
        return -1;
    }
    
    /**
     * Get the forms of contents
     * @param contentIds The ids of contents
     * @param currentLanguage The current language
     * @return The forms
     * @throws FormsException if an error occurred while getting content' form
     * @throws URISyntaxException if an error occurred when URL encoding form's label
     */
    @Callable
    public List<Map<String, Object>> getContentForms (List<String> contentIds, String currentLanguage) throws FormsException, URISyntaxException
    {
        List<Map<String, Object>> result = new ArrayList<>();
        
        for (String contentId : contentIds)
        {
            Content content = getAmetysObjectResolver().resolveById(contentId);
            
            Map<String, Object> contentforms = new HashMap<>();
            
            contentforms.put("id", content.getId());
            contentforms.put("title", content.getTitle(new Locale(currentLanguage)));
            
            List<Map<String, Object>> forms2json = new ArrayList<>();
            
            List<Form> forms = getFormPropertiesManager().getForms(content);
            for (Form form : forms)
            {
                Map<String, Object> form2json = new HashMap<>();
                form2json.put("id", form.getId());
                form2json.put("label", URIUtils.decode(form.getLabel()));
                form2json.put("workflowName", StringUtils.defaultString(form.getWorkflowName(), ""));
                
                forms2json.add(form2json);
            }
            
            contentforms.put("forms", forms2json);
            
            result.add(contentforms);
        }
        
        return result;
    }
    
    /**
     * Get the columns information of a form
     * @param siteName The site name
     * @param formId the identifier of form
     * @return The columns
     * @throws FormsException if an error occurred
     */
    @Callable
    public List<Map<String, Object>> getColumns (String siteName, String formId) throws FormsException
    {
        List<Map<String, Object>> columns2json = new ArrayList<>();
        
        Form form = getFormPropertiesManager().getForm(siteName, formId);
        
        Map<String, FieldValue> columns = getColumns(form);
        for (FieldValue column : columns.values())
        {
            Map<String, Object> column2json = new HashMap<>();
            column2json.put("id", column.getColumnName());
            
            Field field = column.getField();
            column2json.put("label", field.getLabel());
            column2json.put("type", field.getType().toString());
            column2json.put("name", field.getName());
            column2json.put("properties", field.getProperties());
            
            columns2json.add(column2json);
        }
        
        return columns2json;
    }
    
    /**
     * Remove a list of tables
     * @param tableNames the names of the tables to delete
     * @throws FormsException if the drop of the tables is not successful
     */
    @Callable
    public void removeTables(List<String> tableNames) throws FormsException
    {
        for (String tableName : tableNames)
        {
            if (!dropTable(tableName))
            {
                throw new FormsException("An error occurred occured while removing the tables from the database.");
            }
        }
    }
    
    /**
     * Form to tableinfo.
     * @param form the form.
     * @return the tableinfo.
     */
    public Map<String, FieldValue> getColumns(Form form)
    {
        Map<String, FieldValue> columns = new LinkedHashMap<>();
        
        for (Field field : form.getFields())
        {
            final String id = field.getId();
            final String name = field.getName();
            
            FieldValue fdValue = null;
            
            switch (field.getType())
            {
                case TEXT:
                case HIDDEN:
                case PASSWORD:
                case COST:
                case SELECT:
                    fdValue = new FieldValue(id, Types.VARCHAR, null, field);
                    columns.put(id, fdValue);
                    break;
                case TEXTAREA:
                    fdValue = new FieldValue(id, Types.LONGVARCHAR, null, field);
                    columns.put(id, fdValue);
                    break;
                case RADIO:
                    if (!columns.containsKey(name))
                    {
                        String value = field.getProperties().get("value");
                        String label = field.getLabel();
                        int index = 1;
                        field.getProperties().put("size", String.valueOf(index));
                        field.getProperties().put("option-" + index + "-value", value);
                        field.getProperties().put("option-" + index + "-label", label);
                        
                        field.getProperties().remove("value");
                        
                        fdValue = new FieldValue(name, Types.VARCHAR, null, field);
                        columns.put(name, fdValue);
                    }
                    else
                    {
                        // The value exists, clone it, concatenating the label.
                        if (StringUtils.isNotEmpty(field.getLabel()))
                        {
                            Field radioField = columns.get(name).getField();
                            int index = Integer.parseInt(radioField.getProperties().get("size"));
                            index++;
                            
                            String value = field.getProperties().get("value");
                            String label = field.getLabel();
                            
                            radioField.getProperties().put("size", String.valueOf(index));
                            radioField.getProperties().put("option-" + index + "-value", value);
                            radioField.getProperties().put("option-" + index + "-label", label);
                            
                            Field dummyField = new Field(radioField.getId(), radioField.getType(), radioField.getName(), radioField.getLabel() + "/" + field.getLabel(), radioField.getProperties());
                            columns.get(name).setField(dummyField);
                        }
                    }
                    break;
                case CHECKBOX:
                    fdValue = new FieldValue(id, Types.BOOLEAN, null, field);
                    columns.put(id, fdValue);
                    break;
                case FILE:
                    fdValue = new FieldValue(id, Types.BLOB, null, field);
                    columns.put(id, fdValue);
                    break;
                case CAPTCHA:
                    break;
                default:
                    break;
            }
            
        }
        
        return columns;
    }
    
    /**
     * Get the full column type corresponding to the database type (with precision).
     * @param sqlType the sqltype value.
     * @param dbType the database type.
     * @return the real column type identifier (to be included in CREATE statement).
     */
    protected String _getColumnType(int sqlType, String dbType)
    {
        switch (sqlType)
        {
            case Types.LONGVARCHAR:
                return  DbTypeHelper.getTextType(dbType);
            case Types.BOOLEAN:
                return  DbTypeHelper.getBooleanType(dbType);
            case Types.BLOB:
                return  DbTypeHelper.getBinaryType(dbType);
            case Types.VARCHAR:
            default:
                return  DbTypeHelper.getVarcharType(dbType);
        }
    }
    
    /**
     * Get the column type name (without precision) corresponding to the SQL type.
     * @param sqlType the SQL type value.
     * @param dbType the database type.
     * @return the real column type name (without precision).
     */
    protected String _getColumnTypeName(int sqlType, String dbType)
    {
        return StringUtils.substringBefore(_getColumnType(sqlType, dbType), "(");
    }
    
    /**
     * Test if two columns have the same type.
     * @param column the column retrieved from the database.
     * @param newColumn the new column.
     * @param dbType the database type.
     * @return true if the two columns have the same type.
     */
    protected boolean _isSameType(DbColumn column, FieldValue newColumn, String dbType)
    {
        int newColumnType = newColumn.getType();
        int currentColumnType = column.getSqlType();
        String newColumnTypeName = _getColumnTypeName(newColumnType, dbType);
        String currentColumnTypeName = column.getTypeName();
        
        return currentColumnType == newColumnType || currentColumnTypeName.equals(newColumnTypeName);
    }
    
    private boolean _createTable(String formId, Map<String, FieldValue> columns)
    {
        String tableName = TABLE_PREFIX + formId;
        
        Connection connection = null;
        PreparedStatement stmt = null;
        
        try
        {

            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            
            String dbType = ConnectionHelper.getDatabaseType(connection);
            StringBuilder sql = new StringBuilder();
            
            sql.append("CREATE TABLE ");
            sql.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
            sql.append(" ( ").append(ID_FIELD).append(" ").append(DbTypeHelper.getIdentityType(dbType)).append(" ").append(DbTypeHelper.getIdentityMarker(dbType)).append(", ");
            sql.append(CREATION_DATE_FIELD).append(" ").append(DbTypeHelper.getDateTimeType(dbType)).append(" NOT NULL,");
            sql.append(LOGIN_FIELD).append(" ").append(DbTypeHelper.getVarcharType(dbType)).append(" DEFAULT NULL,");
            sql.append(POPULATION_ID_FIELD).append(" ").append(DbTypeHelper.getVarcharType(dbType)).append(" DEFAULT NULL,");
            
            for (Map.Entry<String, FieldValue> column : columns.entrySet())
            {
                int sqlType = column.getValue().getType();
                sql.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, column.getKey())).append(" ");
                sql.append(_getColumnType(sqlType, dbType));
                sql.append(" DEFAULT NULL,");
                
                // Add a column for the file name.
                if (sqlType == Types.BLOB)
                {
                    String fileNameColumn = column.getKey() + FILE_NAME_COLUMN_SUFFIX;
                    String normalizedName = DbTypeHelper.normalizeName(dbType, fileNameColumn);
                    
                    sql.append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, normalizedName)).append(" ");
                    sql.append(DbTypeHelper.getVarcharType(dbType));
                    sql.append(" DEFAULT NULL,");
                }
            }
            
            sql.append("PRIMARY KEY (").append(ID_FIELD).append("))");
            
            if (__logger.isDebugEnabled())
            {
                __logger.debug("Creating table : " + sql.toString());
            }
            
            stmt = connection.prepareStatement(sql.toString());
            stmt.executeUpdate();
            ConnectionHelper.cleanup(stmt);
            
            // create sequence for oracle database
            if (ConnectionHelper.DATABASE_ORACLE.equals(dbType))
            {
                String sqlQuery = "create sequence seq_" + formId; 
                stmt = connection.prepareStatement(sqlQuery);
                stmt.executeUpdate();
            }
            
            return true;
        }
        catch (SQLException e)
        {
            __logger.error("Unable to create table " + tableName, e);
            return false;
        }
        finally
        {
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(connection);
        }
    }
    
    private boolean _alterTable(String formId, Map<String, FieldValue> newColumns)
    {
        String tableName = TABLE_PREFIX + formId;
        
        Connection connection = null;
        String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
        
        try
        {
            connection = ConnectionHelper.getConnection(dataSourceId);
            
            // Start transaction.
            connection.setAutoCommit(false);
            
            // Get the existing columns.
            Map<String, DbColumn> existingColumns = _getExistingColumns(connection, tableName);
            
            // Compute the modifications to make.
            DbColumnModifications modifications = _getColumnsToModify(connection, existingColumns, newColumns);
            
            // Move the columns that were deleted or replaced.
            _moveColumns(connection, modifications.getColumnsToRemove(), existingColumns, tableName);
            
            // Add the new columns.
            _addColumns(connection, modifications.getColumnsToAdd(), tableName);
            
            // Commit the transaction.
            connection.commit();
        }
        catch (SQLException e)
        {
            __logger.error("Unable to alter table " + tableName, e);
            try
            {
                connection.rollback();
            }
            catch (SQLException sqlex)
            {
                // Ignore.
                __logger.error("Error rollbacking the 'alter table' statements for table " + tableName, e);
            }
            return false;
        }
        finally
        {
            ConnectionHelper.cleanup(connection);
        }
        return true;
        
    }
    
    private int _checkTableStatus(String formId, Map<String, FieldValue> newColumns)
    {
        String tableName = TABLE_PREFIX + formId;
        
        Connection connection = null;
        ResultSet tables = null;
        ResultSet rs = null;
        
        Map<String, DbColumn> currentColumns = new HashMap<>();
        
        try
        {

            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            
            String dbType = ConnectionHelper.getDatabaseType(connection);
            
            tableName = DbTypeHelper.filterName(dbType, tableName);
            
            DatabaseMetaData metadata = connection.getMetaData();
            tables = metadata.getTables(connection.getCatalog(), null, tableName, null);
            
            if (!tables.next())
            {
                return TABLE_NOT_CREATED;
            }
            
            currentColumns = _getExistingColumns(connection, tableName);
            
            for (String newColumn : newColumns.keySet())
            {
                String filteredColumn = DbTypeHelper.filterName(dbType, newColumn);
                
                // Test the existence of the column.
                if (!currentColumns.containsKey(filteredColumn))
                {
                    return TABLE_CREATED_BUT_NEED_UPDATE;
                }
                
                // Test that the column type has not changed.
                DbColumn currentColumn = currentColumns.get(filteredColumn);
                FieldValue newColumnFv = newColumns.get(newColumn);
                
                if (!_isSameType(currentColumn, newColumnFv, dbType))
                {
                    return TABLE_CREATED_BUT_NEED_UPDATE;
                }
            }
            
            return TABLE_CREATED_AND_UP_TO_DATE;
        }
        catch (SQLException e)
        {
            __logger.error("Unable to get columns from table " + tableName, e);
            return TABLE_UNKOWN_STATUS;
        }
        finally
        {
            ConnectionHelper.cleanup(tables);
            ConnectionHelper.cleanup(rs);
            ConnectionHelper.cleanup(connection);
        }
    }

    private boolean _createOrUpdateTable(String formId, final Map<String, FieldValue> columns)
    {
        boolean toReturn = true;
        switch (_checkTableStatus(formId, columns))
        {
            case TABLE_NOT_CREATED:
                if (!_createTable(formId, columns))
                {
                    toReturn = false;
                }
                break;
            case TABLE_CREATED_BUT_NEED_UPDATE:
                if (!_alterTable(formId, columns))
                {
                    toReturn = false;
                }
                break;
            case TABLE_CREATED_AND_UP_TO_DATE:
            case TABLE_UNKOWN_STATUS:
            default:
                break;
        }
        return toReturn;
    }

    private DbColumnModifications _getColumnsToModify(Connection con, Map<String, DbColumn> existingColumns, Map<String, FieldValue> newColumns)
    {
        DbColumnModifications modifications = new DbColumnModifications();
        
        String dbType = ConnectionHelper.getDatabaseType(con);
        
        Map<String, FieldValue> columnsToAdd = new LinkedHashMap<>();
        Map<String, DbColumn> columnsToRemove = new HashMap<>();
        for (Map.Entry<String, FieldValue> newColumn : newColumns.entrySet())
        {
            String filteredName = DbTypeHelper.filterName(dbType, newColumn.getKey());
            columnsToAdd.put(filteredName, newColumn.getValue());
        }
        
        if (existingColumns != null)
        {
            for (String existingColName : existingColumns.keySet())
            {
                DbColumn existingColumn = existingColumns.get(existingColName);
                FieldValue newColumn = columnsToAdd.get(existingColName);
                
                // If the column does not already exist or if the type has changed,
                // mark the current column to be removed and keep the new column in the "to be added" list.
                if (newColumn != null && !_isSameType(existingColumn, newColumn, dbType))
                {
                    columnsToRemove.put(existingColName, existingColumn);
                }
                else
                {
                    // The column already exists and its type has not changed: do not touch this one.
                    columnsToAdd.remove(existingColName);
                }
            }
        }
        
        modifications.setColumnsToAdd(columnsToAdd);
        modifications.setColumnsToRemove(columnsToRemove);
        
        return modifications;
    }
    
    private Map<String, DbColumn> _getExistingColumns(Connection con, String table) throws SQLException
    {
        ResultSet columns = null;
        Map<String, DbColumn> toReturn = new LinkedHashMap<>();
        try
        {
            String dbType = ConnectionHelper.getDatabaseType(con);
            String filteredTableName = DbTypeHelper.filterName(dbType, table);
            
            DatabaseMetaData metadata = con.getMetaData();
            columns = metadata.getColumns(con.getCatalog(), null, filteredTableName, null);
            while (columns.next())
            {
                String columnName = columns.getString("COLUMN_NAME");
                Integer sqlType = columns.getInt("DATA_TYPE");
                String typeName = columns.getString("TYPE_NAME");
                Integer colSize = columns.getInt("COLUMN_SIZE");
                
                DbColumn col = new DbColumn(columnName, sqlType, typeName, colSize);
                
                toReturn.put(columnName, col);
            }
        }
        catch (SQLException e)
        {
            __logger.error("Unable to get columns from " + table, e);
            throw e;
        }
        finally
        {
            ConnectionHelper.cleanup(columns);
        }
        return toReturn;
    }
    
    private void _moveColumns(Connection con, Map<String, DbColumn> columnsToRemove, Map<String, DbColumn> existingColumns, String tableName) throws SQLException
    {
        String dbType = ConnectionHelper.getDatabaseType(con);
        PreparedStatement stmt = null;
        
        Set<String> existingColumnNames = existingColumns.keySet();
        
        try
        {
            for (String columnName : columnsToRemove.keySet())
            {
                DbColumn columnToRemove = columnsToRemove.get(columnName);
                
                String newName = _getNewName(con, columnName, existingColumnNames);
                
                String sql = DbTypeHelper.getRenameColumnStatement(tableName, columnToRemove, newName, dbType, getSQLDatabaseTypeExtensionPoint());
                
                if (__logger.isDebugEnabled())
                {
                    __logger.debug("Moving column: " + sql);
                }
                
                stmt = con.prepareStatement(sql);
                stmt.executeUpdate();
                
                // Add a column for the file name.
                if (columnToRemove.getSqlType() == Types.BLOB)
                {
                    // Release the previous statement.
                    ConnectionHelper.cleanup(stmt);
                    
                    String fileNameColumn = columnName + FILE_NAME_COLUMN_SUFFIX;
                    String newFileNameColumn = _getNewName(con, fileNameColumn, existingColumnNames);
                    String varcharType = DbTypeHelper.getVarcharType(dbType);
                    
                    sql = DbTypeHelper.getRenameColumnStatement(tableName, fileNameColumn, newFileNameColumn, varcharType, dbType, getSQLDatabaseTypeExtensionPoint());
                    
                    if (__logger.isDebugEnabled())
                    {
                        __logger.debug("Altering table : " + sql.toString());
                    }
                    
                    stmt = con.prepareStatement(sql.toString());
                    stmt.executeUpdate();
                }
            }
        }
        catch (SQLException e)
        {
            ConnectionHelper.cleanup(stmt);
            throw e;
        }
    }
    
    private void _addColumns(Connection con, Map<String, FieldValue> columnsToAdd, String tableName) throws SQLException
    {
        String dbType = ConnectionHelper.getDatabaseType(con);
        PreparedStatement stmt = null;
        
        try
        {
            for (Entry<String, FieldValue> column : columnsToAdd.entrySet())
            {
                StringBuilder sql = new StringBuilder();
                
                sql.append("ALTER TABLE ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
                
                int sqlType = column.getValue().getType();
                
                sql.append(" ADD ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, column.getKey())).append(" ");
                sql.append(_getColumnType(sqlType, dbType));
                sql.append(" DEFAULT NULL");
                
                if (__logger.isDebugEnabled())
                {
                    __logger.debug("Altering table : " + sql.toString());
                }
                
                stmt = con.prepareStatement(sql.toString());
                stmt.executeUpdate();
                
                // Add a column for the file name.
                if (sqlType == Types.BLOB)
                {
                    // Release the previous statement.
                    ConnectionHelper.cleanup(stmt);
                    
                    sql.setLength(0);
                    sql.append("ALTER TABLE ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
                    
                    String fileNameColumn = column.getKey() + FILE_NAME_COLUMN_SUFFIX;
                    String normalizedName = DbTypeHelper.normalizeName(dbType, fileNameColumn);
                    
                    sql.append(" ADD ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, normalizedName)).append(" ");
                    sql.append(DbTypeHelper.getVarcharType(dbType));
                    sql.append(" DEFAULT NULL");
                    
                    if (__logger.isDebugEnabled())
                    {
                        __logger.debug("Adding column: " + sql.toString());
                    }
                    
                    stmt = con.prepareStatement(sql.toString());
                    stmt.executeUpdate();
                }
            }
        }
        catch (SQLException e)
        {
            ConnectionHelper.cleanup(stmt);
            throw e;
        }
    }
    
    /**
     * Get the new name of a column to be moved.
     * @param con the connection.
     * @param currentName the current name.
     * @param existingColumnNames the names of the existing columns
     * @return the new name.
     */
    private String _getNewName(Connection con, String currentName, Set<String> existingColumnNames)
    {
        String dbType = ConnectionHelper.getDatabaseType(con);
        
        int i = 1;
        String newName = DbTypeHelper.normalizeName(dbType, currentName + "_old" + i);
        String filteredNewName = DbTypeHelper.filterName(dbType, newName);
        
        while (existingColumnNames.contains(filteredNewName))
        {
            i++;
            newName = DbTypeHelper.normalizeName(dbType, currentName + "_old" + i);
            filteredNewName = DbTypeHelper.filterName(dbType, newName);
        }
        
        return newName;
    }
    
    /**
     * Get the SQL type corresponding to a field type.
     * @param fieldType the field
     * @return the sql type as indicated in {@link java.sql.Types}.
     */
    public static int getFieldSqlType(FieldType fieldType)
    {
        int sqlType = Types.VARCHAR;
        switch (fieldType)
        {
            case TEXT:
            case HIDDEN:
            case PASSWORD:
            case SELECT:
            case RADIO:
                sqlType = Types.VARCHAR;
                break;
            case TEXTAREA:
                sqlType = Types.LONGVARCHAR;
                break;
            case CHECKBOX:
                sqlType = Types.BOOLEAN;
                break;
            case FILE:
                sqlType = Types.BLOB;
                break;
            case CAPTCHA:
                sqlType = Types.OTHER;
                break;
            default:
                break;
        }
        return sqlType;
    }
    
    /**
     * Add a workflow id column to the given table
     * @param formId the id of the form to alter
     * @throws FormsException if an error occurs
     */
    public void addWorkflowIdColumn(String formId) throws FormsException
    {
        final String tableName = TABLE_PREFIX + formId;
        
        Connection con = null;
        PreparedStatement stmt = null;
        
        try
        {
            // Connect to the database.
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            con = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(con);
            StringBuilder sql = new StringBuilder();

            sql.append("ALTER TABLE ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
            sql.append(" ADD ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, WORKFLOW_ID_FIELD)).append(" ");
            sql.append(DbTypeHelper.getIntegerType(dbType));

            if (__logger.isDebugEnabled())
            {
                __logger.debug("Adding column: " + sql.toString());
            }

            stmt = con.prepareStatement(sql.toString());
            stmt.executeUpdate();
        }
        catch (SQLException e)
        {
            __logger.error("Error while adding the workflow id column for the table " + tableName, e);
            throw new FormsException("Error while adding a column to the table " + tableName, e);
        }
        finally
        {
            // Clean up connection resources
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(con);
        }
    }
    
    /**
     * Delete the workflow id column of the given table
     * @param formId the id of the form to alter
     * @throws FormsException if an error occurs
     */
    public void dropWorkflowIdColumn(String formId) throws FormsException
    {
        final String tableName = TABLE_PREFIX + formId;
        
        Connection con = null;
        PreparedStatement stmt = null;
        
        try
        {
            // Connect to the database.
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            con = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(con);
            StringBuilder sql = new StringBuilder();

            sql.append("ALTER TABLE ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName));
            sql.append(" DROP COLUMN ").append(getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, WORKFLOW_ID_FIELD));

            if (__logger.isDebugEnabled())
            {
                __logger.debug("Deleting column: " + sql.toString());
            }

            stmt = con.prepareStatement(sql.toString());
            stmt.executeUpdate();
        }
        catch (SQLException e)
        {
            __logger.error("Error while deleting the workflow id column for the table " + tableName, e);
            throw new FormsException("Error while deleting a column from the table " + tableName, e);
        }
        finally
        {
            // Clean up connection resources
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(con);
        }
    }

    /**
     * Set a new workflow id to the column that has the given old workflow id
     * @param form the form
     * @param entryId the id of the entry
     * @param newWorkflowId the new workflow id
     * @throws FormsException if an error occurs
     */
    public void setWorkflowId(Form form, long entryId, long newWorkflowId) throws FormsException
    {
        final String tableName = TABLE_PREFIX + form.getId();

        Connection con = null;
        PreparedStatement stmt = null;
        
        try
        {
            // Connect to the database.
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            con = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(con);
            
            String query = "UPDATE " + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName);
            query += " SET "  + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, WORKFLOW_ID_FIELD) + " = ?";
            query += " WHERE " + ID_FIELD + " = ?";

            stmt = con.prepareStatement(query);

            stmt.setLong(1, newWorkflowId);
            stmt.setLong(2, entryId);

            // Execute the query.
            stmt.executeUpdate();
        }
        catch (SQLException e)
        {
            __logger.error("Error while resetting the workflow id for the table " + tableName, e);
            throw new FormsException("Error while deleting entry for table " + tableName, e);
        }
        finally
        {
            // Clean up connection resources
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(con);
        }
    }
    
    /**
     * Get the workflow id of the given form entry or of all the entries
     * @param formId the id of the form
     * @param entryId the entry to get the workflow id from. If null, all the workflow ids are returned
     * @throws FormsException if an error occurs.
     * @return the list of workflow ids
     */
    public List<Integer> getWorkflowIds(String formId, Integer entryId) throws FormsException
    {
        List<Integer> workflowIds = new ArrayList<> ();

        final String tableName = TABLE_PREFIX + formId;
        
        Connection con = null;
        PreparedStatement stmt = null;
        ResultSet rs = null;
        
        try
        {
            if (!hasWorkflowIdColumn(formId))
            {
                return Collections.EMPTY_LIST;
            }
            
            // Connect to the database.
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            con = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(con);
            
            String query = "SELECT " + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, "workflowId") + " FROM " + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName);
            if (entryId != null)
            {
                query += " WHERE " + ID_FIELD + " = ?";
            }

            stmt = con.prepareStatement(query);

            if (entryId != null)
            {
                stmt.setInt(1, entryId);
            }
            
            // Execute the query.
            rs = stmt.executeQuery();

            // Extract the result(s).
            while (rs.next())
            {
                workflowIds.add(rs.getInt(1));
            }
        }
        catch (SQLException e)
        {
            __logger.error("Error while getting workflow ids from the table " + tableName, e);
            throw new FormsException("Error while getting workflow ids from the table " + tableName, e);
        }
        finally
        {
            // Clean up connection resources
            ConnectionHelper.cleanup(rs);
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(con);
        }

        return workflowIds;
    }
    
    /**
     * Does the given form have a workflow id column in its SQL table ?
     * @param formId the id of the form
     * @return true if the form has a workflow id column, false otherwise
     * @throws SQLException if a database access error occurs
     */
    public boolean hasWorkflowIdColumn (String formId) throws SQLException
    {
        boolean hasWorkflowId = true;
        
        Connection con = null;
        final String tableName = TABLE_PREFIX + formId;
        
        try
        {
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            con = ConnectionHelper.getConnection(dataSourceId);
            Map<String, DbColumn> currentColumns = _getExistingColumns(con, tableName);        
            
            // Check if the column contains a workflow id
            if (!currentColumns.containsKey(WORKFLOW_ID_FIELD))
            {
                hasWorkflowId = false;
            }
            
            return hasWorkflowId;
        }
        finally 
        {
            ConnectionHelper.cleanup(con);
        }
    }
    
    /**
     * Delete a list of entries of a form
     * @param siteName The site name
     * @param formId The form id
     * @param entries The list of entries to delete
     * @throws ProcessingException if the given parameters are wrong
     * @throws FormsException if an error occurs when deleting the form submission
     */
    @Callable
    public void deleteEntry(String siteName, String formId, List<Integer> entries) throws ProcessingException, FormsException
    {
        if (StringUtils.isEmpty(siteName) || StringUtils.isEmpty(formId))
        {
            throw new ProcessingException("The site name and form ID must be provided.");
        }
    
        Form form = getFormPropertiesManager().getForm(siteName, formId);
        
        if (form == null)
        {
            throw new ProcessingException("The form of ID '" + formId + " can't be found in the site '" + siteName + "'.");
        }
        
        final String tableName = TABLE_PREFIX + form.getId();
        
        Connection connection = null;
        PreparedStatement stmt = null;
        
        try
        {
            String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
            connection = ConnectionHelper.getConnection(dataSourceId);
            String dbType = ConnectionHelper.getDatabaseType(connection);
            
            StringBuilder sb = new StringBuilder();
            sb.append("DELETE FROM " + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName) + " WHERE ");

            
            int i;
            int count = entries.size();
            for (i = 0; i < count; i++)
            {
                if (i > 0)
                {
                    sb.append(" OR ");
                }
                sb.append(ID_FIELD).append("=?");
            }
            
            stmt = connection.prepareStatement(sb.toString());
            
            i = 1;
            List<Integer> workflowIds = new ArrayList<> ();
            for (Integer entryId : entries)
            {
                workflowIds.addAll(getWorkflowIds(formId, entryId));
                stmt.setInt(i, entryId);
                i++;
            }
            
            // Delete the corresponding workflow instances and their history
            for (Integer workflowId : workflowIds)
            {
                getJdbcWorkflowStore().clearHistory(workflowId);
                getJdbcWorkflowStore().deleteInstance(workflowId);
            }
            
            // Execute the query.
            stmt.executeUpdate();
        }
        catch (SQLException e)
        {
            __logger.error("Error while deleting entry for table " + tableName, e);
            throw new FormsException("Error while deleting entry for table " + tableName, e);
        }
        finally
        {
            ConnectionHelper.cleanup(stmt);
            ConnectionHelper.cleanup(connection);
        }
    }
    
    /**
     * Clear all entries of the forms corresponding to the given list of ids.
     * @param siteName the name of the site.
     * @param formIds the list of form ids
     * @throws ProcessingException if an error occurs.
     * @throws FormsException if an error occurs.
     */
    @Callable
    public void clearEntries(String siteName, List<String> formIds) throws ProcessingException, FormsException
    {
        if (StringUtils.isEmpty(siteName) || formIds.isEmpty())
        {
            throw new ProcessingException("The site name and form ID must be provided.");
        }
        
        for (String formId : formIds)
        {
            Form form = getFormPropertiesManager().getForm(siteName, formId);
            
            if (form == null)
            {
                throw new ProcessingException("The form of ID '" + formId + " can't be found in the site '" + siteName + "'.");
            }
            
            final String tableName = TABLE_PREFIX + form.getId();
            
            Connection connection = null;
            PreparedStatement stmt = null;
            
            try
            {
                
                String dataSourceId = Config.getInstance().getValue(FORMS_POOL_CONFIG_PARAM);
                connection = ConnectionHelper.getConnection(dataSourceId);
                String dbType = ConnectionHelper.getDatabaseType(connection);
                
                String request = "DELETE FROM " + getSQLDatabaseTypeExtensionPoint().languageEscapeTableName(dbType, tableName);
                
                stmt = connection.prepareStatement(request);
                
                List<Integer> workflowIds = getWorkflowIds(formId, null);
                // Delete the corresponding workflow instances and their history
                for (Integer workflowId : workflowIds)
                {
                    getJdbcWorkflowStore().clearHistory(workflowId);
                    getJdbcWorkflowStore().deleteInstance(workflowId);
                }
                
                // Execute the query.
                stmt.executeUpdate();
            }
            catch (SQLException e)
            {
                __logger.error("Error while deleting entry for table " + tableName, e);
                throw new FormsException("Error while deleting entry for table " + tableName, e);
            }
            finally
            {
                ConnectionHelper.cleanup(stmt);
                ConnectionHelper.cleanup(connection);
            }
        }
        
    }
    
    /**
     * Get the SELECT statement for retrieving the form entries from the database.
     * @param siteName The site name
     * @param formId The form id
     * @return A result map with the SQL query and the table name
     * @throws ProcessingException if the given parameters are wrong
     */
    @Callable
    public Map<String, String> getSelectStatement(String siteName, String formId) throws ProcessingException 
    {
        Map<String, String> result = new HashMap<>();
        
        try
        {
            Form form = getFormPropertiesManager().getForm(siteName, formId);
            
            if (form == null)
            {
                throw new ProcessingException("The form of ID '" + formId + " can't be found in the site '" + siteName + "'.");
            }
            
            String tableName = FormTableManager.TABLE_PREFIX + form.getId();

            result.put("formId", form.getId());
            result.put("tableName", tableName);
            
            List<String> selectClause = new ArrayList<>();
            
            selectClause.add("id as Id");
            selectClause.add("creationDate as '" + getI18nUtils().translate(new I18nizableText("plugin.forms", "PLUGINS_FORMS_CHOOSE_SHOW_FORM_SUBMISSION_DATE")) + "'");
            
            Map<String, FieldValue> columns = getColumns(form);
            for (FieldValue column : columns.values())
            {
                selectClause.add(column.getColumnName() + " as '" + column.getField().getLabel() + "'");
            }

            StringBuilder request = new StringBuilder();
            request.append("SELECT ").append(StringUtils.join(selectClause, ", ")).append("\nFROM ").append(tableName).append(";");
            
            result.put("query", request.toString());
        }
        catch (FormsException e)
        {
            __logger.error("An error occurred while getting the results of a form.", e);
            throw new ProcessingException("An error occurred while getting the results of a form.", e);
        }
        
        return result;
    }
     
    /**
     * Class storing modifications to do on columns.
     */
    class DbColumnModifications
    {
        
        /** The columns to remove. */
        protected Map<String, DbColumn> _columnsToRemove;
        
        /** The columns to remove. */
        protected Map<String, FieldValue> _columnsToAdd;
        
        /**
         * Build a DbColumnModifications object.
         */
        public DbColumnModifications()
        {
            this(new HashMap<String, DbColumn>(), new HashMap<String, FieldValue>());
        }
        
        /**
         * Build a DbColumnModifications object.
         * @param columnsToRemove the columns to remove.
         * @param columnsToAdd the columns to add.
         */
        public DbColumnModifications(Map<String, DbColumn> columnsToRemove, Map<String, FieldValue> columnsToAdd)
        {
            this._columnsToRemove = columnsToRemove;
            this._columnsToAdd = columnsToAdd;
        }
        
        /**
         * Get the columnsToRemove.
         * @return the columnsToRemove
         */
        public Map<String, DbColumn> getColumnsToRemove()
        {
            return _columnsToRemove;
        }
        
        /**
         * Set the columnsToRemove.
         * @param columnsToRemove the columnsToRemove to set
         */
        public void setColumnsToRemove(Map<String, DbColumn> columnsToRemove)
        {
            this._columnsToRemove = columnsToRemove;
        }
        
        /**
         * Get the columnsToAdd.
         * @return the columnsToAdd
         */
        public Map<String, FieldValue> getColumnsToAdd()
        {
            return _columnsToAdd;
        }
        
        /**
         * Set the columnsToAdd.
         * @param columnsToAdd the columnsToAdd to set
         */
        public void setColumnsToAdd(Map<String, FieldValue> columnsToAdd)
        {
            this._columnsToAdd = columnsToAdd;
        }
        
    }

    /**
     * Class representing a db column.
     */
    class DbColumn
    {
        
        /** The columns to remove. */
        protected String _name;
        
        /** The columns to remove. */
        protected int _sqlType;
        
        /** The columns to remove. */
        protected String _typeName;
        
        /** The columns to remove. */
        protected int _columnSize;
        
        /**
         * Build a DB column object.
         */
        public DbColumn()
        {
            this("", 0, "", 0);
        }
        
        /**
         * Build a DB column object.
         * @param name the column name.
         * @param sqlType the sql type.
         * @param typeName the type name.
         * @param columnSize the column size.
         */
        public DbColumn(String name, int sqlType, String typeName, int columnSize)
        {
            this._name = name;
            this._sqlType = sqlType;
            this._typeName = typeName;
            this._columnSize = columnSize;
        }
        
        /**
         * Get the name.
         * @return the name
         */
        public String getName()
        {
            return _name;
        }
        
        /**
         * Set the name.
         * @param name the name to set
         */
        public void setName(String name)
        {
            this._name = name;
        }
        
        /**
         * Get the sqlType.
         * @return the sqlType
         */
        public int getSqlType()
        {
            return _sqlType;
        }
        
        /**
         * Set the sqlType.
         * @param sqlType the sqlType to set
         */
        public void setSqlType(int sqlType)
        {
            this._sqlType = sqlType;
        }
        
        /**
         * Get the typeName.
         * @return the typeName
         */
        public String getTypeName()
        {
            return _typeName;
        }
        
        /**
         * Set the typeName.
         * @param typeName the typeName to set
         */
        public void setTypeName(String typeName)
        {
            this._typeName = typeName;
        }
        
        /**
         * Get the columnSize.
         * @return the columnSize
         */
        public int getColumnSize()
        {
            return _columnSize;
        }
        
        /**
         * Set the columnSize.
         * @param columnSize the columnSize to set
         */
        public void setColumnSize(int columnSize)
        {
            this._columnSize = columnSize;
        }
        
        /**
         * Get a type identifier corresponding to the column (i.e. TEXT, VARCHAR(255), INT(1), and so on.)
         * @return a type identifier  corresponding to the column (i.e. TEXT, VARCHAR(255), INT(1), and so on.)
         */
        public String getColumnTypeIdentifier()
        {
            StringBuilder buff = new StringBuilder();
            
            buff.append(_typeName);
            if (_typeName.equals("VARCHAR") || _typeName.equals("INT") || _typeName.equals("NUMBER"))
            {
                buff.append('(').append(_columnSize).append(')');
            }
            
            return buff.toString();
        }
        
    }
    
}