/*
 * 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.persistence.jdbc;

import static org.apache.openjpa.persistence.jdbc.MappingTag.ASSOC_OVERRIDE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.ATTR_OVERRIDE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.COLLECTION_TABLE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.COLS;
import static org.apache.openjpa.persistence.jdbc.MappingTag.COLUMN_NAME;
import static org.apache.openjpa.persistence.jdbc.MappingTag.COLUMN_RESULT;
import static org.apache.openjpa.persistence.jdbc.MappingTag.DATASTORE_ID_COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.DELIMITED_IDS;
import static org.apache.openjpa.persistence.jdbc.MappingTag.DISCRIM_COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.DISCRIM_VAL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.ENTITY_RESULT;
import static org.apache.openjpa.persistence.jdbc.MappingTag.ENUMERATED;
import static org.apache.openjpa.persistence.jdbc.MappingTag.FIELD_RESULT;
import static org.apache.openjpa.persistence.jdbc.MappingTag.FK;
import static org.apache.openjpa.persistence.jdbc.MappingTag.FK_COL_NAME;
import static org.apache.openjpa.persistence.jdbc.MappingTag.FK_COL_NAMES;
import static org.apache.openjpa.persistence.jdbc.MappingTag.INDEX;
import static org.apache.openjpa.persistence.jdbc.MappingTag.INHERITANCE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.JOIN_COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.JOIN_TABLE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.MAP_KEY_COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.MAP_KEY_ENUMERATED;
import static org.apache.openjpa.persistence.jdbc.MappingTag.MAP_KEY_JOIN_COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.MAP_KEY_TEMPORAL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.NAME;
import static org.apache.openjpa.persistence.jdbc.MappingTag.ORDER_COLUMN;
import static org.apache.openjpa.persistence.jdbc.MappingTag.PK_JOIN_COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.SECONDARY_TABLE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.SQL_RESULT_SET_MAPPING;
import static org.apache.openjpa.persistence.jdbc.MappingTag.TABLE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.TABLE_GEN;
import static org.apache.openjpa.persistence.jdbc.MappingTag.TEMPORAL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.UNIQUE;
import static org.apache.openjpa.persistence.jdbc.MappingTag.VERSION_COL;
import static org.apache.openjpa.persistence.jdbc.MappingTag.VERSION_COLS;

import java.lang.reflect.Modifier;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import jakarta.persistence.DiscriminatorType;
import jakarta.persistence.EnumType;
import jakarta.persistence.InheritanceType;
import jakarta.persistence.TemporalType;

import org.apache.openjpa.jdbc.conf.JDBCConfiguration;
import org.apache.openjpa.jdbc.identifier.DBIdentifier;
import org.apache.openjpa.jdbc.identifier.QualifiedDBIdentifier;
import org.apache.openjpa.jdbc.kernel.EagerFetchModes;
import org.apache.openjpa.jdbc.meta.ClassMapping;
import org.apache.openjpa.jdbc.meta.ClassMappingInfo;
import org.apache.openjpa.jdbc.meta.DiscriminatorMappingInfo;
import org.apache.openjpa.jdbc.meta.FieldMapping;
import org.apache.openjpa.jdbc.meta.FieldMappingInfo;
import org.apache.openjpa.jdbc.meta.MappingInfo;
import org.apache.openjpa.jdbc.meta.MappingRepository;
import org.apache.openjpa.jdbc.meta.QueryResultMapping;
import org.apache.openjpa.jdbc.meta.QueryResultMapping.PCResult;
import org.apache.openjpa.jdbc.meta.SequenceMapping;
import org.apache.openjpa.jdbc.meta.strats.EnumValueHandler;
import org.apache.openjpa.jdbc.meta.strats.FlatClassStrategy;
import org.apache.openjpa.jdbc.meta.strats.FullClassStrategy;
import org.apache.openjpa.jdbc.meta.strats.NoneClassStrategy;
import org.apache.openjpa.jdbc.meta.strats.VerticalClassStrategy;
import org.apache.openjpa.jdbc.schema.Column;
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.meta.SourceTracker;
import org.apache.openjpa.lib.util.Localizer;
import org.apache.openjpa.lib.util.StringUtil;
import org.apache.openjpa.meta.ClassMetaData;
import org.apache.openjpa.meta.FieldMetaData;
import org.apache.openjpa.meta.JavaTypes;
import org.apache.openjpa.persistence.XMLPersistenceMetaDataParser;
import org.apache.openjpa.util.InternalException;
import org.apache.openjpa.util.MetaDataException;
import org.apache.openjpa.util.UserException;
import org.xml.sax.Attributes;
import org.xml.sax.Locator;
import org.xml.sax.SAXException;
/**
 * Custom SAX parser used by the system to parse persistence mapping files.
 *
 * @author Steve Kim
 */
