morningman commented on code in PR #69581: URL: https://github.com/apache/doris/pull/69581#discussion_r4370603201
########## be/src/format_v2/lance/lance_reader_helper.cpp: ########## @@ -0,0 +1,750 @@ +// 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_reader_helper.h" + +#include <arrow/array.h> +#include <arrow/builder.h> +#include <arrow/c/bridge.h> +#include <arrow/extension_type.h> +#include <arrow/type.h> +#include <arrow/util/key_value_metadata.h> +#include <fmt/format.h> +#include <lance/lance.h> + +#include <bit> +#include <limits> +#include <optional> +#include <unordered_set> + +#include "common/config.h" +#include "common/logging.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_factory.hpp" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_nothing.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_struct.h" +#include "core/data_type_serde/arrow_validation.h" +#include "exec/common/endian.h" + +namespace doris::format::lance { +namespace { + +constexpr std::string_view ARROW_EXTENSION_NAME = "ARROW:extension:name"; +constexpr std::string_view ARROW_JSON_EXTENSION = "arrow.json"; +constexpr std::string_view LANCE_JSON_EXTENSION = "lance.json"; +constexpr std::string_view LANCE_BFLOAT16_EXTENSION = "lance.bfloat16"; + +enum class LanceExtensionKind { + NONE, + JSON, + BFLOAT16, +}; + +int arrow_time_precision(arrow::TimeUnit::type unit) { + switch (unit) { + case arrow::TimeUnit::SECOND: + return 0; + case arrow::TimeUnit::MILLI: + return 3; + case arrow::TimeUnit::MICRO: + case arrow::TimeUnit::NANO: + return 6; + } + return 6; +} + +// Extract and validate extension names from metadata and registered Arrow types. +Status get_lance_extension(const std::shared_ptr<arrow::Field>& field, + LanceExtensionKind* extension_kind, + std::shared_ptr<arrow::DataType>* storage_type) { + DORIS_CHECK(field != nullptr); + DORIS_CHECK(extension_kind != nullptr); + DORIS_CHECK(storage_type != nullptr); + *extension_kind = LanceExtensionKind::NONE; + *storage_type = field->type(); + + std::optional<std::string> extension_name; + if (field->HasMetadata()) { + const auto metadata_name = field->metadata()->Get(ARROW_EXTENSION_NAME); + if (metadata_name.ok() && !metadata_name.ValueUnsafe().empty()) { + extension_name = metadata_name.ValueUnsafe(); + } + } + if (field->type()->id() == arrow::Type::EXTENSION) { + const auto extension_type = std::dynamic_pointer_cast<arrow::ExtensionType>(field->type()); + if (extension_type == nullptr) { + return Status::InvalidArgument("invalid Arrow extension type for Lance field '{}'", + field->name()); + } + if (extension_name.has_value() && *extension_name != extension_type->extension_name()) { + return Status::InvalidArgument( + "conflicting Arrow extension names for Lance field '{}': '{}' and '{}'", + field->name(), *extension_name, extension_type->extension_name()); + } + extension_name = extension_type->extension_name(); + *storage_type = extension_type->storage_type(); + } + if (!extension_name.has_value()) { + return Status::OK(); + } + + if (*extension_name == ARROW_JSON_EXTENSION) { + if ((*storage_type)->id() != arrow::Type::STRING && + (*storage_type)->id() != arrow::Type::LARGE_STRING) { + return Status::NotSupported( + "Arrow JSON extension for Lance field '{}' requires UTF8 storage, got {}", + field->name(), (*storage_type)->ToString()); + } + *extension_kind = LanceExtensionKind::JSON; + return Status::OK(); + } + if (*extension_name == LANCE_JSON_EXTENSION) { + if ((*storage_type)->id() != arrow::Type::LARGE_BINARY) { + return Status::NotSupported( + "Lance JSON extension for field '{}' requires LARGE_BINARY storage, got {}", + field->name(), (*storage_type)->ToString()); + } + *extension_kind = LanceExtensionKind::JSON; + return Status::OK(); + } + if (*extension_name == LANCE_BFLOAT16_EXTENSION) { + if ((*storage_type)->id() != arrow::Type::FIXED_SIZE_BINARY || + std::static_pointer_cast<arrow::FixedSizeBinaryType>(*storage_type)->byte_width() != + 2) { + return Status::NotSupported( + "Lance BFloat16 extension for field '{}' requires FIXED_SIZE_BINARY(2) " + "storage, got {}", + field->name(), (*storage_type)->ToString()); + } + *extension_kind = LanceExtensionKind::BFLOAT16; + return Status::OK(); + } + return Status::NotSupported("unsupported Lance Arrow extension type '{}' for field '{}'", + *extension_name, field->name()); +} + +// Map Arrow fields recursively, preserving Null leaves in complex types. +Status arrow_field_to_doris_type(const std::shared_ptr<arrow::Field>& field, + DataTypePtr* doris_type) { + const auto nullable_primitive = [&](PrimitiveType type, int precision = 0, int scale = 0, + int len = -1) { + *doris_type = + DataTypeFactory::instance().create_data_type(type, true, precision, scale, len); + return Status::OK(); + }; + + LanceExtensionKind extension_kind; + std::shared_ptr<arrow::DataType> arrow_type; + RETURN_IF_ERROR(get_lance_extension(field, &extension_kind, &arrow_type)); + switch (extension_kind) { + case LanceExtensionKind::JSON: + return nullable_primitive(TYPE_JSONB); + case LanceExtensionKind::BFLOAT16: + return nullable_primitive(TYPE_FLOAT); + case LanceExtensionKind::NONE: + break; + } + if (arrow_type->id() == arrow::Type::DICTIONARY) { + return Status::NotSupported("unsupported Lance Arrow dictionary type for field '{}': {}", + field->name(), arrow_type->ToString()); + } + + switch (arrow_type->id()) { + case arrow::Type::NA: + // Required Null leaves can bypass NullableSerDe through FE schema reconstruction. + // Reject them before Arrow NA buffers reach the Boolean-backed physical SerDe. + if (!field->nullable()) { + return Status::NotSupported("non-nullable Lance Arrow Null field: {}", field->name()); + } + return nullable_primitive(TYPE_NULL); + case arrow::Type::BOOL: + return nullable_primitive(TYPE_BOOLEAN); + case arrow::Type::INT8: + return nullable_primitive(TYPE_TINYINT); + case arrow::Type::UINT8: + case arrow::Type::INT16: + return nullable_primitive(TYPE_SMALLINT); + case arrow::Type::UINT16: + case arrow::Type::INT32: + return nullable_primitive(TYPE_INT); + case arrow::Type::UINT32: + case arrow::Type::INT64: + return nullable_primitive(TYPE_BIGINT); + case arrow::Type::UINT64: + return nullable_primitive(TYPE_LARGEINT); + case arrow::Type::HALF_FLOAT: + case arrow::Type::FLOAT: + return nullable_primitive(TYPE_FLOAT); + case arrow::Type::DOUBLE: + return nullable_primitive(TYPE_DOUBLE); + case arrow::Type::STRING: + case arrow::Type::LARGE_STRING: + return nullable_primitive(TYPE_STRING); + case arrow::Type::BINARY: + case arrow::Type::LARGE_BINARY: + return nullable_primitive(TYPE_VARBINARY, 0, 0, std::numeric_limits<int32_t>::max()); + case arrow::Type::FIXED_SIZE_BINARY: { + const auto binary = std::static_pointer_cast<arrow::FixedSizeBinaryType>(arrow_type); + return nullable_primitive(TYPE_VARBINARY, 0, 0, binary->byte_width()); + } + case arrow::Type::DATE32: + case arrow::Type::DATE64: + return nullable_primitive(TYPE_DATEV2); + case arrow::Type::TIME32: + case arrow::Type::TIME64: { + const auto time = std::static_pointer_cast<arrow::TimeType>(arrow_type); + return nullable_primitive(TYPE_TIMEV2, 0, arrow_time_precision(time->unit())); + } + case arrow::Type::TIMESTAMP: { + const auto timestamp = std::static_pointer_cast<arrow::TimestampType>(arrow_type); + const auto timestamp_type = + timestamp->timezone().empty() ? TYPE_DATETIMEV2 : TYPE_TIMESTAMPTZ; + return nullable_primitive(timestamp_type, 0, arrow_time_precision(timestamp->unit())); + } + case arrow::Type::DURATION: + return nullable_primitive(TYPE_BIGINT); + case arrow::Type::DECIMAL128: + case arrow::Type::DECIMAL256: { + const auto decimal = std::static_pointer_cast<arrow::DecimalType>(arrow_type); + const int precision = decimal->precision(); + const int scale = decimal->scale(); + if (precision <= 0 || precision > arrow::Decimal256Type::kMaxPrecision || scale < 0 || + scale > precision) { + return Status::NotSupported( + "unsupported Lance Arrow decimal type for field '{}': precision={}, scale={}", + field->name(), precision, scale); + } + const PrimitiveType doris_decimal_type = precision <= 9 ? TYPE_DECIMAL32 + : precision <= 18 ? TYPE_DECIMAL64 + : precision <= 38 ? TYPE_DECIMAL128I + : TYPE_DECIMAL256; Review Comment: Confirmed. Lance keeps the 256-bit carrier of `decimal256(18, 2)`, while this mapping and the FE type the column DECIMAL64 by precision, so `DataTypeDecimalSerDe<TYPE_DECIMAL64>::read_column_from_arrow` dereferenced the null result of `dynamic_cast<const arrow::DecimalArray*>`; the same held for DECIMAL32/128I targets and nested leaves. Fixed in 05437fd656b: `LanceArrowArrayNormalizer` narrows a Decimal256 field with precision <= 38 to `decimal128(p, s)` through a safe Arrow cast, so a value outside the declared precision fails with an error naming the field, and list/struct/map parents are rebuilt around the narrowed leaf. Unit tests cover top-level, sliced nested and overflow cases and a `decimal256(18, 2)` Lance round trip. ########## be/src/format_v2/table/lance_reader.cpp: ########## @@ -0,0 +1,1321 @@ +// 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 <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_TIMER_WITH_LEVEL(_scanner_profile, LANCE_READER_PROFILE, + file_scan_profile::TABLE_READER, 1); + _dataset_open_time = ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceDatasetOpenTime", + LANCE_READER_PROFILE, 1); + _arrow_to_doris_block_time = ADD_CHILD_TIMER_WITH_LEVEL( + _scanner_profile, "LanceArrowToDorisBlockTime", LANCE_READER_PROFILE, 1); + _data_cache_bytes_read_from_cache = + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceDataCacheBytesReadFromCache", + TUnit::BYTES, LANCE_READER_PROFILE, 1); + _data_cache_bytes_read_from_remote = + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceDataCacheBytesReadFromRemote", + TUnit::BYTES, LANCE_READER_PROFILE, 1); + 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)); + _scanner_profile->add_info_string("LanceTopKPlusOffset", std::to_string(top_k + offset)); + } + if (_scan_params->__isset.lance_scan_params && + lance_scan_params.__isset.lance_substrait_filter) { + _scanner_profile->add_info_string("LancePushdownFormat", "SUBSTRAIT"); + _scanner_profile->add_info_string( + "LanceSubstraitFilterBytes", + std::to_string(lance_scan_params.lance_substrait_filter.size())); + } + + 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_read_time); + scan_status = lance_scanner_next(_scanner, &raw_batch); + } + if (scan_status == 1) { + _eof = true; + _close_scanner(); + break; + } + if (scan_status != 0 || raw_batch == nullptr) { + return lance_error("read next Lance batch"); + } + + std::unique_ptr<LanceBatch, LanceBatchDeleter> batch(raw_batch); + size_t rows = 0; + { + SCOPED_TIMER(_arrow_to_doris_block_time); + RETURN_IF_ERROR(_fill_block_from_lance_batch(batch.get(), block, &rows)); + } + _record_scan_rows(rows); + raw_rows += rows; + } + + if (raw_rows == 0) { + DORIS_CHECK(_eof); + *eos = true; + return Status::OK(); + } + + if (block->rows() > 0) { + // Preserve a non-empty final block. The next get_block() observes `_eof` and reports + // split EOF, matching the generic reader contract. + return Status::OK(); + } + if (_eof) { + *eos = true; + return Status::OK(); + } + } +} + +Status LanceTableReader::read_by_row_ids(const TFileRangeDesc& range, + const std::vector<uint64_t>& row_ids, Block* block) { + DORIS_CHECK(block != nullptr); + DORIS_CHECK(block->columns() == _projected_columns.size()); + if (row_ids.empty()) { + return Status::OK(); + } + if (_row_id_take_read_time == nullptr) { + _row_id_take_read_time = ADD_CHILD_TIMER_WITH_LEVEL( + _scanner_profile, "LanceRowIdTakeReadTime", LANCE_READER_PROFILE, 1); + } + if (_row_id_fetch_total_time == nullptr) { + _row_id_fetch_total_time = ADD_CHILD_TIMER_WITH_LEVEL( + _scanner_profile, "LanceRowIdFetchTotalTime", LANCE_READER_PROFILE, 1); + } + auto* fetch_calls = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceRowIdFetchCalls", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + auto* fetch_rows = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceRowIdFetchRows", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + SCOPED_TIMER(_row_id_fetch_total_time); + + // Phase-two row fetch does not execute FTS, so a reader created only for take_rows must not + // collect query-specific global statistics. + RETURN_IF_ERROR(_ensure_dataset_open(range, false)); + std::vector<const char*> columns; + columns.reserve(_projected_columns.size() + 1); + for (const auto& column : _projected_columns) { + columns.emplace_back(column.name.c_str()); + } + columns.emplace_back(nullptr); + + ArrowArrayStream stream {}; + int32_t take_rows_status = 0; + { + SCOPED_TIMER(_row_id_take_read_time); + COUNTER_UPDATE(fetch_calls, 1); + take_rows_status = lance_dataset_take_rows(_dataset, row_ids.data(), row_ids.size(), + columns.data(), &stream); + } + if (take_rows_status != 0) { + if (stream.release != nullptr) { + stream.release(&stream); + } + return lance_error("take Lance rows by row id"); + } + auto imported_reader = arrow::ImportRecordBatchReader(&stream); + if (!imported_reader.ok()) { + if (stream.release != nullptr) { + stream.release(&stream); + } + return Status::InternalError("import Lance take-rows stream failed: {}", + imported_reader.status().message()); + } + + size_t fetched_rows = 0; + auto batch_reader = std::move(imported_reader).ValueUnsafe(); + while (true) { + std::shared_ptr<arrow::RecordBatch> record_batch; + arrow::Status read_status; + { + // Lance may materialize take_rows lazily while its Arrow stream is consumed. + SCOPED_TIMER(_row_id_take_read_time); + read_status = batch_reader->ReadNext(&record_batch); + } + if (!read_status.ok()) { + return Status::InternalError("read Lance take-rows batch failed: {}", + read_status.message()); + } + if (record_batch == nullptr) { + break; + } + size_t rows = 0; + { + SCOPED_TIMER(_arrow_to_doris_block_time); + RETURN_IF_ERROR(_record_batch_converter.convert_record_batch_to_block( + record_batch, block, _global_rowid_context, &rows)); + } + fetched_rows += rows; + COUNTER_UPDATE(fetch_rows, rows); + } + if (fetched_rows != row_ids.size()) { + return Status::InternalError("Lance row-id fetch returned {} rows for {} requested row ids", + fetched_rows, row_ids.size()); + } + return Status::OK(); +} + +Status LanceTableReader::abort_split() { + _close_scanner(); + _eof = true; + return TableReader::abort_split(); +} + +Status LanceTableReader::close() { + _close_scanner(); + _close_dataset(); + _opened_dataset_key.reset(); + _eof = true; + return TableReader::close(); +} + +Status LanceTableReader::_resolve_search_kind() { + DORIS_CHECK(_scan_params != nullptr); + _search_kind = SearchKind::NORMAL; + if (!_scan_params->__isset.lance_scan_params) { + return Status::OK(); + } + const auto& lance_scan_params = _scan_params->lance_scan_params; + if (!lance_scan_params.__isset.external_search_request) { + return Status::OK(); + } + const auto& request = lance_scan_params.external_search_request; + if (!request.__isset.search_query) { + return Status::InvalidArgument("external search request requires search_query"); + } + const bool has_vector = request.search_query.__isset.vector_search; + const bool has_full_text = request.search_query.__isset.full_text_search; + if (has_vector == has_full_text) { + return Status::InvalidArgument("external search query must set exactly one search kind"); + } + _search_kind = has_vector ? SearchKind::VECTOR : SearchKind::FULL_TEXT; + return Status::OK(); +} + +Status LanceTableReader::_validate_external_search_request() const { + // FE validates requests produced by the search TVFs, but this reader consumes a deserialized + // Thrift boundary. Recheck structural invariants and values used for allocation, pointer + // arithmetic, C-string calls, and narrowing conversions before accessing them below. + DORIS_CHECK(_scan_params != nullptr); + DORIS_CHECK(_scan_params->__isset.lance_scan_params); + const auto& lance_scan_params = _scan_params->lance_scan_params; + DORIS_CHECK(lance_scan_params.__isset.external_search_request); + if (lance_scan_params.__isset.lance_substrait_filter) { + return Status::InvalidArgument( + "Lance external search cannot combine its pre-search filter with " + "lance_substrait_filter"); + } + + const auto& request = lance_scan_params.external_search_request; + if (request.schema_version != 1) { + return Status::NotSupported("unsupported external search schema version: {}", + request.schema_version); + } + DORIS_CHECK(request.__isset.search_query); + DORIS_CHECK(_search_kind != SearchKind::NORMAL); + constexpr auto UINT32_MAX_VALUE = static_cast<int64_t>(std::numeric_limits<uint32_t>::max()); + if (_search_kind == SearchKind::VECTOR) { + const auto& vector = request.search_query.vector_search; + if (!vector.__isset.column || vector.column.empty() || + vector.column.find('\0') != std::string::npos) { + return Status::InvalidArgument("Lance vector search requires a non-empty column"); + } + if (!vector.__isset.query_vector) { + return Status::InvalidArgument("Lance vector search requires a query vector"); + } + const auto& query_vector = vector.query_vector; + if (!query_vector.__isset.element_type || !query_vector.__isset.dimension || + !query_vector.__isset.values) { + return Status::InvalidArgument( + "Lance query vector requires element_type, dimension, and values"); + } + if (query_vector.dimension <= 0) { + return Status::InvalidArgument("Lance query vector dimension must be positive: {}", + query_vector.dimension); + } + const auto element_width = lance_vector_element_width(query_vector.element_type); + if (element_width == 0) { + return Status::NotSupported("unsupported Lance query vector element type: {}", + static_cast<int>(query_vector.element_type)); + } + const bool multi_vector = query_vector.__isset.num_vectors; + // The optional count distinguishes a query matrix, including a one-row matrix. + if (multi_vector && query_vector.num_vectors <= 0) { + return Status::InvalidArgument( + "Lance multi-vector queries require positive num_vectors"); + } + if (multi_vector && (query_vector.element_type == TVectorElementType::UINT8 || + query_vector.element_type == TVectorElementType::INT8 || + (vector.__isset.metric && vector.metric == TVectorMetric::HAMMING))) { + return Status::NotSupported( + "Lance multi-vector search requires floating-point vectors and l2, cosine, or " + "dot"); + } + const auto dimension = static_cast<size_t>(query_vector.dimension); + const auto count = multi_vector ? static_cast<size_t>(query_vector.num_vectors) : 1; + if (dimension > std::numeric_limits<size_t>::max() / count / element_width || + query_vector.values.size() != dimension * count * element_width) { + return Status::InvalidArgument( + "Lance query vector byte size {} does not match {} vectors of dimension {} and " + "element width {}", + query_vector.values.size(), count, dimension, element_width); + } + if (!vector.__isset.top_k || vector.top_k <= 0) { + return Status::InvalidArgument("Lance vector search top_k must be positive"); + } + if (!vector.__isset.offset || vector.offset < 0) { + return Status::InvalidArgument("Lance vector search offset must be non-negative"); + } + if (vector.offset > UINT32_MAX_VALUE || vector.top_k > UINT32_MAX_VALUE - vector.offset) { + return Status::InvalidArgument( + "Lance vector search top_k + offset exceeds uint32 range"); + } + // Match FE/C API limits before constructing the per-subvector ANN plan branches. + constexpr int MAX_QUERY_VECTORS = 128; + constexpr int64_t MAX_QUERY_VECTOR_CANDIDATES = 100000; + const auto refine_factor = + request.__isset.vector_search_options && + request.vector_search_options.__isset.refine_factor + ? request.vector_search_options.refine_factor + : 1; + if (multi_vector && + (query_vector.num_vectors > MAX_QUERY_VECTORS || refine_factor <= 0 || + vector.top_k + vector.offset > MAX_QUERY_VECTOR_CANDIDATES / refine_factor || + query_vector.num_vectors > + MAX_QUERY_VECTOR_CANDIDATES / (vector.top_k + vector.offset))) { + return Status::InvalidArgument( + "multi-vector query exceeds 128 subvectors or 100000 subvector-candidates"); + } + } else { + DORIS_CHECK(_search_kind == SearchKind::FULL_TEXT); + const auto& full_text = request.search_query.full_text_search; + if (!full_text.__isset.column || full_text.column.empty() || + full_text.column.find('\0') != std::string::npos) { + return Status::InvalidArgument("Lance full-text search requires a non-empty column"); + } + if (!full_text.__isset.query || full_text.query.empty() || + full_text.query.find('\0') != std::string::npos) { + return Status::InvalidArgument("Lance full-text search requires a non-empty query"); + } + if (!full_text.__isset.top_k || full_text.top_k <= 0) { + return Status::InvalidArgument("Lance full-text search top_k must be positive"); + } + if (!full_text.__isset.offset || full_text.offset < 0) { + return Status::InvalidArgument("Lance full-text search offset must be non-negative"); + } + if (full_text.offset > UINT32_MAX_VALUE || + full_text.top_k > UINT32_MAX_VALUE - full_text.offset) { + return Status::InvalidArgument( + "Lance full-text search top_k + offset exceeds uint32 range"); + } + if (!full_text.__isset.coverage_mode || + (full_text.coverage_mode != TFtsCoverageMode::STRICT && + full_text.coverage_mode != TFtsCoverageMode::INDEX_ONLY)) { + return Status::InvalidArgument( + "Lance full-text search requires STRICT or INDEX_ONLY coverage_mode"); + } + if (full_text.__isset.global_statistics && full_text.global_statistics.empty()) { + return Status::InvalidArgument( + "Lance full-text search global_statistics must not be empty when set"); + } + if (!full_text.__isset.query_type || (full_text.query_type != TFtsQueryType::MATCH && + full_text.query_type != TFtsQueryType::PHRASE)) { + return Status::InvalidArgument( + "Lance full-text search requires MATCH or PHRASE query_type"); + } + if (full_text.query_type == TFtsQueryType::MATCH) { + if (!full_text.__isset.match_operator || + (full_text.match_operator != TFtsMatchOperator::OR && + full_text.match_operator != TFtsMatchOperator::AND)) { + return Status::InvalidArgument( + "Lance MATCH query requires OR or AND match_operator"); + } + if (!full_text.__isset.max_fuzzy_distance || full_text.max_fuzzy_distance < 0) { + return Status::InvalidArgument( + "Lance MATCH query max_fuzzy_distance must be non-negative"); + } + if (full_text.max_fuzzy_distance != 0) { + return Status::NotSupported( + "Lance prepared FTS does not yet support max_fuzzy_distance={}", + full_text.max_fuzzy_distance); + } + if (full_text.__isset.phrase_slop) { + return Status::InvalidArgument("Lance MATCH query cannot set phrase_slop"); + } + } else { + if (!full_text.__isset.phrase_slop || full_text.phrase_slop < 0) { + return Status::InvalidArgument( + "Lance PHRASE query phrase_slop must be non-negative"); + } + if (full_text.__isset.match_operator || full_text.__isset.max_fuzzy_distance) { + return Status::InvalidArgument( + "Lance PHRASE query cannot set MATCH-only parameters"); + } + } + if (request.__isset.vector_search_options) { + return Status::InvalidArgument( + "Lance full-text search cannot set vector_search_options"); + } + } + + if (request.__isset.search_filter) { + const auto& filter = request.search_filter; + if (!filter.__isset.format || !filter.__isset.payload || filter.payload.empty()) { + return Status::InvalidArgument( + "external search filter requires format and non-empty payload"); + } + if (filter.format != TSearchFilterFormat::SQL) { + return Status::NotSupported("unsupported external search filter format: {}", + static_cast<int>(filter.format)); + } + if (filter.payload.find('\0') != std::string::npos) { + return Status::InvalidArgument("Lance SQL search filter contains an embedded NUL byte"); + } + } + + if (_search_kind == SearchKind::VECTOR && request.__isset.vector_search_options) { + const auto& options = request.vector_search_options; + if (options.__isset.query_parallelism && options.query_parallelism < -1) { + return Status::InvalidArgument("Lance query_parallelism must be -1, 0, or positive"); + } + if (options.__isset.nprobes && options.nprobes <= 0) { + return Status::InvalidArgument("Lance nprobes must be positive"); + } + if (options.__isset.refine_factor && options.refine_factor <= 0) { + return Status::InvalidArgument("Lance refine_factor must be positive"); + } + if (options.__isset.ef && options.ef <= 0) { + return Status::InvalidArgument("Lance ef must be positive"); + } + } + return Status::OK(); +} + +Status LanceTableReader::_ensure_dataset_open(const TFileRangeDesc& range, + bool prepare_fts_context) { + DatasetKey key; + RETURN_IF_ERROR(_dataset_key(range, &key)); + if (_dataset == nullptr) { + RETURN_IF_ERROR(_open_dataset(key)); + _opened_dataset_key = key; + } else if (!_opened_dataset_key.has_value() || *_opened_dataset_key != key) { + return Status::InvalidArgument( + "Lance reader cannot mix dataset snapshots or storage options"); + } + if (_search_kind == SearchKind::FULL_TEXT && prepare_fts_context && + _fts_query_context == nullptr) { + RETURN_IF_ERROR(_prepare_fts_query_context()); + } + return Status::OK(); +} + +Status LanceTableReader::_open_dataset(const DatasetKey& key) { + std::vector<const char*> storage_option_ptrs; + storage_option_ptrs.reserve(key.storage_options.size() + 1); + for (const auto& option : key.storage_options) { + storage_option_ptrs.emplace_back(option.c_str()); + } + storage_option_ptrs.emplace_back(nullptr); + + std::unique_ptr<LanceDataset, LanceDatasetDeleter> dataset; + { + SCOPED_TIMER(_dataset_open_time); + LanceDataset* raw_dataset = nullptr; + RETURN_IF_ERROR(LanceSessionManager::instance().open_dataset( + key.uri.c_str(), key.storage_options.empty() ? nullptr : storage_option_ptrs.data(), + static_cast<uint64_t>(key.version), &raw_dataset)); + dataset.reset(raw_dataset); + } + _dataset = dataset.release(); + return Status::OK(); +} + +Status LanceTableReader::_prepare_fts_query_context() { + DORIS_CHECK(_dataset != nullptr); + DORIS_CHECK(_fts_query_context == nullptr); + DORIS_CHECK(_scan_params != nullptr); + const auto& full_text = + _scan_params->lance_scan_params.external_search_request.search_query.full_text_search; + if (full_text.__isset.global_statistics) { + return Status::NotSupported( + "Lance FE-provided FTS global statistics require a lance-c consumer API"); + } + const auto coverage_mode = full_text.coverage_mode == TFtsCoverageMode::STRICT + ? LANCE_FTS_COVERAGE_STRICT + : LANCE_FTS_COVERAGE_INDEX_ONLY; + // Keep statistics preparation at the reader/scanner lifetime today. A future FE-provided + // opaque statistics payload should enter through this boundary and create the same context, + // leaving segment-scoped scanner execution unchanged. + if (full_text.query_type == TFtsQueryType::MATCH) { + const auto match_operator = full_text.match_operator == TFtsMatchOperator::AND + ? LANCE_FTS_MATCH_OPERATOR_AND + : LANCE_FTS_MATCH_OPERATOR_OR; + _fts_query_context = lance_dataset_prepare_fts_match_query( + _dataset, full_text.column.c_str(), full_text.query.c_str(), match_operator, + static_cast<uint32_t>(full_text.max_fuzzy_distance), coverage_mode); + } else { + DORIS_CHECK(full_text.query_type == TFtsQueryType::PHRASE); + _fts_query_context = lance_dataset_prepare_fts_phrase_query( + _dataset, full_text.column.c_str(), full_text.query.c_str(), full_text.phrase_slop, + coverage_mode); + } + if (_fts_query_context == nullptr) { + return lance_error("prepare Lance FTS query context"); + } + return Status::OK(); +} + +void LanceTableReader::_init_scanner_profile() { + if (_scanner_configure_time != nullptr) { + return; + } + + _scanner_configure_time = ADD_CHILD_TIMER_WITH_LEVEL( + _scanner_profile, "LanceScannerConfigureTime", LANCE_READER_PROFILE, 1); + _runtime_filter_sql_time = ADD_CHILD_TIMER_WITH_LEVEL( + _scanner_profile, "LanceRuntimeFilterSqlTime", LANCE_READER_PROFILE, 1); + _scanner_read_time = ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceScannerReadTime", + LANCE_READER_PROFILE, 1); + _execution_iops = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceExecutionIOOps", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + _execution_requests = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceExecutionIORequests", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + _execution_bytes_read = ADD_CHILD_COUNTER_WITH_LEVEL( + _scanner_profile, "LanceExecutionIOBytesRead", TUnit::BYTES, LANCE_READER_PROFILE, 1); + _index_partition_cache_miss_loads = + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceIndexPartitionCacheMissLoads", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + _index_comparisons = ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceIndexComparisons", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + + // Prefilter counters isolate row-id materialization. The generic scan counts below come + // from Lance's FilteredRead execution node and are scan inputs, not ANN result counts. + _lance_count_metrics = { + {"prefilter_loads", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LancePrefilterLoads", TUnit::UNIT, + LANCE_READER_PROFILE, 1)}, + {"prefilter_input_rows", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LancePrefilterInputRows", TUnit::UNIT, + LANCE_READER_PROFILE, 1)}, + {"prefilter_input_batches", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LancePrefilterInputBatches", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"prefilter_row_ids", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LancePrefilterRowIds", TUnit::UNIT, + LANCE_READER_PROFILE, 1)}, + {"fragments_scanned", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceFragmentsScanned", TUnit::UNIT, + LANCE_READER_PROFILE, 1)}, + {"ranges_scanned", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceRowOffsetRangesScanned", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"rows_scanned", ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceRowsScanned", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"partitions_ranked", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceIVFPartitionsRanked", TUnit::UNIT, + LANCE_READER_PROFILE, 1)}, + {"partitions_searched", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceIVFPartitionsSearched", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"deltas_searched", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceVectorIndexSegmentsSearched", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"scalar_segments_requested", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceScalarIndexSegmentsRequested", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"scalar_segments_searched", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceScalarIndexSegmentsSearched", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"scalar_segment_fallbacks", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceScalarIndexSegmentFallbacks", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + {"scalar_segment_candidate_rows", + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LanceScalarIndexCandidateRows", + TUnit::UNIT, LANCE_READER_PROFILE, 1)}, + }; + _lance_time_metrics = { + // Partition stages accumulate across concurrent work and overlap their parent timers. + // DistanceTopK includes fused candidate filtering, scoring, and heap updates. + {"index_open_time", ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexOpenTime", + LANCE_READER_PROFILE, 1)}, + {"index_partition_load_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexPartitionLoadTime", + LANCE_READER_PROFILE, 1)}, + {"index_partition_prepare_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexPartitionPrepareTime", + LANCE_READER_PROFILE, 1)}, + {"index_prefilter_wait_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexPrefilterWaitTime", + LANCE_READER_PROFILE, 1)}, + {"index_cpu_queue_wait_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexCpuQueueWaitTime", + LANCE_READER_PROFILE, 1)}, + {"index_search_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexSearchTime", + LANCE_READER_PROFILE, 1)}, + {"index_query_prepare_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexQueryPrepareTime", + LANCE_READER_PROFILE, 1)}, + {"index_distance_topk_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexDistanceTopKTime", + LANCE_READER_PROFILE, 1)}, + {"index_result_materialize_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIndexResultMaterializeTime", + LANCE_READER_PROFILE, 1)}, + {"ANNIVFPartitionExec_elapsed_compute", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceANNPartitionExecTime", + LANCE_READER_PROFILE, 1)}, + {"ANNSubIndexExec_elapsed_compute", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceANNSubIndexExecTime", + LANCE_READER_PROFILE, 1)}, + {"ANNIvfBatchExec_elapsed_compute", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceANNBatchExecTime", + LANCE_READER_PROFILE, 1)}, + {"SortExec_elapsed_compute", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceSortComputeTime", + LANCE_READER_PROFILE, 1)}, + {"SortPreservingMergeExec_elapsed_compute", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceSortMergeComputeTime", + LANCE_READER_PROFILE, 1)}, + {"TakeExec_elapsed_compute", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceTakeExecTime", LANCE_READER_PROFILE, + 1)}, + {"KNNVectorDistanceExec_elapsed_compute", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceVectorDistanceComputeTime", + LANCE_READER_PROFILE, 1)}, + // These are wall times in the ANN row-id loader. LoadTime includes input polling + // and set construction; it must not be added to its component timers. + {"prefilter_load_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LancePrefilterLoadTime", + LANCE_READER_PROFILE, 1)}, + {"prefilter_input_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LancePrefilterInputTime", + LANCE_READER_PROFILE, 1)}, + {"prefilter_build_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LancePrefilterBuildTime", + LANCE_READER_PROFILE, 1)}, + + // This is wait time reported by the same Lance scan execution node described above, + // rather than Doris scanner scheduling wait time. + {"task_wait_time", ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceTaskWaitTime", + LANCE_READER_PROFILE, 1)}, + {"find_partitions_elapsed", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceIVFPartitionRankingTime", + LANCE_READER_PROFILE, 1)}, + {"scalar_segment_prepare_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceScalarIndexSegmentPrepareTime", + LANCE_READER_PROFILE, 1)}, + {"scalar_segment_search_time", + ADD_CHILD_TIMER_WITH_LEVEL(_scanner_profile, "LanceScalarIndexSegmentSearchTime", + LANCE_READER_PROFILE, 1)}, + }; + if (_search_kind != SearchKind::NORMAL) { + _planned_index_segment_count = + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LancePlannedIndexSegmentCount", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + _planned_indexed_fragment_count = + ADD_CHILD_COUNTER_WITH_LEVEL(_scanner_profile, "LancePlannedIndexedFragmentCount", + TUnit::UNIT, LANCE_READER_PROFILE, 1); + _planned_flat_search_fragment_count = ADD_CHILD_COUNTER_WITH_LEVEL( + _scanner_profile, "LancePlannedFlatSearchFragmentCount", TUnit::UNIT, + LANCE_READER_PROFILE, 1); + } +} + +Status LanceTableReader::_open_scanner(const TFileRangeDesc& range) { + _init_scanner_profile(); + SCOPED_TIMER(_scanner_configure_time); + std::vector<const char*> columns; + columns.reserve(_projected_columns.size() + 1); + for (size_t idx = 0; idx < _projected_columns.size(); ++idx) { + if (_record_batch_converter.is_global_rowid_output(idx)) { + continue; + } + const auto& column = _projected_columns[idx]; + columns.emplace_back(column.name.c_str()); + } + if (_search_kind != SearchKind::NORMAL && columns.empty()) { + // Keep an explicit empty user projection from becoming `nullptr`, which means all dataset + // columns to lance-c. Search execution already returns its generated result column. + columns.emplace_back(_search_kind == SearchKind::VECTOR ? LANCE_DISTANCE_COLUMN.data() + : LANCE_SCORE_COLUMN.data()); + } + columns.emplace_back(nullptr); + + const auto& lance_scan_params = _scan_params->lance_scan_params; + std::string sql_filter; + std::shared_ptr<const LanceRuntimeFilterSql> runtime_filter_sql; + if (_search_kind == SearchKind::NORMAL) { + if (has_lance_runtime_filters(_conjuncts)) { + if (_dataset_schema == nullptr) { + RETURN_IF_ERROR(import_lance_dataset_schema(_dataset, &_dataset_schema)); + } + SCOPED_TIMER(_runtime_filter_sql_time); + runtime_filter_sql = get_or_create_lance_runtime_filter_sql( + _conjuncts, *_dataset_schema, _runtime_filter_cache); + } + } else { + const auto& request = lance_scan_params.external_search_request; + if (request.__isset.search_filter && + request.search_filter.format == TSearchFilterFormat::SQL) { + sql_filter = request.search_filter.payload; + } + } + LanceScanner* scanner = lance_scanner_new(_dataset, columns.data(), + sql_filter.empty() ? nullptr : sql_filter.c_str()); + if (scanner == nullptr) { + return lance_error("create Lance scanner"); + } + std::unique_ptr<LanceScanner, LanceScannerDeleter> scanner_guard(scanner); + const auto collect_scan_statistics = [](void* callback_ctx, + const LanceScanStatistics* statistics) { + LanceTableReader::_collect_scan_statistics(callback_ctx, statistics); + }; + if (lance_scanner_set_statistics_callback(scanner, collect_scan_statistics, this) != 0) { + return lance_error("set Lance scanner statistics callback"); + } + + if (_record_batch_converter.requires_global_rowid() && + lance_scanner_with_row_id(scanner, true) != 0) { + return lance_error("enable Lance row id output"); + } + + if (lance_scan_params.__isset.lance_substrait_filter && + lance_scanner_set_substrait_filter( + scanner, + reinterpret_cast<const uint8_t*>(lance_scan_params.lance_substrait_filter.data()), + lance_scan_params.lance_substrait_filter.size()) != 0) { + return lance_error("set Lance Substrait filter"); + } + if (runtime_filter_sql != nullptr) { + if (!runtime_filter_sql->expression.empty()) { + if (lance_scanner_additional_sql_filter(scanner, + runtime_filter_sql->expression.c_str()) != 0) { + return lance_error("set Lance additional SQL filter"); + } + } + record_lance_runtime_filter_pushdown(_scanner_profile, *runtime_filter_sql); + } + + const auto batch_size = _batch_size > 0 ? _batch_size : _runtime_state->batch_size(); + if (lance_scanner_set_batch_size(scanner, static_cast<int64_t>(batch_size)) != 0) { + return lance_error("set Lance scanner batch size"); + } + RETURN_IF_ERROR(_configure_scan_options(scanner)); + + const auto& lance_params = range.table_format_params.lance_params; + switch (_search_kind) { + case SearchKind::NORMAL: + RETURN_IF_ERROR(_configure_normal_scan(scanner, lance_params)); + break; + case SearchKind::VECTOR: + RETURN_IF_ERROR(_configure_vector_search(scanner, lance_params)); + break; + case SearchKind::FULL_TEXT: + RETURN_IF_ERROR(_configure_full_text_search(scanner, lance_params)); + break; + } + _scanner = scanner_guard.release(); + _scanner_batch_size = batch_size; + return Status::OK(); +} + +Status LanceTableReader::_configure_scan_options(LanceScanner* scanner) const { + DORIS_CHECK(scanner != nullptr); + // Doris runs multiple scanners concurrently. Limit each scanner's read-ahead; + // the I/O budget does not cap its total memory usage. + const auto io_buffer_size = static_cast<uint64_t>(config::lance_io_buffer_size_bytes); + const auto batch_readahead = static_cast<size_t>(config::lance_batch_readahead); + const auto fragment_readahead = static_cast<size_t>(config::lance_fragment_readahead); + constexpr bool scan_in_order = false; + + if (lance_scanner_set_io_buffer_size(scanner, io_buffer_size) != 0) { + return lance_error("set Lance scanner I/O buffer size"); + } + if (lance_scanner_set_batch_readahead(scanner, batch_readahead) != 0) { + return lance_error("set Lance scanner batch readahead"); + } + if (lance_scanner_set_fragment_readahead(scanner, fragment_readahead) != 0) { + return lance_error("set Lance scanner fragment readahead"); + } + // Storage order is not required; query ordering is enforced by Sort/TopN operators. + if (lance_scanner_set_scan_in_order(scanner, scan_in_order) != 0) { + return lance_error("set Lance scanner scan order"); + } + + return Status::OK(); +} + +Status LanceTableReader::_configure_normal_scan(LanceScanner* scanner, + const TLanceFileDesc& lance_params) const { + DORIS_CHECK(scanner != nullptr); + std::vector<uint64_t> fragment_ids; + RETURN_IF_ERROR(parse_fragment_ids(lance_params, &fragment_ids)); + if (!fragment_ids.empty() && + lance_scanner_set_fragment_ids(scanner, fragment_ids.data(), fragment_ids.size()) != 0) { + return lance_error("set Lance scanner fragment ids"); + } + std::vector<uint8_t> segment_uuids; + size_t segment_count = 0; + RETURN_IF_ERROR(parse_index_segment_uuids(lance_params, &segment_uuids, &segment_count)); + if (segment_count > 1) { + return Status::InvalidArgument("normal Lance scan accepts only one scalar index segment"); + } + if (segment_count == 1) { + if (fragment_ids.empty() || !lance_params.__isset.version || lance_params.version <= 0) { + return Status::InvalidArgument( + "Lance scalar index segment requires a fixed version and nonempty fragment " + "ids"); + } + if (lance_params.__isset.use_scalar_index && !lance_params.use_scalar_index) { + return Status::InvalidArgument( + "Lance scalar index segment cannot be combined with use_scalar_index=false"); + } + if (lance_scanner_set_scalar_index_segment(scanner, segment_uuids.data()) != 0) { + return lance_error("set Lance scanner scalar index segment"); + } + } else if (lance_params.__isset.use_scalar_index && + lance_scanner_set_use_scalar_index(scanner, lance_params.use_scalar_index) != 0) { + return lance_error("set Lance scanner scalar index usage"); + } + // FE sets this only when every predicate has been pushed into Lance. + if (lance_params.__isset.limit && lance_params.limit > 0 && + lance_scanner_set_limit(scanner, lance_params.limit) != 0) { + return lance_error("set Lance scanner limit"); + } + return Status::OK(); +} + +Status LanceTableReader::_configure_vector_search(LanceScanner* scanner, + const TLanceFileDesc& lance_params) const { + DORIS_CHECK(scanner != nullptr); + DORIS_CHECK(_scan_params != nullptr); + DORIS_CHECK(_scan_params->__isset.lance_scan_params); + std::vector<uint64_t> fragment_ids; + RETURN_IF_ERROR(parse_fragment_ids(lance_params, &fragment_ids)); + if (!fragment_ids.empty() && + lance_scanner_set_fragment_ids(scanner, fragment_ids.data(), fragment_ids.size()) != 0) { + return lance_error("set Lance vector scanner fragment ids"); + } + std::vector<uint8_t> segment_uuids; + size_t segment_count = 0; + RETURN_IF_ERROR(parse_index_segment_uuids(lance_params, &segment_uuids, &segment_count)); + if (segment_count > 0 && + lance_scanner_set_index_segments(scanner, segment_uuids.data(), segment_count) != 0) { + return lance_error("set Lance vector scanner index segments"); + } + // Fragment-scoped nearest queries require prefiltering before installing the query. The same + // path applies the TVF search filter, when present. + if (lance_scanner_set_prefilter(scanner, true) != 0) { Review Comment: Agreed, fixed in 60aebf8529f: this PR now ports the BE side of #68813. `TExternalSearchRequest.5: optional bool prefilter` has the 4.x id and comment, and both vector and full-text search pass `__isset.prefilter && prefilter` to `lance_scanner_set_prefilter()`, so unset or false means postfilter as on branch-4.1/4.2. `SearchFilterDefaultsToPostfilter` (from #68813) and a new full-text case cover both settings. ########## gensrc/thrift/PlanNodes.thrift: ########## @@ -475,13 +475,142 @@ struct TRemoteDorisFileDesc { 6: optional string password } +// The external search enums below spell out their values because those values are on the wire: +// branch-4.1/4.2 declared the same enums without explicit values, so each member must keep its +// implicit position (FLOAT16 = 0, FLOAT32 = 1, ...) on every branch. + +// Element type and byte layout of a vector search query. The encoded values use little-endian +// byte order and contain exactly `dimension` elements of `element_type`. +enum TVectorElementType { + FLOAT16 = 0, + FLOAT32 = 1, + FLOAT64 = 2, + UINT8 = 3, + INT8 = 4 +} + +enum TVectorMetric { + DEFAULT = 0, + L2 = 1, + COSINE = 2, + DOT_PRODUCT = 3, + HAMMING = 4 +} + +struct TSearchVector { + 1: optional TVectorElementType element_type + 2: optional i32 dimension + 3: optional binary values + // Present only for one multi-vector query; values contains a row-major matrix. + 4: optional i32 num_vectors +} + +// Logical result parameters for one vector query. `top_k` is the number of rows returned after +// skipping `offset`; an adapter may fetch more physical candidates to implement that contract. +struct TVectorSearchParams { + 1: optional string column + 2: optional TSearchVector query_vector + 3: optional i64 top_k + 4: optional i64 offset + 5: optional TVectorMetric metric Review Comment: Agreed, fixed in 7edd9ccffb0: `TVectorSearchParams` 6/7 (`distance_lower_bound` / `distance_upper_bound`) have the 4.x ids and types, the request is validated as in #68815 (finite FLOAT bounds, lower < upper at FLOAT precision, no multi-vector), and `lance_scanner_set_distance_range()` is installed after `nearest()`, using master's pinned lance-c. #68815's fallback for quantized indexes relies on the BE searching a split without index segments flat, which came from #68707, so a66c2a649e7 ports that rule (`use_index` follows the split's segments; a split with segments under `use_index=false` is rejected). #68815's four tests and fixtures come along unchanged. ########## be/src/format_v2/lance/lance_runtime_filter_helper.cpp: ########## @@ -0,0 +1,537 @@ +// 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_runtime_filter_helper.h" + +#include <arrow/type.h> +#include <cctz/time_zone.h> + +#include <algorithm> +#include <cctype> +#include <charconv> +#include <limits> +#include <optional> +#include <set> +#include <string_view> +#include <system_error> + +#include "common/logging.h" +#include "core/data_type/data_type_nullable.h" +#include "core/field.h" +#include "exprs/hybrid_set.h" +#include "exprs/runtime_filter_expr.h" +#include "exprs/vdirect_in_predicate.h" +#include "exprs/vexpr_context.h" +#include "exprs/vliteral.h" +#include "exprs/vslot_ref.h" +#include "format/format_common.h" +#include "runtime/runtime_profile.h" +#include "util/utf8_check.h" + +namespace doris::format::lance { +namespace { + +constexpr std::string_view LANCE_RUNTIME_FILTER_CACHE_KEY_PREFIX = "lance-runtime-filter-sql:"; + +std::string format_filter_ids(const std::vector<int>& filter_ids) { + std::string result; + for (const auto filter_id : filter_ids) { + if (!result.empty()) { + result.append(","); + } + result.append(std::to_string(filter_id)); + } + return result; +} + +std::string quote_sql_identifier(std::string_view identifier) { + // Lance SQL uses backticks for delimited identifiers. Escape an embedded backtick by doubling + // it, matching the SQL parser's quoted-identifier syntax. + std::string quoted("`"); + quoted.reserve(identifier.size() + 2); + for (const char ch : identifier) { + if (ch == '`') { + quoted.append("``"); + } else { + quoted.push_back(ch); + } + } + quoted.push_back('`'); + return quoted; +} + +const RuntimeFilterExpr* get_runtime_filter(const VExprContextSPtr& conjunct) { + if (conjunct == nullptr || conjunct->root() == nullptr) { + return nullptr; + } + return dynamic_cast<const RuntimeFilterExpr*>(conjunct->root().get()); +} + +std::string lowercase_ascii(std::string value) { + std::ranges::transform(value, value.begin(), [](const unsigned char ch) { + return static_cast<char>(std::tolower(ch)); + }); + return value; +} + +std::string quote_string_value(std::string_view value) { + std::string quoted("'"); + quoted.reserve(value.size() + 2); + for (const char ch : value) { + if (ch == '\'') { + quoted.append("''"); + } else { + quoted.push_back(ch); + } + } + quoted.push_back('\''); + return quoted; +} + +std::shared_ptr<arrow::DataType> find_unique_physical_type(const arrow::Schema& schema, + std::string_view column_name) { + const auto fields = schema.GetAllFieldsByName(column_name); + return fields.size() == 1 ? fields.front()->type() : nullptr; +} + +std::optional<int64_t> parse_int64(std::string_view value) { + int64_t parsed = 0; + const auto* begin = value.data(); + const auto* end = begin + value.size(); + const auto result = std::from_chars(begin, end, parsed); + if (result.ec != std::errc() || result.ptr != end) { + return std::nullopt; + } + return parsed; +} + +bool integer_literal_fits_decimal_precision(std::string_view value, int32_t precision) { + if (!value.empty() && (value.front() == '-' || value.front() == '+')) { + value.remove_prefix(1); + } + const auto first_nonzero = value.find_first_not_of('0'); + if (first_nonzero == std::string_view::npos) { + return true; + } + return value.size() - first_nonzero <= static_cast<size_t>(precision); +} + +// This checks Lance planner coercion, not whether stored Lance values fit their own Arrow type. +// Unsigned Arrow columns map to wider signed Doris types (for example UInt8 -> SMALLINT), so an RF +// produced from another input may contain a valid Doris value that cannot be coerced back to the +// narrower physical type. +bool integer_literal_fits_physical_type(std::string_view value, + const arrow::DataType& physical_type) { + const auto parsed = parse_int64(value); + if (!parsed.has_value()) { + return false; + } + const auto literal = *parsed; + switch (physical_type.id()) { + case arrow::Type::INT8: + return literal >= std::numeric_limits<int8_t>::min() && + literal <= std::numeric_limits<int8_t>::max(); + case arrow::Type::UINT8: + return literal >= 0 && literal <= std::numeric_limits<uint8_t>::max(); + case arrow::Type::INT16: + return literal >= std::numeric_limits<int16_t>::min() && + literal <= std::numeric_limits<int16_t>::max(); + case arrow::Type::UINT16: + return literal >= 0 && literal <= std::numeric_limits<uint16_t>::max(); + case arrow::Type::INT32: + return literal >= std::numeric_limits<int32_t>::min() && + literal <= std::numeric_limits<int32_t>::max(); + case arrow::Type::UINT32: + return literal >= 0 && literal <= std::numeric_limits<uint32_t>::max(); + case arrow::Type::INT64: + return true; + case arrow::Type::UINT64: + // The pinned Lance SQL parser first parses a bare integer as i64. Values above INT64_MAX + // become Float64, which cannot be safely coerced to UInt64. + return literal >= 0; + default: + return false; + } +} + +bool physical_type_supports_literal(PrimitiveType logical_type, std::string_view value, + const arrow::DataType& physical_type) { + switch (logical_type) { + case TYPE_BOOLEAN: + return physical_type.id() == arrow::Type::BOOL; + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + return integer_literal_fits_physical_type(value, physical_type); + case TYPE_FLOAT: + // Arrow HALF_FLOAT and FLOAT both materialize as Doris FLOAT, but the pinned Lance planner + // cannot coerce an SQL Int64/Float64 literal to Float16. + return physical_type.id() == arrow::Type::FLOAT; + case TYPE_DOUBLE: + return physical_type.id() == arrow::Type::DOUBLE; + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: { + if (physical_type.id() != arrow::Type::DECIMAL128 && + physical_type.id() != arrow::Type::DECIMAL256) { + return false; + } + const auto& decimal = static_cast<const arrow::DecimalType&>(physical_type); + // A scale-bearing bare SQL token is parsed as Float64, which the pinned planner cannot + // coerce to Decimal. Scale-zero values are safe only while they parse exactly as i64. + return decimal.scale() == 0 && parse_int64(value).has_value() && + integer_literal_fits_decimal_precision(value, decimal.precision()); + } + case TYPE_CHAR: + case TYPE_VARCHAR: + case TYPE_STRING: + return (physical_type.id() == arrow::Type::STRING || + physical_type.id() == arrow::Type::LARGE_STRING) && + value.find('\0') == std::string_view::npos && + validate_utf8(value.data(), value.size()); + case TYPE_DATE: + case TYPE_DATEV2: + // Date64 is only specified to contain a millisecond count that should be day-aligned. + // Doris materialization discards any sub-day remainder, so only Date32 is unconditionally + // comparison-equivalent. + return physical_type.id() == arrow::Type::DATE32; + case TYPE_DATETIME: + case TYPE_DATETIMEV2: { + if (physical_type.id() != arrow::Type::TIMESTAMP) { + return false; + } + const auto& timestamp = static_cast<const arrow::TimestampType&>(physical_type); + // Doris DATETIMEV2 stores at most microseconds, so a nanosecond predicate is not equivalent + // to evaluating the residual after Arrow-to-Doris materialization. + return timestamp.timezone().empty() && timestamp.unit() != arrow::TimeUnit::NANO; + } + default: + return false; + } +} + +std::optional<std::string> to_lance_sql_literal(const VLiteral& literal, + const arrow::DataType& physical_type) { + const auto type = remove_nullable(literal.get_data_type()); + auto options = DataTypeSerDe::get_default_format_options(); + auto timezone = cctz::utc_time_zone(); + options.timezone = &timezone; + const auto value = literal.value(options); + if (!physical_type_supports_literal(type->get_primitive_type(), value, physical_type)) { + // The SQL literal would either fail Lance planning or compare different physical values + // from the Doris residual. Keep the RF residual-only in either case. + return std::nullopt; + } + switch (type->get_primitive_type()) { + case TYPE_BOOLEAN: { + const auto normalized = lowercase_ascii(value); + if (normalized == "0" || normalized == "false") { + return "FALSE"; + } + if (normalized == "1" || normalized == "true") { + return "TRUE"; + } + // Do not pass an implementation-specific boolean spelling through to Lance SQL. + return std::nullopt; + } + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + return value; + case TYPE_FLOAT: + case TYPE_DOUBLE: { + const auto normalized = lowercase_ascii(value); + if (normalized.find("nan") != std::string::npos || + normalized.find("inf") != std::string::npos) { + // Lance SQL has no portable scalar spelling with Doris-equivalent NaN/Inf semantics. + return std::nullopt; + } + return value; + } + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: + return value; + case TYPE_CHAR: + case TYPE_VARCHAR: + case TYPE_STRING: + return quote_string_value(value); + case TYPE_DATE: + case TYPE_DATEV2: + return "DATE " + quote_string_value(value); + case TYPE_DATETIME: + case TYPE_DATETIMEV2: + return "TIMESTAMP " + quote_string_value(value); + default: + return std::nullopt; + } +} + +template <PrimitiveType PT> +std::optional<std::string> in_value_to_lance_sql_literal(const void* raw_value, + const DataTypePtr& data_type, + const arrow::DataType& physical_type) { + if (raw_value == nullptr || data_type == nullptr) { + // A null set entry needs SQL three-valued semantics and cannot be serialized as a scalar. + return std::nullopt; + } + Field field; + if constexpr (is_string_type(PT)) { + const auto* value = static_cast<const StringRef*>(raw_value); + using CppType = typename PrimitiveTypeTraits<PT>::CppType; + field = Field::create_field<PT>(CppType(value->data, value->size)); + } else { + using CppType = typename PrimitiveTypeTraits<PT>::CppType; + field = Field::create_field<PT>(*static_cast<const CppType*>(raw_value)); + } + return to_lance_sql_literal(VLiteral(data_type, field), physical_type); +} + +std::optional<std::string> in_value_to_lance_sql_literal(PrimitiveType primitive_type, + const void* raw_value, + const DataTypePtr& data_type, + const arrow::DataType& physical_type) { +#define DISPATCH_IN_VALUE(TYPE) \ + case TYPE: \ + return in_value_to_lance_sql_literal<TYPE>(raw_value, data_type, physical_type) + switch (primitive_type) { + DISPATCH_IN_VALUE(TYPE_BOOLEAN); + DISPATCH_IN_VALUE(TYPE_TINYINT); + DISPATCH_IN_VALUE(TYPE_SMALLINT); + DISPATCH_IN_VALUE(TYPE_INT); + DISPATCH_IN_VALUE(TYPE_BIGINT); + DISPATCH_IN_VALUE(TYPE_LARGEINT); + DISPATCH_IN_VALUE(TYPE_FLOAT); + DISPATCH_IN_VALUE(TYPE_DOUBLE); + DISPATCH_IN_VALUE(TYPE_DATE); + DISPATCH_IN_VALUE(TYPE_DATETIME); + DISPATCH_IN_VALUE(TYPE_DATEV2); + DISPATCH_IN_VALUE(TYPE_DATETIMEV2); + DISPATCH_IN_VALUE(TYPE_CHAR); + DISPATCH_IN_VALUE(TYPE_VARCHAR); + DISPATCH_IN_VALUE(TYPE_STRING); + DISPATCH_IN_VALUE(TYPE_DECIMALV2); + DISPATCH_IN_VALUE(TYPE_DECIMAL32); + DISPATCH_IN_VALUE(TYPE_DECIMAL64); + DISPATCH_IN_VALUE(TYPE_DECIMAL128I); + DISPATCH_IN_VALUE(TYPE_DECIMAL256); + default: + return std::nullopt; + } +#undef DISPATCH_IN_VALUE +} + +std::optional<std::string> build_in_filter_sql(const VDirectInPredicate& predicate, + const arrow::Schema& physical_schema) { + if (predicate.get_num_children() != 1) { + return std::nullopt; + } + const auto slot = std::dynamic_pointer_cast<VSlotRef>(predicate.get_child(0)); + const auto values = predicate.get_set_func(); + if (slot == nullptr || slot->data_type() == nullptr || values == nullptr) { + // Only a direct slot-backed IN set has a stable Lance column/literal representation. + return std::nullopt; + } + if (values->contain_null()) { + // Dropping NULL would change the Doris IN-filter semantics. + return std::nullopt; + } + + const auto data_type = remove_nullable(slot->data_type()); + const auto physical_type = find_unique_physical_type(physical_schema, slot->column_name()); + if (physical_type == nullptr) { + // A missing or ambiguous top-level field cannot be bound safely by column name. + return std::nullopt; + } + std::string expression("(" + quote_sql_identifier(slot->column_name()) + " IN ("); + auto* iterator = values->begin(); + bool first_value = true; + while (iterator != nullptr && iterator->has_next()) { + auto value = in_value_to_lance_sql_literal( + data_type->get_primitive_type(), iterator->get_value(), data_type, *physical_type); + if (!value.has_value()) { + // Partially pushing an IN set would introduce false negatives, so reject it as a unit. + return std::nullopt; + } + if (!first_value) { + expression.append(", "); + } + expression.append(*value); + first_value = false; + iterator->next(); + } + if (first_value) { + // An empty IN set has no direct Lance SQL representation. + return std::nullopt; + } + expression.append("))"); + return expression; +} + +std::optional<std::string> build_range_filter_sql(const VExpr& predicate, + const arrow::Schema& physical_schema) { + if ((predicate.op() != TExprOpcode::GE && predicate.op() != TExprOpcode::LE) || + predicate.get_num_children() != 2) { + return std::nullopt; + } + const auto slot = std::dynamic_pointer_cast<VSlotRef>(predicate.get_child(0)); + const auto literal = std::dynamic_pointer_cast<VLiteral>(predicate.get_child(1)); + if (slot == nullptr || literal == nullptr) { + // Computed operands are intentionally left to Doris instead of reconstructing expression + // SQL. + return std::nullopt; + } + const auto physical_type = find_unique_physical_type(physical_schema, slot->column_name()); + if (physical_type == nullptr) { + // A missing or ambiguous top-level field cannot be bound safely by column name. + return std::nullopt; + } + const auto sql_literal = to_lance_sql_literal(*literal, *physical_type); + if (!sql_literal.has_value()) { + // Preserve the Doris residual when its bound has no proven-equivalent physical literal. + return std::nullopt; + } + const auto* sql_operator = predicate.op() == TExprOpcode::GE ? ">=" : "<="; + return "(" + quote_sql_identifier(slot->column_name()) + " " + sql_operator + " " + Review Comment: Confirmed. Doris keeps every NaN for `x >= c` (`GreaterThanEqualsFloat` treats NaN as greatest regardless of sign), while the pinned Arrow/DataFusion kernels use totalOrder, so a MIN-only filter could drop a negative-NaN row that Doris returns. `<=` and IN are safe: Lance only returns extra negative NaNs for `<=`, which the residual removes, and IN sets with NaN members are not pushed. Fixed in bf5dcbd9fb1: FLOAT/DOUBLE `>=` runtime filters stay residual-only, the policy `PaimonRustPredicateConverter` already applies, with a helper test and a reader test on 0xffc00000 / 0xfff8000000000000 rows. branch-4.1 and branch-4.2 carry the same helper. -- 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]
