[FLINK-36113] Condition 'transform instanceof PhysicalTransformation' is always 'false'

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Fix Version/s: 1.20.1

Language

java

• Assignee: Unassigned
• Reporter: tiancx
• Created: 20/Aug/24 17:30
• Updated: 21/Aug/24 05:28

Description

The legacyTransform method in the StreamGraphGenerator class has a judgment: the transform instance of PhysicalTransformation is always false

//代码占位符
private Collection<Integer> legacyTransform(Transformation<?> transform) {
    Collection<Integer> transformedIds;
    if (transform instanceof FeedbackTransformation<?>) {
        transformedIds = transformFeedback((FeedbackTransformation<?>) transform);
    } else if (transform instanceof CoFeedbackTransformation<?>) {
        transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);
    } else if (transform instanceof SourceTransformationWrapper<?>) {
        transformedIds = transform(((SourceTransformationWrapper<?>) transform).getInput());
    } else {
        throw new IllegalStateException("Unknown transformation: " + transform);
    }

    if (transform.getBufferTimeout() >= 0) {
        streamGraph.setBufferTimeout(transform.getId(), transform.getBufferTimeout());
    } else {
        streamGraph.setBufferTimeout(transform.getId(), getBufferTimeout());
    }

    if (transform.getUid() != null) {
        streamGraph.setTransformationUID(transform.getId(), transform.getUid());
    }
    if (transform.getUserProvidedNodeHash() != null) {
        streamGraph.setTransformationUserHash(
                transform.getId(), transform.getUserProvidedNodeHash());
    }

    if (!streamGraph.getExecutionConfig().hasAutoGeneratedUIDsEnabled()) {
        if (transform instanceof PhysicalTransformation
                && transform.getUserProvidedNodeHash() == null
                && transform.getUid() == null) {
            throw new IllegalStateException(
                    "Auto generated UIDs have been disabled "
                            + "but no UID or hash has been assigned to operator "
                            + transform.getName());
        }

    }

    if (transform.getMinResources() != null && transform.getPreferredResources() != null) {
        streamGraph.setResources(
                transform.getId(),
                transform.getMinResources(),
                transform.getPreferredResources());
    }

    streamGraph.setManagedMemoryUseCaseWeights(
            transform.getId(),
            transform.getManagedMemoryOperatorScopeUseCaseWeights(),
            transform.getManagedMemorySlotScopeUseCases());

    return transformedIds;
} 

If this needs optimization, I am willing to do so