FrankChen021 commented on code in PR #19548:
URL: https://github.com/apache/druid/pull/19548#discussion_r4325514080
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BaseLeafStageProcessor.java:
##########
@@ -264,6 +267,30 @@ private ProcessorManager<Object, Long>
createBaseLeafProcessorManagerWithHandoff
);
}
);
+
+ // segment map function is shared by the regular and handoff managers
above, so neither of them can own its
+ // lifecycle, close it here, once, when this manager is closed by the
frame processor executor, that happens
+ // after all processors have finished and released their mapped segments
+ return new ProcessorManager<>()
+ {
+ @Override
+ public ListenableFuture<Optional<ProcessorAndCallback<Object>>> next()
+ {
+ return delegate.next();
+ }
+
+ @Override
+ public Long result()
+ {
+ return delegate.result();
+ }
+
+ @Override
+ public void close()
+ {
+ CloseableUtils.closeAndWrapExceptions(() ->
CloseableUtils.closeAll(delegate, segmentMapFunction));
Review Comment:
[P1] Undefined segment map variable prevents compilation
**Finding:** The new ProcessorManager.close implementation references
segmentMapFunction, but no such variable exists in this class or repository;
only segmentMapFns, a list of BaseInputSegmentMapFunction, is in scope. The
multi-stage-query module therefore fails compilation with an unresolved-symbol
error.
**Suggestion:** Close the delegate and every segmentMapFn() from the
correctly scoped segmentMapFns list, then add a compile check.
##########
processing/src/main/java/org/apache/druid/segment/SegmentMapFunction.java:
##########
@@ -27,21 +29,51 @@
* possible.
* <p>
* The {@link Segment} returned by this method, if present, must always be
closed by the caller.
+ * <p>
+ * A {@link SegmentMapFunction} may hold resources acquired when it was
created, such as lookup snapshots pinned for
+ * the duration of a join (see {@link
org.apache.druid.query.lookup.RetainedLookupExtractor}). The owner of a
+ * {@link SegmentMapFunction} obtained from {@link
org.apache.druid.query.DataSource#createSegmentMapFunction} must
+ * {@link #close()} it after all mapped segments have been closed.
Implementations created from lambdas hold no
+ * resources and inherit the no-op default. Implementations that hold
resources must make {@link #close()} idempotent.
*/
@FunctionalInterface
-public interface SegmentMapFunction extends Function<Optional<Segment>,
Optional<Segment>>
+public interface SegmentMapFunction extends Function<Optional<Segment>,
Optional<Segment>>, Closeable
Review Comment:
[P2] UNION map wrappers drop close propagation
**Finding:** After SegmentMapFunction became closeable,
ExecutionVertex.splitUnionSegmentMapFunctions still returns plain lambdas that
capture aboveUnionFn and each armFn and inherit the no-op close. MSQ consumes
these functions through BaseInputSegmentMapFunction, so joins or lookups above
a UNION can retain snapshots until cleaner or timeout.
**Suggestion:** Return close-aware UNION wrappers or one composite owner
that closes every arm and the shared above-UNION function exactly once.
--
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]
