...

...

...

JIRA: 

Released: <Flink Version>

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

A lookup A lookup join is typically used to enrich a table with data that is queried from an external system(backed by a lookup source), but some data may not be updated in external systems in a timely manner, and the user want do delayed retry when encounter such unexpected 'missed' data, but can't elegantly implement it for now. Although flink supports various joins(regular joins, interval joins, temporal joins...) for streams, unfortunately the lookup source is often not available as a stream(e.g., Redis).

...

As discussed in the mailing thread, we're planning to introduce an unified hint to support both sync|async lookup and with|without retry, the support matrix will be:

Non target:

1. The lookup sources that can be connected as streams (can use other types of join)
2. Async lookup with retry is not capable for fixed delayed processing for all input data (should use other lighter ways to solve, e.g., pending source consumption or use sync lookup with retry)
3. Do not support retry on exception (let the sql connectors handle it)

...

The available hint options for each mode::

`ExecutionConfigOptions#TABLE_EXEC_ASYNC_LOOKUP_OUTPUT_MODE`.

For these connectors which can have both capabilities of async and sync lookup, our advice for the connector developers are implementing both sync and async interfaces if both capabilities have suitable use cases, the planner will prefer the async one by default, and users can give different option value 'async'='true|false' via the LOOKUP query hint to suggest the planner,  otherwise choose one interface to implement.

...

LOOKUP('table'='dim1', 'async'='true', 'output-mode'='allow-_unordered', 'capacity'='100', 'timeout'='180s')

...

1. LOOKUP('table'='dim1', 'async'='true', 'output-mode'='allow-_unordered')
2. LOOKUP('table'='dim1', 'async'='true', 'timeout'='300s')

...

1. LOOKUP('table'='dim1', 'async'='true', 'output-mode'='allow-_unordered', 'capacity'='100', 'timeout'='180s')
2. LOOKUP('table'='dim1', 'async'='true', 'output-mode'='ordered', 'capacity'='100', 'timeout'='300s')

2. Retry Support

The hint option 'missretry-retrypredicate'='truelookup_miss'  can enable retry on both sync and async lookup. Retry related hint options:

'retry-strategy'='fixed-_delay'
'fixed-delay'='10s'
'max-attempts'='3'

...

LOOKUP('table'='dim1', 'async'='true', 'missretry-retrypredicate'='truelookup_miss', 'retry-strategy'='fixed-_delay', 'fixed-delay'='10s','max-attempts'='3')

...

LOOKUP('table'='dim1', 'async'='false', 'missretry-retrypredicate'='truelookup_miss', 'retry-strategy'='fixed-_delay', 'fixed-delay'='10s','max-attempts'='3')

...

LOOKUP('table'='dim1', 'missretry-retrypredicate'='truelookup_miss', 'retry-strategy'='fixed-_delay', 'fixed-delay'='10s','max-attempts'='3')

For the retry strategy, we plan to support a fixed-_delay retry strategy first, and this can be extended in the future.

...

-- retry triggered by empty result, using 10s fixed-delay strategy, max attempts 3. 
SELECT /*+ LOOKUP('table'='Customers', 'missretry-retrypredicate'='truelookup_miss', 'retry-strategy'='fixed-_delay', 'fixed-delay'='10s','max-attempts'='3') */ 
	o.order_id, o.total, c.country, c.zip
FROM Orders AS o
  JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
    ON o.customer_id = c.id;

...

This feature is backwards compatible and transparently to all connectors. For existing connectors which support lookup, can easily enable async|sync retry via the new join hint.

Add a followup issue to discuss whether to remove the 'lookup.async' option in HBase connector.

Test Plan

Extend existing cases to cover retry scenarios.

...

-- Customers is backed by the JDBC connector and can be used for lookup joins
CREATE TEMPORARY TABLE Customers (
  id INT,
  name STRING,
  country STRING,
  zip STRING
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',
  'table-name' = 'customers',
  -- lookup-_miss.retry-strategy
  'lookup-_miss.retry-strategy'='fixed-delay', -- 'none' by default
  'lookup-_miss.retry-strategy.fixed-delay.delay' = '10 s',
  'lookup-_miss.retry-strategy.fixed-delay.max-attempts' = '3', 
);

-- join operation stays the same.
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
  JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
    ON o.customer_id = c.id;

...