public static Operation generateOperation(String sql, Map<String, Integer> sqlType) {
        try {
            // now sqlParser don't support first position
            // remove it first and add it later
            boolean endsWithFirst = StringUtils.endsWithIgnoreCase(sql, FIRST);
            if (endsWithFirst) {
                sql = removeFirstFlag(sql);
            }
            Statement statement = CCJSqlParserUtil.parse(sql);
            if (statement instanceof Alter) {
                return parseAlterOperation(
                        (Alter) statement, sqlType, endsWithFirst);
            } else if (statement instanceof CreateTable) {
                return parseCreateTableOperation(
                        (CreateTable) statement, sqlType);
            } else if (statement instanceof Drop) {
                return new DropTableOperation();
            } else if (statement instanceof Truncate) {
                return new TruncateTableOperation();
            } else if (statement instanceof RenameTableStatement) {
                return new RenameTableOperation();
            } else {
                LOG.error("doesn't support sql {}, statement {}", sql, statement);
            }
        } catch (Exception e) {
            LOG.error("parse ddl in sql {} error", sql, e);
        }
        return new UnsupportedOperation();
    }
generate operation from sql and table schema.
@param sql sql from binlog
@param sqlType table sql types
@return Operation
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/utils/OperationUtils.java