Dao.java

package it.inaf.oats.ia2.tapschemamanager.datalayer;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.Persistence;
import javax.sql.DataSource;
import org.postgresql.ds.PGPoolingDataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Utility class that contains static methods for interacting with databases.
 * All the specific SQL clauses of the application are encapsulated in this
 * class. All the other datalayer operations are performed using the JPA.
 *
 * @author Sonia Zorba {@literal <zorba at oats.inaf.it>}
 */
public class Dao {

    private static final Logger log = LoggerFactory.getLogger(Dao.class);

    private static List<String> sortStringsList(List<String> list) {
        Collections.sort(list, String.CASE_INSENSITIVE_ORDER);
        return list;
    }

    /**
     * Retrieve the list of the names of the all the schemas contained into the
     * database specified by the <code>DataSource</code> parameter.
     *
     * @return list of schemas names alphabetically and case insensitively
     * ordered.
     */
    public static List<String> getAllSchemasNames(DataSource dataSource, DatabaseType dbType) throws SQLException {

        String query;
        if (dbType == DatabaseType.MYSQL) {
            query = "SHOW DATABASES";
        } else if (dbType == DatabaseType.POSTGRES) {
            query = "SELECT schema_name FROM information_schema.schemata";
        } else {
            throw new UnsupportedOperationException("Database type " + dbType + " not supported");
        }

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

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

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

        log.debug("{} schemas found", allSchemas.size());

        return sortStringsList(allSchemas);
    }

    public static List<String> getAllTAPSchemasNames(DBWrapper dbs) throws SQLException {
        List<String> allSchemas = getAllSchemasNames(dbs.getTapSchemaDataSource(), dbs.getTapSchemaDatabaseType());
        return getAllTAPSchemasNames(dbs, allSchemas);
    }

