gavinchou commented on code in PR #57072: URL: https://github.com/apache/doris/pull/57072#discussion_r2484233526
########## be/src/io/cache/cache_block_meta_store.cpp: ########## @@ -0,0 +1,465 @@ +// 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 "io/cache/cache_block_meta_store.h" + +#include <butil/logging.h> +#include <bvar/bvar.h> +#include <fmt/format.h> +#include <rocksdb/db.h> +#include <rocksdb/filter_policy.h> +#include <rocksdb/table.h> + +#include <algorithm> +#include <cstring> +#include <filesystem> +#include <optional> +#include <sstream> + +#include "common/status.h" +#include "olap/field.h" +#include "olap/field.h" // For OLAP_FIELD_TYPE_BIGINT +#include "olap/key_coder.h" +#include "olap/olap_common.h" +#include "util/threadpool.h" +#include "vec/common/hex.h" + +namespace doris::io { + +const std::string FILE_CACHE_META_COLUMN_FAMILY = "file_cache_meta"; + +// bvar metrics for rocksdb operation failures +bvar::Adder<uint64_t> g_rocksdb_write_failed_num("file_cache_meta_rocksdb_write_failed_num"); +bvar::Adder<uint64_t> g_rocksdb_delete_failed_num("file_cache_meta_rocksdb_delete_failed_num"); + +CacheBlockMetaStore::CacheBlockMetaStore(const std::string& db_path, size_t queue_size) + : _db_path(db_path), _write_queue(queue_size) { + auto status = init(); + if (!status.ok()) { + LOG(ERROR) << "Failed to initialize CacheBlockMetaStore: " << status.to_string(); + } +} + +CacheBlockMetaStore::~CacheBlockMetaStore() { + _stop_worker.store(true, std::memory_order_release); + if (_write_thread.joinable()) { + _write_thread.join(); + } + + if (_db) { + if (_file_cache_meta_cf_handle) { + _db->DestroyColumnFamilyHandle(_file_cache_meta_cf_handle.release()); + } + _db->Close(); + } +} + +size_t CacheBlockMetaStore::get_write_queue_size() const { + return _write_queue.size_approx(); +} + +Status CacheBlockMetaStore::init() { + std::filesystem::create_directories(_db_path); + + _options.create_if_missing = true; + _options.create_missing_column_families = true; + _options.error_if_exists = false; + _options.compression = rocksdb::kNoCompression; + _options.max_open_files = 1000; + _options.write_buffer_size = 64 * 1024 * 1024; // 64MB + _options.target_file_size_base = 64 * 1024 * 1024; + + rocksdb::BlockBasedTableOptions table_options; + table_options.filter_policy.reset(rocksdb::NewBloomFilterPolicy(10, false)); + table_options.block_size = 16 * 1024; + _options.table_factory.reset(rocksdb::NewBlockBasedTableFactory(table_options)); + + // Create column family descriptors + std::vector<rocksdb::ColumnFamilyDescriptor> column_families; + // Default column family is required + column_families.emplace_back(rocksdb::kDefaultColumnFamilyName, rocksdb::ColumnFamilyOptions()); + // File cache meta column family + column_families.emplace_back(FILE_CACHE_META_COLUMN_FAMILY, rocksdb::ColumnFamilyOptions()); + + std::vector<rocksdb::ColumnFamilyHandle*> handles; + rocksdb::DB* db_ptr = nullptr; + rocksdb::Status status = + rocksdb::DB::Open(_options, _db_path, column_families, &handles, &db_ptr); + + if (!status.ok()) { + LOG(WARNING) << "Failed to open rocksdb: " << status.ToString() + << "Database path: " << _db_path; + return Status::InternalError("Failed to open rocksdb: {}", status.ToString()); + } + _db.reset(db_ptr); + + // Store the file_cache_meta column family handle + // handles[0] is default column family, handles[1] is file_cache_meta + if (handles.size() >= 2) { + _file_cache_meta_cf_handle.reset(handles[1]); + // Close default column family handle as we won't use it + _db->DestroyColumnFamilyHandle(handles[0]); + } else { + return Status::InternalError("Failed to get file_cache_meta column family handle"); + } + + _write_thread = std::thread(&CacheBlockMetaStore::async_write_worker, this); + + return Status::OK(); +} + +void CacheBlockMetaStore::put(const BlockMetaKey& key, const BlockMeta& meta) { + std::string key_str = serialize_key(key); + std::string value_str = serialize_value(meta); + + // Put write task into queue for asynchronous processing + WriteOperation op; + op.type = OperationType::PUT; + op.key = key_str; + op.value = value_str; + _write_queue.enqueue(op); +} + +std::optional<BlockMeta> CacheBlockMetaStore::get(const BlockMetaKey& key) { + // we trade accurate for clean code. so we ignore pending operations in the write queue + // only use data in rocksdb + std::string key_str = serialize_key(key); + std::string value_str; + rocksdb::Status status; + + if (!_db) { + LOG(WARNING) << "Database not initialized, cannot get key"; + return std::nullopt; + } + status = + _db->Get(rocksdb::ReadOptions(), _file_cache_meta_cf_handle.get(), key_str, &value_str); + + if (status.ok()) { + return deserialize_value(value_str); + } else if (status.IsNotFound()) { + return std::nullopt; + } else { + LOG(WARNING) << "Failed to get key from rocksdb: " << status.ToString(); + return std::nullopt; + } +} + +std::unique_ptr<BlockMetaIterator> CacheBlockMetaStore::range_get(int64_t tablet_id) { + // Generate prefix using new serialization format + std::string prefix; + prefix.push_back(0x1); // version byte + auto* tablet_id_coder = get_key_coder(FieldType::OLAP_FIELD_TYPE_BIGINT); + tablet_id_coder->full_encode_ascending(&tablet_id, &prefix); + + class RocksDBIterator : public BlockMetaIterator { + public: + RocksDBIterator(rocksdb::Iterator* iter, const std::string& prefix) + : _iter(iter), _prefix(prefix) { + _iter->Seek(_prefix); + } + + ~RocksDBIterator() override { delete _iter; } + + bool valid() const override { + if (!_iter->Valid()) return false; + Slice key_slice(_iter->key().data(), _prefix.size()); + return key_slice.compare(Slice(_prefix)) == 0; + } + + void next() override { _iter->Next(); } + + BlockMetaKey key() const override { + std::string key_str = _iter->key().ToString(); + Slice slice(key_str); + + // Check version byte + if (slice.size < 1 || slice.data[0] != 0x1) { + LOG(WARNING) << "Invalid key version in range_get"; + return BlockMetaKey(); + } + slice.remove_prefix(1); // skip version byte + + auto* tablet_id_coder = get_key_coder(FieldType::OLAP_FIELD_TYPE_BIGINT); + int64_t tablet_id; + uint64_t hash_high, hash_low; + size_t offset; + + Status st = tablet_id_coder->decode_ascending(&slice, sizeof(int64_t), + reinterpret_cast<uint8_t*>(&tablet_id)); Review Comment: better return an error instead of nothing -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
