Copilot commented on code in PR #12327:
URL: https://github.com/apache/gluten/pull/12327#discussion_r3527633522
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java:
##########
@@ -360,37 +365,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);
}
Review Comment:
snapshotState(...) now always snapshots with checkpointId 0. This is a
regression from using the real checkpoint id and can break checkpoint/restore
semantics (all snapshots look identical to native). Pass
context.getCheckpointId() into task.snapshotState(...) to ensure native state
is associated with the correct checkpoint.
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java:
##########
@@ -219,27 +222,55 @@ private void drainTaskOutput() {
}
UpIterator.State state = task.advance();
if (state == UpIterator.State.AVAILABLE) {
- final StatefulElement statefulElement = task.statefulGet();
- try {
- if (statefulElement.isWatermark()) {
- StatefulWatermark watermark = statefulElement.asWatermark();
- output.emitWatermark(new Watermark(watermark.getTimestamp()));
- } else if (statefulElement.isWatermarkStatus()) {
- output.emitWatermarkStatus(
-
GlutenWatermarkStatuses.toFlinkWatermarkStatus(statefulElement));
- } else {
- outputBridge.collect(
- output, statefulElement.asRecord(),
sessionResource.getAllocator(), outputType);
- }
- } finally {
- statefulElement.close();
- }
+ processAvailableElement();
} else {
break;
}
}
}
+ private void processAvailableElement() {
+ final StatefulElement statefulElement = task.statefulGet();
+ try {
+ if (statefulElement.isWatermark()) {
+ StatefulWatermark watermark = statefulElement.asWatermark();
+ output.emitWatermark(new Watermark(watermark.getTimestamp()));
+ } else {
+ outputBridge.collect(
+ output, statefulElement.asRecord(),
sessionResource.getAllocator(), outputType);
+ }
+ } finally {
+ statefulElement.close();
+ }
+ }
Review Comment:
processAvailableElement(...) no longer handles native watermark-status
elements (previously emitted via output.emitWatermarkStatus(...)). If the
native side can still emit watermark-status StatefulElements, this will
incorrectly route them through asRecord()/collect(...), which is likely
invalid. Either explicitly handle watermark-status elements here (and
reintroduce an equivalent conversion helper), or explicitly drop them with a
clear comment/assert if they are guaranteed to never be emitted.
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java:
##########
@@ -219,27 +222,55 @@ private void drainTaskOutput() {
}
UpIterator.State state = task.advance();
if (state == UpIterator.State.AVAILABLE) {
- final StatefulElement statefulElement = task.statefulGet();
- try {
- if (statefulElement.isWatermark()) {
- StatefulWatermark watermark = statefulElement.asWatermark();
- output.emitWatermark(new Watermark(watermark.getTimestamp()));
- } else if (statefulElement.isWatermarkStatus()) {
- output.emitWatermarkStatus(
-
GlutenWatermarkStatuses.toFlinkWatermarkStatus(statefulElement));
- } else {
- outputBridge.collect(
- output, statefulElement.asRecord(),
sessionResource.getAllocator(), outputType);
- }
- } finally {
- statefulElement.close();
- }
+ processAvailableElement();
} else {
break;
}
}
}
+ private void processAvailableElement() {
+ final StatefulElement statefulElement = task.statefulGet();
+ try {
+ if (statefulElement.isWatermark()) {
+ StatefulWatermark watermark = statefulElement.asWatermark();
+ output.emitWatermark(new Watermark(watermark.getTimestamp()));
+ } else {
+ outputBridge.collect(
+ output, statefulElement.asRecord(),
sessionResource.getAllocator(), outputType);
+ }
+ } finally {
+ statefulElement.close();
+ }
+ }
+
+ // Drains remaining output from the native task after end-of-input. 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;
+ }
+ }
+ }
Review Comment:
finishTask() blocks the mailbox thread when the native task returns BLOCKED
(task.waitFor()). This can deadlock/hang if unblocking requires mailbox
progress (timers, callbacks, downstream backpressure handling). Prefer a
non-blocking approach: on BLOCKED, break/return and rely on the existing native
callback path (scheduleProcessElementOnMailbox) to resume draining when
unblocked, or use a mailbox/timer to retry with a timeout.
##########
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:
Using a fixed Thread.sleep(2000) makes the UT prone to flakiness (slow CI
machines, filesystem latency, partition commit timing). Replace this with a
bounded wait/poll loop (with timeout) that checks for the expected committed
output (e.g., presence of non-empty ORC files and/or the success-file) before
asserting, so the test is deterministic without relying on timing assumptions.
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java:
##########
@@ -206,93 +204,110 @@ 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();
+ output.emitWatermark(new Watermark(watermark.getTimestamp()));
+ } else {
+ outputBridge.collect(
+ output, element.asRecord(), sessionResource.getAllocator(),
outputType);
+ }
+ } finally {
+ element.close();
+ }
+ }
Review Comment:
Similar to the one-input operator, this no longer handles native
watermark-status elements. If StatefulElement can still represent watermark
status, routing it through asRecord()/collect(...) can be incorrect. Consider
restoring explicit watermark-status handling (or explicitly ignoring it) to
avoid misinterpreting non-record elements.
##########
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenTwoInputOperator.java:
##########
@@ -206,93 +204,110 @@ 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();
+ output.emitWatermark(new Watermark(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;
+ }
+ }
+ }
Review Comment:
Same mailbox-blocking risk as in the one-input operator: calling
task.waitFor() from the mailbox thread on BLOCKED can hang the operator under
certain unblocking conditions. Consider making BLOCKED handling non-blocking
(exit and reschedule draining when unblocked) to avoid potential deadlocks.
--
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]
