...

LookupJoinRules would check whether FlinkLogicalJoin contain USE_HASH Hint. If yes, and the rules require the input must have hash distribution on join keys when converting FlinkLogicalJoin to LookupJoin .

Note: If the input stream is not an insert stream, which means, it could contain update_before, update_after or delete record, if it's upsert key is different with join key, update before and update may be sent to different tasks after hash partition which may leads to wrong result.

So hash lookup Join requires that the input stream should be insert_only stream or it's upsert keys contains hash keys.

Compatibility, Deprecation, and Migration Plan

...