    /**
     * Retrieve the list of all TAP_SCHEMA schemas names contained in the
     * TAP_SCHEMA <code>DataSource</code>.<br>
     * TAP_SCHEMA schemas are selected simply checking if they contains the
     * TAP_SCHEMA standard tables. Currently no check on columns is performed.
     *
     * @param allSchemas usually the TAP_SCHEMA schemas list is loaded together
     * the list of all schemas, so this list is passed by parameter to avoid
     * repeating the query twice.
     *
     * @return list of all TAP_SCHEMA schemas names alphabetically and case
     * insensitively ordered.
     */
    public static List<String> getAllTAPSchemasNames(DBWrapper dbs, List<String> allSchemas) throws SQLException {

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

        for (String schemaName : allSchemas) {

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

            DatabaseType dbType = dbs.getTapSchemaDatabaseType();

            String query;
            if (dbType == DatabaseType.MYSQL) {
                query = "SHOW TABLES FROM `" + schemaName + "`";
            } else if (dbType == DatabaseType.POSTGRES) {
                query = "SELECT tablename FROM pg_catalog.pg_tables where schemaname = '" + schemaName + "'";
            } else {
                throw new UnsupportedOperationException("Database type " + dbType + " not supported");
            }

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

            try (Connection connection = dbs.getTapSchemaConnection();
                    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());

        return sortStringsList(allTAPSchemas);
    }

    /**
     * Retrieve the list of the name of the schemas exposed by the TAP_SCHEMA
     * specified by the <code>tapSchemaName</code> parameter.
     *
     * @return list of exposed schemas names alphabetically and case
     * insensitively ordered.
     */
    public static List<String> getExposedSchemas(DBWrapper dbs, String tapSchemaName) throws SQLException {

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

        DatabaseType dbType = dbs.getTapSchemaDatabaseType();

        String query;
        if (dbType == DatabaseType.MYSQL) {
            query = "SELECT schema_name FROM `" + tapSchemaName + "`.`schemas`";
        } else if (dbType == DatabaseType.POSTGRES) {
            query = "SELECT schema_name FROM \"" + tapSchemaName + "\".\"schemas\"";
        } else {
            throw new UnsupportedOperationException("Database type " + dbType + " not supported");
        }

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

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

        return exposedSchemas;
    }

    /**
     * Retrieve the JPA entity manager for given TAP_SCHEMA name and
     * credentials.
     */
    protected static EntityManager getEntityManager(Credentials tapSchemaCredentials, String tapSchemaName) {
        Map<String, String> persistenceMap = new HashMap<>();

        DatabaseType dbType = tapSchemaCredentials.getDatabaseType();

        String jdbcUrl;
        if (dbType == DatabaseType.MYSQL) {
            jdbcUrl = String.format("jdbc:mysql://%s:%s/%s", tapSchemaCredentials.getHostname(), tapSchemaCredentials.getPort(), tapSchemaName);
        } else if (dbType == DatabaseType.POSTGRES) {
            jdbcUrl = String.format("jdbc:postgresql://%s:%s/%s?currentSchema=%s", tapSchemaCredentials.getHostname(), tapSchemaCredentials.getPort(), tapSchemaCredentials.getDatabase(), tapSchemaName);
        } else {
            throw new UnsupportedOperationException("Database type " + dbType + " not supported");
        }

        persistenceMap.put("javax.persistence.jdbc.url", jdbcUrl);
        persistenceMap.put("javax.persistence.jdbc.user", tapSchemaCredentials.getUsername());
        persistenceMap.put("javax.persistence.jdbc.password", tapSchemaCredentials.getPassword());
        persistenceMap.put("javax.persistence.jdbc.driver", tapSchemaCredentials.getDatabaseDriverClass());

        EntityManagerFactory managerFactory = Persistence.createEntityManagerFactory("pu", persistenceMap);

        try {
            Class.forName(tapSchemaCredentials.getDatabaseDriverClass());
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
        EntityManager em = managerFactory.createEntityManager();

//        if (dbType == DatabaseType.POSTGRES) {
//            ((EntityManagerImpl) em).getSession().getLogin().setTableQualifier(tapSchemaName);
//        }
        return em;
    }

    /**
     * Create the TAP_SCHEMA schema into the database.
     */
    protected static void createTapSchemaSchema(DBWrapper dbs, String tapSchemaName) throws SQLException {

        DatabaseType dbType = dbs.getTapSchemaDatabaseType();

        try (Connection connection = dbs.getTapSchemaConnection();
                Statement statement = connection.createStatement()) {

            if (dbType == DatabaseType.MYSQL) {
                statement.executeUpdate("CREATE DATABASE IF NOT EXISTS `" + tapSchemaName + "`");
            } else if (dbType == DatabaseType.POSTGRES) {
                statement.executeUpdate("CREATE SCHEMA IF NOT EXISTS \"" + tapSchemaName + "\"");
            } else {
                throw new UnsupportedOperationException("Database type " + dbType + " not supported");
            }
        }
    }

    /**
     * Retrieve the list of the names of all the tables contained in a schema,
     * given their related <code>DataSource</code> and schema name.
     *
     * @return list of all tables names alphabetically and case insensitively
     * ordered.
     */
    protected static List<String> getAllTablesNames(DataSource dataSource, DatabaseType dbType, String schemaName) throws SQLException {

        String query;
        if (dbType == DatabaseType.MYSQL) {
            query = "SHOW TABLES FROM `" + schemaName + "`";
        } else if (dbType == DatabaseType.POSTGRES) {
            query = "SELECT tablename FROM pg_catalog.pg_tables where schemaname = '" + schemaName + "'";
        } else {
            throw new UnsupportedOperationException("Database type " + dbType + " not supported");
        }

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

        List<String> allTables = new ArrayList<>();
        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(query)) {
            while (resultSet.next()) {
                allTables.add(resultSet.getString(1));
            }
        }

        return sortStringsList(allTables);
    }

    /**
     * Retrieve the association between the tables names and their types
     * (<code>table</code> or <code>view</code>), given a
     * <code>DataSource</code> and a schema name.
     *
     * @return a map which has the tables names as keys and the table types as
     * values.
     */
    protected static Map<String, String> getTablesTypes(DataSource dataSource, DatabaseType dbType, String schemaName) throws SQLException {
        final Map<String, String> tablesTypes = new HashMap<>();

        String query;
        if (dbType == DatabaseType.MYSQL) {
            query = "SELECT table_name, table_type FROM information_schema.tables WHERE table_schema = '" + schemaName + "'";
        } else if (dbType == DatabaseType.POSTGRES) {
            query = "SELECT tablename AS table_name, 'table' AS table_type\n"
                    + "FROM pg_catalog.pg_tables WHERE schemaname = '" + schemaName + "'\n"
                    + "UNION\n"
                    + "SELECT table_name AS table_name, 'view' AS table_type\n"
                    + "FROM INFORMATION_SCHEMA.views\n"
                    + "WHERE table_schema = '" + schemaName + "'";
        } else {
            throw new UnsupportedOperationException("Database type " + dbType + " not supported");
        }

        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;
    }

    /**
     * Generate list of KeyEntity for a given schema, specifying its
     * <code>DataSource</code> and its name.<br>
     * <strong>IMPORTANT</strong>: this keys are without id. The id has to be
     * set when a table is added to a schema.
     */
    protected static List<KeyEntity> getSchemaKeys(DataSource dataSource, DatabaseType dbType, String schemaName) throws SQLException {

        if (dbType == DatabaseType.MYSQL) {

            Map<String, KeyEntity> schemaKeysMap = new HashMap<>();

            String query = "SELECT\n"
                    + "c.`CONSTRAINT_NAME` AS constraint_name,\n"
                    + "k.`TABLE_SCHEMA` AS from_schema,\n"
                    + "k.`TABLE_NAME` AS from_table,\n"
                    + "k.`COLUMN_NAME` AS from_column,\n"
                    + "k.`REFERENCED_TABLE_SCHEMA` AS target_schema,\n"
                    + "k.`REFERENCED_TABLE_NAME` AS target_table,\n"
                    + "k.`REFERENCED_COLUMN_NAME` AS target_column\n"
                    + "FROM information_schema.TABLE_CONSTRAINTS c \n"
                    + "LEFT JOIN information_schema.KEY_COLUMN_USAGE k \n"
                    + "ON c.`CONSTRAINT_NAME` = k.`CONSTRAINT_NAME` AND c.`TABLE_SCHEMA` = k.`TABLE_SCHEMA`\n"
                    + "WHERE c.`CONSTRAINT_TYPE` = 'FOREIGN KEY' \n"
                    + "AND k.`TABLE_SCHEMA` = '" + schemaName + "' OR k.`REFERENCED_TABLE_SCHEMA` = '" + schemaName + "'";

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

                while (resultSet.next()) {
                    String constraintName = resultSet.getString("constraint_name");

                    KeyEntity key = schemaKeysMap.get(constraintName);
                    if (key == null) {
                        key = new KeyEntity(
                                resultSet.getString("from_schema"),
                                resultSet.getString("from_table"),
                                resultSet.getString("target_schema"),
                                resultSet.getString("target_table")
                        );
                        schemaKeysMap.put(constraintName, key);
                    }

                    KeyColumnEntity keyColumnEntity = new KeyColumnEntity();
                    keyColumnEntity.setFromColumn(resultSet.getString("from_column"));
                    keyColumnEntity.setTargetColumn(resultSet.getString("target_column"));
                }
            }

            return new ArrayList<>(schemaKeysMap.values());

        } else if (dbType == DatabaseType.POSTGRES) {

            String databaseName = ((PGPoolingDataSource) dataSource).getDatabaseName();

            List<KeyEntity> schemaKeys = new ArrayList<>();

            String queryKeys = "SELECT\n"
                    + "conname AS constraint_name,\n"
                    + "conrelid::regclass AS from_table, \n"
                    + "confrelid::regclass AS target_table\n"
                    + "FROM pg_catalog.pg_constraint\n"
                    + "WHERE contype = 'f'\n"
                    + "AND ((conrelid::regclass || '' LIKE '" + schemaName + ".%')\n"
                    + "OR (confrelid::regclass || '' LIKE '" + schemaName + ".%'))";

            try (Connection connection = dataSource.getConnection();
                    Statement statement = connection.createStatement();
                    ResultSet resultSet = statement.executeQuery(queryKeys)) {

                log.debug("Executing query {}", queryKeys);

                while (resultSet.next()) {

                    String constraintName = resultSet.getString("constraint_name");

                    String[] fromTableFullNameSplitted = resultSet.getString("from_table").split(Pattern.quote("."));
                    String fromSchema = fromTableFullNameSplitted[0];
                    String fromTable = fromTableFullNameSplitted[1];

                    String[] targetTableFullNameSplitted = resultSet.getString("target_table").split(Pattern.quote("."));
                    String targetSchema = targetTableFullNameSplitted[0];
                    String targetTable = targetTableFullNameSplitted[1];

                    KeyEntity key = new KeyEntity(fromSchema, fromTable, targetSchema, targetTable);
                    schemaKeys.add(key);

                    // conkey conrelid
                    String queryFromKC = "SELECT\n"
                            + "c.column_name AS key_column\n"
                            + "FROM information_schema.columns c\n"
                            + "JOIN pg_catalog.pg_constraint r ON c.ordinal_position = ANY(r.conkey)\n"
                            + "AND (c.table_schema || '.' || c.table_name) = (r.conrelid::regclass || '')\n"
                            + "WHERE r.conname = '" + constraintName + "' AND r.contype = 'f'\n"
                            + "AND c.table_schema = '" + fromSchema + "'\n"
                            + "AND table_catalog = '" + databaseName + "'";

                    // as above, but with confkey and confrelid and different c.table_schema where condition
                    String queryTargetKC = "SELECT\n"
                            + "c.column_name AS key_column\n"
                            + "FROM information_schema.columns c\n"
                            + "JOIN pg_catalog.pg_constraint r ON c.ordinal_position = ANY(r.confkey)\n"
                            + "AND (c.table_schema || '.' || c.table_name) = (r.confrelid::regclass || '')\n"
                            + "WHERE r.conname = '" + constraintName + "' AND r.contype = 'f'\n"
                            + "AND c.table_schema = '" + targetSchema + "'\n"
                            + "AND table_catalog = '" + databaseName + "'";

                    try (Statement statFromKC = connection.createStatement();
                            Statement statTargetKC = connection.createStatement()) {

                        try (ResultSet rsFromKC = statFromKC.executeQuery(queryFromKC);
                                ResultSet rsTargetKC = statTargetKC.executeQuery(queryTargetKC)) {

                            log.debug("Executing query {}", queryFromKC);
                            log.debug("Executing query {}", queryTargetKC);

                            while (rsFromKC.next()) {
                                if (rsTargetKC.next()) {
                                    KeyColumnEntity keyColumn = new KeyColumnEntity();
                                    keyColumn.setFromColumn(rsFromKC.getString("key_column"));
                                    keyColumn.setTargetColumn(rsTargetKC.getString("key_column"));
                                    key.getKeyColumns().add(keyColumn);
                                }
                            }
                        }
                    }
                }
            }

            return schemaKeys;
        } else {
            throw new UnsupportedOperationException("Database type " + dbType + " not supported");
        }
    }

