/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.openjpa.jdbc.meta;

import java.io.Serializable;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

import org.apache.openjpa.jdbc.conf.JDBCConfiguration;
import org.apache.openjpa.jdbc.identifier.DBIdentifier;
import org.apache.openjpa.jdbc.identifier.DBIdentifier.DBIdentifierType;
import org.apache.openjpa.jdbc.identifier.Normalizer;
import org.apache.openjpa.jdbc.identifier.QualifiedDBIdentifier;
import org.apache.openjpa.jdbc.schema.Column;
import org.apache.openjpa.jdbc.schema.ColumnIO;
import org.apache.openjpa.jdbc.schema.ForeignKey;
import org.apache.openjpa.jdbc.schema.Index;
import org.apache.openjpa.jdbc.schema.PrimaryKey;
import org.apache.openjpa.jdbc.schema.Schema;
import org.apache.openjpa.jdbc.schema.SchemaGroup;
import org.apache.openjpa.jdbc.schema.Schemas;
import org.apache.openjpa.jdbc.schema.Table;
import org.apache.openjpa.jdbc.schema.Unique;
import org.apache.openjpa.jdbc.sql.DBDictionary;
import org.apache.openjpa.lib.log.Log;
import org.apache.openjpa.lib.util.ClassUtil;
import org.apache.openjpa.lib.util.Localizer;
import org.apache.openjpa.lib.util.Localizer.Message;
import org.apache.openjpa.lib.util.StringUtil;
import org.apache.openjpa.meta.JavaTypes;
import org.apache.openjpa.meta.MetaDataContext;
import org.apache.openjpa.util.MetaDataException;

/**
 * Base class storing raw mapping information; defines utility methods for
 * converting raw mapping information to full mapping to the schema.
 *
 * @author Abe White
 */
public abstract class MappingInfo implements Serializable {
    private static final long serialVersionUID = 1L;
    public static final int JOIN_NONE = 0;
    public static final int JOIN_FORWARD = 1;
    public static final int JOIN_INVERSE = 2;

    private static final Object NULL = new Object();

    private static final Localizer _loc = Localizer.forPackage
        (MappingInfo.class);

    private String _strategy = null;
    private List<Column> _cols = null;
    private Index _idx = null;
    private Unique _unq = null;
    private ForeignKey _fk = null;
    private boolean _canIdx = true;
    private boolean _canUnq = true;
    private boolean _canFK = true;
    private boolean _implicitRelation = false;
    private int _join = JOIN_NONE;
    private ColumnIO _io = null;

    /**
     * Mapping strategy name.
     */
    public String getStrategy() {
        return _strategy;
    }

    /**
     * Mapping strategy name.
     */
    public void setStrategy(String strategy) {
        _strategy = strategy;
    }

    /**
     * Raw column data.
     */
    public List<Column> getColumns() {
        if (_cols == null) {
            return Collections.emptyList();
        }
        return _cols;
    }

    /**
     * Gets the columns whose table name matches the given table name.
     * @deprecated
     */
    @Deprecated
    public List<Column> getColumns(String tableName) {
        return getColumns(DBIdentifier.newTable(tableName));
    }

    /**
     * Gets the columns whose table name matches the given table name.
     */
    public List<Column> getColumns(DBIdentifier tableName) {
        if (_cols == null)
        	return Collections.emptyList();
        List<Column> result = new ArrayList<>();
        for (Column col : _cols) {
        	if (DBIdentifier.equal(col.getTableIdentifier(),
        			tableName))
        		result.add(col);
        }
        return result;
    }

    /**
     * Raw column data.
     */
    public void setColumns(List<Column> cols) {
        _cols = cols;
    }

    /**
     * Raw index.
     */
    public Index getIndex() {
        return _idx;
    }

    /**
     * Raw index.
     */
    public void setIndex(Index idx) {
        _idx = idx;
    }

    /**
     * The user can mark columns as explicitly non-indexable.
     */
    public boolean canIndex() {
        return _canIdx;
    }

