DBBrokerTemplate.java

/*
 * _____________________________________________________________________________
 * 
 * INAF - OATS National Institute for Astrophysics - Astronomical Observatory of
 * Trieste INAF - IA2 Italian Center for Astronomical Archives
 * _____________________________________________________________________________
 * 
 * Copyright (C) 2017 Istituto Nazionale di Astrofisica
 * 
 * This program is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License Version 3 as published by the
 * Free Software Foundation.
 * 
 * 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 General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
package it.inaf.ia2.tsm.datalayer;

import it.inaf.ia2.tsm.Column;
import it.inaf.ia2.tsm.ColumnHolder;
import it.inaf.ia2.tsm.ConsistencyChecks;
import it.inaf.ia2.tsm.Key;
import it.inaf.ia2.tsm.KeyColumn;
import it.inaf.ia2.tsm.Schema;
import it.inaf.ia2.tsm.Status;
import it.inaf.ia2.tsm.TSMUtil;
import it.inaf.ia2.tsm.Table;
import it.inaf.ia2.tsm.TapSchema;
import it.inaf.ia2.tsm.TapSchemaEntity;
import it.inaf.ia2.tsm.UpdateOperations;
import it.inaf.ia2.tsm.model.ColumnModel;
import it.inaf.ia2.tsm.model.TableModel;
import it.inaf.ia2.tsm.model.SchemaModel;
import it.inaf.ia2.tsm.model.SchemaModels;
import it.inaf.ia2.tsm.model.TypesMapping;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Sonia Zorba {@literal <zorba at oats.inaf.it>}
 */
public abstract class DBBrokerTemplate implements DBBroker {

    private static final Logger LOG = LoggerFactory.getLogger(DBBrokerTemplate.class);

    protected final DataSource dataSource;
    private final char escapeCharacter;
    private final DataTypeMode dataTypeMode;

    public DBBrokerTemplate(DataSource dataSource, char escapeCharacter, DataTypeMode dataTypeMode) {
        this.dataSource = dataSource;
        this.escapeCharacter = escapeCharacter;
        this.dataTypeMode = dataTypeMode;
    }

    protected List<String> getAllItemsNames(String query) throws SQLException {

        List<String> allSchemas = new ArrayList<>();

        LOG.debug("Executing query: {}", query);

        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(query)) {
            while (resultSet.next()) {
                allSchemas.add(resultSet.getString(1));
            }
        }

