github-actions[bot] commented on code in PR #69581: URL: https://github.com/apache/doris/pull/69581#discussion_r4389385861
########## be/src/format_v2/lance/lance_reader_helper.cpp: ########## @@ -0,0 +1,843 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "format_v2/lance/lance_reader_helper.h" + +#include <arrow/array.h> +#include <arrow/builder.h> +#include <arrow/c/bridge.h> +#include <arrow/compute/cast.h> +#include <arrow/extension_type.h> +#include <arrow/type.h> +#include <arrow/util/key_value_metadata.h> +#include <fmt/format.h> +#include <lance/lance.h> + +#include <bit> +#include <limits> +#include <optional> +#include <unordered_set> + +#include "common/config.h" +#include "common/logging.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_factory.hpp" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_nothing.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_struct.h" +#include "core/data_type_serde/arrow_validation.h" +#include "exec/common/endian.h" + +namespace doris::format::lance { + +RuntimeProfile::Counter* add_lance_counter(RuntimeProfile* profile, const char* name, + TUnit::type unit, const char* group) { + // These counters label sections; they do not represent time or a sum of their children. + ADD_CHILD_COUNTER_WITH_LEVEL(profile, group, TUnit::NONE, LANCE_READER_PROFILE, 1); + return ADD_CHILD_COUNTER_WITH_LEVEL(profile, name, unit, group, 1); +} + +namespace { + +constexpr std::string_view ARROW_EXTENSION_NAME = "ARROW:extension:name"; +constexpr std::string_view ARROW_JSON_EXTENSION = "arrow.json"; +constexpr std::string_view LANCE_JSON_EXTENSION = "lance.json"; +constexpr std::string_view LANCE_BFLOAT16_EXTENSION = "lance.bfloat16"; + +enum class LanceExtensionKind { + NONE, + JSON, + BFLOAT16, +}; + +int arrow_time_precision(arrow::TimeUnit::type unit) { + switch (unit) { + case arrow::TimeUnit::SECOND: + return 0; + case arrow::TimeUnit::MILLI: + return 3; + case arrow::TimeUnit::MICRO: + case arrow::TimeUnit::NANO: + return 6; + } + return 6; +} + +// Extract and validate extension names from metadata and registered Arrow types. +Status get_lance_extension(const std::shared_ptr<arrow::Field>& field, + LanceExtensionKind* extension_kind, + std::shared_ptr<arrow::DataType>* storage_type) { + DORIS_CHECK(field != nullptr); + DORIS_CHECK(extension_kind != nullptr); + DORIS_CHECK(storage_type != nullptr); + *extension_kind = LanceExtensionKind::NONE; + *storage_type = field->type(); + + std::optional<std::string> extension_name; + if (field->HasMetadata()) { + const auto metadata_name = field->metadata()->Get(ARROW_EXTENSION_NAME); + if (metadata_name.ok() && !metadata_name.ValueUnsafe().empty()) { + extension_name = metadata_name.ValueUnsafe(); + } + } + if (field->type()->id() == arrow::Type::EXTENSION) { + const auto extension_type = std::dynamic_pointer_cast<arrow::ExtensionType>(field->type()); + if (extension_type == nullptr) { + return Status::InvalidArgument("invalid Arrow extension type for Lance field '{}'", + field->name()); + } + if (extension_name.has_value() && *extension_name != extension_type->extension_name()) { + return Status::InvalidArgument( + "conflicting Arrow extension names for Lance field '{}': '{}' and '{}'", + field->name(), *extension_name, extension_type->extension_name()); + } + extension_name = extension_type->extension_name(); + *storage_type = extension_type->storage_type(); + } + if (!extension_name.has_value()) { + return Status::OK(); + } + + if (*extension_name == ARROW_JSON_EXTENSION) { + if ((*storage_type)->id() != arrow::Type::STRING && + (*storage_type)->id() != arrow::Type::LARGE_STRING) { + return Status::NotSupported( + "Arrow JSON extension for Lance field '{}' requires UTF8 storage, got {}", + field->name(), (*storage_type)->ToString()); + } + *extension_kind = LanceExtensionKind::JSON; + return Status::OK(); + } + if (*extension_name == LANCE_JSON_EXTENSION) { + if ((*storage_type)->id() != arrow::Type::LARGE_BINARY) { + return Status::NotSupported( + "Lance JSON extension for field '{}' requires LARGE_BINARY storage, got {}", + field->name(), (*storage_type)->ToString()); + } + *extension_kind = LanceExtensionKind::JSON; + return Status::OK(); + } + if (*extension_name == LANCE_BFLOAT16_EXTENSION) { + if ((*storage_type)->id() != arrow::Type::FIXED_SIZE_BINARY || + std::static_pointer_cast<arrow::FixedSizeBinaryType>(*storage_type)->byte_width() != + 2) { + return Status::NotSupported( + "Lance BFloat16 extension for field '{}' requires FIXED_SIZE_BINARY(2) " + "storage, got {}", + field->name(), (*storage_type)->ToString()); + } + *extension_kind = LanceExtensionKind::BFLOAT16; + return Status::OK(); + } + return Status::NotSupported("unsupported Lance Arrow extension type '{}' for field '{}'", + *extension_name, field->name()); +} + +// Map Arrow fields recursively, preserving Null leaves in complex types. +Status arrow_field_to_doris_type(const std::shared_ptr<arrow::Field>& field, + DataTypePtr* doris_type) { + const auto nullable_primitive = [&](PrimitiveType type, int precision = 0, int scale = 0, + int len = -1) { + *doris_type = + DataTypeFactory::instance().create_data_type(type, true, precision, scale, len); + return Status::OK(); + }; + + LanceExtensionKind extension_kind; + std::shared_ptr<arrow::DataType> arrow_type; + RETURN_IF_ERROR(get_lance_extension(field, &extension_kind, &arrow_type)); + switch (extension_kind) { + case LanceExtensionKind::JSON: + return nullable_primitive(TYPE_JSONB); + case LanceExtensionKind::BFLOAT16: + return nullable_primitive(TYPE_FLOAT); + case LanceExtensionKind::NONE: + break; + } + if (arrow_type->id() == arrow::Type::DICTIONARY) { + return Status::NotSupported("unsupported Lance Arrow dictionary type for field '{}': {}", + field->name(), arrow_type->ToString()); + } + + switch (arrow_type->id()) { + case arrow::Type::NA: + // Required Null leaves can bypass NullableSerDe through FE schema reconstruction. + // Reject them before Arrow NA buffers reach the Boolean-backed physical SerDe. + if (!field->nullable()) { + return Status::NotSupported("non-nullable Lance Arrow Null field: {}", field->name()); + } + return nullable_primitive(TYPE_NULL); + case arrow::Type::BOOL: + return nullable_primitive(TYPE_BOOLEAN); + case arrow::Type::INT8: + return nullable_primitive(TYPE_TINYINT); + case arrow::Type::UINT8: + case arrow::Type::INT16: + return nullable_primitive(TYPE_SMALLINT); + case arrow::Type::UINT16: + case arrow::Type::INT32: + return nullable_primitive(TYPE_INT); + case arrow::Type::UINT32: + case arrow::Type::INT64: + return nullable_primitive(TYPE_BIGINT); + case arrow::Type::UINT64: + return nullable_primitive(TYPE_LARGEINT); + case arrow::Type::HALF_FLOAT: + case arrow::Type::FLOAT: + return nullable_primitive(TYPE_FLOAT); + case arrow::Type::DOUBLE: + return nullable_primitive(TYPE_DOUBLE); + case arrow::Type::STRING: + case arrow::Type::LARGE_STRING: + return nullable_primitive(TYPE_STRING); + case arrow::Type::BINARY: + case arrow::Type::LARGE_BINARY: + return nullable_primitive(TYPE_VARBINARY, 0, 0, std::numeric_limits<int32_t>::max()); + case arrow::Type::FIXED_SIZE_BINARY: { + const auto binary = std::static_pointer_cast<arrow::FixedSizeBinaryType>(arrow_type); + return nullable_primitive(TYPE_VARBINARY, 0, 0, binary->byte_width()); + } + case arrow::Type::DATE32: + case arrow::Type::DATE64: + return nullable_primitive(TYPE_DATEV2); + case arrow::Type::TIME32: + case arrow::Type::TIME64: { + const auto time = std::static_pointer_cast<arrow::TimeType>(arrow_type); + return nullable_primitive(TYPE_TIMEV2, 0, arrow_time_precision(time->unit())); + } + case arrow::Type::TIMESTAMP: { + const auto timestamp = std::static_pointer_cast<arrow::TimestampType>(arrow_type); + const auto timestamp_type = + timestamp->timezone().empty() ? TYPE_DATETIMEV2 : TYPE_TIMESTAMPTZ; + return nullable_primitive(timestamp_type, 0, arrow_time_precision(timestamp->unit())); + } + case arrow::Type::DURATION: + return nullable_primitive(TYPE_BIGINT); + case arrow::Type::DECIMAL128: + case arrow::Type::DECIMAL256: { + const auto decimal = std::static_pointer_cast<arrow::DecimalType>(arrow_type); + const int precision = decimal->precision(); + const int scale = decimal->scale(); + if (precision <= 0 || precision > arrow::Decimal256Type::kMaxPrecision || scale < 0 || + scale > precision) { + return Status::NotSupported( + "unsupported Lance Arrow decimal type for field '{}': precision={}, scale={}", + field->name(), precision, scale); + } + const PrimitiveType doris_decimal_type = precision <= 9 ? TYPE_DECIMAL32 + : precision <= 18 ? TYPE_DECIMAL64 + : precision <= 38 ? TYPE_DECIMAL128I + : TYPE_DECIMAL256; + return nullable_primitive(doris_decimal_type, precision, 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_field_to_doris_type(list->value_field(), &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_field_to_doris_type(map->key_field(), &key_type)); + RETURN_IF_ERROR(arrow_field_to_doris_type(map->item_field(), &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& child : struct_type->fields()) { + DataTypePtr field_type; + RETURN_IF_ERROR(arrow_field_to_doris_type(child, &field_type)); + field_types.emplace_back(std::move(field_type)); + field_names.emplace_back(child->name()); + } + *doris_type = make_nullable(std::make_shared<DataTypeStruct>(field_types, field_names)); Review Comment: [P2] Reject zero-field structs during Lance schema mapping A valid Lance `STRUCT<>` reaches this branch with no fields and is returned as a supported nullable `DataTypeStruct`; the [pinned native decoder explicitly supports this layout](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance-encoding/src/encodings/logical/struct.rs#L870-L887). Doris cannot materialize it: `DataTypeNullable::create_column()` constructs an empty `ColumnStruct`, and `ColumnNullable` immediately checks its size via `ColumnStruct::size()->columns.at(0)`, throwing `std::out_of_range`. The schema RPC likewise exposes an empty FE `StructType`, whose `toThrift()` rejects an empty field list. Nested empty structs have the same problem. Return `NotSupported` here so these fields use the existing unsupported-column path, or implement empty-struct storage end to end; add top-level and nested schema cases. -- 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]
