[server] Manage historical lookup local cache capacity by luoyuxia · Pull Request #3864 · apache/fluss

luoyuxia · 2026-08-05T07:45:52Z

Purpose

Linked issue: closes #3862

Historical partition lookups materialize local Paimon lookup files on TabletServers. This change bounds their configured disk capacity across tables and keeps cached lookupers consistent when table or lake configuration changes.

Brief change log

Add server-wide and per-table historical lookup cache size options.
Reserve capacity per cached table lookuper and use LRU eviction when admission needs space.
Use stable per-table cache directories and clean stale directories during startup.
Recreate cached lookupers after relevant table or lake configuration changes.
Support dynamic server cache-capacity updates through the cluster configuration procedure.
Add focused cache and historical lookup metrics.

Tests

Common configuration tests: 8 passed.
Server cache budget, lifecycle, validation, and metric tests: 23 passed.
Paimon lake lookuper tests: 9 passed.
Flink 1.20 procedure integration tests: 21 passed.

API and Format

Adds server and table configuration options and extends the evolving lake lookup context with a metric recorder. No RPC wire-format or storage-format changes.

Documentation

Configuration descriptions are included. Cache lifecycle and capacity semantics are covered by the historical lookup local cache management design document.

Generative AI disclosure

Codex (GPT-5), following the repository AGENTS.md guidelines.

Copilot

Pull request overview

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

This PR introduces configurable, capacity-bounded local caching for historical partition lake lookups on TabletServers, including dynamic resizing, cache invalidation on relevant config changes, and new operational/latency metrics.

Changes:

Add server-wide and per-table cache size configs and validate table cache size against the server limit.
Implement a reservation-based global cache budget with LRU eviction and lazy handling of dynamic capacity reductions.
Add metrics for cache utilization/eviction, in-flight requests, and lake lookup latency/materialization; update related RPC request metrics routing.

Reviewed changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
fluss-server/src/test/java/org/apache/fluss/server/utils/HistoricalPartitionTableValidationTest.java	Updates validation tests for new cache-size constraints and new validator signature.
fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLookupCacheBudgetManagerTest.java	Adds tests for reservation, replacement, and dynamic limit behavior of the cache budget manager.
fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java	Expands tests for capacity eviction, lazy reconfigure, lookuper invalidation triggers, and concurrency edge cases.
fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java	Extends table validation to enforce per-table cache size > 0 and <= server cache limit.
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java	Uses a shared RPC utility to detect historical lookup batches (removes local helper).
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Wires historical lookup manager reconfiguration and registers new cache/inflight/eviction metrics; records table-level historical lookup metrics.
fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java	Makes TableConfig snapshot updatable (volatile) and exposes update/get methods used during metadata refresh.
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLookupCacheBudgetManager.java	Adds reservation accounting for configured cache capacity with atomic reserve/replace/release operations.
fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java	Implements budget-based cache admission with LRU eviction, stable per-table cache directories, dynamic reconfigure, and metric recording hook.
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TabletServerMetricGroup.java	Adds a metric group to expose in-flight historical partition operation gauges.
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java	Adds per-table historical lookup counters and lake lookup latency/materialization metrics.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java	Tracks server cache limit for validation and updates validation API to include TablePath and global limit.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigUpdater.java	Adds dynamic-config handler to validate/apply server cache limit changes to MetadataManager.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java	Updates table creation path to pass TablePath into validation.
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java	Registers the new dynamic-config updater for historical lookup cache limit changes.
fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java	Marks the new server cache size option as dynamically reconfigurable.
fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java	Adds shared helper to detect historical lookup content in a LookupRequest.
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java	Splits LOOKUP request metrics into normal vs historical lookups.
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java	Routes LOOKUP request metrics based on whether the request contains historical lookup data.
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java	Adds assertions/coverage for lookup file materialization metric recording.
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java	Uses per-table configured cache size for Paimon lookup cache and records lookup latency/materialization via a metric recorder.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java	Extends set/get/reset cluster config procedure IT to include the new server cache size option.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetClusterConfigsProcedure.java	Documents setting the new server cache size option via the procedure.
fluss-common/src/test/java/org/apache/fluss/config/TableConfigTest.java	Adds unit test for the new per-table cache size accessor.
fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java	Adds validation tests for server cache limit and expiration duration constraints.
fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java	Adds per-table stable directory path construction for historical lookup caches.
fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java	Adds metric name constants for inflight historical requests and lake lookup metrics.
fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java	Extends lookup context with a metric recorder callback used by implementations.
fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java	Adds accessor for per-table historical lookup cache max disk size.
fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java	Adds validation for the new server cache limit and cache expiration duration; marks the table option alterable.
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds new server/table config options for cache limit and idle expiration.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot

