github-actions[bot] commented on code in PR #67650:
URL: https://github.com/apache/doris/pull/67650#discussion_r3972952472
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java:
##########
@@ -302,24 +317,26 @@ private Map<Long, LanceFragmentInfo>
getVisibleFragments(LanceTableMetadata meta
return visible;
}
- private List<Split> createFragmentSplits(LanceTableMetadata metadata,
- Map<Long, LanceFragmentInfo> visibleFragments) {
- long targetRows = 1;
- for (LanceFragmentInfo fragment : visibleFragments.values()) {
- targetRows = Math.max(targetRows,
Math.max(fragment.getPhysicalRows(), 1));
- }
-
- // Keep one fragment per split. Use the largest fragment's physical
row count as the
- // normalization baseline for split weights, so backend scheduling
reflects the relative
- // amount of physical data each fragment scans, including rows covered
by deletion metadata.
- List<Split> splits = new ArrayList<>(visibleFragments.size());
- for (LanceFragmentInfo fragment : visibleFragments.values()) {
- LanceSplit split =
LanceSplit.forFragment(metadata.getDatasetUri(), metadata.getVersion(),
- fragment.getId(), fragment.getPhysicalRows());
- split.setTargetSplitSize(targetRows);
- splits.add(split);
+ private List<Split> createNormalFragmentSplits(LanceTableMetadata metadata,
+ Map<Long, LanceFragmentInfo> visibleFragments, int numBackends) {
+ if (plannedFragmentsPerSplit > 0) {
+ // Keep the debug grouping exact, even when it produces fewer
splits than BEs.
+ IndexSegmentSplitPlan plan = new
IndexSegmentSplitPlan(metadata.getDatasetUri(), metadata.getVersion(), 0);
+ plan.addUncoveredFragments(visibleFragments.values(),
plannedFragmentsPerSplit);
+ return plan.buildSplits();
+ }
+ scalarIndexPlan = LanceScalarIndexPlanner.plan(metadata,
lancePushedConjuncts, visibleFragments);
Review Comment:
[P1] Keep scalar-segment plans executable on old BEs
This new normal-scan path serializes its segment UUID in the pre-existing
`index_segment_uuids` field. A base-version BE does not ignore that field: its
`_configure_normal_scan` explicitly returns `InvalidArgument("normal Lance scan
cannot contain index segment UUIDs")` whenever it is nonempty. Thus a new FE
can send every indexed ordinary split to a smooth-upgrade source BE and fail
the query instead of falling back to the explicit fragment domain. Please use a
new scalar-specific optional field that old BEs can ignore, or gate this
planning mode on a proven BE capability, and cover the new-FE/base-BE contract.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -673,12 +699,11 @@ Status LanceTableReader::_open_dataset(const DatasetKey&
key) {
std::unique_ptr<LanceDataset, LanceDatasetDeleter> dataset;
{
SCOPED_TIMER(_dataset_open_time);
- dataset.reset(lance_dataset_open(
+ LanceDataset* raw_dataset = nullptr;
+ RETURN_IF_ERROR(LanceSessionManager::instance().open_dataset(
Review Comment:
[P1] Make session metadata generation-safe before sharing it
This call now routes every dataset open through one BE-wide Lance session,
but the pinned Lance v11 metadata cache is partitioned only by URI.
`RowIdIndexKey` serializes only `manifest.version`, while the stable-row-ID
`RowAddrMaskKey` uses only that version plus a fragment-subset hash. Dropping
and recreating a dataset at the same URI restarts the version, so indexed/take
and deletion-prefilter paths can consume the prior incarnation's ID mapping or
allow/block mask and return wrong rows or fail. Upstream's later eTag addition
to `RowIdIndexKey` explicitly confirms this previous-incarnation collision, but
that key alone does not protect the mask. Please make every
generation-sensitive entry safe (or avoid/invalidate cross-generation sharing)
and test a warm-cache drop/recreate at the same URI/version.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataLoader.java:
##########
@@ -64,17 +66,10 @@ public static LanceTableMetadata loadLatestForTvf(
* <p>Called by
* {@link LanceExternalCatalog#loadTableMetadata(String, String,
java.util.Optional)} when no
* time-travel version is requested. Schema, version, and fragments are
read from the same
- * opened dataset snapshot.
+ * opened dataset snapshot, together with index coverage for fragment
grouping and external searches.
*/
public static LanceTableMetadata loadLatest(String datasetUri,
Map<String, String> lanceStorageOptions, BufferAllocator
allocator) throws Exception {
- return loadInternal(
- datasetUri, lanceStorageOptions, OptionalLong.empty(),
allocator, false);
- }
-
- /** Loads the latest fixed snapshot together with search-index segment
coverage. */
- public static LanceTableMetadata loadLatestWithIndexSegments(
- String datasetUri, Map<String, String> lanceStorageOptions,
BufferAllocator allocator) throws Exception {
return loadInternal(
datasetUri, lanceStorageOptions, OptionalLong.empty(),
allocator, true);
Review Comment:
[P1] Keep ordinary scans independent of index discovery
Passing `true` here makes `describeUserIndexes` part of every catalog
metadata load, including `LanceExternalTable.initSchema`, unfiltered scans,
row-count fetches, and time travel. That helper intentionally throws on
index/provider inconsistencies and on datasets with more than 256 logical
indexes, so index-only metadata can now make otherwise readable data and schema
unavailable; before this change only the search-specific loader took that
dependency. Scalar segment grouping is an optimization, so please make its
discovery best-effort for ordinary loads (fall back to the already-read fixed
schema/version/fragments), while retaining strict discovery where vector/FTS
semantics require it.
##########
be/src/common/config.cpp:
##########
@@ -1206,6 +1206,26 @@ DEFINE_Validator(variant_max_json_key_length,
DEFINE_Validator(variant_storage_parse_mode,
[](const int config) -> bool { return config >= 0 && config
<= 2; });
+// Lance uses one BE-wide session so metadata/index caches and the optional
Foyer data-file cache
+// 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_data_cache, "true");
Review Comment:
[P1] Fix short cache hits before enabling this by default
The activated Foyer adapter's `assemble_range` treats an incomplete cache
value as a successful hit. In its single-block branch, `end` is clamped to
`block.len()`, so a cached 32 KiB value for a requested 64 KiB block returns
only 32 KiB whenever the start is present. The multi-block branch likewise
breaks on a short block and returns the accumulated output without checking
`requested_len`. Because `cached_ranges` falls back to origin only on `Err`, a
partial/corrupt entry becomes truncated Lance input instead of a cache miss.
Please require exact requested coverage (and verify the final assembled length)
before making this cache default-on.
##########
thirdparty/download-thirdparty.sh:
##########
@@ -718,19 +718,19 @@ if [[ " ${TP_ARCHIVES[*]} " =~ " AZURE " ]]; then
echo "Finished patching ${AZURE_SOURCE}"
fi
-# Apply Doris lance-c patches.
+# Apply Doris lance-c patches as one chain to the pinned release archive.
if [[ " ${TP_ARCHIVES[*]} " =~ " LANCE_C " ]]; then
- if [[ "${LANCE_C_SOURCE}" == "lance-c-0.1.9" ]]; then
- cd "${TP_SOURCE_DIR}/${LANCE_C_SOURCE}"
- if [[ ! -f "${PATCHED_MARK}" ]]; then
- patch --batch --forward --reject-file=- --fuzz=0
--no-backup-if-mismatch -s \
- -p1 <"${TP_PATCH_DIR}/lance-c-0.1.9-pr-73.patch"
+ cd "${TP_SOURCE_DIR}/${LANCE_C_SOURCE}"
+ if [[ ! -f "${PATCHED_MARK}" ]]; then
Review Comment:
[P1] Version the Lance patch marker when changing the patch set
A source tree already prepared by the base revision contains
`thirdparty/src/lance-c-0.1.9/patched_mark`. Because this PR keeps both
`LANCE_C_SOURCE` and the generic marker name unchanged, the condition here
skips every newly added/updated patch on the next `download-thirdparty.sh` run,
while the unpack stage also preserves the existing directory. Incremental
builders can therefore compile the old Lance v10/C ABI even though this
revision's BE expects the new session/scalar-index symbols and FE is on Lance
v11. Please key the marker/source directory to the patch-set revision (or
otherwise invalidate and re-extract it when the chain changes).
##########
be/src/format_v2/lance/lance_session_manager.cpp:
##########
@@ -0,0 +1,218 @@
+// 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) {
+ 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; entries use
WriteOnInsertion and
+ // are persisted to the disk tier immediately.
+ .memory_capacity_bytes =
+
static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes),
+ .disk_capacity_bytes =
+
static_cast<uint64_t>(_config.lance_data_cache_disk_capacity_bytes),
+ .read_block_size_bytes =
+
static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes),
+ };
+ _session = lance_session_new_with_data_cache(
Review Comment:
[P1] Include the effective object store in the data-cache namespace
This creates one Foyer cache for all BE datasets, but the adapter keys
blocks only by Lance's `store_prefix` plus object path. In the pinned Lance v11
code, S3 inherits the default prefix `s3$<bucket>`, which ignores storage
options such as `aws_endpoint`; Doris forwards those options independently for
each catalog. Thus two S3-compatible endpoints containing the same bucket/path
generate identical cache keys: after endpoint A warms a `.lance` data file, a
query against endpoint B can return A's cached size/bytes without consulting B.
Please add a stable identity for the effective endpoint/account to the cache
namespace (without embedding secrets), or fix the provider prefix before
sharing this cache, and test two stores with identical URI paths but different
contents.
##########
be/src/format_v2/lance/lance_session_manager.cpp:
##########
@@ -0,0 +1,218 @@
+// 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) {
+ 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; entries use
WriteOnInsertion and
+ // are persisted to the disk tier immediately.
+ .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),
+ static_cast<uint64_t>(_config.lance_metadata_cache_size_bytes),
+ &data_cache_options);
+ } else {
+ _session =
+
lance_session_new(static_cast<uint64_t>(_config.lance_index_cache_size_bytes),
+
static_cast<uint64_t>(_config.lance_metadata_cache_size_bytes));
+ }
+ if (_session == nullptr) {
+ return lance_error("create shared Lance session");
+ }
+ _metrics = std::make_unique<LanceSessionMetrics>(_session, _config);
+ return Status::OK();
+}
+
+Status LanceSessionManager::open_dataset(const char* uri, const char* const*
storage_options,
+ uint64_t version, LanceDataset**
dataset) {
+ if (uri == nullptr || dataset == nullptr) {
+ return Status::InvalidArgument("Lance dataset URI and output must not
be null");
+ }
+ *dataset = nullptr;
+
+ std::call_once(_initialize_once, [this] { _initialize_status =
_initialize(); });
Review Comment:
[P1] Fall back when the optional data cache cannot initialize
With the new default, `_initialize()` only attempts
`lance_session_new_with_data_cache`. Any cache-specific failure (for example an
inaccessible configured path or a recovery/device I/O error) is stored by this
`call_once`, and every later Lance reader in the BE returns the same failure
even though `lance_session_new` could still serve the table without the disk
cache. Please treat cache initialization as best-effort and fall back to the
ordinary shared session after logging/recording the cause (or validate it as an
explicit BE-startup fatal condition), and cover the failure path.
--
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]
