You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2021/03/09 05:47:51 UTC
[GitHub] [incubator-doris] weizuo93 opened a new pull request #5488: [Sql][Stream Load] Add show stream load function for
weizuo93 opened a new pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488
## Proposed changes
Describe the big picture of your changes here to communicate to the maintainers why we should accept this pull request. If it fixes a bug or resolves a feature request, be sure to link to that issue.
## Types of changes
What types of changes does your code introduce to Doris?
_Put an `x` in the boxes that apply_
- [] Bugfix (non-breaking change which fixes an issue)
- [] New feature (non-breaking change which adds functionality)
- [] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [] Documentation Update (if none of the other choices apply)
- [] Code refactor (Modify the code structure, format the code, etc...)
## Checklist
_Put an `x` in the boxes that apply. You can also fill these out after creating the PR. If you're unsure about any of them, don't hesitate to ask. We're here to help! This is simply a reminder of what we are going to look for before merging your code._
- [] I have created an issue on (Fix #ISSUE) and described the bug/feature there in detail
- [] Compiling and unit tests pass locally with my changes
- [] I have added tests that prove my fix is effective or that my feature works
- [] If these changes need document changes, I have updated the document
- [] Any dependent changes have been merged
## Further comments
If this is a relatively large or complex change, kick off the discussion at dev@doris.apache.org by explaining why you chose the solution you did and what alternatives you considered, etc...
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810298145
> @weizuo93 You are right. I am wrong. This PR could go ahead. (But store some load job meta in FE and store some load job meta in BE, which seems not very good)
@kangkaisen
Thanks for your comment. There is code overlap between this PR and PR #5452(https://github.com/apache/incubator-doris/pull/5452). After discussion with Mingyu Chen, we first push forward PR #5452. About storing stream load meta in BE, it's a tradeoff. If BE reports the stream load result to FE when each stream load job finished, it may result in too much stress for FE RPC. So the stream load results are temporarily stored on BE and then sent to FE by batch. After finished data sending, stream load meta will be removed from BE soon. Do you have any better idea?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810226196
> @weizuo93
> Hi, I mean In `LoadManager`, `dbIdToLabelToLoadJobs` already has the stream load join info, why we still need to store stream load job meta in BE? in other words, I think we don't need to store stream load job meta in BE.
@kangkaisen As far as I know, there is no info about stream load in LoadManagerand dbIdToLabelToLoadJobs. LoadManager only manages bulk load.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] kangkaisen commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
kangkaisen commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-808846461
@weizuo93 Hi, Thanks for your contribution.
I don't understand why we need to store the stream load meta both in FE and BE. why we couldn't get stream load info from FE meta?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-826579935
> `show stream load` is better, like `show routine load`
OK, thank you.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] yangzhg commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
yangzhg commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-826495930
`show stream load` is better, like `show routine load`
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] yangzhg merged pull request #5488: [Feature][Stream Load] Add "show stream load" to show stream load record
Posted by GitBox <gi...@apache.org>.
yangzhg merged pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] yangzhg commented on a change in pull request #5488: [Feature][Stream Load] Add "show stream load" to show stream load record
Posted by GitBox <gi...@apache.org>.
yangzhg commented on a change in pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#discussion_r621735280
##########
File path: fe/fe-core/src/main/java/org/apache/doris/common/Config.java
##########
@@ -605,6 +605,12 @@
@ConfField(mutable = true, masterOnly = true)
public static int max_stream_load_record_size = 5000;
+ /**
+ * Whether to close show stream load and clear stream load records in memory.
+ */
+ @ConfField(mutable = true, masterOnly = true)
+ public static boolean close_show_stream_load = false;
Review comment:
`disable_show_stream_load` is better
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 removed a comment on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 removed a comment on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810079040
> @weizuo93 Hi, Thanks for your contribution.
> I don't understand why we need to store the stream load meta both in FE and BE. why we couldn't get stream load info from FE meta?
Our goal is to store the stream load meta in FE. The reason why the stream load meta is stored on BE first is to send the meta to FE in bulk to avoid calling RPC every time the stream load finished. After a certain period of time, stream load meta will be removed from BE.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] kangkaisen commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
kangkaisen commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810146114
@weizuo93
Hi, I mean In `LoadManager`, `dbIdToLabelToLoadJobs` already has the stream load join info, why we still need to store stream load job meta in BE? in other words, I think we don't need to store stream load job meta in BE.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] kangkaisen commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
kangkaisen commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810698322
@weizuo93 Ok, Thanks for your explain. which is OK for me
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810079040
> @weizuo93 Hi, Thanks for your contribution.
> I don't understand why we need to store the stream load meta both in FE and BE. why we couldn't get stream load info from FE meta?
Our goal is to store the stream load meta in FE. The reason why the stream load meta is stored on BE first is to send the meta to FE in bulk to avoid calling RPC every time the stream load finished. After a certain period of time, stream load meta will be removed from BE.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] yangzhg commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
yangzhg commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-826495930
`show stream load` is better, like `show routine load`
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 removed a comment on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 removed a comment on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810225098
> join
@kangkaisen As far as I know, there is no info about stream load in `LoadManager`and `dbIdToLabelToLoadJobs`. `LoadManager` only manages bulk load.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] acelyc111 commented on a change in pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
acelyc111 commented on a change in pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#discussion_r600261821
##########
File path: be/src/http/action/stream_load.cpp
##########
@@ -132,9 +134,19 @@ void StreamLoadAction::handle(HttpRequest* req) {
str = str + '\n';
HttpChannel::send_reply(req, str);
+ auto stream_load_record = StorageEngine::instance()->get_stream_load_record();
+ if (stream_load_record != nullptr) {
+ auto st = stream_load_record->put(ToStringFromUnixMicros(ctx->start_micros + ctx->load_cost_micros) + "_" + ctx->label, str);
+ if (st.ok()) {
+ LOG(INFO) << "put stream_load_record rocksdb successfully. label: " << ctx->label;
Review comment:
Better to log the whole key, then we can check it in rocksdb.
##########
File path: be/src/olap/olap_server.cpp
##########
@@ -118,6 +118,13 @@ Status StorageEngine::start_bg_threads() {
LOG(INFO) << "path scan/gc threads started. number:" << get_stores().size();
}
+ // stream load record clean thread
+ RETURN_IF_ERROR(Thread::create(
Review comment:
No need to create a thread to gc expired data, you can create a Compaction-Filter on this rocksdb instance.
https://github.com/facebook/rocksdb/wiki/Compaction-Filter
##########
File path: be/src/olap/storage_engine.cpp
##########
@@ -225,6 +226,35 @@ Status StorageEngine::_init_store_map() {
for (auto store : tmp_stores) {
_store_map.emplace(store->path(), store);
}
+
+ auto st = _init_stream_load_record();
+ if (!st.ok()) {
+ LOG(WARNING) << "status=" << st.to_string();
+ }
+
+ return Status::OK();
+}
+
+Status StorageEngine::_init_stream_load_record() {
+ std::string stream_load_record_path;
+ if (_store_map.size() > 0) {
Review comment:
Is it valid when _store_map is empty? Would it be better to explicit define the path in config file?
##########
File path: be/src/runtime/stream_load/stream_load_record.h
##########
@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 <map>
+#include <string>
+
+#include "rocksdb/db.h"
+
+#pragma once
+
+namespace doris {
+
+class Status;
+
+class StreamLoadRecord {
+public:
+ StreamLoadRecord(const std::string& root_path);
+
+ virtual ~StreamLoadRecord();
+
+ Status init();
+
+ Status put(const std::string& key, const std::string& value);
+
+ Status get_batch(const std::string& start, const int batch_size, std::map<std::string, std::string> &stream_load_records);
+
+ Status clean_expired_stream_load_record();
+
+private:
+ std::string _root_path;
+ rocksdb::DB* _db;
+ std::vector<rocksdb::ColumnFamilyHandle*> _handles;
+
+ enum ColumnFamilyIndex {
+ DEFAULT_COLUMN_FAMILY_INDEX = 0,
Review comment:
where does the default column family used for?
##########
File path: be/src/http/action/stream_load.cpp
##########
@@ -132,9 +134,19 @@ void StreamLoadAction::handle(HttpRequest* req) {
str = str + '\n';
HttpChannel::send_reply(req, str);
+ auto stream_load_record = StorageEngine::instance()->get_stream_load_record();
+ if (stream_load_record != nullptr) {
+ auto st = stream_load_record->put(ToStringFromUnixMicros(ctx->start_micros + ctx->load_cost_micros) + "_" + ctx->label, str);
+ if (st.ok()) {
+ LOG(INFO) << "put stream_load_record rocksdb successfully. label: " << ctx->label;
+ }
+ } else {
+ LOG(WARNING) << "stream_load_record is null.";
Review comment:
```suggestion
LOG(WARNING) << "put stream_load_record rocksdb failed, status = " << st.ToString();
```
And log the key.
##########
File path: be/src/runtime/stream_load/stream_load_record.cpp
##########
@@ -0,0 +1,137 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 "runtime/stream_load/stream_load_record.h"
+
+#include "common/config.h"
+#include "common/status.h"
+#include "rocksdb/db.h"
+#include "rocksdb/slice.h"
+#include "rocksdb/options.h"
+#include "rocksdb/slice_transform.h"
+#include "util/time.h"
+
+
+namespace doris {
+const std::string STREAM_LOAD_POSTFIX = "/stream_load";
+const size_t PREFIX_LENGTH = 4;
+
+StreamLoadRecord::StreamLoadRecord(const std::string& root_path)
+ : _root_path(root_path),
+ _db(nullptr) {
+}
+
+StreamLoadRecord::~StreamLoadRecord() {
+ for (auto handle : _handles) {
+ _db->DestroyColumnFamilyHandle(handle);
+ handle = nullptr;
+ }
+ if (_db != nullptr) {
+ delete _db;
+ _db= nullptr;
+ }
+}
+
+Status StreamLoadRecord::init() {
+ // init db
+ rocksdb::DBOptions options;
+ options.IncreaseParallelism();
+ options.create_if_missing = true;
+ options.create_missing_column_families = true;
+ std::string db_path = _root_path + STREAM_LOAD_POSTFIX;
+ std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
+ // default column family is required
+ column_families.emplace_back(DEFAULT_COLUMN_FAMILY, rocksdb::ColumnFamilyOptions());
+ // stream load column family add prefix extractor to improve performance and ensure correctness
+ rocksdb::ColumnFamilyOptions stream_load_column_family;
+ stream_load_column_family.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(PREFIX_LENGTH));
+ column_families.emplace_back(STREAM_LOAD_COLUMN_FAMILY, stream_load_column_family);
+ rocksdb::Status s = rocksdb::DB::Open(options, db_path, column_families, &_handles, &_db);
+ if (!s.ok() || _db == nullptr) {
+ LOG(WARNING) << "rocks db open failed, reason:" << s.ToString();
+ return Status::InternalError("Stream load record rocksdb open failed");
+ }
+ return Status::OK();
+}
+
+Status StreamLoadRecord::put(const std::string& key, const std::string& value) {
+ rocksdb::ColumnFamilyHandle* handle = _handles[1];
+ rocksdb::WriteOptions write_options;
+ write_options.sync = false;
+ rocksdb::Status s = _db->Put(write_options, handle, rocksdb::Slice(key), rocksdb::Slice(value));
+ if (!s.ok()) {
+ LOG(WARNING) << "rocks db put key:" << key << " failed, reason:" << s.ToString();
+ return Status::InternalError("Stream load record rocksdb put failed");
+ }
+ return Status::OK();
+}
+
+Status StreamLoadRecord::get_batch(const std::string& start, const int batch_size, std::map<std::string, std::string> &stream_load_records) {
+ rocksdb::ColumnFamilyHandle* handle = _handles[1];
+ std::unique_ptr<rocksdb::Iterator> it(_db->NewIterator(rocksdb::ReadOptions(), handle));
+ if (start == "") {
Review comment:
```suggestion
if (start.empty()) {
```
##########
File path: be/src/runtime/stream_load/stream_load_context.cpp
##########
@@ -89,6 +103,91 @@ std::string StreamLoadContext::to_json() const {
return s.GetString();
}
+void StreamLoadContext::parse_stream_load_record(std::string stream_load_record, TStreamLoadRecord& stream_load_item) {
+
+ rapidjson::Document document;
+ if (document.Parse(stream_load_record.data()).HasParseError()) {
+ return;
+ }
+
+ if (document.HasMember("Label")) {
+ const rapidjson::Value& label = document["Label"];
+ stream_load_item.__set_label(label.GetString());
Review comment:
```suggestion
stream_load_item.__set_label(document["Label"].GetString());
```
##########
File path: be/src/http/action/stream_load.cpp
##########
@@ -200,6 +212,16 @@ int StreamLoadAction::on_header(HttpRequest* req) {
// add new line at end
str = str + '\n';
HttpChannel::send_reply(req, str);
+
+ auto stream_load_record = StorageEngine::instance()->get_stream_load_record();
+ if (stream_load_record != nullptr) {
+ auto st = stream_load_record->put(ToStringFromUnixMicros(ctx->start_micros + ctx->load_cost_micros) + "_" + ctx->label, str);
+ if (st.ok()) {
+ LOG(INFO) << "put stream_load_record rocksdb successfully. label: " << ctx->label;
+ }
+ } else {
+ LOG(WARNING) << "stream_load_record is null.";
Review comment:
Same as above.
##########
File path: be/src/runtime/stream_load/stream_load_record.cpp
##########
@@ -0,0 +1,137 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 "runtime/stream_load/stream_load_record.h"
+
+#include "common/config.h"
+#include "common/status.h"
+#include "rocksdb/db.h"
+#include "rocksdb/slice.h"
+#include "rocksdb/options.h"
+#include "rocksdb/slice_transform.h"
+#include "util/time.h"
+
+
+namespace doris {
+const std::string STREAM_LOAD_POSTFIX = "/stream_load";
+const size_t PREFIX_LENGTH = 4;
+
+StreamLoadRecord::StreamLoadRecord(const std::string& root_path)
+ : _root_path(root_path),
+ _db(nullptr) {
+}
+
+StreamLoadRecord::~StreamLoadRecord() {
+ for (auto handle : _handles) {
+ _db->DestroyColumnFamilyHandle(handle);
+ handle = nullptr;
+ }
+ if (_db != nullptr) {
+ delete _db;
+ _db= nullptr;
+ }
+}
+
+Status StreamLoadRecord::init() {
+ // init db
+ rocksdb::DBOptions options;
+ options.IncreaseParallelism();
+ options.create_if_missing = true;
+ options.create_missing_column_families = true;
+ std::string db_path = _root_path + STREAM_LOAD_POSTFIX;
+ std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
+ // default column family is required
+ column_families.emplace_back(DEFAULT_COLUMN_FAMILY, rocksdb::ColumnFamilyOptions());
+ // stream load column family add prefix extractor to improve performance and ensure correctness
+ rocksdb::ColumnFamilyOptions stream_load_column_family;
+ stream_load_column_family.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(PREFIX_LENGTH));
+ column_families.emplace_back(STREAM_LOAD_COLUMN_FAMILY, stream_load_column_family);
+ rocksdb::Status s = rocksdb::DB::Open(options, db_path, column_families, &_handles, &_db);
+ if (!s.ok() || _db == nullptr) {
+ LOG(WARNING) << "rocks db open failed, reason:" << s.ToString();
+ return Status::InternalError("Stream load record rocksdb open failed");
+ }
+ return Status::OK();
+}
+
+Status StreamLoadRecord::put(const std::string& key, const std::string& value) {
+ rocksdb::ColumnFamilyHandle* handle = _handles[1];
+ rocksdb::WriteOptions write_options;
+ write_options.sync = false;
+ rocksdb::Status s = _db->Put(write_options, handle, rocksdb::Slice(key), rocksdb::Slice(value));
+ if (!s.ok()) {
+ LOG(WARNING) << "rocks db put key:" << key << " failed, reason:" << s.ToString();
+ return Status::InternalError("Stream load record rocksdb put failed");
+ }
+ return Status::OK();
+}
+
+Status StreamLoadRecord::get_batch(const std::string& start, const int batch_size, std::map<std::string, std::string> &stream_load_records) {
+ rocksdb::ColumnFamilyHandle* handle = _handles[1];
+ std::unique_ptr<rocksdb::Iterator> it(_db->NewIterator(rocksdb::ReadOptions(), handle));
+ if (start == "") {
+ it->SeekToFirst();
+ } else {
+ it->Seek(start);
+ rocksdb::Status status = it->status();
+ if (!status.ok()) {
+ it->SeekToFirst();
Review comment:
why seek to first in this case?
##########
File path: be/src/runtime/stream_load/stream_load_record.cpp
##########
@@ -0,0 +1,137 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 "runtime/stream_load/stream_load_record.h"
+
+#include "common/config.h"
+#include "common/status.h"
+#include "rocksdb/db.h"
+#include "rocksdb/slice.h"
+#include "rocksdb/options.h"
+#include "rocksdb/slice_transform.h"
+#include "util/time.h"
+
+
+namespace doris {
+const std::string STREAM_LOAD_POSTFIX = "/stream_load";
+const size_t PREFIX_LENGTH = 4;
+
+StreamLoadRecord::StreamLoadRecord(const std::string& root_path)
+ : _root_path(root_path),
+ _db(nullptr) {
+}
+
+StreamLoadRecord::~StreamLoadRecord() {
+ for (auto handle : _handles) {
+ _db->DestroyColumnFamilyHandle(handle);
+ handle = nullptr;
+ }
+ if (_db != nullptr) {
+ delete _db;
+ _db= nullptr;
+ }
+}
+
+Status StreamLoadRecord::init() {
+ // init db
+ rocksdb::DBOptions options;
+ options.IncreaseParallelism();
+ options.create_if_missing = true;
+ options.create_missing_column_families = true;
+ std::string db_path = _root_path + STREAM_LOAD_POSTFIX;
+ std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
+ // default column family is required
+ column_families.emplace_back(DEFAULT_COLUMN_FAMILY, rocksdb::ColumnFamilyOptions());
+ // stream load column family add prefix extractor to improve performance and ensure correctness
+ rocksdb::ColumnFamilyOptions stream_load_column_family;
+ stream_load_column_family.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(PREFIX_LENGTH));
+ column_families.emplace_back(STREAM_LOAD_COLUMN_FAMILY, stream_load_column_family);
+ rocksdb::Status s = rocksdb::DB::Open(options, db_path, column_families, &_handles, &_db);
+ if (!s.ok() || _db == nullptr) {
+ LOG(WARNING) << "rocks db open failed, reason:" << s.ToString();
+ return Status::InternalError("Stream load record rocksdb open failed");
+ }
+ return Status::OK();
+}
+
+Status StreamLoadRecord::put(const std::string& key, const std::string& value) {
+ rocksdb::ColumnFamilyHandle* handle = _handles[1];
+ rocksdb::WriteOptions write_options;
+ write_options.sync = false;
+ rocksdb::Status s = _db->Put(write_options, handle, rocksdb::Slice(key), rocksdb::Slice(value));
+ if (!s.ok()) {
+ LOG(WARNING) << "rocks db put key:" << key << " failed, reason:" << s.ToString();
+ return Status::InternalError("Stream load record rocksdb put failed");
+ }
+ return Status::OK();
+}
+
+Status StreamLoadRecord::get_batch(const std::string& start, const int batch_size, std::map<std::string, std::string> &stream_load_records) {
+ rocksdb::ColumnFamilyHandle* handle = _handles[1];
+ std::unique_ptr<rocksdb::Iterator> it(_db->NewIterator(rocksdb::ReadOptions(), handle));
+ if (start == "") {
+ it->SeekToFirst();
+ } else {
+ it->Seek(start);
+ rocksdb::Status status = it->status();
+ if (!status.ok()) {
+ it->SeekToFirst();
+ }
+ }
+ rocksdb::Status status = it->status();
+ if (!status.ok()) {
+ LOG(WARNING) << "rocksdb seek failed. reason:" << status.ToString();
+ return Status::InternalError("Stream load record rocksdb seek failed");
+ }
+ int num = 0;
+ for (it->Next(); it->Valid(); it->Next()) {
Review comment:
Have you skipped the first k-v? should it be
```
for (; it->Valid(); it->Next()) {
```
?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810225098
> join
@kangkaisen As far as I know, there is no info about stream load in `LoadManager`and `dbIdToLabelToLoadJobs`. `LoadManager` only manages bulk load.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 edited a comment on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 edited a comment on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810079432
> @weizuo93 Hi, Thanks for your contribution.
> I don't understand why we need to store the stream load meta both in FE and BE. why we couldn't get stream load info from FE meta?
@kangkaisen Our goal is to store the stream load meta in FE. The reason why the stream load meta is stored on BE first is to send the meta to FE in bulk to avoid calling RPC every time the stream load finished. After a certain period of time, stream load meta will be removed from BE.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] kangkaisen commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
kangkaisen commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810257198
@weizuo93 You are right. I am wrong. This PR could go ahead. (But store some load job meta in FE and store some load job meta in BE, which seems not very good)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 commented on a change in pull request #5488: [Feature][Stream Load] Add "show stream load" to show stream load record
Posted by GitBox <gi...@apache.org>.
weizuo93 commented on a change in pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#discussion_r621763780
##########
File path: fe/fe-core/src/main/java/org/apache/doris/common/Config.java
##########
@@ -605,6 +605,12 @@
@ConfField(mutable = true, masterOnly = true)
public static int max_stream_load_record_size = 5000;
+ /**
+ * Whether to close show stream load and clear stream load records in memory.
+ */
+ @ConfField(mutable = true, masterOnly = true)
+ public static boolean close_show_stream_load = false;
Review comment:
> `disable_show_stream_load` is better
OK, I have modified it.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 edited a comment on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 edited a comment on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810298145
> @weizuo93 You are right. I am wrong. This PR could go ahead. (But store some load job meta in FE and store some load job meta in BE, which seems not very good)
@kangkaisen
Thanks for your comment. There is code overlap between this PR and PR #5452. After discussion with Mingyu Chen, we first push forward PR #5452. About storing stream load meta in BE, it's a tradeoff. If BE reports the stream load result to FE when each stream load job finished, it may result in too much stress for FE RPC. So the stream load results are temporarily stored on BE and then sent to FE by batch. After finished data sending, stream load meta will be removed from BE soon. Do you have any better idea?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 edited a comment on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 edited a comment on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810226196
> @weizuo93
> Hi, I mean In `LoadManager`, `dbIdToLabelToLoadJobs` already has the stream load join info, why we still need to store stream load job meta in BE? in other words, I think we don't need to store stream load job meta in BE.
@kangkaisen As far as I know, there is no info about stream load in `LoadManager` and `dbIdToLabelToLoadJobs`. LoadManager only manages bulk load.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] weizuo93 commented on pull request #5488: [Feature][Stream Load] Add "show stream_load" function to show the result of stream load
Posted by GitBox <gi...@apache.org>.
weizuo93 commented on pull request #5488:
URL: https://github.com/apache/incubator-doris/pull/5488#issuecomment-810079432
> @weizuo93 Hi, Thanks for your contribution.
> I don't understand why we need to store the stream load meta both in FE and BE. why we couldn't get stream load info from FE meta?
Our goal is to store the stream load meta in FE. The reason why the stream load meta is stored on BE first is to send the meta to FE in bulk to avoid calling RPC every time the stream load finished. After a certain period of time, stream load meta will be removed from BE.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org