TapSchema.java

/* 
 * _____________________________________________________________________________
 * 
 * INAF - OATS National Institute for Astrophysics - Astronomical Observatory of
 * Trieste INAF - IA2 Italian Center for Astronomical Archives
 * _____________________________________________________________________________
 * 
 * Copyright (C) 2016 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;

import it.inaf.ia2.tsm.datalayer.DBBroker;
import it.inaf.ia2.tsm.datalayer.DBBrokerFactory;
import it.inaf.ia2.tsm.datalayer.DBWrapper;
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 java.io.Serializable;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Sonia Zorba {@literal <zorba at oats.inaf.it>}
 */
public class TapSchema implements EntitiesContainer<Schema>, Serializable {

    // Mandatory tables constants
    public static final String TABLES_TABLE = "tables";
    public static final String SCHEMAS_TABLE = "schemas";
    public static final String COLUMNS_TABLE = "columns";
    public static final String KEYS_TABLE = "keys";
    public static final String KEY_COLUMNS_TABLE = "key_columns";

    public static final String DESCRIPTION_KEY = "description";

    private static final long serialVersionUID = 1678083091602571256L;

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

    private final Map<String, Schema> schemas;
    private final Set<Key> allKeys;
    private final ConsistencyChecks consistencyChecks;

    private boolean loading;
    private DBWrapper dbWrapper;
    private boolean exists;
    private String tapSchemaVersion;
    private String tapSchemaName;
    private boolean obscore;
    private String obscoreVersion;

    private transient DBBroker sourceDBBroker;
    private transient DBBroker tapSchemaDBBroker;

    public final DBBroker getSourceDBBroker() {
        if (sourceDBBroker == null) {
            sourceDBBroker = DBBrokerFactory.getDBBroker(dbWrapper.getSourceDataSourceWrapper(), tapSchemaVersion);
        }
        return sourceDBBroker;
    }

    public final DBBroker getTapSchemaDBBroker() {
        if (tapSchemaDBBroker == null) {
            tapSchemaDBBroker = DBBrokerFactory.getDBBroker(dbWrapper.getTapSchemaDataSourceWrapper(), tapSchemaVersion);
        }
        return tapSchemaDBBroker;
    }

    private TapSchema() {
        // for serialization
        schemas = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
        allKeys = new HashSet<>();
        consistencyChecks = new ConsistencyChecks();
    }

    public TapSchema(DBWrapper dbWrapper, TapSchemaSettings settings, boolean exists) throws SQLException {
        this();

        loading = true;
        this.dbWrapper = dbWrapper;
        this.exists = exists;
        this.tapSchemaVersion = settings.getTapSchemaVersion();
        this.tapSchemaName = settings.getTapSchemaName();
        this.obscore = settings.isHasObscore();
        this.obscoreVersion = settings.getObscoreVersion();

        // Initializing schemas map
        for (String schemaName : getSourceDBBroker().getAllSchemaNames()) {
            schemas.put(schemaName, null);
        }
        schemas.put(tapSchemaName, null); // the TAP_SCHEMA contains itself

        if (exists) {
            loadSavedTapSchema();
        }

        loading = false;
        checkKeys();
    }

    private void loadSavedProperties(TapSchemaEntity tapSchemaEntity, Map<String, Object> savedProperties) {
        for (Map.Entry<String, Object> entry : savedProperties.entrySet()) {
            String key = entry.getKey();
            Object savedValue = entry.getValue();
            Object currentValue = entry.getValue();
            EntityProperty ep = tapSchemaEntity.getProperty(key);

            if (!ep.isUpdatable() && !Objects.equals(savedValue, currentValue)) {

                InconsistentValue inconsistentValue = new InconsistentValue(
                        TSMUtil.getNaturalLangueName(tapSchemaEntity),
                        TSMUtil.getName(tapSchemaEntity),
                        key,
                        savedValue,
                        savedValue
                );

                consistencyChecks.addInconsistency(inconsistentValue);
            }

            tapSchemaEntity.initProperty(key, savedValue);
        }
    }

