public void setupDeltaTable(BaseDeltaTable delta) {
        Statement stmt = null;
        ResultSet rs = null;
        try {
            dbCon.createStatement();
            stmt = dbCon.createStatement();
            String deltaCountSql = "select count(1)  from information_schema.tables where table_schema='mysqlmv' and table_name = '" + delta.getTable() + "'";
            stmt.execute(deltaCountSql);
            rs = stmt.getResultSet();
            rs.next();
            int deltaCount = rs.getInt(1);
            rs.close();
            if (deltaCount != 0) {
                return;
            }
            String sql = "show create table " + delta.getSchema() + "." + delta.getTable();
            stmt.execute(sql);
            rs = stmt.getResultSet();
            String tblDef = null;
            while (rs.next()) {
                tblDef = rs.getString("Create Table");
            }
            rs.close();
            if (tblDef != null) {
                CreateTableSqlModifier createModifier = new CreateTableSqlModifier();
                String newTable = createModifier.modify(tblDef).getResult();
                stmt.execute(newTable);
                deltaTableList.add(delta);
            }
        } catch (SQLException e) {
            logger.error("Error when setup delta table, schema : " + delta.getSchema()
                    + ", table:" + delta.getTable(), e);
            throw new TransactionException(getTransactionID(), e);
        } finally {
            try {
                stmt.close();
                rs.close();
            } catch (Exception ignore) {
            }
        }
    }