[FLINK-27622] Make `AsyncDataStream.OutputMode` configurable for table module

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.16.0
• Component/s: Table SQL / API
• Release Note: It is recommend to set the new option 'table.exec.async-lookup.output-mode' to 'ALLOW_UNORDERED' when no stritctly output order is needed, this will yield significant performance gains on append-only streams
• Flags: Important
• Assignee: lincoln lee
• Reporter: lincoln lee
• Created: 16/May/22 03:25
• Updated: 18/Sep/22 13:21
• Resolved: 11/Aug/22 04:27

Description

The `AsyncDataStream.OutputMode` is hardcoded to 'AsyncDataStream.OutputMode.ORDERED' for now:

//   org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecLookupJoin

        // force ORDERED output mode currently, optimize it to UNORDERED
        // when the downstream do not need orderness
        return new AsyncWaitOperatorFactory<>(
                asyncFunc, asyncTimeout, asyncBufferCapacity, AsyncDataStream.OutputMode.ORDERED);

It should be configurable to users same as the other two async options 'table.exec.async-lookup.buffer-capacity' & 'table.exec.async-lookup.timeout'.

Also, there must be some plan validation for correctness concern when output mode is unordered(that's the reason I know why not be exposed before).

Further, we should offer more precisely control for async join operation more than job level config, e.g., an async lookup join hint can do this per-join operation.

It's the time to get this work!

Issue Links

• fixes: FLINK-16332 Support un-ordered mode for async lookup join

Sub-Tasks

• Add 'table.exec.async-lookup.output-mode' to ExecutionConfigOptions: Closed
• Add plan validation when 'table.exec.async-lookup.output-mode' is unordered: Closed
• Add query hint for async lookup join: Closed