Lazy segment loading for OFFLINE tables

# Proposal: Lazy Segment Loading for OFFLINE Tables

## Motivation

Today a Pinot server downloads every segment it is assigned as soon as the
Helix OFFLINE→ONLINE transition fires. Local disk (and, once queried, memory)
must therefore hold 100% of every hosted table at all times, even when large
parts of the data are cold — accessed rarely or never. For workloads with a
long cold tail (historical data kept for occasional audits, per-tenant tables
with sparse access patterns), this forces server fleets to be sized by total
data volume rather than by working set.

This proposal decouples segment *assignment* from segment *materialization*:
a server can hold a metadata-only stub for an assigned segment and fetch the
actual data from the deep store only when a query first touches it, evicting
it back to a stub after a configurable idle period. Disk and memory then scale
with the working set, while the deep store remains the authoritative copy.

We have been running this in production on a fork of Pinot 1.5.0
(`codeaeondev/pinot_lazy`) and would like to contribute it upstream.

## Design

Entirely **server-side**: brokers, controllers and clients are unchanged. A
stubbed segment is still reported ONLINE to the cluster and remains routable.

Segment lifecycle on a lazy-enabled table:

```
assignment (OFFLINE→ONLINE) ──► STUB (ZK metadata only; no download)
first query touches segment ──► MATERIALIZING (download tar from deep store, untar, mmap)
──► MATERIALIZED (serves at native speed)
idle > idleEvictionSeconds ──► STUB (offload, release, optionally delete local dir)
```

Key mechanics:

- `OfflineTableDataManager.addOnlineSegment` registers a stub (a
`ConcurrentHashMap` in `BaseTableDataManager`)
instead of calling `addNewOnlineSegment` when lazy loading is enabled — but
only when the deep store holds an authoritative copy (otherwise the
OFFLINE→ONLINE transition attempts the load so a permanent failure surfaces
as a Helix ERROR state, exactly as today) and no local index directory
exists (so a restarted server warms up from its valid local copy instead of
resetting hot segments to stubs).
- `acquireSegment` / `acquireSegments` transparently materialize stubs on the
query path (required and optional segments alike). Materialization runs
under the existing per-segment lock, so concurrent cold queries dedupe into
a single download. Multi-segment queries materialize stubs in parallel on a
bounded pool (`lazy.materialize.parallelism`, threads idle-timeout to zero)
and wait at most `lazy.materialize.timeout.seconds` (default 60): segments
still stubbed after the timeout are reported missing for that query while
their downloads continue in the background, so query workers are never
parked indefinitely behind a cold-cache burst. In-flight materializations
are dedup'd across queries, so retries share the same download instead of
stacking duplicate tasks on the pool. Optional segments never block the
query: their downloads are fired asynchronously (best-effort for the
current query, warm for the next).
- Acquire retries the rare race with the eviction sweeper (eviction registers
the stub before destroying the loaded segment), so a query never sees a
false missing-segment error for a segment that is stubbed and servable.
- ZK metadata is re-fetched at materialization time, so a segment refreshed
while stubbed loads its new version. Refresh messages received while stubbed
update the stub's metadata.
- A single scheduled daemon (`lazy-eviction-sweeper`) sweeps all tables every
`lazy.sweep.interval.seconds` and evicts idle materialized segments back to
stubs. Every acquire AND release stamps `lastAccessTimeMs` (so a query
running longer than the TTL does not leave its segment instantly
evictable); a segment with a query in flight (reference count > 1) is never
evicted. Segments whose ZK metadata has no authoritative deep-store
download URL (e.g. peer-download segments) are never evicted.
- Failed materialization keeps the stub registered and reports the segment
missing for that query (existing upstream semantics), so the next query
retries — self-healing.
- Helix OFFLINE/DROP transitions clear stubs like loaded segments.

### Scope / safety

Lazy loading is only allowed on plain OFFLINE tables. Table-config validation
rejects it for REALTIME, upsert, dedup and dimension tables, which rely on
segments being local at assignment time.

Everything is opt-in twice: an instance-level kill switch **(default off)**
and a per-table config. Tables without `lazyLoadConfig` behave exactly as
today (covered by a regression test).

## Configuration

Table level (`tableIndexConfig` sibling, OFFLINE tables only):

```json
"lazyLoadConfig": {
"enabled": true,
"idleEvictionSeconds": 3600,
"deleteLocalOnEvict": true
}
```

`deleteLocalOnEvict: false` keeps the local index directory on eviction for a
fast re-warm with zero deep-store traffic (two-tier mode: memory freed, disk
kept).

