[server] Manage historical lookup local cache capacity

[luoyuxia]

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]

@platinumhamburg Thanks for review. Comments has been addressed.