[RFC] Tiered remote-store recovery: serve reads through block fetches while hydrating

Summary

Today a hot remote-store shard cannot open its engine until every segment file has been fully downloaded from the remote store. This RFC proposes opening the engine as soon as the commit metadata is known, serving reads through range-level block fetches (the machinery warm indices already use), hydrating full files in the background, and per file atomically switching live readers to the local copy and deleting its cached blocks. When hydration completes the shard has zero cache footprint and is indistinguishable from today's hot shard.

Scope covers both Lucene segment files and parquet (pluggable-dataformat) files, delivered as two tracks: Lucene first, parquet second, NRT replicas as a later phase.

Implementation: #23065 (draft; Track A milestones land incrementally on that branch). Builds on: writable warm Lucene (#21015), writable warm parquet (#22999), multipart parallel download (#23016).

Feedback especially welcome on §10 (risks/decisions), §11 (open questions) and the settings surface in §5.1.

1. Problem

A hot remote-store shard cannot open its engine until every segment file has been fully downloaded ("hydrated") from the remote store.

IndexShard.innerOpenEngineAndTranslog -> syncSegmentsFromRemoteSegmentStore(false) -> copySegmentFiles -> RemoteStoreFileDownloader.download(...), which parks on a PlainActionFuture until every file is on disk (IndexShard.java ~L6272, RemoteStoreFileDownloader.java L90). Only then does indexerFactory.createIndexer(config) run and the shard become searchable / writable.

Every remote-store recovery source on hot nodes pays this:

For a multi-TB shard this is minutes to hours of unavailability, and the cost is paid whether or not the query workload will ever touch most of those bytes.

Warm already solves the read side for both file families:

• Lucene files: TieredDirectory serves reads via SwitchableIndexInput in remote mode, fetching block ranges through TransferManager into FileCache.
• Parquet files: the Rust TieredObjectStore resolves each get_range through the FileRegistry (LOCAL/REMOTE) and serves REMOTE ranges through the foyer block cache.

Neither is wired for hot: IndexService L829 gates TieredDirectory on isWarmIndex() and carries the TODO "remove this check after support for hot indices is added in Composite Directory"; ParquetDataFormatStoreHandler(isWarm=false) returns an EMPTY handle so DataFusion reads plain file:// paths.

2. Goals

1. On a hot remote-store shard, open the engine as soon as the commit metadata is known; serve reads immediately through range-level fetches from the remote store.
2. Hydrate full files in the background, prioritised, throttled, cancellable.
3. Per file, once the full copy is on local disk: atomically switch every live reader to the local file and delete that file's cached blocks (FileCache block entries for Lucene, foyer ranges for parquet).
4. When every file is local, the shard has zero cache footprint and behaves exactly like today's hot shard (fsync on commit, local reads, local uploads).
5. Same guarantees for Lucene and parquet files; no format-specific engine changes.

Non-goals (v1): changing warm residency semantics; changing translog recovery; making hot NRT replicas open readers over blocks at every checkpoint (Phase 2, §9); cross-node block sharing.

3. Current state (code-verified on this branch)

3.1 Lucene files

Warm-only behaviours in TieredDirectory that are wrong for hot: afterSyncToRemote switches the just-uploaded file to blocks and evicts local; sync() is a no-op; createOutput registers a pinned full-file FileCache entry.

3.2 Parquet files

4. Design overview

One shared framework, two format-specific "residency backends":

                     +---------------------------+
   IndexShard -----> |  RemoteStoreHydrator      |  per-shard, background, cancellable
   (after engine     |  work set: files REMOTE   |
    open)            |  order: priority + size   |
                     +------+-------------+------+
                            |             |
             Lucene file    |             |   parquet file
                            v             v
        +-----------------------+   +--------------------------+
        | TieredDirectory       |   | StoreStrategyRegistry    |
        | policy = HOT_LOCAL    |   | (+ Rust FileRegistry)    |
        | per-file state map    |   | seed REMOTE ->           |
        | REMOTE->HYDRATING->   |   |   onWritten LOCAL upsert |
        |   LOCAL               |   | foyer.evictPrefix(file)  |
        | SwitchableIndexInput  |   | FormatSwitchableIndex-   |
        |   .switchToLocal()    |   |   Input.switchToLocal()  |
        | fileCache.remove(     |   +--------------------------+
        |   <file>_block_N)     |
        +-----------------------+

