morningman commented on code in PR #69581: URL: https://github.com/apache/doris/pull/69581#discussion_r4389153994
########## be/src/format_v2/lance/lance_session_manager.cpp: ########## @@ -0,0 +1,239 @@ +// 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 "format_v2/lance/lance_session_manager.h" + +#include <lance/lance.h> + +#include <algorithm> +#include <limits> +#include <string> +#include <string_view> +#include <utility> + +#include "common/config.h" +#include "common/logging.h" +#include "common/metrics/doris_metrics.h" +#include "common/metrics/metrics.h" +#include "format_v2/lance/lance_reader_helper.h" + +namespace doris::format::lance { +namespace { + +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_hits_total, MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_misses_total, + MetricUnit::OPERATIONS); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_hits_total, + MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_misses_total, + MetricUnit::OPERATIONS); + +constexpr std::string_view LANCE_SESSION_CACHE_METRICS_HOOK = "lance_session_cache"; + +int64_t metric_value(uint64_t value) { + return static_cast<int64_t>( + std::min(value, static_cast<uint64_t>(std::numeric_limits<int64_t>::max()))); +} + +LanceSessionManager::Config load_lance_session_config() { + return { + .lance_index_cache_size_bytes = config::lance_index_cache_size_bytes, + .lance_metadata_cache_size_bytes = config::lance_metadata_cache_size_bytes, + .enable_lance_data_cache = config::enable_lance_data_cache, + .lance_data_cache_path = config::lance_data_cache_path, + .lance_data_cache_disk_capacity_bytes = config::lance_data_cache_disk_capacity_bytes, + .lance_data_cache_read_block_size_bytes = + config::lance_data_cache_read_block_size_bytes, + }; +} + +} // namespace + +class LanceSessionMetrics final { +public: + LanceSessionMetrics(LanceSession* session, const LanceSessionManager::Config& config) + : _session(session), _entity(DorisMetrics::instance()->server_entity()) { + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_misses_total); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_misses_total); + + lance_session_index_cache_capacity_bytes->set_value(config.lance_index_cache_size_bytes); + lance_session_metadata_cache_capacity_bytes->set_value( + config.lance_metadata_cache_size_bytes); + _entity->register_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK), + [this]() { update(); }); + update(); + } + + ~LanceSessionMetrics() { + _entity->deregister_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK)); + METRIC_DEREGISTER(_entity, lance_session_index_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_index_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_index_cache_misses_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_misses_total); + } + +private: + void update() { + // Session caches are shared across queries, so publish one process-wide snapshot instead + // of attributing concurrent cache activity to an individual query profile. + LanceSessionCacheStats stats {}; + if (lance_session_get_cache_stats(_session, &stats) != 0) { + LOG_EVERY_N(WARNING, 100) + << lance_error("collect Lance session cache statistics").to_string(); + return; + } + lance_session_index_cache_usage_bytes->set_value( + metric_value(stats.index_cache_size_bytes)); + lance_session_index_cache_entries->set_value(metric_value(stats.index_cache_entries)); + lance_session_index_cache_hits_total->set_value(metric_value(stats.index_cache_hits)); + lance_session_index_cache_misses_total->set_value(metric_value(stats.index_cache_misses)); + lance_session_metadata_cache_usage_bytes->set_value( + metric_value(stats.metadata_cache_size_bytes)); + lance_session_metadata_cache_entries->set_value(metric_value(stats.metadata_cache_entries)); + lance_session_metadata_cache_hits_total->set_value(metric_value(stats.metadata_cache_hits)); + lance_session_metadata_cache_misses_total->set_value( + metric_value(stats.metadata_cache_misses)); + } + + LanceSession* _session; + MetricEntity* _entity; + IntGauge* lance_session_index_cache_capacity_bytes = nullptr; + IntGauge* lance_session_index_cache_usage_bytes = nullptr; + IntGauge* lance_session_index_cache_entries = nullptr; + IntCounter* lance_session_index_cache_hits_total = nullptr; + IntCounter* lance_session_index_cache_misses_total = nullptr; + IntGauge* lance_session_metadata_cache_capacity_bytes = nullptr; + IntGauge* lance_session_metadata_cache_usage_bytes = nullptr; + IntGauge* lance_session_metadata_cache_entries = nullptr; + IntCounter* lance_session_metadata_cache_hits_total = nullptr; + IntCounter* lance_session_metadata_cache_misses_total = nullptr; +}; + +LanceSessionManager& LanceSessionManager::instance() { + // Function-local static initialization is thread safe. Cache configuration is process scoped, + // so changing it requires a BE restart. + static LanceSessionManager manager(load_lance_session_config()); + return manager; +} + +LanceSessionManager::LanceSessionManager(Config config) : _config(std::move(config)) { + LOG(INFO) << "Creating BE-wide Lance session manager: lance_index_cache_size_bytes=" + << _config.lance_index_cache_size_bytes + << ", lance_metadata_cache_size_bytes=" << _config.lance_metadata_cache_size_bytes + << ", enable_lance_data_cache=" << _config.enable_lance_data_cache + << ", lance_data_cache_path=" << _config.lance_data_cache_path + << ", lance_data_cache_disk_capacity_bytes=" + << _config.lance_data_cache_disk_capacity_bytes + << ", lance_data_cache_read_block_size_bytes=" + << _config.lance_data_cache_read_block_size_bytes + << ", foyer_memory_capacity_bytes=" << _config.lance_data_cache_read_block_size_bytes; +} + +LanceSessionManager::~LanceSessionManager() { + _metrics.reset(); + lance_session_close(_session); +} + +Status LanceSessionManager::_initialize() { + if (_config.enable_lance_data_cache) { + // Treat the configured cache mode as a process-level requirement. If an enabled + // data cache cannot initialize, report the failure instead of silently creating a + // session without it. This keeps directory/configuration/device failures visible + // to the operator. Disabling the cache is an explicit configuration change. + const LanceDataCacheOptions data_cache_options { + .directory = _config.lance_data_cache_path.c_str(), + // Foyer's HybridCache requires a memory tier. Keep it at the minimum useful + // capacity of exactly one range-cache block; WriteOnInsertion enqueues disk + // writes on insertion rather than waiting for memory-tier eviction. + .memory_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + .disk_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_disk_capacity_bytes), + .read_block_size_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + }; + _session = lance_session_new_with_data_cache( Review Comment: Confirmed; this is the branch-4.1 / 4.2 design, ported unchanged. The index and metadata caches belong to the lance-c session, which a BE creates on its first Lance scan with the configured capacities (`lance_index_cache_size_bytes`, `lance_metadata_cache_size_bytes`, applied at restart); `enable_lance_data_cache = false` does not affect them, and the `lance_session_*` metrics report their usage. The pinned lance-c exposes only `lance_session_get_cache_stats()`. It has no call to clear, evict or resize either cache, so a CachePolicy could neither prune them nor follow the memory-pressure capacity adjustment. Making them reclaimable needs that API in lance-c first and should land on all branches together, so it is tracked as a follow-up. The PR description now says that these caches are not charged to a MemTracker or managed by CacheManager, and how to size them. ########## be/src/format_v2/lance/lance_session_manager.cpp: ########## @@ -0,0 +1,239 @@ +// 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 "format_v2/lance/lance_session_manager.h" + +#include <lance/lance.h> + +#include <algorithm> +#include <limits> +#include <string> +#include <string_view> +#include <utility> + +#include "common/config.h" +#include "common/logging.h" +#include "common/metrics/doris_metrics.h" +#include "common/metrics/metrics.h" +#include "format_v2/lance/lance_reader_helper.h" + +namespace doris::format::lance { +namespace { + +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_hits_total, MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_misses_total, + MetricUnit::OPERATIONS); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_hits_total, + MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_misses_total, + MetricUnit::OPERATIONS); + +constexpr std::string_view LANCE_SESSION_CACHE_METRICS_HOOK = "lance_session_cache"; + +int64_t metric_value(uint64_t value) { + return static_cast<int64_t>( + std::min(value, static_cast<uint64_t>(std::numeric_limits<int64_t>::max()))); +} + +LanceSessionManager::Config load_lance_session_config() { + return { + .lance_index_cache_size_bytes = config::lance_index_cache_size_bytes, + .lance_metadata_cache_size_bytes = config::lance_metadata_cache_size_bytes, + .enable_lance_data_cache = config::enable_lance_data_cache, + .lance_data_cache_path = config::lance_data_cache_path, + .lance_data_cache_disk_capacity_bytes = config::lance_data_cache_disk_capacity_bytes, + .lance_data_cache_read_block_size_bytes = + config::lance_data_cache_read_block_size_bytes, + }; +} + +} // namespace + +class LanceSessionMetrics final { +public: + LanceSessionMetrics(LanceSession* session, const LanceSessionManager::Config& config) + : _session(session), _entity(DorisMetrics::instance()->server_entity()) { + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_misses_total); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_misses_total); + + lance_session_index_cache_capacity_bytes->set_value(config.lance_index_cache_size_bytes); + lance_session_metadata_cache_capacity_bytes->set_value( + config.lance_metadata_cache_size_bytes); + _entity->register_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK), + [this]() { update(); }); + update(); + } + + ~LanceSessionMetrics() { + _entity->deregister_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK)); + METRIC_DEREGISTER(_entity, lance_session_index_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_index_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_index_cache_misses_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_misses_total); + } + +private: + void update() { + // Session caches are shared across queries, so publish one process-wide snapshot instead + // of attributing concurrent cache activity to an individual query profile. + LanceSessionCacheStats stats {}; + if (lance_session_get_cache_stats(_session, &stats) != 0) { + LOG_EVERY_N(WARNING, 100) + << lance_error("collect Lance session cache statistics").to_string(); + return; + } + lance_session_index_cache_usage_bytes->set_value( + metric_value(stats.index_cache_size_bytes)); + lance_session_index_cache_entries->set_value(metric_value(stats.index_cache_entries)); + lance_session_index_cache_hits_total->set_value(metric_value(stats.index_cache_hits)); + lance_session_index_cache_misses_total->set_value(metric_value(stats.index_cache_misses)); + lance_session_metadata_cache_usage_bytes->set_value( + metric_value(stats.metadata_cache_size_bytes)); + lance_session_metadata_cache_entries->set_value(metric_value(stats.metadata_cache_entries)); + lance_session_metadata_cache_hits_total->set_value(metric_value(stats.metadata_cache_hits)); + lance_session_metadata_cache_misses_total->set_value( + metric_value(stats.metadata_cache_misses)); + } + + LanceSession* _session; + MetricEntity* _entity; + IntGauge* lance_session_index_cache_capacity_bytes = nullptr; + IntGauge* lance_session_index_cache_usage_bytes = nullptr; + IntGauge* lance_session_index_cache_entries = nullptr; + IntCounter* lance_session_index_cache_hits_total = nullptr; + IntCounter* lance_session_index_cache_misses_total = nullptr; + IntGauge* lance_session_metadata_cache_capacity_bytes = nullptr; + IntGauge* lance_session_metadata_cache_usage_bytes = nullptr; + IntGauge* lance_session_metadata_cache_entries = nullptr; + IntCounter* lance_session_metadata_cache_hits_total = nullptr; + IntCounter* lance_session_metadata_cache_misses_total = nullptr; +}; + +LanceSessionManager& LanceSessionManager::instance() { + // Function-local static initialization is thread safe. Cache configuration is process scoped, + // so changing it requires a BE restart. + static LanceSessionManager manager(load_lance_session_config()); + return manager; +} + +LanceSessionManager::LanceSessionManager(Config config) : _config(std::move(config)) { + LOG(INFO) << "Creating BE-wide Lance session manager: lance_index_cache_size_bytes=" + << _config.lance_index_cache_size_bytes + << ", lance_metadata_cache_size_bytes=" << _config.lance_metadata_cache_size_bytes + << ", enable_lance_data_cache=" << _config.enable_lance_data_cache + << ", lance_data_cache_path=" << _config.lance_data_cache_path + << ", lance_data_cache_disk_capacity_bytes=" + << _config.lance_data_cache_disk_capacity_bytes + << ", lance_data_cache_read_block_size_bytes=" + << _config.lance_data_cache_read_block_size_bytes + << ", foyer_memory_capacity_bytes=" << _config.lance_data_cache_read_block_size_bytes; +} + +LanceSessionManager::~LanceSessionManager() { + _metrics.reset(); + lance_session_close(_session); +} + +Status LanceSessionManager::_initialize() { + if (_config.enable_lance_data_cache) { + // Treat the configured cache mode as a process-level requirement. If an enabled + // data cache cannot initialize, report the failure instead of silently creating a + // session without it. This keeps directory/configuration/device failures visible + // to the operator. Disabling the cache is an explicit configuration change. + const LanceDataCacheOptions data_cache_options { + .directory = _config.lance_data_cache_path.c_str(), + // Foyer's HybridCache requires a memory tier. Keep it at the minimum useful + // capacity of exactly one range-cache block; WriteOnInsertion enqueues disk + // writes on insertion rather than waiting for memory-tier eviction. + .memory_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + .disk_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_disk_capacity_bytes), + .read_block_size_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + }; + _session = lance_session_new_with_data_cache( Review Comment: Valid, though it needs more than two endpoints with the same bucket name. The Foyer key is block size + Lance store prefix + object path + block index, and Lance's default prefix for S3 (likewise GCS and OSS) is `scheme$bucket`, so the same bucket name and object path behind two endpoints share entries, and the disk tier keeps them across a BE restart. For a scan to return the other endpoint's rows, the two stores must hold different bytes under the same data-file path. Lance's writers name data files with a random UUID (`generate_random_filename()`) and a copied dataset carries byte-identical immutable files, so this takes data files registered under chosen names, as you describe, or replaced in place. The key is built in `thirdparty/patches/lance-c-foyer.patch`, which branch-4.1 and branch-4.2 share byte for byte, so the fix belongs there: add the physical store identity from the storage options (endpoint, region, addressing style; not credentials) to the cache namespace and bum p `CACHE_KEY_VERSION`, with Rust tests over two origins that hold different bytes under one path. It is tracked as a follow-up for all branches, together with the open-file change to the same patch. Until then, a deployment that serves such files under one bucket name from several endpoints can set `enable_lance_data_cache = false`; the PR description now says so. -- 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]