Instance level (`server.conf`):

```
pinot.server.instance.lazy.load.enabled=true # default false
pinot.server.instance.lazy.sweep.interval.seconds=60
pinot.server.instance.lazy.materialize.parallelism=4
pinot.server.instance.lazy.materialize.timeout.seconds=60
```

## Observability

New server metrics: `LAZY_SEGMENT_COLD_LOADS` (meter),
`LAZY_SEGMENT_EVICTIONS` (meter), `LAZY_SEGMENT_LOAD_TIME_MS` (timer),
`LAZY_STUBBED_SEGMENT_COUNT` (per-table gauge).

## Trade-offs

- First query against a cold segment pays the download+load latency
(observable via `LAZY_SEGMENT_LOAD_TIME_MS`). This is the fundamental trade
and the reason the feature is strictly opt-in per table.
- Deep-store egress increases for segments that oscillate between idle and
active; `idleEvictionSeconds` and `deleteLocalOnEvict=false` are the tuning
knobs.
- The eviction sweeper is a single daemon thread per server, no-op unless
the instance switch is on.

## Known limitations / discussion points

Raised here openly so the design discussion can weigh them; we would address
them in this PR or fast-follows per reviewer preference:

1. **Table-size and quota reporting.** `acquireAllSegments` /
`getAllSegmentsMetadata` iterate only materialized segments, so evicted
segments are missing from the server's `/table/{t}/size` responses and
storage-quota checks under-count lazy tables. Likely fix: include stub
sizes from their ZK metadata in size/metadata endpoints.
2. **Materialization holds the per-segment lock.** A slow cold download
blocks Helix transitions (refresh/offload/drop) on that same segment until
it completes; those transitions can time out on multi-GB segments. Likely
fix: a per-segment materialize-in-progress future so waiters block on the
future rather than the lock, or a dedicated lock for materialization.
3. **Non-query callers of `acquireSegment` can trigger downloads.** Server
metadata/tier/debug REST endpoints materialize stubs as a side effect,
which can re-download evicted segments on monitoring polls. Likely fix: a
non-materializing accessor for metadata-only callers.
4. **Reload of stubbed segments.** Reload paths skip stubs today; with
`deleteLocalOnEvict=false` a `forceDownload` reload does not invalidate an
evicted segment's local copy. Likely fix: drop the local directory (and
refresh stub metadata) for stubbed segments during reload.
5. **The materialize wait is not query-deadline aware.** The multi-segment
wait uses the fixed instance-level timeout rather than the query's
remaining deadline (plumbing the deadline into `acquireSegments` is a
`TableDataManager` interface change). Single-segment materialization
(`acquireSegment`, the lone-stub fast path, and the rare eviction-race
re-materialization) is deliberately synchronous first-query-pays with no
bound.

## Implementation

The change is contained: 14 files, ~870 lines added (of which 404 are tests),
3 lines removed, rebased onto current master.

- `pinot-spi`: new `LazyLoadConfig` (`BaseJsonConfig`), `TableConfig` plumbing,
`InstanceDataManagerConfig` default methods (other implementors keep
compiling).
- `pinot-common`: `TableConfigSerDeUtils` ZK round-trip, three new metrics.
- `pinot-segment-local`: `SegmentDataManager.lastAccessTimeMs`,
`TableConfigUtils.validateLazyLoadConfig`.
- `pinot-core`: stub registry, materialize-on-acquire and eviction in
`BaseTableDataManager`; stub registration in `OfflineTableDataManager`.
- `pinot-server`: instance config keys, eviction sweeper in
`HelixInstanceDataManager`.

Tests (`OfflineTableDataManagerLazyLoadTest`, TestNG, follows the
`BaseTableDataManagerTest` pattern with a `file://` deep store): stub on
assignment + materialize on both acquire paths, concurrent-acquire download
dedupe (8 threads → 1 download), TTL eviction + reload loop, no eviction while
a query is in flight, instance kill switch, and non-lazy regression.

## Production experience

Running in production since June 2026 on a ReelMint k3s cluster with a MinIO
deep store, including a 10 GB / 135M-row scale test. Cold-load and eviction
behavior matched design; brokers required no changes.

Start with BaseTableDataManager and OfflineTableDataManager, then run OfflineTableDataManagerLazyLoadTest following the BaseTableDataManagerTest pattern. Review the pinot-spi, pinot-common, pinot-segment-local, pinot-core, and pinot-server changes named in the proposal; done requires the listed lazy-load, eviction, concurrency, kill-switch, and non-lazy regression behaviors to be covered.

Written by the indexing model from the issue text.