github-actions[bot] commented on code in PR #69581: URL: https://github.com/apache/doris/pull/69581#discussion_r4373822200
########## be/src/format_v2/lance/lance_session_manager.cpp: ########## @@ -0,0 +1,239 @@ +// 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_session_manager.h" + +#include <lance/lance.h> + +#include <algorithm> +#include <limits> +#include <string> +#include <string_view> +#include <utility> + +#include "common/config.h" +#include "common/logging.h" +#include "common/metrics/doris_metrics.h" +#include "common/metrics/metrics.h" +#include "format_v2/lance/lance_reader_helper.h" + +namespace doris::format::lance { +namespace { + +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_hits_total, MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_index_cache_misses_total, + MetricUnit::OPERATIONS); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_capacity_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_usage_bytes, MetricUnit::BYTES); +DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_entries, MetricUnit::NOUNIT); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_hits_total, + MetricUnit::OPERATIONS); +DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(lance_session_metadata_cache_misses_total, + MetricUnit::OPERATIONS); + +constexpr std::string_view LANCE_SESSION_CACHE_METRICS_HOOK = "lance_session_cache"; + +int64_t metric_value(uint64_t value) { + return static_cast<int64_t>( + std::min(value, static_cast<uint64_t>(std::numeric_limits<int64_t>::max()))); +} + +LanceSessionManager::Config load_lance_session_config() { + return { + .lance_index_cache_size_bytes = config::lance_index_cache_size_bytes, + .lance_metadata_cache_size_bytes = config::lance_metadata_cache_size_bytes, + .enable_lance_data_cache = config::enable_lance_data_cache, + .lance_data_cache_path = config::lance_data_cache_path, + .lance_data_cache_disk_capacity_bytes = config::lance_data_cache_disk_capacity_bytes, + .lance_data_cache_read_block_size_bytes = + config::lance_data_cache_read_block_size_bytes, + }; +} + +} // namespace + +class LanceSessionMetrics final { +public: + LanceSessionMetrics(LanceSession* session, const LanceSessionManager::Config& config) + : _session(session), _entity(DorisMetrics::instance()->server_entity()) { + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_index_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_index_cache_misses_total); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_usage_bytes); + INT_GAUGE_METRIC_REGISTER(_entity, lance_session_metadata_cache_entries); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_hits_total); + INT_COUNTER_METRIC_REGISTER(_entity, lance_session_metadata_cache_misses_total); + + lance_session_index_cache_capacity_bytes->set_value(config.lance_index_cache_size_bytes); + lance_session_metadata_cache_capacity_bytes->set_value( + config.lance_metadata_cache_size_bytes); + _entity->register_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK), + [this]() { update(); }); + update(); + } + + ~LanceSessionMetrics() { + _entity->deregister_hook(std::string(LANCE_SESSION_CACHE_METRICS_HOOK)); + METRIC_DEREGISTER(_entity, lance_session_index_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_index_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_index_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_index_cache_misses_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_capacity_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_usage_bytes); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_entries); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_hits_total); + METRIC_DEREGISTER(_entity, lance_session_metadata_cache_misses_total); + } + +private: + void update() { + // Session caches are shared across queries, so publish one process-wide snapshot instead + // of attributing concurrent cache activity to an individual query profile. + LanceSessionCacheStats stats {}; + if (lance_session_get_cache_stats(_session, &stats) != 0) { + LOG_EVERY_N(WARNING, 100) + << lance_error("collect Lance session cache statistics").to_string(); + return; + } + lance_session_index_cache_usage_bytes->set_value( + metric_value(stats.index_cache_size_bytes)); + lance_session_index_cache_entries->set_value(metric_value(stats.index_cache_entries)); + lance_session_index_cache_hits_total->set_value(metric_value(stats.index_cache_hits)); + lance_session_index_cache_misses_total->set_value(metric_value(stats.index_cache_misses)); + lance_session_metadata_cache_usage_bytes->set_value( + metric_value(stats.metadata_cache_size_bytes)); + lance_session_metadata_cache_entries->set_value(metric_value(stats.metadata_cache_entries)); + lance_session_metadata_cache_hits_total->set_value(metric_value(stats.metadata_cache_hits)); + lance_session_metadata_cache_misses_total->set_value( + metric_value(stats.metadata_cache_misses)); + } + + LanceSession* _session; + MetricEntity* _entity; + IntGauge* lance_session_index_cache_capacity_bytes = nullptr; + IntGauge* lance_session_index_cache_usage_bytes = nullptr; + IntGauge* lance_session_index_cache_entries = nullptr; + IntCounter* lance_session_index_cache_hits_total = nullptr; + IntCounter* lance_session_index_cache_misses_total = nullptr; + IntGauge* lance_session_metadata_cache_capacity_bytes = nullptr; + IntGauge* lance_session_metadata_cache_usage_bytes = nullptr; + IntGauge* lance_session_metadata_cache_entries = nullptr; + IntCounter* lance_session_metadata_cache_hits_total = nullptr; + IntCounter* lance_session_metadata_cache_misses_total = nullptr; +}; + +LanceSessionManager& LanceSessionManager::instance() { + // Function-local static initialization is thread safe. Cache configuration is process scoped, + // so changing it requires a BE restart. + static LanceSessionManager manager(load_lance_session_config()); + return manager; +} + +LanceSessionManager::LanceSessionManager(Config config) : _config(std::move(config)) { + LOG(INFO) << "Creating BE-wide Lance session manager: lance_index_cache_size_bytes=" + << _config.lance_index_cache_size_bytes + << ", lance_metadata_cache_size_bytes=" << _config.lance_metadata_cache_size_bytes + << ", enable_lance_data_cache=" << _config.enable_lance_data_cache + << ", lance_data_cache_path=" << _config.lance_data_cache_path + << ", lance_data_cache_disk_capacity_bytes=" + << _config.lance_data_cache_disk_capacity_bytes + << ", lance_data_cache_read_block_size_bytes=" + << _config.lance_data_cache_read_block_size_bytes + << ", foyer_memory_capacity_bytes=" << _config.lance_data_cache_read_block_size_bytes; +} + +LanceSessionManager::~LanceSessionManager() { + _metrics.reset(); + lance_session_close(_session); +} + +Status LanceSessionManager::_initialize() { + if (_config.enable_lance_data_cache) { + // Treat the configured cache mode as a process-level requirement. If an enabled + // data cache cannot initialize, report the failure instead of silently creating a + // session without it. This keeps directory/configuration/device failures visible + // to the operator. Disabling the cache is an explicit configuration change. + const LanceDataCacheOptions data_cache_options { + .directory = _config.lance_data_cache_path.c_str(), + // Foyer's HybridCache requires a memory tier. Keep it at the minimum useful + // capacity of exactly one range-cache block; WriteOnInsertion enqueues disk + // writes on insertion rather than waiting for memory-tier eviction. + .memory_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + .disk_capacity_bytes = + static_cast<uint64_t>(_config.lance_data_cache_disk_capacity_bytes), + .read_block_size_bytes = + static_cast<uint64_t>(_config.lance_data_cache_read_block_size_bytes), + }; + _session = lance_session_new_with_data_cache( Review Comment: [P1] Include the storage endpoint in the shared data-cache identity This creates one Foyer cache for every dataset, but its key in `lance-c-foyer.patch` is block size + store prefix + object path + block number. The [pinned S3 prefix calculation](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance-io/src/object_store/providers.rs#L62-L80) ignores endpoint options and yields `s3$bucket`. With the default cache enabled, two independent endpoints serving `s3://bucket/dataset.lance/data/fragment-0.lance` can therefore share size/data entries: after scanning A, scanning B can return A's rows. Both files can be valid, immutable and independently registered under that name; this requires neither UUID collision nor in-place overwrite. The origin-store registry distinguishes the options, but Foyer discards that distinction. Include effective physical-store identity in the cache namespace, or isolate caches for different stores, and cover equal paths with different endpoint contents. This is distinct from the exist ing memory-reclamation thread. ########## be/src/format_v2/lance/lance_reader_helper.cpp: ########## @@ -0,0 +1,802 @@ +// 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)); + return Status::OK(); + } + default: + return Status::NotSupported("unsupported Lance Arrow type: {}", arrow_type->ToString()); + } +} + +// Widen little-endian Lance BFloat16 values to Arrow Float32 without precision loss. +Status convert_bfloat16_array(const std::shared_ptr<arrow::Array>& array, + arrow::MemoryPool* memory_pool, + std::shared_ptr<arrow::Array>* normalized) { + DORIS_CHECK(array != nullptr); + DORIS_CHECK(memory_pool != nullptr); + DORIS_CHECK(normalized != nullptr); + const auto fixed_binary = std::dynamic_pointer_cast<arrow::FixedSizeBinaryArray>(array); + if (fixed_binary == nullptr || fixed_binary->byte_width() != 2) { + return Status::InvalidArgument("invalid Lance BFloat16 array storage: {}", + array->type()->ToString()); + } + if (config::enable_arrow_input_validation) { + check_arrow_fixed_width_buffer(*fixed_binary, sizeof(uint16_t)); + } + + arrow::FloatBuilder builder(memory_pool); + auto arrow_status = builder.Reserve(fixed_binary->length()); + if (!arrow_status.ok()) { + return Status::InternalError("reserve Lance BFloat16 output failed: {}", + arrow_status.message()); + } + for (int64_t row = 0; row < fixed_binary->length(); ++row) { + if (fixed_binary->IsNull(row)) { + arrow_status = builder.AppendNull(); + } else { + const auto bits = LittleEndian::Load16(fixed_binary->GetValue(row)); + arrow_status = builder.Append(std::bit_cast<float>(static_cast<uint32_t>(bits) << 16)); + } + if (!arrow_status.ok()) { + return Status::InternalError("append Lance BFloat16 value failed: {}", + arrow_status.message()); + } + } + std::shared_ptr<arrow::FloatArray> result; + arrow_status = builder.Finish(&result); + if (!arrow_status.ok()) { + return Status::InternalError("finish Lance BFloat16 conversion failed: {}", + arrow_status.message()); + } + *normalized = std::move(result); + return Status::OK(); +} + +// Narrow a Decimal256 array whose declared precision fits Decimal128. The safe cast keeps every +// in-precision value and its scale, and rejects a value that exceeds the declared precision +// instead of truncating it: unscaled 10^18 stored in a decimal256(18, 2) field fails with +// "Decimal value does not fit in precision 18". +Status narrow_decimal256_array(std::string_view field_name, + const std::shared_ptr<arrow::Array>& array, + const std::shared_ptr<arrow::DataType>& decimal128_type, + arrow::MemoryPool* memory_pool, + std::shared_ptr<arrow::Array>* normalized) { + DORIS_CHECK(array != nullptr); + DORIS_CHECK(memory_pool != nullptr); + DORIS_CHECK(normalized != nullptr); + arrow::compute::ExecContext exec_context(memory_pool); + auto narrowed = arrow::compute::Cast(*array, decimal128_type, + arrow::compute::CastOptions::Safe(), &exec_context); + if (!narrowed.ok()) { + return Status::InvalidArgument("narrow Lance Decimal256 field '{}' to {} failed: {}", + field_name, decimal128_type->ToString(), + narrowed.status().message()); + } + *normalized = std::move(narrowed).ValueUnsafe(); + return Status::OK(); +} + +// Rebuild a nested Arrow type after one or more child arrays changed physical type. +Status set_lance_nested_type(std::string_view field_name, + const std::shared_ptr<arrow::DataType>& source_type, + const arrow::FieldVector& child_fields, + std::shared_ptr<arrow::ArrayData>* data) { + switch (source_type->id()) { + case arrow::Type::LIST: + (*data)->type = arrow::list(child_fields[0]); + break; + case arrow::Type::LARGE_LIST: + (*data)->type = arrow::large_list(child_fields[0]); + break; + case arrow::Type::FIXED_SIZE_LIST: + (*data)->type = arrow::fixed_size_list( + child_fields[0], + std::static_pointer_cast<arrow::FixedSizeListType>(source_type)->list_size()); + break; + case arrow::Type::STRUCT: + (*data)->type = arrow::struct_(child_fields); + break; + case arrow::Type::MAP: { + const auto map_type = std::static_pointer_cast<arrow::MapType>(source_type); + auto normalized_type = arrow::MapType::Make(child_fields[0], map_type->keys_sorted()); + if (!normalized_type.ok()) { + return Status::InvalidArgument("normalize Lance map field '{}' failed: {}", field_name, + normalized_type.status().message()); + } + (*data)->type = std::move(normalized_type).ValueUnsafe(); + break; + } + default: + return Status::InvalidArgument("Lance field '{}' has unexpected child-bearing type {}", + field_name, source_type->ToString()); + } + return Status::OK(); +} + +// Materialize the visible range into an offset-zero Arrow array for Doris SerDes. +Status compact_lance_array(const std::shared_ptr<arrow::Array>& array, + arrow::MemoryPool* memory_pool, + std::shared_ptr<arrow::Array>* compacted) { + DORIS_CHECK(memory_pool != nullptr); + const auto validation = array->ValidateFull(); + if (!validation.ok()) { + return Status::InvalidArgument("validate sliced Lance array failed: {}", + validation.message()); + } + auto builder_result = arrow::MakeBuilder(array->type(), memory_pool); + if (!builder_result.ok()) { + return Status::InternalError("create sliced Lance array builder failed: {}", + builder_result.status().message()); + } + auto builder = std::move(builder_result).ValueUnsafe(); + auto arrow_status = builder->Reserve(array->length()); + if (!arrow_status.ok()) { + return Status::InternalError("reserve sliced Lance array builder failed: {}", + arrow_status.message()); + } + arrow_status = builder->AppendArraySlice(*array->data(), 0, array->length()); + if (!arrow_status.ok()) { + return Status::InternalError("copy sliced Lance array failed: {}", arrow_status.message()); + } + arrow_status = builder->Finish(compacted); + if (!arrow_status.ok()) { + return Status::InternalError("finish sliced Lance array copy failed: {}", + arrow_status.message()); + } + if ((*compacted)->offset() != 0) { + return Status::InternalError("compacted Lance array retained offset {}", + (*compacted)->offset()); + } + return Status::OK(); +} + +// Compact a sliced variable-offset parent and all of its visible descendants. +template <typename ArrayType> +Status compact_lance_offset_array(const std::shared_ptr<arrow::Array>& array, + arrow::MemoryPool* memory_pool, + std::shared_ptr<arrow::Array>* compacted) { + const auto offset_array = std::dynamic_pointer_cast<ArrayType>(array); + if (offset_array == nullptr) { + return Status::InvalidArgument("invalid sliced Lance offset array: {}", + array->type()->ToString()); + } + const auto child_begin = static_cast<int64_t>(offset_array->value_offset(0)); + const auto child_end = static_cast<int64_t>(offset_array->value_offset(offset_array->length())); + const auto& values = offset_array->values(); + if (child_begin < 0 || child_end < child_begin || child_end > values->length()) { + return Status::InvalidArgument("invalid sliced Lance offsets [{}, {}) for child length {}", + child_begin, child_end, values->length()); + } + if (array->offset() == 0 && child_begin == 0 && child_end == values->length()) { + *compacted = array; + return Status::OK(); + } + return compact_lance_array(array, memory_pool, compacted); +} + +// Compact sliced arrays and nested children only when Doris cannot consume their current layout. +Status compact_lance_array_if_needed(const std::shared_ptr<arrow::Array>& array, + arrow::MemoryPool* memory_pool, + std::shared_ptr<arrow::Array>* compacted) { + switch (array->type_id()) { + case arrow::Type::LIST: + return compact_lance_offset_array<arrow::ListArray>(array, memory_pool, compacted); + case arrow::Type::LARGE_LIST: + return compact_lance_offset_array<arrow::LargeListArray>(array, memory_pool, compacted); + case arrow::Type::MAP: + return compact_lance_offset_array<arrow::MapArray>(array, memory_pool, compacted); + case arrow::Type::FIXED_SIZE_LIST: { + const auto list = std::dynamic_pointer_cast<arrow::FixedSizeListArray>(array); + if (list == nullptr) { + return Status::InvalidArgument("invalid sliced Lance fixed-size list array: {}", + array->type()->ToString()); + } + const auto child_begin = list->value_offset(0); + const auto child_length = list->length() * list->value_length(); + const auto& values = list->values(); + if (child_begin < 0 || child_length < 0 || child_begin > values->length() - child_length) { + return Status::InvalidArgument( + "invalid sliced Lance fixed-size list range [{}, {}) for child length {}", + child_begin, child_begin + child_length, values->length()); + } + if (array->offset() == 0 && child_begin == 0 && child_length == values->length()) { + *compacted = array; + return Status::OK(); + } + return compact_lance_array(array, memory_pool, compacted); + } + case arrow::Type::STRUCT: { + const auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(array); + if (struct_array == nullptr) { + return Status::InvalidArgument("invalid sliced Lance struct array: {}", + array->type()->ToString()); + } + bool requires_compaction = array->offset() != 0; + for (const auto& child : array->data()->child_data) { + requires_compaction |= child->length != array->length(); Review Comment: [P2] Compact sliced children of zero-offset structs Equal child lengths do not imply zero child offsets. For a valid legacy/v1 Lance dataset with six rows of `s: STRUCT<b: BOOLEAN>` in one writer batch and Doris `batch_size=3`, the batch reading rows `[3,6)` has a zero-offset struct containing a length-3 Boolean child with bit offset 3: the [pinned plain decoder](https://github.com/lance-format/lance/blob/68c12dfd7efe02f90ad2d7f3a239a7eb64884e57/rust/lance-file/src/versions/v1/encoding/plain.rs#L240-L249) uses `start % 8`, and the struct reader wraps that child without compacting it. Ordinary types skip child normalization, so this branch returns the array unchanged. StructSerDe then passes the child to NullableSerDe, whose `check_arrow_array_range()` rejects the nonzero offset under default `enable_arrow_input_validation=true`; a valid scan fails. Normalize/check descendant layouts before choosing this shortcut and add a legacy struct/boolean scan across a non-byte-aligned batch boundary. -- 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]
