[fix](pipeline) detect unread local exchange data on eos#63588
Open
Mryange wants to merge 1 commit into
Open
Conversation
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
Author
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Found one blocking issue: the new BE unit tests directly access protected/private exchanger members, so the test target will not compile.
Critical checkpoint conclusions:
- Goal/test proof: The production change aims to detect unread local-exchange data after natural EOS, and tests were added, but the new tests currently fail at compile time.
- Scope/focus: The production change is small and focused; the blocker is limited to test construction.
- Concurrency: The runtime check is executed after EOS from the source path. No new threads or dependency types are introduced; existing exchanger queue locking is reused.
- Lifecycle/static initialization: No new static/global lifecycle hazards found.
- Config/compatibility: No new config, storage format, or FE/BE protocol compatibility concern found.
- Parallel paths: The virtual method covers all exchanger implementations through the template override, with an adaptive override for tmp-block state.
- Conditional checks: The no-limit EOS gate is intentional to avoid flagging limit-short-circuit cases.
- Test coverage/results: Tests cover queue residual and adaptive tmp residual intent, but they need to be rewritten to use public behavior or otherwise made legally accessible before they can validate the feature.
- Observability: Error messages include exchange type, channel, queue info, and unread block details; sufficient for this diagnostic path.
- Transaction/persistence/data writes: Not applicable.
- Performance/memory: Check only runs on natural EOS and should not affect hot steady-state paths.
User focus: No additional user-provided review focus was supplied.
| auto wrapper = ExchangerBase::BlockWrapper::create_shared(std::move(in_block), nullptr, -1); | ||
| auto row_idx = std::make_shared<PODArray<uint32_t>>(1); | ||
| (*row_idx)[0] = 0; | ||
| ASSERT_TRUE(exchanger->_data_queue[0].enqueue( |
Contributor
There was a problem hiding this comment.
This test target will not compile because it accesses Exchanger<PartitionedBlock>::_data_queue, which is a protected member of the base class. The next test has the same compile-time problem with AdaptivePassthroughExchanger::_tmp_block and _tmp_eos, which are private. Please construct these residual states through the public sink/get_block behavior (or otherwise add a deliberate test-only access mechanism) instead of directly reaching into class internals.
Contributor
Author
|
/review 我们的beut是有编译选项可以反问private的。 |
Contributor
|
OpenCode automated review failed and did not complete. Error: Review step was skipped (possibly timeout or cancelled) Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
Contributor
Author
|
/review 我们的beut是有编译选项可以访问private的。 |
Contributor
There was a problem hiding this comment.
Review result: no additional blocking issues found in the PR patch.
Critical checkpoint conclusions:
- Goal/test: The change addresses hidden local-exchange eos with unread buffered data by validating queue/tmp-block state after eos on no-limit source operators; added BE unit tests cover queue residuals, adaptive tmp-block residuals, and adaptive copied-slice behavior.
- Scope: The production change is small and focused on local exchange source/exchanger logic.
- Concurrency: The new check runs only after eos from get_block; at that point running sink operators should be finished for the channel, so I did not find a concrete enqueue/check race. Existing exchanger queue locking patterns are preserved.
- Lifecycle: No new static/global lifecycle concerns. The adaptive tmp-block lifecycle is explicitly checked.
- Config/compatibility: No new configs, protocols, storage formats, or FE-BE compatibility concerns.
- Parallel paths: The virtual check covers the Exchanger template used by shuffle, passthrough, pass-to-one, and broadcast; AdaptivePassthrough also checks its extra tmp-block path.
- Tests/results: Tests are BE unit tests and target the new diagnostics. I did not run them in this review environment.
- Observability: Error messages include channel, exchange type, queue info, and block diagnostics, which should be sufficient for this invariant failure.
- Transaction/persistence/data writes: Not applicable.
- Performance: The check is gated behind eos and no-limit, so it is not on the normal per-row/per-block hot path.
Focus response: I noted the existing thread about direct private/protected member access in BE unit tests. Given the user focus that BE UT has compile options for this access, I did not treat that existing concern as a new issue or duplicate it.
Contributor
Author
|
run buildall |
Contributor
TPC-H: Total hot run time: 32034 ms |
Contributor
TPC-DS: Total hot run time: 172689 ms |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
Contributor
Author
|
run cloud_p0 |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Problem Summary:
A local exchange source without a limit can currently trust eos even when unread data is still buffered in the exchanger. In that case the pipeline may finish with fewer output rows than input rows and the failure is hidden behind a normal end-of-stream. Root cause: the source path did not validate exchanger state after eos, and AdaptivePassthroughExchanger also has a tmp block path outside the normal queue state. This change adds a post-eos residual-data check for the no-limit source path, reports detailed queue and buffered-block diagnostics when unread data remains, and adds unit tests that cover both queue residuals and adaptive passthrough tmp-block residuals.
Release note
None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)