[improvement](be) Implement native Parquet decoding for FileScannerV2

be/src/core/column/column_string.h

@@ -26,6 +26,7 @@
 #include <cassert>
 #include <cstdint>
 #include <cstring>
+#include <limits>
 #include <typeinfo>
 #include <vector>
 
@@ -276,6 +277,56 @@ class ColumnStr final : public COWHelper<IColumn, ColumnStr<T>> {
         sanity_check_simple();
     }
 
+    template <typename SelectionRange>
+    void insert_many_parquet_plain_byte_arrays(const char* encoded_data,
+                                               const uint32_t* payload_offsets,
+                                               const uint32_t* value_offsets, size_t num,
+                                               const std::vector<SelectionRange>& value_spans) {
+        if (UNLIKELY(num == 0)) {
+            return;
+        }
+        const size_t old_chars_size = chars.size();
+        const size_t bytes = value_offsets[num];
+        check_chars_length(old_chars_size + bytes, offsets.size() + num);
+        chars.resize(old_chars_size + bytes);
+
+        // Source payloads may be contiguous for decoder-owned buffers even though PLAIN normally
+        // separates them with length prefixes. Coalesce whenever the published layout permits it;
+        // the fallback remains one direct source-to-column copy without a StringRef staging array.
+        size_t covered_values = 0;
+        for (const auto& span : value_spans) {
+            DORIS_CHECK_EQ(span.first, covered_values);
+            DORIS_CHECK_LE(span.first + span.count, num);
+            size_t run_first = span.first;
+            const size_t span_end = span.first + span.count;
+            while (run_first < span_end) {
+                size_t run_end = run_first + 1;
+                while (run_end < span_end &&
+                       payload_offsets[run_end] == payload_offsets[run_first] +
+                                                           value_offsets[run_end] -
+                                                           value_offsets[run_first]) {
+                    ++run_end;
+                }
+                const size_t run_bytes = value_offsets[run_end] - value_offsets[run_first];
+                if (run_bytes != 0) {
+                    memcpy(chars.data() + old_chars_size + value_offsets[run_first],
+                           encoded_data + payload_offsets[run_first], run_bytes);
+                }
+                run_first = run_end;
+            }
+            covered_values = span_end;
+        }
+        DORIS_CHECK_EQ(covered_values, num);
+
+        const size_t old_rows = offsets.size();
+        const auto tail_offset = offsets.back();
+        offsets.resize(old_rows + num);
+        for (size_t row = 0; row < num; ++row) {
+            offsets[old_rows + row] = tail_offset + value_offsets[row + 1];
+        }
+        sanity_check_simple();
+    }
+
     // Insert `num` string entries with real length information but no actual
     // character data. The `lengths` array provides the byte length of each
     // string. Offsets are built with correct cumulative sizes so that
@@ -322,6 +373,35 @@ class ColumnStr final : public COWHelper<IColumn, ColumnStr<T>> {
         sanity_check_simple();
     }
 
