github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3612136693
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -48,22 +39,134 @@
#include "exprs/expr_zonemap_filter.h"
#include "exprs/vexpr_context.h"
#include "format_v2/parquet/parquet_column_schema.h"
+#include "format_v2/parquet/parquet_file_context.h"
+#include "format_v2/parquet/reader/native/block_split_bloom_filter.h"
+#include "format_v2/parquet/reader/native_column_reader.h"
#include "format_v2/timestamp_statistics.h"
#include "runtime/runtime_profile.h"
+#include "storage/index/bloom_filter/bloom_filter.h"
#include "storage/index/zone_map/zone_map_index.h"
#include "storage/index/zone_map/zonemap_eval_context.h"
+#include "util/thrift_util.h"
+#include "util/unaligned.h"
namespace doris::format::parquet {
+namespace detail {
+
+Status validate_native_bloom_filter_layout(int64_t offset, uint32_t
header_size,
+ int64_t payload_size, int64_t
declared_length,
+ size_t file_size) {
+ if (offset < 0 || header_size == 0 || payload_size <
segment_v2::BloomFilter::MINIMUM_BYTES ||
+ payload_size > segment_v2::BloomFilter::MAXIMUM_BYTES || payload_size
% 32 != 0) {
+ return Status::Corruption(
+ "Invalid Parquet Bloom filter layout: offset {}, header {},
payload {}", offset,
+ header_size, payload_size);
+ }
+ const uint64_t unsigned_offset = static_cast<uint64_t>(offset);
+ const uint64_t total_size = static_cast<uint64_t>(header_size) +
payload_size;
+ if (unsigned_offset > file_size || total_size > file_size -
unsigned_offset) {
+ return Status::Corruption("Parquet Bloom filter range exceeds file
size {}", file_size);
+ }
+ if (declared_length >= 0) {
+ const uint64_t unsigned_declared_length =
static_cast<uint64_t>(declared_length);
+ if (unsigned_declared_length < total_size ||
+ unsigned_declared_length > file_size - unsigned_offset) {
+ return Status::Corruption(
+ "Parquet Bloom filter requires {} bytes, metadata declares
{}, file has {}",
+ total_size, declared_length, file_size - unsigned_offset);
+ }
+ }
+ return Status::OK();
+}
+
+bool can_use_native_footer_min_max(const ParquetTypeDescriptor&
type_descriptor,
Review Comment:
[P1] Gate type-defined bounds on column_orders
This accepts `min_value`/`max_value` using only the leaf type, and the
native ColumnIndex path does the same for page bounds. The Parquet Thrift
contract says those fields are undefined when the corresponding
`FileMetaData.column_orders` entry is absent, and readers must ignore an
unsupported `ColumnOrder` union. A legacy or foreign file with missing, short,
or unknown order metadata can therefore supply bounds in an order Doris assumes
incorrectly and have matching rows pruned. Thread the per-leaf order through
both row-group and page-index planning, and use these type-defined bounds only
for a present supported `TYPE_ORDER` entry; keep legacy `min`/`max`
compatibility separate.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -1550,4 +1632,238 @@ Status select_row_group_ranges_by_page_index(
return Status::OK();
}
+namespace {
+
+template <typename ValueType>
+bool set_native_page_scalar_min_max(const tparquet::ColumnIndex& column_index,
+ const ParquetColumnSchema& column_schema,
size_t page_idx,
+ DecodedValueKind kind,
ParquetColumnStatistics* page_statistics,
+ const cctz::time_zone* timezone) {
+ if (page_idx >= column_index.min_values.size() || page_idx >=
column_index.max_values.size() ||
+ column_index.min_values[page_idx].size() != sizeof(ValueType) ||
+ column_index.max_values[page_idx].size() != sizeof(ValueType)) {
+ return false;
+ }
+ const auto min_value =
unaligned_load<ValueType>(column_index.min_values[page_idx].data());
+ const auto max_value =
unaligned_load<ValueType>(column_index.max_values[page_idx].data());
+ if constexpr (std::is_same_v<ValueType, int64_t>) {
+ if (!timestamp_min_max_is_safe(column_schema, min_value, max_value,
timezone)) {
+ return false;
+ }
+ }
+ if (!valid_min_max(min_value, max_value)) {
+ return true;
+ }
+ if (!set_decoded_field(column_schema, kind, min_value,
&page_statistics->min_value, timezone) ||
+ !set_decoded_field(column_schema, kind, max_value,
&page_statistics->max_value, timezone)) {
+ return false;
+ }
+ if (decoded_min_max_is_ordered(*page_statistics)) {
+ page_statistics->has_min_max = true;
+ }
+ return true;
+}
+
+bool build_native_page_statistics(const tparquet::ColumnIndex& column_index,
+ const ParquetColumnSchema& column_schema,
size_t page_idx,
+ ParquetColumnStatistics* page_statistics,
+ const cctz::time_zone* timezone) {
+ DORIS_CHECK(page_statistics != nullptr);
+ *page_statistics = {};
+ if (!column_index.__isset.null_counts || page_idx >=
column_index.null_pages.size() ||
+ page_idx >= column_index.null_counts.size()) {
+ return false;
+ }
+ page_statistics->has_null_count = true;
+ page_statistics->has_null = column_index.null_counts[page_idx] > 0;
Review Comment:
The fix still leaves one contradictory all-null case: for these flat leaves
the paired OffsetIndex gives the exact page span, but any positive
`null_counts[i]` is accepted even when `null_pages[i] = true` and the count
differs from `native_page_row_range(...).length`. That still sets `has_not_null
= false`, so a corrupt optional index can prune real non-null rows for `IS NOT
NULL`. Please make the all-null count equal the page span (otherwise disable
this ColumnIndex) and add the positive-mismatch case to the existing tests.
##########
be/src/format_v2/parquet/reader/native_column_reader.cpp:
##########
@@ -0,0 +1,663 @@
+// 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 <algorithm>
+#include <cstddef>
+#include <limits>
+#include <ranges>
+#include <string>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/config.h"
+#include "core/assert_cast.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_array.h"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
+#include "format_v2/column_data.h"
+#include "format_v2/parquet/parquet_column_schema.h"
+#include "format_v2/parquet/parquet_file_context.h"
+#include "runtime/runtime_state.h"
+
+namespace doris::format::parquet {
+namespace {
+
+constexpr size_t MAX_RETAINED_BATCH_SCRATCH_BYTES = 4UL << 20;
+
+DataTypePtr projected_type(const ParquetColumnSchema& schema,
+ const format::LocalColumnIndex* projection) {
+ if (!format::is_partial_projection(projection)) {
+ return schema.type;
+ }
+ switch (schema.kind) {
+ case ParquetColumnSchemaKind::PRIMITIVE:
+ return schema.type;
+ case ParquetColumnSchemaKind::STRUCT: {
+ DataTypes child_types;
+ Strings child_names;
+ child_types.reserve(projection->children.size());
+ child_names.reserve(projection->children.size());
+ for (const auto& child_projection : projection->children) {
+ const auto child_it = std::ranges::find_if(schema.children,
[&](const auto& child) {
+ return child->local_id == child_projection.local_id();
+ });
+ DORIS_CHECK(child_it != schema.children.end());
+ child_types.push_back(make_nullable(projected_type(**child_it,
&child_projection)));
+ child_names.push_back((*child_it)->name);
+ }
+ DataTypePtr type = std::make_shared<DataTypeStruct>(child_types,
child_names);
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ case ParquetColumnSchemaKind::LIST: {
+ DORIS_CHECK(schema.children.size() == 1);
+ const auto* child_projection =
+ format::find_child_projection(projection,
schema.children[0]->local_id);
+ DORIS_CHECK(child_projection != nullptr);
+ DataTypePtr type = std::make_shared<DataTypeArray>(
+ projected_type(*schema.children[0], child_projection));
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ case ParquetColumnSchemaKind::MAP: {
+ DORIS_CHECK(schema.children.size() == 2);
+ const auto* value_projection =
+ format::find_child_projection(projection,
schema.children[1]->local_id);
+ DORIS_CHECK(value_projection != nullptr);
+ DataTypePtr type = std::make_shared<DataTypeMap>(
+ make_nullable(schema.children[0]->type),
+ make_nullable(projected_type(*schema.children[1],
value_projection)));
+ return schema.type->is_nullable() ? make_nullable(type) : type;
+ }
+ }
+ DORIS_CHECK(false);
+ return nullptr;
+}
+
+const FieldSchema* find_child_field(const FieldSchema& parent, const
ParquetColumnSchema& child) {
+ auto field_it = std::ranges::find_if(parent.children, [&](const
FieldSchema& field) {
+ return (child.parquet_field_id >= 0 && field.field_id ==
child.parquet_field_id) ||
+ field.name == child.name;
+ });
+ return field_it == parent.children.end() ? nullptr : &*field_it;
+}
+
+void collect_projected_ids(const ParquetColumnSchema& schema,
+ const format::LocalColumnIndex* projection,
+ const FieldSchema& native_field,
std::set<uint64_t>* ids) {
+ DORIS_CHECK(ids != nullptr);
+ if (!format::is_partial_projection(projection)) {
Review Comment:
The subtree fix here misses the adjacent implicit MAP-key branch. For a
value-only partial MAP projection, line 142 still retains only the mandatory
key group's ID; a complex key's descendants therefore become
`SkipReadingReader`s/defaults, and `collect_projected_leaf_column_ids()` also
omits those leaves from page-index/MergeRange planning. Please retain the full
mandatory key subtree in both reader-ID and physical-leaf collection, with a
complex-key/value-child-only projection test.
##########
be/src/format_v2/parquet/reader/native/column_reader.cpp:
##########
@@ -0,0 +1,1637 @@
+// 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 <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;
+}
+
+IColumn::Filter* conversion_failure_map(const NativeFieldSchema& field,
+ const DataTypePtr& target_type, bool
strict_mode,
+ IColumn::Filter* output_null_map,
+ IColumn::Filter*
compatibility_scratch) {
+ const auto& schema = field.parquet_schema;
+ const bool is_utc_timestamp =
+ field.physical_type == tparquet::Type::INT96 ||
+ (field.physical_type == tparquet::Type::INT64 &&
+ ((schema.__isset.logicalType &&
schema.logicalType.__isset.TIMESTAMP &&
+ schema.logicalType.TIMESTAMP.isAdjustedToUTC) ||
+ (schema.__isset.converted_type &&
+ (schema.converted_type ==
tparquet::ConvertedType::TIMESTAMP_MILLIS ||
+ schema.converted_type ==
tparquet::ConvertedType::TIMESTAMP_MICROS))));
+ if (!strict_mode && output_null_map != nullptr && is_utc_timestamp &&
+ remove_nullable(target_type)->get_primitive_type() == TYPE_DATETIMEV2)
{
+ // Legacy UTC timestamp conversion kept out-of-range values as
DATETIME defaults. Local
+ // timestamps intentionally keep non-strict NULL-on-overflow behavior
because timezone
+ // conversion is not part of their representation.
+ compatibility_scratch->resize_fill(output_null_map->size(), 0);
+ return compatibility_scratch;
+ }
+ return output_null_map;
+}
+
+void mark_local_timestamp_defaults(const NativeFieldSchema& field, const
DataTypePtr& target_type,
+ bool strict_mode, IColumn& data_column,
+ IColumn::Filter* output_null_map, size_t
start_row) {
+ const auto& schema = field.parquet_schema;
+ const bool is_local_timestamp =
+ field.physical_type == tparquet::Type::INT64 &&
schema.__isset.logicalType &&
+ schema.logicalType.__isset.TIMESTAMP &&
!schema.logicalType.TIMESTAMP.isAdjustedToUTC;
+ if (strict_mode || output_null_map == nullptr || !is_local_timestamp ||
+ remove_nullable(target_type)->get_primitive_type() != TYPE_DATETIMEV2)
{
+ return;
+ }
+ auto& values = assert_cast<ColumnDateTimeV2&>(data_column).get_data();
+ DORIS_CHECK_EQ(values.size(), output_null_map->size());
+ for (size_t row = start_row; row < values.size(); ++row) {
+ if ((*output_null_map)[row] == 0 && !values[row].is_valid_date()) {
+ // Local timestamps before Doris' representable calendar can
materialize as a zero
+ // date without a SerDe error. Preserve non-strict scan semantics
by nulling only that
+ // sentinel; physical NULLs and valid local timestamps keep their
original map bits.
+ (*output_null_map)[row] = 1;
+ }
+ }
+}
+
+// The target SerDe can fuse physical decode with these logical type changes.
Less common schema
+// changes retain the generic file-format converter as a compatibility path:
the decoder still
+// exposes raw spans, but the source SerDe first materializes a reusable
source column before the
+// generic logical cast. Ordinary scans and numeric widening never allocate
that source column.
+bool serde_can_materialize_directly(const DataTypePtr& source_type,
+ const DataTypePtr& target_type) {
+ const auto source = remove_nullable(source_type)->get_primitive_type();
+ const auto target = remove_nullable(target_type)->get_primitive_type();
+ return source == target || (is_direct_integer_type(source) &&
is_direct_integer_type(target)) ||
+ (source == TYPE_FLOAT && target == TYPE_DOUBLE) ||
+ (is_direct_decimal_type(source) && is_direct_decimal_type(target))
||
+ // Parquet STRING and VARBINARY share BYTE_ARRAY bytes.
Materializing through the target
+ // SerDe preserves those bytes and avoids a converter whose scratch
column uses the
+ // String representation instead of the native ColumnVarbinary
representation.
+ (is_direct_binary_type(source) && is_direct_binary_type(target));
+}
+
+Status init_decode_context(const NativeFieldSchema& field, const
cctz::time_zone* ctz,
+ ParquetDecodeContext* context) {
+ DORIS_CHECK(context != nullptr);
+ switch (field.physical_type) {
+ case tparquet::Type::BOOLEAN:
+ context->physical_type = ParquetPhysicalType::BOOLEAN;
+ break;
+ case tparquet::Type::INT32:
+ context->physical_type = ParquetPhysicalType::INT32;
+ break;
+ case tparquet::Type::INT64:
+ context->physical_type = ParquetPhysicalType::INT64;
+ break;
+ case tparquet::Type::INT96:
+ context->physical_type = ParquetPhysicalType::INT96;
+ break;
+ case tparquet::Type::FLOAT:
+ context->physical_type = ParquetPhysicalType::FLOAT;
+ break;
+ case tparquet::Type::DOUBLE:
+ context->physical_type = ParquetPhysicalType::DOUBLE;
+ break;
+ case tparquet::Type::BYTE_ARRAY:
+ context->physical_type = ParquetPhysicalType::BYTE_ARRAY;
+ break;
+ case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
+ context->physical_type = ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY;
+ break;
+ default:
+ return Status::NotSupported("Unsupported Parquet physical type {}",
+ tparquet::to_string(field.physical_type));
+ }
+
+ const auto& schema = field.parquet_schema;
+ context->type_length = schema.__isset.type_length ? schema.type_length :
-1;
+ context->decimal_precision = schema.__isset.precision ? schema.precision :
-1;
+ context->decimal_scale = schema.__isset.scale ? schema.scale : -1;
+ context->timezone = ctz;
+ if (schema.__isset.logicalType) {
Review Comment:
The added validation is still too late for metadata planning:
`validate_physical_annotation()` runs only from `NativeColumnReader::create()`,
after `plan_parquet_row_groups()`. An empty TIMESTAMP `TimeUnit` union is
recorded here as `UNKNOWN` without an unsupported marker, and adjusted-to-UTC
footer/page bounds then reach `floor_timestamp_seconds(UNKNOWN)` and its fatal
check before reader creation. Please perform the requested TIME/TIMESTAMP unit
validation while building the schema/descriptor (and make statistics fail
closed), with footer and page-statistics 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]
