private static List<Index> parseIndexes(CreateTable statement) {

        if (statement.getIndexes() == null) {
            return new ArrayList<>();
        }
        List<Index> indexList = new ArrayList<>();

        for (net.sf.jsqlparser.statement.create.table.Index perIndex : statement.getIndexes()) {
            Index index = new Index();
            switch (perIndex.getType()) {
                case PRIMARY_KEY:
                    index.setIndexType(IndexType.PRIMARY_KEY);
                    break;
                case NORMAL_INDEX:
                    index.setIndexType(IndexType.NORMAL_INDEX);
                    break;
                default:
                    LOG.error("unsupported index type {}", perIndex.getType());
                    break;
            }
            List<String> columns = new ArrayList<>();
            perIndex.getColumnsNames().forEach(columnName -> columns.add(reformatName(columnName)));
            index.setIndexName(reformatName(perIndex.getName()));
            index.setIndexColumns(columns);
            indexList.add(index);
        }

        return indexList;

    }
parse indexes from statement
only support primary key and normal index.
@param statement create table statement
@return list of indexes
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/utils/OperationUtils.java