feat: on-demand partial segment loading in SegmentLocalCacheManager (#19535)

changes:
* adds `acquirePartialSegment` / `acquireCachedPartialSegment` to `SegmentCacheManager` and `SegmentManager` to allow callers to opt-in to async partial segments; MSQ `RegularLoadableSegment` uses the new partial path
* `PartialSegmentMetadataCacheEntry`, `PartialSegmentBundleCacheEntry` are now wired into `SegmentLocalCacheManager`, along with added `PartialBundleAcquirer` helper to pass things like download thread pool and the ability to acquire reference holds on cache entrie
* adds `PartialQueryableIndexSegment`, `PartialQueryableIndexCursorFactory`, `V10TimeBoundaryInspector` for references acquired from the now wired up metadata cache entry, implementing the async cursor holder contract
* moved `DirectoryBackedRangeReader` out of tests to use as the implementation for `LocalLoadSpec` range reader.
* add `FilteredCursorFactory`/`RestrictedCursorFactory` async cursor implementations
* adds config `druid.segmentCache.virtualStoragePartialDownloadsEnabled to enable feature, of by default

Author: clintropolis

embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryVirtualStorageTest.java

@@ -73,6 +73,7 @@ class QueryVirtualStorageTest extends EmbeddedClusterTestBase
   private static final long SIZE_BYTES = 3778338L;
   private static final long CACHE_SIZE = HumanReadableBytes.parse("1MiB");
   private static final long MAX_SIZE = HumanReadableBytes.parse("100MiB");
+  private static final long ESTIMATE_SIZE = HumanReadableBytes.parse("2KiB");
 
   private final EmbeddedBroker broker = new EmbeddedBroker();
   private final EmbeddedIndexer indexer = new EmbeddedIndexer();
@@ -89,6 +90,11 @@ public EmbeddedDruidCluster createCluster()
   {
     historical.setServerMemory(500_000_000)
               .addProperty("druid.segmentCache.virtualStorage", "true")
+              .addProperty("druid.segmentCache.virtualStoragePartialDownloadsEnabled", "true")
+              .addProperty(
+                  "druid.segmentCache.virtualStorageMetadataReservationEstimate",
+                  String.valueOf(ESTIMATE_SIZE)
+              )
               .addProperty("druid.segmentCache.virtualStorageLoadThreads", String.valueOf(Runtime.getRuntime().availableProcessors()))
               .addBeforeStartHook(
                   (cluster, self) -> self.addProperty(
@@ -155,8 +161,10 @@ void testQueryTooMuchData()
         RuntimeException.class,
         () -> cluster.runSql("select count(*) from \"%s\"", dataSource)
     );
-    Assertions.assertTrue(t.getMessage().contains("Unable to load segment"));
-    Assertions.assertTrue(t.getMessage().contains("] on demand, ensure enough disk space has been allocated to load all segments involved in the query"));
+    Assertions.assertTrue(t.getMessage().contains("Unable to reserve bundle"));
+    Assertions.assertTrue(t.getMessage()
+                           .contains(
+                               "ensure enough disk space has been allocated to load all segments involved in the query"));
   }
 
   @Test
@@ -296,11 +304,13 @@ void testQueryTooMuchDataButWithDart()
     MatcherAssert.assertThat(inputChannelSums.bytes(), Matchers.lessThanOrEqualTo(SIZE_BYTES));
 
     // Verify stage 0 (segment read) VSF load counters
-    MatcherAssert.assertThat(inputChannelSums.loadFiles(), Matchers.greaterThan(0L));
+    // partial loading is only partially metered at the moment, so depending on how stuff landed in and was evicted
+    // from the cache,there can be 0 loads (because loads are currently only counted when the metadata entry is mounted)
+    MatcherAssert.assertThat(inputChannelSums.loadFiles(), Matchers.greaterThanOrEqualTo(0L));
     MatcherAssert.assertThat(inputChannelSums.loadFiles(), Matchers.lessThanOrEqualTo(24L));
-    MatcherAssert.assertThat(inputChannelSums.loadTime(), Matchers.greaterThan(0L));
-    MatcherAssert.assertThat(inputChannelSums.loadWait(), Matchers.greaterThan(0L));
-    MatcherAssert.assertThat(inputChannelSums.loadBytes(), Matchers.greaterThan(0L));
+    MatcherAssert.assertThat(inputChannelSums.loadTime(), Matchers.greaterThanOrEqualTo(0L));
+    MatcherAssert.assertThat(inputChannelSums.loadWait(), Matchers.greaterThanOrEqualTo(0L));
+    MatcherAssert.assertThat(inputChannelSums.loadBytes(), Matchers.greaterThanOrEqualTo(0L));
     MatcherAssert.assertThat(inputChannelSums.loadBytes(), Matchers.lessThanOrEqualTo(SIZE_BYTES));
   }
 
@@ -322,7 +332,7 @@ private void assertQueryMetrics(int expectedEventCount, @Nullable Long expectedL
 
     long loadCount = getMetricLatestValue(emitter, DefaultQueryMetrics.QUERY_ON_DEMAND_LOAD_COUNT, expectedEventCount);
     if (expectedLoadCount != null) {
-      Assertions.assertEquals(expectedLoadCount, loadCount);
+      MatcherAssert.assertThat(loadCount, Matchers.lessThanOrEqualTo(expectedLoadCount));
     }
     boolean hasLoads = loadCount > 0;
 

embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/ServerManagerForQueryErrorTest.java

@@ -55,6 +55,7 @@
 import org.apache.druid.segment.Segment;
 import org.apache.druid.segment.SegmentMapFunction;
 import org.apache.druid.segment.SegmentReference;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.server.SegmentManager;
 import org.apache.druid.server.ServerManager;
 import org.apache.druid.server.initialization.ServerConfig;
@@ -225,7 +226,7 @@ protected LeafSegmentsBundle getLeafSegmentsBundle(Query<T> query, SegmentMapFun
             missingSegments.add(segment.getDescriptor());
             continue;
           }
-          Optional<Segment> ref = segmentManager.acquireCachedSegment(dataSegment);
+          Optional<Segment> ref = segmentManager.acquireCachedSegment(dataSegment, AcquireMode.FULL);
           if (ref.isPresent()) {
             segmentReferences.add(
                 new SegmentReference(

indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java

@@ -95,6 +95,7 @@
 import org.apache.druid.segment.indexing.CombinedDataSchema;
 import org.apache.druid.segment.indexing.DataSchema;
 import org.apache.druid.segment.indexing.TuningConfig;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.segment.loading.AcquireSegmentAction;
 import org.apache.druid.segment.loading.SegmentCacheManager;
 import org.apache.druid.segment.projections.AggregateProjectionSchema;
@@ -890,7 +891,8 @@ private static ResourceHolder<QueryableIndex> fetchSegmentInternal(
   {
     final Closer closer = Closer.create();
     try {
-      final AcquireSegmentAction acquireAction = closer.register(segmentCacheManager.acquireSegment(dataSegment));
+      final AcquireSegmentAction acquireAction =
+          closer.register(segmentCacheManager.acquireSegment(dataSegment, AcquireMode.FULL));
       final ReferenceCountedObjectProvider<Segment> segmentProvider =
           FutureUtils.getUnchecked(acquireAction.getSegmentFuture(), true).getReferenceProvider();
       final Segment segment = segmentProvider.acquireReference().map(closer::register).get();

indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java

@@ -132,6 +132,7 @@
 import org.apache.druid.segment.indexing.DataSchema;
 import org.apache.druid.segment.indexing.TuningConfig;
 import org.apache.druid.segment.join.NoopJoinableFactory;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.segment.loading.AcquireSegmentAction;
 import org.apache.druid.segment.loading.AcquireSegmentResult;
 import org.apache.druid.segment.loading.NoopSegmentCacheManager;
@@ -1978,7 +1979,7 @@ public void load(DataSegment segment)
       }
 
       @Override
-      public Optional<Segment> acquireCachedSegment(SegmentId segmentId)
+      public Optional<Segment> acquireCachedSegment(SegmentId segmentId, AcquireMode acquireMode)
       {
         for (Map.Entry<DataSegment, File> entry : segments.entrySet()) {
           if (entry.getKey().getId().equals(segmentId)) {
@@ -1991,7 +1992,7 @@ public Optional<Segment> acquireCachedSegment(SegmentId segmentId)
       }
 
       @Override
-      public AcquireSegmentAction acquireSegment(DataSegment dataSegment)
+      public AcquireSegmentAction acquireSegment(DataSegment dataSegment, AcquireMode acquireMode)
       {
         final Segment segment =
             new QueryableIndexSegment(indexIO.loadIndex(segments.get(dataSegment)), dataSegment.getId());

indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java

@@ -53,6 +53,7 @@
 import org.apache.druid.segment.Segment;
 import org.apache.druid.segment.TestIndex;
 import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.segment.loading.SegmentCacheManager;
 import org.apache.druid.segment.loading.SegmentLoadingException;
 import org.apache.druid.segment.loading.TombstoneLoadSpec;
@@ -260,7 +261,7 @@ private Segment loadSegment(DataSegment dataSegment, File tempSegmentDir)
         .manufacturate(tempSegmentDir, null, false);
     try {
       cacheManager.load(dataSegment);
-      return cacheManager.acquireCachedSegment(dataSegment.getId()).orElseThrow();
+      return cacheManager.acquireCachedSegment(dataSegment.getId(), AcquireMode.FULL).orElseThrow();
     }
     catch (SegmentLoadingException e) {
       throw new RuntimeException(e);

multi-stage-query/src/main/java/org/apache/druid/msq/input/AdaptedLoadableSegment.java

@@ -31,6 +31,7 @@
 import org.apache.druid.query.SegmentDescriptor;
 import org.apache.druid.segment.ReferenceCountedSegmentProvider;
 import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.segment.loading.AcquireSegmentAction;
 import org.apache.druid.segment.loading.AcquireSegmentResult;
 import org.apache.druid.timeline.DataSegment;
@@ -129,16 +130,21 @@ public String description()
   }
 
   /**
-   * Adapted segments are not managed by SegmentManager, so they are never cached.
+   * Adapted segments are not managed by SegmentManager, so they are never cached. The {@code acquireMode} is ignored:
+   * an adapted segment is produced by its own async supplier, not the cache manager's full-vs-partial machinery.
    */
   @Override
-  public Optional<Segment> acquireIfCached()
+  public Optional<Segment> acquireIfCached(AcquireMode acquireMode)
   {
     return Optional.empty();
   }
 
+  /**
+   * The {@code acquireMode} is ignored: an adapted segment is produced by its own async supplier, not the cache
+   * manager's full-vs-partial machinery.
+   */
   @Override
-  public AcquireSegmentAction acquire()
+  public AcquireSegmentAction acquire(AcquireMode acquireMode)
   {
     if (!acquired.compareAndSet(false, true)) {
       throw DruidException.defensive("Segment with descriptor[%s] is already acquired", descriptor);

multi-stage-query/src/main/java/org/apache/druid/msq/input/LoadableSegment.java

@@ -24,6 +24,7 @@
 import org.apache.druid.error.DruidException;
 import org.apache.druid.query.SegmentDescriptor;
 import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.segment.loading.AcquireSegmentAction;
 import org.apache.druid.server.SegmentManager;
 import org.apache.druid.timeline.DataSegment;
@@ -37,7 +38,8 @@
  * {@link PhysicalInputSlice#getLoadableSegments()}, which are in turn generated by {@link InputSliceReader#attach}.
  *
  * These objects do not hold resources and do not have a lifecycle. They are just pointers to data that hasn't
- * been acquired yet. Once {@link #acquire()} or {@link #acquireIfCached()} is called, the lifecycle begins.
+ * been acquired yet. Once {@link #acquire(AcquireMode)} or {@link #acquireIfCached(AcquireMode)} is called, the
+ * lifecycle begins.
  *
  * @see RegularLoadableSegment for segments loaded via {@link SegmentManager}
  * @see AdaptedLoadableSegment for segments adapted from other data sources
@@ -61,23 +63,35 @@ public interface LoadableSegment
    *
    * If the returned {@link Optional} is present, the caller is responsible for closing the {@link Segment}
    * when finished with it. If the returned {@link Optional} is empty, the segment is not cached and the caller
-   * should use {@link #acquire()} to load it.
+   * should use {@link #acquire(AcquireMode)} to load it.
+   *
+   * The {@code acquireMode} selects what counts as a cached hit and how the segment is mounted; see {@link AcquireMode}.
+   * With {@link AcquireMode#PARTIAL} the returned {@link Segment} is mounted but, for a partial-download (virtual
+   * storage) segment, may not be fully downloaded yet: its data is fetched on demand at cursor-build time. Callers must
+   * therefore access it through the async API ({@link org.apache.druid.segment.CursorFactory#makeCursorHolderAsync})
+   * rather than the synchronous {@code makeCursorHolder}, which refuses to download on a processing thread.
    *
    * @return an Optional containing the cached Segment if available, or empty if not cached
    */
-  Optional<Segment> acquireIfCached();
+  Optional<Segment> acquireIfCached(AcquireMode acquireMode);
 
   /**
    * Acquire the actual segment. Non-blocking operation. Once this is called, callers are responsible for closing the
    * {@link AcquireSegmentAction}.
    *
+   * The {@code acquireMode} selects how the segment is loaded; see {@link AcquireMode}. With {@link AcquireMode#PARTIAL}
+   * the returned {@link Segment} is mounted but, for a partial-download (virtual storage) segment, may not be fully
+   * downloaded yet: its data is fetched on demand at cursor-build time. Callers must therefore access it through the
+   * async API ({@link org.apache.druid.segment.CursorFactory#makeCursorHolderAsync}) rather than the
+   * synchronous {@code makeCursorHolder}, which refuses to download on a processing thread.
+   *
    * @throws DruidException if the segment has already been acquired
    */
-  AcquireSegmentAction acquire();
+  AcquireSegmentAction acquire(AcquireMode acquireMode);
 
   /**
    * Returns a future for the {@link DataSegment} object. For {@link RegularLoadableSegment}, the future is created
-   * lazily when this method is first called, or when a segment is acquired through {@link #acquire()}.
+   * lazily when this method is first called, or when a segment is acquired through {@link #acquire(AcquireMode)}.
    * For {@link AdaptedLoadableSegment}, this returns a failed future.
    */
   ListenableFuture<DataSegment> dataSegmentFuture();

multi-stage-query/src/main/java/org/apache/druid/msq/input/RegularLoadableSegment.java

@@ -32,6 +32,7 @@
 import org.apache.druid.query.SegmentDescriptor;
 import org.apache.druid.query.TableDataSource;
 import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.segment.loading.AcquireSegmentAction;
 import org.apache.druid.server.SegmentManager;
 import org.apache.druid.timeline.DataSegment;
@@ -127,13 +128,13 @@ public String description()
   }
 
   @Override
-  public synchronized Optional<Segment> acquireIfCached()
+  public synchronized Optional<Segment> acquireIfCached(AcquireMode acquireMode)
   {
     if (acquired) {
       throw DruidException.defensive("Segment with descriptor[%s] is already acquired", descriptor);
     }
 
-    final Optional<Segment> cachedSegment = segmentManager.acquireCachedSegment(segmentId);
+    final Optional<Segment> cachedSegment = segmentManager.acquireCachedSegment(segmentId, acquireMode);
     if (cachedSegment.isPresent()) {
       acquired = true;
 
@@ -148,7 +149,7 @@ public synchronized Optional<Segment> acquireIfCached()
   }
 
   @Override
-  public synchronized AcquireSegmentAction acquire()
+  public synchronized AcquireSegmentAction acquire(AcquireMode acquireMode)
   {
     if (acquired) {
       throw DruidException.defensive("Segment with descriptor[%s] is already acquired", descriptor);
@@ -157,7 +158,7 @@ public synchronized AcquireSegmentAction acquire()
     acquired = true;
 
     if (cachedDataSegment != null) {
-      final AcquireSegmentAction action = segmentManager.acquireSegment(cachedDataSegment);
+      final AcquireSegmentAction action = segmentManager.acquireSegment(cachedDataSegment, acquireMode);
       return new AcquireSegmentAction(
           () -> LoadableSegmentUtils.countedLoad(
               action.getSegmentFuture(),
@@ -177,7 +178,7 @@ public synchronized AcquireSegmentAction acquire()
           Suppliers.memoize(() -> FutureUtils.transformAsync(
               dataSegmentFutureSupplier.get(),
               dataSegment -> LoadableSegmentUtils.countedLoad(
-                  closer.register(segmentManager.acquireSegment(dataSegment)).getSegmentFuture(),
+                  closer.register(segmentManager.acquireSegment(dataSegment, acquireMode)).getSegmentFuture(),
                   dataSegment.getSize(),
                   inputCounters
               )

multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BaseLeafStageProcessor.java

@@ -55,6 +55,7 @@
 import org.apache.druid.query.filter.SegmentPruner;
 import org.apache.druid.query.planning.ExecutionVertex;
 import org.apache.druid.segment.SegmentMapFunction;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.utils.CollectionUtils;
 
 import javax.annotation.Nullable;
@@ -263,6 +264,23 @@ protected List<PhysicalInputSlice> filterBaseInput(final List<PhysicalInputSlice
     return slices;
   }
 
+
+  /**
+   * Create the {@link ReadableInputQueue} for this stage's base inputs. Override to change how the queue acquires
+   * segments: the default acquires them fully up front ({@link AcquireMode#FULL}), which is safe for any processor.
+   * Leaf processors that read segments through the async cursor API
+   * ({@link org.apache.druid.segment.CursorFactory#makeCursorHolderAsync}) can override to {@link AcquireMode#PARTIAL}
+   * so only the columns a query actually touches are downloaded.
+   */
+  protected ReadableInputQueue makeReadableInputQueue(
+      final StandardPartitionReader partitionReader,
+      final List<PhysicalInputSlice> slices,
+      final int loadahead
+  )
+  {
+    return new ReadableInputQueue(partitionReader, slices, loadahead, AcquireMode.FULL);
+  }
+
   /**
    * Read base inputs, where "base" is meant in the same sense as in {@link ExecutionVertex}: the primary datasource
    * that drives query processing.
@@ -293,7 +311,7 @@ private ReadableInputQueue makeBaseInputQueue(
     }
 
     final List<PhysicalInputSlice> filteredSlices = filterBaseInput(physicalInputSlices);
-    return new ReadableInputQueue(
+    return makeReadableInputQueue(
         new StandardPartitionReader(context),
         filteredSlices,
         context.segmentLoadAheadCount()

multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java

@@ -34,6 +34,7 @@
 import org.apache.druid.msq.input.stage.ReadablePartition;
 import org.apache.druid.segment.Segment;
 import org.apache.druid.segment.SegmentReference;
+import org.apache.druid.segment.loading.AcquireMode;
 import org.apache.druid.segment.loading.AcquireSegmentAction;
 import org.apache.druid.utils.CloseableUtils;
 
@@ -98,16 +99,25 @@ public class ReadableInputQueue implements Closeable
 
   private final StandardPartitionReader partitionReader;
   private final int loadahead;
+
+  /**
+   * How segments in this queue are acquired (fully up front, or partially with on-demand column loading at query
+   * time). Threaded through to {@link LoadableSegment#acquire(AcquireMode)} and
+   * {@link LoadableSegment#acquireIfCached(AcquireMode)}.
+   */
+  private final AcquireMode acquireMode;
   private final AtomicBoolean started = new AtomicBoolean(false);
 
   public ReadableInputQueue(
       final StandardPartitionReader partitionReader,
       final List<PhysicalInputSlice> slices,
-      final int loadahead
+      final int loadahead,
+      final AcquireMode acquireMode
   )
   {
     this.partitionReader = partitionReader;
     this.loadahead = loadahead;
+    this.acquireMode = acquireMode;
 
     for (final PhysicalInputSlice slice : slices) {
       loadableSegments.addAll(slice.getLoadableSegments());
@@ -131,7 +141,7 @@ public void start()
         final List<LoadableSegment> toLoad = new ArrayList<>(); // Temporarily store all non-cached segments
         LoadableSegment loadableSegment;
         while ((loadableSegment = loadableSegments.poll()) != null) {
-          final Optional<Segment> cachedSegment = loadableSegment.acquireIfCached();
+          final Optional<Segment> cachedSegment = loadableSegment.acquireIfCached(acquireMode);
           if (cachedSegment.isPresent()) {
             final SegmentReferenceHolder holder = new SegmentReferenceHolder(
                 new SegmentReference(loadableSegment.descriptor(), cachedSegment, null),
@@ -288,7 +298,7 @@ private ListenableFuture<ReadableInput> loadNextSegment()
         return null;
       }
 
-      final AcquireSegmentAction acquireSegmentAction = nextLoadableSegment.acquire();
+      final AcquireSegmentAction acquireSegmentAction = nextLoadableSegment.acquire(acquireMode);
       loadingSegments.add(acquireSegmentAction);
       return FutureUtils.transform(
           acquireSegmentAction.getSegmentFuture(),