Copilot commented on code in PR #12327:
URL: https://github.com/apache/gluten/pull/12327#discussion_r3563844416
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/client/OffloadedJobGraphGenerator.java:
##########
@@ -283,17 +281,18 @@ 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);
offloadedOpConfig.setStatePartitioner(0, new GlutenKeySelector());
Review Comment:
createOffloadedTwoInputOperator() uses StreamConfig#setStreamOperator, which
will wrap the operator in Flink's SimpleOperatorFactory. That bypasses
GlutenTwoInputOperatorFactory.createStreamOperator(), which is currently the
only place that calls
GlutenAbstractStreamOperator#setProcessingTimeService(...) and binds the Gluten
mailbox at task startup. Without that initialization, GlutenTwoInputOperator (a
GlutenAbstractStreamOperator) can end up with a null ProcessingTimeService
during setup/state initialization.
##########
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());
}
Review Comment:
Utils.getGlutenOperator() no longer recognizes
GlutenTwoInputOperatorFactory. This makes two-input Gluten operators appear
non-offloadable (e.g., OperatorChainSliceGraphGenerator.isOffloadableOperator
uses this), even though the planner still constructs two-input transformations
with GlutenTwoInputOperatorFactory.
--
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]