Pull request overview

Copilot reviewed 31 out of 31 changed files in this pull request and generated no new comments.

Suppressed comments (4)

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java:104

lookupFileMaterializationCount is mutated and read without synchronization/atomicity, but the lookuper can be used concurrently by multiple lookup requests. This introduces a data race (and potentially torn long reads/writes on some platforms) and can also misclassify lookupFileMaterialization in the metric recorder. Use an AtomicLong/LongAdder (and read via get()/sum()) or otherwise synchronize accesses around increment and snapshot/compare.

    private long lookupFileMaterializationCount;

fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java:456

lookupFileMaterializationCount is mutated and read without synchronization/atomicity, but the lookuper can be used concurrently by multiple lookup requests. This introduces a data race (and potentially torn long reads/writes on some platforms) and can also misclassify lookupFileMaterialization in the metric recorder. Use an AtomicLong/LongAdder (and read via get()/sum()) or otherwise synchronize accesses around increment and snapshot/compare.

            lookupFileMaterializationCount++;

fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java:251

This validation rejects only exactly-0 values but would allow negative sizes if they can be parsed/provided (e.g., via table properties), which could later break budget reservation (bytes > 0) or produce misleading behavior. Reject non-positive values by checking <= 0 and keep the error message consistent with that constraint.

        if (tableCacheSize.getBytes() == 0) {
            throw new InvalidConfigException(
                    String.format(
                            "'%s' for table '%s' must be greater than 0 bytes.",
                            ConfigOptions
                                    .TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_SIZE
                                    .key(),
                            tablePath));
        }

fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigUpdater.java:58

Grammar: the message is missing 'to' (i.e., 'greater than or equal to 1 ms'). Updating it improves clarity for operators.

                            "Invalid configuration for %s, it must be greater than or equal 1 ms.",

luoyuxia · 2026-08-06T01:16:16Z

@platinumhamburg Thanks for review. Comments has been addressed.

platinumhamburg

LGTM.

wuchong · 2026-08-10T17:49:47Z

There are "merge" commits, please remove the "merge" commit from the PR's history.

luoyuxia · 2026-08-11T01:56:42Z

There are "merge" commits, please remove the "merge" commit from the PR's history.

merge commit has been removed

wuchong · 2026-08-11T18:18:32Z