+    void insert_many_fixed_length_data(const char* data, size_t value_length, size_t num) {
+        if (num == 0) {
+            return;
+        }
+        if (value_length > std::numeric_limits<size_t>::max() / num) {
+            throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+                                   "ColumnString fixed-length append size overflow");
+        }
+        const size_t old_chars_size = chars.size();
+        const size_t old_offsets_size = offsets.size();
+        const size_t bytes = value_length * num;
+        if (bytes > std::numeric_limits<size_t>::max() - old_chars_size ||
+            num > std::numeric_limits<size_t>::max() - old_offsets_size) {
+            throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+                                   "ColumnString fixed-length append size overflow");
+        }
+        check_chars_length(old_chars_size + bytes, old_offsets_size + num);
+        chars.resize(old_chars_size + bytes);
+        if (bytes != 0) {
+            memcpy(chars.data() + old_chars_size, data, bytes);
+        }
+        offsets.resize(old_offsets_size + num);
+        for (size_t row = 0; row < num; ++row) {
+            offsets[old_offsets_size + row] =
+                    static_cast<T>(old_chars_size + (row + 1) * value_length);
+        }
+        sanity_check_simple();
+    }
+
     template <size_t copy_length>
     void insert_many_strings_fixed_length(const StringRef* strings, size_t num) {
         size_t new_size = 0;

be/src/core/data_type_serde/data_type_datetimev2_serde.cpp

@@ -31,10 +31,13 @@
 #include "core/data_type/primitive_type.h"
 #include "core/data_type_serde/arrow_validation.h"
 #include "core/data_type_serde/decoded_column_view.h"
+#include "core/data_type_serde/parquet_decode_source.h"
+#include "core/data_type_serde/parquet_timestamp.h"
 #include "core/types.h"
 #include "core/value/vdatetime_value.h"
 #include "exprs/function/cast/cast_to_datetimev2_impl.hpp"
 #include "exprs/function/cast/cast_to_string.h"
+#include "util/unaligned.h"
 
 enum {
     DIVISOR_FOR_SECOND = 1,
@@ -48,22 +51,6 @@ static const int64_t micro_to_nano_second = 1000;
 
 namespace {
 
-#pragma pack(1)
-struct DecodedInt96Timestamp {
-    int64_t nanos_of_day;
-    int32_t julian_day;
-
-    int64_t to_timestamp_micros() const {
-        static constexpr int32_t JULIAN_EPOCH_OFFSET_DAYS = 2440588;
-        static constexpr int64_t MICROS_IN_DAY = 86400000000;
-        static constexpr int64_t NANOS_PER_MICROSECOND = 1000;
-        return (julian_day - JULIAN_EPOCH_OFFSET_DAYS) * MICROS_IN_DAY +
-               nanos_of_day / NANOS_PER_MICROSECOND;
-    }
-};
-#pragma pack()
-static_assert(sizeof(DecodedInt96Timestamp) == 12);
-
 Status append_datetimev2_from_epoch_micros(ColumnDateTimeV2::Container& data,
                                            int64_t timestamp_micros) {
     static constexpr int64_t MICROS_PER_SECOND = 1000000;
@@ -105,9 +92,9 @@ Status append_datetimev2_from_epoch_micros(ColumnDateTimeV2::Container& data,
     return Status::OK();
 }
 
-void append_datetimev2_from_utc_epoch_micros(ColumnDateTimeV2::Container& data,
-                                             int64_t timestamp_micros,
-                                             const cctz::time_zone& timezone) {
+Status append_datetimev2_from_utc_epoch_micros(ColumnDateTimeV2::Container& data,
+                                               int64_t timestamp_micros,
+                                               const cctz::time_zone& timezone) {
     static constexpr int64_t MICROS_PER_SECOND = 1000000;
 
     int64_t epoch_seconds = timestamp_micros / MICROS_PER_SECOND;
@@ -120,19 +107,91 @@ void append_datetimev2_from_utc_epoch_micros(ColumnDateTimeV2::Container& data,
     DateV2Value<DateTimeV2ValueType> datetime_value;
     datetime_value.from_unixtime(epoch_seconds, timezone);
     datetime_value.set_microsecond(static_cast<uint32_t>(micros_of_second));
+    if (!datetime_value.is_valid_date()) {
+        return Status::DataQualityError(
+                "Decoded DATETIMEV2 timestamp is outside the target timezone range: micros={}",
+                timestamp_micros);
+    }
     data.push_back(datetime_value);
+    return Status::OK();
 }
 
-int64_t decoded_timestamp_micros(const DecodedColumnView& view, int64_t value) {
+ParquetTimeUnit decoded_parquet_time_unit(const DecodedColumnView& view) {
     if (view.time_unit == DecodedTimeUnit::MILLIS) {
-        return value * 1000;
+        return ParquetTimeUnit::MILLIS;
     }
     if (view.time_unit == DecodedTimeUnit::NANOS) {
-        return value / 1000;
+        return ParquetTimeUnit::NANOS;
     }
-    return value;
+    return ParquetTimeUnit::MICROS;
 }
 
+class DateTimeV2ParquetConsumer final : public ParquetFixedValueConsumer {
+public:
+    DateTimeV2ParquetConsumer(IColumn& column, const ParquetDecodeContext& context,
+                              ParquetMaterializationState* state = nullptr)
+            : _data(assert_cast<ColumnDateTimeV2&>(column).get_data()),
+              _context(context),
+              _state(state) {}
+
+    Status consume(const uint8_t* values, size_t num_values, size_t value_width) override {
+        const size_t old_size = _data.size();
+        static const auto utc_timezone = cctz::utc_time_zone();
+        const auto& timezone = _context.timezone == nullptr ? utc_timezone : *_context.timezone;
+        for (size_t row = 0; row < num_values; ++row) {
+            int64_t timestamp_micros;
+            Status status;
+            if (_context.physical_type == ParquetPhysicalType::INT96) {
+                DORIS_CHECK_EQ(value_width, sizeof(ParquetInt96Timestamp));
+                status = parquet_int96_timestamp_micros(
+                        unaligned_load<ParquetInt96Timestamp>(values +
+                                                              row * sizeof(ParquetInt96Timestamp)),
+                        &timestamp_micros);
+            } else {
+                DORIS_CHECK(_context.physical_type == ParquetPhysicalType::INT64);
+                DORIS_CHECK_EQ(value_width, sizeof(int64_t));
+                status = parquet_timestamp_micros(
+                        _context.time_unit, unaligned_load<int64_t>(values + row * sizeof(int64_t)),
+                        &timestamp_micros);
+            }
+            if (!status.ok()) {
+                if (_state != nullptr && _state->mark_conversion_failure(_data.size())) {
+                    _data.emplace_back();
+                    continue;
+                }
+                _data.resize(old_size);
+                return status;
+            }
+            status = _context.physical_type == ParquetPhysicalType::INT96 ||
+                                     _context.timestamp_is_adjusted_to_utc
+                             ? append_datetimev2_from_utc_epoch_micros(_data, timestamp_micros,
+                                                                       timezone)
+                             : append_datetimev2_from_epoch_micros(_data, timestamp_micros);
+            if (!status.ok()) {
+                if (_state != nullptr && _state->mark_conversion_failure(_data.size())) {
+                    _data.emplace_back();
+                    continue;
+                }
+                _data.resize(old_size);
+                return status;
+            }
+        }
+        return Status::OK();
+    }
+
+private:
+    ColumnDateTimeV2::Container& _data;
+    const ParquetDecodeContext& _context;
+    ParquetMaterializationState* _state;
+};
+
+class RejectDateTimeV2BinaryConsumer final : public ParquetBinaryValueConsumer {
+public:
+    Status consume(const StringRef* values, size_t num_values) override {
+        return Status::NotSupported("Binary Parquet values cannot be materialized as DATETIMEV2");
+    }
+};
+
 } // namespace
 
 // NOLINTBEGIN(readability-function-size)
@@ -559,16 +618,27 @@ Status DataTypeDateTimeV2SerDe::read_column_from_decoded_values(
     auto& data = assert_cast<ColumnDateTimeV2&>(column).get_data();
     const auto old_size = data.size();
     if (view.value_kind == DecodedValueKind::INT96) {
-        const auto* values = reinterpret_cast<const DecodedInt96Timestamp*>(view.values);
+        const auto* values = reinterpret_cast<const ParquetInt96Timestamp*>(view.values);
         static const auto utc_timezone = cctz::utc_time_zone();
         const auto& timezone = view.timezone == nullptr ? utc_timezone : *view.timezone;
         for (int64_t row = 0; row < view.row_count; ++row) {
             if (decoded_column_view_row_is_null(view, row)) {
                 data.push_back(DateV2Value<DateTimeV2ValueType>());
                 continue;
             }
-            append_datetimev2_from_utc_epoch_micros(data, values[row].to_timestamp_micros(),
-                                                    timezone);
+            int64_t timestamp_micros;
+            auto status = parquet_int96_timestamp_micros(values[row], &timestamp_micros);
+            if (status.ok()) {
+                status = append_datetimev2_from_utc_epoch_micros(data, timestamp_micros, timezone);
+            }
+            if (!status.ok()) {
+                if (decoded_column_view_can_null_on_conversion_failure(view)) {
+                    decoded_column_view_insert_null_on_conversion_failure(column, view, row);
+                    continue;
+                }
+                data.resize(old_size);
+                return status;
+            }
         }
         return Status::OK();
     }
@@ -581,18 +651,70 @@ Status DataTypeDateTimeV2SerDe::read_column_from_decoded_values(
             data.push_back(DateV2Value<DateTimeV2ValueType>());
             continue;
         }
-        const int64_t timestamp_micros = decoded_timestamp_micros(view, values[row]);
-        if (view.timestamp_is_adjusted_to_utc) {
-            append_datetimev2_from_utc_epoch_micros(data, timestamp_micros, timezone);
-        } else {
-            auto st = append_datetimev2_from_epoch_micros(data, timestamp_micros);
-            if (!st.ok()) {
-                if (decoded_column_view_can_null_on_conversion_failure(view)) {
-                    decoded_column_view_insert_null_on_conversion_failure(column, view, row);
-                    continue;
-                }
-                data.resize(old_size);
-                return st;
+        int64_t timestamp_micros;
+        auto status = parquet_timestamp_micros(decoded_parquet_time_unit(view), values[row],
+                                               &timestamp_micros);
+        if (status.ok()) {
+            status = view.timestamp_is_adjusted_to_utc
+                             ? append_datetimev2_from_utc_epoch_micros(data, timestamp_micros,
+                                                                       timezone)
+                             : append_datetimev2_from_epoch_micros(data, timestamp_micros);
+        }
+        if (!status.ok()) {
+            if (decoded_column_view_can_null_on_conversion_failure(view)) {
+                decoded_column_view_insert_null_on_conversion_failure(column, view, row);
+                continue;
+            }
+            data.resize(old_size);
+            return status;
+        }
+    }
+    return Status::OK();
+}
+
+Status DataTypeDateTimeV2SerDe::read_parquet_dictionary(IColumn& column,
+                                                        ParquetDecodeSource& source,
+                                                        const ParquetDecodeContext& context) const {
+    DateTimeV2ParquetConsumer consumer(column, context);
+    RejectDateTimeV2BinaryConsumer binary_consumer;
+    return source.decode_dictionary(consumer, binary_consumer);
+}
+
+Status DataTypeDateTimeV2SerDe::read_column_from_parquet(IColumn& column,
+                                                         ParquetDecodeSource& source,
+                                                         const ParquetDecodeContext& context,
+                                                         size_t num_values,
+                                                         ParquetMaterializationState& state) const {
+    if (context.physical_type != ParquetPhysicalType::INT64 &&
+        context.physical_type != ParquetPhysicalType::INT96) {
+        return Status::NotSupported("DATETIMEV2 expects Parquet INT64 or INT96");
+    }
+    DateTimeV2ParquetConsumer consumer(column, context, &state);
+    if (context.encoding != ParquetValueEncoding::DICTIONARY) {
+        return source.decode_fixed_values(num_values, consumer);
+    }
+    if (state.dictionary_generation != source.dictionary_generation()) {
+        state.typed_dictionary = column.clone_empty();
+        auto* output_null_map = state.begin_dictionary_conversion(source.dictionary_size());
+        DateTimeV2ParquetConsumer dictionary_consumer(*state.typed_dictionary, context, &state);
+        RejectDateTimeV2BinaryConsumer binary_consumer;
+        const Status dictionary_status =
+                source.decode_dictionary(dictionary_consumer, binary_consumer);
+        state.end_dictionary_conversion(output_null_map);
+        RETURN_IF_ERROR(dictionary_status);
+        DORIS_CHECK_EQ(state.typed_dictionary->size(), source.dictionary_size());
+        state.dictionary_generation = source.dictionary_generation();
+    }
+    RETURN_IF_ERROR(source.decode_dictionary_indices(num_values, &state.dictionary_indices));
+    DORIS_CHECK_EQ(state.dictionary_indices.size(), num_values);
+    const size_t old_size = column.size();
+    column.insert_indices_from(*state.typed_dictionary, state.dictionary_indices.data(),
+                               state.dictionary_indices.data() + num_values);
+    if (state.can_insert_null_on_conversion_failure()) {
+        for (size_t row = 0; row < num_values; ++row) {
+            if (!state.dictionary_conversion_failures.empty() &&
+                state.dictionary_conversion_failures[state.dictionary_indices[row]] != 0) {
+                state.mark_conversion_failure(old_size + row);
             }
         }
     }

be/src/core/data_type_serde/data_type_datetimev2_serde.h

@@ -90,6 +90,11 @@ class DataTypeDateTimeV2SerDe : public DataTypeNumberSerDe<PrimitiveType::TYPE_D
                                   int64_t end, const cctz::time_zone& ctz) const override;
     Status read_column_from_decoded_values(IColumn& column,
                                            const DecodedColumnView& view) const override;
+    Status read_column_from_parquet(IColumn& column, ParquetDecodeSource& source,
+                                    const ParquetDecodeContext& context, size_t num_values,
+                                    ParquetMaterializationState& state) const override;
+    Status read_parquet_dictionary(IColumn& column, ParquetDecodeSource& source,
+                                   const ParquetDecodeContext& context) const override;
     Status read_column_from_orc(IColumn& column, const OrcDecodedColumnView& view) const override;
 
     Status write_column_to_mysql_binary(const IColumn& column, MysqlRowBinaryBuffer& row_buffer,