diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 8c1c0f62a01ec1..cbb48a518d14f4 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1228,10 +1228,13 @@ DEFINE_Validator(variant_storage_parse_mode, // can be shared by all Lance dataset readers. DEFINE_Int64(lance_index_cache_size_bytes, "10737418240"); // 10 GiB DEFINE_Int64(lance_metadata_cache_size_bytes, "1073741824"); // 1GB +DEFINE_Bool(enable_lance_scalar_index_auto_selection, "true"); DEFINE_Bool(enable_lance_data_cache, "true"); DEFINE_String(lance_data_cache_path, "${DORIS_HOME}/lance_data_cache"); DEFINE_Int64(lance_data_cache_disk_capacity_bytes, "107374182400"); // 100GB DEFINE_Int64(lance_data_cache_read_block_size_bytes, "1048576"); // 1MB +// Zero preserves the existing one-block Foyer memory tier, including custom block sizes. +DEFINE_Int64(lance_data_cache_memory_capacity_bytes, "0"); // I/O buffering budget per Lance scanner, not a cap on its total memory usage. // Runtime changes apply to newly created scanners. diff --git a/be/src/common/config.h b/be/src/common/config.h index afb0e6c58028a9..5c5b11e7748b42 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1264,10 +1264,13 @@ DECLARE_Int32(blocking_pipeline_executor_size); // Lance shared session and optional Foyer data-file cache. DECLARE_Int64(lance_index_cache_size_bytes); DECLARE_Int64(lance_metadata_cache_size_bytes); +// Runtime BTree selection uses page metadata and preserves the explicit fragment domain. +DECLARE_Bool(enable_lance_scalar_index_auto_selection); DECLARE_Bool(enable_lance_data_cache); DECLARE_String(lance_data_cache_path); DECLARE_Int64(lance_data_cache_disk_capacity_bytes); DECLARE_Int64(lance_data_cache_read_block_size_bytes); +DECLARE_Int64(lance_data_cache_memory_capacity_bytes); // I/O buffering budget per Lance scanner, applied when a new scanner is created. DECLARE_mInt64(lance_io_buffer_size_bytes); diff --git a/be/src/format_v2/lance/lance_reader_helper.h b/be/src/format_v2/lance/lance_reader_helper.h index 1f07fbd4bcb66d..d5fa76ede15d9a 100644 --- a/be/src/format_v2/lance/lance_reader_helper.h +++ b/be/src/format_v2/lance/lance_reader_helper.h @@ -141,6 +141,26 @@ inline constexpr LanceProfileMetric LANCE_SCAN_METRICS[] = { .profile_name = "LanceIVFPartitionSelectionTime", .group = LANCE_VECTOR_INDEX_PROFILE, .unit = TUnit::TIME_NS}, + {.native_name = "scalar_index_auto_candidates", + .profile_name = "LanceScalarIndexAutoCandidates", + .group = LANCE_SCALAR_INDEX_PROFILE, + .unit = TUnit::UNIT}, + {.native_name = "scalar_index_auto_changed", + .profile_name = "LanceScalarIndexAutoChanged", + .group = LANCE_SCALAR_INDEX_PROFILE, + .unit = TUnit::UNIT}, + {.native_name = "scalar_index_auto_estimated_pages", + .profile_name = "LanceScalarIndexAutoEstimatedPages", + .group = LANCE_SCALAR_INDEX_PROFILE, + .unit = TUnit::UNIT}, + {.native_name = "scalar_index_auto_estimated_rows", + .profile_name = "LanceScalarIndexAutoEstimatedRows", + .group = LANCE_SCALAR_INDEX_PROFILE, + .unit = TUnit::UNIT}, + {.native_name = "scalar_index_auto_full_scans", + .profile_name = "LanceScalarIndexAutoFullScans", + .group = LANCE_SCALAR_INDEX_PROFILE, + .unit = TUnit::UNIT}, {.native_name = "scalar_segments_requested", .profile_name = "LanceScalarIndexSegmentsRequested", .group = LANCE_SCALAR_INDEX_PROFILE, diff --git a/be/src/format_v2/lance/lance_session_cache_profile.cpp b/be/src/format_v2/lance/lance_session_cache_profile.cpp new file mode 100644 index 00000000000000..82278d37a63157 --- /dev/null +++ b/be/src/format_v2/lance/lance_session_cache_profile.cpp @@ -0,0 +1,114 @@ +// 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_cache_profile.h" + +#include +#include +#include +#include + +#include "common/logging.h" + +namespace doris::format::lance { + +struct LanceSessionCacheProfile::State { + State(RuntimeProfile* profile, RuntimeProfile::Counter* index_misses, + RuntimeProfile::Counter* metadata_misses, Snapshot snapshot) + : profile(profile), + index_misses(index_misses), + metadata_misses(metadata_misses), + snapshot(std::move(snapshot)) {} + + void collect() { + std::lock_guard guard(mutex); + collect_locked(); + } + + void collect_locked() { + if (!active || !valid) { + return; + } + LanceSessionManager::CacheStatistics current; + const auto status = snapshot(¤t); + if (!status.ok()) { + complete = false; + profile->add_info_string("LanceSessionCacheStatsAvailable", "false"); + LOG(WARNING) << status; + return; + } + // Serialization and dataset close share a baseline under this lock. Neither can + // count a delta twice, or extend a closed window with unrelated Session activity. + const auto update = [](RuntimeProfile::Counter* counter, uint64_t before, uint64_t after) { + if (after >= before) { + COUNTER_UPDATE(counter, + static_cast(std::min( + after - before, std::numeric_limits::max()))); + } + }; + update(index_misses, previous.index_cache_misses, current.index_cache_misses); + update(metadata_misses, previous.metadata_cache_misses, current.metadata_cache_misses); + previous = current; + profile->add_info_string("LanceSessionCacheStatsAvailable", complete ? "true" : "false"); + } + + std::mutex mutex; + RuntimeProfile* profile; + RuntimeProfile::Counter* index_misses; + RuntimeProfile::Counter* metadata_misses; + Snapshot snapshot; + LanceSessionManager::CacheStatistics previous; + bool active = false; + bool valid = false; + bool complete = true; +}; + +LanceSessionCacheProfile::LanceSessionCacheProfile(RuntimeProfile* profile, + RuntimeProfile::Counter* index_misses, + RuntimeProfile::Counter* metadata_misses, + Snapshot snapshot) + : _state(std::make_shared(profile, index_misses, metadata_misses, + std::move(snapshot))), + _snapshot_callback( + profile->add_snapshot_callback([state = _state] { state->collect(); })) {} + +LanceSessionCacheProfile::~LanceSessionCacheProfile() { + stop(); +} + +void LanceSessionCacheProfile::start() { + // First-use Session initialization may perform I/O. A profile snapshot must not wait + // for it; the dataset-open window starts only after its baseline is ready. + LanceSessionManager::CacheStatistics previous; + const bool valid = _state->snapshot(&previous).ok(); + std::lock_guard guard(_state->mutex); + _state->active = true; + _state->valid = valid; + _state->previous = previous; + if (!_state->valid) { + _state->complete = false; + _state->profile->add_info_string("LanceSessionCacheStatsAvailable", "false"); + } +} + +void LanceSessionCacheProfile::stop() { + std::lock_guard guard(_state->mutex); + _state->collect_locked(); + _state->active = false; +} + +} // namespace doris::format::lance diff --git a/be/src/format_v2/lance/lance_session_cache_profile.h b/be/src/format_v2/lance/lance_session_cache_profile.h new file mode 100644 index 00000000000000..f4d301d739986e --- /dev/null +++ b/be/src/format_v2/lance/lance_session_cache_profile.h @@ -0,0 +1,53 @@ +// 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. + +#pragma once + +#include +#include + +#include "format_v2/lance/lance_session_manager.h" +#include "runtime/runtime_profile.h" + +namespace doris::format::lance { + +// Observes one reader's dataset-open windows without accessing its native dataset or scanner. +class LanceSessionCacheProfile { +public: + using Snapshot = std::function; + + LanceSessionCacheProfile( + RuntimeProfile* profile, RuntimeProfile::Counter* index_misses, + RuntimeProfile::Counter* metadata_misses, Snapshot snapshot = [](auto* stats) { + return LanceSessionManager::instance().get_cache_statistics(stats); + }); + ~LanceSessionCacheProfile(); + + LanceSessionCacheProfile(const LanceSessionCacheProfile&) = delete; + LanceSessionCacheProfile& operator=(const LanceSessionCacheProfile&) = delete; + + // The reader serializes start/stop; profile serialization may sample concurrently. + void start(); + void stop(); + +private: + struct State; + std::shared_ptr _state; + std::shared_ptr _snapshot_callback; +}; + +} // namespace doris::format::lance diff --git a/be/src/format_v2/lance/lance_session_manager.cpp b/be/src/format_v2/lance/lance_session_manager.cpp index 884eb5db18c5e7..53b0945af7c820 100644 --- a/be/src/format_v2/lance/lance_session_manager.cpp +++ b/be/src/format_v2/lance/lance_session_manager.cpp @@ -64,6 +64,8 @@ LanceSessionManager::Config load_lance_session_config() { .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, + .lance_data_cache_memory_capacity_bytes = + config::lance_data_cache_memory_capacity_bytes, }; } @@ -159,8 +161,8 @@ LanceSessionManager::LanceSessionManager(Config config) : _config(std::move(conf << ", 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; + << _config.lance_data_cache_read_block_size_bytes << ", foyer_memory_capacity_bytes=" + << _config.resolved_data_cache_memory_capacity_bytes(); } LanceSessionManager::~LanceSessionManager() { @@ -170,17 +172,20 @@ LanceSessionManager::~LanceSessionManager() { Status LanceSessionManager::_initialize() { if (_config.enable_lance_data_cache) { + if (_config.lance_data_cache_memory_capacity_bytes < 0) { + return Status::InvalidArgument( + "lance_data_cache_memory_capacity_bytes must be non-negative"); + } // 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. + // Keep the historical one-block tier when the memory capacity is unset. + // Its capacity is independent of the range-cache block granularity otherwise. .memory_capacity_bytes = - static_cast(_config.lance_data_cache_read_block_size_bytes), + static_cast(_config.resolved_data_cache_memory_capacity_bytes()), .disk_capacity_bytes = static_cast(_config.lance_data_cache_disk_capacity_bytes), .read_block_size_bytes = @@ -220,19 +225,37 @@ Status LanceSessionManager::open_dataset(const char* uri, const char* const* sto } *dataset = nullptr; - // Initialize lazily on the first dataset open, not at BE startup. A failed Status is a + RETURN_IF_ERROR(_ensure_initialized()); + + *dataset = lance_dataset_open_with_session(uri, storage_options, version, _session); + if (*dataset == nullptr) { + return lance_error("open Lance dataset with shared session"); + } + return Status::OK(); +} + +Status LanceSessionManager::_ensure_initialized() { + // Initialize lazily on the first use, not at BE startup. A failed Status is a // normal return from this lambda, so call_once completes and retains that failure just // like a successful initialization. All subsequent readers receive the same copied // error; fixing the cache directory alone does not trigger another attempt. This is // intentional: repair the cache configuration/storage (or disable the data cache), then // restart the BE to recreate the process-wide manager and session. std::call_once(_initialize_once, [this] { _initialize_status = _initialize(); }); - RETURN_IF_ERROR(_initialize_status); + return _initialize_status; +} - *dataset = lance_dataset_open_with_session(uri, storage_options, version, _session); - if (*dataset == nullptr) { - return lance_error("open Lance dataset with shared session"); +Status LanceSessionManager::get_cache_statistics(CacheStatistics* statistics) { + if (statistics == nullptr) { + return Status::InvalidArgument("Lance cache statistics output must not be null"); + } + RETURN_IF_ERROR(_ensure_initialized()); + LanceSessionCacheStats stats {}; + if (lance_session_get_cache_stats(_session, &stats) != 0) { + return lance_error("collect Lance session cache statistics"); } + statistics->index_cache_misses = stats.index_cache_misses; + statistics->metadata_cache_misses = stats.metadata_cache_misses; return Status::OK(); } diff --git a/be/src/format_v2/lance/lance_session_manager.h b/be/src/format_v2/lance/lance_session_manager.h index 45dc62acad4442..d1773c2a34ab6d 100644 --- a/be/src/format_v2/lance/lance_session_manager.h +++ b/be/src/format_v2/lance/lance_session_manager.h @@ -43,6 +43,18 @@ class LanceSessionManager final { std::string lance_data_cache_path; int64_t lance_data_cache_disk_capacity_bytes = 0; int64_t lance_data_cache_read_block_size_bytes = 0; + int64_t lance_data_cache_memory_capacity_bytes = 0; + + int64_t resolved_data_cache_memory_capacity_bytes() const { + return lance_data_cache_memory_capacity_bytes == 0 + ? lance_data_cache_read_block_size_bytes + : lance_data_cache_memory_capacity_bytes; + } + }; + + struct CacheStatistics { + uint64_t index_cache_misses = 0; + uint64_t metadata_cache_misses = 0; }; static LanceSessionManager& instance(); @@ -58,8 +70,12 @@ class LanceSessionManager final { Status open_dataset(const char* uri, const char* const* storage_options, uint64_t version, LanceDataset** dataset); + // A process-wide snapshot, including other readers using the shared session. + Status get_cache_statistics(CacheStatistics* statistics); + private: Status _initialize(); + Status _ensure_initialized(); Config _config; std::once_flag _initialize_once; diff --git a/be/src/format_v2/table/lance_reader.cpp b/be/src/format_v2/table/lance_reader.cpp index e0e2884efe7a4a..752ce7460c29e9 100644 --- a/be/src/format_v2/table/lance_reader.cpp +++ b/be/src/format_v2/table/lance_reader.cpp @@ -34,6 +34,7 @@ #include "exec/common/endian.h" #include "format_v2/lance/lance_reader_helper.h" #include "format_v2/lance/lance_runtime_filter_helper.h" +#include "format_v2/lance/lance_session_cache_profile.h" #include "format_v2/lance/lance_session_manager.h" #include "runtime/file_scan_profile.h" #include "runtime/runtime_state.h" @@ -111,12 +112,28 @@ Status LanceTableReader::init(TableReadOptions&& options) { DORIS_CHECK(_scanner_profile != nullptr); DORIS_CHECK(_scan_params != nullptr); RETURN_IF_ERROR(_resolve_search_kind()); + if (_search_kind == SearchKind::NORMAL) { + _scanner_profile->add_info_string( + "LanceScalarIndexAutoSelection", + config::enable_lance_scalar_index_auto_selection ? "true" : "false"); + } const auto& lance_scan_params = _scan_params->lance_scan_params; ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, LANCE_READER_PROFILE, TUnit::NONE, file_scan_profile::TABLE_READER, 1); _dataset_open_time = add_lance_counter(_scanner_profile, "LanceDatasetOpenTime", TUnit::TIME_NS, LANCE_TIMING_PROFILE); + auto* session_index_cache_misses = add_lance_counter( + _scanner_profile, "LanceSessionIndexCacheMisses", TUnit::UNIT, "LanceSessionCache"); + auto* session_metadata_cache_misses = add_lance_counter( + _scanner_profile, "LanceSessionMetadataCacheMisses", TUnit::UNIT, "LanceSessionCache"); + _scanner_profile->add_info_string( + "LanceSessionCacheStatsScope", + "BE-wide changes during dataset-open windows; includes concurrent readers; " + "overlapping windows must not be summed as query-exclusive misses"); + _scanner_profile->add_info_string("LanceSessionCacheStatsAvailable", "false"); + _session_cache_profile = std::make_unique( + _scanner_profile, session_index_cache_misses, session_metadata_cache_misses); _arrow_to_doris_block_time = add_lance_counter(_scanner_profile, "LanceArrowToDorisBlockTime", TUnit::TIME_NS, LANCE_TIMING_PROFILE); _data_cache_bytes_read_from_cache = @@ -647,11 +664,14 @@ Status LanceTableReader::_open_dataset(const DatasetKey& key) { std::unique_ptr dataset; { SCOPED_TIMER(_dataset_open_time); + auto& session = LanceSessionManager::instance(); + _session_cache_profile->start(); LanceDataset* raw_dataset = nullptr; - Status status = LanceSessionManager::instance().open_dataset( + Status status = session.open_dataset( key.uri.c_str(), key.storage_options.empty() ? nullptr : storage_option_ptrs.data(), static_cast(key.version), &raw_dataset); if (!status.ok()) { + _session_cache_profile->stop(); return status.prepend(fmt::format("Lance dataset version {} at {}: ", key.version, display_uri(key.uri))); } @@ -943,6 +963,10 @@ Status LanceTableReader::_configure_normal_scan(LanceScanner* scanner, return Status::InvalidArgument( "Lance scalar index segment cannot be combined with use_scalar_index=false"); } + if (lance_scanner_set_scalar_index_auto_selection( + scanner, config::enable_lance_scalar_index_auto_selection) != 0) { + return lance_error("set Lance runtime scalar index selection"); + } if (lance_scanner_set_scalar_index_segment(scanner, segment_uuids.data()) != 0) { return lance_error("set Lance scanner scalar index segment"); } @@ -1260,6 +1284,9 @@ void LanceTableReader::_close_dataset() { lance_fts_query_context_close(_fts_query_context); _fts_query_context = nullptr; } + if (_session_cache_profile != nullptr) { + _session_cache_profile->stop(); + } if (_dataset != nullptr) { _collect_data_cache_statistics(); lance_dataset_close(_dataset); diff --git a/be/src/format_v2/table/lance_reader.h b/be/src/format_v2/table/lance_reader.h index 205c082d12052d..ba09954afb57d8 100644 --- a/be/src/format_v2/table/lance_reader.h +++ b/be/src/format_v2/table/lance_reader.h @@ -26,6 +26,7 @@ #include "common/status.h" #include "format_v2/lance/lance_record_batch_converter.h" +#include "format_v2/lance/lance_session_cache_profile.h" #include "format_v2/table_reader.h" #include "runtime/runtime_profile.h" @@ -101,6 +102,7 @@ class LanceTableReader final : public TableReader { Status _dataset_key(const TFileRangeDesc& range, DatasetKey* key) const; LanceDataset* _dataset = nullptr; + std::unique_ptr _session_cache_profile; std::shared_ptr _dataset_schema; LanceScanner* _scanner = nullptr; ShardedKVCache* _runtime_filter_cache = nullptr; diff --git a/be/src/runtime/runtime_profile.cpp b/be/src/runtime/runtime_profile.cpp index 311d434f6bcff9..bbd1fc85d85a61 100644 --- a/be/src/runtime/runtime_profile.cpp +++ b/be/src/runtime/runtime_profile.cpp @@ -703,12 +703,40 @@ void RuntimeProfile::pretty_print(std::ostream* s, const std::string& prefix, } } +std::shared_ptr RuntimeProfile::add_snapshot_callback( + SnapshotCallback callback) { + auto registration = std::make_shared(std::move(callback)); + std::lock_guard l(_counter_map_lock); + _snapshot_callbacks.emplace_back(registration); + return registration; +} + +void RuntimeProfile::_run_snapshot_callbacks() { + std::vector> callbacks; + { + std::lock_guard l(_counter_map_lock); + std::erase_if(_snapshot_callbacks, [&](auto& weak) { + if (auto callback = weak.lock()) { + callbacks.emplace_back(std::move(callback)); + return false; + } + return true; + }); + } + // A stopped scanner can remain alive in a native read after pipeline completion. Sample + // before the immutable final copy, without holding locks needed by counter publication. + for (const auto& callback : callbacks) { + (*callback)(); + } +} + void RuntimeProfile::to_thrift(TRuntimeProfileTree* tree, int64_t profile_level) { tree->nodes.clear(); to_thrift(&tree->nodes, profile_level); } void RuntimeProfile::to_thrift(std::vector* nodes, int64_t profile_level) { + _run_snapshot_callbacks(); size_t index = nodes->size(); nodes->push_back(TRuntimeProfileNode()); TRuntimeProfileNode& node = (*nodes)[index]; @@ -756,6 +784,7 @@ void RuntimeProfile::to_proto(PRuntimeProfileTree* tree, int64_t profile_level) void RuntimeProfile::to_proto(google::protobuf::RepeatedPtrField* nodes, int64_t profile_level) { + _run_snapshot_callbacks(); PRuntimeProfileNode* node = nodes->Add(); // allocate new node node->set_name(_name); node->set_metadata(_metadata); diff --git a/be/src/runtime/runtime_profile.h b/be/src/runtime/runtime_profile.h index e63480711ac397..0873cc6ed9b880 100644 --- a/be/src/runtime/runtime_profile.h +++ b/be/src/runtime/runtime_profile.h @@ -584,6 +584,13 @@ class RuntimeProfile { // the key does not exist. const std::string* get_info_string(const std::string& key); + using SnapshotCallback = std::function; + // Refresh observations before serializing counters and info strings, without profile locks. + // Keep the returned token alive to stay registered. In-flight callbacks can outlive the + // token, so callbacks must own their state and synchronize with its producer's shutdown. + // Callbacks must not throw or recursively serialize this profile. + std::shared_ptr add_snapshot_callback(SnapshotCallback callback); + // Returns the counter for the total elapsed time. Counter* total_time_counter() { return &_counter_total_time; } @@ -698,6 +705,10 @@ class RuntimeProfile { // protects _counter_map, _counter_child_map and _bucketing_counters mutable std::mutex _counter_map_lock; + // Protected by _counter_map_lock. Expired registrations are pruned during snapshots. + std::vector> _snapshot_callbacks; + void _run_snapshot_callbacks(); + // Child profiles. Does not own memory. // We record children in both a map (to facilitate updates) and a vector // (to print things in the order they were registered) diff --git a/be/test/format_v2/lance/lance_session_cache_profile_test.cpp b/be/test/format_v2/lance/lance_session_cache_profile_test.cpp new file mode 100644 index 00000000000000..425779d487c5d6 --- /dev/null +++ b/be/test/format_v2/lance/lance_session_cache_profile_test.cpp @@ -0,0 +1,219 @@ +// 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_cache_profile.h" + +#include +#include + +#include +#include +#include +#include +#include + +#include "util/defer_op.h" + +namespace doris::format::lance { + +TEST(LanceSessionCacheProfileTest, EarlyLimitSnapshotsWhileAnotherWorkerIsInNativeRead) { + std::unique_ptr session( + lance_session_new(64 * 1024 * 1024, 64 * 1024 * 1024), lance_session_close); + ASSERT_NE(nullptr, session); + auto snapshot = [&](LanceSessionManager::CacheStatistics* stats) { + LanceSessionCacheStats native {}; + if (lance_session_get_cache_stats(session.get(), &native) != 0) { + return Status::InternalError("native snapshot unavailable"); + } + *stats = {native.index_cache_misses, native.metadata_cache_misses}; + return Status::OK(); + }; + RuntimeProfile profile("early_limit"); + auto* index = profile.add_counter("index", TUnit::UNIT); + auto* metadata = profile.add_counter("metadata", TUnit::UNIT); + LanceSessionCacheProfile slow(&profile, index, metadata, snapshot); + LanceSessionCacheProfile fast(&profile, index, metadata, snapshot); + LanceSessionManager::CacheStatistics before_slow; + ASSERT_TRUE(snapshot(&before_slow).ok()); + slow.start(); + const char* uri = "./be/test/format_v2/table/lance/data/all_types.lance"; + std::unique_ptr slow_dataset( + lance_dataset_open_with_session(uri, nullptr, 0, session.get()), lance_dataset_close); + ASSERT_NE(nullptr, slow_dataset); + LanceSessionManager::CacheStatistics before_fast; + ASSERT_TRUE(snapshot(&before_fast).ok()); + fast.start(); + std::unique_ptr fast_dataset( + lance_dataset_open_with_session(uri, nullptr, 0, session.get()), lance_dataset_close); + ASSERT_NE(nullptr, fast_dataset); + const char* columns[] = {"row_id", nullptr}; + std::unique_ptr slow_scanner( + lance_scanner_new(slow_dataset.get(), columns, nullptr), lance_scanner_close); + ASSERT_NE(nullptr, slow_scanner); + struct BlockedRead { + std::promise entered; + std::shared_future released; + std::atomic held = false; + }; + std::promise release; + BlockedRead blocked; + blocked.released = release.get_future().share(); + auto entered = blocked.entered.get_future(); + ASSERT_EQ(0, lance_scanner_set_statistics_callback( + slow_scanner.get(), + [](void* ctx, const LanceScanStatistics*) { + auto* blocked_read = static_cast(ctx); + if (!blocked_read->held.exchange(true)) { + // This callback runs inside native execution while the + // scanner worker is still waiting for its read to finish. + blocked_read->entered.set_value(); + blocked_read->released.wait(); + } + }, + &blocked)); + std::thread worker([&] { + LanceBatch* batch = nullptr; + int32_t result; + while ((result = lance_scanner_next(slow_scanner.get(), &batch)) == 0) { + lance_batch_free(batch); + } + EXPECT_EQ(1, result); + slow_scanner.reset(); + slow.stop(); + }); + Defer join {[&] { + release.set_value(); + worker.join(); + }}; + ASSERT_EQ(std::future_status::ready, entered.wait_for(std::chrono::seconds(10))); + std::unique_ptr fast_scanner( + lance_scanner_new(fast_dataset.get(), columns, nullptr), lance_scanner_close); + ASSERT_NE(nullptr, fast_scanner); + ASSERT_EQ(0, lance_scanner_set_limit(fast_scanner.get(), 1)); + LanceBatch* batch = nullptr; + ASSERT_EQ(0, lance_scanner_next(fast_scanner.get(), &batch)); + lance_batch_free(batch); + fast_scanner.reset(); + fast.stop(); + LanceSessionManager::CacheStatistics current; + ASSERT_TRUE(snapshot(¤t).ok()); + ASSERT_GT(current.metadata_cache_misses, before_slow.metadata_cache_misses); + TRuntimeProfileTree saved; + profile.to_thrift(&saved); + auto restored = RuntimeProfile::from_thrift(saved); + EXPECT_EQ("true", saved.nodes[0].info_strings.at("LanceSessionCacheStatsAvailable")); + EXPECT_GE(restored->get_counter("metadata")->value(), + (current.metadata_cache_misses - before_slow.metadata_cache_misses) + + (current.metadata_cache_misses - before_fast.metadata_cache_misses)); +} + +TEST(LanceSessionCacheProfileTest, SnapshotOpenWindowsAndDoNotRecountOnClose) { + RuntimeProfile profile("session"); + auto* index = profile.add_counter("index", TUnit::UNIT); + auto* metadata = profile.add_counter("metadata", TUnit::UNIT); + LanceSessionManager::CacheStatistics current {10, 20}; + auto snapshot = [&](auto* stats) { + *stats = current; + return Status::OK(); + }; + LanceSessionCacheProfile first(&profile, index, metadata, snapshot); + LanceSessionCacheProfile second(&profile, index, metadata, snapshot); + first.start(); + current = {12, 23}; + second.start(); + current = {15, 27}; + first.stop(); + + TRuntimeProfileTree saved; + profile.to_thrift(&saved); + auto restored = RuntimeProfile::from_thrift(saved); + EXPECT_EQ(8, restored->get_counter("index")->value()); + EXPECT_EQ(11, restored->get_counter("metadata")->value()); + EXPECT_EQ("true", saved.nodes[0].info_strings.at("LanceSessionCacheStatsAvailable")); + profile.to_thrift(&saved); + second.stop(); + second.stop(); + EXPECT_EQ(8, index->value()); + EXPECT_EQ(11, metadata->value()); + + current = {100, 200}; + profile.to_thrift(&saved); + EXPECT_EQ(8, index->value()); + second.start(); + current = {102, 205}; + second.stop(); + EXPECT_EQ(10, index->value()); + EXPECT_EQ(16, metadata->value()); +} + +TEST(LanceSessionCacheProfileTest, ConcurrentSnapshotAndCloseCountEachDeltaOnce) { + RuntimeProfile profile("session"); + auto* index = profile.add_counter("index", TUnit::UNIT); + auto* metadata = profile.add_counter("metadata", TUnit::UNIT); + std::atomic current = 0; + LanceSessionCacheProfile window(&profile, index, metadata, [&](auto* stats) { + const auto value = current.load(); + *stats = {value, value}; + return Status::OK(); + }); + std::barrier rendezvous(2); + std::thread reporter([&] { + for (int i = 0; i < 100; ++i) { + rendezvous.arrive_and_wait(); + TRuntimeProfileTree saved; + profile.to_thrift(&saved); + rendezvous.arrive_and_wait(); + } + }); + for (int i = 0; i < 100; ++i) { + window.start(); + current.fetch_add(1); + rendezvous.arrive_and_wait(); + window.stop(); + rendezvous.arrive_and_wait(); + } + reporter.join(); + EXPECT_EQ(100, index->value()); + EXPECT_EQ(100, metadata->value()); +} + +TEST(LanceSessionCacheProfileTest, FailedSnapshotsStayUnavailableAndExpiredWindowsAreSafe) { + RuntimeProfile profile("session"); + auto* index = profile.add_counter("index", TUnit::UNIT); + auto* metadata = profile.add_counter("metadata", TUnit::UNIT); + bool fail = false; + { + LanceSessionCacheProfile window(&profile, index, metadata, [&](auto* stats) { + *stats = {1, 1}; + return fail ? Status::InternalError("snapshot unavailable") : Status::OK(); + }); + window.start(); + fail = true; + TRuntimeProfileTree saved; + profile.to_thrift(&saved); + EXPECT_EQ("false", saved.nodes[0].info_strings.at("LanceSessionCacheStatsAvailable")); + fail = false; + profile.to_thrift(&saved); + EXPECT_EQ("false", saved.nodes[0].info_strings.at("LanceSessionCacheStatsAvailable")); + } + TRuntimeProfileTree saved; + profile.to_thrift(&saved); + EXPECT_EQ(0, index->value()); + EXPECT_EQ(0, metadata->value()); +} + +} // namespace doris::format::lance diff --git a/be/test/format_v2/lance/lance_session_manager_test.cpp b/be/test/format_v2/lance/lance_session_manager_test.cpp index e1274219347352..dde7a68e19b4e7 100644 --- a/be/test/format_v2/lance/lance_session_manager_test.cpp +++ b/be/test/format_v2/lance/lance_session_manager_test.cpp @@ -54,6 +54,7 @@ TEST(LanceSessionManagerTest, SessionAndDataCacheConfigurationsAreIndependent) { .lance_data_cache_path = "", .lance_data_cache_disk_capacity_bytes = -1, .lance_data_cache_read_block_size_bytes = -1, + .lance_data_cache_memory_capacity_bytes = -1, }; LanceSessionManager manager(std::move(config)); LanceDataset* raw_dataset = nullptr; @@ -62,6 +63,27 @@ TEST(LanceSessionManagerTest, SessionAndDataCacheConfigurationsAreIndependent) { ASSERT_NE(dataset, nullptr); } +TEST(LanceSessionManagerTest, MemoryCapacityDefaultsToOneConfiguredBlock) { + LanceSessionManager::Config config; + config.lance_data_cache_read_block_size_bytes = 64 * 1024; + EXPECT_EQ(config.resolved_data_cache_memory_capacity_bytes(), 64 * 1024); + config.lance_data_cache_memory_capacity_bytes = 8 * 1024 * 1024; + EXPECT_EQ(config.resolved_data_cache_memory_capacity_bytes(), 8 * 1024 * 1024); + EXPECT_EQ(config.lance_data_cache_read_block_size_bytes, 64 * 1024); +} + +TEST(LanceSessionManagerTest, RejectsNegativeMemoryCapacityBeforeUnsignedConversion) { + LanceSessionManager::Config config; + config.enable_lance_data_cache = true; + config.lance_data_cache_memory_capacity_bytes = -1; + LanceSessionManager manager(std::move(config)); + LanceDataset* dataset = nullptr; + const auto status = manager.open_dataset(lance_fixture_path().c_str(), nullptr, 0, &dataset); + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("lance_data_cache_memory_capacity_bytes"), std::string::npos); + EXPECT_EQ(dataset, nullptr); +} + TEST(LanceSessionManagerTest, PublishesSessionCacheMetrics) { constexpr int64_t INDEX_CACHE_CAPACITY = 8 * 1024 * 1024; constexpr int64_t METADATA_CACHE_CAPACITY = 4 * 1024 * 1024; @@ -114,6 +136,7 @@ TEST(LanceSessionManagerTest, CreatesFoyerBackedSession) { .lance_data_cache_path = cache_path.string(), .lance_data_cache_disk_capacity_bytes = 32 * 1024 * 1024, .lance_data_cache_read_block_size_bytes = 64 * 1024, + .lance_data_cache_memory_capacity_bytes = 8 * 1024 * 1024, }; { LanceSessionManager manager(std::move(config)); diff --git a/be/test/format_v2/table/lance_reader_test.cpp b/be/test/format_v2/table/lance_reader_test.cpp index 6531fc21de72e8..c27190b8fe7751 100644 --- a/be/test/format_v2/table/lance_reader_test.cpp +++ b/be/test/format_v2/table/lance_reader_test.cpp @@ -29,6 +29,7 @@ #include #include #include +#include #include #include @@ -1349,6 +1350,45 @@ TEST(LanceTableReaderSnapshotTest, ProfileShowsTheSnapshotRead) { EXPECT_TRUE(latest_reader.close().ok()); } +TEST(LanceTableReaderSnapshotTest, ProfileReportsObservedSessionCacheMisses) { + LanceSessionManager::CacheStatistics before; + ASSERT_TRUE(LanceSessionManager::instance().get_cache_statistics(&before).ok()); + const std::filesystem::path dataset_uri = + "./be/test/format_v2/table/lance/data/all_types.lance"; + TQueryGlobals query_globals; + RuntimeState state(query_globals); + RuntimeProfile profile("lance_session_cache_profile"); + TFileScanRangeParams scan_params; + const Columns columns {projected_column("row_id", TYPE_BIGINT, false)}; + LanceTableReader reader; + ASSERT_TRUE(init_reader(&reader, columns, &state, &profile, &scan_params).ok()); + ASSERT_TRUE(prepare_range(&reader, make_latest_lance_range(dataset_uri)).ok()); + TRuntimeProfileTree saved; + profile.to_thrift(&saved); + ASSERT_EQ("true", saved.nodes[0].info_strings.at("LanceSessionCacheStatsAvailable")); + LanceSessionManager::CacheStatistics after; + ASSERT_TRUE(LanceSessionManager::instance().get_cache_statistics(&after).ok()); + for (const char* name : {"LanceSessionIndexCacheMisses", "LanceSessionMetadataCacheMisses"}) { + auto* counter = profile.get_counter(name); + ASSERT_NE(counter, nullptr) << name; + EXPECT_GE(counter->value(), 0) << name; + } + EXPECT_EQ(profile.get_counter("LanceSessionIndexCacheMisses")->value(), + after.index_cache_misses - before.index_cache_misses); + EXPECT_EQ(profile.get_counter("LanceSessionMetadataCacheMisses")->value(), + after.metadata_cache_misses - before.metadata_cache_misses); + const auto* scope = profile.get_info_string("LanceSessionCacheStatsScope"); + ASSERT_NE(scope, nullptr); + EXPECT_NE(scope->find("concurrent"), std::string::npos); + const auto* available = profile.get_info_string("LanceSessionCacheStatsAvailable"); + ASSERT_NE(available, nullptr); + EXPECT_EQ(*available, "true"); + auto* misses = profile.get_counter("LanceSessionMetadataCacheMisses"); + const auto before_close = misses->value(); + ASSERT_TRUE(reader.close().ok()); + EXPECT_EQ(misses->value(), before_close); +} + TEST(LanceTableReaderSnapshotTest, OpenErrorsNameTheVersionWithoutTheQueryString) { const std::filesystem::path dataset_uri = "./be/test/format_v2/table/lance/data/all_types.lance"; @@ -2235,87 +2275,100 @@ TEST(LanceTableReaderFilterTest, CombinesStaticSubstraitFilterWithRuntimeFilter) } TEST(LanceTableReaderScalarSegmentTest, FiltersIndexedAndUncoveredDomainsWithoutLosingRows) { - const auto unique_suffix = std::chrono::steady_clock::now().time_since_epoch().count(); - for (const auto index_type : {LANCE_SCALAR_BTREE, LANCE_SCALAR_BITMAP}) { - SCOPED_TRACE(index_type == LANCE_SCALAR_BTREE ? "BTREE" : "BITMAP"); - const auto dataset_uri = std::filesystem::temp_directory_path() / - ("doris_lance_scalar_segment_" + std::to_string(unique_suffix) + - "_" + std::to_string(index_type) + ".lance"); - Defer cleanup {[&] { - std::error_code error; - std::filesystem::remove_all(dataset_uri, error); - }}; - const auto schema = arrow::schema({arrow::field("row_id", arrow::int64(), false)}); - for (int64_t first_id : {1, 3}) { - arrow::Int64Builder values; - ASSERT_TRUE(values.AppendValues({first_id, first_id + 1}).ok()); - auto array = values.Finish(); - ASSERT_TRUE(array.ok()); - auto batch = arrow::RecordBatch::Make(schema, 2, {std::move(array).ValueUnsafe()}); - auto batches = arrow::RecordBatchReader::Make({batch}, schema); - ASSERT_TRUE(batches.ok()); - ArrowArrayStream stream {}; - ASSERT_TRUE( - arrow::ExportRecordBatchReader(std::move(batches).ValueUnsafe(), &stream).ok()); - auto dataset = ::lance::Dataset::write( - dataset_uri.string(), &stream, - first_id == 1 ? ::lance::WriteMode::Create : ::lance::WriteMode::Append); - if (first_id == 1) { - // The second append remains uncovered by this index segment. - dataset.create_scalar_index("row_id", index_type, "row_id_idx"); - } - } - LanceFixtureInfo fixture; - ASSERT_TRUE(get_fixture_info(dataset_uri, &fixture).ok()); - ASSERT_EQ(2, fixture.fragment_ids.size()); - std::vector segments; - ASSERT_TRUE(get_index_segment_uuids(dataset_uri, "row_id_idx", &segments).ok()); - ASSERT_EQ(1, segments.size()); - - const auto read_domain = [&](const std::vector& fragments, bool use_segment, - const std::vector& expected) { - SCOPED_TRACE(::testing::PrintToString(fragments)); - SCOPED_TRACE(use_segment ? "selected segment" : "scalar index disabled"); - TQueryGlobals globals; - RuntimeState state(globals); - RuntimeProfile profile("lance_scalar_segment"); - TFileScanRangeParams scan_params; - const Columns columns {projected_column("row_id", TYPE_BIGINT, false)}; - LanceTableReader reader; - auto filter = create_int64_runtime_in_conjunct("row_id", {2, 4}, 41); - ASSERT_TRUE( - init_reader(&reader, columns, &state, &profile, &scan_params, {filter}).ok()); - auto range = make_lance_range(dataset_uri, fixture.version, fragments); - if (use_segment) { - range.table_format_params.lance_params.__set_index_segment_uuids(segments); - } else { - range.table_format_params.lance_params.__set_use_scalar_index(false); - } - ASSERT_TRUE(prepare_range(&reader, std::move(range)).ok()); - Block block; - add_output_columns(&block, columns); - std::vector actual; - bool eos = false; - while (!eos) { - auto status = reader.get_block(&block, &eos); - ASSERT_TRUE(status.ok()) << status.to_string(); - if (!eos) { - const auto& ids = - assert_cast(*block.get_by_position(0).column); - actual.insert(actual.end(), ids.get_data().begin(), ids.get_data().end()); + const auto previous_auto_selection = config::enable_lance_scalar_index_auto_selection; + Defer restore_config { + [&] { config::enable_lance_scalar_index_auto_selection = previous_auto_selection; }}; + for (const bool auto_selection : {false, true}) { + config::enable_lance_scalar_index_auto_selection = auto_selection; + const auto unique_suffix = std::chrono::steady_clock::now().time_since_epoch().count(); + for (const auto index_type : {LANCE_SCALAR_BTREE, LANCE_SCALAR_BITMAP}) { + SCOPED_TRACE(index_type == LANCE_SCALAR_BTREE ? "BTREE" : "BITMAP"); + const auto dataset_uri = + std::filesystem::temp_directory_path() / + ("doris_lance_scalar_segment_" + std::to_string(unique_suffix) + "_" + + std::to_string(index_type) + ".lance"); + Defer cleanup {[&] { + std::error_code error; + std::filesystem::remove_all(dataset_uri, error); + }}; + const auto schema = arrow::schema({arrow::field("row_id", arrow::int64(), false)}); + for (int64_t first_id : {1, 3}) { + arrow::Int64Builder values; + ASSERT_TRUE(values.AppendValues({first_id, first_id + 1}).ok()); + auto array = values.Finish(); + ASSERT_TRUE(array.ok()); + auto batch = arrow::RecordBatch::Make(schema, 2, {std::move(array).ValueUnsafe()}); + auto batches = arrow::RecordBatchReader::Make({batch}, schema); + ASSERT_TRUE(batches.ok()); + ArrowArrayStream stream {}; + ASSERT_TRUE( + arrow::ExportRecordBatchReader(std::move(batches).ValueUnsafe(), &stream) + .ok()); + auto dataset = ::lance::Dataset::write( + dataset_uri.string(), &stream, + first_id == 1 ? ::lance::WriteMode::Create : ::lance::WriteMode::Append); + if (first_id == 1) { + // The second append remains uncovered by this index segment. + dataset.create_scalar_index("row_id", index_type, "row_id_idx"); } } - std::ranges::sort(actual); - EXPECT_EQ(expected, actual); - EXPECT_TRUE(reader.close().ok()); - }; - read_domain({fixture.fragment_ids[0]}, true, {2}); - read_domain({fixture.fragment_ids[1]}, false, {4}); - // A segment which cannot cover the whole task must filter the entire explicit domain. - read_domain(fixture.fragment_ids, true, {2, 4}); - // Disabling the index must produce the same rows in each read domain. - read_domain({fixture.fragment_ids[0]}, false, {2}); - read_domain(fixture.fragment_ids, false, {2, 4}); + LanceFixtureInfo fixture; + ASSERT_TRUE(get_fixture_info(dataset_uri, &fixture).ok()); + ASSERT_EQ(2, fixture.fragment_ids.size()); + std::vector segments; + ASSERT_TRUE(get_index_segment_uuids(dataset_uri, "row_id_idx", &segments).ok()); + ASSERT_EQ(1, segments.size()); + + const auto read_domain = [&](const std::vector& fragments, bool use_segment, + const std::vector& expected) { + SCOPED_TRACE(::testing::PrintToString(fragments)); + SCOPED_TRACE(use_segment ? "selected segment" : "scalar index disabled"); + TQueryGlobals globals; + RuntimeState state(globals); + RuntimeProfile profile("lance_scalar_segment"); + TFileScanRangeParams scan_params; + const Columns columns {projected_column("row_id", TYPE_BIGINT, false)}; + LanceTableReader reader; + auto filter = create_int64_runtime_in_conjunct("row_id", {2, 4}, 41); + ASSERT_TRUE(init_reader(&reader, columns, &state, &profile, &scan_params, {filter}) + .ok()); + auto range = make_lance_range(dataset_uri, fixture.version, fragments); + if (use_segment) { + range.table_format_params.lance_params.__set_index_segment_uuids(segments); + } else { + range.table_format_params.lance_params.__set_use_scalar_index(false); + } + ASSERT_TRUE(prepare_range(&reader, std::move(range)).ok()); + Block block; + add_output_columns(&block, columns); + std::vector actual; + bool eos = false; + while (!eos) { + auto status = reader.get_block(&block, &eos); + ASSERT_TRUE(status.ok()) << status.to_string(); + if (!eos) { + const auto& ids = + assert_cast(*block.get_by_position(0).column); + actual.insert(actual.end(), ids.get_data().begin(), ids.get_data().end()); + } + } + std::ranges::sort(actual); + EXPECT_EQ(expected, actual); + EXPECT_TRUE(reader.close().ok()); + auto* candidates = profile.get_counter("LanceScalarIndexAutoCandidates"); + ASSERT_NE(nullptr, candidates); + const bool estimable = auto_selection && use_segment && + index_type == LANCE_SCALAR_BTREE && fragments.size() == 1; + EXPECT_EQ(estimable ? 1 : 0, candidates->value()); + }; + read_domain({fixture.fragment_ids[0]}, true, {2}); + read_domain({fixture.fragment_ids[1]}, false, {4}); + // A segment which cannot cover the whole task must filter the entire explicit domain. + read_domain(fixture.fragment_ids, true, {2, 4}); + // Disabling the index must produce the same rows in each read domain. + read_domain({fixture.fragment_ids[0]}, false, {2}); + read_domain(fixture.fragment_ids, false, {2, 4}); + } } } diff --git a/be/test/runtime/runtime_profile_test.cpp b/be/test/runtime/runtime_profile_test.cpp index 1b31c1b24fc9c9..fe5fbba2a4d97f 100644 --- a/be/test/runtime/runtime_profile_test.cpp +++ b/be/test/runtime/runtime_profile_test.cpp @@ -34,6 +34,31 @@ namespace doris { class RuntimeProfileTest : public testing::Test {}; +TEST(RuntimeProfileTest, SnapshotCallbacksRunBeforeSerializationWithoutProfileLocks) { + RuntimeProfile profile("snapshot"); + auto* counter = profile.add_counter("observed", TUnit::UNIT); + auto callback = profile.add_snapshot_callback([&] { + EXPECT_EQ(counter, profile.get_counter("observed")); + counter->update(1); + profile.add_info_string("available", "true"); + }); + TRuntimeProfileTree thrift; + profile.to_thrift(&thrift); + ASSERT_EQ(1, thrift.nodes.size()); + EXPECT_EQ("true", thrift.nodes[0].info_strings.at("available")); + auto restored = RuntimeProfile::from_thrift(thrift); + EXPECT_EQ(1, restored->get_counter("observed")->value()); + + PRuntimeProfileTree proto; + profile.to_proto(&proto); + auto proto_restored = RuntimeProfile::from_proto(proto); + EXPECT_EQ(2, proto_restored->get_counter("observed")->value()); + + callback.reset(); + profile.to_thrift(&thrift); + EXPECT_EQ(2, counter->value()); +} + TEST(RuntimeProfileTest, Basic) { RuntimeProfile profile_a("ProfileA"); RuntimeProfile profile_a1("ProfileA1"); diff --git a/docs/lance-cache-and-index-selection.md b/docs/lance-cache-and-index-selection.md new file mode 100644 index 00000000000000..32b934e67a564e --- /dev/null +++ b/docs/lance-cache-and-index-selection.md @@ -0,0 +1,102 @@ +# Lance cache diagnostics and scalar index selection + +## Cache misses in query profiles + +The `LanceReader / LanceSessionCache` profile group contains: + +| Counter | Meaning | +| --- | --- | +| `LanceSessionIndexCacheMisses` | Increase in the BE-wide Lance Session index-cache miss counter during this reader's dataset-open windows. | +| `LanceSessionMetadataCacheMisses` | Increase in the BE-wide Lance Session metadata-cache miss counter during the same windows, including dataset opening. | + +The reader takes a baseline before opening its dataset. Profile serialization samples +active observation windows before copying counters and availability information, even +if an early LIMIT leaves a worker blocked in a native read. Dataset close publishes +the remaining delta and ends the window. Sampling and close share a synchronized +baseline, so repeated snapshots and close do not count the same delta twice. +Reopening a dataset adds another observation window. Activity after the final Profile +snapshot is outside that saved observation. `LanceSessionCacheStatsAvailable` +is `true` only after statistics have been collected successfully for every observed +window. When it is `false`, zero counters do not establish that the caches had no misses. + +The Session is shared by all readers in one BE. These counters include activity from +concurrent readers and are **not query-exclusive miss counts**. Overlapping windows +must not be summed to obtain a query's exact misses or a cluster-wide miss total. +`LanceSessionCacheStatsScope` records this limitation in the profile. An isolated +query can still have several readers in one BE. For an exact query total, isolate +the query and compare each BE's Session metrics before and after it, rather than +summing reader windows. These counters do not measure the FE's separate metadata cache. +`LanceIndexPartitionCacheMissLoads` remains a separate, narrower execution metric. + +## Automatic scalar index selection + +`enable_lance_scalar_index_auto_selection` is a startup BE configuration, enabled +by default. Restart the BE after changing it. There is no FE session variable. + +FE retains its existing coverage-based grouping and assigns each task an explicit, +disjoint fragment domain. In BE, lance-c can replace the grouping index with a +BTree segment whose known coverage contains the **entire** task domain. It uses +the current snapshot's BTree page lookup (page min/max and null counts), without +Doris column statistics or `ANALYZE`. It estimates matching pages for equality, +range, small IN lists, IS NULL, and supported AND/OR combinations: + +```text +estimated work = matching index pages * rows per index page +``` + +This is a conservative page-based estimate, not an exact candidate count or a +latency measurement. Intersected AND predicates may still match different values +inside the same page. Cache residency and the cost of fetching projected columns +are not modeled. Only a strictly cheaper estimate replaces the grouping index; +ties retain it. If the best estimate is at least the task's physical row count, +BE scans that domain with the complete filter instead of searching an index. + +Each selection examines at most eight segments, an aggregate 8 MiB of serialized +lookup files, and a separate aggregate 8 MiB allowance for decoded lookup buffers. +The decoded allowance is checked from schema and row count before materialization +and charged for both cache hits and new loads. With the current native reader API, +only fixed-width scalar lookup columns have a known decoded-size bound. String +and variable-width binary costs remain unknown: retain such a grouping index or +skip such an alternative without opening its lookup for estimation. These indices +remain usable on the existing segment path. + +A lookup has at most 65,536 pages, an IN list at most 256 values, and +one estimate at most 1,000,000 page/predicate comparisons. Lookup +batches share the bounded Session metadata cache. Probing does not search index +leaf pages or materialize candidate row IDs. Unknown file sizes, unsupported +expressions/formats, probe errors, or an unestimable grouping index retain the +existing segment path. Bitmap and LabelList remain supported on that path, but +are not cost-ranked. Uncovered fragments keep their existing fallback scans. + +`LanceScalarIndexAutoCandidates`, `LanceScalarIndexAutoChanged`, +`LanceScalarIndexAutoEstimatedPages`, `LanceScalarIndexAutoEstimatedRows`, and +`LanceScalarIndexAutoFullScans` expose runtime decisions in Profile. Estimates +are accumulated over tasks. `EXPLAIN`'s `lanceGroupingIndex` identifies FE's task +grouping, which can differ from the runtime index. + +Setting the BE option to `false` retains the previous segment behavior and avoids +these probes. The `lance_fragments_per_split` debug override still bypasses +index-segment grouping. Vector/FTS index selection and external-search TVF filter +semantics are unchanged. + +The native selection implementation and decoded-memory guards are supplied by +upstream lance-c, pinned to an immutable commit in `thirdparty/vars.sh`. Doris +does not carry a local runtime-selection patch. + +## Foyer memory capacity + +`lance_data_cache_memory_capacity_bytes` is a startup BE configuration, independent +of `lance_data_cache_read_block_size_bytes` and the disk capacity. Restart the BE +after changing it. + +| Value | Behavior | +| --- | --- | +| `0` (default) | Use one configured read block, preserving the previous behavior. With the default block size this is 1 MiB. | +| Positive | Set the Foyer memory tier's capacity in bytes. It must be at least one read block. | +| Negative | Invalid when the Lance data cache is enabled. | + +For example, `lance_data_cache_memory_capacity_bytes=10737418240` allows a 10 GiB +memory tier while retaining 1 MiB read blocks. This capacity is shared by Lance +readers in one BE and is separate from index and metadata cache capacities. Cache +bookkeeping and in-flight reads can consume additional memory. When +`enable_lance_data_cache=false`, the Foyer configuration is ignored. diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScalarIndexPlannerTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScalarIndexPlannerTest.java index d94e2f554740ef..4f568b62f45410 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScalarIndexPlannerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/source/LanceScalarIndexPlannerTest.java @@ -53,6 +53,18 @@ import java.util.UUID; public class LanceScalarIndexPlannerTest { + @Test + public void testGroupingDoesNotDependOnPredicateSelectivity() { + Expr predicate = new BinaryPredicate(BinaryPredicate.Operator.EQ, + new SlotRef(null, "key"), new IntLiteral(1)) { + @Override + public double getSelectivity() { + throw new AssertionError("FE grouping must not estimate runtime index costs"); + } + }; + Assertions.assertEquals("key_idx", plan(predicate).indexName); + } + @Test public void testBooleanDriverSelectionPreservesAllBranches() { Expr key = equal("key", 1); diff --git a/regression-test/suites/external_table_p0/lance/test_lance_array_predicate_pushdown.groovy b/regression-test/suites/external_table_p0/lance/test_lance_array_predicate_pushdown.groovy index 7c2d21a8e029ef..efb550153cd22b 100644 --- a/regression-test/suites/external_table_p0/lance/test_lance_array_predicate_pushdown.groovy +++ b/regression-test/suites/external_table_p0/lance/test_lance_array_predicate_pushdown.groovy @@ -86,26 +86,34 @@ suite("test_lance_array_predicate_pushdown", "p0,external") { } } assertEquals(c[1], sql(query).collect { (it[0] as Number).intValue() }) - // A pushed predicate is not necessarily indexed: also verify runtime searches - // and candidate counts, so a non-indexed fallback cannot satisfy this test. if (table != "unindexed") { String profile = profiles.getProfileBySql(token, - ["LanceScalarIndexSegmentsSearched", "LanceScalarIndexCandidateRows", - "LanceScalarIndexSegmentFallbacks"]) + ["LanceScalarIndexSegmentsRequested", "LanceScalarIndexSegmentsSearched", + "LanceScalarIndexCandidateRows", "LanceScalarIndexSegmentFallbacks", + "LanceScalarIndexAutoFullScans"]) def counter = { String name -> def matches = profile =~ /${name}: (?:sum )?(\d+)\b/ assertTrue(matches.find(), "Missing counter ${name}: ${profile}") return matches.group(1).toLong() } - // Broad complements stay as parallel scans without repeated segment searches. - long expectedSearches = c[2] == -1 ? 0L : 1L - assertEquals(expectedSearches, counter("LanceScalarIndexSegmentsSearched")) - long expectedCandidates = c[2] == null ? -1L : + // The cost selector may scan a small indexed domain instead of searching. + // Account for that decision without accepting unrelated index fallbacks. + long expectedRequests = c[2] == -1 ? 0L : 1L + long runtimeFullScans = counter("LanceScalarIndexAutoFullScans") + // Only category has a cost-ranked BTree; label-only predicates + // must continue to exercise their LabelList index. + if (!c[0].contains("category")) { + assertEquals(0L, runtimeFullScans) + } + assertEquals(expectedRequests, counter("LanceScalarIndexSegmentsRequested")) + assertEquals(expectedRequests, + counter("LanceScalarIndexSegmentsSearched") + runtimeFullScans) + long expectedCandidates = runtimeFullScans > 0 ? 0L : c[2] == null ? -1L : (c[table == "partial" ? 3 : 2] as Number).longValue() if (expectedCandidates >= 0) { assertEquals(expectedCandidates, counter("LanceScalarIndexCandidateRows")) } - assertEquals(0L, counter("LanceScalarIndexSegmentFallbacks")) + assertEquals(runtimeFullScans, counter("LanceScalarIndexSegmentFallbacks")) } } // NULL needle matching and ordered subsequence matching have Doris-specific semantics. diff --git a/thirdparty/download-thirdparty.sh b/thirdparty/download-thirdparty.sh index 198e328c79cdd2..175c9b0df05220 100755 --- a/thirdparty/download-thirdparty.sh +++ b/thirdparty/download-thirdparty.sh @@ -719,7 +719,7 @@ if [[ " ${TP_ARCHIVES[*]} " =~ " AZURE " ]]; then fi # Foyer remains a local patch until its cache interface is accepted upstream. -# All search fixes are supplied by the immutable lance-c dependency revision. +# Runtime selection and its memory guards come from the pinned upstream revision. if [[ " ${TP_ARCHIVES[*]} " =~ " LANCE_C " ]]; then foyer_patch_checksum="$(cksum < "${TP_PATCH_DIR}/lance-c-foyer.patch")" foyer_patch_marker="${TP_SOURCE_DIR}/${LANCE_C_SOURCE}/${PATCHED_MARK}_foyer" diff --git a/thirdparty/vars.sh b/thirdparty/vars.sh index 55083e0983eec9..26b83f7dcc234d 100644 --- a/thirdparty/vars.sh +++ b/thirdparty/vars.sh @@ -580,11 +580,11 @@ PUGIXML_SOURCE=pugixml-1.15 PUGIXML_MD5SUM="3b894c29455eb33a40b165c6e2de5895" # lance-c -# Includes lance-c #97: resolve index/column name collisions without serializing prewarm. -LANCE_C_DOWNLOAD="https://codeload.github.com/lance-format/lance-c/tar.gz/4fafcd64525b274b113208390bbcca86bbbc093b" -LANCE_C_NAME="lance-c-4fafcd64525b274b113208390bbcca86bbbc093b.tar.gz" -LANCE_C_SOURCE="lance-c-4fafcd64525b274b113208390bbcca86bbbc093b" -LANCE_C_MD5SUM="65e924fb62137c5dd4d58af3fb33923a" +# Includes lance-c #98: bounded runtime BTree segment selection with decoded-memory guards. +LANCE_C_DOWNLOAD="https://codeload.github.com/lance-format/lance-c/tar.gz/a51ab47a9c814a746cb1714c99fad93dd3a188b8" +LANCE_C_NAME="lance-c-a51ab47a9c814a746cb1714c99fad93dd3a188b8.tar.gz" +LANCE_C_SOURCE="lance-c-a51ab47a9c814a746cb1714c99fad93dd3a188b8" +LANCE_C_MD5SUM="d1282cff3d63acfc0450db1f00d6ab6f" # paimon-rust PAIMON_RUST_DOWNLOAD="https://github.com/apache/paimon-rust/archive/refs/tags/v0.4.0-rc1.tar.gz"