        Collections.sort(allSchemas, String.CASE_INSENSITIVE_ORDER);
        return allSchemas;
    }

    protected String buildColumnsList(String[] columns) {

        StringBuilder sb = new StringBuilder();

        boolean first = true;
        for (String keyColumn : columns) {
            if (!first) {
                sb.append(",");
            }
            first = false;
            sb.append(escape(keyColumn));
        }

        return sb.toString();
    }

    protected String escape(String name) {
        return String.format("%s%s%s", escapeCharacter, name, escapeCharacter);
    }

    protected void appendSize(StringBuilder sb, int size) {
        sb.append(String.format("(%s)", size));
    }

    @Override
    public void createTable(String schemaName, TableModel tableModel) throws SQLException {
        try (Connection conn = dataSource.getConnection()) {
            createTable(schemaName, tableModel, conn);
        }
    }

    protected abstract void createTable(String schemaName, TableModel tableModel, Connection conn) throws SQLException;

    protected abstract String getAddPrimaryKeyQuery(String tapSchemaName, String tableName, String[] keyColumns);

    protected abstract String getAddForeignKeyQuery(String tapSchemaName, String tableName, String[] fromKeyColumns, String targetTableName, String[] toKeyColumns);

    private String getAddPrimaryKeyQuery(String tapSchemaName, String tableName, String keyColumn) {
        return getAddPrimaryKeyQuery(tapSchemaName, tableName, new String[]{keyColumn});
    }

    private String getAddForeignKeyQuery(String tapSchemaName, String tableName, String fromKeyColumn, String targetTableName, String toKeyColumn) {
        return getAddForeignKeyQuery(tapSchemaName, tableName, new String[]{fromKeyColumn}, targetTableName, new String[]{toKeyColumn});
    }

    protected abstract String getCreateDatabaseQuery(String databaseName);

    private void execute(String query, Connection conn) throws SQLException {
        try (Statement stat = conn.createStatement()) {
            LOG.debug("Executing query: {}", query);
            stat.execute(query);
        }
    }

    @Override
    public void createTapSchemaStructure(String tapSchemaName, SchemaModel tapSchemaModel) throws SQLException {

        try (Connection conn = dataSource.getConnection()) {

            execute(getCreateDatabaseQuery(tapSchemaName), conn);

            for (TableModel tableModel : tapSchemaModel.getTables()) {
                createTable(tapSchemaName, tableModel, conn);
            }

            // schemas keys
            execute(getAddPrimaryKeyQuery(tapSchemaName, TapSchema.SCHEMAS_TABLE, Schema.SCHEMA_NAME_KEY), conn);

            // tables keys
            execute(getAddPrimaryKeyQuery(tapSchemaName, TapSchema.TABLES_TABLE, Table.TABLE_NAME_KEY), conn);
            execute(getAddForeignKeyQuery(tapSchemaName, TapSchema.TABLES_TABLE, Table.SCHEMA_NAME_KEY, TapSchema.SCHEMAS_TABLE, Schema.SCHEMA_NAME_KEY), conn);

            // columns keys
            execute(getAddPrimaryKeyQuery(tapSchemaName, TapSchema.COLUMNS_TABLE, new String[]{Column.TABLE_NAME_KEY, Column.COLUMN_NAME_KEY}), conn);
            execute(getAddForeignKeyQuery(tapSchemaName, TapSchema.COLUMNS_TABLE, Column.TABLE_NAME_KEY, TapSchema.TABLES_TABLE, Table.TABLE_NAME_KEY), conn);

            // keys keys
            execute(getAddPrimaryKeyQuery(tapSchemaName, TapSchema.KEYS_TABLE, Key.ID_KEY), conn);
            execute(getAddForeignKeyQuery(tapSchemaName, TapSchema.KEYS_TABLE, Key.FROM_TABLE_KEY, TapSchema.TABLES_TABLE, Table.TABLE_NAME_KEY), conn);
            execute(getAddForeignKeyQuery(tapSchemaName, TapSchema.KEYS_TABLE, Key.TARGET_TABLE_KEY, TapSchema.TABLES_TABLE, Table.TABLE_NAME_KEY), conn);

            // key columns key
            //addPrimaryKey(tapSchemaName, TapSchema.KEY_COLUMNS_TABLE, new String[]{KeyColumn.KEY_ID_KEY, KeyColumn.FROM_COLUMN_KEY, KeyColumn.TARGET_COLUMN_KEY}, conn);
            execute(getAddForeignKeyQuery(tapSchemaName, TapSchema.KEY_COLUMNS_TABLE, Key.ID_KEY, TapSchema.KEYS_TABLE, Key.ID_KEY), conn);
            //addForeignKey(tapSchemaName, TapSchema.KEY_COLUMNS_TABLE, KeyColumn.FROM_COLUMN_KEY, TapSchema.COLUMNS_TABLE, Column.COLUMN_NAME_KEY, conn);
            //addForeignKey(tapSchemaName, TapSchema.KEY_COLUMNS_TABLE, KeyColumn.TARGET_COLUMN_KEY, TapSchema.COLUMNS_TABLE, Column.COLUMN_NAME_KEY, conn);
        }
    }

    @Override
    public void createIvoaSchemaStructure(SchemaModel ivoaSchemaModel) throws SQLException {
        try (Connection conn = dataSource.getConnection()) {
            execute(getCreateDatabaseQuery(ivoaSchemaModel.getName()), conn);
            for (TableModel tableModel : ivoaSchemaModel.getTables()) {
                createTable(ivoaSchemaModel.getName(), tableModel, conn);
            }
        }
    }

    @Override
    public void save(TapSchema tapSchema) throws SQLException {

        LOG.debug("Saving TAP_SCHEMA");

        Connection connection = null;
        PreparedStatement statement = null;
        boolean transactionStarted = false;

        try {

            connection = dataSource.getConnection();

            UpdateOperations operations = new UpdateOperations(tapSchema);

            // Start update
            connection.setAutoCommit(false); // start transaction
            transactionStarted = true;

            String tapSchemaNameEscaped = escape(tapSchema.getRealName());

            // REMOVE ELEMENTS
            if (tapSchema.exists()) {
                for (Key key : operations.getKeysToRemove()) {
                    String keyId = key.getId();

                    String query = String.format("DELETE FROM %s.%s WHERE key_id = ?", tapSchemaNameEscaped, escape("key_columns"));
                    statement = connection.prepareStatement(query);
                    statement.setString(1, keyId);
                    LOG.debug("Executing query {} [key_id={}]", query, keyId);
                    statement.executeUpdate();

                    query = String.format("DELETE FROM %s.%s WHERE key_id = ?", tapSchemaNameEscaped, escape("keys"));
                    statement = connection.prepareStatement(query);
                    statement.setString(1, keyId);
                    LOG.debug("Executing query {} [key_id={}]", query, keyId);
                    statement.executeUpdate();
                }

                for (Column column : operations.getColumnsToRemove()) {
                    String query = String.format("DELETE FROM %s.%s WHERE table_name = ? AND column_name = ?", tapSchemaNameEscaped, escape("columns"));
                    statement = connection.prepareStatement(query);
                    String tableName = column.getTableCompleteName();
                    String columnName = column.getName();
                    statement.setString(1, tableName);
                    statement.setString(2, columnName);
                    LOG.debug("Executing query {} [table_name={}, column_name={}]", query, tableName, columnName);
                    statement.executeUpdate();
                }

                for (Table table : operations.getTablesToRemove()) {
                    String query = String.format("DELETE FROM %s.%s WHERE table_name = ?", tapSchemaNameEscaped, escape("tables"));
                    statement = connection.prepareStatement(query);
                    String tableCompleteName = table.getCompleteName();
                    statement.setString(1, tableCompleteName);
                    LOG.debug("Executing query {} [table_name={}]", query, tableCompleteName);
                    statement.executeUpdate();
                }

                for (Schema schema : operations.getSchemasToRemove()) {
                    String query = String.format("DELETE FROM %s.%s WHERE schema_name = ?", tapSchemaNameEscaped, escape("schemas"));
                    statement = connection.prepareStatement(query);
                    String schemaName = schema.getName();
                    statement.setString(1, schemaName);
                    LOG.debug("Executing query {} [schema_name={}]", query, schemaName);
                    statement.executeUpdate();
                }
            }

            // INSERT ELEMENTS
            if (!operations.getSchemasToAdd().isEmpty()) {
                LOG.debug("Inserting {} new schemas", operations.getSchemasToAdd().size());
            }
            for (Schema schema : operations.getSchemasToAdd()) {
                insertItem(tapSchema.getRealName(), schema, connection);
            }

            if (!operations.getTablesToAdd().isEmpty()) {
                LOG.debug("Inserting {} new tables", operations.getTablesToAdd().size());
            }
            for (Table table : operations.getTablesToAdd()) {
                insertItem(tapSchema.getRealName(), table, connection);
            }

            if (!operations.getColumnsToAdd().isEmpty()) {
                LOG.debug("Inserting {} new columns", operations.getColumnsToAdd().size());
            }
            for (Column column : operations.getColumnsToAdd()) {
                insertItem(tapSchema.getRealName(), column, connection);
            }

            if (!operations.getKeysToAdd().isEmpty()) {
                LOG.debug("Inserting {} new keys", operations.getKeysToAdd().size());
            }
            for (Key key : operations.getKeysToAdd()) {
                // insert new keys and their key columns
                insertItem(tapSchema.getRealName(), key, connection);
                for (KeyColumn keyColumn : key.getKeyColumns()) {
                    insertItem(tapSchema.getRealName(), keyColumn, connection);
                }
            }

            //UPDATE ELEMENTS
            if (tapSchema.exists()) {
                for (Key key : operations.getKeysToUpdate()) {
                    // update keys and their key columns
                    String whereCond = String.format("%s = ?", escape(Key.ID_KEY));
                    updateItem(tapSchema.getRealName(), key, connection, whereCond, key.getId());
                    for (KeyColumn keyColumn : key.getKeyColumns()) {
                        whereCond = String.format("%s = ?", escape(KeyColumn.KEY_ID_KEY));
                        updateItem(tapSchema.getRealName(), keyColumn, connection, whereCond, keyColumn.getKeyId());
                    }
                }

                for (Schema schema : operations.getSchemasToUpdate()) {
                    String whereCond = String.format("%s = ?", escape(Schema.SCHEMA_NAME_KEY));
                    updateItem(tapSchema.getRealName(), schema, connection, whereCond, schema.getName());
                }

                for (Table table : operations.getTablesToUpdate()) {
                    String whereCond = String.format("%s = ?", escape(Table.TABLE_NAME_KEY));
                    updateItem(tapSchema.getRealName(), table, connection, whereCond, table.getCompleteName());
                }

                for (Column column : operations.getColumnsToUpdate()) {
                    String whereCond = String.format("%s = ? AND %s = ?", escape(Column.TABLE_NAME_KEY), escape(Column.COLUMN_NAME_KEY));
                    updateItem(tapSchema.getRealName(), column, connection, whereCond, column.getTableCompleteName(), column.getName());
                }
            }

            connection.commit();

            // Status cleanup after commit
            // added
            for (Key key : operations.getKeysToAdd()) {
                key.save();
            }
            for (Schema schema : operations.getSchemasToAdd()) {
                schema.save();
            }
            for (Table table : operations.getTablesToAdd()) {
                table.save();
            }
            for (Column column : operations.getColumnsToAdd()) {
                column.save();
            }

            // removed
            for (Key key : operations.getKeysToRemove()) {
                key.initProperty(Key.ID_KEY, null);
                for (KeyColumn keyColumn : key.getKeyColumns()) {
                    keyColumn.initProperty(KeyColumn.KEY_ID_KEY, null);
                }
            }
            for (Column column : operations.getColumnsToRemove()) {
                column.setStatus(Status.LOADED);
            }
            for (Column column : operations.getColumnsToClean()) {
                column.setStatus(Status.LOADED);
            }
            for (Table table : operations.getTablesToRemove()) {
                Schema schema = tapSchema.getChild(table.getParent().getName());
                if (schema != null) {
                    schema.cleanTable(table.getName());
                }
            }
            for (Table table : operations.getTablesToClean()) {
                Schema schema = tapSchema.getChild(table.getParent().getName());
                if (schema != null) {
                    schema.cleanTable(table.getName());
                }
            }
            for (Schema schema : operations.getSchemasToRemove()) {
                tapSchema.cleanSchema(schema.getName());
            }
            for (Schema schema : operations.getSchemasToClean()) {
                tapSchema.cleanSchema(schema.getName());
            }

            // updated
            for (Key key : operations.getKeysToUpdate()) {
                key.save();
            }
            for (Schema schema : operations.getSchemasToUpdate()) {
                schema.save();
            }
            for (Table table : operations.getTablesToUpdate()) {
                table.save();
            }
            for (Column column : operations.getColumnsToUpdate()) {
                column.save();
            }
        } catch (SQLException e) {
            LOG.error("Exception caught", e);
            try {
                if (connection != null && transactionStarted) {
                    LOG.debug("Executing rollback");
                    connection.rollback();
                }
            } catch (SQLException e2) {
                LOG.error("Exception caught", e2);
            }
            throw e;
        } finally {
            if (connection != null) {
                try {
                    if (statement != null) {
                        statement.close();
                    }
                    connection.close();
                } catch (SQLException e2) {
                    LOG.error("Exception caught", e2);
                }
            }
        }
    }

    @Override
    public List<Map<String, Object>> getSavedItems(String tapSchemaName, TableModel tableModel, String whereCondition, Object[] whereParams) throws SQLException {

        StringBuilder querySb = new StringBuilder("SELECT ");

        boolean first = true;
        for (ColumnModel cm : tableModel.getColumns()) {
            if (!first) {
                querySb.append(", ");
            }
            first = false;
            querySb.append(escape(cm.getName()));
        }

        querySb.append(" FROM ");

        querySb.append(escape(tapSchemaName));
        querySb.append(".");
        querySb.append(escape(tableModel.getName()));

        // TODO: Manage where condition
        String query = querySb.toString();

        LOG.debug("Executing query {}", query);

        try (Connection conn = dataSource.getConnection()) {
            try (Statement statement = conn.createStatement();
                    ResultSet rs = statement.executeQuery(query)) {

                List<Map<String, Object>> items = new ArrayList<>();

                while (rs.next()) {
                    Map<String, Object> item = new HashMap<>();

                    for (ColumnModel cm : tableModel.getColumns()) {
                        Object value = TSMUtil.getObject(rs, cm.getName(), cm.getJavaType());
                        item.put(cm.getName(), value);
                    }

                    items.add(item);
                }

                return items;
            }
        }
    }

    @Override
    public List<Map<String, Object>> getSavedItems(String tapSchemaName, TableModel tableModel) throws SQLException {
        return getSavedItems(tapSchemaName, tableModel, null, null);
    }

    private int getSQLType(Class type) {
        if (type == String.class) {
            return Types.VARCHAR;
        } else if (type == Integer.class) {
            return Types.INTEGER;
        } else if (type == Long.class) {
            return Types.BIGINT;
        } else if (type == Boolean.class) {
            return Types.BIT;
        } else {
            throw new UnsupportedOperationException("Class type " + type.getCanonicalName() + " not supported yet!");
        }
    }

    @Override
    public void insertItem(String tapSchemaName, TapSchemaEntity tapSchemaItem, Connection conn) throws SQLException {

        StringBuilder querySb = new StringBuilder("INSERT INTO ");
        querySb.append(escape(tapSchemaName));
        querySb.append(".");
        querySb.append(escape(tapSchemaItem.getTableModel().getName()));
        querySb.append(" (");

        boolean first = true;
        for (String key : tapSchemaItem.getPropertiesKeys()) {
            if (!first) {
                querySb.append(", ");
            }
            first = false;
            querySb.append(escape(key));
        }

        querySb.append(") VALUES (");
        first = true;
        for (String key : tapSchemaItem.getPropertiesKeys()) {
            if (!first) {
                querySb.append(",");
            }
            first = false;
            querySb.append("?");
        }
        querySb.append(")");

        String query = querySb.toString();

        try (PreparedStatement statement = conn.prepareStatement(query)) {

            List<Object> values = null;
            if (LOG.isDebugEnabled()) {
                values = new ArrayList<>();
            }

            int i = 1;
            for (String key : tapSchemaItem.getPropertiesKeys()) {
                String adqlType = tapSchemaItem.getTableModel().get(key).getType();
                Class javaType = TypesMapping.getClassFromAdqlType(adqlType);
                Object value = tapSchemaItem.getValue(key, javaType);
                statement.setObject(i, value, getSQLType(javaType));
                i++;

                if (values != null) {
                    values.add(value);
                }
            }

            LOG.debug("Executing query {} {}", query, values);
            statement.executeUpdate();
        }
    }

    @Override
    public void updateItem(String tapSchemaName, TapSchemaEntity tapSchemaItem, Connection conn, String whereCondition, Object... whereParams) throws SQLException {

        StringBuilder querySb = new StringBuilder("UPDATE ");
        querySb.append(escape(tapSchemaName));
        querySb.append(".");
        querySb.append(escape(tapSchemaItem.getTableModel().getName()));
        querySb.append("\nSET");

        boolean first = true;
        for (String key : tapSchemaItem.getPropertiesKeys()) {
            if (!first) {
                querySb.append(",");
            }
            first = false;
            querySb.append(" ");
            querySb.append(escape(key));
            querySb.append(" = ?");
        }

        querySb.append("\nWHERE ");
        querySb.append(whereCondition);

        String query = querySb.toString();

        try (PreparedStatement ps = conn.prepareStatement(query)) {

            int i = 1;

            List<Object> statParams = null;
            if (LOG.isDebugEnabled()) {
                statParams = new ArrayList<>();
            }
            for (String key : tapSchemaItem.getPropertiesKeys()) {
                Object value = tapSchemaItem.getValue(key);
                ps.setObject(i, value, getSQLType(tapSchemaItem.getPropertyType(key)));
                i++;
                if (statParams != null) {
                    statParams.add(value);
                }
            }
            for (Object wp : whereParams) {
                ps.setObject(i, wp, getSQLType(wp.getClass()));
                i++;
                if (statParams != null) {
                    statParams.add(wp);
                }
            }

            LOG.debug("Executing query: {} {}", query, statParams);

            ps.executeUpdate();
        }
    }

    protected abstract String getSchemaTablesQuery(String schemaName);

    @Override
    public List<String> getAllTAPSchemaNames(List<String> allSchemata) throws SQLException {

        List<String> allTAPSchemas = new ArrayList<>();

        for (String schemaName : allSchemata) {

            boolean schemas = false,
                    tables = false,
                    columns = false,
                    keys = false,
                    keyColumns = false;

            String query = getSchemaTablesQuery(schemaName);

            LOG.debug("Executing query {}", query);

            try (Connection connection = dataSource.getConnection();
                    Statement statement = connection.createStatement();
                    ResultSet resultSet = statement.executeQuery(query)) {
                while (resultSet.next()) {
                    String shortTableName = resultSet.getString(1);

                    if (null != shortTableName) {
                        switch (shortTableName) {
                            case "schemas":
                                schemas = true;
                                break;
                            case "tables":
                                tables = true;
                                break;
                            case "columns":
                                columns = true;
                                break;
                            case "keys":
                                keys = true;
                                break;
                            case "key_columns":
                                keyColumns = true;
                                break;
                        }
                    }
                }
            }

            if (schemas && tables && columns && keys && keyColumns) {
                // the schema is a TAP_SCHEMA
                allTAPSchemas.add(schemaName);
            }
        }

        LOG.debug("{} TAP_SCHEMA schemas found", allTAPSchemas.size());

        Collections.sort(allTAPSchemas, String.CASE_INSENSITIVE_ORDER);

        return allTAPSchemas;
    }

    protected abstract String getColumnNamesQuery(String tapSchemaName, String tableName);

    private List<String> getColumns(String tapSchemaName, String tableName, Connection connection) throws SQLException {

        List<String> columns = new ArrayList<>();

        String query = getColumnNamesQuery(tapSchemaName, tableName);
        LOG.debug("Executing query {}", query);

        try (Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(query)) {
            while (resultSet.next()) {
                String columnName = resultSet.getString(1);
                columns.add(columnName);
            }
        }

        return columns;
    }

    private boolean match(TableModel tableModel, List<String> columns) {
        if (tableModel.getColumns().size() != columns.size()) {
            return false;
        }
        for (ColumnModel propertyModel : tableModel.getColumns()) {
            String columnName = propertyModel.getName();
            if (!columns.contains(columnName)) {
                return false;
            }
        }
        return true;
    }

    @Override
    public String detectVersion(String tapSchemaName) throws SQLException {

        List<String> schemasColumns, tablesColumns, columnsColumns, keyColumns, keyColumnsColumns;

        try (Connection connection = dataSource.getConnection()) {
            schemasColumns = getColumns(tapSchemaName, TapSchema.SCHEMAS_TABLE, connection);
            tablesColumns = getColumns(tapSchemaName, TapSchema.TABLES_TABLE, connection);
            columnsColumns = getColumns(tapSchemaName, TapSchema.COLUMNS_TABLE, connection);
            keyColumns = getColumns(tapSchemaName, TapSchema.KEYS_TABLE, connection);
            keyColumnsColumns = getColumns(tapSchemaName, TapSchema.KEY_COLUMNS_TABLE, connection);
        }

        for (SchemaModel tapSchemaModel : SchemaModels.getTapSchemaModels()) {
            if (match(tapSchemaModel.getTable(TapSchema.SCHEMAS_TABLE), schemasColumns)
                    && match(tapSchemaModel.getTable(TapSchema.TABLES_TABLE), tablesColumns)
                    && match(tapSchemaModel.getTable(TapSchema.COLUMNS_TABLE), columnsColumns)
                    && match(tapSchemaModel.getTable(TapSchema.KEYS_TABLE), keyColumns)
                    && match(tapSchemaModel.getTable(TapSchema.KEY_COLUMNS_TABLE), keyColumnsColumns)) {
                return tapSchemaModel.getVersion();
            }
        }

        throw new RuntimeException("Unable to detect TAP_SCHEMA version for " + tapSchemaName);
    }

    protected abstract String getTableTypesQuery(String schemaName);

    @Override
    public Map<String, String> getAllTableTypes(String schemaName) throws SQLException {
        LOG.debug("getTablesTypes");

        final Map<String, String> tablesTypes = new HashMap<>();

        String query = getTableTypesQuery(schemaName);
        LOG.debug("Executing query {}", query);

        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(query)) {
            while (resultSet.next()) {
                String tableName = resultSet.getString("table_name");
                String tableType = resultSet.getString("table_type").equalsIgnoreCase("VIEW") ? "view" : "table";
                tablesTypes.put(tableName, tableType);
            }
        }

        return tablesTypes;
    }

    @Override
    public List<String> getExposedSchemas(String tapSchemaName) throws SQLException {

        final List<String> exposedSchemas = new ArrayList<>();

        String query = String.format("SELECT %s FROM %s.%s", Schema.SCHEMA_NAME_KEY, escape(tapSchemaName), escape(TapSchema.SCHEMAS_TABLE));

        LOG.debug("Executing query " + query);

        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(query)) {
            while (resultSet.next()) {
                exposedSchemas.add(resultSet.getString(1));
            }
        }

        return exposedSchemas;
    }

    protected abstract String getAllSchemaNamesQuery();

    @Override
    public List<String> getAllSchemaNames() throws SQLException {
        String query = getAllSchemaNamesQuery();
        LOG.debug("Executing query: {}", query);
        List<String> allSchemas = getAllItemsNames(query);
        LOG.debug("{} schemas found", allSchemas.size());
        return allSchemas;
    }

    protected abstract String getAllTablesNamesQuery(String schemaName);

    @Override
    public List<String> getAllTablesNames(String schemaName) throws SQLException {
        String query = getAllTablesNamesQuery(schemaName);
        LOG.debug("Executing query: {}", query);
        List<String> allTables = getAllItemsNames(query);
        LOG.debug("{} tables found", allTables.size());
        return allTables;
    }

    @Override
    public List<String> getAllColumnsNames(String schemaName, String tableName) throws SQLException {
        String query = getColumnNamesQuery(schemaName, tableName);
        List<String> allColumns = getAllItemsNames(query);
        LOG.debug("{} columns found", allColumns.size());
        return allColumns;
    }

    protected abstract Map<String, Map<String, Object>> getAllColumnsOriginalMetadata(String schemaName, String tableName) throws SQLException;

    @Override
    public Map<String, Map<String, Object>> getAllColumnsMetadata(String schemaName, String tableName, TableModel tableModel, DataTypeMode dataTypeMode) throws SQLException {

        Map<String, Map<String, Object>> metadata = getAllColumnsOriginalMetadata(schemaName, tableName);

        // Special behavior for data type
        if (tableModel != null) {
            for (Map.Entry<String, Map<String, Object>> entry : metadata.entrySet()) {

                String columnName = entry.getKey();

                // Saving original data type (used for consistency checking)
                Map<String, Object> columnMetadata = entry.getValue();
                String originalDataType = (String) columnMetadata.get(Column.DATATYPE_KEY);
                columnMetadata.put(Column.ORIGINAL_DATATYPE_KEY, originalDataType);

                // Override data type using model definition
                String adqlType = tableModel.get(columnName).getType();
                String definedDataType = TypesMapping.getDataType(adqlType, dataTypeMode);
                columnMetadata.put(Column.DATATYPE_KEY, definedDataType);
            }
        }

        return metadata;
    }

    protected DataTypeMode getDataTypeMode() {
        return dataTypeMode;
    }

    protected abstract String getDataTypeFromADQLType(String adqlType);

    @Override
    public void addColumn(ColumnHolder columnHolder, ColumnModel columnModel) throws SQLException {

        String dataType = getDataTypeFromADQLType(columnModel.getType());
        if (columnModel.getSize() != null) {
            dataType = String.format("%s(%s)", dataType, columnModel.getSize());
        }

        String query = String.format("ALTER TABLE %s.%s ADD COLUMN %s %s",
                escape(columnHolder.getSchemaName()), escape(columnHolder.getTableName()),
                escape(columnHolder.getColumnName()), dataType);

        LOG.debug("Executing query " + query);

        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement()) {
            statement.executeUpdate(query);
        }
    }

    @Override
    public Set<String> getKeysToRemove(String tapSchemaName, String like) throws SQLException {

        Set<String> keysId = new HashSet<>();

        String query = String.format("SELECT key_id from %s.%s WHERE from_table LIKE ? OR target_table LIKE ?",
                escape(tapSchemaName), escape(TapSchema.KEYS_TABLE));

        try (Connection connection = dataSource.getConnection();
                PreparedStatement ps = connection.prepareStatement(query)) {
            ps.setString(1, like + "%");
            ps.setString(2, like + "%");
            LOG.debug("Executing query: {} [{}]", query, like);
            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) {
                    keysId.add(rs.getString("key_id"));
                }
            }
        }

        return keysId;
    }

    @Override
    public Set<String> getKeysToRemoveFromUnexistingColumn(String tapSchemaName, ColumnHolder unexistingColumn) throws SQLException {

        Set<String> keysToRemoveIds = new HashSet<>();

        StringBuilder sb = new StringBuilder();
        sb.append("SELECT k.key_id AS key_id\n");
        sb.append("FROM ");
        sb.append(tapSchemaName);
        sb.append(".`keys` k\n");
        sb.append("JOIN ");
        sb.append(tapSchemaName);
        sb.append(".key_columns c ON k.key_id = c.key_id\n");
        sb.append("WHERE (k.from_table = ? AND c.from_column = ?) OR (k.target_table = ? AND c.target_column = ?)");

        String query = sb.toString();

        String completeTableName = String.format("%s.%s",
                unexistingColumn.getSchemaName(), unexistingColumn.getTableName());

        try (Connection conn = dataSource.getConnection();
                PreparedStatement ps = conn.prepareStatement(query)) {

            ps.setString(1, completeTableName);
            ps.setString(2, unexistingColumn.getColumnName());
            ps.setString(3, completeTableName);
            ps.setString(4, unexistingColumn.getColumnName());

            LOG.debug("Executing query {}", query);

            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) {
                    keysToRemoveIds.add(rs.getString("key_id"));
                }
            }
        }

        return keysToRemoveIds;
    }

    @Override
    public void deleteUnexistingEntities(String tapSchemaName, ConsistencyChecks consistencyChecks, Set<String> keysToRemoveIds) throws SQLException {
        try (Connection conn = dataSource.getConnection()) {

            String query, tapSchemaNameEscaped = escape(tapSchemaName);
            conn.setAutoCommit(false);
            LOG.debug("Starting transaction");

            try {
                // Removing all key_columns
                for (String keyId : keysToRemoveIds) {
                    query = String.format("DELETE FROM %s.%s WHERE key_id = ?", tapSchemaNameEscaped, escape(TapSchema.KEY_COLUMNS_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        ps.setString(1, keyId);
                        LOG.debug("Executing query {} [{}]", query, keyId);
                        ps.executeUpdate();
                    }
                }

                // Removing all keys
                for (String keyId : keysToRemoveIds) {
                    query = String.format("DELETE FROM %s.%s WHERE key_id = ?", tapSchemaNameEscaped, escape(TapSchema.KEYS_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        ps.setString(1, keyId);
                        LOG.debug("Executing query {} [{}]", query, keyId);
                        ps.executeUpdate();
                    }
                }

                // Removing all columns
                for (ColumnHolder unexistingColumn : consistencyChecks.getUnexisingColumns()) {
                    query = String.format("DELETE FROM %s.%s WHERE table_name = ? AND column_name = ?", tapSchemaNameEscaped, escape(TapSchema.COLUMNS_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        String completeTableName = String.format("%s.%s",
                                unexistingColumn.getSchemaName(), unexistingColumn.getTableName());
                        ps.setString(1, completeTableName);
                        ps.setString(2, unexistingColumn.getColumnName());
                        LOG.debug("Executing query {} [{}, {}]", query, completeTableName, unexistingColumn.getColumnName());
                        ps.executeUpdate();
                    }
                }
                for (String table : consistencyChecks.getUnexisingTables()) {
                    query = String.format("DELETE FROM %s.%s WHERE table_name = ?", tapSchemaNameEscaped, escape(TapSchema.COLUMNS_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        ps.setString(1, table);
                        LOG.debug("Executing query {} [{}]", query, table);
                        ps.executeUpdate();
                    }
                }
                for (String schema : consistencyChecks.getUnexisingSchemas()) {
                    query = String.format("DELETE FROM %s.%s WHERE table_name LIKE ?", tapSchemaNameEscaped, escape(TapSchema.COLUMNS_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        ps.setString(1, schema + "%");
                        LOG.debug("Executing query {} [{}%]", query, schema);
                        ps.executeUpdate();
                    }
                }

                // Removing all tables
                for (String table : consistencyChecks.getUnexisingTables()) {
                    query = String.format("DELETE FROM %s.%s WHERE table_name = ?", tapSchemaNameEscaped, escape(TapSchema.TABLES_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        ps.setString(1, table);
                        LOG.debug("Executing query {} [{}]", query, table);
                        ps.executeUpdate();
                    }
                }
                for (String schema : consistencyChecks.getUnexisingSchemas()) {
                    query = String.format("DELETE FROM %s.%s WHERE schema_name = ?", tapSchemaNameEscaped, escape(TapSchema.TABLES_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        ps.setString(1, schema);
                        LOG.debug("Executing query {} [{}]", query, schema);
                        ps.executeUpdate();
                    }
                }

                // Removing all schemas
                for (String schema : consistencyChecks.getUnexisingSchemas()) {
                    query = String.format("DELETE FROM %s.%s WHERE schema_name = ?", tapSchemaNameEscaped, escape(TapSchema.SCHEMAS_TABLE));
                    try (PreparedStatement ps = conn.prepareStatement(query)) {
                        ps.setString(1, schema);
                        LOG.debug("Executing query {} [{}]", query, schema);
                        ps.executeUpdate();
                    }
                }

                conn.commit();
            } catch (SQLException e) {
                LOG.error("Exception detected. Executing rollback!", e);
                try {
                    conn.rollback();
                    conn.setAutoCommit(true);
                } catch (SQLException er) {
                    LOG.error("Exception during rollback", er);
                    throw er;
                }
            }
        }
    }

    @Override
    public void updateTapSchemaColumnValue(String tapSchemaName, String completeTableName, String columnName, String key, Object value) throws SQLException {