morningman commented on code in PR #69581: URL: https://github.com/apache/doris/pull/69581#discussion_r4407190062
########## 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: The native working set of a Lance scan is indeed not charged to the query, but that comes with embedding lance-c and is not added by this port. `_open_scanner()`, `_configure_scan_options()`, `_fill_block_from_lance_batch()` and the converter's tracked pool are the branch-4.1 / 4.2 code, unchanged. Master's other readers of foreign code (ADBC drivers, paimon-rust, the JNI scanners' off-heap batches, Arrow Flight for remote Doris) likewise charge only Doris' own conversion and columns, never the producer's buffers. Three parts of the description are off, though. In the flat top-1 example, lance-c exports a one-row batch, so charging exported Arrow buffers would change nothing; the memory sits in the I/O buffer, the decode readahead and the distance stage inside Lance's plan, and none of that is ever exported. A bounded DataFusion pool would not cap it either. Lance's scan, decode and I/O stages take no reservation from that pool, so it would only bound operators such as the top-k s ort, which holds about one batch here. And the BE does see this memory: it counts toward process memory, which the process limit and memory GC check. What is missing is its attribution to the query and the workload group, so memory GC, which ranks queries by tracked memory, cancels other queries first. A real budget therefore needs lance-c, and for decode and I/O Lance itself, to reserve memory through a caller-supplied pool, the way the Paimon JNI writer takes its pages from Doris' allocator. The pinned C API offers row and byte batch sizes, the I/O buffer size and readahead and parallelism counts, and none of these caps the total. An estimate reserved when the scanner opens would not work either: reserving the 2 GiB I/O bound per scanner would fail ordinary scans, and any smaller estimate can still be exceeded. This is tracked as a follow-up for all branches, together with the session-cache reclamation from discussion_r4370665039, and the small-memory flat top-k test belongs with that change. The PR description now says that a Lance scan's native working set is charged to neither the query nor its workload group, and that on BEs with little memory headroom `lance_io_buffer_size_bytes`, `lance_batch_readahead` and `lance_fragment_readahead` (2 GiB, 5 and 5 per scanner, as on 4. x) should be sized for the number of concurrent Lance scanners. -- 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]