    /**
     * The user can mark columns as explicitly non-indexable.
     */
    public void setCanIndex(boolean indexable) {
        _canIdx = indexable;
    }

    /**
     *  Affirms if this instance represents an implicit relation. For example, a
     *  relation expressed as the value of primary key of the related class and
	 *  not as object reference.
     *
     * @since 1.3.0
     */
    public boolean isImplicitRelation() {
    	return _implicitRelation;
    }

    /**
     * Sets a marker to imply a logical relation that can not have any physical
     * manifest in the database. For example, a relation expressed as the value
     * of primary key of the related class and not as object reference.
     * Populated from @ForeignKey(implicit=true) annotation.
     * The mutator can only transit from false to true but not vice versa.
     *
     * @since 1.3.0
     */
    public void setImplicitRelation(boolean flag) {
    	_implicitRelation |= flag;
    }

    /**
     * Raw foreign key information.
     */
    public ForeignKey getForeignKey() {
        return _fk;
    }

    /**
     * Raw foreign key information.
     */
    public void setForeignKey(ForeignKey fk) {
        _fk = fk;
        if (fk != null && _join == JOIN_NONE)
            _join = JOIN_FORWARD;
    }

    /**
     * The user can mark columns as explicitly not having a foreign key.
     */
    public boolean canForeignKey() {
        return _canFK;
    }

    /**
     * The user can mark columns as explicitly not having a foreign key.
     */
    public void setCanForeignKey(boolean fkable) {
        _canFK = fkable;
    }

    /**
     * Raw unique constraint information.
     */
    public Unique getUnique() {
        return _unq;
    }

    /**
     * Raw unique constraint information.
     */
    public void setUnique(Unique unq) {
        _unq = unq;
    }

    /**
     * The user can mark columns as explicitly not having a unique constraint.
     */
    public boolean canUnique() {
        return _canUnq;
    }

    /**
     * The user can mark columns as explicitly not having a unique constraint.
     */
    public void setCanUnique(boolean uniquable) {
        _canUnq = uniquable;
    }

    /**
     * I/O for the columns created by the last call to {@link #createColumns},
     * or for the foreign key created by the last call to
     * {@link #createForeignKey}. This is also expected to be set correctly
     * prior to calls to {@link #syncColumns} and {@link #syncForeignKey}.
     */
    public ColumnIO getColumnIO() {
        return _io;
    }

    /**
     * I/O for the columns created by the last call to {@link #createColumns},
     * or for the foreign key created by the last call to
     * {@link #createForeignKey}. This is also expected to be set correctly
     * prior to calls to {@link #syncColumns} and {@link #syncForeignKey}.
     */
    public void setColumnIO(ColumnIO io) {
        _io = io;
    }

    /**
     * Direction of the join that the columns of this mapping info form. This
     * is usually automatically set by {@link #createForeignKey}. This flag
     * is also expected to be set correctly prior to calls to
     * {@link #syncForeignKey} if the join is inversed.
     */
    public int getJoinDirection() {
        return _join;
    }

    /**
     * Direction of the join that the columns of this mapping info form. This
     * is usually automatically set by {@link #createForeignKey}. This flag
     * is also expected to be set correctly prior to calls to
     * {@link #syncForeignKey} if the join is inversed.
     */
    public void setJoinDirection(int join) {
        _join = join;
    }

    /**
     * Clear all mapping information.
     */
    public void clear() {
        clear(true);
    }

    /**
     * Clear mapping information.
     *
     * @param canFlags whether to clear information about whether we
     * can place indexed, foreign keys, etc on this mapping
     */
    protected void clear(boolean canFlags) {
        _strategy = null;
        _cols = null;
        _io = null;
        _idx = null;
        _unq = null;
        _fk = null;
        _join = JOIN_NONE;
        if (canFlags) {
            _canIdx = true;
            _canFK = true;
            _canUnq = true;
        }
    }

