private List<StreamField> parseFieldsBySql(String sql) throws JSQLParserException {
        CCJSqlParserManager pm = new CCJSqlParserManager();
        Statement statement = pm.parse(new StringReader(sql));
        List<StreamField> fields = new ArrayList<>();
        if (!(statement instanceof CreateTable)) {
            throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
                    "The SQL statement must be a table creation statement");
        }
        CreateTable createTable = (CreateTable) statement;
        List<ColumnDefinition> columnDefinitions = createTable.getColumnDefinitions();
        // get column definition
        for (int i = 0; i < columnDefinitions.size(); i++) {
            ColumnDefinition definition = columnDefinitions.get(i);
            StreamField streamField = new StreamField();
            // get field name
            String columnName = definition.getColumnName();
            streamField.setFieldName(columnName);

            ColDataType colDataType = definition.getColDataType();
            String sqlDataType = colDataType.getDataType();
            // convert SQL type to Java type
            Class<?> clazz = FieldInfoUtils.sqlTypeToJavaType(sqlDataType);
            if (clazz == Object.class) {
                throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
                        "Unrecognized SQL field type, line: " + (i + 1) + ", type: " + sqlDataType);
            }
            String type = clazz.getSimpleName().toLowerCase();
            streamField.setFieldType(type);
            // get field comment
            List<String> columnSpecs = definition.getColumnSpecs();
            if (CollectionUtils.isNotEmpty(columnSpecs)) {
                int commentIndex = -1;
                for (int csIndex = 0; csIndex < columnSpecs.size(); csIndex++) {
                    String spec = columnSpecs.get(csIndex);
                    if (spec.toUpperCase().startsWith("COMMENT")) {
                        commentIndex = csIndex;
                        break;
                    }
                }
                String comment = null;
                if (-1 != commentIndex && columnSpecs.size() > commentIndex + 1) {
                    comment = columnSpecs.get(commentIndex + 1).replaceAll("['\"]", "");
                }
                streamField.setFieldComment(comment);
            }
            fields.add(streamField);
        }
        return fields;
    }
Parse fields from CSV format

@param statement CSV statement
@return List of StreamField
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java