Copilot commented on code in PR #12327:
URL: https://github.com/apache/gluten/pull/12327#discussion_r3563828874
##########
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 appear
non-offloadable (see
`OperatorChainSliceGraphGenerator.isOffloadableOperator(...)` calling this
method), and can break offloading/chain-slice detection for joins/window-joins
that are wrapped with `GlutenTwoInputOperatorFactory` in the planner.
##########
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(...)` uses
`offloadedOpConfig.setStreamOperator(newTwoInputOp)`, which results in a
`SimpleOperatorFactory`. `SimpleOperatorFactory` only wires
`ProcessingTimeService` for Flink's `AbstractStreamOperator`, but
`GlutenTwoInputOperator` extends `GlutenAbstractStreamOperator` (a fork that
does not extend Flink's base class). Without `GlutenTwoInputOperatorFactory`,
`processingTimeService` will remain null and can cause NPEs when
`GlutenAbstractStreamOperator` constructs the runtime context and initializes
state.
##########
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);
+ if ("q10_orc.sql".equals(queryFileName)) {
+ // Allow filesystem sink to flush and commit partitions after job
completion.
+ Thread.sleep(2000);
+ verifyQ10OrcOutput(queryStartMillis);
+ }
Review Comment:
Using a fixed `Thread.sleep(2000)` to wait for filesystem sink commits makes
the UT timing-dependent/flaky (e.g., slower CI or different FS semantics). It’s
more deterministic to poll/retry `verifyQ10OrcOutput(...)` with a bounded
timeout until output files appear and are non-empty.
--
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]
