[feature](exchange) enable shared exchange sink buffer to reduce RPC concurrency. by Mryange · Pull Request #44850 · apache/doris

Mryange · 2024-12-02T06:47:05Z

What problem does this PR solve?

In the past, each exchange sink had its own sink buffer.
If the query concurrency is n, there would be n * n RPCs running concurrently
in a typical shuffle scenario (each sender instance can send data to all downstream instances).
Here, we introduce support for shared sink buffers.
This does not reduce the total number of RPCs but can limit the number of concurrent RPCs.

Release note

None

Check List (For Author)

Test
- Regression test
- Unit Test
- Manual test (add detailed scripts or steps below)
- No need to test or manual test. Explain why:
  - This is a refactor/code format and no logic has been changed.
  - Previous test can cover this change.
  - No code files have been changed.
  - Other reason
Behavior changed:
- No.
- Yes.
Does this need documentation?
- No.
- Yes.

Check List (For Reviewer who merge this PR)

Confirm the release note
Confirm test cases
Confirm document
Add branch pick label

doris-robot · 2024-12-02T06:47:11Z

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

What problem was fixed (it's best to include specific error reporting information). How it was fixed.
Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
What features were added. Why was this function added?
Which code was refactored and why was this part of the code refactored?
Which functions were optimized and what is the difference before and after the optimization?

Mryange · 2024-12-02T06:47:16Z

run buildall

doris-robot · 2024-12-02T08:43:07Z

TeamCity be ut coverage result:
Function Coverage: 38.39% (9978/25992)
Line Coverage: 29.46% (83688/284085)
Region Coverage: 28.59% (43068/150628)
Branch Coverage: 25.20% (21890/86852)
Coverage Report: http://coverage.selectdb-in.cc/coverage/33eb7399ce6330553fcaeeacc1e1a6ad8139540c_33eb7399ce6330553fcaeeacc1e1a6ad8139540c/report/index.html

Gabriel39 · 2024-12-04T03:15:12Z

+}
+
+std::shared_ptr<ExchangeSinkBuffer> ExchangeSinkOperatorX::get_sink_buffer() {
+ if (_child) {


Is _child always true?

Gabriel39 · 2024-12-04T03:26:41Z

- _buffer = buffer;
- _buffer->register_sink(_fragment_instance_id);
- }
+ void register_exchange_buffer(pipeline::ExchangeSinkBuffer* buffer) { _buffer = buffer; }


change register_exchange_buffer to set_exchange_buffer

Gabriel39 · 2024-12-04T06:08:06Z

- }
+ void register_exchange_buffer(pipeline::ExchangeSinkBuffer* buffer) { _buffer = buffer; }
+
+ InstanceLoId ins_id() const { return _fragment_instance_id.lo; }


Suggested change
InstanceLoIdins_id() const { return_fragment_instance_id.lo; }
InstanceLoIdins_lo_id() const { return_fragment_instance_id.lo; }

Gabriel39 · 2024-12-04T06:50:24Z

 ~ExchangeSinkBuffer() override = default;
