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

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.openjpa.enhance.PersistenceCapable;
import org.apache.openjpa.enhance.ReflectingPersistenceCapable;
import org.apache.openjpa.jdbc.identifier.DBIdentifier;
import org.apache.openjpa.jdbc.kernel.EagerFetchModes;
import org.apache.openjpa.jdbc.kernel.JDBCFetchConfiguration;
import org.apache.openjpa.jdbc.kernel.JDBCStore;
import org.apache.openjpa.jdbc.meta.ClassMapping;
import org.apache.openjpa.jdbc.meta.Embeddable;
import org.apache.openjpa.jdbc.meta.FieldMapping;
import org.apache.openjpa.jdbc.meta.FieldStrategy;
import org.apache.openjpa.jdbc.meta.Joinable;
import org.apache.openjpa.jdbc.meta.MappingInfo;
import org.apache.openjpa.jdbc.meta.ValueMapping;
import org.apache.openjpa.jdbc.meta.ValueMappingImpl;
import org.apache.openjpa.jdbc.meta.ValueMappingInfo;
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.PrimaryKey;
import org.apache.openjpa.jdbc.schema.Table;
import org.apache.openjpa.jdbc.sql.DBDictionary;
import org.apache.openjpa.jdbc.sql.Joins;
import org.apache.openjpa.jdbc.sql.Result;
import org.apache.openjpa.jdbc.sql.Row;
import org.apache.openjpa.jdbc.sql.RowManager;
import org.apache.openjpa.jdbc.sql.SQLBuffer;
import org.apache.openjpa.jdbc.sql.Select;
import org.apache.openjpa.jdbc.sql.SelectExecutor;
import org.apache.openjpa.jdbc.sql.Union;
import org.apache.openjpa.kernel.OpenJPAStateManager;
import org.apache.openjpa.kernel.StateManagerImpl;
import org.apache.openjpa.lib.log.Log;
import org.apache.openjpa.lib.util.Localizer;
import org.apache.openjpa.meta.ClassMetaData;
import org.apache.openjpa.meta.FieldMetaData;
import org.apache.openjpa.meta.JavaTypes;
import org.apache.openjpa.meta.MetaDataModes;
import org.apache.openjpa.util.ApplicationIds;
import org.apache.openjpa.util.ImplHelper;
import org.apache.openjpa.util.InternalException;
import org.apache.openjpa.util.MetaDataException;
import org.apache.openjpa.util.ObjectId;
import org.apache.openjpa.util.OpenJPAId;
import org.apache.openjpa.util.UnsupportedException;


/**
 * Mapping for a single-valued relation to another entity.
 *
 * @author Abe White
 * @since 0.4.0
 */