+     * Collects historical lookup metrics in the KV table scope and separates lake lookup metrics by
+     * whether the lookup materialized local files.
+     */
+    private static class HistoricalLookupMetricGroup extends AbstractMetricGroup {


we may still need a new MetricGroup for this since we need to register metric with same name but different label.

luoyuxia · 2026-08-12T07:45:48Z

@wuchong Thanks for your review. Comments has been addressed

wuchong

@luoyuxia , I only left some minor comments.

wuchong · 2026-08-12T10:22:31Z

+            // Release local cache files after the disk is write-locked. Idle lookupers close
+            // immediately, while active lookupers close after their final lookup releases them.
+            // Run pending cache maintenance now so removal callbacks are processed promptly.
+            lakeTableLookupers.invalidateAll();


I believe we should avoid invalidating and clearing the cache here, as this would significantly impact online historical lookups. We should aim to maintain a certain level of cache stability. We can remove the invlidateAll() and cleanup() in this PR.

In the future, we could introduce a min-disk-ratio (default=0.01) to manage disk usage. When disk usage is high, we can clear the cache down to the min-disk-ratio to free up space for write operations. When disk usage is low, we can allow the cache to expand up to a max-disk-ratio. (Add this as TODO comment as a future work).

wuchong · 2026-08-12T10:38:49Z

+        checkState(started, "Historical lake lookup manager has not been started.");
        TableBucket tableBucket = lookupData.tableBucket();
+        try {
+            ensureDiskWritable();


We should not call ensureDiskWritable() at the entry point of the lookup, as a lookup operation does not necessarily involve writes. Instead, we should check ensureDiskWritable() within org.apache.fluss.lake.paimon.lookup.PaimonLakeTableLookuper.TrackingIOManager#createChannel(java.lang.String). If a DiskWriteLockedException occurs, it should be wrapped in an IOException and propagated upward. Then, in the exception handling block of org.apache.fluss.lake.lakestorage.LakeTableLookuper#lookup, we can unwrap it back to a DiskWriteLockedException (a retriable exception) before returning it to the client.

This approach ensures that pure lookup operations are not blocked; blocking only occurs when downloading cache files.

wuchong · 2026-08-12T10:40:05Z

                                        CachedLakeTableLookuper cachedLookuper,
-                                        RemovalCause ignoredCause) -> {
+                                        RemovalCause cause) -> {
                                    if (cachedLookuper != null) {
-                                        cachedLookuper.invalidate();
+                                        if (cause == RemovalCause.SIZE) {
+                                            capacityEvictions.inc();
+                                            LOG.info(
+                                                    "Evicted historical lookup cache for table {} (table ID {}) because the cache retains at most {} tables.",
+                                                    cachedLookuper.tablePath,
+                                                    cachedLookuper.tableId,
+                                                    MAX_CACHED_TABLES);
+                                        }
+                                        onLookuperRemoved(cachedLookuper);


Could we simplify this listener to a single onLookuperRemoved(cachedLookuper, cause) call and move the null check, size-eviction accounting/logging, and invalidation into that method? That would keep the removal policy in one place and leave the Caffeine builder easier to scan.

Caffeine.newBuilder() .removalListener(this::onLookuperRemoved); private void onLookuperRemoved( Long ignored, @Nullable CachedLakeTableLookuper cachedLookuper, RemovalCause cause) { if (cachedLookuper == null) { return; } if (cause == RemovalCause.SIZE) { capacityEvictions.inc(); LOG.info( "Evicted historical lookup cache for table {} (table ID {}) because the cache retains at most {} tables.", cachedLookuper.tablePath, cachedLookuper.tableId, MAX_CACHED_TABLES); } cachedLookuper.invalidate(); }

wuchong · 2026-08-12T10:40:50Z

-                new File(
-                        new File(conf.get(ConfigOptions.SERVER_IO_TMP_DIR), PAIMON_LOOKUP_DIR_NAME),
-                        String.valueOf(serverId));
+    private synchronized File getOrCreateHistoricalLookupCacheRootDir() {


Create the historicalLookupCacheRootDir directly during startup to avoid acquiring a lock on every lookup. It is just a directory and does not consume significant space.

wuchong · 2026-08-12T10:41:21Z

+        historicalLakeLookupManager.startup();
+        historicalLakeLookupManager.startLookupCacheDiskSizeMonitor(scheduler);


nit: Can we merge this 2 methods? historicalLakeLookupManager.startup(scheduler)

Allow active historical lookup generations to finish after replacement, refresh lookupers when lake configuration changes, and use table-scoped local cache directories. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 148/173 AI-Contributed/UT: 7/7

Add server and table cache capacity settings, validate table limits, reserve configured capacity atomically, and evict lookupers in best-effort LRU order. Expose cached table and eviction metrics and pass per-table limits to Paimon. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 404/638 AI-Contributed/UT: 256/396

Allow ALTER TABLE to update the effective historical lookup cache size. Propagate the latest table configuration to TabletServer replicas and recreate cached lookupers only when the effective size changes. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 72/74 AI-Contributed/UT: 106/106

Allow the global historical lookup cache capacity to be updated through cluster configuration and apply reductions lazily during admission. Keep coordinator table validation synchronized through a dedicated config updater. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 61/151 AI-Contributed/UT: 63/123

Expose request, table, in-flight, and Paimon materialization metrics for historical lookups.\n\nInvalidate cached lookupers when lake configuration changes and initialize Paimon lookup state only from active data files.\n\nCo-Authored-By: Codex <noreply@openai.com>\nAI-Model: gpt-5\nAI-Contributed/Feature: 444/444\nAI-Contributed/UT: 58/58 AI-Contributed/Feature: 378/444 AI-Contributed/UT: 58/58

Make idle expiration dynamically reconfigurable and use Caffeine access order for capacity eviction. Preserve existing lookup metric labels by exposing historical request rates separately, and validate table cache sizes only when historical lookup is enabled. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 170/197 AI-Contributed/UT: 49/73

Clean cache files left by a previous TabletServer process during startup while keeping directory creation lazy. Tighten historical lookup cache configuration validation and isolate temporary directories in replica tests. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 78/86 AI-Contributed/UT: 42/44

Derive global and per-table cache capacities from the total size of the first available data volume. Store lookup files under that data directory, keep the cache root out of tablet recovery, and propagate resolved byte limits to Paimon. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 164/256 AI-Contributed/UT: 191/286

Exclude the delegating TrackingIOManager inner class from the per-class JaCoCo threshold. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 3/3 AI-Contributed/UT: 0/0

Clarify cache metric naming, lookup lifecycle comments, and data-directory parameters. Document the historical lookup metrics exposed to operators. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 66/81 AI-Contributed/UT: 0/0

Determine lookup request type from the first bucket and validate normal lookup conversion so mixed requests are still rejected in either order. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 20/20 AI-Contributed/UT: 0/0

Exclude the internal historical system partition from list partition responses and extend the existing admin integration test. Leave a TODO to return lake-native partitions in the future. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 5/5 AI-Contributed/UT: 16/16

Read the internal historical partition from ZooKeeper now that public partition listings hide it. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 0/0 AI-Contributed/UT: 23/23

Simplify historical lookup cache capacity to a fixed table limit, integrate disk-write protection and historical metrics, and clean up configuration, lifecycle, naming, and cache observability. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 862/1279 AI-Contributed/UT: 456/636

Check disk write protection only when Paimon downloads lookup cache files, preserve cached lookups under write lock, and simplify cache startup and removal lifecycle. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 157/157 AI-Contributed/UT: 188/188

luoyuxia · 2026-08-12T11:24:05Z

@wuchong Thanks for helpful review. Comments has been addressed.

luoyuxia requested a lite review from Copilot August 5, 2026 07:47

Copilot AI reviewed Aug 5, 2026

View reviewed changes

Copilot started reviewing on behalf of luoyuxia August 5, 2026 08:27 View session

platinumhamburg reviewed Aug 5, 2026

View reviewed changes

Comment thread fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java Outdated

Comment thread fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java Outdated

luoyuxia requested a lite review from Copilot August 5, 2026 09:53

Copilot AI reviewed Aug 5, 2026

View reviewed changes

Copilot started reviewing on behalf of luoyuxia August 5, 2026 10:40 View session

platinumhamburg approved these changes Aug 8, 2026

View reviewed changes

Comment thread fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java Outdated

luoyuxia force-pushed the history-partition-support-finnal-cache-pr4 branch from 97491f3 to 93beba2 Compare August 10, 2026 23:12

wuchong reviewed Aug 11, 2026

View reviewed changes

luoyuxia marked this pull request as draft August 12, 2026 04:15

luoyuxia force-pushed the history-partition-support-finnal-cache-pr4 branch 3 times, most recently from 1e6f784 to 706b7e7 Compare August 12, 2026 07:31

luoyuxia marked this pull request as ready for review August 12, 2026 07:45

wuchong reviewed Aug 12, 2026

View reviewed changes

luoyuxia and others added 10 commits August 12, 2026 19:14

[test] Exclude lookup IO wrapper from coverage

e491803

Exclude the delegating TrackingIOManager inner class from the per-class JaCoCo threshold. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 3/3 AI-Contributed/UT: 0/0

luoyuxia and others added 5 commits August 12, 2026 19:14

[test] Fix historical partition lookup IT case

2466d8c

Read the internal historical partition from ZooKeeper now that public partition listings hide it. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-5 AI-Contributed/Feature: 0/0 AI-Contributed/UT: 23/23

luoyuxia force-pushed the history-partition-support-finnal-cache-pr4 branch from 706b7e7 to 4785814 Compare August 12, 2026 11:15

wuchong approved these changes Aug 12, 2026

View reviewed changes

wuchong merged commit 9ee87d0 into apache:main Aug 12, 2026
20 checks passed

		historicalLakeLookupManager.startup();
		historicalLakeLookupManager.startLookupCacheDiskSizeMonitor(scheduler);

Uh oh!

Conversation

luoyuxia commented Aug 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Generative AI disclosure

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

luoyuxia commented Aug 6, 2026

Uh oh!

platinumhamburg left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

wuchong commented Aug 10, 2026

Uh oh!

luoyuxia commented Aug 11, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

wuchong Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

luoyuxia Aug 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

luoyuxia commented Aug 12, 2026

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

wuchong Aug 12, 2026

Choose a reason for hiding this comment

Uh oh!

wuchong Aug 12, 2026

Choose a reason for hiding this comment

Uh oh!

wuchong Aug 12, 2026

Choose a reason for hiding this comment

Uh oh!

wuchong Aug 12, 2026

Choose a reason for hiding this comment

Uh oh!

wuchong Aug 12, 2026

Choose a reason for hiding this comment

Uh oh!

luoyuxia commented Aug 12, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

luoyuxia commented Aug 5, 2026 •

edited

Loading

luoyuxia Aug 12, 2026 •

edited

Loading