    /**
     * Loads saved TAP_SCHEMA data and performs consistency checking.
     *
     * @throws SQLException
     */
    private void loadSavedTapSchema() throws SQLException {
        DBBroker broker = getTapSchemaDBBroker();

        // Schemata
        for (Map<String, Object> schemaProps : broker.getSavedItems(getName(), getTableModel(SCHEMAS_TABLE))) {
            String schemaName = (String) schemaProps.get(Schema.SCHEMA_NAME_KEY);
            Schema schema = addChild(schemaName);
            if (schema == null) {
                consistencyChecks.addUnexistingSchema(schemaName);
            } else {
                loadSavedProperties(schema, schemaProps);
                schema.setStatus(Status.ADDED_PERSISTED);
            }
        }

        // Tables
        for (Map<String, Object> tableProps : broker.getSavedItems(getName(), getTableModel(TABLES_TABLE))) {
            String tableCompleteName = (String) tableProps.get(Table.TABLE_NAME_KEY);
            String[] tableNameSplit = tableCompleteName.split(Pattern.quote("."));
            String schemaName = tableNameSplit[0];
            String tableName = tableNameSplit[1];
            Schema schema = getChild(schemaName, Status.ADDED_PERSISTED);
            if (schema == null) {
                consistencyChecks.addUnexistingSchema(schemaName);
            } else {
                Table table = schema.addChild(tableName);
                if (table == null) {
                    consistencyChecks.addUnexistingTable(schemaName, tableName);
                } else {
                    loadSavedProperties(table, tableProps);
                    table.setStatus(Status.ADDED_PERSISTED);
                }
            }
        }

        // Columns
        for (Map<String, Object> columnProps : broker.getSavedItems(getName(), getTableModel(COLUMNS_TABLE))) {
            String tableCompleteName = (String) columnProps.get(Column.TABLE_NAME_KEY);
            String[] tableNameSplit = tableCompleteName.split(Pattern.quote("."));
            String schemaName = tableNameSplit[0];
            String tableName = tableNameSplit[1];
            String columnName = (String) columnProps.get(Column.COLUMN_NAME_KEY);
            Schema schema = getChild(schemaName, Status.ADDED_PERSISTED);
            if (schema == null) {
                consistencyChecks.addUnexistingSchema(schemaName);
            } else {
                Table table = schema.getChild(tableName, Status.ADDED_PERSISTED);
                if (table == null) {
                    consistencyChecks.addUnexistingTable(schemaName, tableName);
                } else {
                    Column column = table.addChild(columnName);
                    if (column == null) {
                        consistencyChecks.addUnexistingColumn(tableCompleteName, columnName);
                    } else {
                        loadSavedProperties(column, columnProps);
                        column.setStatus(Status.ADDED_PERSISTED);
                    }
                }
            }
        }

        // Keys
        List<Map<String, Object>> keysProps = broker.getSavedItems(getName(), getTableModel(KEYS_TABLE));
        List<Map<String, Object>> keysColumnsProps = broker.getSavedItems(getName(), getTableModel(KEY_COLUMNS_TABLE));
        for (Map<String, Object> keyProp : keysProps) {

            String fromTable = (String) keyProp.get(Key.FROM_TABLE_KEY);
            String targetTable = (String) keyProp.get(Key.TARGET_TABLE_KEY);
            String keyId = (String) keyProp.get(Key.ID_KEY);
            assert keyId != null;

            List<Map<String, Object>> kcPropsById = new ArrayList<>();
            for (Map<String, Object> kcp : keysColumnsProps) {
                String keyColumnId = (String) kcp.get(KeyColumn.KEY_ID_KEY);
                assert keyColumnId != null;
                if (keyColumnId.equals(keyId)) {
                    kcPropsById.add(kcp);
                }
            }

            // Searching the key
            boolean keyFound = false;
            for (Key key : allKeys) {
                if (key.getFromTableCompleteName().equals(fromTable) && key.getTargetTableCompleteName().equals(targetTable)) {
                    // Search the key columns having proper key id

                    // Verifying the matching
                    List<KeyColumn> matchedKeyColumns = new ArrayList<>();
                    if (kcPropsById.size() == key.getKeyColumns().size()) {
                        for (Map<String, Object> kcp : kcPropsById) {
                            String fromColumn = (String) kcp.get(KeyColumn.FROM_COLUMN_KEY);
                            String targetColumn = (String) kcp.get(KeyColumn.TARGET_COLUMN_KEY);
                            for (KeyColumn keyColumn : key.getKeyColumns()) {
                                if (keyColumn.getFromColumn().equals(fromColumn)
                                        && keyColumn.getTargetColumn().equals(targetColumn)) {
                                    matchedKeyColumns.add(keyColumn);
                                }
                            }
                        }
                    }
                    if (kcPropsById.size() == matchedKeyColumns.size()) {
                        keyFound = true;
                        int index = 0;
                        loadSavedProperties(key, keyProp);
                        for (Map<String, Object> kcp : kcPropsById) {
                            KeyColumn kc = matchedKeyColumns.get(index);
                            loadSavedProperties(kc, kcp);
                            index++;
                        }
                    }
                }
            }

            if (!keyFound) {
                boolean setKeyToRemove = true;
                if (setKeyToRemove) {
                    String[] fromColumns = new String[kcPropsById.size()];
                    String[] targetColumns = new String[kcPropsById.size()];
                    int i = 0;
                    for (Map<String, Object> kcp : kcPropsById) {
                        fromColumns[i] = (String) kcp.get(KeyColumn.FROM_COLUMN_KEY);
                        targetColumns[i] = (String) kcp.get(KeyColumn.TARGET_COLUMN_KEY);
                        i++;
                    }
                    consistencyChecks.addUnexistingKey(keyId, fromTable, fromColumns, targetTable, targetColumns);
                }
            }
        }
    }