Common per-file lifecycle (both families):

 shard open          background                  promote (atomic per file)
 ----------          ----------                  -------------------------
 REMOTE  ------->    HYDRATING  ------------->   LOCAL
 (blocks/ranges      (blocks still serve         1. temp file -> checksum verify
  served on demand)   reads; full file            2. localDirectory.rename(tmp, file)
                      streams to <file>.          3. registry/state flip to LOCAL
                      hydrating.tmp)              4. switch live readers to local
                                                  5. evict cached blocks/ranges

Invariant (mirror of the writable-warm invariant "registry flips before the local file dies"): the local full file exists and is verified before any tier metadata says LOCAL, and blocks are evicted only after the flip. A reader can therefore never observe LOCAL without a readable local file, and a block that is evicted early is only a cache miss (re-fetched from remote), never a correctness issue.

5. Shared framework

5.1 Settings and gating

Feature flag: reuse WRITABLE_WARM_INDEX_SETTING (the building blocks live behind it already) or introduce TIERED_REMOTE_RECOVERY; decision in §11.

Shard creation rejects index.remote_store.tiered_recovery.enabled=true on a node whose hydration cache size is 0 with a clear error, rather than silently falling back to full download.

5.2 Directory selection (IndexService.createShard)

• Lucene-only index: isWarmIndex() || isTieredRecoveryEnabled() -> TieredDirectoryFactory, passing ResidencyPolicy.HOT_LOCAL for the non-warm case.
• Pluggable-dataformat index: the L795 branch becomes (isWarmIndex() || isTieredRecoveryEnabled()) && isPluggableDataFormatEnabled() -> StoreStrategyRegistry.open(path, tiered=true, ...) + TieredDataFormatAwareStoreDirectoryFactory (its inner Lucene directory is a TieredDirectory with the same HOT_LOCAL policy). IndexModule L1196 gets the same predicate.
• Delivery order (§13): Track A ships Lucene-only. Until Track B lands, an index with index.remote_store.tiered_recovery.enabled=true AND isPluggableDataFormatEnabled() keeps today's full-download path (logged at INFO once per shard), and the L795 / L1196 predicates stay isWarmIndex(). Nothing in Track A changes parquet behaviour.
• The DataFormatStoreHandlerFactory.create(shardId, isWarm, ...) boolean is renamed to tiered (semantics: "build the native tiered store"), so hot-tiered shards get a live TieredObjectStore wired to the hot node's foyer cache.

5.3 Hot-node caches

Node.java L872: create NodeCacheService when isWarmNode OR node.remote_store.hydration_cache.size > 0. On hot:

• FileCache holds ONLY Lucene block files (LOCAL Lucene files bypass it; see §6.1). Usage == transient hydration footprint. Entries unpinned/evictable; eviction of a block is a cache miss, not data loss.
• BlockCacheProviders (foyer) are instantiated so parquet REMOTE ranges have a cache. Foyer capacity is the plugin's own setting; on hot it holds only not-yet-hydrated ranges.
• Install NodeCacheServiceCleaner on hot as well (Node L618 guard becomes isWarmNode OR hydration cache enabled) with a tiered-recovery branch that only evicts cache entries and never rm's the shard path. The warm rm is redundant on hot rather than destructive (NodeEnvironment deletes the same path immediately after). The genuinely destructive warm behaviour is the startup indicesPath rescan; it stays warm-only. Full lifecycle in §5.3.2.

5.3.1 Cache sizing on hot

