[fix](datatype)Fix unaligned memory access in read_column_from_arrow

Author: Chen768959

be/src/vec/data_types/serde/data_type_array_serde.cpp

@@ -326,11 +326,17 @@ Status DataTypeArraySerDe::read_column_from_arrow(IColumn& column, const arrow::
     auto arrow_offsets_array = concrete_array->offsets();
     auto* arrow_offsets = dynamic_cast<arrow::Int32Array*>(arrow_offsets_array.get());
     auto prev_size = offsets_data.back();
-    auto arrow_nested_start_offset = arrow_offsets->Value(start);
-    auto arrow_nested_end_offset = arrow_offsets->Value(end);
+    int32_t arrow_nested_start_offset;
+    int32_t arrow_nested_end_offset;
+    const auto* offsets_raw_data = arrow_offsets->raw_values();
+    memcpy(&arrow_nested_start_offset, offsets_raw_data + start, sizeof(int32_t));
+    memcpy(&arrow_nested_end_offset, offsets_raw_data + end, sizeof(int32_t));
+
     for (auto i = start + 1; i < end + 1; ++i) {
+        int32_t current_offset;
+        memcpy(&current_offset, offsets_raw_data + i, sizeof(int32_t));
         // convert to doris offset, start from offsets.back()
-        offsets_data.emplace_back(prev_size + arrow_offsets->Value(i) - arrow_nested_start_offset);
+        offsets_data.emplace_back(prev_size + current_offset - arrow_nested_start_offset);
     }
     return nested_serde->read_column_from_arrow(
             column_array.get_data(), concrete_array->values().get(), arrow_nested_start_offset,

be/src/vec/data_types/serde/data_type_datetimev2_serde.cpp

@@ -398,7 +398,11 @@ Status DataTypeDateTimeV2SerDe::read_column_from_arrow(IColumn& column,
         }
         }
         for (auto value_i = start; value_i < end; ++value_i) {
-            auto utc_epoch = static_cast<UInt64>(concrete_array->Value(value_i));
+            int64_t utc_epoch_raw;
+            const auto* raw_data_ptr = concrete_array->raw_values() + value_i;
+            memcpy(&utc_epoch_raw, raw_data_ptr, sizeof(int64_t));
+
+            auto utc_epoch = static_cast<UInt64>(utc_epoch_raw);
 
             DateV2Value<DateTimeV2ValueType> v;
             // convert second

be/src/vec/data_types/serde/data_type_datev2_serde.cpp

@@ -115,8 +115,12 @@ Status DataTypeDateV2SerDe::read_column_from_arrow(IColumn& column, const arrow:
     auto& col_data = static_cast<ColumnDateV2&>(column).get_data();
     const auto* concrete_array = dynamic_cast<const arrow::Date32Array*>(arrow_array);
     for (auto value_i = start; value_i < end; ++value_i) {
+        int32_t date_value;
+        const auto* raw_data_ptr = concrete_array->raw_values() + value_i;
+        memcpy(&date_value, raw_data_ptr, sizeof(int32_t));
+
         DateV2Value<DateV2ValueType> v;
-        v.get_date_from_daynr(concrete_array->Value(value_i) + date_threshold);
+        v.get_date_from_daynr(date_value + date_threshold);
         col_data.emplace_back(binary_cast<DateV2Value<DateV2ValueType>, UInt32>(v));
     }
     return Status::OK();

be/src/vec/data_types/serde/data_type_decimal_serde.cpp

@@ -337,8 +337,10 @@ Status DataTypeDecimalSerDe<T>::read_column_from_arrow(IColumn& column,
     } else if constexpr (T == TYPE_DECIMAL32 || T == TYPE_DECIMAL64 || T == TYPE_DECIMAL128I) {
         const auto* concrete_array = dynamic_cast<const arrow::DecimalArray*>(arrow_array);
         for (auto value_i = start; value_i < end; ++value_i) {
-            column_data.emplace_back(
-                    *reinterpret_cast<const FieldType*>(concrete_array->Value(value_i)));
+			const auto* value = concrete_array->Value(value_i);
+            FieldType decimal_value;
+            memcpy(&decimal_value, value, sizeof(FieldType));
+            column_data.emplace_back(decimal_value);
         }
     } else if constexpr (T == TYPE_DECIMAL256) {
         const auto* concrete_array = dynamic_cast<const arrow::Decimal256Array*>(arrow_array);

be/src/vec/data_types/serde/data_type_number_serde.cpp

@@ -221,8 +221,15 @@ Status DataTypeNumberSerDe<T>::read_column_from_arrow(IColumn& column,
 
         for (size_t offset_i = start; offset_i < end; ++offset_i) {
             if (!concrete_array->IsNull(offset_i)) {
-                const auto* raw_data = buffer->data() + concrete_array->value_offset(offset_i);
-                const auto raw_data_len = concrete_array->value_length(offset_i);
+                int32_t start_offset;
+                int32_t end_offset;
+                const auto* offsets_data = concrete_array->value_offsets()->data();
+                memcpy(&start_offset, offsets_data + offset_i * sizeof(int32_t), sizeof(int32_t));
+                memcpy(&end_offset, offsets_data + (offset_i + 1) * sizeof(int32_t),
+					   sizeof(int32_t));
+
+                const auto* raw_data = buffer->data() + start_offset;
+                const auto raw_data_len = end_offset - start_offset;
 
                 if (raw_data_len == 0) {
                     col_data.emplace_back(Int128()); // Int128() is NULL

be/src/vec/data_types/serde/data_type_string_serde.cpp

@@ -262,12 +262,23 @@ Status DataTypeStringSerDeBase<ColumnType>::read_column_from_arrow(
         arrow_array->type_id() == arrow::Type::BINARY) {
         const auto* concrete_array = dynamic_cast<const arrow::BinaryArray*>(arrow_array);
         std::shared_ptr<arrow::Buffer> buffer = concrete_array->value_data();
+        const auto offsets_buffer = concrete_array->value_offsets();
+        const uint8_t* offsets_data = offsets_buffer->data();
 
         for (auto offset_i = start; offset_i < end; ++offset_i) {
             if (!concrete_array->IsNull(offset_i)) {
-                const auto* raw_data = buffer->data() + concrete_array->value_offset(offset_i);
+                int32_t start_offset;
+                int32_t end_offset;
+
+                memcpy(&start_offset, offsets_data + offset_i * sizeof(int32_t), sizeof(int32_t));
+                memcpy(&end_offset, offsets_data + (offset_i + 1) * sizeof(int32_t),
+                       sizeof(int32_t));
+
+                int32_t length = end_offset - start_offset;
+                const auto* raw_data = buffer->data() + start_offset;
+
                 assert_cast<ColumnType&>(column).insert_data(
-                        (char*)raw_data, concrete_array->value_length(offset_i));
+                        reinterpret_cast<const char*>(raw_data), length);
             } else {
                 assert_cast<ColumnType&>(column).insert_default();
             }
@@ -451,4 +462,4 @@ template class DataTypeStringSerDeBase<ColumnString>;
 template class DataTypeStringSerDeBase<ColumnString64>;
 template class DataTypeStringSerDeBase<ColumnFixedLengthObject>;
 
-} // namespace doris::vectorized
+} // namespace doris::vectorized