This is an automated email from the ASF dual-hosted git repository.
clintropolis pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new b20f29c34bb minor: improve partial load metrics (#19632)
b20f29c34bb is described below
commit b20f29c34bb85b4a16dea426eecaeed8970fdc3e
Author: Clint Wylie <[email protected]>
AuthorDate: Sat Jun 27 10:56:53 2026 -0700
minor: improve partial load metrics (#19632)
---
.../embedded/query/QueryVirtualStorageTest.java | 25 ++++----
.../file/PartialSegmentDownloadListener.java | 62 +++++++++++++++++++
.../segment/file/PartialSegmentFileMapperV10.java | 25 ++++++--
.../PartialQueryableIndexCursorFactoryTest.java | 4 +-
...PartialQueryableIndexCursorFactoryTestBase.java | 4 +-
.../segment/PartialQueryableIndexSegmentTest.java | 4 +-
.../druid/segment/PartialQueryableIndexTest.java | 70 +++++++---------------
.../file/PartialSegmentFileMapperV10Test.java | 51 +++++++---------
.../loading/PartialSegmentMetadataCacheEntry.java | 22 ++++++-
.../segment/loading/SegmentLocalCacheManager.java | 23 ++-----
.../druid/segment/loading/StorageLocation.java | 39 ++++++++++++
.../loading/VirtualStorageLocationStats.java | 23 ++++++-
.../druid/server/metrics/StorageMonitor.java | 42 +++++++++++--
.../loading/PartialSegmentCacheBootstrapTest.java | 44 +++++---------
...SegmentLocalCacheManagerPartialAcquireTest.java | 54 ++++++++++++++++-
.../druid/server/metrics/LatchableEmitter.java | 46 ++++++++++++++
16 files changed, 380 insertions(+), 158 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryVirtualStorageTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryVirtualStorageTest.java
index 24e812a5f34..b9d78d1fa6c 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryVirtualStorageTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryVirtualStorageTest.java
@@ -24,7 +24,6 @@ import org.apache.druid.data.input.impl.LocalInputSource;
import org.apache.druid.indexer.TaskState;
import org.apache.druid.java.util.common.HumanReadableBytes;
import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
import org.apache.druid.msq.indexing.report.MSQTaskReport;
import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
import org.apache.druid.query.DefaultQueryMetrics;
@@ -74,6 +73,9 @@ class QueryVirtualStorageTest extends EmbeddedClusterTestBase
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");
+ // Quiescence wait for the storage monitor: a few times its PT1s emission
period, so a single missed tick doesn't
+ // falsely read as "idle" while still bounding how long we wait once
activity has actually stopped.
+ private static final long MONITOR_QUIESCE_TIMEOUT_MILLIS = 3_000L;
private final EmbeddedBroker broker = new EmbeddedBroker();
private final EmbeddedIndexer indexer = new EmbeddedIndexer();
@@ -191,19 +193,10 @@ class QueryVirtualStorageTest extends
EmbeddedClusterTestBase
LatchableEmitter emitter = historical.latchableEmitter();
LatchableEmitter coordinatorEmitter = coordinator.latchableEmitter();
- // clear out the pipe to get zerod out storage monitor metrics
- ServiceMetricEvent monitorEvent = emitter.waitForNextEvent(event ->
event.hasMetricName(StorageMonitor.VSF_LOAD_BEGIN_COUNT));
- while (monitorEvent != null && monitorEvent.getValue().longValue() > 0) {
- monitorEvent = emitter.waitForNextEvent(event ->
event.hasMetricName(StorageMonitor.VSF_LOAD_BEGIN_COUNT));
- }
- // then flush (which clears out the internal events stores in test
emitter) so we can do clean sums across them
+ // Wait for any in-flight storage activity to settle before taking our
baseline.
+ emitter.awaitMetricQuiescent(StorageMonitor.VSF_LOAD_BEGIN_COUNT,
MONITOR_QUIESCE_TIMEOUT_MILLIS);
emitter.flush();
- emitter.waitForNextEvent(event ->
event.hasMetricName(StorageMonitor.VSF_LOAD_BEGIN_COUNT));
- long beforeLoads =
emitter.getMetricEventLongSum(StorageMonitor.VSF_LOAD_BEGIN_COUNT);
- // confirm flushed
- Assertions.assertEquals(0, beforeLoads);
-
// run the queries in order
Assertions.assertEquals(expectedResults[0],
Long.parseLong(cluster.runSql(queries[0], dataSource)));
assertQueryMetrics(1, expectedLoads[0]);
@@ -234,7 +227,7 @@ class QueryVirtualStorageTest extends
EmbeddedClusterTestBase
Assertions.assertTrue(hits >= expectedTotalHits, "expected " +
expectedTotalHits + " but only got " + hits);
if (expectedTotalHits > 0) {
emitter.waitForNextEvent(event ->
event.hasMetricName(StorageMonitor.VSF_HIT_BYTES));
-
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_HIT_BYTES)
>= 0);
+
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_HIT_BYTES)
> 0);
}
emitter.waitForNextEvent(event ->
event.hasMetricName(StorageMonitor.VSF_LOAD_BEGIN_COUNT));
long loads =
emitter.getMetricEventLongSum(StorageMonitor.VSF_LOAD_BEGIN_COUNT);
@@ -242,8 +235,12 @@ class QueryVirtualStorageTest extends
EmbeddedClusterTestBase
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_LOAD_BEGIN_BYTES)
> 0);
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_LOAD_COUNT)
> 0);
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_LOAD_BYTES)
> 0);
+ emitter.waitForNextEvent(event ->
event.hasMetricName(StorageMonitor.VSF_READ_COUNT));
+
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_READ_COUNT)
> 0);
+
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_READ_BYTES)
> 0);
+
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_READ_TIME)
>= 0);
emitter.waitForNextEvent(event ->
event.hasMetricName(StorageMonitor.VSF_EVICT_COUNT));
-
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_EVICT_COUNT)
>= 0);
+
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_EVICT_COUNT)
> 0);
Assertions.assertTrue(emitter.getMetricEventLongSum(StorageMonitor.VSF_EVICT_BYTES)
> 0);
Assertions.assertEquals(0,
emitter.getMetricEventLongSum(StorageMonitor.VSF_REJECT_COUNT));
Assertions.assertTrue(emitter.getLatestMetricEventValue(StorageMonitor.VSF_USED_BYTES,
0).longValue() > 0);
diff --git
a/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentDownloadListener.java
b/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentDownloadListener.java
new file mode 100644
index 00000000000..111c78b402c
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentDownloadListener.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.file;
+
+/**
+ * Sink for download events emitted by {@link PartialSegmentFileMapperV10} as
it materializes internal files on demand
+ * from deep storage, for collecting metrics.
+ */
+public interface PartialSegmentDownloadListener
+{
+ PartialSegmentDownloadListener NOOP = new PartialSegmentDownloadListener()
+ {
+ @Override
+ public void onBytesDownloaded(long bytes)
+ {
+ // no-op
+ }
+
+ @Override
+ public void onRangeRead(long bytes, long nanos)
+ {
+ // no-op
+ }
+ };
+
+ /**
+ * Bytes were newly materialized on local disk from deep storage and are now
resident and usable: an internal file as
+ * it is accessed (fired exactly once per file, from whichever download path
first records it), or the V10 header on a
+ * genuine cold mount (not a restore from a previous session's on-disk
state). Distinct from
+ * {@link #onRangeRead bytes pulled over the wire}, which can be larger when
a partially-present container is re-read.
+ *
+ * @param bytes the materialized byte count
+ */
+ void onBytesDownloaded(long bytes);
+
+ /**
+ * A single deep-storage range read completed: the actual request
granularity (one read may cover many files when a
+ * whole container is fetched at once). Measures wire bytes and latency, so
it reflects the deep-storage request count
+ * and cost rather than how many files became resident.
+ *
+ * @param bytes the number of bytes read from deep storage in this request
+ * @param nanos the wall-clock time the read took, in nanoseconds
+ */
+ void onRangeRead(long bytes, long nanos);
+}
diff --git
a/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
b/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
index c622584bea9..ea4895a7f2e 100644
---
a/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
+++
b/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
@@ -104,14 +104,16 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
ObjectMapper jsonMapper,
File localCacheDir,
String targetFilename,
- List<String> externals
+ List<String> externals,
+ PartialSegmentDownloadListener downloadListener
) throws IOException
{
final PartialSegmentFileMapperV10 entryPoint = createForFile(
rangeReader,
jsonMapper,
localCacheDir,
- targetFilename
+ targetFilename,
+ downloadListener
);
final Map<String, PartialSegmentFileMapperV10> externalMappers = new
HashMap<>();
@@ -119,7 +121,7 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
for (String filename : externals) {
externalMappers.put(
filename,
- createForFile(rangeReader, jsonMapper, localCacheDir, filename)
+ createForFile(rangeReader, jsonMapper, localCacheDir, filename,
downloadListener)
);
}
}
@@ -139,7 +141,8 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
SegmentRangeReader rangeReader,
ObjectMapper jsonMapper,
File localCacheDir,
- String targetFilename
+ String targetFilename,
+ PartialSegmentDownloadListener downloadListener
) throws IOException
{
FileUtils.mkdirp(localCacheDir);
@@ -172,6 +175,7 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
fetchAndPersistHeader(rangeReader, targetFilename, headerFile);
result = parseHeaderFile(headerFile, jsonMapper);
bitmapBuffer = mmapBitmap(headerFile, result);
+ downloadListener.onBytesDownloaded(headerFile.length());
}
final PartialSegmentFileMapperV10 mapper = new PartialSegmentFileMapperV10(
@@ -180,7 +184,8 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
rangeReader,
targetFilename,
localCacheDir,
- bitmapBuffer
+ bitmapBuffer,
+ downloadListener
);
// bitmap-vs-container repair pre-pass: if the bitmap claims a file is
downloaded but its container file is
@@ -260,6 +265,7 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
private final MappedByteBuffer bitmapBuffer;
private final AtomicLong downloadedBytes = new AtomicLong(0);
private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final PartialSegmentDownloadListener downloadListener;
private PartialSegmentFileMapperV10(
SegmentFileMetadata metadata,
@@ -267,7 +273,8 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
SegmentRangeReader rangeReader,
String targetFilename,
File localCacheDir,
- MappedByteBuffer bitmapBuffer
+ MappedByteBuffer bitmapBuffer,
+ PartialSegmentDownloadListener downloadListener
)
{
this.metadata = metadata;
@@ -276,6 +283,7 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
this.targetFilename = targetFilename;
this.localCacheDir = localCacheDir;
this.bitmapBuffer = bitmapBuffer;
+ this.downloadListener = downloadListener;
// build stable file name ordering for bitmap indexing
this.sortedFileNames = new ArrayList<>(new
TreeSet<>(metadata.getFiles().keySet()));
@@ -788,6 +796,7 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
) throws IOException
{
// stream straight from deep storage to the local container file to avoid
heap-buffering the whole range
+ final long startNanos = System.nanoTime();
try (ResourceHolder<byte[]> bufHolder = CompressedPools.getOutputBytes();
InputStream is = rangeReader.readRange(targetFilename,
absoluteOffset, size);
RandomAccessFile raf = new
RandomAccessFile(containerFiles[containerIndex], "rw")) {
@@ -809,6 +818,9 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
remaining -= read;
}
}
+ // Report the completed deep-storage range read (reached only on success).
One read may cover many files; this is
+ // the actual request granularity, so it measures wire bytes + latency
rather than bytes that became resident.
+ downloadListener.onRangeRead(size, System.nanoTime() - startNanos);
}
/**
@@ -821,6 +833,7 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
{
if (downloadedFiles.add(name)) {
downloadedBytes.addAndGet(size);
+ downloadListener.onBytesDownloaded(size);
markDownloadedInBitmap(name);
}
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTest.java
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTest.java
index 1e2ef260f3f..aac367cce19 100644
---
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTest.java
@@ -47,6 +47,7 @@ import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
import org.apache.druid.segment.data.CompressionStrategy;
import org.apache.druid.segment.file.CountingRangeReader;
+import org.apache.druid.segment.file.PartialSegmentDownloadListener;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.segment.incremental.IncrementalIndexSchema;
import org.apache.druid.segment.projections.Projections;
@@ -323,7 +324,8 @@ class PartialQueryableIndexCursorFactoryTest extends
PartialQueryableIndexCursor
TestHelper.makeJsonMapper(),
cacheDir,
IndexIO.V10_FILE_NAME,
- Collections.emptyList()
+ Collections.emptyList(),
+ PartialSegmentDownloadListener.NOOP
);
try {
final PartialQueryableIndex index =
diff --git
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTestBase.java
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTestBase.java
index 0b8ccf3b11a..334baaa0442 100644
---
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTestBase.java
+++
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTestBase.java
@@ -26,6 +26,7 @@ import org.apache.druid.common.asyncresource.AsyncResources;
import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.segment.column.ColumnConfig;
import org.apache.druid.segment.file.CountingRangeReader;
+import org.apache.druid.segment.file.PartialSegmentDownloadListener;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.testing.InitializedNullHandlingTest;
import org.junit.jupiter.api.io.TempDir;
@@ -62,7 +63,8 @@ abstract class PartialQueryableIndexCursorFactoryTestBase
extends InitializedNul
TestHelper.makeJsonMapper(),
cacheDir,
IndexIO.V10_FILE_NAME,
- Collections.emptyList()
+ Collections.emptyList(),
+ PartialSegmentDownloadListener.NOOP
);
return new IndexAndMapper(
new PartialQueryableIndex(mapper.getSegmentFileMetadata(), mapper,
COLUMN_CONFIG),
diff --git
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexSegmentTest.java
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexSegmentTest.java
index 72156b740a1..c23e1745d3d 100644
---
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexSegmentTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexSegmentTest.java
@@ -37,6 +37,7 @@ import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
import org.apache.druid.segment.data.CompressionStrategy;
import org.apache.druid.segment.file.CountingRangeReader;
+import org.apache.druid.segment.file.PartialSegmentDownloadListener;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.segment.incremental.IncrementalIndexSchema;
import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
@@ -131,7 +132,8 @@ class PartialQueryableIndexSegmentTest extends
InitializedNullHandlingTest
TestHelper.makeJsonMapper(),
cacheDir,
IndexIO.V10_FILE_NAME,
- Collections.emptyList()
+ Collections.emptyList(),
+ PartialSegmentDownloadListener.NOOP
);
return new PartialQueryableIndex(mapper.getSegmentFileMetadata(), mapper,
COLUMN_CONFIG);
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexTest.java
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexTest.java
index f40b0c5523b..20c95f5955c 100644
---
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexTest.java
@@ -38,9 +38,11 @@ import org.apache.druid.segment.column.RowSignature;
import org.apache.druid.segment.column.ValueType;
import org.apache.druid.segment.data.CompressionStrategy;
import org.apache.druid.segment.file.CountingRangeReader;
+import org.apache.druid.segment.file.PartialSegmentDownloadListener;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.segment.incremental.IncrementalIndexSchema;
import org.apache.druid.segment.loading.DirectoryBackedRangeReader;
+import org.apache.druid.segment.loading.SegmentRangeReader;
import org.apache.druid.segment.projections.QueryableProjection;
import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
import org.apache.druid.testing.InitializedNullHandlingTest;
@@ -139,13 +141,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("schema");
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- TestHelper.makeJsonMapper(),
- cacheDir,
- IndexIO.V10_FILE_NAME,
- Collections.emptyList()
- )) {
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
rangeReader.resetCount();
final PartialQueryableIndex index = new PartialQueryableIndex(
@@ -175,13 +171,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("caps");
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- TestHelper.makeJsonMapper(),
- cacheDir,
- IndexIO.V10_FILE_NAME,
- Collections.emptyList()
- )) {
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
rangeReader.resetCount();
final PartialQueryableIndex index = new PartialQueryableIndex(
@@ -219,13 +209,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("colholder");
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- TestHelper.makeJsonMapper(),
- cacheDir,
- IndexIO.V10_FILE_NAME,
- Collections.emptyList()
- )) {
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
rangeReader.resetCount();
final PartialQueryableIndex index = new PartialQueryableIndex(
@@ -255,13 +239,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("projection");
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- TestHelper.makeJsonMapper(),
- cacheDir,
- IndexIO.V10_FILE_NAME,
- Collections.emptyList()
- )) {
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
final PartialQueryableIndex index = new PartialQueryableIndex(
mapper.getSegmentFileMetadata(),
mapper,
@@ -336,13 +314,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("per_col");
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- TestHelper.makeJsonMapper(),
- cacheDir,
- IndexIO.V10_FILE_NAME,
- Collections.emptyList()
- )) {
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
final PartialQueryableIndex index = new PartialQueryableIndex(
mapper.getSegmentFileMetadata(),
mapper,
@@ -389,13 +361,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("no_proj");
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- TestHelper.makeJsonMapper(),
- cacheDir,
- IndexIO.V10_FILE_NAME,
- Collections.emptyList()
- )) {
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
final PartialQueryableIndex index = new PartialQueryableIndex(
mapper.getSegmentFileMetadata(),
mapper,
@@ -421,13 +387,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
try (
QueryableIndex eagerIndex = indexIO.loadIndex(segmentDir);
- PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- TestHelper.makeJsonMapper(),
- cacheDir,
- IndexIO.V10_FILE_NAME,
- Collections.emptyList()
- )
+ PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)
) {
final PartialQueryableIndex partialIndex = new PartialQueryableIndex(
mapper.getSegmentFileMetadata(),
@@ -466,4 +426,16 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
FileUtils.mkdirp(dir);
return dir;
}
+
+ private static PartialSegmentFileMapperV10 createMapper(SegmentRangeReader
rangeReader, File cacheDir) throws IOException
+ {
+ return PartialSegmentFileMapperV10.create(
+ rangeReader,
+ TestHelper.makeJsonMapper(),
+ cacheDir,
+ IndexIO.V10_FILE_NAME,
+ Collections.emptyList(),
+ PartialSegmentDownloadListener.NOOP
+ );
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
b/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
index a2403874757..aa9e420d0d4 100644
---
a/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
+++
b/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
@@ -306,13 +306,7 @@ class PartialSegmentFileMapperV10Test
final File cacheDir = newCacheDir("ext");
final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(baseDir);
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of(externalName)
- )) {
+ try (PartialSegmentFileMapperV10 mapper =
createMapperWithExternal(rangeReader, cacheDir, externalName)) {
// verify main file internal files
for (int i = 0; i < 10; ++i) {
ByteBuffer buf = mapper.mapFile(String.valueOf(i));
@@ -355,13 +349,7 @@ class PartialSegmentFileMapperV10Test
final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(baseDir);
try (SegmentFileMapperV10 eager = SegmentFileMapperV10.create(segmentFile,
JSON_MAPPER, List.of(externalName));
- PartialSegmentFileMapperV10 lazy = PartialSegmentFileMapperV10.create(
- rangeReader,
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of(externalName)
- )
+ PartialSegmentFileMapperV10 lazy =
createMapperWithExternal(rangeReader, cacheDir, externalName)
) {
// verify main files match
for (int i = 0; i < 5; ++i) {
@@ -455,26 +443,14 @@ class PartialSegmentFileMapperV10Test
final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(baseDir);
// create with externals, download some files
- try (PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- rangeReader,
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of(externalName)
- )) {
+ try (PartialSegmentFileMapperV10 mapper =
createMapperWithExternal(rangeReader, cacheDir, externalName)) {
mapper.mapFile("1");
mapper.mapExternalFile(externalName, "7");
}
// restore, previously downloaded files should be available
final DirectoryBackedRangeReader freshReader = new
DirectoryBackedRangeReader(baseDir);
- try (PartialSegmentFileMapperV10 restored =
PartialSegmentFileMapperV10.create(
- freshReader,
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of(externalName)
- )) {
+ try (PartialSegmentFileMapperV10 restored =
createMapperWithExternal(freshReader, cacheDir, externalName)) {
ByteBuffer buf1 = restored.mapFile("1");
Assertions.assertNotNull(buf1);
Assertions.assertEquals(1, buf1.getInt());
@@ -650,7 +626,24 @@ class PartialSegmentFileMapperV10Test
rangeReader,
JSON_MAPPER,
localCacheDir,
- IndexIO.V10_FILE_NAME
+ IndexIO.V10_FILE_NAME,
+ PartialSegmentDownloadListener.NOOP
+ );
+ }
+
+ private static PartialSegmentFileMapperV10 createMapperWithExternal(
+ SegmentRangeReader rangeReader,
+ File localCacheDir,
+ String externalName
+ ) throws IOException
+ {
+ return PartialSegmentFileMapperV10.create(
+ rangeReader,
+ JSON_MAPPER,
+ localCacheDir,
+ IndexIO.V10_FILE_NAME,
+ List.of(externalName),
+ PartialSegmentDownloadListener.NOOP
);
}
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
b/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
index e5f8aa55fab..d597b656da5 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/PartialSegmentMetadataCacheEntry.java
@@ -34,6 +34,7 @@ import org.apache.druid.segment.PartialQueryableIndexSegment;
import org.apache.druid.segment.ReferenceCountingCloseableObject;
import org.apache.druid.segment.Segment;
import org.apache.druid.segment.column.ColumnConfig;
+import org.apache.druid.segment.file.PartialSegmentDownloadListener;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.segment.file.SegmentFileBuilder;
import org.apache.druid.segment.file.SegmentFileMetadata;
@@ -289,8 +290,7 @@ public class PartialSegmentMetadataCacheEntry implements
SegmentCacheEntry, Resi
* the race where the entry's reservation gets evicted (e.g. cache picks a
weak entry whose lone hold was released
* by a concurrent canceler, or {@link StorageLocation#release} fires on the
static entry from a coordinator drop)
* while mount() is still in progress. Without this check, mount would
commit local state for an entry the cache
- * manager no longer knows about, leaking files on disk and memory mappings.
Mirrors the same defensive check in
- * {@code SegmentCacheEntry.mount}. Returns normally if rollback fires;
callers detect via {@link #isMounted}.
+ * manager no longer knows about, leaking files on disk and memory mappings.
*/
private void verifyStillReservedOrRollback(StorageLocation mountLocation)
{
@@ -336,7 +336,8 @@ public class PartialSegmentMetadataCacheEntry implements
SegmentCacheEntry, Resi
jsonMapper,
localCacheDir,
targetFilename,
- externalFilenames
+ externalFilenames,
+ new WeakLoadTracker(mountLocation)
);
final long sizeToAdjust;
@@ -979,4 +980,19 @@ public class PartialSegmentMetadataCacheEntry implements
SegmentCacheEntry, Resi
LOG.warn("Failed to delete header file[%s] during unmount of partial
segment[%s]", file, segmentId);
}
}
+
+ private record WeakLoadTracker(StorageLocation mountLocation) implements
PartialSegmentDownloadListener
+ {
+ @Override
+ public void onBytesDownloaded(long bytes)
+ {
+ mountLocation.trackWeakLoad(bytes);
+ }
+
+ @Override
+ public void onRangeRead(long bytes, long nanos)
+ {
+ mountLocation.trackWeakRangeRead(bytes, nanos);
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
index b19a9e51a8f..f5d2452d9c3 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
@@ -678,34 +678,19 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
final PartialSegmentFileMapperV10 mapper =
reserved.metadata.getFileMapper();
final long loadSizeBytes;
if (fullDownload) {
- // Delta of internal-file bytes downloaded by this task.
The small header range-read isn't
- // counted in getDownloadedBytes; it's the negligible
mount cost.
+ // Delta of internal-file bytes downloaded by this task
final long downloadedBefore = mapper.getDownloadedBytes();
- // Mount every bundle so the containers it owns are
reserved on the location, SIEVE-evictable, and
- // deleted on drop. Writing through the file mapper alone
(ensureAllDownloaded) would leave those
- // containers unmanaged by any cache entry. Each acquire
returns a hold added to loadCleanup, so
- // the fully-materialized segment stays resident for the
eager (sync-cursor) caller until the
- // action closes; on later eviction acquireCachedSegment
returns empty and the caller re-downloads.
- // acquire() mounts each bundle's parents first, so
iteration order doesn't matter.
+ // Mount every bundle so the containers it owns are
reserved on the location
for (String bundleName :
PartialSegmentBundleCacheEntry.bundleNames(mapper)) {
holdHolder.add(reserved.metadata.getBundleAcquirer().acquire(bundleName));
}
mapper.ensureAllDownloaded();
loadSizeBytes = mapper.getDownloadedBytes() -
downloadedBefore;
} else {
- // Lazy mount: report the header bytes when this task
caused the mount; 0 when the entry was
- // already mounted (a concurrent acquirer or earlier query
did the load).
+ // Lazy mount: the header bytes when this task caused the
mount; 0 when the entry was already
+ // mounted (a concurrent acquirer or earlier query did the
load).
loadSizeBytes = wasMounted ? 0L :
mapper.getOnDiskHeaderSize();
}
- // Record the actual-load bytes on the location
(VSF_LOAD_*): the header bytes for a lazy mount, or the
- // full downloaded delta for a full download. The partial
metadata entry is always weak (reservePartial
- // uses addWeakReservationHold). Skip when nothing was
pulled (a cache hit / already-mounted re-acquire
- // reports 0) so we don't log a spurious 0-byte load.
On-demand per-column downloads on the lazy path
- // happen later at query time and are captured by the
query-layer load metrics (the
- // AcquireSegmentResult load size), not by this per-location
stat.
- if (loadSizeBytes > 0) {
- reserved.location.trackWeakLoad(loadSizeBytes);
- }
final long loadNanos = System.nanoTime() - taskStartNanos;
return new AcquireSegmentResult(
reserved.metadata::acquireReference,
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
b/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
index fbc756f8b0b..460a2fa9d2c 100644
--- a/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
+++ b/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
@@ -794,6 +794,16 @@ public class StorageLocation
weakStats.getAndUpdate(s -> s.load(size));
}
+ /**
+ * Record a single on-demand deep-storage range read: {@code bytes} pulled
over the wire in {@code nanos}. One read
+ * may materialize several internal files (whole-container fetch), so this
is the request-level signal that
+ * complements the per-file {@link #trackWeakLoad}.
+ */
+ public void trackWeakRangeRead(long bytes, long nanos)
+ {
+ weakStats.getAndUpdate(s -> s.rangeRead(bytes, nanos));
+ }
+
private void trackWeakHold(WeakCacheEntry entry)
{
currHoldCount.getAndIncrement();
@@ -1234,6 +1244,9 @@ public class StorageLocation
private final AtomicLong evictionCount = new AtomicLong(0);
private final AtomicLong evictionBytes = new AtomicLong(0);
private final AtomicLong unmountCount = new AtomicLong(0);
+ private final AtomicLong readCount = new AtomicLong(0);
+ private final AtomicLong readBytes = new AtomicLong(0);
+ private final AtomicLong readTimeNanos = new AtomicLong(0);
public WeakStats(AtomicLong sizeUsed, AtomicLong holdCount, AtomicLong
holdBytes)
{
@@ -1293,6 +1306,14 @@ public class StorageLocation
return this;
}
+ public WeakStats rangeRead(long bytes, long nanos)
+ {
+ readCount.getAndIncrement();
+ readBytes.getAndAdd(bytes);
+ readTimeNanos.getAndAdd(nanos);
+ return this;
+ }
+
@Override
public long getUsedBytes()
{
@@ -1365,6 +1386,24 @@ public class StorageLocation
return rejectionCount.get();
}
+ @Override
+ public long getReadCount()
+ {
+ return readCount.get();
+ }
+
+ @Override
+ public long getReadBytes()
+ {
+ return readBytes.get();
+ }
+
+ @Override
+ public long getReadTimeNanos()
+ {
+ return readTimeNanos.get();
+ }
+
@VisibleForTesting
public long getUnmountCount()
{
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/VirtualStorageLocationStats.java
b/server/src/main/java/org/apache/druid/segment/loading/VirtualStorageLocationStats.java
index 5f63b08b659..035d30cafff 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/VirtualStorageLocationStats.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/VirtualStorageLocationStats.java
@@ -87,6 +87,24 @@ public interface VirtualStorageLocationStats
*/
long getRejectCount();
+ /**
+ * Number of deep-storage range reads issued during the measurement period
(on-demand partial downloads). One read
+ * may cover several internal files, so this is the actual request count,
distinct from {@link #getLoadCount()}.
+ */
+ long getReadCount();
+
+ /**
+ * Total bytes pulled from deep storage by range reads during the
measurement period. May exceed
+ * {@link #getLoadBytes()} when a partially-present container is re-fetched
in full.
+ */
+ long getReadBytes();
+
+ /**
+ * Total wall-clock time spent in deep-storage range reads during the
measurement period, in nanoseconds. Divide by
+ * {@link #getReadCount()} for the average per-read latency.
+ */
+ long getReadTimeNanos();
+
/**
* Whether any stats are nonzero.
*/
@@ -103,6 +121,9 @@ public interface VirtualStorageLocationStats
|| getLoadBytes() != 0
|| getEvictionCount() != 0
|| getEvictionBytes() != 0
- || getRejectCount() != 0;
+ || getRejectCount() != 0
+ || getReadCount() != 0
+ || getReadBytes() != 0
+ || getReadTimeNanos() != 0;
}
}
diff --git
a/server/src/main/java/org/apache/druid/server/metrics/StorageMonitor.java
b/server/src/main/java/org/apache/druid/server/metrics/StorageMonitor.java
index a0a551d8bb1..23a8bd4bbf4 100644
--- a/server/src/main/java/org/apache/druid/server/metrics/StorageMonitor.java
+++ b/server/src/main/java/org/apache/druid/server/metrics/StorageMonitor.java
@@ -31,6 +31,7 @@ import
org.apache.druid.segment.loading.VirtualStorageLocationStats;
import javax.annotation.Nullable;
import java.util.List;
+import java.util.concurrent.TimeUnit;
/**
* Monitor to emit stats from {@link StorageLocation}.
@@ -114,22 +115,33 @@ public class StorageMonitor extends AbstractMonitor
/**
* Number of weakly-held objects whose load was started during the
measurement period. Incremented when space is
* reserved, before the object has been downloaded.
+ * <p>
+ * For partial (on-demand) V10 segments, reservation is pessimistic: a
bundle reserves its full container size up
+ * front even though a query may download only a subset of that bundle's
columns. So begin (reserved) and complete
+ * (actually downloaded) deliberately diverge on the partial path; begin
reflects reserved space, the
+ * {@link #VSF_LOAD_BYTES} complete metrics reflect bytes actually pulled
from deep storage.
*/
public static final String VSF_LOAD_BEGIN_COUNT =
"storage/virtual/load/begin/count";
/**
- * Total bytes of weakly-held objects whose load was started during the
measurement period.
+ * Total bytes of weakly-held objects whose load was started during the
measurement period. See
+ * {@link #VSF_LOAD_BEGIN_COUNT} for why this (reserved) can exceed {@link
#VSF_LOAD_BYTES} (actually downloaded) for
+ * partial segments.
*/
public static final String VSF_LOAD_BEGIN_BYTES =
"storage/virtual/load/begin/bytes";
/**
- * Number of weakly-held objects whose load completed during the measurement
period. Incremented after the object has
- * been downloaded and is usable.
+ * Number of load completions during the measurement period. For
fully-downloaded weak entries this is incremented
+ * once per object after it has been downloaded and is usable. For partial
(on-demand) V10 segments it additionally
+ * counts each individual internal-file download (a query lazily pulls only
the columns it needs), so on the partial
+ * path this is file-granular and may exceed {@link #VSF_LOAD_BEGIN_COUNT}
(which is per reserved entry).
*/
public static final String VSF_LOAD_COUNT = "storage/virtual/load/count";
/**
- * Total bytes of weakly-held objects whose load completed during the
measurement period.
+ * Total bytes whose load completed during the measurement period: bytes
actually downloaded from deep storage. For
+ * partial segments this includes header range-reads plus every
lazily-downloaded column, and is the accurate measure
+ * of bandwidth used (vs {@link #VSF_LOAD_BEGIN_BYTES}, which is reserved
space).
*/
public static final String VSF_LOAD_BYTES = "storage/virtual/load/bytes";
@@ -149,6 +161,25 @@ public class StorageMonitor extends AbstractMonitor
*/
public static final String VSF_REJECT_COUNT = "storage/virtual/reject/count";
+ /**
+ * Number of deep-storage range reads issued during the measurement period
for on-demand partial downloads. One read
+ * can cover several internal files (a whole-container fetch), so this is
the actual deep-storage request count;
+ * distinct from {@link #VSF_LOAD_COUNT} (file/object load completions).
+ */
+ public static final String VSF_READ_COUNT = "storage/virtual/read/count";
+
+ /**
+ * Total bytes pulled from deep storage by range reads during the
measurement period. Can exceed
+ * {@link #VSF_LOAD_BYTES} when a partially-present container is re-fetched
in full.
+ */
+ public static final String VSF_READ_BYTES = "storage/virtual/read/bytes";
+
+ /**
+ * Total wall-clock time spent in deep-storage range reads during the
measurement period, in milliseconds. Combined
+ * with {@link #VSF_READ_COUNT} this gives average per-read latency.
+ */
+ public static final String VSF_READ_TIME = "storage/virtual/read/time";
+
private final List<StorageLocation> locations;
private final Supplier<ServiceMetricEvent.Builder> builderSupplier;
@@ -193,6 +224,9 @@ public class StorageMonitor extends AbstractMonitor
emitter.emit(builder.setMetric(VSF_EVICT_COUNT,
weakStats.getEvictionCount()));
emitter.emit(builder.setMetric(VSF_EVICT_BYTES,
weakStats.getEvictionBytes()));
emitter.emit(builder.setMetric(VSF_REJECT_COUNT,
weakStats.getRejectCount()));
+ emitter.emit(builder.setMetric(VSF_READ_COUNT,
weakStats.getReadCount()));
+ emitter.emit(builder.setMetric(VSF_READ_BYTES,
weakStats.getReadBytes()));
+ emitter.emit(builder.setMetric(VSF_READ_TIME,
TimeUnit.NANOSECONDS.toMillis(weakStats.getReadTimeNanos())));
}
}
return true;
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentCacheBootstrapTest.java
b/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentCacheBootstrapTest.java
index fb57c807314..c71e209d004 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentCacheBootstrapTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentCacheBootstrapTest.java
@@ -40,6 +40,7 @@ import org.apache.druid.segment.TestHelper;
import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.file.PartialSegmentDownloadListener;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.segment.incremental.IncrementalIndexSchema;
import org.apache.druid.segment.projections.Projections;
@@ -194,13 +195,7 @@ class PartialSegmentCacheBootstrapTest
primeOnDiskState();
// remove the aggregate bundle's container file(s), base's containers stay
- final PartialSegmentFileMapperV10 introspect =
PartialSegmentFileMapperV10.create(
- new DirectoryBackedRangeReader(deepStorageDir),
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of()
- );
+ final PartialSegmentFileMapperV10 introspect =
createMapper(deepStorageDir, cacheDir);
final List<Integer> aggContainers = new ArrayList<>();
final String prefix = AGG_BUNDLE + "/";
for (var entry :
introspect.getSegmentFileMetadata().getFiles().entrySet()) {
@@ -236,13 +231,7 @@ class PartialSegmentCacheBootstrapTest
// Remove base's container files. After this, base is unrestorable on
disk, which makes the aggregate (which
// depends on base) an orphan that must be deleted rather than restored in
a degenerate state.
- final PartialSegmentFileMapperV10 introspect =
PartialSegmentFileMapperV10.create(
- new DirectoryBackedRangeReader(deepStorageDir),
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of()
- );
+ final PartialSegmentFileMapperV10 introspect =
createMapper(deepStorageDir, cacheDir);
final Set<Integer> baseContainers = new HashSet<>();
final Set<Integer> aggContainers = new HashSet<>();
for (var entry :
introspect.getSegmentFileMetadata().getFiles().entrySet()) {
@@ -439,13 +428,7 @@ class PartialSegmentCacheBootstrapTest
{
primeOnDiskState();
// download a file in the aggregate bundle to set a bit, then close
(persists the bitmap)
- final PartialSegmentFileMapperV10 mapper =
PartialSegmentFileMapperV10.create(
- new DirectoryBackedRangeReader(deepStorageDir),
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of()
- );
+ final PartialSegmentFileMapperV10 mapper = createMapper(deepStorageDir,
cacheDir);
final String prefix = AGG_BUNDLE + "/";
String fileInAgg = null;
int aggContainerIdx = -1;
@@ -482,13 +465,7 @@ class PartialSegmentCacheBootstrapTest
Assertions.assertTrue(aggContainer.delete());
// re-open the mapper: the bitmap-vs-container repair should clear the bit
for the missing file
- try (PartialSegmentFileMapperV10 restored =
PartialSegmentFileMapperV10.create(
- new DirectoryBackedRangeReader(deepStorageDir),
- JSON_MAPPER,
- cacheDir,
- IndexIO.V10_FILE_NAME,
- List.of()
- )) {
+ try (PartialSegmentFileMapperV10 restored = createMapper(deepStorageDir,
cacheDir)) {
Assertions.assertFalse(
restored.getDownloadedFiles().contains(fileInAgg),
"bitmap repair should have cleared the bit for " + fileInAgg
@@ -565,4 +542,15 @@ class PartialSegmentCacheBootstrapTest
return metadata;
}
+ private static PartialSegmentFileMapperV10 createMapper(File deepStorageDir,
File cacheDir) throws IOException
+ {
+ return PartialSegmentFileMapperV10.create(
+ new DirectoryBackedRangeReader(deepStorageDir),
+ JSON_MAPPER,
+ cacheDir,
+ IndexIO.V10_FILE_NAME,
+ List.of(),
+ PartialSegmentDownloadListener.NOOP
+ );
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
index 8bffa4cf040..01862eb2b9a 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
@@ -58,6 +58,7 @@ import org.apache.druid.segment.column.ColumnConfig;
import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.file.PartialSegmentDownloadListener;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.segment.incremental.IncrementalIndexSchema;
import org.apache.druid.segment.projections.Projections;
@@ -563,10 +564,58 @@ class SegmentLocalCacheManagerPartialAcquireTest
// Confirm the metadata entry exists on the location and reports fully
downloaded.
final CacheEntry entry = loc.getCacheEntry(new
SegmentCacheEntryIdentifier(SEGMENT_ID));
Assertions.assertInstanceOf(PartialSegmentMetadataCacheEntry.class, entry);
+ final PartialSegmentMetadataCacheEntry metaEntry =
(PartialSegmentMetadataCacheEntry) entry;
Assertions.assertTrue(
- ((PartialSegmentMetadataCacheEntry) entry).isFullyDownloaded(),
+ metaEntry.isFullyDownloaded(),
"force-download path must leave the segment fully downloaded after
acquire returns"
);
+ Assertions.assertEquals(
+ metaEntry.getFileMapper().getOnDiskHeaderSize() +
metaEntry.getFileMapper().getDownloadedBytes(),
+ loc.getWeakStats().getLoadBytes(),
+ "full-download bytes (header + every column file) must each be
recorded exactly once"
+ );
+ }
+
+ @Test
+ void testLazyColumnDownloadsAreRecordedInWeakLoadStats()
+ throws ExecutionException, InterruptedException, IOException
+ {
+ final StorageLocation loc = manager.getLocations().get(0);
+ try (AcquireSegmentAction action = manager.acquireSegment(partialSegment,
AcquireMode.PARTIAL)) {
+ final AcquireSegmentResult result = action.getSegmentFuture().get();
+ try (Segment segment =
result.getReferenceProvider().acquireReference().orElseThrow()) {
+ // After a lazy mount only the metadata header has been pulled; no
column files yet.
+ final long headerLoadBytes = loc.getWeakStats().getLoadBytes();
+ Assertions.assertTrue(headerLoadBytes > 0, "lazy mount records the
header load");
+
+ // A base-table scan downloads the base bundle's columns on demand at
cursor-build time.
+ try (var asyncHolder =
segment.as(CursorFactory.class).makeCursorHolderAsync(CursorBuildSpec.FULL_SCAN))
{
+ final CountDownLatch ready = new CountDownLatch(1);
+ asyncHolder.addReadyCallback(ready::countDown);
+ Assertions.assertTrue(ready.await(15, TimeUnit.SECONDS));
+ try (CursorHolder cursorHolder = asyncHolder.release()) {
+ Assertions.assertNotNull(cursorHolder.asCursor());
+ }
+ }
+
+ // Those on-demand column downloads are now reflected in the
per-location completed-load metrics (previously
+ // they were recorded nowhere): loadBytes grew past the header, and
loadCount grew per downloaded file.
+ final StorageLocation.WeakStats after = loc.getWeakStats();
+ Assertions.assertTrue(
+ after.getLoadBytes() > headerLoadBytes,
+ "on-demand column downloads must add to VSF_LOAD_BYTES; header=" +
headerLoadBytes
+ + " after=" + after.getLoadBytes()
+ );
+ Assertions.assertTrue(after.getLoadCount() > 1, "each downloaded
column file increments the load count");
+
+ // The deep-storage range reads that pulled those columns are recorded
in the request-level VSF_READ_* stats:
+ // one or more range reads, with wire bytes and nonzero latency. (The
header range-read at mount time is not
+ // counted here — it predates the listener and is accounted as the
load header above.)
+ Assertions.assertTrue(after.getReadCount() > 0, "on-demand range reads
must increment VSF_READ_COUNT");
+ Assertions.assertTrue(after.getReadBytes() > 0, "range reads must
record wire bytes in VSF_READ_BYTES");
+ Assertions.assertTrue(after.getReadTimeNanos() > 0, "range reads must
record latency in VSF_READ_TIME");
+ }
+ }
}
@Test
@@ -876,7 +925,8 @@ class SegmentLocalCacheManagerPartialAcquireTest
jsonMapper,
partialDir,
IndexIO.V10_FILE_NAME,
- List.of()
+ List.of(),
+ PartialSegmentDownloadListener.NOOP
)) {
final int numContainers =
seed.getSegmentFileMetadata().getContainers().size();
for (int i = 0; i < numContainers; i++) {
diff --git
a/server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java
b/server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java
index fc4b40af6d6..499e15fee5c 100644
--- a/server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java
+++ b/server/src/test/java/org/apache/druid/server/metrics/LatchableEmitter.java
@@ -194,6 +194,52 @@ public class LatchableEmitter extends StubServiceEmitter
return matcher.matchingEvent.get();
}
+ /**
+ * Like {@link #waitForNextEvent(UnaryOperator)} but treats a timeout as a
valid outcome rather than throwing:
+ * returns {@code true} if a matching event was emitted within {@code
timeoutMillis} (negative for no timeout),
+ * {@code false} if it timed out. Useful when the absence of an event is
itself meaningful (e.g. waiting for a
+ * periodic monitor to fall quiet).
+ */
+ public boolean tryWaitForNextEvent(UnaryOperator<EventMatcher> condition,
long timeoutMillis)
+ {
+ final EventMatcher matcher = condition.apply(new EventMatcher());
+ final WaitCondition waitCondition = new WaitCondition(
+ event -> event instanceof ServiceMetricEvent &&
matcher.test((ServiceMetricEvent) event)
+ );
+ registerWaitConditionNextEvent(waitCondition);
+
+ try {
+ final long awaitTime = timeoutMillis >= 0 ? timeoutMillis :
Long.MAX_VALUE;
+ return waitCondition.countDownLatch.await(awaitTime,
TimeUnit.MILLISECONDS);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ finally {
+ waitConditions.remove(waitCondition);
+ }
+ }
+
+ /**
+ * Wait until the given metric has quiesced: its cumulative value (per
{@link #getMetricEventLongSum}) stops growing
+ * across consecutive emissions, or no matching event is emitted within
{@code timeoutMillis} (which, for a metric
+ * only emitted while there is activity, itself means idle). Unlike {@link
#waitForNextEvent}, this never throws on
+ * timeout, a quiet period is the expected terminating signal. A "wait for
background activity to settle" latch
+ * without a fixed sleep, e.g. to drain stragglers before taking a metric
baseline. Size {@code timeoutMillis} to a
+ * few times the emitting monitor's period so a single missed tick doesn't
read as idle.
+ */
+ public void awaitMetricQuiescent(String metricName, long timeoutMillis)
+ {
+ long previous = getMetricEventLongSum(metricName);
+ while (tryWaitForNextEvent(event -> event.hasMetricName(metricName),
timeoutMillis)) {
+ final long current = getMetricEventLongSum(metricName);
+ if (current == previous) {
+ return;
+ }
+ previous = current;
+ }
+ }
+
/**
* Waits indefinitely until the overall aggregate of matching events
satisfies the given criteria.
* Use {@link Timeout} on the overall test case to get a timeout.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
