github-advanced-security[bot] commented on code in PR #19683:
URL: https://github.com/apache/druid/pull/19683#discussion_r3578921307
##########
processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java:
##########
@@ -487,58 +537,290 @@
}
/**
- * Download an entire container in a single range read and mark every
internal file it holds as downloaded. The whole
- * region streams straight to the container's local sparse file at offset 0.
+ * Ensure the given internal files are resident using as few deep-storage
range reads as possible. Files are grouped
+ * by container; within a container, requested files that are adjacent (they
tile back-to-back with no padding) are
+ * coalesced into a single range read, and two runs separated by unrequested
files are merged into one read when the
+ * separating byte gap is at most {@code coalesceGapBytes}. Bridged gap
files are whole, valid internal files, so
+ * they are marked downloaded too: the "extra" bytes become useful cached
data rather than waste, trading at most
+ * {@code coalesceGapBytes} of extra bandwidth per bridged gap (a run can
bridge several gaps) for one fewer
+ * deep-storage round trip each. Already-resident files are never
re-fetched; a resident file inside a gap always
+ * splits the run. Unknown names are ignored. Runs never span containers.
* <p>
- * No-op when every file in the container is already downloaded. A
partially-downloaded container is re-fetched in
- * full (already-present files are overwritten with byte-identical data).
Holds the container lock for the whole
- * fetch so a concurrent {@link #evictContainer} or container-init can't
race the write; concurrent per-file
- * {@link #ensureFileDownloaded} calls write byte-identical data so they
remain safe, and the download-bookkeeping is
- * gated on the atomic {@link #downloadedFiles} add so neither path
double-counts.
+ * Callers must hold the same eviction-exclusion the {@link #mapFile}
contract requires (see {@link #evictContainer});
+ * only per-file locks are held across the wire fetch, deliberately not the
container lock, so unrelated container
+ * init/eviction and downloads of non-member files proceed in parallel. A
concurrent {@link #mapFile} of a run member
+ * blocks on that member's lock and finds the file resident on wake.
*/
- private void downloadContainer(int containerIndex) throws IOException
+ public void fetchFiles(Collection<String> fileNames) throws IOException
{
- final List<String> fileNames = containerFileNames.get(containerIndex);
- if (fileNames.isEmpty() || downloadedFiles.containsAll(fileNames)) {
- return;
+ for (FetchRun run : planFetch(fileNames)) {
+ fetchRun(run);
}
- containerLocks[containerIndex].lock();
- try {
- checkClosed();
- if (downloadedFiles.containsAll(fileNames)) {
- return;
+ }
+
+ /**
+ * Plan the coalesced range reads that {@link #fetchFiles} would execute for
the given internal files, without
+ * fetching anything. Callers that want the reads to proceed concurrently
(each is an independent deep-storage
+ * request) submit each returned run to an executor as its own {@link
#fetchRun} call instead of using the
+ * sequential {@code fetchFiles}; the runs of one plan cover disjoint file
sets, so they can execute in any order
+ * and in parallel. Plans are advisory snapshots: files that become resident
between planning and fetching are
+ * skipped or trimmed by {@code fetchRun}'s under-lock re-check, so a stale
plan costs at most a redundant no-op.
+ */
+ public List<FetchRun> planFetch(Collection<String> fileNames)
+ {
+ checkClosed();
+
+ // group the not-yet-resident requested files by container, ascending
container index so a plan's run order is
+ // deterministic (locking happens per-file inside fetchRun; runs are
order-independent)
+ final TreeMap<Integer, Set<String>> requestedByContainer = new TreeMap<>();
+ for (String name : fileNames) {
+ final SegmentInternalFileMetadata fileMetadata =
metadata.getFiles().get(name);
+ if (fileMetadata != null && !downloadedFiles.contains(name)) {
+ requestedByContainer.computeIfAbsent(fileMetadata.getContainer(), k ->
new HashSet<>()).add(name);
}
- ensureContainerInitialized(containerIndex);
+ }
- final SegmentFileContainerMetadata containerMeta =
metadata.getContainers().get(containerIndex);
- streamRangeIntoContainer(
- containerIndex,
- headerSize + containerMeta.getStartOffset(),
- 0,
- containerMeta.getSize(),
- StringUtils.format("container[%d]", containerIndex)
+ final List<FetchRun> runs = new ArrayList<>();
+ for (Map.Entry<Integer, Set<String>> entry :
requestedByContainer.entrySet()) {
+ runs.addAll(
+ planFetchRuns(
+ entry.getKey(),
+ containerFileNames.get(entry.getKey()),
+ metadata.getFiles(),
+ entry.getValue(),
+ downloadedFiles,
+ coalesceGapBytes
+ )
);
+ }
+ return runs;
+ }
- for (String name : fileNames) {
- markDownloaded(name, metadata.getFiles().get(name).getSize());
+ /**
+ * {@link #planFetch} for callers that execute the runs concurrently:
additionally splits any run larger than the
+ * configured maximum at file boundaries, so a large fetch spreads across
multiple parallel deep-storage
+ * connections. Do not use for sequential execution — splitting there only
adds request overhead; that's why the cap
+ * doesn't apply to {@link #planFetch}/{@link #fetchFiles} and the eager
download paths built on them.
+ */
+ public List<FetchRun> planParallelFetch(Collection<String> fileNames)
+ {
+ return splitRuns(planFetch(fileNames), metadata.getFiles(),
maxFetchRunBytes);
+ }
+
+ /**
+ * Plan the parallel-capped range reads for every not-yet-resident file in
{@code bundleName}'s containers, across
+ * this mapper AND every attached external mapper so runs come back paired
with their owning mapper, unlike the
+ * self-scoped {@link #planParallelFetch}. Empty for an unknown bundle.
+ */
+ public List<PlannedFetch> planParallelFetchBundle(String bundleName)
+ {
+ checkClosed();
+ final List<PlannedFetch> fetches = new ArrayList<>();
+ for (int containerIndex : getContainerIndicesForBundle(bundleName)) {
+ // the container's file list is already grouped and offset-sorted, so
plan it directly rather than going
+ // through planFetch's group-by-container pass
+ final List<String> fileNames = containerFileNames.get(containerIndex);
+ final List<FetchRun> runs = splitRuns(
+ planFetchRuns(
+ containerIndex,
+ fileNames,
+ metadata.getFiles(),
+ Set.copyOf(fileNames),
+ downloadedFiles,
+ coalesceGapBytes
+ ),
+ metadata.getFiles(),
+ maxFetchRunBytes
+ );
+ for (FetchRun run : runs) {
+ fetches.add(new PlannedFetch(this, run));
}
}
- finally {
- containerLocks[containerIndex].unlock();
+ for (PartialSegmentFileMapperV10 external : externalMappers.values()) {
+ fetches.addAll(external.planParallelFetchBundle(bundleName));
}
+ return fetches;
}
/**
- * Pre-download a set of internal files so that subsequent {@link
#mapFile(String)} calls for these files will not
- * trigger individual downloads. Files that are already downloaded are
skipped. Useful for batch-downloading all
- * files in a bundle at once (see {@link
SegmentFileBuilder#startFileBundle}).
+ * Split any run larger than {@code maxRunBytes} into consecutive sub-runs
of at most that size, cutting only at
+ * file boundaries so each sub-run still marks exactly its own whole files
downloaded (the bitmap tracks residency
+ * per file, so a read must never end mid-file). A single file larger than
the cap forms its own sub-run since
+ * splitting within a file would need sub-file bookkeeping; internal files
are bounded (large columns split into
+ * multiple internal files at write time) so this caps the worst case
acceptably. {@code maxRunBytes <= 0} disables
+ * splitting.
*/
- public void ensureFilesAvailable(Set<String> fileNames) throws IOException
+ @VisibleForTesting
+ static List<FetchRun> splitRuns(
+ List<FetchRun> runs,
+ Map<String, SegmentInternalFileMetadata> fileMetadata,
+ long maxRunBytes
+ )
{
- for (String name : fileNames) {
- final SegmentInternalFileMetadata fileMetadata =
metadata.getFiles().get(name);
- if (fileMetadata != null) {
- ensureFileDownloaded(name, fileMetadata);
+ if (maxRunBytes <= 0) {
+ return runs;
+ }
+ final List<FetchRun> split = new ArrayList<>(runs.size());
+ for (FetchRun run : runs) {
+ if (run.length() <= maxRunBytes) {
+ split.add(run);
+ continue;
+ }
+ List<String> groupFiles = new ArrayList<>();
+ long groupStart = run.startOffset();
+ long groupEnd = groupStart;
+ for (String file : run.files()) {
+ final SegmentInternalFileMetadata fileMeta = fileMetadata.get(file);
+ final long fileEnd = fileMeta.getStartOffset() + fileMeta.getSize();
+ if (!groupFiles.isEmpty() && fileEnd - groupStart > maxRunBytes) {
+ split.add(new FetchRun(run.containerIndex(), groupStart, groupEnd -
groupStart, List.copyOf(groupFiles)));
+ groupFiles = new ArrayList<>();
+ groupStart = fileMeta.getStartOffset();
+ groupEnd = groupStart;
+ }
+ groupFiles.add(file);
+ // max, not assignment: a zero-length file sharing a neighbor's start
offset must never shrink the sub-run
+ groupEnd = Math.max(groupEnd, fileEnd);
+ }
+ split.add(new FetchRun(run.containerIndex(), groupStart, groupEnd -
groupStart, List.copyOf(groupFiles)));
+ }
+ return split;
+ }
+
+ /**
+ * Plan the coalesced range reads for one container. {@code filesByOffset}
is the container's complete file list in
+ * ascending start-offset order; a run opens at a requested non-resident
file and extends across subsequent files,
+ * bridging unrequested non-resident files whenever the byte gap back to the
run's current end is within
+ * {@code gapTolerance} once the next requested file is reached. Resident
files always split (their bytes are never
+ * re-fetched), and a run never ends with unfetched trailing gap bytes. Pure
function over its arguments; the caller
+ * re-checks residency under per-file locks before fetching since {@code
resident} may grow concurrently.
+ */
+ @VisibleForTesting
+ static List<FetchRun> planFetchRuns(
+ int containerIndex,
+ List<String> filesByOffset,
+ Map<String, SegmentInternalFileMetadata> fileMetadata,
+ Set<String> requested,
+ Set<String> resident,
+ long gapTolerance
+ )
+ {
+ final List<FetchRun> runs = new ArrayList<>();
+ List<String> runFiles = null;
+ long runStart = 0;
+ // once a run is open, runEnd only ever grows (Math.max, never plain
assignment): zero-length internal files
+ // (all-null columns) share a start offset with a neighboring file, and a
plain overwrite could shrink the run
+ // and mark a covered file downloaded without having fetched all of its
bytes
+ long runEnd = 0;
+ final List<String> pendingGapFiles = new ArrayList<>();
+
+ for (String name : filesByOffset) {
+ if (resident.contains(name)) {
+ if (runFiles != null) {
+ runs.add(new FetchRun(containerIndex, runStart, runEnd - runStart,
List.copyOf(runFiles)));
+ runFiles = null;
+ }
+ pendingGapFiles.clear();
+ } else if (requested.contains(name)) {
+ final SegmentInternalFileMetadata fileMeta = fileMetadata.get(name);
+ final long fileEnd = fileMeta.getStartOffset() + fileMeta.getSize();
+ if (runFiles != null && fileMeta.getStartOffset() - runEnd <=
gapTolerance) {
+ for (String gapName : pendingGapFiles) {
+ final SegmentInternalFileMetadata gapMeta =
fileMetadata.get(gapName);
+ runEnd = Math.max(runEnd, gapMeta.getStartOffset() +
gapMeta.getSize());
+ }
+ runFiles.addAll(pendingGapFiles);
+ runFiles.add(name);
+ runEnd = Math.max(runEnd, fileEnd);
+ } else {
+ if (runFiles != null) {
+ runs.add(new FetchRun(containerIndex, runStart, runEnd - runStart,
List.copyOf(runFiles)));
+ }
+ runFiles = new ArrayList<>();
+ runFiles.add(name);
+ runStart = fileMeta.getStartOffset();
+ runEnd = fileEnd;
+ }
+ pendingGapFiles.clear();
+ } else if (runFiles != null) {
+ // unrequested and not resident: bridge candidate while a run is open
+ pendingGapFiles.add(name);
+ }
+ }
+ if (runFiles != null) {
+ runs.add(new FetchRun(containerIndex, runStart, runEnd - runStart,
List.copyOf(runFiles)));
+ }
+ return runs;
+ }
+
+ /**
+ * 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.
+ * <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
+ * duration of every run.
+ */
+ public void fetchRun(FetchRun run) throws IOException
+ {
+ final int containerIndex = run.containerIndex();
+ final List<String> runFiles = run.files();
+ final List<ReentrantLock> locks = new ArrayList<>(runFiles.size());
+ int locked = 0;
+ try {
+ for (String name : runFiles) {
+ final ReentrantLock lock = fileLocks.computeIfAbsent(name, k -> new
ReentrantLock());
+ locks.add(lock);
+ lock.lock();
Review Comment:
## CodeQL / Unreleased lock
This lock might not be unlocked or might be locked more times than it is
unlocked.
[Show more
details](https://github.com/apache/druid/security/code-scanning/11335)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
