github-actions[bot] commented on code in PR #69581: URL: https://github.com/apache/doris/pull/69581#discussion_r4400236530
########## be/src/format_v2/table/lance_reader.cpp: ########## @@ -0,0 +1,1248 @@ +// 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/table/lance_reader.h" + +#include <arrow/array.h> +#include <arrow/c/bridge.h> +#include <arrow/record_batch.h> +#include <arrow/type.h> +#include <lance/lance.h> + +#include <algorithm> +#include <bit> +#include <cmath> +#include <cstring> +#include <limits> +#include <memory> + +#include "common/config.h" +#include "common/logging.h" +#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_manager.h" +#include "runtime/file_scan_profile.h" +#include "runtime/runtime_state.h" + +namespace doris::format::lance { + +LanceTableReader::~LanceTableReader() { + static_cast<void>(close()); +} + +Status LanceTableReader::fetch_schema(const TFileRangeDesc& range, + const TFileScanRangeParams& scan_params, + std::vector<std::string>* column_names, + std::vector<DataTypePtr>* column_types) const { + if (column_names == nullptr || column_types == nullptr) { + return Status::InvalidArgument("Lance schema output must not be null"); + } + const auto& params = range.table_format_params.lance_params; + std::vector<std::string> storage_options; + RETURN_IF_ERROR(build_lance_storage_options(&scan_params, &storage_options)); + std::vector<const char*> storage_option_ptrs; + storage_option_ptrs.reserve(storage_options.size() + 1); + for (const auto& option : storage_options) { + storage_option_ptrs.emplace_back(option.c_str()); + } + storage_option_ptrs.emplace_back(nullptr); + + std::unique_ptr<LanceDataset, LanceDatasetDeleter> dataset( + lance_dataset_open(params.dataset_uri.c_str(), + storage_options.empty() ? nullptr : storage_option_ptrs.data(), + static_cast<uint64_t>(params.version))); + if (dataset == nullptr) { + return lance_error("open Lance dataset for schema"); + } + + std::shared_ptr<arrow::Schema> schema; + RETURN_IF_ERROR(import_lance_dataset_schema(dataset.get(), &schema)); + return convert_arrow_schema_to_doris(schema, column_names, column_types); +} + +Status LanceTableReader::init(TableReadOptions&& options) { + RETURN_IF_ERROR(TableReader::init(std::move(options))); + DORIS_CHECK(_runtime_state != nullptr); + DORIS_CHECK(_scanner_profile != nullptr); + DORIS_CHECK(_scan_params != nullptr); + RETURN_IF_ERROR(_resolve_search_kind()); + + 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); + _arrow_to_doris_block_time = add_lance_counter(_scanner_profile, "LanceArrowToDorisBlockTime", + TUnit::TIME_NS, LANCE_TIMING_PROFILE); + _data_cache_bytes_read_from_cache = + add_lance_counter(_scanner_profile, "LanceDataCacheBytesReadFromCache", TUnit::BYTES, + LANCE_DATA_CACHE_PROFILE); + _data_cache_bytes_read_from_remote = + add_lance_counter(_scanner_profile, "LanceDataCacheBytesReadFromRemote", TUnit::BYTES, + LANCE_DATA_CACHE_PROFILE); + if (_search_kind != SearchKind::NORMAL) { + RETURN_IF_ERROR(_validate_external_search_request()); + const auto& request = lance_scan_params.external_search_request; + int64_t top_k; + int64_t offset; + if (_search_kind == SearchKind::VECTOR) { + const auto& vector = request.search_query.vector_search; + top_k = vector.top_k; + offset = vector.offset; + _scanner_profile->add_info_string("LanceSearchType", "VECTOR"); + } else { + DORIS_CHECK(_search_kind == SearchKind::FULL_TEXT); + const auto& full_text = request.search_query.full_text_search; + top_k = full_text.top_k; + offset = full_text.offset; + _scanner_profile->add_info_string("LanceSearchType", "FULL_TEXT"); + _scanner_profile->add_info_string( + "LanceFtsCoverageMode", + full_text.coverage_mode == TFtsCoverageMode::STRICT ? "STRICT" : "INDEX_ONLY"); + if (full_text.query_type == TFtsQueryType::MATCH) { + _scanner_profile->add_info_string("LanceFtsQueryType", "MATCH"); + _scanner_profile->add_info_string( + "LanceFtsMatchOperator", + full_text.match_operator == TFtsMatchOperator::AND ? "AND" : "OR"); + _scanner_profile->add_info_string("LanceFtsMaxFuzzyDistance", + std::to_string(full_text.max_fuzzy_distance)); + } else { + _scanner_profile->add_info_string("LanceFtsQueryType", "PHRASE"); + _scanner_profile->add_info_string("LanceFtsPhraseSlop", + std::to_string(full_text.phrase_slop)); + } + } + _scanner_profile->add_info_string("LanceTopK", std::to_string(top_k)); + _scanner_profile->add_info_string("LanceOffset", std::to_string(offset)); + } + if (_scan_params->__isset.lance_scan_params && + lance_scan_params.__isset.lance_substrait_filter) { + _scanner_profile->add_info_string("LancePushdownFormat", "SUBSTRAIT"); + } + + return _record_batch_converter.init(_runtime_state, _projected_columns, _search_kind); +} + +Status LanceTableReader::prepare_split(const SplitReadOptions& options) { + _close_scanner(); + _eof = false; + _runtime_filter_cache = options.cache; + + RETURN_IF_ERROR(TableReader::prepare_split(options)); + if (current_split_pruned()) { + return Status::OK(); + } + // COUNT(*)/COUNT(1) with no filter is served from Lance metadata. The base class already set + // _remaining_table_level_count from the split's table_level_row_count, so skip opening any + // dataset scanner; get_block() synthesizes the counted rows. + if (_is_table_level_count_active()) { + return Status::OK(); + } + if (_record_batch_converter.requires_global_rowid() && !_global_rowid_context.has_value()) { + return Status::InvalidArgument( + "Lance global row id requested without global row id context"); + } + + RETURN_IF_ERROR(_ensure_dataset_open(options.current_range)); + RETURN_IF_ERROR(_open_scanner(options.current_range)); + return Status::OK(); +} + +Status LanceTableReader::get_block(Block* block, bool* eos) { + DORIS_CHECK(block != nullptr); + DORIS_CHECK(eos != nullptr); + DORIS_CHECK(block->columns() == _projected_columns.size()); + *eos = false; + + if (_eof) { + *eos = true; + return Status::OK(); + } + // Metadata COUNT(*) split: no scanner is opened. Emit synthetic rows for the upper COUNT + // operator directly from the row count the base class parsed out of the split. + if (_is_table_level_count_active()) { + return _read_table_level_count(block, eos); + } + if (_scanner == nullptr) { + return Status::InternalError("Lance scanner is not initialized for the current split"); + } + + const auto target_rows = std::max<size_t>(1, _scanner_batch_size); + while (true) { + block->clear_column_data(_projected_columns.size()); + size_t raw_rows = 0; + while (raw_rows < target_rows) { + if (_io_ctx != nullptr && _io_ctx->should_stop) { + _eof = true; + _close_scanner(); + *eos = true; + return Status::OK(); + } + + LanceBatch* raw_batch = nullptr; + int32_t scan_status = 0; + { + SCOPED_TIMER(_scanner_next_time); + scan_status = lance_scanner_next(_scanner, &raw_batch); Review Comment: [P2] Account for native query memory before starting a scan A flat `vector_search` over wide vectors, projecting only `row_id`/`_distance` with `top_k=1`, can allocate native decode, prefetch and operator buffers before this call returns, while Doris accounts only the tiny converted result. The tracked Arrow pool is used for normalization/copies; it does not charge the Rust buffers [exported through the C Data Interface](https://github.com/lance-format/lance-c/blob/a51ab47a9c814a746cb1714c99fad93dd3a188b8/src/batch.rs#L46-L50). The native scanner uses default execution options and an [unbounded DataFusion pool](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance-datafusion/src/exec.rs#L377-L397), with no Doris query/workload-group budget passed in. Consequently these scans can exceed their assigned memory limit, multiplied by concurrent scanners, even with index/metadata cache capacities zero and the data cache disabled. The I/O buffer setting does not cap decoded/operator memory. Add native reser vation/accounting and enforceable budget propagation before execution, with balanced release on errors/cancellation and a small-memory flat top-k test. This is distinct from the [persistent-session-cache report](https://github.com/apache/doris/pull/69581#discussion_r4370665039). -- 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]