- void register_sink(TUniqueId);
+ void register_sink(InstanceLoId id) {
+ std::lock_guard lc(_init_lock);


Delete this lock.

Gabriel39 · 2024-12-04T06:58:16Z

- void set_broadcast_dependency(std::shared_ptr<Dependency> broadcast_dependency) {
- _broadcast_dependency = broadcast_dependency;
+ ExchangeSinkLocalState* local_state) {
+ std::lock_guard lc(_init_lock);


Better to reduce locking scope

Mryange · 2024-12-04T11:23:14Z

run buildall

Mryange · 2024-12-04T11:58:49Z

run buildall

doris-robot · 2024-12-04T13:23:44Z

TeamCity be ut coverage result:
Function Coverage: 38.47% (10008/26015)
Line Coverage: 29.51% (83893/284295)
Region Coverage: 28.62% (43125/150683)
Branch Coverage: 25.22% (21923/86942)
Coverage Report: http://coverage.selectdb-in.cc/coverage/03d7c1b4f4dbfe8f48eb0bdf014b071e53ec7d1e_03d7c1b4f4dbfe8f48eb0bdf014b071e53ec7d1e/report/index.html

Mryange · 2024-12-04T16:05:20Z

run buildall

doris-robot · 2024-12-04T16:42:21Z

TeamCity be ut coverage result:
Function Coverage: 38.45% (10005/26018)
Line Coverage: 29.50% (83864/284315)
Region Coverage: 28.61% (43108/150692)
Branch Coverage: 25.20% (21913/86946)
Coverage Report: http://coverage.selectdb-in.cc/coverage/36ad1deccaa2d9d7bd738ec1986f3cb5f9cc0ec1_36ad1deccaa2d9d7bd738ec1986f3cb5f9cc0ec1/report/index.html

Gabriel39 · 2024-12-05T02:33:45Z


 int per_fragment_instance_idx() const { return _per_fragment_instance_idx; }

+ const pipeline::PipelineFragmentContext* fragment_context() const { return _fragment_context; }


Get this from pipeline task

Mryange · 2024-12-05T03:37:11Z

run buildall

Mryange · 2024-12-05T03:42:14Z

run buildall

doris-robot · 2024-12-05T04:56:55Z

TeamCity be ut coverage result:
Function Coverage: 38.47% (10008/26015)
Line Coverage: 29.52% (83922/284311)
Region Coverage: 28.62% (43122/150692)
Branch Coverage: 25.21% (21919/86948)
Coverage Report: http://coverage.selectdb-in.cc/coverage/c7834184fd82bef01d7ed9b81c6b68bce1a311ff_c7834184fd82bef01d7ed9b81c6b68bce1a311ff/report/index.html

Mryange · 2024-12-05T06:56:20Z

run buildall

Mryange · 2024-12-05T09:16:49Z

run beut

doris-robot · 2024-12-05T09:47:36Z

TeamCity be ut coverage result:
Function Coverage: 38.48% (10006/26005)
Line Coverage: 29.51% (83920/284349)
Region Coverage: 28.61% (43128/150722)
Branch Coverage: 25.20% (21920/86978)
Coverage Report: http://coverage.selectdb-in.cc/coverage/fcb52aae9d02224a6f5c133c505b7ab11353e21f_fcb52aae9d02224a6f5c133c505b7ab11353e21f/report/index.html

github-actions · 2024-12-05T13:18:27Z

PR approved by at least one committer and no changes requested.

github-actions · 2024-12-05T13:18:29Z

PR approved by anyone and no changes requested.

Mryange · 2024-12-06T16:48:24Z

run buildall

Mryange · 2024-12-07T03:01:26Z

run buildall

doris-robot · 2024-12-07T04:34:49Z

TeamCity be ut coverage result:
Function Coverage: 38.69% (10062/26006)
Line Coverage: 29.61% (84227/284438)
Region Coverage: 28.70% (43276/150771)
Branch Coverage: 25.26% (21980/87006)
Coverage Report: http://coverage.selectdb-in.cc/coverage/d02232ccfaf22044d6ecc62dbc8440861819aaa1_d02232ccfaf22044d6ecc62dbc8440861819aaa1/report/index.html

HappenLee · 2024-12-08T05:51:23Z

+ throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+ "ExchangeSinkOperatorX did not correctly set the child.");
+ }
+ if (std::dynamic_pointer_cast<SortSourceOperatorX>(_child)) {


better if (_state->enable_shared_exchange_sink_buffer() && ! std::dynamic_pointer_cast<LocalExchangeSourceOperatorX>(_child)) && !std::dynamic_pointer_cast<SortSourceOperatorX>(_child))
And add enough comment for the code

HappenLee · 2024-12-08T05:56:09Z

+
+ // _running_sink_count is used to track how many sinks have not finished yet.
+ // It is only decremented when eos is reached.
+ phmap::flat_hash_map<InstanceLoId, int64_t> _running_sink_count;


rethink should we union the maps. seems all maps same key?

HappenLee · 2024-12-08T06:15:09Z

