feat(flink): Off-heap lookup join cache backed by RocksDB

[wombatu-kun]

Describe the issue this Pull Request addresses

When using Hudi as a lookup join dimension table in Flink, the entire table is loaded into the cache on each reload. If dimension table is large - we got OOM.

Summary and Changelog

Introduced LookupCache — a minimal interface (addRow, getRows, clear, close) that abstracts the cache backend from the lookup function.
HeapLookupCache — wraps the original HashMap behavior; default, backward-compatible.
RocksDBLookupCache — off-heap implementation.

Rows are written to RocksDB one-by-one during reload (no intermediate HashMap) so heap never spikes to O(n)
Each row gets its own RocksDB entry with a compound key _, enabling efficient prefix-scan lookup
Flink's TypeSerializer handles binary serialization; a PassThroughSerializer stores the raw bytes without extra Java-serialization overhead
clear() tears down and re-creates the RocksDBDAO (cleaning up the old temp directory automatically)

Added two new config options to FlinkOptions:
lookup.join.cache.type = heap (default) | rocksdb
lookup.join.rocksdb.path = local directory path, default ${java.io.tmpdir}/hudi-lookup-rocksdb

HoodieLookupFunction refactored to use LookupCache interface.

Also fixes a pre-existing bug where currentCommit was never set after a successful load, causing redundant full reloads on every TTL expiry

Impact

Solved OutOfMemoryError on lookup joins with large Hudi tables.

Risk Level

none

Documentation Update

Need to add two new config options to documentation: lookup.join.cache.type and lookup.join.rocksdb.path.

Contributor's checklist

• Read through contributor's guide
• Enough context is provided in the sections above
• Adequate tests were added if applicable

[hudi-bot]

CI report:

• 5df9db0 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 57.27%. Comparing base (057af9e) to head (5df9db0).

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18231      +/-   ##
============================================
- Coverage     57.28%   57.27%   -0.02%     
+ Complexity    18532    18530       -2     
============================================
  Files          1944     1944              
  Lines        106126   106126              
  Branches      13118    13118              
============================================
- Hits          60793    60781      -12     
- Misses        39610    39620      +10     
- Partials       5723     5725       +2     

Flag	Coverage Δ
hadoop-mr-java-client	45.41% <ø> (ø)
spark-java-tests	47.42% <ø> (-0.01%)	⬇️
spark-scala-tests	45.50% <ø> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.
see 10 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.