    /**
     * Copy missing info from the instance to this one.
     */
    public void copy(MappingInfo info) {
        if (_strategy == null)
            _strategy = info.getStrategy();
        if (_canIdx && _idx == null) {
            if (info.getIndex() != null)
                _idx = info.getIndex();
            else
                _canIdx = info.canIndex();
        }
        if (_canUnq && _unq == null) {
            if (info.getUnique() != null)
                _unq = info.getUnique();
            else
                _canUnq = info.canUnique();
        }
        if (_canFK && _fk == null) {
            if (info.getForeignKey() != null)
                _fk = info.getForeignKey();
            else
                _canFK = info.canForeignKey();
        }
        _implicitRelation = info.isImplicitRelation();
        List<Column> cols = getColumns();
        List<Column> icols = info.getColumns();
        if (!icols.isEmpty() && (cols.isEmpty()
            || cols.size() == icols.size())) {
            if (cols.isEmpty())
                cols = new ArrayList<>(icols.size());
            for (int i = 0; i < icols.size(); i++) {
                if (cols.size() == i)
                    cols.add(new Column());
                cols.get(i).copy(icols.get(i));
            }
            setColumns(cols);
        }
    }

    /**
     * Return true if this info has columns, foreign key information, index
     * information, etc.
     */
    public boolean hasSchemaComponents() {
        return (_cols != null && !_cols.isEmpty())
            || _idx != null
            || _unq != null
            || _fk != null
            || !_canIdx
            || !_canFK
            || !_canUnq;
    }

    /**
     * Assert that the user did not supply any columns, index, unique
     * constraint, or foreign key for this mapping.
     */
    public void assertNoSchemaComponents(MetaDataContext context, boolean die) {
        if (_cols == null || _cols.isEmpty()) {
            assertNoIndex(context, die);
            assertNoUnique(context, die);
            assertNoForeignKey(context, die);
            return;
        }

        Message msg = _loc.get("unexpected-cols", context);
        if (die)
            throw new MetaDataException(msg);
        context.getRepository().getLog().warn(msg);
    }

    /**
     * Assert that this info has the given strategy or no strategy.
     */
    public void assertStrategy(MetaDataContext context, Object contextStrat,
        Object expected, boolean die) {
        if (contextStrat == expected)
            return;

        String strat;
        if (contextStrat == null) {
            if (_strategy == null)
                return;
            if (_strategy.equals(expected.getClass().getName()))
                return;
            if (expected instanceof Strategy
                && _strategy.equals(((Strategy) expected).getAlias()))
                return;
            strat = _strategy;
        } else if (contextStrat instanceof Strategy)
            strat = ((Strategy) contextStrat).getAlias();
        else
            strat = contextStrat.getClass().getName();

        Message msg = _loc.get("unexpected-strategy", context, expected,
            strat);
        if (die)
            throw new MetaDataException(msg);
        context.getRepository().getLog().warn(msg);
    }

    /**
     * Assert that the user did not try to place an index on this mapping.
     */
    public void assertNoIndex(MetaDataContext context, boolean die) {
        if (_idx == null)
            return;

        Message msg = _loc.get("unexpected-index", context);
        if (die)
            throw new MetaDataException(msg);
        context.getRepository().getLog().warn(msg);
    }

    /**
     * Assert that the user did not try to place a unique constraint on this
     * mapping.
     */
    public void assertNoUnique(MetaDataContext context, boolean die) {
        if (_unq == null)
            return;

        Message msg = _loc.get("unexpected-unique", context);
        if (die)
            throw new MetaDataException(msg);
        context.getRepository().getLog().warn(msg);
    }

    /**
     * Assert that the user did not try to place a foreign key on this mapping
     * or placed an implicit foreign key.
     */
    public void assertNoForeignKey(MetaDataContext context, boolean die) {
        if (_fk == null || isImplicitRelation())
            return;

        Message msg = _loc.get("unexpected-fk", context);
        if (die)
            throw new MetaDataException(msg);
        context.getRepository().getLog().warn(msg);
    }

