public static DistinctNode createDistinctNode(DeDuplicationDefinition deDuplicationDefinition,
            TransformResponse transformResponse, Map<String, StreamField> constantFieldMap) {
        List<StreamField> streamFields = deDuplicationDefinition.getDupFields();
        List<FieldInfo> distinctFields = streamFields.stream()
                .map(FieldInfoUtils::parseStreamField)
                .collect(Collectors.toList());
        StreamField timingField = deDuplicationDefinition.getTimingField();
        FieldInfo orderField = FieldInfoUtils.parseStreamField(timingField);
        DeDuplicationStrategy deDuplicationStrategy = deDuplicationDefinition.getDeDuplicationStrategy();
        OrderDirection orderDirection;
        switch (deDuplicationStrategy) {
            case RESERVE_LAST:
                orderDirection = OrderDirection.DESC;
                break;
            case RESERVE_FIRST:
                orderDirection = OrderDirection.ASC;
                break;
            default:
                throw new UnsupportedOperationException(
                        String.format("Unsupported deduplication strategy=%s for inlong", deDuplicationStrategy));
        }
        TransformNode transformNode = createNormalTransformNode(transformResponse, constantFieldMap);
        return new DistinctNode(transformNode.getId(),
                transformNode.getName(),
                transformNode.getFields(),
                transformNode.getFieldRelations(),
                transformNode.getFilters(),
                transformNode.getFilterStrategy(),
                distinctFields,
                orderField,
                orderDirection);

    }
Create distinct node based on deDuplicationDefinition
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/TransformNodeUtils.java