Gabriel39 commented on code in PR #67650: URL: https://github.com/apache/doris/pull/67650#discussion_r3975326807
########## fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScalarIndexPlanner.java: ########## @@ -0,0 +1,139 @@ +// 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. + +package org.apache.doris.datasource.lance.source; + +import org.apache.doris.analysis.CompoundPredicate; +import org.apache.doris.analysis.Expr; +import org.apache.doris.analysis.SlotRef; +import org.apache.doris.datasource.lance.LanceFragmentInfo; +import org.apache.doris.datasource.lance.LanceIndexSegmentInfo; +import org.apache.doris.datasource.lance.LanceTableMetadata; + +import org.lance.index.IndexType; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.stream.Collectors; + +/** Assigns one BTree/Bitmap/LabelList segment and a disjoint fragment domain to each ordinary scan task. */ +final class LanceScalarIndexPlanner { + static final class Plan { + final String indexName; + final IndexSegmentSplitPlan splits; + private final long coveredRows; + + Plan(String indexName, IndexSegmentSplitPlan splits, long coveredRows) { + this.indexName = indexName; + this.splits = splits; + this.coveredRows = coveredRows; + } + } + + static Plan plan(LanceTableMetadata metadata, List<Expr> pushedConjuncts, + Map<Long, LanceFragmentInfo> visibleFragments) { + if (metadata.getVersion() <= 0) { + return null; + } + Set<Integer> filterFields = collectFilterFields(metadata, pushedConjuncts); + if (filterFields.isEmpty()) { + return null; + } + // Group all segments of each logical index before checking coverage. Name order + // provides a stable winner when multiple indices cover the same number of rows. + Map<String, List<LanceIndexSegmentInfo>> indices = metadata.getIndexSegments().stream() + .collect(Collectors.groupingBy(LanceIndexSegmentInfo::getIndexName, + TreeMap::new, Collectors.toList())); + Plan selected = null; + for (List<LanceIndexSegmentInfo> segments : indices.values()) { + // PR #79 supports one top-level key in BTree/Bitmap/LabelList indices. Lance + // performs the final typed driver selection and falls back within the same domain. + LanceIndexSegmentInfo index = segments.get(0); + if ((index.getIndexType() != IndexType.BTREE && index.getIndexType() != IndexType.BITMAP + && index.getIndexType() != IndexType.LABEL_LIST) + || index.getFieldIds().size() != 1 || !filterFields.contains(index.getFieldIds().get(0))) { + continue; + } + Plan candidate = groupFragments(metadata, segments, visibleFragments); + if (candidate != null && (selected == null || candidate.coveredRows > selected.coveredRows)) { Review Comment: [P1] Preserve multi-index intersection for conjunctive scalar filters\n\nWhen two pushed conjuncts reference different indexed columns, fully covered indexes normally have the same , so this selects whichever index name sorts first. The selected segment path later unconditionally calls and searches only one leaf belonging to that logical index. As a result, a filter such as no longer lets Lance evaluate and intersect both scalar indexes. If the alphabetically selected predicate is unselective, every task can read nearly its entire segment domain and evaluate the other indexed predicate as a residual, which is a potentially severe regression from ordinary Lance scalar planning. Please either support segment-local intersections, select a driver using selectivity while retaining safe use of the other indexes, or avoid segment mode when multiple independently indexable conjuncts exist. Please also add a two-index test that verifies index-search and data-read work, not only result eq uivalence. ########## be/src/format_v2/lance/lance_session_manager.cpp: ########## @@ -0,0 +1,239 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/lance/lance_session_manager.h" + +#include <lance/lance.h> + +#include <algorithm> +#include <limits> +#include <string> +#include <string_view> +#include <utility> + +#include "common/config.h" +#include "common/logging.h" +#include "common/metrics/doris_metrics.h" +#include "common/metrics/metrics.h" +#include "format_v2/lance/lance_reader_helper.h" + +namespace doris::format::lance { +namespace { + +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_hits_total, MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_misses_total, + MetricUnit::OPERATIONS); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_hits_total, + MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_misses_total, + MetricUnit::OPERATIONS); + +constexpr std::string_view LANCE_SESSION_CACHE_METRICS_HOOK = "lance_session_cache"; + +int64_t metric_value(uint64_t value) { + return static_cast<int64_t>( + std::min(value, static_cast<uint64_t>(std::numeric_limits<int64_t>::max()))); +} + +LanceSessionManager::Config load_lance_session_config() { + return { + .lance_index_cache_size_bytes = config::lance_index_cache_size_bytes, + .lance_metadata_cache_size_bytes = config::lance_metadata_cache_size_bytes, + .enable_lance_data_cache = config::enable_lance_data_cache, + .lance_data_cache_path = config::lance_data_cache_path, + .lance_data_cache_disk_capacity_bytes = config::lance_data_cache_disk_capacity_bytes, + .lance_data_cache_read_block_size_bytes = + config::lance_data_cache_read_block_size_bytes, + }; +} + +} // namespace + +class LanceSessionMetrics final { +public: + LanceSessionMetrics(LanceSession* session, const LanceSessionManager::Config& config) + : _session(session), _entity(DorisMetrics::instance()->server_entity()) { + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_misses_total); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_misses_total); + + lance_session_index_cache_capacity_bytes->set_value(config.lance_index_cache_size_bytes); + lance_session_metadata_cache_capacity_bytes->set_value( + config.lance_metadata_cache_size_bytes); + _entity->register_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK), + [this]() { update(); }); + update(); + } + + ~LanceSessionMetrics() { + _entity->deregister_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK)); + METRIC_DEREGISTER(_entity, lance_session_index_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_index_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_index_cache_misses_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_misses_total); + } + +private: + void update() { + // Session caches are shared across queries, so publish one process-wide snapshot instead + // of attributing concurrent cache activity to an individual query profile. + LanceSessionCacheStats stats {}; + if (lance_session_get_cache_stats(_session, &stats) != 0) { + LOG_EVERY_N(WARNING, 100) + << lance_error("collect Lance session cache statistics").to_string(); + return; + } + lance_session_index_cache_usage_bytes->set_value( + metric_value(stats.index_cache_size_bytes)); + lance_session_index_cache_entries->set_value(metric_value(stats.index_cache_entries)); + lance_session_index_cache_hits_total->set_value(metric_value(stats.index_cache_hits)); + lance_session_index_cache_misses_total->set_value(metric_value(stats.index_cache_misses)); + lance_session_metadata_cache_usage_bytes->set_value( + metric_value(stats.metadata_cache_size_bytes)); + lance_session_metadata_cache_entries->set_value(metric_value(stats.metadata_cache_entries)); + lance_session_metadata_cache_hits_total->set_value(metric_value(stats.metadata_cache_hits)); + lance_session_metadata_cache_misses_total->set_value( + metric_value(stats.metadata_cache_misses)); + } + + LanceSession* _session; + MetricEntity* _entity; + IntGauge* lance_session_index_cache_capacity_bytes = nullptr; + IntGauge* lance_session_index_cache_usage_bytes = nullptr; + IntGauge* lance_session_index_cache_entries = nullptr; + IntCounter* lance_session_index_cache_hits_total = nullptr; + IntCounter* lance_session_index_cache_misses_total = nullptr; + IntGauge* lance_session_metadata_cache_capacity_bytes = nullptr; + IntGauge* lance_session_metadata_cache_usage_bytes = nullptr; + IntGauge* lance_session_metadata_cache_entries = nullptr; + IntCounter* lance_session_metadata_cache_hits_total = nullptr; + IntCounter* lance_session_metadata_cache_misses_total = nullptr; +}; + +LanceSessionManager& LanceSessionManager::instance() { + // Function-local static initialization is thread safe. Cache configuration is process scoped, + // so changing it requires a BE restart. + static LanceSessionManager manager(load_lance_session_config()); + return manager; +} + +LanceSessionManager::LanceSessionManager(Config config) : _config(std::move(config)) { + LOG(INFO) << "Creating BE-wide Lance session manager: lance_index_cache_size_bytes=" + << _config.lance_index_cache_size_bytes + << ", lance_metadata_cache_size_bytes=" << _config.lance_metadata_cache_size_bytes + << ", enable_lance_data_cache=" << _config.enable_lance_data_cache + << ", lance_data_cache_path=" << _config.lance_data_cache_path + << ", lance_data_cache_disk_capacity_bytes=" + << _config.lance_data_cache_disk_capacity_bytes + << ", lance_data_cache_read_block_size_bytes=" + << _config.lance_data_cache_read_block_size_bytes + << ", foyer_memory_capacity_bytes=" << _config.lance_data_cache_read_block_size_bytes; +} + +LanceSessionManager::~LanceSessionManager() { + _metrics.reset(); + lance_session_close(_session); +} + +Status LanceSessionManager::_initialize() { + if (_config.enable_lance_data_cache) { + // Treat the configured cache mode as a process-level requirement. If an enabled + // data cache cannot initialize, report the failure instead of silently creating a + // session without it. This keeps directory/configuration/device failures visible + // to the operator. Disabling the cache is an explicit configuration change. + const LanceDataCacheOptions data_cache_options { + .directory = _config.lance_data_cache_path.c_str(), + // Foyer's HybridCache requires a memory tier. Keep it at the minimum useful + // capacity of exactly one range-cache block; WriteOnInsertion enqueues disk + // writes on insertion rather than waiting for memory-tier eviction. + .memory_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + .disk_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_disk_capacity_bytes), + .read_block_size_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + }; + _session = lance_session_new_with_data_cache( + static_cast<uint64_t>(_config.lance_index_cache_size_bytes), Review Comment: [P2] Validate signed cache sizes before converting them to \n\nThese values come from configuration entries that currently have no validators. On a 64-bit target, a value such as becomes here: index and metadata caches become effectively unbounded, while negative disk or block sizes make the default-on Foyer initialization fail and, because the failure is retained by , disable every Lance read until the BE restarts. Please add non-negative or positive validators for all cache-size settings and reject invalid values before crossing the FFI boundary, with unit coverage for negative values. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