    /**
     * Assert that the user did not try to join.
     */
    public void assertNoJoin(MetaDataContext context, boolean die) {
        boolean join = false;
        if (_cols != null) {
            Column col;
            for (int i = 0; !join && i < _cols.size(); i++) {
                col = _cols.get(i);
                if (!DBIdentifier.isNull(col.getTargetIdentifier()))
                    join = true;
            }
        }
        if (!join)
            return;

        Message msg = _loc.get("unexpected-join", context);
        if (die)
            throw new MetaDataException(msg);
        context.getRepository().getLog().warn(msg);
    }

    /**
     * Find or generate a table for a mapping.
     *
     * @param context the mapping that uses the table
     * @param def default table name provider
     * @param schemaName default schema if known, or null
     * @param given given table name
     * @param adapt whether we can alter the schema or mappings
     * @deprecated
     */
    @Deprecated
    public Table createTable(MetaDataContext context, TableDefaults def,
        String schemaName, String given, boolean adapt) {
        return createTable(context, def, DBIdentifier.newSchema(schemaName),
            DBIdentifier.newTable(given), adapt);
    }

    public Table createTable(MetaDataContext context, TableDefaults def,
        DBIdentifier schemaName, DBIdentifier given, boolean adapt) {
        MappingRepository repos = (MappingRepository) context.getRepository();
        if (DBIdentifier.isNull(given) && (def == null || (!adapt
            && !repos.getMappingDefaults().defaultMissingInfo())))
            throw new MetaDataException(_loc.get("no-table", context));

        if (DBIdentifier.isNull(schemaName)) {
            //Check the configuration first for a set Schema to use
            schemaName = Schemas.getNewTableSchemaIdentifier((JDBCConfiguration)
                repos.getConfiguration());

            // If the schemaName is still NULL type then check for a system default schema name
            // and if available use it.
            if (schemaName != null && (schemaName.getType() == DBIdentifierType.NULL)) {
                String name = repos.getMetaDataFactory().getDefaults().getDefaultSchema();
                schemaName = (name != null ? DBIdentifier.newSchema(name) : schemaName);
            }
        }

        // if no given and adapting or defaulting missing info, use template
        SchemaGroup group = repos.getSchemaGroup();
        Schema schema = null;
        if (DBIdentifier.isNull(given)) {
            schema = group.getSchema(schemaName);
            if (schema == null)
                schema = group.addSchema(schemaName);
            given = def.getIdentifier(schema);
        }

        QualifiedDBIdentifier path = QualifiedDBIdentifier.getPath(given);
        if (DBIdentifier.isNull(path.getSchemaName())) {
            if (!DBIdentifier.isNull(schemaName)) {
                path.setSchemaName(schemaName);
            }
        } else {
            schemaName = path.getSchemaName();
            schema = null;
        }

        // look for named table using full name and findTable, which allows
        // the dynamic schema factory to create the table if needed
        Table table = group.findTable(path);
        if (table != null)
            return table;
        if (!adapt)
            throw new MetaDataException(_loc.get("bad-table", given, context));

        // named table doesn't exist; create it
        if (schema == null) {
            schema = group.getSchema(schemaName);
            if (schema == null)
                schema = group.addSchema(schemaName);
        }
        table = schema.getTable(given);
        if (table == null)
            table = schema.addTable(given);
        return table;
    }

