github-actions[bot] commented on code in PR #69581:
URL: https://github.com/apache/doris/pull/69581#discussion_r4369909220
##########
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:
[P1] Retain the existing vector distance-bound fields
Branch-4.1/4.2 already define [fields 6/7 as
`distance_lower_bound`/`distance_upper_bound`](https://github.com/apache/doris/blob/a7ae6f4611e40dcf5e5444ae1d87bb4f79e4e490/gensrc/thrift/PlanNodes.thrift#L484-L492).
Their FE writes these from `vector_search` properties without synthesizing a
Doris residual. This definition discards both fields during Thrift
deserialization, and the new reader never applies a native distance range while
still accepting `schema_version=1`. During an FE-4.x/BE-master upgrade, an
exact top-1 query whose nearest squared L2 distance is 1 and
`distance_upper_bound=0.5` therefore returns that row instead of an empty
result. Preserve the IDs and enforce the range, or explicitly reject
unsupported bounded requests.
##########
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:
[P1] Handle Decimal256 carriers before selecting a narrower SerDe
A valid Lance `Decimal256(18,2)` is mapped to Doris DECIMAL64 here, but
normalization leaves its `Decimal256Array` unchanged.
`DataTypeDecimalSerDe<TYPE_DECIMAL64>::read_column_from_arrow` casts it to
`arrow::DecimalArray` (Decimal128Array) and then calls `Value` on the null
result, so reading a nonempty column crashes the BE. The [pinned Lance datatype
encoding](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance-core/src/datatypes.rs#L335-L336)
preserves the physical 256-bit width even when precision is <=38; current
fixtures only use Decimal256 at precision 39/76. Normalize and checked-narrow
that carrier, or support both physical widths in the SerDe, including nested
leaves.
##########
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:
[P2] Preserve negative NaNs when pushing floating range filters
For a nested-loop join `L.x > B.y` with finite `B.y=1`, FE can generate the
MIN runtime filter `L.x >= 1`. A Lance FLOAT containing negative quiet-NaN bits
`0xffc00000` satisfies both predicates in Doris, whose `Compare` treats every
NaN as greatest. The pinned [Arrow floating
comparison](https://github.com/apache/arrow-rs/blob/58.3.0/arrow-array/src/arithmetic.rs#L391-L401)
uses total ordering, placing negative NaNs below finite values, and
[DataFusion delegates GE directly to that
kernel](https://github.com/apache/datafusion/blob/54.0.0/datafusion/physical-expr-common/src/datum.rs#L63-L82);
this generated SQL therefore drops a matching probe row. Rejecting nonfinite
bound literals does not protect against NaN input values. Keep floating range
filters residual-only or widen them to preserve every Doris survivor, with
differential negative-NaN coverage.
##########
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] Check DATE32 conversion failures at the Lance boundary
The new adapter advertises Arrow DATE32 as supported DATEV2, including valid
external dates above Doris's maximum year 9999. For example, day count
`2932897` is `10000-01-01`. The shared DATE32 Arrow SerDe ignores
`get_date_from_daynr(false)` and appends its zero-initialized value, while the
nullable wrapper copies the source's non-null validity bit. This call therefore
reports success with a non-null `0000-00-00` value. DATE64 and native Parquet
conversion already reject or explicitly null failed conversions. Add equivalent
checked range handling to this path, including nested date leaves, and cover
the first date above `9999-12-31`.
##########
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:
[P1] Preserve the 4.x search-filter timing flag
The branch-4.1/4.2 FE [serializes `TExternalSearchRequest.prefilter=false`
by
default](https://github.com/apache/doris/blob/a7ae6f4611e40dcf5e5444ae1d87bb4f79e4e490/fe/fe-core/src/main/java/org/apache/doris/tablefunction/LanceExternalSearchTableValuedFunction.java#L202-L216),
but this head omits wire field 5 and unconditionally enables prefiltering
here. With `top_k=1`, `use_index=false`, and `filter='keep = true'`, a nearest
row with `keep=false` followed by a farther `keep=true` row should yield an
empty postfilter result; this BE returns the farther row. The FE keeps that TVF
filter only in the request, so a Doris residual cannot restore the requested
behavior during an FE-4.x/BE-master upgrade. Retain field 5 with its
established default and honor it when configuring the scanner.
--
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]
