FrankChen021 commented on code in PR #19683:
URL: https://github.com/apache/druid/pull/19683#discussion_r3578871904
##########
processing/src/main/java/org/apache/druid/segment/PartialQueryableIndexCursorFactory.java:
##########
@@ -104,77 +104,60 @@ public CursorHolder makeCursorHolder(CursorBuildSpec spec)
@Override
public AsyncCursorHolder makeCursorHolderAsync(CursorBuildSpec spec)
{
- final QueryableProjection<QueryableIndex> matched =
index.getProjection(spec);
- if (matched == null && index.getClusteredBaseSummary() != null) {
- return makeClusteredCursorHolderAsync(spec);
- }
-
- // Aggregate-projection match, or the plain base table: a single bundle
(the projection's, or __base) with the
- // query's required columns.
- final QueryableIndex rowSelector = matched != null ?
matched.getRowSelector() : index;
- final String bundleName = matched != null ? matched.getName() :
Projections.BASE_TABLE_PROJECTION_NAME;
- final DownloadBundle bundle = new DownloadBundle(bundleName, rowSelector,
requiredColumns(rowSelector, matched, spec));
- return buildAsyncCursorHolder(List.of(bundle), () ->
delegate.makeCursorHolderForProjection(spec, matched));
- }
-
- /**
- * Async cursor holder for a clustered base table. Cluster-group resolution
is metadata-only
- * ({@link Projections#planClusterGroupQuery}): only the groups whose
clustering tuples survive the query's filters
- * are downloaded, one bundle (and its required columns) per surviving
group. Once every group is resident, the
- * holder is built via the delegate's clustered path.
- */
- private AsyncCursorHolder makeClusteredCursorHolderAsync(CursorBuildSpec
spec)
- {
- final ClusterGroupQueryPlan plan = Projections.planClusterGroupQuery(
- new ArrayList<>(index.getClusterGroupSchemas()),
- spec
- );
- final List<TableClusterGroupSpec> surviving = plan.survivingGroups();
- if (surviving.isEmpty()) {
- // Filter rules out every group: no bundle to acquire, nothing to
download.
- return AsyncCursorHolder.completed(EmptyCursorHolder.INSTANCE);
- }
-
- final List<DownloadBundle> bundles = new ArrayList<>(surviving.size());
- for (TableClusterGroupSpec group : surviving) {
- final QueryableIndex groupIndex =
index.getClusterGroupQueryableIndex(group, true);
- final CursorBuildSpec groupSpec = plan.rebuildCursorBuildSpec(spec,
group);
- bundles.add(
- new DownloadBundle(
-
Projections.getClusterGroupBundleName(group.getClusteringValueIds()),
- groupIndex,
- requiredColumns(groupIndex, null, groupSpec)
- )
+ // The index resolves what the spec needs (projection match / surviving
cluster groups / base table, required
+ // columns, planned range reads); this factory purely orchestrates the
downloads and cursor wiring.
+ final PartialQueryableIndex.CursorPrefetchPlan plan =
index.planCursorPrefetch(spec);
+ if (plan.clusterGroupPlan() != null) {
+ if (plan.bundles().isEmpty()) {
+ // Filter rules out every cluster group: no bundle to acquire, nothing
to download.
+ return AsyncCursorHolder.completed(EmptyCursorHolder.INSTANCE);
+ }
+ // reuse the plan's cluster resolution
+ return buildAsyncCursorHolder(
+ plan.bundles(),
+ () -> delegate.makeClusteredCursorHolder(spec,
plan.clusterGroupPlan())
);
}
- // Reuse the plan we already computed: delegate.makeClusteredCursorHolder
skips a second planClusterGroupQuery and
- // builds over the now-resident surviving groups (their columns memoized
via getClusterGroupQueryableIndex).
- return buildAsyncCursorHolder(bundles, () ->
delegate.makeClusteredCursorHolder(spec, plan));
+ return buildAsyncCursorHolder(
+ plan.bundles(),
+ () -> delegate.makeCursorHolderForProjection(spec,
plan.matchedProjection())
+ );
}
/**
- * Generalized async-holder build over one or more bundles. Each {@link
DownloadBundle} mounts its cache-layer bundle
- * (its own {@link BundleHoldRelease}, since eviction is per-bundle) and
submits a download per required column; the
- * holder becomes ready once every column of every bundle is downloaded, at
which point {@code innerBuilder} builds the
- * (now fully-resident) cursor holder and the produced holder takes
ownership of {@code inner} plus every bundle hold.
+ * Generalized async-holder build over one or more bundles. Each {@link
PartialQueryableIndex.PrefetchBundle}
+ * mounts its cache-layer bundle (its own {@link BundleHoldRelease}, since
eviction is per-bundle) and submits one
+ * download task per planned range read. Once every run of every bundle is
resident, the ready callback materializes
+ * each bundle's required columns and builds the cursor holder, all inside
every bundle's hold handshake
+ * (deserialization and holder construction both touch the container mmaps,
see {@link #buildHolderUnderHolds}); the
+ * produced holder takes ownership of {@code inner} plus every bundle hold.
*/
- private AsyncCursorHolder buildAsyncCursorHolder(List<DownloadBundle>
bundles, Supplier<CursorHolder> innerBuilder)
+ private AsyncCursorHolder buildAsyncCursorHolder(
+ List<PartialQueryableIndex.PrefetchBundle> bundles,
+ Supplier<CursorHolder> innerBuilder
+ )
{
final List<BundleHoldRelease> holdReleases = new
ArrayList<>(bundles.size());
try {
- final List<AsyncResource<String>> columnDownloads = new ArrayList<>();
- for (DownloadBundle bundle : bundles) {
+ final List<AsyncResource<String>> runDownloads = new ArrayList<>();
+ for (PartialQueryableIndex.PrefetchBundle bundle : bundles) {
final BundleHoldRelease holdRelease = new
BundleHoldRelease(bundleAcquirer.acquire(bundle.bundleName()));
holdReleases.add(holdRelease);
- // submit one materialization task per column so a multi-threaded
download executor can fan them out
- for (String column : bundle.requiredColumns()) {
- columnDownloads.add(submitColumnDownload(bundle.rowSelector(),
column, holdRelease));
+ for (PartialSegmentFileMapperV10.PlannedFetch fetch :
bundle.fetches()) {
+ runDownloads.add(submitRunFetch(bundle.bundleName(), fetch,
holdRelease));
}
}
- final AsyncResource<List<String>> downloaded =
AsyncResources.collect(columnDownloads);
+ if (runDownloads.isEmpty()) {
+ // Everything is already resident, so an empty collect would complete
in its constructor and fire the ready
+ // callback inline on THIS thread — but the callback deserializes
columns and can even hit deep storage
+ // (projection parent columns load lazily), which must not run on a
processing thread. Submit one token task
+ // so the callback hops to the download executor like every other
build.
+ runDownloads.add(bundleAcquirer.submitDownload(() -> "resident"));
Review Comment:
[P1] Keep holder construction off the caller thread
The no-op token can complete before `AsyncResources.collect` and
`addReadyCallback` are wired, and callbacks registered on an already-ready
resource execute synchronously. In that ordering, the callback materializes
columns and may lazily download projection-parent files on the processing
thread, violating the async I/O contract. Schedule the continuation itself on
the download executor instead of relying on the token's completion thread.
--
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]