Warm sizes its cache as the data footprint: node.search.cache.size defaults to 80% of the fileCacheNodePath disk on a dedicated warm node (0 elsewhere), NodeCacheService subtracts the block-cache providers' requested bytes (foyer) and hands the remainder to FileCache. Pinned full files account at file size, blocks at their real size (DEFAULT_BLOCK_SIZE_SHIFT = 23, i.e. 8 MB), and cluster.filecache.remote_data_ratio multiplies capacity into addressable remote data. None of that model applies on hot, because in HOT_LOCAL the data never goes through the cache.

What the hot FileCache actually holds:

Usage is therefore the working set of blocks that queries and merges touched on not-yet-hydrated files, unpinned and LRU-evictable, trending to zero as hydration completes. An evicted block is a re-fetch, never data loss. Foyer on hot holds only REMOTE parquet ranges with the same shape.

Sizing rule for node.remote_store.hydration_cache.size:

• It is a budget for transient blocks, so it must not be expressed as a fraction of the data set. Working-set floor: the metadata-heavy Lucene files (.si, .fnm, .tip, .doc, .nvd, .kdi) of every shard recovering concurrently plus what live queries touch -- typically a few hundred MB to a few GB per recovering shard.
• Proposed default once enabled: min(10% of the data path, 50 GB), overridable as bytes or percent (same RatioValue/ByteSizeValue parser as node.search.cache.size). Leaves ~90% of disk for real data and the transient .hydrating.tmp files, which are the actual disk pressure (up to 2x one file while blocks, temp and final coexist).
• Foyer share on hot: today providers take requestedCapacityBytes(settings, totalBudget) out of the same budget. Keep that partitioning on hot (foyer gets its usual fraction of the hydration budget) unless parquet-heavy fleets need a dedicated node.remote_store.hydration_cache.block_cache.size; decision in §11.
• 0 (default) disables the feature: shard creation with index.remote_store.tiered_recovery.enabled=true fails fast on such a node.

Accounting rules that must hold on hot:

1. Do not publish the cache into shard allocation. fileCacheNodePath.fileCacheReservedSize, remote_data_ratio and NodeDiskEvaluator are warm concepts; setting reservedSize on hot would make the balancer believe 10% of the disk is permanently gone. Leave it unset; the hydration cache competes for free space like any transient usage.
2. FileCache.capacity() is a soft cap on cached bytes, not a disk reservation. Block files live in the shard directory on the data disk, so the disk-threshold decider sees true free space and flood-stage protection still fires normally.
3. Never insert LOCAL files. HOT_LOCAL openInput/createOutput/promotion bypass FileCache entirely; only the block and _switchable entries exist. A unit test asserts fileCache.usage() == 0 after full hydration of a shard.
4. Surface hydration_cache_usage and block_bytes_fetched per node (§5.6) so operators can see the budget is adequate: sustained usage at capacity with high block churn means hydration is starved and the budget or hydration.max_concurrent_files should be raised.

5.3.2 Cleanup on hot

Warm has three cleanup mechanisms. Hot keeps one unchanged, must keep one disabled, and needs a small replacement for the third plus two lifecycle hooks warm never needed.

What warm has today (code-verified):

• Eviction deletes the disk file. FileCacheFactory.createDefaultBuilder (L49-62) installs a removal listener that closes the CachedIndexInput and Files.deleteIfExists(key) on every EVICTED/EXPLICIT removal. Kept as-is on hot; it is what makes fileCache.remove in switchToLocal() actually reclaim disk.
• Shard/index deletion. NodeCacheServiceCleaner.beforeShardPathDeleted (L61-78) walks the shard store path, fileCache.remove()s each file, foyer.evictPrefix(shardDataPath), then IOUtils.rm(shardDataPath). Gated on isRemoteSnapshot() || isWarmIndex(), so it is a no-op for hot indices today.
• Startup restore. NodeCacheService.restoreFileCacheFromDisk (L457-481) rescans fileCachePath always and indicesPath only when isDedicatedWarmNode; FileCache.restoreFromDirectory registers every regular file it finds as a RestoredCachedIndexInput.

The four cleanup moments on hot: