feat: introduce type casting executors for schema evolution

src/paimon/core/casting/boolean_to_decimal_cast_executor.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/boolean_to_decimal_cast_executor.h"
+
+#include <cassert>
+#include <cstdint>
+#include <optional>
+#include <utility>
+
+#include "arrow/array/array_primitive.h"
+#include "arrow/array/builder_decimal.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/decimal.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/decimal_utils.h"
+#include "paimon/data/decimal.h"
+#include "paimon/defs.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class MemoryPool;
+class Array;
+}  // namespace arrow
+
+namespace paimon {
+
+Result<Literal> BooleanToDecimalCastExecutor::Cast(
+    const Literal& literal, const std::shared_ptr<arrow::DataType>& target_type) const {
+    assert(literal.GetType() == FieldType::BOOLEAN);
+    PAIMON_RETURN_NOT_OK(DecimalUtils::CheckDecimalType(*target_type));
+    auto* decimal_type = arrow::internal::checked_cast<arrow::DecimalType*>(target_type.get());
+    assert(decimal_type);
+    if (literal.IsNull()) {
+        return Literal(FieldType::DECIMAL);
+    }
+    bool bool_value = literal.GetValue<bool>();
+    auto scaled_decimal = DecimalUtils::RescaleDecimalWithOverflowCheck(
+        arrow::Decimal128(bool_value), /*src_scale=*/0, decimal_type->precision(),
+        decimal_type->scale());
+    if (scaled_decimal == std::nullopt) {
+        return Literal(FieldType::DECIMAL);
+    }
+    return Literal(Decimal(
+        decimal_type->precision(), decimal_type->scale(),
+        static_cast<Decimal::int128_t>(static_cast<Decimal::uint128_t>(static_cast<uint64_t>(
+                                           scaled_decimal.value().high_bits()))
+                                           << 64 |
+                                       scaled_decimal.value().low_bits())));
+}
+
+Result<std::shared_ptr<arrow::Array>> BooleanToDecimalCastExecutor::Cast(
+    const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::DataType>& target_type,
+    arrow::MemoryPool* pool) const {
+    PAIMON_RETURN_NOT_OK(DecimalUtils::CheckDecimalType(*target_type));
+    auto* boolean_array = arrow::internal::checked_cast<arrow::BooleanArray*>(array.get());
+    assert(boolean_array);
+    auto* decimal_type = arrow::internal::checked_cast<arrow::DecimalType*>(target_type.get());
+    assert(decimal_type);
+    auto decimal_builder = std::make_shared<arrow::Decimal128Builder>(target_type, pool);
+    for (int64_t i = 0; i < boolean_array->length(); ++i) {
+        if (boolean_array->IsNull(i)) {
+            PAIMON_RETURN_NOT_OK_FROM_ARROW(decimal_builder->AppendNull());
+        } else {
+            bool bool_value = boolean_array->Value(i);
+            auto scaled_decimal = DecimalUtils::RescaleDecimalWithOverflowCheck(
+                arrow::Decimal128(bool_value), /*src_scale=*/0, decimal_type->precision(),
+                decimal_type->scale());
+            if (scaled_decimal == std::nullopt) {
+                PAIMON_RETURN_NOT_OK_FROM_ARROW(decimal_builder->AppendNull());
+            } else {
+                PAIMON_RETURN_NOT_OK_FROM_ARROW(decimal_builder->Append(scaled_decimal.value()));
+            }
+        }
+    }
+    std::shared_ptr<arrow::Array> casted_array;
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(decimal_builder->Finish(&casted_array));
+    return casted_array;
+}
+
+}  // namespace paimon

src/paimon/core/casting/boolean_to_decimal_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/array/array_base.h"
+#include "paimon/core/casting/cast_executor.h"
+#include "paimon/data/decimal.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class DataType;
+class MemoryPool;
+}  // namespace arrow
+
+namespace paimon {
+class BooleanToDecimalCastExecutor : public CastExecutor {
+ public:
+    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;
+};
+
+}  // namespace paimon

src/paimon/core/casting/boolean_to_numeric_cast_executor.cpp

