# PrimaryKeyStep.java

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

    import java.util.Map;

    import org.codefilarete.reflection.AccessorDefinition;
    import org.codefilarete.reflection.ReversibleAccessor;
    import org.codefilarete.stalactite.dsl.entity.EntityMappingConfiguration;
    import org.codefilarete.stalactite.dsl.entity.EntityMappingConfiguration.KeyMapping;
    import org.codefilarete.stalactite.dsl.entity.EntityMappingConfiguration.SingleKeyMapping;
    import org.codefilarete.stalactite.dsl.idpolicy.GeneratedKeysPolicy;
    import org.codefilarete.stalactite.dsl.key.CompositeKeyMappingConfiguration;
    import org.codefilarete.stalactite.dsl.naming.ColumnNamingStrategy;
    import org.codefilarete.stalactite.dsl.naming.UniqueConstraintNamingStrategy;
    import org.codefilarete.stalactite.engine.configurer.AbstractIdentification;
    import org.codefilarete.stalactite.engine.configurer.AbstractIdentification.CompositeKeyIdentification;
    import org.codefilarete.stalactite.engine.configurer.AbstractIdentification.SingleColumnIdentification;
    import org.codefilarete.stalactite.engine.configurer.builder.embeddable.EmbeddableMapping;
    import org.codefilarete.stalactite.engine.configurer.builder.embeddable.EmbeddableMappingBuilder;
    import org.codefilarete.stalactite.engine.configurer.entity.CompositeKeyLinkageSupport;
    import org.codefilarete.stalactite.sql.ddl.structure.Column;
    import org.codefilarete.stalactite.sql.ddl.structure.PrimaryKey;
    import org.codefilarete.stalactite.sql.ddl.structure.Table;
    import org.codefilarete.stalactite.sql.statement.binder.ColumnBinderRegistry;
    import org.codefilarete.tool.VisibleForTesting;

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

    public class PrimaryKeyStep<C, I> {

    	/**
    	 * Creates primary key on table owning identification
    	 *
    	 * @param identification information that allow to create primary key
    	 * @return the created {@link PrimaryKey}
    	 */
    	@VisibleForTesting
    	<T extends Table<T>> PrimaryKey<T, I> addIdentifyingPrimarykey(AbstractIdentification<C, I> identification,
    																   Map<EntityMappingConfiguration, Table> tableMap,
    																   ColumnBinderRegistry columnBinderRegistry,
    																   ColumnNamingStrategy columnNamingStrategy,
    																   UniqueConstraintNamingStrategy uniqueConstraintNamingStrategy) {
    		T pkTable = (T) tableMap.get(identification.getIdentificationDefiner());
    		KeyMapping<C, I> keyLinkage = identification.getKeyLinkage();
    		AccessorDefinition identifierDefinition = AccessorDefinition.giveDefinition(identification.getIdAccessor());
    		if (identification instanceof AbstractIdentification.CompositeKeyIdentification) {
    			CompositeKeyMappingConfiguration<I> configuration = ((CompositeKeyLinkageSupport<C, I>) keyLinkage).getCompositeKeyMappingBuilder().getConfiguration();
    			// Note that we won't care about readonly column returned by build(..) since we're on a primary key case, then
    			// some readonly columns would be nonsense
    			EmbeddableMappingBuilder<I, T> compositeKeyBuilder = new EmbeddableMappingBuilder<>(configuration, pkTable, columnBinderRegistry, columnNamingStrategy, uniqueConstraintNamingStrategy);
    			EmbeddableMapping<I, T> build = compositeKeyBuilder.build();
    			Map<ReversibleAccessor<I, Object>, Column<T, Object>> compositeKeyMapping = build.getMapping();
    			compositeKeyMapping.values().forEach(Column::primaryKey);
    			((CompositeKeyIdentification<C, I>) identification).setCompositeKeyMapping(compositeKeyMapping);
    		} else {
    			String columnName = determineColumnName((SingleKeyMapping) keyLinkage, columnNamingStrategy);
    			// nullability = false may not be necessary because of primary key, let for principle
    			Column<?, I> primaryKey = pkTable.addColumn(columnName, identifierDefinition.getMemberType(), ((SingleKeyMapping<C, I>) keyLinkage).getColumnOptions().getColumnSize(), false);
    			primaryKey.primaryKey();
    			if (((SingleColumnIdentification<C, I>) identification).getIdentifierPolicy() instanceof GeneratedKeysPolicy) {
    				primaryKey.autoGenerated();
    			}
    		}
    		return pkTable.getPrimaryKey();
    	}

    	private <O> String determineColumnName(SingleKeyMapping<C, O> keyLinkage, ColumnNamingStrategy columnNamingStrategy) {
    		return nullable(keyLinkage.getColumnOptions().getColumnName())
    				.elseSet(keyLinkage::getFieldName)
    				.elseSet(() -> columnNamingStrategy.giveName(AccessorDefinition.giveDefinition(keyLinkage.getAccessor())))
    				.get();
    	}
    }