[jira] [Created] (FLINK-39170) [Table/SQL] Add async batch lookup join optimization for temporal table joins

["featzhang (Jira)" <ji...@apache.org>]

featzhang created FLINK-39170:
---------------------------------

             Summary: [Table/SQL] Add async batch lookup join optimization for temporal table joins
                 Key: FLINK-39170
                 URL: https://issues.apache.org/jira/browse/FLINK-39170
             Project: Flink
          Issue Type: New Feature
          Components: Table SQL / API, Table SQL / Planner
            Reporter: featzhang


h2. Overview
This feature introduces an async batch lookup join optimization for temporal table joins in Flink's Table API & SQL. The optimization batches multiple lookup requests together to reduce network overhead and improve throughput, particularly beneficial for high-throughput scenarios with frequent dimension table lookups.

h2. Motivation
Currently, Flink's temporal table joins perform individual lookup requests for each input record, which can lead to:
* *High network overhead*: Each lookup request requires a separate network round-trip
* *Suboptimal throughput*: Network latency becomes a bottleneck in high-throughput scenarios  
* *Resource inefficiency*: Frequent small requests don't utilize network bandwidth effectively

h2. Proposed Solution
Implement an async batch lookup join that:
* *Batches lookup requests*: Groups multiple lookup requests together before sending to the external system
* *Configurable batch size*: Allows users to tune batch size based on their specific use case
* *Timeout-based flushing*: Ensures low latency by flushing incomplete batches after a configurable interval
* *Backward compatibility*: Maintains existing behavior when batch mode is disabled (default)

h2. Use Cases
* *Real-time analytics*: High-throughput streaming jobs that need to enrich data with dimension tables
* *Event processing*: Applications processing millions of events per second with frequent lookups
* *Data warehousing*: ETL pipelines that need to join streaming data with slowly changing dimensions

h2. Expected Benefits
* *Improved throughput*: 2-5x improvement in lookup throughput for high-volume scenarios
* *Reduced network overhead*: Fewer network round-trips through batching
* *Better resource utilization*: More efficient use of network bandwidth and external system resources
* *Configurable performance*: Users can tune batch size and flush interval based on their latency/throughput requirements

h2. Configuration Options

h3. table.optimizer.dim-lookup-join.batch.enabled
* *Type*: Boolean
* *Default*: false
* *Description*: Whether to enable the dim table batch lookup join optimization

h3. table.optimizer.dim-lookup-join.batch.size
* *Type*: Integer
* *Default*: 100
* *Description*: The batch size of dim table lookup join. Controls how many lookup requests are batched together.

h3. table.optimizer.dim-lookup-join.batch.flush.millis
* *Type*: Long
* *Default*: 2000L
* *Description*: The flush interval of dim table lookup join in batch mode, in milliseconds. Controls the maximum time to wait before flushing a batch.

h2. Implementation Details

h3. Core Components
# *AsyncBatchLookupJoinRunner*: Main runner for batch async lookup joins without calc
# *AsyncBatchLookupJoinWithCalcRunner*: Runner for batch async lookup joins with post-lookup calculations
# *BatchLookupFunctionWrapper*: Adapter that wraps single lookup functions for batch processing
# *BatchResultFutureWrapper*: Wrapper for handling batch result futures

h3. Architecture
{noformat}
Input Stream → Batch Accumulator → Batch Lookup Function → Result Processor → Output Stream
                     ↓
              Timeout Flusher (configurable interval)
{noformat}

h3. Batch Processing Flow
# *Accumulation*: Input records are accumulated into batches up to the configured batch size
# *Timeout Handling*: If batch size isn't reached within the flush interval, the batch is processed anyway
# *Batch Lookup*: The entire batch is sent to the lookup function as a single request
# *Result Distribution*: Results are distributed back to the corresponding input records
# *Output*: Joined results are emitted to the output stream

h2. Compatibility
* *Backward Compatible*: Existing temporal table joins continue to work unchanged
* *Opt-in Feature*: Must be explicitly enabled via configuration
* *API Compatible*: No changes to existing Table API or SQL syntax

h2. Testing Strategy
* *Unit Tests*: Comprehensive tests for all batch processing components
* *Integration Tests*: End-to-end tests with various batch sizes and flush intervals
* *Performance Tests*: Benchmarks comparing batch vs non-batch performance
* *Configuration Tests*: Validation of all configuration options




--
This message was sent by Atlassian Jira
(v8.20.10#820010)