@@ -0,0 +1,81 @@
+/*
+ * 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/boolean_to_numeric_cast_executor.h"
+
+#include <cassert>
+#include <cstdint>
+#include <string>
+#include <utility>
+
+#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 {
+BooleanToNumericCastExecutor::BooleanToNumericCastExecutor() {
+    literal_cast_executor_map_ = {
+        {FieldType::TINYINT,
+         [&](const Literal& literal) { return CastLiteral<int8_t>(literal, FieldType::TINYINT); }},
+        {FieldType::SMALLINT,
+         [&](const Literal& literal) {
+             return CastLiteral<int16_t>(literal, FieldType::SMALLINT);
+         }},
+        {FieldType::INT,
+         [&](const Literal& literal) { return CastLiteral<int32_t>(literal, FieldType::INT); }},
+        {FieldType::BIGINT,
+         [&](const Literal& literal) { return CastLiteral<int64_t>(literal, FieldType::BIGINT); }},
+        {FieldType::FLOAT,
+         [&](const Literal& literal) { return CastLiteral<float>(literal, FieldType::FLOAT); }},
+        {FieldType::DOUBLE,
+         [&](const Literal& literal) { return CastLiteral<double>(literal, FieldType::DOUBLE); }}};
+}
+
+Result<Literal> BooleanToNumericCastExecutor::Cast(
+    const Literal& literal, const std::shared_ptr<arrow::DataType>& target_type) const {
+    assert(literal.GetType() == FieldType::BOOLEAN);
+    PAIMON_ASSIGN_OR_RAISE(FieldType target_field_type,
+                           FieldTypeUtils::ConvertToFieldType(target_type->id()));
+    auto iter = literal_cast_executor_map_.find(target_field_type);
+    if (iter == literal_cast_executor_map_.end()) {
+        return Status::Invalid(
+            fmt::format("cast literal in BooleanToNumericCastExecutor failed: cannot find cast "
+                        "function from boolean to {}",
+                        target_type->ToString()));
+    }
+    return iter->second(literal);
+}
+
+Result<std::shared_ptr<arrow::Array>> BooleanToNumericCastExecutor::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();
+    return CastingUtils::Cast(array, target_type, options, pool);
+}
+
+}  // namespace paimon

src/paimon/core/casting/boolean_to_numeric_cast_executor.h

@@ -0,0 +1,63 @@
+/*
+ * 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 "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 BooleanToNumericCastExecutor : public CastExecutor {
+ public:
+    BooleanToNumericCastExecutor();
+    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 TargetType>
+    static Literal CastLiteral(const Literal& literal, const FieldType& target_type) {
+        if (literal.IsNull()) {
+            return Literal(target_type);
+        }
+        bool value = literal.GetValue<bool>();
+        return Literal(static_cast<TargetType>(value));
+    }
+
+ private:
+    std::map<FieldType, std::function<Literal(const Literal&)>> literal_cast_executor_map_;
+};
+
+}  // namespace paimon

src/paimon/core/casting/boolean_to_string_cast_executor.cpp

@@ -0,0 +1,61 @@
+/*
+ * 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/boolean_to_string_cast_executor.h"
+
+#include <cassert>
+#include <string>
+#include <utility>
+
+#include "arrow/compute/cast.h"
+#include "arrow/type.h"
+#include "paimon/common/utils/field_type_utils.h"
+#include "paimon/core/casting/casting_utils.h"
+#include "paimon/defs.h"
+
+namespace arrow {
+class MemoryPool;
+class Array;
+}  // namespace arrow
+
+namespace paimon {
+Result<Literal> BooleanToStringCastExecutor::Cast(
+    const Literal& literal, const std::shared_ptr<arrow::DataType>& target_type) const {
+    assert(literal.GetType() == FieldType::BOOLEAN);
+    PAIMON_ASSIGN_OR_RAISE(FieldType target_field_type,
+                           FieldTypeUtils::ConvertToFieldType(target_type->id()));
+    assert(target_field_type == FieldType::STRING);
+    static const std::string TRUE = "true";
+    static const std::string FALSE = "false";
+    if (literal.IsNull()) {
+        return Literal(target_field_type);
+    }
+    if (literal.GetValue<bool>()) {
+        return Literal(target_field_type, TRUE.data(), TRUE.size());
+    }
+    return Literal(target_field_type, FALSE.data(), FALSE.size());
+}
+
+Result<std::shared_ptr<arrow::Array>> BooleanToStringCastExecutor::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();
+    return CastingUtils::Cast(array, target_type, options, pool);
+}
+}  // namespace paimon

src/paimon/core/casting/boolean_to_string_cast_executor.h

@@ -0,0 +1,44 @@
+/*
+ * 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/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 {
+class BooleanToStringCastExecutor : public CastExecutor {
+ public:
+    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;
+};
+
+}  // namespace paimon