[
https://issues.apache.org/jira/browse/FLINK-39154?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
featzhang updated FLINK-39154:
------------------------------
Summary: [Table/SQL] Support Async Batch Lookup Join (with Calc) for
Temporal Table Join (was: [Table]Support Async Batch Lookup Join (with Calc)
for Temporal Table Join)
> [Table/SQL] Support Async Batch Lookup Join (with Calc) for Temporal Table
> Join
> -------------------------------------------------------------------------------
>
> Key: FLINK-39154
> URL: https://issues.apache.org/jira/browse/FLINK-39154
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / API, Table SQL / Planner, Table SQL / Runtime
> Reporter: featzhang
> Priority: Major
>
> This issue introduces Async Batch Lookup Join for temporal table joins,
> enabling batch-based asynchronous lookup of dimension tables.
> Currently, async lookup join performs row-by-row asynchronous invocation,
> where each left input row triggers one async request. This leads to:
> * High RPC overhead under large throughput
> * Inefficient utilization of remote dimension stores
> * Increased latency and resource pressure
> This improvement introduces a batch-based async execution model, where
> multiple input rows are buffered and sent in a single async request.
> In addition, this change supports applying a Calc (projection/filter) on the
> dimension table before evaluating the join condition.
> *Motivation*
> In many production scenarios:
> * Dimension lookup backends support batch key query
> * Per-request overhead dominates total cost
> * High QPS streaming jobs create excessive external calls
> Batching lookup requests:
> * Reduces network round-trips
> * Improves throughput
> * Lowers CPU and serialization overhead
> * Reduces pressure on external systems
> *Proposed Changes*
> *1. Runtime*
> Introduce a new async runner:
> {code:java}
> AsyncBatchLookupJoinRunner
> {code}
> {*}Key behaviors{*}:
> * Buffer left input rows and corresponding ResultFutures
> * Trigger flush when: Batch size reaches configured threshold, OR Flush
> interval timeout is reached
> * Invoke async fetcher with List<RowData>
> * Distribute lookup results back to corresponding left rows
> * Support LEFT OUTER JOIN semantics
> * Reuse ResultFuture instances to reduce allocation cost
> If a Calc exists on the temporal table, use:
> {code:java}
> AsyncBatchLookupJoinWithCalcRunner
> {code}
> which applies:
> * Async fetch
> * Convert to internal RowData
> * Apply generated Calc (projection/filter)
> * Apply join condition
> * Produce joined results
> *2. Planner & Code Generation*
> * Extend LookupJoinCodeGenerator to support batch async mode
> * Integrate with existing generated ResultFuture pipeline
> * Support Calc push-down for temporal table
> * Maintain compatibility with join condition filtering
> A new optimizer option is introduced:
> {code:java}
> table.optimizer.dim-lookup-join.batch-enabled
> {code}
> Default: false
> When enabled, planner generates batch async lookup runner instead of
> row-based async runner.
> *3. Tests*
> Enhancements include:
> * Extend in-memory lookup source to support batch key lookup
> * Add IT cases: Async batch temporal join, Async batch join with Calc
> push-down
> Tests verify:
> * Correct join semantics
> * LEFT OUTER JOIN behavior
> * Calc correctness
> * Result ordering and consistency
> *Compatibility & Migration*
> Fully backward compatible
> * Disabled by default
> * No change in SQL semantics
> * No state format changes
> * No public API changes
> *Performance Impact*
> Expected improvements:
> * Reduced async invocation count
> * Lower RPC overhead
> * Improved throughput
> * Better resource utilization
> Particularly beneficial for:
> * High-throughput streaming jobs
> * Remote dimension stores (e.g., HTTP/KV-based lookups)
> * Latency-sensitive real-time pipelines
> *Future Work*
> * Code-generate a fully integrated JoinedRowResultFuture to simplify layering
> * Adaptive batch size tuning
> * Add metrics for batch flush and async latency
> * Unify async batch logic across connectors
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
