public List<String> getConditionFields(SqlNode conditionNode, String specifyTableName, AbstractSideTableInfo sideTableInfo) {
        List<SqlNode> sqlNodeList = Lists.newArrayList();
        ParseUtils.parseAnd(conditionNode, sqlNodeList);
        List<String> conditionFields = Lists.newArrayList();
        for (SqlNode sqlNode : sqlNodeList) {
            if (!SqlKind.COMPARISON.contains(sqlNode.getKind())) {
                throw new RuntimeException("not compare operator.");
            }

            SqlIdentifier left = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[0];
            SqlIdentifier right = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[1];

            String leftTableName = left.getComponent(0).getSimple();
            String rightTableName = right.getComponent(0).getSimple();

            String tableCol = "";
            if (leftTableName.equalsIgnoreCase(specifyTableName)) {
                tableCol = left.getComponent(1).getSimple();
            } else if (rightTableName.equalsIgnoreCase(specifyTableName)) {
                tableCol = right.getComponent(1).getSimple();
            } else {
                throw new RuntimeException(String.format("side table:%s join condition is wrong", specifyTableName));
            }
            tableCol = sideTableInfo.getPhysicalFields().getOrDefault(tableCol, tableCol);
            conditionFields.add(tableCol);
        }

        return conditionFields;
    }
check whether table exists and whether field is in table.
@param sqlNode
@param joinScope
@param conditionNode
core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java