| // 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 "olap/rowset/beta_rowset.h" |
| |
| #include <crc32c/crc32c.h> |
| #include <ctype.h> |
| #include <errno.h> |
| #include <fmt/format.h> |
| |
| #include <algorithm> |
| #include <filesystem> |
| #include <memory> |
| #include <ostream> |
| #include <utility> |
| |
| #include "beta_rowset.h" |
| #include "common/config.h" |
| #include "common/logging.h" |
| #include "common/status.h" |
| #include "io/fs/file_reader.h" |
| #include "io/fs/file_system.h" |
| #include "io/fs/local_file_system.h" |
| #include "io/fs/path.h" |
| #include "io/fs/remote_file_system.h" |
| #include "olap/olap_common.h" |
| #include "olap/olap_define.h" |
| #include "olap/rowset/beta_rowset_reader.h" |
| #include "olap/rowset/rowset.h" |
| #include "olap/rowset/segment_v2/index_file_reader.h" |
| #include "olap/rowset/segment_v2/inverted_index_cache.h" |
| #include "olap/rowset/segment_v2/inverted_index_desc.h" |
| #include "olap/segment_loader.h" |
| #include "olap/tablet_schema.h" |
| #include "olap/utils.h" |
| #include "util/debug_points.h" |
| #include "util/doris_metrics.h" |
| |
| namespace doris { |
| #include "common/compile_check_begin.h" |
| using namespace ErrorCode; |
| |
| std::string BetaRowset::local_segment_path_segcompacted(const std::string& tablet_path, |
| const RowsetId& rowset_id, int64_t begin, |
| int64_t end) { |
| // {root_path}/data/{shard_id}/{tablet_id}/{schema_hash}/{rowset_id}_{begin_seg}-{end_seg}.dat |
| return fmt::format("{}/{}_{}-{}.dat", tablet_path, rowset_id.to_string(), begin, end); |
| } |
| |
| BetaRowset::BetaRowset(const TabletSchemaSPtr& schema, const RowsetMetaSharedPtr& rowset_meta, |
| std::string tablet_path) |
| : Rowset(schema, rowset_meta, std::move(tablet_path)) {} |
| |
| BetaRowset::~BetaRowset() = default; |
| |
| Status BetaRowset::init() { |
| return Status::OK(); // no op |
| } |
| |
| Status BetaRowset::get_segment_num_rows(std::vector<uint32_t>* segment_rows, |
| OlapReaderStatistics* read_stats) { |
| // `ROWSET_UNLOADING` is state for closed() called but owned by some readers. |
| // So here `ROWSET_UNLOADING` is allowed. |
| DCHECK_NE(_rowset_state_machine.rowset_state(), ROWSET_UNLOADED); |
| |
| RETURN_IF_ERROR(_load_segment_rows_once.call([this, read_stats] { |
| auto segment_count = num_segments(); |
| _segments_rows.resize(segment_count); |
| for (int64_t i = 0; i != segment_count; ++i) { |
| SegmentCacheHandle segment_cache_handle; |
| RETURN_IF_ERROR(SegmentLoader::instance()->load_segment( |
| std::static_pointer_cast<BetaRowset>(shared_from_this()), i, |
| &segment_cache_handle, false, false, read_stats)); |
| const auto& tmp_segments = segment_cache_handle.get_segments(); |
| _segments_rows[i] = tmp_segments[0]->num_rows(); |
| } |
| return Status::OK(); |
| })); |
| segment_rows->assign(_segments_rows.cbegin(), _segments_rows.cend()); |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::get_inverted_index_size(int64_t* index_size) { |
| const auto& fs = _rowset_meta->fs(); |
| if (!fs) { |
| return Status::Error<INIT_FAILED>("get fs failed, resource_id={}", |
| _rowset_meta->resource_id()); |
| } |
| |
| if (_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { |
| for (const auto& index : _schema->inverted_indexes()) { |
| for (int seg_id = 0; seg_id < num_segments(); ++seg_id) { |
| auto seg_path = DORIS_TRY(segment_path(seg_id)); |
| int64_t file_size = 0; |
| |
| std::string inverted_index_file_path = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_path), |
| index->index_id(), index->get_index_suffix()); |
| RETURN_IF_ERROR(fs->file_size(inverted_index_file_path, &file_size)); |
| *index_size += file_size; |
| } |
| } |
| } else { |
| for (int seg_id = 0; seg_id < num_segments(); ++seg_id) { |
| auto seg_path = DORIS_TRY(segment_path(seg_id)); |
| int64_t file_size = 0; |
| |
| std::string inverted_index_file_path = InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)); |
| RETURN_IF_ERROR(fs->file_size(inverted_index_file_path, &file_size)); |
| *index_size += file_size; |
| } |
| } |
| return Status::OK(); |
| } |
| |
| void BetaRowset::clear_inverted_index_cache() { |
| for (int i = 0; i < num_segments(); ++i) { |
| auto seg_path = segment_path(i); |
| if (!seg_path) { |
| continue; |
| } |
| |
| auto index_path_prefix = InvertedIndexDescriptor::get_index_file_path_prefix(*seg_path); |
| for (const auto& column : tablet_schema()->columns()) { |
| auto index_metas = tablet_schema()->inverted_indexs(*column); |
| for (const auto& index_meta : index_metas) { |
| auto inverted_index_file_cache_key = |
| InvertedIndexDescriptor::get_index_file_cache_key( |
| index_path_prefix, index_meta->index_id(), |
| index_meta->get_index_suffix()); |
| (void)segment_v2::InvertedIndexSearcherCache::instance()->erase( |
| inverted_index_file_cache_key); |
| } |
| } |
| } |
| } |
| |
| Status BetaRowset::get_segments_size(std::vector<size_t>* segments_size) { |
| auto fs = _rowset_meta->fs(); |
| if (!fs) { |
| return Status::Error<INIT_FAILED>("get fs failed, resource_id={}", |
| _rowset_meta->resource_id()); |
| } |
| |
| for (int seg_id = 0; seg_id < num_segments(); ++seg_id) { |
| auto seg_path = DORIS_TRY(segment_path(seg_id)); |
| int64_t file_size; |
| RETURN_IF_ERROR(fs->file_size(seg_path, &file_size)); |
| segments_size->push_back(file_size); |
| } |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::load_segments(std::vector<segment_v2::SegmentSharedPtr>* segments) { |
| return load_segments(0, num_segments(), segments); |
| } |
| |
| Status BetaRowset::load_segments(int64_t seg_id_begin, int64_t seg_id_end, |
| std::vector<segment_v2::SegmentSharedPtr>* segments) { |
| int64_t seg_id = seg_id_begin; |
| while (seg_id < seg_id_end) { |
| std::shared_ptr<segment_v2::Segment> segment; |
| RETURN_IF_ERROR(load_segment(seg_id, nullptr, &segment)); |
| segments->push_back(std::move(segment)); |
| seg_id++; |
| } |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::load_segment(int64_t seg_id, OlapReaderStatistics* stats, |
| segment_v2::SegmentSharedPtr* segment) { |
| auto fs = _rowset_meta->fs(); |
| if (!fs) { |
| return Status::Error<INIT_FAILED>("get fs failed"); |
| } |
| |
| DCHECK(seg_id >= 0); |
| auto seg_path = DORIS_TRY(segment_path(seg_id)); |
| io::FileReaderOptions reader_options { |
| .cache_type = config::enable_file_cache ? io::FileCachePolicy::FILE_BLOCK_CACHE |
| : io::FileCachePolicy::NO_CACHE, |
| .is_doris_table = true, |
| .cache_base_path = "", |
| .file_size = _rowset_meta->segment_file_size(static_cast<int>(seg_id)), |
| .tablet_id = _rowset_meta->tablet_id(), |
| }; |
| |
| auto s = segment_v2::Segment::open( |
| fs, seg_path, _rowset_meta->tablet_id(), static_cast<uint32_t>(seg_id), rowset_id(), |
| _schema, reader_options, segment, |
| _rowset_meta->inverted_index_file_info(static_cast<int>(seg_id)), stats); |
| if (!s.ok()) { |
| LOG(WARNING) << "failed to open segment. " << seg_path << " under rowset " << rowset_id() |
| << " : " << s.to_string(); |
| return s; |
| } |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::create_reader(RowsetReaderSharedPtr* result) { |
| // NOTE: We use std::static_pointer_cast for performance |
| result->reset(new BetaRowsetReader(std::static_pointer_cast<BetaRowset>(shared_from_this()))); |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::remove() { |
| if (!is_local()) { |
| DCHECK(false) << _rowset_meta->tablet_id() << ' ' << rowset_id(); |
| return Status::OK(); |
| } |
| |
| // TODO should we close and remove all segment reader first? |
| VLOG_NOTICE << "begin to remove files in rowset " << rowset_id() |
| << ", version:" << start_version() << "-" << end_version() |
| << ", tabletid:" << _rowset_meta->tablet_id(); |
| // If the rowset was removed, it need to remove the fds in segment cache directly |
| clear_cache(); |
| |
| bool success = true; |
| Status st; |
| const auto& fs = io::global_local_filesystem(); |
| for (int i = 0; i < num_segments(); ++i) { |
| auto seg_path = local_segment_path(_tablet_path, rowset_id().to_string(), i); |
| LOG(INFO) << "deleting " << seg_path; |
| st = fs->delete_file(seg_path); |
| if (!st.ok()) { |
| LOG(WARNING) << st.to_string(); |
| success = false; |
| } |
| |
| if (_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { |
| for (const auto& column : _schema->columns()) { |
| auto index_metas = _schema->inverted_indexs(*column); |
| for (const auto& index_meta : index_metas) { |
| std::string inverted_index_file = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_path), |
| index_meta->index_id(), index_meta->get_index_suffix()); |
| st = fs->delete_file(inverted_index_file); |
| if (!st.ok()) { |
| LOG(WARNING) << st.to_string(); |
| success = false; |
| } |
| } |
| } |
| } else { |
| if (_schema->has_inverted_index() || _schema->has_ann_index()) { |
| std::string inverted_index_file = InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)); |
| st = fs->delete_file(inverted_index_file); |
| if (!st.ok()) { |
| LOG(WARNING) << st.to_string(); |
| success = false; |
| } |
| } |
| } |
| } |
| if (!success) { |
| return Status::Error<ROWSET_DELETE_FILE_FAILED>("failed to remove files in rowset {}", |
| rowset_id().to_string()); |
| } |
| return Status::OK(); |
| } |
| |
| void BetaRowset::do_close() { |
| // do nothing. |
| } |
| |
| Status BetaRowset::link_files_to(const std::string& dir, RowsetId new_rowset_id, |
| size_t new_rowset_start_seg_id, |
| std::set<int64_t>* without_index_uids) { |
| if (!is_local()) { |
| DCHECK(false) << _rowset_meta->tablet_id() << ' ' << rowset_id(); |
| return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}", |
| _rowset_meta->tablet_id(), rowset_id().to_string()); |
| } |
| |
| const auto& local_fs = io::global_local_filesystem(); |
| Status status; |
| std::vector<std::string> linked_success_files; |
| Defer remove_linked_files {[&]() { // clear linked files if errors happen |
| if (!status.ok()) { |
| LOG(WARNING) << "will delete linked success files due to error " << status; |
| std::vector<io::Path> paths; |
| for (auto& file : linked_success_files) { |
| paths.emplace_back(file); |
| LOG(WARNING) << "will delete linked success file " << file << " due to error"; |
| } |
| static_cast<void>(local_fs->batch_delete(paths)); |
| LOG(WARNING) << "done delete linked success files due to error " << status; |
| } |
| }}; |
| |
| for (int i = 0; i < num_segments(); ++i) { |
| auto dst_path = |
| local_segment_path(dir, new_rowset_id.to_string(), i + new_rowset_start_seg_id); |
| bool dst_path_exist = false; |
| if (!local_fs->exists(dst_path, &dst_path_exist).ok() || dst_path_exist) { |
| status = Status::Error<FILE_ALREADY_EXIST>( |
| "failed to create hard link, file already exist: {}", dst_path); |
| return status; |
| } |
| auto src_path = local_segment_path(_tablet_path, rowset_id().to_string(), i); |
| // TODO(lingbin): how external storage support link? |
| // use copy? or keep refcount to avoid being delete? |
| if (!local_fs->link_file(src_path, dst_path).ok()) { |
| status = Status::Error<OS_ERROR>("fail to create hard link. from={}, to={}, errno={}", |
| src_path, dst_path, Errno::no()); |
| return status; |
| } |
| linked_success_files.push_back(dst_path); |
| DBUG_EXECUTE_IF("fault_inject::BetaRowset::link_files_to::_link_inverted_index_file", { |
| status = Status::Error<OS_ERROR>("fault_inject link_file error"); |
| return status; |
| }); |
| if (_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { |
| for (const auto& index : _schema->inverted_indexes()) { |
| auto index_id = index->index_id(); |
| if (without_index_uids != nullptr && without_index_uids->count(index_id)) { |
| continue; |
| } |
| std::string inverted_index_src_file_path = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(src_path), |
| index_id, index->get_index_suffix()); |
| std::string inverted_index_dst_file_path = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(dst_path), |
| index_id, index->get_index_suffix()); |
| bool index_file_exists = true; |
| RETURN_IF_ERROR(local_fs->exists(inverted_index_src_file_path, &index_file_exists)); |
| if (index_file_exists) { |
| DBUG_EXECUTE_IF( |
| "fault_inject::BetaRowset::link_files_to::_link_inverted_index_file", { |
| status = Status::Error<OS_ERROR>( |
| "fault_inject link_file error from={}, to={}", |
| inverted_index_src_file_path, inverted_index_dst_file_path); |
| return status; |
| }); |
| if (!local_fs->link_file(inverted_index_src_file_path, |
| inverted_index_dst_file_path) |
| .ok()) { |
| status = Status::Error<OS_ERROR>( |
| "fail to create hard link. from={}, to={}, errno={}", |
| inverted_index_src_file_path, inverted_index_dst_file_path, |
| Errno::no()); |
| return status; |
| } |
| linked_success_files.push_back(inverted_index_dst_file_path); |
| LOG(INFO) << "success to create hard link. from=" |
| << inverted_index_src_file_path << ", " |
| << "to=" << inverted_index_dst_file_path; |
| } else { |
| LOG(WARNING) << "skip create hard link to not existed index file=" |
| << inverted_index_src_file_path; |
| } |
| } |
| } else { |
| if ((_schema->has_inverted_index() || _schema->has_ann_index()) && |
| (without_index_uids == nullptr || without_index_uids->empty())) { |
| std::string inverted_index_file_src = |
| InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(src_path)); |
| std::string inverted_index_file_dst = |
| InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(dst_path)); |
| bool index_dst_path_exist = false; |
| |
| if (!local_fs->exists(inverted_index_file_dst, &index_dst_path_exist).ok() || |
| index_dst_path_exist) { |
| status = Status::Error<FILE_ALREADY_EXIST>( |
| "failed to create hard link, file already exist: {}", |
| inverted_index_file_dst); |
| return status; |
| } |
| if (!local_fs->link_file(inverted_index_file_src, inverted_index_file_dst).ok()) { |
| status = Status::Error<OS_ERROR>( |
| "fail to create hard link. from={}, to={}, errno={}", |
| inverted_index_file_src, inverted_index_file_dst, Errno::no()); |
| return status; |
| } |
| linked_success_files.push_back(inverted_index_file_dst); |
| } |
| } |
| } |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::copy_files_to(const std::string& dir, const RowsetId& new_rowset_id) { |
| if (!is_local()) { |
| DCHECK(false) << _rowset_meta->tablet_id() << ' ' << rowset_id(); |
| return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}", |
| _rowset_meta->tablet_id(), rowset_id().to_string()); |
| } |
| |
| bool exists = false; |
| for (int i = 0; i < num_segments(); ++i) { |
| auto dst_path = local_segment_path(dir, new_rowset_id.to_string(), i); |
| RETURN_IF_ERROR(io::global_local_filesystem()->exists(dst_path, &exists)); |
| if (exists) { |
| return Status::Error<FILE_ALREADY_EXIST>("file already exist: {}", dst_path); |
| } |
| auto src_path = local_segment_path(_tablet_path, rowset_id().to_string(), i); |
| RETURN_IF_ERROR(io::global_local_filesystem()->copy_path(src_path, dst_path)); |
| if (_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { |
| for (const auto& column : _schema->columns()) { |
| // if (column.has_inverted_index()) { |
| auto index_metas = _schema->inverted_indexs(*column); |
| for (const auto& index_meta : index_metas) { |
| std::string inverted_index_src_file_path = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(src_path), |
| index_meta->index_id(), index_meta->get_index_suffix()); |
| std::string inverted_index_dst_file_path = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(dst_path), |
| index_meta->index_id(), index_meta->get_index_suffix()); |
| RETURN_IF_ERROR(io::global_local_filesystem()->copy_path( |
| inverted_index_src_file_path, inverted_index_dst_file_path)); |
| LOG(INFO) << "success to copy file. from=" << inverted_index_src_file_path |
| << ", " |
| << "to=" << inverted_index_dst_file_path; |
| } |
| } |
| } else { |
| if (_schema->has_inverted_index() || _schema->has_ann_index()) { |
| std::string inverted_index_src_file = |
| InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(src_path)); |
| std::string inverted_index_dst_file = |
| InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(dst_path)); |
| RETURN_IF_ERROR(io::global_local_filesystem()->copy_path(inverted_index_src_file, |
| inverted_index_dst_file)); |
| LOG(INFO) << "success to copy file. from=" << inverted_index_src_file << ", " |
| << "to=" << inverted_index_dst_file; |
| } |
| } |
| } |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::upload_to(const StorageResource& dest_fs, const RowsetId& new_rowset_id) { |
| if (!is_local()) { |
| DCHECK(false) << _rowset_meta->tablet_id() << ' ' << rowset_id(); |
| return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}", |
| _rowset_meta->tablet_id(), rowset_id().to_string()); |
| } |
| |
| if (num_segments() < 1) { |
| return Status::OK(); |
| } |
| std::vector<io::Path> local_paths; |
| local_paths.reserve(num_segments()); |
| std::vector<io::Path> dest_paths; |
| dest_paths.reserve(num_segments()); |
| for (int i = 0; i < num_segments(); ++i) { |
| // Note: Here we use relative path for remote. |
| auto remote_seg_path = dest_fs.remote_segment_path(_rowset_meta->tablet_id(), |
| new_rowset_id.to_string(), i); |
| auto local_seg_path = local_segment_path(_tablet_path, rowset_id().to_string(), i); |
| dest_paths.emplace_back(remote_seg_path); |
| local_paths.emplace_back(local_seg_path); |
| if (_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { |
| for (const auto& column : _schema->columns()) { |
| // if (column.has_inverted_index()) { |
| auto index_metas = _schema->inverted_indexs(*column); |
| for (const auto& index_meta : index_metas) { |
| std::string remote_inverted_index_file = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix( |
| remote_seg_path), |
| index_meta->index_id(), index_meta->get_index_suffix()); |
| std::string local_inverted_index_file = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix( |
| local_seg_path), |
| index_meta->index_id(), index_meta->get_index_suffix()); |
| dest_paths.emplace_back(remote_inverted_index_file); |
| local_paths.emplace_back(local_inverted_index_file); |
| } |
| } |
| } else { |
| if (_schema->has_inverted_index() || _schema->has_ann_index()) { |
| std::string remote_inverted_index_file = |
| InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix( |
| remote_seg_path)); |
| std::string local_inverted_index_file = |
| InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix( |
| local_seg_path)); |
| dest_paths.emplace_back(remote_inverted_index_file); |
| local_paths.emplace_back(local_inverted_index_file); |
| } |
| } |
| } |
| auto st = dest_fs.fs->batch_upload(local_paths, dest_paths); |
| if (st.ok()) { |
| DorisMetrics::instance()->upload_rowset_count->increment(1); |
| DorisMetrics::instance()->upload_total_byte->increment(total_disk_size()); |
| } else { |
| DorisMetrics::instance()->upload_fail_count->increment(1); |
| } |
| return st; |
| } |
| |
| Status BetaRowset::check_file_exist() { |
| const auto& fs = _rowset_meta->fs(); |
| if (!fs) { |
| return Status::InternalError("fs is not initialized, resource_id={}", |
| _rowset_meta->resource_id()); |
| } |
| |
| for (int i = 0; i < num_segments(); ++i) { |
| auto seg_path = DORIS_TRY(segment_path(i)); |
| bool seg_file_exist = false; |
| RETURN_IF_ERROR(fs->exists(seg_path, &seg_file_exist)); |
| if (!seg_file_exist) { |
| return Status::InternalError("data file not existed: {}, rowset_id={}", seg_path, |
| rowset_id().to_string()); |
| } |
| } |
| |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::check_current_rowset_segment() { |
| const auto& fs = _rowset_meta->fs(); |
| if (!fs) { |
| return Status::InternalError("fs is not initialized, resource_id={}", |
| _rowset_meta->resource_id()); |
| } |
| |
| for (int seg_id = 0; seg_id < num_segments(); ++seg_id) { |
| auto seg_path = DORIS_TRY(segment_path(seg_id)); |
| |
| std::shared_ptr<segment_v2::Segment> segment; |
| io::FileReaderOptions reader_options { |
| .cache_type = config::enable_file_cache ? io::FileCachePolicy::FILE_BLOCK_CACHE |
| : io::FileCachePolicy::NO_CACHE, |
| .is_doris_table = true, |
| .cache_base_path {}, |
| .file_size = _rowset_meta->segment_file_size(seg_id), |
| }; |
| |
| auto s = segment_v2::Segment::open(fs, seg_path, _rowset_meta->tablet_id(), seg_id, |
| rowset_id(), _schema, reader_options, &segment, |
| _rowset_meta->inverted_index_file_info(seg_id)); |
| if (!s.ok()) { |
| LOG(WARNING) << "segment can not be opened. file=" << seg_path; |
| return s; |
| } |
| } |
| |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::add_to_binlog() { |
| // FIXME(Drogon): not only local file system |
| if (!is_local()) { |
| DCHECK(false) << _rowset_meta->tablet_id() << ' ' << rowset_id(); |
| return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}", |
| _rowset_meta->tablet_id(), rowset_id().to_string()); |
| } |
| |
| const auto& fs = io::global_local_filesystem(); |
| auto segments_num = num_segments(); |
| VLOG_DEBUG << fmt::format("add rowset to binlog. rowset_id={}, segments_num={}", |
| rowset_id().to_string(), segments_num); |
| |
| Status status; |
| std::vector<std::string> linked_success_files; |
| Defer remove_linked_files {[&]() { // clear linked files if errors happen |
| if (!status.ok()) { |
| LOG(WARNING) << "will delete linked success files due to error " |
| << status.to_string_no_stack(); |
| std::vector<io::Path> paths; |
| for (auto& file : linked_success_files) { |
| paths.emplace_back(file); |
| LOG(WARNING) << "will delete linked success file " << file << " due to error"; |
| } |
| static_cast<void>(fs->batch_delete(paths)); |
| LOG(WARNING) << "done delete linked success files due to error " |
| << status.to_string_no_stack(); |
| } |
| }}; |
| |
| // The publish_txn might fail even if the add_to_binlog success, so we need to check |
| // whether a file already exists before linking. |
| auto errno_is_file_exists = []() { return Errno::no() == EEXIST; }; |
| |
| // all segments are in the same directory, so cache binlog_dir without multi times check |
| std::string binlog_dir; |
| for (int i = 0; i < segments_num; ++i) { |
| auto seg_file = local_segment_path(_tablet_path, rowset_id().to_string(), i); |
| |
| if (binlog_dir.empty()) { |
| binlog_dir = std::filesystem::path(seg_file).parent_path().append("_binlog").string(); |
| |
| bool exists = true; |
| RETURN_IF_ERROR(fs->exists(binlog_dir, &exists)); |
| if (!exists) { |
| RETURN_IF_ERROR(fs->create_directory(binlog_dir)); |
| } |
| } |
| |
| auto binlog_file = |
| (std::filesystem::path(binlog_dir) / std::filesystem::path(seg_file).filename()) |
| .string(); |
| VLOG_DEBUG << "link " << seg_file << " to " << binlog_file; |
| if (!fs->link_file(seg_file, binlog_file).ok() && !errno_is_file_exists()) { |
| status = Status::Error<OS_ERROR>("fail to create hard link. from={}, to={}, errno={}", |
| seg_file, binlog_file, Errno::no()); |
| return status; |
| } |
| linked_success_files.push_back(binlog_file); |
| |
| if (_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { |
| for (const auto& index : _schema->inverted_indexes()) { |
| auto index_id = index->index_id(); |
| auto index_file = InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_file), index_id, |
| index->get_index_suffix()); |
| auto binlog_index_file = (std::filesystem::path(binlog_dir) / |
| std::filesystem::path(index_file).filename()) |
| .string(); |
| VLOG_DEBUG << "link " << index_file << " to " << binlog_index_file; |
| if (!fs->link_file(index_file, binlog_index_file).ok() && !errno_is_file_exists()) { |
| status = Status::Error<OS_ERROR>( |
| "fail to create hard link. from={}, to={}, errno={}", index_file, |
| binlog_index_file, Errno::no()); |
| return status; |
| } |
| linked_success_files.push_back(binlog_index_file); |
| } |
| } else { |
| if (_schema->has_inverted_index() || _schema->has_ann_index()) { |
| auto index_file = InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_file)); |
| auto binlog_index_file = (std::filesystem::path(binlog_dir) / |
| std::filesystem::path(index_file).filename()) |
| .string(); |
| VLOG_DEBUG << "link " << index_file << " to " << binlog_index_file; |
| if (!fs->link_file(index_file, binlog_index_file).ok() && !errno_is_file_exists()) { |
| status = Status::Error<OS_ERROR>( |
| "fail to create hard link. from={}, to={}, errno={}", index_file, |
| binlog_index_file, Errno::no()); |
| return status; |
| } |
| linked_success_files.push_back(binlog_index_file); |
| } |
| } |
| } |
| |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::calc_file_crc(uint32_t* crc_value, int64_t* file_count) { |
| const auto& fs = _rowset_meta->fs(); |
| DBUG_EXECUTE_IF("fault_inject::BetaRowset::calc_file_crc", |
| { return Status::Error<OS_ERROR>("fault_inject calc_file_crc error"); }); |
| if (num_segments() < 1) { |
| *crc_value = 0x92a8fc17; // magic code from crc32c table |
| return Status::OK(); |
| } |
| |
| // 1. pick up all the files including dat file and idx file |
| std::vector<io::Path> file_paths; |
| for (int seg_id = 0; seg_id < num_segments(); ++seg_id) { |
| auto seg_path = DORIS_TRY(segment_path(seg_id)); |
| file_paths.emplace_back(seg_path); |
| if (_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { |
| for (const auto& column : _schema->columns()) { |
| auto index_metas = _schema->inverted_indexs(*column); |
| for (const auto& index_meta : index_metas) { |
| std::string inverted_index_file = |
| InvertedIndexDescriptor::get_index_file_path_v1( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_path), |
| index_meta->index_id(), index_meta->get_index_suffix()); |
| file_paths.emplace_back(std::move(inverted_index_file)); |
| } |
| } |
| } else { |
| if (_schema->has_inverted_index() || _schema->has_ann_index()) { |
| std::string inverted_index_file = InvertedIndexDescriptor::get_index_file_path_v2( |
| InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)); |
| file_paths.emplace_back(std::move(inverted_index_file)); |
| } |
| } |
| } |
| *crc_value = 0; |
| *file_count = file_paths.size(); |
| if (!is_local()) { |
| return Status::OK(); |
| } |
| |
| // 2. calculate the md5sum of each file |
| const auto& local_fs = io::global_local_filesystem(); |
| DCHECK(!file_paths.empty()); |
| std::vector<std::string> all_file_md5; |
| all_file_md5.reserve(file_paths.size()); |
| for (const auto& file_path : file_paths) { |
| std::string file_md5sum; |
| auto status = local_fs->md5sum(file_path, &file_md5sum); |
| if (!status.ok()) { |
| return status; |
| } |
| VLOG_CRITICAL << fmt::format("calc file_md5sum finished. file_path={}, md5sum={}", |
| file_path.string(), file_md5sum); |
| all_file_md5.emplace_back(std::move(file_md5sum)); |
| } |
| std::sort(all_file_md5.begin(), all_file_md5.end()); |
| |
| // 3. calculate the crc_value based on all_file_md5 |
| DCHECK(file_paths.size() == all_file_md5.size()); |
| for (auto& i : all_file_md5) { |
| *crc_value = crc32c::Extend(*crc_value, (const uint8_t*)i.data(), i.size()); |
| } |
| |
| return Status::OK(); |
| } |
| |
| Status BetaRowset::show_nested_index_file(rapidjson::Value* rowset_value, |
| rapidjson::Document::AllocatorType& allocator) { |
| const auto& fs = _rowset_meta->fs(); |
| auto storage_format = _schema->get_inverted_index_storage_format(); |
| std::string format_str; |
| switch (storage_format) { |
| case InvertedIndexStorageFormatPB::V1: |
| format_str = "V1"; |
| break; |
| case InvertedIndexStorageFormatPB::V2: |
| format_str = "V2"; |
| break; |
| case InvertedIndexStorageFormatPB::V3: |
| format_str = "V3"; |
| break; |
| default: |
| return Status::InternalError("inverted index storage format error"); |
| break; |
| } |
| auto rs_id = rowset_id().to_string(); |
| rowset_value->AddMember("rowset_id", rapidjson::Value(rs_id.c_str(), allocator), allocator); |
| rowset_value->AddMember("index_storage_format", rapidjson::Value(format_str.c_str(), allocator), |
| allocator); |
| rapidjson::Value segments(rapidjson::kArrayType); |
| for (int seg_id = 0; seg_id < num_segments(); ++seg_id) { |
| rapidjson::Value segment(rapidjson::kObjectType); |
| segment.AddMember("segment_id", rapidjson::Value(seg_id).Move(), allocator); |
| |
| auto seg_path = DORIS_TRY(segment_path(seg_id)); |
| auto index_file_path_prefix = InvertedIndexDescriptor::get_index_file_path_prefix(seg_path); |
| auto index_file_reader = std::make_unique<IndexFileReader>( |
| fs, std::string(index_file_path_prefix), storage_format); |
| RETURN_IF_ERROR(index_file_reader->init()); |
| auto dirs = index_file_reader->get_all_directories(); |
| |
| auto add_file_info_to_json = [&](const std::string& path, |
| rapidjson::Value& json_value) -> Status { |
| json_value.AddMember("idx_file_path", rapidjson::Value(path.c_str(), allocator), |
| allocator); |
| int64_t idx_file_size = 0; |
| auto st = fs->file_size(path, &idx_file_size); |
| if (st != Status::OK()) { |
| LOG(WARNING) << "show nested index file get file size error, file: " << path |
| << ", error: " << st.msg(); |
| return st; |
| } |
| json_value.AddMember("idx_file_size", rapidjson::Value(idx_file_size).Move(), |
| allocator); |
| return Status::OK(); |
| }; |
| |
| auto process_files = [&allocator, &index_file_reader](auto& index_meta, |
| rapidjson::Value& indices, |
| rapidjson::Value& index) -> Status { |
| rapidjson::Value files_value(rapidjson::kArrayType); |
| std::vector<std::string> files; |
| auto ret = index_file_reader->open(&index_meta); |
| if (!ret.has_value()) { |
| LOG(INFO) << "IndexFileReader open error:" << ret.error(); |
| return Status::InternalError("IndexFileReader open error"); |
| } |
| using T = std::decay_t<decltype(ret)>; |
| auto reader = std::forward<T>(ret).value(); |
| reader->list(&files); |
| for (auto& file : files) { |
| rapidjson::Value file_value(rapidjson::kObjectType); |
| auto size = reader->fileLength(file.c_str()); |
| file_value.AddMember("name", rapidjson::Value(file.c_str(), allocator), allocator); |
| file_value.AddMember("size", rapidjson::Value(size).Move(), allocator); |
| files_value.PushBack(file_value, allocator); |
| } |
| index.AddMember("files", files_value, allocator); |
| indices.PushBack(index, allocator); |
| return Status::OK(); |
| }; |
| |
| if (storage_format != InvertedIndexStorageFormatPB::V1) { |
| auto path = InvertedIndexDescriptor::get_index_file_path_v2(index_file_path_prefix); |
| auto st = add_file_info_to_json(path, segment); |
| if (!st.ok()) { |
| return st; |
| } |
| rapidjson::Value indices(rapidjson::kArrayType); |
| for (auto& dir : *dirs) { |
| rapidjson::Value index(rapidjson::kObjectType); |
| auto index_id = dir.first.first; |
| auto index_suffix = dir.first.second; |
| index.AddMember("index_id", rapidjson::Value(index_id).Move(), allocator); |
| index.AddMember("index_suffix", rapidjson::Value(index_suffix.c_str(), allocator), |
| allocator); |
| |
| rapidjson::Value files_value(rapidjson::kArrayType); |
| std::vector<std::string> files; |
| doris::TabletIndexPB index_pb; |
| index_pb.set_index_id(index_id); |
| index_pb.set_index_suffix_name(index_suffix); |
| TabletIndex index_meta; |
| index_meta.init_from_pb(index_pb); |
| |
| auto status = process_files(index_meta, indices, index); |
| if (!status.ok()) { |
| return status; |
| } |
| } |
| segment.AddMember("indices", indices, allocator); |
| segments.PushBack(segment, allocator); |
| } else { |
| rapidjson::Value indices(rapidjson::kArrayType); |
| for (auto column : _rowset_meta->tablet_schema()->columns()) { |
| auto index_metas = _rowset_meta->tablet_schema()->inverted_indexs(*column); |
| for (const auto& index_meta : index_metas) { |
| rapidjson::Value index(rapidjson::kObjectType); |
| auto index_id = index_meta->index_id(); |
| auto index_suffix = index_meta->get_index_suffix(); |
| index.AddMember("index_id", rapidjson::Value(index_id).Move(), allocator); |
| index.AddMember("index_suffix", |
| rapidjson::Value(index_suffix.c_str(), allocator), allocator); |
| auto path = InvertedIndexDescriptor::get_index_file_path_v1( |
| index_file_path_prefix, index_id, index_suffix); |
| RETURN_IF_ERROR(add_file_info_to_json(path, index)); |
| RETURN_IF_ERROR(process_files(*index_meta, indices, index)); |
| } |
| } |
| segment.AddMember("indices", indices, allocator); |
| segments.PushBack(segment, allocator); |
| } |
| } |
| rowset_value->AddMember("segments", segments, allocator); |
| return Status::OK(); |
| } |
| #include "common/compile_check_end.h" |
| } // namespace doris |
