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.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.List;
import java.util.Map;
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));
    }

    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.getName());

            // 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.getName(), schema, connection);
            }

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

            if (!operations.getColumnsToAdd().isEmpty()) {
                LOG.debug("Inserting {} new columns", operations.getColumnsToAdd().size());
            }
            for (Column column : operations.getColumnsToAdd()) {
                insertItem(tapSchema.getName(), 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.getName(), key, connection);
                for (KeyColumn keyColumn : key.getKeyColumns()) {
                    insertItem(tapSchema.getName(), 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.getName(), key, connection, whereCond, key.getId());
                    for (KeyColumn keyColumn : key.getKeyColumns()) {
                        whereCond = String.format("%s = ?", escape(KeyColumn.KEY_ID_KEY));
                        updateItem(tapSchema.getName(), keyColumn, connection, whereCond, keyColumn.getKeyId());
                    }
                }

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

                for (Table table : operations.getTablesToUpdate()) {
                    String whereCond = String.format("%s = ?", escape(Table.TABLE_NAME_KEY));
                    updateItem(tapSchema.getName(), 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.getName(), 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, String tapSchemaVersion) throws SQLException {

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

        // Overriding data type from models
        if (tableModel != null) {
            for (Map.Entry<String, Map<String, Object>> entry : metadata.entrySet()) {
                String columnName = entry.getKey();
                String adqlType = tableModel.get(columnName).getType();
                String tapDataType = TypesMapping.getDataType(adqlType, dataTypeMode);
                entry.getValue().put(Column.DATATYPE_KEY, tapDataType);
            }
        }

        return metadata;
    }

    protected DataTypeMode getDataTypeMode() {
        return dataTypeMode;
    }
}