be/src/format/table/parquet_utils.cpp

Source
// 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.

#include "format/table/parquet_utils.h"

#include <fmt/format.h>

#include <algorithm>
#include <cctype>
#include <cstring>
#include <unordered_map>
#include <utility>

#include "core/assert_cast.h"
#include "core/column/column.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_nullable.h"
#include "format/parquet/parquet_column_convert.h"
#include "util/string_util.h"
#include "util/unaligned.h"

namespace doris::parquet_utils {
namespace {

template <typename ColumnType, typename T>
void insert_numeric_impl(MutableColumnPtr& column, T value) {
    if (auto* nullable_column = check_and_get_column<ColumnNullable>(*column)) {
        auto& nested = nullable_column->get_nested_column();
        assert_cast<ColumnType&>(nested).insert_value(value);
        nullable_column->push_false_to_nullmap(1);
    } else {
        assert_cast<ColumnType&>(*column).insert_value(value);
    }
}

} // namespace

std::string join_path(const std::vector<std::string>& items) {
    return join(items, ".");
}

void insert_int32(MutableColumnPtr& column, Int32 value) {
    insert_numeric_impl<ColumnInt32>(column, value);
}

void insert_int64(MutableColumnPtr& column, Int64 value) {
    insert_numeric_impl<ColumnInt64>(column, value);
}

void insert_bool(MutableColumnPtr& column, bool value) {
    insert_numeric_impl<ColumnUInt8>(column, static_cast<UInt8>(value));
}

void insert_string(MutableColumnPtr& column, const std::string& value) {
    if (auto* nullable = check_and_get_column<ColumnNullable>(*column)) {
        nullable->get_null_map_data().push_back(0);
        auto& nested = nullable->get_nested_column();
        assert_cast<ColumnString&>(nested).insert_data(value.c_str(), value.size());
    } else {
        assert_cast<ColumnString&>(*column).insert_data(value.c_str(), value.size());
    }
}

void insert_null(MutableColumnPtr& column) {
    if (auto* nullable = check_and_get_column<ColumnNullable>(*column)) {
        nullable->get_null_map_data().push_back(1);
        nullable->get_nested_column().insert_default();
    } else {
        column->insert_default();
    }
}

std::string physical_type_to_string(tparquet::Type::type type) {
    switch (type) {
    case tparquet::Type::BOOLEAN:
        return "BOOLEAN";
    case tparquet::Type::INT32:
        return "INT32";
    case tparquet::Type::INT64:
        return "INT64";
    case tparquet::Type::INT96:
        return "INT96";
    case tparquet::Type::FLOAT:
        return "FLOAT";
    case tparquet::Type::DOUBLE:
        return "DOUBLE";
    case tparquet::Type::BYTE_ARRAY:
        return "BYTE_ARRAY";
    case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
        return "FIXED_LEN_BYTE_ARRAY";
    default:
        return "UNKNOWN";
    }
}

std::string compression_to_string(tparquet::CompressionCodec::type codec) {
    switch (codec) {
    case tparquet::CompressionCodec::UNCOMPRESSED:
        return "UNCOMPRESSED";
    case tparquet::CompressionCodec::SNAPPY:
        return "SNAPPY";
    case tparquet::CompressionCodec::GZIP:
        return "GZIP";
    case tparquet::CompressionCodec::LZO:
        return "LZO";
    case tparquet::CompressionCodec::BROTLI:
        return "BROTLI";
    case tparquet::CompressionCodec::LZ4:
        return "LZ4";
    case tparquet::CompressionCodec::ZSTD:
        return "ZSTD";
    case tparquet::CompressionCodec::LZ4_RAW:
        return "LZ4_RAW";
    default:
        return "UNKNOWN";
    }
}

std::string converted_type_to_string(tparquet::ConvertedType::type type) {
    switch (type) {
    case tparquet::ConvertedType::UTF8:
        return "UTF8";
    case tparquet::ConvertedType::MAP:
        return "MAP";
    case tparquet::ConvertedType::MAP_KEY_VALUE:
        return "MAP_KEY_VALUE";
    case tparquet::ConvertedType::LIST:
        return "LIST";
    case tparquet::ConvertedType::ENUM:
        return "ENUM";
    case tparquet::ConvertedType::DECIMAL:
        return "DECIMAL";
    case tparquet::ConvertedType::DATE:
        return "DATE";
    case tparquet::ConvertedType::TIME_MILLIS:
        return "TIME_MILLIS";
    case tparquet::ConvertedType::TIME_MICROS:
        return "TIME_MICROS";
    case tparquet::ConvertedType::TIMESTAMP_MILLIS:
        return "TIMESTAMP_MILLIS";
    case tparquet::ConvertedType::TIMESTAMP_MICROS:
        return "TIMESTAMP_MICROS";
    case tparquet::ConvertedType::UINT_8:
        return "UINT_8";
    case tparquet::ConvertedType::UINT_16:
        return "UINT_16";
    case tparquet::ConvertedType::UINT_32:
        return "UINT_32";
    case tparquet::ConvertedType::UINT_64:
        return "UINT_64";
    case tparquet::ConvertedType::INT_8:
        return "INT_8";
    case tparquet::ConvertedType::INT_16:
        return "INT_16";
    case tparquet::ConvertedType::INT_32:
        return "INT_32";
    case tparquet::ConvertedType::INT_64:
        return "INT_64";
    case tparquet::ConvertedType::JSON:
        return "JSON";
    case tparquet::ConvertedType::BSON:
        return "BSON";
    case tparquet::ConvertedType::INTERVAL:
        return "INTERVAL";
    default:
        return "UNKNOWN";
    }
}

std::string logical_type_to_string(const tparquet::SchemaElement& element) {
    if (element.__isset.logicalType) {
        const auto& logical = element.logicalType;
        if (logical.__isset.STRING) {
            return "STRING";
        } else if (logical.__isset.MAP) {
            return "MAP";
        } else if (logical.__isset.LIST) {
            return "LIST";
        } else if (logical.__isset.ENUM) {
            return "ENUM";
        } else if (logical.__isset.DECIMAL) {
            return "DECIMAL";
        } else if (logical.__isset.DATE) {
            return "DATE";
        } else if (logical.__isset.TIME) {
            return "TIME";
        } else if (logical.__isset.TIMESTAMP) {
            return "TIMESTAMP";
        } else if (logical.__isset.INTEGER) {
            return "INTEGER";
        } else if (logical.__isset.UNKNOWN) {
            return "UNKNOWN";
        } else if (logical.__isset.JSON) {
            return "JSON";
        } else if (logical.__isset.BSON) {
            return "BSON";
        } else if (logical.__isset.UUID) {
            return "UUID";
        } else if (logical.__isset.FLOAT16) {
            return "FLOAT16";
        } else if (logical.__isset.VARIANT) {
            return "VARIANT";
        } else if (logical.__isset.GEOMETRY) {
            return "GEOMETRY";
        } else if (logical.__isset.GEOGRAPHY) {
            return "GEOGRAPHY";
        }
    }
    if (element.__isset.converted_type) {
        return converted_type_to_string(element.converted_type);
    }
    return "";
}

std::string encodings_to_string(const std::vector<tparquet::Encoding::type>& encodings) {
    std::vector<std::string> parts;
    parts.reserve(encodings.size());
    for (auto encoding : encodings) {
        switch (encoding) {
        case tparquet::Encoding::PLAIN:
            parts.emplace_back("PLAIN");
            break;
        case tparquet::Encoding::PLAIN_DICTIONARY:
            parts.emplace_back("PLAIN_DICTIONARY");
            break;
        case tparquet::Encoding::RLE:
            parts.emplace_back("RLE");
            break;
        case tparquet::Encoding::BIT_PACKED:
            parts.emplace_back("BIT_PACKED");
            break;
        case tparquet::Encoding::DELTA_BINARY_PACKED:
            parts.emplace_back("DELTA_BINARY_PACKED");
            break;
        case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
            parts.emplace_back("DELTA_LENGTH_BYTE_ARRAY");
            break;
        case tparquet::Encoding::DELTA_BYTE_ARRAY:
            parts.emplace_back("DELTA_BYTE_ARRAY");
            break;
        case tparquet::Encoding::RLE_DICTIONARY:
            parts.emplace_back("RLE_DICTIONARY");
            break;
        default:
            parts.emplace_back("UNKNOWN");
            break;
        }
    }
    return fmt::format("{}", fmt::join(parts, ","));
}

bool try_get_statistics_encoded_value(const tparquet::Statistics& statistics, bool is_min,
                                      std::string* encoded_value) {
    if (is_min) {
        if (statistics.__isset.min_value) {
            *encoded_value = statistics.min_value;
            return true;
        }
        if (statistics.__isset.min) {
            *encoded_value = statistics.min;
            return true;
        }
    } else {
        if (statistics.__isset.max_value) {
            *encoded_value = statistics.max_value;
            return true;
        }
        if (statistics.__isset.max) {
            *encoded_value = statistics.max;
            return true;
        }
    }
    encoded_value->clear();
    return false;
}

std::string bytes_to_hex_string(const std::string& bytes) {
    static constexpr char kHexDigits[] = "0123456789ABCDEF";
    std::string hex;
    hex.resize(bytes.size() * 2);
    for (size_t i = 0; i < bytes.size(); ++i) {
        auto byte = static_cast<uint8_t>(bytes[i]);
        hex[i * 2] = kHexDigits[byte >> 4];
        hex[i * 2 + 1] = kHexDigits[byte & 0x0F];
    }
    return fmt::format("0x{}", hex);
}

std::string decode_statistics_value(const FieldSchema* schema_field,
                                    tparquet::Type::type physical_type,
                                    const std::string& encoded_value, const cctz::time_zone& ctz) {
    if (encoded_value.empty()) {
        return "";
    }
    if (schema_field == nullptr) {
        return bytes_to_hex_string(encoded_value);
    }

    auto logical_data_type = remove_nullable(schema_field->data_type);
    auto converter = parquet::PhysicalToLogicalConverter::get_converter(
            schema_field, logical_data_type, logical_data_type, &ctz);
    if (!converter || !converter->support()) {
        return bytes_to_hex_string(encoded_value);
    }

    ColumnPtr physical_column;
    switch (physical_type) {
    case tparquet::Type::type::BOOLEAN: {
        if (encoded_value.size() != sizeof(UInt8)) {
            return bytes_to_hex_string(encoded_value);
        }
        auto physical_col = ColumnUInt8::create();
        physical_col->insert_value(doris::unaligned_load<UInt8>(encoded_value.data()));
        physical_column = std::move(physical_col);
        break;
    }
    case tparquet::Type::type::INT32: {
        if (encoded_value.size() != sizeof(Int32)) {
            return bytes_to_hex_string(encoded_value);
        }
        auto physical_col = ColumnInt32::create();
        physical_col->insert_value(doris::unaligned_load<Int32>(encoded_value.data()));
        physical_column = std::move(physical_col);
        break;
    }
    case tparquet::Type::type::INT64: {
        if (encoded_value.size() != sizeof(Int64)) {
            return bytes_to_hex_string(encoded_value);
        }
        auto physical_col = ColumnInt64::create();
        physical_col->insert_value(doris::unaligned_load<Int64>(encoded_value.data()));
        physical_column = std::move(physical_col);
        break;
    }
    case tparquet::Type::type::FLOAT: {
        if (encoded_value.size() != sizeof(Float32)) {
            return bytes_to_hex_string(encoded_value);
        }
        auto physical_col = ColumnFloat32::create();
        physical_col->insert_value(doris::unaligned_load<Float32>(encoded_value.data()));
        physical_column = std::move(physical_col);
        break;
    }
    case tparquet::Type::type::DOUBLE: {
        if (encoded_value.size() != sizeof(Float64)) {
            return bytes_to_hex_string(encoded_value);
        }
        auto physical_col = ColumnFloat64::create();
        physical_col->insert_value(doris::unaligned_load<Float64>(encoded_value.data()));
        physical_column = std::move(physical_col);
        break;
    }
    case tparquet::Type::type::BYTE_ARRAY: {
        auto physical_col = ColumnString::create();
        physical_col->insert_data(encoded_value.data(), encoded_value.size());
        physical_column = std::move(physical_col);
        break;
    }
    case tparquet::Type::type::FIXED_LEN_BYTE_ARRAY: {
        int32_t type_length = schema_field->parquet_schema.__isset.type_length
                                      ? schema_field->parquet_schema.type_length
                                      : 0;
        if (type_length <= 0) {
            type_length = static_cast<int32_t>(encoded_value.size());
        }
        if (static_cast<size_t>(type_length) != encoded_value.size()) {
            return bytes_to_hex_string(encoded_value);
        }
        auto physical_col = ColumnUInt8::create();
        physical_col->resize(type_length);
        memcpy(physical_col->get_data().data(), encoded_value.data(), encoded_value.size());
        physical_column = std::move(physical_col);
        break;
    }
    case tparquet::Type::type::INT96: {
        constexpr size_t kInt96Size = 12;
        if (encoded_value.size() != kInt96Size) {
            return bytes_to_hex_string(encoded_value);
        }
        auto physical_col = ColumnInt8::create();
        physical_col->resize(kInt96Size);
        memcpy(physical_col->get_data().data(), encoded_value.data(), encoded_value.size());
        physical_column = std::move(physical_col);
        break;
    }
    default:
        return bytes_to_hex_string(encoded_value);
    }

    ColumnPtr logical_column;
    if (converter->is_consistent()) {
        logical_column = physical_column;
    } else {
        logical_column = logical_data_type->create_column();
        if (Status st = converter->physical_convert(physical_column, logical_column); !st.ok()) {
            return bytes_to_hex_string(encoded_value);
        }
    }

    if (logical_column->size() != 1) {
        return bytes_to_hex_string(encoded_value);
    }
    DataTypeSerDe::FormatOptions options;
    options.timezone = &ctz;
    return logical_data_type->to_string(*logical_column, 0, options);
}

void build_path_map(const FieldSchema& field, const std::string& prefix,
                    std::unordered_map<std::string, const FieldSchema*>* map) {
    std::string current = prefix.empty() ? field.name : fmt::format("{}.{}", prefix, field.name);
    if (field.children.empty()) {
        (*map)[current] = &field;
    } else {
        for (const auto& child : field.children) {
            build_path_map(child, current, map);
        }
    }
}

#define MERGE_STATS_CASE(ParquetType)                                                     \
    case ParquetType: {                                                                   \
        auto typed_left_stat = std::static_pointer_cast<                                  \
                ::parquet::TypedStatistics<::parquet::PhysicalType<ParquetType>>>(left);  \
        auto typed_right_stat = std::static_pointer_cast<                                 \
                ::parquet::TypedStatistics<::parquet::PhysicalType<ParquetType>>>(right); \
        typed_left_stat->Merge(*typed_right_stat);                                        \
        return;                                                                           \
    }

void merge_stats(const std::shared_ptr<::parquet::Statistics>& left,
                 const std::shared_ptr<::parquet::Statistics>& right) {
    if (left == nullptr || right == nullptr) {
        return;
    }
    DCHECK(left->physical_type() == right->physical_type());

    switch (left->physical_type()) {
        MERGE_STATS_CASE(::parquet::Type::BOOLEAN);
        MERGE_STATS_CASE(::parquet::Type::INT32);
        MERGE_STATS_CASE(::parquet::Type::INT64);
        MERGE_STATS_CASE(::parquet::Type::INT96);
        MERGE_STATS_CASE(::parquet::Type::FLOAT);
        MERGE_STATS_CASE(::parquet::Type::DOUBLE);
        MERGE_STATS_CASE(::parquet::Type::BYTE_ARRAY);
        MERGE_STATS_CASE(::parquet::Type::FIXED_LEN_BYTE_ARRAY);
    default:
        LOG(WARNING) << "Unsupported parquet type for statistics merge: "
                     << static_cast<int>(left->physical_type());
        break;
    }
}

} // namespace doris::parquet_utils

Coverage Report

Created: 2026-03-13 03:47