github-actions[bot] commented on code in PR #65730:
URL: https://github.com/apache/doris/pull/65730#discussion_r3689417093
##########
fe/fe-core/src/main/java/org/apache/doris/common/util/PrintableMap.java:
##########
@@ -61,6 +61,8 @@ public class PrintableMap<K, V> {
SENSITIVE_KEY.add("bos_secret_accesskey");
SENSITIVE_KEY.add("jdbc.password");
SENSITIVE_KEY.add("elasticsearch.password");
+ SENSITIVE_KEY.add("lance.rest.bearer-token");
+ SENSITIVE_KEY.add("lance.rest.api-key");
Review Comment:
[P1] Mask credential-bearing custom REST headers
The catalog accepts arbitrary `lance.rest.header.*` properties and forwards
their values as HTTP headers, but these exact-key additions leave every custom
header printable. For example, a service authenticated with
`lance.rest.header.X-Auth-Token=secret` will expose `secret` through catalog
properties and `SHOW CREATE CATALOG`. Please mask this prefix (or restrict
custom headers to an explicit non-secret allowlist) and add a custom-token
regression.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LanceScanNode.java:
##########
@@ -0,0 +1,272 @@
+// 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.
+
+package org.apache.doris.datasource.lance.source;
+
+import org.apache.doris.analysis.TupleDescriptor;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalUtil;
+import org.apache.doris.datasource.FileQueryScanNode;
+import org.apache.doris.datasource.TableFormatType;
+import org.apache.doris.datasource.lance.LanceExternalCatalog;
+import org.apache.doris.datasource.lance.LanceExternalTable;
+import org.apache.doris.datasource.lance.LanceTableMetadata;
+import org.apache.doris.planner.PlanNodeId;
+import org.apache.doris.planner.ScanContext;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.spi.Split;
+import org.apache.doris.statistics.StatisticalType;
+import org.apache.doris.thrift.TExplainLevel;
+import org.apache.doris.thrift.TExternalSearchRequest;
+import org.apache.doris.thrift.TFileFormatType;
+import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TLanceFileDesc;
+import org.apache.doris.thrift.TTableFormatFileDesc;
+import org.apache.doris.thrift.TVectorMetric;
+import org.apache.doris.thrift.TVectorSearchParams;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Scan node for both ordinary Lance table scans and Lance external-search
scans.
+ *
+ * <p>These modes share dataset metadata, storage properties, and BE
scan-range serialization.
+ * Keeping them in one node prevents those common parts from drifting apart.
The search request is
+ * also an explicit mode marker: ordinary scans are split by fragment, while
the first version of
+ * vector search deliberately sends one whole-snapshot split to one scanner so
Lance can compute a
+ * global TopK result.
+ */
+public class LanceScanNode extends FileQueryScanNode {
+ private LanceExternalTable lanceTable;
+ private LanceTableMetadata plannedMetadata;
+ private TExternalSearchRequest externalSearchRequest;
+ private byte[] lanceSubstraitFilter = new byte[0];
+ private String lancePushdownPredicate = "";
+ private long plannedVersion = -1;
+ private int plannedFragments;
+
+ public LanceScanNode(PlanNodeId id, TupleDescriptor desc, boolean
needCheckColumnPriv,
+ SessionVariable sessionVariable, ScanContext scanContext) {
+ super(id, desc, "LANCE_SCAN_NODE", StatisticalType.LANCE_SCAN_NODE,
+ scanContext, needCheckColumnPriv, sessionVariable);
+ }
+
+ /**
+ * Creates the search mode of this node.
+ *
+ * <p>The tuple descriptor belongs to a FunctionGenTable and contains
generated columns such as
+ * {@code _distance}. Therefore the real Lance table and the metadata
snapshot selected while
+ * analyzing the TVF must be passed separately.
+ */
+ public LanceScanNode(PlanNodeId id, TupleDescriptor desc,
LanceExternalTable lanceTable,
+ LanceTableMetadata plannedMetadata, TExternalSearchRequest
externalSearchRequest,
+ SessionVariable sessionVariable) {
+ super(id, desc, "LANCE_SCAN_NODE", StatisticalType.LANCE_SCAN_NODE,
+
ScanContext.builder().clusterName(sessionVariable.resolveCloudClusterName()).build(),
+ false, sessionVariable);
+ this.lanceTable = lanceTable;
+ this.plannedMetadata = plannedMetadata;
+ this.externalSearchRequest = externalSearchRequest.deepCopy();
+ }
+
+ @Override
+ protected void doInitialize() throws UserException {
+ super.doInitialize();
+
+ if (isExternalSearch()) {
+ // Search output comes from the FunctionGenTable because it adds
generated columns such
+ // as _distance. The real Lance table is still retained for
storage and metadata access.
+ ExternalUtil.initSchemaInfo(params, -1L,
desc.getTable().getColumns());
+ params.setExternalSearchRequest(externalSearchRequest.deepCopy());
+ } else {
+ lanceTable = (LanceExternalTable) desc.getTable();
+ ExternalUtil.initSchemaInfo(params, -1L, lanceTable.getColumns());
+ }
+ }
+
+ @Override
+ protected void convertPredicate() {
+ if (isExternalSearch()) {
+ // The TVF "filter" property is already serialized in
externalSearchRequest and is
+ // evaluated by Lance before vector search. Outer WHERE conjuncts
have different
+ // semantics: Doris must keep them here and evaluate them after
Lance returns TopK.
+ } else {
+ plannedMetadata = lanceTable.loadMetadata();
Review Comment:
[P1] Do not silently ignore the requested Lance snapshot
The planner preserves `FOR VERSION/TIME AS OF` in `queryTableSnapshot`, but
this path always loads the latest Lance metadata and never reads that snapshot.
The resulting splits are pinned consistently—just to the wrong version—so a
historical query silently returns current rows. Please resolve and load the
requested Lance version/time here, or reject snapshot-qualified Lance relations
until time travel is supported.
##########
be/src/core/data_type_serde/data_type_array_serde.cpp:
##########
@@ -326,32 +326,59 @@ Status
DataTypeArraySerDe::read_column_from_arrow(IColumn& column, const arrow::
const cctz::time_zone& ctz)
const {
auto& column_array = static_cast<ColumnArray&>(column);
auto& offsets_data = column_array.get_offsets();
- const auto* concrete_array = dynamic_cast<const
arrow::ListArray*>(arrow_array);
- auto arrow_offsets_array = concrete_array->offsets();
- auto* arrow_offsets =
dynamic_cast<arrow::Int32Array*>(arrow_offsets_array.get());
- if (config::enable_arrow_input_validation) {
- check_arrow_list_offsets(*concrete_array, start, end);
- }
- auto prev_size = offsets_data.back();
- const auto* base_offsets_ptr = reinterpret_cast<const
uint8_t*>(arrow_offsets->raw_values());
- const size_t offset_element_size = sizeof(int32_t);
- int32_t arrow_nested_start_offset = 0;
- int32_t arrow_nested_end_offset = 0;
- const uint8_t* start_offset_ptr = base_offsets_ptr + start *
offset_element_size;
- const uint8_t* end_offset_ptr = base_offsets_ptr + end *
offset_element_size;
- memcpy(&arrow_nested_start_offset, start_offset_ptr, offset_element_size);
- memcpy(&arrow_nested_end_offset, end_offset_ptr, offset_element_size);
-
- for (auto i = start + 1; i < end + 1; ++i) {
- int32_t current_offset = 0;
- const uint8_t* current_offset_ptr = base_offsets_ptr + i *
offset_element_size;
- memcpy(¤t_offset, current_offset_ptr, offset_element_size);
- // convert to doris offset, start from offsets.back()
- offsets_data.emplace_back(prev_size + current_offset -
arrow_nested_start_offset);
- }
- return nested_serde->read_column_from_arrow(
- column_array.get_data(), concrete_array->values().get(),
arrow_nested_start_offset,
- arrow_nested_end_offset, ctz);
+
+ const auto read_list = [&](const auto* concrete_array) -> Status {
+ const int64_t arrow_nested_start_offset =
concrete_array->value_offset(start);
+ const int64_t arrow_nested_end_offset =
concrete_array->value_offset(end);
+ const auto prev_size = offsets_data.back();
+ for (int64_t i = start + 1; i <= end; ++i) {
+ // Convert Arrow offsets to Doris offsets, starting at
offsets.back().
+ offsets_data.emplace_back(prev_size +
concrete_array->value_offset(i) -
+ arrow_nested_start_offset);
+ }
+ return nested_serde->read_column_from_arrow(
+ column_array.get_data(), concrete_array->values().get(),
arrow_nested_start_offset,
+ arrow_nested_end_offset, ctz);
+ };
+
+ switch (arrow_array->type_id()) {
+ case arrow::Type::LIST: {
+ const auto* concrete_array = dynamic_cast<const
arrow::ListArray*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow ListArray, got {}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
+ check_arrow_list_offsets(*concrete_array, start, end);
+ }
+ return read_list(concrete_array);
+ }
+ case arrow::Type::LARGE_LIST: {
+ const auto* concrete_array = dynamic_cast<const
arrow::LargeListArray*>(arrow_array);
+ if (concrete_array == nullptr) {
+ return Status::InvalidArgument("Expected Arrow LargeListArray, got
{}",
+ arrow_array->type()->name());
+ }
+ if (config::enable_arrow_input_validation) {
Review Comment:
[P1] Validate LargeList offsets before reading them
Unlike the existing List branch, this only validates the requested row range
and then `read_list` immediately calls `value_offset()`. A short, negative,
non-monotonic, or child-overrunning 64-bit offsets buffer from an Arrow FFI
producer can therefore be read out of bounds or passed as an invalid nested
range even with input validation enabled. Please add a LargeList offset
validator equivalent to `check_arrow_list_offsets` and malformed-buffer tests.
##########
be/src/core/data_type_serde/data_type_time_serde.cpp:
##########
@@ -281,6 +284,86 @@ Status
DataTypeTimeV2SerDe::from_string_strict_mode(StringRef& str, IColumn& col
return Status::OK();
}
+Status DataTypeTimeV2SerDe::read_column_from_arrow(IColumn& column, const
arrow::Array* arrow_array,
Review Comment:
[P1] Validate temporal Arrow buffers before access
This new TIME32/64 reader calls `IsNull`/`Value` without validating the
requested range, validity bitmap, or fixed-width values buffer. The new
TIMESTAMPTZ path has the same gap, and Date64 only calls
`check_arrow_no_offset`, which validates none of those buffers. A malformed
external Arrow array can therefore read out of bounds even with input
validation enabled; the nullable wrapper still leaves the values buffer
unchecked. Please apply `check_arrow_array_range` plus the appropriate 4/8-byte
fixed-width check to all three temporal paths and add malformed-buffer tests.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalTable.java:
##########
@@ -0,0 +1,81 @@
+// 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.
+
+package org.apache.doris.datasource.lance;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.SchemaCacheValue;
+import org.apache.doris.statistics.AnalysisInfo;
+import org.apache.doris.statistics.BaseAnalysisTask;
+import org.apache.doris.statistics.ExternalAnalysisTask;
+import org.apache.doris.thrift.THiveTable;
+import org.apache.doris.thrift.TTableDescriptor;
+import org.apache.doris.thrift.TTableType;
+
+import org.apache.arrow.vector.types.pojo.Field;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+
+public class LanceExternalTable extends ExternalTable {
+ public LanceExternalTable(long id, String name, String remoteName,
LanceExternalCatalog catalog,
+ LanceExternalDatabase db) {
+ super(id, name, remoteName, catalog, db,
TableType.LANCE_EXTERNAL_TABLE);
+ }
+
+ @Override
+ public Optional<SchemaCacheValue> initSchema() {
+ LanceTableMetadata metadata = loadMetadata();
+ List<Column> columns = new
ArrayList<>(metadata.getSchema().getFields().size());
+ int position = 0;
+ for (Field field : metadata.getSchema().getFields()) {
+ String comment = field.getMetadata() == null ? null :
field.getMetadata().get("comment");
+ columns.add(new Column(field.getName(),
LanceTypeConverter.toDorisType(field), false,
+ null, field.isNullable(), comment, true, position++));
+ }
+ return Optional.of(new SchemaCacheValue(columns));
Review Comment:
[P1] Keep the bound schema and scan on one Lance snapshot
This loads a fixed Lance metadata snapshot but caches only its converted
columns. `LanceScanNode` later builds BE slots from those cached columns and
independently loads the current latest metadata for predicates and splits.
After schema evolution, an INT slot can therefore read a newer Float32 Arrow
array; the numeric SerDe copies same-width bytes without checking the physical
Arrow type, so `1.0f` becomes `1065353216` instead of failing. Please retain
the version/fragments/credentials with this schema and plan from that same
snapshot, or verify and rebind before creating splits.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java:
##########
@@ -169,6 +176,29 @@ public List<Split> getSplits(int numBackends) throws
UserException {
return splits;
}
+ private List<Split> getLanceSplits() throws UserException {
+ if (!sessionVariable.enableFileScannerV2) {
+ throw new UserException("Lance TVF requires
enable_file_scanner_v2=true");
+ }
+ if (tableValuedFunction.getTFileType() == TFileType.FILE_LOCAL) {
+ // A local dataset is visible to its selected BE, not to FE. Keep
exactly one
+ // whole-dataset split; BE resolves version zero to latest when it
opens the dataset.
+ return Collections.singletonList(
+
LanceSplit.wholeDatasetAtLatest(tableValuedFunction.getFilePath()));
Review Comment:
[P1] Pin local TVF execution to the discovered version
Local schema discovery already opens this dataset at version zero, but it
returns only columns and discards the resolved version; this creates a second
version-zero open for execution. If the dataset evolves between analysis and
scan, Doris can bind version-N types and decode version-N+1 values (including
silent same-width bit reinterpretation). Please return the resolved version
with the schema and use that positive version here, as the S3 TVF path does.
##########
be/src/core/data_type_serde/data_type_number_serde.cpp:
##########
@@ -840,6 +856,47 @@ Status
DataTypeNumberSerDe<T>::read_column_from_arrow(IColumn& column,
return Status::OK();
}
+ if constexpr (T == TYPE_FLOAT) {
+ if (arrow_array->type_id() == arrow::Type::HALF_FLOAT) {
Review Comment:
[P1] Validate new numeric buffers before the early returns
The range check above does not validate the values or validity buffers. This
HalfFloat branch and the new unsigned-widening branches below call Arrow's raw
`IsNull`/`Value` accessors and return before the common
`check_arrow_fixed_width_buffer` path, so a missing or short external buffer is
still read out of bounds with validation enabled. Please validate the matching
physical width and bitmap before each early return, and add malformed-buffer
coverage.
##########
be/src/core/data_type_serde/data_type_varbinary_serde.cpp:
##########
@@ -184,6 +186,63 @@ Status DataTypeVarbinarySerDe::write_column_to_arrow(const
IColumn& column, cons
return Status::OK();
}
+Status DataTypeVarbinarySerDe::read_column_from_arrow(IColumn& column,
Review Comment:
[P1] Validate Arrow buffers before VARBINARY access
This shared reader dereferences validity, offset, and value buffers without
any of the checks enabled by `enable_arrow_input_validation`. In the Binary
branch, for example, a short offsets buffer makes the `offset_i + 1` memcpy
read out of bounds, while negative or non-monotonic offsets can form an invalid
pointer or huge length. The adjacent String reader already applies the required
range, bitmap, offset-buffer, per-value, and fixed-width checks. Please mirror
those checks for every VARBINARY branch and cover malformed inputs.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -0,0 +1,912 @@
+// 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/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+#include <lance/lance.h>
+
+#include <bit>
+#include <cstring>
+#include <limits>
+#include <memory>
+
+#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_struct.h"
+#include "exec/common/endian.h"
+
+namespace doris::format::lance {
+namespace {
+
+struct LanceDatasetDeleter {
+ void operator()(LanceDataset* dataset) const {
lance_dataset_close(dataset); }
+};
+
+struct LanceScannerDeleter {
+ void operator()(LanceScanner* scanner) const {
lance_scanner_close(scanner); }
+};
+
+struct LanceBatchDeleter {
+ void operator()(LanceBatch* batch) const { lance_batch_free(batch); }
+};
+
+constexpr std::string_view DISTANCE_COLUMN = "_distance";
+
+size_t vector_element_width(TVectorElementType::type type) {
+ switch (type) {
+ case TVectorElementType::FLOAT16:
+ return sizeof(uint16_t);
+ case TVectorElementType::FLOAT32:
+ return sizeof(float);
+ case TVectorElementType::FLOAT64:
+ return sizeof(double);
+ case TVectorElementType::UINT8:
+ case TVectorElementType::INT8:
+ return sizeof(uint8_t);
+ }
+ return 0;
+}
+
+std::string vector_metric_name(TVectorMetric::type metric) {
+ switch (metric) {
+ case TVectorMetric::DEFAULT:
+ return "default";
+ case TVectorMetric::L2:
+ return "l2";
+ case TVectorMetric::COSINE:
+ return "cosine";
+ case TVectorMetric::DOT_PRODUCT:
+ return "dot";
+ case TVectorMetric::HAMMING:
+ return "hamming";
+ }
+ return "unknown";
+}
+
+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;
+}
+
+Status arrow_type_to_doris_type(const std::shared_ptr<arrow::DataType>&
arrow_type,
+ 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();
+ };
+
+ switch (arrow_type->id()) {
+ 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 doris_type = timestamp->timezone().empty() ?
TYPE_DATETIMEV2 : TYPE_TIMESTAMPTZ;
+ return nullable_primitive(doris_type, 0,
arrow_time_precision(timestamp->unit()));
+ }
+ 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 PrimitiveType doris_decimal_type = precision <= 9 ?
TYPE_DECIMAL32
+ : precision <= 18 ?
TYPE_DECIMAL64
+ : precision <= 38 ?
TYPE_DECIMAL128I
+ :
TYPE_DECIMAL256;
+ return nullable_primitive(doris_decimal_type, precision,
decimal->scale());
+ }
+ case arrow::Type::LIST:
+ case arrow::Type::LARGE_LIST:
+ case arrow::Type::FIXED_SIZE_LIST: {
+ const auto list =
std::static_pointer_cast<arrow::BaseListType>(arrow_type);
+ DataTypePtr value_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(list->value_type(),
&value_type));
+ *doris_type =
make_nullable(std::make_shared<DataTypeArray>(value_type));
+ return Status::OK();
+ }
+ case arrow::Type::MAP: {
+ const auto map = std::static_pointer_cast<arrow::MapType>(arrow_type);
+ DataTypePtr key_type;
+ DataTypePtr item_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(map->key_type(), &key_type));
+ RETURN_IF_ERROR(arrow_type_to_doris_type(map->item_type(),
&item_type));
+ *doris_type = make_nullable(std::make_shared<DataTypeMap>(key_type,
item_type));
+ return Status::OK();
+ }
+ case arrow::Type::STRUCT: {
+ const auto struct_type =
std::static_pointer_cast<arrow::StructType>(arrow_type);
+ DataTypes field_types;
+ Strings field_names;
+ field_types.reserve(struct_type->num_fields());
+ field_names.reserve(struct_type->num_fields());
+ for (const auto& field : struct_type->fields()) {
+ DataTypePtr field_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(field->type(),
&field_type));
+ field_types.emplace_back(std::move(field_type));
+ field_names.emplace_back(field->name());
+ }
+ *doris_type =
make_nullable(std::make_shared<DataTypeStruct>(field_types, field_names));
+ return Status::OK();
+ }
+ default:
+ return Status::NotSupported("unsupported Lance Arrow type: {}",
arrow_type->ToString());
+ }
+}
+
+} // namespace
+
+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");
+ }
+ RETURN_IF_ERROR(_validate_range(range));
+
+ const auto& params = range.table_format_params.lance_params;
+ const auto storage_options = _storage_options(&scan_params);
+ 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");
+ }
+
+ ArrowSchema arrow_schema {};
+ if (lance_dataset_schema(dataset.get(), &arrow_schema) != 0) {
+ return _lance_error("get Lance dataset schema");
+ }
+ auto imported_schema = arrow::ImportSchema(&arrow_schema);
+ if (!imported_schema.ok()) {
+ if (arrow_schema.release != nullptr) {
+ arrow_schema.release(&arrow_schema);
+ }
+ return Status::InternalError("import Lance Arrow schema failed: {}",
+ imported_schema.status().message());
+ }
+
+ const auto schema = std::move(imported_schema).ValueUnsafe();
+ std::vector<std::string> parsed_names;
+ std::vector<DataTypePtr> parsed_types;
+ parsed_names.reserve(schema->num_fields());
+ parsed_types.reserve(schema->num_fields());
+ std::unordered_set<std::string> unique_names;
+ unique_names.reserve(schema->num_fields());
+ for (const auto& field : schema->fields()) {
+ if (!unique_names.emplace(field->name()).second) {
+ return Status::InvalidArgument("duplicate Lance schema column:
{}", field->name());
+ }
+ DataTypePtr doris_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(field->type(), &doris_type));
+ parsed_names.emplace_back(field->name());
+ parsed_types.emplace_back(std::move(doris_type));
+ }
+ *column_names = std::move(parsed_names);
+ *column_types = std::move(parsed_types);
+ return Status::OK();
+}
+
+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);
+
+ _ctz = _runtime_state->timezone_obj();
+ _vector_search = _scan_params->__isset.external_search_request;
+ _search_split_prepared = false;
+ if (_vector_search) {
+ RETURN_IF_ERROR(_validate_external_search_request());
+ const auto& vector =
_scan_params->external_search_request.query.vector;
+ _scanner_profile->add_info_string("ExternalSearchType", "VECTOR");
+ _scanner_profile->add_info_string("LanceVectorColumn", vector.column);
+ _scanner_profile->add_info_string("LanceTopK",
std::to_string(vector.top_k));
+ _scanner_profile->add_info_string("LanceOffset",
std::to_string(vector.offset));
+ _scanner_profile->add_info_string("LanceMetric", vector.__isset.metric
+ ?
vector_metric_name(vector.metric)
+ : "default");
+ _scanner_profile->add_info_string("LanceVectorDimension",
+
std::to_string(vector.query_vector.dimension));
+ }
+ if (_scan_params->__isset.lance_substrait_filter) {
+ _scanner_profile->add_info_string("LancePushdownFormat", "SUBSTRAIT");
+ _scanner_profile->add_info_string(
+ "LanceSubstraitFilterBytes",
+ std::to_string(_scan_params->lance_substrait_filter.size()));
+ }
+
+ _output_name_to_idx.clear();
+ _output_name_to_idx.reserve(_projected_columns.size());
+ for (size_t idx = 0; idx < _projected_columns.size(); ++idx) {
+ const auto& column = _projected_columns[idx];
+ if (column.type == nullptr) {
+ return Status::InvalidArgument("Lance projected column '{}' has no
type", column.name);
+ }
+ if (!_output_name_to_idx.emplace(column.name, idx).second) {
+ return Status::InvalidArgument("duplicate Lance projected column:
{}", column.name);
+ }
+ if (_vector_search && column.name == DISTANCE_COLUMN) {
+ const auto distance_type = remove_nullable(column.type);
+ if (distance_type->get_primitive_type() != TYPE_FLOAT) {
+ return Status::InvalidArgument(
+ "Lance vector search column '{}' must have Doris FLOAT
type, but was {}",
+ DISTANCE_COLUMN, column.type->get_name());
+ }
+ }
+ }
+ return Status::OK();
+}
+
+Status LanceTableReader::prepare_split(const SplitReadOptions& options) {
+ RETURN_IF_ERROR(_validate_range(options.current_range));
+ _close_scanner();
+ _eof = false;
+
+ RETURN_IF_ERROR(TableReader::prepare_split(options));
+ // Lance does not currently provide metadata aggregate pushdown. Do not
let a generic
+ // table-level count supplied by a future planner bypass fragment reads.
+ _remaining_table_level_count = -1;
+ if (current_split_pruned()) {
+ return Status::OK();
+ }
+
+ if (_vector_search) {
+ const auto& lance_params =
options.current_range.table_format_params.lance_params;
+ if (lance_params.version <= 0) {
+ return Status::InvalidArgument(
+ "Lance vector search requires a fixed positive dataset
version");
+ }
+ if (lance_params.__isset.fragment_ids) {
+ return Status::InvalidArgument("Lance vector search split must not
set fragment ids");
+ }
+ if (_search_split_prepared) {
+ return Status::InvalidArgument(
+ "Lance vector search supports exactly one whole-dataset
split");
+ }
+ }
+
+ const auto key = _dataset_key(options.current_range);
+ 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
in one scan");
+ }
+
+ RETURN_IF_ERROR(_prepare_conjuncts());
+ RETURN_IF_ERROR(_open_scanner(options.current_range));
+ _search_split_prepared = _vector_search;
+ 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();
+ }
+ 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;
+ const int32_t 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;
+ RETURN_IF_ERROR(_fill_block_from_arrow(batch.get(), block, &rows));
+ _record_scan_rows(rows);
+ raw_rows += rows;
+ }
+
+ if (raw_rows == 0) {
+ DORIS_CHECK(_eof);
+ *eos = true;
+ return Status::OK();
+ }
+
+ if (!_conjuncts.empty()) {
Review Comment:
[P1] Do not evaluate residual predicates twice
`FileScannerV2` gives this reader a clone of every scanner conjunct, but
`Scanner::get_block()` always filters the returned block with the originals
afterward. Applying all clones here bypasses the generic TableReader safety
gate: a legal residual such as `rand() < 0.5` is evaluated by two independently
seeded contexts, so roughly 25% of rows survive instead of 50%. Please leave
residual evaluation to `Scanner`, or localize only predicates proven safe while
preserving a single semantic owner.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -0,0 +1,912 @@
+// 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/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+#include <lance/lance.h>
+
+#include <bit>
+#include <cstring>
+#include <limits>
+#include <memory>
+
+#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_struct.h"
+#include "exec/common/endian.h"
+
+namespace doris::format::lance {
+namespace {
+
+struct LanceDatasetDeleter {
+ void operator()(LanceDataset* dataset) const {
lance_dataset_close(dataset); }
+};
+
+struct LanceScannerDeleter {
+ void operator()(LanceScanner* scanner) const {
lance_scanner_close(scanner); }
+};
+
+struct LanceBatchDeleter {
+ void operator()(LanceBatch* batch) const { lance_batch_free(batch); }
+};
+
+constexpr std::string_view DISTANCE_COLUMN = "_distance";
+
+size_t vector_element_width(TVectorElementType::type type) {
+ switch (type) {
+ case TVectorElementType::FLOAT16:
+ return sizeof(uint16_t);
+ case TVectorElementType::FLOAT32:
+ return sizeof(float);
+ case TVectorElementType::FLOAT64:
+ return sizeof(double);
+ case TVectorElementType::UINT8:
+ case TVectorElementType::INT8:
+ return sizeof(uint8_t);
+ }
+ return 0;
+}
+
+std::string vector_metric_name(TVectorMetric::type metric) {
+ switch (metric) {
+ case TVectorMetric::DEFAULT:
+ return "default";
+ case TVectorMetric::L2:
+ return "l2";
+ case TVectorMetric::COSINE:
+ return "cosine";
+ case TVectorMetric::DOT_PRODUCT:
+ return "dot";
+ case TVectorMetric::HAMMING:
+ return "hamming";
+ }
+ return "unknown";
+}
+
+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;
+}
+
+Status arrow_type_to_doris_type(const std::shared_ptr<arrow::DataType>&
arrow_type,
+ 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();
+ };
+
+ switch (arrow_type->id()) {
+ 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 doris_type = timestamp->timezone().empty() ?
TYPE_DATETIMEV2 : TYPE_TIMESTAMPTZ;
+ return nullable_primitive(doris_type, 0,
arrow_time_precision(timestamp->unit()));
+ }
+ 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 PrimitiveType doris_decimal_type = precision <= 9 ?
TYPE_DECIMAL32
+ : precision <= 18 ?
TYPE_DECIMAL64
+ : precision <= 38 ?
TYPE_DECIMAL128I
+ :
TYPE_DECIMAL256;
+ return nullable_primitive(doris_decimal_type, precision,
decimal->scale());
Review Comment:
[P1] Reject negative-scale decimals before schema construction
Arrow permits negative decimal scales, but this passes the signed scale
directly into `DataTypeFactory`; conversion to the factory's unsigned scale
makes `create_decimal` throw. `fetch_table_schema` runs this path in a
heavy-work-pool task with no exception guard, so a local Lance dataset such as
`decimal(10,-2)` can terminate the BE instead of returning an
unsupported-schema error. Please validate precision/scale and return
`Status::NotSupported` before the factory call (matching `LanceTypeConverter`),
and add negative-scale local schema-fetch coverage.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -0,0 +1,912 @@
+// 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/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+#include <lance/lance.h>
+
+#include <bit>
+#include <cstring>
+#include <limits>
+#include <memory>
+
+#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_struct.h"
+#include "exec/common/endian.h"
+
+namespace doris::format::lance {
+namespace {
+
+struct LanceDatasetDeleter {
+ void operator()(LanceDataset* dataset) const {
lance_dataset_close(dataset); }
+};
+
+struct LanceScannerDeleter {
+ void operator()(LanceScanner* scanner) const {
lance_scanner_close(scanner); }
+};
+
+struct LanceBatchDeleter {
+ void operator()(LanceBatch* batch) const { lance_batch_free(batch); }
+};
+
+constexpr std::string_view DISTANCE_COLUMN = "_distance";
+
+size_t vector_element_width(TVectorElementType::type type) {
+ switch (type) {
+ case TVectorElementType::FLOAT16:
+ return sizeof(uint16_t);
+ case TVectorElementType::FLOAT32:
+ return sizeof(float);
+ case TVectorElementType::FLOAT64:
+ return sizeof(double);
+ case TVectorElementType::UINT8:
+ case TVectorElementType::INT8:
+ return sizeof(uint8_t);
+ }
+ return 0;
+}
+
+std::string vector_metric_name(TVectorMetric::type metric) {
+ switch (metric) {
+ case TVectorMetric::DEFAULT:
+ return "default";
+ case TVectorMetric::L2:
+ return "l2";
+ case TVectorMetric::COSINE:
+ return "cosine";
+ case TVectorMetric::DOT_PRODUCT:
+ return "dot";
+ case TVectorMetric::HAMMING:
+ return "hamming";
+ }
+ return "unknown";
+}
+
+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;
+}
+
+Status arrow_type_to_doris_type(const std::shared_ptr<arrow::DataType>&
arrow_type,
+ 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();
+ };
+
+ switch (arrow_type->id()) {
+ 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 doris_type = timestamp->timezone().empty() ?
TYPE_DATETIMEV2 : TYPE_TIMESTAMPTZ;
+ return nullable_primitive(doris_type, 0,
arrow_time_precision(timestamp->unit()));
+ }
+ 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 PrimitiveType doris_decimal_type = precision <= 9 ?
TYPE_DECIMAL32
+ : precision <= 18 ?
TYPE_DECIMAL64
+ : precision <= 38 ?
TYPE_DECIMAL128I
+ :
TYPE_DECIMAL256;
+ return nullable_primitive(doris_decimal_type, precision,
decimal->scale());
+ }
+ case arrow::Type::LIST:
+ case arrow::Type::LARGE_LIST:
+ case arrow::Type::FIXED_SIZE_LIST: {
+ const auto list =
std::static_pointer_cast<arrow::BaseListType>(arrow_type);
+ DataTypePtr value_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(list->value_type(),
&value_type));
+ *doris_type =
make_nullable(std::make_shared<DataTypeArray>(value_type));
+ return Status::OK();
+ }
+ case arrow::Type::MAP: {
+ const auto map = std::static_pointer_cast<arrow::MapType>(arrow_type);
+ DataTypePtr key_type;
+ DataTypePtr item_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(map->key_type(), &key_type));
+ RETURN_IF_ERROR(arrow_type_to_doris_type(map->item_type(),
&item_type));
+ *doris_type = make_nullable(std::make_shared<DataTypeMap>(key_type,
item_type));
+ return Status::OK();
+ }
+ case arrow::Type::STRUCT: {
+ const auto struct_type =
std::static_pointer_cast<arrow::StructType>(arrow_type);
+ DataTypes field_types;
+ Strings field_names;
+ field_types.reserve(struct_type->num_fields());
+ field_names.reserve(struct_type->num_fields());
+ for (const auto& field : struct_type->fields()) {
+ DataTypePtr field_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(field->type(),
&field_type));
+ field_types.emplace_back(std::move(field_type));
+ field_names.emplace_back(field->name());
+ }
+ *doris_type =
make_nullable(std::make_shared<DataTypeStruct>(field_types, field_names));
+ return Status::OK();
+ }
+ default:
+ return Status::NotSupported("unsupported Lance Arrow type: {}",
arrow_type->ToString());
+ }
+}
+
+} // namespace
+
+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");
+ }
+ RETURN_IF_ERROR(_validate_range(range));
+
+ const auto& params = range.table_format_params.lance_params;
+ const auto storage_options = _storage_options(&scan_params);
+ 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");
+ }
+
+ ArrowSchema arrow_schema {};
+ if (lance_dataset_schema(dataset.get(), &arrow_schema) != 0) {
+ return _lance_error("get Lance dataset schema");
+ }
+ auto imported_schema = arrow::ImportSchema(&arrow_schema);
+ if (!imported_schema.ok()) {
+ if (arrow_schema.release != nullptr) {
+ arrow_schema.release(&arrow_schema);
+ }
+ return Status::InternalError("import Lance Arrow schema failed: {}",
+ imported_schema.status().message());
+ }
+
+ const auto schema = std::move(imported_schema).ValueUnsafe();
+ std::vector<std::string> parsed_names;
+ std::vector<DataTypePtr> parsed_types;
+ parsed_names.reserve(schema->num_fields());
+ parsed_types.reserve(schema->num_fields());
+ std::unordered_set<std::string> unique_names;
+ unique_names.reserve(schema->num_fields());
+ for (const auto& field : schema->fields()) {
+ if (!unique_names.emplace(field->name()).second) {
+ return Status::InvalidArgument("duplicate Lance schema column:
{}", field->name());
+ }
+ DataTypePtr doris_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(field->type(), &doris_type));
+ parsed_names.emplace_back(field->name());
+ parsed_types.emplace_back(std::move(doris_type));
+ }
+ *column_names = std::move(parsed_names);
+ *column_types = std::move(parsed_types);
+ return Status::OK();
+}
+
+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);
+
+ _ctz = _runtime_state->timezone_obj();
+ _vector_search = _scan_params->__isset.external_search_request;
+ _search_split_prepared = false;
+ if (_vector_search) {
+ RETURN_IF_ERROR(_validate_external_search_request());
+ const auto& vector =
_scan_params->external_search_request.query.vector;
+ _scanner_profile->add_info_string("ExternalSearchType", "VECTOR");
+ _scanner_profile->add_info_string("LanceVectorColumn", vector.column);
+ _scanner_profile->add_info_string("LanceTopK",
std::to_string(vector.top_k));
+ _scanner_profile->add_info_string("LanceOffset",
std::to_string(vector.offset));
+ _scanner_profile->add_info_string("LanceMetric", vector.__isset.metric
+ ?
vector_metric_name(vector.metric)
+ : "default");
+ _scanner_profile->add_info_string("LanceVectorDimension",
+
std::to_string(vector.query_vector.dimension));
+ }
+ if (_scan_params->__isset.lance_substrait_filter) {
+ _scanner_profile->add_info_string("LancePushdownFormat", "SUBSTRAIT");
+ _scanner_profile->add_info_string(
+ "LanceSubstraitFilterBytes",
+ std::to_string(_scan_params->lance_substrait_filter.size()));
+ }
+
+ _output_name_to_idx.clear();
+ _output_name_to_idx.reserve(_projected_columns.size());
+ for (size_t idx = 0; idx < _projected_columns.size(); ++idx) {
+ const auto& column = _projected_columns[idx];
+ if (column.type == nullptr) {
+ return Status::InvalidArgument("Lance projected column '{}' has no
type", column.name);
+ }
+ if (!_output_name_to_idx.emplace(column.name, idx).second) {
+ return Status::InvalidArgument("duplicate Lance projected column:
{}", column.name);
+ }
+ if (_vector_search && column.name == DISTANCE_COLUMN) {
+ const auto distance_type = remove_nullable(column.type);
+ if (distance_type->get_primitive_type() != TYPE_FLOAT) {
+ return Status::InvalidArgument(
+ "Lance vector search column '{}' must have Doris FLOAT
type, but was {}",
+ DISTANCE_COLUMN, column.type->get_name());
+ }
+ }
+ }
+ return Status::OK();
+}
+
+Status LanceTableReader::prepare_split(const SplitReadOptions& options) {
+ RETURN_IF_ERROR(_validate_range(options.current_range));
+ _close_scanner();
+ _eof = false;
+
+ RETURN_IF_ERROR(TableReader::prepare_split(options));
+ // Lance does not currently provide metadata aggregate pushdown. Do not
let a generic
+ // table-level count supplied by a future planner bypass fragment reads.
+ _remaining_table_level_count = -1;
+ if (current_split_pruned()) {
+ return Status::OK();
+ }
+
+ if (_vector_search) {
+ const auto& lance_params =
options.current_range.table_format_params.lance_params;
+ if (lance_params.version <= 0) {
+ return Status::InvalidArgument(
+ "Lance vector search requires a fixed positive dataset
version");
+ }
+ if (lance_params.__isset.fragment_ids) {
+ return Status::InvalidArgument("Lance vector search split must not
set fragment ids");
+ }
+ if (_search_split_prepared) {
+ return Status::InvalidArgument(
+ "Lance vector search supports exactly one whole-dataset
split");
+ }
+ }
+
+ const auto key = _dataset_key(options.current_range);
+ 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
in one scan");
+ }
+
+ RETURN_IF_ERROR(_prepare_conjuncts());
+ RETURN_IF_ERROR(_open_scanner(options.current_range));
+ _search_split_prepared = _vector_search;
+ 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();
+ }
+ 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;
+ const int32_t 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;
+ RETURN_IF_ERROR(_fill_block_from_arrow(batch.get(), block, &rows));
+ _record_scan_rows(rows);
+ raw_rows += rows;
+ }
+
+ if (raw_rows == 0) {
+ DORIS_CHECK(_eof);
+ *eos = true;
+ return Status::OK();
+ }
+
+ if (!_conjuncts.empty()) {
+ RETURN_IF_ERROR(VExprContext::filter_block(_conjuncts, block,
block->columns()));
+ }
+ 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::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::_validate_range(const TFileRangeDesc& range) const {
+ if (!range.__isset.table_format_params ||
!range.table_format_params.__isset.lance_params) {
+ return Status::InvalidArgument("Lance split requires lance_params in
table format params");
+ }
+ const auto& params = range.table_format_params.lance_params;
+ if (!params.__isset.dataset_uri || params.dataset_uri.empty()) {
+ return Status::InvalidArgument("Lance split requires a non-empty
dataset_uri");
+ }
+ if (!params.__isset.version || params.version < 0) {
+ return Status::InvalidArgument("Lance split requires a non-negative
dataset version");
+ }
+ std::unordered_set<int64_t> unique_ids;
+ for (const auto fragment_id : params.fragment_ids) {
+ if (fragment_id < 0) {
+ return Status::InvalidArgument("Lance fragment id must be
non-negative: {}",
+ fragment_id);
+ }
+ if (!unique_ids.emplace(fragment_id).second) {
+ return Status::InvalidArgument("Lance split contains duplicate
fragment id: {}",
+ fragment_id);
+ }
+ }
+ return Status::OK();
+}
+
+Status LanceTableReader::_validate_external_search_request() const {
+ DORIS_CHECK(_scan_params != nullptr);
+ DORIS_CHECK(_scan_params->__isset.external_search_request);
+ if (_scan_params->__isset.lance_substrait_filter) {
+ return Status::InvalidArgument(
+ "Lance vector search cannot combine its pre-search filter with
"
+ "lance_substrait_filter");
+ }
+ const auto& request = _scan_params->external_search_request;
+ if (request.__isset.schema_version && request.schema_version != 1) {
+ return Status::NotSupported("unsupported external search schema
version: {}",
+ request.schema_version);
+ }
+ if (!request.__isset.query) {
+ return Status::InvalidArgument("external search request requires
query");
+ }
+
+ const bool has_vector = request.query.__isset.vector;
+ const bool has_full_text = request.query.__isset.full_text;
+ if (has_vector == has_full_text) {
+ return Status::InvalidArgument("external search query must set exactly
one search kind");
+ }
+ if (has_full_text) {
+ return Status::NotSupported("Lance Format V2 reader does not yet
support full-text search");
+ }
+
+ const auto& vector = request.query.vector;
+ 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 = 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 auto dimension = static_cast<size_t>(query_vector.dimension);
+ if (dimension > std::numeric_limits<size_t>::max() / element_width ||
+ query_vector.values.size() != dimension * element_width) {
+ return Status::InvalidArgument(
+ "Lance query vector byte size {} does not match dimension {}
and element width {}",
+ query_vector.values.size(), 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");
+ }
+ constexpr auto UINT32_MAX_VALUE =
static_cast<int64_t>(std::numeric_limits<uint32_t>::max());
+ 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");
+ }
+
+ if (request.__isset.filter) {
+ const auto& filter = request.filter;
+ if (!filter.__isset.format || !filter.__isset.payload ||
filter.payload.empty()) {
+ return Status::InvalidArgument(
+ "external search filter requires format and non-empty
payload");
+ }
+ switch (filter.format) {
+ case TSearchFilterFormat::SQL:
+ if (filter.payload.find('\0') != std::string::npos) {
+ return Status::InvalidArgument(
+ "Lance SQL search filter contains an embedded NUL
byte");
+ }
+ break;
+ case TSearchFilterFormat::SUBSTRAIT:
+ break;
+ default:
+ return Status::NotSupported("unsupported external search filter
format: {}",
+ static_cast<int>(filter.format));
+ }
+ }
+
+ if (request.__isset.lance_options) {
+ const auto& options = request.lance_options;
+ 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::_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);
+
+ _dataset = lance_dataset_open(
+ key.uri.c_str(), key.storage_options.empty() ? nullptr :
storage_option_ptrs.data(),
+ static_cast<uint64_t>(key.version));
+ if (_dataset == nullptr) {
+ return _lance_error("open Lance dataset");
+ }
+ return Status::OK();
+}
+
+Status LanceTableReader::_open_scanner(const TFileRangeDesc& range) {
+ std::vector<const char*> columns;
+ columns.reserve(_projected_columns.size() + 1);
+ for (const auto& column : _projected_columns) {
+ columns.emplace_back(column.name.c_str());
+ }
+ if (_vector_search && _projected_columns.empty()) {
+ columns.emplace_back(DISTANCE_COLUMN.data());
+ }
+ columns.emplace_back(nullptr);
+
+ const char* sql_filter = nullptr;
+ if (_vector_search) {
+ const auto& request = _scan_params->external_search_request;
+ if (request.__isset.filter && request.filter.format ==
TSearchFilterFormat::SQL) {
+ sql_filter = request.filter.payload.c_str();
+ }
+ }
+ LanceScanner* scanner =
+ lance_scanner_new(_dataset, columns.size() == 1 ? nullptr :
columns.data(), sql_filter);
+ if (scanner == nullptr) {
+ return _lance_error("create Lance scanner");
+ }
+ std::unique_ptr<LanceScanner, LanceScannerDeleter> scanner_guard(scanner);
+
+ if (_scan_params->__isset.lance_substrait_filter &&
+ !_scan_params->lance_substrait_filter.empty()) {
+ const auto& filter = _scan_params->lance_substrait_filter;
+ if (lance_scanner_set_substrait_filter(
+ scanner, reinterpret_cast<const uint8_t*>(filter.data()),
filter.size()) != 0) {
+ return _lance_error("set Lance Substrait filter");
+ }
+ }
+ if (_vector_search) {
+ const auto& request = _scan_params->external_search_request;
+ if (request.__isset.filter && request.filter.format ==
TSearchFilterFormat::SUBSTRAIT) {
+ const auto& filter = request.filter.payload;
+ if (lance_scanner_set_substrait_filter(scanner,
+ reinterpret_cast<const
uint8_t*>(filter.data()),
+ filter.size()) != 0) {
+ return _lance_error("set Lance vector search Substrait
filter");
+ }
+ }
+ }
+
+ const auto batch_size = _batch_size > 0 ? _batch_size :
_runtime_state->batch_size();
Review Comment:
[P2] Keep Lance batches within the adaptive byte budget
`FORMAT_LANCE` is absent from FileScannerV2's adaptive-batch list, so this
normally configures thousands of rows immediately instead of the 32-row probe
used to learn wide/nested row size. The size is also copied into
`_scanner_batch_size`, so later `set_batch_size()` updates cannot affect the
open scanner. A Lance table with large strings, arrays, maps, or vectors can
therefore materialize an enormous Arrow batch plus its Doris copy despite
`preferred_block_size_bytes`. Please enable the probe and make subsequent
predicted sizes reach lance-c, or enforce an equivalent byte bound.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java:
##########
@@ -0,0 +1,551 @@
+// 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.
+
+package org.apache.doris.datasource.lance;
+
+import org.apache.doris.common.DdlException;
+import org.apache.doris.datasource.CatalogProperty;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.InitCatalogLog;
+import org.apache.doris.datasource.SessionContext;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.lance.namespace.LanceNamespace;
+import org.lance.namespace.model.DescribeTableRequest;
+import org.lance.namespace.model.DescribeTableResponse;
+import org.lance.namespace.model.ListNamespacesRequest;
+import org.lance.namespace.model.ListNamespacesResponse;
+import org.lance.namespace.model.ListTablesRequest;
+import org.lance.namespace.model.ListTablesResponse;
+
+import java.net.URI;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/** Read-only Lance Directory or REST Namespace catalog. */
+public class LanceExternalCatalog extends ExternalCatalog {
+ public static final String LANCE_CATALOG_TYPE = "lance.catalog.type";
+ public static final String LANCE_FILESYSTEM = "filesystem";
+ public static final String LANCE_REST = "rest";
+ public static final String WAREHOUSE = "warehouse";
+ public static final String NAMESPACE_PARENT = "lance.namespace.parent";
+ public static final String NAMESPACE_DELIMITER =
"lance.namespace.delimiter";
+ public static final String ROOT_DATABASE = "lance.namespace.root_database";
+ public static final String REST_URI = "lance.rest.uri";
+ public static final String REST_SECURITY_TYPE = "lance.rest.security.type";
+ public static final String REST_BEARER_TOKEN = "lance.rest.bearer-token";
+ public static final String REST_API_KEY = "lance.rest.api-key";
+ public static final String REST_HEADER_PREFIX = "lance.rest.header.";
+
+ private static final String DEFAULT_DELIMITER = "$";
+ private static final String DEFAULT_ROOT_DATABASE = "default";
+ private static final String DATABASE_NAMESPACE_DELIMITER = ".";
+ private static final String REST_SECURITY_NONE = "none";
+ private static final String REST_SECURITY_BEARER = "bearer";
+ private static final String REST_SECURITY_API_KEY = "api_key";
+ private static final Pattern HTTP_HEADER_NAME =
Pattern.compile("^[!#$%&'*+.^_`|~0-9A-Za-z-]+$");
+ private static final int PAGE_SIZE = 1000;
+ private static final long ALLOCATOR_LIMIT = 256L * 1024 * 1024;
+
+ private transient LanceNamespace namespace;
+ private transient BufferAllocator allocator;
+ private transient List<String> parentNamespace = Collections.emptyList();
+ private transient String catalogType;
+ private transient String namespaceDelimiter;
+ private transient String rootDatabase;
+ private transient Map<String, String> javaStorageOptions =
Collections.emptyMap();
+ private transient Map<String, String> backendStorageOptions =
Collections.emptyMap();
+ private transient Object namespaceLock = new Object();
+
+ public LanceExternalCatalog(long catalogId, String name, String resource,
Map<String, String> props,
+ String comment) {
+ super(catalogId, name, InitCatalogLog.Type.LANCE, comment);
+ catalogProperty = new CatalogProperty(resource, props);
+ }
+
+ @Override
+ protected void initLocalObjectsImpl() {
+ try {
+ namespaceLock = new Object();
+ catalogType = normalizedCatalogType();
+ namespaceDelimiter =
catalogProperty.getOrDefault(NAMESPACE_DELIMITER, DEFAULT_DELIMITER);
+ rootDatabase = catalogProperty.getOrDefault(ROOT_DATABASE,
DEFAULT_ROOT_DATABASE);
+ parentNamespace = LanceNamespaceName.parseParent(
+ catalogProperty.getOrDefault(NAMESPACE_PARENT, ""),
namespaceDelimiter);
+ backendStorageOptions =
catalogProperty.getBackendStorageProperties();
+ javaStorageOptions =
LanceStorageOptions.forJavaSdk(backendStorageOptions);
+
+ allocator = new RootAllocator(ALLOCATOR_LIMIT);
+ namespace = connectNamespace(catalogType, allocator,
javaStorageOptions);
+ } catch (Exception e) {
+ closeLanceObjects();
+ throw new RuntimeException("Failed to initialize Lance " +
normalizedCatalogType() + " catalog '"
+ + getName()
+ + "': " + sanitizedRootCauseMessage(e), safeCause(e));
+ }
+ }
+
+ @Override
+ public void checkWhenCreating() throws DdlException {
+ checkProperties();
+ boolean testConnection = Boolean.parseBoolean(
+ catalogProperty.getOrDefault(TEST_CONNECTION,
String.valueOf(DEFAULT_TEST_CONNECTION)));
+ if (!testConnection) {
+ return;
+ }
+
+ Map<String, String> storageOptions = LanceStorageOptions.forJavaSdk(
+ catalogProperty.getBackendStorageProperties());
+ String delimiter = catalogProperty.getOrDefault(NAMESPACE_DELIMITER,
DEFAULT_DELIMITER);
+ List<String> parent = LanceNamespaceName.parseParent(
+ catalogProperty.getOrDefault(NAMESPACE_PARENT, ""), delimiter);
+ String type = normalizedCatalogType();
+ try (BufferAllocator testAllocator = new
RootAllocator(ALLOCATOR_LIMIT)) {
+ LanceNamespace testNamespace = connectNamespace(type,
testAllocator, storageOptions);
+ try {
+ testNamespace.listTables(new
ListTablesRequest().id(parent).limit(1));
+ testNamespace.listNamespaces(new
ListNamespacesRequest().id(parent).limit(1));
+ } finally {
+ closeNamespace(testNamespace);
+ }
+ } catch (Exception e) {
+ throw new DdlException("Lance " + type + " catalog connectivity
test failed: "
+ + sanitizedRootCauseMessage(e), safeCause(e));
+ }
+ }
+
+ private LanceNamespace connectNamespace(String type, BufferAllocator
namespaceAllocator,
+ Map<String, String> storageOptions) {
+ if (LANCE_FILESYSTEM.equals(type)) {
+ return connectDirectoryNamespace(namespaceAllocator,
storageOptions);
+ }
+ if (LANCE_REST.equals(type)) {
+ return connectRestNamespace(namespaceAllocator);
+ }
+ throw new IllegalArgumentException("Unsupported Lance catalog type '"
+ type + "'");
+ }
+
+ private LanceNamespace connectDirectoryNamespace(BufferAllocator
namespaceAllocator,
+ Map<String, String> storageOptions) {
+ Map<String, String> namespaceProperties = new HashMap<>();
+ namespaceProperties.put("root",
catalogProperty.getOrDefault(WAREHOUSE, ""));
+ storageOptions.forEach((key, value) ->
namespaceProperties.put("storage." + key, value));
+ return LanceNamespace.connect("dir", namespaceProperties,
namespaceAllocator);
+ }
+
+ private LanceNamespace connectRestNamespace(BufferAllocator
namespaceAllocator) {
+ Map<String, String> namespaceProperties = new HashMap<>();
+ namespaceProperties.put("uri", normalizedRestUri());
+ namespaceProperties.put("delimiter",
+ catalogProperty.getOrDefault(NAMESPACE_DELIMITER,
DEFAULT_DELIMITER));
+
+ Map<String, String> properties = catalogProperty.getProperties();
+ properties.forEach((key, value) -> {
+ if (key.startsWith(REST_HEADER_PREFIX)) {
+ namespaceProperties.put("header." +
key.substring(REST_HEADER_PREFIX.length()), value);
+ }
+ });
+ String securityType = properties.getOrDefault(REST_SECURITY_TYPE,
REST_SECURITY_NONE)
+ .trim().toLowerCase(Locale.ROOT);
+ if (REST_SECURITY_BEARER.equals(securityType)) {
+ namespaceProperties.put("header.Authorization", "Bearer " +
properties.get(REST_BEARER_TOKEN));
+ } else if (REST_SECURITY_API_KEY.equals(securityType)) {
+ namespaceProperties.put("header.x-api-key",
properties.get(REST_API_KEY));
+ }
+ return LanceNamespace.connect("rest", namespaceProperties,
namespaceAllocator);
+ }
+
+ @Override
+ public void checkProperties() throws DdlException {
+ super.checkProperties();
+ Map<String, String> properties = catalogProperty.getProperties();
+ String type = normalizedCatalogType();
+ if (!LANCE_FILESYSTEM.equals(type) && !LANCE_REST.equals(type)) {
+ throw new DdlException("Property '" + LANCE_CATALOG_TYPE
+ + "' must be 'filesystem' or 'rest', but was '" + type +
"'");
+ }
+ validateCommonProperties(properties);
+ if (LANCE_FILESYSTEM.equals(type)) {
+ validateFilesystemProperties(properties);
+ } else {
+ validateRestProperties(properties);
+ }
+ }
+
+ private void validateCommonProperties(Map<String, String> properties)
throws DdlException {
+ String delimiter = properties.getOrDefault(NAMESPACE_DELIMITER,
DEFAULT_DELIMITER);
+ if (delimiter.isEmpty() || delimiter.indexOf('\\') >= 0) {
+ throw new DdlException("Property '" + NAMESPACE_DELIMITER
+ + "' cannot be empty or contain the escape character
'\\'");
+ }
+ String rootDb = properties.getOrDefault(ROOT_DATABASE,
DEFAULT_ROOT_DATABASE);
+ if (StringUtils.isBlank(rootDb)) {
+ throw new DdlException("Property '" + ROOT_DATABASE
+ + "' must be non-empty");
+ }
+
LanceNamespaceName.parseParent(properties.getOrDefault(NAMESPACE_PARENT, ""),
delimiter);
+ }
+
+ private void validateFilesystemProperties(Map<String, String> properties)
throws DdlException {
+ String warehouse = properties.get(WAREHOUSE);
+ if (StringUtils.isBlank(warehouse)) {
+ throw new DdlException("Missing required property 'warehouse' for
Lance filesystem catalog");
+ }
+ validateWarehouse(warehouse);
+ for (String key : properties.keySet()) {
+ if (key.startsWith("lance.rest.")) {
+ throw new DdlException("Property '" + key + "' is not valid
for Lance filesystem catalog");
+ }
+ }
+ }
+
+ private void validateRestProperties(Map<String, String> properties) throws
DdlException {
+ if (properties.containsKey(WAREHOUSE)) {
+ throw new DdlException("Property 'warehouse' is not valid for
Lance REST catalog");
+ }
+ String restUri = properties.get(REST_URI);
+ if (StringUtils.isBlank(restUri)) {
+ throw new DdlException("Missing required property '" + REST_URI +
"' for Lance REST catalog");
+ }
+ validateRestUri(restUri);
+
+ String securityType = properties.getOrDefault(REST_SECURITY_TYPE,
REST_SECURITY_NONE)
+ .trim().toLowerCase(Locale.ROOT);
+ boolean bearerTokenConfigured =
properties.containsKey(REST_BEARER_TOKEN);
+ boolean apiKeyConfigured = properties.containsKey(REST_API_KEY);
+ boolean hasBearerToken =
StringUtils.isNotBlank(properties.get(REST_BEARER_TOKEN));
+ boolean hasApiKey =
StringUtils.isNotBlank(properties.get(REST_API_KEY));
+ switch (securityType) {
+ case REST_SECURITY_NONE:
+ if (bearerTokenConfigured || apiKeyConfigured) {
+ throw new DdlException("Lance REST security type 'none'
cannot configure '"
+ + REST_BEARER_TOKEN + "' or '" + REST_API_KEY +
"'");
+ }
+ break;
+ case REST_SECURITY_BEARER:
+ if (!hasBearerToken || apiKeyConfigured) {
+ throw new DdlException("Lance REST security type 'bearer'
requires only '"
+ + REST_BEARER_TOKEN + "'");
+ }
+ validateCredentialHeaderValue(REST_BEARER_TOKEN,
properties.get(REST_BEARER_TOKEN));
+ break;
+ case REST_SECURITY_API_KEY:
+ if (!hasApiKey || bearerTokenConfigured) {
+ throw new DdlException("Lance REST security type 'api_key'
requires only '"
+ + REST_API_KEY + "'");
+ }
+ validateCredentialHeaderValue(REST_API_KEY,
properties.get(REST_API_KEY));
+ break;
+ default:
+ throw new DdlException("Property '" + REST_SECURITY_TYPE
+ + "' must be 'none', 'bearer', or 'api_key'");
+ }
+
+ Set<String> supportedKeys = new HashSet<>();
+ Collections.addAll(supportedKeys, REST_URI, REST_SECURITY_TYPE,
REST_BEARER_TOKEN, REST_API_KEY);
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ String key = entry.getKey();
+ if (!key.startsWith("lance.rest.") || supportedKeys.contains(key))
{
+ continue;
+ }
+ if (!key.startsWith(REST_HEADER_PREFIX)) {
+ throw new DdlException("Unsupported Lance REST property '" +
key + "'");
+ }
+ validateRestHeader(key.substring(REST_HEADER_PREFIX.length()),
entry.getValue());
+ }
+ }
+
+ private void validateRestHeader(String headerName, String headerValue)
throws DdlException {
+ if (StringUtils.isBlank(headerName) ||
!HTTP_HEADER_NAME.matcher(headerName).matches()) {
+ throw new DdlException("Invalid HTTP header name in property '" +
REST_HEADER_PREFIX
+ + headerName + "'");
+ }
+ if ("authorization".equalsIgnoreCase(headerName) ||
"x-api-key".equalsIgnoreCase(headerName)) {
+ throw new DdlException("Authentication header '" + headerName + "'
must be configured through '"
+ + REST_SECURITY_TYPE + "'");
+ }
+ if (headerValue == null || headerValue.indexOf('\r') >= 0 ||
headerValue.indexOf('\n') >= 0) {
+ throw new DdlException("Invalid HTTP header value in property '" +
REST_HEADER_PREFIX
+ + headerName + "'");
+ }
+ }
+
+ private void validateCredentialHeaderValue(String propertyName, String
value) throws DdlException {
+ if (value.indexOf('\r') >= 0 || value.indexOf('\n') >= 0) {
+ throw new DdlException("Invalid HTTP credential value in property
'" + propertyName + "'");
+ }
+ }
+
+ private void validateRestUri(String value) throws DdlException {
+ URI uri;
+ try {
+ uri = URI.create(value.trim());
+ } catch (IllegalArgumentException e) {
+ throw new DdlException("Invalid Lance REST URI in property '" +
REST_URI + "'", e);
+ }
+ String scheme = uri.getScheme();
+ if (scheme == null || (!("http".equalsIgnoreCase(scheme)) &&
!("https".equalsIgnoreCase(scheme)))) {
+ throw new DdlException("Property '" + REST_URI + "' must use http
or https");
+ }
+ if (StringUtils.isBlank(uri.getRawAuthority()) || uri.getRawUserInfo()
!= null
+ || uri.getRawQuery() != null || uri.getRawFragment() != null) {
+ throw new DdlException("Property '" + REST_URI
+ + "' must contain an authority and cannot contain
user-info, query, or fragment");
+ }
+ }
+
+ private String normalizedCatalogType() {
+ return catalogProperty.getOrDefault(LANCE_CATALOG_TYPE,
LANCE_FILESYSTEM)
+ .trim().toLowerCase(Locale.ROOT);
+ }
+
+ private String normalizedRestUri() {
+ String uri = catalogProperty.getOrDefault(REST_URI, "").trim();
+ while (uri.endsWith("/")) {
+ uri = uri.substring(0, uri.length() - 1);
+ }
+ return uri;
+ }
+
+ private void validateWarehouse(String warehouse) throws DdlException {
+ URI uri;
+ try {
+ uri = URI.create(warehouse);
+ } catch (IllegalArgumentException e) {
+ throw new DdlException("Invalid Lance warehouse URI: " +
warehouse, e);
+ }
+ if (uri.getScheme() == null) {
+ Path path = Paths.get(warehouse);
+ if (!path.isAbsolute()) {
+ throw new DdlException("Local Lance warehouse must be an
absolute path: " + warehouse);
+ }
+ return;
+ }
+ String scheme = uri.getScheme().toLowerCase();
+ if (!"file".equals(scheme) && !"s3".equals(scheme)) {
+ throw new DdlException("Unsupported Lance filesystem warehouse
scheme '" + scheme
+ + "'; first phase supports local/file and s3");
Review Comment:
[P1] Do not advertise FE-local warehouses without a placement contract
This explicitly accepts bare and `file:` warehouses, which the FE opens
successfully for namespace and metadata work. The resulting `LanceSplit` has no
host locality, however, so ordinary scan planning assigns its local URI to
arbitrary BEs; on a normal multi-host deployment those BEs see a different or
nonexistent filesystem and queries fail after catalog validation succeeded.
Please reject local/file catalog warehouses, or require an explicit
shared-mount contract and constrain placement to BEs known to see that path.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -0,0 +1,912 @@
+// 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/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/type.h>
+#include <lance/lance.h>
+
+#include <bit>
+#include <cstring>
+#include <limits>
+#include <memory>
+
+#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_struct.h"
+#include "exec/common/endian.h"
+
+namespace doris::format::lance {
+namespace {
+
+struct LanceDatasetDeleter {
+ void operator()(LanceDataset* dataset) const {
lance_dataset_close(dataset); }
+};
+
+struct LanceScannerDeleter {
+ void operator()(LanceScanner* scanner) const {
lance_scanner_close(scanner); }
+};
+
+struct LanceBatchDeleter {
+ void operator()(LanceBatch* batch) const { lance_batch_free(batch); }
+};
+
+constexpr std::string_view DISTANCE_COLUMN = "_distance";
+
+size_t vector_element_width(TVectorElementType::type type) {
+ switch (type) {
+ case TVectorElementType::FLOAT16:
+ return sizeof(uint16_t);
+ case TVectorElementType::FLOAT32:
+ return sizeof(float);
+ case TVectorElementType::FLOAT64:
+ return sizeof(double);
+ case TVectorElementType::UINT8:
+ case TVectorElementType::INT8:
+ return sizeof(uint8_t);
+ }
+ return 0;
+}
+
+std::string vector_metric_name(TVectorMetric::type metric) {
+ switch (metric) {
+ case TVectorMetric::DEFAULT:
+ return "default";
+ case TVectorMetric::L2:
+ return "l2";
+ case TVectorMetric::COSINE:
+ return "cosine";
+ case TVectorMetric::DOT_PRODUCT:
+ return "dot";
+ case TVectorMetric::HAMMING:
+ return "hamming";
+ }
+ return "unknown";
+}
+
+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;
+}
+
+Status arrow_type_to_doris_type(const std::shared_ptr<arrow::DataType>&
arrow_type,
+ 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();
+ };
+
+ switch (arrow_type->id()) {
+ 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 doris_type = timestamp->timezone().empty() ?
TYPE_DATETIMEV2 : TYPE_TIMESTAMPTZ;
+ return nullable_primitive(doris_type, 0,
arrow_time_precision(timestamp->unit()));
+ }
+ 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 PrimitiveType doris_decimal_type = precision <= 9 ?
TYPE_DECIMAL32
+ : precision <= 18 ?
TYPE_DECIMAL64
+ : precision <= 38 ?
TYPE_DECIMAL128I
+ :
TYPE_DECIMAL256;
+ return nullable_primitive(doris_decimal_type, precision,
decimal->scale());
+ }
+ case arrow::Type::LIST:
+ case arrow::Type::LARGE_LIST:
+ case arrow::Type::FIXED_SIZE_LIST: {
+ const auto list =
std::static_pointer_cast<arrow::BaseListType>(arrow_type);
+ DataTypePtr value_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(list->value_type(),
&value_type));
+ *doris_type =
make_nullable(std::make_shared<DataTypeArray>(value_type));
+ return Status::OK();
+ }
+ case arrow::Type::MAP: {
+ const auto map = std::static_pointer_cast<arrow::MapType>(arrow_type);
+ DataTypePtr key_type;
+ DataTypePtr item_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(map->key_type(), &key_type));
+ RETURN_IF_ERROR(arrow_type_to_doris_type(map->item_type(),
&item_type));
+ *doris_type = make_nullable(std::make_shared<DataTypeMap>(key_type,
item_type));
+ return Status::OK();
+ }
+ case arrow::Type::STRUCT: {
+ const auto struct_type =
std::static_pointer_cast<arrow::StructType>(arrow_type);
+ DataTypes field_types;
+ Strings field_names;
+ field_types.reserve(struct_type->num_fields());
+ field_names.reserve(struct_type->num_fields());
+ for (const auto& field : struct_type->fields()) {
+ DataTypePtr field_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(field->type(),
&field_type));
+ field_types.emplace_back(std::move(field_type));
+ field_names.emplace_back(field->name());
+ }
+ *doris_type =
make_nullable(std::make_shared<DataTypeStruct>(field_types, field_names));
+ return Status::OK();
+ }
+ default:
+ return Status::NotSupported("unsupported Lance Arrow type: {}",
arrow_type->ToString());
+ }
+}
+
+} // namespace
+
+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");
+ }
+ RETURN_IF_ERROR(_validate_range(range));
+
+ const auto& params = range.table_format_params.lance_params;
+ const auto storage_options = _storage_options(&scan_params);
+ 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");
+ }
+
+ ArrowSchema arrow_schema {};
+ if (lance_dataset_schema(dataset.get(), &arrow_schema) != 0) {
+ return _lance_error("get Lance dataset schema");
+ }
+ auto imported_schema = arrow::ImportSchema(&arrow_schema);
+ if (!imported_schema.ok()) {
+ if (arrow_schema.release != nullptr) {
+ arrow_schema.release(&arrow_schema);
+ }
+ return Status::InternalError("import Lance Arrow schema failed: {}",
+ imported_schema.status().message());
+ }
+
+ const auto schema = std::move(imported_schema).ValueUnsafe();
+ std::vector<std::string> parsed_names;
+ std::vector<DataTypePtr> parsed_types;
+ parsed_names.reserve(schema->num_fields());
+ parsed_types.reserve(schema->num_fields());
+ std::unordered_set<std::string> unique_names;
+ unique_names.reserve(schema->num_fields());
+ for (const auto& field : schema->fields()) {
+ if (!unique_names.emplace(field->name()).second) {
+ return Status::InvalidArgument("duplicate Lance schema column:
{}", field->name());
+ }
+ DataTypePtr doris_type;
+ RETURN_IF_ERROR(arrow_type_to_doris_type(field->type(), &doris_type));
+ parsed_names.emplace_back(field->name());
+ parsed_types.emplace_back(std::move(doris_type));
+ }
+ *column_names = std::move(parsed_names);
+ *column_types = std::move(parsed_types);
+ return Status::OK();
+}
+
+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);
+
+ _ctz = _runtime_state->timezone_obj();
+ _vector_search = _scan_params->__isset.external_search_request;
+ _search_split_prepared = false;
+ if (_vector_search) {
+ RETURN_IF_ERROR(_validate_external_search_request());
+ const auto& vector =
_scan_params->external_search_request.query.vector;
+ _scanner_profile->add_info_string("ExternalSearchType", "VECTOR");
+ _scanner_profile->add_info_string("LanceVectorColumn", vector.column);
+ _scanner_profile->add_info_string("LanceTopK",
std::to_string(vector.top_k));
+ _scanner_profile->add_info_string("LanceOffset",
std::to_string(vector.offset));
+ _scanner_profile->add_info_string("LanceMetric", vector.__isset.metric
+ ?
vector_metric_name(vector.metric)
+ : "default");
+ _scanner_profile->add_info_string("LanceVectorDimension",
+
std::to_string(vector.query_vector.dimension));
+ }
+ if (_scan_params->__isset.lance_substrait_filter) {
+ _scanner_profile->add_info_string("LancePushdownFormat", "SUBSTRAIT");
+ _scanner_profile->add_info_string(
+ "LanceSubstraitFilterBytes",
+ std::to_string(_scan_params->lance_substrait_filter.size()));
+ }
+
+ _output_name_to_idx.clear();
+ _output_name_to_idx.reserve(_projected_columns.size());
+ for (size_t idx = 0; idx < _projected_columns.size(); ++idx) {
+ const auto& column = _projected_columns[idx];
+ if (column.type == nullptr) {
+ return Status::InvalidArgument("Lance projected column '{}' has no
type", column.name);
+ }
+ if (!_output_name_to_idx.emplace(column.name, idx).second) {
+ return Status::InvalidArgument("duplicate Lance projected column:
{}", column.name);
+ }
+ if (_vector_search && column.name == DISTANCE_COLUMN) {
+ const auto distance_type = remove_nullable(column.type);
+ if (distance_type->get_primitive_type() != TYPE_FLOAT) {
+ return Status::InvalidArgument(
+ "Lance vector search column '{}' must have Doris FLOAT
type, but was {}",
+ DISTANCE_COLUMN, column.type->get_name());
+ }
+ }
+ }
+ return Status::OK();
+}
+
+Status LanceTableReader::prepare_split(const SplitReadOptions& options) {
+ RETURN_IF_ERROR(_validate_range(options.current_range));
+ _close_scanner();
+ _eof = false;
+
+ RETURN_IF_ERROR(TableReader::prepare_split(options));
+ // Lance does not currently provide metadata aggregate pushdown. Do not
let a generic
+ // table-level count supplied by a future planner bypass fragment reads.
+ _remaining_table_level_count = -1;
+ if (current_split_pruned()) {
+ return Status::OK();
+ }
+
+ if (_vector_search) {
+ const auto& lance_params =
options.current_range.table_format_params.lance_params;
+ if (lance_params.version <= 0) {
+ return Status::InvalidArgument(
+ "Lance vector search requires a fixed positive dataset
version");
+ }
+ if (lance_params.__isset.fragment_ids) {
+ return Status::InvalidArgument("Lance vector search split must not
set fragment ids");
+ }
+ if (_search_split_prepared) {
+ return Status::InvalidArgument(
+ "Lance vector search supports exactly one whole-dataset
split");
+ }
+ }
+
+ const auto key = _dataset_key(options.current_range);
+ 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
in one scan");
+ }
+
+ RETURN_IF_ERROR(_prepare_conjuncts());
+ RETURN_IF_ERROR(_open_scanner(options.current_range));
+ _search_split_prepared = _vector_search;
+ 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();
+ }
+ 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;
+ const int32_t scan_status = lance_scanner_next(_scanner,
&raw_batch);
Review Comment:
[P1] Integrate Lance I/O with Doris resource governance
This synchronous lance-c call owns the physical local/S3 reads, but it has
no Doris `FileReader`/`IOContext`, byte/call/time reporting, workload-group
throttle, or in-flight cancellation handle. Consequently Lance scans report
zero scan bytes and remote/local attribution, bypass the S3 remote-scan
limiter, and cannot stop a slow object-store request until it returns. Please
route physical I/O through the common layer or add callbacks that provide
accounting, throttling, and cancellation before enabling this scan path.
--
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]