@@ -271,14 +285,16 @@ Status ExchangeSinkBuffer::_send_rpc(InstanceLoId id) {
 } else if (!s.ok()) {
 _failed(id,


should union the failed check？

Mryange · 2024-12-08T06:43:01Z

run buildall

doris-robot · 2024-12-08T07:22:54Z

TeamCity be ut coverage result:
Function Coverage: 38.69% (10061/26006)
Line Coverage: 29.61% (84210/284437)
Region Coverage: 28.69% (43265/150777)
Branch Coverage: 25.25% (21973/87010)
Coverage Report: http://coverage.selectdb-in.cc/coverage/682969d3e6e502c8f35e63bfa1c23cc639d800e8_682969d3e6e502c8f35e63bfa1c23cc639d800e8/report/index.html

github-actions · 2024-12-09T02:12:00Z

PR approved by at least one committer and no changes requested.

…concurrency. (apache#44850) In the past, each exchange sink had its own sink buffer. If the query concurrency is n, there would be n * n RPCs running concurrently in a typical shuffle scenario (each sender instance can send data to all downstream instances). Here, we introduce support for shared sink buffers. This does not reduce the total number of RPCs but can limit the number of concurrent RPCs.

…concurrency. (apache#44850) ### What problem does this PR solve? In the past, each exchange sink had its own sink buffer. If the query concurrency is n, there would be n * n RPCs running concurrently in a typical shuffle scenario (each sender instance can send data to all downstream instances). Here, we introduce support for shared sink buffers. This does not reduce the total number of RPCs but can limit the number of concurrent RPCs.

Gabriel39 reviewed Dec 4, 2024
View reviewed changes

Mryange force-pushed the shared-sink-buffer-dev-12-2 branch from 33eb739 to 35209a5Compare December 4, 2024 11:23

Gabriel39 reviewed Dec 5, 2024
View reviewed changes

Mryange force-pushed the shared-sink-buffer-dev-12-2 branch from 36ad1de to db66a83Compare December 5, 2024 03:36

Mryange force-pushed the shared-sink-buffer-dev-12-2 branch from db66a83 to c783418Compare December 5, 2024 03:41

Mryange force-pushed the shared-sink-buffer-dev-12-2 branch from c783418 to fcb52aaCompare December 5, 2024 06:55

Gabriel39 previously approved these changes Dec 5, 2024
View reviewed changes

github-actionsBot added the approved Indicates a PR has been approved by one committer. label Dec 5, 2024

github-actionsBot added the reviewed label Dec 5, 2024

Mryange added 2 commits December 6, 2024 14:26

upd
8ccb0de

add be ut
d02232c

Mryange dismissed Gabriel39’s stale review via d02232cDecember 6, 2024 16:47

Mryange force-pushed the shared-sink-buffer-dev-12-2 branch from fcb52aa to d02232cCompare December 6, 2024 16:47

github-actionsBot removed the approved Indicates a PR has been approved by one committer. label Dec 6, 2024

HappenLee reviewed Dec 8, 2024
View reviewed changes

Comment threadbe/src/pipeline/exec/exchange_sink_buffer.cpp

comment
682969d

Gabriel39 approved these changes Dec 9, 2024
View reviewed changes

github-actionsBot added the approved Indicates a PR has been approved by one committer. label Dec 9, 2024

BiteTheDDDDt approved these changes Dec 9, 2024
View reviewed changes

BiteTheDDDDt merged commit 2bc011f into apache:masterDec 9, 2024

Gabriel39 mentioned this pull request Dec 16, 2024
[refactor](exchange) remove duplicate code #45466
Merged
16 tasks

Mryange mentioned this pull request Jan 10, 2025
[feature](exchange) enable shared exchange sink buffer to reduce RPC concurrency #46764
Closed
16 tasks

	InstanceLoIdins_id() const { return_fragment_instance_id.lo; }
	InstanceLoIdins_lo_id() const { return_fragment_instance_id.lo; }


		int per_fragment_instance_idx() const { return _per_fragment_instance_idx; }

		const pipeline::PipelineFragmentContext* fragment_context() const { return _fragment_context; }

		@@ -271,14 +285,16 @@ Status ExchangeSinkBuffer::_send_rpc(InstanceLoId id) {
		} else if (!s.ok()) {
		_failed(id,

Uh oh!

Conversation

Mryange commented Dec 2, 2024• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What problem does this PR solve?

Release note

Check List (For Author)

Check List (For Reviewer who merge this PR)

Uh oh!

doris-robot commented Dec 2, 2024

Uh oh!

Mryange commented Dec 2, 2024

Uh oh!

doris-robot commented Dec 2, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Mryange commented Dec 4, 2024

Uh oh!

Mryange commented Dec 4, 2024

Uh oh!

doris-robot commented Dec 4, 2024

Uh oh!

Mryange commented Dec 4, 2024

Uh oh!

doris-robot commented Dec 4, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Mryange commented Dec 5, 2024

Uh oh!

Mryange commented Dec 5, 2024

Uh oh!

doris-robot commented Dec 5, 2024

Uh oh!

Mryange commented Dec 5, 2024

Uh oh!

Mryange commented Dec 5, 2024

Uh oh!

doris-robot commented Dec 5, 2024

Uh oh!

github-actionsBot commented Dec 5, 2024

Uh oh!

github-actionsBot commented Dec 5, 2024

Uh oh!

Mryange commented Dec 6, 2024

Uh oh!

Mryange commented Dec 7, 2024

Uh oh!

doris-robot commented Dec 7, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HappenLeeDec 8, 2024• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Mryange commented Dec 8, 2024

Uh oh!

doris-robot commented Dec 8, 2024

Uh oh!

github-actionsBot commented Dec 9, 2024

Uh oh!

Reviewers

Mryange commented Dec 2, 2024•
edited
Loading

HappenLeeDec 8, 2024•
edited
Loading