Package org.elasticsearch.xpack.esql.datasources.cache

Class ExternalSourceCacheService

java.lang.Object
org.elasticsearch.xpack.esql.datasources.cache.ExternalSourceCacheService
All Implemented Interfaces:
Closeable, AutoCloseable
public class ExternalSourceCacheService extends Object implements Closeable
Coordinator-only, in-memory cache service for external source metadata. Maintains two independent caches:
  • Schema cache (20% of budget, 5m TTL) — shared across users
  • Listing cache (80% of budget, 30s TTL) — isolated by credential hash
Uses hard TTL via setExpireAfterWrite for the initial implementation. Lazy TTL with ETag revalidation is deferred to a follow-up PR.

  • Constructor Details

    • ExternalSourceCacheService

      public ExternalSourceCacheService(Settings settings)

  • Method Details

    • getOrComputeSchema

      public SchemaCacheEntry getOrComputeSchema(SchemaCacheKey key, CacheLoader<SchemaCacheKey,SchemaCacheEntry> loader) throws Exception
      Returns a cached schema entry or computes it via the loader. The loader is only invoked on a cache miss. When the cache is disabled, the loader is called directly (bypassing the cache).
      Throws:
      Exception

    • getSchemaIfPresent

      public SchemaCacheEntry getSchemaIfPresent(SchemaCacheKey key)
      Returns a cached schema entry, or null on a miss (or when the cache is disabled). Unlike getOrComputeSchema(org.elasticsearch.xpack.esql.datasources.cache.SchemaCacheKey, org.elasticsearch.common.cache.CacheLoader<org.elasticsearch.xpack.esql.datasources.cache.SchemaCacheKey, org.elasticsearch.xpack.esql.datasources.cache.SchemaCacheEntry>), this never invokes a loader — it is the peek half of the async resolve path, which fetches on a miss without holding an executor thread and then stores the result via putSchema(org.elasticsearch.xpack.esql.datasources.cache.SchemaCacheKey, org.elasticsearch.xpack.esql.datasources.cache.SchemaCacheEntry). This trades strict thundering-herd coalescing (two concurrent misses for the same key may both fetch) for the ability to resolve asynchronously.

    • putSchema

      public void putSchema(SchemaCacheKey key, SchemaCacheEntry entry)
      Stores a schema entry. No-op when the cache is disabled. Pairs with getSchemaIfPresent(org.elasticsearch.xpack.esql.datasources.cache.SchemaCacheKey).

    • getDatasetAggregate

      @Nullable public Map<String,Object> getDatasetAggregate(SchemaCacheKey key)
      Returns the dataset-level aggregate stats stored under key (see SchemaCacheKey.forDatasetAggregate(java.lang.String, org.elasticsearch.xpack.esql.datasources.FileSetFingerprint, java.lang.String, java.util.Map<java.lang.String, java.lang.Object>)), or null on a miss. The map carries only dataset-INDEPENDENT-of-declaration keys — today just _stats.row_count — never per-column stats, so serving it can never leak a wrongly-normalized MIN/MAX (those keep re-scanning until the per-file rail serves them). A found entry is re-put to refresh the expireAfterWrite clock and LRU position: the entry is the warm path's single survival dependency, so a hot dataset must not decay mid-use while its per-file siblings churn (same revive the per-file reconcile applies to recovered entries). Does NOT touch the hit/miss counters — those are resolver-driven at the serve decision (see recordDatasetAggregateHit() / recordDatasetAggregateMiss()).

    • putDatasetAggregate

      public void putDatasetAggregate(SchemaCacheKey key, long rowCount, String sourceType, String location)
      Stores the dataset-level row-count aggregate for one resolved file set. The entry is a synthetic SchemaCacheEntry (no columns; safeMetadata = the row count) so all existing schema cache plumbing — weigher, TTL, enable/disable, clearAll, usage stats — applies unchanged.

    • registerPendingDatasetAggregate

      public void registerPendingDatasetAggregate(SchemaCacheKey datasetKey, Map<String,Long> pathToMtimeMillis, int expectedFileCount, String configFingerprint, String sourceType, String location)
      Registers a ExternalSourceCacheService.PendingDatasetAggregate promise for the reconcile of the in-flight cold scan to fulfill. Doubly bounded (64 descriptors / 65536 total stored paths, oldest dropped) and expired on PENDING_DATASET_AGGREGATE_TTL_NANOS, so an abandoned promise (query failed, scan partial) costs nothing. Re-registering the same dataset key replaces the previous promise (same content, refreshed clock).

      expectedFileCount is the resolved listing's file COUNT (a multiset count: a comma-separated source list can name the same file twice, and the scan then counts its rows twice). pathToMtimeMillis is a map and therefore deduplicates; when the two disagree the promise-side sum over unique paths would publish an undercount relative to the scan, so registration is refused — the dataset stays on the re-scan path (correct-or-miss).

    • recordStatsAggregateIncomplete

      public void recordStatsAggregateIncomplete()
      Counts a resolve whose per-file stats aggregate came back incomplete (observability). Only aggregate-QUALIFYING resolves are counted (multi-file, text-format, file-set-fingerprint-bearing, cacheable provider): the caller sits after the datasetKey == null early-return in ExternalSourceResolver#applyDatasetAggregate, so non-qualifying incompletes never reach it.

      Today stats_aggregate.incomplete == dataset_aggregate.hits + dataset_aggregate.misses exactly (every incomplete then resolves to precisely one of hit/miss on the same needed path). Kept as its own counter so the "the per-file merge was incomplete" signal survives a future serve path that grows a third outcome and breaks that identity.

    • recordDatasetAggregateHit

      public void recordDatasetAggregateHit()
      Counts a dataset-aggregate fallback that was NEEDED (the per-file merge came back incomplete) AND PRESENT — i.e. the memoized aggregate was actually served. Resolver-driven and symmetric with recordDatasetAggregateMiss() so that hits / (hits + misses) is a true fallback hit rate over the resolves that actually needed the aggregate.

    • recordDatasetAggregateMiss

      public void recordDatasetAggregateMiss()
      Counts a dataset-aggregate fallback that was NEEDED (the per-file merge came back incomplete) but ABSENT. Deliberately resolver-driven rather than incremented inside getDatasetAggregate(org.elasticsearch.xpack.esql.datasources.cache.SchemaCacheKey): every multi-file resolve prefetches the aggregate — including healthy warm resolves whose per-file merge succeeds and never consumes it — so a get-side counter would count non-events. Its hit twin (recordDatasetAggregateHit()) is counted at the same serve decision, so the two share a denominator (resolves that needed the fallback) and their ratio is meaningful.

    • getOrComputeListing

      public FileList getOrComputeListing(ListingCacheKey key, CacheLoader<ListingCacheKey,FileList> loader) throws Exception
      Returns a cached file listing or stores the provided one. The loader is only invoked on a cache miss. When the cache is disabled, the loader is called directly (bypassing the cache).
      Throws:
      Exception

    • reconcileSourceStatsFromContributions

      public void reconcileSourceStatsFromContributions(Map<String,List<Map<String,Object>>> contributionsPerFile)
      Coordinator-side entry point. Takes the DriverCompletionInfo.capturedSourceMetadata payload — raw per-file contribution lists shipped back from every data node — merges each list via SourceStatisticsSerializer.mergeStatistics (Parquet's existing multi-row- group merge algorithm), then enriches the matching SchemaCacheEntry so the next query's planning-time lookup short-circuits on the merged stats.

    • reconcileSourceStats

      public void reconcileSourceStats(Map<String,Map<String,Object>> mergedStatsPerFile)
      Reconciles already-merged data-node-captured source stats into the schema cache. For each (path, mergedStats) entry, finds the cached SchemaCacheEntry whose location and mtime match and replaces it with a new entry whose safeMetadata folds in the merged _stats.* keys. Entries with no cache match are ignored (the warm path will just trigger a fresh metadata() call on the next query).

    • setEnabled

      public void setEnabled(boolean enabled)

    • isEnabled

      public boolean isEnabled()

    • clearAll

      public void clearAll()

    • usageStats

      public Map<String,Object> usageStats()

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable