This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new c9d28771d88 [enhancement](lance) Add cache diagnostics and BE runtime
index selection (#69625)
c9d28771d88 is described below
commit c9d28771d887fcc5b6385146a9c1e1400d22da48
Author: Gabriel <[email protected]>
AuthorDate: Tue Oct 27 16:07:20 2026 +0800
[enhancement](lance) Add cache diagnostics and BE runtime index selection
(#69625)
### What problem does this PR solve?
Lance profiles do not expose Session cache-miss observation windows,
scalar-index task grouping cannot compare runtime costs, and Foyer
memory capacity is tied to the data-cache read block size.
This PR targets **branch-4.2** and adds:
- `LanceSessionIndexCacheMisses` and `LanceSessionMetadataCacheMisses`,
with availability and scope information. They report BE-wide Session
deltas observed during a reader's dataset-open windows, including
opening. Concurrent readers can overlap; these are not query-exclusive
counts and must not be summed as an exact query total.
- The startup BE option `enable_lance_scalar_index_auto_selection`
(default `true`). FE keeps coverage-based task grouping and does not
estimate selectivity or add a session variable. In BE, lance-c compares
current BTree page min/max and null-count metadata for segments covering
the entire task. Estimated work is matching pages times rows per page; a
broad estimate can select a scoped full scan. Serialized lookup bytes,
decoded buffers and probe work are bounded. Unknown estimates retain the
existing segment path. The current native API cannot bound
variable-width expansion, so string/binary costs remain unknown: retain
such a grouping index or skip such an alternative before opening its
lookup. Bitmap/LabelList remain supported on their existing path and are
not cost-ranked; vector/FTS semantics are unchanged.
- `lance_data_cache_memory_capacity_bytes`. Default `0` preserves the
previous one-configured-block memory tier, including nondefault block
sizes. Explicit capacities configure Foyer independently of block and
disk capacities; negative values are rejected before unsigned
conversion.
The native selector and decoded-memory fix are now merged upstream in
[lance-format/lance-c#98](https://github.com/lance-format/lance-c/pull/98).
This PR pins upstream main commit
`a51ab47a9c814a746cb1714c99fad93dd3a188b8`, refreshes the source archive
checksum, and removes the local runtime-selection patch together with
its application/checksum handling. The pre-existing Foyer patch is
unchanged and applies with zero fuzz.
Runtime decisions are exposed through Profile counters. Usage,
cost-model limits and cache-counter scope are documented in
`docs/lance-cache-and-index-selection.md`.
### Release note
Add Lance Session cache-miss diagnostics, bounded BE runtime BTree
selection controlled by a BE configuration, and independently
configurable Foyer memory capacity.
### Validation
- On the newly downloaded upstream archive plus the unchanged Foyer
patch, `cargo test --offline --locked` passed all 515 tests, including
82 unit tests and 385 C API tests. The normal command ignores three
C/C++ executable compilation tests. Coverage includes the compressed
long-string allocation regression, decoded-budget limits, competing
indices, scan fallback, fragment scope, NULL/deletion/stable-ID behavior
and LIMIT/OFFSET.
- C and C++ API sources passed syntax-only compilation against the newly
patched headers.
- Verified archive MD5, clean zero-fuzz patch application, cached no-op,
stale-checksum source replacement, and equality of the resulting tested
source. Selector/scanner/segment sources match merged upstream exactly.
- All nine affected C++ files passed the repository formatting runner
with clang-format 16.0.6. Shell syntax and staged whitespace checks
passed.
- Existing FE validation passed 74 focused tests and FE Checkstyle; this
dependency synchronization changes no FE code.
- Updated the array-predicate regression profile checks to account for
explicit runtime cost-selected scans while retaining request/search/scan
accounting, exact candidate counts and strict LabelList coverage. The
actual assertion block passes six valid profile paths and rejects five
invalid profiles; the previous assertion reproduces the observed
failure. The focused native runtime-selection test and full-suite Groovy
4.0.19 parsing pass. Full external regression execution remains for CI.
- Fixed Session counter publication for early LIMIT: active dataset-open
windows are sampled before Thrift/protobuf profile copies, with
synchronized delta accounting shared with close. All 14 focused C++
tests pass with ASAN/UBSAN and leak detection enabled, including a real
Lance native read held in its statistics callback while another worker
completes LIMIT 1. Disabling the snapshot hook reproduces both
missing-window regressions (native saved misses 9 versus at least 19).
The full reader and test sources pass syntax checks, and all 15 affected
PR C++ files pass clang-format 16.0.6.
- Full Doris BE build and unit-suite validation remain for CI and have
not been verified locally.
### Check List (For Author)
- Test
- [x] Unit Test
- [x] Manual test: native source/patch lifecycle and header
compatibility checks above.
- [ ] Regression test
- Behavior changed:
- [x] Yes. BE runtime BTree selection is enabled by default and can be
disabled. Foyer's default memory behavior is preserved.
- Does this need documentation?
- [x] Yes. Usage and native memory limits are documented.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/common/config.cpp | 3 +
be/src/common/config.h | 3 +
be/src/format_v2/lance/lance_reader_helper.h | 20 ++
.../lance/lance_session_cache_profile.cpp | 114 +++++++++++
.../format_v2/lance/lance_session_cache_profile.h | 53 +++++
be/src/format_v2/lance/lance_session_manager.cpp | 45 +++--
be/src/format_v2/lance/lance_session_manager.h | 16 ++
be/src/format_v2/table/lance_reader.cpp | 29 ++-
be/src/format_v2/table/lance_reader.h | 2 +
be/src/runtime/runtime_profile.cpp | 29 +++
be/src/runtime/runtime_profile.h | 11 ++
.../lance/lance_session_cache_profile_test.cpp | 219 +++++++++++++++++++++
.../format_v2/lance/lance_session_manager_test.cpp | 23 +++
be/test/format_v2/table/lance_reader_test.cpp | 211 ++++++++++++--------
be/test/runtime/runtime_profile_test.cpp | 25 +++
docs/lance-cache-and-index-selection.md | 102 ++++++++++
.../lance/source/LanceScalarIndexPlannerTest.java | 12 ++
.../test_lance_array_predicate_pushdown.groovy | 26 ++-
thirdparty/download-thirdparty.sh | 2 +-
thirdparty/vars.sh | 10 +-
20 files changed, 849 insertions(+), 106 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index bc35931fc07..98d10247eab 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 e9cca29fc25..966b91743a8 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 1f07fbd4bcb..d5fa76ede15 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 00000000000..82278d37a63
--- /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 <algorithm>
+#include <limits>
+#include <mutex>
+#include <utility>
+
+#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<std::mutex> 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<int64_t>(std::min<uint64_t>(
+ after - before,
std::numeric_limits<int64_t>::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<State>(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<std::mutex> 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<std::mutex> 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 00000000000..f4d301d7399
--- /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 <functional>
+#include <memory>
+
+#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<Status(LanceSessionManager::CacheStatistics*)>;
+
+ 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> _state;
+ std::shared_ptr<RuntimeProfile::SnapshotCallback> _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 884eb5db18c..53b0945af7c 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<uint64_t>(_config.lance_data_cache_read_block_size_bytes),
+
static_cast<uint64_t>(_config.resolved_data_cache_memory_capacity_bytes()),
.disk_capacity_bytes =
static_cast<uint64_t>(_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 45dc62acad4..d1773c2a34a 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 @@ public:
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 @@ public:
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 e0e2884efe7..752ce7460c2 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<LanceSessionCacheProfile>(
+ _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<LanceDataset, LanceDatasetDeleter> 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<uint64_t>(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 205c082d120..ba09954afb5 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 @@ private:
Status _dataset_key(const TFileRangeDesc& range, DatasetKey* key) const;
LanceDataset* _dataset = nullptr;
+ std::unique_ptr<LanceSessionCacheProfile> _session_cache_profile;
std::shared_ptr<arrow::Schema> _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 311d434f6bc..bbd1fc85d85 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::SnapshotCallback>
RuntimeProfile::add_snapshot_callback(
+ SnapshotCallback callback) {
+ auto registration =
std::make_shared<SnapshotCallback>(std::move(callback));
+ std::lock_guard<std::mutex> l(_counter_map_lock);
+ _snapshot_callbacks.emplace_back(registration);
+ return registration;
+}
+
+void RuntimeProfile::_run_snapshot_callbacks() {
+ std::vector<std::shared_ptr<SnapshotCallback>> callbacks;
+ {
+ std::lock_guard<std::mutex> 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<TRuntimeProfileNode>* 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<PRuntimeProfileNode>*
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 e63480711ac..0873cc6ed9b 100644
--- a/be/src/runtime/runtime_profile.h
+++ b/be/src/runtime/runtime_profile.h
@@ -584,6 +584,13 @@ public:
// the key does not exist.
const std::string* get_info_string(const std::string& key);
+ using SnapshotCallback = std::function<void()>;
+ // 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<SnapshotCallback> 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 @@ private:
// 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<std::weak_ptr<SnapshotCallback>> _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 00000000000..425779d487c
--- /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 <gtest/gtest.h>
+#include <lance/lance.h>
+
+#include <atomic>
+#include <barrier>
+#include <chrono>
+#include <future>
+#include <thread>
+
+#include "util/defer_op.h"
+
+namespace doris::format::lance {
+
+TEST(LanceSessionCacheProfileTest,
EarlyLimitSnapshotsWhileAnotherWorkerIsInNativeRead) {
+ std::unique_ptr<LanceSession, decltype(&lance_session_close)> 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<LanceDataset, decltype(&lance_dataset_close)> 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<LanceDataset, decltype(&lance_dataset_close)> 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<LanceScanner, decltype(&lance_scanner_close)> slow_scanner(
+ lance_scanner_new(slow_dataset.get(), columns, nullptr),
lance_scanner_close);
+ ASSERT_NE(nullptr, slow_scanner);
+ struct BlockedRead {
+ std::promise<void> entered;
+ std::shared_future<void> released;
+ std::atomic<bool> held = false;
+ };
+ std::promise<void> 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<BlockedRead*>(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<LanceScanner, decltype(&lance_scanner_close)> 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<uint64_t> 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 e1274219347..dde7a68e19b 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 6531fc21de7..c27190b8fe7 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 <arrow/type.h>
#include <arrow/util/decimal.h>
#include <arrow/util/key_value_metadata.h>
+#include <gtest/gtest.h>
#include <lance/lance.h>
#include <algorithm>
@@ -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<std::string> 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<int64_t>& fragments,
bool use_segment,
- const std::vector<int64_t>& 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<int64_t> 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<const
ColumnInt64&>(*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<std::string> 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<int64_t>&
fragments, bool use_segment,
+ const std::vector<int64_t>& 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<int64_t> 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<const
ColumnInt64&>(*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 1b31c1b24fc..fe5fbba2a4d 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 00000000000..32b934e67a5
--- /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 d94e2f55474..4f568b62f45 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.Map;
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 7c2d21a8e02..efb550153cd 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 198e328c79c..175c9b0df05 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 55083e0983e..26b83f7dcc2 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";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
