package cn.com.ty.lift.common.sql;

import com.baomidou.mybatisplus.core.parser.SqlParserHelper;
import lombok.Data;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.expression.Alias;
import net.sf.jsqlparser.expression.BinaryExpression;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.Function;
import net.sf.jsqlparser.expression.operators.conditional.OrExpression;
import net.sf.jsqlparser.expression.operators.relational.InExpression;
import net.sf.jsqlparser.expression.operators.relational.ItemsList;
import net.sf.jsqlparser.expression.operators.relational.NotEqualsTo;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Column;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.delete.Delete;
import net.sf.jsqlparser.statement.select.*;
import net.sf.jsqlparser.statement.update.Update;
import org.apache.ibatis.executor.statement.StatementHandler;
import org.apache.ibatis.mapping.MappedStatement;
import org.apache.ibatis.mapping.SqlCommandType;
import org.apache.ibatis.plugin.*;
import org.apache.ibatis.reflection.MetaObject;
import org.apache.ibatis.reflection.SystemMetaObject;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * 由于开发人员水平参差不齐，即使订了开发规范很多人也不遵守
 * <p>SQL是影响系统性能最重要的因素，所以拦截掉垃圾SQL语句</p>
 * <br>
 * <p>拦截SQL类型的场景</p>
 * <p>1.必须使用到索引，包含left join连接字段，符合索引最左原则</p>
 * <p>必须使用索引好处，</p>
 * <p>1.1 如果因为动态SQL，bug导致update的where条件没有带上，全表更新上万条数据</p>
 * <p>1.2 如果检查到使用了索引，SQL性能基本不会太差</p>
 * <br>
 * <p>2.SQL尽量单表执行，有查询left join的语句，必须在注释里面允许该SQL运行，否则会被拦截，有left join的语句，如果不能拆成单表执行的SQL，请leader商量在做</p>
 * <p>https://gaoxianglong.github.io/shark</p>
 * <p>SQL尽量单表执行的好处</p>
 * <p>2.1 查询条件简单、易于开理解和维护；</p>
 * <p>2.2 扩展性极强；（可为分库分表做准备）</p>
 * <p>2.3 缓存利用率高；</p>
 * <p>2.在字段上使用函数</p>
 * <br>
 * <p>3.where条件为空</p>
 * <p>4.where条件使用了 !=</p>
 * <p>5.where条件使用了 not 关键字</p>
 * <p>6.where条件使用了 or 关键字</p>
 * <p>7.where条件使用了 使用子查询</p>
 *
 * @author willenfoo
 * @since 2018-03-22
 */
@Slf4j
@Intercepts({@Signature(type = StatementHandler.class, method = "prepare", args = {Connection.class, Integer.class})})
public class SqlIllegalInterceptor implements Interceptor {

    /**
     * cache the legal sql result.
     */
    private final Set<String> legalSqlCacheSet  = new HashSet<>();
    /**
     * cache the index info exist on the column.
     */
    private static final Set<String> indexInfoCacheSet = new CopyOnWriteArraySet<>();
    /**
     * cache the table had load index info from the connection.
     */
    private static final Set<String> tableCacheSet     = new HashSet<>();
    @Getter
    @Setter
    private       boolean     devMode           = true;
    @Getter
    @Setter
    private       boolean     sqlWrite          = false;
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        StatementHandler statementHandler = SqlUtils.realTarget(invocation.getTarget());
        MetaObject metaObject = SystemMetaObject.forObject(statementHandler);
        // 如果是insert操作， 或者 @SqlParser(filter = true) 跳过该方法解析 ， 不进行验证
        MappedStatement mappedStatement = SqlUtils.getMappedStatement(metaObject);
        if (SqlCommandType.INSERT.equals(mappedStatement.getSqlCommandType()) || SqlParserHelper.getSqlParserInfo(metaObject)) {
            return invocation.proceed();
        }
        String originalSql = statementHandler.getBoundSql().getSql();
        log.info("Verifying SQL ID: {}", mappedStatement.getId());
        if (sqlWrite) {
            log.info("Verifying SQL:");
            log.info(originalSql);
        }
        String sqlEncode = SqlUtils.sqlEncode(originalSql);
        if (legalSqlCacheSet.contains(sqlEncode)) {
            log.info("Original Sql is legal, No need verify.");
            return invocation.proceed();
        }

