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 8447abf feat: add type casting infrastructure for schema evolution
(#79)
8447abf is described below
commit 8447abf703f8456d80fa7c8fbf974ee616e79056
Author: lszskye <[email protected]>
AuthorDate: Wed Jun 17 19:16:39 2026 -0700
feat: add type casting infrastructure for schema evolution (#79)
---
src/paimon/core/casting/cast_executor.h | 45 ++++
src/paimon/core/casting/cast_executor_factory.cpp | 191 ++++++++++++++++
src/paimon/core/casting/cast_executor_factory.h | 45 ++++
.../core/casting/cast_executor_factory_test.cpp | 236 ++++++++++++++++++++
src/paimon/core/casting/casted_row.cpp | 51 +++++
src/paimon/core/casting/casted_row.h | 231 +++++++++++++++++++
src/paimon/core/casting/casted_row_test.cpp | 245 +++++++++++++++++++++
src/paimon/core/casting/casting_utils.cpp | 98 +++++++++
src/paimon/core/casting/casting_utils.h | 105 +++++++++
src/paimon/core/casting/casting_utils_test.cpp | 165 ++++++++++++++
.../casting/numeric_primitive_cast_executor.cpp | 216 ++++++++++++++++++
.../core/casting/numeric_primitive_cast_executor.h | 64 ++++++
12 files changed, 1692 insertions(+)
diff --git a/src/paimon/core/casting/cast_executor.h
b/src/paimon/core/casting/cast_executor.h
new file mode 100644
index 0000000..b10e3b0
--- /dev/null
+++ b/src/paimon/core/casting/cast_executor.h
@@ -0,0 +1,45 @@
+/*
+ * 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/api.h"
+#include "arrow/array/array_base.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class DataType;
+class MemoryPool;
+} // namespace arrow
+
+namespace paimon {
+class CastExecutor {
+ public:
+ virtual ~CastExecutor() = default;
+ virtual Result<Literal> Cast(const Literal& literal,
+ const std::shared_ptr<arrow::DataType>&
target_type) const = 0;
+ virtual Result<std::shared_ptr<arrow::Array>> Cast(
+ const std::shared_ptr<arrow::Array>& array,
+ const std::shared_ptr<arrow::DataType>& target_type,
arrow::MemoryPool* pool) const = 0;
+};
+} // namespace paimon
diff --git a/src/paimon/core/casting/cast_executor_factory.cpp
b/src/paimon/core/casting/cast_executor_factory.cpp
new file mode 100644
index 0000000..2c0a411
--- /dev/null
+++ b/src/paimon/core/casting/cast_executor_factory.cpp
@@ -0,0 +1,191 @@
+/*
+ * 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/casting/cast_executor_factory.h"
+
+#include <utility>
+
+#include "paimon/core/casting/binary_to_string_cast_executor.h"
+#include "paimon/core/casting/boolean_to_decimal_cast_executor.h"
+#include "paimon/core/casting/boolean_to_numeric_cast_executor.h"
+#include "paimon/core/casting/boolean_to_string_cast_executor.h"
+#include "paimon/core/casting/date_to_string_cast_executor.h"
+#include "paimon/core/casting/date_to_timestamp_cast_executor.h"
+#include "paimon/core/casting/decimal_to_decimal_cast_executor.h"
+#include "paimon/core/casting/decimal_to_numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/numeric_primitive_to_decimal_cast_executor.h"
+#include "paimon/core/casting/numeric_primitive_to_timestamp_cast_executor.h"
+#include "paimon/core/casting/numeric_to_boolean_cast_executor.h"
+#include "paimon/core/casting/numeric_to_string_cast_executor.h"
+#include "paimon/core/casting/string_to_binary_cast_executor.h"
+#include "paimon/core/casting/string_to_boolean_cast_executor.h"
+#include "paimon/core/casting/string_to_date_cast_executor.h"
+#include "paimon/core/casting/string_to_decimal_cast_executor.h"
+#include "paimon/core/casting/string_to_numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/string_to_timestamp_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_date_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_string_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_timestamp_cast_executor.h"
+#include "paimon/defs.h"
+
+namespace paimon {
+#define REGISTER_CAST_EXECUTOR(TARGET, SRC, EXECUTOR) \
+ executor_map_[FieldType::TARGET][FieldType::SRC] =
std::make_shared<EXECUTOR>();
+
+CastExecutorFactory* CastExecutorFactory::GetCastExecutorFactory() {
+ static std::unique_ptr<CastExecutorFactory> executor_factory =
+ std::unique_ptr<CastExecutorFactory>(new CastExecutorFactory());
+ return executor_factory.get();
+}
+
+std::shared_ptr<CastExecutor> CastExecutorFactory::GetCastExecutor(const
FieldType& src,
+ const
FieldType& target) const {
+ auto target_iter = executor_map_.find(target);
+ if (target_iter == executor_map_.end()) {
+ return nullptr;
+ }
+ auto src_iter = target_iter->second.find(src);
+ if (src_iter == target_iter->second.end()) {
+ return nullptr;
+ }
+ return src_iter->second;
+}
+
+CastExecutorFactory::CastExecutorFactory() {
+ REGISTER_CAST_EXECUTOR(TINYINT, BOOLEAN, BooleanToNumericCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, BOOLEAN, BooleanToNumericCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, BOOLEAN, BooleanToNumericCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, BOOLEAN, BooleanToNumericCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, BOOLEAN, BooleanToNumericCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, BOOLEAN, BooleanToNumericCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(DECIMAL, BOOLEAN, BooleanToDecimalCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(STRING, BOOLEAN, BooleanToStringCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(TINYINT, TINYINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(TINYINT, SMALLINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(TINYINT, INT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(TINYINT, BIGINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(TINYINT, FLOAT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(TINYINT, DOUBLE, NumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(SMALLINT, TINYINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, SMALLINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, INT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, BIGINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, FLOAT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, DOUBLE, NumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(INT, TINYINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, SMALLINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, INT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, BIGINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, FLOAT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, DOUBLE, NumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(BIGINT, TINYINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, SMALLINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, INT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, BIGINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, FLOAT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, DOUBLE, NumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(FLOAT, TINYINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, SMALLINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, INT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, BIGINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, FLOAT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, DOUBLE, NumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(DOUBLE, TINYINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, SMALLINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, INT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, BIGINT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, FLOAT, NumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, DOUBLE, NumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(BOOLEAN, TINYINT, NumericToBooleanCastExecutor);
+ REGISTER_CAST_EXECUTOR(BOOLEAN, SMALLINT, NumericToBooleanCastExecutor);
+ REGISTER_CAST_EXECUTOR(BOOLEAN, INT, NumericToBooleanCastExecutor);
+ REGISTER_CAST_EXECUTOR(BOOLEAN, BIGINT, NumericToBooleanCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(BOOLEAN, STRING, StringToBooleanCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(TINYINT, STRING,
StringToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, STRING,
StringToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, STRING, StringToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, STRING,
StringToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, STRING,
StringToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, STRING,
StringToNumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(STRING, TINYINT, NumericToStringCastExecutor);
+ REGISTER_CAST_EXECUTOR(STRING, SMALLINT, NumericToStringCastExecutor);
+ REGISTER_CAST_EXECUTOR(STRING, INT, NumericToStringCastExecutor);
+ REGISTER_CAST_EXECUTOR(STRING, BIGINT, NumericToStringCastExecutor);
+ REGISTER_CAST_EXECUTOR(STRING, FLOAT, NumericToStringCastExecutor);
+ REGISTER_CAST_EXECUTOR(STRING, DOUBLE, NumericToStringCastExecutor);
+ REGISTER_CAST_EXECUTOR(STRING, DECIMAL, NumericToStringCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(BINARY, STRING, StringToBinaryCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(STRING, BINARY, BinaryToStringCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(STRING, DATE, DateToStringCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(TIMESTAMP, DATE, DateToTimestampCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(TIMESTAMP, INT,
NumericPrimitiveToTimestampCastExecutor);
+ REGISTER_CAST_EXECUTOR(TIMESTAMP, BIGINT,
NumericPrimitiveToTimestampCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(DATE, STRING, StringToDateCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(STRING, TIMESTAMP, TimestampToStringCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(DATE, TIMESTAMP, TimestampToDateCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(INT, TIMESTAMP,
TimestampToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, TIMESTAMP,
TimestampToNumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(TIMESTAMP, STRING, StringToTimestampCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(TINYINT, DECIMAL,
DecimalToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(SMALLINT, DECIMAL,
DecimalToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(INT, DECIMAL,
DecimalToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(BIGINT, DECIMAL,
DecimalToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(FLOAT, DECIMAL,
DecimalToNumericPrimitiveCastExecutor);
+ REGISTER_CAST_EXECUTOR(DOUBLE, DECIMAL,
DecimalToNumericPrimitiveCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(DECIMAL, TINYINT,
NumericPrimitiveToDecimalCastExecutor);
+ REGISTER_CAST_EXECUTOR(DECIMAL, SMALLINT,
NumericPrimitiveToDecimalCastExecutor);
+ REGISTER_CAST_EXECUTOR(DECIMAL, INT,
NumericPrimitiveToDecimalCastExecutor);
+ REGISTER_CAST_EXECUTOR(DECIMAL, BIGINT,
NumericPrimitiveToDecimalCastExecutor);
+ REGISTER_CAST_EXECUTOR(DECIMAL, FLOAT,
NumericPrimitiveToDecimalCastExecutor);
+ REGISTER_CAST_EXECUTOR(DECIMAL, DOUBLE,
NumericPrimitiveToDecimalCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(DECIMAL, STRING, StringToDecimalCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(DECIMAL, DECIMAL, DecimalToDecimalCastExecutor);
+
+ REGISTER_CAST_EXECUTOR(TIMESTAMP, TIMESTAMP,
TimestampToTimestampCastExecutor);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/casting/cast_executor_factory.h
b/src/paimon/core/casting/cast_executor_factory.h
new file mode 100644
index 0000000..50f70b0
--- /dev/null
+++ b/src/paimon/core/casting/cast_executor_factory.h
@@ -0,0 +1,45 @@
+/*
+ * 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 <map>
+#include <memory>
+
+#include "paimon/core/casting/cast_executor.h"
+#include "paimon/defs.h"
+
+namespace paimon {
+enum class FieldType;
+
+class CastExecutorFactory {
+ public:
+ static CastExecutorFactory* GetCastExecutorFactory();
+
+ std::shared_ptr<CastExecutor> GetCastExecutor(const FieldType& src,
+ const FieldType& target)
const;
+
+ private:
+ CastExecutorFactory();
+
+ private:
+ // {target type: {src type: cast executor}}
+ std::map<FieldType, std::map<FieldType, std::shared_ptr<CastExecutor>>>
executor_map_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/casting/cast_executor_factory_test.cpp
b/src/paimon/core/casting/cast_executor_factory_test.cpp
new file mode 100644
index 0000000..56acbe4
--- /dev/null
+++ b/src/paimon/core/casting/cast_executor_factory_test.cpp
@@ -0,0 +1,236 @@
+/*
+ * 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/casting/cast_executor_factory.h"
+
+#include "gtest/gtest.h"
+#include "paimon/core/casting/binary_to_string_cast_executor.h"
+#include "paimon/core/casting/boolean_to_decimal_cast_executor.h"
+#include "paimon/core/casting/boolean_to_numeric_cast_executor.h"
+#include "paimon/core/casting/boolean_to_string_cast_executor.h"
+#include "paimon/core/casting/date_to_string_cast_executor.h"
+#include "paimon/core/casting/date_to_timestamp_cast_executor.h"
+#include "paimon/core/casting/decimal_to_decimal_cast_executor.h"
+#include "paimon/core/casting/decimal_to_numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/numeric_primitive_to_decimal_cast_executor.h"
+#include "paimon/core/casting/numeric_primitive_to_timestamp_cast_executor.h"
+#include "paimon/core/casting/numeric_to_boolean_cast_executor.h"
+#include "paimon/core/casting/numeric_to_string_cast_executor.h"
+#include "paimon/core/casting/string_to_binary_cast_executor.h"
+#include "paimon/core/casting/string_to_boolean_cast_executor.h"
+#include "paimon/core/casting/string_to_date_cast_executor.h"
+#include "paimon/core/casting/string_to_decimal_cast_executor.h"
+#include "paimon/core/casting/string_to_numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/string_to_timestamp_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_date_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_numeric_primitive_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_string_cast_executor.h"
+#include "paimon/core/casting/timestamp_to_timestamp_cast_executor.h"
+#include "paimon/defs.h"
+
+namespace paimon::test {
+TEST(CastExecutorFactoryTest, TestRegister) {
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::TINYINT,
FieldType::INT);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<NumericPrimitiveCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::INT,
FieldType::BIGINT);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<NumericPrimitiveCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::DECIMAL,
FieldType::TIMESTAMP);
+ ASSERT_FALSE(cast_executor);
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::BOOLEAN,
FieldType::TINYINT);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<BooleanToNumericCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::BOOLEAN,
FieldType::STRING);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<BooleanToStringCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::INT,
FieldType::BOOLEAN);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<NumericToBooleanCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::STRING,
FieldType::BOOLEAN);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<StringToBooleanCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::STRING,
FieldType::BIGINT);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<StringToNumericPrimitiveCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::FLOAT,
FieldType::STRING);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<NumericToStringCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::STRING,
FieldType::BINARY);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<StringToBinaryCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::BINARY,
FieldType::STRING);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<BinaryToStringCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::DATE,
FieldType::STRING);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<DateToStringCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::DATE,
FieldType::TIMESTAMP);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<DateToTimestampCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::INT,
FieldType::TIMESTAMP);
+ ASSERT_TRUE(cast_executor);
+ ASSERT_TRUE(
+
std::dynamic_pointer_cast<NumericPrimitiveToTimestampCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::STRING,
FieldType::DATE);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<StringToDateCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::TIMESTAMP,
FieldType::STRING);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<TimestampToStringCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::TIMESTAMP,
FieldType::DATE);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<TimestampToDateCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::TIMESTAMP,
FieldType::BIGINT);
+ ASSERT_TRUE(cast_executor);
+ ASSERT_TRUE(
+
std::dynamic_pointer_cast<TimestampToNumericPrimitiveCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::STRING,
FieldType::TIMESTAMP);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<StringToTimestampCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::DECIMAL,
FieldType::TINYINT);
+ ASSERT_TRUE(cast_executor);
+ ASSERT_TRUE(
+
std::dynamic_pointer_cast<DecimalToNumericPrimitiveCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::BIGINT,
FieldType::DECIMAL);
+ ASSERT_TRUE(cast_executor);
+ ASSERT_TRUE(
+
std::dynamic_pointer_cast<NumericPrimitiveToDecimalCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::STRING,
FieldType::DECIMAL);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<StringToDecimalCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::DECIMAL,
FieldType::DECIMAL);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<DecimalToDecimalCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::TIMESTAMP,
FieldType::TIMESTAMP);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<TimestampToTimestampCastExecutor>(cast_executor));
+ }
+ {
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::BOOLEAN,
FieldType::DECIMAL);
+ ASSERT_TRUE(cast_executor);
+
ASSERT_TRUE(std::dynamic_pointer_cast<BooleanToDecimalCastExecutor>(cast_executor));
+ }
+ {
+ // test non-exist cast executor
+ auto* factory = CastExecutorFactory::GetCastExecutorFactory();
+ ASSERT_FALSE(factory->executor_map_.empty());
+ auto cast_executor = factory->GetCastExecutor(FieldType::ARRAY,
FieldType::MAP);
+ ASSERT_FALSE(cast_executor);
+ }
+}
+} // namespace paimon::test
diff --git a/src/paimon/core/casting/casted_row.cpp
b/src/paimon/core/casting/casted_row.cpp
new file mode 100644
index 0000000..bd8b424
--- /dev/null
+++ b/src/paimon/core/casting/casted_row.cpp
@@ -0,0 +1,51 @@
+/*
+ * 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/casting/casted_row.h"
+
+#include "arrow/type.h"
+#include "paimon/common/types/data_field.h"
+
+namespace paimon {
+Result<std::unique_ptr<CastedRow>> CastedRow::Create(
+ const std::vector<std::shared_ptr<CastExecutor>>& cast_executors,
+ const std::vector<DataField>& src_fields, const std::vector<DataField>&
target_fields,
+ const std::shared_ptr<InternalRow>& row) {
+ if (src_fields.size() != target_fields.size() || src_fields.size() !=
cast_executors.size() ||
+ src_fields.size() != static_cast<size_t>(row->GetFieldCount())) {
+ return Status::Invalid(
+ "CastedRow create failed, src_fields & target_fields &
cast_executors & row size "
+ "mismatch");
+ }
+ std::vector<InternalRow::FieldGetterFunc> field_getters;
+ field_getters.reserve(src_fields.size());
+ std::vector<arrow::Type::type> src_types;
+ src_types.reserve(src_fields.size());
+ for (size_t i = 0; i < src_fields.size(); ++i) {
+ const auto& type = src_fields[i].Type();
+ PAIMON_ASSIGN_OR_RAISE(InternalRow::FieldGetterFunc getter,
+ InternalRow::CreateFieldGetter(i, type,
/*use_view=*/false));
+ field_getters.push_back(std::move(getter));
+ src_types.push_back(type->id());
+ }
+ return std::unique_ptr<CastedRow>(
+ new CastedRow(cast_executors, std::move(field_getters),
std::move(src_types), row));
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/casting/casted_row.h
b/src/paimon/core/casting/casted_row.h
new file mode 100644
index 0000000..677232d
--- /dev/null
+++ b/src/paimon/core/casting/casted_row.h
@@ -0,0 +1,231 @@
+/*
+ * 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 <cassert>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include "arrow/type_fwd.h"
+#include "fmt/format.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/data_define.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/types/row_kind.h"
+#include "paimon/common/utils/date_time_utils.h"
+#include "paimon/core/casting/cast_executor.h"
+#include "paimon/data/decimal.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+class DataField;
+class InternalArray;
+class InternalMap;
+class Literal;
+class RowKind;
+
+/// An implementation of `InternalRow` which provides a casted view of the
underlying `InternalRow`.
+///
+/// It reads data from underlying `InternalRow` according to source logical
type and casts
+/// it with specific `CastExecutor`.
+/// @note When use CastedRow, we need to catch exceptions
+class CastedRow : public InternalRow {
+ public:
+ static Result<std::unique_ptr<CastedRow>> Create(
+ const std::vector<std::shared_ptr<CastExecutor>>& cast_executors,
+ const std::vector<DataField>& src_fields, const
std::vector<DataField>& target_fields,
+ const std::shared_ptr<InternalRow>& row);
+
+ int32_t GetFieldCount() const override {
+ return row_->GetFieldCount();
+ }
+
+ Result<const RowKind*> GetRowKind() const override {
+ return row_->GetRowKind();
+ }
+
+ void SetRowKind(const RowKind* kind) override {
+ row_->SetRowKind(kind);
+ }
+
+ bool IsNullAt(int32_t pos) const override {
+ return row_->IsNullAt(pos);
+ }
+
+ template <typename T>
+ T GetValue(int32_t pos, const std::shared_ptr<arrow::DataType>&
target_type) const {
+ assert(static_cast<size_t>(pos) < field_getters_.size());
+ auto value = field_getters_[pos](*row_);
+ if (!cast_executors_[pos]) {
+ return DataDefine::GetVariantValue<T>(value);
+ }
+ return CastValue<T>(pos, value, target_type);
+ }
+
+ template <typename T>
+ T CastValue(int32_t pos, const VariantType& value,
+ const std::shared_ptr<arrow::DataType>& target_type) const {
+ auto literal = DataDefine::VariantValueToLiteral(value,
src_types_[pos]);
+ if (!literal.ok()) {
+ throw std::invalid_argument(literal.status().ToString());
+ }
+ Result<Literal> cast_ret = cast_executors_[pos]->Cast(literal.value(),
target_type);
+ if (!cast_ret.ok()) {
+ throw std::invalid_argument(cast_ret.status().ToString());
+ }
+ return cast_ret.value().GetValue<T>();
+ }
+
+ template <typename T>
+ T GetNestedValue(int32_t pos) const {
+ assert(static_cast<size_t>(pos) < field_getters_.size());
+ auto value = field_getters_[pos](*row_);
+ assert(!cast_executors_[pos]);
+ return DataDefine::GetVariantValue<T>(value);
+ }
+
+ bool GetBoolean(int32_t pos) const override {
+ return GetValue<bool>(pos, arrow::boolean());
+ }
+
+ char GetByte(int32_t pos) const override {
+ // TODO(liancheng.lsz): use char rather than int8_t
+ assert(static_cast<size_t>(pos) < field_getters_.size());
+ auto value = field_getters_[pos](*row_);
+ if (!cast_executors_[pos]) {
+ return DataDefine::GetVariantValue<char>(value);
+ }
+ return CastValue<int8_t>(pos, value, arrow::int8());
+ }
+
+ int16_t GetShort(int32_t pos) const override {
+ return GetValue<int16_t>(pos, arrow::int16());
+ }
+
+ int32_t GetInt(int32_t pos) const override {
+ return GetValue<int32_t>(pos, arrow::int32());
+ }
+
+ int32_t GetDate(int32_t pos) const override {
+ return GetValue<int32_t>(pos, arrow::date32());
+ }
+
+ int64_t GetLong(int32_t pos) const override {
+ return GetValue<int64_t>(pos, arrow::int64());
+ }
+
+ float GetFloat(int32_t pos) const override {
+ return GetValue<float>(pos, arrow::float32());
+ }
+
+ double GetDouble(int32_t pos) const override {
+ return GetValue<double>(pos, arrow::float64());
+ }
+
+ BinaryString GetString(int32_t pos) const override {
+ assert(static_cast<size_t>(pos) < field_getters_.size());
+ auto value = field_getters_[pos](*row_);
+ if (!cast_executors_[pos]) {
+ return DataDefine::GetVariantValue<BinaryString>(value);
+ }
+ auto str = CastValue<std::string>(pos, value, arrow::utf8());
+ return BinaryString::FromString(str, GetDefaultPool().get());
+ }
+
+ std::string_view GetStringView(int32_t pos) const override {
+ assert(false);
+ throw std::invalid_argument("cannot get string view in casted row");
+ }
+
+ Decimal GetDecimal(int32_t pos, int32_t precision, int32_t scale) const
override {
+ return GetValue<Decimal>(pos, arrow::decimal128(precision, scale));
+ }
+
+ Timestamp GetTimestamp(int32_t pos, int32_t precision) const override {
+ // timestamp does not support casting
+ Result<std::shared_ptr<arrow::DataType>> ts_type =
+ DateTimeUtils::GetTypeFromPrecision(precision,
/*with_timezone=*/false);
+ if (!ts_type.ok()) {
+ throw std::invalid_argument(ts_type.status().ToString());
+ }
+ return GetValue<Timestamp>(pos, ts_type.value());
+ }
+
+ std::shared_ptr<Bytes> GetBinary(int32_t pos) const override {
+ assert(static_cast<size_t>(pos) < field_getters_.size());
+ auto value = field_getters_[pos](*row_);
+ if (!cast_executors_[pos]) {
+ return DataDefine::GetVariantValue<std::shared_ptr<Bytes>>(value);
+ }
+ auto str = CastValue<std::string>(pos, value, arrow::binary());
+ return std::make_shared<Bytes>(str, GetDefaultPool().get());
+ }
+
+ std::shared_ptr<InternalArray> GetArray(int32_t pos) const override {
+ return GetNestedValue<std::shared_ptr<InternalArray>>(pos);
+ }
+
+ std::shared_ptr<InternalMap> GetMap(int32_t pos) const override {
+ return GetNestedValue<std::shared_ptr<InternalMap>>(pos);
+ }
+
+ std::shared_ptr<InternalRow> GetRow(int32_t pos, int32_t num_fields) const
override {
+ return GetNestedValue<std::shared_ptr<InternalRow>>(pos);
+ }
+
+ std::string ToString() const override {
+ return fmt::format("casted row, inner row = {}", row_->ToString());
+ }
+
+ private:
+ CastedRow(const std::vector<std::shared_ptr<CastExecutor>>& cast_executors,
+ std::vector<InternalRow::FieldGetterFunc>&& field_getters,
+ std::vector<arrow::Type::type>&& src_types, const
std::shared_ptr<InternalRow>& row)
+ : cast_executors_(cast_executors),
+ field_getters_(std::move(field_getters)),
+ src_types_(std::move(src_types)),
+ row_(row) {
+ assert(row_);
+ }
+
+ private:
+ std::vector<std::shared_ptr<CastExecutor>> cast_executors_;
+ std::vector<InternalRow::FieldGetterFunc> field_getters_;
+ std::vector<arrow::Type::type> src_types_;
+ std::shared_ptr<InternalRow> row_;
+};
+} // namespace paimon
diff --git a/src/paimon/core/casting/casted_row_test.cpp
b/src/paimon/core/casting/casted_row_test.cpp
new file mode 100644
index 0000000..4525e7e
--- /dev/null
+++ b/src/paimon/core/casting/casted_row_test.cpp
@@ -0,0 +1,245 @@
+/*
+ * 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/casting/casted_row.h"
+
+#include "arrow/api.h"
+#include "arrow/array/array_nested.h"
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/columnar/columnar_row.h"
+#include "paimon/common/data/internal_array.h"
+#include "paimon/common/data/internal_map.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/types/row_kind.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+TEST(CastedRowTest, TestSimpleWithNoCasting) {
+ auto pool = GetDefaultPool();
+
+ std::vector<DataField> fields = {DataField(0, arrow::field("f0",
arrow::boolean())),
+ DataField(1, arrow::field("f1",
arrow::int8())),
+ DataField(2, arrow::field("f2",
arrow::int16())),
+ DataField(3, arrow::field("f3",
arrow::int32())),
+ DataField(4, arrow::field("field_null",
arrow::int32())),
+ DataField(5, arrow::field("f4",
arrow::int64())),
+ DataField(6, arrow::field("f5",
arrow::float32())),
+ DataField(7, arrow::field("f6",
arrow::float64())),
+ DataField(8, arrow::field("f7",
arrow::utf8())),
+ DataField(9, arrow::field("f8",
arrow::binary()))};
+ auto arrow_type = DataField::ConvertDataFieldsToArrowStructType(fields);
+
+ std::string data =
+ R"([[true, 0, 32767, 2147483647, null, 4294967295, 0.5, 1.141592659,
"2025-03-27", "banana"],
+ [true, -2, -32768, -2147483648, null, -4294967298, 2.0,
3.141592657, "2025-03-26", "mouse"]])";
+ auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow_type,
data).ValueOrDie();
+ ASSERT_TRUE(array);
+ auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(array);
+ ASSERT_TRUE(struct_array);
+ auto row = std::make_shared<ColumnarRow>(struct_array->fields(), pool,
/*row_id=*/0);
+
+ // all null cast executor
+ std::vector<std::shared_ptr<CastExecutor>> cast_executors(fields.size(),
nullptr);
+ ASSERT_OK_AND_ASSIGN(auto casted_row, CastedRow::Create(cast_executors,
/*src_fields=*/fields,
+
/*target_fields=*/fields, row));
+ ASSERT_EQ(casted_row->GetFieldCount(), 10);
+ ASSERT_EQ(casted_row->GetRowKind().value(), RowKind::Insert());
+ ASSERT_EQ(casted_row->GetBoolean(0), true);
+ ASSERT_EQ(casted_row->GetByte(1), static_cast<char>(0));
+ ASSERT_EQ(casted_row->GetShort(2), static_cast<int16_t>(32767));
+ ASSERT_FALSE(casted_row->IsNullAt(3));
+ ASSERT_EQ(casted_row->GetInt(3), static_cast<int32_t>(2147483647));
+ ASSERT_TRUE(casted_row->IsNullAt(4));
+ ASSERT_EQ(casted_row->GetLong(5), static_cast<int64_t>(4294967295));
+ ASSERT_EQ(casted_row->GetFloat(6), static_cast<float>(0.5));
+ ASSERT_EQ(casted_row->GetDouble(7), static_cast<double>(1.141592659));
+ ASSERT_EQ(casted_row->GetString(8).ToString(), "2025-03-27");
+ ASSERT_EQ(*casted_row->GetBinary(9), Bytes("banana", pool.get()));
+
+ ASSERT_EQ("casted row, inner row = ColumnarRow, row_id 0",
casted_row->ToString());
+}
+
+TEST(CastedRowTest, TestSimpleWithCasting) {
+ auto pool = GetDefaultPool();
+
+ std::vector<DataField> fields = {
+ DataField(0, arrow::field("f0", arrow::boolean())),
+ DataField(1, arrow::field("f1", arrow::int8())),
+ DataField(2, arrow::field("f2", arrow::int16())),
+ DataField(3, arrow::field("f3", arrow::int32())),
+ DataField(4, arrow::field("field_null", arrow::int32())),
+ DataField(5, arrow::field("f4", arrow::int64())),
+ DataField(6, arrow::field("f5", arrow::float32())),
+ DataField(7, arrow::field("f6", arrow::float64())),
+ DataField(8, arrow::field("f7",
arrow::timestamp(arrow::TimeUnit::SECOND))),
+ DataField(9, arrow::field("f8", arrow::binary())),
+ DataField(10, arrow::field("f9", arrow::int16()))};
+ auto arrow_type = DataField::ConvertDataFieldsToArrowStructType(fields);
+
+ std::string data =
+ R"([[true, 0, 32767, 2147483647, null, 4294967295, 0.5, 1.141592659,
"2025-03-27", "banana", 5],
+ [true, -2, -32768, -2147483648, null, -4294967298, 2.0,
3.141592657, "2025-03-26", "mouse", 2]])";
+ auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow_type,
data).ValueOrDie();
+ ASSERT_TRUE(array);
+ auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(array);
+ ASSERT_TRUE(struct_array);
+ auto row = std::make_shared<ColumnarRow>(struct_array->fields(), pool,
/*row_id=*/0);
+
+ // create cast executor
+ std::vector<DataField> target_fields = {
+ DataField(0, arrow::field("f0", arrow::utf8())),
+ DataField(1, arrow::field("f1", arrow::utf8())),
+ DataField(2, arrow::field("f2", arrow::utf8())),
+ DataField(3, arrow::field("f3", arrow::utf8())),
+ DataField(4, arrow::field("field_null", arrow::utf8())),
+ DataField(5, arrow::field("f4", arrow::utf8())),
+ DataField(6, arrow::field("f5", arrow::utf8())),
+ DataField(7, arrow::field("f6", arrow::utf8())),
+ DataField(8, arrow::field("f7",
arrow::timestamp(arrow::TimeUnit::SECOND))),
+ DataField(9, arrow::field("f8", arrow::utf8())),
+ DataField(10, arrow::field("f9", arrow::int8()))};
+
+ ASSERT_OK_AND_ASSIGN(auto cast_executors,
+
FieldMappingBuilder::CreateDataCastExecutors(target_fields, fields));
+ ASSERT_OK_AND_ASSIGN(auto casted_row, CastedRow::Create(cast_executors,
/*src_fields=*/fields,
+
/*target_fields=*/target_fields, row));
+ ASSERT_EQ(casted_row->GetFieldCount(), 11);
+ ASSERT_EQ(casted_row->GetRowKind().value(), RowKind::Insert());
+ ASSERT_EQ(casted_row->GetString(0).ToString(), "true");
+ ASSERT_EQ(casted_row->GetString(1).ToString(), "0");
+ ASSERT_EQ(casted_row->GetString(2).ToString(), "32767");
+ ASSERT_FALSE(casted_row->IsNullAt(3));
+ ASSERT_EQ(casted_row->GetString(3).ToString(), "2147483647");
+ ASSERT_TRUE(casted_row->IsNullAt(4));
+ ASSERT_EQ(casted_row->GetString(5).ToString(), "4294967295");
+ ASSERT_EQ(casted_row->GetString(6).ToString(), "0.5");
+ ASSERT_EQ(casted_row->GetString(7).ToString(), "1.141592659");
+ ASSERT_EQ(casted_row->GetTimestamp(8, /*precision=*/0),
Timestamp(1743033600000l, 0l))
+ << casted_row->GetTimestamp(8, /*precision=*/0).ToString();
+ ASSERT_EQ(casted_row->GetString(9).ToString(), "banana");
+ ASSERT_EQ(casted_row->GetByte(10), 5);
+
+ ASSERT_EQ("casted row, inner row = ColumnarRow, row_id 0",
casted_row->ToString());
+}
+
+TEST(CastedRowTest, TestNestedTypeWithCasting) {
+ auto pool = GetDefaultPool();
+ std::vector<DataField> fields = {
+ DataField(0, arrow::field("f1", arrow::map(arrow::int8(),
arrow::int16()))),
+ DataField(1, arrow::field("f2", arrow::list(arrow::float32()))),
+ DataField(2, arrow::field("f3", arrow::struct_({arrow::field("f0",
arrow::boolean()),
+ arrow::field("f1",
arrow::int64())}))),
+ DataField(3, arrow::field("f4",
arrow::timestamp(arrow::TimeUnit::NANO, "Asia/Shanghai"))),
+ DataField(4, arrow::field("f5", arrow::date32())),
+ DataField(5, arrow::field("f6", arrow::decimal128(2, 2)))};
+
+ auto arrow_type = DataField::ConvertDataFieldsToArrowStructType(fields);
+ std::string data = R"([
+ [[[10, 20]], [0.1, 0.2], [true, 2], "1970-01-01 00:02:03.123123",
2456, "0.22"],
+ [[[11, 64], [12, 32]], [2.2, 3.2], [true, 2], "1970-01-01
00:00:00.123123", 24, "0.78"]
+ ])";
+
+ auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow_type,
data).ValueOrDie();
+ ASSERT_TRUE(array);
+ auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(array);
+ ASSERT_TRUE(struct_array);
+ auto row = std::make_shared<ColumnarRow>(struct_array->fields(), pool,
/*row_id=*/0);
+
+ // create cast executor
+ std::vector<DataField> target_fields = {
+ fields[0],
+ fields[1],
+ fields[2],
+ DataField(3, arrow::field("f4",
arrow::timestamp(arrow::TimeUnit::NANO, "Asia/Shanghai"))),
+ DataField(4, arrow::field("f5", arrow::utf8())),
+ DataField(5, arrow::field("f6", arrow::utf8()))};
+
+ ASSERT_OK_AND_ASSIGN(auto cast_executors,
+
FieldMappingBuilder::CreateDataCastExecutors(target_fields, fields));
+ ASSERT_OK_AND_ASSIGN(auto casted_row, CastedRow::Create(cast_executors,
/*src_fields=*/fields,
+
/*target_fields=*/target_fields, row));
+ ASSERT_EQ(casted_row->GetFieldCount(), 6);
+ ASSERT_EQ(casted_row->GetRowKind().value(), RowKind::Insert());
+
+ ASSERT_EQ(casted_row->GetMap(0)->KeyArray()->ToByteArray().value(),
std::vector<char>({10}));
+ ASSERT_EQ(casted_row->GetMap(0)->ValueArray()->ToShortArray().value(),
+ std::vector<int16_t>({20}));
+
+ ASSERT_EQ(casted_row->GetArray(1)->ToFloatArray().value(),
std::vector<float>({0.1, 0.2}));
+
+ auto inner_row = casted_row->GetRow(2, 2);
+ ASSERT_EQ(inner_row->GetBoolean(0), true);
+ ASSERT_EQ(inner_row->GetLong(1), 2l);
+
+ ASSERT_FALSE(casted_row->IsNullAt(3));
+ ASSERT_EQ(casted_row->GetTimestamp(3, /*precision=*/9).ToString(),
+ "1970-01-01 00:02:03.123123000");
+ ASSERT_EQ(casted_row->GetString(4).ToString(), "1976-09-22");
+ ASSERT_EQ(casted_row->GetString(5).ToString(), "0.22");
+
+ ASSERT_EQ("casted row, inner row = ColumnarRow, row_id 0",
casted_row->ToString());
+}
+
+TEST(CastedRowTest, TestInvalidCast) {
+ auto pool = GetDefaultPool();
+
+ std::vector<DataField> fields = {DataField(0, arrow::field("f0",
arrow::utf8())),
+ DataField(1, arrow::field("f1",
arrow::utf8())),
+ DataField(2, arrow::field("f2",
arrow::utf8()))};
+ auto arrow_type = DataField::ConvertDataFieldsToArrowStructType(fields);
+
+ std::string data = R"([["apple", "noo", "2024-11-21T09:91:56.1"]])";
+ auto array = arrow::ipc::internal::json::ArrayFromJSON(arrow_type,
data).ValueOrDie();
+ ASSERT_TRUE(array);
+ auto struct_array = std::dynamic_pointer_cast<arrow::StructArray>(array);
+ ASSERT_TRUE(struct_array);
+ auto row = std::make_shared<ColumnarRow>(struct_array->fields(), pool,
/*row_id=*/0);
+
+ // create cast executor
+ std::vector<DataField> target_fields = {
+ DataField(0, arrow::field("f0", arrow::binary())),
+ DataField(1, arrow::field("f1", arrow::boolean())),
+ DataField(2, arrow::field("f2",
arrow::timestamp(arrow::TimeUnit::NANO)))};
+
+ ASSERT_OK_AND_ASSIGN(auto cast_executors,
+
FieldMappingBuilder::CreateDataCastExecutors(target_fields, fields));
+ ASSERT_OK_AND_ASSIGN(auto casted_row, CastedRow::Create(cast_executors,
/*src_fields=*/fields,
+
/*target_fields=*/target_fields, row));
+ ASSERT_EQ(casted_row->GetFieldCount(), 3);
+ ASSERT_EQ(casted_row->GetRowKind().value(), RowKind::Insert());
+
+ Bytes f0_bytes("apple", pool.get());
+ ASSERT_EQ(*casted_row->GetBinary(0), f0_bytes);
+ ASSERT_THROW(casted_row->GetBoolean(1), std::invalid_argument);
+ ASSERT_THROW(casted_row->GetTimestamp(2, 9), std::invalid_argument);
+}
+
+TEST(CastedRowTest, TestInvalidCastedRowCreate) {
+ std::vector<DataField> fields = {DataField(0, arrow::field("f0",
arrow::utf8()))};
+ // cast_executors.size() != fields.size()
+ ASSERT_NOK_WITH_MSG(
+ CastedRow::Create(/*cast_executors=*/{}, /*src_fields=*/fields,
+ /*target_fields=*/fields, nullptr),
+ "CastedRow create failed, src_fields & target_fields & cast_executors
& row size mismatch");
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/casting/casting_utils.cpp
b/src/paimon/core/casting/casting_utils.cpp
new file mode 100644
index 0000000..c5820f9
--- /dev/null
+++ b/src/paimon/core/casting/casting_utils.cpp
@@ -0,0 +1,98 @@
+/*
+ * 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/casting/casting_utils.h"
+
+#include <memory>
+
+namespace paimon {
+Result<std::shared_ptr<arrow::Array>> CastingUtils::Cast(
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::DataType>& target_type, const
arrow::compute::CastOptions& options,
+ arrow::MemoryPool* pool) {
+ auto src_type = src_array->type();
+ if (!arrow::compute::CanCast(*src_type, *target_type)) {
+ return Status::Invalid(fmt::format("cast arrow array failed: cannot
cast from {} to {}",
+ src_type->ToString(),
target_type->ToString()));
+ }
+ arrow::compute::ExecContext ctx(pool);
+ arrow::TypeHolder type_holder(target_type.get());
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array>
casted_array,
+ arrow::compute::Cast(*src_array,
type_holder, options, &ctx));
+ return casted_array;
+}
+
+Result<std::shared_ptr<arrow::Array>>
CastingUtils::TimestampToTimestampWithTimezone(
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::TimestampType>& target_type,
arrow::MemoryPool* pool) {
+ auto src_ts_type =
+
arrow::internal::checked_pointer_cast<arrow::TimestampType>(src_array->type());
+ assert(src_ts_type);
+ if (src_ts_type->unit() != target_type->unit()) {
+ return Status::Invalid("in timezone converter, time unit of src and
target type mismatch");
+ }
+ if (!src_ts_type->timezone().empty() || target_type->timezone().empty()) {
+ return Status::Invalid(
+ "in TimestampToTimestampWithTimezone, src value must be local time
(no tz), target "
+ "value must be UTC (with tz)");
+ }
+ arrow::compute::ExecContext ctx(pool);
+ arrow::compute::AssumeTimezoneOptions options(target_type->timezone());
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow::Datum target_array,
+ arrow::compute::AssumeTimezone(arrow::Datum(src_array), options,
&ctx));
+ return target_array.make_array();
+}
+
+Result<std::shared_ptr<arrow::Array>>
CastingUtils::TimestampWithTimezoneToTimestamp(
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::TimestampType>& target_type,
arrow::MemoryPool* pool) {
+ auto src_ts_type =
+
arrow::internal::checked_pointer_cast<arrow::TimestampType>(src_array->type());
+ assert(src_ts_type);
+ if (src_ts_type->unit() != target_type->unit()) {
+ return Status::Invalid("in timezone converter, time unit of src and
target type mismatch");
+ }
+ if (src_ts_type->timezone().empty() || !target_type->timezone().empty()) {
+ return Status::Invalid(
+ "in TimestampWithTimezoneToTimestamp, src value must be UTC (with
tz), target value "
+ "must be local time (no tz)");
+ }
+ arrow::compute::ExecContext ctx(pool);
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow::Datum target_array,
arrow::compute::LocalTimestamp(arrow::Datum(src_array), &ctx));
+ return target_array.make_array();
+}
+
+int64_t CastingUtils::GetLongValueFromLiteral(const Literal& literal) {
+ auto type = literal.GetType();
+ switch (type) {
+ case FieldType::TINYINT:
+ return static_cast<int64_t>(literal.GetValue<int8_t>());
+ case FieldType::SMALLINT:
+ return static_cast<int64_t>(literal.GetValue<int16_t>());
+ case FieldType::INT:
+ return static_cast<int64_t>(literal.GetValue<int32_t>());
+ case FieldType::BIGINT:
+ return literal.GetValue<int64_t>();
+ default:
+ return -1;
+ }
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/casting/casting_utils.h
b/src/paimon/core/casting/casting_utils.h
new file mode 100644
index 0000000..453abf8
--- /dev/null
+++ b/src/paimon/core/casting/casting_utils.h
@@ -0,0 +1,105 @@
+/*
+ * 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 <string>
+
+#include "arrow/compute/api.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/field_type_utils.h"
+#include "paimon/data/decimal.h"
+#include "paimon/predicate/literal.h"
+
+namespace paimon {
+class CastingUtils {
+ public:
+ CastingUtils() = delete;
+ ~CastingUtils() = delete;
+
+ // make sure src and casted literals are both integer (i.e., TINYINT,
SMALLINT, INT, BIGINT)
+ static bool IsIntegerLiteralCastedOverflow(const Literal& src_literal,
+ const Literal& casted_literal) {
+ return GetLongValueFromLiteral(src_literal) !=
GetLongValueFromLiteral(casted_literal);
+ }
+
+ static Result<std::shared_ptr<arrow::Array>> Cast(
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::DataType>& target_type,
+ const arrow::compute::CastOptions& options, arrow::MemoryPool* pool);
+
+ template <typename SrcScalar, typename SrcDataType, typename TargetScalar,
+ typename TargetDataType>
+ static Result<Literal> Cast(const Literal& literal,
+ const std::shared_ptr<arrow::DataType>&
src_type,
+ const std::shared_ptr<arrow::DataType>&
target_type,
+ const arrow::compute::CastOptions& options) {
+ PAIMON_ASSIGN_OR_RAISE(FieldType target_field_type,
+
FieldTypeUtils::ConvertToFieldType(target_type->id()));
+ if (literal.IsNull()) {
+ return Literal(target_field_type);
+ }
+ auto src_value = literal.GetValue<SrcDataType>();
+ if (!arrow::compute::CanCast(*src_type, *target_type)) {
+ return Status::Invalid(fmt::format("cast literal failed: cannot
cast from {} to {}",
+ src_type->ToString(),
target_type->ToString()));
+ }
+ std::shared_ptr<arrow::Scalar> src_scalar;
+ if constexpr (std::is_same_v<SrcDataType, Decimal>) {
+ src_scalar = std::make_shared<SrcScalar>(
+ arrow::Decimal128(src_value.HighBits(), src_value.LowBits()),
src_type);
+ } else {
+ src_scalar = std::make_shared<SrcScalar>(src_value);
+ }
+ arrow::TypeHolder type_holder(target_type.get());
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ arrow::Datum casted_result,
+ arrow::compute::Cast(arrow::Datum(src_scalar), type_holder,
options));
+ auto* casted_scalar =
+
arrow::internal::checked_cast<TargetScalar*>(casted_result.scalar().get());
+ if (!casted_scalar) {
+ return Status::Invalid(fmt::format("cast literal failed: cannot
cast to {} scalar",
+ target_type->ToString()));
+ }
+ if constexpr (std::is_same_v<TargetDataType, std::string>) {
+ std::string_view casted_data = casted_scalar->view();
+ return Literal(FieldType::STRING, casted_data.data(),
casted_data.size());
+ } else {
+ const auto* casted_data = static_cast<const
TargetDataType*>(casted_scalar->data());
+ assert(casted_data);
+ return Literal(*casted_data);
+ }
+ }
+
+ // src_array is local time (no tz), target type is utc (with tz)
+ static Result<std::shared_ptr<arrow::Array>>
TimestampToTimestampWithTimezone(
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::TimestampType>& target_type,
arrow::MemoryPool* pool);
+
+ // src_array is utc (with tz) , target type is local time (no tz),
+ static Result<std::shared_ptr<arrow::Array>>
TimestampWithTimezoneToTimestamp(
+ const std::shared_ptr<arrow::Array>& src_array,
+ const std::shared_ptr<arrow::TimestampType>& target_type,
arrow::MemoryPool* pool);
+
+ private:
+ static int64_t GetLongValueFromLiteral(const Literal& literal);
+};
+} // namespace paimon
diff --git a/src/paimon/core/casting/casting_utils_test.cpp
b/src/paimon/core/casting/casting_utils_test.cpp
new file mode 100644
index 0000000..861789c
--- /dev/null
+++ b/src/paimon/core/casting/casting_utils_test.cpp
@@ -0,0 +1,165 @@
+/*
+ * 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/casting/casting_utils.h"
+
+#include <memory>
+
+#include "arrow/ipc/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+class CastingUtilsTest : public ::testing::Test {
+ std::shared_ptr<arrow::MemoryPool> arrow_pool_ =
GetArrowPool(GetDefaultPool());
+};
+
+TEST_F(CastingUtilsTest, TestDictionaryToString) {
+ auto dict =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::utf8(), R"(["foo",
"bar", "bazr"])")
+ .ValueOrDie();
+ auto dict_type = arrow::dictionary(arrow::int32(), arrow::utf8());
+ auto indices =
+ arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), "[1, 2, 0,
2, 0]").ValueOrDie();
+ std::shared_ptr<arrow::DictionaryArray> dict_array =
+ std::make_shared<arrow::DictionaryArray>(dict_type, indices, dict);
+ auto string_array = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::utf8(), R"(["bar", "bazr", "foo", "bazr",
"foo"])")
+ .ValueOrDie();
+
+ auto pool = GetArrowPool(GetDefaultPool());
+ arrow::compute::CastOptions options = arrow::compute::CastOptions::Safe();
+ ASSERT_OK_AND_ASSIGN(
+ auto result_array,
+ CastingUtils::Cast(dict_array, /*target_type=*/arrow::utf8(), options,
pool.get()));
+ ASSERT_TRUE(result_array->Equals(string_array));
+}
+
+TEST_F(CastingUtilsTest, TestTimestampToTimestampWithTimezone) {
+ // local no tz -> utc tz
+ auto src_array = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::timestamp(arrow::TimeUnit::SECOND),
R"(["1970-01-01 00:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(src_array);
+ auto target_type = arrow::timestamp(arrow::TimeUnit::SECOND,
"Asia/Shanghai");
+ auto target_ts_type =
arrow::internal::checked_pointer_cast<arrow::TimestampType>(target_type);
+ auto target_array =
+ arrow::ipc::internal::json::ArrayFromJSON(target_type, R"(["1969-12-31
16:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(target_array);
+ ASSERT_OK_AND_ASSIGN(auto result_array,
CastingUtils::TimestampToTimestampWithTimezone(
+ src_array, target_ts_type,
arrow_pool_.get()));
+ ASSERT_TRUE(target_array->Equals(result_array));
+}
+
+TEST_F(CastingUtilsTest, TestTimestampToTimestampWithTimezoneInvalid) {
+ // local no tz -> utc tz
+ {
+ auto src_array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::timestamp(arrow::TimeUnit::SECOND),
+ R"(["1970-01-01
00:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(src_array);
+ auto target_type = arrow::timestamp(arrow::TimeUnit::NANO,
"Asia/Shanghai");
+ auto target_ts_type =
+
arrow::internal::checked_pointer_cast<arrow::TimestampType>(target_type);
+ ASSERT_NOK_WITH_MSG(CastingUtils::TimestampToTimestampWithTimezone(
+ src_array, target_ts_type, arrow_pool_.get()),
+ "time unit of src and target type mismatch");
+ }
+ {
+ auto src_array =
+
arrow::ipc::internal::json::ArrayFromJSON(arrow::timestamp(arrow::TimeUnit::SECOND),
+ R"(["1970-01-01
00:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(src_array);
+ auto target_type = arrow::timestamp(arrow::TimeUnit::SECOND);
+ auto target_ts_type =
+
arrow::internal::checked_pointer_cast<arrow::TimestampType>(target_type);
+ ASSERT_NOK_WITH_MSG(
+ CastingUtils::TimestampToTimestampWithTimezone(src_array,
target_ts_type,
+ arrow_pool_.get()),
+ "src value must be local time (no tz), target value must be UTC
(with tz)");
+ }
+ {
+ auto src_array = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::timestamp(arrow::TimeUnit::SECOND),
+ R"(["2015-03-29 02:30:00", "2015-03-29
03:30:00"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(src_array);
+ auto target_type = arrow::timestamp(arrow::TimeUnit::SECOND,
"Europe/Warsaw");
+ auto target_ts_type =
+
arrow::internal::checked_pointer_cast<arrow::TimestampType>(target_type);
+ ASSERT_NOK_WITH_MSG(CastingUtils::TimestampToTimestampWithTimezone(
+ src_array, target_ts_type, arrow_pool_.get()),
+ "Timestamp doesn't exist in timezone
'Europe/Warsaw': 2015-03-29 "
+ "02:30:00 is in a gap between");
+ }
+}
+
+TEST_F(CastingUtilsTest, TestTimestampWithTimezoneToTimestamp) {
+ // utc tz -> local no tz
+ auto src_array = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::timestamp(arrow::TimeUnit::SECOND,
"Asia/Shanghai"),
+ R"(["1970-01-01 00:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(src_array);
+ auto target_type = arrow::timestamp(arrow::TimeUnit::SECOND);
+ auto target_ts_type =
arrow::internal::checked_pointer_cast<arrow::TimestampType>(target_type);
+ auto target_array =
+ arrow::ipc::internal::json::ArrayFromJSON(target_type, R"(["1970-01-01
08:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(target_array);
+ ASSERT_OK_AND_ASSIGN(auto result_array,
CastingUtils::TimestampWithTimezoneToTimestamp(
+ src_array, target_ts_type,
arrow_pool_.get()));
+ ASSERT_TRUE(target_array->Equals(result_array));
+}
+
+TEST_F(CastingUtilsTest, TestTimestampWithTimezoneToTimestampInvalid) {
+ // utc tz -> local no tz
+ {
+ auto src_array = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::timestamp(arrow::TimeUnit::NANO,
"Asia/Shanghai"),
+ R"(["1970-01-01 00:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(src_array);
+ auto target_type = arrow::timestamp(arrow::TimeUnit::SECOND);
+ auto target_ts_type =
+
arrow::internal::checked_pointer_cast<arrow::TimestampType>(target_type);
+ ASSERT_NOK_WITH_MSG(CastingUtils::TimestampWithTimezoneToTimestamp(
+ src_array, target_ts_type, arrow_pool_.get()),
+ "in timezone converter, time unit of src and
target type mismatch");
+ }
+ {
+ auto src_array = arrow::ipc::internal::json::ArrayFromJSON(
+ arrow::timestamp(arrow::TimeUnit::SECOND,
"Asia/Shanghai"),
+ R"(["1970-01-01 00:00:01"])")
+ .ValueOr(nullptr);
+ ASSERT_TRUE(src_array);
+ auto target_type = arrow::timestamp(arrow::TimeUnit::SECOND,
"Asia/Tokyo");
+ auto target_ts_type =
+
arrow::internal::checked_pointer_cast<arrow::TimestampType>(target_type);
+ ASSERT_NOK_WITH_MSG(CastingUtils::TimestampWithTimezoneToTimestamp(
+ src_array, target_ts_type, arrow_pool_.get()),
+ "target value must be local time (no tz)");
+ }
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/casting/numeric_primitive_cast_executor.cpp
b/src/paimon/core/casting/numeric_primitive_cast_executor.cpp
new file mode 100644
index 0000000..785189e
--- /dev/null
+++ b/src/paimon/core/casting/numeric_primitive_cast_executor.cpp
@@ -0,0 +1,216 @@
+/*
+ * 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/casting/numeric_primitive_cast_executor.h"
+
+#include <cstdint>
+#include <string>
+
+#include "arrow/compute/cast.h"
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/field_type_utils.h"
+#include "paimon/core/casting/casting_utils.h"
+#include "paimon/defs.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class MemoryPool;
+class Array;
+} // namespace arrow
+
+namespace paimon {
+NumericPrimitiveCastExecutor::NumericPrimitiveCastExecutor() {
+ literal_cast_executor_map_ = {
+ {std::make_pair(FieldType::TINYINT, FieldType::TINYINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int8_t, int8_t>(literal, FieldType::TINYINT);
+ }},
+ {std::make_pair(FieldType::TINYINT, FieldType::SMALLINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int8_t, int16_t>(literal, FieldType::SMALLINT);
+ }},
+ {std::make_pair(FieldType::TINYINT, FieldType::INT),
+ [&](const Literal& literal) {
+ return CastLiteral<int8_t, int32_t>(literal, FieldType::INT);
+ }},
+ {std::make_pair(FieldType::TINYINT, FieldType::BIGINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int8_t, int64_t>(literal, FieldType::BIGINT);
+ }},
+ {std::make_pair(FieldType::TINYINT, FieldType::FLOAT),
+ [&](const Literal& literal) {
+ return CastLiteral<int8_t, float>(literal, FieldType::FLOAT);
+ }},
+ {std::make_pair(FieldType::TINYINT, FieldType::DOUBLE),
+ [&](const Literal& literal) {
+ return CastLiteral<int8_t, double>(literal, FieldType::DOUBLE);
+ }},
+
+ {std::make_pair(FieldType::SMALLINT, FieldType::TINYINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int16_t, int8_t>(literal, FieldType::TINYINT);
+ }},
+ {std::make_pair(FieldType::SMALLINT, FieldType::SMALLINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int16_t, int16_t>(literal,
FieldType::SMALLINT);
+ }},
+ {std::make_pair(FieldType::SMALLINT, FieldType::INT),
+ [&](const Literal& literal) {
+ return CastLiteral<int16_t, int32_t>(literal, FieldType::INT);
+ }},
+ {std::make_pair(FieldType::SMALLINT, FieldType::BIGINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int16_t, int64_t>(literal, FieldType::BIGINT);
+ }},
+ {std::make_pair(FieldType::SMALLINT, FieldType::FLOAT),
+ [&](const Literal& literal) {
+ return CastLiteral<int16_t, float>(literal, FieldType::FLOAT);
+ }},
+ {std::make_pair(FieldType::SMALLINT, FieldType::DOUBLE),
+ [&](const Literal& literal) {
+ return CastLiteral<int16_t, double>(literal, FieldType::DOUBLE);
+ }},
+
+ {std::make_pair(FieldType::INT, FieldType::TINYINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int32_t, int8_t>(literal, FieldType::TINYINT);
+ }},
+ {std::make_pair(FieldType::INT, FieldType::SMALLINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int32_t, int16_t>(literal,
FieldType::SMALLINT);
+ }},
+ {std::make_pair(FieldType::INT, FieldType::INT),
+ [&](const Literal& literal) {
+ return CastLiteral<int32_t, int32_t>(literal, FieldType::INT);
+ }},
+ {std::make_pair(FieldType::INT, FieldType::BIGINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int32_t, int64_t>(literal, FieldType::BIGINT);
+ }},
+ {std::make_pair(FieldType::INT, FieldType::FLOAT),
+ [&](const Literal& literal) {
+ return CastLiteral<int32_t, float>(literal, FieldType::FLOAT);
+ }},
+ {std::make_pair(FieldType::INT, FieldType::DOUBLE),
+ [&](const Literal& literal) {
+ return CastLiteral<int32_t, double>(literal, FieldType::DOUBLE);
+ }},
+
+ {std::make_pair(FieldType::BIGINT, FieldType::TINYINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int64_t, int8_t>(literal, FieldType::TINYINT);
+ }},
+ {std::make_pair(FieldType::BIGINT, FieldType::SMALLINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int64_t, int16_t>(literal,
FieldType::SMALLINT);
+ }},
+ {std::make_pair(FieldType::BIGINT, FieldType::INT),
+ [&](const Literal& literal) {
+ return CastLiteral<int64_t, int32_t>(literal, FieldType::INT);
+ }},
+ {std::make_pair(FieldType::BIGINT, FieldType::BIGINT),
+ [&](const Literal& literal) {
+ return CastLiteral<int64_t, int64_t>(literal, FieldType::BIGINT);
+ }},
+ {std::make_pair(FieldType::BIGINT, FieldType::FLOAT),
+ [&](const Literal& literal) {
+ return CastLiteral<int64_t, float>(literal, FieldType::FLOAT);
+ }},
+ {std::make_pair(FieldType::BIGINT, FieldType::DOUBLE),
+ [&](const Literal& literal) {
+ return CastLiteral<int64_t, double>(literal, FieldType::DOUBLE);
+ }},
+
+ {std::make_pair(FieldType::FLOAT, FieldType::TINYINT),
+ [&](const Literal& literal) {
+ return CastLiteral<float, int8_t>(literal, FieldType::TINYINT);
+ }},
+ {std::make_pair(FieldType::FLOAT, FieldType::SMALLINT),
+ [&](const Literal& literal) {
+ return CastLiteral<float, int16_t>(literal, FieldType::SMALLINT);
+ }},
+ {std::make_pair(FieldType::FLOAT, FieldType::INT),
+ [&](const Literal& literal) {
+ return CastLiteral<float, int32_t>(literal, FieldType::INT);
+ }},
+ {std::make_pair(FieldType::FLOAT, FieldType::BIGINT),
+ [&](const Literal& literal) {
+ return CastLiteral<float, int64_t>(literal, FieldType::BIGINT);
+ }},
+ {std::make_pair(FieldType::FLOAT, FieldType::FLOAT),
+ [&](const Literal& literal) {
+ return CastLiteral<float, float>(literal, FieldType::FLOAT);
+ }},
+ {std::make_pair(FieldType::FLOAT, FieldType::DOUBLE),
+ [&](const Literal& literal) {
+ return CastLiteral<float, double>(literal, FieldType::DOUBLE);
+ }},
+
+ {std::make_pair(FieldType::DOUBLE, FieldType::TINYINT),
+ [&](const Literal& literal) {
+ return CastLiteral<double, int8_t>(literal, FieldType::TINYINT);
+ }},
+ {std::make_pair(FieldType::DOUBLE, FieldType::SMALLINT),
+ [&](const Literal& literal) {
+ return CastLiteral<double, int16_t>(literal, FieldType::SMALLINT);
+ }},
+ {std::make_pair(FieldType::DOUBLE, FieldType::INT),
+ [&](const Literal& literal) {
+ return CastLiteral<double, int32_t>(literal, FieldType::INT);
+ }},
+ {std::make_pair(FieldType::DOUBLE, FieldType::BIGINT),
+ [&](const Literal& literal) {
+ return CastLiteral<double, int64_t>(literal, FieldType::BIGINT);
+ }},
+ {std::make_pair(FieldType::DOUBLE, FieldType::FLOAT),
+ [&](const Literal& literal) {
+ return CastLiteral<double, float>(literal, FieldType::FLOAT);
+ }},
+ {std::make_pair(FieldType::DOUBLE, FieldType::DOUBLE), [&](const
Literal& literal) {
+ return CastLiteral<double, double>(literal, FieldType::DOUBLE);
+ }}};
+}
+
+Result<Literal> NumericPrimitiveCastExecutor::Cast(
+ const Literal& literal, const std::shared_ptr<arrow::DataType>&
target_type) const {
+ FieldType src_type = literal.GetType();
+ PAIMON_ASSIGN_OR_RAISE(FieldType target_field_type,
+
FieldTypeUtils::ConvertToFieldType(target_type->id()));
+ auto iter = literal_cast_executor_map_.find(std::make_pair(src_type,
target_field_type));
+ if (iter == literal_cast_executor_map_.end()) {
+ return Status::Invalid(
+ fmt::format("cast literal in NumericPrimitiveCastExecutor failed:
cannot find cast "
+ "function from {} to {}",
+ FieldTypeUtils::FieldTypeToString(src_type),
+ FieldTypeUtils::FieldTypeToString(target_field_type)));
+ }
+ return iter->second(literal);
+}
+
+Result<std::shared_ptr<arrow::Array>> NumericPrimitiveCastExecutor::Cast(
+ const std::shared_ptr<arrow::Array>& array, const
std::shared_ptr<arrow::DataType>& target_type,
+ arrow::MemoryPool* pool) const {
+ arrow::compute::CastOptions options = arrow::compute::CastOptions::Safe();
+ options.allow_int_overflow = true;
+ options.allow_float_truncate = true;
+ return CastingUtils::Cast(array, target_type, options, pool);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/casting/numeric_primitive_cast_executor.h
b/src/paimon/core/casting/numeric_primitive_cast_executor.h
new file mode 100644
index 0000000..d84cd9f
--- /dev/null
+++ b/src/paimon/core/casting/numeric_primitive_cast_executor.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 <functional>
+#include <map>
+#include <memory>
+#include <utility>
+
+#include "arrow/array/array_base.h"
+#include "paimon/core/casting/cast_executor.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class DataType;
+class MemoryPool;
+} // namespace arrow
+
+namespace paimon {
+enum class FieldType;
+
+class NumericPrimitiveCastExecutor : public CastExecutor {
+ public:
+ NumericPrimitiveCastExecutor();
+ Result<Literal> Cast(const Literal& literal,
+ const std::shared_ptr<arrow::DataType>& target_type)
const override;
+
+ Result<std::shared_ptr<arrow::Array>> Cast(const
std::shared_ptr<arrow::Array>& array,
+ const
std::shared_ptr<arrow::DataType>& target_type,
+ arrow::MemoryPool* pool) const
override;
+
+ private:
+ template <typename SrcType, typename TargetType>
+ static Literal CastLiteral(const Literal& literal, const FieldType&
target_type) {
+ if (literal.IsNull()) {
+ return Literal(target_type);
+ }
+ SrcType value = literal.GetValue<SrcType>();
+ return Literal(static_cast<TargetType>(value));
+ }
+
+ private:
+ std::map<std::pair<FieldType, FieldType>, std::function<Literal(const
Literal&)>>
+ literal_cast_executor_map_;
+};
+
+} // namespace paimon
