[echcement](hive)support read hive table that change timestamp column to bigint.

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

@@ -278,6 +278,23 @@ static std::unique_ptr<ColumnTypeConverter> _numeric_to_decimal_converter(
     }
 }
 
+static std::unique_ptr<ColumnTypeConverter> _datetime_to_numeric_converter(
+        const DataTypePtr& src_type, const DataTypePtr& dst_type) {
+    PrimitiveType dst_primitive_type = dst_type->get_primitive_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 DataTypePtr& src_type, const DataTypePtr& dst_type) {
     PrimitiveType src_primitive_type = src_type->get_primitive_type();
@@ -404,6 +421,11 @@ std::unique_ptr<ColumnTypeConverter> ColumnTypeConverter::get_converter(const Da
         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

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <absl/strings/numbers.h>
+#include <cctz/time_zone.h>
 
 #include <cstdint>
 #include <utility>
@@ -592,6 +593,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 {

be/src/vec/exec/format/parquet/parquet_column_convert.cpp

@@ -246,7 +246,7 @@ std::unique_ptr<PhysicalToLogicalConverter> PhysicalToLogicalConverter::get_conv
             convert_params->reset_time_scale_if_missing(9);
             physical_converter = std::make_unique<Int96toTimestamp>();
         } else if (src_physical_type == tparquet::Type::INT64) {
-            convert_params->reset_time_scale_if_missing(dst_logical_type->get_scale());
+            convert_params->reset_time_scale_if_missing(src_logical_type->get_scale());
             physical_converter = std::make_unique<Int64ToTimestamp>();
         } else {
             physical_converter =