Self join with cache will cause repeated results

For a workflow with a self-join (see example below), it will have incorrect output reading from cache.

Say normally the join operator would take 1K input tuples, because the 500 output tuples from distinct is joining themselves (500 + 500 = 1K).

However, if we cache the result of the distinct operator, the next run from the cache will result in more input data (more than 1K) for the join: