You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "aglinxinyuan (via GitHub)" <gi...@apache.org> on 2026/10/09 04:55:58 UTC
[PR] fix(amber): count a materialized input port as completed once its END is processed [texera]
aglinxinyuan opened a new pull request, #8938:
URL: https://github.com/apache/texera/pull/8938
### What changes were proposed in this PR?
`InputManager.isPortCompleted` now treats a port read from materialization the same as a port fed by a link: it counts as completed once END_CHANNEL has been processed on it. `EndChannelHandler` marks that by setting `WorkerPort.completed`.
Before, such a port reported its reader thread's `finished` flag. The thread sets that flag only after enqueuing END. So a DP thread waiting on the queue, typically the worker that got no tuples, could run `EndChannelHandler` before the flag was set. It then saw the port as not completed and skipped `finalizeOutput`. Nothing called it again, so the worker never finished and its region hung until the test deadline. That is the intermittent `TimeoutException: 1.minutes+30.seconds` in `LoopIntegrationSpec`.
```
Before: reader enqueues END -> DP processes END, flag not set yet -> output never finalized -> region hangs
After: reader enqueues END -> DP processes END, port completed -> output finalized -> region completes
```
Two related effects of the change:
- The flag could also be set while END was still queued. A worker with a second materialized input no longer finalizes before that port's END has been processed.
- `startInputPortReaderThreads` uses the same check to skip ports that already completed. The second `StartWorker` of a two-phase region arrives only after the coordinator has the port's `PortCompleted`, which the worker sends after processing END. So it skips the same ports as before.
### Any related issues, documentation, discussions?
Closes #8937. The two `LoopIntegrationSpec` cases added in #8639 hit the race most often; the race itself dates from #3460.
### How was this PR tested?
Tests first: the first commit adds the tests and fails on `main`, and the second commit makes them pass.
- `InputManagerSpec`: a materialized port is completed once END has been processed, before the reader thread sets its flag. It is not completed while END is still queued, even after the flag is set.
- `EndChannelHandlerSpec`: when END arrives on a materialized input port before its reader thread sets the flag, the worker emits the output `FinalizePort` and `FinalizeExecutor`. On `main` it emits only the input port's `FinalizePort`, which is the hang.
```
sbt "WorkflowExecutionService/testOnly *InputManagerSpec *EndChannelHandlerSpec *StartHandlerSpec *DataProcessorSpec *EndHandlerSpec *StartChannelHandlerSpec *AssignPortHandlerSpec *DPThreadSpec *WorkerSpec"
sbt "WorkflowExecutionService/scalafmtCheckAll" "WorkflowExecutionService/scalafixAll --check"
```
`LoopIntegrationSpec`, run locally on Windows with the sbt JVM and its Python workers pinned to 3 CPUs (affinity mask `0x7`), one case per JVM:
| Case | `main` | This PR |
|---|---|---|
| bind a `$i` reference after two branches of the loop body meet again | 3 of 5 runs hung on the first attempt; 1 also hung on the retry and failed as on CI | 10 of 10 passed on the first attempt, 42–44 s each |
| bind the inner loop's variable where it shadows the outer loop's | (fails on CI, see #8937) | 10 of 10 passed on the first attempt, 64–70 s each |
Unpinned, the whole spec passes with this PR: 8 of 8 cases, 29–59 s each, no retries.
```
AMBER_TEST_FILTER= STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly *LoopIntegrationSpec -- -z \"after two branches of the loop body meet again\""
AMBER_TEST_FILTER=integration-only STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly *LoopIntegrationSpec"
```
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "Xiao-zhen-Liu (via GitHub)" <gi...@apache.org>.
Xiao-zhen-Liu commented on code in PR #8938:
URL: https://github.com/apache/texera/pull/8938#discussion_r4263187206
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala:
##########
@@ -116,20 +116,12 @@ class InputManager(
def getPort(portId: PortIdentity): WorkerPort = ports(portId)
/**
- * For ports that read from materialization, the port completion is marked by the finish of the reader thread.
- * For other ports that connect to upstream links, the completion is marked by the completion the port.
+ * Whether END_CHANNEL has been processed on every channel of the port (`EndChannelHandler` then
+ * marks it completed), for a port read from materialization as well. Its reader thread flags
+ * itself finished only after enqueuing END, so the DP thread can process END before the flag is
+ * set, and the flag can be set while END is still queued.
*/
- def isPortCompleted(portId: PortIdentity): Boolean = {
- if (
- !this.inputPortMaterializationReaderThreads
- .contains(portId) || this.inputPortMaterializationReaderThreads(portId).isEmpty
- ) {
- this.getPort(portId).completed
- } else {
- val existingThread = this.inputPortMaterializationReaderThreads(portId).head
- existingThread.finished
- }
- }
+ def isPortCompleted(portId: PortIdentity): Boolean = this.getPort(portId).completed
Review Comment:
Nit: after this change nothing in production reads the reader thread's `finished` flag; only the new tests do. I'd remove it, so it can't be used for completion again.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #8938:
URL: https://github.com/apache/texera/pull/8938#issuecomment-6074563891
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- No candidates found from `git blame` history.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #8938:
URL: https://github.com/apache/texera/pull/8938#discussion_r4264506987
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala:
##########
@@ -116,20 +116,12 @@ class InputManager(
def getPort(portId: PortIdentity): WorkerPort = ports(portId)
/**
- * For ports that read from materialization, the port completion is marked by the finish of the reader thread.
- * For other ports that connect to upstream links, the completion is marked by the completion the port.
+ * Whether END_CHANNEL has been processed on every channel of the port (`EndChannelHandler` then
+ * marks it completed), for a port read from materialization as well. Its reader thread flags
+ * itself finished only after enqueuing END, so the DP thread can process END before the flag is
+ * set, and the flag can be set while END is still queued.
*/
- def isPortCompleted(portId: PortIdentity): Boolean = {
- if (
- !this.inputPortMaterializationReaderThreads
- .contains(portId) || this.inputPortMaterializationReaderThreads(portId).isEmpty
- ) {
- this.getPort(portId).completed
- } else {
- val existingThread = this.inputPortMaterializationReaderThreads(portId).head
- existingThread.finished
- }
- }
+ def isPortCompleted(portId: PortIdentity): Boolean = this.getPort(portId).completed
Review Comment:
Good call — removed it in bb46d20ad7, along with its `AtomicBoolean`. The tests that read it now check what it stood for: the END_CHANNEL the reader enqueued, or the thread's own state (`NEW` / `TERMINATED`) in the two port-completion tests. Also merged main.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "carloea2 (via GitHub)" <gi...@apache.org>.
carloea2 commented on PR #8938:
URL: https://github.com/apache/texera/pull/8938#issuecomment-6179488907
@Xiao-zhen-Liu, required checks now pass on Linux and macOS, and the review thread is resolved. Could you merge this when you have time? I can then refresh #8142 and #8144 to trigger CI with the loop fix.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #8938:
URL: https://github.com/apache/texera/pull/8938#issuecomment-6074564005
<!-- backport-auto-label-report -->
### Backport auto-label report
This `fix:` PR was checked against each actively-supported release branch. A `release/*` label nominates a backport target; the branch's release manager approving this PR is what sends the fix there. The required **Backport Approvals** check stays red until every label below is approved, so each manager either approves or removes their own label — which is why the labels left on a merged PR are exactly the branches it reached.
| Release branch | Analysis |
| --- | --- |
| ✅ `release/v1.3` | Change detected on this branch — label added; this fix is queued to backport here. @mengw15 decides: approving sends the fix here, removing this label declines it. The merge waits on one or the other. Review requested. |
| ✅ `release/v1.2` | Change detected on this branch — label added; this fix is queued to backport here. @xuang7 decides: approving sends the fix here, removing this label declines it. The merge waits on one or the other. Review requested. |
_[Auto-label run](https://github.com/apache/texera/actions/runs/37886173034)._
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] refactor(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #8938:
URL: https://github.com/apache/texera/pull/8938
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #8938:
URL: https://github.com/apache/texera/pull/8938#issuecomment-6074621226
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 0 better · 🔴 8 worse · ⚪ 7 noise (<±5%) · 0 without baseline
> Compared against main `be65bf8` benchmarked on **this same runner**, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.
[Dashboard](https://apache.github.io/texera/dev/bench/) · [Run](https://github.com/apache/texera/actions/runs/37886220716)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 706 | 0.431 | 13,605/22,820/22,820 us | 🔴 +17.6% / 🔴 +68.5% |
| 🔴 | bs=100 sw=10 sl=64 | 1,580 | 0.965 | 61,512/83,009/83,009 us | 🔴 +12.9% / 🟢 -29.3% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,959 | 1.196 | 503,531/561,934/561,934 us | ⚪ within ±5% / 🟢 +50.8% |
<details><summary>Baseline details</summary>
Latest main `be65bf8` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 706 tuples/sec | 809 tuples/sec | 983 tuples/sec | -12.7% | -28.2% |
| bs=10 sw=10 sl=64 | MB/s | 0.431 MB/s | 0.494 MB/s | 0.6 MB/s | -12.8% | -28.2% |
| bs=10 sw=10 sl=64 | p50 | 13,605 us | 11,572 us | 10,903 us | +17.6% | +24.8% |
| bs=10 sw=10 sl=64 | p95 | 22,820 us | 19,600 us | 13,539 us | +16.4% | +68.5% |
| bs=10 sw=10 sl=64 | p99 | 22,820 us | 19,600 us | 16,174 us | +16.4% | +41.1% |
| bs=100 sw=10 sl=64 | throughput | 1,580 tuples/sec | 1,752 tuples/sec | 1,259 tuples/sec | -9.8% | +25.5% |
| bs=100 sw=10 sl=64 | MB/s | 0.965 MB/s | 1.069 MB/s | 0.769 MB/s | -9.7% | +25.5% |
| bs=100 sw=10 sl=64 | p50 | 61,512 us | 54,504 us | 86,981 us | +12.9% | -29.3% |
| bs=100 sw=10 sl=64 | p95 | 83,009 us | 79,573 us | 93,289 us | +4.3% | -11.0% |
| bs=100 sw=10 sl=64 | p99 | 83,009 us | 79,573 us | 105,695 us | +4.3% | -21.5% |
| bs=1000 sw=10 sl=64 | throughput | 1,959 tuples/sec | 1,994 tuples/sec | 1,300 tuples/sec | -1.8% | +50.7% |
| bs=1000 sw=10 sl=64 | MB/s | 1.196 MB/s | 1.217 MB/s | 0.793 MB/s | -1.7% | +50.8% |
| bs=1000 sw=10 sl=64 | p50 | 503,531 us | 490,896 us | 854,105 us | +2.6% | -41.0% |
| bs=1000 sw=10 sl=64 | p95 | 561,934 us | 567,490 us | 901,686 us | -1.0% | -37.7% |
| bs=1000 sw=10 sl=64 | p99 | 561,934 us | 567,490 us | 924,532 us | -1.0% | -39.2% |
</details>
<details><summary>Raw CSV</summary>
```csv
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,283.47,200,128000,706,0.431,13605.12,22819.84,22819.84
1,100,10,64,20,1265.52,2000,1280000,1580,0.965,61511.53,83009.20,83009.20
2,1000,10,64,20,10207.45,20000,12800000,1959,1.196,503530.74,561934.41,561934.41
```
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(amber): count a materialized input port as completed once its END is processed [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #8938:
URL: https://github.com/apache/texera/pull/8938#issuecomment-6074722280
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/8938?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 92.60%. Comparing base ([`be65bf8`](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`e193699`](https://app.codecov.io/gh/apache/texera/commit/e1936998c8bda0ac9a06103dd2b0baaca2952727?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:white_check_mark: All tests successful. No failed tests found.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #8938 +/- ##
============================================
- Coverage 92.60% 92.60% -0.01%
+ Complexity 5046 5042 -4
============================================
Files 1252 1252
Lines 53586 53582 -4
Branches 6673 6672 -1
============================================
- Hits 49625 49618 -7
Misses 2317 2317
- Partials 1644 1647 +3
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| [access-control-service](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `77.38% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [agent-service](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `99.16% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [amber](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `88.09% <100.00%> (-0.02%)` | :arrow_down: | |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `60.48% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [config-service](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `87.37% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [file-service](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.53% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [frontend](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.59% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [notebook-migration-service](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.73% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pyamber](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `98.52% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/8938/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.09% <ø> (ø)` | | Carriedforward from [be65bf8](https://app.codecov.io/gh/apache/texera/commit/be65bf8564a32eea4196aa0190a04f73884bfe70?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/texera/pull/8938?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org