Unbounded growth of BaseHoodieTableFileIndex.cachedAllInputFileSlices in a long-lived Spark session

[linliu-code]

Bug Description

BaseHoodieTableFileIndex.cachedAllInputFileSlices has no eviction of any kind. In a
long-lived, read-only Spark session it grows monotonically with the number of distinct
partitions queried and is never reset, because Spark reuses one HoodieFileIndex instance per
catalog table for the life of the session.

Mechanism

cachedAllInputFileSlices (BaseHoodieTableFileIndex.java:136, a plain HashMap by default)
is inserted into only via putAll at :267, and shrinks only when doRefresh() replaces it
wholesale at :576/:580. doRefresh() has exactly two call sites — the constructor at
:201 and refresh() at :455 — and neither is on the read query path. There is no remove,
no clear, no size cap, no LRU and no TTL.

Because Spark caches the resolved relation for a catalog table, the same HoodieFileIndex
serves every query in the session, so nothing resets the map in a read-only workload.

Exposure grows as (partitions touched x tables queried), bounded above only by Spark's
spark.sql.filesourceTableRelationCacheSize (default 1000). A long-lived thrift server,
notebook, or SQL gateway over a daily-partitioned table accumulates indefinitely.

Note that the caching itself is working as intended and is valuable: an exact repeat of a query
costs no file-slice resolution at all, and a changed filter loads only the partitions it newly
needs (ensurePreloadedPartitions, :262-267). The gap is that the retention has no ceiling.

Reproduction

A JUnit/Scala probe on master @ a833f881ebca, against a partitioned COW table with 200
partitions registered in the catalog, issuing one query per distinct partition in a single
session, and reading the private map's size by reflection after each sample:

SOAK partitions_queried=200
SOAK index_identity_stable=true
SOAK after_query=1    slice_cache_entries=1
SOAK after_query=50   slice_cache_entries=50
SOAK after_query=100  slice_cache_entries=100
SOAK after_query=150  slice_cache_entries=150
SOAK after_query=200  slice_cache_entries=200
SOAK verdict_monotonic_growth=true
SOAK verdict_bounded=false

Exactly 1:1, with a single index instance serving all 200 queries (index_identity_stable=true
rules out the growth being an artifact of the index being rebuilt).

The probe queries one distinct partition per query deliberately: repeating a single
partition leaves the map flat whether or not eviction exists, so it could not distinguish the
two cases.

What I am not claiming

• No magnitude claim. The fixture was 200 partitions x 1 row, so its heap delta (71 -> 81 MB)
  reflects the fixture, not a real table, and should not be extrapolated. On a real table each
  entry is a List<FileSlice> per partition, with base and log files per file group.
• I have not measured GC or steady-state query latency over a long session.
• The DataFrame reader path (spark.read.format("hudi").load(path)) builds a fresh index per
  load() call and so does not accumulate this way — this is specific to the catalog/SQL path.
• Correctness is not affected in the case I tested: a commit landing between two identical
  queries is picked up (count moved 1 -> 2).

Possibly related but not a duplicate

#17292 ("Improve memory management in File Index with MDT") concerns closing cached MDT
readers, which is a different mechanism from the file-slice map's retention.

Possible directions

• A bounded eviction policy on the map, configurable with a stated default.
• hoodie.file.index.cache.use.spillable.map (default false) bounds heap when enabled, but
  its disk tier has no cap either, so it relocates the growth rather than bounding it.
• Any bound has to keep the existing correctness behaviour: eviction must not resurrect stale
  slices across a commit.

Happy to work up a PR if maintainers agree on the direction — in particular whether a size or
weight bound is preferred, and what the default should be.

Environment

• Hudi version: 1.3.0-SNAPSHOT (master @ a833f881ebca)
• Spark version: 3.5
• Scala: 2.12
• JDK: Temurin 11 (arm64)
• Table type: COPY_ON_WRITE, partitioned, registered in the Spark catalog
• Storage: local filesystem

Logs and Stack Trace

No exception; this is a retention issue rather than a failure. The relevant stock log line on
the refill path is BaseHoodieTableFileIndex:287:

[table=probe_tbl] HoodieFileIndex.listPartitionPathFiles took 24 ms (1 partitions, queryInstant=...)

In a three-query probe (repeat, then a changed filter) only two refills were logged — the exact
repeat issued none — which is the reuse behaviour described above.