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 aedca205992 feat: PartialSegmentFileMapper no longer lazy loads
(#19693)
aedca205992 is described below
commit aedca2059923a1906f8382b9984cd3dd4c2acfc2
Author: Clint Wylie <[email protected]>
AuthorDate: Thu Jul 16 22:06:17 2026 -0700
feat: PartialSegmentFileMapper no longer lazy loads (#19693)
---
.../druid/segment/PartialQueryableIndex.java | 156 ++++++++++++---
.../PartialQueryableIndexCursorFactory.java | 3 +-
.../segment/file/PartialSegmentFileMapperV10.java | 118 +++++++-----
.../PartialQueryableIndexCursorFactoryTest.java | 98 ++++++++++
.../druid/segment/PartialQueryableIndexTest.java | 122 ++++++++----
.../file/PartialSegmentFileMapperV10Test.java | 213 +++++++--------------
.../PartialSegmentBundleCacheEntryTest.java | 2 +
.../loading/PartialSegmentCacheBootstrapTest.java | 8 +-
8 files changed, 458 insertions(+), 262 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndex.java
b/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndex.java
index 4b95af14fb6..e36d5916d0e 100644
---
a/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndex.java
+++
b/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndex.java
@@ -70,10 +70,13 @@ import java.util.function.Function;
import java.util.stream.Collectors;
/**
- * A {@link QueryableIndex} that loads projection and base table columns on
demand from a
- * {@link PartialSegmentFileMapperV10}. Schema queries (column names, types,
intervals, metadata) are answered from
- * the {@link SegmentFileMetadata} alone without triggering any downloads.
Column data is only downloaded when a column
- * is accessed via {@link #getColumnHolder(String)} or {@link
#getProjection(CursorBuildSpec)}.
+ * A {@link QueryableIndex} over a {@link PartialSegmentFileMapperV10} whose
columns are made resident by explicit,
+ * planned downloads. Schema queries (column names, types, intervals,
metadata) are answered from the
+ * {@link SegmentFileMetadata} alone without touching column data. {@link
#planCursorPrefetch} resolves what a
+ * {@link CursorBuildSpec} needs and plans the coalesced range reads that make
it resident; only after those fetches
+ * execute may columns be read. Column deserialization stays lazy (memoized
per-column suppliers), but the underlying
+ * files must already be resident: mapping a non-resident file throws rather
than downloading, so a read outside a
+ * plan fails loudly instead of blocking a processing thread on deep-storage
I/O.
* <p>
* Projection matching uses only metadata ({@link
SegmentFileMetadata#getColumnDescriptors()} keys) to determine if
* a projection can satisfy a query, avoiding downloads of projection data
that won't be used.
@@ -102,9 +105,9 @@ public class PartialQueryableIndex implements QueryableIndex
// segment-internal file prefix for the base table projection, used to
translate column names to descriptor keys
private final String baseProjectionPrefix;
- // base table columns, built at construction time. each entry's supplier
defers both mapFile() and column
- // deserialization until the column is actually accessed, so queries only
trigger downloads for the specific
- // columns they use.
+ // base table columns, built at construction time. each entry's supplier
defers mapFile() and column
+ // deserialization until the column is actually accessed; the files must
already be resident by then (made so by a
+ // planned fetch), since mapFile() throws rather than downloads.
private final Map<String, Supplier<BaseColumnHolder>> baseColumns;
// projection columns, keyed by projection name. built on demand
(per-projection) when the projection is matched.
@@ -304,8 +307,9 @@ public class PartialQueryableIndex implements QueryableIndex
}
/**
- * Answers from metadata without triggering column downloads. The default
implementation in {@link QueryableIndex}
- * calls {@link #getColumnHolder(String)}, which would force a base table
load.
+ * Answers from metadata without touching column data. The default
implementation in {@link QueryableIndex}
+ * calls {@link #getColumnHolder(String)}, which would force base table
column deserialization (and throw if the
+ * column isn't resident).
*/
@Nullable
@Override
@@ -324,7 +328,7 @@ public class PartialQueryableIndex implements QueryableIndex
* resolve from the summary's typed clustering signature; data columns (and
{@code __time}) resolve from the first
* cluster group's {@link ColumnDescriptor} (all groups share the same
per-group shape). Mirrors the eager
* {@link SimpleQueryableIndex} clustered branch, but reads the descriptor
directly rather than routing through a
- * group sub-index's {@code getColumnHolder} (which would trigger a
download).
+ * group sub-index's {@code getColumnHolder} (which would deserialize the
column, throwing if it isn't resident).
*/
@Nullable
private ColumnCapabilities getClusteredColumnCapabilities(String column)
@@ -474,6 +478,7 @@ public class PartialQueryableIndex implements QueryableIndex
final Map<String, Supplier<BaseColumnHolder>> baseColumns =
clusterGroupColumnsByIndex.computeIfAbsent(
groupIndex,
i -> buildColumnSuppliers(
+
Projections.getClusterGroupBundleName(groupSpec.getClusteringValueIds()),
clusteredBaseSummary.getTimeColumnName(),
groupSpec.getNumRows(),
clusteredBaseSummary.getGroupColumnNames(),
@@ -549,7 +554,7 @@ public class PartialQueryableIndex implements QueryableIndex
Projections.getClusterGroupBundleName(group.getClusteringValueIds()),
groupIndex,
required,
- planClusterGroupPrefetch(group, required)
+ () -> planClusterGroupPrefetch(group, required)
)
);
}
@@ -562,25 +567,32 @@ public class PartialQueryableIndex implements
QueryableIndex
final QueryableIndex rowSelector = matched != null ?
matched.getRowSelector() : this;
final String bundleName = matched != null ? matched.getName() :
Projections.BASE_TABLE_PROJECTION_NAME;
final Set<String> required = requiredColumns(rowSelector, matched, spec);
- final List<PartialSegmentFileMapperV10.PlannedFetch> fetches =
- matched != null ? planProjectionPrefetch(matched.getName(), required)
: planBaseTablePrefetch(required);
final List<PrefetchBundle> bundles = new ArrayList<>(2);
- bundles.add(new PrefetchBundle(bundleName, rowSelector, required,
fetches));
+ bundles.add(
+ new PrefetchBundle(
+ bundleName,
+ rowSelector,
+ required,
+ matched != null
+ ? () -> planProjectionPrefetch(matched.getName(), required)
+ : () -> planBaseTablePrefetch(required)
+ )
+ );
if (matched != null) {
final Set<String> parents = projectionParentColumns(matched.getName(),
required);
if (!parents.isEmpty()) {
// Materializing a projection column also materializes its same-named
base column when one exists
// (buildColumnSuppliers reads through the parent, e.g. for dictionary
reuse), so those parents are part of
// this query's working set: plan their files and hold the base bundle
so the parent mmaps are
- // eviction-excluded for the holder's whole lifetime. When the
metadata predates recorded column file lists
- // the parent files can't be enumerated; the bundle is then hold-only
(no planned fetches) and the parents
- // download lazily during materialization, under the hold, on the
download executor.
+ // eviction-excluded for the holder's whole lifetime. Nothing
downloads lazily (mapFile throws on
+ // non-resident files), so when the metadata predates recorded column
file lists and the parent files can't
+ // be enumerated, planBaseTablePrefetch's fallback covers the whole
base bundle.
bundles.add(
new PrefetchBundle(
Projections.BASE_TABLE_PROJECTION_NAME,
this,
parents,
- metadata.getColumnFiles() == null ? List.of() :
planBaseTablePrefetch(parents)
+ () -> planBaseTablePrefetch(parents)
)
);
}
@@ -636,8 +648,8 @@ public class PartialQueryableIndex implements QueryableIndex
/**
* Build a map of column name to per-column supplier for the given
projection. Each supplier defers both
- * {@link SegmentFileMapper#mapFile} and {@link ColumnDescriptor#read} until
the column is actually accessed, so
- * queries only trigger downloads for the specific columns they use.
+ * {@link SegmentFileMapper#mapFile} and {@link ColumnDescriptor#read} until
the column is actually accessed, by
+ * which point the column's files must be resident (made so by a planned
fetch).
*/
private Map<String, Supplier<BaseColumnHolder>>
buildProjectionColumnSuppliers(
ProjectionMetadata projectionSpec,
@@ -645,6 +657,7 @@ public class PartialQueryableIndex implements QueryableIndex
)
{
return buildColumnSuppliers(
+ projectionSpec.getSchema().getName(),
projectionSpec.getSchema().getTimeColumnName(),
projectionSpec.getNumRows(),
projectionSpec.getSchema().getColumnNames(),
@@ -654,12 +667,15 @@ public class PartialQueryableIndex implements
QueryableIndex
}
/**
- * Shared builder for lazy per-column suppliers. Each supplier is memoized
and defers both
- * {@link SegmentFileMapper#mapFile} and {@link ColumnDescriptor#read} until
the column is actually accessed, so
- * queries only trigger downloads for the specific columns they use. {@code
fileNameFn} maps a logical column name to
- * its segment-internal (smoosh) file name in the right bundle namespace.
+ * Shared builder for lazy per-column suppliers. Each supplier memoizes with
eviction awareness (see
+ * {@link EvictionAwareColumnSupplier}) and defers both {@link
SegmentFileMapper#mapFile} and
+ * {@link ColumnDescriptor#read} until the column is actually accessed, by
which point the column's files must be
+ * resident (made so by a planned fetch, mapFile throws otherwise). {@code
bundleName} is the cache-layer bundle the
+ * columns live in (their eviction unit); {@code fileNameFn} maps a logical
column name to its segment-internal
+ * (smoosh) file name in that bundle's namespace.
*/
private Map<String, Supplier<BaseColumnHolder>> buildColumnSuppliers(
+ String bundleName,
@Nullable String timeColumnName,
int numRows,
List<String> columnNames,
@@ -678,7 +694,10 @@ public class PartialQueryableIndex implements
QueryableIndex
}
final String internedColumnName =
SmooshedFileMapper.STRING_INTERNER.intern(column);
- final Supplier<BaseColumnHolder> columnSupplier = Suppliers.memoize(()
-> {
+ // a column with a same-named base parent deserializes through the
parent's buffers (dictionary reuse), so its
+ // memoization must also invalidate when the parent's (__base) bundle is
evicted
+ final boolean readsParent = parentColumns.containsKey(column);
+ final Supplier<BaseColumnHolder> columnSupplier = new
EvictionAwareColumnSupplier(bundleName, readsParent, () -> {
try {
final ByteBuffer colBuffer = fileMapper.mapFile(smooshName);
final BaseColumnHolder parentColumn =
@@ -930,17 +949,94 @@ public class PartialQueryableIndex implements
QueryableIndex
}
/**
- * One bundle's worth of planned download work: the cache-layer {@code
bundleName} to mount, the row selector whose
- * {@code getColumnHolder} materializes the columns, the columns to
materialize once resident, and the planned
- * coalesced range reads that make them resident, each paired with the (main
or external) mapper that executes it
- * and each executable concurrently via {@link
PartialSegmentFileMapperV10.PlannedFetch#fetch()}.
+ * One bundle's worth of download work: the cache-layer {@code bundleName}
to mount, the row selector whose
+ * {@code getColumnHolder} materializes the columns, the columns to
materialize once resident, and a planner for
+ * the coalesced range reads that make them resident, each paired with the
(main or external) mapper that executes
+ * it and each executable concurrently via {@link
PartialSegmentFileMapperV10.PlannedFetch#fetch()}.
*/
record PrefetchBundle(
String bundleName,
QueryableIndex rowSelector,
Set<String> requiredColumns,
- List<PartialSegmentFileMapperV10.PlannedFetch> fetches
+ Supplier<List<PartialSegmentFileMapperV10.PlannedFetch>> fetchPlanner
)
{
+ /**
+ * Plan this bundle's range reads against CURRENT residency. Call only
once the bundle is protected from
+ * eviction (its cache-layer hold is acquired); planning drops
already-resident files from the runs, so a
+ * residency snapshot taken while the bundle is unheld can be invalidated
by a concurrent eviction
+ * ({@code evictContainer} clears the residency bitmap), leaving a plan
whose runs never restore the
+ * formerly-resident files. Files that become resident after a held plan
only shrink the work
+ * ({@code fetchRun} re-checks under its locks); they can never grow it.
+ */
+ List<PartialSegmentFileMapperV10.PlannedFetch> planFetches()
+ {
+ return fetchPlanner.get();
+ }
+ }
+
+ /**
+ * Memoizing column supplier that invalidates on bundle eviction. This index
(and its supplier maps) is cached
+ * across acquisitions, but the deserialized {@link BaseColumnHolder}s wrap
slices of container mmaps, and a bundle
+ * eviction between queries unmaps those containers; a plainly-memoized
holder would then be a dangling reference
+ * into freed memory that never consults {@link SegmentFileMapper#mapFile}
again (so the non-resident throw can't
+ * catch it). Each {@code get()} compares the {@link
PartialSegmentFileMapperV10#getBundleGeneration bundle
+ * generation} captured at deserialization time against the current one and
rebuilds on mismatch; a column that
+ * reads through a base-table parent also captures the parent bundle's
generation, since its holder references the
+ * parent's buffers (dictionary reuse) and must be rebuilt when only {@code
__base} was evicted.
+ * <p>
+ * The comparison is only meaningful under the bundle's eviction exclusion
(the caller's cache-layer hold), which
+ * is where the cursor factory materializes columns: under the hold the
generation cannot advance, so a holder
+ * returned here stays valid for as long as the hold is held.
+ */
+ private final class EvictionAwareColumnSupplier implements
Supplier<BaseColumnHolder>
+ {
+ private final String bundleName;
+ private final boolean readsParent;
+ private final Supplier<BaseColumnHolder> delegate;
+
+ @Nullable
+ private volatile Memoized memoized = null;
+
+ private EvictionAwareColumnSupplier(String bundleName, boolean
readsParent, Supplier<BaseColumnHolder> delegate)
+ {
+ this.bundleName = bundleName;
+ this.readsParent = readsParent;
+ this.delegate = delegate;
+ }
+
+ @Override
+ public BaseColumnHolder get()
+ {
+ final long generation = currentGeneration();
+ Memoized current = memoized;
+ if (current != null && current.generation == generation) {
+ return current.holder;
+ }
+ synchronized (this) {
+ // re-read both under the lock: a concurrent get() may have rebuilt
already
+ final long lockedGeneration = currentGeneration();
+ current = memoized;
+ if (current != null && current.generation == lockedGeneration) {
+ return current.holder;
+ }
+ final BaseColumnHolder holder = delegate.get();
+ memoized = new Memoized(holder, lockedGeneration);
+ return holder;
+ }
+ }
+
+ private long currentGeneration()
+ {
+ long generation = fileMapper.getBundleGeneration(bundleName);
+ if (readsParent) {
+ generation +=
fileMapper.getBundleGeneration(Projections.BASE_TABLE_PROJECTION_NAME);
+ }
+ return generation;
+ }
+
+ private record Memoized(BaseColumnHolder holder, long generation)
+ {
+ }
}
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndexCursorFactory.java
b/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndexCursorFactory.java
index c4b69bc7db3..31edef76062 100644
---
a/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndexCursorFactory.java
+++
b/processing/src/main/java/org/apache/druid/segment/PartialQueryableIndexCursorFactory.java
@@ -143,7 +143,8 @@ public class PartialQueryableIndexCursorFactory implements
CursorFactory
for (PartialQueryableIndex.PrefetchBundle bundle : bundles) {
final BundleHoldRelease holdRelease = new
BundleHoldRelease(bundleAcquirer.acquire(bundle.bundleName()));
holdReleases.add(holdRelease);
- for (PartialSegmentFileMapperV10.PlannedFetch fetch :
bundle.fetches()) {
+ // Plan the bundle's range reads only now, under its freshly-acquired
hold
+ for (PartialSegmentFileMapperV10.PlannedFetch fetch :
bundle.planFetches()) {
runDownloads.add(submitRunFetch(bundle.bundleName(), fetch,
holdRelease));
}
}
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 5a32513a8c6..df571ff2647 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
@@ -58,12 +58,20 @@ import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongArray;
import java.util.concurrent.locks.ReentrantLock;
/**
- * A {@link SegmentFileMapper} that downloads internal files on demand from
deep storage via a
- * {@link SegmentRangeReader}. This enables partial segment downloads where
only the files needed for a query are
- * fetched, rather than downloading the entire segment.
+ * A {@link SegmentFileMapper} that fetches internal files from deep storage
via a {@link SegmentRangeReader},
+ * enabling partial segment downloads where only the files needed for a query
are fetched rather than the entire
+ * segment.
+ * <p>
+ * <b>Downloads are always explicit.</b> Callers decide what to load and when:
plan coalesced range reads with
+ * {@link #planFetch}/{@link #planParallelFetch}/{@link
#planParallelFetchBundle} and execute them via
+ * {@link #fetchFiles}/{@link FetchRun}/{@link PlannedFetch}, or force whole
units resident with
+ * {@link #ensureBundleDownloaded}/{@link #ensureAllDownloaded}. {@link
#mapFile} never downloads: it only slices
+ * already-resident files and throws for a file that hasn't been fetched, so a
caller that forgot to plan fails
+ * loudly instead of triggering a synchronous deep-storage read from whatever
thread touched the column.
* <p>
* Locally, this mapper mirrors the original V10 container structure: each
container from the segment file is
* represented as a local sparse file at its original size, and only the byte
ranges for downloaded internal files are
@@ -83,7 +91,7 @@ import java.util.concurrent.locks.ReentrantLock;
* different file in the segment's storage location.
* <p>
* Thread-safe for concurrent access from multiple queries. Per-file locks
prevent duplicate downloads of the same
- * internal file.
+ * internal file across concurrently executing runs.
*
* @see SegmentFileMapperV10
* @see SegmentRangeReader
@@ -277,6 +285,8 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
private final MappedByteBuffer[] containers;
private final File[] containerFiles;
private final ReentrantLock[] containerLocks;
+ // per-container eviction generation; see getBundleGeneration
+ private final AtomicLongArray containerGenerations;
// bundle name -> indices (into metadata.getContainers()) of this single
mapper's containers in that bundle.
// Computed once at construction from the immutable container metadata.
Single-mapper scope only: stitching bundles
@@ -338,6 +348,7 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
this.containers = new MappedByteBuffer[numContainers];
this.containerFiles = new File[numContainers];
this.containerLocks = new ReentrantLock[numContainers];
+ this.containerGenerations = new AtomicLongArray(numContainers);
final Map<String, List<Integer>> bundleIndices = new HashMap<>();
for (int i = 0; i < numContainers; i++) {
this.containerLocks[i] = new ReentrantLock();
@@ -458,7 +469,16 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
return null;
}
- ensureFileDownloaded(name, fileMetadata);
+ // Mapping never downloads, every caller is expected to have made the file
resident through an explicit plan before
+ // reading it
+ if (!downloadedFiles.contains(name)) {
+ throw DruidException.defensive(
+ "Internal file[%s] of segment file[%s] is not resident; downloads
must be planned and fetched explicitly "
+ + "before the file can be mapped",
+ name,
+ targetFilename
+ );
+ }
// slice from the container mmap
final MappedByteBuffer container = containers[fileMetadata.getContainer()];
@@ -755,12 +775,11 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
/**
* Execute one planned run: take every member's per-file lock in offset
order (the canonical acquisition order
- * shared by all {@link #fetchFiles} callers, so overlapping concurrent runs
can't deadlock, and
- * {@link #ensureFileDownloaded} holds at most one of these locks so it
can't participate in a cycle either), trim
- * files that became resident since planning off the run's edges, then
stream the remaining span in a single range
- * read and mark each covered file downloaded after the bytes are on disk
(bytes-before-bits, preserving the bitmap
- * corruption invariant). Interior files that became resident mid-plan are
re-fetched with byte-identical data;
- * {@link #markDownloaded}'s add-gate keeps the accounting straight.
+ * shared by all {@link #fetchFiles} callers; this is the only code path
that takes file locks, so overlapping
+ * concurrent runs can't deadlock), trim files that became resident since
planning off the run's edges, then stream
+ * the remaining span in a single range read and mark each covered file
downloaded after the bytes are on disk
+ * (bytes-before-bits, preserving the bitmap corruption invariant). Interior
files that became resident mid-plan are
+ * re-fetched with byte-identical data; {@link #markDownloaded}'s add-gate
keeps the accounting straight.
* <p>
* Runs from one {@link #planFetch} cover disjoint files, so callers may
execute them concurrently (each is one
* deep-storage request); the caller must hold the same eviction-exclusion
{@link #mapFile} requires for the
@@ -918,43 +937,11 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
return headerSize + container.getStartOffset() +
fileMetadata.getStartOffset();
}
- private void ensureFileDownloaded(String name, SegmentInternalFileMetadata
fileMetadata) throws IOException
- {
- // already downloaded, nothing to do
- if (downloadedFiles.contains(name)) {
- return;
- }
-
- final ReentrantLock lock = fileLocks.computeIfAbsent(name, k -> new
ReentrantLock());
- lock.lock();
- try {
- checkClosed();
-
- if (downloadedFiles.contains(name)) {
- return;
- }
-
- ensureContainerInitialized(fileMetadata.getContainer());
- streamRangeIntoContainer(
- fileMetadata.getContainer(),
- computeAbsoluteOffset(fileMetadata),
- fileMetadata.getStartOffset(),
- fileMetadata.getSize(),
- StringUtils.format("file[%s]", name)
- );
- markDownloaded(name, fileMetadata.getSize());
- }
- finally {
- lock.unlock();
- fileLocks.remove(name, lock);
- }
- }
-
/**
* Public entry point for cache-layer code that wants to ensure a container
is materialized before any data is
* downloaded into it (e.g. when a per-bundle cache entry is mounted, the
entry pre-allocates its container files
- * so that subsequent {@link #mapFile} calls have somewhere to write into
and the cache layer can charge the
- * reservation up front).
+ * so that subsequent fetches have somewhere to write into and the cache
layer can charge the reservation up
+ * front).
*/
public void initializeContainer(int containerIndex) throws IOException
{
@@ -968,9 +955,9 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
* in this container.
* <p>
* Used by per-bundle cache entries on unmount/eviction to release the disk
and memory footprint of one bundle
- * without affecting other bundles sharing the same {@link
PartialSegmentFileMapperV10}. After eviction, subsequent
- * {@link #mapFile} calls for files in this container will re-trigger
downloads via {@link #initializeContainer}
- * and the bitmap will be repopulated incrementally.
+ * without affecting other bundles sharing the same {@link
PartialSegmentFileMapperV10}. After eviction, the
+ * container's files are non-resident again: {@link #mapFile} throws for
them until a subsequent fetch re-downloads
+ * them (re-initializing the container and repopulating the bitmap
incrementally).
* <p>
* <b>Concurrency contract.</b> The caller is responsible for ensuring no
concurrent {@link #mapFile} (or
* {@link #fetchFiles}/{@link #fetchRun}) call is in flight for any file in
this container. This is enforced one layer up
@@ -1028,6 +1015,41 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
}
clearBitmapBit(fileName);
}
+
+ // last: readers that observe the bumped generation must also observe the
cleared residency above
+ containerGenerations.incrementAndGet(containerIndex);
+ }
+
+ /**
+ * Monotonic eviction generation for {@code bundleName}: increments every
time one of the bundle's containers is
+ * evicted ({@link #evictContainer}), summed across this mapper and every
attached external mapper (a bundle's
+ * containers can span them). Readers that cache objects deserialized from
container mmaps (e.g.
+ * {@code PartialQueryableIndex}'s memoized column holders) compare
generations to detect that a cached object may
+ * reference since-unmapped memory and must be rebuilt. A comparison is only
meaningful while the caller holds the
+ * bundle's eviction exclusion (its cache-layer hold): under the hold the
generation cannot advance, so a matching
+ * value stays valid for as long as the hold is held.
+ * <p>
+ * A bundle name with no containers in a mapper conservatively reflects
every container of that mapper (the name
+ * may resolve to a catch-all bundle at the cache layer, e.g. untagged
legacy segments, so any eviction must
+ * invalidate).
+ */
+ public long getBundleGeneration(String bundleName)
+ {
+ long generation = 0;
+ final List<Integer> indices = bundleToContainerIndices.get(bundleName);
+ if (indices == null) {
+ for (int i = 0; i < containerGenerations.length(); i++) {
+ generation += containerGenerations.get(i);
+ }
+ } else {
+ for (int i : indices) {
+ generation += containerGenerations.get(i);
+ }
+ }
+ for (PartialSegmentFileMapperV10 external : externalMappers.values()) {
+ generation += external.getBundleGeneration(bundleName);
+ }
+ return generation;
}
private void clearBitmapBit(String 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 6e539530b8e..7b940fd7c30 100644
---
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexCursorFactoryTest.java
@@ -399,6 +399,104 @@ class PartialQueryableIndexCursorFactoryTest extends
PartialQueryableIndexCursor
}
}
+ @Test
+ void testResidentBundleEvictedBeforeHoldAcquisitionIsRefetched() throws
IOException
+ {
+ // A resident-but-unheld bundle can be evicted between
makeCursorHolderAsync's plan resolution and the bundle
+ // hold acquisition; evictContainer clears the residency bitmap, so a
residency snapshot taken before the hold
+ // would omit the formerly-resident files and never restore them, making
materialization throw on their
+ // non-resident mapFile. Fetch planning is deferred until the bundle's
hold is acquired (PrefetchBundle
+ // .planFetches), so the plan sees post-eviction residency and re-fetches.
Model the worst-case timing by
+ // evicting inside acquire() itself, immediately before the hold exists.
+ final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader, "evict_window")) {
+ final PartialQueryableIndex index = opened.index();
+ final PartialSegmentFileMapperV10 mapper = opened.mapper();
+
+ // the query's columns are resident before the query arrives
+ mapper.fetchFiles(Set.of("__base/__time", "__base/dim1"));
+
Assertions.assertTrue(mapper.getDownloadedFiles().containsAll(Set.of("__base/__time",
"__base/dim1")));
+
+ final ListeningExecutorService exec = directExec();
+ final PartialBundleAcquirer evictingAcquirer = new
PartialBundleAcquirer()
+ {
+ @Override
+ public Closeable acquire(String bundleName)
+ {
+ // the eviction lands in the unheld window: every container's
residency is wiped just before the hold
+ for (int i = 0; i <
mapper.getSegmentFileMetadata().getContainers().size(); i++) {
+ mapper.evictContainer(i);
+ }
+ return () -> {};
+ }
+
+ @Override
+ public <T> AsyncResource<T> submitDownload(Callable<T> task)
+ {
+ return AsyncResources.fromFutureUnmanaged(exec.submit(task));
+ }
+ };
+ final PartialQueryableIndexCursorFactory factory = new
PartialQueryableIndexCursorFactory(
+ index,
+ QueryableIndexTimeBoundaryInspector.create(index),
+ evictingAcquirer
+ );
+
+ final CursorBuildSpec spec =
CursorBuildSpec.builder().setPhysicalColumns(Set.of("dim1")).build();
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(spec);
+ CursorHolder holder = asyncHolder.release()) {
+ Assertions.assertNotNull(holder);
+ Assertions.assertTrue(
+ mapper.getDownloadedFiles().containsAll(Set.of("__base/__time",
"__base/dim1")),
+ "evicted files must be re-fetched by the post-hold plan; got: " +
mapper.getDownloadedFiles()
+ );
+ }
+ }
+ }
+
+ @Test
+ void testMemoizedColumnsSurviveBundleEvictionBetweenQueries() throws
IOException
+ {
+ // Query 1 materializes (and memoizes) its columns; with all holds
released, the bundle evicts; query 2
+ // re-acquires and re-fetches. The holders memoized by query 1 wrap the
unmapped old container mmap — serving
+ // them to query 2 would read freed memory without ever consulting mapFile
(so the non-resident throw can't
+ // catch it). The eviction-aware suppliers detect the bundle-generation
change and rebuild against the fresh
+ // container, so query 2 reads correct values.
+ final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
+ try (IndexAndMapper opened = openIndex(rangeReader,
"evict_between_queries")) {
+ final PartialQueryableIndex index = opened.index();
+ final PartialSegmentFileMapperV10 mapper = opened.mapper();
+ final PartialQueryableIndexCursorFactory factory = new
PartialQueryableIndexCursorFactory(
+ index,
+ QueryableIndexTimeBoundaryInspector.create(index),
+ noOpAcquirer(directExec())
+ );
+ final CursorBuildSpec spec =
CursorBuildSpec.builder().setPhysicalColumns(Set.of("dim1")).build();
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(spec);
+ CursorHolder holder = asyncHolder.release()) {
+ Assertions.assertNotNull(holder.asCursor());
+ }
+
+ // every hold is released; the bundle evicts
+ for (int i = 0; i <
mapper.getSegmentFileMetadata().getContainers().size(); i++) {
+ mapper.evictContainer(i);
+ }
+
+ try (AsyncCursorHolder asyncHolder = factory.makeCursorHolderAsync(spec);
+ CursorHolder holder = asyncHolder.release()) {
+ final Cursor cursor = holder.asCursor();
+ final ColumnValueSelector<?> dim1 =
cursor.getColumnSelectorFactory().makeColumnValueSelector("dim1");
+ final List<Object> values = new ArrayList<>();
+ while (!cursor.isDone()) {
+ values.add(dim1.getObject());
+ cursor.advance();
+ }
+ Assertions.assertEquals(List.of("a", "a", "b", "b"), values);
+ }
+ }
+ }
+
@Test
void testMatchedProjectionDownloadsOnlyRequestedColumns() throws IOException
{
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 64fd5b0b244..b2953cfec24 100644
---
a/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/PartialQueryableIndexTest.java
@@ -219,7 +219,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
}
@Test
- void testGetColumnHolderTriggersBaseTableLoad() throws IOException
+ void testGetColumnHolderRequiresResidentFiles() throws IOException
{
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("colholder");
@@ -233,13 +233,14 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
COLUMN_CONFIG
);
- // no downloads yet
+ // no downloads yet, and column access never triggers one: an unfetched
column fails loudly
Assertions.assertEquals(0, rangeReader.getReadCount());
+ Assertions.assertThrows(DruidException.class, () ->
index.getColumnHolder(ColumnHolder.TIME_COLUMN_NAME));
+ Assertions.assertEquals(0, rangeReader.getReadCount(), "the failed
access must not have downloaded anything");
- // accessing a column holder should trigger downloads
+ // after an explicit fetch, deserialization finds the bytes resident
+ mapper.fetchFiles(Set.of("__base/__time", "__base/dim1"));
Assertions.assertNotNull(index.getColumnHolder(ColumnHolder.TIME_COLUMN_NAME));
- Assertions.assertTrue(rangeReader.getReadCount() > 0);
-
Assertions.assertNotNull(index.getColumnHolder("dim1"));
Assertions.assertNull(index.getColumnHolder("nonexistent"));
@@ -249,7 +250,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
}
@Test
- void testGetProjectionMatchesFromMetadataAndLoadsLazily() throws IOException
+ void testGetProjectionMatchesFromMetadataAndReadsAfterPlannedFetch() throws
IOException
{
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("projection");
@@ -276,40 +277,43 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final QueryableIndex projIndex = projection.getRowSelector();
Assertions.assertNotNull(projIndex);
Assertions.assertEquals(0, rangeReader.getReadCount(), "this should not
download files either");
- // actually accessing a column on the projection triggers the column's
download
-
Assertions.assertNotNull(projIndex.getColumnHolder(ColumnHolder.TIME_COLUMN_NAME));
- Assertions.assertTrue(rangeReader.getReadCount() > 0, "accessing a
projection column should download");
+
+ // column access never downloads: an unfetched projection column fails
loudly
+ Assertions.assertThrows(DruidException.class, () ->
projIndex.getColumnHolder(ColumnHolder.TIME_COLUMN_NAME));
+ Assertions.assertEquals(0, rangeReader.getReadCount(), "the failed
access must not have downloaded anything");
+
+ // execute the plan for this spec: the projection bundle plus the __base
parent bundle
+ final PartialQueryableIndex.CursorPrefetchPlan plan =
index.planCursorPrefetch(matchingSpec);
+ for (PartialQueryableIndex.PrefetchBundle bundle : plan.bundles()) {
+ for (PartialSegmentFileMapperV10.PlannedFetch fetch :
bundle.planFetches()) {
+ fetch.fetch();
+ }
+ }
+ final int readsAfterFetch = rangeReader.getReadCount();
+ Assertions.assertTrue(readsAfterFetch > 0, "executing the plan
downloads");
Assertions.assertEquals(Set.of(IndexIO.V10_FILE_NAME),
rangeReader.getReadFilenames());
- // downloaded files are scoped to the matched projection's namespace,
not the base table (if no shared parts)
+ // now reads work, including dim1 which reads through its base parent,
with no further downloads
+
Assertions.assertNotNull(projIndex.getColumnHolder(ColumnHolder.TIME_COLUMN_NAME));
+ Assertions.assertNotNull(projIndex.getColumnHolder("dim1"));
+ Assertions.assertEquals(readsAfterFetch, rangeReader.getReadCount(),
"reads must find everything resident");
+
+ // the plan covered the projection's namespace plus the dim1 parent from
__base, and nothing else from base
final Set<String> downloaded = mapper.getDownloadedFiles();
Assertions.assertTrue(
downloaded.stream().anyMatch(name ->
name.startsWith("dim1_hourly_metric1_sum/")),
"expected at least one file from the matched projection's namespace,
got " + downloaded
);
- Assertions.assertTrue(
- downloaded.stream().noneMatch(name -> name.startsWith("__base/")),
- "no base table files should be downloaded when only the projection
was accessed, got " + downloaded
- );
-
- // fetching a projection column which has a base table parent does
download base table stuff
- Assertions.assertNotNull(projIndex.getColumnHolder("dim1"));
- final Set<String> downloadedAfterDim1 = mapper.getDownloadedFiles();
- Assertions.assertTrue(
- downloadedAfterDim1.stream().anyMatch(name ->
name.startsWith("dim1_hourly_metric1_sum/")),
- "expected at least one file from the matched projection's namespace,
got " + downloadedAfterDim1
- );
- Assertions.assertTrue(
- downloadedAfterDim1.stream().anyMatch(name ->
name.startsWith("__base/")),
- "base table files should be downloaded when a projection column
shares data with a base table parent, got " + downloadedAfterDim1
- );
+ Assertions.assertTrue(downloaded.contains("__base/dim1"), "parent must
be planned, got " + downloaded);
+ Assertions.assertFalse(downloaded.contains("__base/metric1"), "got " +
downloaded);
}
}
@Test
- void testPerColumnLaziness() throws IOException
+ void testPerColumnFetchIndependence() throws IOException
{
- // verify that accessing one column of a projection doesn't download other
columns
+ // verify that fetching + reading one column leaves other columns
untouched, and that reading an unfetched
+ // column fails instead of downloading
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
final File cacheDir = newCacheDir("per_col");
@@ -322,26 +326,29 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
rangeReader.resetCount();
- // access one base table column
+ // fetch + access one base table column
+ mapper.fetchFiles(Set.of("__base/dim1"));
Assertions.assertNotNull(index.getColumnHolder("dim1"));
final int countAfterDim1 = rangeReader.getReadCount();
- Assertions.assertTrue(countAfterDim1 > 0, "accessing dim1 should trigger
downloads");
+ Assertions.assertTrue(countAfterDim1 > 0, "fetching dim1 should trigger
downloads");
- // dim1's smoosh entry is downloaded; metric1's is not
+ // dim1's smoosh entry is downloaded; metric1's is not, and accessing it
fails rather than downloading
final Set<String> filesAfterDim1 = mapper.getDownloadedFiles();
Assertions.assertTrue(filesAfterDim1.contains("__base/dim1"), "expected
__base/dim1 in " + filesAfterDim1);
- Assertions.assertFalse(filesAfterDim1.contains("__base/metric1"),
"metric1 should not be downloaded yet");
+ Assertions.assertFalse(filesAfterDim1.contains("__base/metric1"),
"metric1 should not be downloaded");
+ Assertions.assertThrows(DruidException.class, () ->
index.getColumnHolder("metric1"));
// access the same column again should not trigger more downloads
Assertions.assertNotNull(index.getColumnHolder("dim1"));
Assertions.assertEquals(countAfterDim1, rangeReader.getReadCount(),
"re-access should be cached");
Assertions.assertEquals(filesAfterDim1, mapper.getDownloadedFiles(),
"re-access should not download new files");
- // access a different column should trigger additional downloads for its
files
+ // after fetching metric1's file, the previously-failing access works
+ mapper.fetchFiles(Set.of("__base/metric1"));
Assertions.assertNotNull(index.getColumnHolder("metric1"));
Assertions.assertTrue(
rangeReader.getReadCount() > countAfterDim1,
- "accessing metric1 should trigger additional downloads"
+ "fetching metric1 should trigger additional downloads"
);
// metric1's smoosh entry is now also downloaded
@@ -443,11 +450,12 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
Assertions.assertEquals(0, rangeReader.getReadCount(), "planning must
not download");
Assertions.assertEquals(1, plan.bundles().size());
final PartialQueryableIndex.PrefetchBundle bundle =
plan.bundles().get(0);
- Assertions.assertEquals(2, bundle.fetches().size(), "one run per mapper:
" + bundle.fetches());
+ final List<PartialSegmentFileMapperV10.PlannedFetch> fetches =
bundle.planFetches();
+ Assertions.assertEquals(2, fetches.size(), "one run per mapper: " +
fetches);
boolean sawMain = false;
boolean sawExternal = false;
- for (PartialSegmentFileMapperV10.PlannedFetch fetch : bundle.fetches()) {
+ for (PartialSegmentFileMapperV10.PlannedFetch fetch : fetches) {
if (fetch.mapper() == mapper) {
sawMain = true;
// __time and dimA are adjacent in the main container; dimB is an
unrequested trailing file, never fetched
@@ -501,7 +509,7 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
final PartialQueryableIndex.CursorPrefetchPlan plan =
index.planCursorPrefetch(
CursorBuildSpec.builder().setPhysicalColumns(Set.of("dimA")).build()
);
- for (PartialSegmentFileMapperV10.PlannedFetch fetch :
plan.bundles().get(0).fetches()) {
+ for (PartialSegmentFileMapperV10.PlannedFetch fetch :
plan.bundles().get(0).planFetches()) {
fetch.fetch();
}
@@ -519,6 +527,41 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
}
}
+ @Test
+ void testColumnHolderRebuildsAfterBundleEviction() throws IOException
+ {
+ // The index (and its memoized column suppliers) outlives bundle
evictions: a memoized holder wraps slices of
+ // the container mmap that eviction unmaps, and once memoized it never
consults mapFile again, so the
+ // non-resident throw can't catch the staleness. The eviction-aware
supplier compares bundle generations and
+ // rebuilds instead of returning the dangling holder.
+ final CountingRangeReader rangeReader = new
CountingRangeReader(segmentDir);
+ final File cacheDir = newCacheDir("evict_rebuild");
+
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
+ final PartialQueryableIndex index = new PartialQueryableIndex(
+ mapper.getSegmentFileMetadata(),
+ mapper,
+ COLUMN_CONFIG
+ );
+
+ mapper.fetchFiles(Set.of("__base/dim1"));
+ final ColumnHolder before = index.getColumnHolder("dim1");
+ Assertions.assertNotNull(before);
+ Assertions.assertSame(before, index.getColumnHolder("dim1"), "no
eviction: the holder is memoized");
+
+ // evict the base bundle's containers, then make the files resident
again (as a re-acquired query would)
+ for (int containerIndex :
mapper.getContainerIndicesForBundle(Projections.BASE_TABLE_PROJECTION_NAME)) {
+ mapper.evictContainer(containerIndex);
+ }
+ mapper.fetchFiles(Set.of("__base/dim1"));
+
+ final ColumnHolder after = index.getColumnHolder("dim1");
+ Assertions.assertNotNull(after);
+ Assertions.assertNotSame(before, after, "an evicted bundle's holder must
be rebuilt, not returned stale");
+ Assertions.assertSame(after, index.getColumnHolder("dim1"), "the rebuilt
holder memoizes again");
+ }
+ }
+
@Test
void testPlanCursorPrefetchIncludesProjectionParentBundle() throws
IOException
{
@@ -544,14 +587,15 @@ class PartialQueryableIndexTest extends
InitializedNullHandlingTest
Assertions.assertEquals(Projections.BASE_TABLE_PROJECTION_NAME,
parentBundle.bundleName());
Assertions.assertEquals(Set.of("dim1"), parentBundle.requiredColumns());
+ final List<PartialSegmentFileMapperV10.PlannedFetch> parentFetches =
parentBundle.planFetches();
final Set<String> parentFiles = new TreeSet<>();
- for (PartialSegmentFileMapperV10.PlannedFetch fetch :
parentBundle.fetches()) {
+ for (PartialSegmentFileMapperV10.PlannedFetch fetch : parentFetches) {
parentFiles.addAll(fetch.run().files());
}
Assertions.assertEquals(Set.of("__base/dim1"), parentFiles);
// executing the parent fetches makes exactly the parent's file
resident, nothing else from the base bundle
- for (PartialSegmentFileMapperV10.PlannedFetch fetch :
parentBundle.fetches()) {
+ for (PartialSegmentFileMapperV10.PlannedFetch fetch : parentFetches) {
fetch.fetch();
}
Assertions.assertEquals(Set.of("__base/dim1"),
mapper.getDownloadedFiles());
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 561d84dac7d..21e730ca954 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
@@ -22,6 +22,7 @@ package org.apache.druid.segment.file;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.io.Files;
import com.google.common.primitives.Ints;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
@@ -63,10 +64,10 @@ class PartialSegmentFileMapperV10Test
File tempDir;
@Test
- void testMapFileDownloadsOnDemand() throws IOException
+ void testMapFileThrowsUntilFetchedThenSlices() throws IOException
{
final File segmentFile = buildTestSegment(20, CompressionStrategy.NONE);
- final File cacheDir = newCacheDir("demand");
+ final File cacheDir = newCacheDir("explicit");
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentFile.getParentFile());
@@ -74,16 +75,21 @@ class PartialSegmentFileMapperV10Test
// reset count after create fetched metadata
rangeReader.resetCount();
- // access a single file - should trigger exactly one download
+ // mapFile never downloads: a known-but-unfetched file fails loudly (a
silent slice would be all zeros)
+ Assertions.assertThrows(DruidException.class, () -> mapper.mapFile("5"));
+ Assertions.assertEquals(0, rangeReader.getReadCount(), "the failed map
must not have downloaded anything");
+
+ mapper.fetchFiles(Set.of("5"));
+ Assertions.assertEquals(1, rangeReader.getReadCount());
+ Assertions.assertEquals(4, mapper.getDownloadedBytes());
+
ByteBuffer buf = mapper.mapFile("5");
Assertions.assertNotNull(buf);
Assertions.assertEquals(0, buf.position());
Assertions.assertEquals(4, buf.remaining());
Assertions.assertEquals(5, buf.getInt());
- Assertions.assertEquals(1, rangeReader.getReadCount());
- Assertions.assertEquals(4, mapper.getDownloadedBytes());
- // access the same file again - should NOT trigger another download
+ // mapping again slices the same resident bytes without any further reads
ByteBuffer buf2 = mapper.mapFile("5");
Assertions.assertNotNull(buf2);
Assertions.assertEquals(5, buf2.getInt());
@@ -100,6 +106,7 @@ class PartialSegmentFileMapperV10Test
final CountingRangeReader rangeReader = new
CountingRangeReader(segmentFile.getParentFile());
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
+ mapper.ensureAllDownloaded();
for (int i = 0; i < 20; ++i) {
ByteBuffer buf = mapper.mapFile(String.valueOf(i));
Assertions.assertNotNull(buf);
@@ -258,7 +265,7 @@ class PartialSegmentFileMapperV10Test
// large tolerance so only residency, not gaps, can split the span
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir, 1024)) {
- mapper.mapFile("4");
+ mapper.fetchFiles(Set.of("4"));
rangeReader.resetCount();
mapper.fetchFiles(Set.of("2", "3", "4", "5", "6", "7"));
@@ -340,12 +347,12 @@ class PartialSegmentFileMapperV10Test
}
@Test
- void testConcurrentMapFileVsFetchFiles() throws Exception
+ void testMapFileDuringInFlightFetch() throws Exception
{
final File segmentFile = buildTestSegment(20, CompressionStrategy.NONE);
final File cacheDir = newCacheDir("concurrent_mixed");
- // Gate the first fetch mid-flight so the concurrent mapFile calls
genuinely contend with an in-progress run.
+ // Gate the second fetch mid-flight so concurrent mapFile calls genuinely
observe an in-progress run.
// Discriminate by thread identity, not offset: createMapper's header
fetch also reads at offset > 0 (the
// metadata bytes after the fixed preamble) on this thread, and gating
that would self-deadlock the test.
final Thread mainThread = Thread.currentThread();
@@ -371,8 +378,10 @@ class PartialSegmentFileMapperV10Test
};
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir, 1024)) {
+ // make a file outside the run resident up front (on this thread, which
the gate ignores)
+ mapper.fetchFiles(Set.of("10"));
rangeReader.resetCount();
- final ExecutorService exec = Execs.multiThreaded(3, "mixed-test-%d");
+ final ExecutorService exec = Execs.multiThreaded(2, "mixed-test-%d");
try {
final Future<Void> fetchFuture = exec.submit(() -> {
mapper.fetchFiles(Set.of("3", "4", "5"));
@@ -380,23 +389,21 @@ class PartialSegmentFileMapperV10Test
});
Assertions.assertTrue(fetchStarted.await(10, TimeUnit.SECONDS));
- // a run member: must block on the per-file lock and find the file
resident after the run finishes
- final Future<ByteBuffer> memberFuture = exec.submit(() ->
mapper.mapFile("4"));
- // a non-member: proceeds in parallel while the run is gated
- final Future<ByteBuffer> outsideFuture = exec.submit(() ->
mapper.mapFile("10"));
-
- final ByteBuffer outside = outsideFuture.get(10, TimeUnit.SECONDS);
+ // reads never wait on downloads: a resident file slices immediately
while the run is parked, and a run
+ // member that isn't resident yet fails fast instead of blocking for it
+ final ByteBuffer outside = mapper.mapFile("10");
Assertions.assertNotNull(outside);
Assertions.assertEquals(10, outside.getInt());
+ Assertions.assertThrows(DruidException.class, () ->
mapper.mapFile("4"));
releaseFetch.countDown();
fetchFuture.get(10, TimeUnit.SECONDS);
- final ByteBuffer member = memberFuture.get(10, TimeUnit.SECONDS);
+ final ByteBuffer member = mapper.mapFile("4");
Assertions.assertNotNull(member);
Assertions.assertEquals(4, member.getInt());
- // one read for the run, one for the non-member; the member never
re-downloaded
- Assertions.assertEquals(2, rangeReader.getReadCount());
+ // one read for the run; the throwing mapFile and the resident reads
downloaded nothing
+ Assertions.assertEquals(1, rangeReader.getReadCount());
Assertions.assertEquals(16, mapper.getDownloadedBytes());
}
finally {
@@ -678,8 +685,8 @@ class PartialSegmentFileMapperV10Test
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
// download just 2 files
- mapper.mapFile("3");
- mapper.mapFile("7");
+ mapper.fetchFiles(Set.of("3"));
+ mapper.fetchFiles(Set.of("7"));
// verify local container files exist (all files share one container in
this test)
File containerFile = new File(cacheDir, IndexIO.V10_FILE_NAME +
".container.00000");
@@ -694,53 +701,6 @@ class PartialSegmentFileMapperV10Test
}
}
- @Test
- void testConcurrentMapFile() throws Exception
- {
- final File segmentFile = buildTestSegment(20, CompressionStrategy.NONE);
- final File cacheDir = newCacheDir("concurrent");
-
- final CountingRangeReader rangeReader = new
CountingRangeReader(segmentFile.getParentFile());
-
- try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
- rangeReader.resetCount();
-
- final int numThreads = 8;
- final ExecutorService exec = Execs.multiThreaded(numThreads,
"lazy-test-%d");
- try {
- final CountDownLatch startLatch = new CountDownLatch(1);
- final List<Future<Void>> futures = new ArrayList<>();
-
- // all threads try to access all 20 files concurrently
- for (int t = 0; t < numThreads; t++) {
- futures.add(exec.submit(() -> {
- startLatch.await();
- for (int i = 0; i < 20; ++i) {
- ByteBuffer buf = mapper.mapFile(String.valueOf(i));
- Assertions.assertNotNull(buf);
- Assertions.assertEquals(4, buf.remaining());
- Assertions.assertEquals(i, buf.getInt());
- }
- return null;
- }));
- }
-
- startLatch.countDown();
-
- for (Future<Void> f : futures) {
- f.get();
- }
-
- // each file should have been downloaded exactly once despite
concurrent access
- Assertions.assertEquals(20, rangeReader.getReadCount());
- Assertions.assertEquals(80, mapper.getDownloadedBytes());
- }
- finally {
- exec.shutdownNow();
- }
- }
- }
-
@Test
void testProjectionStyleFileNames() throws IOException
{
@@ -760,6 +720,7 @@ class PartialSegmentFileMapperV10Test
final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(baseDir);
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
+ mapper.ensureAllDownloaded();
for (int i = 0; i < 5; ++i) {
ByteBuffer buf =
mapper.mapFile(StringUtils.format("myProjection/col_%d", i));
Assertions.assertNotNull(buf);
@@ -768,70 +729,6 @@ class PartialSegmentFileMapperV10Test
}
}
- @Test
- void testMatchesEagerMapper() throws IOException
- {
- // verify that the lazy mapper produces identical ByteBuffer contents as
the eager mapper
- final File segmentFile = buildTestSegment(20, CompressionStrategy.NONE);
- final File cacheDir = newCacheDir("match_eager");
-
- final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(segmentFile.getParentFile());
- try (SegmentFileMapperV10 eager = SegmentFileMapperV10.create(segmentFile,
JSON_MAPPER);
- PartialSegmentFileMapperV10 lazy = createMapper(rangeReader, cacheDir)
- ) {
- Assertions.assertEquals(eager.getInternalFilenames(),
lazy.getInternalFilenames());
- for (String name : eager.getInternalFilenames()) {
- ByteBuffer eagerBuf = eager.mapFile(name);
- ByteBuffer lazyBuf = lazy.mapFile(name);
- Assertions.assertNotNull(eagerBuf);
- Assertions.assertNotNull(lazyBuf);
- Assertions.assertEquals(eagerBuf, lazyBuf);
- }
- }
-
- }
-
- @Test
- void testExternalFiles() throws IOException
- {
- final String externalName = "external.segment";
- final File baseDir = new File(tempDir, "base_" +
ThreadLocalRandom.current().nextInt());
- FileUtils.mkdirp(baseDir);
-
- try (SegmentFileBuilderV10 builder =
SegmentFileBuilderV10.create(JSON_MAPPER, baseDir)) {
- for (int i = 0; i < 10; ++i) {
- File tmpFile = new File(tempDir, StringUtils.format("main-%s.bin", i));
- Files.write(Ints.toByteArray(i), tmpFile);
- builder.add(StringUtils.format("%d", i), tmpFile);
- }
- SegmentFileBuilder external = builder.getExternalBuilder(externalName);
- for (int i = 10; i < 20; ++i) {
- File tmpFile = new File(tempDir, StringUtils.format("ext-%s.bin", i));
- Files.write(Ints.toByteArray(i), tmpFile);
- external.add(StringUtils.format("%d", i), tmpFile);
- }
- }
-
- final File cacheDir = newCacheDir("ext");
- final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(baseDir);
-
- 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));
- Assertions.assertNotNull(buf);
- Assertions.assertEquals(i, buf.getInt());
- }
-
- // verify external file internal files via mapExternalFile
- for (int i = 10; i < 20; ++i) {
- ByteBuffer buf = mapper.mapExternalFile(externalName,
String.valueOf(i));
- Assertions.assertNotNull(buf);
- Assertions.assertEquals(i, buf.getInt());
- }
- }
- }
-
@Test
void testExternalFilesMatchEagerMapper() throws IOException
{
@@ -860,6 +757,7 @@ class PartialSegmentFileMapperV10Test
try (SegmentFileMapperV10 eager = SegmentFileMapperV10.create(segmentFile,
JSON_MAPPER, List.of(externalName));
PartialSegmentFileMapperV10 lazy =
createMapperWithExternal(rangeReader, cacheDir, externalName)
) {
+ lazy.ensureAllDownloaded();
// verify main files match
for (int i = 0; i < 5; ++i) {
ByteBuffer eagerBuf = eager.mapFile(String.valueOf(i));
@@ -888,11 +786,11 @@ class PartialSegmentFileMapperV10Test
final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(segmentFile.getParentFile());
- // fetches from range reader and persists header + bitmap
+ // fetches from range reader and persists header + bitmap; separate
single-file fetches so nothing bridges
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
- mapper.mapFile("2");
- mapper.mapFile("5");
- mapper.mapFile("8");
+ mapper.fetchFiles(Set.of("2"));
+ mapper.fetchFiles(Set.of("5"));
+ mapper.fetchFiles(Set.of("8"));
Assertions.assertEquals(12, mapper.getDownloadedBytes());
}
@@ -920,6 +818,7 @@ class PartialSegmentFileMapperV10Test
Assertions.assertEquals(8, buf8.getInt());
// downloading a new file should still work after restore
+ restored.fetchFiles(Set.of("0"));
ByteBuffer buf0 = restored.mapFile("0");
Assertions.assertNotNull(buf0);
Assertions.assertEquals(0, buf0.getInt());
@@ -951,10 +850,10 @@ class PartialSegmentFileMapperV10Test
final File cacheDir = newCacheDir("ext_persist");
final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(baseDir);
- // create with externals, download some files
+ // create with externals, download some files (external files fetch
through their own mapper)
try (PartialSegmentFileMapperV10 mapper =
createMapperWithExternal(rangeReader, cacheDir, externalName)) {
- mapper.mapFile("1");
- mapper.mapExternalFile(externalName, "7");
+ mapper.fetchFiles(Set.of("1"));
+ mapper.getExternalMapper(externalName).fetchFiles(Set.of("7"));
}
// restore, previously downloaded files should be available
@@ -980,7 +879,7 @@ class PartialSegmentFileMapperV10Test
// populate normally
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
- mapper.mapFile("3");
+ mapper.fetchFiles(Set.of("3"));
}
// corrupt the header file
@@ -995,6 +894,7 @@ class PartialSegmentFileMapperV10Test
try (PartialSegmentFileMapperV10 recovered = createMapper(rangeReader,
cacheDir)) {
Assertions.assertEquals(0, recovered.getDownloadedBytes());
+ recovered.fetchFiles(Set.of("3"));
ByteBuffer buf = recovered.mapFile("3");
Assertions.assertNotNull(buf);
Assertions.assertEquals(3, buf.getInt());
@@ -1036,6 +936,7 @@ class PartialSegmentFileMapperV10Test
try (SegmentFileMapperV10 eager = SegmentFileMapperV10.create(segmentFile,
JSON_MAPPER);
PartialSegmentFileMapperV10 lazy = createMapper(freshReader,
newCacheDir("ensure_all_bulk_match"))) {
lazy.ensureAllDownloaded();
+ Assertions.assertEquals(eager.getInternalFilenames(),
lazy.getInternalFilenames());
for (String name : eager.getInternalFilenames()) {
Assertions.assertEquals(eager.mapFile(name), lazy.mapFile(name),
"file[" + name + "]");
}
@@ -1053,7 +954,7 @@ class PartialSegmentFileMapperV10Test
try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
// make one interior file of the middle container resident first
- mapper.mapFile("b1/col_1");
+ mapper.fetchFiles(Set.of("b1/col_1"));
rangeReader.resetCount();
mapper.ensureAllDownloaded();
@@ -1068,6 +969,36 @@ class PartialSegmentFileMapperV10Test
}
}
+ @Test
+ void testEvictContainerBumpsBundleGeneration() throws IOException
+ {
+ final File segmentFile = buildMultiBundleSegment(3, 4);
+ final File cacheDir = newCacheDir("generation");
+ final CountingRangeReader rangeReader = new
CountingRangeReader(segmentFile.getParentFile());
+
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
+ final long b1Before = mapper.getBundleGeneration("b1");
+ final long b0Before = mapper.getBundleGeneration("b0");
+ final long unknownBefore = mapper.getBundleGeneration("no-such-bundle");
+
+ mapper.ensureBundleDownloaded("b1");
+ Assertions.assertEquals(b1Before, mapper.getBundleGeneration("b1"),
"downloads must not advance the generation");
+
+ for (int containerIndex : mapper.getContainerIndicesForBundle("b1")) {
+ mapper.evictContainer(containerIndex);
+ }
+
+ Assertions.assertTrue(
+ mapper.getBundleGeneration("b1") > b1Before,
+ "eviction must advance the evicted bundle's generation"
+ );
+ Assertions.assertEquals(b0Before, mapper.getBundleGeneration("b0"),
"other bundles' generations must not move");
+ // a name with no containers in this mapper (the cache layer can resolve
names to a catch-all bundle)
+ // conservatively reflects every eviction
+ Assertions.assertTrue(mapper.getBundleGeneration("no-such-bundle") >
unknownBefore);
+ }
+ }
+
@Test
void testEnsureBundleDownloadedFetchesOnlyThatBundle() throws IOException
{
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntryTest.java
b/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntryTest.java
index ac63d61c8f8..a6293b5aed2 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntryTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/PartialSegmentBundleCacheEntryTest.java
@@ -357,6 +357,7 @@ class PartialSegmentBundleCacheEntryTest
final PartialSegmentFileMapperV10 mapper = metadata.getFileMapper();
Assertions.assertNotNull(mapper);
final String anyFile =
mapper.getInternalFilenames().stream().findFirst().orElseThrow();
+ mapper.fetchFiles(Set.of(anyFile));
Assertions.assertNotNull(mapper.mapFile(anyFile));
Assertions.assertEquals(1, mapper.getDownloadedFiles().size());
final PartialSegmentBundleCacheEntry.BundleContainerRef evictedRef =
baseEntry.getContainerRefs().getFirst();
@@ -481,6 +482,7 @@ class PartialSegmentBundleCacheEntryTest
final PartialSegmentFileMapperV10 mapper = metadata.getFileMapper();
Assertions.assertNotNull(mapper);
final String anyFile =
mapper.getInternalFilenames().stream().findFirst().orElseThrow();
+ mapper.fetchFiles(Set.of(anyFile));
Assertions.assertNotNull(mapper.mapFile(anyFile));
final PartialSegmentBundleCacheEntry.BundleContainerRef containerRef =
base.getContainerRefs().getFirst();
final String mapperFilename =
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 8a036fb3955..c972f9eee77 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
@@ -363,9 +363,9 @@ class PartialSegmentCacheBootstrapTest
}
@Test
- void testRestoredEntryCanLazilyFetchUndownloadedFile() throws IOException
+ void testRestoredEntryCanFetchUndownloadedFile() throws IOException
{
- // a bootstrap-restored entry must keep the segment's real deep-storage
range reader so a later query can lazily
+ // a bootstrap-restored entry must keep the segment's real deep-storage
range reader so a later query can
// fetch a bundle/column that wasn't on disk at startup. Previously the
entry held a throwing disk-only reader, so
// this fetch failed with "bootstrap should only read from local disk".
primeOnDiskState();
@@ -385,9 +385,10 @@ class PartialSegmentCacheBootstrapTest
"precondition: " + fileToFetch + " should not be downloaded yet"
);
+ mapper.fetchFiles(Set.of(fileToFetch));
Assertions.assertNotNull(
mapper.mapFile(fileToFetch),
- "restored entry must lazily fetch an un-downloaded file from deep
storage"
+ "restored entry must be able to fetch an un-downloaded file from deep
storage"
);
Assertions.assertTrue(mapper.getDownloadedFiles().contains(fileToFetch));
}
@@ -457,6 +458,7 @@ class PartialSegmentCacheBootstrapTest
mapper.close();
return;
}
+ mapper.fetchFiles(Set.of(fileInAgg));
Assertions.assertNotNull(mapper.mapFile(fileInAgg), "expected file " +
fileInAgg + " to be downloadable");
Assertions.assertTrue(mapper.getDownloadedFiles().contains(fileInAgg));
mapper.close();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
