github-actions[bot] commented on code in PR #65674: URL: https://github.com/apache/doris/pull/65674#discussion_r3609979129
########## be/src/format_v2/parquet/reader/native/column_reader.cpp: ########## @@ -0,0 +1,1840 @@ +// 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_v2/parquet/reader/native/column_reader.h" + +#include <gen_cpp/parquet_types.h> +#include <limits.h> +#include <sys/types.h> + +#include <algorithm> +#include <string_view> +#include <utility> + +#include "common/cast_set.h" +#include "common/status.h" +#include "core/column/column.h" +#include "core/column/column_array.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.h" +#include "core/column/column_vector.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_map.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_struct.h" +#include "core/data_type/define_primitive_type.h" +#include "format_v2/parquet/native_schema_desc.h" +#include "format_v2/parquet/reader/native/column_chunk_reader.h" +#include "format_v2/parquet/reader/native/level_decoder.h" +#include "io/fs/tracing_file_reader.h" +#include "runtime/runtime_profile.h" + +namespace doris::format::parquet::native { +namespace { + +ParquetTimeUnit parquet_time_unit(const tparquet::TimeUnit& unit) { + if (unit.__isset.MILLIS) { + return ParquetTimeUnit::MILLIS; + } + if (unit.__isset.MICROS) { + return ParquetTimeUnit::MICROS; + } + if (unit.__isset.NANOS) { + return ParquetTimeUnit::NANOS; + } + return ParquetTimeUnit::UNKNOWN; +} + +bool is_direct_integer_type(PrimitiveType type) { + switch (type) { + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_LARGEINT: + return true; + default: + return false; + } +} + +bool is_direct_decimal_type(PrimitiveType type) { + switch (type) { + case TYPE_DECIMALV2: + case TYPE_DECIMAL32: + case TYPE_DECIMAL64: + case TYPE_DECIMAL128I: + case TYPE_DECIMAL256: + return true; + default: + return false; + } +} + +template <typename T> +bool release_vector_if_oversized(std::vector<T>* values, size_t max_retained_bytes) { + DORIS_CHECK(values != nullptr); + if (values->capacity() * sizeof(T) <= max_retained_bytes) { + return false; + } + std::vector<T>().swap(*values); + return true; +} + +size_t retained_set_bytes(const std::unordered_set<size_t>& values) { + return values.bucket_count() * sizeof(void*) + values.size() * sizeof(size_t); +} + +bool is_direct_binary_type(PrimitiveType type) { + return is_string_type(type) || type == TYPE_VARBINARY; +} + +Status validate_decimal_physical_type(const NativeFieldSchema& field, int precision, int scale) { + if (precision <= 0 || scale < 0 || scale > precision) { + return Status::Corruption("Parquet decimal field {} has invalid precision {} and scale {}", + field.name, precision, scale); + } + switch (field.physical_type) { + case tparquet::Type::INT32: + if (precision <= 9) { + return Status::OK(); + } + break; + case tparquet::Type::INT64: + if (precision <= 18) { + return Status::OK(); + } + break; + case tparquet::Type::BYTE_ARRAY: + return Status::OK(); + case tparquet::Type::FIXED_LEN_BYTE_ARRAY: { + const int length = + field.parquet_schema.__isset.type_length ? field.parquet_schema.type_length : -1; + if (length > 0) { + const int64_t max_precision = (static_cast<int64_t>(length) * 8 - 1) * 30103 / 100000; + if (precision <= max_precision) { + return Status::OK(); + } + } + break; + } + default: + break; + } + return Status::Corruption("Parquet decimal field {} has incompatible physical type {}", + field.name, tparquet::to_string(field.physical_type)); +} + +Status validate_physical_annotation(const NativeFieldSchema& field) { + const auto& schema = field.parquet_schema; + if (field.physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY && + (!schema.__isset.type_length || schema.type_length <= 0)) { + return Status::Corruption("Parquet fixed-length field {} has invalid width {}", field.name, + schema.__isset.type_length ? schema.type_length : -1); + } + auto require = [&](bool valid, std::string_view annotation) -> Status { + if (valid) { + return Status::OK(); + } + return Status::Corruption("Parquet {} field {} is incompatible with physical type {}", + annotation, field.name, tparquet::to_string(field.physical_type)); + }; + + if (schema.__isset.logicalType) { + const auto& logical = schema.logicalType; + if (logical.__isset.STRING || logical.__isset.ENUM || logical.__isset.JSON || + logical.__isset.BSON) { + return require(field.physical_type == tparquet::Type::BYTE_ARRAY, "string"); + } + if (logical.__isset.DECIMAL) { + return validate_decimal_physical_type(field, logical.DECIMAL.precision, + logical.DECIMAL.scale); + } + if (logical.__isset.DATE) { + return require(field.physical_type == tparquet::Type::INT32, "date"); + } + if (logical.__isset.TIME) { + const auto unit = parquet_time_unit(logical.TIME.unit); + return require( + (unit == ParquetTimeUnit::MILLIS && + field.physical_type == tparquet::Type::INT32) || + ((unit == ParquetTimeUnit::MICROS || unit == ParquetTimeUnit::NANOS) && + field.physical_type == tparquet::Type::INT64), + "time"); + } + if (logical.__isset.TIMESTAMP) { + return require( + field.physical_type == tparquet::Type::INT64 && + parquet_time_unit(logical.TIMESTAMP.unit) != ParquetTimeUnit::UNKNOWN, + "timestamp"); + } + if (logical.__isset.INTEGER) { + const int width = logical.INTEGER.bitWidth; + return require(((width == 8 || width == 16 || width == 32) && + field.physical_type == tparquet::Type::INT32) || + (width == 64 && field.physical_type == tparquet::Type::INT64), + "integer"); + } + if (logical.__isset.UUID) { + return require(field.physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY && + schema.type_length == 16, + "UUID"); + } + if (logical.__isset.FLOAT16) { + return require(field.physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY && + schema.type_length == 2, + "FLOAT16"); + } + if (logical.__isset.UNKNOWN) { + return Status::OK(); + } + return Status::Corruption("Unsupported Parquet logical annotation on field {}", field.name); Review Comment: [P1] Preserve valid geospatial BYTE_ARRAY leaves The checked-in Parquet schema permits both GEOMETRY and GEOGRAPHY on BYTE_ARRAY, and `NativeFieldDescriptor::get_doris_type()` intentionally falls back to the physical STRING/VARBINARY type when it cannot interpret a logical annotation. The native schema layer therefore accepts those columns, but this catch-all makes every scan that reads either column fail at reader initialization; the established resolver instead falls back to the physical string type. Handle these two valid BYTE_ARRAY pairings as raw bytes (leaving the decode context unannotated), while still rejecting invalid physical pairings, and add plain/dictionary coverage. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
