Copilot commented on code in PR #12327:
URL: https://github.com/apache/gluten/pull/12327#discussion_r3560022849
##########
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);
Review Comment:
createOffloadedTwoInputOperator() switches to
offloadedOpConfig.setStreamOperator(...), which will typically create a
SimpleOperatorFactory. GlutenTwoInputOperator extends
GlutenAbstractStreamOperator and relies on
GlutenTwoInputOperatorFactory.createStreamOperator(...) to inject
ProcessingTimeService (and bind the mailbox at task startup). Without the
custom factory, GlutenAbstractStreamOperator.setup(...) can run with a null
ProcessingTimeService and break runtime context/state initialization.
##########
gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java:
##########
@@ -242,11 +267,48 @@ private void executeQuery(StreamTableEnvironment tEnv,
String queryFileName, boo
assertThat(checkJobRunningStatus(insertResult, 30000) == true);
} else {
waitForJobCompletion(insertResult, 30000);
+ // Allow filesystem sink to flush and commit partitions after job
completion.
+ Thread.sleep(2000);
+ if ("q10_orc.sql".equals(queryFileName)) {
+ verifyQ10OrcOutput(queryStartMillis);
+ }
Review Comment:
Thread.sleep(2000) is applied to every non-Kafka Nexmark query. Since
testAllNexmarkSourceQueries runs all queries, this adds ~2s per query and can
significantly inflate CI runtime. It looks like the delay is only needed for
the ORC filesystem sink verification, so it should be conditional on
q10_orc.sql.
##########
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. This makes two-input Gluten operators (e.g.,
joins created via GlutenTwoInputOperatorFactory) look non-Gluten to callers
like OperatorChainSliceGraphGenerator/OffloadedJobGraphGenerator, which can
disable offloading/vectorization decisions or cause Optional.get() failures.
--
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]
