Copilot commented on code in PR #12327:
URL: https://github.com/apache/gluten/pull/12327#discussion_r3576896631
##########
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. That will cause
OperatorChainSliceGraphGenerator/OffloadedJobGraphGenerator to treat two-input
Gluten operators as non-offloadable, even though they are valid Gluten
operators (and are still created via GlutenTwoInputOperatorFactory in the
planner).
##########
gluten-flink/planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java:
##########
@@ -325,30 +324,15 @@ protected Transformation<RowData> translateToPlanInternal(
}
final RowType returnType = (RowType) getOutputType();
- final TwoInputTransformation<RowData, RowData, RowData> transform;
- if (operator instanceof GlutenTwoInputOperator) {
- transform =
- ExecNodeUtil.createTwoInputTransformation(
- leftTransform,
- rightTransform,
- createTransformationMeta(JOIN_TRANSFORMATION, config),
- new GlutenTwoInputOperatorFactory<RowData, RowData, RowData>(
- (StreamOperator<RowData>) operator),
- InternalTypeInfo.of(returnType),
- leftTransform.getParallelism(),
- 0L,
- false);
- } else {
- transform =
- ExecNodeUtil.createTwoInputTransformation(
- leftTransform,
- rightTransform,
- createTransformationMeta(JOIN_TRANSFORMATION, config),
- operator,
- InternalTypeInfo.of(returnType),
- leftTransform.getParallelism(),
- false);
- }
+ final TwoInputTransformation<RowData, RowData, RowData> transform =
+ ExecNodeUtil.createTwoInputTransformation(
+ leftTransform,
+ rightTransform,
+ createTransformationMeta(JOIN_TRANSFORMATION, config),
+ new GlutenTwoInputOperatorFactory<>(operator),
Review Comment:
GlutenTwoInputOperatorFactory throws if the operator is not a
GlutenOperator, but `operator` can be a non-Gluten Flink join operator (e.g.,
StreamingSemiAntiJoinOperator, MiniBatchStreamingJoinOperator) in the branches
above. This will fail plan translation with RuntimeException for SEMI/ANTI
joins or when mini-batch join is enabled.
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java:
##########
@@ -360,37 +437,22 @@ public String getRightId() {
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
// TODO: notify velox
- processElementInternal();
super.prepareSnapshotPreBarrier(checkpointId);
}
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
// TODO: implement it
- snapshotNativeState(context.getCheckpointId());
+ task.snapshotState(0);
super.snapshotState(context);
}
@Override
public void initializeState(StateInitializationContext context) throws
Exception {
- // TODO: implement it
- initializeNativeState();
- super.initializeState(context);
- }
-
- private void snapshotNativeState(long checkpointId) {
- if (task != null) {
- task.snapshotState(checkpointId);
- }
- }
-
- private void initializeNativeState() {
- if (stateInitialized) {
- return;
- }
initSession();
+ // TODO: implement it
task.initializeState(0, null);
- stateInitialized = true;
+ super.initializeState(context);
Review Comment:
initializeState() calls initSession() (which binds the native callback
target) before ensuring the mailbox executor is initialized. For operators
configured via SimpleOperatorFactory (e.g., offloaded job graph), a native
callback before open() would hit GlutenMailboxOperatorHelper.scheduleDrain()
with mailboxExecutor==null and run draining on the callback thread, violating
the mailbox-thread requirement and risking races.
##########
gluten-flink/ut/src/test/java/org/apache/gluten/table/runtime/stream/custom/NexmarkTest.java:
##########
@@ -242,11 +271,62 @@ 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();
+ }
Review Comment:
Using a fixed Thread.sleep(2000) to wait for filesystem partition commits is
prone to CI flakiness (slow disks, busy runners). It’s better to poll until the
output is committed (no .inprogress files + part files present) with a timeout
and then verify.
--
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]