    public DBBroker getDBBroker(String schemaName) {
        if (schemaName.equals(tapSchemaName)) {
            return getTapSchemaDBBroker();
        } else {
            return getSourceDBBroker();
        }
    }

    /**
     * The name of the TAP_SCHEMA schema.
     */
    public final String getName() {
        return tapSchemaName;
    }

    /**
     * The version selected for this TAP_SCHEMA.
     */
    public String getVersion() {
        return tapSchemaVersion;
    }

    private void loadSchemaKeysMetadata(String schemaName) throws SQLException {
        allKeys.addAll(getDBBroker(schemaName).getKeys(this, schemaName));
    }

    public Set<Key> getAllKeys() {
        return allKeys;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final Schema addChild(String schemaName) throws SQLException {
        LOG.debug("Adding schema {}", schemaName);

        Schema schema;

        if (!schemas.containsKey(schemaName)) {
            schema = null;
        } else {

            schema = schemas.get(schemaName);

            if (schema == null) {
                schema = new Schema(this, schemaName);
                schema.setStatus(Status.ADDED_NOT_PERSISTED);
                schemas.put(schemaName, schema);
                loadSchemaKeysMetadata(schemaName);
            } else {
                switch (schema.getStatus()) {
                    case TO_REMOVE:
                        schema.setStatus(Status.ADDED_PERSISTED);
                        break;
                    case REMOVED_NOT_PERSISTED:
                        schema.setStatus(Status.ADDED_NOT_PERSISTED);
                        break;
                    default:
                        throw new IllegalArgumentException("Cannot add the schema " + schemaName + ". Invalid status. Schema status is " + schema.getStatus());
                }
            }

            checkKeys();
        }

        return schema;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void removeChild(String schemaName) {
        LOG.debug("Removing schema {}", schemaName);

        if (!schemas.containsKey(schemaName)) {
            throw new IllegalArgumentException("The database doesn't contains a schema named " + schemaName);
        }

        Schema schema = schemas.get(schemaName);
        if (schema == null || schema.getStatus() == Status.LOADED) {
            throw new IllegalArgumentException("Cannot remove the schema " + schemaName + ". It has never been added.");
        }

        if (schema.getStatus() == Status.ADDED_NOT_PERSISTED) {
            schema.setStatus(Status.REMOVED_NOT_PERSISTED);
        } else if (schema.getStatus() == Status.ADDED_PERSISTED) {
            schema.setStatus(Status.TO_REMOVE);
        }

        checkKeys();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final Schema getChild(String childName, Status... statuses) {
        return TSMUtil.getChild(schemas, childName, statuses);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public List<Schema> getChildren(Status... statuses) {
        return TSMUtil.getChildrenByStatus(schemas.values(), statuses);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public List<String> getAddableChildrenNames() {
        return TSMUtil.getAddableChildrenNames(schemas);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public List<Schema> getAddedChildren() {
        return getChildren(Status.ADDED_PERSISTED, Status.ADDED_NOT_PERSISTED);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public List<Schema> getAddedOrRemovedChildren() {
        return getChildren(Status.ADDED_PERSISTED, Status.ADDED_NOT_PERSISTED, Status.TO_REMOVE, Status.REMOVED_NOT_PERSISTED);
    }

    /**
     * This method has to be used after TAP_SCHEMA modifications are committed
     * to the database, in order to remove from the memory the schemas with
     * status {@code Status.TO_REMOVE} or {@code Status.REMOVED_NOT_PERSISTED}.
     */
    public void cleanSchema(String schemaName) {
        if (!schemas.containsKey(schemaName)) {
            throw new IllegalArgumentException("The TAP_SCHEMA doesn't contain the schema " + schemaName);
        }
        schemas.put(schemaName, null);
    }

    public SchemaModel getIvoaSchemaModel() {
        if (obscore) {
            return SchemaModels.getIvoaSchemaModel(obscoreVersion);
        }
        return null;
    }

    private void addEntireSchema(String schemaName) throws SQLException {
        Schema schema = addChild(schemaName);
        for (String tableName : schema.getAddableChildrenNames()) {
            Table table = schema.addChild(tableName);
            for (String columnName : table.getAddableChildrenNames()) {
                table.addChild(columnName);
            }
        }
    }

    /**
     * Save or update the TAP_SCHEMA changes into the database.
     */
    public void save() throws SQLException {

        DBBroker broker = getTapSchemaDBBroker();

        if (!exists) {
            SchemaModel tapSchemaModel = getTapSchemaModel();
            broker.createTapSchemaStructure(tapSchemaName, tapSchemaModel);

            // Adding TAP_SCHEMA into TAP_SCHEMA
            addEntireSchema(tapSchemaName);
            fillColumnProperties(tapSchemaModel, tapSchemaName);

            if (obscore) {
                SchemaModel ivoaSchemaModel = getIvoaSchemaModel();

                // ivoa schema has to be created into source database
                getSourceDBBroker().createIvoaSchemaStructure(ivoaSchemaModel);
                // Initializing ivoa schema slot in schemata maps
                schemas.put(ivoaSchemaModel.getName(), null);

                // Add ivoa schema into TAP_SCHEMA
                addEntireSchema(ivoaSchemaModel.getName());
                fillColumnDescriptionsAndStd(ivoaSchemaModel);
            }
        }

        fillKeyIds();

        broker.save(this);
        exists = true;

        consistencyChecks.getInconsistencies().clear();
    }

    /**
     * Retrieve the maximum key id from all the schemas that are added into the
     * TAP_SCHEMA.
     *
     * @return the maximum key, if it exists, zero otherwise.
     */
    private int getMaxKeyId() {
        int maxKeyId = 0;
        for (Key key : allKeys) {
            if (key.getId() != null) {
                int keyId = Integer.parseInt(key.getId());
                if (keyId > maxKeyId) {
                    maxKeyId = keyId;
                }
            }
        }
        return maxKeyId;
    }

    private void fillKeyIds() {

        List<Key> newKeys = new ArrayList<>();
        for (Key key : allKeys) {
            if (key.isVisible()) {
                newKeys.add(key);
            }
        }

        int maxKeyId = getMaxKeyId();
        for (Key newKey : newKeys) {
            maxKeyId++;
            newKey.setId(maxKeyId + "");
        }
    }

    /**
     * Set keys visibility based on other entities visibility (a key is visible
     * if all schemas, tables and columns involved have
     * {@link Status} {@code ADDED_PERSISTED} or {@code ADDED_NOT_PERSISTED}).
     */
    public final void checkKeys() {
        if (!loading) {

            for (Key key : allKeys) {

                // Check if key should be exposed in TAP_SCHEMA
                boolean keyVisible = true;

                for (KeyColumn keyColumn : key.getKeyColumns()) {
                    String schemaName = keyColumn.getParent().getFromSchemaName();
                    String tableName = keyColumn.getParent().getFromTableSimpleName();
                    String columnName = keyColumn.getFromColumn();
                    if (!isColumnVisible(schemaName, tableName, columnName)) {
                        keyVisible = false;
                        break;
                    }

                    schemaName = keyColumn.getParent().getTargetSchemaName();
                    tableName = keyColumn.getParent().getTargetTableSimpleName();
                    columnName = keyColumn.getTargetColumn();
                    if (!isColumnVisible(schemaName, tableName, columnName)) {
                        keyVisible = false;
                        break;
                    }
                }

                // TODO: use status instead of set visibile [?]
                key.setVisible(keyVisible);
            }
        }
    }

    /**
     * Print all TAP_SCHEMA tree (useful for debugging).
     */
    @Override
    public String toString() {

        StringBuilder sb = new StringBuilder("\n");

        sb.append(String.format(">> TAP_SCHEMA %s <<\n", tapSchemaName));

        for (Schema schema : getChildren()) {
            sb.append("--");
            sb.append(schema.getName());
            sb.append(String.format(" [%s]", schema.getStatus()));
            sb.append("\n");
            List<Table> tables = schema.getChildren();
            for (int i = 0; i < tables.size(); i++) {
                Table table = tables.get(i);
                sb.append("   |--");
                sb.append(table.getName());
                sb.append(String.format(" [%s]", table.getStatus()));
                sb.append("\n");

                String padder = i < tables.size() - 1 ? "   |   " : "       ";

                for (Column column : table.getChildren()) {
                    sb.append(padder);
                    sb.append("|--");
                    sb.append(column.getName());
                    sb.append(String.format(" [%s]", column.getStatus()));
                    sb.append("\n");
                }

                sb.append("\n");
            }
        }

        sb.append("** Keys **\n");
        for (Key key : getVisibileKeys()) {
            sb.append(key);
            sb.append("\n");
        }

        return sb.toString();
    }

    public boolean isSchemaVisible(String schemaName) {
        Schema schema = schemas.get(schemaName);
        if (schema == null) {
            return false;
        }
        return schema.getStatus() == Status.ADDED_PERSISTED
                || schema.getStatus() == Status.ADDED_NOT_PERSISTED;
    }

    public boolean isTableVisible(String schemaName, String tableName) {
        if (!isSchemaVisible(schemaName)) {
            return false;
        }

        Table table = schemas.get(schemaName).getChild(tableName);
        if (table == null) {
            return false;
        }

        if (table.getStatus() == Status.ADDED_PERSISTED
                || table.getStatus() == Status.ADDED_NOT_PERSISTED) {
            return isSchemaVisible(schemaName);
        }

        return false;
    }

    public boolean isColumnVisible(String schemaName, String tableName, String columnName) {
        if (!isTableVisible(schemaName, tableName)) {
            return false;
        }

        Column column = schemas.get(schemaName).getChild(tableName).getChild(columnName);
        if (column == null) {
            return false;
        }

        if (column.getStatus() == Status.ADDED_PERSISTED
                || column.getStatus() == Status.ADDED_NOT_PERSISTED) {
            return isTableVisible(schemaName, tableName);
        }

        return false;
    }

    /**
     * Define if the TAP_SCHEMA schema was already written into the database.
     */
    public boolean exists() {
        return exists;
    }

    public ConsistencyChecks getConsistencyChecks() {
        return consistencyChecks;
    }

    public SchemaModel getTapSchemaModel() {
        return SchemaModels.getTapSchemaModel(tapSchemaVersion);
    }

    public final TableModel getTableModel(String tableName) {
        return getTapSchemaModel().getTable(tableName);
    }

    public Map<String, Object> getSchemaMetadata(String schemaName) {
        Map<String, Object> metadata = new HashMap<>();
        metadata.put(Schema.SCHEMA_NAME_KEY, schemaName);
        return metadata;
    }

    public List<Key> getVisibileKeys() {
        List<Key> visibleKeys = new ArrayList<>();
        for (Key key : allKeys) {
            if (key.isVisible()) {
                visibleKeys.add(key);
            }
        }
        return visibleKeys;
    }

    private Integer getIntAsBool(Boolean value) {
        if (value == null) {
            return null;
        }
        return value ? 1 : 0;
    }

    public boolean isHasObscore() {
        return obscore;
    }
    
    /**
     * Fill descriptions of the TAP_SCHEMA schema entities for a given
     * SchemaModel (TAP_SCHEMA or ivoa).
     */
    private void fillColumnProperties(SchemaModel schemaModel, String schemaName) {

        // check only on std, but valid also for principal (it depends on TS version)
        boolean useIntegerAsBool = getTapSchemaModel().getTable(COLUMNS_TABLE).get(Column.STD_KEY).getJavaType() == Integer.class;

        Schema schema = getChild(schemaName);
        schema.setValue(DESCRIPTION_KEY, schemaModel.getDescription());
        for (TableModel tableModel : schemaModel.getTables()) {
            Table table = schema.getChild(tableModel.getName());
            table.setValue(DESCRIPTION_KEY, tableModel.getDescription());
            for (ColumnModel propertyModel : tableModel.getColumns()) {
                Column column = table.getChild(propertyModel.getName());
                column.setValue(DESCRIPTION_KEY, propertyModel.getDescription());
                column.setValue(Column.UCD_KEY, propertyModel.getUcd());
                column.setValue(Column.UNIT_KEY, propertyModel.getUnit());
                column.setValue(Column.UTYPE_KEY, propertyModel.getUtype());

                Object compatibleStd = useIntegerAsBool ? getIntAsBool(propertyModel.isStandard()) : propertyModel.isStandard();
                Object compatiblePrincipal = useIntegerAsBool ? getIntAsBool(propertyModel.isPrincipal()) : propertyModel.isPrincipal();
                column.setValue(Column.STD_KEY, compatibleStd);
                column.setValue(Column.PRINCIPAL_KEY, compatiblePrincipal);
            }
        }
    }

    private void fillColumnDescriptionsAndStd(SchemaModel schemaModel) {
        fillColumnProperties(schemaModel, schemaModel.getName());
    }
}