[flink] Avoid slot deadlock when sorting normalized index keys

[leaves12138]

Purpose

Non-identity sorted indexes, such as multivalue indexes, normalize one source value into multiple index keys. The normalized-key path used keyBy before the heap-and-spill sorter. Replacing it with a custom partitioner avoids Flink inserting a redundant managed-memory sorter, but a pipelined exchange can require the reader and sorter stages to acquire slots concurrently. A constrained batch deployment may then make no progress once the upstream network buffers fill.

Changes

• Partition normalized index keys by build task without keyBy.
• Explicitly use StreamExchangeMode.BATCH, matching the deadlock-breaking approach used by range shuffle.
• Preserve the existing build-task co-location semantics.
• Add tests for the partition mapping and exchange mode.

BTree and scalar bitmap indexes use the identity/global-sort path and are unaffected.

Tests

• SortedIndexTopoBuilderTest
• SortedGlobalIndexITCase#testMultiValueIndex
• A constrained-slot validation with 1,000,006 rows completed with full multivalue-index coverage and content checks.

[JingsongLi]

+1