lxy-9602 commented on code in PR #198: URL: https://github.com/apache/paimon-cpp/pull/198#discussion_r3793373684
########## src/paimon/core/io/vector_file_batch_reader.cpp: ########## @@ -0,0 +1,325 @@ +/* + * 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 <cstdint> +#include <limits> +#include <memory> +#include <utility> +#include <vector> + +#include "arrow/array.h" +#include "arrow/array/array_nested.h" +#include "arrow/array/builder_primitive.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/checked_cast.h" +#include "paimon/status.h" + +namespace paimon { +namespace { + +bool ContainsVectorType(const std::shared_ptr<arrow::DataType>& type) { + if (type->id() == arrow::Type::FIXED_SIZE_LIST) { + return true; + } + for (const auto& field : type->fields()) { + if (ContainsVectorType(field->type())) { + return true; + } + } + return false; +} + +std::shared_ptr<arrow::DataType> GetPhysicalReadType( + const std::shared_ptr<arrow::DataType>& logical_type) { + switch (logical_type->id()) { + case arrow::Type::FIXED_SIZE_LIST: { + const auto& vector_type = checked_cast<const arrow::FixedSizeListType&>(*logical_type); + return arrow::list( + vector_type.value_field()->WithType(GetPhysicalReadType(vector_type.value_type()))); + } + case arrow::Type::STRUCT: { + arrow::FieldVector fields; Review Comment: One thing to note here: for data written by Java, vectors are indeed read back as `list`, while for data written by Python/Rust, vectors seem to already come back as `fixed size list`. So I’d suggest narrowing this conversion to only the case where the **file schema is `list`** but the **read schema is `fixed size list`**. Also, could you please add test data generated from both Rust and Java for vector columns, and verify that C++ can read them correctly? ########## src/paimon/core/io/vector_file_batch_reader.cpp: ########## @@ -0,0 +1,325 @@ +/* + * 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 <cstdint> +#include <limits> +#include <memory> +#include <utility> +#include <vector> + +#include "arrow/array.h" +#include "arrow/array/array_nested.h" +#include "arrow/array/builder_primitive.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/checked_cast.h" +#include "paimon/status.h" + +namespace paimon { +namespace { + +bool ContainsVectorType(const std::shared_ptr<arrow::DataType>& type) { + if (type->id() == arrow::Type::FIXED_SIZE_LIST) { + return true; + } + for (const auto& field : type->fields()) { + if (ContainsVectorType(field->type())) { + return true; + } + } + return false; +} + +std::shared_ptr<arrow::DataType> GetPhysicalReadType( + const std::shared_ptr<arrow::DataType>& logical_type) { + switch (logical_type->id()) { + case arrow::Type::FIXED_SIZE_LIST: { + const auto& vector_type = checked_cast<const arrow::FixedSizeListType&>(*logical_type); + return arrow::list( + vector_type.value_field()->WithType(GetPhysicalReadType(vector_type.value_type()))); + } + case arrow::Type::STRUCT: { + arrow::FieldVector fields; + fields.reserve(logical_type->num_fields()); + for (const auto& field : logical_type->fields()) { + fields.push_back(field->WithType(GetPhysicalReadType(field->type()))); + } + return arrow::struct_(fields); + } + case arrow::Type::LIST: + return arrow::list(logical_type->field(0)->WithType( + GetPhysicalReadType(logical_type->field(0)->type()))); + case arrow::Type::MAP: { + const auto& map_type = checked_cast<const arrow::MapType&>(*logical_type); + return std::make_shared<arrow::MapType>( + map_type.key_field()->WithType(GetPhysicalReadType(map_type.key_type())), + map_type.item_field()->WithType(GetPhysicalReadType(map_type.item_type())), + map_type.keys_sorted()); + } + default: + return logical_type; + } +} + +Status ValidateVectorElements(const arrow::FixedSizeListArray& array, int32_t vector_length) { + const std::shared_ptr<arrow::Array>& values = array.values(); + if (values->null_count() == 0) { + return Status::OK(); + } + for (int64_t i = 0; i < array.length(); ++i) { + if (array.IsNull(i)) { + continue; + } + int64_t value_offset = (array.offset() + i) * vector_length; + for (int32_t j = 0; j < vector_length; ++j) { + if (values->IsNull(value_offset + j)) { + return Status::Invalid("VECTOR cannot contain null elements"); + } + } + } + return Status::OK(); +} + +Result<int64_t> GetIndexCapacity(int64_t row_count, int32_t vector_length) { + if (vector_length < 1) { + return Status::Invalid("VECTOR length must be positive"); + } + if (row_count > std::numeric_limits<int64_t>::max() / vector_length) { + return Status::Invalid("VECTOR values exceed the supported Arrow array length"); + } + return row_count * vector_length; +} + +Result<std::shared_ptr<arrow::Array>> ConvertListToVector( + const std::shared_ptr<arrow::Array>& array, + const std::shared_ptr<arrow::FixedSizeListType>& read_type, arrow::MemoryPool* pool) { + int32_t vector_length = read_type->list_size(); + if (array->type()->id() == arrow::Type::FIXED_SIZE_LIST) { + const auto& source_type = checked_cast<const arrow::FixedSizeListType&>(*array->type()); + if (source_type.list_size() != vector_length || + !source_type.value_type()->Equals(read_type->value_type())) { + return Status::Invalid(fmt::format("VECTOR type mismatch: data {} vs read {}", + array->type()->ToString(), read_type->ToString())); + } + const auto& vector_array = checked_cast<const arrow::FixedSizeListArray&>(*array); + PAIMON_RETURN_NOT_OK(ValidateVectorElements(vector_array, vector_length)); + std::shared_ptr<arrow::ArrayData> data = array->data()->Copy(); + data->type = read_type; + return arrow::MakeArray(data); + } + if (array->type()->id() != arrow::Type::LIST) { + return Status::Invalid( + fmt::format("Cannot restore VECTOR from type {}", array->type()->ToString())); + } + + const auto& list_array = checked_cast<const arrow::ListArray&>(*array); + if (!list_array.value_type()->Equals(read_type->value_type())) { + return Status::Invalid(fmt::format("VECTOR element type mismatch: data {} vs read {}", + list_array.value_type()->ToString(), + read_type->value_type()->ToString())); + } + + arrow::Int64Builder indices_builder(pool); + PAIMON_ASSIGN_OR_RAISE(int64_t index_capacity, + GetIndexCapacity(list_array.length(), vector_length)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(indices_builder.Reserve(index_capacity)); + arrow::BooleanBuilder validity_builder(pool); + PAIMON_RETURN_NOT_OK_FROM_ARROW(validity_builder.Reserve(list_array.length())); + Review Comment: I’m wondering whether we really need to rebuild the array with a builder here to make a deep copy. It seems Arrow provides `arrow::compute::Cast(array, read_type, ...)`, which may support a more lightweight conversion and avoid deep copying where possible. ########## src/paimon/format/parquet/parquet_vector_converter.cpp: ########## @@ -0,0 +1,198 @@ +/* + * 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/format/parquet/parquet_vector_converter.h" + +#include <cstdint> +#include <limits> +#include <memory> +#include <vector> + +#include "arrow/array.h" +#include "arrow/array/array_nested.h" +#include "arrow/array/builder_primitive.h" +#include "arrow/compute/api.h" +#include "arrow/type.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/status.h" + +namespace paimon::parquet { +namespace { + +bool ContainsVectorType(const std::shared_ptr<arrow::DataType>& type) { + if (type->id() == arrow::Type::FIXED_SIZE_LIST) { + return true; + } + for (const auto& field : type->fields()) { + if (ContainsVectorType(field->type())) { + return true; + } + } + return false; +} + +Status ValidateVectorElements(const arrow::FixedSizeListArray& array, int32_t vector_length) { + const std::shared_ptr<arrow::Array>& values = array.values(); + if (values->null_count() == 0) { + return Status::OK(); + } + for (int64_t i = 0; i < array.length(); ++i) { + if (array.IsNull(i)) { + continue; + } + int64_t value_offset = (array.offset() + i) * vector_length; + for (int32_t j = 0; j < vector_length; ++j) { + if (values->IsNull(value_offset + j)) { + return Status::Invalid("VECTOR cannot contain null elements"); + } + } + } + return Status::OK(); +} + +Result<int64_t> GetIndexCapacity(int64_t row_count, int32_t vector_length) { + if (vector_length < 1) { + return Status::Invalid("VECTOR length must be positive"); + } + if (row_count > std::numeric_limits<int64_t>::max() / vector_length) { + return Status::Invalid("VECTOR values exceed the supported Arrow array length"); + } + return row_count * vector_length; +} + +Result<std::shared_ptr<arrow::Array>> ConvertVectorToList( + const std::shared_ptr<arrow::Array>& array, arrow::MemoryPool* pool) { + const auto& vector_array = checked_cast<const arrow::FixedSizeListArray&>(*array); + const auto& vector_type = checked_cast<const arrow::FixedSizeListType&>(*array->type()); + int32_t vector_length = vector_type.list_size(); + PAIMON_RETURN_NOT_OK(ValidateVectorElements(vector_array, vector_length)); + + arrow::Int32Builder offsets_builder(pool); + arrow::Int64Builder indices_builder(pool); + arrow::BooleanBuilder validity_builder(pool); + PAIMON_RETURN_NOT_OK_FROM_ARROW(offsets_builder.Reserve(vector_array.length() + 1)); + PAIMON_ASSIGN_OR_RAISE(int64_t index_capacity, + GetIndexCapacity(vector_array.length(), vector_length)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(indices_builder.Reserve(index_capacity)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(validity_builder.Reserve(vector_array.length())); + PAIMON_RETURN_NOT_OK_FROM_ARROW(offsets_builder.Append(0)); + + int32_t offset = 0; + for (int64_t i = 0; i < vector_array.length(); ++i) { Review Comment: Similar to the read path, it seems the write path could also use `arrow::cast`. We could validate nulls before casting. -- 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]
