SteNicholas commented on code in PR #198: URL: https://github.com/apache/paimon-cpp/pull/198#discussion_r3803574743
########## src/paimon/core/io/vector_file_batch_reader.cpp: ########## @@ -0,0 +1,274 @@ +/* + * 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 "paimon/core/io/vector_file_batch_reader.h" + +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "arrow/array.h" +#include "arrow/c/abi.h" +#include "arrow/c/bridge.h" +#include "arrow/compute/api.h" +#include "arrow/type.h" +#include "fmt/format.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/arrow/vector_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/status.h" + +namespace paimon { +namespace { + +std::shared_ptr<arrow::Field> FindField(const std::shared_ptr<arrow::DataType>& type, + const std::string& name) { + for (const auto& field : type->fields()) { + if (field->name() == name) { + return field; + } + } + return nullptr; +} + +/// Rebuilds `map_type` with new key and item types, keeping the name and metadata of its +/// entries field. +std::shared_ptr<arrow::DataType> MakeMapType(const arrow::MapType& map_type, + const std::shared_ptr<arrow::Field>& key_field, + const std::shared_ptr<arrow::Field>& item_field) { + return std::make_shared<arrow::MapType>( + map_type.value_field()->WithType(arrow::struct_({key_field, item_field})), + map_type.keys_sorted()); +} + +/// Returns the type to request from the file format plugin. A VECTOR is only read back as a +/// LIST when the file itself stores it as one: writers such as Paimon Java expose VECTOR +/// columns as Arrow LIST, while Paimon Rust and Python expose them as FixedSizeList. +std::shared_ptr<arrow::DataType> GetPhysicalReadType( + const std::shared_ptr<arrow::DataType>& logical_type, + const std::shared_ptr<arrow::DataType>& file_type) { + switch (logical_type->id()) { + case arrow::Type::FIXED_SIZE_LIST: { + if (!file_type || file_type->id() != arrow::Type::LIST) { + return logical_type; + } + const auto& vector_type = checked_cast<const arrow::FixedSizeListType&>(*logical_type); + const auto& list_type = checked_cast<const arrow::ListType&>(*file_type); + return arrow::list(vector_type.value_field()->WithType( + GetPhysicalReadType(vector_type.value_type(), list_type.value_type()))); + } + case arrow::Type::STRUCT: { + if (!file_type || file_type->id() != arrow::Type::STRUCT) { + return logical_type; + } + arrow::FieldVector fields; + fields.reserve(logical_type->num_fields()); + for (const auto& field : logical_type->fields()) { + std::shared_ptr<arrow::Field> file_field = FindField(file_type, field->name()); + fields.push_back(field->WithType( + GetPhysicalReadType(field->type(), file_field ? file_field->type() : nullptr))); + } + return arrow::struct_(fields); + } + case arrow::Type::LIST: { + if (!file_type || file_type->id() != arrow::Type::LIST) { + return logical_type; + } + return arrow::list(logical_type->field(0)->WithType( + GetPhysicalReadType(logical_type->field(0)->type(), file_type->field(0)->type()))); + } + case arrow::Type::MAP: { + if (!file_type || file_type->id() != arrow::Type::MAP) { + return logical_type; + } + const auto& map_type = checked_cast<const arrow::MapType&>(*logical_type); + const auto& file_map_type = checked_cast<const arrow::MapType&>(*file_type); + return MakeMapType(map_type, + map_type.key_field()->WithType(GetPhysicalReadType( + map_type.key_type(), file_map_type.key_type())), + map_type.item_field()->WithType(GetPhysicalReadType( + map_type.item_type(), file_map_type.item_type()))); + } + default: + return logical_type; + } +} + +Result<std::shared_ptr<arrow::Array>> CastListToVector( + const std::shared_ptr<arrow::Array>& array, + const std::shared_ptr<arrow::FixedSizeListType>& read_type, arrow::MemoryPool* pool) { + if (array->type_id() != arrow::Type::LIST) { + return Status::Invalid( + fmt::format("Cannot restore VECTOR from type {}", array->type()->ToString())); + } + PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(*array)); + arrow::compute::ExecContext exec_context(pool); + arrow::TypeHolder type_holder(read_type.get()); + arrow::compute::CastOptions options = arrow::compute::CastOptions::Safe(); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::Array> result, + arrow::compute::Cast(*array, type_holder, options, &exec_context)); + return result; +} + +std::shared_ptr<arrow::DataType> RebuildNestedType( + const std::shared_ptr<arrow::DataType>& read_type, + const std::vector<std::shared_ptr<arrow::ArrayData>>& children) { + if (read_type->id() == arrow::Type::STRUCT) { + arrow::FieldVector fields; + fields.reserve(children.size()); + for (int32_t i = 0; i < static_cast<int32_t>(children.size()); ++i) { + fields.push_back(read_type->field(i)->WithType(children[i]->type)); + } + return arrow::struct_(fields); + } + if (read_type->id() == arrow::Type::LIST) { + return arrow::list(read_type->field(0)->WithType(children[0]->type)); + } + + const auto& entries_type = checked_cast<const arrow::StructType&>(*children[0]->type); + const auto& map_type = checked_cast<const arrow::MapType&>(*read_type); + return MakeMapType(map_type, map_type.key_field()->WithType(entries_type.field(0)->type()), + map_type.item_field()->WithType(entries_type.field(1)->type())); +} + +Result<std::shared_ptr<arrow::Array>> ConvertToReadType( + const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::DataType>& read_type, + arrow::MemoryPool* pool) { + if (!VectorUtils::ContainsVectorType(read_type)) { + return array; + } + switch (read_type->id()) { + case arrow::Type::FIXED_SIZE_LIST: { + if (array->type_id() == arrow::Type::FIXED_SIZE_LIST) { + const auto& source_type = + checked_cast<const arrow::FixedSizeListType&>(*array->type()); + const auto& vector_type = checked_cast<const arrow::FixedSizeListType&>(*read_type); + if (source_type.list_size() != vector_type.list_size() || + !source_type.value_type()->Equals(vector_type.value_type())) { + return Status::Invalid(fmt::format("VECTOR type mismatch: data {} vs read {}", + source_type.ToString(), + vector_type.ToString())); + } + PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(*array)); + return array; Review Comment: **[P1] Normalize the FixedSizeList output type** This returns the file FixedSizeList type unchanged after comparing only the list size and value data type. The physical child field name/nullability can therefore remain different from the requested logical schema (for example, the Rust fixture uses `element: float not null`, while the C++ schema uses `item: float`). Java/C++ `LIST` files are cast to the logical type, so reading mixed encodings yields unequal batch types and `ChunkedArray::Make` fails with “Array chunks must all be same type.” Please zero-copy rewrap a copy of `array->data()` with `read_type` after validation, and add a regression test that reads mixed `LIST`/`FixedSizeList` files using the actual table schema. -- 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]
