Gabriel39 commented on code in PR #65730:
URL: https://github.com/apache/doris/pull/65730#discussion_r3689218590
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java:
##########
@@ -87,11 +87,15 @@ public LocalTableValuedFunction(Map<String, String>
properties) throws AnalysisE
throw new AnalysisException("'backend_id' is required when
'shared_storage' is false.");
}
- // 3. parse file
- getFileListFromBackend();
+ initializeBackendForRequest();
+ if (!isLanceFormat()) {
Review Comment:
[Blocker] Please preserve the local TVF secure-path validation here.
getFileListFromBackend() currently reaches LocalFileSystem::safe_glob(), which
prefixes user_files_secure_path, rejects .., and canonicalizes containment.
Skipping the call for Lance makes the raw user path flow into LanceSplit and
eventually lance_dataset_open(), so absolute or traversal paths can escape the
intended sandbox and relative paths resolve from the BE working directory
instead of user_files_secure_path. Lance does need a single dataset-level
split, but please add a directory resolver that performs the same security
checks without enumerating dataset files, and add tests for relative, absolute,
.., and symlink-escape paths.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/source/LancePredicateConverter.java:
##########
@@ -0,0 +1,822 @@
+// 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.BinaryPredicate;
+import org.apache.doris.analysis.BoolLiteral;
+import org.apache.doris.analysis.CompoundPredicate;
+import org.apache.doris.analysis.DateLiteral;
+import org.apache.doris.analysis.DecimalLiteral;
+import org.apache.doris.analysis.Expr;
+import org.apache.doris.analysis.FloatLiteral;
+import org.apache.doris.analysis.InPredicate;
+import org.apache.doris.analysis.IntLiteral;
+import org.apache.doris.analysis.IsNullPredicate;
+import org.apache.doris.analysis.LargeIntLiteral;
+import org.apache.doris.analysis.LiteralExpr;
+import org.apache.doris.analysis.NullLiteral;
+import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.analysis.StringLiteral;
+
+import io.substrait.expression.Expression;
+import io.substrait.expression.ExpressionCreator;
+import io.substrait.expression.FieldReference;
+import io.substrait.expression.proto.ExpressionProtoConverter;
+import io.substrait.extension.DefaultExtensionCatalog;
+import io.substrait.extension.ExtensionCollector;
+import io.substrait.extension.SimpleExtension;
+import io.substrait.proto.ExpressionReference;
+import io.substrait.proto.ExtendedExpression;
+import io.substrait.proto.NamedStruct;
+import io.substrait.proto.Type.Nullability;
+import io.substrait.relation.RelProtoConverter;
+import io.substrait.type.Type;
+import io.substrait.type.TypeCreator;
+import org.apache.arrow.vector.types.DateUnit;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.time.DateTimeException;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Converts predicates with identical Doris and Lance semantics to a
Substrait ExtendedExpression. */
+public class LancePredicateConverter {
+ private static final int MAX_SUBSTRAIT_DECIMAL_PRECISION = 38;
+ // DataFusion represents Arrow unsigned integers as the corresponding
Substrait signed
+ // integer type class with type variation 1.
+ private static final int UNSIGNED_INTEGER_TYPE_VARIATION_REFERENCE = 1;
+ // DataFusion distinguishes Arrow LargeUtf8 from Utf8 with the Substrait
+ // large-container type variation.
+ private static final int LARGE_CONTAINER_TYPE_VARIATION_REFERENCE = 1;
+ private static final String UNSUPPORTED_FIELD_PREFIX =
"__unlikely_name_placeholder_doris_";
+ private static final TypeCreator REQUIRED = TypeCreator.of(false);
+ private static final SimpleExtension.ExtensionCollection EXTENSIONS =
loadExtensions();
+
+ private final Schema schema;
+ private final Map<String, ResolvedField> fieldsByLowerCaseName = new
HashMap<>();
+
+ public LancePredicateConverter(Schema schema) {
+ this.schema = schema;
+ List<Field> fields = schema.getFields();
+ for (int ordinal = 0; ordinal < fields.size(); ordinal++) {
+ Field field = fields.get(ordinal);
+ fieldsByLowerCaseName.putIfAbsent(
+ field.getName().toLowerCase(Locale.ROOT), new
ResolvedField(field, ordinal));
+ }
+ }
+
+ public ConversionResult convert(List<Expr> conjuncts) {
+ List<Expression> filters = new ArrayList<>();
+ List<Expr> pushedConjuncts = new ArrayList<>();
+ for (Expr conjunct : conjuncts) {
+ Optional<Expression> filter = convert(conjunct);
+ if (filter.isPresent()) {
+ filters.add(filter.get());
+ pushedConjuncts.add(conjunct);
+ }
+ }
+ if (filters.isEmpty()) {
+ return new ConversionResult(new byte[0], "", pushedConjuncts);
+ }
+
+ Expression combined = filters.size() == 1 ? filters.get(0) :
booleanFunction("and:bool", filters);
+ String debugPredicate = pushedConjuncts.stream()
+ .map(Expr::toSql)
+ .map(LancePredicateConverter::parenthesize)
+ .collect(Collectors.joining(" AND "));
+ return new ConversionResult(serialize(combined), debugPredicate,
pushedConjuncts);
+ }
+
+ private Optional<Expression> convert(Expr expr) {
+ if (expr instanceof BinaryPredicate) {
+ return convertBinary((BinaryPredicate) expr);
+ }
+ if (expr instanceof CompoundPredicate) {
+ return convertCompound((CompoundPredicate) expr);
+ }
+ if (expr instanceof InPredicate) {
+ return convertIn((InPredicate) expr);
+ }
+ if (expr instanceof IsNullPredicate) {
+ return convertIsNull((IsNullPredicate) expr);
+ }
+ return Optional.empty();
+ }
+
+ private Optional<Expression> convertBinary(BinaryPredicate predicate) {
+ SlotRef slot = directSlot(predicate.getChild(0));
+ LiteralExpr literal = directLiteral(predicate.getChild(1));
+ BinaryPredicate.Operator operator = predicate.getOp();
+ if (slot == null || literal == null) {
+ slot = directSlot(predicate.getChild(1));
+ literal = directLiteral(predicate.getChild(0));
+ operator = reverse(operator);
+ }
+ if (slot == null || literal == null || operator == null) {
+ return Optional.empty();
+ }
+
+ ResolvedField field = findField(slot);
+ if (field == null || !isPushdownType(field.field.getType())) {
+ return Optional.empty();
+ }
+ Expression fieldReference = fieldReference(field);
+ if (literal instanceof NullLiteral) {
+ if (operator == BinaryPredicate.Operator.EQ_FOR_NULL) {
+ return Optional.of(comparisonFunction("is_null:any",
fieldReference));
+ }
+ return Optional.empty();
+ }
+ Optional<Expression> value = convertLiteral(field.field.getType(),
literal);
+ if (!value.isPresent()) {
+ return Optional.empty();
+ }
+
+ String function;
+ switch (operator) {
+ case EQ:
+ case EQ_FOR_NULL:
Review Comment:
[Correctness] EQ_FOR_NULL cannot be translated to ordinary equal for a
non-null literal when the result can appear under NOT or another boolean
composition. For example, NOT(nullable_col <=> 1) must be TRUE for a NULL value
in Doris, while Substrait NOT(equal(NULL, 1)) evaluates to NULL and the Lance
prefilter drops the row. Please encode a non-null boolean expression that
preserves <=> semantics, or leave this form as a residual predicate, and add
nullable NOT/AND/OR tests.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/VectorSearchTableValuedFunction.java:
##########
@@ -0,0 +1,322 @@
+// 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.tablefunction;
+
+import org.apache.doris.analysis.TableName;
+import org.apache.doris.analysis.TupleDescriptor;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.CatalogIf;
+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.datasource.lance.LanceTypeConverter;
+import org.apache.doris.datasource.lance.LanceVectorQuery;
+import org.apache.doris.datasource.lance.source.LanceScanNode;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.planner.PlanNodeId;
+import org.apache.doris.planner.ScanNode;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TExternalSearchQuery;
+import org.apache.doris.thrift.TExternalSearchRequest;
+import org.apache.doris.thrift.TLanceVectorSearchOptions;
+import org.apache.doris.thrift.TSearchFilter;
+import org.apache.doris.thrift.TSearchFilterFormat;
+import org.apache.doris.thrift.TSearchVector;
+import org.apache.doris.thrift.TVectorMetric;
+import org.apache.doris.thrift.TVectorSearchParams;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.arrow.vector.types.pojo.Field;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+/** Relation TVF for a whole-snapshot Lance vector search. */
+public class VectorSearchTableValuedFunction extends TableValuedFunctionIf {
+ public static final String NAME = "vector_search";
+ public static final String DISTANCE_COLUMN = "_distance";
+
+ private static final String TABLE = "table";
+ private static final String COLUMN = "column";
+ private static final String QUERY_VECTOR = "query_vector";
+ private static final String TOP_K = "top_k";
+ private static final String OFFSET = "offset";
+ private static final String METRIC = "metric";
+ private static final String FILTER = "filter";
+ private static final String NPROBES = "nprobes";
+ private static final String REFINE_FACTOR = "refine_factor";
+ private static final String EF = "ef";
+ private static final String USE_INDEX = "use_index";
+ private static final long UINT32_MAX = 0xFFFF_FFFFL;
+ private static final Set<String> PROPERTIES = ImmutableSet.of(
+ TABLE, COLUMN, QUERY_VECTOR, TOP_K, OFFSET, METRIC, FILTER,
+ NPROBES, REFINE_FACTOR, EF, USE_INDEX);
+
+ private final TableName sourceTableName;
+ private final LanceExternalTable sourceTable;
+ private final LanceTableMetadata metadata;
+ private final List<Column> columns;
+ private final TExternalSearchRequest searchRequest;
+
+ public VectorSearchTableValuedFunction(Map<String, String> properties)
+ throws AnalysisException {
+ Map<String, String> params = normalizeProperties(properties);
+ sourceTableName = parseTableName(required(params, TABLE));
+ checkSelectPrivilege(sourceTableName);
+ sourceTable = resolveLanceTable(sourceTableName);
+ try {
+ metadata = sourceTable.loadMetadata();
+ } catch (RuntimeException e) {
+ throw new AnalysisException("Failed to load Lance metadata for
vector search on "
+ + sourceTableName + ": " + e.getMessage(), e);
+ }
+ if (metadata.getVersion() <= 0) {
+ throw new AnalysisException("Lance vector search requires a fixed
positive dataset version");
+ }
+
+ Field vectorField = LanceVectorQuery.resolveVectorField(
+ metadata.getSchema(), required(params, COLUMN));
+ TSearchVector queryVector = LanceVectorQuery.encode(
+ vectorField, required(params, QUERY_VECTOR));
+ long topK = parseLong(params.getOrDefault(TOP_K, "10"), TOP_K, 1,
Long.MAX_VALUE);
+ long offset = parseLong(params.getOrDefault(OFFSET, "0"), OFFSET, 0,
Long.MAX_VALUE);
+ if (offset > UINT32_MAX || topK > UINT32_MAX - offset) {
+ throw new AnalysisException("'top_k + offset' must not exceed " +
UINT32_MAX);
+ }
+
+ TVectorSearchParams vectorParams = new TVectorSearchParams()
+ .setColumn(vectorField.getName())
+ .setQueryVector(queryVector)
+ .setTopK(topK)
+ .setOffset(offset);
+ if (params.containsKey(METRIC)) {
+ vectorParams.setMetric(parseMetric(params.get(METRIC)));
+ }
+
+ searchRequest = new TExternalSearchRequest()
+ .setSchemaVersion(1)
+ .setQuery(TExternalSearchQuery.vector(vectorParams));
+ if (params.containsKey(FILTER)) {
+ String filter = params.get(FILTER);
+ if (filter == null || filter.trim().isEmpty()) {
+ throw new AnalysisException("'filter' must not be empty");
+ }
+ searchRequest.setFilter(new TSearchFilter()
+ .setFormat(TSearchFilterFormat.SQL)
+ .setPayload(filter.getBytes(StandardCharsets.UTF_8)));
+ }
+
+ TLanceVectorSearchOptions lanceOptions = new
TLanceVectorSearchOptions();
+ boolean hasLanceOptions = false;
+ if (params.containsKey(NPROBES)) {
+ lanceOptions.setNprobes(parsePositiveInt(params.get(NPROBES),
NPROBES));
+ hasLanceOptions = true;
+ }
+ if (params.containsKey(REFINE_FACTOR)) {
+ lanceOptions.setRefineFactor(
+ parsePositiveInt(params.get(REFINE_FACTOR),
REFINE_FACTOR));
+ hasLanceOptions = true;
+ }
+ if (params.containsKey(EF)) {
+ lanceOptions.setEf(parsePositiveInt(params.get(EF), EF));
+ hasLanceOptions = true;
+ }
+ if (params.containsKey(USE_INDEX)) {
+ lanceOptions.setUseIndex(parseBoolean(params.get(USE_INDEX),
USE_INDEX));
+ hasLanceOptions = true;
+ }
+ if (hasLanceOptions) {
+ searchRequest.setLanceOptions(lanceOptions);
+ }
+ columns = buildOutputColumns(metadata);
+ }
+
+ public LanceExternalTable getSourceTable() {
+ return sourceTable;
+ }
+
+ public LanceTableMetadata getMetadata() {
+ return metadata;
+ }
+
+ public TExternalSearchRequest getSearchRequest() {
+ return searchRequest.deepCopy();
+ }
+
+ @Override
+ public String getTableName() {
+ return "VectorSearchTableValuedFunction<" + sourceTableName + ">";
+ }
+
+ @Override
+ public List<Column> getTableColumns() {
+ return columns;
+ }
+
+ @Override
+ public ScanNode getScanNode(PlanNodeId id, TupleDescriptor desc,
SessionVariable sv) {
+ return new LanceScanNode(id, desc, sourceTable, metadata,
+ searchRequest, sv);
+ }
+
+ private static Map<String, String> normalizeProperties(Map<String, String>
properties)
+ throws AnalysisException {
+ Map<String, String> normalized = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ String key = entry.getKey().toLowerCase(Locale.ROOT);
+ if (!PROPERTIES.contains(key)) {
+ throw new AnalysisException("'" + entry.getKey()
+ + "' is an invalid property for vector_search()");
+ }
+ if (normalized.put(key, entry.getValue()) != null) {
+ throw new AnalysisException("Duplicate vector_search()
property '" + key + "'");
+ }
+ }
+ return normalized;
+ }
+
+ private static String required(Map<String, String> params, String key)
+ throws AnalysisException {
+ String value = params.get(key);
+ if (value == null || value.trim().isEmpty()) {
+ throw new AnalysisException("Missing required vector_search()
property '" + key + "'");
+ }
+ return value.trim();
+ }
+
+ private static TableName parseTableName(String value) throws
AnalysisException {
+ String[] names = value.split("\\.", -1);
Review Comment:
[Correctness] This parser makes vector_search() unusable for the multi-level
namespaces supported by this PR. A documented database such as doris.analytics
produces lance_catalog.doris.analytics.items, which splits into four parts;
putting backticks inside this string does not make String.split() quote-aware.
Please use a qualified-name parser with quoting support or pass
catalog/database/table separately, and add a vector_search() test for a
multi-level namespace.
##########
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));
Review Comment:
[Correctness/docs] Local TVF schema discovery aborts on the first
unsupported Arrow type, so it cannot support the documented behavior that DESC
FUNCTION shows unsupported columns while queries can project only supported
columns. This conversion also only receives field->type() and therefore misses
Field metadata used to identify Lance extension and Dictionary semantics.
Please use a Field-aware conversion and preserve unsupported fields with a
sentinel, or narrow the documentation explicitly, so supported columns remain
queryable.
--
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]
