# TablePerClassPolymorphicRelationJoinNode.java

    package org.codefilarete.stalactite.engine.runtime.load;

    import java.util.HashSet;
    import java.util.IdentityHashMap;
    import java.util.Set;
    import javax.annotation.Nullable;

    import org.codefilarete.reflection.PropertyAccessPoint;
    import org.codefilarete.stalactite.engine.runtime.load.EntityJoinTree.JoinType;
    import org.codefilarete.stalactite.engine.runtime.load.EntityTreeInflater.RelationIdentifier;
    import org.codefilarete.stalactite.engine.runtime.load.EntityTreeInflater.TreeInflationContext;
    import org.codefilarete.stalactite.engine.runtime.load.JoinRowConsumer.ForkJoinRowConsumer;
    import org.codefilarete.stalactite.query.api.JoinLink;
    import org.codefilarete.stalactite.query.api.QueryStatement;
    import org.codefilarete.stalactite.query.api.QueryStatement.PseudoTable;
    import org.codefilarete.stalactite.query.api.Selectable;
    import org.codefilarete.stalactite.query.model.Union;
    import org.codefilarete.stalactite.sql.ddl.structure.Column;
    import org.codefilarete.stalactite.sql.ddl.structure.Key;
    import org.codefilarete.stalactite.sql.ddl.structure.Table;
    import org.codefilarete.stalactite.sql.result.BeanRelationFixer;
    import org.codefilarete.stalactite.sql.result.ColumnedRow;
    import org.codefilarete.tool.Reflections;
    import org.codefilarete.tool.collection.Iterables;

    import static org.codefilarete.tool.Nullable.nullable;

    /**
     * Particular {@link JoinNode} made to handle relation from an entity to a collection of some another polymorphic one. Actually relation doesn't
     * make the complexity of that class: polymorphic entity instantiation is the core focus of it. Here are the hot spots:
     * identifier is given by the subclass which find its id in the row (see {@link TablePerClassPolymorphicRelationJoinRowConsumer#giveIdentifier}),
     * and while doing it, it remembers which consumer made it. Then while instantiating entity we invoke it to get right entity type (parent mapping
     * would only give parent entity, which, out of being a wrong approach, can be an abstract type). Then instance is filled up with parent properties
     * by calling merging method (see line 98).
     * Finally, {@link TablePerClassPolymorphicRelationJoinRowConsumer} must extend {@link ForkJoinRowConsumer} to give next branch to be consumed by
     * {@link EntityTreeInflater} to avoid "dead" branch to be read : we give it according to the consumer which found the identifier.
     *
     * @author Guillaume Mary
     */
    public class TablePerClassPolymorphicRelationJoinNode<C, T1 extends Table<T1>, JOINCOLTYPE, I> extends RelationJoinNode<C, T1, PseudoTable, JOINCOLTYPE, I> {

    	private static IdentityHashMap<JoinLink<?, ?>, JoinLink<?, ?>> collectColumnClones(Key<?, ?> rightJoinLink, Set<? extends JoinLink<?, ?>> columnsToSelect) {
    		IdentityHashMap<JoinLink<?, ?>, JoinLink<?, ?>> result = new IdentityHashMap<>();
    		rightJoinLink.getTable().getColumns().forEach(column -> {
    			result.put((JoinLink<?, ?>) column, (JoinLink<?, ?>) column);
    		});
    		columnsToSelect.forEach(column -> {
    			result.put(column, column);
    		});
    		return result;
    	}

    	private final Set<SubPersisterAndDiscriminator<? extends C>> subPersisters = new HashSet<>();

    	private final PseudoTable pseudoTable;
    	private final Union.PseudoColumn<Integer> discriminatorColumn;
    	private final Key<PseudoTable, JOINCOLTYPE> pseudoRightJoinLink;

    	public TablePerClassPolymorphicRelationJoinNode(JoinNode<?, T1> parent,
    													Union subPersistersUnion,
    													// First generic type is unknown because it's the relation owning type.
    													// Second generic type should be C, or a type containing C (like a Collection<C>or even Map<K, C>) which makes it
    													// impossible to implement; thus, we have to use the ? wilcard.
    													PropertyAccessPoint<?, ?> propertyAccessor,
    													Key<T1, JOINCOLTYPE> leftJoinColumn,
    													Key<?, JOINCOLTYPE> rightJoinColumn,
    													JoinType joinType,
    													Set<? extends JoinLink<?, ?>> columnsToSelect,
    													@Nullable String tableAlias,
    													EntityInflater<C, I> entityInflater,
    													BeanRelationFixer<Object, C> beanRelationFixer,
    													Union.PseudoColumn<Integer> discriminatorColumn) {
    		super(parent, propertyAccessor, leftJoinColumn, (Key) rightJoinColumn, joinType, columnsToSelect, tableAlias, entityInflater, beanRelationFixer, null,
    				collectColumnClones(rightJoinColumn, columnsToSelect));
    		this.pseudoTable = subPersistersUnion.asPseudoTable(getTableAlias());
    		super.getOriginalColumnsToLocalOnes().clear();
    		super.getOriginalColumnsToLocalOnes().putAll(collectColumnClones(rightJoinColumn, pseudoTable.getColumns()));
    		this.discriminatorColumn = discriminatorColumn;
    		// rebuilding the right join link to make it match the union table, because the given one has columns from the abstract (parent) entity
    		// which doesn't have a matching / existing table
    		Key.KeyBuilder<PseudoTable, JOINCOLTYPE> pseudoRightJoinLinkBuilder = Key.from(pseudoTable);
    		rightJoinColumn.getColumns()
    				.forEach(column -> {
    					QueryStatement.PseudoColumn pseudoColumn1 = Iterables.find(pseudoTable.getColumns(), pseudoColumn -> pseudoColumn.getExpression().equals(column.getExpression()));
    					pseudoRightJoinLinkBuilder.addColumn(pseudoColumn1);
    				});
    		this.pseudoRightJoinLink = pseudoRightJoinLinkBuilder.build();
    	}

    	@Override
    	public PseudoTable getRightTable() {
    		return pseudoTable;
    	}

    	@Override
    	public Key<PseudoTable, JOINCOLTYPE> getRightJoinLink() {
    		return pseudoRightJoinLink;
    	}

    	/**
    	 * Overridden, else it returns the ones given at construction time which are from Union, meaning getOwner() is not
    	 * {@link PseudoTable} which, out of throwing a {@link ClassCastException}, avoid to give access to correct
    	 * {@link Column}
    	 */
    	@Override
    	public Set<Selectable<?>> getColumnsToSelect() {
    		return (Set) pseudoTable.getColumns();
    	}

    	@Override
    	public TablePerClassPolymorphicRelationJoinRowConsumer toConsumer(JoinNode<C, PseudoTable> joinNode) {
    		return new TablePerClassPolymorphicRelationJoinRowConsumer(joinNode, discriminatorColumn, getConsumptionListener());
    	}

    	public <D extends C> void addSubPersisterJoin(PolymorphicMergeJoinRowConsumer<D, I> subPersisterJoin, int discriminatorValue) {
    		this.subPersisters.add(new SubPersisterAndDiscriminator<>(subPersisterJoin, discriminatorValue));
    	}

    	private class SubPersisterAndDiscriminator<D extends C> {

    		private final PolymorphicMergeJoinRowConsumer<D, I> subPersisterJoin;
    		private final int discriminatorValue;

    		public SubPersisterAndDiscriminator(PolymorphicMergeJoinRowConsumer<D, I> subPersisterJoin, int discriminatorValue) {
    			this.subPersisterJoin = subPersisterJoin;
    			this.discriminatorValue = discriminatorValue;
    		}
    	}

    	public class TablePerClassPolymorphicRelationJoinRowConsumer implements RelationJoinRowConsumer<C, I>, ForkJoinRowConsumer {

    		private final ThreadLocal<RowIdentifier<? extends C>> currentlyFoundConsumer = new ThreadLocal<>();

    		private final JoinNode<C, PseudoTable> joinNode;

    		/** Optional listener of ResultSet decoding */
    		@Nullable
    		private final EntityTreeJoinNodeConsumptionListener<C> consumptionListener;
    		private final Union.PseudoColumn<Integer> discriminatorColumn;

    		private TablePerClassPolymorphicRelationJoinRowConsumer(JoinNode<C, PseudoTable> joinNode,
    																Union.PseudoColumn<Integer> discriminatorColumn,
    																@Nullable EntityTreeJoinNodeConsumptionListener<C> consumptionListener) {
    			this.joinNode = joinNode;
    			this.consumptionListener = consumptionListener;
    			this.discriminatorColumn = discriminatorColumn;
    		}

    		@Override
    		public JoinNode<C, ?> getNode() {
    			return joinNode;
    		}

    		@Override
    		public EntityReference<C, I> applyRelatedEntity(EntityReference<?, ?> parentJoinEntity, ColumnedRow row, TreeInflationContext context) {
    			RowIdentifier<C> rowIdentifier = giveIdentifier(row);
    			currentlyFoundConsumer.set(rowIdentifier);
    			if (rowIdentifier == null) {
    				return null;
    			} else {
    				I rightIdentifier = rowIdentifier.entityIdentifier;
    				// we avoid treating twice same relation, overall to avoid adding twice same instance to a collection (one-to-many list cases)
    				// in case of multiple collections in ResultSet because it creates similar data (through cross join) which are treated as many as
    				// collections cross with each other. This also works for one-to-one relations but produces no bugs. It can also be seen as a performance
    				// enhancement even if it hasn't been measured.
    				RelationIdentifier eventuallyApplied = new RelationIdentifier(
    						parentJoinEntity.getIdentifier(),
    						getEntityType(),
    						rightIdentifier,
    						this);
    				// primary key null means no entity => nothing to do
    				if (rightIdentifier != null && context.isTreatedOrAppend(eventuallyApplied)) {
    					C rightEntity = context.giveEntityFromCache(getEntityType(), rightIdentifier, () -> rowIdentifier.rowConsumer.transform(row));
    					getBeanRelationFixer().apply(parentJoinEntity.getEntity(), rightEntity);
    					if (this.consumptionListener != null) {
    						this.consumptionListener.onNodeConsumption(rightEntity, row);
    					}
    					return new EntityReference<>(rightEntity, rightIdentifier);
    				} else {
    					return null;
    				}
    			}
    		}

    		@Nullable
    		/* Optimized, from 530 000 nanos to 65 000 nanos at 1st exec, from 40 000 nanos to 12 000 nanos on usual run */
    		private <D extends C> RowIdentifier<D> giveIdentifier(ColumnedRow row) {
    			// @Optimized : use for & return instead of stream().map().filter(notNull).findFirst()
    			Integer discriminatorValue = row.get(discriminatorColumn);
    			if (discriminatorValue != null) {
    				SubPersisterAndDiscriminator<D> discriminatorConsumer = (SubPersisterAndDiscriminator) Iterables.find(subPersisters, o -> o.discriminatorValue == discriminatorValue);
    				if (discriminatorConsumer != null) {
    					I identifier = discriminatorConsumer.subPersisterJoin.giveIdentifier(row);
    					if (identifier != null) {
    						return new RowIdentifier<>(identifier, discriminatorConsumer.subPersisterJoin);
    					}
    				}
    			}
    			return null;
    		}

    		@Override
    		public JoinRowConsumer giveNextConsumer() {
    			return nullable(currentlyFoundConsumer.get()).map(rowIdentifier -> rowIdentifier.rowConsumer).get();
    		}

    		@Override
    		public void afterRowConsumption(TreeInflationContext context) {
    			currentlyFoundConsumer.remove();
    		}

    		private class RowIdentifier<D extends C> {

    			private final I entityIdentifier;
    			private final PolymorphicMergeJoinRowConsumer<D, I> rowConsumer;

    			private RowIdentifier(I entityIdentifier, PolymorphicMergeJoinRowConsumer<D, I> rowConsumer) {
    				this.entityIdentifier = entityIdentifier;
    				this.rowConsumer = rowConsumer;
    			}
    		}

    		/**
    		 * Implemented for debug. DO NOT RELY ON IT for anything else.
    		 */
    		@Override
    		public String toString() {
    			return Reflections.toString(this.getClass());
    		}
    	}
    }