private String parseFields(List<FieldInfo> fields, Node node, String filterPrimaryKey) {
        StringBuilder sb = new StringBuilder();
        for (FieldInfo field : fields) {
            if (StringUtils.isNotBlank(filterPrimaryKey) && field.getName().equals(filterPrimaryKey)) {
                continue;
            }
            sb.append("    `").append(field.getName()).append("` ");
            if (field instanceof MetaFieldInfo) {
                if (!(node instanceof Metadata)) {
                    throw new IllegalArgumentException(String.format("Node: %s is not instance of Metadata",
                            node.getClass().getSimpleName()));
                }
                MetaFieldInfo metaFieldInfo = (MetaFieldInfo) field;
                Metadata metadataNode = (Metadata) node;
                if (!metadataNode.supportedMetaFields().contains(metaFieldInfo.getMetaField())) {
                    throw new UnsupportedOperationException(String.format("Unsupported meta field for %s: %s",
                            metadataNode.getClass().getSimpleName(), metaFieldInfo.getMetaField()));
                }
                sb.append(metadataNode.format(metaFieldInfo.getMetaField()));
            } else {
                sb.append(TableFormatUtils.deriveLogicalType(field.getFormatInfo()).asSummaryString());
            }
            sb.append(",\n");
        }
        if (sb.length() > 0) {
            sb.delete(sb.lastIndexOf(","), sb.length());
        }
        return sb.toString();
    }
Parse fields

@param fields The fields defined in node
@param node The abstract of extract, transform, load
@param filterPrimaryKey filter PrimaryKey, use for mongo
@return Field formats in select sql
inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/parser/impl/FlinkSqlParser.java