public abstract class BaseImportor extends AbstractStep implements Importor {
    private final static Log log = LogFactory.getLog(BaseImportor.class);
    
    private static final String batchID = "http://www.dbunit.org/features/batchedStatements";
    private static DataSource ds;
    
    protected IDataSet srcDataSet;

    protected String _type = TYPE_CLEAN_INSERT;
    private boolean _transaction = false;
    private DatabaseOperation _operation;
    protected boolean _forwardOperation = true;
    
    /**
     * 通过spring注射dataSource
     * @param dataSource
     */
    public final void setDataSource(final DataSource dataSource){
    	ds = dataSource;
    }
    public BaseImportor() {
        
    }
    /**
     * 获取连接
     * @return
     * @throws DatabaseUnitException
     */
    protected IDatabaseConnection getConnection() throws DatabaseUnitException{
    	IDatabaseConnection iDataBaseConnection = null;
		try {
			String schema = ds.getConnection().getMetaData().getUserName();
			iDataBaseConnection = new DatabaseConnection(ds.getConnection(), schema);
			DatabaseConfig config = iDataBaseConnection.getConfig();
			if (!config.getFeature(batchID)) {
			    config.setFeature(batchID, true);
			}
			log.debug("get database connection with schema '"+schema+"'");
		} catch (SQLException e) {
			log.error(e);
			throw new DatabaseUnitException(e);
		}
        return iDataBaseConnection;
    }

    public String getType()
    {
        return _type;
    }

    public DatabaseOperation getDbOperation()
    {
        return _operation;
    }

    public boolean isTransaction()
    {
        return _transaction;
    }
    /**
     * 设置操作类型
     */
    public void setType(String type)
    {
        if (TYPE_UPDATE.equals(type))
        {
            _operation = DatabaseOperation.UPDATE;
            _forwardOperation = true;
        }
        else if (TYPE_INSERT.equals(type))
        {
            _operation = DatabaseOperation.INSERT;
            _forwardOperation = true;
        }
        else if (TYPE_REFRESH.equals(type))
        {
            _operation = DatabaseOperation.REFRESH;
            _forwardOperation = true;
        }
        else if (TYPE_DELETE.equals(type))
        {
            _operation = DatabaseOperation.DELETE;
            _forwardOperation = false;
        }
        else if (TYPE_DELETE_ALL.equals(type))
        {
            _operation = DatabaseOperation.DELETE_ALL;
            _forwardOperation = false;
        }
        else if (TYPE_CLEAN_INSERT.equals(type))
        {
            _operation = DatabaseOperation.CLEAN_INSERT;
            _forwardOperation = false;
        }
        else if (TYPE_NONE.equals(type))
        {
            _operation = DatabaseOperation.NONE;
            _forwardOperation = true;
        }
        else if (TYPE_MSSQL_CLEAN_INSERT.equals(type))
        {
            _operation = InsertIdentityOperation.CLEAN_INSERT;
            _forwardOperation = false;
        }
        else if (TYPE_MSSQL_INSERT.equals(type))
        {
            _operation = InsertIdentityOperation.INSERT;
            _forwardOperation = true;
        }
        else if (TYPE_MSSQL_REFRESH.equals(type))
        {
            _operation = InsertIdentityOperation.REFRESH;
            _forwardOperation = true;
        }
        else
        {
            throw new IllegalArgumentException("Type must be one of: UPDATE, INSERT,"
                    + " REFRESH, DELETE, DELETE_ALL, CLEAN_INSERT, MSSQL_INSERT, "
                    + " or MSSQL_REFRESH but was: " + type);
        }
        _type = type;
    }


    public void setTransaction(boolean transaction)
    {
        _transaction = transaction;
    }

    public void execute(IDatabaseConnection connection) throws DatabaseUnitException
    {
        if (_operation == null)
        {