github-actions[bot] commented on code in PR #69581: URL: https://github.com/apache/doris/pull/69581#discussion_r4370665038
########## be/src/format_v2/lance/lance_record_batch_converter.cpp: ########## @@ -0,0 +1,261 @@ +// 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_record_batch_converter.h" + +#include <arrow/array.h> +#include <arrow/memory_pool.h> +#include <arrow/record_batch.h> +#include <arrow/type.h> + +#include <limits> + +#include "common/consts.h" +#include "common/exception.h" +#include "core/block/block.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/data_type/data_type_nullable.h" +#include "format_v2/lance/lance_reader_helper.h" +#include "runtime/exec_env.h" +#include "runtime/runtime_state.h" +#include "storage/utils.h" + +namespace doris::format::lance { +namespace { + +constexpr size_t IGNORED_OUTPUT_INDEX = std::numeric_limits<size_t>::max(); + +} // namespace + +arrow::MemoryPool* LanceRecordBatchConverter::_query_arrow_memory_pool( + RuntimeState* runtime_state) { + DORIS_CHECK(runtime_state != nullptr); + if (runtime_state->exec_env() != nullptr) { + if (auto* memory_pool = runtime_state->exec_env()->arrow_memory_pool(); + memory_pool != nullptr) { + return memory_pool; + } + } + return arrow::default_memory_pool(); +} + +Status LanceRecordBatchConverter::init(RuntimeState* runtime_state, + const std::vector<ColumnDefinition>& projected_columns, + SearchKind search_kind) { + DORIS_CHECK(runtime_state != nullptr); + _search_kind = search_kind; + _timezone = runtime_state->timezone_obj(); + _memory_pool = _query_arrow_memory_pool(runtime_state); + _output_columns.clear(); + _output_columns.reserve(projected_columns.size()); + _output_index_by_input_name.clear(); + _output_index_by_input_name.reserve(projected_columns.size()); + _global_rowid_output_index.reset(); + reset_schema(); + + for (const auto& column : projected_columns) { + if (column.type == nullptr) { + return Status::InvalidArgument("Lance projected column '{}' has no type", column.name); + } + if (column.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) { + if (_search_kind == SearchKind::NORMAL) { + return Status::NotSupported( + "Lance global row id is currently supported only for external search"); + } + if (_global_rowid_output_index.has_value()) { + return Status::InvalidArgument("duplicate Lance global row id projected column: {}", + column.name); + } + if (remove_nullable(column.type)->get_primitive_type() != TYPE_STRING) { + return Status::InvalidArgument( + "Lance global row id column '{}' must have Doris STRING type, but was {}", + column.name, column.type->get_name()); + } + _global_rowid_output_index = _output_columns.size(); + // Lance returns the native `_rowid`, while Doris exposes it through a generated + // global-rowid column. One input column cannot also satisfy a regular `_rowid` + // projection, so reject an existing binding instead of silently replacing it. + if (!_output_index_by_input_name + .emplace(LANCE_ROW_ID_COLUMN, *_global_rowid_output_index) + .second) { + return Status::InvalidArgument( + "Lance global row id cannot be combined with projected column '{}'", + LANCE_ROW_ID_COLUMN); + } + _output_columns.push_back({column.name, nullptr}); + continue; + } + const auto output_index = _output_columns.size(); + if (!_output_index_by_input_name.emplace(column.name, output_index).second) { + return Status::InvalidArgument("duplicate Lance projected column: {}", column.name); + } + if (_search_kind == SearchKind::VECTOR && column.name == LANCE_DISTANCE_COLUMN && + remove_nullable(column.type)->get_primitive_type() != TYPE_FLOAT) { + return Status::InvalidArgument( + "Lance vector search column '{}' must have Doris FLOAT type, but was {}", + LANCE_DISTANCE_COLUMN, column.type->get_name()); + } + if (_search_kind == SearchKind::FULL_TEXT && column.name == LANCE_SCORE_COLUMN && + remove_nullable(column.type)->get_primitive_type() != TYPE_FLOAT) { + return Status::InvalidArgument( + "Lance full-text search column '{}' must have Doris FLOAT type, but was {}", + LANCE_SCORE_COLUMN, column.type->get_name()); + } + _output_columns.push_back({column.name, column.type->get_serde()}); + } + return Status::OK(); +} + +Status LanceRecordBatchConverter::_bind_schema(const std::shared_ptr<arrow::Schema>& schema) { + DORIS_CHECK(schema != nullptr); + if (_bound_schema == schema || + (_bound_schema != nullptr && _bound_schema->Equals(*schema, true))) { + return Status::OK(); + } + + std::vector<size_t> arrow_column_to_output(static_cast<size_t>(schema->num_fields()), + IGNORED_OUTPUT_INDEX); + std::vector<LanceArrowArrayNormalizer> array_normalizer_for_arrow_column( + static_cast<size_t>(schema->num_fields())); + std::vector<bool> output_has_source_column(_output_columns.size(), false); + for (int arrow_index = 0; arrow_index < schema->num_fields(); ++arrow_index) { + const auto& field = schema->field(arrow_index); + const auto output = _output_index_by_input_name.find(field->name()); + if (output == _output_index_by_input_name.end()) { + const bool is_omitted_search_result = + (_search_kind == SearchKind::VECTOR && + field->name() == LANCE_DISTANCE_COLUMN) || + (_search_kind == SearchKind::FULL_TEXT && field->name() == LANCE_SCORE_COLUMN); + // Lance may return this generated result even when Doris did not project it. + if (is_omitted_search_result) { + continue; + } + return Status::InternalError("Lance returned unknown column '{}'", field->name()); + } + const auto output_index = output->second; + if (output_has_source_column[output_index]) { + return Status::InternalError("Lance returned duplicate column '{}'", field->name()); + } + output_has_source_column[output_index] = true; + arrow_column_to_output[arrow_index] = output_index; + RETURN_IF_ERROR(LanceArrowArrayNormalizer::create( + field, &array_normalizer_for_arrow_column[static_cast<size_t>(arrow_index)])); + } + for (size_t output_index = 0; output_index < _output_columns.size(); ++output_index) { + if (!output_has_source_column[output_index]) { + return Status::InternalError("Lance did not return requested column '{}'", + _output_columns[output_index].name); + } + } + _bound_schema = schema; + _arrow_column_to_output_index = std::move(arrow_column_to_output); + _array_normalizer_for_arrow_column = std::move(array_normalizer_for_arrow_column); + return Status::OK(); +} + +Status LanceRecordBatchConverter::convert_record_batch_to_block( + const std::shared_ptr<arrow::RecordBatch>& record_batch, Block* block, + const std::optional<GlobalRowIdContext>& global_rowid_context, size_t* rows) { + DORIS_CHECK(record_batch != nullptr); + DORIS_CHECK(block != nullptr); + DORIS_CHECK(rows != nullptr); + RETURN_IF_ERROR(_bind_schema(record_batch->schema())); + + const auto row_count = static_cast<size_t>(record_batch->num_rows()); + auto columns_guard = block->mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + for (int arrow_index = 0; arrow_index < record_batch->num_columns(); ++arrow_index) { + const auto output_index = _arrow_column_to_output_index[static_cast<size_t>(arrow_index)]; + if (output_index == IGNORED_OUTPUT_INDEX) { + continue; + } + const auto& output = _output_columns[output_index]; + const auto& arrow_column = record_batch->column(arrow_index); + if (is_global_rowid_output(output_index)) { + if (!global_rowid_context.has_value()) { + return Status::InvalidArgument( + "Lance global row id requested without global row id context"); + } + RETURN_IF_ERROR(_append_global_row_ids(arrow_column, columns[output_index], + *global_rowid_context)); + continue; + } + const auto& field = record_batch->schema()->field(arrow_index); + try { + if (arrow_column->type_id() == arrow::Type::NA) { + columns[output_index]->insert_many_defaults(row_count); + continue; + } + std::shared_ptr<arrow::Array> normalized_column; + RETURN_IF_ERROR( + _array_normalizer_for_arrow_column[static_cast<size_t>(arrow_index)] + .normalize_for_doris(arrow_column, _memory_pool, &normalized_column)); + RETURN_IF_ERROR(output.serde->read_column_from_arrow( Review Comment: [P2] Reject out-of-range timestamps before narrowing the year A valid Lance `timestamp(us)` value `2005980768000000000` represents `65537-01-01 00:00:00`, but this call returns non-null `0001-01-01`. The schema mapper selects DATETIMEV2(6), and the shared timestamp SerDe calls [`from_unixtime()`](https://github.com/apache/doris/blob/a1c36365694010ca74a1870337f16890751708a5/be/src/core/value/vdatetime_value.cpp#L2797-L2808), which passes the civil year to `unchecked_set_time(uint16_t year, ...)`, narrowing 65537 to 1. `transform_date_scale(6, 6)` only copies, and NullableSerDe retains the source's non-null bit. The repaired DATE32 check is on another conversion path and cannot catch this, including nested timestamp leaves. Validate the calendar range before narrowing and add first-out-of-range and year-wrap timestamp cases. ########## 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( Review Comment: [P2] Make persistent session caches trackable and reclaimable This singleton gives the native index and metadata caches 10 GiB and 1 GiB by default, but neither cache is charged to a Doris MemTracker or registered with CacheManager. The [pinned native session](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance/src/session.rs#L117-L126) retains these caches after dataset/query teardown; publishing size gauges does not add a reclaim path. Once indexed reads populate them, Doris's memory-pressure capacity adjustment and `/api/clear_cache/all` only visit registered CachePolicy objects, leaving these idle entries live even while other queries encounter process memory pressure. Disabling the Foyer data cache does not disable these two caches. Add process-cache accounting and a CacheManager eviction/capacity bridge, and cover reclamation after the owning queries finish. -- 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]
