xiangfu0 commented on code in PR #20017:
URL: https://github.com/apache/pinot/pull/20017#discussion_r4366341258
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/routing/WorkerManager.java:
##########
@@ -1231,6 +1269,10 @@ private void
setSegmentsForReplicatedLeafFragment(DispatchablePlanMetadata metad
/// segment lists (or the whole map) on the metadata, rather than by editing
them in place: part of what the
/// assignment is built from is the broker's published partition metadata,
shared across queries and read concurrently
/// by other planning threads. What is handed over is nevertheless kept safe
to edit in place.
+ ///
+ /// The optional segments of each worker are held apart from its segment
lists, in
+ /// [DispatchablePlanMetadata#getWorkerIdToOptionalSegmentsMap()], under the
same worker ids; an override that filters
Review Comment:
Good call documenting that an override must now filter the optional map too.
Worth knowing there are no in-tree overrides of `filterLeafStageSegments`, so
any subclass that prunes segments (e.g. a downstream fork) won't filter
optional segments until updated — the doc is the right mitigation, just
flagging the behavioral obligation for downstream maintainers.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestUtils.java:
##########
@@ -203,18 +206,33 @@ public static List<InstanceRequest>
constructServerQueryRequests(OpChainExecutio
return List.of(
compileInstanceRequest(executionContext, new PinotQuery(pinotQuery),
timeBoundary, TableType.OFFLINE,
offlineTableDataManager.getTableName(),
offlineTableConfigAndSchema.getLeft(),
- offlineTableConfigAndSchema.getRight(), offlineSegments, null),
+ offlineTableConfigAndSchema.getRight(), offlineSegments,
+ copyOptionalSegments(optionalSegmentsMap,
TableType.OFFLINE.name()), null),
compileInstanceRequest(executionContext, pinotQuery, timeBoundary,
TableType.REALTIME,
realtimeTableDataManager.getTableName(),
realtimeTableConfigAndSchema.getLeft(),
- realtimeTableConfigAndSchema.getRight(), realtimeSegments,
null));
+ realtimeTableConfigAndSchema.getRight(), realtimeSegments,
+ copyOptionalSegments(optionalSegmentsMap,
TableType.REALTIME.name()), null));
}
}
+ /// Returns a copy of the optional segments under the given key, or `null`
when there are none. A copy because the
+ /// list is handed over to [InstanceRequest], whose consumers may append to
it (e.g. the newly added segments of an
+ /// upsert table), while the original belongs to the worker metadata.
+ @Nullable
+ private static List<String> copyOptionalSegments(@Nullable Map<String,
List<String>> optionalSegmentsMap,
+ String key) {
+ if (optionalSegmentsMap == null) {
+ return null;
+ }
+ List<String> optionalSegments = optionalSegmentsMap.get(key);
+ return CollectionUtils.isNotEmpty(optionalSegments) ? new
ArrayList<>(optionalSegments) : null;
Review Comment:
The defensive copy here is correct and load-bearing:
`SingleTableExecutionInfo` appends newly-added upsert segments onto this list
in place, so handing over the worker-metadata-owned list directly would mutate
shared planning state. Nice that the test asserts `assertNotSame` on it.
--
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]
