Copilot commented on code in PR #12327:
URL: https://github.com/apache/gluten/pull/12327#discussion_r3558659700
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/util/Utils.java:
##########
@@ -65,8 +64,6 @@ public static Optional<GlutenOperator> getGlutenOperator(
}
} else if (operatorFactory instanceof GlutenOneInputOperatorFactory) {
return Optional.of(((GlutenOneInputOperatorFactory)
operatorFactory).getOperator());
- } else if (operatorFactory instanceof GlutenTwoInputOperatorFactory) {
- return Optional.of(((GlutenTwoInputOperatorFactory)
operatorFactory).getOperator());
}
return Optional.empty();
Review Comment:
Utils.getGlutenOperator() no longer recognizes
GlutenTwoInputOperatorFactory. For two-input Gluten operators (e.g. joins
created via GlutenTwoInputOperatorFactory), this will make
Utils.getGlutenOperator(...) return Optional.empty(), and downstream logic that
uses this to detect/offload Gluten operators (e.g.
OperatorChainSliceGraphGenerator) will treat them as non-Gluten.
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java:
##########
@@ -206,93 +217,169 @@ private void drainTaskOutput() {
}
UpIterator.State state = task.advance();
if (state == UpIterator.State.AVAILABLE) {
- final StatefulElement element = task.statefulGet();
- try {
- if (element.isWatermark()) {
- StatefulWatermark watermark = element.asWatermark();
- output.emitWatermark(new Watermark(watermark.getTimestamp()));
- } else if (element.isWatermarkStatus()) {
-
output.emitWatermarkStatus(GlutenWatermarkStatuses.toFlinkWatermarkStatus(element));
- } else {
- outputBridge.collect(
- output, element.asRecord(), sessionResource.getAllocator(),
outputType);
- }
- } finally {
- element.close();
- }
+ processAvailableElement();
} else {
break;
}
}
}
+ private void processAvailableElement() {
+ final StatefulElement element = task.statefulGet();
+ try {
+ if (element.isWatermark()) {
+ StatefulWatermark watermark = element.asWatermark();
+ emitWatermarkIfAdvanced(watermark.getTimestamp());
+ } else {
+ outputBridge.collect(
+ output, element.asRecord(), sessionResource.getAllocator(),
outputType);
+ }
+ } finally {
+ element.close();
+ }
+ }
+
+ // Drains remaining output from the native task after both inputs end. The
+ // BLOCKED case calls task.waitFor() which blocks the mailbox thread. This is
+ // safe as long as unblocking only depends on native-side I/O (e.g. flushing
+ // to filesystem) and does not require mailbox progress. If future async
+ // operators introduce mailbox-dependent BLOCKED, this will need to become
+ // non-blocking (break on BLOCKED and rely on native callback ->
+ // scheduleProcessElementOnMailbox to continue draining).
+ private void finishTask() {
+ while (true) {
+ UpIterator.State state = task.advance();
+ switch (state) {
+ case AVAILABLE:
+ processAvailableElement();
+ break;
+ case BLOCKED:
+ task.waitFor();
+ break;
+ case FINISHED:
+ return;
+ default:
+ // Treat unknown states as terminal, consistent with
GlutenSourceFunction.
+ LOG.warn("Unexpected Velox task state in finishTask: {}", state);
+ return;
+ }
+ }
+ }
+
+ private void emitWatermarkIfAdvanced(long timestamp) {
+ if (timestamp > lastEmittedWatermark) {
+ lastEmittedWatermark = timestamp;
+ output.emitWatermark(new Watermark(timestamp));
+ }
+ }
+
+ private void updateInputStatus(int inputIndex, boolean idle) {
+ boolean wasIdle = outputIdle;
+ if (inputIndex == 0) {
+ leftInputIdle = idle;
+ } else {
+ rightInputIdle = idle;
+ }
+ outputIdle = leftInputIdle && rightInputIdle;
+ if (wasIdle != outputIdle) {
+ output.emitWatermarkStatus(outputIdle ? WatermarkStatus.IDLE :
WatermarkStatus.ACTIVE);
+ }
Review Comment:
When one input transitions to IDLE, Flink semantics allow the combined
watermark to advance to the other (still-active) input’s watermark immediately.
updateInputStatus(...) currently only emits WatermarkStatus changes; it does
not emit an updated combined watermark on an IDLE transition, which can leave
downstream event-time progress stuck until another watermark arrives.
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java:
##########
@@ -283,10 +279,7 @@ private void createOffloadedTwoInputOperator(
sourceOperator.getOutputTypes(),
inClass,
outClass);
- // setStreamOperator would wrap this in Flink's SimpleOperatorFactory,
which only initializes
- // ProcessingTimeService for Flink AbstractStreamOperator.
GlutenTwoInputOperator uses
- // GlutenAbstractStreamOperator so it needs the Gluten-specific factory.
- offloadedOpConfig.setStreamOperatorFactory(new
GlutenTwoInputOperatorFactory<>(newTwoInputOp));
+ offloadedOpConfig.setStreamOperator(newTwoInputOp);
Review Comment:
createOffloadedTwoInputOperator() now uses
StreamConfig.setStreamOperator(newTwoInputOp). That bypasses
GlutenTwoInputOperatorFactory.createStreamOperator(...), which is currently
where GlutenAbstractStreamOperator subclasses get their ProcessingTimeService
injected (via setProcessingTimeService(...)) and where
GlutenMailboxOperatorHelper.bindAtTaskStartup(...) is invoked. Without the
factory, processingTimeService/mailbox binding can remain unset for
GlutenTwoInputOperator, risking NPEs or incorrect timer/mailbox behavior at
runtime.
--
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]