public class XMLPersistenceMappingParser
    extends XMLPersistenceMetaDataParser {

    private static final Map<String, MappingTag> _elems =
        new HashMap<>();

    static {
        _elems.put("association-override", ASSOC_OVERRIDE);
        _elems.put("attribute-override", ATTR_OVERRIDE);
        _elems.put("collection-table", COLLECTION_TABLE);
        _elems.put("column", COL);
        _elems.put("columns", COLS);
        _elems.put("column-name", COLUMN_NAME);
        _elems.put("column-result", COLUMN_RESULT);
        _elems.put("data-store-id-column", DATASTORE_ID_COL);
        _elems.put("delimited-identifiers", DELIMITED_IDS);
        _elems.put("discriminator-column", DISCRIM_COL);
        _elems.put("discriminator-value", DISCRIM_VAL);
        _elems.put("entity-result", ENTITY_RESULT);
        _elems.put("enumerated", ENUMERATED);
        _elems.put("field-result", FIELD_RESULT);
        _elems.put("foreign-key", FK);
        _elems.put("fk_column-names", FK_COL_NAMES);
        _elems.put("fk_column_name", FK_COL_NAME);
        _elems.put("inheritance", INHERITANCE);
        _elems.put("index", INDEX);
        _elems.put("join-column", JOIN_COL);
        _elems.put("inverse-join-column", COL);
        _elems.put("join-table", JOIN_TABLE);
        _elems.put("map-key-enumerated", MAP_KEY_ENUMERATED);
        _elems.put("map-key-column", MAP_KEY_COL);
        _elems.put("map-key-join-column", MAP_KEY_JOIN_COL);
        _elems.put("map-key-temporal", MAP_KEY_TEMPORAL);
        _elems.put("name", NAME);
        _elems.put("order-column", ORDER_COLUMN);
        _elems.put("primary-key-join-column", PK_JOIN_COL);
        _elems.put("secondary-table", SECONDARY_TABLE);
        _elems.put("sql-result-set-mapping", SQL_RESULT_SET_MAPPING);
        _elems.put("table", TABLE);
        _elems.put("table-generator", TABLE_GEN);
        _elems.put("temporal", TEMPORAL);
        _elems.put("unique-constraint", UNIQUE);
        _elems.put("version-columns", VERSION_COLS);
        _elems.put("version-column", VERSION_COL);
    }

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

    private String _override = null;
    private String _schema = null;
    private String _colTable = null;
    private String _secondaryTable = null;
    private List<Column> _cols = null;
    private List<Column> _joinCols = null;
    private List<Column> _supJoinCols = null;
    private boolean _lob = false;
    private TemporalType _temporal = null;
    private EnumSet<UniqueFlag> _unique = EnumSet.noneOf(UniqueFlag.class);
    private DiscriminatorType _discType;
    private Column _discCol;
    private int _resultIdx = 0;
    private final DBDictionary _dict;

    // ForeignKey info
    private Attributes _foreignKeyAttributes = null;
    private List<String> _columnNamesList = null;
    private String[] _columnNames = {};

    private List<Column> _versionColumnsList = null;

    private final Map<Class<?>, ArrayList<DeferredEmbeddableOverrides>>
        _deferredMappings = new HashMap<>();

    /**
     * Constructor; supply configuration.
     */
    public XMLPersistenceMappingParser(JDBCConfiguration conf) {
        super(conf);
        _dict = conf.getDBDictionaryInstance();
    }

    @Override
    protected void reset() {
        super.reset();
        clearColumnInfo();
        clearClassInfo();
        clearSecondaryTableInfo();
        _override = null;
        _schema = null;
        _resultIdx = 0;
    }

    @Override
    protected Object startSystemMappingElement(String name, Attributes attrs)
        throws SAXException {
        MappingTag tag = _elems.get(name);
        if (tag == null) {
            if ("schema".equals(name))
                return name;
            return null;
        }

        boolean ret;
        switch (tag) {
            case TABLE_GEN:
                ret = startTableGenerator(attrs);
                break;
            case SQL_RESULT_SET_MAPPING:
                ret = startSQLResultSetMapping(attrs);
                break;
            case ENTITY_RESULT:
                ret = startEntityResult(attrs);
                break;
            case FIELD_RESULT:
                ret = startFieldResult(attrs);
                break;
            case COLUMN_RESULT:
                ret = startColumnResult(attrs);
                break;
            default:
                ret = false;
        }
        return (ret) ? tag : null;
    }

    @Override
    protected void endSystemMappingElement(String name)
        throws SAXException {
        MappingTag tag = _elems.get(name);
        if (tag == null) {
            if ("schema".equals(name)) {
                _schema = currentText();
                getRepository().getMetaDataFactory().getDefaults().setDefaultSchema(_schema);
            }
            return;
        }

        switch (tag) {
            case SQL_RESULT_SET_MAPPING:
                endSQLResultSetMapping();
                break;
            case ENTITY_RESULT:
                endEntityResult();
                break;
        }
    }

    @Override
    protected Object startClassMappingElement(String name, Attributes attrs)
        throws SAXException {
        MappingTag tag = _elems.get(name);
        if (tag == null)
            return null;

        boolean ret;
        switch (tag) {
            case TABLE:
                ret = startTable(attrs);
                break;
            case SECONDARY_TABLE:
                ret = startSecondaryTable(attrs);
                break;
            case DISCRIM_COL:
                parseDiscriminatorColumn(attrs);
                _discCol = parseColumn(attrs);
                ret = true;
                break;
            case DISCRIM_VAL:
                ret = true;
                break;
            case INHERITANCE:
                ret = startInheritance(attrs);
                break;
            case ASSOC_OVERRIDE:
            case ATTR_OVERRIDE:
                ret = startAttributeOverride(attrs);
                break;
            case PK_JOIN_COL:
                ret = startPrimaryKeyJoinColumn(attrs);
                break;
            case COL:
                ret = startColumn(attrs);
                break;
            case COLS:
                ret = true;
                break;
            case JOIN_COL:
                ret = startJoinColumn(attrs);
                break;
            case JOIN_TABLE:
                ret = startJoinTable(attrs);
                break;
            case TABLE_GEN:
                ret = startTableGenerator(attrs);
                break;
            case UNIQUE:
                ret = startUniqueConstraint(attrs);
                break;
            case NAME:
                ret = true;
                break;
            case TEMPORAL:
            case ENUMERATED:
            case MAP_KEY_ENUMERATED:
            case MAP_KEY_TEMPORAL:
                ret = true;
                break;
            case SQL_RESULT_SET_MAPPING:
                ret = startSQLResultSetMapping(attrs);
                break;
            case ENTITY_RESULT:
                ret = startEntityResult(attrs);
                break;
            case FIELD_RESULT:
                ret = startFieldResult(attrs);
                break;
            case COLUMN_RESULT:
                ret = startColumnResult(attrs);
                break;
            case COLUMN_NAME:
                ret = true;
                break;
            case COLLECTION_TABLE:
                ret = startCollectionTable(attrs);
                break;
            case MAP_KEY_COL:
                ret = startMapKeyColumn(attrs);
                break;
            case MAP_KEY_JOIN_COL:
                ret = startMapKeyJoinColumn(attrs);
                break;
            case DATASTORE_ID_COL:
                ret = startDatastoreIdCol(attrs);
                break;
            case INDEX:
                ret = startIndex(attrs);
                break;
            case FK:
                ret = startForeignKey(attrs);
                break;
            case FK_COL_NAMES:
                ret = startFKColumnNames(attrs);
                break;
            case FK_COL_NAME:
                ret = true;
                break;
            case VERSION_COLS:
                ret = startVersionColumns(attrs);
                break;
            case VERSION_COL:
                ret = startVersionColumn(attrs);
                break;
            default:
                ret = false;
        }
        return (ret) ? tag : null;
    }

    private boolean endName() {
        String name = this.currentText();
        if (StringUtil.isNotEmpty(name)) {
            Object current = currentElement();
            if (current instanceof Unique) {
                Unique unq = (Unique)current;
                unq.setIdentifier(DBIdentifier.newConstraint(name, delimit()));
            }
        }

        return true;
    }

    @Override
    protected void endClassMappingElement(String name)
        throws SAXException {
        MappingTag tag = _elems.get(name);
        if (tag == null)
            return;

        switch (tag) {
            case SECONDARY_TABLE:
                endSecondaryTable();
                break;
            case DISCRIM_VAL:
                endDiscriminatorValue();
                break;
            case ASSOC_OVERRIDE:
            case ATTR_OVERRIDE:
                endAttributeOverride();
                break;
            case JOIN_TABLE:
                endJoinTable();
                break;
            case TEMPORAL:
                endTemporal();
                break;
            case MAP_KEY_TEMPORAL:
                endMapKeyTemporal();
                break;
            case ENUMERATED:
                endEnumerated();
                break;
            case MAP_KEY_ENUMERATED:
                endMapKeyEnumerated();
                break;
            case SQL_RESULT_SET_MAPPING:
                endSQLResultSetMapping();
                break;
            case ENTITY_RESULT:
                endEntityResult();
                break;
            case UNIQUE:
                endUniqueConstraint();
                break;
            case COLUMN_NAME:
                endColumnName();
                break;
            case TABLE_GEN:
            	endTableGenerator();
            	break;
            case NAME:
                endName();
                break;
            case FK:
                endForeignKey();
                break;
            case FK_COL_NAMES:
                endFKColumnNames();
                break;
            case FK_COL_NAME:
                endFKColumnName();
                break;
            case VERSION_COLS:
                endVersionColumns();
                break;
        }
    }

    @Override
    protected void startClassMapping(ClassMetaData meta, boolean mappedSuper,
        Attributes attrs)
        throws SAXException {
        if (mappedSuper)
            ((ClassMapping) meta).getMappingInfo().setStrategy
                (NoneClassStrategy.ALIAS);
    }

    @Override
    protected void endClassMapping(ClassMetaData meta)
        throws SAXException {
        ClassMapping cm = (ClassMapping) meta;
        if (_schema != null)
            cm.getMappingInfo().setSchemaIdentifier(DBIdentifier.newSchema(_schema, delimit()));

        if (_supJoinCols != null)
            cm.getMappingInfo().setColumns(_supJoinCols);

        if (_discCol != null) {
            DiscriminatorMappingInfo dinfo = cm.getDiscriminator()
                    .getMappingInfo();
            switch (_discType) {
                case CHAR:
                    _discCol.setJavaType(JavaTypes.CHAR);
                    cm.getDiscriminator().setJavaType(JavaTypes.CHAR);
                    break;
                case INTEGER:
                    _discCol.setJavaType(JavaTypes.INT);
                    cm.getDiscriminator().setJavaType(JavaTypes.INT);
                    break;
                default:
                    _discCol.setJavaType(JavaTypes.STRING);
                    cm.getDiscriminator().setJavaType(JavaTypes.STRING);
                    break;
            }
            dinfo.setColumns(Arrays.asList(new Column[]{ _discCol }));
        }
        clearClassInfo();
    }

    /**
     * Clear cached class mapping info.
     */
    private void clearClassInfo() {
        _supJoinCols = null;
        _discCol = null;
        _discType = null;
    }

    /**
     * Start tracking secondary table information and columns
     */
    private boolean startSecondaryTable(Attributes attrs)
        throws SAXException {
        _secondaryTable = toTableIdentifier(attrs.getValue("schema"),
            attrs.getValue("name")).getName();
        ((ClassMapping)currentElement()).getMappingInfo()
        	.addSecondaryTable(DBIdentifier.newTable(_secondaryTable));
        return true;
    }

    /**
     * Set the secondary table information back to the owning class mapping.
     */
    private void endSecondaryTable() {
        ClassMapping cm = (ClassMapping) currentElement();
        ClassMappingInfo info = cm.getMappingInfo();
        info.setSecondaryTableJoinColumns(DBIdentifier.newTable(_secondaryTable, delimit()), _joinCols);
        clearSecondaryTableInfo();
    }

    /**
     * Clear cached secondary table info.
     */
    private void clearSecondaryTableInfo() {
        _joinCols = null;
        _secondaryTable = null;
    }

    /**
     * Parse table-generator.
     */
    private boolean startTableGenerator(Attributes attrs) {
        String name = attrs.getValue("name");
        Log log = getLog();
        if (log.isTraceEnabled())
            log.trace(_loc.get("parse-gen", name));
        if (getRepository().getCachedSequenceMetaData(name) != null
            && log.isWarnEnabled())
            log.warn(_loc.get("override-gen", name));

        SequenceMapping seq = (SequenceMapping) getRepository().
            addSequenceMetaData(name);
        seq.setSequencePlugin(SequenceMapping.IMPL_VALUE_TABLE);
        seq.setTableIdentifier(toTableIdentifier(attrs.getValue("schema"),
            attrs.getValue("table")));
        seq.setPrimaryKeyColumnIdentifier(DBIdentifier.newColumn(attrs.getValue("pk-column-name"), delimit()));
        seq.setSequenceColumnIdentifier(DBIdentifier.newColumn(attrs.getValue("value-column-name"), delimit()));
        seq.setPrimaryKeyValue(attrs.getValue("pk-column-value"));
        String val = attrs.getValue("initial-value");
        if (val != null)
            seq.setInitialValue(Integer.parseInt(val));
        val = attrs.getValue("allocation-size");
        if (val != null)
            seq.setAllocate(Integer.parseInt(val));

        Object cur = currentElement();
        Object scope = (cur instanceof ClassMetaData)
            ? ((ClassMetaData) cur).getDescribedType() : null;
        seq.setSource(getSourceFile(), scope, SourceTracker.SRC_XML);
        Locator locator = getLocation().getLocator();
        if (locator != null) {
            seq.setLineNumber(locator.getLineNumber());
            seq.setColNumber(locator.getColumnNumber());
        }
        pushElement(seq);
        return true;
    }

    private void endTableGenerator() {
    	popElement();
    }

    /**
     * Parse inheritance.
     */
    private boolean startInheritance(Attributes attrs) {
        String val = attrs.getValue("strategy");
        if (val == null)
            return true;

        ClassMapping cm = (ClassMapping) currentElement();
        ClassMappingInfo info = cm.getMappingInfo();
        switch (Enum.valueOf(InheritanceType.class, val)) {
            case SINGLE_TABLE:
                info.setHierarchyStrategy(FlatClassStrategy.ALIAS);
                break;
            case JOINED:
                info.setHierarchyStrategy(VerticalClassStrategy.ALIAS);
                break;
            case TABLE_PER_CLASS:
                info.setHierarchyStrategy(FullClassStrategy.ALIAS);
                break;
        }
        return true;
    }

    /**
     * Parse discriminator-value.
     */
    private void endDiscriminatorValue() {
        String val = currentText();
        if (StringUtil.isEmpty(val))
            return;

        ClassMapping cm = (ClassMapping) currentElement();
        cm.getDiscriminator().getMappingInfo().setValue(val);

        if (Modifier.isAbstract(cm.getDescribedType().getModifiers())
                && getLog().isInfoEnabled()) {
            getLog().info(
                    _loc.get("discriminator-on-abstract-class", cm
                            .getDescribedType().getName()));
        }
    }

    /**
     * Parse temporal.
     */
    private void endTemporal() {
        String temp = currentText();
        if (!StringUtil.isEmpty(temp))
            _temporal = Enum.valueOf(TemporalType.class, temp);
    }

    /**
     * Parse temporal.
     */
    private void endMapKeyTemporal() {
        String temp = currentText();
        TemporalType _mapKeyTemporal = null;
        if (!StringUtil.isEmpty(temp))
            _mapKeyTemporal = Enum.valueOf(TemporalType.class, temp);
        FieldMapping fm = (FieldMapping) currentElement();
        List<Column> cols = fm.getKeyMapping().getValueInfo().getColumns();
        if (cols.isEmpty()) {
            cols = Arrays.asList(new Column[]{ new Column() });
            fm.getKeyMapping().getValueInfo().setColumns(cols);
        }

        Column col = (Column) cols.get(0);
        switch (_mapKeyTemporal) {
            case DATE:
                col.setType(Types.DATE);
                break;
            case TIME:
                col.setType(Types.TIME);
                break;
            case TIMESTAMP:
                col.setType(Types.TIMESTAMP);
                break;
        }
    }

    /**
     * Parse enumerated.
     */
    private void endEnumerated() {
        String text = currentText();
        if (StringUtil.isEmpty(text))
            return;
        EnumType type = Enum.valueOf(EnumType.class, text);

        FieldMapping fm = (FieldMapping) currentElement();
        String strat = EnumValueHandler.class.getName() + "(StoreOrdinal="
            + (type == EnumType.ORDINAL) + ")";
        if (fm.isElementCollection())
            fm.getElementMapping().getValueInfo().setStrategy(strat);
        else
            fm.getValueInfo().setStrategy(strat);
    }

    /**
     * Parse map-key-enumerated.
     */
    private void endMapKeyEnumerated() {
        String text = currentText();
        if (StringUtil.isEmpty(text))
            return;
        EnumType type = Enum.valueOf(EnumType.class, text);

        FieldMapping fm = (FieldMapping) currentElement();
        String strat = EnumValueHandler.class.getName() + "(StoreOrdinal="
            + (type == EnumType.ORDINAL) + ")";
        fm.getKeyMapping().getValueInfo().setStrategy(strat);
    }

    @Override
    protected boolean startLob(Attributes attrs)
        throws SAXException {
        if (super.startLob(attrs)) {
            _lob = true;
            return true;
        }
        return false;
    }

    /**
     * Extend to clear annotation mapping info.
     */
    @Override
    protected void startFieldMapping(FieldMetaData field, Attributes attrs)
        throws SAXException {
        super.startFieldMapping(field, attrs);
        if (getAnnotationParser() != null) {
            FieldMapping fm = (FieldMapping) field;
            fm.getMappingInfo().clear();
            fm.getValueInfo().clear();
            fm.getElementMapping().getValueInfo().clear();
            fm.getKeyMapping().getValueInfo().clear();
        }
    }

    /**
     * Extend to set the columns.
     */
    @Override
    protected void endFieldMapping(FieldMetaData field)
        throws SAXException {
        // setup columns with cached lob and temporal info
        FieldMapping fm = (FieldMapping) field;
        if (_lob || _temporal != null) {
            int typeCode = fm.isElementCollection() ? fm.getElement().getDeclaredTypeCode() :
                fm.getDeclaredTypeCode();
            Class<?> type = fm.isElementCollection() ? fm.getElement().getDeclaredType() : fm.getDeclaredType();
            if (_cols == null) {
                _cols = new ArrayList<>(1);
                _cols.add(new Column());
            }
            for (Column col : _cols) {
                if (_lob && (typeCode == JavaTypes.STRING
                    || type == char[].class
                    || type == Character[].class)) {
                    col.setSize(-1);
                    col.setType(Types.CLOB);
                } else if (_lob)
                    col.setType(Types.BLOB);
                else {
                    switch (_temporal) {
                        case DATE:
                            col.setType(Types.DATE);
                            break;
                        case TIME:
                            col.setType(Types.TIME);
                            break;
                        case TIMESTAMP:
                            col.setType(Types.TIMESTAMP);
                            break;
                    }
                }
            }
        }

        if (_cols != null) {
            switch (fm.getDeclaredTypeCode()) {
                case JavaTypes.ARRAY:
                    Class<?> type = fm.getDeclaredType();
                    if (type == byte[].class || type == Byte[].class
                        || type == char[].class || type == Character[].class ) {
                        fm.getValueInfo().setColumns(_cols);
                        break;
                    }
                    // else no break
                case JavaTypes.COLLECTION:
                    if (!fm.getValue().isSerialized()) {
                        fm.getElementMapping().getValueInfo().setColumns(_cols);
                    } else  {
                        fm.getValueInfo().setColumns(_cols);
                    }
                    break;
                case JavaTypes.MAP:
                    fm.getElementMapping().getValueInfo().setColumns(_cols);
                    break;
                default:
                    fm.getValueInfo().setColumns(_cols);
            }
            if (_colTable != null)
                fm.getMappingInfo().setTableIdentifier(DBIdentifier.newTable(_colTable, delimit()));
            setUnique(fm);
        }
        clearColumnInfo();
    }

    /**
     * Set unique for field.
     */
    private void setUnique(FieldMapping fm) {
        setUnique(fm, _unique);
    }

    private void setUnique(FieldMapping fm, EnumSet<UniqueFlag> unique) {
        if (unique.size() == 2) // i.e. TRUE & FALSE
            getLog().warn(_loc.get("inconsist-col-attrs", fm));
        else if (unique.contains(UniqueFlag.TRUE))
            fm.getValueInfo().setUnique(new Unique());
    }

    /**
     * Clear field level column information.
     */
    private void clearColumnInfo() {
        _cols = null;
        _joinCols = null;
        _colTable = null;
        _lob = false;
        _temporal = null;
        _unique.clear();
    }

    /**
     * Parse attribute-override.
     */
    private boolean startAttributeOverride(Attributes attr) {
        _override = attr.getValue("name");
        return true;
    }

    /**
     * Set attribute override into proper mapping.
     */
    private void endAttributeOverride()
        throws SAXException {
        Object elem = currentElement();
        FieldMapping fm = null;
        if (elem instanceof ClassMapping)
            fm = getAttributeOverride((ClassMapping) elem);
        else {
            FieldMapping basefm = (FieldMapping) elem;

            fm = getAttributeOverrideForEmbeddable(basefm, _override, false);
            if (fm == null) {
                DeferredEmbeddableOverrides dfm =
                    getDeferredFieldMappingInfo(
                        AnnotationPersistenceMappingParser.
                        getEmbeddedClassType(basefm, _override),
                        basefm, _override, true);
                dfm._defCols = _cols;
                dfm._defTable = DBIdentifier.newTable(_colTable, delimit());
                dfm._attrName = _override;
                dfm._unique = _unique;
            }
        }
        if (fm != null && _cols != null) {
            fm.getValueInfo().setColumns(_cols);
            if (_colTable != null)
                fm.getMappingInfo().setTableIdentifier(DBIdentifier.newTable(_colTable, delimit()));
            setUnique(fm);
        }
        clearColumnInfo();
        _override = null;
    }

    /**
     * Return the proper override.
     */
    private FieldMapping getAttributeOverride(ClassMapping cm) {
        FieldMapping sup = (FieldMapping) cm.getDefinedSuperclassField
            (_override);
        if (sup == null)
            sup = (FieldMapping) cm.addDefinedSuperclassField(_override,
                Object.class, Object.class);
        return sup;
    }

    /**
     * Return the proper override.
     */
    private FieldMapping getAttributeOverrideForEmbeddable(FieldMapping fm,
        String attrName, boolean mustExist)
    throws SAXException {
        return AnnotationPersistenceMappingParser.getEmbeddedFieldMapping(fm,
            attrName, mustExist);
    }

    /**
     * Parse table.
     */
    private boolean startTable(Attributes attrs)
        throws SAXException {
        ClassMapping mapping = (ClassMapping) currentElement();
        if (mapping.isAbstract())
            throw new UserException(_loc.get("table-not-allowed", mapping));
        DBIdentifier table = toTableIdentifier(attrs.getValue("schema"),
            attrs.getValue("name"));
        if (!DBIdentifier.isNull(table))
            mapping.getMappingInfo().setTableIdentifier(table);
        return true;
    }

    /**
     * Parse join-table.
     */
    private boolean startJoinTable(Attributes attrs)
        throws SAXException {
        DBIdentifier sTable = toTableIdentifier(attrs.getValue("schema"),
            attrs.getValue("name"));
        if (!DBIdentifier.isNull(sTable)) {
            Object elem = currentElement();
            FieldMapping fm = null;
            if (elem instanceof FieldMapping) {
                fm = (FieldMapping) elem;
                if (_override != null) {
                    FieldMapping basefm = (FieldMapping) elem;
                    fm = getAttributeOverrideForEmbeddable(basefm,
                        _override, false);
                    if (fm == null) {
                        DeferredEmbeddableOverrides dfm =
                            getDeferredFieldMappingInfo(
                                AnnotationPersistenceMappingParser.
                                getEmbeddedClassType(basefm, _override),
