[enhancement](lance) Add cache diagnostics and BE runtime index selection

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.

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);

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,

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();
 }
 

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;

be/src/format_v2/table/lance_reader.cpp

@@ -111,12 +111,26 @@ 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);
+    _session_index_cache_misses = add_lance_counter(
+            _scanner_profile, "LanceSessionIndexCacheMisses", TUnit::UNIT, "LanceSessionCache");
+    _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");
     _arrow_to_doris_block_time = add_lance_counter(_scanner_profile, "LanceArrowToDorisBlockTime",
                                                    TUnit::TIME_NS, LANCE_TIMING_PROFILE);
     _data_cache_bytes_read_from_cache =
@@ -647,8 +661,15 @@ Status LanceTableReader::_open_dataset(const DatasetKey& key) {
     std::unique_ptr<LanceDataset, LanceDatasetDeleter> dataset;
     {
         SCOPED_TIMER(_dataset_open_time);
+        auto& session = LanceSessionManager::instance();
+        _session_cache_statistics_valid =
+                session.get_cache_statistics(&_session_cache_statistics).ok();
+        if (!_session_cache_statistics_valid) {
+            _session_cache_statistics_complete = false;
+            _scanner_profile->add_info_string("LanceSessionCacheStatsAvailable", "false");
+        }
         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()) {
@@ -943,6 +964,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");
         }
@@ -1261,6 +1286,7 @@ void LanceTableReader::_close_dataset() {
         _fts_query_context = nullptr;
     }
     if (_dataset != nullptr) {
+        _collect_session_cache_statistics();
         _collect_data_cache_statistics();
         lance_dataset_close(_dataset);
         _dataset = nullptr;
@@ -1271,6 +1297,36 @@ void LanceTableReader::_close_dataset() {
     _record_batch_converter.reset_schema();
 }
 
+void LanceTableReader::_collect_session_cache_statistics() {
+    if (!_session_cache_statistics_valid) {
+        return;
+    }
+    LanceSessionManager::CacheStatistics current;
+    const auto status = LanceSessionManager::instance().get_cache_statistics(&current);
+    if (!status.ok()) {
+        _session_cache_statistics_complete = false;
+        _scanner_profile->add_info_string("LanceSessionCacheStatsAvailable", "false");
+        LOG(WARNING) << status;
+        return;
+    }
+    // Session counters are monotonic and shared. Subtract the last snapshot exactly once
+    // per window; sampling them never resets counters used by other readers or metrics.
+    const auto update = [](RuntimeProfile::Counter* counter, uint64_t before, uint64_t after) {
+        if (counter != nullptr && after >= before) {
+            const auto delta =
+                    std::min<uint64_t>(after - before, std::numeric_limits<int64_t>::max());
+            COUNTER_UPDATE(counter, static_cast<int64_t>(delta));
+        }
+    };
+    update(_session_index_cache_misses, _session_cache_statistics.index_cache_misses,
+           current.index_cache_misses);
+    update(_session_metadata_cache_misses, _session_cache_statistics.metadata_cache_misses,
+           current.metadata_cache_misses);
+    _session_cache_statistics = current;
+    _scanner_profile->add_info_string("LanceSessionCacheStatsAvailable",
+                                      _session_cache_statistics_complete ? "true" : "false");
+}
+
 void LanceTableReader::_collect_data_cache_statistics() {
     if (_dataset == nullptr) {
         return;

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_manager.h"
 #include "format_v2/table_reader.h"
 #include "runtime/runtime_profile.h"
 
@@ -92,6 +93,7 @@ class LanceTableReader final : public TableReader {
     // Collect the cumulative statistics owned by this dataset handle. The Lance-C API returns an
     // absolute snapshot, so this method replaces rather than increments the profile counters.
     void _collect_data_cache_statistics();
+    void _collect_session_cache_statistics();
     // Keep lance-c's anonymous statistics typedef out of this header. _open_scanner installs the
     // strongly typed C callback adapter before forwarding the borrowed value here.
     static void _collect_scan_statistics(void* callback_ctx, const void* opaque_statistics);
@@ -101,6 +103,11 @@ class LanceTableReader final : public TableReader {
     Status _dataset_key(const TFileRangeDesc& range, DatasetKey* key) const;
 
     LanceDataset* _dataset = nullptr;
+    LanceSessionManager::CacheStatistics _session_cache_statistics;
+    bool _session_cache_statistics_valid = false;
+    bool _session_cache_statistics_complete = true;
+    RuntimeProfile::Counter* _session_index_cache_misses = nullptr;
+    RuntimeProfile::Counter* _session_metadata_cache_misses = nullptr;
     std::shared_ptr<arrow::Schema> _dataset_schema;
     LanceScanner* _scanner = nullptr;
     ShardedKVCache* _runtime_filter_cache = nullptr;

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));