package gu.sql2java;

import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.MoreObjects;
import com.google.common.base.Predicate;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableBiMap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import gu.sql2java.ForeignKeyMetaData.ForeignKeyRule;
import gu.sql2java.Manager.AutoKeyRetrieveType;
import gu.sql2java.exception.DaoException;
import gu.sql2java.exception.DataAccessException;
import gu.sql2java.exception.DataRetrievalException;
import gu.sql2java.exception.ObjectRetrievalException;
import gu.sql2java.exception.OptimisticLockingException;
import gu.sql2java.exception.RuntimeDaoException;

import static com.google.common.base.Preconditions.*;
import static gu.sql2java.SimpleLog.*;

/**
 * implementation of {@link TableManager} 
 * @author guyadong
 *
 * @param <B>  java bean type
 */
public class BaseTableManager<B extends BaseBean> implements TableManager<B>,Constant{
    protected final RowMetaData metaData;
    /** lazy load */
        private volatile Map<String,TableListener<BaseBean>> foreignKeyDeleteListeners;
        /** lazy load */
        private volatile ListenerContainer<B> listenerContainer;
        /** lazy load */
        private volatile String generatedkeyStatement;
        private static boolean debug = false;
    protected BaseTableManager(String tablename){
        metaData = RowMetaData.getMedaData(tablename);
    }

    public ListenerContainer<B> getListenerContainer() {
                // double checking
                if(listenerContainer == null){
                        synchronized (this) {
                                if(listenerContainer == null){
                                        listenerContainer = new ListenerContainer<B>();
                                }
                        }
                }
                return listenerContainer;
        }

    /**
     * @return map with foreignKey name TO TableListener
     */
    public Map<String, TableListener<BaseBean>> getForeignKeyDeleteListeners(){
        // double checking
        if(foreignKeyDeleteListeners == null){
                synchronized (this) {
                        if(foreignKeyDeleteListeners == null){
                                LinkedHashMap<String, TableListener<BaseBean>> map = Maps.newLinkedHashMap();
                                for(ForeignKeyMetaData fk : metaData.getForeignKeysForListener()){
                                        map.put(fk.name, new DeleteRuleListener<BaseBean>(fk.name));
                                }
                                foreignKeyDeleteListeners = Collections.unmodifiableMap(map);
                        }
                        }
        }
        return foreignKeyDeleteListeners;
    }
        private String getGeneratedkeyStatement(Connection c){
                // double check
                if(generatedkeyStatement == null){
                        synchronized (this) {
                                if(generatedkeyStatement == null){
                                        generatedkeyStatement = checkNotNull(getManager().getGeneratedkeyStatement(c),"INVALID generatedkeyStatement")
                                                .replaceAll("<TABLE>", metaData.tablename )
                                                .replaceAll("<KEY>", metaData.columnNameOf(metaData.autoincrementColumnId));
                                        if(debug){
                                                log("generatedkeyStatement={}",generatedkeyStatement);
                                        }
                                }
                        }
                }
                return generatedkeyStatement;           
        }

        /**
     * Creates a new B instance.
     *
     * @return the new B instance
     */
        @SuppressWarnings("unchecked")
    protected final B createBean()
    {
        try {
                        return (B) metaData.beanType.newInstance();
                } catch (Exception e) {
                        Throwables.throwIfUnchecked(e);
                        throw new RuntimeException(e);
                }
    }
        /**
         * Creates a new B instance.
         * @param primaryValues values of primary keys
         * @return B instance
         */
        protected final B createBean(Object... primaryValues)
    {
                checkArgument(primaryValues != null && primaryValues.length== metaData.primaryKeyNames.length,"INVALID primaryValues");
                B bean = createBean();
                int[] pkIds = metaData.primaryKeyIds;
                for(int i=0;i<primaryValues.length;++i){
                        int columnId = pkIds[i];
                        Object value = primaryValues[i];
                        checkArgument(null == value || metaData.columnTypeOf(columnId).isInstance(value),
                                        "INVALID primkey type for %s,%s required",metaData.columnNameOf(columnId),metaData.columnTypeOf(columnId).getName());
                        bean.setValue(columnId, value);
                }
                return bean;
    }
    private static int indexOfFirstNull(Object...objects) {
                if(objects != null){
                        for(int i=0;i< objects.length;++i){
                                if(null == objects[i]){
                                        return i;
                                }
                        }
                        return -1;
                }
                return -2;
        }
        /**
         * @param objects 
         * @return true if any one of object is null or objects is null 
         */
        private static boolean hasNull(Object...objects) {
                return indexOfFirstNull(objects) != -1;
        }
        
        /**
         * @param bean
         * @return true if any primary key of B is null or B is null
         */
        private static <T extends BaseBean >boolean hasNullPk(T bean){
                return (null == bean || hasNull(bean.primaryValues()));
        }
        private void prepareAutoincrement( Connection c,PreparedStatement ps, B bean) throws SQLException
    {
        if (!bean.isModified(metaData.autoincrementColumnId))
        {
            PreparedStatement ps2 = null;
            ResultSet rs = null;
            try {
                AutoKeyRetrieveType retrieveType = getManager().getGeneratedkeyRetrieveType(c);
                if(AutoKeyRetrieveType.auto.equals(retrieveType)){
                        rs = ps.getGeneratedKeys();
                }else{
                    ps2 = c.prepareStatement(getGeneratedkeyStatement(c));
                        rs = ps2.executeQuery();
                }
                if(rs.next()) {
                    setColumnValue(bean, 
                                metaData.autoincrementColumnId, 
                                Manager.getObject(rs,1,metaData.columnTypes.get(metaData.autoincrementColumnId)));
                } else {
                    throw new IllegalStateException(logString("ATTENTION: Could not retrieve generated key!(retrieveType:{})",retrieveType));
                }
            } finally {
               getManager().close(ps2, rs);
            }
        }
    }
        
    //13
    /**
     * Insert the B bean into the database.
     * 
     * @param bean the B bean to be saved
     * @return the inserted bean
     * @throws RuntimeDaoException
     */
    protected B insert(final B bean)
    {
        // mini checks
        if (null == bean || !bean.isModified()) {
            return bean; 
        }
        if (!bean.isNew()){
            return this.update(bean);
        }

        Connection c = null;
        PreparedStatement ps = null;

                try
        {
            c = this.getConnection();
            final AutoKeyRetrieveType retrieveType = getManager().getGeneratedkeyRetrieveType(c);
            //-------------writePreInsert
            if(metaData.autoincrementColumnId >= 0 
                        && AutoKeyRetrieveType.before.equals(retrieveType)){
                        prepareAutoincrement(c, null, bean);
                }
            //------------/writePreInsert
            // listener callback
            getListenerContainer().beforeInsert(bean);
            StringBuilder sql = new StringBuilder("INSERT into " + metaData.tablename + " (");
            List<String> modifiedList = Lists.newArrayList(Iterables.filter(metaData.columnNames,new Predicate<String>() {

                        @Override
                        public boolean apply(String input) {
                                return bean.isModified(input);
                        }
                }));
            String fields=Joiner.on(",").join(modifiedList);
            // fields
            sql.append(fields);
            sql.append(") values (");
            // values
            sql.append(fields.replaceAll("[^,]+", "?"));
            sql.append(")");
            if(debug){
                log("insert : " + sql.toString());
            }
            if(metaData.autoincrementColumnId >= 0 && AutoKeyRetrieveType.auto.equals(retrieveType)){
                ps = c.prepareStatement(sql.toString(), Statement.RETURN_GENERATED_KEYS);
            }else{
                ps = c.prepareStatement(sql.toString(), ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            }
            
            fillPreparedStatement(ps, bean, SEARCH_EXACT, true);

            ps.executeUpdate();
            //------------------writePostInsert
            if(metaData.autoincrementColumnId >= 0 && !AutoKeyRetrieveType.before.equals(retrieveType)){
                        prepareAutoincrement(c, ps, bean);
                }
            //-------------------/writePostInsert
            bean.setNew(false);
            bean.resetIsModified();
            // listener callback
            getListenerContainer().afterInsert(bean);
            return bean;
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            // listener callback
            getListenerContainer().done();
            getManager().close(ps);
            freeConnection(c);
        }
    }
    
    private static final Function<String, String> SQL_FUN1 = new Function<String,String>(){

                @Override
                public String apply(String input) {
                        return input + "=?";
                }};
    //14
    /**
     * Update the B bean record in the database according to the changes.
     *
     * @param bean the B bean to be updated
     * @return the updated bean
     * @throws RuntimeDaoException
     */
    protected B update(final B bean) throws RuntimeDaoException
    {
        // mini checks
        if (null == bean || !bean.isModified()) {
            return bean;
        }
        if (bean.isNew()){
            return this.insert(bean);
        }

        Connection c = null;
        PreparedStatement ps = null;

        try
        {
            c = this.getConnection();

            // listener callback
            getListenerContainer().beforeUpdate(bean); 
            Object oldLockValue = null;
            if(metaData.lockColumnType != null){
                oldLockValue = bean.getValue(metaData.lockColumnName);
                        // lockColumnType is String or Long
                        if(String.class == metaData.lockColumnType){
                                bean.setValue(metaData.lockColumnName,String.valueOf(System.currentTimeMillis())); 
                        }else if(Long.class ==metaData.lockColumnType){
                                bean.setValue(metaData.lockColumnName,System.currentTimeMillis()); 
                        }else{
                                throw new RuntimeException(logString("INVALID LOCK COLUMN TYPE:{},String or Long required",metaData.lockColumnType));
                        }
            }
            StringBuilder sql = new StringBuilder("UPDATE " + metaData.tablename + " SET ");
            List<String> modified = Lists.newArrayList(Iterables.filter(metaData.columnNames,new Predicate<String>() {

                        @Override
                        public boolean apply(String input) {
                                return bean.isModified(input);
                        }
                }));
            sql.append(Joiner.on(",").join(Lists.transform(modified,SQL_FUN1)));

            sql.append(" WHERE ");
            sql.append(Joiner.on(" AND ").join(
                    Lists.transform(Arrays.asList(metaData.primaryKeyNames), SQL_FUN1)));
            if(metaData.lockColumnType != null){
                if(metaData.primaryKeyNames.length > 0){
                        sql.append(" AND ");
                }
                checkArgument(metaData.lockColumnName != null, "NOT DEFINED lock column name");
                sql.append(metaData.lockColumnName + "=?");
            }
            if(debug){
                log("update : " + sql.toString());
            }
            ps = c.prepareStatement(sql.toString(),
                                    ResultSet.TYPE_SCROLL_INSENSITIVE,
                                    ResultSet.CONCUR_READ_ONLY);

            int dirtyCount = this.fillPreparedStatement(ps, bean, SEARCH_EXACT,true);

            if (dirtyCount == 0) {
                if(debug){
                        log("The bean to look is not initialized... do not update.");
                }
                return bean;
            }
            int[] pkIds = metaData.primaryKeyIds;
            for(int i = 0; i<pkIds.length; ++i){
                setPreparedStatement(ps, ++dirtyCount, bean, pkIds[i]);
            }
            if(metaData.lockColumnType != null){
                setPreparedStatement(ps, ++dirtyCount, oldLockValue, metaData.columnIDOf(metaData.lockColumnName));
            }
            int rowCount = ps.executeUpdate();
            if (metaData.lockColumnType != null && rowCount==0) {
                throw new OptimisticLockingException("sql2java.exception.optimisticlock");
            }
            // listener callback
            getListenerContainer().afterUpdate(bean); 
            bean.resetIsModified();

            return bean;
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            // listener callback
            getListenerContainer().done();
            getManager().close(ps);
            freeConnection(c);
        }
    }
    
    class ListAction implements TableManager.Action<B> {
        final List<B> list;
        public ListAction() {
            list=new LinkedList<>();
        }

        public List<B> getList() {
            return list;
        }

        @Override
        public void call(B bean) {
            list.add(bean);
        }

    }
    
    @Override
    public int countAll()throws RuntimeDaoException{
        return this.countWhere("");
    }

    @Override
    public int countUsingTemplate(B bean)throws RuntimeDaoException{
        return this.countUsingTemplate(bean, SEARCH_EXACT);
    }

    @Override
    public int deleteAll()throws RuntimeDaoException{
        return this.deleteByWhere("");
    }

    @Override
    public B[] loadAll()throws RuntimeDaoException{
        return this.loadByWhere(null);
    }

    @Override
    public int loadAll(TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByWhere(null, action);
    }

    @SuppressWarnings("unchecked")
        @Override
    public B[] loadAll(int startRow, int numRows)throws RuntimeDaoException{
        return this.loadByWhereAsList(null, null, startRow, numRows).toArray((B[]) Array.newInstance(metaData.beanType, 0));
    }

    @Override
    public int loadAll(int startRow, int numRows, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByWhereForAction(null, null, startRow, numRows, action);
    }

    @Override
    public List<B> loadAllAsList()throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(null,1, -1, SEARCH_EXACT);
    }