    /**
     * Retrieve/create columns on the given table by merging the given
     * template information with any user-provided information.
     *
     * @param context the mapping we're retrieving columns for
     * @param prefix localized error message key prefix
     * @param tmplates template columns
     * @param table the table for the columns
     * @param adapt whether we can modify the existing mapping or schema
     */
    protected Column[] createColumns(MetaDataContext context, String prefix,
        Column[] tmplates, Table table, boolean adapt) {
        assertTable(context, table);
        if (prefix == null)
            prefix = "generic";

        // the user has to give the right number of expected columns for this
        // mapping, or none at all if we're adapting.  can't just given one of
        // n columns because we don't know which of the n columns the info
        // applies to
        List<Column> given = getColumns();

        if (context instanceof FieldMapping && ((FieldMapping)context).hasMapsIdCols())
            given = ((FieldMapping)context).getValueInfo().getMapsIdColumns();

        boolean fill = ((MappingRepository) context.getRepository()).
            getMappingDefaults().defaultMissingInfo();
        if ((!given.isEmpty() || (!adapt && !fill))
            && given.size() != tmplates.length) {
            // also consider when this info has columns from multiple tables
            given = getColumns(table.getIdentifier());
            if ((!adapt && !fill) && given.size() != tmplates.length) {
                // try default table
                given = getColumns("");
                if ((!adapt && !fill) && given.size() != tmplates.length) {
                    throw new MetaDataException(_loc.get(prefix + "-num-cols",
                            context, String.valueOf(tmplates.length),
                            String.valueOf(given.size())));
                }
            }
        }

        Column[] cols = new Column[tmplates.length];
        _io = null;
        Column col;
        for (int i = 0; i < tmplates.length; i++) {
            col = (given.isEmpty()) ? null : (Column) given.get(i);
            cols[i] = mergeColumn(context, prefix, tmplates[i], true, col,
                table, adapt, fill);
            setIOFromColumnFlags(col, i);
        }
        return cols;
    }

    /**
     * Set the proper internal column I/O metadata for the given column's flags.
     */
    private void setIOFromColumnFlags(Column col, int i) {
        if (col == null || (!col.getFlag(Column.FLAG_UNINSERTABLE)
            && !col.getFlag(Column.FLAG_UNUPDATABLE)
            && !col.isNotNull()))
            return;

        if (_io == null)
            _io = new ColumnIO();
        _io.setInsertable(i, !col.getFlag(Column.FLAG_UNINSERTABLE));
        _io.setUpdatable(i, !col.getFlag(Column.FLAG_UNUPDATABLE));
        _io.setNullInsertable(i, !col.isNotNull());
        _io.setNullUpdatable(i, !col.isNotNull());
    }

    /**
     * Assert that the given table is non-null.
     */
    private static void assertTable(MetaDataContext context, Table table) {
        if (table == null)
            throw new MetaDataException(_loc.get("unmapped", context));
    }

