add wal function #6
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
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 |
|---|---|---|
|
|
@@ -25,6 +25,7 @@ | |
| #include "common/status.h" | ||
| #include "util/threadpool.h" | ||
| #include "vec/core/block.h" | ||
| #include "vec/sink/writer/vwal_writer.h" | ||
|
|
||
| namespace doris { | ||
| class ExecEnv; | ||
|
|
@@ -53,6 +54,10 @@ class LoadBlockQueue { | |
| Status add_load_id(const UniqueId& load_id); | ||
| void remove_load_id(const UniqueId& load_id); | ||
| void cancel(const Status& st); | ||
| Status create_wal(int64_t db_id, int64_t tb_id, int64_t wal_id, const std::string& import_label, | ||
| WalManager* wal_manager, std::vector<TSlotDescriptor>& slot_desc, | ||
| int be_exe_version); | ||
| Status close_wal(); | ||
|
|
||
| static constexpr size_t MAX_BLOCK_QUEUE_ADD_WAIT_TIME = 1000; | ||
| UniqueId load_instance_id; | ||
|
|
@@ -80,6 +85,7 @@ class LoadBlockQueue { | |
| std::shared_ptr<std::atomic_size_t> _single_block_queue_bytes; | ||
| // group commit interval in ms, can be changed by 'ALTER TABLE my_table SET ("group_commit_interval_ms"="1000");' | ||
| int64_t _group_commit_interval_ms; | ||
| std::shared_ptr<vectorized::VWalWriter> _v_wal_writer; | ||
| }; | ||
|
|
||
| class GroupCommitTable { | ||
|
|
@@ -94,12 +100,14 @@ class GroupCommitTable { | |
| Status get_first_block_load_queue(int64_t table_id, int64_t base_schema_version, | ||
| const UniqueId& load_id, | ||
| std::shared_ptr<vectorized::Block> block, | ||
| std::shared_ptr<LoadBlockQueue>& load_block_queue, | ||
| int be_exe_version); | ||
| Status get_load_block_queue(const TUniqueId& instance_id, | ||
| std::shared_ptr<LoadBlockQueue>& load_block_queue); | ||
|
|
||
| private: | ||
| Status _create_group_commit_load(std::shared_ptr<LoadBlockQueue>& load_block_queue, | ||
| int be_exe_version); | ||
| Status _exec_plan_fragment(int64_t db_id, int64_t table_id, const std::string& label, | ||
| int64_t txn_id, bool is_pipeline, | ||
| const TExecPlanFragmentParams& params, | ||
|
|
@@ -134,7 +142,8 @@ class GroupCommitMgr { | |
| Status get_first_block_load_queue(int64_t db_id, int64_t table_id, int64_t base_schema_version, | ||
| const UniqueId& load_id, | ||
| std::shared_ptr<vectorized::Block> block, | ||
| std::shared_ptr<LoadBlockQueue>& load_block_queue, | ||
| int be_exe_version); | ||
|
|
||
| private: | ||
| ExecEnv* _exec_env = nullptr; | ||
|
|
||
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 |
|---|---|---|
|
|
@@ -122,6 +122,17 @@ Status GroupCommitBlockSink::send(RuntimeState* state, vectorized::Block* input_ | |
| bool has_filtered_rows = false; | ||
| RETURN_IF_ERROR(_block_convertor->validate_and_convert_block( | ||
| state, input_block, block, _output_vexpr_ctxs, rows, has_filtered_rows)); | ||
| if (_block_convertor->num_filtered_rows() > 0) { | ||
| auto cloneBlock = block->clone_without_columns(); | ||
| auto res_block = vectorized::MutableBlock::build_mutable_block(&cloneBlock); | ||
| for (int i = 0; i < rows; ++i) { | ||
| if (_block_convertor->filter_map()[i]) { | ||
| continue; | ||
| } | ||
| res_block.add_row(block.get(), i); | ||
| } | ||
| block->swap(res_block.to_block()); | ||
| } | ||
| // add block into block queue | ||
| return _add_block(state, block); | ||
| } | ||
|
|
@@ -153,10 +164,15 @@ Status GroupCommitBlockSink::_add_block(RuntimeState* state, | |
| load_id.__set_hi(_load_id.hi); | ||
| load_id.__set_lo(_load_id.lo); | ||
| if (_load_block_queue == nullptr) { | ||
| if (state->exec_env()->wal_mgr()->is_running()) { | ||
| RETURN_IF_ERROR(state->exec_env()->group_commit_mgr()->get_first_block_load_queue( | ||
| _db_id, _table_id, _base_schema_version, load_id, block, _load_block_queue, | ||
| state->be_exec_version())); | ||
| state->set_import_label(_load_block_queue->label); | ||
| state->set_wal_id(_load_block_queue->txn_id); | ||
| } else { | ||
| return Status::InternalError("be is stopping"); | ||
| } | ||
| } | ||
| RETURN_IF_ERROR(_load_block_queue->add_block(output_block)); | ||
| return Status::OK(); | ||
|
|
||
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 |
|---|---|---|
|
|
@@ -27,7 +27,6 @@ | |
|
|
||
| #include "common/compiler_util.h" | ||
| #include "common/status.h" | ||
| #include "runtime/client_cache.h" | ||
| #include "runtime/descriptors.h" | ||
| #include "runtime/runtime_state.h" | ||
|
|
@@ -43,67 +42,44 @@ | |
| namespace doris { | ||
| namespace vectorized { | ||
|
|
||
| VWalWriter::VWalWriter(int64_t db_id, int64_t tb_id, int64_t wal_id, | ||
| const std::string& import_label, WalManager* wal_manager, | ||
| std::vector<TSlotDescriptor>& slot_desc, int be_exe_version) | ||
| : _db_id(db_id), | ||
| _tb_id(tb_id), | ||
| _wal_id(wal_id), | ||
| _label(import_label), | ||
| _wal_manager(wal_manager), | ||
| _slot_descs(slot_desc), | ||
| _be_exe_version(be_exe_version) {} | ||
|
|
||
| VWalWriter::~VWalWriter() {} | ||
|
|
||
| Status VWalWriter::init() { | ||
| RETURN_IF_ERROR(_wal_manager->add_wal_path(_db_id, _tb_id, _wal_id, _label)); | ||
| RETURN_IF_ERROR(_wal_manager->create_wal_writer(_wal_id, _wal_writer)); | ||
| _wal_manager->add_wal_status_queue(_tb_id, _wal_id, WalManager::WAL_STATUS::CREATE); | ||
| std::stringstream ss; | ||
| for (auto slot_desc : _slot_descs) { | ||
| if (slot_desc.col_unique_id < 0) { | ||
| continue; | ||
| } | ||
| ss << std::to_string(slot_desc.col_unique_id) << ","; | ||
| } | ||
| std::string col_ids = ss.str().substr(0, ss.str().size() - 1); | ||
| RETURN_IF_ERROR(_wal_writer->append_header(_version, col_ids)); | ||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status VWalWriter::write_wal(vectorized::Block* block) { | ||
|
There was a problem hiding this comment. warning: method 'write_wal' can be made static [readability-convert-member-functions-to-static] be/src/vec/sink/writer/vwal_writer.h:90: - Status write_wal(vectorized::Block* block);
+ static Status write_wal(vectorized::Block* block); |
||
| PBlock pblock; | ||
| size_t uncompressed_bytes = 0, compressed_bytes = 0; | ||
| RETURN_IF_ERROR(block->serialize(_be_exe_version, &pblock, &uncompressed_bytes, | ||
| &compressed_bytes, segment_v2::CompressionTypePB::SNAPPY)); | ||
| RETURN_IF_ERROR(_wal_writer->append_blocks(std::vector<PBlock*> {&pblock})); | ||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status VWalWriter::close() { | ||
| if (_wal_writer != nullptr) { | ||
| RETURN_IF_ERROR(_wal_writer->finalize()); | ||
|
|
||
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 |
|---|---|---|
|
|
@@ -56,6 +56,7 @@ | |
| #include "exec/data_sink.h" | ||
| #include "exec/tablet_info.h" | ||
| #include "gutil/ref_counted.h" | ||
| #include "olap/wal_manager.h" | ||
| #include "olap/wal_writer.h" | ||
| #include "runtime/decimalv2_value.h" | ||
| #include "runtime/exec_env.h" | ||
|
|
@@ -82,16 +83,12 @@ namespace vectorized { | |
|
|
||
| class VWalWriter { | ||
| public: | ||
| VWalWriter(int64_t db_id, int64_t tb_id, int64_t wal_id, const std::string& import_label, | ||
| WalManager* wal_manager, std::vector<TSlotDescriptor>& slot_desc, | ||
| int be_exe_version); | ||
| ~VWalWriter(); | ||
| Status init(); | ||
| Status write_wal(vectorized::Block* block); | ||
| Status close(); | ||
|
|
||
| private: | ||
|
|
@@ -100,8 +97,9 @@ class VWalWriter { | |
| int64_t _wal_id; | ||
| uint32_t _version = 0; | ||
| std::string _label; | ||
| WalManager* _wal_manager; | ||
| std::vector<TSlotDescriptor>& _slot_descs; | ||
| int _be_exe_version = 0; | ||
| std::shared_ptr<WalWriter> _wal_writer; | ||
| }; | ||
| } // namespace vectorized | ||
|
|
||
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
warning: 'gen_cpp/PaloInternalService_types.h' file not found [clang-diagnostic-error]