-
Notifications
You must be signed in to change notification settings - Fork 411
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
storage: Support pipeline model for disaggregated S3 read (#7552)
ref #6827
- Loading branch information
1 parent
874c2c4
commit 2402b48
Showing
7 changed files
with
385 additions
and
45 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
139 changes: 139 additions & 0 deletions
139
dbms/src/Storages/DeltaMerge/Remote/RNSegmentSourceOp.cpp
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,139 @@ | ||
| // Copyright 2023 PingCAP, Ltd. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| #include <Common/TiFlashMetrics.h> | ||
| #include <Storages/DeltaMerge/Remote/RNReadTask.h> | ||
| #include <Storages/DeltaMerge/Remote/RNSegmentSourceOp.h> | ||
| #include <Storages/DeltaMerge/Remote/RNWorkers.h> | ||
|
|
||
| #include <magic_enum.hpp> | ||
|
|
||
| namespace DB::DM::Remote | ||
| { | ||
|
|
||
| void RNSegmentSourceOp::operateSuffixImpl() | ||
| { | ||
| LOG_INFO( | ||
| log, | ||
| "Finished reading remote segments, rows={} read_segments={} total_wait_ready_task={:.3f}s total_read={:.3f}s", | ||
| action.totalRows(), | ||
| processed_seg_tasks, | ||
| duration_wait_ready_task_sec, | ||
| duration_read_sec); | ||
|
|
||
| // This metric is per-stream. | ||
| GET_METRIC(tiflash_disaggregated_breakdown_duration_seconds, type_stream_wait_next_task).Observe(duration_wait_ready_task_sec); | ||
| // This metric is per-stream. | ||
| GET_METRIC(tiflash_disaggregated_breakdown_duration_seconds, type_stream_read).Observe(duration_read_sec); | ||
| } | ||
|
|
||
| void RNSegmentSourceOp::operatePrefixImpl() | ||
| { | ||
| workers->startInBackground(); | ||
| } | ||
|
|
||
| OperatorStatus RNSegmentSourceOp::startGettingNextReadyTask() | ||
| { | ||
| // Start timing the time of get next ready task. | ||
| wait_stop_watch.start(); | ||
| // A quick try to get the next task to reduce the overhead of switching to WaitReactor. | ||
| return awaitImpl(); | ||
| } | ||
|
|
||
| OperatorStatus RNSegmentSourceOp::readImpl(Block & block) | ||
| { | ||
| if unlikely (done) | ||
| { | ||
| block = {}; | ||
| return OperatorStatus::HAS_OUTPUT; | ||
| } | ||
|
|
||
| if (t_block.has_value()) | ||
| { | ||
| std::swap(block, t_block.value()); | ||
| action.transform(block, current_seg_task->meta.physical_table_id); | ||
| t_block.reset(); | ||
| return OperatorStatus::HAS_OUTPUT; | ||
| } | ||
|
|
||
| return current_seg_task ? OperatorStatus::IO : startGettingNextReadyTask(); | ||
| } | ||
|
|
||
| OperatorStatus RNSegmentSourceOp::awaitImpl() | ||
| { | ||
| if unlikely (done || t_block.has_value()) | ||
| { | ||
| duration_wait_ready_task_sec += wait_stop_watch.elapsedSeconds(); | ||
| return OperatorStatus::HAS_OUTPUT; | ||
| } | ||
|
|
||
| if unlikely (current_seg_task) | ||
| { | ||
| duration_wait_ready_task_sec += wait_stop_watch.elapsedSeconds(); | ||
| return OperatorStatus::IO; | ||
| } | ||
|
|
||
| auto pop_result = workers->getReadyChannel()->tryPop(current_seg_task); | ||
| switch (pop_result) | ||
| { | ||
| case MPMCQueueResult::OK: | ||
| processed_seg_tasks += 1; | ||
| RUNTIME_CHECK(current_seg_task != nullptr); | ||
| duration_wait_ready_task_sec += wait_stop_watch.elapsedSeconds(); | ||
| return OperatorStatus::IO; | ||
| case MPMCQueueResult::EMPTY: | ||
| return OperatorStatus::WAITING; | ||
| case MPMCQueueResult::FINISHED: | ||
| current_seg_task = nullptr; | ||
| done = true; | ||
| duration_wait_ready_task_sec += wait_stop_watch.elapsedSeconds(); | ||
| return OperatorStatus::HAS_OUTPUT; | ||
| case MPMCQueueResult::CANCELLED: | ||
| current_seg_task = nullptr; | ||
| done = true; | ||
| duration_wait_ready_task_sec += wait_stop_watch.elapsedSeconds(); | ||
| throw Exception(workers->getReadyChannel()->getCancelReason()); | ||
| default: | ||
| current_seg_task = nullptr; | ||
| done = true; | ||
| duration_wait_ready_task_sec += wait_stop_watch.elapsedSeconds(); | ||
| throw Exception(fmt::format("Unexpected pop result {}", magic_enum::enum_name(pop_result))); | ||
| } | ||
| } | ||
|
|
||
| OperatorStatus RNSegmentSourceOp::executeIOImpl() | ||
| { | ||
| if unlikely (done || t_block.has_value()) | ||
| return OperatorStatus::HAS_OUTPUT; | ||
|
|
||
| if unlikely (!current_seg_task) | ||
| return startGettingNextReadyTask(); | ||
|
|
||
| FilterPtr filter_ignored = nullptr; | ||
| Stopwatch w{CLOCK_MONOTONIC_COARSE}; | ||
| Block res = current_seg_task->getInputStream()->read(filter_ignored, false); | ||
| duration_read_sec += w.elapsedSeconds(); | ||
| if likely (res) | ||
| { | ||
| t_block.emplace(std::move(res)); | ||
| return OperatorStatus::HAS_OUTPUT; | ||
| } | ||
| else | ||
| { | ||
| // Current stream is drained, try to get next ready task. | ||
| current_seg_task = nullptr; | ||
| return startGettingNextReadyTask(); | ||
| } | ||
| } | ||
| } // namespace DB::DM::Remote |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| // Copyright 2023 PingCAP, Ltd. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| #pragma once | ||
|
|
||
| #include <DataStreams/AddExtraTableIDColumnTransformAction.h> | ||
| #include <Operators/Operator.h> | ||
| #include <Storages/DeltaMerge/Remote/RNReadTask_fwd.h> | ||
| #include <Storages/DeltaMerge/Remote/RNWorkers_fwd.h> | ||
|
|
||
| namespace DB::DM::Remote | ||
| { | ||
|
|
||
| class RNSegmentSourceOp : public SourceOp | ||
| { | ||
| static constexpr auto NAME = "RNSegment"; | ||
|
|
||
| public: | ||
| struct Options | ||
| { | ||
| std::string_view debug_tag; | ||
| PipelineExecutorStatus & exec_status; | ||
| const RNWorkersPtr & workers; | ||
| const ColumnDefines & columns_to_read; | ||
| int extra_table_id_index; | ||
| }; | ||
|
|
||
| explicit RNSegmentSourceOp(const Options & options) | ||
| : SourceOp(options.exec_status, String(options.debug_tag)) | ||
| , workers(options.workers) | ||
| , action(options.columns_to_read, options.extra_table_id_index) | ||
| { | ||
| setHeader(action.getHeader()); | ||
| } | ||
|
|
||
| static SourceOpPtr create(const Options & options) | ||
| { | ||
| return std::make_unique<RNSegmentSourceOp>(options); | ||
| } | ||
|
|
||
| String getName() const override { return NAME; } | ||
|
|
||
| IOProfileInfoPtr getIOProfileInfo() const override { return IOProfileInfo::createForLocal(profile_info_ptr); } | ||
|
|
||
| protected: | ||
| void operateSuffixImpl() override; | ||
|
|
||
| void operatePrefixImpl() override; | ||
|
|
||
| OperatorStatus readImpl(Block & block) override; | ||
|
|
||
| OperatorStatus awaitImpl() override; | ||
|
|
||
| OperatorStatus executeIOImpl() override; | ||
|
|
||
| private: | ||
| OperatorStatus startGettingNextReadyTask(); | ||
|
|
||
| private: | ||
| const RNWorkersPtr workers; | ||
| AddExtraTableIDColumnTransformAction action; | ||
|
|
||
| // Temporarily store the block read from current_seg_task->stream and pass it to downstream operators in readImpl. | ||
| std::optional<Block> t_block = std::nullopt; | ||
|
|
||
| RNReadSegmentTaskPtr current_seg_task = nullptr; | ||
| bool done = false; | ||
|
|
||
| // Count the number of segment tasks obtained. | ||
| size_t processed_seg_tasks = 0; | ||
|
|
||
| // Count the time spent waiting for segment tasks to be ready. | ||
| double duration_wait_ready_task_sec = 0; | ||
| Stopwatch wait_stop_watch{CLOCK_MONOTONIC_COARSE}; | ||
|
|
||
| // Count the time consumed by reading blocks in the stream of segment tasks. | ||
| double duration_read_sec = 0; | ||
| }; | ||
|
|
||
| } // namespace DB::DM::Remote |
This file contains 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
Oops, something went wrong.