    @Override
    public List<B> loadAllAsList(int startRow, int numRows)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(null, startRow, numRows, SEARCH_EXACT);
    }

    @Override
        public B loadByPrimaryKey(B bean)throws RuntimeDaoException{
                return bean==null ? null : loadByPrimaryKey(bean.primaryValues());
        }

        @Override
        public B loadByPrimaryKeyChecked(B bean)throws RuntimeDaoException,ObjectRetrievalException{
            return loadByPrimaryKeyChecked(checkNotNull(bean,"bean is null").primaryValues());
        }
        @Override
        public final B loadByPrimaryKeyChecked(Object ...keys)throws RuntimeDaoException,ObjectRetrievalException{
                // {{check parameters
                if(metaData.primaryKeyNames.length == 0){
                        throw new UnsupportedOperationException();
                }
            String[] pkNames = metaData.primaryKeyNames;
            int[] pkIds = metaData.primaryKeyIds;
                if(hasNull(keys)){
                        throw new ObjectRetrievalException(new NullPointerException("primary key must not be null"));
                }
                checkArgument(keys.length == pkNames.length,
                                "INVALID ARGUMENT NUM, %s required",pkNames.length);
            for(int i=0; i<pkNames.length; ++i){
                Object key = keys[i];
                Class<?> type = metaData.columnTypeOf(pkIds[i]);
                checkArgument(type.isAssignableFrom(key.getClass()),
                                "INVALID type for pk %s,%s required",pkNames[i],type.getName());
            }
         // }}check parameters
            return doLoadByPrimaryKeyChecked(keys);
        }
        
        protected B doLoadByPrimaryKeyChecked(Object ...keys)throws RuntimeDaoException,ObjectRetrievalException{
                if(metaData.primaryKeyNames.length == 0){
                        throw new UnsupportedOperationException();
                }
            String[] pkNames = metaData.primaryKeyNames;
            int[] pkIds = metaData.primaryKeyIds;
                if(hasNull(keys)){
                        throw new ObjectRetrievalException(new NullPointerException("primary key must not be null"));
                }
                checkArgument(keys.length == pkNames.length,
                                "INVALID ARGUMENT NUM, %s required",pkNames.length);
            for(int i=0; i<pkNames.length; ++i){
                Object key = keys[i];
                Class<?> type = metaData.columnTypeOf(pkIds[i]);
                checkArgument(type.isAssignableFrom(key.getClass()),
                                "INVALID type for pk %s,%s required",pkNames[i],type.getName());
            }
            Connection c = null;
            PreparedStatement ps = null;
            try
            {
                c = this.getConnection();
                StringBuilder sql = new StringBuilder("SELECT " + metaData.columnFields + " FROM " + metaData.tablename + " WHERE ");
                sql.append(Joiner.on(" AND ").join(
                            Lists.transform(Arrays.asList(pkNames), SQL_FUN1)));
                if(debug){
                        log("LOAD BY PK: " + sql.toString());
                }
                ps = c.prepareStatement(sql.toString(),
                                        ResultSet.TYPE_SCROLL_INSENSITIVE,
                                        ResultSet.CONCUR_READ_ONLY);
                        for(int i = 0; i<pkNames.length; ++i){
                                setPreparedStatement(ps,  i+1, keys[i], pkIds[i]);
                        }
                List<B> pReturn = this.loadByPreparedStatementAsList(ps,null,1,-1);
                if (1 == pReturn.size()) {
                    return pReturn.get(0);
                } else {
                    throw new ObjectRetrievalException();
                }
            }
            catch(ObjectRetrievalException e)
            {
                throw e;
            }
            catch(SQLException e)
            {
                throw new RuntimeDaoException(new DataRetrievalException(e));
            }
            finally
            {
                this.getManager().close(ps);
                this.freeConnection(c);
            }
        }

        @Override
        public B loadByPrimaryKey(Object ...keys)throws RuntimeDaoException{
            try{
                return loadByPrimaryKeyChecked(keys);
            }catch(ObjectRetrievalException e){
                // not found
                return null;
            }
        }

        protected <K> List<B> loadByPks(Collection<K> keys){
                checkState(metaData.primaryKeyCount == 1,"UNSUPPORTED OPERATION");
            if(null == keys){
                return Collections.emptyList();
            }
            ArrayList<B> list = new ArrayList<B>(keys.size());
            for(K key:keys){
                list.add(loadByPrimaryKey(key));
            }
            return list;
        }
        
        @SuppressWarnings("unchecked")
        protected <K> List<B> loadByPks(K... keys){
            if(null == keys){
                return Collections.emptyList();
            }
                return loadByPks(Arrays.asList(keys));
        }

        @Override
    public boolean existsByPrimaryKey(B bean)throws RuntimeDaoException{
        return null == bean ? false : existsPrimaryKey(bean.primaryValues());
    }
    @Override
    public B checkDuplicate(B bean)throws RuntimeDaoException,ObjectRetrievalException{
        if(existsByPrimaryKey(bean)){
            throw new ObjectRetrievalException("Duplicate entry ("+ bean.primaryValues() +") for key 'PRIMARY'");
        }
        return bean;   
    }
    @Override
    public final boolean existsPrimaryKey(Object ...keys)throws RuntimeDaoException{
        // {{check parameters
        if(metaData.primaryKeyNames.length == 0){
                throw new UnsupportedOperationException();
        }
        String[] pkNames = metaData.primaryKeyNames;
        int[] pkIds = metaData.primaryKeyIds;
        if(null == keys || hasNull(keys)){
                return false;
        }
        checkArgument(keys.length == pkNames.length,
                        "INVALID ARGUMENT NUM, %s required",pkNames.length);
        for(int i=0; i<pkNames.length; ++i){
                Object key = keys[i];
                Class<?> type = metaData.columnTypeOf(pkIds[i]);
                checkArgument(type.isInstance(key),
                                "INVALID type for pk %s,%s required",pkNames[i],type.getName());
        }
        // }}check parameters
        return doExistsPrimaryKey(keys);
    }
    protected boolean doExistsPrimaryKey(Object ...keys)throws RuntimeDaoException{
        String[] pkNames = metaData.primaryKeyNames;
        int[] pkIds = metaData.primaryKeyIds;
        Connection c = null;
        PreparedStatement ps = null;
        try{
            c = this.getConnection();
            StringBuilder sql = new StringBuilder("SELECT COUNT(*) AS MCOUNT FROM "  + metaData.tablename + " WHERE ");
            sql.append(Joiner.on(" AND ").join(Lists.transform(Arrays.asList(pkNames),SQL_FUN1)));
            if(debug){
                log("ExistsPrimaryKey: " + sql);
            }
            ps = c.prepareStatement(sql.toString(),
                                    ResultSet.TYPE_SCROLL_INSENSITIVE,
                                    ResultSet.CONCUR_READ_ONLY);

                        for(int i = 0; i<pkNames.length; ++i){
                                 setPreparedStatement(ps, i+1, keys[i], pkIds[i]);
                        }
            return 1 == getManager().runPreparedStatementForValue(Long.class, ps);
        }catch(SQLException e){
            throw new RuntimeDaoException(new ObjectRetrievalException(e));
        }finally{
            this.getManager().close(ps);
            this.freeConnection(c);
        }
    }
    protected <T>T checkDuplicateByPk(T primaryKeyValue)throws ObjectRetrievalException{
        if(metaData.primaryKeyNames.length != 1){
                throw new UnsupportedOperationException();
        }
            if(existsPrimaryKey(primaryKeyValue)){
                throw new ObjectRetrievalException("Duplicate entry '"+ primaryKeyValue +"' for key 'PRIMARY'");
            }
            return primaryKeyValue;
        }

        @Override
    public B[] loadByWhere(String where)throws RuntimeDaoException{
        return this.loadByWhere(where, (int[])null);
    }

    @Override
    public int loadByWhere(String where, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByWhere(where, null, action);
    }

    @Override
    public B[] loadByWhere(String where, int[] fieldList)throws RuntimeDaoException{
        return this.loadByWhere(where, fieldList, 1, -1);
    }

    @Override
    public int loadByWhere(String where, int[] fieldList, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByWhere(where, fieldList, 1, -1, action);
    }

    @SuppressWarnings("unchecked")
    @Override
    public B[] loadByWhere(String where, int[] fieldList, int startRow, int numRows)throws RuntimeDaoException{
        return this.loadByWhereAsList(where, fieldList, startRow, numRows).toArray((B[])Array.newInstance(metaData.beanType,0));
    }

    @Override
    public int loadByWhere(String where, int[] fieldList, int startRow, int numRows,
            TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadByWhereForAction(where, fieldList, startRow, numRows, action);
    }

    @Override
    public List<B> loadByWhereAsList(String where)throws RuntimeDaoException{
        return this.loadByWhereAsList(where, null, 1, -1);
    }

    @Override
    public List<B> loadByWhereAsList(String where, int[] fieldList)throws RuntimeDaoException{
        return this.loadByWhereAsList(where, fieldList, 1, -1);
    }

    @Override
    public List<B> loadByWhereAsList(String where, int[] fieldList, int startRow, int numRows)throws RuntimeDaoException{
        ListAction action = new ListAction();
        loadByWhereForAction(where, fieldList, startRow, numRows, action);              
        return action.getList();
    }

    @Override
    public int loadByWhereForAction(String where, int[] fieldList, int startRow, int numRows,TableManager.Action<B> action)throws RuntimeDaoException{
        String sql=createSelectSql(fieldList, where);
        return this.loadBySqlForAction(sql, null, fieldList, startRow, numRows, action);
    }

    @Override
    public B[] loadUsingTemplate(B bean)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, 1, -1, SEARCH_EXACT);
    }

    @Override
    public int loadUsingTemplate(B bean, TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, null, 1, -1, SEARCH_EXACT, action);
    }

    @Override
    public B[] loadUsingTemplate(B bean, int startRow, int numRows)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, startRow, numRows, SEARCH_EXACT);
    }

    @Override
    public int loadUsingTemplate(B bean, int startRow, int numRows,
            TableManager.Action<B> action)throws RuntimeDaoException{
        return this.loadUsingTemplate(bean, null, startRow, numRows,SEARCH_EXACT, action);
    }

    @SuppressWarnings("unchecked")
    @Override
    public B[] loadUsingTemplate(B bean, int startRow, int numRows, int searchType)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(bean, startRow, numRows, searchType).toArray((B[])Array.newInstance(metaData.beanType,0));
    }

    @Override
    public List<B> loadUsingTemplateAsList(B bean)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(bean, 1, -1, SEARCH_EXACT);
    }

    @Override
    public List<B> loadUsingTemplateAsList(B bean, int startRow, int numRows)throws RuntimeDaoException{
        return this.loadUsingTemplateAsList(bean, startRow, numRows, SEARCH_EXACT);
    }

    @Override
    public List<B> loadUsingTemplateAsList(B bean, int startRow, int numRows, int searchType)throws RuntimeDaoException{
        ListAction action = new ListAction();
        loadUsingTemplate(bean,null,startRow,numRows,searchType, action);
        return action.getList();
    }
    //18
    @Override
    public B loadUniqueUsingTemplate(B bean)
    {
        try {
                return loadUniqueUsingTemplateChecked(bean);    
                } catch (ObjectRetrievalException e) {
                        return null;
                }
     }
    //18-1
    @Override
    public B loadUniqueUsingTemplateChecked(B bean) throws ObjectRetrievalException
    {
         List<B> beans = this.loadUsingTemplateAsList(bean);
         switch(beans.size()){
         case 0:
             throw new ObjectRetrievalException("Not found element !!");
         case 1:
             return beans.get(0);
         default:
             throw new RuntimeDaoException(new ObjectRetrievalException("More than one element !!"));
         }
     }
    
    @Override
    public int loadUsingTemplate(B bean, int[] fieldList, int startRow, int numRows,int searchType, Action<B> action)
    {
        StringBuilder sqlWhere = new StringBuilder("");
        String sql=createSelectSql(fieldList,
                        this.fillWhere(sqlWhere, bean, searchType) > 0 ? " WHERE "+ sqlWhere.toString() : null);
        PreparedStatement ps = null;
        Connection connection = null;
        try {
            connection = this.getConnection();
            if(debug){
                log("loadUsingTemplate:" + sql);
            }
            ps = connection.prepareStatement(sql,
                    ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY);
            this.fillPreparedStatement(ps, bean, searchType,false);
            return this.loadByPreparedStatement(ps, fieldList, startRow, numRows, action);
        } catch (DaoException e) {
            throw new RuntimeDaoException(e);
        }catch (SQLException e) {
            throw new RuntimeDaoException(new DataAccessException(e));
        } finally {
            this.getManager().close(ps);
            this.freeConnection(connection);
        }
    }
    /**
     * @param <F> bean type of foreign table 
     * @param left
     * @param columnsMap
     * @param startRow
     * @param numRows
     * @return list of B or empty list
     */
    private <F extends BaseBean>
        List<B> loadByForeignKeyAsList(F left,Map<Integer,Integer>columnsMap,int startRow, int numRows)
        {
            if(null == left){
                return Collections.emptyList();
            }
                checkArgument(columnsMap != null && !columnsMap.isEmpty(),"columnsMap is null or empty");
            B bean = createBean().copy(left, columnsMap);
            return loadUsingTemplateAsList(bean,startRow,numRows);
        }
    /**
     * @param <F> bean type of foreign table
     * @param left
     * @param fkName
     * @param startRow
     * @param numRows
     * @return list of B or empty list
     */
    public <F extends BaseBean>
        List<B> loadByForeignKeyAsList(F left,String fkName,int startRow, int numRows)
        {
            ImmutableBiMap<Integer, Integer> columnsMap = metaData.foreignKeyIdMapOf(fkName).inverse();
            return loadByForeignKeyAsList(left,columnsMap,startRow,numRows);
        }

        @Override
    public B save(B bean)throws RuntimeDaoException{
        if(null != bean){
            if (bean.isNew()) {
                this.insert(bean);
            } else {
                this.update(bean);
            }
        }
        return bean;
    }
    
    @Override
    public B[] save(B[] beans)throws RuntimeDaoException{
        if(null != beans){
            for (B bean : beans) 
            {
                this.save(bean);
            }
        }
        return beans;
    }

    @Override
    public <C extends Collection<B>> C save(C beans)throws RuntimeDaoException{
        if(null != beans){
            for (B bean : beans) 
            {
                this.save(bean);
            }
        }
        return beans;
    }
    
    @Override
    public <C extends Collection<B>> C saveAsTransaction(final C beans)throws RuntimeDaoException{
        return this.runAsTransaction(new Callable<C>(){
            @Override
            public C call() throws Exception {
                return save(beans);
            }});
    }

    @Override
    public B[] saveAsTransaction(final B[] beans)throws RuntimeDaoException{
        return this.runAsTransaction(new Callable<B[]>(){
            @Override
            public B[] call() throws Exception {
                return save(beans);
            }});
    }

    private String checkWhere(String where){
        where = MoreObjects.firstNonNull(where, "").trim();
        checkArgument(where.isEmpty() || where.toUpperCase().startsWith("WHERE "),
                        "WHERE expression must start with 'WHERE'(case insensitive)");
        return where;
    }
    
    @SuppressWarnings("unchecked")
        @Override
    public <T> List<T> loadColumnAsList(String column,boolean distinct,String where,int startRow,int numRows)throws RuntimeDaoException{
        int columnId = metaData.columnIDOf(column);
        checkArgument(columnId>=0,"INVALID column name %s",column);
        String sql = String.format("SELECT %s " + metaData.columnNameOf(columnId) + " FROM %s %s",
                distinct ? "DISTINCT" : "",
                metaData.tablename,
                checkWhere(where));
        return runSqlAsList((Class<T>)metaData.columnTypes.get(columnId), sql);
    }
    /**
     * generate SQL query(SELECT) statement,such as: 'SELECT id,name from mytable WHERE id=1'
     * @param fieldList
     * @param where where condition expression statement that start with 'WHERE',or {@code null},or empty string
     * @return SQL statement string
     * @throws IllegalArgumentException where condition expression don't start with 'WHERE'
     */
    private String createSelectSql(int[] fieldList, String where){
        StringBuffer sql = new StringBuffer(128);
        String fullFields = metaData.columnFullFields;
        if(null == fieldList || 0 == fieldList.length) {
            sql.append("SELECT ").append(fullFields);
        } else{
            sql.append("SELECT ");
            String[] names=fullFields.split(",");
            for(int i = 0; i < fieldList.length; ++i){
                if(i > 0) {
                    sql.append(",");
                }
                sql.append(names[fieldList[i]]);
            }      
        }
        sql.append(" FROM " + this.metaData.tablename + " ");
        sql.append(checkWhere(where));
        return sql.toString();
    }

    //2.2

    @Override
    public int delete(B bean){
        if(metaData.primaryKeyNames.length==0){
                throw new UnsupportedOperationException();
        }
        if(hasNullPk(bean)){
                return 0;
        }
        Connection c = null;
        PreparedStatement ps = null;
        String [] pks = metaData.primaryKeyNames;
        int[] pkIds = metaData.primaryKeyIds;
        try
        {
            // listener callback
            getListenerContainer().beforeDelete(bean);
            c = this.getConnection();
            StringBuilder sql = new StringBuilder("DELETE  FROM " + metaData.tablename +" WHERE ");
            sql.append(Joiner.on(" AND ").join(Lists.transform(Arrays.asList(pks),SQL_FUN1)));
            if(debug){
                log("deleteByPrimaryKey: " + sql);
            }
            ps = c.prepareStatement(sql.toString(),
                                    ResultSet.TYPE_SCROLL_INSENSITIVE,
                                    ResultSet.CONCUR_READ_ONLY);
            for(int i = 0; i<pkIds.length; ++i){
                 setPreparedStatement(ps, i+1, bean, pkIds[i]);
            }
            int rows=ps.executeUpdate();
            if(rows>0){
                // listener callback
                getListenerContainer().afterDelete(bean);
            }
            return rows;
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            // listener callback
            getListenerContainer().done();
            getManager().close(ps);
            freeConnection(c);
        }
    }

    protected <K> int deleteByPks(Collection<K> keys){
                checkState(metaData.primaryKeyCount == 1,"UNSUPPORTED OPERATION");
            int count = 0;
            if(null != keys){
                for(K key :keys){
                    count += deleteByPrimaryKey(key);
                }
            }
            return count;
        }

        @SuppressWarnings("unchecked")
        protected <K> int deleteByPks(K... keys){
            if(null == keys){
                return 0;
            }
            return deleteByPks(Arrays.asList(keys));
        }

        @Override
        public int deleteByPrimaryKey(Object ...keys)throws RuntimeDaoException{
            if(hasNull(keys)){
                return 0;
            }
                checkArgument(keys.length == metaData.primaryKeyCount,"argument number mismatch with primary key number");
            return delete(createBean(keys));
        }

        //2.4 

        @SuppressWarnings("unchecked")
        @Override
        public int delete(B... beans)throws RuntimeDaoException{
            int count = 0;
            if(null != beans){
                for(B bean :beans){
                    count += delete(bean);
                }
            }
            return count;
        }

        //2.5  

        @Override
        public int delete(Collection<B> beans)throws RuntimeDaoException{
            int count = 0;
            if(null != beans){
                for(B bean :beans){
                    count += delete(bean);
                }
            }
            return count;
        }

        @SuppressWarnings("unchecked")
    @Override
        public <T extends BaseBean> T getReferencedBean(B bean, String fkName) throws RuntimeDaoException{
        if(null == bean){
                return null;
        }
        ForeignKeyMetaData foreignkey = checkNotNull(metaData.foreignKeys.get(fkName),
                        "INVALID fkName %s for table %s", fkName, metaData.tablename);
                BaseTableManager<T> foreignManager = (BaseTableManager<T>)managerOf(foreignkey.foreignTable);
        T t = foreignManager.createBean().copy(bean, metaData.foreignKeyIdMapOf(fkName));
        return foreignManager.loadByPrimaryKey(t);
    }
    
    @SuppressWarnings("unchecked")
    @Override
        public <T extends BaseBean> T setReferencedBean(B bean, T beanToSet, String fkName) throws RuntimeDaoException{
        if(null != bean){
                ForeignKeyMetaData foreignkey = checkNotNull(metaData.foreignKeys.get(fkName),
                                "INVALID fkName %s for table %s", fkName, metaData.tablename);
                BaseTableManager<T> foreignManager = (BaseTableManager<T>)managerOf(foreignkey.foreignTable);
                foreignManager.save(beanToSet);
                bean.copy(beanToSet, metaData.foreignKeyIdMapOf(fkName).inverse());
        }
        return beanToSet;
    }
    @SuppressWarnings("unchecked")
    @Override
        public <T extends BaseBean> T[] getImportedBeans(B bean,String fkName) throws RuntimeDaoException{
        ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
        RowMetaData foreignMetaData = RowMetaData.getMedaData(foreignkey.foreignTable);
        return getImportedBeansAsList(bean,fkName).toArray((T[]) Array.newInstance(foreignMetaData.beanType, 0));
    }

    @SuppressWarnings("unchecked")
    @Override
        public <T extends BaseBean> List<T> getImportedBeansAsList(B bean, String fkName)throws RuntimeDaoException{
        ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
                BaseTableManager<T> foreignManager = (BaseTableManager<T>)managerOf(foreignkey.ownerTable);
                return foreignManager.loadByForeignKeyAsList(bean,fkName, 1, -1);
    }
    protected <T extends BaseBean> List<T> getImportedBeansAsList(String fkName,Object...keys) throws RuntimeDaoException{          
        return getImportedBeansAsList(createBean(keys),fkName);
    }
        @SuppressWarnings("unchecked")
        protected <T extends BaseBean> T[] getImportedBeans(String fkName,Object...keys) throws RuntimeDaoException{              
        ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
        RowMetaData foreignMetaData = RowMetaData.getMedaData(foreignkey.foreignTable);
        return getImportedBeansAsList(fkName,keys).toArray((T[]) Array.newInstance(foreignMetaData.beanType, 0));
    }
    @SuppressWarnings("unchecked")
    @Override
        public <T extends BaseBean, C extends Collection<T>> C setImportedBeans(B bean, C importedBeans,
            String fkName)throws RuntimeDaoException{
        if(null != importedBeans){
                ForeignKeyMetaData foreignkey =metaData.getImportedKey(fkName);
                BaseTableManager<T> foreignManager = (BaseTableManager<T>)managerOf(foreignkey.ownerTable);
                for( T importBean : importedBeans ){
                        foreignManager.setReferencedBean(importBean , bean,fkName);
            }
        }
        return importedBeans;
    }
    
    @Override
    public <T extends BaseBean> T[] setImportedBeans(B bean, T[] importedBeans, String fkName) throws RuntimeDaoException{
        if(null != importedBeans){
                setImportedBeans(bean, Arrays.asList(importedBeans), fkName);
        }
        return importedBeans;
    }
    @SuppressWarnings("unchecked")
    @Override
        public int deleteImportedBeans(B bean,String fkName){
        if(bean == null){
                return 0;
        }
        ForeignKeyMetaData foreignkey = metaData.getImportedKey(fkName);
        RowMetaData foreignMetaData = RowMetaData.getMedaData(foreignkey.foreignTable);
                BaseTableManager<BaseBean> foreignManager = (BaseTableManager<BaseBean>)managerOf(foreignkey.ownerTable);
                BaseBean tmpl = foreignManager.createBean().copy(bean,foreignMetaData.foreignKeyIdMapOf(fkName));
                return foreignManager.deleteUsingTemplate(tmpl);
    }
    protected int deleteImportedBeans(Map<Integer, Object> idValueMap, String fkName) throws RuntimeDaoException{         
        B bean = createBean().copy(idValueMap);
        return deleteImportedBeans(bean,fkName);
    }
    
    protected int deleteImportedBeans(String fkName,Object...keys) throws RuntimeDaoException{          
        return deleteImportedBeans(createBean(keys),fkName);
    }
    private Map<Integer, Object> makeIndexValueMap(String indexName,Object ...indexValues){
        IndexMetaData indexMetaData = metaData.indices.get(indexName);
        checkArgument(null != indexMetaData,"INVALID index name %s",indexName);
        checkArgument(null != indexValues && indexValues.length == indexMetaData.columns.size(),"INVALID index value");
        ImmutableMap.Builder<Integer,Object> builder = ImmutableMap.<Integer,Object>builder();
        for(int i=0;i<indexValues.length;++i){
                Object value = indexValues[i];
                if(null != value){
                        String column = indexMetaData.columns.get(i);
                        Class<?> columnType = metaData.columnTypeOf(column);
                        checkArgument(columnType.isInstance(value),
                                        "INVALID value type %s for %s,%s required",value.getClass(),column,columnType);
                        builder.put(metaData.columnIDOf(column), value);
                }
        }
        return builder.build();
    }
    //2.4 override IDeviceManager

    @SuppressWarnings("unchecked")
    @Override
    public B[] loadByIndex(String indexName,Object ...keys)throws RuntimeDaoException{
        return this.loadByIndexAsList(indexName,keys).toArray((B[])Array.newInstance(metaData.beanType,0));
    }
    @Override
    public List<B> loadByIndexAsList(String indexName,Object ...indexValues)throws RuntimeDaoException{
        Map<Integer, Object> map = makeIndexValueMap(indexName,indexValues);
        return loadUsingTemplateAsList(map);
    }
    @Override
    public final B loadUniqueByIndex(String indexName,Object ...indexValues)throws RuntimeDaoException{
        return doLoadUniqueByIndex(indexName, indexValues);
    }
    protected B doLoadUniqueByIndex(String indexName,Object ...indexValues)throws RuntimeDaoException{
        Map<Integer, Object> keys = makeIndexValueMap(indexName,indexValues);
        B bean = createBean().copy(keys);
        return loadUniqueUsingTemplate(bean);
    }
    @Override
    public final B loadUniqueByIndexChecked(String indexName,Object ...indexValues)throws ObjectRetrievalException{
                if(hasNull(indexValues)){
                        throw new ObjectRetrievalException(new NullPointerException("index keys must not be null"));
                }
        return doLoadUniqueByIndexChecked(indexName, indexValues);
    }
    protected B doLoadUniqueByIndexChecked(String indexName,Object ...indexValues)throws ObjectRetrievalException{
        Map<Integer, Object> keys = makeIndexValueMap(indexName,indexValues);
        B bean = createBean().copy(keys);
        return loadUniqueUsingTemplateChecked(bean);
    }
    @Override
    public int deleteByIndex(String indexName,Object ...indexValues)throws RuntimeDaoException{
        Map<Integer, Object> map = makeIndexValueMap(indexName,indexValues);
        return deleteUsingTemplate(map);
    }
    
    private List<B> loadUsingTemplateAsList(Map<Integer, Object>keys){
        B bean = createBean().copy(keys);
        return loadUsingTemplateAsList(bean);
    }
    
    private int deleteUsingTemplate(Map<Integer, Object>keys){
        B bean = createBean().copy(keys);
        return deleteUsingTemplate(bean);
    }
    protected <T>List<B> loadByIndexForIndices(String indexName,Collection<T> indexs)
    {
        IndexMetaData indexMetaData = metaData.indices.get(indexName);
        checkArgument(null != indexMetaData,"INVALID index name %s",indexName);
        checkArgument(1 == indexMetaData.columns.size(),"column count of  index must be 1");
        checkArgument(indexMetaData.unique,"index must be unique");
        if(null == indexs ){
            return Collections.emptyList();
        }
        List<B> list = new ArrayList<B>(indexs.size());
        for(T key: indexs){
            list.add(loadUniqueByIndex(indexName,key));
        }
        return list;
    }
    @SuppressWarnings("unchecked")
        protected <T>List<B> loadByIndexForIndices(String indexName,T... indexs){
        if(indexs == null || indexs.length == 0){
                return Collections.emptyList();
        }
        return loadByIndexForIndices(indexName,Arrays.asList(indexs));
    }
    protected <T>int deleteByIndexForIndices(String indexName,Collection<T> indexs)
    {
        IndexMetaData indexMetaData = metaData.indices.get(indexName);
        checkArgument(null != indexMetaData,"INVALID index name %s",indexName);
        checkArgument(1 == indexMetaData.columns.size(),"column count of  index must be 1");

        if(null == indexs ){
            return 0;
        }
        int count = 0;
        for(T key: indexs){
                if(key != null){
                        count += deleteByIndex(indexName, key);
                }
        }
        return count;
    }
    @SuppressWarnings("unchecked")
        protected <T>int deleteByIndexForIndices(String indexName,T... indexs){
        if(indexs == null || indexs.length == 0){
                return 0;
        }
        return deleteByIndexForIndices(indexName,Arrays.asList(indexs));
    }
    private boolean checkPkValid(B bean)
        {
                if(metaData.primaryKeyNames.length ==0){
                        return false;
                }
        for(String name:metaData.primaryKeyNames){
                if(! (bean.isInitialized(name) && null != bean.getValue(name))){
                        return false;   
                }
        }       
        return true;
        }
    //21

    @Override
    public int deleteUsingTemplate(B bean)
    {
        if(bean == null || !bean.isModified()){
                return 0;
        }
        if(checkPkValid(bean)){
            return this.deleteByPrimaryKey(bean);
        }
        if( !getListenerContainer().isEmpty()){
            final DeleteBeanAction action=new DeleteBeanAction(); 
            this.loadUsingTemplate(bean,action);
            return action.getCount();
        }
        Connection c = null;
        PreparedStatement ps = null;
        StringBuilder sql = new StringBuilder("DELETE FROM " + metaData.tablename +" ");
        StringBuilder sqlWhere = new StringBuilder("");

        try
        {
            fillWhere(sqlWhere, bean, SEARCH_EXACT);
            sql.append(" WHERE ").append(sqlWhere);
            c = this.getConnection();
            if(debug){
                log("deleteUsingTemplate: " + sql.toString());
            }
            ps = c.prepareStatement(sql.toString(),
                                    ResultSet.TYPE_SCROLL_INSENSITIVE,
                                    ResultSet.CONCUR_READ_ONLY);
            this.fillPreparedStatement(ps, bean, SEARCH_EXACT, false);

            return ps.executeUpdate();
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            this.getManager().close(ps);
            this.freeConnection(c);
        }
    }
    
    //11

    @Override
    public int deleteByWhere(String where)
    {
        if( !getListenerContainer().isEmpty()){
            final DeleteBeanAction action = new DeleteBeanAction(); 
            this.loadByWhere(where,action);
            return action.getCount();
        }
        Connection c = null;
        PreparedStatement ps = null;

        try
        {
            c = this.getConnection();
            StringBuilder sql = new StringBuilder("DELETE FROM " + metaData.tablename + " " + where);
            if(debug){
                log("deleteByWhere: " + sql);
            }
            ps = c.prepareStatement(sql.toString());
            return ps.executeUpdate();
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            this.getManager().close(ps);
            this.freeConnection(c);
        }
    }
    private void setPreparedStatement(PreparedStatement ps,int pos,Object value,int columnId)
                throws SQLException {
        Manager.fillPreparedStatement(ps, pos, value, metaData.sqlTypes[columnId]);
    }
    private void setPreparedStatement(PreparedStatement ps,int pos,B bean,int columnId)
                throws SQLException {
        setPreparedStatement(ps,pos,bean.getValue(columnId),columnId);
    }

    @Override
        public B save(B bean,Map<String, BaseBean> referenceBeans,Map<String, Collection<BaseBean>> importedBeans) throws RuntimeDaoException{
            if(null == bean){
                return null;
            }
            referenceBeans = MoreObjects.firstNonNull(referenceBeans, Collections.<String, BaseBean>emptyMap());
            importedBeans = MoreObjects.firstNonNull(importedBeans, Collections.<String, Collection<BaseBean>>emptyMap());
            for(Entry<String, BaseBean> entry:referenceBeans.entrySet()){
                BaseBean beanToSet = entry.getValue();
                if(beanToSet != null){
                        setReferencedBean(bean, beanToSet, entry.getKey());
                }
            }
            bean = this.save( bean );
            for(Entry<String, Collection<BaseBean>> entry:importedBeans.entrySet()){
                String ikName = entry.getKey();
                Collection<BaseBean> importeds = entry.getValue(); 
                if(null != importeds){
                        setImportedBeans(bean, importeds, ikName);
                        TableManager<BaseBean> impManager = managerOf(metaData.getImportedKey(ikName).ownerTable);
                        impManager.save(importeds);
                }
            }
                return bean;            
        }
    
    @Override
    public B saveAsTransaction(final B bean,final Map<String, BaseBean> referenceBeans,final Map<String, Collection<BaseBean>> importedBeans)throws RuntimeDaoException{
        return this.runAsTransaction(new Callable<B>(){
            @Override
            public B call() throws Exception {
                return save(bean , referenceBeans, importedBeans );
            }});
    }
    
    //_____________________________________________________________________
    //
    // COUNT
    //_____________________________________________________________________
    //25

    @Override
    public int countWhere(String where)
    {
        String sql = new StringBuffer("SELECT COUNT(*) AS MCOUNT FROM " + metaData.tablename + " ")
                    .append(null == where ? "" : where).toString();
        return runSqlForValue(Long.class, sql).intValue();
    }
    //20
    /**
     * count the number of elements of a specific B bean given the search type
     *
     * @param bean the B template to look for
     * @param searchType exact ?  like ? starting like ?
     * @return the number of rows returned
     */
    @Override
    public int countUsingTemplate(B bean, int searchType)
    {
        Connection c = null;
        PreparedStatement ps = null;
        StringBuilder sql = new StringBuilder("SELECT COUNT(*) AS MCOUNT FROM " + metaData.tablename);
        StringBuilder sqlWhere = new StringBuilder("");

        try
        {
            if (this.fillWhere(sqlWhere, bean, SEARCH_EXACT) > 0) {
                sql.append(" WHERE ").append(sqlWhere);
            } else {
                if(debug){
                        log("The bean to look is not initialized... counting all...");
                }
            }

            c = this.getConnection();
            if(debug){
                log("countUsingTemplate: " + sql.toString());
            }
            ps = c.prepareStatement(sql.toString(),
                                    ResultSet.TYPE_SCROLL_INSENSITIVE,
                                    ResultSet.CONCUR_READ_ONLY);
            this.fillPreparedStatement(ps, bean, searchType,false);
            return getManager().runPreparedStatementForValue(Long.class,ps).intValue();
        }
        catch(SQLException e)
        {
            throw new RuntimeDaoException(new DataAccessException(e));
        }
        finally
        {
            this.getManager().close(ps);
            this.freeConnection(c);
            sql = null;
            sqlWhere = null;
        }
    }
    /**
     * fills the given StringBuilder with the sql where clauses constructed using the bean and the search type
     * @param sqlWhere the StringBuilder that will be filled
     * @param bean the bean to use for creating the where clauses
     * @param searchType exact ?  like ? starting like ?
     * @return the number of clauses returned
     */
    private int fillWhere(StringBuilder sqlWhere, B bean, int searchType)
    {
        if (bean == null) {
                return 0;
        }
        int dirtyCount = 0;
        String sqlEqualsOperation = searchType == SEARCH_EXACT ? "=" : " like ";
        List<String> fields = metaData.columnNames;
        for(int i=0; i<fields.size(); ++i){
                if(bean.isModified(i)){
                        ++dirtyCount;
                        if(bean.getValue(i) == null){
                                sqlWhere.append((sqlWhere.length() == 0) ? " " : " AND ").append(metaData.columnNameOf(i)).append(" IS NULL");
                        }else{
                                if(metaData.columnTypeOf(i) == String.class){
                                        sqlWhere.append((sqlWhere.length() == 0) ? " " : " AND ").append(metaData.columnNameOf(i)).append(" ").append(sqlEqualsOperation).append("?");
                                } else{
                                        sqlWhere.append((sqlWhere.length() == 0) ? " " : " AND ").append(metaData.columnNameOf(i)).append(" = ?");
                                }
                        }
                }
        }

        return dirtyCount;
    }
    /**
     * fill the given prepared statement with the bean values and a search type
     * @param ps the PreparedStatement that will be filled
     * @param bean the bean to use for creating the where clauses
     * @param searchType exact ?  like ? starting like ?
     * @param fillNull wether fill null for null field
     * @return the number of clauses returned
     */
    private int fillPreparedStatement(PreparedStatement ps, B bean, int searchType,boolean fillNull) throws DaoException
    {
        if (bean == null) {
            return 0;
        }
        int dirtyCount = 0;
        try
        {
                List<String> fields = metaData.columnNames;
                for(int columnId=0; columnId<fields.size(); ++columnId){
                        Object value = bean.getValue(columnId);
                        if(null != value || fillNull){
                                if(bean.isModified(columnId)){
                                        if(String.class == metaData.columnTypeOf(columnId)){
                                                switch (searchType) {
                                                case SEARCH_EXACT:
                                                        setPreparedStatement(ps, ++dirtyCount , value, columnId);
                                                        break;
                                                case SEARCH_LIKE:
                                                        setPreparedStatement(ps, ++dirtyCount , SQL_LIKE_WILDCARD + value + SQL_LIKE_WILDCARD, columnId);
                                                        break;
                                                case SEARCH_STARTING_LIKE:
                                                        setPreparedStatement(ps, ++dirtyCount , SQL_LIKE_WILDCARD + value, columnId);
                                                        break;
                                                case SEARCH_ENDING_LIKE:
                                                        setPreparedStatement(ps, ++dirtyCount , value + SQL_LIKE_WILDCARD, columnId);
                                                        break;
                                                default:
                                                        throw new DaoException("Unknown search type : " + searchType);
                                                }
                                        }else{
                                                setPreparedStatement(ps, ++dirtyCount , bean, columnId);
                                        } 
                                }                               
                        }
                }
        }
        catch(SQLException e)
        {
            throw new DataAccessException(e);
        }
        return dirtyCount;
    }
    
    /**
     * 检查指定的字段是否为{@code null}
     * @author guyadong
     *
     */
    private static class FieldNullChecker implements Predicate<String>{

                private final BaseBean bean;

                public FieldNullChecker(BaseBean bean) {
                        this.bean = checkNotNull(bean,"bean is null");
                }

                @Override
                public boolean apply(String input) {
                        return null == bean.getValue(input);
                }
        
    }
    //_____________________________________________________________________
    //
    // MANY TO MANY: LOAD OTHER BEAN VIA JUNCTION TABLE
    //_____________________________________________________________________

    /**
     * Retrieves an list of R using the junction table junction table(junctionTablename), given a linked table bean, 
     * specifying the start row and the number of rows.
     * @param <L>
     * @param <R>
     * @param left
     * @param rightType
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @return a list of R
     */
        @SuppressWarnings("unchecked")
        public <L extends BaseBean,R extends BaseBean>
    List<R> loadViaJunctionTableAsList( L left,Class<R> rightType, int startRow, int numRows)
    {
        checkState(!metaData.getJunctionTablePkMap().isEmpty(),"%s is not junction table",metaData.tablename);
        checkArgument(null != left,"left is null");
        checkArgument(null != rightType,"rightType is null");
        RowMetaData leftMetaData = RowMetaData.getMedaData(left.tableName());
        RowMetaData rightMetaData = RowMetaData.getMedaData(rightType);
        checkArgument(!leftMetaData.equals(rightMetaData),"same metadata  FOR left AND right type");
        checkArgument(metaData.isLinkedTable(leftMetaData.tablename),"INVALID left type %s",left.getClass());
        checkArgument(metaData.isLinkedTable(rightMetaData.tablename),"INVALID right type %s",rightType);
                Map<String, String> leftFields = metaData.junctionMapOf(leftMetaData.tablename);
        Map<String, String> rightFields = metaData.junctionMapOf(rightMetaData.tablename);
        if(Iterables.tryFind(leftFields.values(), new FieldNullChecker(left)).isPresent()){
                return Collections.emptyList();
        }
        Connection c = null;
        PreparedStatement ps = null;
        String sql = " SELECT " + rightMetaData.columnFullFields
                        + " FROM "+ metaData.tablename + ", " + rightMetaData.tablename
                        + " WHERE "
                        + Joiner.on(" AND ").join(Iterables.transform(leftFields.keySet(), 
                                        new Function<String,String>(){
                                                @Override
                                                public String apply(String input) {
                                                        return input + "=?";
                                                }}))
                        + " AND "
                        + Joiner.on(" AND ").join(Iterables.transform(rightFields.entrySet(), 
                                        new Function<Map.Entry<String,String>,String>(){
                
                                                                        @Override
                                                                        public String apply(Map.Entry<String, String> input) {
                                                                                return input.getKey() + "=" + input.getValue();
                                                                        }}));
        
        try
        {
            c = this.getConnection();
            if(debug){
                log("loadViaJunctionTableAsList" + sql);
            }
            ps = c.prepareStatement(sql,
                                    ResultSet.TYPE_SCROLL_INSENSITIVE,
                                    ResultSet.CONCUR_READ_ONLY);
            int pos =1;
            for(String key:leftFields.keySet()){
                String fullName = leftFields.get(key);
                int columnId = leftMetaData.columnIDOf(fullName);
                Manager.fillPreparedStatement(ps, pos++, 
                                left.getValue(columnId), 
                                leftMetaData.sqlTypeOf(columnId));      
            }
            return ((BaseTableManager<R>) managerOf(rightMetaData.beanType))
                        .loadByPreparedStatementAsList(ps, null, startRow, numRows);
        }
        catch (SQLException e)
        {
            throw new RuntimeDaoException(new DaoException(e.getMessage(), e));
        }
        finally
        {
           this.getManager().close(ps);
           this.freeConnection(c);
        }
    }

    private <L extends BaseBean,R extends BaseBean> B makeJunctionBean(L left,R right){
        Map<String, Object[]> map = metaData.getJunctionTablePkMap();
        checkState(!map.isEmpty(), "%s is not junction table",metaData.tablename);
        Object[] primaryValues = new Object[metaData.primaryKeyCount];
                for(int i=0; i<primaryValues.length; ++i){
                        String pk = metaData.columnNameOf(i);
                        Object[] data = map.get(pk);
                        checkArgument(data!=null,"PRIMARY KEY %s NOT DEFINED",pk);
                        String linkedTableName = (String)data[0];
                        int linkedColumnId = ((Integer)data[1]).intValue();

                        if(left.tableName().equals(linkedTableName)){
                                primaryValues[i] = left.getValueChecked(linkedColumnId);
                        }else if (right.tableName().equals(linkedTableName)){
                                primaryValues[i] = right.getValueChecked(linkedColumnId);
                        } else {
                                throw new IllegalArgumentException(String.format("%s NOT linked table  %s", linkedTableName,metaData.tablename));
                        }
                }
                return createBean(primaryValues);
    }

    /**
     * add junction between L and R if junction not exists
     * @param <L>
     * @param <R>
     * @param left
     * @param right
     */    
    public <L extends BaseBean,R extends BaseBean> 
    void addJunction(L left,R right){
        if(hasNullPk(left) || hasNullPk(right)){
            return ;
        }
        B junction = makeJunctionBean(left,right);
        if(!existsByPrimaryKey(junction)){
            save(junction);
        }
    }

    /**
     * remove junction between L and R if junction not exists
     * @param <L>
     * @param <R>
     * @param left
     * @param right
     */    
    public <L extends BaseBean,R extends BaseBean> 
    int deleteJunction(L left,R right){
        if(hasNullPk(left) || hasNullPk(right)){
            return 0;
        }
        B junction = makeJunctionBean(left,right);
        return delete(junction);
    }
    
    @SuppressWarnings("unchecked")
    public <L extends BaseBean,R extends BaseBean>
    void addJunction(L left,R... rights){
        if(null != rights){
                addJunction(left, Arrays.asList(rights));
        }
    }

    public <L extends BaseBean,R extends BaseBean> 
    void addJunction(L left,Collection<R> rights){
        if(null != rights){
            for(R linked:rights){
                addJunction(left,linked);
            }
        }
    }
    @SuppressWarnings("unchecked")
    public <L extends BaseBean,R extends BaseBean>
    int deleteJunction(L left,R... rights){
        if(null != rights){
                return deleteJunction(left, Arrays.asList(rights));
        }
        return 0;
    }
    public <L extends BaseBean,R extends BaseBean>
    int deleteJunction(L left,Collection<R> rights){
        int count = 0;
        if(null != rights){
            for(R right:rights){
                count += deleteJunction(left,right);
            }
        }
        return count;
    }

    //28-2
    /** decode a resultset and call action
     * @param rs the resultset to decode
     * @param fieldList table of the field's associated constants
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @param action interface obj for do something
     * @return the count dealt by action  
     * @throws IllegalArgumentException
     */
    protected int actionOnResultSet(ResultSet rs, int[] fieldList, int startRow, int numRows, Action<B> action) throws DaoException{
        try{
            int count = 0;
            if(0!=numRows){
                checkArgument(startRow>=1,"invalid argument:startRow (must >=1)");
                checkArgument(null !=action && null != rs,"invalid argument:action OR rs (must not be null)");
                        rs.absolute(startRow - 1);
                if (fieldList == null) {
                    if(numRows<0){
                        for(;rs.next();++count){
                            action.call(decodeRow(rs));
                        }
                    }else{
                        for(;rs.next() && count<numRows;++count){
                            action.call(decodeRow(rs));
                        }
                    }
                }else {
                    if(numRows<0){
                        for(;rs.next();++count){
                            action.call(decodeRow(rs, fieldList));
                        }
                    }else{
                        for(;rs.next() && count<numRows;++count){
                            action.call(decodeRow(rs, fieldList));
                        }
                    }
                }
            }
            return count;
        }catch(DaoException e){
            throw e;
        }catch(SQLException e){
            throw new DataAccessException(e);
        }
    }

        @SuppressWarnings("unchecked")
        protected <T> T getColumnValue(ResultSet resultSet, int columnId) throws SQLException {
                return (T) Manager.getObject(resultSet,columnId + 1, metaData.columnTypeOf(columnId));
        }
        protected void setColumnValue(B bean,int columnId, Object value ){
                if(value instanceof byte[] && ByteBuffer.class.equals(metaData.columnTypeOf(columnId))){
                        value = ByteBuffer.wrap((byte[])value);
                }
                bean.setValue(columnId, value);
        }
    //29
    /**    
     * Transforms a ResultSet iterating on a B bean.
     *
     * @param rs the ResultSet to be transformed
     * @return bean resulting B bean
     */
    private B decodeRow(ResultSet rs) throws DaoException
    {
        B   bean = createBean();
        try
        {
                for(int i=0; i<metaData.columnNames.size(); ++i){
                        setColumnValue(bean, i, getColumnValue(rs,i));
                }
        }
        catch(SQLException e)
        {
            throw new DataAccessException(e);
        }
        bean.setNew(false);
        bean.resetIsModified();

        return bean;
    }

    //30
    /**
     * Transforms a ResultSet iterating on a B bean according to a list of fields.
     *
     * @param rs the ResultSet to be transformed
     * @param fieldList table of the field's associated constants
     * @return bean resulting B bean
     */
    private B decodeRow(ResultSet rs, int[] fieldList) throws DaoException
    {
        fieldList = MoreObjects.firstNonNull(fieldList, metaData.defaultColumnIdList);
        B bean = createBean();
        try
        {
            for(int i = 0; i < fieldList.length; i++)
            {
                if(fieldList[i]>=0 && fieldList[i]<metaData.columnNames.size()){
                        int columnId = fieldList[i];
                        setColumnValue(bean, columnId, getColumnValue(rs, columnId) );
                }else{
                        throw new DaoException("Unknown field id " + fieldList[i]);
                }
            }
        }
        catch(SQLException e)
        {
            throw new DataAccessException(e);
        }
        bean.setNew(false);
        bean.resetIsModified();

        return bean;
    }
    //////////////////////////////////////
    // PREPARED STATEMENT LOADER
    //////////////////////////////////////

    //34-1
    /**
     * Loads all the elements using a prepared statement specifying a list of fields to be retrieved,
     * and specifying the start row and the number of rows.
     *
     * @param ps the PreparedStatement to be used
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @param fieldList table of the field's associated constants
     * @return an array of B
     */
    protected List<B> loadByPreparedStatementAsList(PreparedStatement ps, int[] fieldList, int startRow, int numRows) throws DaoException
    {
        ListAction action = new ListAction();
        loadByPreparedStatement(ps,fieldList,startRow,numRows,action);
        return action.getList();
    }
    //34-2
    /**
     * Loads each element using a prepared statement specifying a list of fields to be retrieved,
     * and specifying the start row and the number of rows 
     * and dealt by action.
     *
     * @param ps the PreparedStatement to be used
     * @param startRow the start row to be used (first row = 1, last row = -1)
     * @param numRows the number of rows to be retrieved (all rows = a negative number)
     * @param fieldList table of the field's associated constants
     * @param action Action object for do something(not null)
     * @return the count dealt by action
     */     
    private int loadByPreparedStatement(PreparedStatement ps, int[] fieldList, int startRow, int numRows,Action<B> action) throws DaoException
    {
        ResultSet rs =  null;
        try { 
            ps.setFetchSize(100);
            rs = ps.executeQuery();
            return this.actionOnResultSet(rs, fieldList, startRow, numRows, action);
        } catch (DaoException e) {
            throw e;
        } catch (SQLException e) {
            throw new DataAccessException(e);
        } finally {
            this.getManager().close(rs);
        }
    }
    //_____________________________________________________________________
    //
    // LISTENER
    //_____________________________________________________________________    

    /** foreign key listener for DEELTE RULE */
    protected class  DeleteRuleListener<F extends BaseBean> extends BaseForeignKeyListener<F,B>{
        protected final String fkName;
        private final ForeignKeyRule deleteRule;
        private final int[] foreignKeyIds;
        DeleteRuleListener(String fkName) {
                checkArgument(metaData.foreignKeys.containsKey(fkName),"INVALID foreign key name %s",fkName);
                this.fkName = fkName;
                this.deleteRule = metaData.foreignKeys.get(fkName).deleteRule;
                this.foreignKeyIds = metaData.foreignKeyIdArrayOf(fkName);
        }
        @Override
        protected List<B> getImportedBeans(F bean) {
                return getListenerContainer().isEmpty() 
                                ? Collections.<B>emptyList()
                                : loadByForeignKeyAsList(bean,fkName,1,-1);
        }
                @SuppressWarnings("unchecked")
                @Override
                protected void onRemove(List<B> affectedBeans) throws DaoException {
                        switch (deleteRule) {
                        case SET_NULL:
                                for(B bean : affectedBeans){
                                        BaseRow updated = ((BaseRow)bean).clone();
                                        for(int i=0; i<foreignKeyIds.length; ++i){
                                                updated.setValue(foreignKeyIds[i], null);
                                        }
                                        fire(Event.UPDATE, (B) updated);
                                }
                                break;
                        case SET_DEFAULT:
                                B tmpl = createBean();
                                for(B bean : affectedBeans){
                                        BaseRow updated = ((BaseRow)bean).clone();
                                        for(int i=0; i<foreignKeyIds.length; ++i){
                                                updated.setValue(foreignKeyIds[i], tmpl.getValue(foreignKeyIds[i]));
                                        }
                                        fire(Event.UPDATE, (B) updated);
                                }
                                break;
                        default:
                                if(!deleteRule.isNoAction()){
                                        Event event = Event.valueOf(deleteRule.eventOfDeleteRule);
                                        for(B bean : affectedBeans){
                                                fire(event, bean);
                                        }
                                }
                                break;
                        }
                }
                
                /**
                 * fire current event by  {@link TableListener}
                 * @param event
                 * @param bean
                 * @throws RuntimeDaoException
                 */
                private void fire(Event event,B bean)throws RuntimeDaoException {

                        ListenerContainer<B> container = getListenerContainer();
                        switch(event){
                        case UPDATE:
                                container.beforeUpdate(bean);
                                container.afterUpdate(bean);
                                container.done();
                                break;
                        case DELETE:
                                container.beforeDelete(bean);
                                container.afterDelete(bean);
                                container.done();
                                break;
                        case INSERT:
                                // DO NOTHING
                                // container.beforeInsert(bean);
                                // container.afterInsert(bean);
                                // container.done();
                        default:
                                break;
                        }
                }

                @Override
                public String toString() {
                        StringBuilder builder = new StringBuilder();
                        builder.append("DeleteRuleListener [fkName=");
                        builder.append(fkName);
                        builder.append(", deleteRule=");
                        builder.append(deleteRule);
                        builder.append(", foreignKeyIds=");
                        builder.append(metaData.columnNamesOf(foreignKeyIds));
                        builder.append("]");
                        return builder.toString();
                }       
    }

    //35

    @Override
    public void registerListener(TableListener<B> listener)
    {
        this.getListenerContainer().add(listener);
                if(debug){
                        log("REGISTER TABLE LISTENER {}",listener);
                }
    }

    //36

    @Override
    public void unregisterListener(TableListener<B> listener)
    {
        this.getListenerContainer().remove(listener);
                if(debug){
                        log("UNREGISTER TABLE LISTENER {}",listener);
                }
    }
    
    //37-2
    /**
     * bind foreign key listener to foreign table: <br>
     * DELETE RULE <br>
     */
    public void bindForeignKeyListenerForDeleteRule(){
        for(Entry<String, TableListener<BaseBean>> entry : getForeignKeyDeleteListeners().entrySet()){
                TableManager<BaseBean> manager = managerOf(metaData.foreignKeys.get(entry.getKey()).foreignTable);
                manager.registerListener(entry.getValue());
        }
    }
    
  //37-3
    /**
     * unbind foreign key listener from all of foreign tables <br>
     * @see #bindForeignKeyListenerForDeleteRule()
     */
    public void unbindForeignKeyListenerForDeleteRule(){
        for(Entry<String, TableListener<BaseBean>> entry : getForeignKeyDeleteListeners().entrySet()){
                TableManager<BaseBean> manager = managerOf(metaData.foreignKeys.get(entry.getKey()).foreignTable);
                manager.unregisterListener(entry.getValue());
        }
    }
    //_____________________________________________________________________
    //
    // UTILS
    //_____________________________________________________________________
    //40
    /**
     * Retrieves the manager object used to get connections.
     *
     * @return the manager used
     */
    protected Manager getManager()
    {
        return Manager.getInstance();
    }

    //41
    /**
     * Frees the connection.
     *
     * @param c the connection to release
     */
    protected void freeConnection(Connection c)
    {
        // back to pool
        this.getManager().releaseConnection(c);
    }

    //42
    /**
     * Gets the connection.
     */
    protected Connection getConnection() throws DaoException
    {
        try
        {
            return this.getManager().getConnection();
        }
        catch(SQLException e)
        {
            throw new DataAccessException(e);
        }
    }
    
    private int loadBySqlForAction(String sql, Object[] argList, int[] fieldList,int startRow, int numRows,Action<B> action){
        PreparedStatement ps = null;
        Connection connection = null;
        try {
            connection = this.getConnection();
            if(debug){
                log("loadBySqlForAction:" + sql );
            }
            ps = connection.prepareStatement(sql,
                    ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY);
            Manager.fillPrepareStatement(ps, argList);
            return this.loadByPreparedStatement(ps, fieldList, startRow, numRows, action);
        } catch (SQLException e) {
            throw new RuntimeDaoException(new DataAccessException(e));
        } finally {
            this.getManager().close(ps);
            this.freeConnection(connection);
        }
    }

    @Override
        public List<BaseBean> runSqlAsList(String sql, Object... argList) throws RuntimeDaoException{
        return getManager().runSqlAsList(sql, argList);
    }
    
    @Override
        public List<Map<String, Object>> runSqlForMap(Map<String,Class<?>> targetTypes, String sql,Object... argList) throws RuntimeDaoException{
        return getManager().runSqlForMap(targetTypes, sql, argList);
    }
    
    @Override
        public <T>List<T> runSqlAsList(Class<T> targetType, String sql,Object... argList) throws RuntimeDaoException{
        return getManager().runSqlAsList(targetType, sql, argList);
    }
    
    @Override
        public <T> T runSqlForValue(Class<T> targetType,String sql, Object... argList) throws RuntimeDaoException{
        return getManager().runSqlForValue(targetType, sql, argList);
    }
    
    @Override
    public <T>T runAsTransaction(Callable<T> fun) {
       return getManager().runAsTransaction(fun);
    }
    
    @Override
    public void runAsTransaction(Runnable fun) {
        getManager().runAsTransaction(fun);
    }    
    protected class DeleteBeanAction implements Action<B>{
        private final AtomicInteger count=new AtomicInteger(0);
        public DeleteBeanAction() {
                }
                @Override
        public void call(B bean){
                delete(bean);
                count.incrementAndGet();
        }
        public int getCount(){
            return count.get();
        }
    }
    
    //45 

    /**
     * @param <T> PK type
     * @param type
     * @param beans
     * @return return a primary key list from B array
     * @see #toPrimaryKeyList(Class,Collection)
     */
    @SuppressWarnings("unchecked")
        protected <T> List<T> toPrimaryKeyList(Class<T>type,B... beans){        
        if(null == beans || beans.length ==0){
            return Collections.emptyList();
        }
        return toPrimaryKeyList(type,Arrays.asList(beans));
    }
    
    private class ColumnTransformer<T> implements Function<B, T>{
        private int columnId;
                public ColumnTransformer(int columnId) {
                        checkArgument(columnId >=0 && columnId < metaData.columnCount,"INVALID column id %s",columnId);
                        this.columnId = columnId;
                }
                @Override
                public T apply(B input) {
                        return input == null ? null : input.<T>getValue(columnId);
                }       
    }
    /**
         * listener event:<br>
         * {@code INSERT} insert a bean<br>
         * {@code UPDATE} update a bean<br>
         * {@code DELETE} delete a bean<br>
         * @author guyadong
         *
         */
        public static enum Event{        
            /** insert a bean */INSERT,
            /** update a bean */UPDATE,
            /** delete a bean */DELETE
        }

        //46 
    /**
     * return a primary key list from B collection<br>
     * throw {@link UnsupportedOperationException} if there is more than a primary key
     * @param <T> PK type
     * @param type PK type
     * @param beans input beans
     */
    protected <T> List<T> toPrimaryKeyList(Class<T>type,Collection<B> beans){        
        if(metaData.primaryKeyNames.length != 1){
                throw new UnsupportedOperationException();
        }
        if(null == beans){
            return Collections.emptyList();
        }
        checkArgument(metaData.columnTypeOf(metaData.primaryKeyIds[0]) != type,"INVALID primary key type: " + type);
        return ImmutableList.copyOf(Iterables.transform(beans, new ColumnTransformer<T>(metaData.primaryKeyIds[0])));
    }
    private Object[] toPkValues(B bean,int[] selfFkIds){
        checkArgument(selfFkIds.length == metaData.primaryKeyNames.length,"MISMATCH SIZE of primary keys");
        return bean.asValueArray(selfFkIds);
    }
    
    protected List<B> listOfSelfRef(String fkName,Object... primaryKeys){
        List<B> list = new ArrayList<>();
        int[] selfFks = metaData.foreignKeyIdArrayOf(fkName); 
        for(B ref = loadByPrimaryKey(primaryKeys)
                ; null != ref
                ; ref = loadByPrimaryKey(toPkValues(ref,selfFks))){
            list.add(ref);
            Object[] refPk = ref.primaryValues();
            Object[] refSelf = toPkValues(ref,selfFks);
            
            if(Arrays.equals(refPk,refSelf)
                        || (list.size() > 1 && Arrays.equals(refPk,primaryKeys))){
                // cycle reference
                break;
            }
        }
        Collections.reverse(list);
        return list;
    }
    protected int levelOfSelfRef(String fkName,Object... primaryKeys){
        int count = 0 ;
        int[] selfFks = metaData.foreignKeyIdArrayOf(fkName); 
        for(B ref = loadByPrimaryKey(primaryKeys)
           ; null != ref
           ; ++count,ref = loadByPrimaryKey(toPkValues(ref,selfFks))){
            Object[] refPk = ref.primaryValues();
            Object[] refSelf = toPkValues(ref,selfFks);
            if(    (Arrays.equals(refPk,refSelf))
                || (count > 0 && Arrays.equals(refPk,primaryKeys))){
                // cycle reference
                return -1;
            }
        }
        return count;
    }
    protected B topOfSelfRef(String fkName,Object... primaryKeys){
        checkArgument(!hasNull(primaryKeys),"primaryKeys has null element");
        int[] selfFks = metaData.foreignKeyIdArrayOf(fkName); 

        B ref = loadByPrimaryKey(primaryKeys);
        int count = 0 ;
        Object[] refSelf = toPkValues(ref,selfFks);

        for(;null != ref && !hasNull(refSelf);){
            Object[] refPk = ref.primaryValues();
            refSelf = toPkValues(ref,selfFks);

                if(   (Arrays.equals(refPk,refSelf))
                || (++count > 1 && Arrays.equals(refPk,primaryKeys))){
                // cycle reference
                throw new IllegalStateException("cycle on field: " + "parent");
            }
                checkState(Arrays.equals(refPk,refSelf)
                || (++count > 1 && Arrays.equals(refPk,primaryKeys)));
            ref = loadByPrimaryKeyChecked(refSelf);
        }
        return ref;
    }
    protected LinkedHashSet<B> doListOfChild(B bean, LinkedHashSet<B> set,String fkName){
        if(existsByPrimaryKey(bean)){
            checkState(!set.contains(bean),"cycle on foreign key: " + fkName);
            set.add(bean);
            List<B> childs = loadByForeignKeyAsList(bean, fkName, 1, -1);
            for(B c:childs){
                doListOfChild(c,set,fkName);
            }
        }
        return set;
    }
        @Override
        public int hashCode() {
                final int prime = 31;
                int result = 1;
                result = prime * result + ((metaData == null) ? 0 : metaData.hashCode());
                return result;
        }

        @Override
        public boolean equals(Object obj) {
                if (this == obj) {
                        return true;
                }
                if (obj == null) {
                        return false;
                }
                if (getClass() != obj.getClass()) {
                        return false;
                }
                BaseTableManager<?> other = (BaseTableManager<?>) obj;
                if (metaData == null) {
                        if (other.metaData != null) {
                                return false;
                        }
                } else if (!metaData.equals(other.metaData)) {
                        return false;
                }
                return true;
        }

        @Override
        public String toString() {
                StringBuilder builder = new StringBuilder();
                builder.append(getClass().getSimpleName() + " [metaData=");
                builder.append(metaData);
                builder.append("]");
                return builder.toString();
        }

        /**
         * set debug flag that determine if output log message,default : false
         * @param debug flag for debug message output
         */
        public static void setDebug(boolean debug) {
                BaseTableManager.debug = debug;
        }
    private static final ImmutableMap<String, TableManager<? extends BaseBean>> 
        tableManagerInstances = loadTableManager(); 
    private static final  ImmutableMap<Class<?>, TableManager<? extends BaseBean>> 
        tableManagerTypeMaps = asTypeMap(tableManagerInstances);
    private static final  ImmutableMap<Class<?>, TableManager<? extends BaseBean>> 
        tableManagerBeanTypeMaps = asBeanTypeMap(tableManagerInstances);
        /**
         * SPI(Service Provider Interface)机制加载 {@link TableManager}所有实例
         * @return 表名和 {@link TableManager}实例的映射对象 
         */
        @SuppressWarnings({ "rawtypes" })
        private static ImmutableMap<String, TableManager<? extends BaseBean>> loadTableManager() {          
                ServiceLoader<TableManager> providers = ServiceLoader.load(TableManager.class);
                Iterator<TableManager> itor = providers.iterator();
                TableManager<?> node;
                ImmutableMap.Builder<String, TableManager<? extends BaseBean>> builder = ImmutableMap.builder();
                while(itor.hasNext()){
                        node = itor.next();
                        if(node instanceof BaseTableManager){
                                BaseTableManager<?> manager = (BaseTableManager<?>)node;
                                builder.put(manager.metaData.tablename, manager);
                        }
                }
                return builder.build();
        }
        private static final ImmutableMap<Class<?>, TableManager<? extends BaseBean>> asTypeMap(Map<String, TableManager<? extends BaseBean>>input){
                return Maps.uniqueIndex(input.values(), new Function<TableManager<? extends BaseBean>,Class<?>>(){

                        @Override
                        public Class<?> apply(TableManager<? extends BaseBean> input) {
                                return input.getClass();
                        }});
        }
        private static final ImmutableMap<Class<?>, TableManager<? extends BaseBean>> asBeanTypeMap(Map<String, TableManager<? extends BaseBean>>input){
                return Maps.uniqueIndex(input.values(), new Function<TableManager<? extends BaseBean>,Class<?>>(){

                        @Override
                        public Class<?> apply(TableManager<? extends BaseBean> input) {
                                return ((BaseTableManager<?>)input).metaData.beanType;
                        }});
        }
        public static ImmutableMap<String, TableManager<? extends BaseBean>> getTableManagers() {
                return tableManagerInstances;
        }

        @SuppressWarnings("unchecked")
        public static final <M extends TableManager<?>>M 
        getTableManager(Class<M>managerType) {
                TableManager<? extends BaseBean> manager = tableManagerTypeMaps.get(managerType);
                return checkNotNull((M) manager,"INVALID manager type %s",managerType);
        }
        @SuppressWarnings("unchecked")
        public static final <T extends BaseBean,M extends TableManager<T>>M 
        getTableManagerByBeanType(Class<T>beanType) {
                TableManager<? extends BaseBean> manager = tableManagerBeanTypeMaps.get(beanType);
                return checkNotNull((M) manager,"INVALID bean type %s",beanType);
        }
        @SuppressWarnings("unchecked")
        public static final <M extends TableManager<?>>M 
        getTableManager(String tablename) {
                TableManager<? extends BaseBean> manager = tableManagerInstances.get(tablename);
                return checkNotNull((M) manager,"INVALID tablename %s",tablename);
        }
        
    private static final  Map<Class<?>, TableManager<? extends BaseBean>> cacheManagers = Maps.newHashMap();
    private static final  Map<Class<?>, TableManager<? extends BaseBean>> cacheBeanTypeManagers = Maps.newHashMap();
    private static final  Map<String, TableManager<? extends BaseBean>> cacheNameManagers = Maps.newHashMap();
        /**
         * 注册cache manager
         * @param cacheManager
         */
        public synchronized static void registerCacheManager(ICacheManager cacheManager) {              
                if(cacheManager instanceof BaseTableManager<?>){
                        BaseTableManager<?> manager = (BaseTableManager<?>)cacheManager;
                        Class<? extends ICacheManager> clazz = cacheManager.getClass();
                        cacheManagers.put(clazz.getSuperclass(), manager);
                        cacheNameManagers.put(manager.metaData.tablename, manager);
                        cacheBeanTypeManagers.put(manager.metaData.beanType, manager);
                        if(debug){
                                log("REGISTER CACHE MANAGER {}",manager);
                        }
                }
        }

        public static Map<Class<?>, TableManager<? extends BaseBean>> getCacheManagers() {
                return Collections.unmodifiableMap(cacheManagers);
        }

        /**
         * 根据目标类型返回对应的 {@link TableManager}实例
         * @param targetType 目标类型
         * @return {@link TableManager}实例
         * @throws NoSuchElementException 找不到时抛出异常
         */
        @SuppressWarnings("unchecked")
        public static final <M extends TableManager<? extends BaseBean>>M 
        getCacheManager(final Class<M>targetType) throws NoSuchElementException {
                checkArgument(targetType != null,"targetType is null");
                TableManager<? extends BaseBean> manager;
                if(null != (manager = cacheManagers.get(targetType))){
                        return (M) manager;
                }
                return (M) Iterables.find(cacheManagers.values(), new Predicate<TableManager<?>>() {

                        @Override
                        public boolean apply(TableManager<?> input) {
                                return targetType.isInstance(input);
                        }
                });
        }
        @SuppressWarnings("unchecked")
        public static final <M extends TableManager<?>> M 
        getCacheManagerByBeanType(Class<?> beanType)  {
                TableManager<? extends BaseBean> manager = cacheBeanTypeManagers.get(beanType);
                return (M) checkNotNull(manager,"INVALID bean type %s",beanType);
        }
        @SuppressWarnings("unchecked")
        public static final <M extends TableManager<?>> M 
        getCacheManager(String tablename)  {
                TableManager<? extends BaseBean> manager = cacheNameManagers.get(tablename);
                return (M) checkNotNull(manager,"INVALID table name %s",tablename);
        }

        public static <T extends BaseBean,M extends TableManager<? extends BaseBean>>M 
        instanceOf(Class<M>targetType) {
            try {
                return getCacheManager(targetType);
            } catch (Exception e) {
                return getTableManager(targetType);
            } 
        }

        public static <M extends TableManager<? extends BaseBean>>M managerOf(String tablename) {
            try {
                return getCacheManager(tablename);
            } catch (Exception e) {
                return getTableManager(tablename);
            } 
        }

        public static <T extends BaseBean,M extends TableManager<T>>M managerOf(Class<T> beanType) {
            try {
                return getCacheManagerByBeanType(beanType);
            } catch (Exception e) {
                return getTableManagerByBeanType(beanType);
            } 
        }
}