| // 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 <cctz/time_zone.h> |
| #include <gen_cpp/parquet_types.h> |
| #include <libdivide.h> |
| |
| #include <chrono> |
| #include <limits> |
| |
| #include "common/cast_set.h" |
| #include "core/column/column_fixed_length_object.h" |
| #include "core/column/column_varbinary.h" |
| #include "core/column/column_vector.h" |
| #include "core/data_type/data_type_factory.hpp" |
| #include "core/data_type/primitive_type.h" |
| #include "core/extended_types.h" |
| #include "core/field.h" |
| #include "core/types.h" |
| #include "format/column_type_convert.h" |
| #include "format/format_common.h" |
| #include "format/parquet/decoder.h" |
| #include "format/parquet/parquet_common.h" |
| #include "format/parquet/schema_desc.h" |
| #include "util/timezone_utils.h" |
| |
| namespace doris::parquet { |
| namespace detail { |
| |
| struct EpochSecondsAndMicros { |
| int64_t seconds; |
| uint32_t microseconds; |
| }; |
| |
| inline EpochSecondsAndMicros split_epoch_time(int64_t timestamp, int64_t units_per_second, |
| int64_t nanos_per_unit) { |
| int64_t seconds = timestamp / units_per_second; |
| int64_t subsecond = timestamp % units_per_second; |
| // C++ division truncates toward zero. Normalize to floor division so the fractional part is |
| // always non-negative for timestamps before the Unix epoch. |
| if (subsecond < 0) { |
| subsecond += units_per_second; |
| --seconds; |
| } |
| return {.seconds = seconds, |
| .microseconds = static_cast<uint32_t>(subsecond * nanos_per_unit / 1000)}; |
| } |
| |
| inline bool try_split_local_time(int64_t local_time, uint16_t* year, uint8_t* month, uint8_t* day, |
| uint8_t* hour, uint8_t* minute, uint8_t* second) { |
| static const libdivide::divider<int64_t> fast_div_86400(86400); |
| static const libdivide::divider<int64_t> fast_div_3600(3600); |
| static const libdivide::divider<int64_t> fast_div_60(60); |
| static constexpr int64_t kMinSupportedDays = -365LL * 10000; |
| static constexpr int64_t kMaxSupportedDays = 365LL * 10000; |
| |
| int64_t days = local_time / fast_div_86400; |
| int64_t second_of_day = local_time - days * 86400; |
| if (second_of_day < 0) { |
| second_of_day += 86400; |
| --days; |
| } |
| if (days < kMinSupportedDays || days > kMaxSupportedDays) { |
| return false; |
| } |
| |
| const auto ymd = std::chrono::year_month_day {std::chrono::sys_days {std::chrono::days {days}}}; |
| const int y = static_cast<int>(ymd.year()); |
| if (y < 0 || y > std::numeric_limits<uint16_t>::max()) { |
| return false; |
| } |
| |
| const int64_t h = second_of_day / fast_div_3600; |
| const int64_t minute_second = second_of_day - h * 3600; |
| const int64_t m = minute_second / fast_div_60; |
| const int64_t s = minute_second - m * 60; |
| |
| *year = static_cast<uint16_t>(y); |
| *month = static_cast<uint8_t>(static_cast<unsigned>(ymd.month())); |
| *day = static_cast<uint8_t>(static_cast<unsigned>(ymd.day())); |
| *hour = static_cast<uint8_t>(h); |
| *minute = static_cast<uint8_t>(m); |
| *second = static_cast<uint8_t>(s); |
| return true; |
| } |
| |
| template <typename DateType> |
| inline bool try_convert_timestamp_with_fixed_offset(DateType& value, int64_t epoch_seconds, |
| int32_t offset_seconds) { |
| uint16_t year = 0; |
| uint8_t month = 0; |
| uint8_t day = 0; |
| uint8_t hour = 0; |
| uint8_t minute = 0; |
| uint8_t second = 0; |
| if (!try_split_local_time(epoch_seconds + offset_seconds, &year, &month, &day, &hour, &minute, |
| &second)) { |
| return false; |
| } |
| // The caller sets sub-second precision immediately after this conversion. |
| value.unchecked_set_time(year, month, day, hour, minute, second, 0); |
| return true; |
| } |
| |
| template <typename DateType> |
| inline bool try_convert_timestamp_with_lookup(DateType& value, int64_t epoch_seconds, |
| const cctz::time_zone& ctz) { |
| static const auto epoch = std::chrono::time_point_cast<cctz::sys_seconds>( |
| std::chrono::system_clock::from_time_t(0)); |
| cctz::time_point<cctz::sys_seconds> t = epoch + cctz::seconds(epoch_seconds); |
| const int32_t offset = ctz.lookup_offset(t).offset; |
| return try_convert_timestamp_with_fixed_offset(value, epoch_seconds, offset); |
| } |
| |
| } // namespace detail |
| |
| struct ConvertParams { |
| // schema.logicalType.TIMESTAMP.isAdjustedToUTC == false |
| static const cctz::time_zone utc0; |
| // schema.logicalType.TIMESTAMP.isAdjustedToUTC == true, we should set local time zone |
| const cctz::time_zone* ctz = nullptr; |
| bool is_fixed_offset = false; |
| int32_t fixed_offset_seconds = 0; |
| int64_t second_mask = 1; |
| int64_t scale_to_nano_factor = 1; |
| const FieldSchema* field_schema = nullptr; |
| |
| //For UInt8 -> Int16,UInt16 -> Int32,UInt32 -> Int64,UInt64 -> Int128. |
| bool is_type_compatibility = false; |
| |
| /** |
| * Some frameworks like paimon maybe writes non-standard parquet files. Timestamp field doesn't have |
| * logicalType or converted_type to indicates its precision. We have to reset the time mask. |
| */ |
| void reset_time_scale_if_missing(int scale) { |
| const auto& schema = field_schema->parquet_schema; |
| if (!schema.__isset.logicalType && !schema.__isset.converted_type) { |
| int ts_scale = 9; |
| if (scale <= 3) { |
| ts_scale = 3; |
| } else if (scale <= 6) { |
| ts_scale = 6; |
| } |
| second_mask = common::exp10_i64(ts_scale); |
| scale_to_nano_factor = common::exp10_i64(9 - ts_scale); |
| |
| // The missing parque metadata makes it impossible for us to know the time zone information, |
| // so we default to UTC here. |
| if (ctz == nullptr) { |
| ctz = &utc0; |
| } |
| } |
| } |
| |
| void init(const FieldSchema* field_schema_, const cctz::time_zone* ctz_) { |
| field_schema = field_schema_; |
| if (ctz_ != nullptr) { |
| ctz = ctz_; |
| } |
| const auto& schema = field_schema->parquet_schema; |
| if (schema.__isset.logicalType && schema.logicalType.__isset.TIMESTAMP) { |
| const auto& timestamp_info = schema.logicalType.TIMESTAMP; |
| if (!timestamp_info.isAdjustedToUTC) { |
| // should set timezone to utc+0 |
| // Reference: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#instant-semantics-timestamps-normalized-to-utc |
| // If isAdjustedToUTC = false, the reader should display the same value no mater what local time zone is. For example: |
| // When a timestamp is stored as `1970-01-03 12:00:00`, |
| // if isAdjustedToUTC = true, UTC8 should read as `1970-01-03 20:00:00`, UTC6 should read as `1970-01-03 18:00:00` |
| // if isAdjustedToUTC = false, UTC8 and UTC6 should read as `1970-01-03 12:00:00`, which is the same as `1970-01-03 12:00:00` in UTC0 |
| ctz = &utc0; |
| } |
| const auto& time_unit = timestamp_info.unit; |
| if (time_unit.__isset.MILLIS) { |
| second_mask = 1000; |
| scale_to_nano_factor = 1000000; |
| } else if (time_unit.__isset.MICROS) { |
| second_mask = 1000000; |
| scale_to_nano_factor = 1000; |
| } else if (time_unit.__isset.NANOS) { |
| second_mask = 1000000000; |
| scale_to_nano_factor = 1; |
| } |
| } else if (schema.__isset.converted_type) { |
| const auto& converted_type = schema.converted_type; |
| if (converted_type == tparquet::ConvertedType::TIMESTAMP_MILLIS) { |
| second_mask = 1000; |
| scale_to_nano_factor = 1000000; |
| } else if (converted_type == tparquet::ConvertedType::TIMESTAMP_MICROS) { |
| second_mask = 1000000; |
| scale_to_nano_factor = 1000; |
| } |
| } |
| |
| if (ctz != nullptr) { |
| is_fixed_offset = |
| TimezoneUtils::try_get_fixed_offset_seconds(*ctz, &fixed_offset_seconds); |
| } |
| is_type_compatibility = field_schema_->is_type_compatibility; |
| } |
| }; |
| |
| inline IColumn* get_mutable_inner_column(ColumnPtr& column) { |
| column = IColumn::mutate(std::move(column)); |
| auto mutable_column = column->assert_mutable(); |
| if (is_column_nullable(*mutable_column)) { |
| return &assert_cast<ColumnNullable*>(mutable_column.get())->get_nested_column(); |
| } |
| return mutable_column.get(); |
| } |
| |
| inline size_t get_mutable_inner_column_size(const ColumnPtr& column) { |
| if (is_column_nullable(*column)) { |
| const auto* nullable = assert_cast<const ColumnNullable*>(column.get()); |
| return nullable->get_nested_column().size(); |
| } |
| return column->size(); |
| } |
| |
| inline size_t get_null_map_size_or_inner_column_size(const ColumnPtr& column) { |
| if (is_column_nullable(*column)) { |
| const auto* nullable = assert_cast<const ColumnNullable*>(column.get()); |
| return nullable->get_null_map_column().size(); |
| } |
| return column->size(); |
| } |
| |
| inline size_t get_appended_null_map_start(const ColumnPtr& column, size_t new_rows) { |
| if (!is_column_nullable(*column)) { |
| return 0; |
| } |
| const auto* nullable = assert_cast<const ColumnNullable*>(column.get()); |
| const size_t null_map_size = nullable->get_null_map_column().size(); |
| DCHECK_GE(null_map_size, new_rows); |
| return null_map_size - new_rows; |
| } |
| |
| inline void align_null_map(ColumnPtr& src_column, ColumnPtr& dst_column, size_t old_null_map_size, |
| size_t new_rows, size_t src_null_map_start = 0) { |
| if (!is_column_nullable(*dst_column)) { |
| return; |
| } |
| |
| dst_column = IColumn::mutate(std::move(dst_column)); |
| auto* dst_nullable = assert_cast<ColumnNullable*>(dst_column->assert_mutable().get()); |
| auto& dst_null_map = dst_nullable->get_null_map_column(); |
| const size_t expected_rows = old_null_map_size + new_rows; |
| if (dst_null_map.size() == expected_rows) { |
| return; |
| } |
| DCHECK_EQ(dst_null_map.size(), old_null_map_size); |
| if (is_column_nullable(*src_column)) { |
| const auto* src_nullable = assert_cast<const ColumnNullable*>(src_column.get()); |
| DCHECK_GE(src_nullable->get_null_map_column().size(), src_null_map_start + new_rows); |
| dst_null_map.insert_range_from(src_nullable->get_null_map_column(), src_null_map_start, |
| new_rows); |
| } else { |
| dst_null_map.insert_many_vals(0, new_rows); |
| } |
| } |
| |
| struct FixedLengthPhysicalData { |
| const uint8_t* data = nullptr; |
| size_t byte_size = 0; |
| size_t rows = 0; |
| }; |
| |
| inline FixedLengthPhysicalData get_fixed_length_physical_data(const IColumn& column, |
| size_t type_length) { |
| if (const auto* fixed_length_column = check_and_get_column<ColumnFixedLengthObject>(column)) { |
| DCHECK_EQ(fixed_length_column->item_size(), type_length); |
| return {fixed_length_column->get_data().data(), fixed_length_column->byte_size(), |
| fixed_length_column->size()}; |
| } |
| |
| const auto& uint8_column = assert_cast<const ColumnUInt8&>(column); |
| DCHECK_EQ(uint8_column.size() % type_length, 0); |
| return {uint8_column.get_data().data(), uint8_column.size(), uint8_column.size() / type_length}; |
| } |
| |
| /** |
| * Convert parquet physical column to logical column |
| * In parquet document(https://github.com/apache/parquet-format/blob/master/LogicalTypes.md), |
| * Logical or converted type is the data type of column, physical type is the stored type of column chunk. |
| * eg, decimal type can be stored as INT32, INT64, BYTE_ARRAY, FIXED_LENGTH_BYTE_ARRAY. |
| * So there is a convert process from physical type to logical type. |
| * In addition, Schema change will bring about a change in logical type. |
| * |
| * `PhysicalToLogicalConverter` strips away the conversion of logical type, and reuse `ColumnTypeConverter` |
| * to resolve schema change, allowing parquet reader to only focus on the conversion of physical types. |
| * |
| * Therefore, tow layers converters are designed: |
| * First, read parquet data with the physical type |
| * Second, convert physical type to logical type |
| * Third, convert logical type to the final type planned by FE(schema change) |
| * |
| * Ultimate performance optimization: |
| * 1. If process of (First => Second) is consistent, eg. from BYTE_ARRAY to string, no additional copies and conversions will be introduced; |
| * 2. If process of (Second => Third) is consistent, no additional copies and conversions will be introduced; |
| * 3. Null maps are owned by each temporary nullable column, and only appended null slices are |
| * copied between conversion stages; |
| * 4. Only create one physical column in physical conversion, and reused in each loop; |
| * 5. Only create one logical column in logical conversion, and reused in each loop; |
| * 6. FIXED_LENGTH_BYTE_ARRAY is read as ColumnFixedLengthObject instead of ColumnString, so |
| * the decoder can copy fixed-size values as a whole while keeping nullable row counts valid. |
| */ |
| class PhysicalToLogicalConverter { |
| protected: |
| ColumnPtr _cached_src_physical_column = nullptr; |
| DataTypePtr _cached_src_physical_type = nullptr; |
| std::unique_ptr<converter::ColumnTypeConverter> _logical_converter = nullptr; |
| |
| std::string _error_msg; |
| |
| std::unique_ptr<ConvertParams> _convert_params; |
| |
| public: |
| static std::unique_ptr<PhysicalToLogicalConverter> get_converter( |
| const FieldSchema* field_schema, DataTypePtr src_logical_type, |
| const DataTypePtr& dst_logical_type, const cctz::time_zone* ctz, |
| bool is_dict_filter = false, bool preserve_binary_uuid = false); |
| |
| static bool is_parquet_native_type(PrimitiveType type); |
| |
| static bool is_decimal_type(PrimitiveType type); |
| |
| PhysicalToLogicalConverter() = default; |
| virtual ~PhysicalToLogicalConverter() = default; |
| |
| virtual Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) { |
| return Status::OK(); |
| } |
| |
| Status convert(ColumnPtr& src_physical_col, DataTypePtr src_logical_type, |
| const DataTypePtr& dst_logical_type, ColumnPtr& dst_logical_col, |
| bool is_dict_filter) { |
| if (is_dict_filter) { |
| src_logical_type = DataTypeFactory::instance().create_data_type( |
| PrimitiveType::TYPE_INT, dst_logical_type->is_nullable()); |
| } |
| if (is_consistent() && _logical_converter->is_consistent()) { |
| dst_logical_col = std::move(src_physical_col); |
| return Status::OK(); |
| } |
| if (_logical_converter->is_consistent()) { |
| const size_t old_rows = get_mutable_inner_column_size(dst_logical_col); |
| const size_t old_null_map_size = |
| get_null_map_size_or_inner_column_size(dst_logical_col); |
| RETURN_IF_ERROR(physical_convert(src_physical_col, dst_logical_col)); |
| const size_t new_rows = get_mutable_inner_column_size(dst_logical_col) - old_rows; |
| align_null_map(src_physical_col, dst_logical_col, old_null_map_size, new_rows, |
| get_appended_null_map_start(src_physical_col, new_rows)); |
| return Status::OK(); |
| } |
| |
| ColumnPtr src_logical_column; |
| if (is_consistent()) { |
| src_logical_column = src_physical_col; |
| } else { |
| src_logical_column = _logical_converter->get_column(src_logical_type, dst_logical_col, |
| dst_logical_type); |
| } |
| const size_t src_old_rows = get_mutable_inner_column_size(src_logical_column); |
| const size_t src_old_null_map_size = |
| get_null_map_size_or_inner_column_size(src_logical_column); |
| RETURN_IF_ERROR(physical_convert(src_physical_col, src_logical_column)); |
| const size_t src_new_rows = |
| get_mutable_inner_column_size(src_logical_column) - src_old_rows; |
| align_null_map(src_physical_col, src_logical_column, src_old_null_map_size, src_new_rows, |
| get_appended_null_map_start(src_physical_col, src_new_rows)); |
| |
| dst_logical_col = IColumn::mutate(std::move(dst_logical_col)); |
| const size_t dst_old_rows = get_mutable_inner_column_size(dst_logical_col); |
| const size_t dst_old_null_map_size = |
| get_null_map_size_or_inner_column_size(dst_logical_col); |
| auto converted_column = dst_logical_col->assert_mutable(); |
| RETURN_IF_ERROR(_logical_converter->convert(src_logical_column, converted_column)); |
| const size_t dst_new_rows = get_mutable_inner_column_size(dst_logical_col) - dst_old_rows; |
| align_null_map(src_logical_column, dst_logical_col, dst_old_null_map_size, dst_new_rows, |
| get_appended_null_map_start(src_logical_column, dst_new_rows)); |
| return Status::OK(); |
| } |
| |
| virtual ColumnPtr get_physical_column(tparquet::Type::type src_physical_type, |
| DataTypePtr src_logical_type, |
| ColumnPtr& dst_logical_column, |
| const DataTypePtr& dst_logical_type, bool is_dict_filter); |
| |
| DataTypePtr& get_physical_type() { return _cached_src_physical_type; } |
| |
| bool read_directly_into_dst_logical_column() { |
| return !_convert_params->is_type_compatibility && is_consistent() && |
| _logical_converter->is_consistent(); |
| } |
| |
| virtual bool is_consistent() { return false; } |
| |
| virtual bool support() { return true; } |
| |
| std::string get_error_msg() { return _error_msg; } |
| }; |
| |
| class ConsistentPhysicalConverter : public PhysicalToLogicalConverter { |
| bool is_consistent() override { return true; } |
| }; |
| |
| class UnsupportedConverter : public PhysicalToLogicalConverter { |
| public: |
| UnsupportedConverter(std::string error_msg) { _error_msg = error_msg; } |
| |
| UnsupportedConverter(tparquet::Type::type src_physical_type, |
| const DataTypePtr& src_logical_type) { |
| std::string src_physical_str = tparquet::to_string(src_physical_type); |
| std::string src_logical_str = src_logical_type->get_name(); |
| _error_msg = src_physical_str + " => " + src_logical_str; |
| } |
| |
| bool support() override { return false; } |
| |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| return Status::InternalError("Unsupported physical to logical type: {}", _error_msg); |
| } |
| }; |
| |
| // for tinyint, smallint |
| template <PrimitiveType IntPrimitiveType> |
| class LittleIntPhysicalConverter : public PhysicalToLogicalConverter { |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| using DstCppType = typename PrimitiveTypeTraits<IntPrimitiveType>::CppType; |
| using DstColumnType = typename PrimitiveTypeTraits<IntPrimitiveType>::ColumnType; |
| ColumnPtr from_col = remove_nullable(src_physical_col); |
| IColumn* to_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = from_col->size(); |
| // always comes from tparquet::Type::INT32 |
| auto& src_data = assert_cast<const ColumnInt32*>(from_col.get())->get_data(); |
| size_t start_idx = to_col->size(); |
| to_col->resize(start_idx + rows); |
| auto& data = assert_cast<DstColumnType&>(*to_col).get_data(); |
| for (int i = 0; i < rows; ++i) { |
| data[start_idx + i] = static_cast<DstCppType>(src_data[i]); |
| } |
| |
| return Status::OK(); |
| } |
| }; |
| |
| template <PrimitiveType type> |
| struct UnsignedTypeTraits; |
| |
| template <> |
| struct UnsignedTypeTraits<TYPE_SMALLINT> { |
| using UnsignedCppType = UInt8; |
| //https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#unsigned-integers |
| //INT(8, false), INT(16, false), and INT(32, false) must annotate an int32 primitive type and INT(64, false) |
| //must annotate an int64 primitive type. |
| using StorageCppType = Int32; |
| using StorageColumnType = ColumnInt32; |
| }; |
| |
| template <> |
| struct UnsignedTypeTraits<TYPE_INT> { |
| using UnsignedCppType = UInt16; |
| using StorageCppType = Int32; |
| using StorageColumnType = ColumnInt32; |
| }; |
| |
| template <> |
| struct UnsignedTypeTraits<TYPE_BIGINT> { |
| using UnsignedCppType = UInt32; |
| using StorageCppType = Int32; |
| using StorageColumnType = ColumnInt32; |
| }; |
| |
| template <> |
| struct UnsignedTypeTraits<TYPE_LARGEINT> { |
| using UnsignedCppType = UInt64; |
| using StorageCppType = Int64; |
| using StorageColumnType = ColumnInt64; |
| }; |
| |
| template <PrimitiveType IntPrimitiveType> |
| class UnsignedIntegerConverter : public PhysicalToLogicalConverter { |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| using UnsignedCppType = typename UnsignedTypeTraits<IntPrimitiveType>::UnsignedCppType; |
| using StorageCppType = typename UnsignedTypeTraits<IntPrimitiveType>::StorageCppType; |
| using StorageColumnType = typename UnsignedTypeTraits<IntPrimitiveType>::StorageColumnType; |
| using DstColumnType = typename PrimitiveTypeTraits<IntPrimitiveType>::ColumnType; |
| |
| ColumnPtr from_col = remove_nullable(src_physical_col); |
| IColumn* to_col = get_mutable_inner_column(src_logical_column); |
| auto& src_data = assert_cast<const StorageColumnType*>(from_col.get())->get_data(); |
| |
| size_t rows = src_data.size(); |
| size_t start_idx = to_col->size(); |
| to_col->resize(start_idx + rows); |
| auto& data = assert_cast<DstColumnType&>(*to_col).get_data(); |
| |
| for (int i = 0; i < rows; i++) { |
| StorageCppType src_value = src_data[i]; |
| auto unsigned_value = static_cast<UnsignedCppType>(src_value); |
| data[start_idx + i] = unsigned_value; |
| } |
| |
| return Status::OK(); |
| } |
| }; |
| |
| class FixedSizeBinaryConverter : public PhysicalToLogicalConverter { |
| private: |
| int _type_length; |
| |
| public: |
| FixedSizeBinaryConverter(int type_length) : _type_length(type_length) {} |
| |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr from_col = remove_nullable(src_physical_col); |
| IColumn* to_col = get_mutable_inner_column(src_logical_column); |
| |
| const auto src_data = get_fixed_length_physical_data(*from_col, _type_length); |
| size_t length = src_data.byte_size; |
| size_t num_values = src_data.rows; |
| auto& string_col = static_cast<ColumnString&>(*to_col); |
| auto& offsets = string_col.get_offsets(); |
| auto& chars = string_col.get_chars(); |
| |
| size_t origin_size = chars.size(); |
| chars.resize(origin_size + length); |
| memcpy(chars.data() + origin_size, src_data.data, length); |
| |
| origin_size = offsets.size(); |
| offsets.resize(origin_size + num_values); |
| auto end_offset = offsets[origin_size - 1]; |
| for (int i = 0; i < num_values; ++i) { |
| end_offset += _type_length; |
| offsets[origin_size + i] = end_offset; |
| } |
| |
| return Status::OK(); |
| } |
| }; |
| |
| class Float16PhysicalConverter : public PhysicalToLogicalConverter { |
| private: |
| int _type_length; |
| |
| public: |
| Float16PhysicalConverter(int type_length) : _type_length(type_length) { |
| DCHECK_EQ(_type_length, 2); |
| } |
| |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr from_col = remove_nullable(src_physical_col); |
| IColumn* to_col = get_mutable_inner_column(src_logical_column); |
| |
| const auto src_data = get_fixed_length_physical_data(*from_col, _type_length); |
| size_t num_values = src_data.rows; |
| auto* to_float_column = assert_cast<ColumnFloat32*>(to_col); |
| size_t start_idx = to_float_column->size(); |
| to_float_column->resize(start_idx + num_values); |
| auto& to_float_column_data = to_float_column->get_data(); |
| const auto* ptr = src_data.data; |
| for (int i = 0; i < num_values; ++i) { |
| size_t offset = i * _type_length; |
| const auto* data_ptr = ptr + offset; |
| uint16_t raw; |
| memcpy(&raw, data_ptr, sizeof(uint16_t)); |
| float value = half_to_float(raw); |
| to_float_column_data[start_idx + i] = value; |
| } |
| |
| return Status::OK(); |
| } |
| |
| float half_to_float(uint16_t h) { |
| // uint16_t h: half precision floating point |
| // bit 15: sign（1 bit） |
| // bits 14..10 : exponent（5 bits） |
| // bits 9..0 : mantissa（10 bits） |
| |
| // sign bit placed to float32 bit31 |
| uint32_t sign = (h & 0x8000U) << 16; // 0x8000 << 16 = 0x8000_0000 |
| // exponent:（5 bits） |
| uint32_t exp = (h & 0x7C00U) >> 10; // 0x7C00 = 0111 1100 0000 (half exponent mask) |
| // mantissa（10 bits） |
| uint32_t mant = (h & 0x03FFU); // 10-bit fraction |
| |
| // cases：Zero/Subnormal, Normal, Inf/NaN |
| if (exp == 0) { |
| // exp==0: Zero or Subnormal ---------- |
| if (mant == 0) { |
| // ±0.0 |
| // sign = either 0x00000000 or 0x80000000 |
| return std::bit_cast<float>(sign); |
| } else { |
| // ---------- Subnormal ---------- |
| // half subnormal: |
| // value = (-1)^sign * (mant / 2^10) * 2^(1 - bias) |
| // half bias = 15 → exponent = 1 - 15 = -14 |
| float f = (static_cast<float>(mant) / 1024.0F) * std::powf(2.0F, -14.0F); |
| return sign ? -f : f; |
| } |
| } else if (exp == 0x1F) { |
| // exp==31: Inf or NaN ---------- |
| // float32: |
| // exponent = 255 (0xFF) |
| // mantissa = mant << 13 |
| uint32_t f = sign | 0x7F800000U | (mant << 13); |
| return std::bit_cast<float>(f); |
| } else { |
| // Normalized ---------- |
| // float32 exponent: |
| // exp32 = exp16 - bias16 + bias32 |
| // bias16 = 15 |
| // bias32 = 127 |
| // |
| // so: exp32 = exp + (127 - 15) |
| uint32_t f = sign | ((exp + (127 - 15)) << 23) // place to float32 exponent |
| | (mant << 13); // mantissa align to 23 bits |
| return std::bit_cast<float>(f); |
| } |
| } |
| }; |
| |
| class UUIDVarBinaryConverter : public PhysicalToLogicalConverter { |
| public: |
| UUIDVarBinaryConverter(int type_length) : _type_length(type_length) {} |
| |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| DCHECK(!is_column_const(*src_physical_col)) << src_physical_col->dump_structure(); |
| DCHECK(!is_column_const(*src_logical_column)) << src_logical_column->dump_structure(); |
| const ColumnPtr from_col = remove_nullable(src_physical_col); |
| const auto src_data = get_fixed_length_physical_data(*from_col, _type_length); |
| |
| IColumn* to_col = get_mutable_inner_column(src_logical_column); |
| auto* to_varbinary_column = assert_cast<ColumnVarbinary*>(to_col); |
| size_t num_values = src_data.rows; |
| const auto* ptr = src_data.data; |
| |
| for (int i = 0; i < num_values; ++i) { |
| auto offset = i * _type_length; |
| const char* data_ptr = reinterpret_cast<const char*>(ptr + offset); |
| to_varbinary_column->insert_data(data_ptr, _type_length); |
| } |
| return Status::OK(); |
| } |
| |
| private: |
| int _type_length; |
| }; |
| |
| template <PrimitiveType DecimalPType> |
| class FixedSizeToDecimal : public PhysicalToLogicalConverter { |
| public: |
| using DecimalType = typename PrimitiveTypeTraits<DecimalPType>::CppType; |
| FixedSizeToDecimal(int32_t type_length) : _type_length(type_length) {} |
| |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| #define M(FixedTypeLength, ValueCopyType) \ |
| case FixedTypeLength: \ |
| return _convert_internal<FixedTypeLength, ValueCopyType>(src_col, dst_col); |
| |
| #define APPLY_FOR_DECIMALS() \ |
| M(1, int64_t) \ |
| M(2, int64_t) \ |
| M(3, int64_t) \ |
| M(4, int64_t) \ |
| M(5, int64_t) \ |
| M(6, int64_t) \ |
| M(7, int64_t) \ |
| M(8, int64_t) \ |
| M(9, int128_t) \ |
| M(10, int128_t) \ |
| M(11, int128_t) \ |
| M(12, int128_t) \ |
| M(13, int128_t) \ |
| M(14, int128_t) \ |
| M(15, int128_t) \ |
| M(16, int128_t) \ |
| M(17, wide::Int256) \ |
| M(18, wide::Int256) \ |
| M(19, wide::Int256) \ |
| M(20, wide::Int256) \ |
| M(21, wide::Int256) \ |
| M(22, wide::Int256) \ |
| M(23, wide::Int256) \ |
| M(24, wide::Int256) \ |
| M(25, wide::Int256) \ |
| M(26, wide::Int256) \ |
| M(27, wide::Int256) \ |
| M(28, wide::Int256) \ |
| M(29, wide::Int256) \ |
| M(30, wide::Int256) \ |
| M(31, wide::Int256) \ |
| M(32, wide::Int256) |
| |
| switch (_type_length) { |
| APPLY_FOR_DECIMALS() |
| default: |
| throw Exception(Status::FatalError("__builtin_unreachable")); |
| } |
| return Status::OK(); |
| #undef APPLY_FOR_DECIMALS |
| #undef M |
| } |
| |
| template <int fixed_type_length, typename ValueCopyType> |
| Status _convert_internal(ColumnPtr& src_col, IColumn* dst_col) { |
| const auto src_data = get_fixed_length_physical_data(*src_col, fixed_type_length); |
| size_t rows = src_data.rows; |
| const auto* buf = src_data.data; |
| size_t start_idx = dst_col->size(); |
| dst_col->resize(start_idx + rows); |
| |
| auto& data = static_cast<ColumnDecimal<DecimalPType>*>(dst_col)->get_data(); |
| size_t offset = 0; |
| for (int i = 0; i < rows; i++) { |
| // When Decimal in parquet is stored in byte arrays, binary and fixed, |
| // the unscaled number must be encoded as two's complement using big-endian byte order. |
| ValueCopyType value = 0; |
| memcpy(reinterpret_cast<char*>(&value), buf + offset, sizeof(value)); |
| offset += fixed_type_length; |
| value = to_endian<std::endian::big>(value); |
| value = value >> ((sizeof(value) - fixed_type_length) * 8); |
| auto& v = reinterpret_cast<DecimalType&>(data[start_idx + i]); |
| v = (DecimalType)value; |
| } |
| |
| return Status::OK(); |
| } |
| |
| private: |
| int32_t _type_length; |
| }; |
| |
| template <PrimitiveType DecimalPType> |
| class StringToDecimal : public PhysicalToLogicalConverter { |
| using DecimalType = typename PrimitiveTypeTraits<DecimalPType>::CppType; |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| using ValueCopyType = DecimalType::NativeType; |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = src_col->size(); |
| auto buf = static_cast<const ColumnString*>(src_col.get())->get_chars().data(); |
| auto& offset = static_cast<const ColumnString*>(src_col.get())->get_offsets(); |
| size_t start_idx = dst_col->size(); |
| dst_col->resize(start_idx + rows); |
| |
| auto& data = static_cast<ColumnDecimal<DecimalPType>*>(dst_col)->get_data(); |
| for (int i = 0; i < rows; i++) { |
| size_t len = offset[i] - offset[i - 1]; |
| // When Decimal in parquet is stored in byte arrays, binary and fixed, |
| // the unscaled number must be encoded as two's complement using big-endian byte order. |
| ValueCopyType value = 0; |
| if (len > 0) { |
| memcpy(reinterpret_cast<char*>(&value), buf + offset[i - 1], len); |
| value = to_endian<std::endian::big>(value); |
| value = value >> ((sizeof(value) - len) * 8); |
| } |
| auto& v = reinterpret_cast<DecimalType&>(data[start_idx + i]); |
| v = (DecimalType)value; |
| } |
| |
| return Status::OK(); |
| } |
| }; |
| |
| template <PrimitiveType NumberType, PrimitiveType DecimalPType> |
| class NumberToDecimal : public PhysicalToLogicalConverter { |
| using DecimalType = typename PrimitiveTypeTraits<DecimalPType>::CppType; |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| using ValueCopyType = typename DecimalType::NativeType; |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = src_col->size(); |
| auto* src_data = |
| static_cast<const ColumnVector<NumberType>*>(src_col.get())->get_data().data(); |
| size_t start_idx = dst_col->size(); |
| dst_col->resize(start_idx + rows); |
| |
| auto* data = static_cast<ColumnDecimal<DecimalPType>*>(dst_col)->get_data().data(); |
| |
| for (int i = 0; i < rows; i++) { |
| ValueCopyType value; |
| if constexpr (std::is_same_v<DecimalType, Decimal256>) { |
| value = src_data[i]; |
| } else { |
| value = cast_set<ValueCopyType, typename PrimitiveTypeTraits<NumberType>::CppType, |
| false>(src_data[i]); |
| } |
| |
| data[start_idx + i] = (DecimalType)value; |
| } |
| return Status::OK(); |
| } |
| }; |
| |
| class Int32ToDate : public PhysicalToLogicalConverter { |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = src_col->size(); |
| size_t start_idx = dst_col->size(); |
| dst_col->reserve(start_idx + rows); |
| |
| auto& src_data = static_cast<const ColumnInt32*>(src_col.get())->get_data(); |
| auto& data = static_cast<ColumnDateV2*>(dst_col)->get_data(); |
| date_day_offset_dict& date_dict = date_day_offset_dict::get(); |
| |
| for (int i = 0; i < rows; i++) { |
| data.push_back_without_reserve(date_dict[src_data[i]].to_date_int_val()); |
| } |
| |
| return Status::OK(); |
| } |
| }; |
| |
| struct Int64ToTimestamp : public PhysicalToLogicalConverter { |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = src_col->size(); |
| size_t start_idx = dst_col->size(); |
| dst_col->resize(start_idx + rows); |
| |
| auto src_data = static_cast<const ColumnInt64*>(src_col.get())->get_data().data(); |
| auto& data = static_cast<ColumnDateTimeV2*>(dst_col)->get_data(); |
| |
| for (int i = 0; i < rows; i++) { |
| int64_t x = src_data[i]; |
| auto& num = data[start_idx + i]; |
| auto& value = reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(num); |
| const auto epoch_time = detail::split_epoch_time(x, _convert_params->second_mask, |
| _convert_params->scale_to_nano_factor); |
| if (_convert_params->is_fixed_offset) { |
| if (!detail::try_convert_timestamp_with_fixed_offset( |
| value, epoch_time.seconds, _convert_params->fixed_offset_seconds)) { |
| value.from_unixtime(epoch_time.seconds, *_convert_params->ctz); |
| } |
| } else if (!detail::try_convert_timestamp_with_lookup(value, epoch_time.seconds, |
| *_convert_params->ctz)) { |
| value.from_unixtime(epoch_time.seconds, *_convert_params->ctz); |
| } |
| value.set_microsecond(epoch_time.microseconds); |
| } |
| return Status::OK(); |
| } |
| }; |
| |
| struct Int64ToTimestampTz : public PhysicalToLogicalConverter { |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = src_col->size(); |
| size_t start_idx = dst_col->size(); |
| dst_col->resize(start_idx + rows); |
| |
| const auto& src_data = assert_cast<const ColumnInt64*>(src_col.get())->get_data(); |
| auto& dest_data = assert_cast<ColumnTimeStampTz*>(dst_col)->get_data(); |
| static const cctz::time_zone UTC = cctz::utc_time_zone(); |
| |
| for (int i = 0; i < rows; i++) { |
| int64_t x = src_data[i]; |
| auto& tz = dest_data[start_idx + i]; |
| const auto epoch_time = detail::split_epoch_time(x, _convert_params->second_mask, |
| _convert_params->scale_to_nano_factor); |
| tz.from_unixtime(epoch_time.seconds, UTC); |
| tz.set_microsecond(epoch_time.microseconds); |
| } |
| return Status::OK(); |
| } |
| }; |
| |
| struct Int96toTimestamp : public PhysicalToLogicalConverter { |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = src_col->size() / sizeof(ParquetInt96); |
| auto& src_data = static_cast<const ColumnInt8*>(src_col.get())->get_data(); |
| auto ParquetInt96_data = (ParquetInt96*)src_data.data(); |
| size_t start_idx = dst_col->size(); |
| dst_col->resize(start_idx + rows); |
| auto& data = static_cast<ColumnDateTimeV2*>(dst_col)->get_data(); |
| |
| for (int i = 0; i < rows; i++) { |
| ParquetInt96 src_cell_data = ParquetInt96_data[i]; |
| auto& dst_value = |
| reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(data[start_idx + i]); |
| |
| const auto epoch_time = |
| detail::split_epoch_time(src_cell_data.to_timestamp_micros(), 1000000, 1000); |
| if (_convert_params->is_fixed_offset) { |
| if (!detail::try_convert_timestamp_with_fixed_offset( |
| dst_value, epoch_time.seconds, _convert_params->fixed_offset_seconds)) { |
| dst_value.from_unixtime(epoch_time.seconds, *_convert_params->ctz); |
| } |
| } else if (!detail::try_convert_timestamp_with_lookup(dst_value, epoch_time.seconds, |
| *_convert_params->ctz)) { |
| dst_value.from_unixtime(epoch_time.seconds, *_convert_params->ctz); |
| } |
| dst_value.set_microsecond(epoch_time.microseconds); |
| } |
| return Status::OK(); |
| } |
| }; |
| |
| struct Int96toTimestampTz : public PhysicalToLogicalConverter { |
| Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override { |
| ColumnPtr src_col = remove_nullable(src_physical_col); |
| IColumn* dst_col = get_mutable_inner_column(src_logical_column); |
| |
| size_t rows = src_col->size() / sizeof(ParquetInt96); |
| const auto& src_data = assert_cast<const ColumnInt8*>(src_col.get())->get_data(); |
| auto* ParquetInt96_data = (ParquetInt96*)src_data.data(); |
| size_t start_idx = dst_col->size(); |
| dst_col->resize(start_idx + rows); |
| auto& data = assert_cast<ColumnTimeStampTz*>(dst_col)->get_data(); |
| static const cctz::time_zone UTC = cctz::utc_time_zone(); |
| |
| for (int i = 0; i < rows; i++) { |
| ParquetInt96 src_cell_data = ParquetInt96_data[i]; |
| auto& tz = data[start_idx + i]; |
| const auto epoch_time = |
| detail::split_epoch_time(src_cell_data.to_timestamp_micros(), 1000000, 1000); |
| tz.from_unixtime(epoch_time.seconds, UTC); |
| tz.set_microsecond(epoch_time.microseconds); |
| } |
| return Status::OK(); |
| } |
| }; |
| |
| } // namespace doris::parquet |
