# TablePerClassPolymorphismBuilder.java

    package org.codefilarete.stalactite.engine.configurer.polymorphism;

    import java.util.HashMap;
    import java.util.Map;
    import java.util.Set;
    import java.util.function.BiFunction;

    import org.codefilarete.reflection.ReversibleAccessor;
    import org.codefilarete.reflection.ValueAccessPoint;
    import org.codefilarete.reflection.ValueAccessPointMap;
    import org.codefilarete.reflection.ValueAccessPointSet;
    import org.codefilarete.stalactite.dsl.PolymorphismPolicy;
    import org.codefilarete.stalactite.dsl.PolymorphismPolicy.TablePerClassPolymorphism;
    import org.codefilarete.stalactite.dsl.idpolicy.GeneratedKeysPolicy;
    import org.codefilarete.stalactite.dsl.subentity.SubEntityMappingConfiguration;
    import org.codefilarete.stalactite.engine.configurer.AbstractIdentification;
    import org.codefilarete.stalactite.engine.configurer.AbstractIdentification.SingleColumnIdentification;
    import org.codefilarete.stalactite.engine.configurer.builder.embeddable.EmbeddableMappingBuilder;
    import org.codefilarete.stalactite.engine.configurer.builder.embeddable.EmbeddableMapping;
    import org.codefilarete.stalactite.engine.configurer.NamingConfiguration;
    import org.codefilarete.stalactite.engine.configurer.builder.PersisterBuilderContext;
    import org.codefilarete.stalactite.engine.configurer.builder.InheritanceMappingStep.Mapping;
    import org.codefilarete.stalactite.engine.configurer.builder.MainPersisterStep;
    import org.codefilarete.stalactite.engine.configurer.builder.PrimaryKeyPropagationStep;
    import org.codefilarete.stalactite.engine.runtime.AbstractPolymorphismPersister;
    import org.codefilarete.stalactite.engine.runtime.ConfiguredRelationalPersister;
    import org.codefilarete.stalactite.engine.runtime.SimpleRelationalEntityPersister;
    import org.codefilarete.stalactite.engine.runtime.tableperclass.TablePerClassPolymorphismPersister;
    import org.codefilarete.stalactite.mapping.DefaultEntityMapping;
    import org.codefilarete.stalactite.sql.ConnectionConfiguration;
    import org.codefilarete.stalactite.sql.Dialect;
    import org.codefilarete.stalactite.sql.ddl.structure.Column;
    import org.codefilarete.stalactite.sql.ddl.structure.Table;
    import org.codefilarete.stalactite.sql.statement.binder.ColumnBinderRegistry;
    import org.codefilarete.tool.Reflections;
    import org.codefilarete.tool.collection.Arrays;
    import org.codefilarete.tool.exception.NotImplementedException;
    import org.codefilarete.tool.function.Converter;

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

    /**
     * @author Guillaume Mary
     */
    class TablePerClassPolymorphismBuilder<C, I, T extends Table<T>> extends AbstractPolymorphicPersisterBuilder<C, I, T> {

    	/**
    	 * Creates equivalent {@link Column}s of those given in <code>propertyToColumn</code> onto given target table.
    	 * May do nothing if columns already exist.
    	 *
    	 * @param propertyToColumn mapping between some property accessors and {@link Column}s from default target {@link Table}
    	 * @param targetTable the table on which columns must be added
    	 * @return a mapping between properties of given {@link Map} keys and their column in given {@link Table}
    	 */
    	public static <C, T extends Table<T>, A extends ValueAccessPoint<C>> Map<A, Column<T, Object>>
    	projectColumns(Map<? extends A, ? extends Column<?, Object>> propertyToColumn,
    				   T targetTable,
    				   BiFunction<? super A, Column<?, Object>, String> columnNameSupplier) {
    		Map<A, Column<T, Object>> localResult = new HashMap<>();
    		propertyToColumn.forEach((accessor, column) -> {
    			Column<T, Object> projectedColumn = targetTable.addColumn(columnNameSupplier.apply(accessor, column), column.getJavaType(), column.getSize(), column.isNullable());
    			projectedColumn.setAutoGenerated(column.isAutoGenerated());
    			localResult.put(accessor, projectedColumn);
    		});
    		return localResult;
    	}

    	private final Map<ReversibleAccessor<C, Object>, Column<T, Object>> mainMapping;
    	private final Map<ReversibleAccessor<C, Object>, Column<T, Object>> mainReadonlyMapping;
    	private final ValueAccessPointMap<C, Converter<Object, Object>> mainReadConverters;
    	private final ValueAccessPointMap<C, Converter<Object, Object>> mainWriteConverters;

    	TablePerClassPolymorphismBuilder(TablePerClassPolymorphism<C> polymorphismPolicy,
    									 AbstractIdentification<C, I> identification,
    									 ConfiguredRelationalPersister<C, I> mainPersister,
    									 Map<? extends ReversibleAccessor<C, Object>, Column<T, Object>> mainMapping,
    									 Map<? extends ReversibleAccessor<C, Object>, ? extends Column<T, Object>> mainReadonlyMapping,
    									 ValueAccessPointMap<C, ? extends Converter<Object, Object>> mainReadConverters,
    									 ValueAccessPointMap<C, ? extends Converter<Object, Object>> mainWriteConverters,
    									 ColumnBinderRegistry columnBinderRegistry,
    									 NamingConfiguration namingConfiguration,
    									 PersisterBuilderContext persisterBuilderContext) {
    		super(polymorphismPolicy, identification, mainPersister, columnBinderRegistry, namingConfiguration, persisterBuilderContext);
    		this.mainMapping = (Map<ReversibleAccessor<C, Object>, Column<T, Object>>) mainMapping;
    		this.mainReadonlyMapping = (Map<ReversibleAccessor<C, Object>, Column<T, Object>>) mainReadonlyMapping;
    		this.mainReadConverters = (ValueAccessPointMap<C, Converter<Object, Object>>) mainReadConverters;
    		this.mainWriteConverters = (ValueAccessPointMap<C, Converter<Object, Object>>) mainWriteConverters;
    	}

    	@Override
    	public AbstractPolymorphismPersister<C, I> build(Dialect dialect, ConnectionConfiguration connectionConfiguration) {
    		if (this.identification instanceof AbstractIdentification.SingleColumnIdentification && ((SingleColumnIdentification<C, I>) this.identification).getIdentifierPolicy() instanceof GeneratedKeysPolicy) {
    			throw new UnsupportedOperationException("Table-per-class polymorphism is not compatible with auto-incremented primary key");
    		}
    		Map<Class<C>, ConfiguredRelationalPersister<C, I>> persisterPerSubclass = collectSubClassPersister(dialect, connectionConfiguration);

    		// Note that registering the cascades to sub-persisters must be done BEFORE the creation of the main persister to make it have all
    		// entities joins and let it build a consistent entity graph load; without it, we miss sub-relations when loading main entities
    		registerSubEntitiesRelations(persisterPerSubclass, dialect, connectionConfiguration);

    		TablePerClassPolymorphismPersister<C, I, T> result = new TablePerClassPolymorphismPersister<>(
    				mainPersister, persisterPerSubclass, connectionConfiguration.getConnectionProvider(), dialect);

    		// we propagate shadow columns through TablePerClassPolymorphismPersister.getMapping() because it has a
    		// mechanism that projects them to sub-persisters (but columns were added in buildSubclassPersister()
    		// which is not very consistent)
    		((DefaultEntityMapping<C, I, T>) mainPersister.getMapping()).getShadowColumnsForInsert().forEach(columnProvider -> {
    			result.getMapping().addShadowColumnInsert(columnProvider);
    		});
    		((DefaultEntityMapping<C, I, T>) mainPersister.getMapping()).getShadowColumnsForUpdate().forEach(columnProvider -> {
    			result.getMapping().addShadowColumnUpdate(columnProvider);
    		});
    		return result;
    	}

    	private <D extends C> Map<Class<D>, ConfiguredRelationalPersister<D, I>> collectSubClassPersister(Dialect dialect, ConnectionConfiguration connectionConfiguration) {
    		Map<Class<D>, ConfiguredRelationalPersister<D, I>> persisterPerSubclass = new HashMap<>();

    		for (SubEntityMappingConfiguration<D> subConfiguration : ((Set<SubEntityMappingConfiguration<D>>) (Set) polymorphismPolicy.getSubClasses())) {
    			SimpleRelationalEntityPersister<D, I, ?> subclassPersister = buildSubclassPersister(dialect, connectionConfiguration, subConfiguration);
    			persisterPerSubclass.put(subConfiguration.getEntityType(), subclassPersister);
    		}

    		return persisterPerSubclass;
    	}

    	private <D extends C, SUBTABLE extends Table<SUBTABLE>> SimpleRelationalEntityPersister<D, I, SUBTABLE>
    	buildSubclassPersister(Dialect dialect,
    						   ConnectionConfiguration connectionConfiguration,
    						   SubEntityMappingConfiguration<D> subConfiguration) {
    		// first we'll use table of columns defined in embedded override
    		// then the one defined by inheritance
    		// if both are null we'll create a new one
    		Table tableDefinedByInheritanceConfiguration = ((TablePerClassPolymorphism<D>) polymorphismPolicy).giveTable(subConfiguration);
    		Table tableDefinedByColumnOverride = EmbeddableMappingBuilder.giveTargetTable(subConfiguration.getPropertiesMapping(), tableDefinedByInheritanceConfiguration);

    		SUBTABLE subTable = (SUBTABLE) nullable(tableDefinedByColumnOverride)
    				.elseSet(tableDefinedByInheritanceConfiguration)
    				.getOr(() -> new Table(namingConfiguration.getTableNamingStrategy().giveName(subConfiguration.getEntityType())));

    		EmbeddableMappingBuilder<D, SUBTABLE> embeddableMappingBuilder = new EmbeddableMappingBuilder<>(subConfiguration.getPropertiesMapping(), subTable,
    				this.columnBinderRegistry, this.namingConfiguration.getColumnNamingStrategy(), this.namingConfiguration.getIndexNamingStrategy());
    		EmbeddableMapping<D, SUBTABLE> embeddableMapping = embeddableMappingBuilder.build();
    		Map<ReversibleAccessor<D, Object>, Column<SUBTABLE, Object>> subEntityPropertiesMapping = embeddableMapping.getMapping();
    		Map<ReversibleAccessor<D, Object>, Column<SUBTABLE, Object>> subEntityReadonlyPropertiesMapping = embeddableMapping.getReadonlyMapping();
    		ValueAccessPointMap<D, Converter<Object, Object>> subEntityPropertiesReadConverters = embeddableMapping.getReadConverters();
    		ValueAccessPointMap<D, Converter<Object, Object>> subEntityPropertiesWriteConverters = embeddableMapping.getWriteConverters();
    		// in table-per-class polymorphism, main properties must be transferred to sub-entities ones, because CRUD operations are dispatched to them
    		// by a proxy and main persister is not so much used
    		addPrimaryKey(subTable);
    		Map<ReversibleAccessor<C, Object>, Column<SUBTABLE, Object>> projectedMainMapping = projectColumns(mainMapping, subTable, (accessor, c) -> c.getName());
    		subEntityPropertiesMapping.putAll((Map) projectedMainMapping);
    		Map<ReversibleAccessor<C, Object>, Column<SUBTABLE, Object>> projectedMainReadonlyMapping = projectColumns(mainReadonlyMapping, subTable, (accessor, c) -> c.getName());
    		subEntityReadonlyPropertiesMapping.putAll((Map) projectedMainReadonlyMapping);
    		subEntityPropertiesReadConverters.putAll((Map) mainReadConverters);
    		subEntityPropertiesWriteConverters.putAll((Map) mainWriteConverters);
    		DefaultEntityMapping<D, I, SUBTABLE> entityMapping = MainPersisterStep.createEntityMapping(
    				true,    // given Identification (which is parent one) contains identifier policy
    				subTable,
    				subEntityPropertiesMapping,
    				subEntityReadonlyPropertiesMapping,
    				subEntityPropertiesReadConverters,
    				subEntityPropertiesWriteConverters,
    				new ValueAccessPointSet<>(),    // TODO: implement properties set by constructor feature in table-per-class polymorphism
    				(AbstractIdentification<D, I>) identification,
    				subConfiguration.getPropertiesMapping().getBeanType(),
    				null);

    		((DefaultEntityMapping<C, I, T>) mainPersister.getMapping()).getShadowColumnsForInsert().forEach(columnProvider -> {
    			columnProvider.getColumns().forEach(c -> {
    				subTable.addColumn(c.getName(), c.getJavaType(), c.getSize(), c.isNullable());
    			});
    		});
    		((DefaultEntityMapping<C, I, T>) mainPersister.getMapping()).getShadowColumnsForUpdate().forEach(columnProvider -> {
    			columnProvider.getColumns().forEach(c -> {
    				subTable.addColumn(c.getName(), c.getJavaType(), c.getSize(), c.isNullable());
    			});
    		});

    		return new SimpleRelationalEntityPersister<>(entityMapping, dialect, connectionConfiguration);
    	}

    	private void addPrimaryKey(Table table) {
    		PrimaryKeyPropagationStep.propagatePrimaryKey(this.mainPersister.getMapping().getTargetTable().getPrimaryKey(), Arrays.asSet(table));
    	}

    	@Override
    	void assertSubPolymorphismIsSupported(PolymorphismPolicy<? extends C> subPolymorphismPolicy) {
    		// Everything else than joined-tables and single-table is not implemented (meaning table-per-class)
    		// Written as a negative condition to explicitly say what we support
    		if (subPolymorphismPolicy instanceof PolymorphismPolicy.TablePerClassPolymorphism) {
    			throw new NotImplementedException("Combining table-per-class polymorphism policy with " + Reflections.toString(subPolymorphismPolicy.getClass()));
    		}
    	}
    }