    private static boolean equalsOneOf(String string, String... values) {
        for (String value : values) {
            if (string.equals(value)) {
                return true;
            }
        }
        return false;
    }

    /**
     * A list of <strong>ALL</strong> the possible {@link ColumnEntity} for a
     * given <code>DataSource</code> and a given {@link TableEntity}.<br>
     * <strong>IMPORTANT</strong>: this list includes also the
     * <code>ColumnEntity</code> instances that aren't exposed by the
     * TAP_SCHEMA. This choice was done for performance reasons, to avoid
     * repeating the query for adding new columns. A <code>ColumnEntity</code>
     * will be persisted to the TAP_SCHEMA if it was added to its
     * <code>TableEntity</code> using the methods
     * {@link TableEntity.addColumn()}.
     */
    protected static List<ColumnEntity> getAllColumns(DataSource dataSource, DatabaseType dbType, TableEntity table) throws SQLException {
        final List<ColumnEntity> allColumns = new ArrayList<>();

        String query;
        if (dbType == DatabaseType.MYSQL) {
            query = String.format("SHOW COLUMNS FROM `%s`.`%s`", table.getSchemaName(), table.getShortTableName());
        } else if (dbType == DatabaseType.POSTGRES) {
            query = "SELECT c.column_name, c.data_type, r.contype AS column_type, c.character_maximum_length, c.numeric_precision\n" //, c.numeric_precision_radix
                    + "FROM information_schema.columns c\n"
                    + "JOIN pg_catalog.pg_tables t ON c.table_schema = t.schemaname AND c.table_name = t.tablename\n"
                    + "LEFT JOIN pg_catalog.pg_constraint r ON c.ordinal_position = ANY(r.conkey) AND r.conrelid = (t.schemaname || '.' || t.tablename)::regclass::oid\n"
                    + "WHERE t.schemaname = '" + table.getSchemaName() + "' AND t.tablename = '" + table.getShortTableName() + "'";
        } else {
            throw new UnsupportedOperationException("Database type " + dbType + " not supported");
        }

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

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

            while (resultSet.next()) {

                String columnName;
                if (dbType == DatabaseType.MYSQL) {
                    columnName = resultSet.getString("Field");
                } else if (dbType == DatabaseType.POSTGRES) {
                    columnName = resultSet.getString("column_name");
                } else {
                    throw new UnsupportedOperationException("Database type " + dbType + " not supported");
                }

                ColumnEntity columnEntity = DLUtil.getEntityByName(table.getColumns(), columnName);
                boolean alreadyLoaded = columnEntity != null;

                if (columnEntity == null) {
                    columnEntity = new ColumnEntity(columnName);
                }

                columnEntity.setFullTableName(table.getName());
                columnEntity.setTable(table);

                // Key type
                if (dbType == DatabaseType.MYSQL) {
                    String key = resultSet.getString("Key");
                    columnEntity.setPrimaryKey(key.equals("PRI"));
                    columnEntity.setIndexed(equalsOneOf(key, "PRI", "UNI", "MUL") ? 1 : 0);
                    if (!alreadyLoaded) {
                        columnEntity.setPrincipal(key.equals("PRI") ? 1 : 0);
                    }
                } else if (dbType == DatabaseType.POSTGRES) {
                    String columnType = resultSet.getString("column_type");
                    if (columnType != null) {
                        columnEntity.setPrimaryKey("p".equals(columnType));
                        columnEntity.setIndexed(equalsOneOf(columnType, "p", "f", "u") ? 1 : 0);
                    }
                    if (!alreadyLoaded) {
                        columnEntity.setPrincipal("p".equals(columnType) ? 1 : 0);
                    }
                } else {
                    throw new UnsupportedOperationException("Database type " + dbType + " not supported");
                }

                // Datatype and Size
                int size = 0;
                String datatype;

                if (dbType == DatabaseType.MYSQL) {
                    String type = resultSet.getString("Type").toLowerCase();

                    if (type.startsWith("int")) {
                        datatype = "adql:INTEGER";
                    } else if (type.startsWith("smallint")) {
                        datatype = "adql:SMALLINT";
                    } else if (type.startsWith("bigint")) {
                        datatype = "adql:BIGINT";
                    } else if (type.startsWith("float")) {
                        datatype = "adql:REAL";
                    } else if (type.startsWith("char")) {
                        int beginIndex = type.indexOf('(');
                        int endIndex = type.indexOf(')');
                        size = Integer.parseInt(type.substring(beginIndex + 1, endIndex));
                        datatype = "adql:CHAR";
                    } else if (type.startsWith("varchar")) {
                        int beginIndex = type.indexOf('(');
                        int endIndex = type.indexOf(')');
                        size = Integer.parseInt(type.substring(beginIndex + 1, endIndex));
                        datatype = "adql:VARCHAR";
                    } else if (type.contains("timestamp")) {
                        datatype = "adql:TIMESTAMP";
                    } else {
                        datatype = "adql:" + type.toUpperCase();
                    }
                } else if (dbType == DatabaseType.POSTGRES) {
                    String type = resultSet.getString("data_type");

                    if (type.startsWith("int")) {
                        datatype = "adql:INTEGER";
                    } else if (type.startsWith("smallint")) {
                        datatype = "adql:SMALLINT";
                    } else if (type.startsWith("bigint")) {
                        datatype = "adql:BIGINT";
                    } else if (type.startsWith("double") || type.startsWith("real")) {
                        datatype = "adql:REAL";
                    } else if (type.startsWith("character varying")) {
                        datatype = "adql:VARCHAR";
                        size = resultSet.getInt("character_maximum_length");
                    } else if (type.startsWith("char")) {
                        datatype = "adql:CHAR";
                        size = resultSet.getInt("character_maximum_length");
                    } else if (type.contains("timestamp")) {
                        datatype = "adql:TIMESTAMP";
                    } else {
                        datatype = "adql:" + type.toUpperCase();
                    }
                } else {
                    throw new UnsupportedOperationException("Database type " + dbType + " not supported");
                }

                columnEntity.setDatatype(datatype);
                columnEntity.setSize(size);

                if (!alreadyLoaded) {
                    columnEntity.setStd(0);
                }

                allColumns.add(columnEntity);
            }
        }

        return allColumns;
    }
}