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 2023/01/11 13:12:33 UTC
[GitHub] [doris] compasses opened a new pull request, #15839: [Feature] support segment builder tool
compasses opened a new pull request, #15839:
URL: https://github.com/apache/doris/pull/15839
# Proposed changes
Issue Number: just one part of #11640
## Problem summary
Describe your changes.
## Checklist(Required)
1. Does it affect the original behavior:
- [ ] Yes
- [ ✓] No
- [ ] I don't know
2. Has unit tests been added:
- [ ] Yes
- [ ] No
- [ ✓] No Need
3. Has document been added or modified:
- [ ] Yes
- [ ✓] No
- [ ] No Need
4. Does it need to update dependencies:
- [ ] Yes
- [ ✓] No
5. Are there any changes that cannot be rolled back:
- [ ] Yes (If Yes, please explain WHY)
- [ ✓] No
## Further comments
This PR is one part of our bulk load implementation, which provide the tool to build the segment file of a tablet in an external way.
It's support build local and HDFS, which means you need provide the meta file and the data file like this:
```
./segment_builder --meta_file=/path/to/hdr/88409.hdr --data_path=/path/to/data/file --format=parquet --is_remote=false
ll /path/to/data/file
xxx1..gz.parquet
xxx2..gz.parquet
...
```
If the file all from the HDFS, the path should be the HDFS path. Currently only support parquet.
Since from internal we use the privately-owned HDFS lib, *** so this PR HDFS related code may not work ***. I don't have such open source HDFS environment to test it.
From above picture you can see the final work flow:
1. Read the hdr file from the meta path, do some validation and system initialization.
2. Build the HDFS scanner, and read the parquet file from HDFS directly, and generate the segment file on local disk.
3. At last upload the segment file to HDFS, same path with the hdr file, and all these files will be used by the load segment statement.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] morningman commented on pull request #15839: [Feature] support segment builder tool
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on PR #15839:
URL: https://github.com/apache/doris/pull/15839#issuecomment-1401381547
Some questions and suggestions:
1. `builder_scanner` doesn't seem to be used? Only `builder_scanner_memtable` is used?
2. Need to unify the inputs and outputs:
1. Inputs:
* header file in json and data file in parquet(can be orc or other supported file format)
* In the code, the reading methods of different file systems can be unified, and there is no need to use `isHDFS` to judge.
2. Outputs
* new header file in json and Doris segment data file
3. The final upload logic can be encapsulated without being limited to HDFS
4. I think we can generate a manifest file to save all output file list. So that the downstream system can read 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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] freemandealer commented on a diff in pull request #15839: [Feature] support segment builder tool
Posted by "freemandealer (via GitHub)" <gi...@apache.org>.
freemandealer commented on code in PR #15839:
URL: https://github.com/apache/doris/pull/15839#discussion_r1102289435
##########
be/src/tools/segment_builder.cpp:
##########
@@ -0,0 +1,128 @@
+// 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 <gflags/gflags.h>
+#include <unistd.h>
+
+#include <chrono>
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_helper.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+DEFINE_string(meta_file, "", "tablet header meta file");
+DEFINE_string(data_path, "", "this tablet's data to be build");
+DEFINE_bool(is_remote, true, "input data's format, just support parquet");
Review Comment:
misdescription for is_remote
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] github-actions[bot] closed pull request #15839: [Feature] support segment builder tool
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #15839: [Feature] support segment builder tool
URL: https://github.com/apache/doris/pull/15839
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] freemandealer commented on a diff in pull request #15839: [Feature] support segment builder tool
Posted by "freemandealer (via GitHub)" <gi...@apache.org>.
freemandealer commented on code in PR #15839:
URL: https://github.com/apache/doris/pull/15839#discussion_r1102286757
##########
be/src/olap/tablet_manager.h:
##########
@@ -151,7 +150,7 @@ class TabletManager {
// Status::Error<UNINITIALIZED>(), if not inited
Status _add_tablet_unlocked(TTabletId tablet_id, const TabletSharedPtr& tablet,
Review Comment:
As a gentle reminder, strip the '_' for the public method
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] hello-stephen commented on pull request #15839: [Feature] support segment builder tool
Posted by GitBox <gi...@apache.org>.
hello-stephen commented on PR #15839:
URL: https://github.com/apache/doris/pull/15839#issuecomment-1379039479
TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 35.77 seconds
load time: 496 seconds
storage size: 17122683656 Bytes
https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/tmp/20230111160607_clickbench_pr_77941.html
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] github-actions[bot] commented on pull request #15839: [Feature] support segment builder tool
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15839:
URL: https://github.com/apache/doris/pull/15839#issuecomment-1672342106
We're closing this PR because it hasn't been updated in a while.
This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and feel free a maintainer to remove the Stale tag!
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] spaces-X commented on a diff in pull request #15839: [Feature] support segment builder tool
Posted by GitBox <gi...@apache.org>.
spaces-X commented on code in PR #15839:
URL: https://github.com/apache/doris/pull/15839#discussion_r1067650287
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
Review Comment:
why need to redefine these config variable here?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] github-actions[bot] commented on a diff in pull request #15839: [Feature] support segment builder tool
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on code in PR #15839:
URL: https://github.com/apache/doris/pull/15839#discussion_r1066983013
##########
be/src/runtime/descriptor_helper.h:
##########
@@ -70,6 +70,10 @@ class TSlotDescriptorBuilder {
_slot_desc.slotType.types[0].scalar_type.__set_scale(scale);
return *this;
}
+ TSlotDescriptorBuilder& length(int len) {
+ _slot_desc.slotType.types[0].scalar_type.__set_len(len);
+ return *this;
+ }
TSlotDescriptorBuilder& string_type(int len) {
_slot_desc.slotType = get_common_type(to_thrift(TYPE_VARCHAR));
Review Comment:
warning: use of undeclared identifier 'TYPE_VARCHAR' [clang-diagnostic-error]
```cpp
_slot_desc.slotType = get_common_type(to_thrift(TYPE_VARCHAR));
^
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!std::filesystem::exists(path)) { LOG(FATAL) << "file not exist:" << file;
}
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!std::filesystem::is_directory(std::filesystem::status(dir_path))) {
LOG(FATAL) << "build dir should be a directory";
}
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
+ IOContext io_ctx;
+ io::FileReaderSPtr read_ptr = nullptr;
+ st = dfs->open_file(_meta_file, &read_ptr, &io_ctx);
+ if (!st.ok() || !read_ptr)
+ LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
+
+ buf.resize(size);
+
+ size_t read_size = 0;
+ read_ptr->read_at(0, Slice(buf.data(), size), io_ctx, &read_size);
+ if (read_size < size) {
+ LOG(FATAL) << "read meta file size abnormal.."; //<< dfs->GetLastError();
+ }
+
+ read_ptr->close();
+ } else {
+ buf = read_local_file(_meta_file);
+ }
+
+ FileHeader<TabletMetaPB> file_header;
+ buf = buf.substr(file_header.size());
+ // init tablet
+ TabletMeta* tablet_meta = new TabletMeta();
+ Status status = tablet_meta->deserialize(buf);
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to load tablet meta :" << status.to_string();
+ return;
+ }
+
+ LOG(INFO) << "table id:" << tablet_meta->table_id() << " tablet id:" << tablet_meta->tablet_id()
+ << " shard id:" << tablet_meta->shard_id();
+
+ auto data_dir = StorageEngine::instance()->get_store(_build_dir);
+ TabletMetaSharedPtr tablet_meta_ptr(tablet_meta);
+ TabletSharedPtr new_tablet = doris::Tablet::create_tablet_from_meta(tablet_meta_ptr, data_dir);
+ status = StorageEngine::instance()->tablet_manager()->_add_tablet_unlocked(
+ tablet_meta->tablet_id(), new_tablet, false, true);
+
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to add tablet to storage :" << status.to_string();
+ return;
+ }
+
+ std::vector<std::string> files;
+ if (_isHDFS) {
+ auto* dfs = hdfsFileSystem.get();
+ std::vector<io::Path> names;
+ dfs->list(_data_path, &names);
+ for (const auto& name : names) {
+ std::string filename = name.filename().c_str();
+ if (filename.substr(filename.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << name;
+ files.emplace_back(filename);
+ }
+ }
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(_data_path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << file_path;
+ files.emplace_back(file_path);
+ }
+ }
+ }
+
+ BuilderScannerMemtable scanner(new_tablet, _build_dir, _file_type, _isHDFS);
+ scanner.doSegmentBuild(files);
+
+ std::string local_new_header =
+ _build_dir + "/" + std::to_string(new_tablet->table_id()) + ".hdr";
+ TabletMetaSharedPtr new_tablet_meta = std::make_shared<TabletMeta>();
+ new_tablet->generate_tablet_meta_copy(new_tablet_meta);
+ new_tablet_meta->save(local_new_header);
+
+ if (_isHDFS) {
+ //upload segment to meta file directory
+ auto pos = _meta_file.rfind("/");
Review Comment:
warning: 'rfind' called with a string literal consisting of a single character; consider using the more effective overload accepting a character [performance-faster-string-find]
```suggestion
auto pos = _meta_file.rfind('/');
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
+ IOContext io_ctx;
+ io::FileReaderSPtr read_ptr = nullptr;
+ st = dfs->open_file(_meta_file, &read_ptr, &io_ctx);
+ if (!st.ok() || !read_ptr)
+ LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
+
+ buf.resize(size);
+
+ size_t read_size = 0;
+ read_ptr->read_at(0, Slice(buf.data(), size), io_ctx, &read_size);
+ if (read_size < size) {
+ LOG(FATAL) << "read meta file size abnormal.."; //<< dfs->GetLastError();
+ }
+
+ read_ptr->close();
+ } else {
+ buf = read_local_file(_meta_file);
+ }
+
+ FileHeader<TabletMetaPB> file_header;
+ buf = buf.substr(file_header.size());
+ // init tablet
+ TabletMeta* tablet_meta = new TabletMeta();
+ Status status = tablet_meta->deserialize(buf);
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to load tablet meta :" << status.to_string();
+ return;
+ }
+
+ LOG(INFO) << "table id:" << tablet_meta->table_id() << " tablet id:" << tablet_meta->tablet_id()
+ << " shard id:" << tablet_meta->shard_id();
+
+ auto data_dir = StorageEngine::instance()->get_store(_build_dir);
+ TabletMetaSharedPtr tablet_meta_ptr(tablet_meta);
+ TabletSharedPtr new_tablet = doris::Tablet::create_tablet_from_meta(tablet_meta_ptr, data_dir);
+ status = StorageEngine::instance()->tablet_manager()->_add_tablet_unlocked(
+ tablet_meta->tablet_id(), new_tablet, false, true);
+
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to add tablet to storage :" << status.to_string();
+ return;
+ }
+
+ std::vector<std::string> files;
+ if (_isHDFS) {
+ auto* dfs = hdfsFileSystem.get();
+ std::vector<io::Path> names;
+ dfs->list(_data_path, &names);
+ for (const auto& name : names) {
+ std::string filename = name.filename().c_str();
+ if (filename.substr(filename.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << name;
+ files.emplace_back(filename);
+ }
+ }
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(_data_path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << file_path;
+ files.emplace_back(file_path);
+ }
+ }
+ }
+
+ BuilderScannerMemtable scanner(new_tablet, _build_dir, _file_type, _isHDFS);
+ scanner.doSegmentBuild(files);
+
+ std::string local_new_header =
+ _build_dir + "/" + std::to_string(new_tablet->table_id()) + ".hdr";
+ TabletMetaSharedPtr new_tablet_meta = std::make_shared<TabletMeta>();
+ new_tablet->generate_tablet_meta_copy(new_tablet_meta);
+ new_tablet_meta->save(local_new_header);
+
+ if (_isHDFS) {
+ //upload segment to meta file directory
+ auto pos = _meta_file.rfind("/");
+ std::string remote_path = _meta_file.substr(0, pos + 1);
+ LOG(INFO) << "Upload to remote fs:" << remote_path;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(remote_path, &exist);
+ if (!st.ok() || !exist) {
+ st = dfs->create_directory(remote_path);
+ if (!st.ok()) LOG(INFO) << "fail to create directory:" << remote_path;
+ }
+ std::string local_segment_path = _build_dir + "/segment";
+ std::string data_suffix = ".dat";
+ std::vector<std::string> segments;
+ std::vector<std::string> remote_segments;
+
+ for (const auto& file : std::filesystem::directory_iterator(local_segment_path)) {
+ if (!file.is_regular_file()) continue;
+ std::string file_path = file.path().string();
+ if (file_path.substr(file_path.size() - data_suffix.size()) == data_suffix) {
+ LOG(INFO) << "get segment file:" << file_path;
+ segments.emplace_back(file_path);
+ auto pos = file_path.rfind("/");
Review Comment:
warning: 'rfind' called with a string literal consisting of a single character; consider using the more effective overload accepting a character [performance-faster-string-find]
```suggestion
auto pos = file_path.rfind('/');
```
##########
be/src/tools/builder_scanner_memtable.cpp:
##########
@@ -0,0 +1,346 @@
+#include "tools/builder_scanner_memtable.h"
+
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Types_types.h"
+#include "hdfs/hdfs.h"
+#include "olap/delta_writer.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+static const int BATCH_SIZE = 8192;
+
+BuilderScannerMemtable::BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()),
+ _tablet(tablet),
+ _build_dir(build_dir),
+ _file_type(file_type),
+ _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = BATCH_SIZE;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+ _runtime_state.init_mem_trackers(uid);
+}
+
+void BuilderScannerMemtable::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+TPrimitiveType::type BuilderScannerMemtable::getPrimitiveType(FieldType t) {
+ switch (t) {
+ case FieldType::OLAP_FIELD_TYPE_OBJECT: {
+ return TPrimitiveType::OBJECT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_HLL: {
+ return TPrimitiveType::HLL;
+ }
+ case FieldType::OLAP_FIELD_TYPE_CHAR: {
+ return TPrimitiveType::CHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ return TPrimitiveType::VARCHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_STRING: {
+ return TPrimitiveType::STRING;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ return TPrimitiveType::DATE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ return TPrimitiveType::DATETIME;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ return TPrimitiveType::DATEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ return TPrimitiveType::DATETIMEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ return TPrimitiveType::DECIMAL32;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ return TPrimitiveType::DECIMAL64;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ return TPrimitiveType::DECIMAL128I;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ return TPrimitiveType::JSONB;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ return TPrimitiveType::BOOLEAN;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ return TPrimitiveType::TINYINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT: {
+ return TPrimitiveType::SMALLINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ return TPrimitiveType::INT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ return TPrimitiveType::BIGINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT: {
+ return TPrimitiveType::LARGEINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ return TPrimitiveType::FLOAT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ return TPrimitiveType::DOUBLE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ return TPrimitiveType::ARRAY;
+ }
+ default: {
+ LOG(FATAL) << "unknown type error:" << t;
+ }
+ }
+}
+TDescriptorTable BuilderScannerMemtable::create_descriptor_tablet() {
+ TDescriptorTableBuilder dtb;
+
+ // build DST table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ // build SRC table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ return dtb.desc_tbl();
+}
+
+void BuilderScannerMemtable::init_desc_table() {
+ TDescriptorTable t_desc_table = create_descriptor_tablet();
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScannerMemtable::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i;
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScannerMemtable::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::vector<std::string>& files) {
+ LOG(INFO) << "build scan ranges for files size:" << files.size() << " file_type:" << _file_type;
+ for (const auto& file : files) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file;
+ range.file_type = _isHDFS ? TFileType::FILE_HDFS : TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScannerMemtable::doSegmentBuild(const std::vector<std::string>& files) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, files);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!status.ok()) { LOG(FATAL) << "open scan node fail:" << status.to_string();
}
```
##########
be/src/tools/builder_scanner.cpp:
##########
@@ -0,0 +1,366 @@
+#include "tools/builder_scanner.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+
+BuilderScanner::BuilderScanner(TabletSharedPtr tablet, const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()), _tablet(tablet), _file_type(file_type), _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = 8192;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+}
+
+void BuilderScanner::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+int BuilderScanner::create_src_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 1;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ scalar_type.__set_type(TPrimitiveType::FLOAT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ scalar_type.__set_type(TPrimitiveType::DOUBLE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ scalar_type.__set_type(TPrimitiveType::DATETIME);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATETIMEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::TINYINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_UNSIGNED_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::SMALLINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ scalar_type.__set_type(TPrimitiveType::INT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL32);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL64);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL128I);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ scalar_type.__set_type(TPrimitiveType::JSONB);
+ break;
+ }
+ default:
+ LOG(FATAL) << "unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ // Skip the first 8 bytes These 8 bytes are used to indicate whether the field is a null value
+ slot_desc.byteOffset = i * 16 + 8;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ {
+ // TTupleDescriptor source
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_SRC;
+ //Here 8 bytes in order to handle null values
+ t_tuple_desc.byteSize = _tablet->num_columns() * 16 + 8;
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+int BuilderScanner::create_dst_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ int32_t byteOffset = 8;
+ for (int i = 0; i < _tablet->num_columns(); i++, byteOffset += 16) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 0;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ default:
+ LOG(FATAL) << " unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ slot_desc.byteOffset = byteOffset;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ t_desc_table.__isset.slotDescriptors = true;
+ {
+ // TTupleDescriptor dest
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_DST;
+ t_tuple_desc.byteSize = byteOffset + 8; //Here 8 bytes in order to handle null values
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+void BuilderScanner::init_desc_table() {
+ TDescriptorTable t_desc_table;
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ int next_slot_id = 0;
+
+ next_slot_id = create_dst_tuple(t_desc_table, next_slot_id);
+ next_slot_id = create_src_tuple(t_desc_table, next_slot_id);
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScanner::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i; // log_time id in src tuple
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScanner::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::string& path) {
+ LOG(INFO) << "build scan ranges for:" << path << " file_type:" << _file_type;
+ if (_isHDFS) {
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file_path;
+ range.file_type = TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+ }
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScanner::doSegmentBuild(const std::string& path) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, path);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
+
+ std::unique_ptr<RowsetWriter> rowset_writer;
+ PUniqueId load_id;
+ load_id.set_hi(1);
+ load_id.set_lo(1);
+ int64_t transaction_id = 1;
+
+ RowsetWriterContext context;
+ context.txn_id = transaction_id;
+ context.load_id = load_id;
+ context.rowset_state = PREPARED;
+ context.segments_overlap = OVERLAP_UNKNOWN;
+ context.tablet_schema = _tablet->tablet_schema();
+
+ status = _tablet->create_rowset_writer(context, &rowset_writer);
+
+ if (!status.ok()) LOG(FATAL) << "create rowset error:" << status.to_string();
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!status.ok()) { LOG(FATAL) << "create rowset error:" << status.to_string();
}
```
##########
be/src/tools/builder_scanner_memtable.cpp:
##########
@@ -0,0 +1,346 @@
+#include "tools/builder_scanner_memtable.h"
+
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Types_types.h"
+#include "hdfs/hdfs.h"
+#include "olap/delta_writer.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+static const int BATCH_SIZE = 8192;
+
+BuilderScannerMemtable::BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()),
+ _tablet(tablet),
+ _build_dir(build_dir),
+ _file_type(file_type),
+ _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = BATCH_SIZE;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+ _runtime_state.init_mem_trackers(uid);
+}
+
+void BuilderScannerMemtable::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+TPrimitiveType::type BuilderScannerMemtable::getPrimitiveType(FieldType t) {
+ switch (t) {
+ case FieldType::OLAP_FIELD_TYPE_OBJECT: {
+ return TPrimitiveType::OBJECT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_HLL: {
+ return TPrimitiveType::HLL;
+ }
+ case FieldType::OLAP_FIELD_TYPE_CHAR: {
+ return TPrimitiveType::CHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ return TPrimitiveType::VARCHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_STRING: {
+ return TPrimitiveType::STRING;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ return TPrimitiveType::DATE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ return TPrimitiveType::DATETIME;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ return TPrimitiveType::DATEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ return TPrimitiveType::DATETIMEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ return TPrimitiveType::DECIMAL32;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ return TPrimitiveType::DECIMAL64;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ return TPrimitiveType::DECIMAL128I;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ return TPrimitiveType::JSONB;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ return TPrimitiveType::BOOLEAN;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ return TPrimitiveType::TINYINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT: {
+ return TPrimitiveType::SMALLINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ return TPrimitiveType::INT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ return TPrimitiveType::BIGINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT: {
+ return TPrimitiveType::LARGEINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ return TPrimitiveType::FLOAT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ return TPrimitiveType::DOUBLE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ return TPrimitiveType::ARRAY;
+ }
+ default: {
+ LOG(FATAL) << "unknown type error:" << t;
+ }
+ }
+}
+TDescriptorTable BuilderScannerMemtable::create_descriptor_tablet() {
+ TDescriptorTableBuilder dtb;
+
+ // build DST table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ // build SRC table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ return dtb.desc_tbl();
+}
+
+void BuilderScannerMemtable::init_desc_table() {
+ TDescriptorTable t_desc_table = create_descriptor_tablet();
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScannerMemtable::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i;
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScannerMemtable::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::vector<std::string>& files) {
+ LOG(INFO) << "build scan ranges for files size:" << files.size() << " file_type:" << _file_type;
+ for (const auto& file : files) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file;
+ range.file_type = _isHDFS ? TFileType::FILE_HDFS : TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScannerMemtable::doSegmentBuild(const std::vector<std::string>& files) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, files);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
+
+ // std::unique_ptr<RowsetWriter> rowset_writer;
+ PUniqueId load_id;
+ load_id.set_hi(1);
+ load_id.set_lo(1);
+ int64_t transaction_id = 1;
+
+ // delta writer
+ TupleDescriptor* tuple_desc = _desc_tbl->get_tuple_descriptor(TUPLE_ID_DST);
+ WriteRequest write_req = {_tablet->tablet_meta()->tablet_id(),
+ _tablet->schema_hash(),
+ WriteType::LOAD,
+ transaction_id,
+ _tablet->partition_id(),
+ load_id,
+ tuple_desc,
+ &(tuple_desc->slots())};
+
+ DeltaWriter* delta_writer = nullptr;
+ DeltaWriter::open(&write_req, &delta_writer, load_id);
+ status = delta_writer->init();
+ if (!status.ok()) LOG(FATAL) << "delta_writer init fail:" << status.to_string();
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!status.ok()) { LOG(FATAL) << "delta_writer init fail:" << status.to_string();
}
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
+ IOContext io_ctx;
+ io::FileReaderSPtr read_ptr = nullptr;
+ st = dfs->open_file(_meta_file, &read_ptr, &io_ctx);
+ if (!st.ok() || !read_ptr)
+ LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!st.ok() || !read_ptr) {
LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
}
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
+ IOContext io_ctx;
+ io::FileReaderSPtr read_ptr = nullptr;
+ st = dfs->open_file(_meta_file, &read_ptr, &io_ctx);
+ if (!st.ok() || !read_ptr)
+ LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
+
+ buf.resize(size);
+
+ size_t read_size = 0;
+ read_ptr->read_at(0, Slice(buf.data(), size), io_ctx, &read_size);
+ if (read_size < size) {
+ LOG(FATAL) << "read meta file size abnormal.."; //<< dfs->GetLastError();
+ }
+
+ read_ptr->close();
+ } else {
+ buf = read_local_file(_meta_file);
+ }
+
+ FileHeader<TabletMetaPB> file_header;
+ buf = buf.substr(file_header.size());
+ // init tablet
+ TabletMeta* tablet_meta = new TabletMeta();
+ Status status = tablet_meta->deserialize(buf);
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to load tablet meta :" << status.to_string();
+ return;
+ }
+
+ LOG(INFO) << "table id:" << tablet_meta->table_id() << " tablet id:" << tablet_meta->tablet_id()
+ << " shard id:" << tablet_meta->shard_id();
+
+ auto data_dir = StorageEngine::instance()->get_store(_build_dir);
+ TabletMetaSharedPtr tablet_meta_ptr(tablet_meta);
+ TabletSharedPtr new_tablet = doris::Tablet::create_tablet_from_meta(tablet_meta_ptr, data_dir);
+ status = StorageEngine::instance()->tablet_manager()->_add_tablet_unlocked(
+ tablet_meta->tablet_id(), new_tablet, false, true);
+
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to add tablet to storage :" << status.to_string();
+ return;
+ }
+
+ std::vector<std::string> files;
+ if (_isHDFS) {
+ auto* dfs = hdfsFileSystem.get();
+ std::vector<io::Path> names;
+ dfs->list(_data_path, &names);
+ for (const auto& name : names) {
+ std::string filename = name.filename().c_str();
+ if (filename.substr(filename.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << name;
+ files.emplace_back(filename);
+ }
+ }
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(_data_path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << file_path;
+ files.emplace_back(file_path);
+ }
+ }
+ }
+
+ BuilderScannerMemtable scanner(new_tablet, _build_dir, _file_type, _isHDFS);
+ scanner.doSegmentBuild(files);
+
+ std::string local_new_header =
+ _build_dir + "/" + std::to_string(new_tablet->table_id()) + ".hdr";
+ TabletMetaSharedPtr new_tablet_meta = std::make_shared<TabletMeta>();
+ new_tablet->generate_tablet_meta_copy(new_tablet_meta);
+ new_tablet_meta->save(local_new_header);
+
+ if (_isHDFS) {
+ //upload segment to meta file directory
+ auto pos = _meta_file.rfind("/");
+ std::string remote_path = _meta_file.substr(0, pos + 1);
+ LOG(INFO) << "Upload to remote fs:" << remote_path;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(remote_path, &exist);
+ if (!st.ok() || !exist) {
+ st = dfs->create_directory(remote_path);
+ if (!st.ok()) LOG(INFO) << "fail to create directory:" << remote_path;
+ }
+ std::string local_segment_path = _build_dir + "/segment";
+ std::string data_suffix = ".dat";
+ std::vector<std::string> segments;
+ std::vector<std::string> remote_segments;
+
+ for (const auto& file : std::filesystem::directory_iterator(local_segment_path)) {
+ if (!file.is_regular_file()) continue;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!file.is_regular_file()) { continue;
}
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
else {
_data_path = data_path;
}
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!st.ok() || size <= 0) {
LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
}
```
##########
be/src/tools/builder_helper.h:
##########
@@ -0,0 +1,32 @@
+#pragma once
+#include "common/status.h"
+#include "io/fs/hdfs_file_system.h"
+
+namespace doris {
+
+class BuilderHelper {
+public:
+ static BuilderHelper* init_instance();
+ ~BuilderHelper();
+ // Return global instance.
+ static BuilderHelper* instance() { return _s_instance; }
+
+ void initial_build_env();
+ void open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS);
+ void upload_to_HDFS(std::string& read_path, std::string& write_path, void* bufferPtr);
+ void build();
+
+private:
+ BuilderHelper() {}
Review Comment:
warning: use '= default' to define a trivial default constructor [modernize-use-equals-default]
```suggestion
BuilderHelper() = default;
```
##########
be/src/tools/builder_scanner.h:
##########
@@ -0,0 +1,76 @@
+#pragma once
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+class BuilderScanner {
+public:
+ BuilderScanner(TabletSharedPtr tablet, const std::string& file_type, bool isHDFS);
+ ~BuilderScanner() {}
Review Comment:
warning: use '= default' to define a trivial destructor [modernize-use-equals-default]
```suggestion
~BuilderScanner() = default;
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (data_path.at(data_path.size() - 1) != '/') {
```
be/src/tools/builder_helper.cpp:105:
```diff
- else
+ } else
```
##########
be/src/tools/builder_scanner.cpp:
##########
@@ -0,0 +1,366 @@
+#include "tools/builder_scanner.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+
+BuilderScanner::BuilderScanner(TabletSharedPtr tablet, const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()), _tablet(tablet), _file_type(file_type), _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = 8192;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+}
+
+void BuilderScanner::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+int BuilderScanner::create_src_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 1;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ scalar_type.__set_type(TPrimitiveType::FLOAT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ scalar_type.__set_type(TPrimitiveType::DOUBLE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ scalar_type.__set_type(TPrimitiveType::DATETIME);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATETIMEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::TINYINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_UNSIGNED_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::SMALLINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ scalar_type.__set_type(TPrimitiveType::INT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL32);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL64);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL128I);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ scalar_type.__set_type(TPrimitiveType::JSONB);
+ break;
+ }
+ default:
+ LOG(FATAL) << "unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ // Skip the first 8 bytes These 8 bytes are used to indicate whether the field is a null value
+ slot_desc.byteOffset = i * 16 + 8;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ {
+ // TTupleDescriptor source
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_SRC;
+ //Here 8 bytes in order to handle null values
+ t_tuple_desc.byteSize = _tablet->num_columns() * 16 + 8;
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+int BuilderScanner::create_dst_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ int32_t byteOffset = 8;
+ for (int i = 0; i < _tablet->num_columns(); i++, byteOffset += 16) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 0;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ default:
+ LOG(FATAL) << " unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ slot_desc.byteOffset = byteOffset;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ t_desc_table.__isset.slotDescriptors = true;
+ {
+ // TTupleDescriptor dest
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_DST;
+ t_tuple_desc.byteSize = byteOffset + 8; //Here 8 bytes in order to handle null values
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+void BuilderScanner::init_desc_table() {
+ TDescriptorTable t_desc_table;
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ int next_slot_id = 0;
+
+ next_slot_id = create_dst_tuple(t_desc_table, next_slot_id);
+ next_slot_id = create_src_tuple(t_desc_table, next_slot_id);
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScanner::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i; // log_time id in src tuple
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScanner::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::string& path) {
+ LOG(INFO) << "build scan ranges for:" << path << " file_type:" << _file_type;
+ if (_isHDFS) {
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file_path;
+ range.file_type = TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+ }
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScanner::doSegmentBuild(const std::string& path) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!status.ok()) { LOG(FATAL) << "prepare scan node fail:" << status.to_string();
}
```
##########
be/src/tools/builder_scanner.cpp:
##########
@@ -0,0 +1,366 @@
+#include "tools/builder_scanner.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+
+BuilderScanner::BuilderScanner(TabletSharedPtr tablet, const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()), _tablet(tablet), _file_type(file_type), _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = 8192;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+}
+
+void BuilderScanner::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+int BuilderScanner::create_src_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 1;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ scalar_type.__set_type(TPrimitiveType::FLOAT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ scalar_type.__set_type(TPrimitiveType::DOUBLE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ scalar_type.__set_type(TPrimitiveType::DATETIME);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATETIMEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::TINYINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_UNSIGNED_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::SMALLINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ scalar_type.__set_type(TPrimitiveType::INT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL32);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL64);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL128I);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ scalar_type.__set_type(TPrimitiveType::JSONB);
+ break;
+ }
+ default:
+ LOG(FATAL) << "unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ // Skip the first 8 bytes These 8 bytes are used to indicate whether the field is a null value
+ slot_desc.byteOffset = i * 16 + 8;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ {
+ // TTupleDescriptor source
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_SRC;
+ //Here 8 bytes in order to handle null values
+ t_tuple_desc.byteSize = _tablet->num_columns() * 16 + 8;
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+int BuilderScanner::create_dst_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ int32_t byteOffset = 8;
+ for (int i = 0; i < _tablet->num_columns(); i++, byteOffset += 16) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 0;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ default:
+ LOG(FATAL) << " unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ slot_desc.byteOffset = byteOffset;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ t_desc_table.__isset.slotDescriptors = true;
+ {
+ // TTupleDescriptor dest
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_DST;
+ t_tuple_desc.byteSize = byteOffset + 8; //Here 8 bytes in order to handle null values
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+void BuilderScanner::init_desc_table() {
+ TDescriptorTable t_desc_table;
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ int next_slot_id = 0;
+
+ next_slot_id = create_dst_tuple(t_desc_table, next_slot_id);
+ next_slot_id = create_src_tuple(t_desc_table, next_slot_id);
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScanner::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i; // log_time id in src tuple
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScanner::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::string& path) {
+ LOG(INFO) << "build scan ranges for:" << path << " file_type:" << _file_type;
+ if (_isHDFS) {
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file_path;
+ range.file_type = TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+ }
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!ranges.size()) { LOG(FATAL) << "cannot get valid scan file!";
}
```
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
+ IOContext io_ctx;
+ io::FileReaderSPtr read_ptr = nullptr;
+ st = dfs->open_file(_meta_file, &read_ptr, &io_ctx);
+ if (!st.ok() || !read_ptr)
+ LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
+
+ buf.resize(size);
+
+ size_t read_size = 0;
+ read_ptr->read_at(0, Slice(buf.data(), size), io_ctx, &read_size);
+ if (read_size < size) {
+ LOG(FATAL) << "read meta file size abnormal.."; //<< dfs->GetLastError();
+ }
+
+ read_ptr->close();
+ } else {
+ buf = read_local_file(_meta_file);
+ }
+
+ FileHeader<TabletMetaPB> file_header;
+ buf = buf.substr(file_header.size());
+ // init tablet
+ TabletMeta* tablet_meta = new TabletMeta();
+ Status status = tablet_meta->deserialize(buf);
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to load tablet meta :" << status.to_string();
+ return;
+ }
+
+ LOG(INFO) << "table id:" << tablet_meta->table_id() << " tablet id:" << tablet_meta->tablet_id()
+ << " shard id:" << tablet_meta->shard_id();
+
+ auto data_dir = StorageEngine::instance()->get_store(_build_dir);
+ TabletMetaSharedPtr tablet_meta_ptr(tablet_meta);
+ TabletSharedPtr new_tablet = doris::Tablet::create_tablet_from_meta(tablet_meta_ptr, data_dir);
+ status = StorageEngine::instance()->tablet_manager()->_add_tablet_unlocked(
+ tablet_meta->tablet_id(), new_tablet, false, true);
+
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to add tablet to storage :" << status.to_string();
+ return;
+ }
+
+ std::vector<std::string> files;
+ if (_isHDFS) {
+ auto* dfs = hdfsFileSystem.get();
+ std::vector<io::Path> names;
+ dfs->list(_data_path, &names);
+ for (const auto& name : names) {
+ std::string filename = name.filename().c_str();
+ if (filename.substr(filename.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << name;
+ files.emplace_back(filename);
+ }
+ }
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(_data_path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << file_path;
+ files.emplace_back(file_path);
+ }
+ }
+ }
+
+ BuilderScannerMemtable scanner(new_tablet, _build_dir, _file_type, _isHDFS);
+ scanner.doSegmentBuild(files);
+
+ std::string local_new_header =
+ _build_dir + "/" + std::to_string(new_tablet->table_id()) + ".hdr";
+ TabletMetaSharedPtr new_tablet_meta = std::make_shared<TabletMeta>();
+ new_tablet->generate_tablet_meta_copy(new_tablet_meta);
+ new_tablet_meta->save(local_new_header);
+
+ if (_isHDFS) {
+ //upload segment to meta file directory
+ auto pos = _meta_file.rfind("/");
+ std::string remote_path = _meta_file.substr(0, pos + 1);
+ LOG(INFO) << "Upload to remote fs:" << remote_path;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(remote_path, &exist);
+ if (!st.ok() || !exist) {
+ st = dfs->create_directory(remote_path);
+ if (!st.ok()) LOG(INFO) << "fail to create directory:" << remote_path;
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!st.ok()) { LOG(INFO) << "fail to create directory:" << remote_path;
}
```
##########
be/src/tools/builder_scanner.cpp:
##########
@@ -0,0 +1,366 @@
+#include "tools/builder_scanner.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+
+BuilderScanner::BuilderScanner(TabletSharedPtr tablet, const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()), _tablet(tablet), _file_type(file_type), _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = 8192;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+}
+
+void BuilderScanner::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+int BuilderScanner::create_src_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 1;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ scalar_type.__set_type(TPrimitiveType::FLOAT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ scalar_type.__set_type(TPrimitiveType::DOUBLE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ scalar_type.__set_type(TPrimitiveType::DATETIME);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATETIMEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::TINYINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_UNSIGNED_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::SMALLINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ scalar_type.__set_type(TPrimitiveType::INT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL32);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL64);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL128I);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ scalar_type.__set_type(TPrimitiveType::JSONB);
+ break;
+ }
+ default:
+ LOG(FATAL) << "unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ // Skip the first 8 bytes These 8 bytes are used to indicate whether the field is a null value
+ slot_desc.byteOffset = i * 16 + 8;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ {
+ // TTupleDescriptor source
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_SRC;
+ //Here 8 bytes in order to handle null values
+ t_tuple_desc.byteSize = _tablet->num_columns() * 16 + 8;
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+int BuilderScanner::create_dst_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ int32_t byteOffset = 8;
+ for (int i = 0; i < _tablet->num_columns(); i++, byteOffset += 16) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 0;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ default:
+ LOG(FATAL) << " unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ slot_desc.byteOffset = byteOffset;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ t_desc_table.__isset.slotDescriptors = true;
+ {
+ // TTupleDescriptor dest
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_DST;
+ t_tuple_desc.byteSize = byteOffset + 8; //Here 8 bytes in order to handle null values
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+void BuilderScanner::init_desc_table() {
+ TDescriptorTable t_desc_table;
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ int next_slot_id = 0;
+
+ next_slot_id = create_dst_tuple(t_desc_table, next_slot_id);
+ next_slot_id = create_src_tuple(t_desc_table, next_slot_id);
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScanner::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i; // log_time id in src tuple
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScanner::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::string& path) {
+ LOG(INFO) << "build scan ranges for:" << path << " file_type:" << _file_type;
+ if (_isHDFS) {
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file_path;
+ range.file_type = TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+ }
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScanner::doSegmentBuild(const std::string& path) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, path);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!status.ok()) { LOG(FATAL) << "open scan node fail:" << status.to_string();
}
```
##########
be/src/tools/builder_scanner_memtable.cpp:
##########
@@ -0,0 +1,346 @@
+#include "tools/builder_scanner_memtable.h"
+
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Types_types.h"
+#include "hdfs/hdfs.h"
+#include "olap/delta_writer.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+static const int BATCH_SIZE = 8192;
+
+BuilderScannerMemtable::BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()),
+ _tablet(tablet),
+ _build_dir(build_dir),
+ _file_type(file_type),
+ _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = BATCH_SIZE;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+ _runtime_state.init_mem_trackers(uid);
+}
+
+void BuilderScannerMemtable::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+TPrimitiveType::type BuilderScannerMemtable::getPrimitiveType(FieldType t) {
+ switch (t) {
+ case FieldType::OLAP_FIELD_TYPE_OBJECT: {
+ return TPrimitiveType::OBJECT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_HLL: {
+ return TPrimitiveType::HLL;
+ }
+ case FieldType::OLAP_FIELD_TYPE_CHAR: {
+ return TPrimitiveType::CHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ return TPrimitiveType::VARCHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_STRING: {
+ return TPrimitiveType::STRING;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ return TPrimitiveType::DATE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ return TPrimitiveType::DATETIME;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ return TPrimitiveType::DATEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ return TPrimitiveType::DATETIMEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ return TPrimitiveType::DECIMAL32;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ return TPrimitiveType::DECIMAL64;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ return TPrimitiveType::DECIMAL128I;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ return TPrimitiveType::JSONB;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ return TPrimitiveType::BOOLEAN;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ return TPrimitiveType::TINYINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT: {
+ return TPrimitiveType::SMALLINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ return TPrimitiveType::INT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ return TPrimitiveType::BIGINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT: {
+ return TPrimitiveType::LARGEINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ return TPrimitiveType::FLOAT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ return TPrimitiveType::DOUBLE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ return TPrimitiveType::ARRAY;
+ }
+ default: {
+ LOG(FATAL) << "unknown type error:" << t;
+ }
+ }
+}
+TDescriptorTable BuilderScannerMemtable::create_descriptor_tablet() {
+ TDescriptorTableBuilder dtb;
+
+ // build DST table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ // build SRC table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ return dtb.desc_tbl();
+}
+
+void BuilderScannerMemtable::init_desc_table() {
+ TDescriptorTable t_desc_table = create_descriptor_tablet();
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScannerMemtable::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i;
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScannerMemtable::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::vector<std::string>& files) {
+ LOG(INFO) << "build scan ranges for files size:" << files.size() << " file_type:" << _file_type;
+ for (const auto& file : files) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file;
+ range.file_type = _isHDFS ? TFileType::FILE_HDFS : TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScannerMemtable::doSegmentBuild(const std::vector<std::string>& files) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, files);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
+
+ // std::unique_ptr<RowsetWriter> rowset_writer;
+ PUniqueId load_id;
+ load_id.set_hi(1);
+ load_id.set_lo(1);
+ int64_t transaction_id = 1;
+
+ // delta writer
+ TupleDescriptor* tuple_desc = _desc_tbl->get_tuple_descriptor(TUPLE_ID_DST);
+ WriteRequest write_req = {_tablet->tablet_meta()->tablet_id(),
+ _tablet->schema_hash(),
+ WriteType::LOAD,
+ transaction_id,
+ _tablet->partition_id(),
+ load_id,
+ tuple_desc,
+ &(tuple_desc->slots())};
+
+ DeltaWriter* delta_writer = nullptr;
+ DeltaWriter::open(&write_req, &delta_writer, load_id);
+ status = delta_writer->init();
+ if (!status.ok()) LOG(FATAL) << "delta_writer init fail:" << status.to_string();
+
+ std::filesystem::path segment_path(std::filesystem::path(_build_dir + "/segment"));
+ std::filesystem::remove_all(segment_path);
+ if (!std::filesystem::create_directory(segment_path)) LOG(FATAL) << "create segment path fail.";
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!std::filesystem::create_directory(segment_path)) { LOG(FATAL) << "create segment path fail.";
}
```
##########
be/src/tools/builder_scanner_memtable.cpp:
##########
@@ -0,0 +1,346 @@
+#include "tools/builder_scanner_memtable.h"
+
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Types_types.h"
+#include "hdfs/hdfs.h"
+#include "olap/delta_writer.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+static const int BATCH_SIZE = 8192;
+
+BuilderScannerMemtable::BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()),
+ _tablet(tablet),
+ _build_dir(build_dir),
+ _file_type(file_type),
+ _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = BATCH_SIZE;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+ _runtime_state.init_mem_trackers(uid);
+}
+
+void BuilderScannerMemtable::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+TPrimitiveType::type BuilderScannerMemtable::getPrimitiveType(FieldType t) {
+ switch (t) {
+ case FieldType::OLAP_FIELD_TYPE_OBJECT: {
+ return TPrimitiveType::OBJECT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_HLL: {
+ return TPrimitiveType::HLL;
+ }
+ case FieldType::OLAP_FIELD_TYPE_CHAR: {
+ return TPrimitiveType::CHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ return TPrimitiveType::VARCHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_STRING: {
+ return TPrimitiveType::STRING;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ return TPrimitiveType::DATE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ return TPrimitiveType::DATETIME;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ return TPrimitiveType::DATEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ return TPrimitiveType::DATETIMEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ return TPrimitiveType::DECIMAL32;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ return TPrimitiveType::DECIMAL64;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ return TPrimitiveType::DECIMAL128I;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ return TPrimitiveType::JSONB;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ return TPrimitiveType::BOOLEAN;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ return TPrimitiveType::TINYINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT: {
+ return TPrimitiveType::SMALLINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ return TPrimitiveType::INT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ return TPrimitiveType::BIGINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT: {
+ return TPrimitiveType::LARGEINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ return TPrimitiveType::FLOAT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ return TPrimitiveType::DOUBLE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ return TPrimitiveType::ARRAY;
+ }
+ default: {
+ LOG(FATAL) << "unknown type error:" << t;
+ }
+ }
+}
+TDescriptorTable BuilderScannerMemtable::create_descriptor_tablet() {
+ TDescriptorTableBuilder dtb;
+
+ // build DST table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ // build SRC table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ return dtb.desc_tbl();
+}
+
+void BuilderScannerMemtable::init_desc_table() {
+ TDescriptorTable t_desc_table = create_descriptor_tablet();
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScannerMemtable::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i;
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScannerMemtable::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::vector<std::string>& files) {
+ LOG(INFO) << "build scan ranges for files size:" << files.size() << " file_type:" << _file_type;
+ for (const auto& file : files) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file;
+ range.file_type = _isHDFS ? TFileType::FILE_HDFS : TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!ranges.size()) { LOG(FATAL) << "cannot get valid scan file!";
}
```
##########
be/src/tools/builder_scanner_memtable.h:
##########
@@ -0,0 +1,68 @@
+#pragma once
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/primitive_type.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+
+namespace doris {
+
+class BuilderScannerMemtable {
+public:
+ BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS);
+ ~BuilderScannerMemtable() {}
Review Comment:
warning: use '= default' to define a trivial destructor [modernize-use-equals-default]
```suggestion
~BuilderScannerMemtable() = default;
```
##########
be/src/tools/builder_scanner.cpp:
##########
@@ -0,0 +1,366 @@
+#include "tools/builder_scanner.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+
+BuilderScanner::BuilderScanner(TabletSharedPtr tablet, const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()), _tablet(tablet), _file_type(file_type), _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = 8192;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+}
+
+void BuilderScanner::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+int BuilderScanner::create_src_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 1;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ scalar_type.__set_type(TPrimitiveType::FLOAT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ scalar_type.__set_type(TPrimitiveType::DOUBLE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ scalar_type.__set_type(TPrimitiveType::DATETIME);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ scalar_type.__set_type(TPrimitiveType::DATETIMEV2);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::TINYINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_UNSIGNED_TINYINT: {
+ scalar_type.__set_type(TPrimitiveType::SMALLINT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ scalar_type.__set_type(TPrimitiveType::INT);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL32);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL64);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ scalar_type.__set_type(TPrimitiveType::DECIMAL128I);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ scalar_type.__set_type(TPrimitiveType::JSONB);
+ break;
+ }
+ default:
+ LOG(FATAL) << "unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ // Skip the first 8 bytes These 8 bytes are used to indicate whether the field is a null value
+ slot_desc.byteOffset = i * 16 + 8;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ {
+ // TTupleDescriptor source
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_SRC;
+ //Here 8 bytes in order to handle null values
+ t_tuple_desc.byteSize = _tablet->num_columns() * 16 + 8;
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+int BuilderScanner::create_dst_tuple(TDescriptorTable& t_desc_table, int next_slot_id) {
+ int32_t byteOffset = 8;
+ for (int i = 0; i < _tablet->num_columns(); i++, byteOffset += 16) {
+ TSlotDescriptor slot_desc;
+
+ slot_desc.id = next_slot_id++;
+ slot_desc.parent = 0;
+ const auto& col = _tablet->tablet_schema()->column(i);
+ TTypeDesc type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ switch (col.type()) {
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ scalar_type.__set_type(TPrimitiveType::DATE);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ break;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ scalar_type.__set_type(TPrimitiveType::BIGINT);
+ break;
+ }
+ default:
+ LOG(FATAL) << " unknown type error:" << col.type();
+ }
+
+ scalar_type.__set_len(col.length());
+ node.__set_scalar_type(scalar_type);
+ type.types.push_back(node);
+ }
+ slot_desc.slotType = type;
+ slot_desc.columnPos = i;
+ slot_desc.byteOffset = byteOffset;
+ slot_desc.nullIndicatorByte = i / 8;
+ slot_desc.nullIndicatorBit = i % 8;
+ slot_desc.colName = _tablet->tablet_schema()->column(i).name();
+ slot_desc.slotIdx = i + 1;
+ slot_desc.isMaterialized = true;
+
+ t_desc_table.slotDescriptors.push_back(slot_desc);
+ }
+
+ t_desc_table.__isset.slotDescriptors = true;
+ {
+ // TTupleDescriptor dest
+ TTupleDescriptor t_tuple_desc;
+ t_tuple_desc.id = TUPLE_ID_DST;
+ t_tuple_desc.byteSize = byteOffset + 8; //Here 8 bytes in order to handle null values
+ t_tuple_desc.numNullBytes = 0;
+ t_tuple_desc.tableId = _tablet->table_id();
+ t_tuple_desc.__isset.tableId = true;
+ t_desc_table.tupleDescriptors.push_back(t_tuple_desc);
+ }
+ return next_slot_id;
+}
+
+void BuilderScanner::init_desc_table() {
+ TDescriptorTable t_desc_table;
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ int next_slot_id = 0;
+
+ next_slot_id = create_dst_tuple(t_desc_table, next_slot_id);
+ next_slot_id = create_src_tuple(t_desc_table, next_slot_id);
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScanner::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i; // log_time id in src tuple
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScanner::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::string& path) {
+ LOG(INFO) << "build scan ranges for:" << path << " file_type:" << _file_type;
+ if (_isHDFS) {
+ } else {
+ for (const auto& file : std::filesystem::directory_iterator(path)) {
+ auto file_path = file.path().string();
+ if (file_path.substr(file_path.size() - _file_type.size()) == _file_type) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file_path;
+ range.file_type = TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+ }
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScanner::doSegmentBuild(const std::string& path) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, path);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
+
+ std::unique_ptr<RowsetWriter> rowset_writer;
+ PUniqueId load_id;
+ load_id.set_hi(1);
+ load_id.set_lo(1);
+ int64_t transaction_id = 1;
+
+ RowsetWriterContext context;
+ context.txn_id = transaction_id;
+ context.load_id = load_id;
+ context.rowset_state = PREPARED;
+ context.segments_overlap = OVERLAP_UNKNOWN;
+ context.tablet_schema = _tablet->tablet_schema();
+
+ status = _tablet->create_rowset_writer(context, &rowset_writer);
+
+ if (!status.ok()) LOG(FATAL) << "create rowset error:" << status.to_string();
+
+ std::filesystem::path segment_path(std::filesystem::path(path + "/segment"));
+ if (!std::filesystem::exists(segment_path)) {
+ LOG(INFO) << "create segment path.";
+ if (!std::filesystem::create_directory(segment_path))
+ LOG(FATAL) << "create segment path fail.";
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!std::filesystem::create_directory(segment_path)) {
LOG(FATAL) << "create segment path fail.";
}
```
##########
be/src/tools/builder_scanner_memtable.cpp:
##########
@@ -0,0 +1,346 @@
+#include "tools/builder_scanner_memtable.h"
+
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Types_types.h"
+#include "hdfs/hdfs.h"
+#include "olap/delta_writer.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+static const int BATCH_SIZE = 8192;
+
+BuilderScannerMemtable::BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()),
+ _tablet(tablet),
+ _build_dir(build_dir),
+ _file_type(file_type),
+ _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = BATCH_SIZE;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+ _runtime_state.init_mem_trackers(uid);
+}
+
+void BuilderScannerMemtable::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+TPrimitiveType::type BuilderScannerMemtable::getPrimitiveType(FieldType t) {
+ switch (t) {
+ case FieldType::OLAP_FIELD_TYPE_OBJECT: {
+ return TPrimitiveType::OBJECT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_HLL: {
+ return TPrimitiveType::HLL;
+ }
+ case FieldType::OLAP_FIELD_TYPE_CHAR: {
+ return TPrimitiveType::CHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ return TPrimitiveType::VARCHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_STRING: {
+ return TPrimitiveType::STRING;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ return TPrimitiveType::DATE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ return TPrimitiveType::DATETIME;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ return TPrimitiveType::DATEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ return TPrimitiveType::DATETIMEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ return TPrimitiveType::DECIMAL32;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ return TPrimitiveType::DECIMAL64;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ return TPrimitiveType::DECIMAL128I;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ return TPrimitiveType::JSONB;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ return TPrimitiveType::BOOLEAN;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ return TPrimitiveType::TINYINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT: {
+ return TPrimitiveType::SMALLINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ return TPrimitiveType::INT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ return TPrimitiveType::BIGINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT: {
+ return TPrimitiveType::LARGEINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ return TPrimitiveType::FLOAT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ return TPrimitiveType::DOUBLE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ return TPrimitiveType::ARRAY;
+ }
+ default: {
+ LOG(FATAL) << "unknown type error:" << t;
+ }
+ }
+}
+TDescriptorTable BuilderScannerMemtable::create_descriptor_tablet() {
+ TDescriptorTableBuilder dtb;
+
+ // build DST table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ // build SRC table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ return dtb.desc_tbl();
+}
+
+void BuilderScannerMemtable::init_desc_table() {
+ TDescriptorTable t_desc_table = create_descriptor_tablet();
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScannerMemtable::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i;
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScannerMemtable::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::vector<std::string>& files) {
+ LOG(INFO) << "build scan ranges for files size:" << files.size() << " file_type:" << _file_type;
+ for (const auto& file : files) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file;
+ range.file_type = _isHDFS ? TFileType::FILE_HDFS : TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScannerMemtable::doSegmentBuild(const std::vector<std::string>& files) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
if (!status.ok()) { LOG(FATAL) << "prepare scan node fail:" << status.to_string();
}
```
##########
be/src/tools/builder_scanner_memtable.cpp:
##########
@@ -0,0 +1,346 @@
+#include "tools/builder_scanner_memtable.h"
+
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Types_types.h"
+#include "hdfs/hdfs.h"
+#include "olap/delta_writer.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+static const int BATCH_SIZE = 8192;
+
+BuilderScannerMemtable::BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()),
+ _tablet(tablet),
+ _build_dir(build_dir),
+ _file_type(file_type),
+ _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = BATCH_SIZE;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+ _runtime_state.init_mem_trackers(uid);
+}
+
+void BuilderScannerMemtable::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+TPrimitiveType::type BuilderScannerMemtable::getPrimitiveType(FieldType t) {
+ switch (t) {
+ case FieldType::OLAP_FIELD_TYPE_OBJECT: {
+ return TPrimitiveType::OBJECT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_HLL: {
+ return TPrimitiveType::HLL;
+ }
+ case FieldType::OLAP_FIELD_TYPE_CHAR: {
+ return TPrimitiveType::CHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ return TPrimitiveType::VARCHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_STRING: {
+ return TPrimitiveType::STRING;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ return TPrimitiveType::DATE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ return TPrimitiveType::DATETIME;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ return TPrimitiveType::DATEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ return TPrimitiveType::DATETIMEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ return TPrimitiveType::DECIMAL32;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ return TPrimitiveType::DECIMAL64;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ return TPrimitiveType::DECIMAL128I;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ return TPrimitiveType::JSONB;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ return TPrimitiveType::BOOLEAN;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ return TPrimitiveType::TINYINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT: {
+ return TPrimitiveType::SMALLINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ return TPrimitiveType::INT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ return TPrimitiveType::BIGINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT: {
+ return TPrimitiveType::LARGEINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ return TPrimitiveType::FLOAT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ return TPrimitiveType::DOUBLE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ return TPrimitiveType::ARRAY;
+ }
+ default: {
+ LOG(FATAL) << "unknown type error:" << t;
+ }
+ }
+}
+TDescriptorTable BuilderScannerMemtable::create_descriptor_tablet() {
+ TDescriptorTableBuilder dtb;
+
+ // build DST table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ // build SRC table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ return dtb.desc_tbl();
+}
+
+void BuilderScannerMemtable::init_desc_table() {
+ TDescriptorTable t_desc_table = create_descriptor_tablet();
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScannerMemtable::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i;
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScannerMemtable::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::vector<std::string>& files) {
+ LOG(INFO) << "build scan ranges for files size:" << files.size() << " file_type:" << _file_type;
+ for (const auto& file : files) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file;
+ range.file_type = _isHDFS ? TFileType::FILE_HDFS : TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScannerMemtable::doSegmentBuild(const std::vector<std::string>& files) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, files);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
+
+ // std::unique_ptr<RowsetWriter> rowset_writer;
+ PUniqueId load_id;
+ load_id.set_hi(1);
+ load_id.set_lo(1);
+ int64_t transaction_id = 1;
+
+ // delta writer
+ TupleDescriptor* tuple_desc = _desc_tbl->get_tuple_descriptor(TUPLE_ID_DST);
+ WriteRequest write_req = {_tablet->tablet_meta()->tablet_id(),
+ _tablet->schema_hash(),
+ WriteType::LOAD,
+ transaction_id,
+ _tablet->partition_id(),
+ load_id,
+ tuple_desc,
+ &(tuple_desc->slots())};
+
+ DeltaWriter* delta_writer = nullptr;
+ DeltaWriter::open(&write_req, &delta_writer, load_id);
+ status = delta_writer->init();
+ if (!status.ok()) LOG(FATAL) << "delta_writer init fail:" << status.to_string();
+
+ std::filesystem::path segment_path(std::filesystem::path(_build_dir + "/segment"));
+ std::filesystem::remove_all(segment_path);
+ if (!std::filesystem::create_directory(segment_path)) LOG(FATAL) << "create segment path fail.";
+
+ delta_writer->set_writer_path(segment_path.string());
+ // Get block
+ vectorized::Block block;
+ bool eof = false;
+
+ std::vector<int> rowidx;
+ for (size_t i = 0; i < BATCH_SIZE; ++i) {
+ rowidx.push_back(i);
+ }
+
+ while (!eof) {
+ status = scan_node.get_next(&_runtime_state, &block, &eof);
+ if (!status.ok()) {
+ LOG(FATAL) << "scan error: " << status.to_string();
+ break;
+ }
+
+ if (block.rows() != BATCH_SIZE) {
+ std::vector<int> index;
+ for (size_t i = 0; i < block.rows(); ++i) {
+ index.push_back(i);
+ }
+ status = delta_writer->write(&block, index);
+ } else
+ status = delta_writer->write(&block, rowidx);
Review Comment:
warning: statement should be inside braces [readability-braces-around-statements]
```suggestion
} else {
status = delta_writer->write(&block, rowidx);
}
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] spaces-X commented on a diff in pull request #15839: [Feature] support segment builder tool
Posted by GitBox <gi...@apache.org>.
spaces-X commented on code in PR #15839:
URL: https://github.com/apache/doris/pull/15839#discussion_r1067656119
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
+ IOContext io_ctx;
+ io::FileReaderSPtr read_ptr = nullptr;
+ st = dfs->open_file(_meta_file, &read_ptr, &io_ctx);
+ if (!st.ok() || !read_ptr)
+ LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
+
+ buf.resize(size);
+
+ size_t read_size = 0;
+ read_ptr->read_at(0, Slice(buf.data(), size), io_ctx, &read_size);
+ if (read_size < size) {
+ LOG(FATAL) << "read meta file size abnormal.."; //<< dfs->GetLastError();
+ }
+
+ read_ptr->close();
+ } else {
+ buf = read_local_file(_meta_file);
+ }
+
+ FileHeader<TabletMetaPB> file_header;
+ buf = buf.substr(file_header.size());
+ // init tablet
+ TabletMeta* tablet_meta = new TabletMeta();
+ Status status = tablet_meta->deserialize(buf);
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to load tablet meta :" << status.to_string();
+ return;
+ }
+
+ LOG(INFO) << "table id:" << tablet_meta->table_id() << " tablet id:" << tablet_meta->tablet_id()
+ << " shard id:" << tablet_meta->shard_id();
+
+ auto data_dir = StorageEngine::instance()->get_store(_build_dir);
+ TabletMetaSharedPtr tablet_meta_ptr(tablet_meta);
+ TabletSharedPtr new_tablet = doris::Tablet::create_tablet_from_meta(tablet_meta_ptr, data_dir);
+ status = StorageEngine::instance()->tablet_manager()->_add_tablet_unlocked(
+ tablet_meta->tablet_id(), new_tablet, false, true);
+
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to add tablet to storage :" << status.to_string();
+ return;
+ }
+
+ std::vector<std::string> files;
+ if (_isHDFS) {
+ auto* dfs = hdfsFileSystem.get();
+ std::vector<io::Path> names;
+ dfs->list(_data_path, &names);
+ for (const auto& name : names) {
+ std::string filename = name.filename().c_str();
+ if (filename.substr(filename.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << name;
+ files.emplace_back(filename);
+ }
Review Comment:
suggest to add some warning log when the file type are not matched both in hdfs and local dir.
##########
be/src/tools/builder_scanner_memtable.cpp:
##########
@@ -0,0 +1,346 @@
+#include "tools/builder_scanner_memtable.h"
+
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/Types_types.h"
+#include "hdfs/hdfs.h"
+#include "olap/delta_writer.h"
+#include "runtime/descriptor_helper.h"
+#include "runtime/descriptors.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+
+static const int TUPLE_ID_DST = 0;
+static const int TUPLE_ID_SRC = 1;
+static const int BATCH_SIZE = 8192;
+
+BuilderScannerMemtable::BuilderScannerMemtable(TabletSharedPtr tablet, const std::string& build_dir,
+ const std::string& file_type, bool isHDFS)
+ : _runtime_state(TQueryGlobals()),
+ _tablet(tablet),
+ _build_dir(build_dir),
+ _file_type(file_type),
+ _isHDFS(isHDFS) {
+ init();
+ _runtime_state.init_scanner_mem_trackers();
+ TUniqueId uid;
+ uid.hi = 1;
+ uid.lo = 1;
+ TQueryOptions _options;
+ _options.batch_size = BATCH_SIZE;
+ _options.enable_vectorized_engine = true;
+ auto* _exec_env = ExecEnv::GetInstance();
+ _runtime_state.init(uid, _options, TQueryGlobals(), _exec_env);
+ _runtime_state.init_mem_trackers(uid);
+}
+
+void BuilderScannerMemtable::init() {
+ create_expr_info();
+ init_desc_table();
+
+ // Node Id
+ _tnode.node_id = 0;
+ _tnode.node_type = TPlanNodeType::SCHEMA_SCAN_NODE;
+ _tnode.num_children = 0;
+ _tnode.limit = -1;
+ _tnode.row_tuples.push_back(0);
+ _tnode.nullable_tuples.push_back(false);
+ _tnode.broker_scan_node.tuple_id = 0;
+ _tnode.__isset.broker_scan_node = true;
+}
+
+TPrimitiveType::type BuilderScannerMemtable::getPrimitiveType(FieldType t) {
+ switch (t) {
+ case FieldType::OLAP_FIELD_TYPE_OBJECT: {
+ return TPrimitiveType::OBJECT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_HLL: {
+ return TPrimitiveType::HLL;
+ }
+ case FieldType::OLAP_FIELD_TYPE_CHAR: {
+ return TPrimitiveType::CHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_VARCHAR: {
+ return TPrimitiveType::VARCHAR;
+ }
+ case FieldType::OLAP_FIELD_TYPE_STRING: {
+ return TPrimitiveType::STRING;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ return TPrimitiveType::DATE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ return TPrimitiveType::DATETIME;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ return TPrimitiveType::DATEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: {
+ return TPrimitiveType::DATETIMEV2;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL:
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ return TPrimitiveType::DECIMAL32;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ return TPrimitiveType::DECIMAL64;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ return TPrimitiveType::DECIMAL128I;
+ }
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ return TPrimitiveType::JSONB;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ return TPrimitiveType::BOOLEAN;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT: {
+ return TPrimitiveType::TINYINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT: {
+ return TPrimitiveType::SMALLINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_INT: {
+ return TPrimitiveType::INT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_BIGINT: {
+ return TPrimitiveType::BIGINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT: {
+ return TPrimitiveType::LARGEINT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_FLOAT: {
+ return TPrimitiveType::FLOAT;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE: {
+ return TPrimitiveType::DOUBLE;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ return TPrimitiveType::ARRAY;
+ }
+ default: {
+ LOG(FATAL) << "unknown type error:" << t;
+ }
+ }
+}
+TDescriptorTable BuilderScannerMemtable::create_descriptor_tablet() {
+ TDescriptorTableBuilder dtb;
+
+ // build DST table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ // build SRC table descriptor
+ {
+ TTupleDescriptorBuilder tuple_builder;
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ const auto& col = _tablet->tablet_schema()->column(i);
+
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(thrift_to_type(getPrimitiveType(col.type())))
+ .column_name(col.name())
+ .column_pos(i)
+ .length(col.length())
+ .build());
+ }
+ tuple_builder.build(&dtb);
+ }
+
+ return dtb.desc_tbl();
+}
+
+void BuilderScannerMemtable::init_desc_table() {
+ TDescriptorTable t_desc_table = create_descriptor_tablet();
+
+ // table descriptors
+ TTableDescriptor t_table_desc;
+
+ t_table_desc.id = _tablet->table_id();
+ t_table_desc.tableType = TTableType::OLAP_TABLE;
+ t_table_desc.numCols = _tablet->num_columns();
+ t_table_desc.numClusteringCols = 0;
+ t_desc_table.tableDescriptors.push_back(t_table_desc);
+ t_desc_table.__isset.tableDescriptors = true;
+
+ DescriptorTbl::create(&_obj_pool, t_desc_table, &_desc_tbl);
+
+ _runtime_state.set_desc_tbl(_desc_tbl);
+}
+
+void BuilderScannerMemtable::create_expr_info() {
+ TTypeDesc varchar_type;
+ {
+ TTypeNode node;
+ node.__set_type(TTypeNodeType::SCALAR);
+ TScalarType scalar_type;
+ scalar_type.__set_type(TPrimitiveType::VARCHAR);
+ scalar_type.__set_len(65535);
+ node.__set_scalar_type(scalar_type);
+ varchar_type.types.push_back(node);
+ }
+ for (int i = 0; i < _tablet->num_columns(); i++) {
+ auto col = _tablet->tablet_schema()->column(i);
+
+ TExprNode slot_ref;
+ slot_ref.node_type = TExprNodeType::SLOT_REF;
+ slot_ref.type = varchar_type;
+ slot_ref.num_children = 0;
+ slot_ref.__isset.slot_ref = true;
+ slot_ref.slot_ref.slot_id = _tablet->num_columns() + i;
+ slot_ref.slot_ref.tuple_id = 1;
+
+ TExpr expr;
+ expr.nodes.push_back(slot_ref);
+
+ _params.expr_of_dest_slot.emplace(i, expr);
+ _params.src_slot_ids.push_back(_tablet->num_columns() + i);
+ }
+
+ // _params.__isset.expr_of_dest_slot = true;
+ _params.__set_dest_tuple_id(TUPLE_ID_DST);
+ _params.__set_src_tuple_id(TUPLE_ID_SRC);
+}
+
+void BuilderScannerMemtable::build_scan_ranges(std::vector<TBrokerRangeDesc>& ranges,
+ const std::vector<std::string>& files) {
+ LOG(INFO) << "build scan ranges for files size:" << files.size() << " file_type:" << _file_type;
+ for (const auto& file : files) {
+ TBrokerRangeDesc range;
+ range.start_offset = 0;
+ range.size = -1;
+ range.format_type = TFileFormatType::FORMAT_PARQUET;
+ range.splittable = true;
+
+ range.path = file;
+ range.file_type = _isHDFS ? TFileType::FILE_HDFS : TFileType::FILE_LOCAL;
+ ranges.push_back(range);
+ }
+
+ if (!ranges.size()) LOG(FATAL) << "cannot get valid scan file!";
+}
+
+void BuilderScannerMemtable::doSegmentBuild(const std::vector<std::string>& files) {
+ vectorized::VBrokerScanNode scan_node(&_obj_pool, _tnode, *_desc_tbl);
+ scan_node.init(_tnode);
+ auto status = scan_node.prepare(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "prepare scan node fail:" << status.to_string();
+
+ // set scan range
+ std::vector<TScanRangeParams> scan_ranges;
+ {
+ TScanRangeParams scan_range_params;
+
+ TBrokerScanRange broker_scan_range;
+ broker_scan_range.params = _params;
+ build_scan_ranges(broker_scan_range.ranges, files);
+ scan_range_params.scan_range.__set_broker_scan_range(broker_scan_range);
+ scan_ranges.push_back(scan_range_params);
+ }
+
+ scan_node.set_scan_ranges(scan_ranges);
+ status = scan_node.open(&_runtime_state);
+ if (!status.ok()) LOG(FATAL) << "open scan node fail:" << status.to_string();
+
+ // std::unique_ptr<RowsetWriter> rowset_writer;
+ PUniqueId load_id;
+ load_id.set_hi(1);
+ load_id.set_lo(1);
+ int64_t transaction_id = 1;
+
+ // delta writer
+ TupleDescriptor* tuple_desc = _desc_tbl->get_tuple_descriptor(TUPLE_ID_DST);
+ WriteRequest write_req = {_tablet->tablet_meta()->tablet_id(),
+ _tablet->schema_hash(),
+ WriteType::LOAD,
+ transaction_id,
+ _tablet->partition_id(),
+ load_id,
+ tuple_desc,
+ &(tuple_desc->slots())};
+
+ DeltaWriter* delta_writer = nullptr;
+ DeltaWriter::open(&write_req, &delta_writer, load_id);
+ status = delta_writer->init();
+ if (!status.ok()) LOG(FATAL) << "delta_writer init fail:" << status.to_string();
+
+ std::filesystem::path segment_path(std::filesystem::path(_build_dir + "/segment"));
+ std::filesystem::remove_all(segment_path);
Review Comment:
Nit::
Can we guarantee that the `segment_path` is always a new empty directory ?
If so it had better to failed rather than `remove all` when there are files under this path.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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] [doris] spaces-X commented on a diff in pull request #15839: [Feature] support segment builder tool
Posted by GitBox <gi...@apache.org>.
spaces-X commented on code in PR #15839:
URL: https://github.com/apache/doris/pull/15839#discussion_r1067656119
##########
be/src/tools/builder_helper.cpp:
##########
@@ -0,0 +1,295 @@
+#include "tools/builder_helper.h"
+
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <string>
+
+#include "common/object_pool.h"
+#include "common/status.h"
+#include "env/env.h"
+#include "exec/parquet_scanner.h"
+#include "exprs/cast_functions.h"
+#include "gen_cpp/Descriptors_types.h"
+#include "gen_cpp/PaloInternalService_types.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/olap_file.pb.h"
+#include "gen_cpp/segment_v2.pb.h"
+#include "gutil/strings/numbers.h"
+#include "gutil/strings/split.h"
+#include "gutil/strings/substitute.h"
+#include "io/buffered_reader.h"
+#include "io/file_factory.h"
+#include "io/file_reader.h"
+#include "io/local_file_reader.h"
+#include "json2pb/pb_to_json.h"
+#include "olap/data_dir.h"
+#include "olap/file_helper.h"
+#include "olap/olap_define.h"
+#include "olap/options.h"
+#include "olap/row.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset_id_generator.h"
+#include "olap/rowset/rowset_meta_manager.h"
+#include "olap/rowset/segment_v2/binary_plain_page.h"
+#include "olap/rowset/segment_v2/column_reader.h"
+#include "olap/schema_change.h"
+#include "olap/storage_engine.h"
+#include "olap/storage_policy_mgr.h"
+#include "olap/tablet.h"
+#include "olap/tablet_meta.h"
+#include "olap/tablet_meta_manager.h"
+#include "olap/tablet_schema.h"
+#include "olap/tablet_schema_cache.h"
+#include "olap/utils.h"
+#include "runtime/descriptors.h"
+#include "runtime/exec_env.h"
+#include "runtime/row_batch.h"
+#include "runtime/runtime_state.h"
+#include "runtime/tuple.h"
+#include "runtime/user_function_cache.h"
+#include "tools/builder_scanner.h"
+#include "tools/builder_scanner_memtable.h"
+#include "util/coding.h"
+#include "util/crc32c.h"
+#include "util/disk_info.h"
+#include "util/file_utils.h"
+#include "util/runtime_profile.h"
+#include "util/time.h"
+#include "vec/exec/format/parquet/vparquet_file_metadata.h"
+#include "vec/exec/vbroker_scan_node.h"
+
+namespace doris {
+#define BUFFER_SIZE 1048576
+
+BuilderHelper* BuilderHelper::_s_instance = nullptr;
+
+BuilderHelper* BuilderHelper::init_instance() {
+ // DCHECK(_s_instance == nullptr);
+ static BuilderHelper instance;
+ _s_instance = &instance;
+ return _s_instance;
+}
+
+void BuilderHelper::initial_build_env() {
+ char doris_home[] = "DORIS_HOME=/tmp";
+ putenv(doris_home);
+
+ if (!doris::config::init(nullptr, true, false, true)) {
+ LOG(FATAL) << "init config fail";
+ exit(-1);
+ }
+ CpuInfo::init();
+ DiskInfo::init();
+ MemInfo::init();
+
+ // write buffer size before flush
+ config::write_buffer_size = 14097152000;
+ // max buffer size used in memtable for the aggregated table
+ config::memtable_max_buffer_size = 18194304000;
+ doris::thread_context()->thread_mem_tracker_mgr->set_check_limit(false);
+ doris::TabletSchemaCache::create_global_schema_cache();
+ doris::ChunkAllocator::init_instance(4096);
+
+ doris::MemInfo::init();
+}
+
+void BuilderHelper::open(const std::string& meta_file, const std::string& build_dir,
+ const std::string& data_path, const std::string& file_type, bool isHDFS) {
+ _meta_file = meta_file;
+ _build_dir = build_dir;
+ if (data_path.at(data_path.size() - 1) != '/')
+ _data_path = data_path + "/";
+ else
+ _data_path = data_path;
+
+ _file_type = file_type;
+ _isHDFS = isHDFS;
+
+ // TODO: need adapt for open HDFS
+ if (_isHDFS) {
+ THdfsParams hdfs_params;
+ hdfsFileSystem = std::make_unique<io::HdfsFileSystem>(hdfs_params, _data_path);
+ }
+
+ std::filesystem::path dir_path(std::filesystem::absolute(std::filesystem::path(build_dir)));
+ if (!std::filesystem::is_directory(std::filesystem::status(dir_path)))
+ LOG(FATAL) << "build dir should be a directory";
+
+ // init and open storage engine
+ std::vector<doris::StorePath> paths;
+ auto olap_res = doris::parse_conf_store_paths(_build_dir, &paths);
+ if (!olap_res) {
+ LOG(FATAL) << "parse config storage path failed, path=" << doris::config::storage_root_path;
+ exit(-1);
+ }
+ doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths);
+
+ doris::EngineOptions options;
+ options.store_paths = paths;
+ options.backend_uid = doris::UniqueId::gen_uid();
+ doris::StorageEngine* engine = nullptr;
+ auto st = doris::StorageEngine::open(options, &engine);
+ if (!st.ok()) {
+ LOG(FATAL) << "fail to open StorageEngine, res=" << st;
+ exit(-1);
+ }
+}
+
+std::string BuilderHelper::read_local_file(const std::string& file) {
+ std::filesystem::path path(std::filesystem::absolute(std::filesystem::path(file)));
+ if (!std::filesystem::exists(path)) LOG(FATAL) << "file not exist:" << file;
+
+ std::ifstream f(path, std::ios::in | std::ios::binary);
+ const auto sz = std::filesystem::file_size(path);
+ std::string result(sz, '\0');
+ f.read(result.data(), sz);
+
+ return result;
+}
+
+void BuilderHelper::build() {
+ // load tablet
+ std::string buf;
+ if (_isHDFS) {
+ FileSystemProperties system_properties;
+ LOG(INFO) << "read meta from HDFS:" << _meta_file;
+ auto* dfs = hdfsFileSystem.get();
+ bool exist = false;
+ auto st = dfs->exists(_meta_file, &exist);
+ if (!st.ok() || !exist) {
+ LOG(FATAL) << "file not exist:" /*<< dfs->GetLastError() */ << _meta_file;
+ }
+
+ size_t size = 0;
+ st = dfs->file_size(_meta_file, &size);
+ if (!st.ok() || size <= 0)
+ LOG(FATAL) << "meta file size abnormal.."; //<< dfs->GetLastError();
+ IOContext io_ctx;
+ io::FileReaderSPtr read_ptr = nullptr;
+ st = dfs->open_file(_meta_file, &read_ptr, &io_ctx);
+ if (!st.ok() || !read_ptr)
+ LOG(FATAL) << "cannot open file:" /*<< dfs->GetLastError()*/ << _meta_file;
+
+ buf.resize(size);
+
+ size_t read_size = 0;
+ read_ptr->read_at(0, Slice(buf.data(), size), io_ctx, &read_size);
+ if (read_size < size) {
+ LOG(FATAL) << "read meta file size abnormal.."; //<< dfs->GetLastError();
+ }
+
+ read_ptr->close();
+ } else {
+ buf = read_local_file(_meta_file);
+ }
+
+ FileHeader<TabletMetaPB> file_header;
+ buf = buf.substr(file_header.size());
+ // init tablet
+ TabletMeta* tablet_meta = new TabletMeta();
+ Status status = tablet_meta->deserialize(buf);
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to load tablet meta :" << status.to_string();
+ return;
+ }
+
+ LOG(INFO) << "table id:" << tablet_meta->table_id() << " tablet id:" << tablet_meta->tablet_id()
+ << " shard id:" << tablet_meta->shard_id();
+
+ auto data_dir = StorageEngine::instance()->get_store(_build_dir);
+ TabletMetaSharedPtr tablet_meta_ptr(tablet_meta);
+ TabletSharedPtr new_tablet = doris::Tablet::create_tablet_from_meta(tablet_meta_ptr, data_dir);
+ status = StorageEngine::instance()->tablet_manager()->_add_tablet_unlocked(
+ tablet_meta->tablet_id(), new_tablet, false, true);
+
+ if (!status.ok()) {
+ LOG(FATAL) << "fail to add tablet to storage :" << status.to_string();
+ return;
+ }
+
+ std::vector<std::string> files;
+ if (_isHDFS) {
+ auto* dfs = hdfsFileSystem.get();
+ std::vector<io::Path> names;
+ dfs->list(_data_path, &names);
+ for (const auto& name : names) {
+ std::string filename = name.filename().c_str();
+ if (filename.substr(filename.size() - _file_type.size()) == _file_type) {
+ LOG(INFO) << "get file:" << name;
+ files.emplace_back(filename);
+ }
Review Comment:
suggest to add some warning log when the file types are not matched both in hdfs and local dir.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
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