    /**
     * Merge the given columns if possible.
     *
     * @param context the mapping we're retrieving columns for
     * @param prefix localized error message key prefix
     * @param tmplate template for expected column information
     * @param compat whether the existing column type must be compatible
     * with the type of the template column
     * @param given the given column information from mapping info
     * @param table the table for the columns
     * @param adapt whether we can modify the existing mapping or schema
     * @param fill whether to default missing column information
     */
    protected static Column mergeColumn(MetaDataContext context, String prefix,
        Column tmplate, boolean compat, Column given, Table table,
        boolean adapt, boolean fill) {
        assertTable(context, table);

        // if not adapting must provide column name at a minimum
        DBIdentifier colName = (given == null) ? DBIdentifier.NULL : given.getIdentifier();
        if (DBIdentifier.isNull(colName) && !adapt && !fill)
            throw new MetaDataException(_loc.get(prefix + "-no-col-name",
                context));

        MappingRepository repos = (MappingRepository) context.getRepository();
        DBDictionary dict = repos.getDBDictionary();

        // determine the column name based on given info, or template if none;
        // also make sure that if the user gave a column name, he didn't try
        // to put the column in an unexpected table
        if (DBIdentifier.isNull(colName))
            colName = tmplate.getIdentifier();
        QualifiedDBIdentifier path = QualifiedDBIdentifier.getPath(colName);
        if (path.isUnqualifiedColumn()) {
            colName = path.getIdentifier();
        } else if (!DBIdentifier.isNull(path.getObjectTableName())) {
            findTable(context, path.getObjectTableName(), table,
                null, null);
            colName = path.getUnqualifiedName();
        }

        // find existing column
        Column col = table.getColumn(colName);
        if (col == null && !adapt) {
            //
            // See if column name has already been validated in a dynamic table.
            // If so then want to use that validated column name instead. This
            // should seldom if ever occur as long as the database dictionaries
            // are kept up-to-date.
            //
            if ((colName.getName().length() > dict.maxColumnNameLength) ||
               dict.getInvalidColumnWordSet().contains(DBIdentifier.toUpper(colName).getName()) &&
              !(table.getClass().getName().contains("DynamicTable"))) {
                colName=dict.getValidColumnName(colName, new Table());
                col = table.getColumn(colName);
                if (col == null && !adapt) {
                    throw new MetaDataException(_loc.
                        get(prefix + "-bad-col-name", context, colName, table));
                }
            }
            else {
                throw new MetaDataException(_loc.
                    get(prefix + "-bad-col-name", context, colName, table));
            }
        }

        // use information from template column by default, allowing any
        // user-given specifics to override it
        int type = tmplate.getType();
        int size = tmplate.getSize();
        if (type == Types.OTHER) {
            int precis = 0;
            int scale = 0;
            if(given != null) {
                precis = given.getSize();
                scale = given.getDecimalDigits();
            }
            type =
                dict.getJDBCType(tmplate.getJavaType(), size == -1, precis,
                    scale, tmplate.isXML());
        }

        boolean ttype = true;
        int otype = type;
        String typeName = tmplate.getTypeName();
        Boolean notNull = null;
        if (tmplate.isNotNullExplicit())
            notNull = (tmplate.isNotNull()) ? Boolean.TRUE : Boolean.FALSE;
        int decimals = tmplate.getDecimalDigits();
        String defStr = tmplate.getDefaultString();
        boolean autoAssign = tmplate.isAutoAssigned();
        boolean relationId = tmplate.isRelationId();
        boolean implicitRelation = tmplate.isImplicitRelation();
        String targetField = tmplate.getTargetField();
        if (given != null) {
            // use given type if provided, but warn if it isn't compatible with
            // the expected column type
            if (given.getType() != Types.OTHER) {
                ttype = false;
                if (compat && !given.isCompatible(type, typeName, size,
                    decimals)) {
                    Log log = repos.getLog();
                    if (log.isWarnEnabled())
                        log.warn(_loc.get(prefix + "-incompat-col",
                            context, colName, Schemas.getJDBCName(type)));
                }
                otype = given.getType();
                type = dict.getPreferredType(otype);
            }
            typeName = given.getTypeName();
            if (given.getSize() > 0)
                size = given.getSize();
            decimals = given.getDecimalDigits();

            // leave this info as the template defaults unless the user
            // explicitly turns it on in the given column
            if (given.isNotNullExplicit())
                notNull = (given.isNotNull()) ? Boolean.TRUE : Boolean.FALSE;
            if (given.getDefaultString() != null)
                defStr = given.getDefaultString();
            if (given.isAutoAssigned())
                autoAssign = true;
            if (given.isRelationId())
                relationId = true;
            if (given.isImplicitRelation())
                implicitRelation = true;
        }

        // default char column size if original type is char (test original
        // type rather than final type because orig might be clob, translated
        // to an unsized varchar, which is supported by some dbs)
        if (size == 0 && (otype == Types.VARCHAR || otype == Types.CHAR))
            size = dict.characterColumnSize;

        // create column, or make sure existing column matches expected type
        if (col == null) {
            col = table.addColumn(colName);
            col.setType(type);
        } else if ((compat || !ttype) &&
                !col.isCompatible(type, typeName, size, decimals)) {
            // if existing column isn't compatible with desired type, die if
            // can't adapt, else warn and change the existing column type
            Message msg = _loc.get(prefix + "-bad-col", context,
                Schemas.getJDBCName(type), col.getDescription());
            if (!adapt && !dict.disableSchemaFactoryColumnTypeErrors)
                throw new MetaDataException(msg);
            Log log = repos.getLog();
            if (log.isWarnEnabled())
                log.warn(msg);

            col.setType(type);
        } else if (given != null && given.getType() != Types.OTHER) {
            // as long as types are compatible, set column to expected type
            col.setType(type);
        }

        // always set the java type and autoassign to expected values, even on
        // an existing column, since we don't get this from the DB
        if (compat)
            col.setJavaType(tmplate.getJavaType());
        else if (col.getJavaType() == JavaTypes.OBJECT) {
            if (given != null && given.getJavaType() != JavaTypes.OBJECT)
                col.setJavaType(given.getJavaType());
            else
                col.setJavaType(JavaTypes.getTypeCode
                    (Schemas.getJavaType(col.getType(), col.getSize(),
                        col.getDecimalDigits())));
        }
        col.setAutoAssigned(autoAssign);
        col.setRelationId(relationId);
        col.setImplicitRelation(implicitRelation);
        col.setTargetField(targetField);

        // we need this for runtime, and the dynamic schema factory might
        // not know it, so set it even if not adapting
        if (defStr != null)
            col.setDefaultString(defStr);
        if (notNull != null)
            col.setNotNull(notNull);

        // add other details if adapting
        if (adapt) {
            if (typeName != null)
                col.setTypeName(typeName);
            if (size != 0)
                col.setSize(size);
            if (decimals != 0)
                col.setDecimalDigits(decimals);
        }

        if (tmplate.hasComment())
            col.setComment(tmplate.getComment());
        if (tmplate.isXML())
            col.setXML(tmplate.isXML());
        return col;
    }

