This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new c662861 feat(common): introduce arrow utilities (#110)
c662861 is described below
commit c66286180e2d52e9b70b01d88e660db0d872c64a
Author: Zhang Jiawei <[email protected]>
AuthorDate: Thu Jun 25 10:27:33 2026 +0800
feat(common): introduce arrow utilities (#110)
---
.../utils/arrow/arrow_input_stream_adapter.cpp | 166 +++++++
.../utils/arrow/arrow_input_stream_adapter.h | 64 +++
.../utils/arrow/arrow_output_stream_adapter.cpp | 70 +++
.../utils/arrow/arrow_output_stream_adapter.h | 48 ++
.../utils/arrow/arrow_stream_adapter_test.cpp | 94 ++++
src/paimon/common/utils/arrow/arrow_utils.cpp | 179 ++++++++
src/paimon/common/utils/arrow/arrow_utils.h | 65 +++
src/paimon/common/utils/arrow/arrow_utils_test.cpp | 490 +++++++++++++++++++++
src/paimon/common/utils/arrow/mem_utils.cpp | 87 ++++
src/paimon/common/utils/arrow/mem_utils.h | 36 ++
src/paimon/common/utils/arrow/mem_utils_test.cpp | 77 ++++
.../common/utils/arrow/status_utils_test.cpp | 60 +++
12 files changed, 1436 insertions(+)
diff --git a/src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp
b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp
new file mode 100644
index 0000000..941c18d
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp
@@ -0,0 +1,166 @@
+/*
+ * 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/common/utils/arrow/arrow_input_stream_adapter.h"
+
+#include <cstdint>
+#include <utility>
+
+#include "arrow/api.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/common/utils/options_utils.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/macros.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+namespace {
+
+template <typename To, typename From>
+arrow::Status ValidateArrowIoRange(From value, const char* name) {
+ if (!InRange<To>(value)) {
+ return arrow::Status::Invalid(fmt::format("{} value {} is out of bound
of type {}", name,
+ value,
OptionsUtils::GetTypeName<To>()));
+ }
+ return arrow::Status::OK();
+}
+
+} // namespace
+
+ArrowInputStreamAdapter::ArrowInputStreamAdapter(
+ const std::shared_ptr<paimon::InputStream>& input_stream,
+ const std::shared_ptr<arrow::MemoryPool>& pool, uint64_t file_size)
+ : input_stream_(input_stream), pool_(pool), file_size_(file_size) {}
+
+ArrowInputStreamAdapter::~ArrowInputStreamAdapter() {
+ [[maybe_unused]] auto status = DoClose();
+}
+
+arrow::Status ArrowInputStreamAdapter::Seek(int64_t position) {
+ return ToArrowStatus(input_stream_->Seek(position,
SeekOrigin::FS_SEEK_SET));
+}
+
+arrow::Result<int64_t> ArrowInputStreamAdapter::Read(int64_t nbytes, void*
out) {
+ ARROW_RETURN_NOT_OK(ValidateArrowIoRange<uint32_t>(nbytes, "nbytes"));
+ Result<int32_t> read_bytes =
+ input_stream_->Read(static_cast<char*>(out),
static_cast<uint32_t>(nbytes));
+ if (!read_bytes.ok()) {
+ return ToArrowStatus(read_bytes.status());
+ }
+ return read_bytes.value();
+}
+
+arrow::Result<std::shared_ptr<arrow::Buffer>>
ArrowInputStreamAdapter::Read(int64_t nbytes) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::ResizableBuffer> buffer,
+ arrow::AllocateResizableBuffer(nbytes, pool_.get()));
+ ARROW_ASSIGN_OR_RAISE(int64_t read_bytes, Read(nbytes,
buffer->mutable_data()));
+ if (read_bytes < nbytes) {
+ ARROW_RETURN_NOT_OK(buffer->Resize(read_bytes));
+ }
+ return std::shared_ptr<arrow::Buffer>(std::move(buffer));
+}
+
+arrow::Result<int64_t> ArrowInputStreamAdapter::ReadAt(int64_t position,
int64_t nbytes,
+ void* out) {
+ ARROW_RETURN_NOT_OK(ValidateArrowIoRange<uint64_t>(position, "position"));
+ ARROW_RETURN_NOT_OK(ValidateArrowIoRange<uint32_t>(nbytes, "nbytes"));
+ Result<int32_t> read_bytes = input_stream_->Read(
+ static_cast<char*>(out), static_cast<uint32_t>(nbytes),
static_cast<uint64_t>(position));
+ if (!read_bytes.ok()) {
+ return ToArrowStatus(read_bytes.status());
+ }
+ return read_bytes.value();
+}
+
+arrow::Result<std::shared_ptr<arrow::Buffer>>
ArrowInputStreamAdapter::ReadAt(int64_t position,
+
int64_t nbytes) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<arrow::ResizableBuffer> buffer,
+ arrow::AllocateResizableBuffer(nbytes, pool_.get()));
+ ARROW_ASSIGN_OR_RAISE(int64_t read_bytes, ReadAt(position, nbytes,
buffer->mutable_data()));
+ if (read_bytes < nbytes) {
+ ARROW_RETURN_NOT_OK(buffer->Resize(read_bytes));
+ }
+ return std::shared_ptr<arrow::Buffer>(std::move(buffer));
+}
+
+arrow::Future<std::shared_ptr<arrow::Buffer>>
ArrowInputStreamAdapter::ReadAsync(
+ const arrow::io::IOContext& io_context, int64_t position, int64_t nbytes) {
+ auto fut = arrow::Future<std::shared_ptr<arrow::Buffer>>::Make();
+ auto range_status = ValidateArrowIoRange<uint64_t>(position, "position");
+ if (!range_status.ok()) {
+ fut.MarkFinished(range_status);
+ return fut;
+ }
+ range_status = ValidateArrowIoRange<uint32_t>(nbytes, "nbytes");
+ if (!range_status.ok()) {
+ fut.MarkFinished(range_status);
+ return fut;
+ }
+
+ arrow::Result<std::shared_ptr<arrow::Buffer>> buffer_result =
+ arrow::AllocateResizableBuffer(nbytes, pool_.get());
+ if (PAIMON_UNLIKELY(!buffer_result.ok())) {
+ fut.MarkFinished(buffer_result.status());
+ return fut;
+ }
+ std::shared_ptr<arrow::Buffer> buffer =
std::move(buffer_result).ValueUnsafe();
+ input_stream_->ReadAsync(reinterpret_cast<char*>(buffer->mutable_data()),
+ static_cast<uint32_t>(nbytes),
static_cast<uint64_t>(position),
+ [fut, buffer](Status callback_status) mutable {
+ if (callback_status.ok()) {
+ fut.MarkFinished(std::move(buffer));
+ } else {
+
fut.MarkFinished(ToArrowStatus(callback_status));
+ }
+ });
+ return fut;
+}
+
+arrow::Result<int64_t> ArrowInputStreamAdapter::Tell() const {
+ Result<int64_t> position = input_stream_->GetPos();
+ if (!position.ok()) {
+ return ToArrowStatus(position.status());
+ }
+ return position.value();
+}
+
+arrow::Result<int64_t> ArrowInputStreamAdapter::GetSize() {
+ return static_cast<int64_t>(file_size_);
+}
+
+bool ArrowInputStreamAdapter::closed() const {
+ return closed_;
+}
+
+arrow::Status ArrowInputStreamAdapter::DoClose() {
+ if (!closed_) {
+ Status status = input_stream_->Close();
+ if (!status.ok()) {
+ return ToArrowStatus(status);
+ }
+ closed_ = true;
+ }
+ return arrow::Status::OK();
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/arrow_input_stream_adapter.h
b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.h
new file mode 100644
index 0000000..00ee164
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.h
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+
+#include "arrow/api.h"
+#include "arrow/io/interfaces.h"
+#include "arrow/util/future.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class InputStream;
+
+class PAIMON_EXPORT ArrowInputStreamAdapter : public
arrow::io::RandomAccessFile {
+ public:
+ ArrowInputStreamAdapter(const std::shared_ptr<paimon::InputStream>&
input_stream,
+ const std::shared_ptr<arrow::MemoryPool>& pool,
uint64_t file_size);
+ ~ArrowInputStreamAdapter() override;
+
+ // NOTE: In paimon file system definition, position + nbytes should not
exceed file_size_.
+ arrow::Result<int64_t> Read(int64_t nbytes, void* out) override;
+ arrow::Result<std::shared_ptr<arrow::Buffer>> Read(int64_t nbytes)
override;
+ arrow::Result<int64_t> ReadAt(int64_t position, int64_t nbytes, void* out)
override;
+ arrow::Result<std::shared_ptr<arrow::Buffer>> ReadAt(int64_t position,
int64_t nbytes) override;
+ arrow::Future<std::shared_ptr<arrow::Buffer>> ReadAsync(const
arrow::io::IOContext& io_context,
+ int64_t position,
+ int64_t nbytes)
override;
+ arrow::Status Seek(int64_t position) override;
+ arrow::Result<int64_t> Tell() const override;
+ arrow::Result<int64_t> GetSize() override;
+ arrow::Status Close() override {
+ return DoClose();
+ }
+ bool closed() const override;
+
+ private:
+ arrow::Status DoClose();
+
+ std::shared_ptr<paimon::InputStream> input_stream_;
+ std::shared_ptr<arrow::MemoryPool> pool_;
+ uint64_t file_size_;
+ bool closed_ = false;
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/arrow_output_stream_adapter.cpp
b/src/paimon/common/utils/arrow/arrow_output_stream_adapter.cpp
new file mode 100644
index 0000000..c74c090
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_output_stream_adapter.cpp
@@ -0,0 +1,70 @@
+/*
+ * 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/common/utils/arrow/arrow_output_stream_adapter.h"
+
+#include "arrow/result.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/math.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+ArrowOutputStreamAdapter::ArrowOutputStreamAdapter(const
std::shared_ptr<paimon::OutputStream>& out)
+ : out_(out) {}
+
+arrow::Status ArrowOutputStreamAdapter::Close() {
+ // output stream close is called by paimon framework(such as single file
writer), no need to
+ // close here
+ closed_ = true;
+ return arrow::Status::OK();
+}
+
+arrow::Result<int64_t> ArrowOutputStreamAdapter::Tell() const {
+ paimon::Result<int64_t> pos = out_->GetPos();
+ if (!pos.ok()) {
+ return ToArrowStatus(pos.status());
+ }
+ return pos.value();
+}
+
+bool ArrowOutputStreamAdapter::closed() const {
+ return closed_;
+}
+
+arrow::Status ArrowOutputStreamAdapter::Write(const void* data, int64_t
nbytes) {
+ if (!InRange<uint32_t>(nbytes)) {
+ return arrow::Status::Invalid(
+ fmt::format("nbytes value {} is out of bound of uint32_t",
nbytes));
+ }
+ Result<int32_t> len =
+ out_->Write(static_cast<const char*>(data),
static_cast<uint32_t>(nbytes));
+ if (!len.ok()) {
+ return ToArrowStatus(len.status());
+ }
+ return arrow::Status::OK();
+}
+
+arrow::Status ArrowOutputStreamAdapter::Flush() {
+ return ToArrowStatus(out_->Flush());
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/arrow_output_stream_adapter.h
b/src/paimon/common/utils/arrow/arrow_output_stream_adapter.h
new file mode 100644
index 0000000..75f791d
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_output_stream_adapter.h
@@ -0,0 +1,48 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+
+#include "arrow/io/interfaces.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class OutputStream;
+
+class PAIMON_EXPORT ArrowOutputStreamAdapter : public arrow::io::OutputStream {
+ public:
+ explicit ArrowOutputStreamAdapter(const
std::shared_ptr<paimon::OutputStream>& out);
+
+ arrow::Status Close() override;
+ arrow::Result<int64_t> Tell() const override;
+ bool closed() const override;
+ arrow::Status Write(const void* data, int64_t nbytes) override;
+ arrow::Status Flush() override;
+
+ private:
+ std::shared_ptr<paimon::OutputStream> out_;
+ bool closed_ = false;
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/arrow_stream_adapter_test.cpp
b/src/paimon/common/utils/arrow/arrow_stream_adapter_test.cpp
new file mode 100644
index 0000000..04fba95
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_stream_adapter_test.cpp
@@ -0,0 +1,94 @@
+/*
+ * 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 <cstdint>
+#include <memory>
+#include <string>
+
+#include "arrow/api.h"
+#include "arrow/io/type_fwd.h"
+#include "arrow/util/future.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/arrow/arrow_input_stream_adapter.h"
+#include "paimon/common/utils/arrow/arrow_output_stream_adapter.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(ArrowStreamAdapterTest, TestInputAndOutputStream) {
+ auto test_root_dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(test_root_dir);
+ std::string test_root = test_root_dir->Str();
+ std::shared_ptr<FileSystem> file_system =
std::make_shared<LocalFileSystem>();
+ ASSERT_OK(file_system->Mkdirs(test_root));
+ std::string file_name = test_root + "/stream.arrow";
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> out,
+ file_system->Create(file_name, /*overwrite=*/true));
+ auto out_stream = std::make_unique<ArrowOutputStreamAdapter>(out);
+ ASSERT_EQ(out_stream->Tell().ValueOrDie(), 0);
+ ASSERT_FALSE(out_stream->closed());
+
+ std::string data = "hello";
+ ASSERT_TRUE(out_stream->Write(data.data(), data.length()).ok());
+ ASSERT_TRUE(out_stream->Flush().ok());
+ ASSERT_EQ(out_stream->Tell().ValueOrDie(), 5);
+ // noted that ArrowOutputStreamAdapter::Close() api do nothing except set
the closed_ flag.
+ ASSERT_TRUE(out_stream->Close().ok());
+ ASSERT_TRUE(out_stream->closed());
+ ASSERT_OK(out_stream->out_->Close());
+
+ // in stream
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in,
file_system->Open(file_name));
+ ASSERT_OK_AND_ASSIGN(uint64_t length, in->Length());
+ auto in_stream =
+ std::make_unique<ArrowInputStreamAdapter>(in,
GetArrowPool(GetDefaultPool()), length);
+ ASSERT_EQ(in_stream->GetSize().ValueOrDie(),
static_cast<int64_t>(data.length()));
+ ASSERT_EQ(in_stream->Tell().ValueOrDie(), 0);
+ ASSERT_FALSE(in_stream->closed());
+
+ char ret[10] = {};
+ int64_t read_len = in_stream->Read(data.length(), ret).ValueOrDie();
+ ASSERT_EQ(read_len, data.length());
+ ASSERT_EQ(std::string(ret, read_len), data);
+ ASSERT_EQ(in_stream->Tell().ValueOrDie(), 5);
+
+ ASSERT_TRUE(in_stream->Seek(/*position=*/3).ok());
+ std::shared_ptr<arrow::Buffer> buffer =
in_stream->Read(/*nbytes=*/2).ValueOrDie();
+ ASSERT_EQ(buffer->ToString(), "lo");
+
+ auto fut = in_stream->ReadAsync(arrow::io::default_io_context(),
/*position=*/0, /*nbytes=*/5);
+ auto buffer2 = fut.result().ValueOrDie();
+ ASSERT_EQ(data, buffer2->ToString());
+
+ auto buffer3 = in_stream->ReadAt(/*position=*/1,
/*nbytes=*/2).ValueOrDie();
+ ASSERT_EQ(buffer3->ToString(), "el");
+ int64_t read_len2 = in_stream->ReadAt(/*position=*/4, /*nbytes=*/1,
ret).ValueOrDie();
+ ASSERT_EQ(read_len2, 1);
+ ASSERT_EQ(std::string(ret, read_len2), "o");
+
+ ASSERT_TRUE(in_stream->Close().ok());
+ ASSERT_TRUE(in_stream->closed());
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/arrow/arrow_utils.cpp
b/src/paimon/common/utils/arrow/arrow_utils.cpp
new file mode 100644
index 0000000..88623b3
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_utils.cpp
@@ -0,0 +1,179 @@
+/*
+ * 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/common/utils/arrow/arrow_utils.h"
+
+#include "arrow/array/array_base.h"
+#include "arrow/array/array_nested.h"
+#include "arrow/util/compression.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/string_utils.h"
+
+namespace paimon {
+Result<std::shared_ptr<arrow::Schema>> ArrowUtils::DataTypeToSchema(
+ const std::shared_ptr<arrow::DataType>& data_type) {
+ if (data_type->id() != arrow::Type::STRUCT) {
+ return Status::Invalid(
+ fmt::format("Expected struct data type, actual data type: {}",
data_type->ToString()));
+ }
+ const auto& struct_type =
std::static_pointer_cast<arrow::StructType>(data_type);
+ return std::make_shared<arrow::Schema>(struct_type->fields());
+}
+
+Result<std::vector<int32_t>> ArrowUtils::CreateProjection(
+ const std::shared_ptr<arrow::Schema>& src_schema, const
arrow::FieldVector& target_fields) {
+ std::vector<int32_t> target_to_src_mapping;
+ target_to_src_mapping.reserve(target_fields.size());
+ for (const auto& field : target_fields) {
+ auto src_field_idx = src_schema->GetFieldIndex(field->name());
+ if (src_field_idx < 0) {
+ return Status::Invalid(
+ fmt::format("Field '{}' not found or duplicate in src schema",
field->name()));
+ }
+ target_to_src_mapping.push_back(src_field_idx);
+ }
+ return target_to_src_mapping;
+}
+
+Status ArrowUtils::CheckNullabilityMatch(const std::shared_ptr<arrow::Schema>&
schema,
+ const std::shared_ptr<arrow::Array>&
data) {
+ auto struct_array =
arrow::internal::checked_pointer_cast<arrow::StructArray>(data);
+ if (struct_array->num_fields() != schema->num_fields()) {
+ return Status::Invalid(fmt::format(
+ "CheckNullabilityMatch failed, data field count {} mismatch schema
field count {}",
+ struct_array->num_fields(), schema->num_fields()));
+ }
+ for (int32_t i = 0; i < schema->num_fields(); i++) {
+ PAIMON_RETURN_NOT_OK(InnerCheckNullabilityMatch(schema->field(i),
struct_array->field(i)));
+ }
+ return Status::OK();
+}
+
+void ArrowUtils::TraverseArray(const std::shared_ptr<arrow::Array>& array) {
+ arrow::Type::type type = array->type()->id();
+ switch (type) {
+ case arrow::Type::type::DICTIONARY: {
+ auto* dict_array =
arrow::internal::checked_cast<arrow::DictionaryArray*>(array.get());
+ [[maybe_unused]] auto dict = dict_array->dictionary();
+ return;
+ }
+ case arrow::Type::type::STRUCT: {
+ auto* struct_array =
arrow::internal::checked_cast<arrow::StructArray*>(array.get());
+ for (const auto& field : struct_array->fields()) {
+ TraverseArray(field);
+ }
+ return;
+ }
+ case arrow::Type::type::MAP: {
+ auto* map_array =
arrow::internal::checked_cast<arrow::MapArray*>(array.get());
+ TraverseArray(map_array->keys());
+ TraverseArray(map_array->items());
+ return;
+ }
+ case arrow::Type::type::LIST: {
+ auto* list_array =
arrow::internal::checked_cast<arrow::ListArray*>(array.get());
+ TraverseArray(list_array->values());
+ return;
+ }
+ default:
+ return;
+ }
+}
+
+bool ArrowUtils::EqualsIgnoreNullable(const std::shared_ptr<arrow::DataType>&
type,
+ const std::shared_ptr<arrow::DataType>&
other_type) {
+ if (type->id() != other_type->id() || type->num_fields() !=
other_type->num_fields()) {
+ return false;
+ }
+ for (int32_t i = 0; i < type->num_fields(); ++i) {
+ const auto& field = type->field(i);
+ const auto& other_field = other_type->field(i);
+ if (field->name() != other_field->name()) {
+ return false;
+ }
+ if (!EqualsIgnoreNullable(field->type(), other_field->type())) {
+ return false;
+ }
+ }
+ return true;
+}
+
+Status ArrowUtils::InnerCheckNullabilityMatch(const
std::shared_ptr<arrow::Field>& field,
+ const
std::shared_ptr<arrow::Array>& data) {
+ if (PAIMON_UNLIKELY(!field->nullable() && data->null_count() != 0)) {
+ return Status::Invalid(fmt::format(
+ "CheckNullabilityMatch failed, field {} not nullable while data
have null value",
+ field->name()));
+ }
+ auto type = field->type();
+ if (type->id() == arrow::Type::STRUCT) {
+ auto struct_type =
arrow::internal::checked_pointer_cast<arrow::StructType>(field->type());
+ auto struct_array =
arrow::internal::checked_pointer_cast<arrow::StructArray>(data);
+ for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
+ PAIMON_RETURN_NOT_OK(
+ InnerCheckNullabilityMatch(struct_type->field(i),
struct_array->field(i)));
+ }
+ } else if (type->id() == arrow::Type::LIST) {
+ auto list_type =
arrow::internal::checked_pointer_cast<arrow::ListType>(field->type());
+ auto list_array =
arrow::internal::checked_pointer_cast<arrow::ListArray>(data);
+ PAIMON_RETURN_NOT_OK(
+ InnerCheckNullabilityMatch(list_type->value_field(),
list_array->values()));
+ } else if (type->id() == arrow::Type::MAP) {
+ auto map_type =
arrow::internal::checked_pointer_cast<arrow::MapType>(field->type());
+ auto map_array =
arrow::internal::checked_pointer_cast<arrow::MapArray>(data);
+ PAIMON_RETURN_NOT_OK(InnerCheckNullabilityMatch(map_type->key_field(),
map_array->keys()));
+ PAIMON_RETURN_NOT_OK(
+ InnerCheckNullabilityMatch(map_type->item_field(),
map_array->items()));
+ }
+ return Status::OK();
+}
+
+Result<std::shared_ptr<arrow::StructArray>>
ArrowUtils::RemoveFieldFromStructArray(
+ const std::shared_ptr<arrow::StructArray>& struct_array, const
std::string& field_name) {
+ auto struct_type =
std::static_pointer_cast<arrow::StructType>(struct_array->type());
+ int32_t field_idx = struct_type->GetFieldIndex(field_name);
+ if (field_idx == -1) {
+ return struct_array;
+ }
+ std::vector<std::shared_ptr<arrow::Array>> new_arrays;
+ std::vector<std::shared_ptr<arrow::Field>> new_fields;
+ for (int32_t i = 0; i < struct_type->num_fields(); ++i) {
+ if (i != field_idx) {
+ new_arrays.emplace_back(struct_array->field(i));
+ new_fields.emplace_back(struct_type->field(i));
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::StructArray> array,
+ arrow::StructArray::Make(new_arrays, new_fields,
struct_array->null_bitmap(),
+ struct_array->null_count(),
struct_array->offset()));
+ return array;
+}
+
+Result<arrow::Compression::type> ArrowUtils::GetCompressionType(const
std::string& compression) {
+ std::string normalized = StringUtils::ToLowerCase(compression);
+ if (normalized.empty() || normalized == "none") {
+ normalized = "uncompressed";
+ }
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(arrow::Compression::type
compression_type,
+
arrow::util::Codec::GetCompressionType(normalized));
+ return compression_type;
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/arrow_utils.h
b/src/paimon/common/utils/arrow/arrow_utils.h
new file mode 100644
index 0000000..f6dea3d
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_utils.h
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/util/type_fwd.h"
+#include "fmt/format.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+class PAIMON_EXPORT ArrowUtils {
+ public:
+ ArrowUtils() = delete;
+ ~ArrowUtils() = delete;
+
+ static Result<std::shared_ptr<arrow::Schema>> DataTypeToSchema(
+ const std::shared_ptr<arrow::DataType>& data_type);
+
+ static Result<std::vector<int32_t>> CreateProjection(
+ const std::shared_ptr<arrow::Schema>& src_schema, const
arrow::FieldVector& target_fields);
+
+ static Status CheckNullabilityMatch(const std::shared_ptr<arrow::Schema>&
schema,
+ const std::shared_ptr<arrow::Array>&
data);
+
+ // For struct array, arrow is unsafe for fields() and field(); for dict
array, arrow is unsafe
+ // for dictionary(). Therefore, access array in advance before merge sort
and projection to
+ // avoid subsequent multi-threading problems.
+ static void TraverseArray(const std::shared_ptr<arrow::Array>& array);
+
+ static Result<std::shared_ptr<arrow::StructArray>>
RemoveFieldFromStructArray(
+ const std::shared_ptr<arrow::StructArray>& struct_array, const
std::string& field_name);
+
+ static bool EqualsIgnoreNullable(const std::shared_ptr<arrow::DataType>&
type,
+ const std::shared_ptr<arrow::DataType>&
other_type);
+
+ /// Normalize and resolve a compression string to an Arrow compression
type.
+ /// Handles "none" and empty string by mapping them to "uncompressed".
+ static Result<arrow::Compression::type> GetCompressionType(const
std::string& compression);
+
+ private:
+ static Status InnerCheckNullabilityMatch(const
std::shared_ptr<arrow::Field>& field,
+ const
std::shared_ptr<arrow::Array>& data);
+};
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/arrow_utils_test.cpp
b/src/paimon/common/utils/arrow/arrow_utils_test.cpp
new file mode 100644
index 0000000..8908f67
--- /dev/null
+++ b/src/paimon/common/utils/arrow/arrow_utils_test.cpp
@@ -0,0 +1,490 @@
+/*
+ * 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/common/utils/arrow/arrow_utils.h"
+
+#include "arrow/api.h"
+#include "arrow/ipc/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/testing/utils/testharness.h"
+namespace paimon::test {
+
+TEST(ArrowUtilsTest, TestCreateProjection) {
+ arrow::FieldVector file_fields = {
+ arrow::field("k0", arrow::int32()), arrow::field("k1",
arrow::int32()),
+ arrow::field("p1", arrow::int32()), arrow::field("s1",
arrow::utf8()),
+ arrow::field("v0", arrow::float64()), arrow::field("v1",
arrow::boolean()),
+ arrow::field("s0", arrow::utf8())};
+ auto file_schema = arrow::schema(file_fields);
+
+ {
+ // normal case
+ arrow::FieldVector read_fields = {
+ arrow::field("k1", arrow::int32()), arrow::field("p1",
arrow::int32()),
+ arrow::field("s1", arrow::utf8()), arrow::field("v0",
arrow::float64()),
+ arrow::field("v1", arrow::boolean())};
+ auto read_schema = arrow::schema(read_fields);
+ ASSERT_OK_AND_ASSIGN(std::vector<int32_t> projection,
+ ArrowUtils::CreateProjection(file_schema,
read_schema->fields()));
+ std::vector<int32_t> expected_projection = {1, 2, 3, 4, 5};
+ ASSERT_EQ(projection, expected_projection);
+ }
+ {
+ // duplicate read field
+ arrow::FieldVector read_fields = {
+ arrow::field("k1", arrow::int32()), arrow::field("p1",
arrow::int32()),
+ arrow::field("s1", arrow::utf8()), arrow::field("v0",
arrow::float64()),
+ arrow::field("v0", arrow::float64()), arrow::field("v1",
arrow::boolean())};
+ auto read_schema = arrow::schema(read_fields);
+ ASSERT_OK_AND_ASSIGN(std::vector<int32_t> projection,
+ ArrowUtils::CreateProjection(file_schema,
read_schema->fields()));
+ std::vector<int32_t> expected_projection = {1, 2, 3, 4, 4, 5};
+ ASSERT_EQ(projection, expected_projection);
+ }
+ {
+ // duplicate read field, and sizeof(read_fields) > sizeof(file_fields)
+ arrow::FieldVector read_fields = {
+ arrow::field("k1", arrow::int32()), arrow::field("p1",
arrow::int32()),
+ arrow::field("s1", arrow::utf8()), arrow::field("v0",
arrow::float64()),
+ arrow::field("v0", arrow::float64()), arrow::field("v0",
arrow::float64()),
+ arrow::field("v0", arrow::float64()), arrow::field("v0",
arrow::float64()),
+ arrow::field("v1", arrow::boolean())};
+ auto read_schema = arrow::schema(read_fields);
+ ASSERT_OK_AND_ASSIGN(std::vector<int32_t> projection,
+ ArrowUtils::CreateProjection(file_schema,
read_schema->fields()));
+ std::vector<int32_t> expected_projection = {1, 2, 3, 4, 4, 4, 4, 4, 5};
+ ASSERT_EQ(projection, expected_projection);
+ }
+ {
+ // read field not found in src schema
+ arrow::FieldVector read_fields = {
+ arrow::field("k1", arrow::int32()), arrow::field("p1",
arrow::int32()),
+ arrow::field("s1", arrow::utf8()), arrow::field("v2",
arrow::float64()),
+ arrow::field("v1", arrow::boolean())};
+ auto read_schema = arrow::schema(read_fields);
+ ASSERT_NOK_WITH_MSG(ArrowUtils::CreateProjection(file_schema,
read_schema->fields()),
+ "Field 'v2' not found or duplicate in src schema");
+ }
+ {
+ // duplicate field in src schema
+ arrow::FieldVector file_fields_dup = {
+ arrow::field("k0", arrow::int32()), arrow::field("k1",
arrow::int32()),
+ arrow::field("p1", arrow::int32()), arrow::field("s1",
arrow::utf8()),
+ arrow::field("v0", arrow::float64()), arrow::field("v1",
arrow::boolean()),
+ arrow::field("v1", arrow::boolean()), arrow::field("s0",
arrow::utf8())};
+ auto file_schema_dup = arrow::schema(file_fields_dup);
+ arrow::FieldVector read_fields = {
+ arrow::field("k1", arrow::int32()), arrow::field("p1",
arrow::int32()),
+ arrow::field("s1", arrow::utf8()), arrow::field("v1",
arrow::float64()),
+ arrow::field("v1", arrow::boolean())};
+ auto read_schema = arrow::schema(read_fields);
+ ASSERT_NOK_WITH_MSG(ArrowUtils::CreateProjection(file_schema_dup,
read_schema->fields()),
+ "Field 'v1' not found or duplicate in src schema");
+ }
+ {
+ arrow::FieldVector read_fields = {
+ arrow::field("k1", arrow::int32()), arrow::field("p1",
arrow::int32()),
+ arrow::field("s1", arrow::utf8()), arrow::field("v0",
arrow::float64()),
+ arrow::field("v1", arrow::boolean())};
+ auto read_schema = arrow::schema(read_fields);
+ ASSERT_OK_AND_ASSIGN(std::vector<int32_t> projection,
+ ArrowUtils::CreateProjection(file_schema,
read_schema->fields()));
+ std::vector<int32_t> expected_projection = {1, 2, 3, 4, 5};
+ ASSERT_EQ(projection, expected_projection);
+ }
+}
+
+TEST(ArrowUtilsTest, TestCheckNullableMatchSimple) {
+ auto field = arrow::field("column1", arrow::int32(), /*nullable=*/false);
+ auto schema = arrow::schema({field});
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({field}),
R"([
+ [20],
+ [null],
+ [10]
+])")
+ .ValueOrDie();
+
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field column1 not nullable while
data have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({field}),
R"([
+ [20],
+ [10]
+])")
+ .ValueOrDie();
+
+ ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
+ }
+}
+
+TEST(ArrowUtilsTest, TestCheckNullableMatchWithStruct) {
+ auto child1 = arrow::field("child1", arrow::int32(), /*nullable=*/false);
+ auto child2 = arrow::field("child2", arrow::float64(), /*nullable=*/true);
+ auto struct_field =
+ arrow::field("parent", arrow::struct_({child1, child2}),
/*nullable=*/false);
+ auto schema = arrow::schema({struct_field});
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({struct_field}), R"([
+ [null]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field parent not nullable while
data have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({struct_field}), R"([
+ [[1, null]],
+ [[null, 10.0]]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field child1 not nullable while
data have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({struct_field}), R"([
+ [[1, null]],
+ [[2, 10.0]]
+])")
+ .ValueOrDie();
+ ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
+ }
+}
+
+TEST(ArrowUtilsTest, TestCheckNullableMatchWithList) {
+ auto value_field = arrow::field("value", arrow::int32(),
/*nullable=*/false);
+ auto list_field = arrow::field("list_column", arrow::list(value_field),
/*nullable=*/false);
+ auto schema = arrow::schema({list_field});
+
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({list_field}), R"([
+ [[1, 2, null, 4, 5]],
+ [null]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field list_column
not nullable while "
+ "data have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({list_field}), R"([
+ [[1, 2, null, 4, 5]]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field value not nullable while data
have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({list_field}), R"([
+ [[1, 2, 3, 4, 5]]
+])")
+ .ValueOrDie();
+ ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
+ }
+}
+
+TEST(ArrowUtilsTest, TestCheckNullableMatchWithMap) {
+ auto key_field = arrow::field("key", arrow::int32(), /*nullable=*/false);
+ auto value_field = arrow::field("value", arrow::int32(),
/*nullable=*/true);
+ auto map_type = std::make_shared<arrow::MapType>(key_field, value_field);
+ auto map_field = arrow::field("map_column", map_type, /*nullable=*/false);
+ auto schema = arrow::schema({map_field});
+
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({map_field}), R"([
+ [null]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field map_column
not nullable while "
+ "data have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({map_field}), R"([
+ [[[1, null]]]
+])")
+ .ValueOrDie();
+ ASSERT_OK(ArrowUtils::CheckNullabilityMatch(schema, array));
+ }
+}
+
+TEST(ArrowUtilsTest, TestCheckNullableMatchComplex) {
+ auto key_field = arrow::field("key", arrow::int32(), /*nullable=*/false);
+ auto value_field = arrow::field("value", arrow::int32(),
/*nullable=*/false);
+
+ auto inner_child1 =
+ arrow::field("inner1",
+ arrow::map(arrow::utf8(), arrow::field("inner_list",
arrow::list(value_field),
+
/*nullable=*/true)),
+ /*nullable=*/false);
+ auto inner_child2 = arrow::field(
+ "inner2",
+ arrow::map(arrow::utf8(), arrow::field("inner_map",
arrow::map(arrow::utf8(), value_field),
+ /*nullable=*/true)),
+ /*nullable=*/false);
+ auto inner_child3 = arrow::field(
+ "inner3",
+ arrow::map(arrow::utf8(),
+ arrow::field("inner_struct", arrow::struct_({key_field,
value_field}),
+ /*nullable=*/true)),
+ /*nullable=*/false);
+
+ auto schema = arrow::schema({inner_child1, inner_child2, inner_child3});
+ // test inner1
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({inner_child1, inner_child2, inner_child3}),
R"([
+[[["outer_key", [1, 2, 3, null]]], [["outer_key", [["key1", 1]]]],
[["outer_key", [100, 200]]]]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field value not nullable while data
have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({inner_child1, inner_child2, inner_child3}),
R"([
+[[["outer_key", [1, 2, 3]]], [["outer_key", [["key1", 1]]]], [["outer_key",
[100, 200]]]],
+[[["outer_key", null]], [["outer_key", [["key1", 1]]]], [["outer_key", [100,
200]]]],
+[null, [["outer_key", [["key1", 1]]]], [["outer_key", [100, 200]]]]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field inner1 not nullable while
data have null value");
+ }
+ // test inner2
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({inner_child1, inner_child2, inner_child3}),
R"([
+[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", [100,
200]]]],
+[[["outer_key", null]], [["outer_key", [["key1", null]]]], [["outer_key",
[100, 200]]]]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field value not nullable while data
have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({inner_child1, inner_child2, inner_child3}),
R"([
+[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", [100,
200]]]],
+[[["outer_key", [1, 2, 3]]], null, [["outer_key", [100, 200]]]]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field inner2 not nullable while
data have null value");
+ }
+ // test inner3
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({inner_child1, inner_child2, inner_child3}),
R"([
+[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", null]]],
+[[["outer_key", null]], [["outer_key", [["key1", 2]]]], [["outer_key", [100,
null]]]]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field value not nullable while data
have null value");
+ }
+ {
+ std::shared_ptr<arrow::Array> array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::struct_({inner_child1, inner_child2, inner_child3}),
R"([
+[[["outer_key", [1, 2, 3]]], [["outer_key", null]], [["outer_key", null]]],
+[[["outer_key", null]], [["outer_key", [["key1", 2]]]], null]
+])")
+ .ValueOrDie();
+ ASSERT_NOK_WITH_MSG(
+ ArrowUtils::CheckNullabilityMatch(schema, array),
+ "CheckNullabilityMatch failed, field inner3 not nullable while
data have null value");
+ }
+}
+
+TEST(ArrowUtilsTest, TestRemoveFieldFromStructArrayFieldNotFound) {
+ auto struct_type =
+ arrow::struct_({arrow::field("a", arrow::int32()), arrow::field("b",
arrow::utf8())});
+ auto src_array = arrow::ipc::internal::json::ArrayFromJSON(
+ struct_type,
R"([{"a":1,"b":"x"},{"a":2,"b":"y"},{"a":3,"b":"z"}])")
+ .ValueOrDie();
+ auto src_struct_array =
std::static_pointer_cast<arrow::StructArray>(src_array);
+
+ ASSERT_OK_AND_ASSIGN(auto result,
+
ArrowUtils::RemoveFieldFromStructArray(src_struct_array, "missing"));
+
+ ASSERT_TRUE(result->Equals(src_struct_array));
+ ASSERT_EQ(result->type()->num_fields(), 2);
+}
+
+TEST(ArrowUtilsTest, TestRemoveFieldFromStructArraySuccess) {
+ auto struct_type =
+ arrow::struct_({arrow::field("a", arrow::int32()), arrow::field("b",
arrow::utf8()),
+ arrow::field("c", arrow::int64())});
+ auto src_array =
+ arrow::ipc::internal::json::ArrayFromJSON(
+ struct_type,
+
R"([{"a":1,"b":"x","c":10},{"a":2,"b":"y","c":20},{"a":3,"b":"z","c":30}])")
+ .ValueOrDie();
+ auto src_struct_array =
std::static_pointer_cast<arrow::StructArray>(src_array);
+
+ ASSERT_OK_AND_ASSIGN(auto result,
+
ArrowUtils::RemoveFieldFromStructArray(src_struct_array, "b"));
+
+ auto expected_type =
+ arrow::struct_({arrow::field("a", arrow::int32()), arrow::field("c",
arrow::int64())});
+ auto expected_array = arrow::ipc::internal::json::ArrayFromJSON(
+ expected_type,
R"([{"a":1,"c":10},{"a":2,"c":20},{"a":3,"c":30}])")
+ .ValueOrDie();
+ auto expected_struct_array =
std::static_pointer_cast<arrow::StructArray>(expected_array);
+
+ ASSERT_EQ(result->type()->num_fields(), 2);
+ ASSERT_EQ(result->type()->field(0)->name(), "a");
+ ASSERT_EQ(result->type()->field(1)->name(), "c");
+ ASSERT_TRUE(result->Equals(expected_struct_array));
+}
+
+TEST(ArrowUtilsTest, TestEqualsIgnoreNullable) {
+ {
+ // test simple
+ ASSERT_FALSE(ArrowUtils::EqualsIgnoreNullable(arrow::int32(),
arrow::int64()));
+ ASSERT_TRUE(ArrowUtils::EqualsIgnoreNullable(arrow::int32(),
arrow::int32()));
+ }
+ {
+ // test struct
+ auto child1 = arrow::field("child1", arrow::int32(),
/*nullable=*/false);
+ auto child2 = arrow::field("child2", arrow::int32(),
/*nullable=*/false);
+ auto child3 = arrow::field("child1", arrow::int32(),
/*nullable=*/true);
+ auto struct_type1 = arrow::struct_({child1});
+ auto struct_type2 = arrow::struct_({child2});
+ auto struct_type3 = arrow::struct_({child3});
+ auto struct_type4 = arrow::struct_({child3, child1});
+ ASSERT_FALSE(ArrowUtils::EqualsIgnoreNullable(struct_type1,
struct_type2));
+ ASSERT_TRUE(ArrowUtils::EqualsIgnoreNullable(struct_type1,
struct_type3));
+ ASSERT_FALSE(ArrowUtils::EqualsIgnoreNullable(struct_type1,
struct_type4));
+ }
+ {
+ // test complex
+ auto key_field = arrow::field("key", arrow::int32(),
/*nullable=*/false);
+ auto value_field = arrow::field("value", arrow::int32(),
/*nullable=*/false);
+ auto inner_child1 = arrow::field(
+ "inner1",
+ arrow::map(arrow::utf8(), arrow::field("inner_list",
arrow::list(value_field),
+ /*nullable=*/true)),
+ /*nullable=*/false);
+ auto inner_child2 = arrow::field(
+ "inner2",
+ arrow::map(arrow::utf8(),
+ arrow::field("inner_map", arrow::map(arrow::utf8(),
value_field),
+ /*nullable=*/true)),
+ /*nullable=*/false);
+ auto inner_child3 = arrow::field(
+ "inner3",
+ arrow::map(arrow::utf8(),
+ arrow::field("inner_struct", arrow::struct_({key_field,
value_field}),
+ /*nullable=*/true)),
+ /*nullable=*/false);
+ auto struct_type1 = arrow::struct_({inner_child1, inner_child2,
inner_child3});
+
+ auto key_field_other = arrow::field("key", arrow::int32(),
/*nullable=*/true);
+ auto value_field_other = arrow::field("value", arrow::int32(),
/*nullable=*/true);
+ auto inner_child1_other = arrow::field(
+ "inner1",
+ arrow::map(arrow::utf8(), arrow::field("inner_list",
arrow::list(value_field_other),
+ /*nullable=*/false)),
+ /*nullable=*/true);
+ auto inner_child2_other = arrow::field(
+ "inner2",
+ arrow::map(arrow::utf8(),
+ arrow::field("inner_map", arrow::map(arrow::utf8(),
value_field_other),
+ /*nullable=*/false)),
+ /*nullable=*/true);
+ auto inner_child3_other = arrow::field(
+ "inner3",
+ arrow::map(
+ arrow::utf8(),
+ arrow::field("inner_struct", arrow::struct_({key_field_other,
value_field_other}),
+ /*nullable=*/false)),
+ /*nullable=*/true);
+ auto struct_type2 =
+ arrow::struct_({inner_child1_other, inner_child2_other,
inner_child3_other});
+ ASSERT_TRUE(ArrowUtils::EqualsIgnoreNullable(struct_type1,
struct_type2));
+ }
+}
+
+TEST(ArrowUtilsTest, TestGetCompressionType) {
+ {
+ ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType(""));
+ ASSERT_EQ(type, arrow::Compression::UNCOMPRESSED);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(auto type,
ArrowUtils::GetCompressionType("none"));
+ ASSERT_EQ(type, arrow::Compression::UNCOMPRESSED);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(auto type,
ArrowUtils::GetCompressionType("uncompressed"));
+ ASSERT_EQ(type, arrow::Compression::UNCOMPRESSED);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(auto type,
ArrowUtils::GetCompressionType("zstd"));
+ ASSERT_EQ(type, arrow::Compression::ZSTD);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(auto type,
ArrowUtils::GetCompressionType("ZSTD"));
+ ASSERT_EQ(type, arrow::Compression::ZSTD);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("lz4"));
+ ASSERT_EQ(type, arrow::Compression::LZ4_FRAME);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(auto type,
ArrowUtils::GetCompressionType("snappy"));
+ ASSERT_EQ(type, arrow::Compression::SNAPPY);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(auto type,
ArrowUtils::GetCompressionType("gzip"));
+ ASSERT_EQ(type, arrow::Compression::GZIP);
+ }
+ {
+ ASSERT_NOK(ArrowUtils::GetCompressionType("invalid_codec"));
+ }
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/arrow/mem_utils.cpp
b/src/paimon/common/utils/arrow/mem_utils.cpp
new file mode 100644
index 0000000..4ecb644
--- /dev/null
+++ b/src/paimon/common/utils/arrow/mem_utils.cpp
@@ -0,0 +1,87 @@
+/*
+ * 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/common/utils/arrow/mem_utils.h"
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include "arrow/memory_pool.h"
+#include "arrow/status.h"
+#include "paimon/memory/memory_pool.h"
+
+namespace paimon {
+
+class ArrowMemPoolAdaptor : public arrow::MemoryPool {
+ public:
+ explicit ArrowMemPoolAdaptor(const std::shared_ptr<paimon::MemoryPool>&
pool)
+ : pool_(*pool), life_holder_(pool) {}
+
+ arrow::Status Allocate(int64_t size, int64_t alignment, uint8_t** out)
override {
+ *out = reinterpret_cast<uint8_t*>(pool_.Malloc(size, alignment));
+ stats_.DidAllocateBytes(size);
+ return arrow::Status::OK();
+ }
+
+ arrow::Status Reallocate(int64_t old_size, int64_t new_size, int64_t
alignment,
+ uint8_t** ptr) override {
+ *ptr = reinterpret_cast<uint8_t*>(pool_.Realloc(*ptr, old_size,
new_size, alignment));
+ stats_.DidReallocateBytes(old_size, new_size);
+ return arrow::Status::OK();
+ }
+
+ void Free(uint8_t* buffer, int64_t size, int64_t alignment) override {
+ pool_.Free(buffer, size, alignment);
+ stats_.DidFreeBytes(size);
+ }
+
+ int64_t bytes_allocated() const override {
+ return stats_.bytes_allocated();
+ }
+
+ int64_t max_memory() const override {
+ return stats_.max_memory();
+ }
+
+ std::string backend_name() const override {
+ return "Paimon Pool";
+ }
+
+ /// The number of bytes that were allocated.
+ int64_t total_bytes_allocated() const override {
+ return stats_.total_bytes_allocated();
+ }
+
+ /// The number of allocations or reallocations that were requested.
+ int64_t num_allocations() const override {
+ return stats_.num_allocations();
+ }
+
+ private:
+ paimon::MemoryPool& pool_;
+ std::shared_ptr<paimon::MemoryPool> life_holder_;
+ arrow::internal::MemoryPoolStats stats_;
+};
+
+std::unique_ptr<arrow::MemoryPool> GetArrowPool(const
std::shared_ptr<MemoryPool>& pool) {
+ return std::make_unique<ArrowMemPoolAdaptor>(pool);
+}
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/mem_utils.h
b/src/paimon/common/utils/arrow/mem_utils.h
new file mode 100644
index 0000000..b59c403
--- /dev/null
+++ b/src/paimon/common/utils/arrow/mem_utils.h
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "arrow/memory_pool.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class MemoryPool;
+
+PAIMON_EXPORT std::unique_ptr<arrow::MemoryPool> GetArrowPool(MemoryPool&
pool);
+
+PAIMON_EXPORT std::unique_ptr<arrow::MemoryPool> GetArrowPool(
+ const std::shared_ptr<MemoryPool>& pool);
+
+} // namespace paimon
diff --git a/src/paimon/common/utils/arrow/mem_utils_test.cpp
b/src/paimon/common/utils/arrow/mem_utils_test.cpp
new file mode 100644
index 0000000..d5d28a7
--- /dev/null
+++ b/src/paimon/common/utils/arrow/mem_utils_test.cpp
@@ -0,0 +1,77 @@
+/*
+ * 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/common/utils/arrow/mem_utils.h"
+
+#include "gtest/gtest.h"
+#include "paimon/memory/memory_pool.h"
+
+namespace paimon::test {
+
+TEST(MemUtilsTest, TestSimple) {
+ const int64_t alignment = 64;
+ auto pool = GetArrowPool(GetDefaultPool());
+ ASSERT_EQ("Paimon Pool", pool->backend_name());
+ ASSERT_EQ(0, pool->total_bytes_allocated());
+ ASSERT_EQ(0, pool->num_allocations());
+
+ uint8_t* ptr1 = nullptr;
+ ASSERT_TRUE(pool->Allocate(10, alignment, &ptr1).ok());
+ ASSERT_TRUE(ptr1);
+ ASSERT_EQ(10, pool->total_bytes_allocated());
+ ASSERT_EQ(10, pool->bytes_allocated());
+ ASSERT_EQ(10, pool->max_memory());
+ ASSERT_EQ(1, pool->num_allocations());
+
+ // test malloc and free
+ uint8_t* ptr2 = nullptr;
+ ASSERT_TRUE(pool->Allocate(20, alignment, &ptr2).ok());
+ ASSERT_TRUE(ptr2);
+ ASSERT_EQ(30, pool->bytes_allocated());
+ ASSERT_EQ(30, pool->max_memory());
+ pool->Free(ptr2, 20, alignment);
+ ASSERT_EQ(10, pool->bytes_allocated());
+ ASSERT_EQ(30, pool->max_memory());
+ ASSERT_EQ(2, pool->num_allocations());
+
+ // test realloc with nullptr
+ uint8_t* ptr3 = nullptr;
+ ASSERT_TRUE(pool->Reallocate(/*old_size=*/0, /*new_size=*/40, alignment,
&ptr3).ok());
+ ASSERT_TRUE(ptr3);
+ ASSERT_EQ(50, pool->bytes_allocated());
+ ASSERT_EQ(50, pool->max_memory());
+ ASSERT_EQ(3, pool->num_allocations());
+
+ uint8_t* ptr3_old = ptr3;
+ // test realloc with same size
+ ASSERT_TRUE(pool->Reallocate(/*old_size=*/40, /*new_size=*/40, alignment,
&ptr3).ok());
+ ASSERT_EQ(ptr3_old, ptr3);
+ ASSERT_EQ(50, pool->bytes_allocated());
+ ASSERT_EQ(50, pool->max_memory());
+ ASSERT_EQ(3, pool->num_allocations());
+
+ pool->Free(ptr1, 10, alignment);
+ pool->Free(ptr3, 40, alignment);
+ ASSERT_EQ(0, pool->bytes_allocated());
+ ASSERT_EQ(70, pool->total_bytes_allocated());
+ ASSERT_EQ(3, pool->num_allocations());
+ ASSERT_EQ(50, pool->max_memory());
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/common/utils/arrow/status_utils_test.cpp
b/src/paimon/common/utils/arrow/status_utils_test.cpp
new file mode 100644
index 0000000..b96c993
--- /dev/null
+++ b/src/paimon/common/utils/arrow/status_utils_test.cpp
@@ -0,0 +1,60 @@
+/*
+ * 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/common/utils/arrow/status_utils.h"
+
+#include "gtest/gtest.h"
+
+namespace paimon::test {
+
+TEST(StatusUtilsTest, ToArrowStatus) {
+ ASSERT_EQ(arrow::Status::OK(), ToArrowStatus(Status::OK()));
+ std::string msg = "msg";
+ ASSERT_EQ(arrow::Status::OutOfMemory(msg),
ToArrowStatus(Status::OutOfMemory(msg)));
+ ASSERT_EQ(arrow::Status::KeyError(msg),
ToArrowStatus(Status::KeyError(msg)));
+ ASSERT_EQ(arrow::Status::TypeError(msg),
ToArrowStatus(Status::TypeError(msg)));
+ ASSERT_EQ(arrow::Status::Invalid(msg),
ToArrowStatus(Status::Invalid(msg)));
+ ASSERT_EQ(arrow::Status::IOError(msg),
ToArrowStatus(Status::IOError(msg)));
+ ASSERT_EQ(arrow::Status::CapacityError(msg),
ToArrowStatus(Status::CapacityError(msg)));
+ ASSERT_EQ(arrow::Status::IndexError(msg),
ToArrowStatus(Status::IndexError(msg)));
+ ASSERT_EQ(arrow::Status::UnknownError(msg),
ToArrowStatus(Status::UnknownError(msg)));
+ ASSERT_EQ(arrow::Status::NotImplemented(msg),
ToArrowStatus(Status::NotImplemented(msg)));
+ ASSERT_EQ(arrow::Status::SerializationError(msg),
+ ToArrowStatus(Status::SerializationError(msg)));
+ ASSERT_EQ(arrow::Status::Invalid(msg),
ToArrowStatus(Status::Invalid(msg)));
+}
+
+TEST(StatusUtilsTest, ToPaimonStatus) {
+ ASSERT_EQ(Status::OK(), ToPaimonStatus(arrow::Status::OK()));
+ std::string msg = "msg";
+ ASSERT_EQ(Status::OutOfMemory(msg),
ToPaimonStatus(arrow::Status::OutOfMemory(msg)));
+ ASSERT_EQ(Status::KeyError(msg),
ToPaimonStatus(arrow::Status::KeyError(msg)));
+ ASSERT_EQ(Status::TypeError(msg),
ToPaimonStatus(arrow::Status::TypeError(msg)));
+ ASSERT_EQ(Status::Invalid(msg),
ToPaimonStatus(arrow::Status::Invalid(msg)));
+ ASSERT_EQ(Status::IOError(msg),
ToPaimonStatus(arrow::Status::IOError(msg)));
+ ASSERT_EQ(Status::CapacityError(msg),
ToPaimonStatus(arrow::Status::CapacityError(msg)));
+ ASSERT_EQ(Status::IndexError(msg),
ToPaimonStatus(arrow::Status::IndexError(msg)));
+ ASSERT_EQ(Status::UnknownError(msg),
ToPaimonStatus(arrow::Status::UnknownError(msg)));
+ ASSERT_EQ(Status::NotImplemented(msg),
ToPaimonStatus(arrow::Status::NotImplemented(msg)));
+ ASSERT_EQ(Status::SerializationError(msg),
+ ToPaimonStatus(arrow::Status::SerializationError(msg)));
+ ASSERT_EQ(Status::Invalid(msg),
ToPaimonStatus(arrow::Status::Invalid(msg)));
+}
+
+} // namespace paimon::test
