Add lazy segment loading for OFFLINE tables

pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerGauge.java

@@ -73,6 +73,8 @@ public enum ServerGauge implements AbstractMetrics.Gauge {
   // Segment operation metrics - count is the current number of segments undergoing the given operation.
   // Incremented when the semaphore is acquired and decremented when the semaphore is released
   SEGMENT_TABLE_DOWNLOAD_COUNT("segmentTableDownloadCount", false),
+  // Number of lazily stubbed (assigned but not materialized) segments per table
+  LAZY_STUBBED_SEGMENT_COUNT("segments", false),
   SEGMENT_DOWNLOAD_COUNT("segmentDownloadCount", true),
   SEGMENT_ALL_PREPROCESS_COUNT("segmentAllPreprocessCount", true),
   SEGMENT_STARTREE_PREPROCESS_COUNT("segmentStartreePreprocessCount", true),

pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java

@@ -102,6 +102,9 @@ public enum ServerMeter implements AbstractMetrics.Meter {
   SEGMENT_DOWNLOAD_FAILURES("segments", false),
   SEGMENT_DOWNLOAD_FROM_REMOTE_FAILURES("segments", false),
   SEGMENT_DOWNLOAD_FROM_PEERS_FAILURES("segments", false),
+  // Lazy segment loading
+  LAZY_SEGMENT_COLD_LOADS("segments", false),
+  LAZY_SEGMENT_EVICTIONS("segments", false),
   SEGMENT_BUILD_FAILURE("segments", false),
   SEGMENT_UPLOAD_FAILURE("segments", false),
   SEGMENT_UPLOAD_SUCCESS("segments", false),

pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerTimer.java

@@ -107,6 +107,8 @@ public enum ServerTimer implements AbstractMetrics.Timer {
       "Time taken to download a segment from deep store (including untar and move operations)"),
   SEGMENT_DOWNLOAD_FROM_PEERS_TIME_MS("millis", false,
       "Time taken to download a segment from peers (including untar and move operations)"),
+  LAZY_SEGMENT_LOAD_TIME_MS("millis", false,
+      "Time taken to materialize (download, untar and load) a lazily stubbed segment on first query access"),
 
   // Ingestion metrics
   INGESTION_DELAY_TRACKING_MS("milliseconds", false, "Time taken to run a trackIngestionDelay cycle"),

pinot-common/src/main/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtils.java

@@ -31,6 +31,7 @@
 import org.apache.pinot.spi.config.table.DimensionTableConfig;
 import org.apache.pinot.spi.config.table.FieldConfig;
 import org.apache.pinot.spi.config.table.IndexingConfig;
+import org.apache.pinot.spi.config.table.LazyLoadConfig;
 import org.apache.pinot.spi.config.table.QueryConfig;
 import org.apache.pinot.spi.config.table.QuotaConfig;
 import org.apache.pinot.spi.config.table.RoutingConfig;
@@ -204,6 +205,13 @@ public static TableConfig fromZNRecord(ZNRecord znRecord)
             tableSamplerConfigs, isMaterializedView);
     tableConfig.setDescription(description);
     tableConfig.setTags(tags);
+
+    // NOTE: lazyLoadConfig is not part of the @JsonCreator constructor, set it explicitly. Any TableConfig
+    // sub-config missing from this class is silently dropped when the controller persists the config to ZK.
+    String lazyLoadConfigString = simpleFields.get(TableConfig.LAZY_LOAD_CONFIG_KEY);
+    if (lazyLoadConfigString != null) {
+      tableConfig.setLazyLoadConfig(JsonUtils.stringToObject(lazyLoadConfigString, LazyLoadConfig.class));
+    }
     return tableConfig;
   }
 
@@ -292,6 +300,10 @@ public static ZNRecord toZNRecord(TableConfig tableConfig)
     if (tags != null && !tags.isEmpty()) {
       simpleFields.put(TableConfig.TAGS_KEY, JsonUtils.objectToString(tags));
     }
+    LazyLoadConfig lazyLoadConfig = tableConfig.getLazyLoadConfig();
+    if (lazyLoadConfig != null) {
+      simpleFields.put(TableConfig.LAZY_LOAD_CONFIG_KEY, JsonUtils.objectToString(lazyLoadConfig));
+    }
 
     ZNRecord znRecord = new ZNRecord(tableConfig.getTableName());
     znRecord.setSimpleFields(simpleFields);

pinot-core/src/main/java/org/apache/pinot/core/data/manager/offline/OfflineTableDataManager.java

@@ -77,7 +77,18 @@ protected void doAddOnlineSegment(String segmentName)
     indexLoadingConfig.setSegmentTier(zkMetadata.getTier());
     SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
     if (segmentDataManager == null) {
+      if (isLazyLoadEnabled() && canRegisterStub(segmentName, zkMetadata)) {
+        // Lazy loading: register a metadata-only stub instead of downloading. The first query that touches the
+        // segment materializes it from the deep store. Segments without an authoritative deep-store copy or with an
+        // existing local copy load eagerly (see canRegisterStub).
+        registerStubbedSegment(segmentName, zkMetadata);
+        return;
+      }
       addNewOnlineSegment(zkMetadata, indexLoadingConfig);
+      // Clear any stale stub (e.g. re-sent ONLINE after an eviction that kept the local copy) only AFTER the eager
+      // load succeeds: while loading, concurrent queries can still self-heal through the stub, and if the load
+      // throws, the stub keeps the segment servable from the deep store.
+      unregisterStubbedSegment(segmentName);
     } else {
       replaceSegmentIfCrcMismatch(segmentDataManager, zkMetadata, indexLoadingConfig);
     }