    /**
     * Find the table named by a column or target.
     *
     * @param context context for error messages, etc.
     * @param name the table name, possibly including schema
     * @param expected the expected table; may be null
     * @param inverse the possible inverse table; may be null
     * @param rel if we're finding the target table of a join, the
     * joined-to type; allows us to also look in its superclass tables
     */
    private static Table findTable(MetaDataContext context, DBIdentifier name,
        Table expected, Table inverse, ClassMapping rel) {
        // is this the expected table?
        if (expected == null && rel != null)
            expected = rel.getTable();
        if (expected != null && isTableName(name, expected))
            return expected;

        // check for inverse
        if (inverse != null && isTableName(name, inverse))
            return inverse;

        // superclass table?
        if (rel != null)
            rel = rel.getJoinablePCSuperclassMapping();
        while (rel != null) {
            if (isTableName(name, rel.getTable()))
                return rel.getTable();
            rel = rel.getJoinablePCSuperclassMapping();
        }

        // none of the possible tables
        throw new MetaDataException(_loc.get("col-wrong-table", context,
            expected, name.getName()));
    }

    /**
     * Return whether the given name matches the given table.
     */
    private static boolean isTableName(DBIdentifier name, Table table) {
        return DBIdentifier.equal(name, table.getIdentifier())
            || DBIdentifier.equal(name, table.getFullIdentifier());
    }

    /**
     * Retrieve/create an index on the given columns by merging the given
     * template information with any user-provided information.
     *
     * @param context the mapping we're retrieving an index for
     * @param prefix localized error message key prefix
     * @param tmplate template for expected index information
     * @param cols the indexed columns
     * @param adapt whether we can modify the existing mapping or schema
     */
    protected Index createIndex(MetaDataContext context, String prefix,
        Index tmplate, Column[] cols, boolean adapt) {
        if (prefix == null)
            prefix = "generic";

        // can't create an index if there are no cols
        if (cols == null || cols.length == 0) {
            if (_idx != null)
                throw new MetaDataException(_loc.get(prefix
                    + "-no-index-cols", context));
            return null;
        }

        // look for an existing index on these columns
        Table table = cols[0].getTable();
        Index[] idxs = table.getIndexes();
        Index exist = null;
        for (Index index : idxs) {
            if (index.columnsMatch(cols)) {
                exist = index;
                break;
            }
        }

        // remove existing index?
        if (!_canIdx) {
            if (exist == null)
                return null;
            if (!adapt)
                throw new MetaDataException(_loc.get(prefix + "-index-exists",
                    context));
            table.removeIndex(exist);
            return null;
        }

        // if we have an existing index, merge given info into it
        if (exist != null) {