        //get databaseMetaData & catalog & schema from connection, and put to local thread.
        Object firstArg = invocation.getArgs()[0];
        if(!(firstArg instanceof Connection)){
            return invocation.proceed();
        }
        Connection con = (Connection) firstArg;
        //the illegal rule.
        IllegalRule illegalRule = new IllegalRule(con.getMetaData(), con.getCatalog(), con.getSchema(), originalSql);

        // if any to verify, just return.
        if (!illegalRule.verifiable()) {
            illegalRule.error();
            return invocation.proceed();
        }

        // if the sql is legal, cache the sql encode.
        if (illegalRule.verify()) {
            legalSqlCacheSet.add(sqlEncode);
        } else {
            if (devMode) {
                illegalRule.warn();
            } else {
                illegalRule.error();
            }
        }
        return invocation.proceed();
    }

    @Override
    public Object plugin(Object target) {
        if (target instanceof StatementHandler) {
            return Plugin.wrap(target, this);
        }
        return target;
    }

    @Override
    public void setProperties(Properties prop) {
        String devMode = prop.getProperty("devMode");
        String sqlWrite = prop.getProperty("sqlWrite");
        if (SqlUtils.isBoolean(devMode)) {
            this.devMode = Boolean.valueOf(devMode);
        }
        if (SqlUtils.isBoolean(devMode)) {
            this.sqlWrite = Boolean.valueOf(sqlWrite);
        }
    }

    /**
     * the rules of illegal sql.
     */
    private static class IllegalRule {
        // the list of where exp parse from the sql to verify.
        private List<Expression> wheres         = new ArrayList<>();
        // the list of index info parse from the sql to verify.
        private List<IndexInfo>  indexInfos     = new ArrayList<>();
        // the verify info parse from sql, include WHERE & TABLE & JOINS.
        private List<VerifyInfo> verifyInfos    = new ArrayList<>();
        // the verify message.
        private List<String>     verifyMessage  = new ArrayList<>();
        private DatabaseMetaData databaseMetaData;
        private String           catalog;
        private String           schema;

        /**
         * the constructor with a original sql.
         *
         * @param originalSql the original sql string.
         */
        private IllegalRule(DatabaseMetaData databaseMetaData, String catalog, String schema, String originalSql) {
            this.databaseMetaData = databaseMetaData;
            this.catalog = catalog;
            this.schema = schema;
            // parse original sql
            parseSql(originalSql);
        }

        private void addMessage(String message, Object... values){
            if(message.contains("%s") && null != values && values.length > 0){
                verifyMessage.add(String.format(message, values));
            }else {
                verifyMessage.add(message);
            }
        }
        /**
         * verify no use WHERE.
         *
         * @param where a where {@link Expression}
         */
        private int whereNoUse(Expression where) {
            if (SqlUtils.isNull(where)) {
                addMessage("illegal sql. not have WHERE condition.");
                return 0;
            }
            return 1;
        }

        /**
         * verify use OR expression in WHERE.
         *
         * @param where a where {@link Expression}
         */
        private int whereHasOr(Expression where) {
            if (where instanceof OrExpression) {
                OrExpression or = (OrExpression) where;
                addMessage("illegal use [%s] in WHERE, will cause full table scan, try UNION ALL. sql: %s", or.getStringExpression(), or.toString());
                return 0;
            }
            return 1;
        }

        /**
         * verify use NotEqualsTo expression in WHERE.
         *
         * @param where a where {@link Expression}
         */
        private int whereHasNotEqualsTo(Expression where) {
            if (where instanceof NotEqualsTo) {
                NotEqualsTo notEqualsTo = (NotEqualsTo) where;
                addMessage("illegal use [%s] in WHERE, will cause full table scan. sql: %s", notEqualsTo.getStringExpression(), notEqualsTo.toString());
                return 0;
            }
            return 1;
        }

        /**
         * verify use binary expression in WHERE.
         * 1. NOT binary expression
         * 2. the left of binary is FUNCTION.
         * 3. the right of binary is SubSelect.
         *
         * @param where a where {@link Expression}
         */
        private int whereHasNotBinary(Expression where) {
            if (where instanceof BinaryExpression) {
                BinaryExpression binary = (BinaryExpression) where;
                if (binary.isNot()) {
                    addMessage("illegal use [%s] in WHERE, will cause full table scan. Sql: %s", binary.getStringExpression(), binary.toString());
                    return 0;
                }
                Expression leftExpression = binary.getLeftExpression();
                if (leftExpression instanceof Function) {
                    Function function = (Function) leftExpression;
                    addMessage("illegal use db function in WHERE, will cause full table scan. sql: %s", function.toString());
                    return 0;
                }
                Expression rightExpression = binary.getRightExpression();
                if (rightExpression instanceof SubSelect) {
                    SubSelect subSelect = (SubSelect) rightExpression;
                    addMessage("illegal use subSelect in WHERE, will cause full table scan. sql: %s", subSelect.toString());
                    return 0;
                }
            }
            return 1;
        }

        /**
         * verify use IN expression and the rightItem of IN is SubSelect in WHERE.
         *
         * @param where a where {@link Expression}
         */
        private int whereHasInSubSelect(Expression where) {
            if (where instanceof InExpression) {
                InExpression in = (InExpression) where;
                ItemsList rightItemsList = in.getRightItemsList();
                if (rightItemsList instanceof SubSelect) {
                    SubSelect subSelect = (SubSelect) rightItemsList;
                    addMessage("illegal use IN (subSelect) in WHERE, Replacing IN with EXISTS(subSelect) is better. sql: %s", subSelect.toString());
                    return 0;
                }
            }
            return 1;
        }

        /**
         * verify check NULL in WHERE.
         *
         * @param where a where {@link Expression}
         */
        private int whereCheckNull(Expression where) {
            String lower = where.toString().toLowerCase();
            boolean checkNull = lower.contains("is null") || lower.contains("is not null");
            if (checkNull) {
                addMessage("illegal check null in WHERE, will cause full table scan, set a default value on column and equals it. sql: %s", where);
                return 0;
            }
            return 1;
        }

        /**
         * verify whether use index on the column.
         *
         * @param indexInfo include catalog & schema & tableName & columnName to Specify a unique index.
         */
        private int columnNoUseIndex(IndexInfo indexInfo) {
            String fullyQualifiedColumn = indexInfo.fullyQualifiedColumn();
            //if the column has set. and the cache contains the column.
            if (indexInfo.fullyColumn() && indexInfoCacheSet.contains(fullyQualifiedColumn)) {
                return 1;
            }
            //if the table has set. and the cache do not contains the table.
            String fullyQualifiedTable = indexInfo.fullyQualifiedTable();
            // if new create index on the column. must to reload index,
            if (indexInfo.fullyTable() && tableCacheSet.contains(fullyQualifiedTable)) {
                addMessage("Index not use on %s. Example sql: CREATE [UNIQUE|CLUSTERED] INDEX idx_%s ON %s(%s)", fullyQualifiedColumn, indexInfo.columnName, indexInfo.tableName, indexInfo.columnName);
                return 0;
            }
            //get index info from the connection.
            try (ResultSet resultSet = databaseMetaData.getIndexInfo(indexInfo.catalog, indexInfo.schema, indexInfo.tableName, false, true)) {
                int indexCount = 0;
                while (resultSet.next()) {
                    //索引中的列序列号等于1，才有效, index: 8, label: ORDINAL_POSITION
                    if (resultSet.getShort("ORDINAL_POSITION") == 1) {
                        //在索引中的列名列 index：9, label: COLUMN_NAME
                        IndexInfo index = IndexInfo.newInstance();
                        //TABLE_CAT
                        index.setCatalog(indexInfo.catalog);
                        //TABLE_SCHEM
                        index.setSchema(indexInfo.schema);
                        //TABLE_NAME
                        index.setTableName(indexInfo.tableName);
                        //COLUMN_NAME
                        index.setColumnName(resultSet.getString("COLUMN_NAME"));
                        indexInfoCacheSet.add(index.fullyQualifiedColumn());
                        indexCount++;
                        //cache table, no reload index every time.
                        tableCacheSet.add(index.fullyQualifiedTable());
                    }
                }
                if (indexCount > 6) {
                    addMessage("Too many indexes (%s > 6) use on table (%s).", indexCount, fullyQualifiedTable);
                }
                log.info("Cache index: {}, table: {}", indexInfoCacheSet.size(), tableCacheSet.size());
            } catch (SQLException e) {
                log.error("get index info from DatabaseMetaData fail.");
            }
            if (indexInfo.fullyColumn() && indexInfoCacheSet.contains(fullyQualifiedColumn)) {
                return 1;
            }
            addMessage("Index not use on %s. Example sql: CREATE [UNIQUE|CLUSTERED] INDEX idx_%s ON %s(%s)", fullyQualifiedColumn, indexInfo.columnName, indexInfo.tableName, indexInfo.columnName);
            return 0;
        }

        /**
         * parse original sql to WHERE & TABLE & JOIN TABLES.
         */
        private void parseSql(String originalSql) {
            try {
                log.info("Parsing original sql.");
                if(null == originalSql || originalSql.trim().isEmpty()){
                    addMessage("The original sql is null or empty.");
                    return;
                }
                Statement statement = CCJSqlParserUtil.parse(originalSql);
                //select <SelectItem> from <FromItem> where <Expression>
                if (isSelect(statement)) {
                    //SELECT t.* FROM (SELECT * FROM t1 LEFT JOIN t2 ON t1.c1 = t2.c1 WHERE ...) t WHERE t....
                    log.info("original sql is SELECT.");
                    SelectBody selectBody = ((Select) statement).getSelectBody();
                    if(isPlainSelect(selectBody)){
                        parseSelect((PlainSelect) selectBody);
                    }
                } else if (isUpdate(statement)) {
                    //UPDATE table SET column1 = v1 WHERE column2 = v2
                    log.info("original sql is UPDATE.");
                    verifyInfos.add(VerifyInfo.newInstance().parse((Update) statement));
                } else if (isDelete(statement)) {
                    //DELETE FROM table WHERE column = v
                    log.info("original sql is DELETE.");
                    verifyInfos.add(VerifyInfo.newInstance().parse((Delete) statement));
                }
            } catch (Exception e) {
                verifyInfos.clear();
                addMessage("Parse original sql fail. %s", e.getMessage());
            }
        }

        /**
         * parse plain select to WHERE & TABLE & JOIN TABLES.
         *
         * @param plainSelect a plain select.
         */
        private void parseSelect(PlainSelect plainSelect) {
            verifyInfos.add(VerifyInfo.newInstance().parse(plainSelect));
            FromItem fromItem = plainSelect.getFromItem();
            //if is SubSelect, recursion parse select.
            if (isSubSelect(fromItem)) {
                SelectBody selectBody = ((SubSelect) fromItem).getSelectBody();
                if (isPlainSelect(selectBody)) {
                    parseSelect((PlainSelect) selectBody);
                }
            }
        }

        /**
         * parse a join to WHERE & INDEX
         */
        private void parseJoin(VerifyInfo verifyInfo) {
            Expression on = verifyInfo.getWhere();
            Table table = verifyInfo.getTable();
            List<Table> joinTables = verifyInfo.getJoinTables();
            wheres.add(on);
            if (!isBinary(on)) {
                return;
            }
            BinaryExpression binary = (BinaryExpression) on;
            //获得左边表达式
            Expression leftExp = binary.getLeftExpression();
            //获得右边表达式
            Expression rightExp = binary.getRightExpression();
            wheres.add(rightExp);
            if (isBinary(leftExp)) {
                verifyInfo.setWhere(leftExp);
                parseJoin(verifyInfo);
            } else if (isColumn(leftExp) && isColumn(rightExp)) {
                // the alias of primary table
                Alias alias = table.getAlias();
                Column leftCol = (Column) leftExp;
                String leftColumn = leftCol.getColumnName();
                Table leftTable = leftCol.getTable();
                Column rightCol = (Column) rightExp;
                String rightColumn = rightCol.getColumnName();
                Table rightTable = rightCol.getTable();
                // if left or right belong to the main table, that verify the index info.
                // if the alias or the table name equals.
                if (SqlUtils.notNull(leftTable, rightTable)) {
                    String leftTableName = leftTable.getName();
                    String rightTableName = rightTable.getName();
                    if (SqlUtils.notNull(alias) && Objects.equals(leftTableName, alias.getName())) {
                        parseTable(table, leftColumn);
                    } else if (Objects.equals(leftTableName, table.getName())) {
                        parseTable(table, leftColumn);
                    } else if (SqlUtils.notNull(alias) && Objects.equals(rightTableName, alias.getName())) {
                        parseTable(table, rightColumn);
                    } else if (Objects.equals(rightTableName, table.getName())) {
                        parseTable(table, rightColumn);
                    } else if (verifyInfo.hasJoinTables()) {
                        // if left and right both don't belong to main table, find in the join tables.
                        addMessage("Join column does not contain primary table (%s). sql: %s", table, binary);
                        // the column is belong to the join tables.
                        for (Table joinTable : joinTables) {
                            Alias joinTableAlias = joinTable.getAlias();
                            if (SqlUtils.notNull(joinTableAlias) && Objects.equals(joinTableAlias.getName(), leftTableName)) {
                                parseTable(joinTable, leftColumn);
                            } else if (SqlUtils.notNull(joinTableAlias) && Objects.equals(joinTableAlias.getName(), rightTableName)) {
                                parseTable(joinTable, rightColumn);
                            } else if (Objects.equals(joinTable.getName(), leftTableName)) {
                                parseTable(joinTable, leftColumn);
                            } else if (Objects.equals(joinTable.getName(), rightTableName)) {
                                parseTable(joinTable, rightColumn);
                            }
                        }
                    }
                }
            }
        }

        /**
         * parse where exp to WHERE & INDEX.
         */
        private void parseWhere(VerifyInfo verifyInfo) {
            Expression where = verifyInfo.getWhere();
            Table table = verifyInfo.getTable();
            List<Table> joinTables = verifyInfo.getJoinTables();
            wheres.add(where);
            if (!isBinary(where)) {
                return;
            }
            BinaryExpression binary = (BinaryExpression) where;
            //获得右边表达式
            Expression rightExp = binary.getRightExpression();
            wheres.add(rightExp);
            //获得左边表达式
            Expression leftExp = binary.getLeftExpression();
            if (isBinary(leftExp)) {
                verifyInfo.setWhere(leftExp);
                parseWhere(verifyInfo);
            } else if (isColumn(leftExp)) {
                Column leftCol = (Column) leftExp;
                Table leftTable = leftCol.getTable();
                Alias alias = table.getAlias();
                String leftColumn = leftCol.getColumnName();
                // with table alias
                if (SqlUtils.notNull(leftTable, alias) && Objects.equals(leftTable.getName(), alias.getName())) {
                    parseTable(table, leftColumn);
                    // single table select without alias
                } else if (SqlUtils.isNull(leftTable, alias)) {
                    parseTable(table, leftColumn);
                } else if (verifyInfo.hasJoinTables()) {
                    // the column is belong to join tables.
                    addMessage("Column in the WHERE is not of the primary table (%s). sql: %s", table, binary);
                    for (Table joinTable : joinTables) {
                        Alias joinTableAlias = joinTable.getAlias();
                        if (SqlUtils.notNull(joinTableAlias) && Objects.equals(joinTableAlias.getName(), leftTable.getName())) {
                            parseTable(joinTable, leftColumn);
                        } else if (Objects.equals(joinTable.getName(), leftTable.getName())) {
                            parseTable(joinTable, leftColumn);
                        }
                    }
                }
            }
        }

        /**
         * parse joins exp to to WHERE & INDEX
         */
        private void parseJoins(VerifyInfo verifyInfo) {
            //允许执行join，验证join是否使用索引等等
            if (verifyInfo.hasJoins()) {
                for (Join join : verifyInfo.getJoins()) {
                    verifyInfo.setWhere(join.getOnExpression());
                    parseJoin(verifyInfo);
                }
            }
        }

        private void parseOrderBy(VerifyInfo verifyInfo) {
            List<OrderByElement> orderByElements = verifyInfo.getOrderByElements();
            Table table = verifyInfo.getTable();
            List<Table> joinTables = verifyInfo.getJoinTables();
            if (SqlUtils.notEmpty(orderByElements)) {
                for (OrderByElement orderByElement : orderByElements) {
                    Expression orderBy = orderByElement.getExpression();
                    if (!isColumn(orderBy)) {
                        continue;
                    }
                    Column orderByCol = (Column) orderBy;
                    String orderByColumnName = orderByCol.getColumnName();
                    Table orderByTable = orderByCol.getTable();
                    Alias alias = table.getAlias();
                    if (SqlUtils.notNull(orderByTable, alias) && Objects.equals(orderByTable.getName(), alias.getName())) {
                        parseTable(table, orderByColumnName);
                    } else if (SqlUtils.isNull(orderByTable, alias)) {
                        parseTable(table, orderByColumnName);
                    } else if (verifyInfo.hasJoinTables()) {
                        // the column is belong to join tables.
                        addMessage("Column in ORDER BY is not of the primary table (%s). sql: %s", table, orderByElement);
                        for (Table joinTable : joinTables) {
                            Alias joinTableAlias = joinTable.getAlias();
                            if (SqlUtils.notNull(joinTableAlias) && Objects.equals(joinTableAlias.getName(), orderByTable.getName())) {
                                parseTable(joinTable, orderByColumnName);
                            } else if (Objects.equals(joinTable.getName(), orderByTable.getName())) {
                                parseTable(joinTable, orderByColumnName);
                            }
                        }
                    }
                }
            }
        }

        /**
         * parse table and it's column to get index.
         *
         * @param table      a table.
         * @param columnName the column belong to the table.
         */
        private void parseTable(Table table, String columnName) {
            String tableInfo = table.getName();
            //table : db.tableName
            IndexInfo indexInfo = new IndexInfo();
            String[] tableArray = tableInfo.split("\\.");
            if (tableArray.length == 1) {
                indexInfo.setTableName(tableArray[0]);
            } else {
                indexInfo.setTableName(tableArray[1]);
            }
            indexInfo.setCatalog(catalog);
            indexInfo.setSchema(schema);
            indexInfo.setColumnName(columnName);
            indexInfos.add(indexInfo);
        }

        /**
         * check whether the sql is verifiable.
         */
        private boolean verifiable() {
            return !verifyInfos.isEmpty();
        }

        /**
         * verify the where exp & index info parse from the sql. return finally.
         */
        private boolean verify() {
            if (verifyInfos.isEmpty()) {
                return false;
            }
            log.info("Parsing all expressions of sql.");
            for (VerifyInfo verifyInfo : verifyInfos) {
                if (verifyInfo.hasTable()) {
                    log.info("===============================");
                    parseWhere(verifyInfo);
                    parseJoins(verifyInfo);
                    parseOrderBy(verifyInfo);
                } else {
                    wheres.add(verifyInfo.getWhere());
                }
            }

            int whereTotal = wheres.size();
            int indexTotal = indexInfos.size();

            log.info("parse expression complete, WHERE: {}, INDEX: {}.", whereTotal, indexTotal);
            int verifyCount = 0;
            if(!wheres.isEmpty()){
                log.info("Verifying all WHERE expressions.");
                for (int i = 0; i < whereTotal; i++) {
                    Expression where = wheres.get(i);
                    log.info("{}. where: {}", (i + 1), where);
                    int whereNoUse = whereNoUse(where);
                    verifyCount += whereNoUse;
                    if (whereNoUse > 0) {
                        verifyCount += whereHasOr(where);
                        verifyCount += whereHasNotEqualsTo(where);
                        verifyCount += whereHasNotBinary(where);
                        verifyCount += whereHasInSubSelect(where);
                        verifyCount += whereCheckNull(where);
                    }
                }
            }
            if(!indexInfos.isEmpty()){
                log.info("Verifying all INDEX on column.");
                for (int i = 0; i < indexTotal; i++) {
                    IndexInfo indexInfo = indexInfos.get(i);
                    log.info("{}. indexInfo: {}", (i + 1), indexInfo.fullyQualifiedColumn());
                    verifyCount += columnNoUseIndex(indexInfo);
                }
            }
            return verifyCount == whereTotal * 6 + indexTotal;
        }

        public void warn() {
            if (verifyMessage.isEmpty()) {
                log.warn("no any message.");
            } else {
                verifyMessage.forEach(m -> {
                    log.warn("Warn: {}", m);
                });
            }
        }

        public void error() {
            if (verifyMessage.isEmpty()) {
                log.error("no any message.");
            } else {
                verifyMessage.forEach(m -> {
                    log.error("Error: {}", m);
                });
            }
        }
        private boolean isSelect(Statement statement) {
            return (statement instanceof Select);
        }

        private boolean isUpdate(Statement statement) {
            return (statement instanceof Update);
        }

        private boolean isDelete(Statement statement) {
            return (statement instanceof Delete);
        }

        private boolean isPlainSelect(Object object) {
            return (object instanceof PlainSelect);
        }

        private boolean isSubSelect(Object object) {
            return (object instanceof SubSelect);
        }

        private boolean isBinary(Expression expression) {
            return (expression instanceof BinaryExpression);
        }

        private boolean isColumn(Expression expression) {
            return (expression instanceof Column);
        }
    }

    @Data
    @Accessors(chain = true)
    @ToString
    private static class VerifyInfo {
        private Expression           where;
        private Table                table;
        private List<Join>           joins;
        private List<OrderByElement> orderByElements;
        // the list tables in the JOIN condition.
        private List<Table>          joinTables = new ArrayList<>();

        private static VerifyInfo newInstance(){
            return new VerifyInfo();
        }

        private void parseJoinTables() {
            if (hasJoins()) {
                for (Join join : joins) {
                    FromItem rightItem = join.getRightItem();
                    if(rightItem instanceof Table){
                        joinTables.add((Table) rightItem);
                    }
                }
            }
        }

        private VerifyInfo parse(PlainSelect plainSelect) {
            this.setWhere(plainSelect.getWhere());
            this.setJoins(plainSelect.getJoins());
            this.setOrderByElements(plainSelect.getOrderByElements());
            FromItem fromItem = plainSelect.getFromItem();
            // if is Table
            if (fromItem instanceof Table) {
                this.setTable((Table) fromItem);
            }
            parseJoinTables();
            return this;
        }

        private VerifyInfo parse(Update update) {
            this.setWhere(update.getWhere());
            this.setJoins(update.getJoins());
            List<Table> tables = update.getTables();
            if (!tables.isEmpty()) {
                this.setTable(tables.get(0));
            }
            parseJoinTables();
            return this;
        }

        private VerifyInfo parse(Delete delete) {
            this.setWhere(delete.getWhere());
            this.setJoins(delete.getJoins());
            this.setTable(delete.getTable());
            parseJoinTables();
            return this;
        }

        private boolean hasWhere() {
            return null != where;
        }

        private boolean hasTable() {
            return null != table;
        }

        private boolean hasJoins() {
            return null != joins && !joins.isEmpty();
        }

        private boolean hasJoinTables() {
            return !joinTables.isEmpty();
        }
    }

    /**
     * the information of the index.
     */
    @Data
    @Accessors(chain = true)
    @ToString
    private static class IndexInfo {
        private String catalog;
        private String schema;
        private String tableName;
        private String columnName;

        private static IndexInfo newInstance() {
            return new IndexInfo();
        }

        private boolean fullyColumn() {
            return SqlUtils.notNull(getTableName(), getColumnName());
        }

        private boolean fullyTable() {
            return SqlUtils.notNull(getTableName());
        }

        private String fullyQualifiedColumn() {
            return fullyColumn() ? String.format("%s.%s.%s.%s", fixCatalog(), fixSchema(), getTableName(), getColumnName()) : null;
        }

        private String fullyQualifiedTable() {
            return fullyTable() ? String.format("%s.%s.%s", fixCatalog(), fixSchema(), getTableName()) : null;
        }

        private String fixCatalog() {
            return null == this.catalog ? "<catalog>" : this.catalog;
        }

        private String fixSchema() {
            return null == this.schema ? "<schema>" : this.schema;
        }
    }
}