lxy-9602 commented on code in PR #198:
URL: https://github.com/apache/paimon-cpp/pull/198#discussion_r3795124961
##########
src/paimon/format/parquet/parquet_format_writer.h:
##########
@@ -72,7 +72,8 @@ class ParquetFormatWriter : public FormatWriter {
ParquetFormatWriter(std::unique_ptr<::parquet::arrow::FileWriter> writer,
const std::shared_ptr<ArrowOutputStreamAdapter>& out,
const std::shared_ptr<arrow::Schema>& schema, uint64_t
max_memory_use,
- const std::shared_ptr<arrow::MemoryPool>& pool);
+ const std::shared_ptr<arrow::MemoryPool>& pool,
+ bool needs_vector_conversion);
Review Comment:
Could you please move the newly added parameter to come before `pool`?
##########
src/paimon/core/io/vector_file_batch_reader.cpp:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/array.h"
+#include "arrow/array/array_nested.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::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;
+}
+
+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;
Review Comment:
`HasSameNestedProjectionShape` in `parquet_file_batch_reader.cpp` does not
seem to handle this case yet. Would this cause issues when reading data like
`List<FixedSizeList<Float32, 3>>`?
Also, there seem to be several places in the Parquet code path that do not
yet include branches for `fixed size list`.
##########
src/paimon/format/parquet/parquet_vector_converter.cpp:
##########
@@ -0,0 +1,126 @@
+/*
+ * 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 <memory>
+
+#include "arrow/array.h"
+#include "arrow/array/array_nested.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 std::shared_ptr<arrow::Array>& array) {
Review Comment:
`ValidateVectorElements` and `ContainsVectorType` could go in either
`arrow_utils` or a separate `vector_utils`, but right now there is too much
duplicated implementation.
##########
test/test_data/parquet/vector_compatibility/README.md:
##########
@@ -0,0 +1,41 @@
+<!--
+ ~ 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.
+-->
+
Review Comment:
The license for the test data can be omitted.
##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -648,4 +673,26 @@ Status SchemaValidation::ValidateMapStorageLayout(const
TableSchema& schema,
return Status::OK();
}
+Status SchemaValidation::ValidateVectorFields(const TableSchema& schema,
+ const CoreOptions& options) {
+ bool has_vector = false;
+ for (const auto& field : schema.Fields()) {
+ if (ContainsVectorField(field.ArrowField())) {
Review Comment:
`ContainsVectorField` appears in many places. Please consolidate it into a
single shared location and keep only one implementation.
##########
src/paimon/core/operation/abstract_split_read.cpp:
##########
@@ -208,6 +209,9 @@ Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::CreateFieldMappingRe
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileBatchReader> file_reader,
CreateFileBatchReader(file_format_identifier,
data_file_path,
file_meta->file_size,
reader_builder));
+ if (VectorFileBatchReader::ContainsVector(read_schema)) {
+ file_reader =
std::make_unique<VectorFileBatchReader>(std::move(file_reader), pool_);
Review Comment:
Does the nested-type cast exemption list in field_mapping.cpp (lines
186–192) need to include FIXED_SIZE_LIST as well?
##########
src/paimon/core/io/vector_file_batch_reader.cpp:
##########
@@ -0,0 +1,315 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/array.h"
+#include "arrow/array/array_nested.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::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;
+}
+
+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 std::make_shared<arrow::MapType>(
+ 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())),
+ map_type.keys_sorted());
+ }
+ default:
+ return logical_type;
+ }
+}
+
+Status ValidateVectorElements(const arrow::ListArray& array) {
+ for (int64_t i = 0; i < array.length(); ++i) {
+ if (array.IsNull(i)) {
+ continue;
+ }
+ int64_t value_offset = array.value_offset(i);
+ int64_t value_length = array.value_length(i);
+ for (int64_t j = 0; j < value_length; ++j) {
+ if (array.values()->IsNull(value_offset + j)) {
+ return Status::Invalid(fmt::format(
+ "VECTOR cannot contain null elements, found one at row {}
position {}", i, j));
+ }
+ }
+ }
+ return Status::OK();
+}
+
+Status ValidateVectorElements(const arrow::FixedSizeListArray& array) {
+ const auto& vector_type = checked_cast<const
arrow::FixedSizeListType&>(*array.type());
+ for (int64_t i = 0; i < array.length(); ++i) {
+ if (array.IsNull(i)) {
+ continue;
+ }
+ int64_t value_offset = (array.offset() + i) * vector_type.list_size();
+ for (int32_t j = 0; j < vector_type.list_size(); ++j) {
+ if (array.values()->IsNull(value_offset + j)) {
+ return Status::Invalid(fmt::format(
+ "VECTOR cannot contain null elements, found one at row {}
position {}", i, j));
+ }
+ }
+ }
+ return Status::OK();
+}
+
+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(ValidateVectorElements(checked_cast<const
arrow::ListArray&>(*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 std::make_shared<arrow::MapType>(entries_type.field(0),
entries_type.field(1),
+ map_type.keys_sorted());
Review Comment:
Why do both the list and struct branches preserve the original field, while
only the MAP branch does not? This drops the original entries field name and
metadata.
##########
src/paimon/common/utils/arrow/arrow_utils.cpp:
##########
Review Comment:
It looks like there is no zero-copy branch for `FIXED_SIZE_LIST` here.
Should we add one? Also, please add an end-to-end test with predicate pushdown
to verify that `FIXED_SIZE_LIST` still works correctly when predicates are
pushed down on other fields.
--
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]
