branch-3.1:[feature](external) Support reading Hudi/Paimon/Iceberg tables after schema changes. (#51341)

be/src/olap/push_handler.cpp

@@ -643,11 +643,11 @@ Status PushBrokerReader::_get_next_reader() {
                         const_cast<cctz::time_zone*>(&_runtime_state->timezone_obj()),
                         _io_ctx.get(), _runtime_state.get());
 
-        std::vector<std::string> place_holder;
         init_status = parquet_reader->init_reader(
-                _all_col_names, place_holder, _colname_to_value_range, _push_down_exprs,
-                _real_tuple_desc, _default_val_row_desc.get(), _col_name_to_slot_id,
-                &_not_single_slot_filter_conjuncts, &_slot_id_to_filter_conjuncts, false);
+                _all_col_names, _colname_to_value_range, _push_down_exprs, _real_tuple_desc,
+                _default_val_row_desc.get(), _col_name_to_slot_id,
+                &_not_single_slot_filter_conjuncts, &_slot_id_to_filter_conjuncts,
+                vectorized::TableSchemaChangeHelper::ConstNode::get_instance(), false);
         _cur_reader = std::move(parquet_reader);
         if (!init_status.ok()) {
             return Status::InternalError("failed to init reader for file {}, err: {}", range.path,

be/src/runtime/descriptors.h

@@ -47,6 +47,12 @@ class RepeatedField;
 
 namespace doris {
 
+#ifdef BE_TEST
+#define MOCK_REMOVE(str)
+#else
+#define MOCK_REMOVE(str) str
+#endif
+
 class ObjectPool;
 class PTupleDescriptor;
 class PSlotDescriptor;
@@ -105,7 +111,7 @@ class SlotDescriptor {
     const TupleId _parent;
     const int _col_pos;
     bool _is_nullable;
-    const std::string _col_name;
+    MOCK_REMOVE(const) std::string _col_name;
     const std::string _col_name_lower_case;
 
     const int32_t _col_unique_id;

be/src/vec/exec/format/column_type_convert.cpp

@@ -279,6 +279,24 @@ static std::unique_ptr<ColumnTypeConverter> _numeric_to_decimal_converter(
     }
 }
 
+static std::unique_ptr<ColumnTypeConverter> _datetime_to_numeric_converter(
+        const TypeDescriptor& src_type, const DataTypePtr& dst_type) {
+    PrimitiveType dst_primitive_type =
+            remove_nullable(dst_type)->get_type_as_type_descriptor().type;
+
+    switch (dst_primitive_type) {
+#define DISPATCH(DST_TYPE)                                               \
+    case DST_TYPE: {                                                     \
+        return std::make_unique<DateTimeToNumericConverter<DST_TYPE>>(); \
+    }
+        FOR_LOGICAL_INTEGER_TYPES(DISPATCH)
+#undef DISPATCH
+    default: {
+        return std::make_unique<UnsupportedConverter>(src_type, dst_type);
+    }
+    };
+}
+
 static std::unique_ptr<ColumnTypeConverter> _decimal_to_numeric_converter(
         const TypeDescriptor& src_type, const DataTypePtr& dst_type) {
     PrimitiveType src_primitive_type = src_type.type;
@@ -403,6 +421,11 @@ std::unique_ptr<ColumnTypeConverter> ColumnTypeConverter::get_converter(
         return std::make_unique<TimeV2Converter<TYPE_DATETIMEV2, TYPE_DATEV2>>();
     }
 
+    // datetime to bigint (ms)
+    if (src_primitive_type == TYPE_DATETIMEV2 && _is_numeric_type(dst_primitive_type)) {
+        return _datetime_to_numeric_converter(src_type, dst_type);
+    }
+
     // numeric to decimal
     if (_is_numeric_type(src_primitive_type) && _is_decimal_type(dst_primitive_type)) {
         return _numeric_to_decimal_converter(src_type, dst_type);

be/src/vec/exec/format/column_type_convert.h

@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <cctz/time_zone.h>
+
 #include <cstdint>
 #include <utility>
 
@@ -594,6 +596,62 @@ class CastStringConverter : public ColumnTypeConverter {
     }
 };
 
+template <PrimitiveType DstPrimitiveType>
+class DateTimeToNumericConverter : public ColumnTypeConverter {
+public:
+    Status convert(ColumnPtr& src_col, MutableColumnPtr& dst_col) override {
+        using SrcColumnType = typename PrimitiveTypeTraits<TYPE_DATETIMEV2>::ColumnType;
+        using DstColumnType = typename PrimitiveTypeTraits<DstPrimitiveType>::ColumnType;
+        using SrcCppType = typename PrimitiveTypeTraits<TYPE_DATETIMEV2>::CppType;
+        using DstCppType = typename PrimitiveTypeTraits<DstPrimitiveType>::CppType;
+
+        ColumnPtr from_col = remove_nullable(src_col);
+        MutableColumnPtr to_col = remove_nullable(dst_col->get_ptr())->assume_mutable();
+
+        NullMap* null_map = nullptr;
+        if (dst_col->is_nullable()) {
+            null_map = &reinterpret_cast<vectorized::ColumnNullable*>(dst_col.get())
+                                ->get_null_map_data();
+        }
+
+        size_t rows = from_col->size();
+        auto& src_data = static_cast<const SrcColumnType*>(from_col.get())->get_data();
+        size_t start_idx = to_col->size();
+        to_col->resize(start_idx + rows);
+        auto& data = static_cast<DstColumnType&>(*to_col.get()).get_data();
+
+        for (int i = 0; i < rows; ++i) {
+            const SrcCppType& src_value = src_data[i];
+            auto& dst_value = reinterpret_cast<DstCppType&>(data[start_idx + i]);
+
+            int64_t ts_s = 0;
+            if (!src_value.unix_timestamp(&ts_s, cctz::utc_time_zone())) {
+                if (null_map == nullptr) {
+                    return Status::InternalError("Failed to cast value '{}' to {} column",
+                                                 src_data[i], dst_col->get_name());
+                } else {
+                    (*null_map)[start_idx + i] = 1;
+                }
+            }
+            auto micro = src_value.microsecond();
+            int64_t ts_ms = ts_s * 1000 + micro / 1000;
+            if constexpr (DstPrimitiveType != TYPE_LARGEINT && DstPrimitiveType != TYPE_BIGINT) {
+                if ((Int64)std::numeric_limits<DstCppType>::min() > ts_ms ||
+                    ts_ms > (Int64)std::numeric_limits<DstCppType>::max()) {
+                    if (null_map == nullptr) {
+                        return Status::InternalError("Failed to cast value '{}' to {} column",
+                                                     src_data[i], dst_col->get_name());
+                    } else {
+                        (*null_map)[start_idx + i] = 1;
+                    }
+                }
+            }
+            dst_value = static_cast<DstCppType>(ts_ms);
+        }
+        return Status::OK();
+    }
+};
+
 // only support date & datetime v2
 template <PrimitiveType SrcPrimitiveType, PrimitiveType DstPrimitiveType>
 class TimeV2Converter : public ColumnTypeConverter {