[flink] Delta Join additional IT tests and docs improvement

[xuyangzhong]

Purpose

This pr is based on #2268 and do some refactoring work.

Linked issue: close #2231

Brief change log

• Added tests for normal PK tables with table.delete.behavior=IGNORE (not just first_row merge engine)
• Added tests for joining on bucket key (not just full primary key)
• Added tests verifying LEFT/RIGHT/FULL OUTER joins fail with appropriate validation error
• Added test verifying cascade joins are not supported
• Updated documentation for Flink 2.2 Delta Join with new supported features and limitations

Tests

• testDeltaJoinWithPrimaryKeyTableNoDeletes - normal PK table with delete.behavior=IGNORE
• testDeltaJoinOnBucketKey - join on bucket key only
• testDeltaJoinFailsWhenFilterOnNonUpsertKeys - filter on non-upsert-key columns fails
• testDeltaJoinOnBucketKey - join on bucket key only (not full PK)
• testDeltaJoinFailsWhenSourceHasDelete - source with DELETE records fails
• testDeltaJoinFailsWhenJoinKeyNotContainIndex - join key not containing index fail
• testDeltaJoinFailsWithLeftJoin - LEFT JOIN wouldn't be converted to DeltaJoin
• testDeltaJoinFailsWithRightJoin - RIGHT JOIN wouldn't be converted to DeltaJoin
• testDeltaJoinFailsWithFullOuterJoin - FULL OUTER JOIN wouldn't be converted to DeltaJoin
• testDeltaJoinFailsWithCascadeJoin - cascade join wouldn't be converted to DeltaJoin

API and Format

No

Documentation

Yes - updated docs/engine-flink/delta-joins.md in Flink 2.2 part.

[wuchong]

Thanks @xuyangzhong and @fresh-borzoni , this is a great and thorough test coverage.

Could you also add tests for partitioned tables?

[wuchong]

I think we should configure all tests to use table.delete.behavior and join on the bucket key by default, as this aligns with the recommended approach documented for users implementing delta joins.

Then I suggest to rename this method to testDeltaJoinOnPrimaryKey and keep this test as is. And rename testDeltaJoinOnBucketKey to testDeltaJoin.

[wuchong]

Add TODO comment above this

        // TODO we can add a UpsertFilterOperator that can convert the un-match-filter UPSERT record
        //  into DELETE record.

[wuchong]

Please add TODO on this

// TODO: this depends on Fluss supports MVCC/point-in-time lookup to support change upsert keys

[wuchong]

Please add comment on this

        // TODO this should be supported in Flink in future for non-deterministic functions before sinking

[fresh-borzoni]

@xuyangzhong Ack. Since you’re pushing new commits, let's continue with this PR.