public class RelationFieldStrategy
    extends AbstractFieldStrategy
    implements Joinable, Embeddable {

    
    private static final long serialVersionUID = 1L;

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

    private Boolean _fkOid = null;

    @Override
    public void map(boolean adapt) {
        if (field.getTypeCode() != JavaTypes.PC || field.isEmbeddedPC())
            throw new MetaDataException(_loc.get("not-relation", field));

        field.getKeyMapping().getValueInfo().assertNoSchemaComponents
            (field.getKey(), !adapt);
        if (!field.isNonDefaultMappingUsingJoinTableStrategy())
            field.getElementMapping().getValueInfo().assertNoSchemaComponents
                (field.getElement(), !adapt);
        boolean criteria = field.getValueInfo().getUseClassCriteria();

        // check for named inverse
        FieldMapping mapped = field.getMappedByMapping();
        if (mapped != null) {
            field.getMappingInfo().assertNoSchemaComponents(field, !adapt);
            field.getValueInfo().assertNoSchemaComponents(field, !adapt);
            mapped.resolve(MetaDataModes.MODE_META | MetaDataModes.MODE_MAPPING);

            if (!mapped.isMapped() || mapped.isSerialized())
                throw new MetaDataException(_loc.get("mapped-by-unmapped",
                    field, mapped));

            if (mapped.getTypeCode() == JavaTypes.PC) {
                if (mapped.getJoinDirection() == ValueMapping.JOIN_FORWARD) {
                    field.setJoinDirection(ValueMapping.JOIN_INVERSE);
                    field.setColumns(mapped.getDefiningMapping().
                        getPrimaryKeyColumns());
                } else if (isTypeUnjoinedSubclass(mapped))
                    throw new MetaDataException(_loc.get
                        ("mapped-inverse-unjoined", field.getName(),
                            field.getDefiningMapping(), mapped));

                field.setForeignKey(mapped.getForeignKey
                    (field.getDefiningMapping()));
            } else if (mapped.getElement().getTypeCode() == JavaTypes.PC) {
                if (isTypeUnjoinedSubclass(mapped.getElementMapping()))
                    throw new MetaDataException(_loc.get
                        ("mapped-inverse-unjoined", field.getName(),
                            field.getDefiningMapping(), mapped));

                // warn the user about making the collection side the owner
                Log log = field.getRepository().getLog();
                if (log.isInfoEnabled())
                    log.info(_loc.get("coll-owner", field, mapped));
                field.setForeignKey(mapped.getElementMapping().
                    getForeignKey());
            } else
                throw new MetaDataException(_loc.get("not-inv-relation",
                    field, mapped));

            field.setUseClassCriteria(criteria);
            return;
        }

        // this is necessary to support openjpa 3 mappings, which didn't
        // differentiate between secondary table joins and relations built
        // around an inverse key: check to see if we're mapped as a secondary
        // table join but we're in the table of the related type, and if so
        // switch our join mapping info to our value mapping info
        DBIdentifier tableName = field.getMappingInfo().getTableIdentifier();
        Table table = field.getTypeMapping().getTable();
        ValueMappingInfo vinfo = field.getValueInfo();
        if (!DBIdentifier.isNull(tableName) && table != null
            && (tableName.equals(table.getIdentifier())
            || tableName.equals(table.getFullIdentifier()))) {
            vinfo.setJoinDirection(MappingInfo.JOIN_INVERSE);
            vinfo.setColumns(field.getMappingInfo().getColumns());
            field.getMappingInfo().setTableIdentifier(DBIdentifier.NULL);
            field.getMappingInfo().setColumns(null);
        }

        if (!field.isBiMTo1JT())
            field.mapJoin(adapt, false);
        if (field.getTypeMapping().isMapped()) {
            if (field.getMappedByIdValue() != null)
                setMappedByIdColumns();

            if (!field.isBiMTo1JT()) {
                ForeignKey fk = vinfo.getTypeJoin(field, field.getName(), true,
                    adapt);
                field.setForeignKey(fk);
            }
            field.setColumnIO(vinfo.getColumnIO());
            if (vinfo.getJoinDirection() == MappingInfo.JOIN_INVERSE)
                field.setJoinDirection(ValueMapping.JOIN_INVERSE);
        } else
            RelationStrategies.mapRelationToUnmappedPC(field, field.getName(),
                adapt);

        field.setUseClassCriteria(criteria);
        field.mapPrimaryKey(adapt);
        PrimaryKey pk = field.getTable().getPrimaryKey();
        if (field.isPrimaryKey()) {
            Column[] cols = field.getColumns();
            if (pk != null && (adapt || pk.isLogical()))
                for (Column col : cols) {
                    pk.addColumn(col);
                }
            for (Column col : cols) {
                field.getDefiningMapping().setJoinable(col, this);
            }
        }

        // map constraints after pk so we don't re-index / re-unique pk col
        field.mapConstraints(field.getName(), adapt);
    }

    /**
     * When there is MappedById annotation, the owner of the one-to-one/
     * many-to-one relationship will use its primary key to represent
     * foreign key relation. No need to create a separate foreign key
     * column.
     */
    private void setMappedByIdColumns() {
        ClassMetaData owner = field.getDefiningMetaData();
        FieldMetaData[] pks = owner.getPrimaryKeyFields();
        for (FieldMetaData pk : pks) {
            FieldMapping fm = (FieldMapping) pk;
            ValueMappingImpl val = (ValueMappingImpl) field.getValue();
            ValueMappingInfo info = val.getValueInfo();
            if (info.getColumns().size() == 0)
                info.setColumns(getMappedByIdColumns(fm));
        }
    }

    private List getMappedByIdColumns(FieldMapping pk) {
        ClassMetaData embeddedId = ((ValueMappingImpl)pk.getValue()).
            getEmbeddedMetaData();
        Column[] pkCols = null;
        List cols = new ArrayList();
        String mappedByIdValue = field.getMappedByIdValue();
        if (embeddedId != null) {
            FieldMetaData[] fmds = embeddedId.getFields();
            for (FieldMetaData fmd : fmds) {
                if ((fmd.getName().equals(mappedByIdValue)) ||
                        mappedByIdValue.length() == 0) {
                    if (fmd.getValue().getEmbeddedMetaData() != null) {
                        EmbedValueHandler.getEmbeddedIdCols(
                                (FieldMapping) fmd, cols);
                    }
                    else
                        EmbedValueHandler.getIdColumns(
                                (FieldMapping) fmd, cols);
                }
            }
            return cols;
        } else { // primary key is single-value
            Class pkType = pk.getDeclaredType();
            FieldMetaData[] pks = field.getValue().getDeclaredTypeMetaData().
                    getPrimaryKeyFields();
            if (pks.length != 1 || pks[0].getDeclaredType() != pkType)
                return Collections.EMPTY_LIST;
            pkCols = pk.getColumns();
            for (Column pkCol : pkCols) {
                cols.add(pkCol);
            }
            return cols;
        }
    }

    /**
     * Return whether our defining mapping is an unjoined subclass of
     * the type of the given value.
     */
    private boolean isTypeUnjoinedSubclass(ValueMapping mapped) {
        ClassMapping def = field.getDefiningMapping();
        for (; def != null; def = def.getJoinablePCSuperclassMapping())
            if (def == mapped.getTypeMapping())
                return false;
        return true;
    }

    @Override
    public void initialize() {
        field.setUsesIntermediate(true);

        ForeignKey fk = field.getForeignKey();
        if (fk == null)
            _fkOid = Boolean.TRUE;
        else if (field.getJoinDirection() != ValueMapping.JOIN_INVERSE)
            _fkOid = field.getTypeMapping().isForeignKeyObjectId(fk);
    }

    @Override
    public void insert(OpenJPAStateManager sm, JDBCStore store, RowManager rm)
        throws SQLException {
        if (field.getMappedBy() != null)
            return;
        Row row = null;
        OpenJPAStateManager rel = RelationStrategies.getStateManager
            (sm.fetchObjectField(field.getIndex()), store.getContext());
        // Checks if the field being inserted is a MapsId field and
        // the related object is using auto-assigned identity
        // If the above conditions are satisfied and the related instance has
        // already been inserted in the RowManger, then returns without further
        // processing
        if (sm instanceof StateManagerImpl) {
	        List<FieldMetaData> mappedByIdFields = ((StateManagerImpl)sm).getMappedByIdFields();
	        if (rel != null && ((ClassMapping)rel.getMetaData()).getTable().getAutoAssignedColumns().length > 0
	        &&  mappedByIdFields!= null && mappedByIdFields.contains(field)) {
	        	row = rm.getRow(((ClassMapping)rel.getMetaData()).getTable(), Row.ACTION_INSERT, rel, false);
	        	if (row != null) return;
	        }
        }
        if (field.getJoinDirection() == ValueMapping.JOIN_INVERSE)
            updateInverse(sm, rel, store, rm);
        else {
            if (row == null) row =  field.getRow(sm, store, rm, Row.ACTION_INSERT);
            if (row != null && !field.isBiMTo1JT()) {
                field.setForeignKey(row, rel);
                // this is for bi-directional maps, the key and value of the
                // map are stored in the table of the mapped-by entity
                setMapKey(sm, rel, store, row);
            }
        }
    }

    private void setMapKey(OpenJPAStateManager sm, OpenJPAStateManager rel,
        JDBCStore store, Row row) throws SQLException {
        if (rel == null)
            return;
        ClassMetaData meta = rel.getMetaData();
        FieldMapping mapField = getMapField(meta);

        // there is no bi-directional map field
        if (mapField == null)
            return;

        Map mapObj = (Map)rel.fetchObjectField(mapField.getIndex());
        Object keyObj = getMapKeyObj(mapObj, sm.getPersistenceCapable());
        ValueMapping key = mapField.getKeyMapping();
        if (!key.isEmbedded()) {
            if (keyObj instanceof PersistenceCapable) {
                OpenJPAStateManager keySm = RelationStrategies.
                    getStateManager(keyObj, store.getContext());
                // key is an entity
                ForeignKey fk = mapField.getKeyMapping().
                    getForeignKey();
                ColumnIO io = new ColumnIO();
                row.setForeignKey(fk, io, keySm);
            }
        } else {
            // key is an embeddable or basic type
            FieldStrategy strategy = mapField.getStrategy();
            if (strategy instanceof
                    HandlerRelationMapTableFieldStrategy) {
                HandlerRelationMapTableFieldStrategy strat =
                    (HandlerRelationMapTableFieldStrategy) strategy;
                Column[] kcols = strat.getKeyColumns((ClassMapping)meta);
                ColumnIO kio = strat.getKeyColumnIO();
                HandlerStrategies.set(key, keyObj, store, row, kcols,
                        kio, true);
            }
        }
    }

    private FieldMapping getMapField(ClassMetaData meta) {
        FieldMapping[] fields = ((ClassMapping)meta).getFieldMappings();
        for (FieldMapping fieldMapping : fields) {
            FieldMetaData mappedBy = fieldMapping.getMappedByMetaData();
            if (fieldMapping.getDeclaredTypeCode() == JavaTypes.MAP &&
                    mappedBy == field)
                return fieldMapping;
        }
        return null;
    }

    private Object getMapKeyObj(Map mapObj, Object value) {
        if (value instanceof ReflectingPersistenceCapable)
            value = ((ReflectingPersistenceCapable)value).getManagedInstance();

        Set<Map.Entry> entries = mapObj.entrySet();
        for (Map.Entry entry : entries) {
            if (entry.getValue() == value)
                return entry.getKey();
        }

        return null;
    }

    @Override
    public void update(OpenJPAStateManager sm, JDBCStore store, RowManager rm)
        throws SQLException {
        if (field.getMappedBy() != null)
            return;

        OpenJPAStateManager rel = RelationStrategies.getStateManager
            (sm.fetchObjectField(field.getIndex()), store.getContext());

        if (field.getJoinDirection() == ValueMapping.JOIN_INVERSE) {
            nullInverse(sm, rm);
            updateInverse(sm, rel, store, rm);
        } else {
            int action = (rel == null &&
                    field.isBidirectionalJoinTableMappingNonOwner()) ?
                    Row.ACTION_DELETE : Row.ACTION_UPDATE;
            Row row = field.getRow(sm, store, rm, action);
            if (row != null && !field.isBiMTo1JT()) {
                field.setForeignKey(row, rel);
                // this is for bi-directional maps, the key and value of the
                // map are stored in the table of the mapped-by entity
                setMapKey(sm, rel, store, row);
            }

            if (field.isBiMTo1JT()) { // also need to update the join table
                PersistenceCapable invPC = (PersistenceCapable)sm.fetchObject(
                    field.getBi_1ToM_JTField().getIndex());
                Row secondaryRow = null;
                if (invPC != null) {
                    secondaryRow = rm.getSecondaryRow(field.getBi1ToMJoinFK().getTable(),
                        Row.ACTION_INSERT);
                    secondaryRow.setForeignKey(field.getBi1ToMElemFK(), null, sm);
                    secondaryRow.setForeignKey(field.getBi1ToMJoinFK(), null,
                        RelationStrategies.getStateManager(invPC,
                        store.getContext()));
                    rm.flushSecondaryRow(secondaryRow);
                }
            }
        }
    }

    @Override
    public void delete(OpenJPAStateManager sm, JDBCStore store, RowManager rm)
        throws SQLException {
        if (field.getMappedBy() != null)
            return;

        if (field.getJoinDirection() == ValueMapping.JOIN_INVERSE) {
            if (sm.getLoaded().get(field.getIndex())) {
                OpenJPAStateManager rel = RelationStrategies.getStateManager(sm.
                    fetchObjectField(field.getIndex()), store.getContext());
                updateInverse(sm, rel, store, rm);
            } else
                nullInverse(sm, rm);
        } else {
            field.deleteRow(sm, store, rm);

            // if our foreign key has a delete action, we need to set the
            // related object so constraints can be evaluated
            Object lastRelPc = sm.fetchObjectField(field.getIndex());
            if( lastRelPc == null) {
            	lastRelPc = sm.fetchInitialField(field.getIndex());
            }
            OpenJPAStateManager rel = RelationStrategies.getStateManager
                (lastRelPc, store.getContext());
            if (rel != null) {
                ForeignKey fk = field.getForeignKey((ClassMapping)
                    rel.getMetaData());
                if (fk.getDeleteAction() == ForeignKey.ACTION_RESTRICT ||
                    fk.getDeleteAction() == ForeignKey.ACTION_CASCADE) {
                    Row row = field.getRow(sm, store, rm, Row.ACTION_DELETE);
                    row.setForeignKey(fk, null, rel);
                    // this is for bi-directional maps, the key and value of the
                    // map are stored in the table of the mapped-by entity
                    setMapKey(sm, rel, store, row);
                }
            }
        }
    }

    /**
     * Null inverse relations that reference the given object.
     */
    private void nullInverse(OpenJPAStateManager sm, RowManager rm)
        throws SQLException {
        if (field.getUseClassCriteria())
            return;

        ForeignKey fk = field.getForeignKey();
        ColumnIO io = field.getColumnIO();
        if (!io.isAnyUpdatable(fk, true))
            return;

        // null inverse if not already enforced by fk
        if (field.getIndependentTypeMappings().length != 1)
            throw RelationStrategies.uninversable(field);
        Row row = rm.getAllRows(fk.getTable(), Row.ACTION_UPDATE);
        row.setForeignKey(fk, io, null);
        row.whereForeignKey(fk, sm);
        rm.flushAllRows(row);
    }

    /**
     * This method updates the inverse columns of our relation
     * with the given object.
     */
    private void updateInverse(OpenJPAStateManager sm, OpenJPAStateManager rel,
        JDBCStore store, RowManager rm)
        throws SQLException {
        if (rel == null)
            return;

        ForeignKey fk = field.getForeignKey();
        ColumnIO io = field.getColumnIO();

        int action;
        if (rel.isNew() && !rel.isFlushed()) {
            if (sm.isDeleted() || !io.isAnyInsertable(fk, false))
                return;
            action = Row.ACTION_INSERT;
        } else if (rel.isDeleted()) {
            if (rel.isFlushed() || !sm.isDeleted())
                return;
            action = Row.ACTION_DELETE;
        } else {
            if (sm.isDeleted())
                sm = null;
            if (!io.isAnyUpdatable(fk, sm == null))
                return;
            action = Row.ACTION_UPDATE;
        }

        if (field.getIndependentTypeMappings().length != 1)
            throw RelationStrategies.uninversable(field);

        // get the row for the inverse object; the row might be in a secondary
        // table if there is a field controlling the foreign key
        Row row = null;
        FieldMapping[] invs = field.getInverseMappings();
        for (FieldMapping inv : invs) {
            if (inv.getMappedByMetaData() == field
                    && inv.getTypeCode() == JavaTypes.PC) {
                row = inv.getRow(rel, store, rm, action);
                break;
            }
        }
        ClassMapping relMapping = field.getTypeMapping();
        if (row == null)
            row = rm.getRow(relMapping.getTable(), action, rel, true);

        // if this is an update, this might be the only mod to the row, so
        // make sure the where condition is set
        if (action == Row.ACTION_UPDATE
            && row.getTable() == relMapping.getTable())
            row.wherePrimaryKey(rel);

        // update the inverse pointer with our oid value
        row.setForeignKey(fk, io, sm);
    }

    @Override
    public int supportsSelect(Select sel, int type, OpenJPAStateManager sm,
        JDBCStore store, JDBCFetchConfiguration fetch) {
        if (type == Select.TYPE_JOINLESS)
            return (field.getJoinDirection() != ValueMapping.JOIN_INVERSE
                && sel.isSelected(field.getTable())) ? 1 : 0;
        if (type == Select.TYPE_TWO_PART)
            return 1;

        // already cached?
        if (sm != null) {
            Object oid = sm.getIntermediate(field.getIndex());
            if (store.getContext().findCached(oid, null) != null)
                return 0;
        }

        ClassMapping[] clss = field.getIndependentTypeMappings();
        switch (type) {
            case Select.EAGER_PARALLEL:
                return clss.length;
            case Select.EAGER_OUTER:
                return (clss.length == 1 && store.getDBDictionary().canOuterJoin
                    (sel.getJoinSyntax(), field.getForeignKey(clss[0]))) ? 1 :
                    0;
            case Select.EAGER_INNER:
                return (clss.length == 1) ? 1 : 0;
            default:
                return 0;
        }
    }

    @Override
    public void selectEagerParallel(SelectExecutor sel,
        final OpenJPAStateManager sm, final JDBCStore store,
        final JDBCFetchConfiguration fetch, final int eagerMode) {
        final ClassMapping[] clss = field.getIndependentTypeMappings();
        if (!(sel instanceof Union))
            selectEagerParallel((Select) sel, clss[0], store, fetch, eagerMode);
        else {
            Union union = (Union) sel;
            if (fetch.getSubclassFetchMode (field.getTypeMapping())
                != EagerFetchModes.EAGER_JOIN)
                union.abortUnion();
            union.select(new Union.Selector() {
                @Override
                public void select(Select sel, int idx) {
                    selectEagerParallel(sel, clss[idx], store, fetch,
                        eagerMode);
                }
            });
        }
    }

    /**
     * Perform an eager parallel select.
     */
    private void selectEagerParallel(Select sel, ClassMapping cls,
        JDBCStore store, JDBCFetchConfiguration fetch, int eagerMode) {
        if (field.isBiMTo1JT())
            return;
        sel.selectPrimaryKey(field.getDefiningMapping());
        // set a variable name that does not conflict with any in the query;
        // using a variable guarantees that the selected data will use different
        // aliases and joins than any existing WHERE conditions on this field
        // that might otherwise limit the relations that match
        Joins joins = sel.newJoins().setVariable("*");
        eagerJoin(joins, cls, true);
        sel.select(cls, field.getSelectSubclasses(), store, fetch, eagerMode,
            joins);
    }

    @Override
    public void selectEagerJoin(Select sel, OpenJPAStateManager sm,
        JDBCStore store, JDBCFetchConfiguration fetch, int eagerMode) {
        if (field.isBiMTo1JT())
            return;

        // limit the eager mode to single on recursive eager fetching b/c
        // at this point the select has been modified and an attempt to
        // clone it for a to-many eager select can result in a clone that
        // produces invalid SQL
        ClassMapping cls = field.getIndependentTypeMappings()[0];
        boolean forceInner = fetch.hasFetchInnerJoin(field.getFullName(false)) ?
                true : false;
        sel.select(cls, field.getSelectSubclasses(), store, fetch,
            EagerFetchModes.EAGER_JOIN,
            eagerJoin(sel.newJoins(), cls, forceInner));
    }

    /**
     * Add the joins needed to select/load eager data.
     */
    private Joins eagerJoin(Joins joins, ClassMapping cls, boolean forceInner) {
        boolean inverse = field.getJoinDirection() == ValueMapping.JOIN_INVERSE;
        if (!inverse) {
            joins = join(joins, false);
            joins = setEmbeddedVariable(joins);
        }

        // and join into relation
        ForeignKey fk = field.getForeignKey(cls);
        if (!forceInner && field.getNullValue() != FieldMetaData.NULL_EXCEPTION)
            return joins.outerJoinRelation(field.getName(), fk, field.
                getTypeMapping(), field.getSelectSubclasses(), inverse, false);
        return joins.joinRelation(field.getName(), fk, field.getTypeMapping(),
            field.getSelectSubclasses(), inverse, false);
    }

    /**
     * If joining from an embedded owner, use variable to create a unique
     * alias in case owner contains other same-typed embedded relations.
     */
    private Joins setEmbeddedVariable(Joins joins) {
        if (field.getDefiningMetaData().getEmbeddingMetaData() == null)
            return joins;
        return joins.setVariable(field.getDefiningMetaData().
            getEmbeddingMetaData().getFieldMetaData().getName());
    }

    @Override
    public int select(Select sel, OpenJPAStateManager sm, JDBCStore store,
        JDBCFetchConfiguration fetch, int eagerMode) {
        if (field.getJoinDirection() == ValueMapping.JOIN_INVERSE)
            return -1;
        // already cached oid?
        if (sm != null && sm.getIntermediate(field.getIndex()) != null)
            return -1;
        if (!Boolean.TRUE.equals(_fkOid))
            return -1;
        sel.select(field.getColumns(), field.join(sel));
        return 0;
    }

    @Override
    public Object loadEagerParallel(OpenJPAStateManager sm, JDBCStore store,
        JDBCFetchConfiguration fetch, Object res)
        throws SQLException {
        // process batched results if we haven't already
        Map rels;
        if (res instanceof Result)
            rels = processEagerParallelResult(sm, store, fetch, (Result) res);
        else
            rels = (Map) res;

        // store object for this oid in instance
        sm.storeObject(field.getIndex(), rels.remove(sm.getObjectId()));
        return rels;
    }

    /**
     * Process the given batched result.
     */
    private Map processEagerParallelResult(OpenJPAStateManager sm,
        JDBCStore store, JDBCFetchConfiguration fetch, Result res)
        throws SQLException {
        // do same joins as for load
        //### cheat: we know typical result joins only care about the relation
        //### path; thus we can ignore different mappings
        ClassMapping[] clss = field.getIndependentTypeMappings();
        Joins joins = res.newJoins().setVariable("*");
        eagerJoin(joins, clss[0], true);

        Map rels = new HashMap();
        ClassMapping owner = field.getDefiningMapping();
        ClassMapping cls;
        Object oid;
        while (res.next()) {
            cls = res.getBaseMapping();
            if (cls == null)
                cls = clss[0];
            oid = owner.getObjectId(store, res, null, true, null);
            rels.put(oid, res.load(cls, store, fetch, joins));
        }
        res.close();

        return rels;
    }

    @Override
    public void loadEagerJoin(OpenJPAStateManager sm, JDBCStore store,
        JDBCFetchConfiguration fetch, Result res)
        throws SQLException {
        if (field.isBiMTo1JT())
            return;
        ClassMapping cls = field.getIndependentTypeMappings()[0];

        // for inverseEager field
        FieldMapping mappedByFieldMapping = field.getMappedByMapping();
        PersistenceCapable mappedByValue = null;

        if (mappedByFieldMapping != null) {
        	ValueMapping val = mappedByFieldMapping.getValueMapping();
        	ClassMetaData decMeta = val.getTypeMetaData();
            // eager loading a child from its toOne parent and
            // the parent has @OneToOne(mappedBy="parent") child relation.
            // By saving the mapped-by info in 'res' is to
            // avoid unneeded SQL pushdown that would otherwise gets
            // generated.
            if (decMeta != null && !sm.isEmbedded()) {
        	    mappedByValue = sm.getPersistenceCapable();
        	    res.setMappedByFieldMapping(mappedByFieldMapping);
        	    res.setMappedByValue(mappedByValue);
        	}
        }

        boolean isLocked = res.isLocking();
        try {
            if (store.getLockManager() != null)
                res.setLocking(store.getLockManager().skipRelationFieldLock());
            sm.storeObject(field.getIndex(), res.load(cls, store, fetch,
                    eagerJoin(res.newJoins(), cls, false)));
        } finally {
            res.setLocking(isLocked);
        }

        // reset mapped by is needed for OneToOne bidirectional relations
        // having a mapped-by parent to correctly set the parent-child
        // relation.
        res.setMappedByFieldMapping(null);
        res.setMappedByValue(null);
    }

    @Override
    public void load(OpenJPAStateManager sm, JDBCStore store,
        JDBCFetchConfiguration fetch, Result res)
        throws SQLException {
        if (field.getJoinDirection() == ValueMapping.JOIN_INVERSE)
            return;
        // cached oid?
        if (sm != null && sm.getIntermediate(field.getIndex()) != null)
            return;
        if (!Boolean.TRUE.equals(_fkOid))
            return;
        if (!res.containsAll(field.getColumns()))
            return;

        // get the related object's oid
        ClassMapping relMapping = field.getTypeMapping();
        Object oid = null;
        if (relMapping.isMapped() && !field.isBiMTo1JT()) {
            oid = relMapping.getObjectId(store, res, field.getForeignKey(),
                    field.getPolymorphic() != ValueMapping.POLY_FALSE, null);
        } else {
            Column[] cols = field.getColumns();
            if (relMapping.getIdentityType() == ClassMetaData.ID_DATASTORE) {
                long id = res.getLong(cols[0]);
                if (!res.wasNull())
                    oid = store.newDataStoreId(id, relMapping, true);
            } else {
                // application id
                if (cols.length == 1) {
                    Object val = res.getObject(cols[0], null, null);
                    if (val != null)
                        oid = ApplicationIds.fromPKValues(new Object[]{ val },
                            relMapping);
                } else {
                    Object[] vals = new Object[cols.length];
                    for (int i = 0; i < cols.length; i++) {
                        vals[i] = res.getObject(cols[i], null, null);
                        if (vals[i] == null)
                            break;
                        if (i == cols.length - 1)
