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.xmlmodel.PropertyModel;
import it.inaf.ia2.tsm.xmlmodel.TableModel;
import it.inaf.ia2.tsm.xmlmodel.TapSchemaModel;
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;

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

    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 abstract void createTable(String tapSchemaName, TableModel tableModel, Connection conn) throws SQLException;

    protected abstract void addPrimaryKey(String tapSchemaName, String tableName, String[] keyColumns, Connection conn) throws SQLException;

    protected abstract void addForeignKey(String tapSchemaName, String tableName, String[] fromKeyColumns, String targetTableName, String[] toKeyColumns, Connection conn) throws SQLException;

    private void addPrimaryKey(String tapSchemaName, String tableName, String keyColumn, Connection conn) throws SQLException {
        addPrimaryKey(tapSchemaName, tableName, new String[]{keyColumn}, conn);
    }

    private void addForeignKey(String tapSchemaName, String tableName, String fromKeyColumn, String targetTableName, String toKeyColumn, Connection conn) throws SQLException {
        addForeignKey(tapSchemaName, tableName, new String[]{fromKeyColumn}, targetTableName, new String[]{toKeyColumn}, conn);
    }

    protected abstract void createDatabase(String databaseName, Connection conn) throws SQLException;

    private void createTapSchemaStructure(String tapSchemaName, TapSchemaModel tapSchemaModel) throws SQLException {

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

            createDatabase(tapSchemaName, conn);

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

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

            // tables keys
            addPrimaryKey(tapSchemaName, TapSchema.TABLES_TABLE, Table.TABLE_NAME_KEY, conn);
            addForeignKey(tapSchemaName, TapSchema.TABLES_TABLE, Table.SCHEMA_NAME_KEY, TapSchema.SCHEMAS_TABLE, Schema.SCHEMA_NAME_KEY, conn);

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

            // keys keys
            addPrimaryKey(tapSchemaName, TapSchema.KEYS_TABLE, Key.ID_KEY, conn);
            addForeignKey(tapSchemaName, TapSchema.KEYS_TABLE, Key.FROM_TABLE_KEY, TapSchema.TABLES_TABLE, Table.TABLE_NAME_KEY, conn);
            addForeignKey(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);
            addForeignKey(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 save(TapSchema tapSchema) throws SQLException {

        LOG.debug("Saving TAP_SCHEMA");

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

        try {

            if (!tapSchema.exists()) {
                createTapSchemaStructure(tapSchema.getName(), tapSchema.getTapSchemaModel());
            }

            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);
                // TODO: INSERT COLUMNS!!!
            }

            //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 (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 (PropertyModel pm : tableModel.getProperties().values()) {
            if (!first) {
                querySb.append(", ");
            }
            first = false;
            querySb.append(escape(pm.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 (PropertyModel pm : tableModel.getProperties().values()) {
                        Object value = rs.getObject(pm.getName(), pm.getType());
                        item.put(pm.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()) {
                Class type = tapSchemaItem.getTableModel().get(key).getType();
                Object value = tapSchemaItem.getValue(key, type);
                statement.setObject(i, value, getSQLType(type));
                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 = 0;

            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(value.getClass()));
                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 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;
    }
}