github-actions[bot] commented on code in PR #65674:
URL: https://github.com/apache/doris/pull/65674#discussion_r3627308190
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -1388,127 +1065,150 @@ void collect_request_leaf_schemas(
}
}
-bool build_page_skip_plan_for_leaf(
- const std::shared_ptr<::parquet::RowGroupPageIndexReader>& row_group,
- const ParquetColumnSchema& column_schema, const std::vector<RowRange>&
selected_ranges,
- int64_t row_group_rows, ParquetPageSkipPlan* page_skip_plan) {
- DORIS_CHECK(page_skip_plan != nullptr);
- *page_skip_plan = ParquetPageSkipPlan {};
- if (column_schema.kind != ParquetColumnSchemaKind::PRIMITIVE ||
- column_schema.descriptor == nullptr || column_schema.leaf_column_id <
0 ||
- column_schema.descriptor->max_repetition_level() != 0) {
+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;
}
-
- std::shared_ptr<::parquet::OffsetIndex> offset_index;
- try {
- offset_index = row_group->GetOffsetIndex(column_schema.leaf_column_id);
- } catch (const ::parquet::ParquetException&) {
+ 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;
- } catch (const std::exception&) {
+ }
+ 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,
+ int64_t page_rows, 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;
}
- if (offset_index == nullptr) {
+ const int64_t null_count = column_index.null_counts[page_idx];
+ const bool all_null = column_index.null_pages[page_idx];
+ if (page_rows < 0 || null_count < 0 || null_count > page_rows ||
+ all_null != (null_count == page_rows)) {
+ // The caller supplies the exact flat page or row-group span.
Contradictory optional null
+ // metadata must disable pruning instead of turning a partial span
into an all-null proof.
return false;
}
-
- const auto page_count = offset_index->page_locations().size();
- page_skip_plan->leaf_column_id = column_schema.leaf_column_id;
- page_skip_plan->skipped_pages.resize(page_count);
- page_skip_plan->skipped_page_compressed_sizes.resize(page_count);
- const auto& page_locations = offset_index->page_locations();
- for (size_t page_idx = 0; page_idx < page_count; ++page_idx) {
- const RowRange row_range = page_row_range(*offset_index, page_idx,
row_group_rows);
- if (row_range.length == 0 || ranges_intersect(selected_ranges,
row_range)) {
- continue;
+ page_statistics->has_null_count = true;
+ page_statistics->has_null = null_count > 0;
+ page_statistics->has_not_null = !all_null;
+ if (!page_statistics->has_not_null) {
+ return true;
+ }
+ switch (column_schema.type_descriptor.physical_type) {
+ case tparquet::Type::BOOLEAN:
+ return set_native_page_scalar_min_max<uint8_t>(column_index,
column_schema, page_idx,
Review Comment:
[P1] Decode BOOLEAN bounds with Parquet's bit packing
Parquet PLAIN BOOLEAN is bit-packed LSB-first, but this loads the entire
statistics byte and labels its address `BOOL`; NumberSerDe then reinterprets it
as a C++ `bool`. A byte such as `0x02` has a false value bit (the replaced
Arrow decoder read that bit), whereas this path creates a noncanonical `bool`
that may behave as true or otherwise invoke undefined behavior. Footer
statistics reuse this helper and ColumnIndex calls it directly, so the
misdecoded bound can drive false-negative pruning or wrong MIN/MAX. Decode bit
0 into a canonical Boolean before constructing the decoded view, and cover
footer/page-index bytes with high padding bits such as `0x02`.
##########
be/src/format_v2/parquet/parquet_statistics.cpp:
##########
@@ -109,8 +238,15 @@ Status read_decoded_field(const ParquetColumnSchema&
column_schema, DecodedColum
view.fixed_length = column_schema.type_descriptor.fixed_length;
view.timestamp_is_adjusted_to_utc =
column_schema.type_descriptor.timestamp_is_adjusted_to_utc;
view.timezone = timezone;
- return
column_schema.type->get_serde()->read_field_from_decoded_value(*column_schema.type,
-
field, view);
+ // Statistics are pruning proofs, not row materialization. A malformed
non-NULL bound must
+ // disable pruning instead of being converted to NULL under permissive
scan semantics.
+ view.enable_strict_mode = true;
+
RETURN_IF_ERROR(column_schema.type->get_serde()->read_field_from_decoded_value(
Review Comment:
[P1] Validate TIME bounds before publishing a ZoneMap
Strict mode does not validate TIME here:
`DataTypeTimeV2SerDe::read_column_from_decoded_values()` accepts any
INT32/INT64 carrier, while normal Parquet materialization rejects values
outside `[0, 24h)`. Thus a footer or ColumnIndex bound of 90,000,000
TIME_MILLIS becomes 25:00 and can make `time_col < 13:00` prune a row group
containing a valid 12:00 value; `INT64_MIN` also reaches `std::abs(INT64_MIN)`.
Reuse the raw-unit one-day-domain check before publishing min/max, and make
invalid bounds unusable for pruning/aggregation.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -0,0 +1,848 @@
+// 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/native_schema_desc.h"
+
+#include <ctype.h>
+
+#include <algorithm>
+#include <ostream>
+#include <utility>
+
+#include "common/cast_set.h"
+#include "common/exception.h"
+#include "common/logging.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_map.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/define_primitive_type.h"
+#include "util/slice.h"
+#include "util/string_util.h"
+
+namespace doris::format::parquet {
+
+static bool is_group_node(const tparquet::SchemaElement& schema) {
+ return schema.num_children > 0;
+}
+
+static bool is_list_node(const tparquet::SchemaElement& schema) {
+ // Writers may emit only the modern logical type, so collection shape must
not depend on the
+ // deprecated converted_type field being duplicated in the footer.
+ return (schema.__isset.converted_type &&
+ schema.converted_type == tparquet::ConvertedType::LIST) ||
+ (schema.__isset.logicalType && schema.logicalType.__isset.LIST);
+}
+
+static bool is_map_node(const tparquet::SchemaElement& schema) {
+ return (schema.__isset.converted_type &&
+ (schema.converted_type == tparquet::ConvertedType::MAP ||
+ schema.converted_type == tparquet::ConvertedType::MAP_KEY_VALUE))
||
+ (schema.__isset.logicalType && schema.logicalType.__isset.MAP);
+}
+
+static bool has_primitive_only_annotation(const tparquet::SchemaElement&
schema) {
+ if (schema.__isset.logicalType) {
+ const auto& logical = schema.logicalType;
+ if (logical.__isset.STRING || logical.__isset.ENUM ||
logical.__isset.DECIMAL ||
+ logical.__isset.DATE || logical.__isset.TIME ||
logical.__isset.TIMESTAMP ||
+ logical.__isset.INTEGER || logical.__isset.UNKNOWN ||
logical.__isset.JSON ||
+ logical.__isset.BSON || logical.__isset.UUID ||
logical.__isset.FLOAT16 ||
+ logical.__isset.GEOMETRY || logical.__isset.GEOGRAPHY) {
+ return true;
+ }
+ }
+ if (!schema.__isset.converted_type) {
+ return false;
+ }
+ return schema.converted_type != tparquet::ConvertedType::MAP &&
+ schema.converted_type != tparquet::ConvertedType::MAP_KEY_VALUE &&
+ schema.converted_type != tparquet::ConvertedType::LIST;
+}
+
+static bool is_repeated_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REPEATED;
+}
+
+static bool is_required_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::REQUIRED;
+}
+
+static bool is_optional_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.repetition_type &&
+ schema.repetition_type == tparquet::FieldRepetitionType::OPTIONAL;
+}
+
+static int num_children_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.num_children ? schema.num_children : 0;
+}
+
+static Status validate_native_schema_structure(
+ const std::vector<tparquet::SchemaElement>& schemas) {
+ if (schemas.empty()) {
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+
+ const auto& root = schemas[0];
+ if (root.__isset.type || !root.__isset.num_children || root.num_children <
0 ||
+ (root.__isset.repetition_type &&
+ root.repetition_type != tparquet::FieldRepetitionType::REQUIRED)) {
+ // Writers may encode the root's implicit REQUIRED repetition
explicitly, and a zero-child
+ // root is a valid metadata-only schema. Optional or repeated roots
remain ambiguous.
+ return Status::InvalidArgument("Wrong parquet root schema element");
+ }
+
+ struct PendingGroup {
+ size_t remaining_children;
+ size_t depth;
+ };
+ const auto root_children = num_children_node(schemas[0]);
+ if (root_children < 0 || static_cast<size_t>(root_children) >
schemas.size() - 1) {
+ return Status::InvalidArgument("Invalid parquet root child count {}",
root_children);
+ }
+ std::vector<PendingGroup> pending {{static_cast<size_t>(root_children),
0}};
+ for (size_t pos = 1; pos < schemas.size(); ++pos) {
+ while (!pending.empty() && pending.back().remaining_children == 0) {
+ pending.pop_back();
+ }
+ if (pending.empty()) {
+ return Status::InvalidArgument("Schema element {} is not reachable
from the root", pos);
+ }
+
+ const size_t depth = pending.back().depth + 1;
+ --pending.back().remaining_children;
+ const auto& schema = schemas[pos];
+ if (!schema.__isset.repetition_type) {
+ return Status::InvalidArgument("Schema element {} has no
repetition type", pos);
+ }
+ const bool has_children = schema.__isset.num_children &&
schema.num_children > 0;
+ if (schema.__isset.type == has_children) {
+ // Some legacy parquet-cpp files explicitly encode num_children=0
on primitive nodes.
+ // Only a positive count denotes a group, preserving strict
rejection of real dual-kind
+ // nodes without dropping those otherwise valid files.
+ return Status::InvalidArgument("Schema element {} has ambiguous
primitive/group kind",
+ pos);
+ }
+ if (schema.__isset.num_children && schema.num_children < 0) {
+ return Status::InvalidArgument("Schema element {} has a negative
child count", pos);
+ }
+ const int children = num_children_node(schemas[pos]);
+ if (children < 0 || static_cast<size_t>(children) > schemas.size() -
pos - 1) {
+ return Status::InvalidArgument("Invalid child count {} at schema
element {}", children,
+ pos);
+ }
+ if (children > 0) {
+ // Bound the tree before any resize or recursion so an untrusted
footer cannot
+ // turn a tiny schema into an unbounded allocation or parser stack.
+ if (depth > MAX_NATIVE_SCHEMA_DEPTH) {
+ return Status::InvalidArgument("Parquet schema depth {}
exceeds limit {}", depth,
+ MAX_NATIVE_SCHEMA_DEPTH);
+ }
+ pending.push_back({static_cast<size_t>(children), depth});
+ }
+ }
+ while (!pending.empty() && pending.back().remaining_children == 0) {
+ pending.pop_back();
+ }
+ if (!pending.empty()) {
+ return Status::InvalidArgument("Parquet schema ended before all
children were parsed");
+ }
+ return Status::OK();
+}
+
+/**
+ * `repeated_parent_def_level` is the definition level of the first ancestor
node whose repetition_type equals REPEATED.
+ * Empty array/map values are not stored in doris columns, so have to use
`repeated_parent_def_level` to skip the
+ * empty or null values in ancestor node.
+ *
+ * For instance, considering an array of strings with 3 rows like the
following:
+ * null, [], [a, b, c]
+ * We can store four elements in data column: null, a, b, c
+ * and the offsets column is: 1, 1, 4
+ * and the null map is: 1, 0, 0
+ * For the i-th row in array column: range from `offsets[i - 1]` until
`offsets[i]` represents the elements in this row,
+ * so we can't store empty array/map values in doris data column.
+ * As a comparison, spark does not require `repeated_parent_def_level`,
+ * because the spark column stores empty array/map values , and use anther
length column to indicate empty values.
+ * Please reference:
https://github.com/apache/spark/blob/master/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java
+ *
+ * Furthermore, we can also avoid store null array/map values in doris data
column.
+ * The same three rows as above, We can only store three elements in data
column: a, b, c
+ * and the offsets column is: 0, 0, 3
+ * and the null map is: 1, 0, 0
+ *
+ * Inherit the repetition and definition level from parent node, if the parent
node is repeated,
+ * we should set repeated_parent_def_level = definition_level, otherwise as
repeated_parent_def_level.
+ * @param parent parent node
+ * @param repeated_parent_def_level the first ancestor node whose
repetition_type equals REPEATED
+ */
+static void set_child_node_level(NativeFieldSchema* parent, int16_t
repeated_parent_def_level) {
+ for (auto& child : parent->children) {
+ child.repetition_level = parent->repetition_level;
+ child.definition_level = parent->definition_level;
+ child.repeated_parent_def_level = repeated_parent_def_level;
+ }
+}
+
+static bool is_struct_list_node(const tparquet::SchemaElement& schema,
+ const std::string& enclosing_list_name) {
+ // The legacy Parquet exception is exact: accepting every "*_tuple"
wrapper changes a standard
+ // one-child LIST wrapper from ARRAY<T> to ARRAY<STRUCT<T>>.
+ return schema.name == "array" || schema.name == enclosing_list_name +
"_tuple";
+}
+
+static bool has_logical_annotation(const tparquet::SchemaElement& schema) {
+ return schema.__isset.logicalType || schema.__isset.converted_type;
+}
+
+std::string NativeFieldSchema::debug_string() const {
+ std::stringstream ss;
+ ss << "NativeFieldSchema(name=" << name << ", R=" << repetition_level
+ << ", D=" << definition_level;
+ if (children.size() > 0) {
+ ss << ", type=" << data_type->get_name() << ", children=[";
+ for (int i = 0; i < children.size(); ++i) {
+ if (i != 0) {
+ ss << ", ";
+ }
+ ss << children[i].debug_string();
+ }
+ ss << "]";
+ } else {
+ ss << ", physical_type=" << physical_type;
+ ss << " , doris_type=" << data_type->get_name();
+ }
+ ss << ")";
+ return ss.str();
+}
+
+Status NativeFieldDescriptor::parse_from_thrift(
+ const std::vector<tparquet::SchemaElement>& t_schemas) {
+ _fields.clear();
+ _physical_fields.clear();
+ _name_to_field.clear();
+ RETURN_IF_ERROR(validate_native_schema_structure(t_schemas));
+ const auto& root_schema = t_schemas[0];
+ _fields.resize(root_schema.num_children);
+ _next_schema_pos = 1;
+
+ for (int i = 0; i < root_schema.num_children; ++i) {
+ RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos,
&_fields[i]));
+ if (_name_to_field.find(_fields[i].name) != _name_to_field.end()) {
+ return Status::InvalidArgument("Duplicated field name: {}",
_fields[i].name);
+ }
+ _name_to_field.emplace(_fields[i].name, &_fields[i]);
+ }
+
+ if (_next_schema_pos != t_schemas.size()) {
+ return Status::InvalidArgument("Remaining {} unparsed schema elements",
+ t_schemas.size() - _next_schema_pos);
+ }
+
+ return Status::OK();
+}
+
+Status NativeFieldDescriptor::parse_node_field(
+ const std::vector<tparquet::SchemaElement>& t_schemas, size_t curr_pos,
+ NativeFieldSchema* node_field) {
+ if (curr_pos >= t_schemas.size()) {
+ return Status::InvalidArgument("Out-of-bounds index of schema
elements");
+ }
+ auto& t_schema = t_schemas[curr_pos];
+ if (is_group_node(t_schema)) {
+ // nested structure or nullable list
+ return parse_group_field(t_schemas, curr_pos, node_field);
+ }
+ if (is_repeated_node(t_schema)) {
+ // repeated <primitive-type> <name> (LIST)
+ // produce required list<element>
+ node_field->repetition_level++;
+ node_field->definition_level++;
+ node_field->children.resize(1);
+ set_child_node_level(node_field, node_field->definition_level);
+ auto child = &node_field->children[0];
+ parse_physical_field(t_schema, false, child);
+
+ node_field->name = t_schema.name;
+ node_field->lower_case_name = to_lower(t_schema.name);
+ node_field->data_type =
std::make_shared<DataTypeArray>(make_nullable(child->data_type));
+ _next_schema_pos = curr_pos + 1;
+ node_field->field_id = t_schema.__isset.field_id ? t_schema.field_id :
-1;
+ } else {
+ bool is_optional = is_optional_node(t_schema);
+ if (is_optional) {
+ node_field->definition_level++;
+ }
+ parse_physical_field(t_schema, is_optional, node_field);
+ _next_schema_pos = curr_pos + 1;
+ }
+ return Status::OK();
+}
+
+void NativeFieldDescriptor::parse_physical_field(const
tparquet::SchemaElement& physical_schema,
+ bool is_nullable,
+ NativeFieldSchema*
physical_field) {
+ physical_field->name = physical_schema.name;
+ physical_field->lower_case_name = to_lower(physical_field->name);
+ physical_field->parquet_schema = physical_schema;
+ physical_field->physical_type = physical_schema.type;
+ physical_field->column_id = NATIVE_UNASSIGNED_COLUMN_ID; // Initialize
column_id
+ _physical_fields.push_back(physical_field);
+ physical_field->physical_column_index =
cast_set<int>(_physical_fields.size() - 1);
+ auto type = get_doris_type(physical_schema, is_nullable);
+ physical_field->data_type = type.first;
+ physical_field->is_type_compatibility = type.second;
+ physical_field->field_id = physical_schema.__isset.field_id ?
physical_schema.field_id : -1;
+}
+
+std::pair<DataTypePtr, bool> NativeFieldDescriptor::get_doris_type(
+ const tparquet::SchemaElement& physical_schema, bool nullable) {
+ std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(),
false};
+ try {
+ if (physical_schema.__isset.logicalType) {
+ ans = convert_to_doris_type(physical_schema.logicalType, nullable);
+ } else if (physical_schema.__isset.converted_type) {
+ ans = convert_to_doris_type(physical_schema, nullable);
+ }
+ } catch (...) {
Review Comment:
[P1] Preserve unsupported DECIMAL conversion failures
A valid `BYTE_ARRAY DECIMAL(77,s)` reaches `create_decimal()`, which throws
because Doris tops out at precision 76; this catch suppresses that failure and
the fallback below assigns STRING/VARBINARY. `fill_native_type_descriptor()`
still labels the leaf decimal, but the scalar reader selects StringSerDe from
the fallback file type and returns its two's-complement payload as raw bytes.
Parquet does not limit BYTE_ARRAY decimal precision, so preserve/record the
logical conversion failure and fail an actual projection instead of silently
changing the column's meaning. Please cover a precision-over-76 BYTE_ARRAY
decimal.
--
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]
